diff --git a/remote-worker/internal/session/loop.go b/remote-worker/internal/session/loop.go index 2c1b268..5b5f066 100644 --- a/remote-worker/internal/session/loop.go +++ b/remote-worker/internal/session/loop.go @@ -15,14 +15,41 @@ 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. + // 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 // DefaultHeartbeat is liveness plus NAT/proxy keepalive (spec §7 item 4). 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 @@ -33,6 +60,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 +130,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 +170,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 +212,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 +376,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 +396,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 +410,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 +439,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 +450,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 +463,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 +482,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..468b97b 100644 --- a/remote-worker/internal/session/loop_test.go +++ b/remote-worker/internal/session/loop_test.go @@ -24,9 +24,20 @@ 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 + // 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,16 +46,54 @@ 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) { + f.mu.Lock() + f.recvCalls++ + f.mu.Unlock() select { case fr := <-f.in: return fr, nil @@ -53,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 { @@ -159,16 +220,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 +262,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 +599,250 @@ 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. + 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 + // 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() + 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) {