From b0da40559a94b053eb6dd06251060f211696d530 Mon Sep 17 00:00:00 2001 From: Paolo Dettori Date: Mon, 7 Sep 2026 11:58:43 -0400 Subject: [PATCH 1/2] fix(worker): reserve outbound capacity for terminal frames (#173 item 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A `busy: queue full` refusal could be dropped outright, leaving its caller with no answer at all. accept runs on the recv goroutine, which must never block — an Abort queued behind a stalled dispatch is what would free the pool — so it sends through the non-blocking trySend. But every frame accept sends is TERMINAL, refusals are never cached, and outbound was shared with the chunk stream of every running exec. Chunks win by volume, so the drop correlated with the overload that produced it. The deferral comment justified this as "survivable only because the harness timeout is dual-ended", written when the relay ceiling was 120s. #182 replaced that with DEFAULT_EXEC_TIMEOUT_S = 30 minutes, so the stall it excused had grown 15x under exactly the condition that causes it. Reserved capacity, not a priority lane. outbound keeps ONE channel and gains TerminalReserve (QueueCap/4 = 16) slots; a chunkSlots semaphore holds chunk-class frames to QueueCap residency, so those last slots are reachable only from the recv goroutine. The sender releases a slot when it TAKES a non-reserved frame rather than after Send, so the accounting measures residency in the channel and does not depend on Send's latency. Frames carry the accounting they were admitted under (outFrame.reserved) instead of it being re-derived from frame type, which would break silently the day a kind is sent from both paths. A second channel with sender-side priority — the remedy the old comment named — was rejected: spec §8 requires Chunk* then End per req_id, and a priority lane lets a cache replay overtake the original's still-queued chunks, or a colliding refusal overtake an earlier exec's frames under the same id. The harness would settle the exec on whichever arrived first and discard the real output behind it. Making that safe needs per-req_id pending tracking; reserved capacity needs none, because one FIFO channel cannot reorder. Exhaustion is now a different condition, and handled as one. With chunk producers fenced off, failing to place a terminal frame means nothing is draining at all — so accept returns ErrEgressWedged, recvLoop propagates it, and Serve returns it for main.go to re-dial, with the dedup cache answering redeliveries (spec §5, §6.2). Seconds instead of half an hour. Note a dropped cache-hit replay escalates too: it looks like "a duplicate of a frame already delivered once", but the harness redelivers precisely because it never got the first answer. Also fixes a teardown wedge that predates this and was uncovered: enqueue now gives up if connCtx is done. A Send that BLOCKS (rather than fails) parks the sender, so a producer waiting for room waited forever and wg.Wait() never reached zero — Serve could then never return, which is the only thing that makes main.go re-dial. TestSendFailureDoesNotWedgeTeardown misses this because a FAILING Send lets the sender keep draining. Serve still cannot return while Send is blocked forever, since teardown ends at wgSender.Wait(); in production that is bounded by gRPC's keepalive, and the new test says so rather than implying otherwise. Three tests, each watched failing first, and each checked to discriminate: - TestBusyRefusalSurvivesAChunkBacklog — the defect itself, plus FIFO: the refusal must arrive AFTER the chunks queued before it. - TestExhaustedReserveEndsTheSession — the session ends on its own initiative; the stream is deliberately never closed. - TestBlockedSendDoesNotWedgeProducers — verified red by making the semaphore acquire uncancellable. The first test passed vacuously three times before it was made honest: a gate predicate disabled by a `> 0` guard, a saturation check that fired on a 5ms scheduling hiccup, and releasing the gate before recvLoop had consumed the flood. Saturation is now proven by chunk delivery stalling at a floor of QueueCap+1. No wire or proto change; §8 is enforced, not amended. accept gains an error return, which is not #173 item 4's plumbing refactor — that stays open. Assisted-By: Claude (Anthropic AI) Signed-off-by: Paolo Dettori --- remote-worker/internal/session/loop.go | 186 ++++++++++-- remote-worker/internal/session/loop_test.go | 304 +++++++++++++++++++- 2 files changed, 459 insertions(+), 31 deletions(-) diff --git a/remote-worker/internal/session/loop.go b/remote-worker/internal/session/loop.go index 2c1b268..7d18be3 100644 --- a/remote-worker/internal/session/loop.go +++ b/remote-worker/internal/session/loop.go @@ -15,8 +15,23 @@ import ( const ( // QueueCap bounds queued execs. Overflow is refused rather than blocking the - // recv loop — see Serve. + // recv loop — see Serve. It also bounds how many chunk-class frames may sit in + // outbound at once, which is what leaves the reserve below reachable. QueueCap = 64 + // TerminalReserve is outbound capacity that ONLY the recv goroutine may use, for + // the terminal frames it must not drop (#173 item 1). Chunk producers are held + // to QueueCap residency, so these slots are always free for a refusal. + // + // It cannot be a proof, and is not sized as if it were: accept emits one + // terminal frame per refused exec and nothing bounds how many arrive between two + // drains, so a large enough burst still exhausts it. What the reserve buys is + // that a transient chunk backlog — the common case, and the one that produced + // the bug — can no longer squeeze a refusal out. Exhaustion is then a genuinely + // different condition (nothing is draining at all) and is handled as one, by + // failing the connection rather than losing the frame. Deliberately modest for + // that reason: a big reserve would only buffer more frames behind a wedged + // sender before anyone noticed. + TerminalReserve = QueueCap / 4 // DefaultConcurrency is the pool size, advertised as Hello.capacity_max. DefaultConcurrency = 4 // DefaultHeartbeat is liveness plus NAT/proxy keepalive (spec §7 item 4). @@ -33,6 +48,26 @@ type Config struct { Heartbeat time.Duration } +// ErrEgressWedged ends a session whose terminal-frame reserve could not be +// queued. It is not "busy": chunk producers are held to QueueCap residency, so +// TerminalReserve slots are reachable only from the recv goroutine — failing to +// place one means nothing is draining at all. Serve returns it so main.go +// re-dials; the dedup cache then answers redeliveries of anything that completed +// (spec §5, §6.2). The alternative was to log the loss and keep serving a +// connection whose callers each wait out DEFAULT_EXEC_TIMEOUT_S, 30 minutes since +// #182 (#173 item 1). +var ErrEgressWedged = errors.New("egress wedged: the terminal-frame reserve could not be queued") + +// outFrame is a frame plus the accounting it was admitted under. reserved frames +// came through the terminal reserve and hold no chunkSlot, so the sender must not +// release one when it forwards them — hence carrying the fact explicitly rather +// than re-deriving it from the frame's type. Deriving it would silently break the +// accounting the day a frame kind is sent from both paths. +type outFrame struct { + frame *pb.WorkerFrame + reserved bool +} + // Stream is one Attach connection. pb.SandboxWorker_AttachClient satisfies it // directly, so production needs no adapter. type Stream interface { @@ -83,6 +118,13 @@ type slot struct { // goroutine and a bounded-blocking enqueue (correct backpressure) for everyone // else. // +// A non-blocking enqueue can only fail by DROPPING, though, and everything the +// recv goroutine sends is terminal — which is how a refusal used to vanish and +// leave its caller waiting out a 30-minute deadline (#173 item 1). So the single +// channel carries TerminalReserve slots that chunk producers cannot reach: see +// chunkSlots and trySend below. One channel, not two, because ordering is a wire +// contract (spec §8) and a priority lane would break it. +// // PRECONDITION on ctx: it MUST be the context the Attach stream was created from // (in production, the same attachCtx handed to client.Attach). recvLoop blocks in // st.Recv() and has NO select on ctx, so cancelling ctx does not by itself stop @@ -116,20 +158,40 @@ func (s *Session) Serve(ctx context.Context, st Stream) error { return fmt.Errorf("send hello: %w", err) } - outbound := make(chan *pb.WorkerFrame, QueueCap) + // ONE channel carries every frame, and that is load-bearing rather than + // incidental: spec §8 requires Chunk* then End per req_id, so a second + // "priority" channel for terminal frames would let a cache-hit replay overtake + // the original's still-queued chunks, or a colliding refusal overtake an earlier + // exec's frames under the same id. The harness would settle the exec on the + // frame that arrived first and discard the real output behind it. Ordering is + // therefore kept by construction, and the starvation problem (#173 item 1) is + // solved with RESERVED CAPACITY instead of with priority. + outbound := make(chan outFrame, QueueCap+TerminalReserve) + // chunkSlots caps how many chunk-class frames may be RESIDENT in outbound at + // once. Whatever the chunk stream does, TerminalReserve slots stay free for the + // terminal frames accept must not drop. + chunkSlots := make(chan struct{}, QueueCap) var wgSender sync.WaitGroup wgSender.Add(1) go func() { defer wgSender.Done() failed := false - for f := range outbound { + for of := range outbound { + if !of.reserved { + // Release the instant the frame LEAVES the channel, before Send rather + // than after. The semaphore counts residency in the channel, so a frame + // held in this goroutine's hand must not keep a slot — otherwise a + // blocked Send would shrink the effective chunk budget by one and, worse, + // make the reserve arithmetic depend on Send's latency. + <-chunkSlots + } if failed { // Keep draining rather than returning: if this goroutine exited early, // every later blocking enqueue below would block forever once the // buffer filled, and wg.Wait() in Serve would never reach zero. continue } - if err := st.Send(f); err != nil { + if err := st.Send(of.frame); err != nil { failed = true cancelConn() } @@ -138,31 +200,72 @@ func (s *Session) Serve(ctx context.Context, st Stream) error { // enqueue is the blocking sender used by producers (heartbeat, the pool). // Backpressure here is correct: neither is the recv goroutine, so blocking - // them cannot stall a read of an Abort frame. The sender above always keeps - // draining outbound (forwarding or discarding), so this never blocks forever - // even after a send failure. - enqueue := func(f *pb.WorkerFrame) { outbound <- f } + // them cannot stall a read of an Abort frame. + // + // Both waits give up if connCtx is done, and that is not defensive garnish. A + // Send that BLOCKS forever (rather than failing) parks the sender, outbound + // fills, and a producer waiting here would never return — so wg.Wait() in + // Serve's teardown would never reach zero and Serve could never return, which + // is the only thing that makes main.go re-dial. That hazard predates the + // reserve; the semaphore just adds a second place to hit it. Frames abandoned + // this way are not silently lost work: the connection is already dying, the + // harness re-dials, and the dedup cache answers the redelivery (spec §5, §6.2). + enqueue := func(f *pb.WorkerFrame) { + // Non-blocking attempts first, so behaviour is unchanged whenever there is + // room. select picks at random among ready cases, so without these a done + // connCtx could abandon a frame that would have fit. + select { + case chunkSlots <- struct{}{}: + default: + select { + case chunkSlots <- struct{}{}: + case <-connCtx.Done(): + return + } + } + // A held slot does NOT guarantee room here, so this wait is real rather than + // belt-and-braces: reserved frames are bounded only by the channel's own + // capacity, so a burst of refusals can fill outbound while chunk-class + // residency is low, and a producer holding a slot then finds the channel + // full. That starves chunk producers in favour of terminal frames, which is + // the intended priority — producers are exactly the ones allowed to block. + select { + case outbound <- outFrame{frame: f}: + return + default: + } + select { + case outbound <- outFrame{frame: f}: + case <-connCtx.Done(): + <-chunkSlots // hand the slot back; nothing will ever drain this frame + } + } // trySend is the non-blocking sender used by the recv goroutine (via accept). // It must never block: the recv goroutine has to stay free to read the next - // Abort, so a frame is dropped rather than stalling the stream. + // Abort, so it cannot wait for room. + // + // Every frame accept routes through here is TERMINAL: a cache-hit replay, or a + // refusal ("busy: queue full", or a req_id collision). Dropping a replay would + // only lose a duplicate of a frame already delivered once, but dropping a + // refusal loses it outright, since refusals are never cached — and the drop + // correlated with the exact overload that produced it, so the caller got + // nothing and waited out its own deadline. #182 made that deadline + // DEFAULT_EXEC_TIMEOUT_S = 30 minutes, which is what retired the old comment's + // "survivable because the harness timeout is dual-ended". // - // Be honest about the cost — the dropped frame is not advisory. Every frame - // accept routes through trySend is TERMINAL: a cache-hit replay, or a refusal - // ("busy: queue full", or a req_id collision). Dropping a replay loses a - // duplicate of a frame already delivered once. Dropping a refusal loses it - // outright, since refusals are never cached — and the drop correlates with the - // exact overload that produced the refusal, so under load the caller gets - // nothing and waits out its own deadline. That is survivable only because the - // harness timeout is dual-ended. Giving terminal frames a priority path is the - // real fix and is deliberately out of scope here; the warning below is what - // makes the loss diagnosable in the field instead of invisible. + // It no longer competes with the chunk stream for room: chunkSlots holds + // chunk-class residency to QueueCap, so the last TerminalReserve slots are + // reachable only from here. Failure here therefore no longer means "busy" — it + // means the reserve ITSELF has not drained, i.e. egress is wedged rather than + // merely behind, which the caller handles by giving up on the connection. trySend := func(f *pb.WorkerFrame) bool { select { - case outbound <- f: + case outbound <- outFrame{frame: f, reserved: true}: return true default: - log.Printf("session: WARNING outbound full, DROPPED the terminal frame for req_id %d; "+ - "that exec is now unanswered and its caller will wait out its own timeout", reqIDOf(f)) + log.Printf("session: WARNING the terminal-frame reserve (%d slots) is undrained, so the "+ + "terminal frame for req_id %d cannot be queued; treating egress as wedged and "+ + "dropping the connection so the harness re-dials", TerminalReserve, reqIDOf(f)) return false } } @@ -261,7 +364,13 @@ func (s *Session) recvLoop( } switch m := sf.Msg.(type) { case *pb.ServerFrame_Exec: - s.accept(ctx, trySend, queue, inflight, mu, m.Exec) + // accept's only error is ErrEgressWedged, and it is fatal to the + // CONNECTION rather than to the exec: returning it here ends this session + // so the caller re-dials, instead of continuing to accept work that can + // never be answered (#173 item 1). + if err := s.accept(ctx, trySend, queue, inflight, mu, m.Exec); err != nil { + return err + } case *pb.ServerFrame_Abort: // Cancel only — do NOT remove the slot. runOne owns the terminal frame in // both cases: a running exec's Run returns ErrAborted, and a queued exec @@ -275,6 +384,13 @@ func (s *Session) recvLoop( // accept decides an exec's fate without blocking: cached, queued, coalesced // into an already-running duplicate, or refused. Every send here uses trySend, // since this runs on the recv goroutine. +// +// It returns ErrEgressWedged, and only that, when a terminal frame could not be +// queued even in the reserve. Note the error describes the CONNECTION, not this +// exec — every trySend failure is treated the same way, including a dropped +// cache-hit replay. A replay looks harmless ("a duplicate of a frame already +// delivered once") but usually is not: the harness redelivers precisely because it +// never got the first answer, so dropping the replay strands that caller too. func (s *Session) accept( ctx context.Context, trySend func(*pb.WorkerFrame) bool, @@ -282,7 +398,7 @@ func (s *Session) accept( inflight map[uint64]*slot, mu *sync.Mutex, e *pb.Exec, -) { +) error { reqID := e.GetReqId() fp := Fingerprint(e.GetCommand(), e.GetStdin(), e.GetTimeoutS(), e.GetStreaming()) @@ -311,7 +427,7 @@ func (s *Session) accept( // nowhere to put it. Silently coalesce instead — the exec already owes // exactly one terminal frame, and it is coming. log.Printf("session: req_id %d already in flight; coalescing duplicate delivery", reqID) - return + return nil } // NOT a redelivery: different command+stdin under an id already in flight, // which a req_id salt collision across harness replicas still makes reachable @@ -322,8 +438,10 @@ func (s *Session) accept( // frame for the same logical exec, precisely because it is a different one. log.Printf("session: req_id %d reused for a different command while the original is still "+ "in flight; refusing it (req_id is only probabilistically unique across replicas — see spec §3.1)", reqID) - trySend(errFrame(reqID, "req_id collision: a different command is already in flight for this id")) - return + if !trySend(errFrame(reqID, "req_id collision: a different command is already in flight for this id")) { + return ErrEgressWedged + } + return nil } // Consulted before enqueue: a redelivery of a COMPLETED exec must not consume // a queue slot or a pool goroutine. Held under mu so accept's decision is @@ -333,8 +451,10 @@ func (s *Session) accept( if hit { mu.Unlock() cancel() - trySend(frame) - return + if !trySend(frame) { + return ErrEgressWedged + } + return nil } inflight[reqID] = &slot{ctx: slotCtx, cancel: cancel, fp: fp} mu.Unlock() @@ -350,8 +470,14 @@ func (s *Session) accept( delete(inflight, reqID) mu.Unlock() cancel() - trySend(errFrame(reqID, "busy: queue full")) + // The frame this item was filed about. The exec never ran and never will, so + // losing this refusal loses the caller's only answer — hence the reserve, and + // hence ending the connection if even the reserve cannot take it. + if !trySend(errFrame(reqID, "busy: queue full")) { + return ErrEgressWedged + } } + return nil } // frameSink turns runner output into Chunk frames. It only ever enqueues: with diff --git a/remote-worker/internal/session/loop_test.go b/remote-worker/internal/session/loop_test.go index 4849979..e80bbeb 100644 --- a/remote-worker/internal/session/loop_test.go +++ b/remote-worker/internal/session/loop_test.go @@ -27,6 +27,15 @@ type fakeStream struct { // failAfter, when > 0, makes Send fail once sendCalls exceeds it. failAfter int sendCalls int + // gateAfter, when > 0, makes Send BLOCK once sendCalls exceeds it, until + // releaseGate. This is a different fault from failAfter and not a variation on + // it: a FAILING Send lets the sender goroutine keep draining outbound, whereas + // a BLOCKING one stops the drain dead. Only the latter fills outbound, which is + // the precondition for a dropped terminal frame (#173 item 1) — and for the + // teardown wedge that a blocked producer causes. + gateAfter int + gate chan struct{} + gateOnce sync.Once } func newFakeStream() *fakeStream { @@ -35,15 +44,50 @@ func newFakeStream() *fakeStream { func (f *fakeStream) Send(fr *pb.WorkerFrame) error { f.mu.Lock() - defer f.mu.Unlock() f.sendCalls++ if f.failAfter > 0 && f.sendCalls > f.failAfter { + f.mu.Unlock() return errors.New("stream gone") } + // Keyed on the gate's existence, not on gateAfter > 0: gateAfter == 0 is the + // useful case ("park every Send from here on"), and a > 0 guard would silently + // disable exactly that, leaving a test green because it never gated anything. + gate, gated := f.gate, f.gate != nil && f.sendCalls > f.gateAfter + f.mu.Unlock() + + if gated { + // Deliberately NOT holding mu across the block: sent() takes it, and a test + // has to be able to inspect the wire while a Send is parked here. + <-gate + } + + f.mu.Lock() + defer f.mu.Unlock() f.out = append(f.out, fr) return nil } +// gateSendAfter makes Send succeed n times and then block until releaseGate, so a +// test can hold the drain still and let outbound fill behind it. +func (f *fakeStream) gateSendAfter(n int) { + f.mu.Lock() + defer f.mu.Unlock() + f.gateAfter = n + f.gate = make(chan struct{}) + f.sendCalls = 0 +} + +// releaseGate lets every parked and future Send through. Idempotent, so it is safe +// both as a t.Cleanup and as an explicit step mid-test. +func (f *fakeStream) releaseGate() { + f.mu.Lock() + g := f.gate + f.mu.Unlock() + if g != nil { + f.gateOnce.Do(func() { close(g) }) + } +} + func (f *fakeStream) Recv() (*pb.ServerFrame, error) { select { case fr := <-f.in: @@ -159,16 +203,31 @@ type scriptedRunner struct { // chunks, when > 0, makes Run emit that many stdout chunks via sink before // any block/ctx handling — used to drive more Sends than outbound can buffer. chunks int + // delivered counts Chunk calls that RETURNED. With the sender parked, it stops + // advancing at exactly the point outbound is full and the next enqueue blocks, + // which is how a test proves saturation instead of sleeping and hoping. + delivered int + // finished counts Run calls that RETURNED, which is the only way to observe that + // a pool goroutine escaped a blocked enqueue. + finished int } func (r *scriptedRunner) Run(ctx context.Context, s wexec.Spec, sink wexec.Sink) (int32, error) { r.mu.Lock() r.specs = append(r.specs, s) r.mu.Unlock() + defer func() { + r.mu.Lock() + r.finished++ + r.mu.Unlock() + }() for i := 0; i < r.chunks; i++ { if err := sink.Chunk(pb.Stream_STREAM_STDOUT, []byte("x")); err != nil { return -1, err } + r.mu.Lock() + r.delivered++ + r.mu.Unlock() } if r.block != nil { select { @@ -186,6 +245,18 @@ func (r *scriptedRunner) count() int { return len(r.specs) } +func (r *scriptedRunner) chunksDelivered() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.delivered +} + +func (r *scriptedRunner) runsFinished() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.finished +} + func testConfig() session.Config { return session.Config{ SandboxID: "sbx-test-1", @@ -511,6 +582,237 @@ func TestQueueOverflowIsRefusedNotBlocking(t *testing.T) { close(r.block) } +// countBusyRefusals reports how many "busy: queue full" refusals reached the wire. +func countBusyRefusals(frames []*pb.WorkerFrame) int { + n := 0 + for _, f := range frames { + if e := f.GetError(); e != nil && e.GetMessage() == "busy: queue full" { + n++ + } + } + return n +} + +// #173 item 1. The sibling test above refuses an overflow with an EMPTY outbound, +// which is the easy half. This is the half that mattered: the refusal has to +// survive a chunk backlog. +// +// Why it was dropped. accept runs on the recv goroutine, which must never block — +// an Abort queued behind a stalled dispatch is exactly what would free the pool — +// so it sends through the non-blocking trySend. But every frame accept sends is +// TERMINAL, refusals are never cached, and outbound is shared with the chunk +// stream of every running exec. So the drop correlated with the overload that +// produced it, and the caller then waited out its own deadline: since #182 made +// DEFAULT_EXEC_TIMEOUT_S 30 minutes, up to half an hour of nothing. +// +// The gate is what makes this reproducible rather than probabilistic. A blocking +// Send parks the sender, the pool goroutine fills outbound behind it, and +// chunksDelivered stops advancing at exactly the point the next enqueue blocks — +// so saturation is OBSERVED, not slept for. Only then is the queue overflowed. +func TestBusyRefusalSurvivesAChunkBacklog(t *testing.T) { + st := newFakeStream() + // Far more chunks than any buffer here can hold, so the backlog is not a + // near-miss; block keeps the worker occupied if it ever drains. + r := &scriptedRunner{chunks: 8 * session.QueueCap, block: make(chan struct{})} + cfg := testConfig() + cfg.MaxConcurrent = 1 // one worker, so every later exec has to queue + cfg.Heartbeat = time.Hour // heartbeats must not compete for the buffer + s := session.New(cfg, r) + serve(t, s, st) + waitFor(t, "hello", func() bool { return len(st.sent()) >= 1 }) + + st.gateSendAfter(0) // Hello is already out; from here nothing drains + t.Cleanup(st.releaseGate) + defer close(r.block) + + // Occupy the only worker with an exec whose output floods outbound. + st.exec(&pb.Exec{ReqId: 1, Command: "yes", Streaming: true}) + // Saturation needs BOTH conditions. "Unchanged between two polls" alone fires on + // a scheduling hiccup — the producer simply not having run for 5ms — which left + // this test green against the unfixed code because outbound still had room. The + // floor is what makes it real: the sender is parked holding one frame, so the + // channel cannot be full until QueueCap+1 chunks have been accepted. + saturated := -1 + waitFor(t, "outbound to saturate (chunk delivery to stall at a full buffer)", func() bool { + n := r.chunksDelivered() + if n >= session.QueueCap+1 && n == saturated { + return true + } + saturated = n + return false + }) + + // 1 running + QueueCap queued + 2 too many, so at least one exec is refused. + for i := uint64(2); i <= uint64(session.QueueCap+3); i++ { + st.exec(&pb.Exec{ReqId: i, Command: "sleep 30", Streaming: true}) + } + + // The refusal must be ATTEMPTED while outbound is still full — that is the whole + // scenario. st.exec only buffers into the fake's Recv channel (cap 16), so + // returning from the loop above proves nothing about what recvLoop has seen; + // releasing the gate here let the sender drain first and the refusal then sailed + // into a buffer with room, which is how this test passed against the unfixed + // code. Same sequencing discipline as TestAbortWhileQueuedStillEmitsTerminal. + waitFor(t, "recvLoop to consume the whole flood", func() bool { return st.pending() == 0 }) + time.Sleep(50 * time.Millisecond) // and for the last accept to finish its trySend + + // Only now let the wire drain. The refusal must be ON it, not dropped. + st.releaseGate() + waitFor(t, "a busy refusal on the wire despite the chunk backlog", func() bool { + return countBusyRefusals(st.sent()) >= 1 + }) + + // And it must not have jumped the queue to get there. Reserved capacity is not + // a priority lane: spec §8 requires Chunk* then End per req_id, so a terminal + // frame overtaking buffered chunks would let the harness settle an exec and + // then discard the real output that followed. + frames := st.sent() + firstRefusal := -1 + for i, f := range frames { + if e := f.GetError(); e != nil && e.GetMessage() == "busy: queue full" { + firstRefusal = i + break + } + } + chunksBefore := 0 + for _, f := range frames[:firstRefusal] { + if f.GetChunk() != nil { + chunksBefore++ + } + } + if chunksBefore < session.QueueCap { + t.Errorf("refusal arrived after only %d chunks, want >= %d: it overtook frames enqueued "+ + "before it, so egress is no longer FIFO", chunksBefore, session.QueueCap) + } +} + +// A producer parked on a full outbound must be released when the connection dies, +// even though Send is still blocked. This hazard PREDATES the reserve and nothing +// covered it: TestSendFailureDoesNotWedgeTeardown uses a Send that FAILS, and a +// failing Send lets the sender keep draining, so producers never park. A Send that +// BLOCKS stops the drain, and a producer waiting for room then waits forever — +// wg.Wait() in Serve's teardown never reaches zero. +// +// Scope, stated plainly: this pins that PRODUCERS are released. Serve itself still +// cannot return here, because teardown ends with wgSender.Wait() and the sender is +// parked inside Send — in production that is bounded by gRPC's keepalive killing +// the stream, not by anything this package does. Releasing the producers is what +// makes the escalation path reachable at all, so it is worth its own guard. +func TestBlockedSendDoesNotWedgeProducers(t *testing.T) { + st := newFakeStream() + r := &scriptedRunner{chunks: 8 * session.QueueCap, block: make(chan struct{})} + cfg := testConfig() + cfg.MaxConcurrent = 1 + cfg.Heartbeat = time.Hour + s := session.New(cfg, r) + serve(t, s, st) + waitFor(t, "hello", func() bool { return len(st.sent()) >= 1 }) + + st.gateSendAfter(0) + t.Cleanup(st.releaseGate) + defer close(r.block) + + st.exec(&pb.Exec{ReqId: 1, Command: "yes", Streaming: true}) + saturated := -1 + waitFor(t, "the producer to park on a full outbound", func() bool { + n := r.chunksDelivered() + if n >= session.QueueCap+1 && n == saturated { + return true + } + saturated = n + return false + }) + if r.runsFinished() != 0 { + t.Fatalf("runner already returned (%d): it never parked, so this proves nothing", r.runsFinished()) + } + + // Kill the connection. Send stays blocked throughout — the gate is untouched. + st.close() + + waitFor(t, "the parked producer to be released by the dying connection", func() bool { + return r.runsFinished() == 1 + }) +} + +// The reserve is a probability argument, not a proof — a burst larger than +// terminalReserve still exhausts it. What must NOT happen then is the old +// behaviour: log the loss and carry on serving a connection that cannot answer. +// Exhaustion means nothing is draining at all, which is a different condition from +// "busy", so the session gives up and lets main.go re-dial; the dedup cache is what +// makes that safe (spec §5, §6.2). +// +// On promptness, honestly: Serve can only return once teardown joins the sender, so +// a Send blocked FOREVER delays this until gRPC's own keepalive kills the stream. +// The gate is released below for exactly that reason. Escalation earns its keep in +// the reachable case — a sender that is slow rather than dead, where the reserve was +// emptied by a burst and Send does return. +func TestExhaustedReserveEndsTheSession(t *testing.T) { + st := newFakeStream() + r := &scriptedRunner{chunks: 8 * session.QueueCap, block: make(chan struct{})} + cfg := testConfig() + cfg.MaxConcurrent = 1 + cfg.Heartbeat = time.Hour + s := session.New(cfg, r) + sv := serve(t, s, st) + waitFor(t, "hello", func() bool { return len(st.sent()) >= 1 }) + + st.gateSendAfter(0) + t.Cleanup(st.releaseGate) + defer close(r.block) + + st.exec(&pb.Exec{ReqId: 1, Command: "yes", Streaming: true}) + saturated := -1 + waitFor(t, "outbound to saturate", func() bool { + n := r.chunksDelivered() + if n >= session.QueueCap+1 && n == saturated { + return true + } + saturated = n + return false + }) + + // 1 running + QueueCap queued, then comfortably more refusals than the reserve + // can hold, so exhaustion is reached rather than approached. + // + // Pushed from a goroutine, and that is required rather than tidy: escalation + // makes recvLoop RETURN, so nothing drains the fake's Recv channel afterwards + // and the tail of this flood blocks forever. Driving it from the test goroutine + // deadlocked the test against its own fix — it never reached releaseGate, so the + // parked sender was never freed and teardown could not join it. + go func() { + last := uint64(session.QueueCap + 2*session.TerminalReserve + 4) + for i := uint64(2); i <= last; i++ { + select { + case st.in <- &pb.ServerFrame{Msg: &pb.ServerFrame_Exec{Exec: &pb.Exec{ + ReqId: i, Command: "sleep 30", Streaming: true, + }}}: + case <-st.done: + return // the stream is gone; stop pushing + } + } + }() + + // Wait for escalation, observed rather than slept for. Once accept returns + // ErrEgressWedged, recvLoop returns and Serve cancels connCtx — which makes + // every remaining enqueue give up immediately, so the runner's whole chunk + // budget drains in an instant. Nothing else in this test cancels connCtx: there + // are no Send failures and the stream is never closed. + waitFor(t, "the session to give up on the connection", func() bool { + return r.chunksDelivered() >= 8*session.QueueCap + }) + + st.releaseGate() // let teardown join the parked sender + + // Note what is NOT done here: the stream is never closed. Serve must end on its + // own initiative, which is the whole point — before this, it went on serving a + // connection whose callers would each wait out a 30-minute deadline. + err := sv.wait(t, "of the terminal-frame reserve being exhausted") + if !errors.Is(err, session.ErrEgressWedged) { + t.Errorf("Serve returned %v, want ErrEgressWedged: an unanswerable connection must end, "+ + "not keep accepting work", err) + } +} + // When the stream dies, Serve returns the recv error and stops cleanly rather // than leaking its pool or heartbeat goroutines. func TestServeReturnsOnStreamError(t *testing.T) { From 26c689b39711b6e86b20b697ce65528f182d0992 Mon Sep 17 00:00:00 2001 From: Paolo Dettori Date: Mon, 7 Sep 2026 16:24:58 -0400 Subject: [PATCH 2/2] =?UTF-8?q?fix(worker):=20address=20PR=20#230=20review?= =?UTF-8?q?=20=E2=80=94=20observed=20barrier,=20guard=20the=20reserve=20co?= =?UTF-8?q?nstant?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two non-blocking review comments, both correct. 1. TestBusyRefusalSurvivesAChunkBacklog still rested on a 50ms sleep, and that sleep was the test's SOLE vacuity guard. If it expired before the last accept reached its trySend, the gate released, the sender drained, and the refusal landed in a buffer with room — the test passing against the unfixed code, which is precisely the third vacuity mode this PR already confesses to. The FIFO assertion does not backstop it: in that ordering the sender has already drained >= QueueCap chunks, so chunksBefore >= QueueCap holds and it passes too. Replaced with an observed condition, and without adding a test-only hook to the session. recvLoop is a single goroutine alternating Recv and dispatch, so ENTERING Recv for the (N+1)th time proves dispatch of the Nth frame returned. fakeStream now counts Recv entries and the test waits on that count. Every synchronisation point in this test is now observed rather than timed. The reviewer also noted settle() is this same sleep; no inline sleep remains, so that is moot here. settle()'s other uses are #173 item 9, still open. 2. TerminalReserve = QueueCap / 4 degrades SILENTLY: at QueueCap < 4 it is 0, the reserve disappears, and trySend then fails on any full buffer — turning a transient chunk backlog into a dropped connection, a worse failure than the dropped frame this PR exists to prevent. Added the suggested compile-time assertion, `var _ [TerminalReserve - 1]struct{}`, which fails the build with "invalid array length TerminalReserve - 1 (untyped int constant -1)". Verified by temporarily setting the divisor to 128. Same ethos as outFrame carrying its own accounting: make the plausible mistake impossible, not merely documented. Also re-verified the rewritten test still catches the original defect, since a green test proves nothing by itself: with the reserve removed it fails. Note it now fails AT THE BARRIER rather than at the refusal assertion, because escalation ends the session mid-flood once the reserve cannot absorb the refusals — so the barrier's message names that cause explicitly. loop.go's only functional change is the assertion. Full Go suite green, gofmt and vet clean, no temporary code left. Assisted-By: Claude (Anthropic AI) Signed-off-by: Paolo Dettori --- remote-worker/internal/session/loop.go | 12 ++++++ remote-worker/internal/session/loop_test.go | 46 +++++++++++++++++---- 2 files changed, 50 insertions(+), 8 deletions(-) diff --git a/remote-worker/internal/session/loop.go b/remote-worker/internal/session/loop.go index 7d18be3..5b5f066 100644 --- a/remote-worker/internal/session/loop.go +++ b/remote-worker/internal/session/loop.go @@ -31,6 +31,11 @@ const ( // failing the connection rather than losing the frame. Deliberately modest for // that reason: a big reserve would only buffer more frames behind a wedged // sender before anyone noticed. + // Derived from QueueCap, so it degrades badly if QueueCap is ever lowered: at + // QueueCap < 4 this is 0, the reserve vanishes, and trySend then fails on any + // full buffer — turning a transient chunk backlog into a dropped connection, + // which is a WORSE failure than the dropped frame this all exists to prevent. + // The assertion below the const block makes that a compile error instead. TerminalReserve = QueueCap / 4 // DefaultConcurrency is the pool size, advertised as Hello.capacity_max. DefaultConcurrency = 4 @@ -38,6 +43,13 @@ const ( DefaultHeartbeat = 15 * time.Second ) +// TerminalReserve must be at least 1, or the reserve silently disappears and every +// full buffer becomes a dropped connection. A negative array length is a compile +// error, so this fails at build time rather than in production. Same intent as +// outFrame carrying its own accounting: make the next maintainer's plausible +// mistake impossible rather than merely documented. +var _ [TerminalReserve - 1]struct{} + // Config is everything the session needs that does not come off the wire. type Config struct { SandboxID string diff --git a/remote-worker/internal/session/loop_test.go b/remote-worker/internal/session/loop_test.go index e80bbeb..468b97b 100644 --- a/remote-worker/internal/session/loop_test.go +++ b/remote-worker/internal/session/loop_test.go @@ -24,6 +24,8 @@ type fakeStream struct { // stream, so a test that also closes it mid-run (to observe a reconnect, or // Serve's return value) must not panic on the second close. closeOnce sync.Once + // recvCalls counts entries into Recv — see recvEntries. + recvCalls int // failAfter, when > 0, makes Send fail once sendCalls exceeds it. failAfter int sendCalls int @@ -89,6 +91,9 @@ func (f *fakeStream) releaseGate() { } func (f *fakeStream) Recv() (*pb.ServerFrame, error) { + f.mu.Lock() + f.recvCalls++ + f.mu.Unlock() select { case fr := <-f.in: return fr, nil @@ -97,6 +102,18 @@ func (f *fakeStream) Recv() (*pb.ServerFrame, error) { } } +// recvEntries reports how many times recvLoop has ENTERED Recv. That count is a +// synchronisation primitive, not a statistic: recvLoop is a single goroutine +// running Recv -> dispatch -> Recv, so entering Recv for the (N+1)th time proves +// dispatch of the Nth frame RETURNED. It turns "has accept finished with the frame +// I just pushed?" — otherwise only answerable with a sleep — into an observable +// condition, with no test-only hook in the session itself. +func (f *fakeStream) recvEntries() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.recvCalls +} + func (f *fakeStream) exec(e *pb.Exec) { f.in <- &pb.ServerFrame{Msg: &pb.ServerFrame_Exec{Exec: e}} } func (f *fakeStream) close() { f.closeOnce.Do(func() { close(f.done) }) } func (f *fakeStream) sent() []*pb.WorkerFrame { @@ -643,18 +660,31 @@ func TestBusyRefusalSurvivesAChunkBacklog(t *testing.T) { }) // 1 running + QueueCap queued + 2 too many, so at least one exec is refused. + pushed := 1 // the chunk-flooding exec above for i := uint64(2); i <= uint64(session.QueueCap+3); i++ { st.exec(&pb.Exec{ReqId: i, Command: "sleep 30", Streaming: true}) + pushed++ } - // The refusal must be ATTEMPTED while outbound is still full — that is the whole - // scenario. st.exec only buffers into the fake's Recv channel (cap 16), so - // returning from the loop above proves nothing about what recvLoop has seen; - // releasing the gate here let the sender drain first and the refusal then sailed - // into a buffer with room, which is how this test passed against the unfixed - // code. Same sequencing discipline as TestAbortWhileQueuedStillEmitsTerminal. - waitFor(t, "recvLoop to consume the whole flood", func() bool { return st.pending() == 0 }) - time.Sleep(50 * time.Millisecond) // and for the last accept to finish its trySend + // The refusal must be ATTEMPTED while outbound is still full — that IS the + // scenario, and this is the only thing standing between this test and vacuity. + // st.exec merely buffers into the fake's Recv channel, so returning from the loop + // above proves nothing about what accept has done; release the gate too early and + // the sender drains first, the refusal sails into a buffer with room, and the test + // passes against the unfixed code. The FIFO assertion below does not backstop that + // — in the racy ordering the sender has already drained >= QueueCap chunks, so it + // passes too. + // + // So this is an observed condition rather than a sleep: entering Recv for the + // (pushed+1)th time proves dispatch of the last pushed frame returned, because + // recvLoop is one goroutine alternating Recv and dispatch. + // A timeout here has one other cause worth naming, since the message is what a + // future maintainer will read: if TerminalReserve is ever lowered below the + // couple of refusals this test provokes, accept escalates instead, recvLoop + // returns, and Recv is never entered again. + waitFor(t, "accept to finish with every pushed exec (the last one refused) — or, if this "+ + "timed out, the reserve was too small to absorb them and the session escalated instead", + func() bool { return st.recvEntries() >= pushed+1 }) // Only now let the wire drain. The refusal must be ON it, not dropped. st.releaseGate()