fix(h2): queue onto the stream worker pool only against a parked worker (#520)

protocol/h2/stream/pool_starvation_test.go

@@ -0,0 +1,123 @@
+package stream
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// blockingPoolHandler blocks every stream handler until release is closed and
+// reports how many handlers actually started.
+type blockingPoolHandler struct {
+	started atomic.Int32
+	entered chan struct{}
+	release chan struct{}
+}
+
+func (h *blockingPoolHandler) HandleStream(_ context.Context, _ *Stream) error {
+	h.started.Add(1)
+	select {
+	case h.entered <- struct{}{}:
+	default:
+	}
+	<-h.release
+	return nil
+}
+
+func (h *blockingPoolHandler) RouteAsync(_, _ string) bool { return true }
+func (h *blockingPoolHandler) HasAsyncRoutes() bool        { return true }
+
+// TestH2Pool_StreamingHandlersDoNotStarveLaterStreams pins celeris#520.
+//
+// An H2 stream handler is not guaranteed to return: an SSE stream, a long
+// poll or any streaming handler holds its worker until the peer goes away.
+// The pool used to queue onto a buffered channel whenever the channel had
+// room, so once `size` such handlers were running every later stream sat in
+// the buffer behind workers that would never come back -- and because the
+// pool is process-global, that starved every H2 connection in the binary,
+// not just the one that filled it.
+//
+// The pool is deliberately tiny here: the failure needs more concurrent
+// streaming handlers than workers, which on a 4-vCPU CI runner (16 workers)
+// is 17 SSE clients and on a 32-core host is 129. Sizing the pool down is
+// what makes the bug reproducible on any machine.
+func TestH2Pool_StreamingHandlersDoNotStarveLaterStreams(t *testing.T) {
+	const size = 2
+	const streams = size * 4
+
+	p := newH2WorkerPool(size)
+	h := &blockingPoolHandler{entered: make(chan struct{}, streams), release: make(chan struct{})}
+	defer close(h.release)
+
+	var wg sync.WaitGroup
+	for i := 0; i < streams; i++ {
+		proc := NewProcessor(h, newTestFrameWriter(), newTestResponseWriter())
+		s := proc.manager.CreateStream(uint32(2*i + 1))
+		if s == nil {
+			t.Fatalf("stream %d: CreateStream returned nil", i)
+		}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			p.Submit(proc, s)
+		}()
+	}
+	wg.Wait()
+
+	deadline := time.Now().Add(5 * time.Second)
+	for h.started.Load() < int32(streams) && time.Now().Before(deadline) {
+		time.Sleep(5 * time.Millisecond)
+	}
+	if got := h.started.Load(); got != int32(streams) {
+		t.Fatalf("%d/%d streaming handlers started on a %d-worker pool: the rest are queued behind "+
+			"workers that never return, and every later H2 stream in the process is stuck with them "+
+			"(celeris#520)", got, streams, size)
+	}
+}
+
+// TestH2Pool_IdleCreditTracksParkedWorkers pins the invariant Submit relies
+// on: idle == (workers parked in the receive) - (tasks queued in work). With
+// no work submitted every worker is parked, so idle must equal the pool size;
+// it must never go negative.
+func TestH2Pool_IdleCreditTracksParkedWorkers(t *testing.T) {
+	const size = 4
+	p := newH2WorkerPool(size)
+
+	deadline := time.Now().Add(2 * time.Second)
+	for p.idle.Load() != size && time.Now().Before(deadline) {
+		time.Sleep(2 * time.Millisecond)
+	}
+	if got := p.idle.Load(); got != size {
+		t.Fatalf("idle=%d with an empty queue, want %d (every worker parked)", got, size)
+	}
+
+	// Run a burst of handlers that DO return: the credit must come back.
+	h := &blockingPoolHandler{entered: make(chan struct{}, 64), release: make(chan struct{})}
+	close(h.release) // handlers return immediately
+	for i := 0; i < 64; i++ {
+		proc := NewProcessor(h, newTestFrameWriter(), newTestResponseWriter())
+		s := proc.manager.CreateStream(uint32(2*i + 1))
+		if s == nil {
+			t.Fatalf("stream %d: CreateStream returned nil", i)
+		}
+		p.Submit(proc, s)
+		if got := p.idle.Load(); got < 0 {
+			t.Fatalf("idle went negative (%d) after %d submits: a task was queued with no parked worker", got, i+1)
+		}
+	}
+	deadline = time.Now().Add(5 * time.Second)
+	for h.started.Load() < 64 && time.Now().Before(deadline) {
+		time.Sleep(5 * time.Millisecond)
+	}
+	if got := h.started.Load(); got != 64 {
+		t.Fatalf("%d/64 returning handlers ran", got)
+	}
+	for p.idle.Load() != size && time.Now().Before(deadline) {
+		time.Sleep(5 * time.Millisecond)
+	}
+	if got := p.idle.Load(); got != size {
+		t.Fatalf("idle=%d after the burst drained, want %d: credits leak", got, size)
+	}
+}

