| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "os" |
| 6 | "path/filepath" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "syscall" |
| 10 | "testing" |
| 11 | "time" |
| 12 | ) |
| 13 | |
| 14 | // The relay exists because the agent's adopt loops run inside |
| 15 | // carlos-edge.service (ProtectSystem=strict, no /opt write): they resolve a |
| 16 | // promotion in seconds and can stage none of it, while the only process that |
| 17 | // can runs hourly. These tests pin the two halves that make the relay safe — |
| 18 | // it fires, and it does not fire too often. |
| 19 | |
| 20 | func TestStageRequesterWritesMarker(t *testing.T) { |
| 21 | dd := t.TempDir() |
| 22 | req := stageRequester(dd) |
| 23 | req("echo.example.com", "abc1234", syscall.EROFS) |
| 24 | |
| 25 | b, err := os.ReadFile(filepath.Join(dd, stageRequestMarker)) |
| 26 | if err != nil { |
| 27 | t.Fatalf("marker not written: %v", err) |
| 28 | } |
| 29 | // A timestamp, like requestRootSync's: systemd watches the path, not the |
| 30 | // bytes, but an operator reading /data should see when the last request |
| 31 | // went out. |
| 32 | if _, err := time.Parse(time.RFC3339, strings.TrimSpace(string(b))); err != nil { |
| 33 | t.Errorf("marker content %q is not an RFC3339 timestamp: %v", b, err) |
| 34 | } |
| 35 | } |
| 36 | |
| 37 | // The throttle is the whole reason this is safe to wire into a loop that ticks |
| 38 | // every 2-3 seconds. Without it a blocked stage — which repeats every tick, |
| 39 | // unlike a config etag moving — would ask systemd to start a root pass ~30 |
| 40 | // times a minute, each one re-blocking behind the last. |
| 41 | func TestStageRequesterThrottles(t *testing.T) { |
| 42 | dd := t.TempDir() |
| 43 | req := stageRequester(dd) |
| 44 | marker := filepath.Join(dd, stageRequestMarker) |
| 45 | |
| 46 | req("echo.example.com", "abc1234", syscall.EROFS) |
| 47 | first, err := os.Stat(marker) |
| 48 | if err != nil { |
| 49 | t.Fatalf("first request wrote nothing: %v", err) |
| 50 | } |
| 51 | if err := os.Remove(marker); err != nil { |
| 52 | t.Fatal(err) |
| 53 | } |
| 54 | // No request inside the window writes at once — proven by the marker not |
| 55 | // coming back after we removed it. (They are deferred, not dropped: |
| 56 | // TestStageThrottleDefersToTheWindowEnd.) |
| 57 | for i := 0; i < 20; i++ { |
| 58 | req("echo.example.com", "abc1234", syscall.EROFS) |
| 59 | } |
| 60 | if _, err := os.Stat(marker); !errors.Is(err, os.ErrNotExist) { |
| 61 | t.Errorf("throttle let a second request through inside %v (marker recreated)", stageRequestEvery) |
| 62 | } |
| 63 | _ = first |
| 64 | } |
| 65 | |
| 66 | // Each loop gets its own budget: the fast converge loop asking must not |
| 67 | // silence the hourly one. stageRequester returns a fresh closure per call, |
| 68 | // so this is a property of the API shape rather than of the clock. |
| 69 | func TestStageRequesterBudgetIsPerRequester(t *testing.T) { |
| 70 | dd := t.TempDir() |
| 71 | marker := filepath.Join(dd, stageRequestMarker) |
| 72 | |
| 73 | stageRequester(dd)("a.example.com", "v1", syscall.EROFS) |
| 74 | if err := os.Remove(marker); err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | stageRequester(dd)("b.example.com", "v1", syscall.EROFS) |
| 78 | if _, err := os.Stat(marker); err != nil { |
| 79 | t.Errorf("a second requester was throttled by the first's clock: %v", err) |
| 80 | } |
| 81 | } |
| 82 | |
| 83 | // fakeClock drives stageThrottle without sleeping: after only records the |
| 84 | // deferred write, and the test fires it when it chooses. |
| 85 | type fakeClock struct { |
| 86 | t time.Time |
| 87 | fires []func() |
| 88 | waits []time.Duration |
| 89 | } |
| 90 | |
| 91 | func (c *fakeClock) throttle(writes *[]string) *stageThrottle { |
| 92 | return &stageThrottle{every: 10 * time.Second, now: func() time.Time { return c.t }, |
| 93 | after: func(d time.Duration, f func()) { c.waits = append(c.waits, d); c.fires = append(c.fires, f) }, |
| 94 | write: func(host, ver string) { *writes = append(*writes, host+" "+ver) }} |
| 95 | } |
| 96 | |
| 97 | // A request inside the window is deferred to the end of it, never dropped. |
| 98 | // Dropped, it waited for whatever happened to bump the serial next: on the |
| 99 | // flagship on 2026-10-06 a deploy 41 seconds after the previous one sat |
| 100 | // unstaged for 98 seconds (#231). The rest of the window's requests ride the |
| 101 | // one deferred write. |
| 102 | func TestStageThrottleDefersToTheWindowEnd(t *testing.T) { |
| 103 | c := &fakeClock{t: time.Unix(1754640000, 0)} |
| 104 | var writes []string |
| 105 | th := c.throttle(&writes) |
| 106 | |
| 107 | th.request("a", "v1") |
| 108 | c.t = c.t.Add(3 * time.Second) |
| 109 | for i := 0; i < 5; i++ { |
| 110 | th.request("b", "v2") |
| 111 | } |
| 112 | if len(writes) != 1 || len(c.fires) != 1 { |
| 113 | t.Fatalf("writes %v, deferred %d, want one write now and one deferred", writes, len(c.fires)) |
| 114 | } |
| 115 | if c.waits[0] != 7*time.Second { |
| 116 | t.Errorf("deferred by %v, want the 7s left in the window", c.waits[0]) |
| 117 | } |
| 118 | c.t = c.t.Add(7 * time.Second) |
| 119 | c.fires[0]() |
| 120 | if len(writes) != 2 || writes[1] != "b v2" { |
| 121 | t.Fatalf("writes %v, want the deferred request written at the window end", writes) |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | // A deferred write whose timer fires late still owns the window: a request |
| 126 | // arriving after the deadline but before the timer rides it rather than |
| 127 | // writing at once, so two writes are never closer than the window. |
| 128 | func TestStageThrottleLateTimerStillCoalesces(t *testing.T) { |
| 129 | c := &fakeClock{t: time.Unix(1754640000, 0)} |
| 130 | var writes []string |
| 131 | th := c.throttle(&writes) |
| 132 | |
| 133 | th.request("a", "v1") |
| 134 | c.t = c.t.Add(time.Second) |
| 135 | th.request("b", "v2") // deferred to t+10s |
| 136 | c.t = c.t.Add(30 * time.Second) |
| 137 | th.request("c", "v3") // the timer is overdue and has not run |
| 138 | if len(writes) != 1 || len(c.fires) != 1 { |
| 139 | t.Fatalf("writes %v, deferred %d: a request beside an overdue timer wrote or scheduled again", writes, len(c.fires)) |
| 140 | } |
| 141 | c.fires[0]() |
| 142 | th.request("d", "v4") // just after the deferred write: deferred again |
| 143 | if len(writes) != 2 || len(c.fires) != 2 { |
| 144 | t.Fatalf("writes %v, deferred %d, want the late write then a new deferral", writes, len(c.fires)) |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | // A write in progress holds the throttle: the window is measured from when it |
| 149 | // finished, so a slow write to /data cannot let the next one land right |
| 150 | // beside it. A request made while it is under way may concern a pointer root |
| 151 | // has already read, so it is not folded into that write: it waits for it and |
| 152 | // gets the next window's. |
| 153 | func TestStageThrottleRequestDuringAWriteGetsTheNextWindow(t *testing.T) { |
| 154 | t.Parallel() |
| 155 | var mu sync.Mutex |
| 156 | now := time.Unix(1754640000, 0) |
| 157 | var writes []string |
| 158 | var waits []time.Duration |
| 159 | release := make(chan struct{}) |
| 160 | started := make(chan struct{}, 1) |
| 161 | th := &stageThrottle{every: 10 * time.Second, |
| 162 | now: func() time.Time { mu.Lock(); defer mu.Unlock(); return now }, |
| 163 | after: func(d time.Duration, f func()) { mu.Lock(); waits = append(waits, d); mu.Unlock() }, |
| 164 | write: func(host, ver string) { |
| 165 | mu.Lock() |
| 166 | writes = append(writes, host+" "+ver) // the marker has changed |
| 167 | mu.Unlock() |
| 168 | started <- struct{}{} |
| 169 | <-release |
| 170 | }} |
| 171 | |
| 172 | done := make(chan struct{}) |
| 173 | go func() { th.request("a", "v1"); close(done) }() |
| 174 | <-started |
| 175 | mu.Lock() |
| 176 | now = now.Add(20 * time.Second) // the write stalls past a whole window |
| 177 | mu.Unlock() |
| 178 | bDone := make(chan struct{}) |
| 179 | go func() { th.request("b", "v2"); close(bDone) }() |
| 180 | select { |
| 181 | case <-bDone: |
| 182 | t.Fatal("a request during a write returned before the write finished; it rode a write that cannot answer it") |
| 183 | case <-time.After(50 * time.Millisecond): |
| 184 | } |
| 185 | close(release) |
| 186 | <-done |
| 187 | <-bDone |
| 188 | mu.Lock() |
| 189 | defer mu.Unlock() |
| 190 | if len(writes) != 1 || len(waits) != 1 || waits[0] != 10*time.Second { |
| 191 | t.Fatalf("writes %v, deferred %v: want one write and b deferred a full window from its end", writes, waits) |
| 192 | } |
| 193 | } |
| 194 | |
amadan