From fc5fc5349d8e209c3724c3ae6c73905af192ffca Mon Sep 17 00:00:00 2001 From: AdaAibaby <8872412+adababys@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:54:44 +0800 Subject: [PATCH] fix(envd): buffer subscriber channels so a ghost does not wedge the process-stream fan-out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A subscriber whose client TCP connection died after a sandbox resume (its FIN arrived after the VM snapshot) leaves its consumer goroutine wedged inside stream.Send until the kernel's retransmit timeout fires — ~924s (~15 min) at the default tcp_retries2=15. MultiplexedChannel used unbuffered subscriber channels. The fan-out loop serialises over all subscribers; at the ghost both `case s.ch <- v` (nobody reading) and `case <-s.done` (never cancelled, because the consumer never returns to run its deferred cancel) block, stalling delivery to every other subscriber for the full timeout. This is #3292: stdout/stderr goes silent for exactly ~15 min, then resumes. Fix, scoped to multiplex.go: 1. Buffered subscriber channels (subscriberBufSize = 64) absorb normal bursts past a briefly-slow consumer. 2. A non-blocking overflow path (`default:`) treats a consumer that falls a full buffer behind as a ghost and drops it: cancel() so later iterations skip it, closeCh() so its own goroutine's range exits. Both are sync.Once-guarded, so double-close is impossible. chOnce is kept separate from once so cancel (done) and closeCh (ch) stay independent. Healthy subscribers are untouched — their sends are buffered or complete immediately. Adds multiplex_ghost_test.go with three regression tests for the case the existing Abandoned* tests do not cover — a ghost that never reads AND never cancels: - GhostNeverReadingDoesNotWedgeFanOut: flood subscriberBufSize+10 past a ghost; the healthy subscriber still receives every value in order. - GhostChannelClosedAfterOverflow: after overflow the ghost channel is closed, so a ranging consumer terminates instead of leaking. - MultipleHealthyUnaffectedByOneGhost: 3 healthy subscribers all receive the full stream alongside one ghost. All three fail (producer back-pressured to timeout) against the unbuffered implementation and pass under -race with the fix. Scoped to this one fix. The original #3344 also bundled a separate stream.Send write-deadline mechanism (writedeadline.go) and connect/start wiring; that is a distinct concern and is left out here so this change stays reviewable. Closes #3292. Supersedes #3344. --- .../services/process/handler/multiplex.go | 29 +++- .../process/handler/multiplex_ghost_test.go | 159 ++++++++++++++++++ 2 files changed, 186 insertions(+), 2 deletions(-) create mode 100644 packages/envd/internal/services/process/handler/multiplex_ghost_test.go diff --git a/packages/envd/internal/services/process/handler/multiplex.go b/packages/envd/internal/services/process/handler/multiplex.go index 456d0936a6..2ac223ba6f 100644 --- a/packages/envd/internal/services/process/handler/multiplex.go +++ b/packages/envd/internal/services/process/handler/multiplex.go @@ -17,10 +17,20 @@ type MultiplexedChannel[T any] struct { exited atomic.Bool } +// subscriberBufSize is how many fan-out events a subscriber channel buffers +// before the consumer is treated as a ghost. It absorbs normal bursts past a +// briefly-slow consumer; a consumer that falls this far behind (e.g. one whose +// client TCP connection died after a resume and is stuck in stream.Send) is +// dropped so it cannot wedge the fan-out for everyone else. +const subscriberBufSize = 64 + type subscriber[T any] struct { ch chan T done chan struct{} once sync.Once + // chOnce guards closing ch; separate from once (which guards done) so cancel + // and closeCh are independent and each idempotent. + chOnce sync.Once } // cancel marks the subscriber as gone. Idempotent and non-blocking. @@ -30,6 +40,13 @@ func (s *subscriber[T]) cancel() { }) } +// closeCh closes the consumer channel so a ranging consumer exits. Idempotent. +func (s *subscriber[T]) closeCh() { + s.chOnce.Do(func() { + close(s.ch) + }) +} + // isCancelled reports whether cancel has been called. func (s *subscriber[T]) isCancelled() bool { select { @@ -67,6 +84,14 @@ func (m *MultiplexedChannel[T]) run() { select { case s.ch <- v: case <-s.done: + default: + // Buffer full and the consumer is not draining it — a ghost + // (e.g. a dead client TCP connection stuck in stream.Send). + // Drop it: cancel so later iterations skip it, and close its + // channel so its consumer goroutine exits. Healthy subscribers + // are untouched; their sends buffered or completed above. + s.cancel() + s.closeCh() } } } @@ -79,7 +104,7 @@ func (m *MultiplexedChannel[T]) run() { for _, s := range m.channels { s.cancel() - close(s.ch) + s.closeCh() } m.channels = nil } @@ -120,7 +145,7 @@ func (m *MultiplexedChannel[T]) Fork() (<-chan T, func()) { } s := &subscriber[T]{ - ch: make(chan T), + ch: make(chan T, subscriberBufSize), done: make(chan struct{}), } diff --git a/packages/envd/internal/services/process/handler/multiplex_ghost_test.go b/packages/envd/internal/services/process/handler/multiplex_ghost_test.go new file mode 100644 index 0000000000..9c16e7bae7 --- /dev/null +++ b/packages/envd/internal/services/process/handler/multiplex_ghost_test.go @@ -0,0 +1,159 @@ +package handler + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Regression tests for #3292: a ghost subscriber — one whose client TCP +// connection died after a sandbox resume so its consumer goroutine is wedged +// inside stream.Send — never reads its channel AND never calls its cancel func. +// +// This is the case the existing TestMultiplexedChannel_AbandonedConsumer* tests +// do NOT cover: those model a consumer that stops reading but then calls +// cancel(), which closes done and lets the fan-out's `case <-s.done` unblock. +// A real ghost never returns from stream.Send, so cancel never runs. On an +// unbuffered implementation the fan-out loop would then block on both +// `case s.ch <- v` (nobody reading) and `case <-s.done` (never cancelled) for +// the full TCP retransmit timeout (~15 min at tcp_retries2=15), stalling every +// other subscriber. The buffered channel plus non-blocking overflow drop is +// what prevents that; these tests fail (hang to timeout) without it. + +// forkGhost registers a subscriber whose consumer never reads and never cancels. +// It returns nothing on purpose: the ghost channel is deliberately leaked to the +// fan-out so only the overflow-drop path can reclaim it. +func forkGhost[T any](m *MultiplexedChannel[T]) { + _, _ = m.Fork() +} + +// TestMultiplexedChannel_GhostNeverReadingDoesNotWedgeFanOut is the core +// regression. A ghost is present that never reads and never cancels. We push +// subscriberBufSize+10 values; every send must stay deliverable and the one +// healthy subscriber must receive all of them well within the timeout. Without +// the overflow drop, the fan-out wedges on the ghost once its buffer fills and +// the healthy subscriber stops receiving. +func TestMultiplexedChannel_GhostNeverReadingDoesNotWedgeFanOut(t *testing.T) { + t.Parallel() + + m := NewMultiplexedChannel[int](1) + t.Cleanup(func() { close(m.Source) }) + + healthy, cancelHealthy := m.Fork() + t.Cleanup(cancelHealthy) + + healthyGot := make(chan int, subscriberBufSize+32) + go func() { + for v := range healthy { + healthyGot <- v + } + }() + + // A ghost that will never read its channel and never cancel. + forkGhost(m) + + const total = subscriberBufSize + 10 + + for i := 1; i <= total; i++ { + require.Truef(t, + sendOrTimeout(t, m.Source, i, multiplexTestTimeout), + "send %d must not be back-pressured by a ghost subscriber", i, + ) + } + + // The healthy subscriber must observe every value in order. + for i := 1; i <= total; i++ { + got, ok := recvOrTimeout(t, healthyGot, multiplexTestTimeout) + require.Truef(t, ok, "healthy subscriber should receive value %d, not stall behind the ghost", i) + assert.Equalf(t, i, got, "healthy subscriber received out-of-order value") + } +} + +// TestMultiplexedChannel_GhostChannelClosedAfterOverflow proves the overflow +// path closes the ghost's channel, so a consumer goroutine ranging over it +// exits promptly instead of leaking. We take the ghost channel ourselves but +// never drain it during the flood, forcing its buffer to fill and overflow. +func TestMultiplexedChannel_GhostChannelClosedAfterOverflow(t *testing.T) { + t.Parallel() + + m := NewMultiplexedChannel[int](1) + t.Cleanup(func() { close(m.Source) }) + + healthy, cancelHealthy := m.Fork() + t.Cleanup(cancelHealthy) + go func() { + for range healthy { //nolint:revive // drain to keep the healthy path moving + } + }() + + ghost, _ := m.Fork() + + // Flood past the buffer without reading the ghost, triggering overflow. + const total = subscriberBufSize + 5 + for i := 1; i <= total; i++ { + require.Truef(t, + sendOrTimeout(t, m.Source, i, multiplexTestTimeout), + "send %d should pass; the ghost must not back-pressure the producer", i, + ) + } + + // A consumer ranging over the ghost channel must terminate: the channel + // holds its buffered values and is then closed by the overflow drop. + drained := make(chan struct{}) + go func() { + for range ghost { //nolint:revive // we only care that the range terminates + } + close(drained) + }() + + select { + case <-drained: + case <-time.After(multiplexTestTimeout): + t.Fatal("ghost channel was not closed after overflow; a ranging consumer would leak") + } +} + +// TestMultiplexedChannel_MultipleHealthyUnaffectedByOneGhost shows that with +// several healthy subscribers and one ghost, every healthy subscriber receives +// the full stream; only the ghost is dropped. +func TestMultiplexedChannel_MultipleHealthyUnaffectedByOneGhost(t *testing.T) { + t.Parallel() + + m := NewMultiplexedChannel[int](1) + t.Cleanup(func() { close(m.Source) }) + + const healthyCount = 3 + collectors := make([]chan int, healthyCount) + for i := range collectors { + ch, cancel := m.Fork() + t.Cleanup(cancel) + + collectors[i] = make(chan int, subscriberBufSize+32) + go func(src <-chan int, sink chan int) { + for v := range src { + sink <- v + } + }(ch, collectors[i]) + } + + // One ghost among the healthy subscribers. + forkGhost(m) + + const total = subscriberBufSize + 10 + for i := 1; i <= total; i++ { + require.Truef(t, + sendOrTimeout(t, m.Source, i, multiplexTestTimeout), + "send %d must not be back-pressured by the ghost", i, + ) + } + + for c, sink := range collectors { + for i := 1; i <= total; i++ { + got, ok := recvOrTimeout(t, sink, multiplexTestTimeout) + require.Truef(t, ok, "healthy subscriber %d should receive value %d", c, i) + assert.Equalf(t, i, got, "healthy subscriber %d received out-of-order value", c) + } + } +}