protocol/h2/stream/processor.go

@@ -28,8 +28,26 @@ var headersSlicePoolIn = sync.Pool{New: func() any { s := make([][2]string, 0, 1
 // h2WorkerPool is a global goroutine pool for executing H2 stream handlers.
 // A single pool is shared across all connections to avoid per-connection
 // goroutine overhead (8 goroutines × N connections was catastrophic).
+//
+// A task is only ever QUEUED when a worker is parked waiting for one. That
+// rule is the whole design: an H2 handler is not guaranteed to return. A
+// Server-Sent Events stream, a long poll or any handler that streams for the
+// life of its client holds its worker until the peer goes away, so a pool
+// that queues into a buffered channel whenever the channel has room wedges as
+// soon as `size` such handlers are running -- every later stream sits in the
+// buffer behind workers that will never come back, on EVERY connection in the
+// process, because the pool is global (celeris#520). Work that finds no
+// parked worker runs on its own goroutine instead, which is what net/http
+// does for every stream unconditionally.
 type h2WorkerPool struct {
 	work chan h2Task
+	// idle is (workers parked in the receive) - (tasks sitting in work).
+	// A worker credits it before blocking on the channel; Submit claims a
+	// credit before it is allowed to queue. It is therefore positive only
+	// when a parked worker will pick the task up promptly, and drops to
+	// zero the moment the queue catches up with the parked workers -- see
+	// the invariant proof in TestH2Pool_IdleCreditTracksParkedWorkers.
+	idle atomic.Int32
 }
 
 type h2Task struct {
@@ -49,19 +67,57 @@ func newH2WorkerPool(size int) *h2WorkerPool {
 	return p
 }
 
+// run is one pool worker.
 func (p *h2WorkerPool) run() {
-	for task := range p.work {
+	for {
+		// Credit the pool before parking. Submit spends exactly one
+		// credit per queued task, so the counter stays equal to
+		// parked-workers minus queued-tasks no matter which worker
+		// ends up taking which task.
+		p.idle.Add(1)
+		task, ok := <-p.work
+		if !ok {
+			p.idle.Add(-1)
+			return
+		}
 		task.proc.executeHandler(task.stream)
 	}
 }
 
-// Submit tries to dispatch to a pooled worker. If all workers are busy
-// and the channel is full, falls back to a one-shot goroutine to avoid
-// blocking the event loop (which would stall frame processing).
+// Submit dispatches a stream handler. It queues onto the shared pool while a
+// worker is parked waiting for work, and otherwise runs the handler on its
+// own goroutine.
+//
+// It never blocks the event loop: blocking there would stall frame
+// processing for every stream on the connection.
+//
+// Growing the pool instead of spawning a one-shot goroutine was measured and
+// rejected: a surplus worker that joins the pool and retires on an idle TTL
+// made the dispatch benchmark SLOWER (865 ns/op vs 793) while adding a
+// window in which a traffic spike is held as parked goroutines. The spawn is
+// what net/http pays for every stream, unconditionally.
 func (p *h2WorkerPool) Submit(proc *Processor, s *Stream) {
+	for {
+		n := p.idle.Load()
+		if n <= 0 {
+			// No parked worker. Queueing here is what used to wedge the
+			// pool (celeris#520): a streaming handler holds its worker
+			// for the life of the stream, so a queued task would wait
+			// behind handlers that never return.
+			go proc.executeHandler(s)
+			return
+		}
+		if p.idle.CompareAndSwap(n, n-1) {
+			break
+		}
+	}
 	select {
 	case p.work <- h2Task{proc, s}:
 	default:
+		// Unreachable while the buffer is size*16 and we only queue
+		// against a parked worker, but a dropped task would hang a
+		// stream forever, so hand it a goroutine rather than trust that.
+		p.idle.Add(1)
 		go proc.executeHandler(s)
 	}
 }