amadan
host: a stage request made during a marker write gets the next window
Astra round 3 (accept with fixes): the throttle cleared busy only after the write returned, so a request made once the marker had changed, but before busy cleared, rode a write root may already have acted on, and nothing asked again. The write now runs under the lock; such a request waits for it and is deferred to the next window.
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
Holds the throttle lock through the marker write so a request during the write is deferred, not folded in.
Agent notes
Mutants: write outside the lock, window from write start, no busy coalesce, drop instead of defer; each fails a test. -race clean, count=3.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+36
−24
internal/host/agent.go+9 −6internal/host/stage_relay_test.go+27 −18
CI passed — run details
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 1fcc1e537..328c83bef 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -1235,11 +1235,10 @@ func stageRequester(dd string) func(host, ver string, err error) { |
| } |
| // 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. |
| +// 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 |
| @@ -1257,9 +1256,13 @@ func (t *stageThrottle) request(host, ver string) { |
| 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.write(host, ver) |
| t.mu.Lock() |
| + t.write(host, ver) |
| t.busy, t.last = false, t.now() |
| t.mu.Unlock() |
| } |
diff --git a/internal/host/stage_relay_test.go b/internal/host/stage_relay_test.go| index 09784e96b..ebfa82293 100644 |
| --- a/internal/host/stage_relay_test.go |
| +++ b/internal/host/stage_relay_test.go |
| @@ -145,39 +145,48 @@ func TestStageThrottleLateTimerStillCoalesces(t *testing.T) { |
| } |
| } |
| -// 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)} |
| +// 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{}) |
| - 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) }, |
| + 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) { |
| - close(started) |
| - <-release |
| mu.Lock() |
| - writes = append(writes, host+" "+ver) |
| + 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() |
| - c.t = c.t.Add(20 * time.Second) // the write stalls past a whole window |
| + now = now.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)) |
| + 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 |
| - 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) |
| + <-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) |
| } |
| } |