amadan

carlos / platform Public

host: drop the repeat pass; the stage window is 15s and runs from the end of a write

Astra's second round accepted the deferred request with fixes, and three of
its four findings were in stagePasses: a repeat pass forgets a restart that
just failed and retries it, and it adopts against config synced only before
the first pass. Rather than carry a second root-side loop with its own
ordering rules, it is gone. The window instead outlasts a root pass: 15s
against the ~8s measured, so a deferred write lands after the pass it
followed and carlos-stage.path sees it. A request written mid-pass when a
pass runs longer than the window still waits for the next trigger, as it
did before this branch; the level-triggered marker that would close that
is a unit change and is noted on #231.

The throttle now measures its window from the end of a write and stays
busy while a write is in progress, so a write that stalls on /data cannot
let the next one land beside it.
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

Removes the root-side repeat pass, widens the stage window to 15s, and measures it from the end of each write.
Agent notes
Mutants checked: window from write start, no busy coalesce, drop instead of defer, busy never set; each fails a test. -race clean on the Stage tests.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 879db237d3e376d6a31c73879e6b0d2e1537ea36 parent f5657d8
2 files changed, +69 −111
  • internal/host/agent.go +34 −70
  • internal/host/stage_relay_test.go +35 −41

CI cancelled — superseded by a newer push — run details

diff --git a/internal/host/agent.go b/internal/host/agent.go
index b350a9db8..1fcc1e537 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -1204,13 +1204,14 @@ const stageRequestMarker = "stage-requested"
// 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.
//
-// Ten 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; one written while it is still running is caught by the pass itself
-// (stagePasses), not by the path unit.
-const stageRequestEvery = 10 * time.Second
+// 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
@@ -1233,72 +1234,42 @@ func stageRequester(dd string) func(host, ver string, err error) {
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.
+// 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 or in progress
+// (busy), every request rides it: a deferred write owns the window however
+// late its timer fires, and a request made during a write is answered by
+// that write's marker change. A request inside the window schedules one
+// write for the window's end, so none waits longer than every.
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
+ 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.pending {
+ if t.busy {
t.mu.Unlock()
return
}
- now := t.now()
- if since := now.Sub(t.last); !t.last.IsZero() && since < t.every {
- t.pending = true
+ t.busy = true
+ do := func() {
+ t.write(host, ver)
+ t.mu.Lock()
+ t.busy, t.last = false, t.now()
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)
- })
+ }
+ if since := t.now().Sub(t.last); !t.last.IsZero() && since < t.every {
+ t.mu.Unlock()
+ t.after(t.every-since, do)
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
- }
- if i >= stageRepeats || stamp() == before {
- return nil
- }
- log.Printf("adopt: a stage request arrived during the pass; running another")
- }
+ do()
}
// configGenLoop is the fast-converge shape applied to issue #44: a config
@@ -3688,15 +3659,8 @@ func RunAdopt(args []string) {
if !*stageOnly {
passOpts = append(passOpts, release.NagExpiredCanaries(os.Stdout))
}
- 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)
+ if err := release.Pass(ctx, reg, store, pub, release.OptDir(), restart, passOpts...); err != nil {
+ log.Fatalf("adopt: %v", err)
}
// Step 4: the advisories. The two capability advisories that fall out of
diff --git a/internal/host/stage_relay_test.go b/internal/host/stage_relay_test.go
index b599422cd..09784e96b 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"
@@ -144,46 +145,39 @@ func TestStageThrottleLateTimerStillCoalesces(t *testing.T) {
}
}
-// 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 err := os.Chtimes(marker, time.Now(), time.Unix(int64(1754640000+i), 0)); err != nil {
- t.Fatal(err)
- }
- }
- 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)
+// A write still in progress owns the window: a request made during it rides
+// it, and the window is measured from when it finished, so a slow write to
+// /data cannot let the next one land right beside it.
+func TestStageThrottleWriteInProgressOwnsTheWindow(t *testing.T) {
+ c := &fakeClock{t: time.Unix(1754640000, 0)}
+ var mu sync.Mutex
+ var writes []string
+ release := make(chan struct{})
+ started := make(chan struct{})
+ th := &stageThrottle{every: 10 * time.Second, now: func() time.Time { mu.Lock(); defer mu.Unlock(); 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) {
+ close(started)
+ <-release
+ mu.Lock()
+ writes = append(writes, host+" "+ver)
+ mu.Unlock()
+ }}
+
+ done := make(chan struct{})
+ go func() { th.request("a", "v1"); close(done) }()
+ <-started
+ mu.Lock()
+ c.t = c.t.Add(20 * time.Second) // the write stalls past a whole window
+ mu.Unlock()
+ th.request("b", "v2")
+ if len(c.fires) != 0 {
+ t.Fatalf("a request during a write scheduled another (%d)", len(c.fires))
+ }
+ close(release)
+ <-done
+ th.request("c", "v3") // just after the stalled write finished
+ if len(writes) != 1 || len(c.fires) != 1 || c.waits[0] != 10*time.Second {
+ t.Fatalf("writes %v, deferred %d %v: want one write and the next deferred a full window from its end", writes, len(c.fires), c.waits)
}
}