amadan
host: a throttled stage request is deferred, not dropped; a request during a root pass runs another
Measured on the flagship's IE box on 2026-10-06 after the c2782c13 roll: one probe deploy took 23s, and the next, 41 seconds later, took 123s. The agent asked root to stage p8 at 06:00:13. When it found p9 blocked at 06:00:54, the one-request-a-minute throttle dropped the ask, and nothing asked again until the next serial move at 06:02:35. stageRequester now defers a request that lands inside its window to the end of the window, and the window is 10s instead of a minute; a root pass takes about 8s. A deferred write owns the window until it has happened, however late its timer fires, so two writes are never closer than the window. carlos-stage.path does not act on a marker change made while carlos-stage.service is running, so `carlos adopt --stage-only` now checks the marker itself after each pass and runs again if it moved (at most three more times). Replaces the first version of this branch, which also swept a blocked host in the fast loop until it was staged. Astra rejected that: it bypassed the staging backoff, never retired a superseded target, and could re-trigger a failing restart every ten seconds. Requests now come only from where they came before.
Human
Prompt
Keep investigating overnight until it's sorted.
From this commit's own message. The prompt ledger has no entries for it — which is what a squashed, rebased or amended commit looks like, and what every commit looks like in a repo without a ledger.
Summary
Stage requests inside the throttle window are deferred instead of dropped (window 60s to 10s), and the root stage pass re-runs when a request arrives while it is running.
Agent notes
Discussion #231. Throttle tests use an injected clock, no sleeps. Mutants checked: no coalescing beside an overdue timer, drop instead of defer, no repeat, unbounded repeat, blind marker stamp. carlos-stage.service's comment still says "one a minute"; left alone so the unit file does not change.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+195
−130
internal/host/agent.go+92 −73internal/host/agent_test.go+0 −25internal/host/stage_relay_test.go+103 −32
CI failed — run details
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 569fbc37c..b350a9db8 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -1202,17 +1202,15 @@ const stageRequestMarker = "stage-requested" |
| // touched only when a generation etag MOVES, which is naturally rare — a |
| // blocked stage repeats on every tick of a loop that runs every 2-3 seconds, |
| // so an untouched throttle would ask systemd to start a root pass ~30 times a |
| -// minute and each pass would re-block behind the one before it. One request a |
| -// minute is far below the hourly floor it improves on and far above the rate a |
| -// pass takes to finish. |
| +// minute and each pass would re-block behind the one before it. |
| // |
| // Ten seconds, and a request inside the window is DEFERRED to its end, never |
| -// dropped (stageRequester). It was a minute and dropped, and a second release |
| +// dropped (stageThrottle). It was a minute and dropped, and a second release |
| // inside the minute then waited for whatever next moved the serial: 98 extra |
| // seconds on the flagship on 2026-10-06 (#231). A root pass takes about eight |
| -// seconds, so ten keeps requests from piling up behind it. A var only so the |
| -// tests can shorten it. |
| -var stageRequestEvery = 10 * time.Second |
| +// seconds; one written while it is still running is caught by the pass itself |
| +// (stagePasses), not by the path unit. |
| +const stageRequestEvery = 10 * time.Second |
| // stageRequester returns a release.OnStageBlocked callback that touches the |
| // marker at most once per stageRequestEvery. The closure holds the clock, so |
| @@ -1222,41 +1220,84 @@ var stageRequestEvery = 10 * time.Second |
| // Failures are logged and swallowed, like requestRootSync's: a request that |
| // cannot be written costs latency, never correctness, because |
| // carlos-adopt.timer remains the floor underneath it. |
| -// |
| -// A request inside the window is not dropped: the first one schedules a single |
| -// write for the moment the window ends, and the rest of the window's requests |
| -// ride it. Every request is therefore answered within stageRequestEvery. |
| func stageRequester(dd string) func(host, ver string, err error) { |
| - var mu sync.Mutex |
| - var last time.Time |
| - var deferred bool |
| - write := func(host, ver string) { |
| - if err := os.WriteFile(filepath.Join(dd, stageRequestMarker), |
| - []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644); err != nil { |
| - log.Printf("agent: cannot request root stage for %s %s: %v", host, ver, err) |
| - return |
| - } |
| - log.Printf("agent: %s needs %s staged and this process cannot write the version store — asked for a root pass", host, ver) |
| - } |
| - return func(host, ver string, _ error) { |
| - mu.Lock() |
| - every := stageRequestEvery |
| - if since := time.Since(last); !last.IsZero() && since < every { |
| - if !deferred { |
| - deferred = true |
| - time.AfterFunc(every-since, func() { |
| - mu.Lock() |
| - deferred, last = false, time.Now() |
| - mu.Unlock() |
| - write(host, ver) |
| - }) |
| + t := &stageThrottle{every: stageRequestEvery, now: time.Now, after: func(d time.Duration, f func()) { time.AfterFunc(d, f) }, |
| + write: func(host, ver string) { |
| + if err := os.WriteFile(filepath.Join(dd, stageRequestMarker), |
| + []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644); err != nil { |
| + log.Printf("agent: cannot request root stage for %s %s: %v", host, ver, err) |
| + return |
| } |
| - mu.Unlock() |
| - return |
| + log.Printf("agent: %s needs %s staged and this process cannot write the version store — asked for a root pass", host, ver) |
| + }} |
| + return func(host, ver string, _ error) { t.request(host, ver) } |
| +} |
| + |
| +// stageThrottle writes at most once per every. A request inside the window |
| +// schedules one write for the window's end and every later request rides it |
| +// until that write has happened, however late its timer fires, so two writes |
| +// are never closer than every apart and no request goes unanswered for longer. |
| +type stageThrottle struct { |
| + every time.Duration |
| + now func() time.Time |
| + after func(time.Duration, func()) |
| + write func(host, ver string) |
| + mu sync.Mutex |
| + last time.Time |
| + pending bool |
| +} |
| + |
| +func (t *stageThrottle) request(host, ver string) { |
| + t.mu.Lock() |
| + if t.pending { |
| + t.mu.Unlock() |
| + return |
| + } |
| + now := t.now() |
| + if since := now.Sub(t.last); !t.last.IsZero() && since < t.every { |
| + t.pending = true |
| + t.mu.Unlock() |
| + t.after(t.every-since, func() { |
| + t.mu.Lock() |
| + t.pending, t.last = false, t.now() |
| + t.mu.Unlock() |
| + t.write(host, ver) |
| + }) |
| + return |
| + } |
| + t.last = now |
| + t.mu.Unlock() |
| + t.write(host, ver) |
| +} |
| + |
| +// stageRepeats bounds stagePasses: a pass that keeps finding a fresh request |
| +// behind it runs at most this many more times, then leaves the rest to the |
| +// next trigger. |
| +const stageRepeats = 3 |
| + |
| +// stagePasses runs pass, then again while the stage marker changed during |
| +// the pass that just ran. carlos-stage.path watches the marker with |
| +// PathChanged, and systemd does not act on a change made while |
| +// carlos-stage.service is still running; without this, a request the agent |
| +// wrote mid-pass (a second release inside one pass) would wait for the next |
| +// serial move or the hourly timer. |
| +func stagePasses(marker string, pass func() error) error { |
| + stamp := func() string { |
| + fi, err := os.Stat(marker) |
| + if err != nil { |
| + return "" |
| + } |
| + return fmt.Sprintf("%d/%d", fi.ModTime().UnixNano(), fi.Size()) |
| + } |
| + for i := 0; ; i++ { |
| + before := stamp() |
| + if err := pass(); err != nil { |
| + return err |
| } |
| - last = time.Now() |
| - mu.Unlock() |
| - write(host, ver) |
| + if i >= stageRepeats || stamp() == before { |
| + return nil |
| + } |
| + log.Printf("adopt: a stage request arrived during the pass; running another") |
| } |
| } |
| @@ -1420,7 +1461,6 @@ type fastDeps struct { |
| notBefore map[string]time.Time |
| step map[string]time.Duration |
| restartHold map[string]*restartFailure // hosts held out of the sweep after a failed restart; see noteRestartFailed |
| - stageBlocked map[string]string // host -> a version this loop could not stage; swept until it is (observeStageBlocked) |
| serialBroken bool // last ReadSerial failed and was logged; reset on success |
| } |
| @@ -1461,7 +1501,7 @@ type restartFailure struct { |
| // for: a canary is promoted to be picked up in seconds, and staging is |
| // the one step of that pickup this process cannot perform. Throttled |
| // inside stageRequester — this tick is 2-3 seconds, the request is at |
| -// most one a minute (the requester is built ONCE per loop, so the budget |
| +// most one per stageRequestEvery (the requester is built ONCE per loop, so the budget |
| // is the loop's, not one tick's). |
| // |
| // RestartHibernating IS wired here too, and for the same reason as on |
| @@ -1538,20 +1578,6 @@ func (d *fastDeps) fastMembers(now time.Time) []string { |
| members = append(members, rt.Host) |
| continue |
| } |
| - // A host whose stage this loop asked root for stays in the sweep |
| - // until it is staged, so the ask is repeated (throttled, inside |
| - // stageRequester) instead of waiting on the next serial move. That |
| - // covers a request carlos-stage.path never acted on: one written |
| - // while carlos-stage.service was still running is not a new trigger. |
| - if want, ok := d.stageBlocked[rt.Host]; ok { |
| - if v, err := release.LiveVersion(d.optDir, rt.Host); err == nil && v == want { |
| - delete(d.stageBlocked, rt.Host) |
| - } else { |
| - seen[rt.Host] = true |
| - members = append(members, rt.Host) |
| - continue |
| - } |
| - } |
| // A LiveVersion error reads as unstaged: the pass is the thing that |
| // can repair a broken link, and the backoff bounds the retry cost. |
| if v, err := release.LiveVersion(d.optDir, rt.Host); err == nil && v != "" { |
| @@ -1773,20 +1799,6 @@ func logSlow(what string, start time.Time, detail string) { |
| } |
| } |
| -// observeStageBlocked wraps the loop's stage requester so the loop remembers |
| -// which host and version it could not stage (fastMembers sweeps those until |
| -// they are). The entry ends when the host's live version is the one asked |
| -// for, or when the route leaves the registry (fastMembers sees no row). |
| -func (d *fastDeps) observeStageBlocked(req func(host, ver string, err error)) func(host, ver string, err error) { |
| - return func(host, ver string, err error) { |
| - if d.stageBlocked == nil { |
| - d.stageBlocked = map[string]string{} |
| - } |
| - d.stageBlocked[host] = ver |
| - req(host, ver, err) |
| - } |
| -} |
| - |
| // fastTick is one fast-converge tick — two independent jobs (spec §3-§4): |
| // |
| // 1. Serial check: one ReadSerial. The FIRST successful read only seeds |
| @@ -1945,7 +1957,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| restart := d.observeRestart(makeRestartRoute(reg, systemctlTryRestart, systemctlTryRestartUnit, adoptBounceFn(svc))) |
| // Built once, used by both pass shapes below — see fastPassOpts for the |
| // DO-NOT list and why the stage requester is per-loop, not per-tick. |
| - opts := fastPassOpts(dd, d.observeStageBlocked(stageRequester(dd)), restart, makeStartRoute(reg, systemctlStartUnit)) |
| + opts := fastPassOpts(dd, stageRequester(dd), restart, makeStartRoute(reg, systemctlStartUnit)) |
| // The restart check's memory (issue #113, narrowed requests added by |
| // discussion #16), seeded ONCE here — two reads per app at arming (the |
| // app-wide generation and the narrowed request log) and never again except |
| @@ -3654,7 +3666,7 @@ func RunAdopt(args []string) { |
| // |
| // -stage-only drops the nag and keeps everything else. It is what |
| // carlos-stage.service runs when the agent relays a version it could not |
| - // stage: that request arrives at most once a minute per loop, and the nag |
| + // stage: that request arrives at most once per stageRequestEvery per loop, and the nag |
| // is an HOURLY visibility feature — firing it per request would flood the |
| // journal with the same expired-canary lines, which is precisely what the |
| // option's own doc comment says never to do. The capability converge and |
| @@ -3676,8 +3688,15 @@ func RunAdopt(args []string) { |
| if !*stageOnly { |
| passOpts = append(passOpts, release.NagExpiredCanaries(os.Stdout)) |
| } |
| - if err := release.Pass(ctx, reg, store, pub, release.OptDir(), restart, passOpts...); err != nil { |
| - log.Fatalf("adopt: %v", err) |
| + pass := func() error { return release.Pass(ctx, reg, store, pub, release.OptDir(), restart, passOpts...) } |
| + var passErr error |
| + if *stageOnly { |
| + passErr = stagePasses(filepath.Join(dataDir(), stageRequestMarker), pass) |
| + } else { |
| + passErr = pass() |
| + } |
| + if passErr != nil { |
| + log.Fatalf("adopt: %v", passErr) |
| } |
| // Step 4: the advisories. The two capability advisories that fall out of |
diff --git a/internal/host/agent_test.go b/internal/host/agent_test.go| index 7edf75ba4..ca22797b1 100644 |
| --- a/internal/host/agent_test.go |
| +++ b/internal/host/agent_test.go |
| @@ -2343,28 +2343,3 @@ func TestEdgeDoorOptions(t *testing.T) { |
| t.Fatalf("the public address was read %d time(s), want once", reads) |
| } |
| } |
| - |
| -// A host whose stage this loop could not perform stays in the fast sweep |
| -// until it is staged, so the loop keeps asking for a root pass (throttled) |
| -// rather than waiting for the next serial bump. A request the path unit |
| -// missed — written while carlos-stage.service was still running the last |
| -// one — is then asked again within seconds. |
| -func TestFastSweepKeepsABlockedStageUntilItIsStaged(t *testing.T) { |
| - env := newFastFixture(t) |
| - env.addRoute(t, "edge.test", "edge") |
| - env.stageLive(t, "edge.test", "v1") |
| - d := env.deps(nil) |
| - now := time.Unix(1754640000, 0) |
| - if got := d.fastMembers(now); len(got) != 0 { |
| - t.Fatalf("members = %v before any blocked stage", got) |
| - } |
| - req := d.observeStageBlocked(func(string, string, error) {}) |
| - req("edge.test", "v2", nil) |
| - if got := d.fastMembers(now); !slices.Equal(got, []string{"edge.test"}) { |
| - t.Fatalf("members = %v after a blocked stage of v2, want edge.test", got) |
| - } |
| - env.stageLive(t, "edge.test", "v2") |
| - if got := d.fastMembers(now); len(got) != 0 { |
| - t.Fatalf("members = %v once v2 is staged, want none", got) |
| - } |
| -} |
diff --git a/internal/host/stage_relay_test.go b/internal/host/stage_relay_test.go| index f8972c20b..b599422cd 100644 |
| --- a/internal/host/stage_relay_test.go |
| +++ b/internal/host/stage_relay_test.go |
| @@ -38,9 +38,6 @@ func TestStageRequesterWritesMarker(t *testing.T) { |
| // unlike a config etag moving — would ask systemd to start a root pass ~30 |
| // times a minute, each one re-blocking behind the last. |
| func TestStageRequesterThrottles(t *testing.T) { |
| - prev := stageRequestEvery |
| - stageRequestEvery = time.Hour // the deferred write lands long after this test |
| - t.Cleanup(func() { stageRequestEvery = prev }) |
| dd := t.TempDir() |
| req := stageRequester(dd) |
| marker := filepath.Join(dd, stageRequestMarker) |
| @@ -53,8 +50,9 @@ func TestStageRequesterThrottles(t *testing.T) { |
| if err := os.Remove(marker); err != nil { |
| t.Fatal(err) |
| } |
| - // Every subsequent request inside the window must be dropped — proven by |
| - // the marker not coming back after we removed it. |
| + // No request inside the window writes at once — proven by the marker not |
| + // coming back after we removed it. (They are deferred, not dropped: |
| + // TestStageThrottleDefersToTheWindowEnd.) |
| for i := 0; i < 20; i++ { |
| req("echo.example.com", "abc1234", syscall.EROFS) |
| } |
| @@ -81,38 +79,111 @@ func TestStageRequesterBudgetIsPerRequester(t *testing.T) { |
| } |
| } |
| -// A request that lands inside the throttle window is deferred to the end of |
| -// it, never dropped. Dropped, it waited for whatever happened to bump the |
| -// serial next: on the flagship on 2026-10-06 a deploy 41 seconds after the |
| -// previous one sat unstaged for 98 seconds (#231). |
| -func TestStageRequesterDefersARequestInsideTheWindow(t *testing.T) { |
| - prev := stageRequestEvery |
| - stageRequestEvery = 100 * time.Millisecond |
| - t.Cleanup(func() { stageRequestEvery = prev }) |
| - dd := t.TempDir() |
| - marker := filepath.Join(dd, stageRequestMarker) |
| - req := stageRequester(dd) |
| +// fakeClock drives stageThrottle without sleeping: after only records the |
| +// deferred write, and the test fires it when it chooses. |
| +type fakeClock struct { |
| + t time.Time |
| + fires []func() |
| + waits []time.Duration |
| +} |
| - req("a.example.com", "v1", syscall.EROFS) |
| - if err := os.Remove(marker); err != nil { |
| - t.Fatal(err) |
| - } |
| +func (c *fakeClock) throttle(writes *[]string) *stageThrottle { |
| + return &stageThrottle{every: 10 * time.Second, now: func() time.Time { return c.t }, |
| + after: func(d time.Duration, f func()) { c.waits = append(c.waits, d); c.fires = append(c.fires, f) }, |
| + write: func(host, ver string) { *writes = append(*writes, host+" "+ver) }} |
| +} |
| + |
| +// A request inside the window is deferred to the end of it, never dropped. |
| +// Dropped, it waited for whatever happened to bump the serial next: on the |
| +// flagship on 2026-10-06 a deploy 41 seconds after the previous one sat |
| +// unstaged for 98 seconds (#231). The rest of the window's requests ride the |
| +// one deferred write. |
| +func TestStageThrottleDefersToTheWindowEnd(t *testing.T) { |
| + c := &fakeClock{t: time.Unix(1754640000, 0)} |
| + var writes []string |
| + th := c.throttle(&writes) |
| + |
| + th.request("a", "v1") |
| + c.t = c.t.Add(3 * time.Second) |
| for i := 0; i < 5; i++ { |
| - req("b.example.com", "v2", syscall.EROFS) // inside the window |
| + th.request("b", "v2") |
| } |
| - if _, err := os.Stat(marker); !errors.Is(err, os.ErrNotExist) { |
| - t.Fatal("a request inside the window was written at once; the throttle is gone") |
| + if len(writes) != 1 || len(c.fires) != 1 { |
| + t.Fatalf("writes %v, deferred %d, want one write now and one deferred", writes, len(c.fires)) |
| + } |
| + if c.waits[0] != 7*time.Second { |
| + t.Errorf("deferred by %v, want the 7s left in the window", c.waits[0]) |
| } |
| - deadline := time.Now().Add(5 * time.Second) |
| - for { |
| - if _, err := os.Stat(marker); err == nil { |
| - break |
| + c.t = c.t.Add(7 * time.Second) |
| + c.fires[0]() |
| + if len(writes) != 2 || writes[1] != "b v2" { |
| + t.Fatalf("writes %v, want the deferred request written at the window end", writes) |
| + } |
| +} |
| + |
| +// A deferred write whose timer fires late still owns the window: a request |
| +// arriving after the deadline but before the timer rides it rather than |
| +// writing at once, so two writes are never closer than the window. |
| +func TestStageThrottleLateTimerStillCoalesces(t *testing.T) { |
| + c := &fakeClock{t: time.Unix(1754640000, 0)} |
| + var writes []string |
| + th := c.throttle(&writes) |
| + |
| + th.request("a", "v1") |
| + c.t = c.t.Add(time.Second) |
| + th.request("b", "v2") // deferred to t+10s |
| + c.t = c.t.Add(30 * time.Second) |
| + th.request("c", "v3") // the timer is overdue and has not run |
| + if len(writes) != 1 || len(c.fires) != 1 { |
| + t.Fatalf("writes %v, deferred %d: a request beside an overdue timer wrote or scheduled again", writes, len(c.fires)) |
| + } |
| + c.fires[0]() |
| + th.request("d", "v4") // just after the deferred write: deferred again |
| + if len(writes) != 2 || len(c.fires) != 2 { |
| + t.Fatalf("writes %v, deferred %d, want the late write then a new deferral", writes, len(c.fires)) |
| + } |
| +} |
| + |
| +// carlos-stage.path does not act on a marker change made while |
| +// carlos-stage.service is running, so the pass checks for one itself and |
| +// runs again; it stops when the marker is still, and after stageRepeats more. |
| +func TestStagePassesRunsAgainForARequestDuringThePass(t *testing.T) { |
| + marker := filepath.Join(t.TempDir(), stageRequestMarker) |
| + touch := func(i int) { |
| + if err := os.WriteFile(marker, []byte("x"), 0o644); err != nil { |
| + t.Fatal(err) |
| } |
| - if time.Now().After(deadline) { |
| - t.Fatal("a request inside the throttle window was dropped, not deferred") |
| + if err := os.Chtimes(marker, time.Now(), time.Unix(int64(1754640000+i), 0)); err != nil { |
| + t.Fatal(err) |
| } |
| - time.Sleep(10 * time.Millisecond) |
| } |
| - // Let the deferred write's timer finish before the temp dir goes. |
| - time.Sleep(2 * stageRequestEvery) |
| + touch(0) |
| + |
| + runs := 0 |
| + if err := stagePasses(marker, func() error { |
| + runs++ |
| + if runs == 1 { |
| + touch(1) // the agent asks again mid-pass |
| + } |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + if runs != 2 { |
| + t.Errorf("ran %d passes, want 2: one asked for, one for the request that arrived during it", runs) |
| + } |
| + |
| + runs = 0 |
| + if err := stagePasses(marker, func() error { runs++; touch(100 + runs); return nil }); err != nil { |
| + t.Fatal(err) |
| + } |
| + if runs != 1+stageRepeats { |
| + t.Errorf("ran %d passes under a marker that never stops moving, want %d", runs, 1+stageRepeats) |
| + } |
| + |
| + runs = 0 |
| + boom := errors.New("boom") |
| + if err := stagePasses(marker, func() error { runs++; touch(200 + runs); return boom }); !errors.Is(err, boom) || runs != 1 { |
| + t.Errorf("err %v after %d passes, want the first pass's error and no repeat", err, runs) |
| + } |
| } |