perf(h2,overload): fix write-queue sharding + Reap GC storm + realize inline-response bypass (#406 #407 #408)

internal/conn/h2.go

@@ -104,7 +104,14 @@ type h2QueueShard struct {
 // Enqueue appends pre-encoded frame bytes to the write queue.
 // Called from handler goroutines. Shards by stream ID.
 func (q *h2ShardedQueue) Enqueue(streamID uint32, data *[]byte) {
-	shard := &q.shards[streamID%h2QueueShards]
+	// Client request streams are always odd (RFC 9113 §5.1.1) and Celeris does
+	// no server push, so streamID%h2QueueShards would only ever hit the odd
+	// residues (shards 1,3) and leave 0,2 dead — a 2-shard queue in disguise.
+	// Shift off the constant low bit first so odd IDs 1,3,5,7,… spread across
+	// ALL shards. Still a pure function of streamID → every frame of a stream
+	// maps to one shard (per-stream HEADERS-before-DATA FIFO + single-writer
+	// drain preserved). See celeris#406.
+	shard := &q.shards[(streamID>>1)%h2QueueShards]
 	shard.mu.Lock()
 	shard.bufs = append(shard.bufs, data)
 	shard.mu.Unlock()
@@ -519,6 +526,9 @@ func NewH2State(handler stream.Handler, cfg H2Config, write func([]byte), wakeup
 	// peer's SETTINGS lands.
 	s.adapter.manager = mgr
 	s.inlineAdapter.manager = mgr
+	// The inline adapter delegates incremental streaming to the queue adapter
+	// so StreamWriter()-based responses stay safe on inline routes (celeris#408).
+	s.inlineAdapter.queue = &s.adapter
 
 	p := frame.NewParser()
 	p.InitReader(&s.inBuf)

internal/conn/h2_inline_test.go

@@ -0,0 +1,82 @@
+package conn
+
+import (
+	"bytes"
+	"context"
+	"sync"
+	"testing"
+
+	"golang.org/x/net/http2"
+
+	"github.com/goceleris/celeris/protocol/h2/frame"
+	"github.com/goceleris/celeris/protocol/h2/stream"
+)
+
+// inlineStreamerHandler records the writer it was handed: whether it is the
+// direct-to-outBuf inline adapter (the #408 optimization) and whether it is a
+// stream.Streamer (SSE-safety), then writes a buffered response.
+type inlineStreamerHandler struct {
+	isInline   *bool
+	streamerOK *bool
+}
+
+func (h inlineStreamerHandler) HandleStream(_ context.Context, s *stream.Stream) error {
+	if s.ResponseWriter == nil {
+		return nil
+	}
+	_, *h.isInline = s.ResponseWriter.(*h2InlineResponseAdapter)
+	_, *h.streamerOK = s.ResponseWriter.(stream.Streamer)
+	return s.ResponseWriter.WriteResponse(s, 200, [][2]string{{"content-type", "text/plain"}}, []byte("ok"))
+}
+
+// TestH2InlineResponse_DirectOutBuf_StaysStreamer is the regression guard for
+// celeris#408. Realizing the inline-outBuf bypass must (a) actually take the
+// direct-to-outBuf path (skip the sharded queue + eventfd self-wake), and
+// (b) keep the inline ResponseWriter a stream.Streamer — otherwise
+// Context.StreamWriter() returns nil and SSE/chunked-over-H2 hard-500s.
+//
+//	old connWriter path:  streamer=true,  queue.pending=TRUE  → fails (b)-check (not optimized)
+//	naive InlineWriter:   streamer=FALSE, queue.pending=false → fails (a)-check (SSE broken)
+//	this fix:             streamer=true,  queue.pending=false → passes both
+func TestH2InlineResponse_DirectOutBuf_StaysStreamer(t *testing.T) {
+	var isInline, streamerOK bool
+	h := inlineStreamerHandler{isInline: &isInline, streamerOK: &streamerOK}
+	var mu sync.Mutex
+	var writes []byte
+	write := func(b []byte) { mu.Lock(); writes = append(writes, b...); mu.Unlock() }
+
+	state := NewH2State(h, H2Config{}, write, -1) // wakeupFD=-1: no eventfd
+
+	// Client preface + empty SETTINGS + a single inline-eligible GET (END_STREAM).
+	hdr := encodeH2Headers(t, [][2]string{
+		{":method", "GET"}, {":scheme", "http"}, {":path", "/"}, {":authority", "x"},
+	})
+	var in bytes.Buffer
+	in.WriteString(frame.ClientPreface)
+	fr := http2.NewFramer(&in, nil)
+	if err := fr.WriteSettings(); err != nil {
+		t.Fatalf("WriteSettings: %v", err)
+	}
+	if err := fr.WriteRawFrame(http2.FrameHeaders,
+		http2.FlagHeadersEndStream|http2.FlagHeadersEndHeaders, 1, hdr); err != nil {
+		t.Fatalf("WriteHeaders: %v", err)
+	}
+	if err := ProcessH2(context.Background(), in.Bytes(), state, h, write, H2Config{}); err != nil {
+		t.Fatalf("ProcessH2: %v", err)
+	}
+
+	// (a) the optimization: the inline handler must be handed the direct-to-outBuf
+	// InlineWriter, not the queue-backed connWriter (else the Enqueue + eventfd
+	// self-wake the inline path was meant to skip still fire).
+	if !isInline {
+		t.Fatal("inline handler's ResponseWriter is not the direct-outBuf InlineWriter (celeris#408 optimization not applied)")
+	}
+	// (b) SSE-safety: the inline writer MUST be a stream.Streamer, else
+	// Context.StreamWriter() returns nil and SSE/chunked-over-H2 hard-500s.
+	if !streamerOK {
+		t.Fatal("inline handler's ResponseWriter is not a stream.Streamer — Context.StreamWriter()/SSE over H2 would 500 (celeris#408)")
+	}
+	if len(writes) == 0 {
+		t.Fatal("no bytes written for the inline GET response")
+	}
+}

internal/conn/h2_sharding_test.go

@@ -0,0 +1,58 @@
+package conn
+
+import "testing"
+
+// TestH2ShardedQueue_OddStreamShardDistribution is a regression guard for
+// celeris#406. HTTP/2 client request streams are always odd (RFC 9113 §5.1.1)
+// and Celeris does no server push, so the old streamID%h2QueueShards collapsed
+// every response onto the odd residues (shards 1,3) and left 0,2 permanently
+// dead — a 4-shard queue that behaved like 2. The (streamID>>1) hash must
+// spread odd IDs across ALL shards while keeping each stream affine to one.
+func TestH2ShardedQueue_OddStreamShardDistribution(t *testing.T) {
+	var q h2ShardedQueue
+	q.wakeupFD = -1 // pure in-memory; no eventfd signaling
+
+	// Enqueue one frame per odd stream ID (1,3,5,…), like real client streams.
+	const perShard = 4
+	n := perShard * h2QueueShards
+	for i := 0; i < n; i++ {
+		q.Enqueue(uint32(2*i+1), getH2FrameBuf())
+	}
+
+	// Every shard must be exercised, evenly. Fails on the old %4 (shards 0,2
+	// get 0 bufs).
+	for i := range q.shards {
+		if got := len(q.shards[i].bufs); got != perShard {
+			t.Fatalf("shard %d got %d bufs, want %d: odd stream IDs not spread across all %d shards (celeris#406)",
+				i, got, perShard, h2QueueShards)
+		}
+	}
+
+	// Stream affinity: the same stream ID always lands in exactly one shard.
+	var q2 h2ShardedQueue
+	q2.wakeupFD = -1
+	const affID = 5
+	for i := 0; i < 3; i++ {
+		q2.Enqueue(affID, getH2FrameBuf())
+	}
+	nonEmpty := 0
+	for i := range q2.shards {
+		switch len(q2.shards[i].bufs) {
+		case 0:
+		case 3:
+			nonEmpty++
+		default:
+			t.Fatalf("stream %d split across shards: shard %d has %d bufs", affID, i, len(q2.shards[i].bufs))
+		}
+	}
+	if nonEmpty != 1 {
+		t.Fatalf("stream %d landed in %d shards, want exactly 1", affID, nonEmpty)
+	}
+
+	// DrainTo must return every buffer, in order, from the single-writer drain.
+	got := 0
+	q.DrainTo(func([]byte) { got++ })
+	if got != n {
+		t.Fatalf("DrainTo wrote %d bufs, want %d", got, n)
+	}
+}

internal/conn/response.go

@@ -518,6 +518,13 @@ type h2InlineResponseAdapter struct {
 	maxFrameSize uint32          // peer's advertised MAX_FRAME_SIZE (updated via Refresh)
 	manager      *stream.Manager // peer-SETTINGS-aware source of max frame size
 	enc          h2StreamEncoder // per-connection encoder (no sync.Pool needed for inline path)
+	// queue is this connection's sharded-queue adapter. Incremental streaming
+	// (the Streamer methods: WriteHeader/Write/Flush/Close) MUST delegate to it
+	// — a streaming response (SSE, chunked) may be produced by a detached/async
+	// goroutine that cannot write outBuf directly (outBuf is event-loop-thread +
+	// H2State.mu only). Only WriteResponse — the buffered inline GET/HEAD fast
+	// path, always on the event loop — writes straight to outBuf. See celeris#408.
+	queue *h2ResponseAdapter
 }
 
 // peerMaxFrame returns the peer's currently-advertised SETTINGS_MAX_FRAME_SIZE.
@@ -629,6 +636,31 @@ func (a *h2InlineResponseAdapter) WriteRSTStreamPriority(_ uint32, _ http2.ErrCo
 
 func (a *h2InlineResponseAdapter) CloseConn() error { return nil }
 
+// h2InlineResponseAdapter implements stream.Streamer by delegating incremental
+// streaming to the sharded write queue (a.queue). This is what keeps
+// StreamWriter()-based responses (SSE, chunked) working when an inline-eligible
+// route streams: Context.StreamWriter() type-asserts ResponseWriter.(Streamer),
+// so the inline writer MUST be a Streamer or streaming 500s. The queue path is
+// safe from a detached/async streaming goroutine; direct-outBuf (WriteResponse)
+// is not. See celeris#408.
+func (a *h2InlineResponseAdapter) WriteHeader(s *stream.Stream, status int, headers [][2]string) error {
+	return a.queue.WriteHeader(s, status, headers)
+}
+
+func (a *h2InlineResponseAdapter) Write(s *stream.Stream, data []byte) error {
+	return a.queue.Write(s, data)
+}
+
+func (a *h2InlineResponseAdapter) Flush(s *stream.Stream) error {
+	return a.queue.Flush(s)
+}
+
+func (a *h2InlineResponseAdapter) Close(s *stream.Stream) error {
+	return a.queue.Close(s)
+}
+
+var _ stream.Streamer = (*h2InlineResponseAdapter)(nil)
+
 type h2ResponseAdapter struct {
 	write        func([]byte)
 	outBuf       *bytes.Buffer

middleware/overload/config.go

@@ -158,8 +158,10 @@ type Config struct {
 	// can measure the effect.
 	EnableReap bool
 
-	// ReapAggressiveness: 1=GC hint, 2=GC + encourage pool drain.
-	// Default: 1.
+	// ReapAggressiveness gates the opt-in GC fired on entry into StageReap:
+	// any value >= 1 triggers a one-shot runtime.GC(). Level 2 ("GC + encourage
+	// pool drain") is reserved and currently behaves identically to level 1 —
+	// no pool-drain primitive is wired yet (celeris#407). Default: 1.
 	ReapAggressiveness int
 
 	// Skip defines a function to skip this middleware for certain

middleware/overload/overload.go

@@ -231,6 +231,12 @@ func run(ctx context.Context, cfg Config,
 	defer close(stopped)
 	t := time.NewTicker(cfg.PollInterval)
 	defer t.Stop()
+	// prevStage tracks the stage applied on the previous tick so the opt-in
+	// Reap GC fires only on the transition INTO Reap, not on every tick the
+	// stage lingers in the band (celeris#407). Goroutine-local: run() is the
+	// sole writer of the `stage` atomic and the sole owner of prevStage, so
+	// this does not affect the stage value the hot path reads.
+	prevStage := StageNormal
 	for {
 		select {
 		case <-ctx.Done():
@@ -256,11 +262,19 @@ func run(ctx context.Context, cfg Config,
 				newStage = latencyStage
 			}
 			stage.Store(int32(newStage))
-			if newStage == StageReap && cfg.EnableReap {
+			// Fire the opt-in GC only on ENTRY into Reap. A forced GC every
+			// PollInterval while the stage lingers in the Reap band (celeris#407)
+			// wrecks tail latency and defeats the stage's purpose — a Reap GC is
+			// a one-shot capacity reclaim, not a periodic sweep. Re-entry
+			// (Reap→Reorder→Reap) correctly re-fires. The col==nil / cpu<0 skip
+			// branches above `continue` before both stage.Store and this
+			// assignment, so a skipped tick advances neither.
+			if cfg.EnableReap && newStage == StageReap && prevStage != StageReap {
 				if cfg.ReapAggressiveness >= 1 {
 					runtime.GC()
 				}
 			}
+			prevStage = newStage
 			_ = inFlight // surfaced through Controller; not needed in poll
 		}
 	}

middleware/overload/overload_reap_test.go

@@ -0,0 +1,52 @@
+package overload
+
+import (
+	"runtime"
+	"testing"
+	"time"
+)
+
+// TestReapGCFiresOnceOnEntryNotPerTick is a regression guard for celeris#407:
+// the opt-in Reap GC must fire on the transition INTO Reap, not on every poll
+// tick the stage lingers in the Reap band. Pre-fix this forced a GC (STW) every
+// PollInterval under sustained moderate load.
+func TestReapGCFiresOnceOnEntryNotPerTick(t *testing.T) {
+	var ms runtime.MemStats
+	runtime.ReadMemStats(&ms)
+	before := ms.NumForcedGC
+
+	m := &mockCPU{}
+	m.set(0.82) // in the Reap band [Reap=0.80, Reorder=0.85)
+	poll := 5 * time.Millisecond
+	_, ctrl := NewWithController(Config{
+		CollectorProvider:  withMock(m),
+		PollInterval:       poll,
+		EnableReap:         true,
+		ReapAggressiveness: 1,
+	})
+	defer ctrl.Stop()
+
+	waitForStage(t, ctrl, StageReap)
+	// waitForStage returns the instant stage.Store publishes StageReap, but the
+	// entry runtime.GC() runs after that in the same tick — wait a few ticks so
+	// it (and only it) has completed before sampling.
+	time.Sleep(3 * poll)
+
+	// Entering Reap must fire the opt-in GC exactly once.
+	runtime.ReadMemStats(&ms)
+	atEntry := ms.NumForcedGC
+	if atEntry <= before {
+		t.Fatalf("entering Reap did not fire the opt-in GC (NumForcedGC %d -> %d)", before, atEntry)
+	}
+
+	// Linger in the Reap band for ~30 poll intervals. Pre-fix (celeris#407)
+	// forces a GC every tick (~30 more); post-fix fires 0 more.
+	time.Sleep(30 * poll)
+	if ctrl.Stage() != StageReap {
+		t.Fatalf("expected to stay in Reap, got %s", ctrl.Stage())
+	}
+	runtime.ReadMemStats(&ms)
+	if lingering := ms.NumForcedGC - atEntry; lingering > 1 {
+		t.Fatalf("Reap GC fired %d extra times while lingering in the band; want <=1 (celeris#407: GC only on entry)", lingering)
+	}
+}

protocol/h2/stream/processor.go

@@ -557,12 +557,22 @@ func (p *Processor) runHandler(stream *Stream) {
 }
 
 // executeHandlerInline runs the handler synchronously on the event loop.
-// Response writes go directly to outBuf (via connWriter) instead of the
-// async write queue, eliminating goroutine dispatch, shard mutex ops,
-// eventfd syscall, and drain loop overhead.
+// Because it runs ON the event-loop thread (under H2State.mu), a buffered
+// response (WriteResponse — the common inline GET/HEAD END_STREAM path) goes
+// straight to outBuf via InlineWriter, skipping the sharded write queue's mutex
+// + pooled buffer and — the real win — the eventfd self-wake syscall Enqueue
+// fires on every response. Incremental streaming (StreamWriter) still routes
+// through the queue: InlineWriter is a stream.Streamer that delegates its
+// streaming methods to connWriter, so SSE/chunked stays safe (celeris#408).
+// Falls back to connWriter when InlineWriter is unset (Processors built without
+// the conn layer, e.g. unit tests).
 func (p *Processor) executeHandlerInline(stream *Stream) {
 	p.InlineCount++
-	stream.ResponseWriter = p.connWriter
+	if p.InlineWriter != nil {
+		stream.ResponseWriter = p.InlineWriter
+	} else {
+		stream.ResponseWriter = p.connWriter
+	}
 
 	if p.InlineCachedCtx != nil {
 		stream.CachedCtx = p.InlineCachedCtx