amadan

carlos / platform Public

fix-stage-request-deferred

Squashed from fix-stage-request-deferred into main.
claude f3f6cf7303679fcb334fda5056c9ee4761a31cc0 parent c2782c1
2 files changed, +177 −24
  • internal/host/agent.go +62 −22
  • internal/host/stage_relay_test.go +115 −2

CI passed — run details

diff --git a/internal/host/agent.go b/internal/host/agent.go
index 9d28492b5..20a5af745 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -1190,10 +1190,16 @@ 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.
-const stageRequestEvery = time.Minute
+// minute and each pass would re-block behind the one before it.
+//
+// Fifteen seconds, and a request inside the window is DEFERRED to its end,
+// never 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, and the window has to outlast it: carlos-stage.path
+// watches with PathChanged, and systemd does not act on a change made while
+// carlos-stage.service is still running.
+const stageRequestEvery = 15 * time.Second
// stageRequester returns a release.OnStageBlocked callback that touches the
// marker at most once per stageRequestEvery. The closure holds the clock, so
@@ -1204,23 +1210,57 @@ const stageRequestEvery = time.Minute
// cannot be written costs latency, never correctness, because
// carlos-adopt.timer remains the floor underneath it.
func stageRequester(dd string) func(host, ver string, err error) {
- var mu sync.Mutex
- var last time.Time
- return func(host, ver string, _ error) {
- mu.Lock()
- if !last.IsZero() && time.Since(last) < stageRequestEvery {
- 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)
+ 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
+ }
+ 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, measured from the end of one
+// write to the start of the next. While a write is scheduled (busy), every
+// request rides it: a deferred write owns the window however late its timer
+// fires. A request inside the window schedules one write for the window's
+// end, so none waits longer than every after the write before it.
+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 // when the last write finished
+ busy bool
+}
+
+func (t *stageThrottle) request(host, ver string) {
+ t.mu.Lock()
+ if t.busy {
+ t.mu.Unlock()
+ return
+ }
+ t.busy = true
+ // The write runs under the lock: a request made once the marker has
+ // changed may concern a pointer root has already read, so it waits for
+ // the write to finish and schedules the next window, rather than
+ // riding a write that cannot answer it.
+ do := func() {
+ t.mu.Lock()
+ t.write(host, ver)
+ t.busy, t.last = false, t.now()
+ t.mu.Unlock()
+ }
+ if since := t.now().Sub(t.last); !t.last.IsZero() && since < t.every {
+ t.mu.Unlock()
+ t.after(t.every-since, do)
+ return
}
+ t.mu.Unlock()
+ do()
}
// configGenLoop is the fast-converge shape applied to issue #44: a config
@@ -1423,7 +1463,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
@@ -3577,7 +3617,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
diff --git a/internal/host/stage_relay_test.go b/internal/host/stage_relay_test.go
index 41f5d5f51..ebfa82293 100644
--- a/internal/host/stage_relay_test.go
+++ b/internal/host/stage_relay_test.go
@@ -5,6 +5,7 @@ import (
"os"
"path/filepath"
"strings"
+ "sync"
"syscall"
"testing"
"time"
@@ -50,8 +51,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)
}
@@ -77,3 +79,114 @@ func TestStageRequesterBudgetIsPerRequester(t *testing.T) {
t.Errorf("a second requester was throttled by the first's clock: %v", err)
}
}
+
+// 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
+}
+
+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++ {
+ th.request("b", "v2")
+ }
+ 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])
+ }
+ 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))
+ }
+}
+
+// A write in progress holds the throttle: the window is measured from when it
+// finished, so a slow write to /data cannot let the next one land right
+// beside it. A request made while it is under way may concern a pointer root
+// has already read, so it is not folded into that write: it waits for it and
+// gets the next window's.
+func TestStageThrottleRequestDuringAWriteGetsTheNextWindow(t *testing.T) {
+ var mu sync.Mutex
+ now := time.Unix(1754640000, 0)
+ var writes []string
+ var waits []time.Duration
+ release := make(chan struct{})
+ started := make(chan struct{}, 1)
+ th := &stageThrottle{every: 10 * time.Second,
+ now: func() time.Time { mu.Lock(); defer mu.Unlock(); return now },
+ after: func(d time.Duration, f func()) { mu.Lock(); waits = append(waits, d); mu.Unlock() },
+ write: func(host, ver string) {
+ mu.Lock()
+ writes = append(writes, host+" "+ver) // the marker has changed
+ mu.Unlock()
+ started <- struct{}{}
+ <-release
+ }}
+
+ done := make(chan struct{})
+ go func() { th.request("a", "v1"); close(done) }()
+ <-started
+ mu.Lock()
+ now = now.Add(20 * time.Second) // the write stalls past a whole window
+ mu.Unlock()
+ bDone := make(chan struct{})
+ go func() { th.request("b", "v2"); close(bDone) }()
+ select {
+ case <-bDone:
+ t.Fatal("a request during a write returned before the write finished; it rode a write that cannot answer it")
+ case <-time.After(50 * time.Millisecond):
+ }
+ close(release)
+ <-done
+ <-bDone
+ mu.Lock()
+ defer mu.Unlock()
+ if len(writes) != 1 || len(waits) != 1 || waits[0] != 10*time.Second {
+ t.Fatalf("writes %v, deferred %v: want one write and b deferred a full window from its end", writes, waits)
+ }
+}