From b3efdab0e7caf5593cb4ae6a1cbf0aa2e0d38e00 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 21:07:27 +0100
Subject: [PATCH 01/12] epoll: test that one drainRead call on a conn whose
 peer keeps its socket full returns after a bounded read (fails on main;
 v1.6.0, Refs #885 item 41)

---
 .../epoll/drainread_flood_885_linux_test.go   | 348 ++++++++++++++++++
 1 file changed, 348 insertions(+)
 create mode 100644 internal/engine/epoll/drainread_flood_885_linux_test.go

diff --git a/internal/engine/epoll/drainread_flood_885_linux_test.go b/internal/engine/epoll/drainread_flood_885_linux_test.go
new file mode 100644
index 000000000..2b39285cf
--- /dev/null
+++ b/internal/engine/epoll/drainread_flood_885_linux_test.go
@@ -0,0 +1,348 @@
+//go:build linux
+
+package epoll
+
+// celeris#885 item 41: drainRead reads an HTTP conn until a read returns less
+// than the buffer holds, with no bound on the number of reads, so a client
+// that keeps its socket full holds the loop, and the conns on it wait.
+//
+// Two measurements. TestDrainReadIsBoundedUnderAFlood885 is the gate: one
+// drainRead call on a conn whose peer never lets the socket empty must return
+// after a bounded number of bytes (deterministic: it counts bytes, not time).
+// TestProbeDrainReadFloodLatency885 is the end-to-end probe behind it, run by
+// hand (CELERIS_PROBE_885=1): conn A pipelines requests as fast as its socket
+// takes them, conn B on A's loop sends one request every few milliseconds and
+// times the answer, and conn C on the other loop is the control (a stall that
+// C shares is CPU contention of the test process, not the loop).
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+
+	"github.com/goceleris/celeris/internal/engine"
+	"github.com/goceleris/celeris/internal/protocol/h2/stream"
+	"github.com/goceleris/celeris/internal/resource"
+)
+
+// floodHandler885 answers every request with 200; /who carries the id of the
+// worker (loop) that owns the conn in its body.
+type floodHandler885 struct{}
+
+func (floodHandler885) HandleStream(_ context.Context, s *stream.Stream) error {
+	if s.ResponseWriter == nil {
+		return nil
+	}
+	body := []byte("ok")
+	if s.Path == "/who" {
+		body = []byte(strconv.Itoa(int(s.WorkerID)))
+	}
+	return s.ResponseWriter.WriteResponse(s, 200, [][2]string{{"content-type", "text/plain"}}, body)
+}
+
+func startFloodEngine885(t *testing.T, bufSize int) *Engine {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("pick port: %v", err)
+	}
+	addr := ln.Addr().String()
+	_ = ln.Close()
+	cfg := resource.Config{
+		Addr:     addr,
+		Protocol: engine.HTTP1,
+		Resources: resource.Resources{
+			Workers:    resource.MinWorkers,
+			BufferSize: bufSize,
+		},
+	}
+	e, err := New(cfg, floodHandler885{})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	ctx, cancel := context.WithCancel(t.Context())
+	errCh := make(chan error, 1)
+	go func() { errCh <- e.Listen(ctx) }()
+	t.Cleanup(func() { cancel(); <-errCh })
+	for dl := time.Now().Add(5 * time.Second); time.Now().Before(dl) && e.Addr() == nil; {
+		time.Sleep(5 * time.Millisecond)
+	}
+	if e.Addr() == nil {
+		t.Fatal("engine did not bind")
+	}
+	return e
+}
+
+// pingConn885 is a client conn that asks /who and times requests.
+type pingConn885 struct {
+	c  net.Conn
+	br *bufio.Reader
+}
+
+func (p *pingConn885) ask(path string) (string, time.Duration, error) {
+	start := time.Now()
+	_ = p.c.SetDeadline(time.Now().Add(30 * time.Second))
+	if _, err := p.c.Write([]byte("GET " + path + " HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil {
+		return "", 0, err
+	}
+	resp, err := http.ReadResponse(p.br, nil)
+	if err != nil {
+		return "", 0, err
+	}
+	var sb strings.Builder
+	buf := make([]byte, 64)
+	for {
+		n, rerr := resp.Body.Read(buf)
+		sb.Write(buf[:n])
+		if rerr != nil {
+			break
+		}
+	}
+	_ = resp.Body.Close()
+	return sb.String(), time.Since(start), nil
+}
+
+// threeConns885 dials until it holds conn a, conn same on a's loop and conn
+// other on the other loop (the engine has two), and closes the rest.
+func threeConns885(t *testing.T, e *Engine) (a, same, other *pingConn885) {
+	t.Helper()
+	byLoop := map[string][]*pingConn885{}
+	var all []*pingConn885
+	for i := 0; i < 64; i++ {
+		c, err := net.DialTimeout("tcp", e.Addr().String(), 3*time.Second)
+		if err != nil {
+			t.Fatalf("dial: %v", err)
+		}
+		p := &pingConn885{c: c, br: bufio.NewReader(c)}
+		all = append(all, p)
+		who, _, err := p.ask("/who")
+		if err != nil {
+			t.Fatalf("who: %v", err)
+		}
+		byLoop[who] = append(byLoop[who], p)
+		if len(byLoop) < 2 {
+			continue
+		}
+		for _, v := range byLoop {
+			if len(v) >= 2 {
+				a, same = v[0], v[1]
+			} else {
+				other = v[0]
+			}
+		}
+		if a != nil && other != nil {
+			break
+		}
+	}
+	if a == nil || other == nil {
+		t.Fatal("did not get two conns on one loop and one on the other in 64 dials")
+	}
+	for _, p := range all {
+		if p != a && p != same && p != other {
+			_ = p.c.Close()
+		}
+	}
+	t.Cleanup(func() { _ = a.c.Close(); _ = same.c.Close(); _ = other.c.Close() })
+	return a, same, other
+}
+
+type floodStats885 struct {
+	n             int
+	p50, p99, max time.Duration
+}
+
+func (s floodStats885) String() string {
+	return fmt.Sprintf("n=%d p50=%v p99=%v max=%v", s.n, s.p50, s.p99, s.max)
+}
+
+func summarize885(lat []time.Duration) floodStats885 {
+	if len(lat) == 0 {
+		return floodStats885{}
+	}
+	sort.Slice(lat, func(i, j int) bool { return lat[i] < lat[j] })
+	return floodStats885{n: len(lat), p50: lat[len(lat)/2], p99: lat[len(lat)*99/100], max: lat[len(lat)-1]}
+}
+
+// runFlood885 floods a (when flood is set) for dur while each of the pingers
+// asks /who every gap, and returns each one's latencies, and how many
+// requests the flood sent.
+func runFlood885(t *testing.T, a *pingConn885, pingers []*pingConn885, flood bool, dur, gap time.Duration) ([]floodStats885, int64) {
+	t.Helper()
+	var stop atomic.Bool
+	var wg sync.WaitGroup
+	var sent atomic.Int64
+	if flood {
+		req := "GET /flood HTTP/1.1\r\nHost: x\r\n\r\n"
+		chunk := []byte(strings.Repeat(req, (64<<10)/len(req)))
+		per := int64(len(chunk) / len(req))
+		wg.Add(2)
+		go func() { // reader: discard the responses
+			defer wg.Done()
+			buf := make([]byte, 256<<10)
+			for !stop.Load() {
+				_ = a.c.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
+				if _, err := a.c.Read(buf); err != nil {
+					if ne, ok := err.(net.Error); ok && ne.Timeout() {
+						continue
+					}
+					return
+				}
+			}
+		}()
+		go func() { // writer: as fast as the socket takes it
+			defer wg.Done()
+			for !stop.Load() {
+				_ = a.c.SetWriteDeadline(time.Now().Add(200 * time.Millisecond))
+				n, err := a.c.Write(chunk)
+				sent.Add(per * int64(n) / int64(len(chunk)))
+				if err != nil {
+					if ne, ok := err.(net.Error); ok && ne.Timeout() {
+						continue
+					}
+					return
+				}
+			}
+		}()
+		time.Sleep(50 * time.Millisecond) // let the flood build up
+	}
+	lats := make([][]time.Duration, len(pingers))
+	var pw sync.WaitGroup
+	end := time.Now().Add(dur)
+	for i, p := range pingers {
+		pw.Add(1)
+		go func() {
+			defer pw.Done()
+			for time.Now().Before(end) {
+				_, d, err := p.ask("/who")
+				if err != nil {
+					t.Errorf("ping %d: %v", i, err)
+					return
+				}
+				lats[i] = append(lats[i], d)
+				time.Sleep(gap)
+			}
+		}()
+	}
+	pw.Wait()
+	stop.Store(true)
+	wg.Wait()
+	out := make([]floodStats885, len(pingers))
+	for i := range pingers {
+		out[i] = summarize885(lats[i])
+	}
+	return out, sent.Load()
+}
+
+// TestProbeDrainReadFloodLatency885 is the probe, not a gate: it logs the
+// latency of a conn on the flooded conn's loop and of one on the other loop,
+// with and without the flood.
+func TestProbeDrainReadFloodLatency885(t *testing.T) {
+	if os.Getenv("CELERIS_PROBE_885") == "" {
+		t.Skip("a measurement, not a gate: CELERIS_PROBE_885=1 runs it (about 10 s per buffer size)")
+	}
+	for _, buf := range []int{0, 65536} {
+		t.Run(fmt.Sprintf("buf%d", buf), func(t *testing.T) {
+			e := startFloodEngine885(t, buf)
+			a, same, other := threeConns885(t, e)
+			base, _ := runFlood885(t, a, []*pingConn885{same, other}, false, 2*time.Second, 5*time.Millisecond)
+			t.Logf("no flood      same-loop B: %v | other-loop C: %v", base[0], base[1])
+			fl, sent := runFlood885(t, a, []*pingConn885{same, other}, true, 3*time.Second, 5*time.Millisecond)
+			t.Logf("under flood   same-loop B: %v | other-loop C: %v | flood requests sent=%d", fl[0], fl[1], sent)
+		})
+	}
+}
+
+// floodFeed885 writes pipelined "GET /flood" requests to peer as fast as its
+// socket takes them until stop, keeping the stream request-aligned across
+// partial writes. queued counts the bytes the socket took.
+func floodFeed885(peer int, stop *atomic.Bool, queued *atomic.Int64) {
+	req := "GET /flood HTTP/1.1\r\nHost: x\r\n\r\n"
+	pattern := []byte(strings.Repeat(req, (64<<10)/len(req)+2))
+	var total int64
+	for !stop.Load() {
+		off := int(total % int64(len(req)))
+		n, err := unix.Write(peer, pattern[off:off+(64<<10)])
+		if n > 0 {
+			total += int64(n)
+			queued.Store(total)
+		}
+		if err != nil {
+			time.Sleep(20 * time.Microsecond)
+		}
+	}
+}
+
+// TestDrainReadIsBoundedUnderAFlood885 is celeris#885 item 41 as an assertion:
+// the peer keeps the conn's socket full (4 MiB of requests queued before the
+// read, more written while it goes on, and the loop parses far slower than the
+// peer writes), and one drainRead call, which is one readiness event, must
+// return after reading a bounded number of bytes instead of when the peer
+// pauses. A watchdog stops the peer after 2 s so that an unbounded drainRead
+// ends the test instead of hanging it.
+func TestDrainReadIsBoundedUnderAFlood885(t *testing.T) {
+	l := newLedgerLoop(t)
+	l.handler = floodHandler885{}
+	fd, peer, cs := adoptedPair885(t, l)
+	var b [8]byte
+	_, _ = unix.Read(fd, b[:]) // the byte adoptedPair885 wrote to have the ADD report EPOLLIN
+	for _, s := range []int{fd, peer} {
+		_ = unix.SetsockoptInt(s, unix.SOL_SOCKET, unix.SO_SNDBUF, 4<<20)
+	}
+
+	var stop atomic.Bool
+	var queued atomic.Int64
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() { defer wg.Done(); floodFeed885(peer, &stop, &queued) }()
+	go func() { // the peer takes the responses
+		defer wg.Done()
+		buf := make([]byte, 256<<10)
+		for !stop.Load() {
+			if _, err := unix.Read(peer, buf); err != nil {
+				time.Sleep(50 * time.Microsecond)
+			}
+		}
+	}()
+	t.Cleanup(func() { stop.Store(true); wg.Wait() })
+
+	const preload = 1 << 20
+	for dl := time.Now().Add(5 * time.Second); queued.Load() < preload; time.Sleep(time.Millisecond) {
+		if time.Now().After(dl) {
+			t.Fatalf("apparatus: the peer queued only %d bytes in 5 s", queued.Load())
+		}
+	}
+	watchdog := time.AfterFunc(2*time.Second, func() { stop.Store(true) })
+	defer watchdog.Stop()
+
+	before := l.bytesReadBatch
+	start := time.Now()
+	l.drainRead(fd, time.Now().UnixNano())
+	took := time.Since(start)
+	read := l.bytesReadBatch - before
+	fedDuring := queued.Load()
+	stoppedByWatchdog := stop.Load()
+
+	// The bound is far above any sensible budget (a few hundred KiB) and far
+	// below what an unbounded read takes in 2 s.
+	const bound = 2 << 20
+	if read > bound || stoppedByWatchdog {
+		t.Fatalf("celeris#885 item 41: one drainRead read %d bytes in %v (the peer had written %d) and returned only when the peer stopped (watchdog fired: %v); "+
+			"a conn whose peer keeps its socket full holds the loop for as long as it does, and the other conns on the loop wait",
+			read, took, fedDuring, stoppedByWatchdog)
+	}
+	if l.conns[fd] != cs {
+		t.Fatalf("the conn was closed by the read (read %d bytes)", read)
+	}
+}

From 27357b47ba533cc898ec9d9df8ba720b7d580def Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 21:54:07 +0100
Subject: [PATCH 02/12] epoll: tests that a reclaimed hand-off does not count
 its carried recv twice and that the async gate reads cs.protocol without a
 race (both fail on main; v1.6.0, Refs #885 item 28, #886 item 25)

---
 .../async_protocol_edge_885_linux_test.go     | 117 ++++++++++++++++++
 .../epoll/drainread_flood_885_linux_test.go   |   6 +-
 .../epoll/reclaim_reqcount_886_linux_test.go  |  53 ++++++++
 3 files changed, 173 insertions(+), 3 deletions(-)
 create mode 100644 internal/engine/epoll/async_protocol_edge_885_linux_test.go
 create mode 100644 internal/engine/epoll/reclaim_reqcount_886_linux_test.go

diff --git a/internal/engine/epoll/async_protocol_edge_885_linux_test.go b/internal/engine/epoll/async_protocol_edge_885_linux_test.go
new file mode 100644
index 000000000..982a2eb4e
--- /dev/null
+++ b/internal/engine/epoll/async_protocol_edge_885_linux_test.go
@@ -0,0 +1,117 @@
+//go:build linux
+
+package epoll
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"net"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/hpack"
+
+	"github.com/goceleris/celeris/internal/protocol/h2/stream"
+	"github.com/goceleris/celeris/internal/resource"
+)
+
+// celeris#885 item 28. drainRead's async gate reads cs.protocol on the loop
+// thread, while the conn's dispatch goroutine writes it (switchToH2Local, on an
+// h2c upgrade it parses). A client that waits for the 101 before it sends
+// anything else is ordered after the write by the 101's flush (x/sys's
+// unix.Write and unix.Read are race-annotated as one synchronisation, which is
+// all that keeps the detector quiet for it). A client that sends while the
+// goroutine is still in the switch is not: this test holds the goroutine inside
+// the switch (the handler of an HTTP/2 stream that came in the same write runs
+// inline after the protocol is written) and sends a frame meanwhile. Run it
+// under -race.
+
+type edge885Handler struct {
+	started chan struct{}
+	release chan struct{}
+}
+
+func (h *edge885Handler) HandleStream(_ context.Context, s *stream.Stream) error {
+	var path string
+	for _, hd := range s.GetHeaders() {
+		if hd[0] == ":path" {
+			path = hd[1]
+		}
+	}
+	if path == "/hold" {
+		select {
+		case h.started <- struct{}{}:
+		default:
+		}
+		<-h.release
+	}
+	if s.ResponseWriter == nil {
+		return nil
+	}
+	return s.ResponseWriter.WriteResponse(s, 200,
+		[][2]string{{"content-type", "text/plain"}, {"content-length", strconv.Itoa(len(path))}}, []byte(path))
+}
+
+// Only /a is async: it promotes the conn, and the upgrade request behind it is
+// parsed by the dispatch goroutine.
+func (*edge885Handler) RouteAsync(_, path string) bool { return path == "/a" }
+func (*edge885Handler) HasAsyncRoutes() bool           { return true }
+
+func TestAsyncGateProtocolReadHasAnEdgeWithTheUpgradeSwitch885(t *testing.T) {
+	h := &edge885Handler{started: make(chan struct{}, 1), release: make(chan struct{})}
+	_, addr := startE2E885(t, true, h, func(c *resource.Config) { c.EnableH2Upgrade = true })
+	c, err := net.DialTimeout("tcp", addr, 3*time.Second)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer func() { _ = c.Close() }()
+
+	var hb bytes.Buffer
+	enc := hpack.NewEncoder(&hb)
+	for _, f := range [][2]string{{":method", "GET"}, {":scheme", "http"}, {":path", "/hold"}, {":authority", "x"}} {
+		_ = enc.WriteField(hpack.HeaderField{Name: f[0], Value: f[1]})
+	}
+	var frames bytes.Buffer
+	fr := http2.NewFramer(&frames, nil)
+	_ = fr.WriteSettings()
+	_ = fr.WriteHeaders(http2.HeadersFrameParam{StreamID: 3, BlockFragment: hb.Bytes(), EndStream: true, EndHeaders: true})
+	// One write: a request that makes the conn async, the h2c upgrade request
+	// behind it, and the client preface, SETTINGS and a request on stream 3.
+	all := "GET /a HTTP/1.1\r\nHost: x\r\n\r\n" + h2cUpgradeHead908() + http2.ClientPreface + frames.String()
+	if _, err := c.Write([]byte(all)); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	select {
+	case <-h.started:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the goroutine never reached the HTTP/2 stream's handler after the switch")
+	}
+	// The goroutine is inside the switch, the protocol written. The client
+	// sends a PING: the loop reads it and reads cs.protocol.
+	var ping bytes.Buffer
+	_ = http2.NewFramer(&ping, nil).WritePing(false, [8]byte{1})
+	if _, err := c.Write(ping.Bytes()); err != nil {
+		t.Fatalf("write the PING: %v", err)
+	}
+	time.Sleep(300 * time.Millisecond)
+	close(h.release)
+
+	// The conn goes on: the response to /a and the 101 arrive.
+	br := bufio.NewReader(c)
+	_ = c.SetReadDeadline(time.Now().Add(10 * time.Second))
+	var got strings.Builder
+	for {
+		line, err := br.ReadString('\n')
+		got.WriteString(line)
+		if err != nil {
+			t.Fatalf("read the response to /a and the 101: %v (got %q)", err, got.String())
+		}
+		if strings.Contains(line, "HTTP/1.1 101") {
+			return
+		}
+	}
+}
diff --git a/internal/engine/epoll/drainread_flood_885_linux_test.go b/internal/engine/epoll/drainread_flood_885_linux_test.go
index 2b39285cf..4fb85483b 100644
--- a/internal/engine/epoll/drainread_flood_885_linux_test.go
+++ b/internal/engine/epoll/drainread_flood_885_linux_test.go
@@ -52,7 +52,7 @@ func (floodHandler885) HandleStream(_ context.Context, s *stream.Stream) error {
 	return s.ResponseWriter.WriteResponse(s, 200, [][2]string{{"content-type", "text/plain"}}, body)
 }
 
-func startFloodEngine885(t *testing.T, bufSize int) *Engine {
+func startFloodEngine885(t testing.TB, bufSize int) *Engine {
 	t.Helper()
 	ln, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
@@ -116,7 +116,7 @@ func (p *pingConn885) ask(path string) (string, time.Duration, error) {
 
 // threeConns885 dials until it holds conn a, conn same on a's loop and conn
 // other on the other loop (the engine has two), and closes the rest.
-func threeConns885(t *testing.T, e *Engine) (a, same, other *pingConn885) {
+func threeConns885(t testing.TB, e *Engine) (a, same, other *pingConn885) {
 	t.Helper()
 	byLoop := map[string][]*pingConn885{}
 	var all []*pingConn885
@@ -178,7 +178,7 @@ func summarize885(lat []time.Duration) floodStats885 {
 // runFlood885 floods a (when flood is set) for dur while each of the pingers
 // asks /who every gap, and returns each one's latencies, and how many
 // requests the flood sent.
-func runFlood885(t *testing.T, a *pingConn885, pingers []*pingConn885, flood bool, dur, gap time.Duration) ([]floodStats885, int64) {
+func runFlood885(t testing.TB, a *pingConn885, pingers []*pingConn885, flood bool, dur, gap time.Duration) ([]floodStats885, int64) {
 	t.Helper()
 	var stop atomic.Bool
 	var wg sync.WaitGroup
diff --git a/internal/engine/epoll/reclaim_reqcount_886_linux_test.go b/internal/engine/epoll/reclaim_reqcount_886_linux_test.go
new file mode 100644
index 000000000..ac4c8ce2c
--- /dev/null
+++ b/internal/engine/epoll/reclaim_reqcount_886_linux_test.go
@@ -0,0 +1,53 @@
+//go:build linux
+
+package epoll
+
+import (
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// celeris#886 item 25: a hand-off that the target refuses puts the conn back on
+// the loop it came from (reclaimTransplant), with the bytes it had carried: the
+// start of a pipelined next request that drainRead had read, and counted as
+// the recv that read them, before tryTransplant moved the conn. The re-adoption
+// replays them as the recv event of a fresh adoption, and counted one more: the
+// same bytes counted twice by the loop that never gave them up.
+func TestReclaimTransplantDoesNotCountTheCarriedRecvTwice886(t *testing.T) {
+	l := newHandoffLoop(t, nil)
+	l.handler = newPathHandler885(16)
+	target := &refusingTarget{}
+	l.transplant.Store(&transplantState{target: target})
+	fd, peer, cs := adoptedPair885(t, l)
+	var b [8]byte
+	_, _ = unix.Read(fd, b[:]) // the byte adoptedPair885 wrote to have the ADD report EPOLLIN
+
+	// One whole request, then the first bytes of the next: the parser is at a
+	// request boundary with those bytes buffered, which is what a sync source
+	// carries.
+	if _, err := unix.Write(peer, []byte("GET /a HTTP/1.1\r\nHost: x\r\n\r\nGET /b HT")); err != nil {
+		t.Fatalf("peer write: %v", err)
+	}
+	l.drainRead(fd, time.Now().UnixNano())
+	if l.reqBatch != 1 {
+		t.Fatalf("apparatus: drainRead counted %d recvs, want 1", l.reqBatch)
+	}
+	if !cs.h1State.AtRequestBoundary() || !cs.h1State.HasPendingData() {
+		t.Fatalf("apparatus: the conn is not at a boundary with the next request's first bytes buffered")
+	}
+
+	adoptedBefore := l.transplantAdopted.Load()
+	l.tryTransplant(fd)
+
+	if target.calls.Load() != 1 {
+		t.Fatalf("apparatus: the target was asked %d times, want 1 (the refusal is the scenario)", target.calls.Load())
+	}
+	if l.conns[fd] == nil || l.transplantAdopted.Load() != adoptedBefore+1 {
+		t.Fatalf("apparatus: the conn was not reclaimed onto the loop (adopted %d, was %d)", l.transplantAdopted.Load(), adoptedBefore)
+	}
+	if l.reqBatch != 1 {
+		t.Fatalf("celeris#886 item 25: the loop counted %d recvs for one read of the conn: the refused hand-off's reclaim replayed the carried bytes drainRead had counted, and counted them again", l.reqBatch)
+	}
+}

From 88f7f4407c52700a8de601c98d80db64ba8a6bd2 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:03:48 +0100
Subject: [PATCH 03/12] epoll: RegisterConn's comment names what keeps the
 first byte now, the registration's generation and driverMu (comment only;
 v1.6.0, Refs #885 item 55)

---
 internal/engine/epoll/driver.go | 22 +++++++++++++++-------
 1 file changed, 15 insertions(+), 7 deletions(-)

diff --git a/internal/engine/epoll/driver.go b/internal/engine/epoll/driver.go
index 1800e7992..87bc12680 100644
--- a/internal/engine/epoll/driver.go
+++ b/internal/engine/epoll/driver.go
@@ -323,13 +323,21 @@ func (l *Loop) RegisterConn(fd int, onRecv func([]byte), onClose func(error)) er
 			onRecv:  onRecv,
 			onClose: onClose,
 		}
-		// Publish the conn before arming the descriptor (celeris#770). The
-		// worker looks a driver conn up only while hasDriverConns is set, and
-		// the event is edge triggered: armed first, a first event the worker
-		// took before hasDriverConns was set went to the HTTP path, which read
-		// nothing, and never came again. Published first, a racing worker finds
-		// hasDriverConns set and waits in lookupDriver for driverMu, which is
-		// held here until the conn is in the map.
+		// What keeps the first byte (celeris#770, celeris#771, celeris#885
+		// item 55): the conn is in the map, and the descriptor armed, with
+		// driverMu held for both. The event is edge triggered, so a first
+		// event that reached the HTTP path, which reads nothing for a number
+		// that is no HTTP conn's, would never come again. It cannot: a
+		// worker that takes the event routes it by its Pad, the registration's
+		// generation (driverCandidate; the ADD below puts it there), whether
+		// or not hasDriverConns is set yet, and lookupDriver then waits for
+		// driverMu, which is held until the ADD is done, so it finds the conn.
+		// The order of the insert and the ADD inside the lock does not matter
+		// (a mutant with the ADD first, still under driverMu, passes
+		// TestRegisterConnDeliversBytesQueuedBeforeIt); an ADD outside the lock
+		// loses the byte. Before the generation, the order was the whole
+		// guard: only hasDriverConns made the worker look, so the conn had to
+		// be published before the descriptor was armed.
 		l.driverConns[fd] = dc
 		l.hasDriverConns.Store(true)
 		// Level-triggered EPOLLOUT is added lazily (on EAGAIN) by

From 78f44fca2814ba0fe6d5321a667b27d8abc77ba6 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:19:03 +0100
Subject: [PATCH 04/12] epoll: drainRead ends a read turn after 64 KiB and
 re-arms the conn with an EPOLL_CTL_MOD; the half-close, error and transplant
 steps of that event wait for the turn that drains the socket (v1.6.0, Refs
 #885 item 41)

---
 .../epoll/drainread_budget_885_linux_test.go  | 342 ++++++++++++++++++
 .../drainread_budget_bench_885_linux_test.go  | 219 +++++++++++
 .../epoll/drainread_flood_885_linux_test.go   |  10 +-
 .../epoll/http_fd_identity_885_linux_test.go  |   2 +-
 internal/engine/epoll/loop.go                 |  80 +++-
 .../epoll/transplant_accounting_test.go       |   4 +-
 6 files changed, 647 insertions(+), 10 deletions(-)
 create mode 100644 internal/engine/epoll/drainread_budget_885_linux_test.go
 create mode 100644 internal/engine/epoll/drainread_budget_bench_885_linux_test.go

diff --git a/internal/engine/epoll/drainread_budget_885_linux_test.go b/internal/engine/epoll/drainread_budget_885_linux_test.go
new file mode 100644
index 000000000..65b07e77f
--- /dev/null
+++ b/internal/engine/epoll/drainread_budget_885_linux_test.go
@@ -0,0 +1,342 @@
+//go:build linux
+
+package epoll
+
+// celeris#885 item 41, the other half of the read budget. The gate that fails
+// without it is TestDrainReadIsBoundedUnderAFlood885: a turn of drainRead is
+// bounded. These tests are guards of what the bound must not break, and none of
+// them fails without it (a conn with no budget reads everything): a turn that
+// stops at the budget leaves bytes in the socket, edge-triggered epoll will not
+// report them again, and the conn must be given another turn (yieldRead's
+// EPOLL_CTL_MOD) or the rest of a burst, a body or a pipeline waits for bytes
+// the peer will never send. Each is checked against a mutant that drops the MOD.
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+
+	"github.com/goceleris/celeris/internal/protocol/h2/stream"
+	"github.com/goceleris/celeris/internal/resource"
+)
+
+// withBudget885 sets drainReadBudget to n for the test.
+func withBudget885(t *testing.T, n int) {
+	t.Helper()
+	old := drainReadBudget
+	drainReadBudget = n
+	t.Cleanup(func() { drainReadBudget = old })
+}
+
+// readyForFD885 reports the event ready for fd on l's epoll set right now.
+func readyForFD885(l *Loop, fd int) (unix.EpollEvent, bool) {
+	evs := make([]unix.EpollEvent, 16)
+	n, _ := unix.EpollWait(l.epollFD, evs, 0)
+	for i := 0; i < n; i++ {
+		if int(evs[i].Fd) == fd {
+			return evs[i], true
+		}
+	}
+	return unix.EpollEvent{}, false
+}
+
+// budgetConn885 is a conn of the engine's own making (attachAdoptedFD) over a
+// socketpair, whose responses go to a 4 MiB queue.
+func budgetConn885(t *testing.T, bufSize int) (l *Loop, fd, peer int, cs *connState) {
+	t.Helper()
+	l = newLedgerLoop(t)
+	l.resolved = resource.ResolvedResources{BufferSize: bufSize}
+	l.handler = newPathHandler885(1 << 20)
+	fd, peer, cs = adoptedPair885(t, l)
+	var b [8]byte
+	_, _ = unix.Read(fd, b[:]) // the byte adoptedPair885 wrote to have the ADD report EPOLLIN
+	for _, s := range []int{fd, peer} {
+		_ = unix.SetsockoptInt(s, unix.SOL_SOCKET, unix.SO_SNDBUF, 4<<20)
+	}
+	drainEvents885(l)
+	return l, fd, peer, cs
+}
+
+// sendAll885 writes data to peer, in chunks, retrying on EAGAIN for up to 5 s.
+// Everything the tests send is in the socket before the loop reads it, so it
+// must fit the queue the kernel gives a socketpair (about 210 KB at the default
+// net.core.wmem_max, which is what a CI runner has): the largest payload here is
+// 170 KB.
+func sendAll885(t *testing.T, peer int, data []byte) {
+	t.Helper()
+	dl := time.Now().Add(5 * time.Second)
+	for off := 0; off < len(data); {
+		n, err := unix.Write(peer, data[off:min(off+64<<10, len(data))])
+		if n > 0 {
+			off += n
+		}
+		if err != nil && err != unix.EAGAIN {
+			t.Fatalf("peer write: %v", err)
+		}
+		if err == unix.EAGAIN && time.Now().After(dl) {
+			t.Fatalf("apparatus: the socket took only %d of %d bytes: its queue is smaller than the payload", off, len(data))
+		}
+	}
+}
+
+// takeResponses885 reads the peer's end and returns what it holds.
+func takeResponses885(peer int) []byte {
+	var out []byte
+	buf := make([]byte, 256<<10)
+	for {
+		n, err := unix.Read(peer, buf)
+		if n <= 0 || err != nil {
+			return out
+		}
+		out = append(out, buf[:n]...)
+	}
+}
+
+// responses885 counts the responses the peer takes, as it takes them: the loop
+// parks a conn whose responses the peer has not read once the socket's queue
+// is full, which is not what these tests are about.
+type responses885 struct {
+	peer int
+	buf  []byte
+}
+
+// take reads what the peer holds.
+func (r *responses885) take() { r.buf = append(r.buf, takeResponses885(r.peer)...) }
+
+// count is the number of responses taken.
+func (r *responses885) count() int { return bytes.Count(r.buf, []byte("HTTP/1.1 200")) }
+
+// pumpReads885 plays the run loop for fd: while the epoll set reports an event
+// for it, drainRead, and the peer takes the responses. It returns the number of
+// turns it took.
+func pumpReads885(l *Loop, fd int, got *responses885) int {
+	turns := 0
+	for {
+		ev, ok := readyForFD885(l, fd)
+		if !ok {
+			return turns
+		}
+		if ev.Events&unix.EPOLLIN != 0 {
+			turns++
+			l.drainRead(fd, time.Now().UnixNano())
+			got.take()
+		}
+	}
+}
+
+// TestDrainReadBudgetGivesTheConnAnotherTurn885: 3000 pipelined requests are
+// already in the socket (about 100 KB) and the budget is 16 KiB, so the first
+// turn stops with the socket not empty. The conn must be ready again, with its
+// generation, and the turns that follow must answer every request.
+func TestDrainReadBudgetGivesTheConnAnotherTurn885(t *testing.T) {
+	withBudget885(t, 16<<10)
+	l, fd, peer, cs := budgetConn885(t, 4096)
+	const reqs = 3000
+	sendAll885(t, peer, []byte(strings.Repeat("GET /small HTTP/1.1\r\nHost: x\r\n\r\n", reqs)))
+
+	before := l.bytesReadBatch
+	l.drainRead(fd, time.Now().UnixNano())
+	read := int(l.bytesReadBatch - before)
+	if read < 16<<10 || read > 16<<10+4096 {
+		t.Fatalf("the first turn read %d bytes, want the budget (16384) and at most one buffer more", read)
+	}
+	ev, ok := readyForFD885(l, fd)
+	if !ok {
+		t.Fatalf("the conn is not ready after a turn that stopped at the budget with the socket not empty: its rest would wait for bytes the peer does not send")
+	}
+	if ev.Events&unix.EPOLLIN == 0 || uint32(ev.Pad) != cs.gen {
+		t.Fatalf("the event after the turn is %#x with Pad %#x, want EPOLLIN and the conn's generation %#x", ev.Events, uint32(ev.Pad), cs.gen)
+	}
+	// The event was consumed by the check above: ready again only through a
+	// MOD, so pump from the state the run loop is in (it would have this event).
+	got := &responses885{peer: peer}
+	l.drainRead(fd, time.Now().UnixNano())
+	got.take()
+	turns := 2 + pumpReads885(l, fd, got)
+	if got.count() != reqs {
+		t.Fatalf("%d of %d requests were answered after %d turns: readiness was lost", got.count(), reqs, turns)
+	}
+	if turns < 5 {
+		t.Fatalf("the burst took %d turns at a 16 KiB budget, want more than 5: the apparatus does not reach the budget", turns)
+	}
+}
+
+// TestDrainReadBudgetAfterADirectBody885: the budget runs out on the direct-body
+// path (a POST body received straight into H1State's buffer, a read of the
+// whole body), with 2000 pipelined requests behind it still in the socket; the
+// turn that the body ended must give the conn another one.
+func TestDrainReadBudgetAfterADirectBody885(t *testing.T) {
+	withBudget885(t, 16<<10)
+	l, fd, peer, _ := budgetConn885(t, 4096)
+	const bodyLen = 100 << 10
+	req := "POST /small HTTP/1.1\r\nHost: x\r\nContent-Length: " + fmt.Sprint(bodyLen) + "\r\n\r\n" + strings.Repeat("b", bodyLen)
+	const tail = 2000
+	sendAll885(t, peer, []byte(req+strings.Repeat("GET /small HTTP/1.1\r\nHost: x\r\n\r\n", tail)))
+
+	got := &responses885{peer: peer}
+	l.drainRead(fd, time.Now().UnixNano())
+	got.take()
+	turns := 1 + pumpReads885(l, fd, got)
+	if got.count() != 1+tail {
+		t.Fatalf("%d of %d requests (a %d-byte POST, then %d GETs) were answered after %d turns: readiness was lost", got.count(), 1+tail, bodyLen, tail, turns)
+	}
+	if turns < 2 {
+		t.Fatalf("the request took %d turn, want more than 1: the apparatus does not reach the budget", turns)
+	}
+}
+
+// TestDrainReadBudgetKeepsEPOLLOUT885: a conn parked on a response the peer has
+// not taken (EPOLLOUT armed) yields at the budget; the MOD must keep EPOLLOUT in
+// the interest, or the flush that waits for the socket's room never resumes.
+func TestDrainReadBudgetKeepsEPOLLOUT885(t *testing.T) {
+	withBudget885(t, 16<<10)
+	l, fd, peer, cs := budgetConn885(t, 4096)
+	// A small send queue, so that the response parks.
+	_ = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_SNDBUF, 4096)
+	sendAll885(t, peer, []byte("GET /big HTTP/1.1\r\nHost: x\r\n\r\n"))
+	l.drainRead(fd, time.Now().UnixNano())
+	if !cs.epollOut {
+		t.Fatalf("apparatus: the conn is not parked on its response (epollOut false)")
+	}
+	drainEvents885(l)
+
+	sendAll885(t, peer, []byte(strings.Repeat("GET /small HTTP/1.1\r\nHost: x\r\n\r\n", 2000)))
+	l.drainRead(fd, time.Now().UnixNano()) // stops at the budget, MODs
+	if !cs.epollOut {
+		t.Fatalf("the turn disarmed EPOLLOUT in the conn's own state")
+	}
+	drainEvents885(l)
+	// The peer takes the response: the socket is writable again, an edge that
+	// only an interest with EPOLLOUT reports.
+	for i := 0; i < 8; i++ {
+		takeResponses885(peer)
+		time.Sleep(5 * time.Millisecond)
+	}
+	ev, ok := readyForFD885(l, fd)
+	for i := 0; i < 20 && (!ok || ev.Events&unix.EPOLLOUT == 0); i++ {
+		time.Sleep(5 * time.Millisecond)
+		takeResponses885(peer)
+		ev, ok = readyForFD885(l, fd)
+	}
+	if !ok || ev.Events&unix.EPOLLOUT == 0 {
+		t.Fatalf("no EPOLLOUT event (%#x, ready %v) after the peer took the response: the turn's MOD dropped EPOLLOUT from the interest", ev.Events, ok)
+	}
+}
+
+// TestDrainReadBudgetSparesAClosingConn885: a conn whose close is deferred until
+// its response has gone out (peerClosed; closeWhenFlushed's interest has no
+// EPOLLIN by design) is not given a MOD with EPOLLIN at the budget.
+func TestDrainReadBudgetSparesAClosingConn885(t *testing.T) {
+	withBudget885(t, 16<<10)
+	l, fd, peer, cs := budgetConn885(t, 4096)
+	cs.peerClosed = true
+	sendAll885(t, peer, []byte(strings.Repeat("GET /small HTTP/1.1\r\nHost: x\r\n\r\n", 3000)))
+	drainEvents885(l)
+	before := l.bytesReadBatch
+	l.drainRead(fd, time.Now().UnixNano())
+	if read := int(l.bytesReadBatch - before); read <= 16<<10+4096 {
+		t.Fatalf("a closing conn stopped after %d bytes: the turn must go on as it did without a budget", read)
+	}
+}
+
+// slowFirst885 is pathHandler885 whose /slow answers after 300 ms: the loop
+// (a sync engine) or the conn's dispatch goroutine (an async one) is held on
+// the first request of a burst while the rest of it, written at once into the
+// socket's large buffers, is all in the kernel when reading starts: bytes the
+// budget leaves in the socket are not followed by any new segment, so only the
+// re-armed readiness reads them.
+type slowFirst885 struct{ *pathHandler885 }
+
+func (h slowFirst885) HandleStream(ctx context.Context, s *stream.Stream) error {
+	if s.Path == "/slow" {
+		time.Sleep(300 * time.Millisecond)
+	}
+	return h.pathHandler885.HandleStream(ctx, s)
+}
+
+// burst885 sends first and then n requests req down one conn in one write
+// (larger than the budget) and reads the n+1 responses back, then waits for
+// nothing more. With halfClose the client shuts its write side right after the
+// burst, so that the FIN rides the same readable edge as the bytes the turn
+// leaves in the socket: the EPOLLRDHUP of the event that ended a turn at its
+// budget must not close the conn before the rest has been read and answered.
+func burst885(t *testing.T, addr, first, req string, n int, halfClose bool) {
+	t.Helper()
+	c, err := net.DialTimeout("tcp", addr, 3*time.Second)
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer func() { _ = c.Close() }()
+	_ = c.(*net.TCPConn).SetWriteBuffer(8 << 20)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		_, _ = io.Copy(c, strings.NewReader(first+strings.Repeat(req, n)))
+		if halfClose {
+			_ = c.(*net.TCPConn).CloseWrite()
+		}
+	}()
+	br := bufio.NewReader(c)
+	for i := 0; i <= n; i++ {
+		_ = c.SetReadDeadline(time.Now().Add(15 * time.Second))
+		resp, err := http.ReadResponse(br, nil)
+		if err != nil {
+			t.Fatalf("response %d of %d: %v (the rest of the burst was never read)", i, n+1, err)
+		}
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+	}
+	wg.Wait()
+}
+
+// TestDrainReadBudgetEndToEnd885: the default budget, a real engine, a burst of
+// up to 2 MiB of pipelined requests (or 1 MiB of POST body) written at once
+// behind a request that takes 300 ms, and then silence, or the client's FIN:
+// every response must arrive. On the sync engine and on the async one, where the loop hands the
+// bytes to the conn's dispatch goroutine.
+func TestDrainReadBudgetEndToEnd885(t *testing.T) {
+	const slow = "GET /slow HTTP/1.1\r\nHost: x\r\n\r\n"
+	for _, async := range []bool{false, true} {
+		t.Run(fmt.Sprintf("async=%v", async), func(t *testing.T) {
+			_, addr := startE2E885(t, async, slowFirst885{newPathHandler885(1 << 10)}, func(c *resource.Config) {
+				c.Resources.BufferSize = 8192
+				c.Resources.SocketRecv = 8 << 20
+			})
+			t.Run("pipeline", func(t *testing.T) {
+				const req = "GET /small HTTP/1.1\r\nHost: x\r\n\r\n"
+				// 2 MiB; the async engine takes less: its dispatch goroutine
+				// answers a whole batch before it flushes, and a conn whose
+				// queued responses pass the write cap is closed, which is not
+				// what this test asks.
+				size := 2 << 20
+				if async {
+					size = 768 << 10
+				}
+				burst885(t, addr, slow, req, size/len(req), false)
+				if !async {
+					// The async engine answers none of a burst that ends in a FIN
+					// while its goroutine is still in the first request, with or
+					// without a budget (the conn is closed with the bytes in
+					// asyncInBuf): not what this test asks.
+					burst885(t, addr, slow, req, (192<<10)/len(req), true)
+				}
+			})
+			t.Run("body", func(t *testing.T) {
+				body := strings.Repeat("p", 1<<20)
+				post := "POST /small HTTP/1.1\r\nHost: x\r\nContent-Length: " + fmt.Sprint(len(body)) + "\r\n\r\n" + body
+				burst885(t, addr, slow, post, 1, false)
+			})
+		})
+	}
+}
diff --git a/internal/engine/epoll/drainread_budget_bench_885_linux_test.go b/internal/engine/epoll/drainread_budget_bench_885_linux_test.go
new file mode 100644
index 000000000..9cdd87c5b
--- /dev/null
+++ b/internal/engine/epoll/drainread_budget_bench_885_linux_test.go
@@ -0,0 +1,219 @@
+//go:build linux
+
+package epoll
+
+// Benchmarks of the read budget (celeris#885 item 41), for an A/B of drainRead
+// with and without it. The ping-pong shapes (one short read per request: the
+// budget costs an add and a compare) are BenchmarkEpollRecvShapes884 and
+// BenchmarkKeepAliveDispatch; these are the two shapes that reach the budget:
+//
+//   - BenchmarkDrainReadFloodThroughput885: one conn, the loop otherwise idle,
+//     so a yield costs a MOD and an epoll_wait and buys nothing. "pipeline" is
+//     pipelined GETs written 64 KiB at a time; "upload" is a 4 MiB POST body (just under the default body limit), a
+//     read straight into the body buffer. One op is one write (or one POST).
+//   - BenchmarkDrainReadFloodBystander885: the conn floods and a second conn on
+//     the same loop does ping-pong; one op is one round trip of the second, and
+//     the latency it sees while the first floods is reported as p99 and max.
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"sort"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+func BenchmarkDrainReadFloodThroughput885(b *testing.B) {
+	for _, shape := range []string{"pipeline", "upload"} {
+		b.Run(shape, func(b *testing.B) {
+			e := startFloodEngine885(b, 0)
+			c, err := net.DialTimeout("tcp", e.Addr().String(), 3*time.Second)
+			if err != nil {
+				b.Fatalf("dial: %v", err)
+			}
+			defer func() { _ = c.Close() }()
+			cr := &countReader885{r: c}
+			br := bufio.NewReaderSize(cr, 256<<10)
+			var chunk, onePost []byte
+			var perOp int
+			if shape == "pipeline" {
+				req := "GET /flood HTTP/1.1\r\nHost: x\r\n\r\n"
+				perOp = (64 << 10) / len(req)
+				chunk = []byte(strings.Repeat(req, perOp))
+				onePost = []byte(req)
+				b.SetBytes(int64(len(chunk)))
+			} else {
+				body := strings.Repeat("u", 4<<20)
+				chunk = []byte("POST /flood HTTP/1.1\r\nHost: x\r\nContent-Length: " + fmt.Sprint(len(body)) + "\r\n\r\n" + body)
+				onePost = chunk
+				perOp = 1
+				b.SetBytes(int64(len(chunk)))
+			}
+			// One response, to learn its size: the flood's responses are then
+			// counted in bytes, which is all the reader can keep up with.
+			if _, err := c.Write(onePost); err != nil {
+				b.Fatalf("probe write: %v", err)
+			}
+			r, err := http.ReadResponse(br, nil)
+			if err != nil {
+				b.Fatalf("probe: %v", err)
+			}
+			_, _ = io.Copy(io.Discard, r.Body)
+			_ = r.Body.Close()
+			if r.StatusCode != 200 {
+				b.Fatalf("probe: status %d", r.StatusCode)
+			}
+			respLen := cr.n.Load() - int64(br.Buffered())
+			var got atomic.Int64
+			done := make(chan struct{})
+			go func() { // count the response bytes
+				defer close(done)
+				buf := make([]byte, 256<<10)
+				got.Add(int64(br.Buffered()))
+				_, _ = br.Discard(br.Buffered())
+				for {
+					n, err := c.Read(buf)
+					got.Add(int64(n))
+					if err != nil {
+						return
+					}
+				}
+			}()
+			// At most window ops are unanswered: a peer that lets its
+			// responses pile up past the write cap is closed, by design, and
+			// that is not what is measured here.
+			window := int64(8)
+			if shape == "upload" {
+				window = 2
+			}
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				for dl := time.Now().Add(60 * time.Second); got.Load() < (int64(i)-window)*int64(perOp)*respLen; time.Sleep(20 * time.Microsecond) {
+					if time.Now().After(dl) {
+						b.Fatalf("op %d: the server answered %d bytes", i, got.Load())
+					}
+				}
+				if _, err := c.Write(chunk); err != nil {
+					b.Fatalf("write: %v", err)
+				}
+			}
+			want := int64(b.N*perOp) * respLen
+			for dl := time.Now().Add(60 * time.Second); got.Load() < want; time.Sleep(50 * time.Microsecond) {
+				if time.Now().After(dl) {
+					b.Fatalf("got %d of %d response bytes", got.Load(), want)
+				}
+			}
+			b.StopTimer()
+			_ = c.Close()
+			<-done
+		})
+	}
+}
+
+func BenchmarkDrainReadFloodBystander885(b *testing.B) {
+	e := startFloodEngine885(b, 0)
+	a, same, _ := threeConns885(b, e)
+	var stop atomic.Bool
+	var wg sync.WaitGroup
+	req := "GET /flood HTTP/1.1\r\nHost: x\r\n\r\n"
+	chunk := []byte(strings.Repeat(req, (64<<10)/len(req)))
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		buf := make([]byte, 256<<10)
+		for !stop.Load() {
+			_ = a.c.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
+			if _, err := a.c.Read(buf); err != nil {
+				if ne, ok := err.(net.Error); ok && ne.Timeout() {
+					continue
+				}
+				return
+			}
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for !stop.Load() {
+			_ = a.c.SetWriteDeadline(time.Now().Add(200 * time.Millisecond))
+			if _, err := a.c.Write(chunk); err != nil {
+				if ne, ok := err.(net.Error); ok && ne.Timeout() {
+					continue
+				}
+				return
+			}
+		}
+	}()
+	// Stops the flood on every way out, a Fatalf too: a leaked flood goes on
+	// loading the process for the benchmarks that follow.
+	defer func() {
+		stop.Store(true)
+		_ = a.c.Close()
+		wg.Wait()
+	}()
+	time.Sleep(100 * time.Millisecond)
+	lat := make([]time.Duration, 0, b.N)
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, d, err := same.ask("/who")
+		if err != nil {
+			b.Fatalf("ask: %v", err)
+		}
+		lat = append(lat, d)
+	}
+	b.StopTimer()
+	sort.Slice(lat, func(i, j int) bool { return lat[i] < lat[j] })
+	b.ReportMetric(float64(lat[len(lat)*99/100]), "p99-ns")
+	b.ReportMetric(float64(lat[len(lat)-1]), "max-ns")
+}
+
+// countReader885 counts the bytes read through it.
+type countReader885 struct {
+	r io.Reader
+	n atomic.Int64
+}
+
+func (c *countReader885) Read(p []byte) (int, error) {
+	n, err := c.r.Read(p)
+	c.n.Add(int64(n))
+	return n, err
+}
+
+// BenchmarkDrainReadLoop885 is drainRead alone: a conn of the loop's own making
+// over a socketpair, no engine, no client scheduler. One op writes n requests
+// from the peer, has the loop read them (one drainRead call: the read, ProcessH1
+// and the handler, the inline flush) and the peer takes the responses. The cost
+// of the read budget on this path is an add and a compare per read, and of
+// flushInline, one call per read (a call per recv, not per request).
+func BenchmarkDrainReadLoop885(b *testing.B) {
+	for _, n := range []int{1, 16} {
+		b.Run(fmt.Sprintf("reqs=%d", n), func(b *testing.B) {
+			l := newLedgerLoop(b)
+			l.handler = floodHandler885{}
+			fd, peer, _ := adoptedPair885(b, l)
+			var one [8]byte
+			_, _ = unix.Read(fd, one[:]) // the byte adoptedPair885 wrote
+			req := []byte(strings.Repeat("GET /flood HTTP/1.1\r\nHost: x\r\n\r\n", n))
+			out := make([]byte, 64<<10)
+			now := time.Now().UnixNano()
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				if _, err := unix.Write(peer, req); err != nil {
+					b.Fatalf("write: %v", err)
+				}
+				l.drainRead(fd, now)
+				if _, err := unix.Read(peer, out); err != nil {
+					b.Fatalf("read: %v", err)
+				}
+			}
+		})
+	}
+}
diff --git a/internal/engine/epoll/drainread_flood_885_linux_test.go b/internal/engine/epoll/drainread_flood_885_linux_test.go
index 4fb85483b..fe270c4d6 100644
--- a/internal/engine/epoll/drainread_flood_885_linux_test.go
+++ b/internal/engine/epoll/drainread_flood_885_linux_test.go
@@ -285,8 +285,8 @@ func floodFeed885(peer int, stop *atomic.Bool, queued *atomic.Int64) {
 }
 
 // TestDrainReadIsBoundedUnderAFlood885 is celeris#885 item 41 as an assertion:
-// the peer keeps the conn's socket full (4 MiB of requests queued before the
-// read, more written while it goes on, and the loop parses far slower than the
+// the peer keeps the conn's socket full (its queue full before the read, more
+// written while it goes on, and the loop parses far slower than the
 // peer writes), and one drainRead call, which is one readiness event, must
 // return after reading a bounded number of bytes instead of when the peer
 // pauses. A watchdog stops the peer after 2 s so that an unbounded drainRead
@@ -297,6 +297,10 @@ func TestDrainReadIsBoundedUnderAFlood885(t *testing.T) {
 	fd, peer, cs := adoptedPair885(t, l)
 	var b [8]byte
 	_, _ = unix.Read(fd, b[:]) // the byte adoptedPair885 wrote to have the ADD report EPOLLIN
+	// A bigger queue where the host allows one (net.core.wmem_max): the more is
+	// queued before the read, the longer the peer can stall without the socket
+	// emptying. At the default cap the queue is still about 210 KB, three times
+	// the budget.
 	for _, s := range []int{fd, peer} {
 		_ = unix.SetsockoptInt(s, unix.SOL_SOCKET, unix.SO_SNDBUF, 4<<20)
 	}
@@ -317,7 +321,7 @@ func TestDrainReadIsBoundedUnderAFlood885(t *testing.T) {
 	}()
 	t.Cleanup(func() { stop.Store(true); wg.Wait() })
 
-	const preload = 1 << 20
+	const preload = 128 << 10
 	for dl := time.Now().Add(5 * time.Second); queued.Load() < preload; time.Sleep(time.Millisecond) {
 		if time.Now().After(dl) {
 			t.Fatalf("apparatus: the peer queued only %d bytes in 5 s", queued.Load())
diff --git a/internal/engine/epoll/http_fd_identity_885_linux_test.go b/internal/engine/epoll/http_fd_identity_885_linux_test.go
index 369ef3b14..f23afd050 100644
--- a/internal/engine/epoll/http_fd_identity_885_linux_test.go
+++ b/internal/engine/epoll/http_fd_identity_885_linux_test.go
@@ -81,7 +81,7 @@ func drainEvents885(l *Loop) {
 // (attachAdoptedFD, on this goroutine) and returns its descriptor, its peer
 // and its slot. The peer has written a byte before the adoption, so the ADD
 // reports EPOLLIN at once.
-func adoptedPair885(t *testing.T, l *Loop) (fd, peer int, cs *connState) {
+func adoptedPair885(t testing.TB, l *Loop) (fd, peer int, cs *connState) {
 	t.Helper()
 	pair, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_CLOEXEC|unix.SOCK_NONBLOCK, 0)
 	if err != nil {
diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index bf0bf25a9..fc8dc703a 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -321,6 +321,11 @@ type Loop struct {
 	httpGen        uint32
 	hasDriverConns atomic.Bool
 	driverReadBuf  []byte // scratch buffer for driver EPOLLIN drains (worker-local)
+	// readYielded is set by yieldRead, on the loop thread, when a read turn ends
+	// at its budget with bytes possibly left in the socket; the run loop reads
+	// it after drainRead and skips the half-close and error handling of the
+	// event that turn served (celeris#885 item 41).
+	readYielded bool
 	// driverReadQ holds the driver conns whose last read turn used up
 	// driverReadBudget with bytes possibly left (loop thread only).
 	// Edge-triggered epoll does not report bytes already in the socket again,
@@ -731,8 +736,11 @@ func (l *Loop) run(ctx context.Context) {
 
 			// Process EPOLLIN before EPOLLHUP: a peer may send data and
 			// immediately close, producing both flags in one event.
+			yielded := false
 			if ev.Events&unix.EPOLLIN != 0 {
+				l.readYielded = false
 				l.drainRead(fd, now)
+				yielded = l.readYielded
 			}
 
 			// EPOLLOUT: the socket became writable again for a conn that hit
@@ -752,11 +760,17 @@ func (l *Loop) run(ctx context.Context) {
 			}
 
 			// EPOLLRDHUP: peer half-closed (sent FIN). See onPeerHalfClose.
-			if ev.Events&unix.EPOLLRDHUP != 0 {
+			// Not while the turn above ended at its budget (yielded): the
+			// conn's bytes before the FIN are not all read, and the close
+			// onPeerHalfClose makes would cut them off. The MOD that gave the
+			// conn its next turn re-polls the descriptor, so the next event
+			// carries EPOLLRDHUP (and EPOLLHUP, EPOLLERR) again, after the last
+			// of the bytes has been read (celeris#885 item 41).
+			if !yielded && ev.Events&unix.EPOLLRDHUP != 0 {
 				l.onPeerHalfClose(fd)
 			}
 
-			if ev.Events&(unix.EPOLLERR|unix.EPOLLHUP) != 0 {
+			if !yielded && ev.Events&(unix.EPOLLERR|unix.EPOLLHUP) != 0 {
 				if fd >= 0 && fd < len(l.conns) && l.conns[fd] != nil {
 					l.closeConn(fd)
 				}
@@ -766,8 +780,10 @@ func (l *Loop) run(ctx context.Context) {
 			// a request cycle (EPOLLIN → drainRead) and may now be at a clean
 			// keep-alive boundary — try to transplant it. tryTransplant re-checks
 			// the slot, so a close above is harmless. The pointer load is the
-			// whole cost when no drain is active.
-			if l.transplant.Load() != nil {
+			// whole cost when no drain is active. Not after a turn that ended at
+			// its budget (yielded): that conn is mid-flood, its next turn will
+			// find it at a boundary or not.
+			if !yielded && l.transplant.Load() != nil {
 				l.tryTransplant(fd)
 			}
 		}
@@ -1523,6 +1539,51 @@ func readPromoted(fd int, cs *connState, buf []byte) (int, error) {
 	return n, err
 }
 
+// drainReadBudget is the number of bytes drainRead reads from one conn in one
+// turn (one readiness event) before it gives the loop back: a conn whose peer
+// keeps its socket full, a client that pipelines faster than the loop parses or
+// uploads a large body, would otherwise hold the loop in drainRead for as long
+// as its inflow does not pause, and every other conn on the loop would wait
+// (celeris#885 item 41, the HTTP twin of driverReadBudget, celeris#930). In
+// bytes, not reads: the buffer is 4 KiB to 256 KiB (Config.BufferSize) and a
+// read straight into a request body can be larger still. A turn that stops at
+// the budget re-arms the conn's readiness (yieldRead), so the rest of the
+// socket is read in the next round, after the events of the batch that were
+// already ready: a conn on the loop waits for at most one turn of the flooding
+// one, about 0.2 ms of pipelined tiny requests at 64 KiB (0.5 ms at 256 KiB,
+// 0.1 ms at 16 KiB), and the flooding conn pays one EPOLL_CTL_MOD and one
+// epoll_wait round per turn, 4% of a flood of such requests at 64 KiB (1% at
+// 256 KiB, 7% at 16 KiB: BenchmarkDrainReadFloodThroughput885 and
+// BenchmarkDrainReadFloodBystander885, measured with a 2 vCPU container). A
+// variable for the tests, which lower it; never written while a loop runs.
+var drainReadBudget = 64 << 10
+
+// yieldRead ends cs's read turn at the budget, with bytes possibly left in the
+// socket. Edge-triggered epoll reports bytes already there once, so nothing
+// would read them until the peer sent more: the interest is set again with an
+// EPOLL_CTL_MOD, which re-polls the descriptor and puts it back at the end of
+// the ready list if it is readable (ep_modify), and the event comes with the
+// next epoll_wait carrying the conn's generation, as every event of the conn
+// does (httpEpollEvent): a conn that has closed by then, or whose number
+// another conn has taken, is dropped by the run loop's identity check. The
+// mask is the conn's own (epollInterest: EPOLLOUT if armed, no EPOLLIN while
+// recv is paused). It reports whether drainRead should return: true when the
+// MOD was issued, or when an async conn has been hijacked, and the conn is
+// gone from this loop. False means keep reading, as before the budget: the MOD
+// failed, or the conn is closing (closeWhenFlushed's mask has no EPOLLIN by
+// design and a MOD with it would undo that). Loop thread.
+func (l *Loop) yieldRead(cs *connState) bool {
+	if cs.peerClosed {
+		return false
+	}
+	issued, err := l.modEpollOut(cs, cs.epollInterest(cs.epollOut))
+	if !issued || err == nil {
+		l.readYielded = true
+		return true
+	}
+	return false
+}
+
 func (l *Loop) drainRead(fd int, now int64) {
 	if fd < 0 || fd >= len(l.conns) {
 		return
@@ -1537,7 +1598,17 @@ func (l *Loop) drainRead(fd int, now int64) {
 		return
 	}
 
+	// Bytes read in this turn; at drainReadBudget the turn ends (yieldRead).
+	// Checked at the top so that every path back to it (a partial body, a
+	// protocol not yet decided, an async hand-off, an upgrade) is bounded.
+	spent := 0
 	for {
+		if spent >= drainReadBudget {
+			if l.yieldRead(cs) {
+				return
+			}
+			spent = 0 // not re-armed: read on, and ask again a budget later
+		}
 		// Zero-copy body recv: when H1 is in a partial-body state we read
 		// directly into the tail of H1State.bodyBuf — skipping the
 		// cs.buf → bodyBuf memcpy that would otherwise happen on every
@@ -1586,6 +1657,7 @@ func (l *Loop) drainRead(fd int, now int64) {
 		// n > 0 here (the err and EOF cases returned above): bytes read on
 		// this iteration, into either cs.buf or the zero-copy bodyBuf.
 		l.bytesReadBatch += uint64(n)
+		spent += n
 
 		// Direct-into-bodyBuf completion path: we read straight into
 		// H1State.bodyBuf; dispatch the handler if the body is full,
diff --git a/internal/engine/epoll/transplant_accounting_test.go b/internal/engine/epoll/transplant_accounting_test.go
index b943f5831..a6919e4df 100644
--- a/internal/engine/epoll/transplant_accounting_test.go
+++ b/internal/engine/epoll/transplant_accounting_test.go
@@ -31,7 +31,7 @@ import (
 
 // newLedgerLoop builds a bare Loop with a real epoll fd and the #383 ledger
 // counters wired to fresh atomics, mirroring what Engine.Listen does per loop.
-func newLedgerLoop(t *testing.T) *Loop {
+func newLedgerLoop(t testing.TB) *Loop {
 	t.Helper()
 	epfd, err := unix.EpollCreate1(unix.EPOLL_CLOEXEC)
 	if err != nil {
@@ -58,7 +58,7 @@ func newLedgerLoop(t *testing.T) *Loop {
 
 // socketpairFD returns one end of a connected AF_UNIX socketpair, small enough
 // to land inside the loop's conn table, with the peer closed on cleanup.
-func socketpairFD(t *testing.T, l *Loop) int {
+func socketpairFD(t testing.TB, l *Loop) int {
 	t.Helper()
 	pair, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM, 0)
 	if err != nil {

From 71f38c976a9bed5299d78dfd1deda19437f22a43 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:19:14 +0100
Subject: [PATCH 05/12] epoll: reclaimTransplant does not count a recv that
 drainRead counted (v1.6.0, Refs #886 item 25)

---
 internal/engine/epoll/adopt.go      | 29 +++++++++++++++++++++++------
 internal/engine/epoll/transplant.go |  4 +++-
 2 files changed, 26 insertions(+), 7 deletions(-)

diff --git a/internal/engine/epoll/adopt.go b/internal/engine/epoll/adopt.go
index eae7b4581..625c0c8ab 100644
--- a/internal/engine/epoll/adopt.go
+++ b/internal/engine/epoll/adopt.go
@@ -150,6 +150,17 @@ func (l *Loop) refuseAdopt(fd int, carry engine.Carryover) {
 // clean HTTP/1 request boundary; any pending request bytes wait in the kernel
 // socket receive buffer and are picked up by the EPOLLIN edge after registration.
 func (l *Loop) attachAdoptedFD(ctx context.Context, fd int, carry engine.Carryover, now int64) {
+	l.attachFD(ctx, fd, carry, now, false)
+}
+
+// attachFD is attachAdoptedFD with the replay's recv count settled by the
+// caller: recvCounted says that the carried bytes were read, and counted, by
+// this loop already, which is a reclaim (reclaimTransplant: drainRead counted
+// the recv that read them before tryTransplant moved the conn), so the replay
+// must not count them again (celeris#886 item 25). A conn adopted from another
+// engine has had its bytes counted there, and counts them here as its first
+// recv (celeris#886).
+func (l *Loop) attachFD(ctx context.Context, fd int, carry engine.Carryover, now int64, recvCounted bool) {
 	if fd < 0 {
 		l.refuseAdopt(fd, carry)
 		return
@@ -232,7 +243,7 @@ func (l *Loop) attachAdoptedFD(ctx context.Context, fd int, carry engine.Carryov
 	// answer to a request no handler saw (celeris#543).
 	if len(carry.Buffered) > 0 {
 		if l.async {
-			l.replayCarriedAsync(cs, fd, carry.Buffered)
+			l.replayCarriedAsync(cs, fd, carry.Buffered, recvCounted)
 			return
 		}
 		perr := conn.ProcessH1(cs.ctx, carry.Buffered, cs.h1State, l.handler, cs.writeFn)
@@ -240,8 +251,11 @@ func (l *Loop) attachAdoptedFD(ctx context.Context, fd int, carry engine.Carryov
 		// one, whatever the verdict (celeris#886). drainRead counts a recv
 		// after ProcessH1 too, but returns before it on an h2c upgrade and
 		// never counts a direct-body recv: a carry that is an upgrade request
-		// counts one here and its recv would have counted none.
-		l.reqBatch++
+		// counts one here and its recv would have counted none. Not when this
+		// loop read and counted the bytes itself (a reclaim: recvCounted).
+		if !recvCounted {
+			l.reqBatch++
+		}
 		if perr != nil {
 			l.endCarriedReplay(cs, fd, perr)
 			return
@@ -259,15 +273,18 @@ func (l *Loop) attachAdoptedFD(ctx context.Context, fd int, carry engine.Carryov
 // thread; the conn is fresh from attachAdoptedFD, so no dispatch goroutine
 // exists yet and the loop owns cs.h1State until the promotion hands it over.
 // The conn's EPOLLIN is already registered, so nothing is left to arm.
-func (l *Loop) replayCarriedAsync(cs *connState, fd int, data []byte) {
+func (l *Loop) replayCarriedAsync(cs *connState, fd int, data []byte, recvCounted bool) {
 	cs.h1State.InlineMode = true
 	perr := conn.ProcessH1(cs.ctx, data, cs.h1State, l.handler, cs.writeFn)
 	// One request counted for the replay, whichever way it ends below: it is
 	// the recv event of this conn's first bytes (the promotion included, as
 	// in drainRead, celeris#626; celeris#886). Unlike drainRead it also counts
 	// a carry that ends in an h2c upgrade, which that path returns before
-	// counting.
-	l.reqBatch++
+	// counting. Not when this loop counted the recv already (a reclaim,
+	// recvCounted).
+	if !recvCounted {
+		l.reqBatch++
+	}
 	// A handler that hijacked inline has had cs released inside the Hijack
 	// call, as in drainRead, so cs is not touched again after ErrHijacked.
 	if errors.Is(perr, conn.ErrHijacked) {
diff --git a/internal/engine/epoll/transplant.go b/internal/engine/epoll/transplant.go
index 64e88324c..83189a54e 100644
--- a/internal/engine/epoll/transplant.go
+++ b/internal/engine/epoll/transplant.go
@@ -222,7 +222,9 @@ func (l *Loop) reclaimTransplant(fd int, carry engine.Carryover, bucket *atomic.
 	if now == 0 {
 		now = time.Now().UnixNano()
 	}
-	l.attachAdoptedFD(ctx, fd, carry, now)
+	// The carried bytes were read by this loop, and counted as the recv that
+	// read them, before the hand-off (celeris#886 item 25).
+	l.attachFD(ctx, fd, carry, now, true)
 }
 
 // flushedAtBoundary reports whether cs is at a clean HTTP/1 request boundary with

From f783a622edeb34789dd6345111abb1099c1a538e Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:19:41 +0100
Subject: [PATCH 06/12] epoll: the async gate of drainRead reads h2Switched,
 set under asyncInMu by the dispatch goroutine's switch to HTTP/2, not
 cs.protocol (v1.6.0, Refs #885 item 28)

---
 internal/engine/epoll/conn.go | 12 ++++++++++++
 internal/engine/epoll/loop.go | 36 +++++++++++++++++++++++++++++++++--
 2 files changed, 46 insertions(+), 2 deletions(-)

diff --git a/internal/engine/epoll/conn.go b/internal/engine/epoll/conn.go
index f51157015..ec0694bc8 100644
--- a/internal/engine/epoll/conn.go
+++ b/internal/engine/epoll/conn.go
@@ -366,6 +366,17 @@ type connState struct {
 	// straight to the dispatch goroutine instead of retrying the inline
 	// fast path. Reset on release.
 	asyncPromoted bool
+	// h2Switched is set by switchToH2Local, under asyncInMu, after it has
+	// written cs.h1State, cs.h2State and cs.protocol, and is what the loop
+	// thread reads of them for a promoted conn (drainRead's async gate, under
+	// the same lock): a true read orders every later read of the three after
+	// those writes, which the dispatch goroutine makes holding cs.detachMu and
+	// the loop thread does not take. A conn that is not switched is HTTP/1 by
+	// construction (a promoted conn started as one), so the gate never reads
+	// cs.protocol while the goroutine may be writing it (celeris#885 item 28).
+	// Declared beside asyncPromoted, in the byte of padding before the atomic
+	// that follows (TestConnStateStaysInItsSizeClass). Reset on release.
+	h2Switched bool
 	// asyncH2Promoted signals that runAsyncHandler observed ErrUpgradeH2C
 	// and completed the cs-local H1→H2 swap under detachMu. The worker
 	// must finish the promotion in drainDetachQueue (append fd to
@@ -512,6 +523,7 @@ func releaseConnState(cs *connState) {
 	cs.dirtyNext = nil
 	cs.dirtyPrev = nil
 	cs.protocol = 0
+	cs.h2Switched = false
 	cs.detected = false
 	cs.detectN = 0
 	cs.detectDeadline = 0
diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index fc8dc703a..f451bb318 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -1584,6 +1584,22 @@ func (l *Loop) yieldRead(cs *connState) bool {
 	return false
 }
 
+// lockAsyncInUnlessSwitched is the loop thread's test of drainRead's async
+// gate: it takes asyncInMu and, if the dispatch goroutine has not switched the
+// conn to HTTP/2 (h2Switched), returns true with the lock held, for the
+// append of the bytes just read. A switched conn returns false with the lock
+// released; the loop then reads protocol, h1State and h2State, which the
+// goroutine's switchToH2Local wrote before it set h2Switched under the same
+// lock (celeris#885 item 28).
+func (cs *connState) lockAsyncInUnlessSwitched() bool {
+	cs.asyncInMu.Lock()
+	if cs.h2Switched {
+		cs.asyncInMu.Unlock()
+		return false
+	}
+	return true
+}
+
 func (l *Loop) drainRead(fd int, now int64) {
 	if fd < 0 || fd >= len(l.conns) {
 		return
@@ -1793,8 +1809,17 @@ func (l *Loop) drainRead(fd int, now int64) {
 		// with ErrAsyncDispatch on the first async route, promoting the
 		// conn — so sync routes run inline on the event loop on a server
 		// that mixes sync + async handlers.
-		if l.async && cs.asyncPromoted && cs.protocol == engine.HTTP1 {
-			cs.asyncInMu.Lock()
+		//
+		// The gate reads h2Switched, not cs.protocol: for a promoted conn the
+		// dispatch goroutine writes protocol (and h1State, h2State) in
+		// switchToH2Local, holding detachMu, and the loop thread must not read
+		// them unordered: a client that sends HTTP/2 frames before it has read
+		// the 101 reaches this line while the goroutine is in the switch
+		// (celeris#885 item 28). A conn the goroutine has not switched is
+		// HTTP/1; one it has switched is read below, after the lock that
+		// orders those reads. lockAsyncInUnlessSwitched returns with
+		// cs.asyncInMu held when it is HTTP/1, which the block releases.
+		if l.async && cs.asyncPromoted && cs.lockAsyncInUnlessSwitched() {
 			// Backpressure: drop the conn if the dispatch goroutine is
 			// falling behind. A client that pipelines faster than we
 			// can process would otherwise grow asyncInBuf without bound.
@@ -3148,6 +3173,13 @@ func (l *Loop) switchToH2Local(cs *connState, writeFn func([]byte)) error {
 	cs.h2State = h2State
 	cs.h2State.SetRemoteAddr(cs.remoteAddr)
 	cs.protocol = engine.H2C
+	// Ordered after the three writes above for the loop thread's gate, which
+	// reads the flag under asyncInMu (taken here under detachMu, the order
+	// every site keeps, and for these few instructions only: not across
+	// ProcessH2 below, which runs handlers).
+	cs.asyncInMu.Lock()
+	cs.h2Switched = true
+	cs.asyncInMu.Unlock()
 
 	var processErr error
 	if len(info.Remaining) > 0 {

From 67342698b847555f83d12e840165d5f070ccee3b Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:19:56 +0100
Subject: [PATCH 07/12] epoll: the transplant replay's flush is drainRead's
 inline flush, one function (flushInline) (v1.6.0, Refs #885 item 17)

---
 internal/engine/epoll/adopt.go |  59 +-------------
 internal/engine/epoll/loop.go  | 145 +++++++++++++++++++--------------
 2 files changed, 86 insertions(+), 118 deletions(-)

diff --git a/internal/engine/epoll/adopt.go b/internal/engine/epoll/adopt.go
index 625c0c8ab..d051d24e9 100644
--- a/internal/engine/epoll/adopt.go
+++ b/internal/engine/epoll/adopt.go
@@ -359,7 +359,8 @@ func (l *Loop) endCarriedReplay(cs *connState, fd int, perr error) {
 }
 
 // flushCarriedReplay sends the responses a replay's ProcessH1 queued, as
-// drainRead's inline flush does after a recv (celeris#908). The replay used
+// drainRead's inline flush does after a recv (celeris#908): it is that flush,
+// flushInline, which the two share since celeris#885 item 17. The replay used
 // to flush once and stop: a response larger than the socket buffers left its
 // rest in cs.writeBuf with no EPOLLOUT armed and the dirty list untouched
 // (EPOLLIN is edge-triggered, and nothing reads from the socket again until
@@ -367,59 +368,7 @@ func (l *Loop) endCarriedReplay(cs *connState, fd int, perr error) {
 // It also left pendingBytes at the sum of the replayed responses, which the
 // write hooks read as a backlog, and ignored a refused write.
 //
-// It stays a copy of drainRead's inline flush on purpose: one shared function
-// would put a call on the per-request path of every recv, and the copy runs
-// once per adoption (celeris#885 item 17 asks for the verdict and upgrade
-// blocks to be shared, which h1verdict.go does; this block is not one of them).
-//
-// Loop thread, on a conn fresh from attachAdoptedFD. cs.detachMu is released
-// before any close, which takes it again.
+// Loop thread, on a conn fresh from attachAdoptedFD.
 func (l *Loop) flushCarriedReplay(cs *connState, fd int) {
-	mu := cs.detachMu
-	if mu != nil {
-		mu.Lock()
-	}
-	if csWritePending(cs) {
-		queued := 0
-		if cs.writeSince != 0 {
-			queued = csPendingBytes(cs) // before this flush (celeris#885)
-		}
-		if fErr := l.flushWrites(cs, true); fErr != nil {
-			if cs.h1State != nil && cs.h1State.OnError != nil {
-				cs.h1State.OnError(fErr)
-			}
-			if mu != nil {
-				mu.Unlock()
-			}
-			if cs.dirty {
-				l.removeDirty(cs)
-			}
-			l.closeConn(fd)
-			return
-		}
-		if !csWritePending(cs) {
-			cs.pendingBytes = 0
-			cs.endWrite()
-			if cs.dirty {
-				l.removeDirty(cs)
-			}
-			l.disarmEpollOut(cs)
-		} else {
-			// The kernel send buffer is full: sync pendingBytes and arm
-			// EPOLLOUT, which handleWritable carries on from (a pending
-			// sendfile too), for a detached conn as for any other: the dirty
-			// list is a 0 ms epoll_wait for as long as the peer does not read
-			// (celeris#885 item 42).
-			cs.pendingBytes = csPendingBytes(cs)
-			cs.noteWriteProgress(queued, cs.pendingBytes)
-			l.armEpollOut(cs)
-		}
-	}
-	refused := cs.writeRefused
-	if mu != nil {
-		mu.Unlock()
-	}
-	if refused {
-		l.closeWhenFlushed(cs)
-	}
+	l.flushInline(cs, fd)
 }
diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index f451bb318..23810e7bd 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -1992,69 +1992,7 @@ func (l *Loop) drainRead(fd int, now int64) {
 		// connections). This sends the response one event-batch earlier than
 		// the dirty list pass, reducing per-request latency by up to N×2µs
 		// (where N is the number of other events in the same epoll_wait batch).
-		if mu := cs.detachMu; mu != nil {
-			mu.Lock()
-		}
-		if csWritePending(cs) {
-			queued := 0
-			if cs.writeSince != 0 {
-				queued = csPendingBytes(cs) // before this flush (celeris#885)
-			}
-			if fErr := l.flushWrites(cs, true); fErr != nil {
-				if cs.h1State != nil && cs.h1State.OnError != nil {
-					cs.h1State.OnError(fErr)
-				}
-				if mu := cs.detachMu; mu != nil {
-					mu.Unlock()
-				}
-				if cs.dirty {
-					l.removeDirty(cs)
-				}
-				l.closeConn(fd)
-				return
-			}
-			if !csWritePending(cs) {
-				// Fully flushed (including any sendfile body) — no dirty
-				// list / EPOLLOUT needed.
-				cs.pendingBytes = 0
-				cs.endWrite()
-				if cs.dirty {
-					l.removeDirty(cs)
-				}
-				l.disarmEpollOut(cs)
-			} else {
-				// Partial write — the kernel send buffer is full (write
-				// backpressure). Sync pendingBytes and arm EPOLLOUT
-				// instead of busy-retrying via the dirty list, which
-				// would spin epoll_wait(0)→write(EAGAIN) at 100% CPU, a
-				// truly-detached WS/SSE conn included (celeris#885 item 42).
-				// A pending sendfile is resumed by handleWritable on the next
-				// EPOLLOUT edge.
-				cs.pendingBytes = csPendingBytes(cs)
-				// A client that sends while it reads (an HTTP/2 PING or
-				// WINDOW_UPDATE, a pipelined request) raises an EPOLLIN event
-				// per segment, and this flush takes the room it made: that is
-				// progress for the write clock, and no EPOLLOUT edge may come
-				// to show it (celeris#885 items 35 and 45).
-				cs.noteWriteProgress(queued, cs.pendingBytes)
-				l.armEpollOut(cs)
-			}
-		}
-		// Read writeRefused inside the lock so the check below is safe
-		// against concurrent goroutine writes via the guarded writeFn.
-		refused := cs.writeRefused
-		if mu := cs.detachMu; mu != nil {
-			mu.Unlock()
-		}
-
-		// Back-pressure close: a write hook refused bytes because the
-		// backlog before them was over writeCap, a peer that stopped
-		// reading while it kept sending requests. A backlog over the cap
-		// by itself is not a reason to close: one response larger than the
-		// cap puts it there, and so does a sendfile body. Closing on it cut
-		// such a response off mid-body (celeris#761).
-		if refused {
-			l.closeWhenFlushed(cs)
+		if l.flushInline(cs, fd) {
 			return
 		}
 
@@ -2080,6 +2018,87 @@ func (l *Loop) drainRead(fd int, now int64) {
 	}
 }
 
+// flushInline is drainRead's inline flush: it sends the responses the request
+// handler queued on cs right after the handler returned, and settles the
+// conn's write state for what the kernel did not take (celeris#885 item 17: the
+// transplant replay's flushCarriedReplay is this function, so the replay cannot
+// drift from the recv path). It reports whether the caller is done with the
+// conn, which is closed or closing: a failed flush closes it (after the
+// detached middleware's OnError), and a write the hooks refused ends it once
+// what is queued has gone out.
+//
+//   - Fully flushed (sendfile body included): no dirty list or EPOLLOUT is
+//     needed, so both are cleared.
+//   - Partial: the kernel send buffer is full (write backpressure), so
+//     pendingBytes is synced and EPOLLOUT armed instead of the dirty list,
+//     which would spin epoll_wait(0) then write(EAGAIN) at 100% CPU, a
+//     truly-detached WS/SSE conn included (celeris#885 item 42). A pending
+//     sendfile is resumed by handleWritable on the next EPOLLOUT edge. A
+//     client that sends while it reads (an HTTP/2 PING or WINDOW_UPDATE, a
+//     pipelined request) raises an EPOLLIN event per segment, and this flush
+//     takes the room it made: that is progress for the write clock, and no
+//     EPOLLOUT edge may come to show it (celeris#885 items 35 and 45).
+//   - Refused: a write hook refused bytes because the backlog before them was
+//     over writeCap, a peer that stopped reading while it kept sending
+//     requests. A backlog over the cap by itself is not a reason to close: one
+//     response larger than the cap puts it there, and so does a sendfile body.
+//     Closing on it cut such a response off mid-body (celeris#761).
+//
+// writeRefused is read inside the lock, against the dispatch goroutine's
+// guarded writeFn; cs.detachMu is released before any close, which takes it
+// again. Loop thread.
+//
+// The direct-body block of drainRead has a flush of its own and is not this
+// one: it does not call OnError, does not drop the conn from the dirty list on
+// an error, and does not disarm EPOLLOUT when everything went out, because it
+// is sync-only (intoBody is gated on !l.async: no detached conn).
+func (l *Loop) flushInline(cs *connState, fd int) (done bool) {
+	mu := cs.detachMu
+	if mu != nil {
+		mu.Lock()
+	}
+	if csWritePending(cs) {
+		queued := 0
+		if cs.writeSince != 0 {
+			queued = csPendingBytes(cs) // before this flush (celeris#885)
+		}
+		if fErr := l.flushWrites(cs, true); fErr != nil {
+			if cs.h1State != nil && cs.h1State.OnError != nil {
+				cs.h1State.OnError(fErr)
+			}
+			if mu != nil {
+				mu.Unlock()
+			}
+			if cs.dirty {
+				l.removeDirty(cs)
+			}
+			l.closeConn(fd)
+			return true
+		}
+		if !csWritePending(cs) {
+			cs.pendingBytes = 0
+			cs.endWrite()
+			if cs.dirty {
+				l.removeDirty(cs)
+			}
+			l.disarmEpollOut(cs)
+		} else {
+			cs.pendingBytes = csPendingBytes(cs)
+			cs.noteWriteProgress(queued, cs.pendingBytes)
+			l.armEpollOut(cs)
+		}
+	}
+	refused := cs.writeRefused
+	if mu != nil {
+		mu.Unlock()
+	}
+	if refused {
+		l.closeWhenFlushed(cs)
+		return true
+	}
+	return false
+}
+
 // drainH2Writes drains the H2 async write queues. Handler goroutines enqueue
 // response frame bytes; the loop drains them into writeBuf and flushes to the
 // wire. Extracted from run so the pass can be driven by a test (celeris#889

From e8571ad0b33a3ce4b9ef9e32c1291465bd182077 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 23:05:20 +0100
Subject: [PATCH 08/12] epoll: tests that refusesRecv reads cs.protocol in
 order with the h2c switch (fails on main under -race) and that the run loop
 handles a half-close after a yielded turn (v1.6.0, Refs #885 items 28, 41)

---
 .../async_protocol_edge_885_linux_test.go     | 76 ++++++++++++++++++-
 .../epoll/drainread_budget_885_linux_test.go  | 44 +++++++++++
 2 files changed, 118 insertions(+), 2 deletions(-)

diff --git a/internal/engine/epoll/async_protocol_edge_885_linux_test.go b/internal/engine/epoll/async_protocol_edge_885_linux_test.go
index 982a2eb4e..e1cb697d5 100644
--- a/internal/engine/epoll/async_protocol_edge_885_linux_test.go
+++ b/internal/engine/epoll/async_protocol_edge_885_linux_test.go
@@ -6,9 +6,12 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"io"
 	"net"
+	"net/http"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -33,6 +36,7 @@ import (
 type edge885Handler struct {
 	started chan struct{}
 	release chan struct{}
+	eng     atomic.Pointer[Engine]
 }
 
 func (h *edge885Handler) HandleStream(_ context.Context, s *stream.Stream) error {
@@ -42,6 +46,14 @@ func (h *edge885Handler) HandleStream(_ context.Context, s *stream.Stream) error
 			path = hd[1]
 		}
 	}
+	if path == "/arm" {
+		// Runs inline on the loop thread of the conn it came on: the pool wait's
+		// gate (poolWaitGate) sets refuseRecv there, and so does this.
+		h.eng.Load().loops[s.WorkerID].refuseRecv = true
+	}
+	if path == "/who" {
+		path = strconv.Itoa(int(s.WorkerID))
+	}
 	if path == "/hold" {
 		select {
 		case h.started <- struct{}{}:
@@ -62,13 +74,62 @@ func (*edge885Handler) RouteAsync(_, path string) bool { return path == "/a" }
 func (*edge885Handler) HasAsyncRoutes() bool           { return true }
 
 func TestAsyncGateProtocolReadHasAnEdgeWithTheUpgradeSwitch885(t *testing.T) {
+	runEdge885(t, false)
+}
+
+// TestRefusesRecvReadsProtocolInOrderWithTheUpgradeSwitch885 is the same
+// scenario with the pool wait's gate up (refuseRecv): drainRead then asks
+// refusesRecv about the conn before anything else, and refusesRecv read
+// cs.protocol, which the dispatch goroutine writes in the switch. The flag is
+// set by a request on a second conn of the same loop, handled inline on the
+// loop thread.
+func TestRefusesRecvReadsProtocolInOrderWithTheUpgradeSwitch885(t *testing.T) {
+	runEdge885(t, true)
+}
+
+func runEdge885(t *testing.T, gate bool) {
 	h := &edge885Handler{started: make(chan struct{}, 1), release: make(chan struct{})}
-	_, addr := startE2E885(t, true, h, func(c *resource.Config) { c.EnableH2Upgrade = true })
+	e, addr := startE2E885(t, true, h, func(c *resource.Config) { c.EnableH2Upgrade = true })
+	h.eng.Store(e)
 	c, err := net.DialTimeout("tcp", addr, 3*time.Second)
 	if err != nil {
 		t.Fatalf("dial: %v", err)
 	}
 	defer func() { _ = c.Close() }()
+	br := bufio.NewReader(c)
+
+	var other net.Conn
+	if gate {
+		who := func(c net.Conn, br *bufio.Reader, path string) string {
+			_ = c.SetDeadline(time.Now().Add(10 * time.Second))
+			if _, err := c.Write([]byte("GET " + path + " HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil {
+				t.Fatalf("write %s: %v", path, err)
+			}
+			resp, err := http.ReadResponse(br, nil)
+			if err != nil {
+				t.Fatalf("read %s: %v", path, err)
+			}
+			b, _ := io.ReadAll(resp.Body)
+			_ = resp.Body.Close()
+			return string(b)
+		}
+		w := who(c, br, "/who")
+		for i := 0; i < 64 && other == nil; i++ {
+			oc, err := net.DialTimeout("tcp", addr, 3*time.Second)
+			if err != nil {
+				t.Fatalf("dial: %v", err)
+			}
+			if who(oc, bufio.NewReader(oc), "/who") == w {
+				other = oc
+			} else {
+				_ = oc.Close()
+			}
+		}
+		if other == nil {
+			t.Fatal("no second conn on the first conn's loop in 64 dials")
+		}
+		defer func() { _ = other.Close() }()
+	}
 
 	var hb bytes.Buffer
 	enc := hpack.NewEncoder(&hb)
@@ -90,6 +151,18 @@ func TestAsyncGateProtocolReadHasAnEdgeWithTheUpgradeSwitch885(t *testing.T) {
 	case <-time.After(10 * time.Second):
 		t.Fatal("the goroutine never reached the HTTP/2 stream's handler after the switch")
 	}
+	if gate {
+		// The gate goes up on the loop: this request runs inline on its thread.
+		_ = other.SetDeadline(time.Now().Add(10 * time.Second))
+		if _, err := other.Write([]byte("GET /arm HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil {
+			t.Fatalf("write /arm: %v", err)
+		}
+		resp, err := http.ReadResponse(bufio.NewReader(other), nil)
+		if err != nil {
+			t.Fatalf("read /arm: %v", err)
+		}
+		_ = resp.Body.Close()
+	}
 	// The goroutine is inside the switch, the protocol written. The client
 	// sends a PING: the loop reads it and reads cs.protocol.
 	var ping bytes.Buffer
@@ -101,7 +174,6 @@ func TestAsyncGateProtocolReadHasAnEdgeWithTheUpgradeSwitch885(t *testing.T) {
 	close(h.release)
 
 	// The conn goes on: the response to /a and the 101 arrive.
-	br := bufio.NewReader(c)
 	_ = c.SetReadDeadline(time.Now().Add(10 * time.Second))
 	var got strings.Builder
 	for {
diff --git a/internal/engine/epoll/drainread_budget_885_linux_test.go b/internal/engine/epoll/drainread_budget_885_linux_test.go
index 65b07e77f..36529c0d0 100644
--- a/internal/engine/epoll/drainread_budget_885_linux_test.go
+++ b/internal/engine/epoll/drainread_budget_885_linux_test.go
@@ -340,3 +340,47 @@ func TestDrainReadBudgetEndToEnd885(t *testing.T) {
 		})
 	}
 }
+
+// TestRunLoopHandlesAHalfCloseAfterAYieldedTurn885: the run loop skips the
+// half-close, error and transplant steps of an event whose turn yielded
+// (Loop.readYielded), and clears the flag before every drainRead. Without the
+// clear, the first yield on a loop would leave the flag set for good: every
+// later read event on it would skip its steps, and an edge-triggered EPOLLRDHUP
+// is not reported again. Two conns on one loop: A floods and yields (the budget
+// is 16 KiB, the burst 256 KiB); then B sends a request and its FIN, and the
+// server must end B.
+func TestRunLoopHandlesAHalfCloseAfterAYieldedTurn885(t *testing.T) {
+	withBudget885(t, 16<<10) // before the engine starts: the loops read it
+	e := startFloodEngine885(t, 0)
+	a, b, _ := threeConns885(t, e)
+
+	const req = "GET /flood HTTP/1.1\r\nHost: x\r\n\r\n"
+	n := (256 << 10) / len(req)
+	go func() { _, _ = io.Copy(a.c, strings.NewReader(strings.Repeat(req, n))) }()
+	for i := 0; i < n; i++ {
+		_ = a.c.SetReadDeadline(time.Now().Add(10 * time.Second))
+		resp, err := http.ReadResponse(a.br, nil)
+		if err != nil {
+			t.Fatalf("flood response %d of %d: %v", i, n, err)
+		}
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+	}
+
+	_ = b.c.SetDeadline(time.Now().Add(10 * time.Second))
+	if _, err := b.c.Write([]byte(req)); err != nil {
+		t.Fatalf("write B: %v", err)
+	}
+	if err := b.c.(*net.TCPConn).CloseWrite(); err != nil {
+		t.Fatalf("CloseWrite: %v", err)
+	}
+	resp, err := http.ReadResponse(b.br, nil)
+	if err != nil {
+		t.Fatalf("B's response: %v", err)
+	}
+	_, _ = io.Copy(io.Discard, resp.Body)
+	_ = resp.Body.Close()
+	if _, err := b.br.ReadByte(); err != io.EOF {
+		t.Fatalf("B sent its FIN after a yielded turn on its loop and the server did not end it (read: %v): the half-close was skipped", err)
+	}
+}

From 84e17ab4b9a9afe3c85e89b91f813cce950f04f4 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 23:06:32 +0100
Subject: [PATCH 09/12] epoll: refusesRecv reads h2Switched under asyncInMu for
 a promoted conn, not cs.protocol; comments on the conservative skips (v1.6.0,
 Refs #885 items 28, 41)

---
 .../epoll/drainread_budget_885_linux_test.go  |  9 +++++---
 internal/engine/epoll/loop.go                 | 23 ++++++++++++++++---
 internal/engine/epoll/poolwait_gate_885.go    | 17 +++++++++++---
 3 files changed, 40 insertions(+), 9 deletions(-)

diff --git a/internal/engine/epoll/drainread_budget_885_linux_test.go b/internal/engine/epoll/drainread_budget_885_linux_test.go
index 36529c0d0..a6a4a304a 100644
--- a/internal/engine/epoll/drainread_budget_885_linux_test.go
+++ b/internal/engine/epoll/drainread_budget_885_linux_test.go
@@ -252,9 +252,12 @@ func TestDrainReadBudgetSparesAClosingConn885(t *testing.T) {
 // slowFirst885 is pathHandler885 whose /slow answers after 300 ms: the loop
 // (a sync engine) or the conn's dispatch goroutine (an async one) is held on
 // the first request of a burst while the rest of it, written at once into the
-// socket's large buffers, is all in the kernel when reading starts: bytes the
-// budget leaves in the socket are not followed by any new segment, so only the
-// re-armed readiness reads them.
+// socket's large buffers, collects in the kernel: most of it where the buffers
+// are capped (a CI runner's net.core.wmem_max and rmem_max give about 425 KB,
+// and an explicit SO_RCVBUF turns autotuning off), all of it where they are
+// not. The bytes the budget leaves in the socket at the end of the burst are
+// not followed by any new segment, so only the re-armed readiness reads them:
+// that tail is what tests the re-arm.
 type slowFirst885 struct{ *pathHandler885 }
 
 func (h slowFirst885) HandleStream(ctx context.Context, s *stream.Stream) error {
diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index 23810e7bd..7a01c4126 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -770,6 +770,9 @@ func (l *Loop) run(ctx context.Context) {
 				l.onPeerHalfClose(fd)
 			}
 
+			// Conservative: EPOLLHUP with bytes unread needs both directions shut
+			// and an error purges the receive queue, so the bytes are rarely
+			// there; the re-poll reports both again on the next turn.
 			if !yielded && ev.Events&(unix.EPOLLERR|unix.EPOLLHUP) != 0 {
 				if fd >= 0 && fd < len(l.conns) && l.conns[fd] != nil {
 					l.closeConn(fd)
@@ -782,7 +785,9 @@ func (l *Loop) run(ctx context.Context) {
 			// the slot, so a close above is harmless. The pointer load is the
 			// whole cost when no drain is active. Not after a turn that ended at
 			// its budget (yielded): that conn is mid-flood, its next turn will
-			// find it at a boundary or not.
+			// find it at a boundary or not. Conservative: a hand-off with bytes
+			// still in the socket would be safe, the target's first read gets
+			// them.
 			if !yielded && l.transplant.Load() != nil {
 				l.tryTransplant(fd)
 			}
@@ -1570,8 +1575,10 @@ var drainReadBudget = 64 << 10
 // recv is paused). It reports whether drainRead should return: true when the
 // MOD was issued, or when an async conn has been hijacked, and the conn is
 // gone from this loop. False means keep reading, as before the budget: the MOD
-// failed, or the conn is closing (closeWhenFlushed's mask has no EPOLLIN by
-// design and a MOD with it would undo that). Loop thread.
+// failed, or the conn is closing (peerClosed: closeWhenFlushed's mask has no
+// EPOLLIN by design and a MOD with it would undo that; onPeerHalfClose's mark
+// keeps EPOLLIN, but after a FIN the inflow is bounded by the receive queue).
+// Loop thread.
 func (l *Loop) yieldRead(cs *connState) bool {
 	if cs.peerClosed {
 		return false
@@ -1584,6 +1591,16 @@ func (l *Loop) yieldRead(cs *connState) bool {
 	return false
 }
 
+// switchedToH2 reports h2Switched, read under asyncInMu: true orders the
+// caller's later reads of protocol, h1State and h2State after the dispatch
+// goroutine's writes (celeris#885 item 28). Loop thread.
+func (cs *connState) switchedToH2() bool {
+	cs.asyncInMu.Lock()
+	v := cs.h2Switched
+	cs.asyncInMu.Unlock()
+	return v
+}
+
 // lockAsyncInUnlessSwitched is the loop thread's test of drainRead's async
 // gate: it takes asyncInMu and, if the dispatch goroutine has not switched the
 // conn to HTTP/2 (h2Switched), returns true with the lock held, for the
diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go
index 03e59c0b1..68c5a1eaf 100644
--- a/internal/engine/epoll/poolwait_gate_885.go
+++ b/internal/engine/epoll/poolwait_gate_885.go
@@ -62,16 +62,27 @@ const gateClosesPerPass = 64
 // io_uring does not make: a request in progress is read on, and answered, as
 // std's Shutdown answers the request of an active conn (headers in pieces, a
 // body still arriving, the bytes of a protocol decision begun). A conn whose
-// dispatch goroutine owns its parse state (asyncPromoted) is refused as before:
-// its state is not the loop's to read. Loop thread.
+// dispatch goroutine owns its parse state (asyncPromoted) is refused as before
+// while it is HTTP/1: its state is not the loop's to read. Loop thread.
 func refusesRecv(cs *connState) bool {
 	if !cs.detected {
 		return cs.detectN == 0
 	}
+	if cs.asyncPromoted {
+		// protocol, h1State and h2State are the dispatch goroutine's, which
+		// writes them in switchToH2Local under detachMu: not read here. The
+		// conn is HTTP/1 until the goroutine sets h2Switched (under asyncInMu,
+		// which is held for an instant, never across a handler, so waiting for
+		// it cannot stall the pass), and an HTTP/1 conn it owns is refused, as
+		// before; one it has switched is HTTP/2 and is served (celeris#885
+		// item 28). Not refused on asyncPromoted alone: the pool wait is for
+		// the HTTP/2 conns.
+		return !cs.switchedToH2()
+	}
 	if cs.protocol != engine.HTTP1 {
 		return false
 	}
-	return cs.asyncPromoted || cs.h1State == nil || !cs.h1State.HasPendingData()
+	return cs.h1State == nil || !cs.h1State.HasPendingData()
 }
 
 // poolWaitGate applies the gate once the loop has waited poolWaitGrace for the

From 69fefc20b355f07e88d6cb1dacb361c0aab4dec2 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 23:38:43 +0100
Subject: [PATCH 10/12] ci: the root race step's skip allow-list names the
 opt-in measurement TestProbeDrainReadFloodLatency885 (v1.6.0, Refs #885 item
 41)

---
 .github/workflows/ci.yml | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index b5183977c..ba2e16b03 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -222,9 +222,10 @@ jobs:
           #     workers and the runner's 8 MiB funds one; the step "celeris#592
           #     settled-route retime rig" below runs them with memlock raised and
           #     CELERIS_REQUIRE_IOURING_WORKERS=1, where a skip is a failure.
-          #   * TestRouteAdaptive_SettleReopenCost and
-          #     TestWriteBufBackpressureClosesSlowConsumer: opt-in measurements
-          #     (CELERIS_592_COST=1, GOTEST_BACKPRESSURE=1), not checks.
+          #   * TestRouteAdaptive_SettleReopenCost,
+          #     TestWriteBufBackpressureClosesSlowConsumer and
+          #     TestProbeDrainReadFloodLatency885: opt-in measurements
+          #     (CELERIS_592_COST=1, GOTEST_BACKPRESSURE=1, CELERIS_PROBE_885=1), not checks.
           #   * the four alloc-count tests in middleware/sse: they skip
           #     under -race and every step of this job is -race (celeris#684);
           #     the step "celeris#890 allocation guards and alloc-count tests,
@@ -239,7 +240,7 @@ jobs:
           # too. That is safe only while this step excludes /engine/iouring$ (the
           # pkgs line above; the step after it runs that package with its own list).
           # A name added here allows that name in every package this step runs.
-          allowed='TestAdaptiveSettledRouteRetime592/iouring/settled|TestAdaptiveSettledRouteRetime592/iouring/negctrl_async|TestAdaptiveSettledRouteRetime592/iouring/negctrl_learning|TestRouteAdaptive_SettleReopenCost|TestWriteBufBackpressureClosesSlowConsumer|TestBrokerFanOutFormatsOnce|TestRingBufferAppendAllocs|TestSlowClientSendNonFullQueueAllocs|TestSlowClientDropPathAllocs888|TestPauseAcceptServesDeferredIdleConnectionsSynackRetriesZero|TestPauseAcceptNoGuardLosesAtSynackRetriesZero|TestProbeNonLinux|TestValidateEngineRequiresLinux'
+          allowed='TestAdaptiveSettledRouteRetime592/iouring/settled|TestAdaptiveSettledRouteRetime592/iouring/negctrl_async|TestAdaptiveSettledRouteRetime592/iouring/negctrl_learning|TestRouteAdaptive_SettleReopenCost|TestWriteBufBackpressureClosesSlowConsumer|TestProbeDrainReadFloodLatency885|TestBrokerFanOutFormatsOnce|TestRingBufferAppendAllocs|TestSlowClientSendNonFullQueueAllocs|TestSlowClientDropPathAllocs888|TestPauseAcceptServesDeferredIdleConnectionsSynackRetriesZero|TestPauseAcceptNoGuardLosesAtSynackRetriesZero|TestProbeNonLinux|TestValidateEngineRequiresLinux'
           required='TestSendfileEndToEndLargeFile|TestSendfileHEADNoBody|TestSendfileRangeRequest|TestSendfileSubThresholdFallback'
           want=$(( $(tr '|' '\n' <<<"$required" | wc -l) ))
           ran=$(grep -cE "^=== RUN   (${required})\$" /tmp/root.log || true)

From 8ec0b6cfe75aa8fd3e7391ae287222e30adb5224 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Tue, 27 Oct 2026 00:19:59 +0100
Subject: [PATCH 11/12] epoll: tests that stopDetachedProducers reads h1State
 in order with the h2c switch (fails on main under -race), that a switched
 promoted conn is served past the gate (PING ACK), and a half-close test that
 holds the loop (v1.6.0, Refs #885 items 28, 41)

---
 .../async_protocol_edge_885_linux_test.go     | 48 +++++++++++++++++--
 .../epoll/drainread_budget_885_linux_test.go  | 25 ++++++++--
 .../epoll/drainread_flood_885_linux_test.go   |  8 +++-
 3 files changed, 72 insertions(+), 9 deletions(-)

diff --git a/internal/engine/epoll/async_protocol_edge_885_linux_test.go b/internal/engine/epoll/async_protocol_edge_885_linux_test.go
index e1cb697d5..3dabdd2d6 100644
--- a/internal/engine/epoll/async_protocol_edge_885_linux_test.go
+++ b/internal/engine/epoll/async_protocol_edge_885_linux_test.go
@@ -51,6 +51,11 @@ func (h *edge885Handler) HandleStream(_ context.Context, s *stream.Stream) error
 		// gate (poolWaitGate) sets refuseRecv there, and so does this.
 		h.eng.Load().loops[s.WorkerID].refuseRecv = true
 	}
+	if path == "/sdp" {
+		// The pool wait gate's pass over the detached conns, which runs on the
+		// loop thread before closeIdleH1.
+		h.eng.Load().loops[s.WorkerID].stopDetachedProducers()
+	}
 	if path == "/who" {
 		path = strconv.Itoa(int(s.WorkerID))
 	}
@@ -74,7 +79,7 @@ func (*edge885Handler) RouteAsync(_, path string) bool { return path == "/a" }
 func (*edge885Handler) HasAsyncRoutes() bool           { return true }
 
 func TestAsyncGateProtocolReadHasAnEdgeWithTheUpgradeSwitch885(t *testing.T) {
-	runEdge885(t, false)
+	runEdge885(t, "")
 }
 
 // TestRefusesRecvReadsProtocolInOrderWithTheUpgradeSwitch885 is the same
@@ -84,10 +89,25 @@ func TestAsyncGateProtocolReadHasAnEdgeWithTheUpgradeSwitch885(t *testing.T) {
 // set by a request on a second conn of the same loop, handled inline on the
 // loop thread.
 func TestRefusesRecvReadsProtocolInOrderWithTheUpgradeSwitch885(t *testing.T) {
-	runEdge885(t, true)
+	runEdge885(t, "/arm")
+}
+
+// TestStopDetachedProducersReadsH1StateInOrderWithTheUpgradeSwitch885: the same
+// scenario, the gate's pass over the detached conns (stopDetachedProducers) run
+// on the loop thread, from a request on a second conn, while the goroutine is
+// inside the switch. The pass read cs.h1State, which the switch sets to nil,
+// before it took detachMu.
+func TestStopDetachedProducersReadsH1StateInOrderWithTheUpgradeSwitch885(t *testing.T) {
+	runEdge885(t, "/sdp")
 }
 
-func runEdge885(t *testing.T, gate bool) {
+// runEdge885 runs the scenario; via names the request, on a second conn of the
+// same loop, whose handler runs a loop-thread step of the pool wait's gate while
+// the dispatch goroutine is inside the switch ("" for none). After the 101 the
+// conn must speak HTTP/2: a PING is answered, so a gate that refused every
+// promoted conn (including a switched one) fails it.
+func runEdge885(t *testing.T, via string) {
+	gate := via != ""
 	h := &edge885Handler{started: make(chan struct{}, 1), release: make(chan struct{})}
 	e, addr := startE2E885(t, true, h, func(c *resource.Config) { c.EnableH2Upgrade = true })
 	h.eng.Store(e)
@@ -154,7 +174,7 @@ func runEdge885(t *testing.T, gate bool) {
 	if gate {
 		// The gate goes up on the loop: this request runs inline on its thread.
 		_ = other.SetDeadline(time.Now().Add(10 * time.Second))
-		if _, err := other.Write([]byte("GET /arm HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil {
+		if _, err := other.Write([]byte("GET " + via + " HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil {
 			t.Fatalf("write /arm: %v", err)
 		}
 		resp, err := http.ReadResponse(bufio.NewReader(other), nil)
@@ -183,7 +203,25 @@ func runEdge885(t *testing.T, gate bool) {
 			t.Fatalf("read the response to /a and the 101: %v (got %q)", err, got.String())
 		}
 		if strings.Contains(line, "HTTP/1.1 101") {
-			return
+			for {
+				l, err := br.ReadString('\n')
+				if err != nil {
+					t.Fatalf("the 101's headers: %v", err)
+				}
+				if l == "\r\n" {
+					break
+				}
+			}
+			fr := http2.NewFramer(nil, br)
+			for {
+				f, err := fr.ReadFrame()
+				if err != nil {
+					t.Fatalf("no PING ACK on the switched conn (via %q): %v", via, err)
+				}
+				if pf, ok := f.(*http2.PingFrame); ok && pf.IsAck() {
+					return
+				}
+			}
 		}
 	}
 }
diff --git a/internal/engine/epoll/drainread_budget_885_linux_test.go b/internal/engine/epoll/drainread_budget_885_linux_test.go
index a6a4a304a..88e00a204 100644
--- a/internal/engine/epoll/drainread_budget_885_linux_test.go
+++ b/internal/engine/epoll/drainread_budget_885_linux_test.go
@@ -344,17 +344,31 @@ func TestDrainReadBudgetEndToEnd885(t *testing.T) {
 	}
 }
 
+// sleepFlood885 is floodHandler885 whose /sleep holds the loop thread for 300 ms
+// (a sync engine runs the handler inline).
+type sleepFlood885 struct{ floodHandler885 }
+
+func (h sleepFlood885) HandleStream(ctx context.Context, s *stream.Stream) error {
+	if s.Path == "/sleep" {
+		time.Sleep(300 * time.Millisecond)
+	}
+	return h.floodHandler885.HandleStream(ctx, s)
+}
+
 // TestRunLoopHandlesAHalfCloseAfterAYieldedTurn885: the run loop skips the
 // half-close, error and transplant steps of an event whose turn yielded
 // (Loop.readYielded), and clears the flag before every drainRead. Without the
 // clear, the first yield on a loop would leave the flag set for good: every
 // later read event on it would skip its steps, and an edge-triggered EPOLLRDHUP
 // is not reported again. Two conns on one loop: A floods and yields (the budget
-// is 16 KiB, the burst 256 KiB); then B sends a request and its FIN, and the
-// server must end B.
+// is 16 KiB, the burst 256 KiB); then A holds the loop thread in a 300 ms
+// handler, and B's request and its FIN are both queued behind it, so that B's
+// read is a short read that does not see the FIN and only the run loop's
+// EPOLLRDHUP step can end B (a FIN that comes after the read is an event of
+// its own, read as EOF whatever the flag says). The server must end B.
 func TestRunLoopHandlesAHalfCloseAfterAYieldedTurn885(t *testing.T) {
 	withBudget885(t, 16<<10) // before the engine starts: the loops read it
-	e := startFloodEngine885(t, 0)
+	e := startFloodEngineWith885(t, 0, sleepFlood885{})
 	a, b, _ := threeConns885(t, e)
 
 	const req = "GET /flood HTTP/1.1\r\nHost: x\r\n\r\n"
@@ -369,6 +383,11 @@ func TestRunLoopHandlesAHalfCloseAfterAYieldedTurn885(t *testing.T) {
 		_, _ = io.Copy(io.Discard, resp.Body)
 		_ = resp.Body.Close()
 	}
+	// Hold the loop thread, then queue B's request and FIN behind it.
+	if _, err := a.c.Write([]byte("GET /sleep HTTP/1.1\r\nHost: x\r\n\r\n")); err != nil {
+		t.Fatalf("write /sleep: %v", err)
+	}
+	time.Sleep(100 * time.Millisecond)
 
 	_ = b.c.SetDeadline(time.Now().Add(10 * time.Second))
 	if _, err := b.c.Write([]byte(req)); err != nil {
diff --git a/internal/engine/epoll/drainread_flood_885_linux_test.go b/internal/engine/epoll/drainread_flood_885_linux_test.go
index fe270c4d6..b18278d42 100644
--- a/internal/engine/epoll/drainread_flood_885_linux_test.go
+++ b/internal/engine/epoll/drainread_flood_885_linux_test.go
@@ -53,6 +53,12 @@ func (floodHandler885) HandleStream(_ context.Context, s *stream.Stream) error {
 }
 
 func startFloodEngine885(t testing.TB, bufSize int) *Engine {
+	t.Helper()
+	return startFloodEngineWith885(t, bufSize, floodHandler885{})
+}
+
+// startFloodEngineWith885 is startFloodEngine885 with the handler h.
+func startFloodEngineWith885(t testing.TB, bufSize int, h stream.Handler) *Engine {
 	t.Helper()
 	ln, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
@@ -68,7 +74,7 @@ func startFloodEngine885(t testing.TB, bufSize int) *Engine {
 			BufferSize: bufSize,
 		},
 	}
-	e, err := New(cfg, floodHandler885{})
+	e, err := New(cfg, h)
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}

From 2d73cd05de2bfd429f1449da37af81831910f2be Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Tue, 27 Oct 2026 00:27:08 +0100
Subject: [PATCH 12/12] epoll: stopDetachedProducers reads a promoted conn's
 h1State under detachMu, as its io_uring twin does; the comment on the ERR/HUP
 skip says what an RST leaves (v1.6.0, Refs #885 item 28)

---
 internal/engine/epoll/loop.go              |  8 +++++---
 internal/engine/epoll/poolwait_gate_885.go | 13 ++++++++++++-
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index 7a01c4126..6bc92dfab 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -770,9 +770,11 @@ func (l *Loop) run(ctx context.Context) {
 				l.onPeerHalfClose(fd)
 			}
 
-			// Conservative: EPOLLHUP with bytes unread needs both directions shut
-			// and an error purges the receive queue, so the bytes are rarely
-			// there; the re-poll reports both again on the next turn.
+			// Conservative: bytes may be unread here (an RST does not purge the
+			// receive queue on Linux: a read returns them, and only the next one
+			// ECONNRESET), and nothing can be answered on the conn, but its next
+			// turn reads them and then the error, bounded by the receive queue;
+			// the re-poll reports ERR and HUP again with it.
 			if !yielded && ev.Events&(unix.EPOLLERR|unix.EPOLLHUP) != 0 {
 				if fd >= 0 && fd < len(l.conns) && l.conns[fd] != nil {
 					l.closeConn(fd)
diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go
index 68c5a1eaf..01b57ca19 100644
--- a/internal/engine/epoll/poolwait_gate_885.go
+++ b/internal/engine/epoll/poolwait_gate_885.go
@@ -191,12 +191,23 @@ func (l *Loop) stopDetachedProducers() {
 	for i := len(l.liveConns) - 1; i >= 0; i-- {
 		cs := l.liveConns[i]
 		mu := cs.detachMu
-		if mu == nil || cs.hijacked.Load() || cs.h1State == nil || !cs.h1State.Detached.Load() {
+		if mu == nil || cs.hijacked.Load() {
+			continue
+		}
+		// h1State is read before the lock only on a conn the loop owns: on a
+		// promoted conn the dispatch goroutine sets it to nil in the h2c switch,
+		// under detachMu, so it is read under the lock (celeris#885 item 28).
+		promoted := cs.asyncPromoted
+		if !promoted && (cs.h1State == nil || !cs.h1State.Detached.Load()) {
 			continue
 		}
 		if !mu.TryLock() {
 			continue
 		}
+		if promoted && (cs.h1State == nil || !cs.h1State.Detached.Load()) {
+			mu.Unlock()
+			continue
+		}
 		// Re-tested under the lock, as closeConn's ownership check is: an async
 		// Hijack takes the conn and closes its descriptor under this lock.
 		if cs.hijacked.Load() || (!cs.detachClosed && cs.asyncDetachPending) {