amadan
host: a throttled stage request is deferred, not dropped; a blocked stage stays in the fast sweep
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. A staged edge route is not a fast-sweep member, so no tick revisited it. 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. The fast loop also remembers each host and version whose stage it could not perform, and sweeps that host until its live version is the one asked for. That covers a request carlos-stage.path does not act on because it was written while carlos-stage.service was still running the previous one.
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 a blocked stage keeps its host in the fast sweep until it is staged.
Agent notes
Discussion #231. Mutants checked: dropping instead of deferring, and not sweeping a blocked host; each fails its test. The eight internal/host tests that fail in the sandbox are socket tests and fail on main too.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+127
−9
internal/host/agent.go+63 −9internal/host/agent_test.go+25 −0internal/host/stage_relay_test.go+39 −0
CI passed — run details
diff --git a/internal/host/agent.go b/internal/host/agent.go| index b38bf0ae4..569fbc37c 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -1205,7 +1205,14 @@ const stageRequestMarker = "stage-requested" |
| // 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. |
| -const stageRequestEvery = time.Minute |
| +// |
| +// 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 |
| +// 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 |
| // stageRequester returns a release.OnStageBlocked callback that touches the |
| // marker at most once per stageRequestEvery. The closure holds the clock, so |
| @@ -1215,23 +1222,41 @@ const stageRequestEvery = time.Minute |
| // 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() |
| - if !last.IsZero() && time.Since(last) < stageRequestEvery { |
| + 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) |
| + }) |
| + } |
| mu.Unlock() |
| return |
| } |
| last = time.Now() |
| mu.Unlock() |
| - 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) |
| + write(host, ver) |
| } |
| } |
| @@ -1395,6 +1420,7 @@ 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 |
| } |
| @@ -1512,6 +1538,20 @@ 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 != "" { |
| @@ -1733,6 +1773,20 @@ 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 |
| @@ -1891,7 +1945,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, stageRequester(dd), restart, makeStartRoute(reg, systemctlStartUnit)) |
| + opts := fastPassOpts(dd, d.observeStageBlocked(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 |
diff --git a/internal/host/agent_test.go b/internal/host/agent_test.go| index ca22797b1..7edf75ba4 100644 |
| --- a/internal/host/agent_test.go |
| +++ b/internal/host/agent_test.go |
| @@ -2343,3 +2343,28 @@ 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 41f5d5f51..f8972c20b 100644 |
| --- a/internal/host/stage_relay_test.go |
| +++ b/internal/host/stage_relay_test.go |
| @@ -38,6 +38,9 @@ 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) |
| @@ -77,3 +80,39 @@ func TestStageRequesterBudgetIsPerRequester(t *testing.T) { |
| t.Errorf("a second requester was throttled by the first's clock: %v", err) |
| } |
| } |
| + |
| +// 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) |
| + |
| + req("a.example.com", "v1", syscall.EROFS) |
| + if err := os.Remove(marker); err != nil { |
| + t.Fatal(err) |
| + } |
| + for i := 0; i < 5; i++ { |
| + req("b.example.com", "v2", syscall.EROFS) // inside the window |
| + } |
| + 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") |
| + } |
| + deadline := time.Now().Add(5 * time.Second) |
| + for { |
| + if _, err := os.Stat(marker); err == nil { |
| + break |
| + } |
| + if time.Now().After(deadline) { |
| + t.Fatal("a request inside the throttle window was dropped, not deferred") |
| + } |
| + time.Sleep(10 * time.Millisecond) |
| + } |
| + // Let the deferred write's timer finish before the temp dir goes. |
| + time.Sleep(2 * stageRequestEvery) |
| +} |