From 7117555cb684a8ed92730ce0b64d5f786beba419 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 21:52:28 +0100
Subject: [PATCH 01/17] test: failing assertions for the io_uring shutdown gate
 (idle/incomplete HTTP/1 conns, detached streams, requests in progress) and
 the ring-wait ceiling (v1.6.0, Refs #884)

---
 .../ring_wait_ceiling_884_linux_test.go       | 161 +++++++
 .../shutdown_iouring_gate_884_linux_test.go   | 410 ++++++++++++++++++
 2 files changed, 571 insertions(+)
 create mode 100644 internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
 create mode 100644 test/server/shutdown_iouring_gate_884_linux_test.go

diff --git a/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go b/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
new file mode 100644
index 000000000..33fea78e2
--- /dev/null
+++ b/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
@@ -0,0 +1,161 @@
+//go:build linux
+
+package iouring
+
+import (
+	"bufio"
+	"fmt"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// celeris#884 item 72: ringWaitContinues had no ceiling. A ring wait that
+// answers a retryable error (EBUSY, EAGAIN) for good, with nothing in the CQ
+// ring to reap, is a ring that is failing, not one whose CQ ring is full; the
+// loop retried it for ever (a 50 us sleep per pass), where before PR #1026 it
+// shut the worker down on the first error. The ceiling restores that end for a
+// run of about a second of retries that reaped nothing, and nothing shorter or
+// less stuck: a retry that finds a completion in the CQ ring (the loop's own
+// pass reaps it: the overloaded case the retry exists for) or one that follows
+// a wait that succeeded is not part of the run.
+
+// ceilingErr884 is the error of a stuck ring.
+func ceilingErr884() error { return fmt.Errorf("io_uring_enter submit+wait: %w", unix.EBUSY) }
+
+// TestRingWaitContinuesGivesUpOnAStuckRing884: retryable errors, one after the
+// other, on a CQ ring with nothing in it end the worker after about a second.
+func TestRingWaitContinuesGivesUpOnAStuckRing884(t *testing.T) {
+	ring := newTestRing(t)
+	logs := &warnLog884{}
+	w := &Worker{ring: ring, id: 3, logger: slog.New(logs)}
+
+	start := time.Now()
+	retries := 0
+	var gaveUpAfter time.Duration
+	for time.Since(start) < 5*time.Second {
+		if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
+			gaveUpAfter = time.Since(start)
+			break
+		}
+		retries++
+	}
+	t.Logf("celeris884: %d consecutive EBUSY waits with nothing to reap; gave up after %v", retries, gaveUpAfter)
+	if gaveUpAfter == 0 {
+		t.Fatalf("%d consecutive EBUSY waits over %v, with nothing in the CQ ring to reap, and the worker is still retrying: a failing ring is never given up on",
+			retries, time.Since(start).Round(time.Millisecond))
+	}
+	if gaveUpAfter < 500*time.Millisecond {
+		t.Errorf("the worker gave up on a ring after %v of retries (%d): too soon for a wait that only meets a full CQ ring", gaveUpAfter, retries)
+	}
+	if gaveUpAfter > 3*time.Second {
+		t.Errorf("the worker gave up on a ring after %v of retries: the ceiling is about a second", gaveUpAfter)
+	}
+	var errs int
+	logs.mu.Lock()
+	for _, r := range logs.recs {
+		if r.Level >= slog.LevelError {
+			errs++
+		}
+	}
+	logs.mu.Unlock()
+	if errs == 0 {
+		t.Error("the worker stopped on a failing ring and logged no error: its operator would see a stopped worker with no reason")
+	}
+}
+
+// TestRingWaitContinuesNeverGivesUpWhileItReaps884: the same error with a
+// completion waiting in the CQ ring at each retry is the loop being behind its
+// ring, not a failing ring: the loop's pass reaps, and the retry goes on for as
+// long as that lasts, well past the ceiling.
+func TestRingWaitContinuesNeverGivesUpWhileItReaps884(t *testing.T) {
+	ring := newTestRing(t)
+	w := &Worker{ring: ring, id: 3, logger: slog.New(&warnLog884{})}
+	postNOP884(t, ring)
+	start := time.Now()
+	n := 0
+	for time.Since(start) < 2500*time.Millisecond {
+		if !w.ringWaitContinues(waitModeSubmitSends, ceilingErr884()) {
+			t.Fatalf("the worker gave up after %v and %d retries, each of which found a completion in the CQ ring to reap",
+				time.Since(start).Round(time.Millisecond), n)
+		}
+		n++
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// TestRingWaitContinuesCountsOnlyConsecutiveRetries884: retries that come
+// apart, each after waits that succeeded (an idle worker's timed wait takes up
+// to a second; a retry 400 ms after the last), are not a stuck ring: a worker that meets an EBUSY now and
+// then, with nothing to reap each time, runs on.
+func TestRingWaitContinuesCountsOnlyConsecutiveRetries884(t *testing.T) {
+	ring := newTestRing(t)
+	w := &Worker{ring: ring, id: 3, logger: slog.New(&warnLog884{})}
+	start := time.Now()
+	n := 0
+	for time.Since(start) < 3*time.Second {
+		if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
+			t.Fatalf("the worker gave up after %v and %d retries, each apart from the last by 400 ms of waits that succeeded",
+				time.Since(start).Round(time.Millisecond), n)
+		}
+		n++
+		time.Sleep(400 * time.Millisecond)
+	}
+}
+
+// TestRunStopsOnAStuckRingWait884: the engine's own arm. One worker's waits
+// answer EBUSY for good (the seam) on an engine with nothing to do, so there is
+// nothing to reap; it stops within a few seconds, as it does for any other ring
+// failure (TestRunStopsOnARingWaitFailure884), and the other worker is left to
+// serve.
+func TestRunStopsOnAStuckRingWait884(t *testing.T) {
+	inj := &ebusyInject884{err: ebusyErr884(), n: 1 << 40, firstOnly: true}
+	inj.only.Store(-1)
+	setRingWaitSeam884(t, inj.seam)
+	e, addr := startFDLEngine(t, transplantTestHandler{}, nil)
+
+	time.Sleep(200 * time.Millisecond)
+	armedAt := time.Now()
+	inj.armed.Store(true)
+	var dead *Worker
+	for deadline := time.Now().Add(6 * time.Second); time.Now().Before(deadline) && dead == nil; time.Sleep(10 * time.Millisecond) {
+		for _, w := range e.workers {
+			if w.exited.Load() {
+				dead = w
+			}
+		}
+	}
+	t.Logf("celeris884 stuck ring: injected=%d%s", inj.total(-1), inj.summary())
+	if inj.total(-1) == 0 {
+		t.Fatal("the injected EBUSY never reached a wait: the test would prove nothing")
+	}
+	if dead == nil {
+		t.Fatalf("a worker whose every ring wait answered EBUSY for %v, with nothing to reap, was still running: the retry has no ceiling", time.Since(armedAt).Round(time.Millisecond))
+	}
+	t.Logf("celeris884 stuck ring: worker %d stopped %v after the injection began", dead.id, time.Since(armedAt).Round(time.Millisecond))
+	if dead != inj.chosen.Load() {
+		t.Errorf("worker %d stopped, but the EBUSY was answered to worker %d", dead.id, inj.chosen.Load().id)
+	}
+	for _, w := range e.workers {
+		if w != dead && w.exited.Load() {
+			t.Errorf("worker %d stopped though only worker %d met the failure", w.id, dead.id)
+		}
+	}
+	c, err := net.DialTimeout("tcp", addr, time.Second)
+	if err != nil {
+		t.Fatalf("dial after the stuck worker stopped: %v", err)
+	}
+	defer func() { _ = c.Close() }()
+	if _, err := io.WriteString(c, "GET / HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
+		t.Fatal(err)
+	}
+	_ = c.SetReadDeadline(time.Now().Add(3 * time.Second))
+	if _, err := http.ReadResponse(bufio.NewReader(c), nil); err != nil {
+		t.Errorf("the surviving worker did not answer: %v", err)
+	}
+}
diff --git a/test/server/shutdown_iouring_gate_884_linux_test.go b/test/server/shutdown_iouring_gate_884_linux_test.go
new file mode 100644
index 000000000..814d4663a
--- /dev/null
+++ b/test/server/shutdown_iouring_gate_884_linux_test.go
@@ -0,0 +1,410 @@
+//go:build linux
+
+package celeris_test
+
+import (
+	"bufio"
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/goceleris/celeris"
+)
+
+// TestIOUringShutdownGateConns884 pins what the io_uring engine does with the
+// HTTP/1 conns it stops reading once its shutdown has passed the gate
+// (celeris#884 item 73, the io_uring twin of
+// celeris#885 item 49 / PR #1116 on epoll). The gate is the first 250 ms of
+// the wait for HTTP/2 pool handlers ("pool-wait") or of the send drain
+// ("send-drain"); it is held open for the whole 20 s budget by clients that
+// take nothing, so every assertion below is about the gate, not about the end
+// of the wait (which closes everything). Per arm:
+//
+//   - an idle keep-alive conn, a conn that was accepted and sent nothing (a
+//     browser's preconnect), and, on a server with an Async route, an idle conn
+//     that route promoted to a dispatch goroutine each get a FIN (EOF, not a
+//     reset, not silence) at once, as std's Shutdown closes its idle conns;
+//   - a detached stream (Server-Sent Events) the gate ended receives what was
+//     queued for it, whole, and then a FIN, on a sync route and on an Async one
+//     (a promoted conn: epoll's r3 M1);
+//   - a request in progress when the shutdown begins (a 2 MiB POST of which
+//     64 KiB has arrived, the same on an Async route, a GET whose headers are
+//     in pieces) is read on and answered, as std's Shutdown answers the request
+//     of an active conn, with Connection: close, and its conn is then closed;
+//   - a 16 MiB download queued for a client that reads slowly arrives whole and
+//     its conn is then closed.
+//
+// On main the gate reads nothing past it and closes nothing: every FIN and every
+// answer above is missing until the end of the wait.
+func TestIOUringShutdownGateConns884(t *testing.T) {
+	type arm struct {
+		name  string
+		eng   celeris.EngineType
+		setup func(*testing.T)
+		gate  string // "pool-wait" or "send-drain"
+		async bool   // a server with an Async route, whose conns can be promoted
+	}
+	noEnv := func(*testing.T) {}
+	var arms []arm
+	for _, gate := range []string{"pool-wait", "send-drain"} {
+		for _, async := range []bool{false, true} {
+			name := gate + "/sync"
+			if async {
+				name = gate + "/async-route"
+			}
+			arms = append(arms, arm{name, celeris.IOUring, noEnv, gate, async})
+		}
+	}
+	arms = append(arms, arm{"adaptive-iouring/pool-wait/sync", celeris.Adaptive,
+		func(t *testing.T) { t.Setenv("CELERIS_ADAPTIVE_START", "iouring") }, "pool-wait", false})
+
+	for _, a := range arms {
+		t.Run(a.name, func(t *testing.T) {
+			a.setup(t)
+			runIOUringGateArm884(t, a.eng, a.gate, a.async)
+		})
+	}
+}
+
+func runIOUringGateArm884(t *testing.T, eng celeris.EngineType, gate string, async bool) {
+	const (
+		nBulk, chunk = 4, 256 << 10
+		upSize       = 2 << 20
+		upFirst      = 64 << 10
+		holdSize     = 4 << 20
+		nHold        = 48
+	)
+	dlBody := strings.Repeat("d", 16<<20)
+	holdBody := strings.Repeat("h", holdSize)
+	var okBytes [nBulk]atomic.Int64
+	srv := startH2PoolServer759(t, eng, 20*time.Second, func(s *celeris.Server) {
+		s.GET("/big", func(c *celeris.Context) error { return c.String(http.StatusOK, "%s", holdBody) })
+		s.GET("/dl", func(c *celeris.Context) error { return c.String(http.StatusOK, "%s", dlBody) })
+		s.POST("/up", func(c *celeris.Context) error { return c.String(http.StatusOK, "got %d", len(c.Body())) })
+		bulk := s.GET("/bulk/:i", func(c *celeris.Context) error {
+			i, convErr := strconv.Atoi(c.Param("i"))
+			done := c.Detach()
+			sw := c.StreamWriter()
+			if sw == nil || convErr != nil || i < 0 || i >= nBulk {
+				done()
+				return errors.New("no stream writer")
+			}
+			if err := sw.WriteHeader(http.StatusOK, [][2]string{{"content-type", "application/octet-stream"}}); err != nil {
+				done()
+				return err
+			}
+			go func() {
+				defer done()
+				buf := bytes.Repeat([]byte{'x'}, chunk)
+				for {
+					if _, err := sw.Write(buf); err != nil {
+						return
+					}
+					okBytes[i].Add(chunk)
+					time.Sleep(10 * time.Millisecond)
+				}
+			}()
+			return nil
+		})
+		if async {
+			bulk.Async()
+			s.GET("/aping", func(c *celeris.Context) error { return c.String(http.StatusOK, "a") }).Async()
+			s.POST("/aup", func(c *celeris.Context) error { return c.String(http.StatusOK, "got %d", len(c.Body())) }).Async()
+		}
+	})
+
+	// What holds the gate open: HTTP/2 responses nobody gives window to (the
+	// pool wait), or HTTP/1 responses queued for clients that read nothing (the
+	// send drain). Enough of them that every worker has some.
+	var holders []net.Conn
+	switch gate {
+	case "pool-wait":
+		for range nHold {
+			c, fr := dialH2759(t, srv.addr)
+			holders = append(holders, c)
+			writeH2Get759(t, fr, 1, "/big")
+		}
+	case "send-drain":
+		for range nHold {
+			c := dialSmallRcvbuf884(t, srv.addr)
+			holders = append(holders, c)
+			if _, err := io.WriteString(c, "GET /big HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	time.Sleep(300 * time.Millisecond)
+
+	keepAlive := func(path string) (net.Conn, *bufio.Reader) {
+		c, err := net.Dial("tcp", srv.addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = c.Close() })
+		br := bufio.NewReader(c)
+		if _, err := io.WriteString(c, "GET "+path+" HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
+			t.Fatal(err)
+		}
+		_ = c.SetReadDeadline(time.Now().Add(10 * time.Second))
+		resp, err := http.ReadResponse(br, nil)
+		if err != nil {
+			t.Fatalf("%s before the shutdown: %v", path, err)
+		}
+		_, _ = io.Copy(io.Discard, resp.Body)
+		return c, br
+	}
+	idle := make([]net.Conn, 6)
+	idleBr := make([]*bufio.Reader, 6)
+	for i := range idle {
+		idle[i], idleBr[i] = keepAlive("/ping")
+	}
+	silent := make([]net.Conn, 2)
+	for i := range silent {
+		c, err := net.Dial("tcp", srv.addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = c.Close() })
+		silent[i] = c
+	}
+	var promoted net.Conn
+	var promotedBr *bufio.Reader
+	if async {
+		promoted, promotedBr = keepAlive("/aping")
+	}
+
+	// Detached streams to clients that read nothing until later.
+	bulk := make([]net.Conn, nBulk)
+	for i := range bulk {
+		c, err := net.Dial("tcp", srv.addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = c.Close() })
+		bulk[i] = c
+		if _, err := io.WriteString(c, fmt.Sprintf("GET /bulk/%d HTTP/1.1\r\nHost: x\r\n\r\n", i)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for deadline := time.Now().Add(5 * time.Second); ; {
+		ready := true
+		for i := range okBytes {
+			ready = ready && okBytes[i].Load() >= 4*chunk
+		}
+		if ready {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("the producers wrote under 1 MiB each within 5s")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	// A download queued in the engine for a slow reader.
+	dl := dialSmallRcvbuf884(t, srv.addr)
+	if _, err := io.WriteString(dl, "GET /dl HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
+		t.Fatal(err)
+	}
+
+	// Requests in progress: the head and the first 64 KiB of a 2 MiB body, on a
+	// sync route and (when the server has an Async route) on an Async one, and a
+	// GET whose headers are in pieces.
+	type progress struct {
+		name string
+		c    net.Conn
+		rest []byte
+		want string
+	}
+	var inProgress []*progress
+	startPost := func(name, path string) {
+		c, err := net.Dial("tcp", srv.addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = c.Close() })
+		if _, err := io.WriteString(c, "POST "+path+" HTTP/1.1\r\nHost: x\r\nContent-Length: "+strconv.Itoa(upSize)+"\r\n\r\n"); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := c.Write(make([]byte, upFirst)); err != nil {
+			t.Fatal(err)
+		}
+		inProgress = append(inProgress, &progress{name, c, make([]byte, upSize-upFirst), "got " + strconv.Itoa(upSize)})
+	}
+	startPost("POST /up (sync route)", "/up")
+	if async {
+		startPost("POST /aup (Async route)", "/aup")
+	}
+	{
+		c, err := net.Dial("tcp", srv.addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = c.Close() })
+		if _, err := io.WriteString(c, "GET /ping HTTP/1.1\r\nHost: x\r\n"); err != nil {
+			t.Fatal(err)
+		}
+		inProgress = append(inProgress, &progress{"GET /ping with headers in pieces", c, []byte("\r\n"), "ok"})
+	}
+
+	// TCP_DEFER_ACCEPT hands the engine a conn with no byte after about a
+	// second: the silent conns are in its table by then.
+	time.Sleep(1300 * time.Millisecond)
+
+	begin := time.Now()
+	srv.beginShutdown("Shutdown", 20*time.Second)
+
+	expectFIN := func(name string, c net.Conn, br *bufio.Reader) {
+		t.Helper()
+		_ = c.SetReadDeadline(begin.Add(2500 * time.Millisecond))
+		var err error
+		if br != nil {
+			_, err = br.ReadByte()
+		} else {
+			_, err = c.Read(make([]byte, 1))
+		}
+		if err != io.EOF {
+			t.Errorf("%s: read %v at %v, want a FIN (io.EOF) once the gate has passed: the gate left the conn open with no answer", name, err, time.Since(begin).Round(time.Millisecond))
+		}
+	}
+	for i, c := range idle {
+		expectFIN("idle keep-alive "+strconv.Itoa(i), c, idleBr[i])
+	}
+	for i, c := range silent {
+		expectFIN("conn with no byte sent "+strconv.Itoa(i), c, nil)
+	}
+	if promoted != nil {
+		expectFIN("idle conn on an Async route", promoted, promotedBr)
+	}
+
+	// The rest of every request in progress arrives after the gate.
+	time.Sleep(time.Until(begin.Add(1200 * time.Millisecond)))
+	var pwg sync.WaitGroup
+	for _, p := range inProgress {
+		pwg.Add(1)
+		go func() {
+			defer pwg.Done()
+			_ = p.c.SetWriteDeadline(time.Now().Add(3 * time.Second))
+			if _, err := p.c.Write(p.rest); err != nil {
+				t.Errorf("%s: the rest of the request, sent %v into the shutdown: %v: a request in progress was cut", p.name, time.Since(begin).Round(time.Millisecond), err)
+				return
+			}
+			_ = p.c.SetReadDeadline(time.Now().Add(4 * time.Second))
+			br := bufio.NewReader(p.c)
+			resp, err := http.ReadResponse(br, nil)
+			if err != nil {
+				t.Errorf("%s: the response: %v: a request in progress when the gate came was not answered", p.name, err)
+				return
+			}
+			b, _ := io.ReadAll(resp.Body)
+			if string(b) != p.want {
+				t.Errorf("%s: the response body %q, want %q", p.name, b, p.want)
+			}
+			if !resp.Close {
+				t.Errorf("%s: the response, written after the gate, has no Connection: close", p.name)
+			}
+			_ = p.c.SetReadDeadline(time.Now().Add(3 * time.Second))
+			if _, err := br.ReadByte(); err != io.EOF {
+				t.Errorf("%s: after the response the conn read %v, want a FIN (io.EOF)", p.name, err)
+			}
+		}()
+	}
+
+	// What was queued for the detached streams is flushed, and then the conns
+	// are closed: every stream read at once.
+	type streamResult struct {
+		got, ok int64
+		err     error
+		end     error
+	}
+	results := make([]streamResult, nBulk)
+	var rwg sync.WaitGroup
+	for i, c := range bulk {
+		ok := okBytes[i].Load()
+		rwg.Add(1)
+		go func() {
+			defer rwg.Done()
+			_ = c.SetReadDeadline(time.Now().Add(15 * time.Second))
+			resp, err := http.ReadResponse(bufio.NewReader(c), nil)
+			if err != nil {
+				results[i] = streamResult{ok: ok, err: err}
+				return
+			}
+			var got int64
+			buf := make([]byte, 64<<10)
+			// okBytes was read before the producers were ended: what reached
+			// the engine after it may or may not go out, so the body is read
+			// to its end below, and only what had been written is required.
+			for got < ok {
+				n, err := resp.Body.Read(buf)
+				got += int64(n)
+				if err != nil {
+					results[i] = streamResult{got: got, ok: ok, err: err}
+					return
+				}
+			}
+			_ = c.SetReadDeadline(time.Now().Add(3 * time.Second))
+			var endErr error
+			for endErr == nil {
+				_, endErr = resp.Body.Read(buf)
+			}
+			results[i] = streamResult{got: got, ok: ok, end: endErr}
+		}()
+	}
+	rwg.Wait()
+	for i, r := range results {
+		t.Logf("celeris884 %s stream %d: %d of %d bytes written before the gate were received (%v)", gate, i, r.got, r.ok, r.err)
+		if r.err != nil || r.got < r.ok-chunk {
+			t.Errorf("stream %d: the client received %d of the %d bytes whose writes had succeeded (%v): what was queued was not flushed after the gate", i, r.got, r.ok, r.err)
+		} else if !errors.Is(r.end, io.EOF) && !errors.Is(r.end, io.ErrUnexpectedEOF) {
+			t.Errorf("stream %d: after the flushed bytes the client read %v, 3 s on, want a FIN (EOF, not a reset or silence): the gate ended the stream but did not close its conn", i, r.end)
+		}
+	}
+	pwg.Wait()
+
+	// The queued download arrives whole, and then its conn is closed.
+	_ = dl.SetReadDeadline(time.Now().Add(30 * time.Second))
+	dbr := bufio.NewReader(dl)
+	resp, err := http.ReadResponse(dbr, nil)
+	if err != nil {
+		t.Fatalf("the download's head: %v", err)
+	}
+	if n, err := io.Copy(io.Discard, resp.Body); n != int64(len(dlBody)) {
+		t.Errorf("the queued download delivered %d of %d bytes (%v): the gate cut a response that was still going out", n, len(dlBody), err)
+	}
+	_ = dl.SetReadDeadline(time.Now().Add(3 * time.Second))
+	if _, err := dbr.ReadByte(); err != io.EOF {
+		t.Errorf("after the download the conn read %v, want a FIN (io.EOF): a drained conn was left open", err)
+	}
+
+	for _, c := range holders { // the holders go: the wait ends
+		_ = c.Close()
+	}
+	srv.waitStart(t, 25*time.Second)
+}
+
+// dialSmallRcvbuf884 opens an HTTP/1 client whose receive buffer is small, so
+// a response queues in the engine instead of the kernel.
+func dialSmallRcvbuf884(t *testing.T, addr string) net.Conn {
+	t.Helper()
+	d := net.Dialer{Control: func(_, _ string, rc syscall.RawConn) error {
+		return rc.Control(func(fd uintptr) {
+			_ = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_RCVBUF, 32<<10)
+		})
+	}}
+	c, err := d.Dial("tcp", addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = c.Close() })
+	return c
+}

From 5e6b94c2352c0de4a4775a88acc2a92af9219183 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:00:51 +0100
Subject: [PATCH 02/17] iouring: the shutdown gate closes idle and ended HTTP/1
 conns and reads on requests in progress; a ring wait that fails for good
 stops the worker (v1.6.0, Refs #884)

---
 .../ring_wait_ceiling_884_linux_test.go       |  16 +-
 internal/engine/iouring/shutdown_gate_884.go  | 202 ++++++++++++++++++
 internal/engine/iouring/shutdown_ring_wait.go |   9 +
 internal/engine/iouring/worker.go             |  58 +++--
 .../shutdown_iouring_gate_884_linux_test.go   |  35 ++-
 5 files changed, 296 insertions(+), 24 deletions(-)
 create mode 100644 internal/engine/iouring/shutdown_gate_884.go

diff --git a/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go b/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
index 33fea78e2..8493c63c6 100644
--- a/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
+++ b/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
@@ -90,22 +90,26 @@ func TestRingWaitContinuesNeverGivesUpWhileItReaps884(t *testing.T) {
 }
 
 // TestRingWaitContinuesCountsOnlyConsecutiveRetries884: retries that come
-// apart, each after waits that succeeded (an idle worker's timed wait takes up
-// to a second; a retry 400 ms after the last), are not a stuck ring: a worker that meets an EBUSY now and
-// then, with nothing to reap each time, runs on.
+// apart, each after waits that succeeded, are not a stuck ring: a worker that
+// meets an EBUSY now and then, with nothing to reap each time, runs on. The
+// loop pass count (tickCounter) moves once per pass, so a retry that follows
+// the last by more than one pass has a wait that succeeded between them: the
+// test models it by advancing the count, with the time between the retries a
+// stuck run's own (50 us sleeps) so that only the count tells the two apart.
 func TestRingWaitContinuesCountsOnlyConsecutiveRetries884(t *testing.T) {
 	ring := newTestRing(t)
 	w := &Worker{ring: ring, id: 3, logger: slog.New(&warnLog884{})}
 	start := time.Now()
 	n := 0
-	for time.Since(start) < 3*time.Second {
+	for time.Since(start) < 2500*time.Millisecond {
 		if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
-			t.Fatalf("the worker gave up after %v and %d retries, each apart from the last by 400 ms of waits that succeeded",
+			t.Fatalf("the worker gave up after %v and %d retries, each apart from the last by a pass whose wait succeeded",
 				time.Since(start).Round(time.Millisecond), n)
 		}
 		n++
-		time.Sleep(400 * time.Millisecond)
+		w.tickCounter += 2 // the retried pass, then a pass whose wait succeeded
 	}
+	t.Logf("celeris884: %d retries, every other pass, over %v: the worker runs on", n, time.Since(start).Round(time.Millisecond))
 }
 
 // TestRunStopsOnAStuckRingWait884: the engine's own arm. One worker's waits
diff --git a/internal/engine/iouring/shutdown_gate_884.go b/internal/engine/iouring/shutdown_gate_884.go
new file mode 100644
index 000000000..f5b719bc4
--- /dev/null
+++ b/internal/engine/iouring/shutdown_gate_884.go
@@ -0,0 +1,202 @@
+//go:build linux
+
+package iouring
+
+import (
+	"golang.org/x/sys/unix"
+
+	"github.com/goceleris/celeris/internal/engine"
+)
+
+// The conns the shutdown gate stops reading (celeris#884 item 73), the
+// io_uring twin of epoll's closeIdleH1 and refusesRecv (celeris#885 item 49,
+// poolwait_gate_885.go). Past the first shutdownSendDrainNanos of the send
+// drain, or of the wait for the HTTP/2 pool handlers before it, an HTTP/1 conn
+// is read no more (refuseRecv): the drain, which lasts as long as the budget,
+// would never find its queues empty under steady keep-alive traffic. That left
+// the conns the gate had stopped reading open with no answer until the end of
+// the wait (up to the budget of the shutdown): an idle keep-alive client, a
+// conn that was accepted and has sent nothing, a detached stream the gate had
+// ended, and a request whose bytes were still arriving at the gate (an upload,
+// headers in pieces), which was dropped, never answered, and cut at the end.
+// std's Shutdown closes the idle conns at once and answers the request of an
+// active conn, so a client of either fails or finishes at once and retries
+// elsewhere. The gate now does the same:
+//
+//   - a conn that is idle, holds nothing queued (user space or the kernel's
+//     send queue) and has no request in progress is closed (closeIdleH1): an
+//     HTTP/1 keep-alive conn, one that has sent no byte, and a detached stream
+//     the gate has ended once what was queued for it has gone out (its client
+//     gets a FIN and reconnects), whatever kind of route it is on;
+//   - a request in progress is read on and answered (gateReadsOn): headers in
+//     pieces, a fixed-length or chunked body still arriving, the bytes of a
+//     protocol decision begun. The response carries Connection: close
+//     (H1State.CloseAfter, as net/http's response written after Shutdown began
+//     does) and the conn is closed once it has gone out;
+//   - the HTTP/2 conns the wait is for are served on: they have had GOAWAY since
+//     the wait began.
+//
+// A conn whose dispatch goroutine owns its parse state (asyncPromoted: a conn on
+// an Async route, or, on a server with any Async route, one whose read left a
+// partial request) is looked at under the goroutine's own locks, never without:
+// asyncInMu says whether input is waiting for it or it is inside a handler,
+// and detachMu, taken with TryLock as every worker-thread site does (celeris#593),
+// is free only while the goroutine is between requests, so h1State is read under
+// it. Epoll cannot yet look inside such a conn (its follow-up F2); this engine
+// can.
+
+// gateClosesPerPass caps the conns closeIdleH1 closes in one pass, so that a
+// worker with thousands of idle conns does not stop serving the conns the gate
+// is waiting for while it closes them all (a cancel SQE and a few syscalls
+// each): the next pass, h2PoolDrainPoll on, goes on.
+const gateClosesPerPass = 64
+
+// gateReadsOn reports whether handleRecv, past the gate, reads cs's bytes on
+// instead of dropping them: a request in progress (see the list above). It is
+// called only when refuseRecv is set and the conn is HTTP/1 or undecided, and
+// it marks the conn close-after when it says yes. Worker thread.
+func (w *Worker) gateReadsOn(cs *connState) bool {
+	if !cs.detected {
+		// The bytes of a protocol decision begun. Nothing yet (no earlier
+		// bytes): a request that begins after the gate.
+		return len(cs.detectAccum) > 0
+	}
+	h := cs.h1State
+	if h == nil || h.Detached.Load() {
+		return false // a detached stream is read no more: its writes were ended with the gate
+	}
+	if cs.asyncPromoted.Load() {
+		return w.promotedInProgress(cs)
+	}
+	// A conn the worker parses itself: its h1State is the worker's to read.
+	if cs.recvIntoBody || h.HasPendingData() || !h.AtRequestBoundary() {
+		h.CloseAfter = true
+		return true
+	}
+	return false
+}
+
+// promotedInProgress is gateReadsOn for a conn whose dispatch goroutine owns the
+// parse state: true while input is waiting for the goroutine, it is inside a
+// handler (or a guarded write holds detachMu), or its parser holds part of a
+// request. It marks an idle conn close-after too: the worker does not know
+// that no request will follow before the pass closes the conn. Worker thread.
+func (w *Worker) promotedInProgress(cs *connState) bool {
+	if dispatchHasWork(cs) {
+		return true
+	}
+	mu := cs.detachMu
+	if mu == nil || !mu.TryLock() {
+		return true
+	}
+	defer mu.Unlock()
+	h := cs.h1State
+	if h == nil || h.Detached.Load() {
+		return false
+	}
+	h.CloseAfter = true
+	return h.HasPendingData() || !h.AtRequestBoundary()
+}
+
+// dispatchHasWork reports whether a conn's dispatch goroutine has a request to
+// serve or is serving one: input waiting in asyncInBuf, or the goroutine alive
+// and out of its park loop (dispatchBusy's test, which also covers the window
+// after it took the input and before it holds detachMu), all read under
+// asyncInMu. A goroutine that has released detachMu for good at a Detach is not
+// counted: such a conn is a stream, judged by detachClosed. Worker thread.
+func dispatchHasWork(cs *connState) bool {
+	cs.asyncInMu.Lock()
+	busy := len(cs.asyncInBuf) > 0 || (cs.asyncRun && !cs.asyncParked && !cs.asyncDetachUnlocked)
+	cs.asyncInMu.Unlock()
+	return busy
+}
+
+// closeIdleH1 is the gate's pass over the live conns: every HTTP/1 conn it can
+// look at is marked close-after (H1State.CloseAfter, so that a request it still
+// reads, one in progress, is answered with Connection: close), and the ones that
+// hold nothing are closed (gateCloseable). A conn whose response is still going
+// out, or whose lock is busy, is met again by the next pass. closeConn removes
+// a conn from liveConns by swapping it with the last entry, so the walk goes
+// from the top down, as checkTimeouts does. Worker thread.
+func (w *Worker) closeIdleH1() {
+	closed := 0
+	for i := len(w.liveConns) - 1; i >= 0; i-- {
+		fd := w.liveConns[i]
+		cs := w.conns[fd]
+		if cs == nil || cs.closing {
+			continue
+		}
+		if cs.detected && engine.Protocol(cs.protocol.Load()) != engine.HTTP1 {
+			continue
+		}
+		if !w.gateCloseable(cs) {
+			continue
+		}
+		if closed >= gateClosesPerPass {
+			continue
+		}
+		// The kernel's send queue: a conn whose response left user space but
+		// not the host is not idle either (a close with unsent bytes would race
+		// them with the FIN). A fixed-file conn's slot is an index into the
+		// ring's file table, not a descriptor, so the ioctl cannot ask about it:
+		// such a conn is judged by its user-space queues alone
+		// (ownsDescriptorNumber).
+		if ownsDescriptorNumber(cs) {
+			if q, err := unix.IoctlGetInt(cs.fd, unix.TIOCOUTQ); err != nil || q != 0 {
+				continue
+			}
+		}
+		w.closeConn(fd)
+		closed++
+	}
+}
+
+// gateCloseable reports whether the gate may close cs: it is HTTP/1 (or not
+// decided, with no byte), has no request in progress, holds nothing queued and,
+// if it is a detached stream, the gate has ended it (a stream that still runs
+// is ended by stopDetachedProducers first). It marks every non-detached HTTP/1
+// conn it can look at close-after. Worker thread.
+func (w *Worker) gateCloseable(cs *connState) bool {
+	if !cs.detected {
+		// No byte (an undecided conn holding some is a request begun) and no
+		// dispatch goroutine yet: the first bytes decide that.
+		return len(cs.detectAccum) == 0 && !connSendPending(cs) && !cs.zcNotifPending
+	}
+	if cs.asyncPromoted.Load() && dispatchHasWork(cs) {
+		return false
+	}
+	if mu := cs.detachMu; mu != nil {
+		if !mu.TryLock() {
+			return false // inside a handler or a guarded write
+		}
+		defer mu.Unlock()
+	}
+	h := cs.h1State
+	if h == nil {
+		return false
+	}
+	if h.Detached.Load() {
+		if !cs.detachClosed {
+			return false // the stream still runs
+		}
+	} else {
+		h.CloseAfter = true
+		if h.HasPendingData() || !h.AtRequestBoundary() {
+			return false
+		}
+	}
+	return !connSendPending(cs) && !cs.zcNotifPending && len(cs.heldSends) == 0
+}
+
+// closeIdlePaced runs closeIdleH1 at most once per h2PoolDrainPoll, for the
+// send drain's gate, whose own walk over the detached conns is not paced (the
+// drain is not a serving phase). It shares producersStopNext with the pool
+// wait's gate, which never runs in the same loop pass. now is the pass's
+// UnixNano. Worker thread.
+func (w *Worker) closeIdlePaced(now int64) {
+	if now < w.producersStopNext {
+		return
+	}
+	w.producersStopNext = now + int64(h2PoolDrainPoll)
+	w.closeIdleH1()
+}
diff --git a/internal/engine/iouring/shutdown_ring_wait.go b/internal/engine/iouring/shutdown_ring_wait.go
index 59ae9e322..2129e6a9e 100644
--- a/internal/engine/iouring/shutdown_ring_wait.go
+++ b/internal/engine/iouring/shutdown_ring_wait.go
@@ -4,6 +4,7 @@ package iouring
 
 import (
 	"errors"
+	"time"
 
 	"golang.org/x/sys/unix"
 )
@@ -28,3 +29,11 @@ var shutdownRingWait = (*Ring).SubmitAndWaitTimeout
 func ringWaitRetryable(err error) bool {
 	return errors.Is(err, unix.EBUSY) || errors.Is(err, unix.EAGAIN)
 }
+
+// ringWaitCeiling is how long Worker.run retries a ring wait that answers a
+// retryable error, one pass after the other, with nothing in the CQ ring to
+// reap, before it takes the ring for a failing one and stops the worker, as it
+// did for every error before the retries existed (ringWaitContinues, celeris#884
+// item 72). A full CQ ring is cleared by the loop's own reap within a pass or
+// two, so a second of retries that reap nothing is a ring that will not recover.
+const ringWaitCeiling = time.Second
diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go
index 22cb9a1b1..658cb296a 100644
--- a/internal/engine/iouring/worker.go
+++ b/internal/engine/iouring/worker.go
@@ -962,9 +962,15 @@ type Worker struct {
 	// ringWaitWarned is set once the first retried wait was logged
 	// (ringWaitContinues). Worker thread only.
 	ringWaitWarned bool
-	// producersStopNext is when (UnixNano) the pool wait's gate may walk the
-	// detached conns again (stopProducersPaced). Worker thread only.
+	// producersStopNext is when (UnixNano) the shutdown gate may walk the
+	// conns again (stopProducersPaced, closeIdlePaced). Worker thread only.
 	producersStopNext int64
+	// ringWaitRunStart is when (UnixNano) the run of consecutive retried ring
+	// waits that reaped nothing began, and ringWaitRunTick the loop pass
+	// (tickCounter) of the last of them: ringWaitContinues' ceiling
+	// (celeris#884 item 72). Zero start: no run. Worker thread only.
+	ringWaitRunStart int64
+	ringWaitRunTick  uint32
 }
 
 // recvArmStats are the recv-arming witnesses behind celeris#484 / #560,
@@ -1594,7 +1600,9 @@ func (w *Worker) run(ctx context.Context) {
 				// so a slow callback delays the HTTP/2 conns served here.
 				if now := time.Now().UnixNano(); now > w.h2DrainStart+shutdownSendDrainNanos {
 					w.refuseRecv = true
-					w.stopProducersPaced(now)
+					if w.stopProducersPaced(now) {
+						w.closeIdleH1()
+					}
 				}
 			} else {
 				now := time.Now().UnixNano()
@@ -1614,6 +1622,7 @@ func (w *Worker) run(ctx context.Context) {
 					w.stopAccepting(ctx)
 					w.refuseRecv = true
 					w.stopDetachedProducers()
+					w.closeIdlePaced(now)
 				}
 			}
 		}
@@ -3969,7 +3978,7 @@ func (w *Worker) handleRecv(c *completionEntry, fd int, now int64) {
 	// HTTP/2 conns are served on: they have had GOAWAY since the pool wait,
 	// and what their responses may still wait for was waited for there or
 	// is out of time.
-	if w.refuseRecv && (!cs.detected || w.h1Only || engine.Protocol(cs.protocol.Load()) == engine.HTTP1) {
+	if w.refuseRecv && (!cs.detected || w.h1Only || engine.Protocol(cs.protocol.Load()) == engine.HTTP1) && !w.gateReadsOn(cs) {
 		if cqeHasBuffer(c.Flags) && w.bufRing != nil {
 			w.bufRing.PushBuffer(cqeBufferID(c.Flags))
 			w.hasBufReturns = true
@@ -8004,13 +8013,21 @@ func (w *Worker) stopDetachedProducers() {
 // flushed, or EAGAIN) says the CQ ring has no room, not that the ring failed:
 // the SQEs the enter did not take stay pending (retryPending), and the loop's
 // own pass over the CQ ring after this returns is the reap that makes room.
-// The error is counted and, the first time on a worker, logged as a Warn;
-// there is no ceiling on the retries (a ring that fails for good with EBUSY
-// or EAGAIN is retried until the context is cancelled: the checklist of
-// celeris#884 records it). With nothing in the ring to reap the worker gives
-// the kernel a moment (cancelAccept's drain does the same) rather than enter
-// again at once. Any other error is the ring failing: the worker stops, and
-// closes what it owns.
+// The error is counted and, the first time on a worker, logged as a Warn.
+// With nothing in the ring to reap the worker gives the kernel a moment
+// (cancelAccept's drain does the same) rather than enter again at once.
+//
+// A ring that answers for good, with nothing in its CQ ring to reap, is not
+// one whose CQ ring is full but one that is failing, and the retries end as
+// they did before they existed: after ringWaitCeiling of consecutive retries
+// that reaped nothing the worker stops (false: the caller shuts it down) and
+// logs an Error (celeris#884 item 72). A retry that finds a completion in the
+// CQ ring, which the loop's own pass reaps, is the loop being behind its ring
+// and ends the run; so does a wait that succeeded in between (the loop pass
+// count, which moves once per pass, jumps by more than one: a retry is part
+// of the run only if it follows the last one at once), and a run is not
+// carried over a gap longer than a wait takes. Any other error is the ring
+// failing: the worker stops, and closes what it owns.
 func (w *Worker) ringWaitContinues(mode ringWaitMode, err error) bool {
 	if h := ringWaitSeam; h != nil {
 		err = h(w, mode, err)
@@ -8032,9 +8049,24 @@ func (w *Worker) ringWaitContinues(mode ringWaitMode, err error) bool {
 				"worker", w.id, "wait", mode.String(), "err", err)
 		}
 	}
-	if h, t := w.ring.BeginCQ(); h == t {
-		time.Sleep(50 * time.Microsecond)
+	h, t := w.ring.BeginCQ()
+	if h != t {
+		w.ringWaitRunStart = 0 // the loop reaps: not a ring that is stuck
+		return true
+	}
+	now := time.Now().UnixNano()
+	if w.ringWaitRunStart == 0 || w.tickCounter-w.ringWaitRunTick > 1 {
+		w.ringWaitRunStart = now // a new run: a wait succeeded since the last retry
+	} else if now-w.ringWaitRunStart >= int64(ringWaitCeiling) {
+		if w.logger != nil {
+			w.logger.Error("io_uring: a ring wait has answered a retryable error for the whole ceiling with nothing to reap; the ring is failing, the worker stops",
+				"worker", w.id, "wait", mode.String(), "err", err, "ceiling", ringWaitCeiling,
+				"retries", w.ringWaitRetries.Load())
+		}
+		return false
 	}
+	w.ringWaitRunTick = w.tickCounter
+	time.Sleep(50 * time.Microsecond)
 	return true
 }
 
diff --git a/test/server/shutdown_iouring_gate_884_linux_test.go b/test/server/shutdown_iouring_gate_884_linux_test.go
index 814d4663a..563ed8ea9 100644
--- a/test/server/shutdown_iouring_gate_884_linux_test.go
+++ b/test/server/shutdown_iouring_gate_884_linux_test.go
@@ -78,7 +78,7 @@ func TestIOUringShutdownGateConns884(t *testing.T) {
 
 func runIOUringGateArm884(t *testing.T, eng celeris.EngineType, gate string, async bool) {
 	const (
-		nBulk, chunk = 4, 256 << 10
+		nBulk, chunk = 4, 128 << 10
 		upSize       = 2 << 20
 		upFirst      = 64 << 10
 		holdSize     = 4 << 20
@@ -288,6 +288,17 @@ func runIOUringGateArm884(t *testing.T, eng celeris.EngineType, gate string, asy
 
 	// The rest of every request in progress arrives after the gate.
 	time.Sleep(time.Until(begin.Add(1200 * time.Millisecond)))
+	// The download's client sends more after the gate, twice, which the engine
+	// does not read (the conn is not idle, but the gate reads a request only
+	// when one is in progress): bytes that stay unread in the kernel, so a close
+	// of that conn while any of the response is still unacknowledged in the
+	// kernel's send queue would be a reset that throws the tail of it away.
+	for range 2 {
+		if _, err := io.WriteString(dl, "GET /ping HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
+			t.Errorf("a request written on the download's conn after the gate: %v", err)
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
 	var pwg sync.WaitGroup
 	for _, p := range inProgress {
 		pwg.Add(1)
@@ -378,12 +389,26 @@ func runIOUringGateArm884(t *testing.T, eng celeris.EngineType, gate string, asy
 	if err != nil {
 		t.Fatalf("the download's head: %v", err)
 	}
-	if n, err := io.Copy(io.Discard, resp.Body); n != int64(len(dlBody)) {
-		t.Errorf("the queued download delivered %d of %d bytes (%v): the gate cut a response that was still going out", n, len(dlBody), err)
+	// Read slowly (about 20 MiB/s), so that the kernel's send queue (a few MiB)
+	// is still draining when the engine's own queue has emptied.
+	var dlN int64
+	var dlErr error
+	for chunk := make([]byte, 64<<10); dlErr == nil; {
+		var n int
+		n, dlErr = resp.Body.Read(chunk)
+		dlN += int64(n)
+		time.Sleep(3 * time.Millisecond)
+	}
+	if dlN != int64(len(dlBody)) {
+		t.Errorf("the queued download delivered %d of %d bytes (%v): the gate cut a response that was still going out", dlN, len(dlBody), dlErr)
 	}
+	// Closed once drained: a FIN, or, because the client's two requests after
+	// the gate were never read and close(2) with unread bytes resets, an RST that
+	// comes after the whole response has been received (the length check above).
+	// Silence is what the gate must not leave.
 	_ = dl.SetReadDeadline(time.Now().Add(3 * time.Second))
-	if _, err := dbr.ReadByte(); err != io.EOF {
-		t.Errorf("after the download the conn read %v, want a FIN (io.EOF): a drained conn was left open", err)
+	if _, err := dbr.ReadByte(); err != io.EOF && !errors.Is(err, syscall.ECONNRESET) {
+		t.Errorf("after the download the conn read %v, want a close (FIN, or a reset after the whole response): a drained conn was left open", err)
 	}
 
 	for _, c := range holders { // the holders go: the wait ends

From 87f4fd5dd09ab4cfd5a929abfd1fffe8d9c61d11 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:07:19 +0100
Subject: [PATCH 03/17] iouring: unit tests for the gate's look at a conn, its
 per-pass cap and the kernel send queue (v1.6.0, Refs #884)

---
 internal/engine/iouring/shutdown_gate_884.go  |   2 +-
 .../iouring/shutdown_gate_884_linux_test.go   | 363 ++++++++++++++++++
 2 files changed, 364 insertions(+), 1 deletion(-)
 create mode 100644 internal/engine/iouring/shutdown_gate_884_linux_test.go

diff --git a/internal/engine/iouring/shutdown_gate_884.go b/internal/engine/iouring/shutdown_gate_884.go
index f5b719bc4..49bae688a 100644
--- a/internal/engine/iouring/shutdown_gate_884.go
+++ b/internal/engine/iouring/shutdown_gate_884.go
@@ -181,7 +181,7 @@ func (w *Worker) gateCloseable(cs *connState) bool {
 		}
 	} else {
 		h.CloseAfter = true
-		if h.HasPendingData() || !h.AtRequestBoundary() {
+		if cs.recvIntoBody || h.HasPendingData() || !h.AtRequestBoundary() {
 			return false
 		}
 	}
diff --git a/internal/engine/iouring/shutdown_gate_884_linux_test.go b/internal/engine/iouring/shutdown_gate_884_linux_test.go
new file mode 100644
index 000000000..c10d5bda3
--- /dev/null
+++ b/internal/engine/iouring/shutdown_gate_884_linux_test.go
@@ -0,0 +1,363 @@
+//go:build linux
+
+package iouring
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+
+	"github.com/goceleris/celeris/internal/conn"
+	"github.com/goceleris/celeris/internal/engine"
+	"github.com/goceleris/celeris/internal/engine/internal/errclass"
+	"github.com/goceleris/celeris/internal/protocol/h2/stream"
+)
+
+// celeris#884 item 73: the unit half of the shutdown gate's look at a conn
+// (shutdown_gate_884.go). The wire-level half, with the engine and clients, is
+// test/server TestIOUringShutdownGateConns884.
+
+// gateNoHandler884 fails the test if a request reaches it: the parser is fed
+// partial requests here, which it buffers.
+type gateNoHandler884 struct{ t *testing.T }
+
+func (h gateNoHandler884) HandleStream(_ context.Context, _ *stream.Stream) error {
+	h.t.Error("a handler ran for a partial request")
+	return nil
+}
+
+// gateConn884 returns an HTTP/1 conn (decided, state fresh) that is idle.
+func gateConn884(fd int) *connState {
+	cs := &connState{fd: fd, liveIdx: -1, detected: true, h1State: conn.NewH1State()}
+	cs.protocol.Store(int32(engine.HTTP1))
+	return cs
+}
+
+// feed884 hands the parser part of a request: it keeps it, in progress.
+func feed884(t *testing.T, cs *connState, part string) {
+	t.Helper()
+	write := func([]byte) {}
+	if err := conn.ProcessH1(context.Background(), []byte(part), cs.h1State, gateNoHandler884{t}, write); err != nil {
+		t.Fatalf("ProcessH1(%q): %v", part, err)
+	}
+	if !cs.h1State.HasPendingData() {
+		t.Fatalf("setup: %q left no pending state", part)
+	}
+}
+
+// TestGateLookAtAConn884 is the table of what the gate does with a conn: whether
+// handleRecv reads its bytes on (gateReadsOn), whether the pass may close it
+// (gateCloseable), and whether it is marked close-after. A conn the gate reads
+// on must not be closeable, and the other way round for an idle one.
+func TestGateLookAtAConn884(t *testing.T) {
+	w := &Worker{}
+	for _, tc := range []struct {
+		name       string
+		build      func(t *testing.T) *connState
+		readsOn    bool
+		closeable  bool
+		closeAfter bool // set on the conn by the look at it
+	}{
+		{"idle-keepalive", func(t *testing.T) *connState { return gateConn884(7) }, false, true, true},
+		{"no-byte-undetected", func(t *testing.T) *connState { return &connState{fd: 7, liveIdx: -1} }, false, true, false},
+		{"undecided-with-bytes", func(t *testing.T) *connState {
+			return &connState{fd: 7, liveIdx: -1, detectAccum: []byte("PRI ")}
+		}, true, false, false},
+		{"headers-in-pieces", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			feed884(t, cs, "GET / HTTP/1.1\r\nHost: x\r\n")
+			return cs
+		}, true, false, true},
+		{"body-arriving", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			feed884(t, cs, "POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nabc")
+			return cs
+		}, true, false, true},
+		{"body-into-bodybuf", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.recvIntoBody = true
+			return cs
+		}, true, false, true},
+		{"response-queued", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.writeBuf = []byte("HTTP/1.1 200 OK\r\n")
+			return cs
+		}, false, false, true},
+		{"send-in-flight", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.sending = true
+			return cs
+		}, false, false, true},
+		{"zc-notification-owed", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.zcNotifPending = true
+			return cs
+		}, false, false, true},
+		{"detached-stream-running", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.h1State.Detached.Store(true)
+			return cs
+		}, false, false, false},
+		{"detached-stream-ended", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.h1State.Detached.Store(true)
+			cs.detachClosed = true
+			return cs
+		}, false, true, false},
+		{"detached-stream-ended-queue-left", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.h1State.Detached.Store(true)
+			cs.detachClosed = true
+			cs.writeBuf = []byte("tick")
+			return cs
+		}, false, false, false},
+		{"promoted-idle-parked", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.asyncPromoted.Store(true)
+			cs.asyncRun, cs.asyncParked = true, true
+			return cs
+		}, false, true, true},
+		{"promoted-goroutine-gone", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.asyncPromoted.Store(true)
+			return cs
+		}, false, true, true},
+		{"promoted-input-waiting", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.asyncPromoted.Store(true)
+			cs.asyncRun, cs.asyncParked = true, true
+			cs.asyncInBuf = []byte("GET / HTTP/1.1\r\n")
+			return cs
+		}, true, false, false},
+		{"promoted-inside-a-handler", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.detachMu.Lock() // the goroutine holds it across the handler
+			cs.asyncPromoted.Store(true)
+			cs.asyncRun = true
+			return cs
+		}, true, false, false},
+		{"promoted-partial-request", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.asyncPromoted.Store(true)
+			cs.asyncRun, cs.asyncParked = true, true
+			feed884(t, cs, "POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 100\r\n\r\nabc")
+			return cs
+		}, true, false, true},
+		{"promoted-lock-busy-parked", func(t *testing.T) *connState {
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.detachMu.Lock() // a guarded write holds it
+			cs.asyncPromoted.Store(true)
+			cs.asyncRun, cs.asyncParked = true, true
+			return cs
+		}, true, false, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cs := tc.build(t)
+			if got := w.gateReadsOn(cs); got != tc.readsOn {
+				t.Errorf("gateReadsOn=%v, want %v", got, tc.readsOn)
+			}
+			if cs.h1State != nil {
+				cs.h1State.CloseAfter = false
+			}
+			if got := w.gateCloseable(cs); got != tc.closeable {
+				t.Errorf("gateCloseable=%v, want %v", got, tc.closeable)
+			}
+			if h := cs.h1State; h != nil && h.CloseAfter != tc.closeAfter {
+				t.Errorf("after the pass's look the conn's CloseAfter is %v, want %v: a request answered after the gate must carry Connection: close", h.CloseAfter, tc.closeAfter)
+			}
+			if tc.readsOn && tc.closeable {
+				t.Error("the table has a conn the gate both reads on and may close")
+			}
+		})
+	}
+}
+
+// gateRig884 is a worker with a ring and n conns on real descriptors (one end
+// of a socketpair each), each idle HTTP/1, as closeConn needs to close them.
+func gateRig884(t *testing.T, n int) (*Worker, []*connState, []int) {
+	t.Helper()
+	ring := newTestRing(t)
+	w := &Worker{
+		ring:        ring,
+		liveConns:   make([]int, 0, n),
+		errs:        &errclass.Counters{},
+		activeConns: &atomic.Int64{},
+		closeCount:  &atomic.Uint64{},
+		recvArm:     &recvArmStats{},
+		handoffLoss: &handoffLossStats{},
+		listenFD:    -1,
+	}
+	w.cachedNow = time.Now().UnixNano()
+	var css []*connState
+	var peers []int
+	for range n {
+		fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
+		if err != nil {
+			t.Fatalf("socketpair: %v", err)
+		}
+		local, peer := fds[0], fds[1]
+		t.Cleanup(func() { _ = unix.Close(peer) })
+		cs := gateConn884(local)
+		cs.generation = 9
+		cs.buf = make([]byte, 4096)
+		css = append(css, cs)
+		peers = append(peers, peer)
+	}
+	w.conns = make([]*connState, 0)
+	maxfd := 0
+	for _, cs := range css {
+		maxfd = max(maxfd, cs.fd)
+	}
+	w.conns = make([]*connState, maxfd+1)
+	for _, cs := range css {
+		w.conns[cs.fd] = cs
+		w.addLiveConn(cs)
+		w.connCount++
+		w.activeConns.Add(1)
+	}
+	t.Cleanup(func() {
+		// What the test left open: closeConn's close is deferred behind the
+		// recv-cancel's completion, which no loop runs here.
+		for _, cs := range css {
+			_ = unix.Close(cs.fd)
+		}
+	})
+	return w, css, peers
+}
+
+func liveGate884(w *Worker, css []*connState) (live int) {
+	for _, cs := range css {
+		if w.conns[cs.fd] == cs {
+			live++
+		}
+	}
+	return live
+}
+
+// TestGatePassClosesAtMostTheCapPerPass884: the pass closes gateClosesPerPass
+// idle conns and no more, and the next one goes on, so that a worker with
+// thousands of idle conns is not stopped for the length of all their closes.
+func TestGatePassClosesAtMostTheCapPerPass884(t *testing.T) {
+	n := gateClosesPerPass + 6
+	w, css, _ := gateRig884(t, n)
+	w.closeIdleH1()
+	if got, want := liveGate884(w, css), n-gateClosesPerPass; got != want {
+		t.Fatalf("after one pass %d of %d conns are open, want %d: the pass must close exactly the cap (%d)", got, n, want, gateClosesPerPass)
+	}
+	w.closeIdleH1()
+	if got := liveGate884(w, css); got != 0 {
+		t.Fatalf("after the second pass %d conns are still open, want 0", got)
+	}
+}
+
+// TestGatePassSkipsHTTP2AndBusyConns884: only HTTP/1 conns are the pass's (an
+// HTTP/2 conn has had GOAWAY and is served on), and one with a response going
+// out is left for the next pass.
+func TestGatePassSkipsHTTP2AndBusyConns884(t *testing.T) {
+	w, css, _ := gateRig884(t, 4)
+	css[0].protocol.Store(int32(engine.H2C))
+	css[1].writeBuf = []byte("HTTP/1.1 200 OK\r\n\r\n")
+	w.closeIdleH1()
+	for i, want := range []bool{true, true, false, false} {
+		if got := w.conns[css[i].fd] == css[i]; got != want {
+			t.Errorf("conn %d open=%v after the pass, want %v", i, got, want)
+		}
+	}
+}
+
+// TestGatePassWaitsForTheKernelSendQueue884: a conn whose response has left
+// the engine's queues but not the kernel's (the peer has not taken it) is not
+// idle: it is closed once the peer has, not before.
+func TestGatePassWaitsForTheKernelSendQueue884(t *testing.T) {
+	w, css, peers := gateRig884(t, 1)
+	// A TCP pair in place of the socketpair, to have a send queue the kernel
+	// holds: SIOCOUTQ is the unacknowledged bytes.
+	lfd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = unix.Close(lfd) }()
+	if err := unix.Bind(lfd, &unix.SockaddrInet4{Addr: [4]byte{127, 0, 0, 1}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := unix.Listen(lfd, 1); err != nil {
+		t.Fatal(err)
+	}
+	sa, _ := unix.Getsockname(lfd)
+	peer, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = unix.Close(peer) }()
+	if err := unix.SetsockoptInt(peer, unix.SOL_SOCKET, unix.SO_RCVBUF, 4096); err != nil {
+		t.Fatal(err)
+	}
+	if err := unix.Connect(peer, sa); err != nil {
+		t.Fatal(err)
+	}
+	local, _, err := unix.Accept4(lfd, unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Move the conn onto the TCP socket.
+	old := css[0]
+	w.removeLiveConn(old)
+	w.clearConnSlot(old.fd)
+	w.connCount--
+	_ = unix.Close(old.fd)
+	_ = peers
+	cs := gateConn884(local)
+	cs.generation = 9
+	cs.buf = make([]byte, 4096)
+	if local >= len(w.conns) {
+		w.conns = append(w.conns, make([]*connState, local+1-len(w.conns))...)
+	}
+	w.conns[local] = cs
+	w.addLiveConn(cs)
+	w.connCount++
+	// Fill the kernel's send queue: the peer reads nothing.
+	chunk := make([]byte, 64<<10)
+	for {
+		if _, err := unix.Write(local, chunk); err != nil {
+			break
+		}
+	}
+	if q, _ := unix.IoctlGetInt(local, unix.TIOCOUTQ); q == 0 {
+		t.Fatal("setup: the kernel's send queue is empty")
+	}
+
+	w.closeIdleH1()
+	if w.conns[local] != cs {
+		t.Fatal("the pass closed a conn whose response is still in the kernel's send queue, unread by its peer: a close with unread bytes from the peer would reset it away")
+	}
+	// The peer reads it all; the queue empties, and the next pass closes the conn.
+	buf := make([]byte, 64<<10)
+	for end := time.Now().Add(5 * time.Second); time.Now().Before(end); {
+		_, _ = unix.Read(peer, buf)
+		if q, _ := unix.IoctlGetInt(local, unix.TIOCOUTQ); q == 0 {
+			break
+		}
+		time.Sleep(time.Millisecond)
+	}
+	if q, _ := unix.IoctlGetInt(local, unix.TIOCOUTQ); q != 0 {
+		t.Fatalf("setup: the kernel's send queue still holds %d bytes after the peer read", q)
+	}
+	w.closeIdleH1()
+	if w.conns[local] == cs {
+		t.Error("the pass left an idle conn with an empty send queue open")
+	}
+	_ = unix.Close(local)
+}

From fc7baeb4a29f683527a14ae7ca83d8f7d1953263 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:09:01 +0100
Subject: [PATCH 04/17] iouring: shutdown drain trials for a linked recv after
 its SEND CQE and a re-armed multishot recv (v1.6.0, Refs #884 item 16)

---
 .../shutdown_drain_trials_884_linux_test.go   | 122 ++++++++++++++++++
 1 file changed, 122 insertions(+)
 create mode 100644 internal/engine/iouring/shutdown_drain_trials_884_linux_test.go

diff --git a/internal/engine/iouring/shutdown_drain_trials_884_linux_test.go b/internal/engine/iouring/shutdown_drain_trials_884_linux_test.go
new file mode 100644
index 000000000..4fd9a4044
--- /dev/null
+++ b/internal/engine/iouring/shutdown_drain_trials_884_linux_test.go
@@ -0,0 +1,122 @@
+//go:build linux
+
+package iouring
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// celeris#884 item 16, trials (a) and (b). The shutdown drain
+// (endOwedOpsAtShutdown) ends the ops the kernel still owes on a live
+// connection's descriptor before shutdown closes it. TestShutdownDrainSurvivesAFullCQRing873
+// and the 880 tests pin it for a standalone recv; PR #1099 recorded two forms
+// they do not reach, and its open question, keep or remove the drain, is
+// answered by what these show: the drain is the only thing that ends them.
+//
+//   - (a) a recv linked behind a SEND whose CQE the loop has already read. The
+//     kernel consumed both SQEs at the submit; the recv is issued, and its
+//     descriptor number resolved, only when the SEND completes, so it can be
+//     in no cancel table at the moment the descriptor is closed.
+//   - (b) a multishot recv the loop re-armed after the kernel ended it with
+//     ENOBUFS (the provided-buffer ring ran dry): the new SQE is in the SQ ring,
+//     not yet submitted, and counts as owed from the moment it is placed.
+//
+// Each holds the worker at that point (the test is the loop) and calls the
+// drain. With the drain disabled (mutant M1: `if false && w.ring != nil`) both
+// leave the recv owed, kernelInflight 1, which is what shutdown would close
+// the descriptor under. Mutant logs: the lane's mutants/ directory.
+
+// reapReal884 dispatches the CQEs in the ring the way the loop does, until stop
+// returns true for one, and returns whether it did.
+func reapReal884(t *testing.T, w *Worker, stop func(c *completionEntry) bool) bool {
+	t.Helper()
+	for end := time.Now().Add(2 * time.Second); time.Now().Before(end); time.Sleep(time.Millisecond) {
+		head, tail := w.ring.BeginCQ()
+		found := false
+		for ; head != tail; head++ {
+			c := w.ring.cqeAt(head)
+			w.processCQE(context.Background(), c, time.Now().UnixNano())
+			if stop(c) {
+				found = true
+			}
+		}
+		w.ring.EndCQ(head)
+		if found {
+			return true
+		}
+	}
+	return false
+}
+
+func drainEnded884(t *testing.T, f *fdlFixture) {
+	t.Helper()
+	w, cs := f.w, f.cs
+	start := time.Now()
+	w.endOwedOpsAtShutdown()
+	took := time.Since(start)
+	t.Logf("celeris884 drain: took=%v kernelInflight_after=%d gaveUp=%d", took.Round(time.Microsecond), cs.kernelInflight, w.handoffLoss.shutdownDrainGaveUp.Load())
+	if cs.kernelInflight != 0 {
+		t.Fatalf("the drain ended with %d op(s) still owed on fd %d: shutdown would close the descriptor under them", cs.kernelInflight, f.fd)
+	}
+	if g := w.handoffLoss.shutdownDrainGaveUp.Load(); g != 0 {
+		t.Fatalf("shutdownDrainGaveUp=%d after a drain that finished", g)
+	}
+	if took > shutdownFDDrainBound873 {
+		t.Fatalf("the drain took %v (bound %v)", took, shutdownFDDrainBound873)
+	}
+}
+
+// TestShutdownDrainEndsALinkedRecvAfterItsSendCQE884 is trial (a).
+func TestShutdownDrainEndsALinkedRecvAfterItsSendCQE884(t *testing.T) {
+	f := newFDLFixture(t, false)
+	w, cs := f.w, f.cs
+	cs.writeBuf = append(cs.writeBuf, "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"...)
+	if retry := w.flushSendLink(cs); retry || !cs.recvLinked || !cs.recvArmed || !cs.sending || cs.kernelInflight != 2 {
+		t.Fatalf("no linked SEND+RECV placed: retry=%v recvLinked=%v recvArmed=%v sending=%v kernelInflight=%d",
+			retry, cs.recvLinked, cs.recvArmed, cs.sending, cs.kernelInflight)
+	}
+	if _, err := w.ring.Submit(); err != nil {
+		t.Fatalf("submit: %v", err)
+	}
+	if !reapReal884(t, w, func(c *completionEntry) bool { return c.UserData&udMask == udSend }) {
+		t.Fatal("the SEND's completion never reached the ring")
+	}
+	// The worker is held here: the SEND's CQE read, its linked recv owed, no
+	// completion for it (the peer sends nothing).
+	if cs.sending || cs.kernelInflight != 1 || !cs.recvArmed {
+		t.Fatalf("after the SEND's CQE: sending=%v kernelInflight=%d recvArmed=%v, want false 1 true", cs.sending, cs.kernelInflight, cs.recvArmed)
+	}
+	time.Sleep(50 * time.Millisecond)
+	if head, tail := w.ring.BeginCQ(); head != tail {
+		t.Fatalf("%d completion(s) in the ring with the peer silent: the recv did not stay owed", tail-head)
+	}
+	if !fdOwed(cs) {
+		t.Fatal("setup: the conn owes the kernel nothing, so there is nothing for the drain to end")
+	}
+	drainEnded884(t, f)
+}
+
+// TestShutdownDrainEndsAReArmedMultishotRecv884 is trial (b).
+func TestShutdownDrainEndsAReArmedMultishotRecv884(t *testing.T) {
+	f := newMultishotFixture869(t)
+	w, cs := f.w, f.cs
+	f.armFirstRecv() // the first recv's SQE, taken off the ring: its completion is the test's
+	if !cs.recvArmed || cs.kernelInflight != 1 {
+		t.Fatalf("setup: recvArmed=%v kernelInflight=%d after the first arm", cs.recvArmed, cs.kernelInflight)
+	}
+	// The kernel ends the multishot with ENOBUFS (-105); handleRecv publishes
+	// the buffers and re-arms.
+	f.process(f.recvCQE(-105))
+	// The worker is held here: the re-armed recv's SQE is in the SQ ring and
+	// owed, and no enter has submitted it.
+	if !cs.recvArmed || cs.kernelInflight != 1 || w.ring.Pending() == 0 {
+		t.Fatalf("after the ENOBUFS: recvArmed=%v kernelInflight=%d pending SQEs=%d, want a re-armed recv, owed and unsubmitted",
+			cs.recvArmed, cs.kernelInflight, w.ring.Pending())
+	}
+	if !fdOwed(cs) {
+		t.Fatal("setup: the conn owes the kernel nothing, so there is nothing for the drain to end")
+	}
+	drainEnded884(t, f)
+}

From b0b735582baf540800ed779dae40698a454e54f2 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:10:48 +0100
Subject: [PATCH 05/17] iouring: park retraction test with a close that owes
 the kernel nothing (v1.6.0, Refs #884 item 80)

---
 .../iouring/park_retracts_residue_test.go     | 69 +++++++++++++++++--
 1 file changed, 65 insertions(+), 4 deletions(-)

diff --git a/internal/engine/iouring/park_retracts_residue_test.go b/internal/engine/iouring/park_retracts_residue_test.go
index 06729957c..2b8867f46 100644
--- a/internal/engine/iouring/park_retracts_residue_test.go
+++ b/internal/engine/iouring/park_retracts_residue_test.go
@@ -23,9 +23,14 @@ package iouring
 // close owed does not park, so the next iteration's sweep() retracts. The arm
 // goes red with both removed, not with the retraction alone (checked in
 // celeris#885 item 11); the retraction is what a close with nothing in flight
-// would need.
+// would need. TestParkedWorkerRetractsResidueOfACloseWithNothingInFlight884 is
+// that close (celeris#884 item 80): the conn's recv is ended and reaped inside
+// the sweep's own pass, so its close lands after sweep() with nothing owed to
+// the kernel, the gate does not hold the worker, and only the retraction at
+// the park clears the gauges (red with it removed).
 
 import (
+	"context"
 	"io"
 	"log/slog"
 	"net"
@@ -100,7 +105,12 @@ func parkWaitIou(d time.Duration, f func() bool) bool {
 // checkTimeouts (postSweep) or by the client, right after PauseAccept
 // (!postSweep), and asserts that a worker that has parked holding nothing
 // publishes nothing. The header timeout is 60 s in both arms: far from either.
-func parkResidueIou(t *testing.T, postSweep bool) {
+func parkResidueIou(t *testing.T, postSweep bool) { parkResidueIouMode(t, postSweep, false) }
+
+// parkResidueIouMode is parkResidueIou with a third kind of close: nothingOwed
+// replaces ReadTimeout's close by one made after sweep() with no op owed to the
+// kernel (see afterSweepNothingOwed884).
+func parkResidueIouMode(t *testing.T, postSweep, nothingOwed bool) {
 	var disc atomic.Int64
 	e, addr := startParkEngine711(t, fdlHandler{}, func(c *resource.Config) {
 		c.OnDisconnect = func(string) { disc.Add(1) }
@@ -110,7 +120,7 @@ func parkResidueIou(t *testing.T, postSweep bool) {
 		// listener, the only kind that parks.
 		c.DisableDeferAccept = true
 		c.ReadHeaderTimeout = 60 * time.Second // neither the header timer nor its checkTimeouts close can come first
-		if postSweep {
+		if postSweep && !nothingOwed {
 			c.ReadTimeout = 300 * time.Millisecond
 			c.IdleTimeout = 10 * time.Minute
 		}
@@ -156,6 +166,12 @@ func parkResidueIou(t *testing.T, postSweep bool) {
 		t.Fatalf("celeris711 PREMISE: the sweep never published a residue of 1 (sum=%d adopted=%d)",
 			residueSum(e.Metrics()), tgt.adopted.Load())
 	}
+	var manufactured atomic.Int32
+	if nothingOwed {
+		hook := afterSweepNothingOwed884(&manufactured)
+		testHookCheckTimeouts.Store(&hook)
+		t.Cleanup(func() { testHookCheckTimeouts.Store(nil) })
+	}
 	if err := e.PauseAccept(); err != nil {
 		t.Fatalf("pause: %v", err)
 	}
@@ -163,7 +179,7 @@ func parkResidueIou(t *testing.T, postSweep bool) {
 		t.Fatalf("celeris711 PREMISE: the connection closed (active=%d) before PauseAccept returned, so not "+
 			"on a worker without a listener", n)
 	}
-	if !postSweep {
+	if !postSweep && !nothingOwed {
 		// The client leaves, far before any deadline: the worker reads the
 		// EOF from a CQE, and CQEs are dispatched before sweep().
 		_ = c.Close()
@@ -171,6 +187,9 @@ func parkResidueIou(t *testing.T, postSweep bool) {
 	if !parkWaitIou(10*time.Second, func() bool { return e.Metrics().ActiveConnections == 0 }) {
 		t.Fatalf("celeris711 PREMISE: the connection was never closed")
 	}
+	if nothingOwed && manufactured.Load() != 1 {
+		t.Fatalf("celeris884 PREMISE: the sweep hook ended the conn's recv %d times, want 1", manufactured.Load())
+	}
 	if !parkWaitIou(3*time.Second, allParked) {
 		t.Fatalf("celeris711 PREMISE: not every worker parked")
 	}
@@ -202,3 +221,45 @@ func TestParkedWorkerRetractsResidueOfAPostSweepClose(t *testing.T) { parkResidu
 // connection closed by the client's EOF, a CQE dispatched before sweep(),
 // which retracted before this fix as well.
 func TestParkedWorkerRetractsResidueOfAPreSweepClose(t *testing.T) { parkResidueIou(t, false) }
+
+// afterSweepNothingOwed884 is a checkTimeouts hook (it runs on the worker's
+// thread, after sweep() in the same pass) that gives a worker with one live
+// conn, no listener and a recv armed on it the close celeris#884 item 80 asks
+// for: it cancels the recv, submits, and reaps the cancelled recv's CQE the way
+// the loop does, which closes the conn (a failed recv closes it) with nothing
+// left in flight. The close is therefore after sweep() and owes the kernel
+// nothing, so closeFDOwed stays 0 and the worker parks in the same pass; once.
+func afterSweepNothingOwed884(done *atomic.Int32) func(*Worker) {
+	return func(w *Worker) {
+		if done.Load() != 0 || w.listenFD >= 0 || !w.acceptPaused.Load() || len(w.liveConns) != 1 {
+			return
+		}
+		cs := w.conns[w.liveConns[0]]
+		if cs == nil || !cs.recvArmed || cs.closing {
+			return
+		}
+		fd := cs.fd
+		if w.cancelConnOps(fd, cs) {
+			return // the cancel was not placed; the next sweep tries again
+		}
+		done.Store(1)
+		for end := time.Now().Add(2 * time.Second); w.conns[fd] == cs && time.Now().Before(end); {
+			if err := w.ring.SubmitAndWaitTimeout(10 * time.Millisecond); err != nil && !ringWaitRetryable(err) {
+				return
+			}
+			head, tail := w.ring.BeginCQ()
+			for ; head != tail; head++ {
+				w.processCQE(context.Background(), w.ring.cqeAt(head), time.Now().UnixNano())
+			}
+			w.ring.EndCQ(head)
+		}
+	}
+}
+
+// TestParkedWorkerRetractsResidueOfACloseWithNothingInFlight884 is the arm item
+// 80 asks for: the last conn closed after sweep() with nothing owed to the
+// kernel, which the closeFDOwed gate does not hold off the park. It goes red
+// with the park's sweepRetract removed.
+func TestParkedWorkerRetractsResidueOfACloseWithNothingInFlight884(t *testing.T) {
+	parkResidueIouMode(t, true, true)
+}

From 50b21e5ac9b1ab6e65cec5bcc0c2a3f964fc37b1 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:12:08 +0100
Subject: [PATCH 06/17] iouring: exit-order and leaveGroupAfterExit tests
 (v1.6.0, Refs #884 item 71)

---
 .../worker_exit_order_884_linux_test.go       | 150 ++++++++++++++++++
 1 file changed, 150 insertions(+)
 create mode 100644 internal/engine/iouring/worker_exit_order_884_linux_test.go

diff --git a/internal/engine/iouring/worker_exit_order_884_linux_test.go b/internal/engine/iouring/worker_exit_order_884_linux_test.go
new file mode 100644
index 000000000..1a37fa74a
--- /dev/null
+++ b/internal/engine/iouring/worker_exit_order_884_linux_test.go
@@ -0,0 +1,150 @@
+//go:build linux
+
+package iouring
+
+import (
+	"errors"
+	"net"
+	"runtime"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// celeris#884 item 71 (PR #1020 review r2): no test checked that a worker whose
+// goroutine ended without returning has left the SO_REUSEPORT group BEFORE it
+// is marked exited and the pause notified (reportWorkerExit), nor that
+// leaveGroupAfterExit closes the ring. The tests of worker_exit_pause_lifecycle
+// pass if the shutdown(SHUT_RD) and the close move after the flag and the
+// notify, because abnormalWorkerExits is counted first and a pause's own
+// wait hides the rest. The epoll half of the item (stopAccepting against the
+// flags and the notify) is left to that engine's lane.
+
+// TestLeaveGroupAfterExitClosesTheRingAndTheListener884 is the unit half: after
+// leaveGroupAfterExit the listen socket is out of the group (a connect is
+// refused, which the shutdown(SHUT_RD) alone brings about at once, before any
+// close), its descriptor and the ring's are closed, and the worker says so
+// (listenFDClosed, which PauseAccept waits on). Red with the ring's Close
+// removed, or with the listener's.
+func TestLeaveGroupAfterExitClosesTheRingAndTheListener884(t *testing.T) {
+	ring, err := NewRing(64, 0, 0)
+	if err != nil {
+		skipOrFail656(t, "io_uring unavailable: %v", err)
+	}
+	ringFD := ring.fd
+	lfd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := unix.SetsockoptInt(lfd, unix.SOL_SOCKET, unix.SO_REUSEPORT, 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := unix.Bind(lfd, &unix.SockaddrInet4{Addr: [4]byte{127, 0, 0, 1}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := unix.Listen(lfd, 8); err != nil {
+		t.Fatal(err)
+	}
+	sa, _ := unix.Getsockname(lfd)
+	addr := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: sa.(*unix.SockaddrInet4).Port}
+	t.Cleanup(func() {
+		if fdOpen704(lfd) {
+			_ = unix.Close(lfd)
+		}
+		if fdOpen704(ringFD) {
+			_ = ring.Close()
+		}
+	})
+	// The control: the listener takes a connection before.
+	if c, err := net.DialTimeout("tcp", addr.String(), time.Second); err != nil {
+		t.Fatalf("setup: the listener refused a connection: %v", err)
+	} else {
+		_ = c.Close()
+	}
+
+	w := &Worker{id: 3, ring: ring, listenFD: lfd, handoffLoss: &handoffLossStats{}}
+	w.leaveGroupAfterExit()
+
+	if _, err := net.DialTimeout("tcp", addr.String(), time.Second); !errors.Is(err, syscall.ECONNREFUSED) {
+		t.Errorf("a connect after leaveGroupAfterExit gave %v, want connection refused: the listener is still in the group", err)
+	}
+	if w.ring != nil || fdOpen704(ringFD) {
+		t.Errorf("the ring is not closed after leaveGroupAfterExit: w.ring=%v fd open=%v", w.ring != nil, fdOpen704(ringFD))
+	}
+	if w.listenFD != -1 || fdOpen704(lfd) {
+		t.Errorf("the listener is not closed after leaveGroupAfterExit: listenFD=%d fd open=%v", w.listenFD, fdOpen704(lfd))
+	}
+	if !w.listenFDClosed.Load() {
+		t.Error("listenFDClosed is not set after leaveGroupAfterExit: a pause would wait for it")
+	}
+}
+
+// TestGoexitedWorkerLeavesTheGroupBeforeItIsMarkedExited884 is the engine half,
+// in the child process every test that kills a worker runs in. A watcher per
+// worker spins on exited; at the instant it sees it set, the listener must
+// already be closed (listenFDClosed is stored by leaveGroupAfterExit after its
+// shutdown(SHUT_RD) and its closes), which is what lets PauseAccept, woken by
+// the notify that follows the flag, promise the group has shed the worker. Red
+// with reportWorkerExit's leaveGroupAfterExit moved after its exited.Store.
+func TestGoexitedWorkerLeavesTheGroupBeforeItIsMarkedExited884(t *testing.T) {
+	if !child848(t) {
+		return
+	}
+	e, _, cancel, done := start848(t, "127.0.0.1:0", true)
+	addr := e.Addr().String()
+	e.mu.Lock()
+	ws := append([]*Worker(nil), e.workers...)
+	e.mu.Unlock()
+
+	type seen struct {
+		w            *Worker
+		listenClosed bool
+		at           time.Time
+	}
+	results := make(chan seen, len(ws))
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+	for _, w := range ws {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for !w.exited.Load() {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				runtime.Gosched()
+			}
+			results <- seen{w, w.listenFDClosed.Load(), time.Now()}
+		}()
+	}
+	// The worker that serves this connection runs the handler that calls
+	// runtime.Goexit.
+	boom, err := net.DialTimeout("tcp", addr, 2*time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = boom.Close() }()
+	_, _ = boom.Write([]byte("GET /boom HTTP/1.1\r\nHost: x\r\n\r\n"))
+	var got seen
+	select {
+	case got = <-results:
+	case <-time.After(5 * time.Second):
+		close(stop)
+		t.Fatal("PREMISE: no worker was marked exited within 5s of the Goexit")
+	}
+	close(stop)
+	wg.Wait()
+	if !got.listenClosed {
+		t.Errorf("worker %d was seen marked exited with its listener not closed: the worker left the SO_REUSEPORT group after the flag and the notify a PauseAccept waits on, not before", got.w.id)
+	}
+	if e.abnormalWorkerExits.Load() != 1 {
+		t.Errorf("PREMISE: abnormalWorkerExits = %d, want 1", e.abnormalWorkerExits.Load())
+	}
+	stop848(t, cancel, done)
+	t.Logf("celeris848 RESULT iouring-exit-order worker=%d listener_closed_at_exited=%v", got.w.id, got.listenClosed)
+}

From 2bfebedda2c4b63522f4ff19e6303668ff9d1a5d Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:13:00 +0100
Subject: [PATCH 07/17] iouring: reportWorkerExit order test at the function
 (v1.6.0, Refs #884 item 71)

---
 .../worker_exit_order_884_linux_test.go       | 58 +++++++++++++++++++
 1 file changed, 58 insertions(+)

diff --git a/internal/engine/iouring/worker_exit_order_884_linux_test.go b/internal/engine/iouring/worker_exit_order_884_linux_test.go
index 1a37fa74a..7245cf5a4 100644
--- a/internal/engine/iouring/worker_exit_order_884_linux_test.go
+++ b/internal/engine/iouring/worker_exit_order_884_linux_test.go
@@ -12,6 +12,8 @@ import (
 	"time"
 
 	"golang.org/x/sys/unix"
+
+	"github.com/goceleris/celeris/internal/deferlinger"
 )
 
 // celeris#884 item 71 (PR #1020 review r2): no test checked that a worker whose
@@ -148,3 +150,59 @@ func TestGoexitedWorkerLeavesTheGroupBeforeItIsMarkedExited884(t *testing.T) {
 	stop848(t, cancel, done)
 	t.Logf("celeris848 RESULT iouring-exit-order worker=%d listener_closed_at_exited=%v", got.w.id, got.listenClosed)
 }
+
+// TestReportWorkerExitLeavesTheGroupBeforeTheFlag884 is the same order at the
+// function, many times over (a watcher that sees the flag set between the flag
+// and the end of the group's shedding is a race the engine arm catches once in
+// a few runs): reportWorkerExit on a worker holding a listening socket, with a
+// goroutine spinning on exited; whenever it sees the flag, listenFDClosed, which
+// leaveGroupAfterExit stores after its shutdown(SHUT_RD) and its closes, must
+// be set; and a goroutine told by the pause's channel (the notify) must find it
+// set too. Red with leaveGroupAfterExit moved after the flag and the notify.
+func TestReportWorkerExitLeavesTheGroupBeforeTheFlag884(t *testing.T) {
+	const rounds = 400
+	e := &Engine{}
+	early := 0
+	for round := range rounds {
+		lfd, err := unix.Socket(unix.AF_INET, unix.SOCK_STREAM|unix.SOCK_CLOEXEC, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := unix.Bind(lfd, &unix.SockaddrInet4{Addr: [4]byte{127, 0, 0, 1}}); err != nil {
+			_ = unix.Close(lfd)
+			t.Fatal(err)
+		}
+		if err := unix.Listen(lfd, 1); err != nil {
+			_ = unix.Close(lfd)
+			t.Fatal(err)
+		}
+		w := &Worker{id: round, listenFD: lfd, handoffLoss: &handoffLossStats{}, pause: &deferlinger.PauseState{}}
+		changed := w.pause.Changed()
+		saw := make(chan bool, 1)
+		go func() {
+			for !w.exited.Load() {
+				runtime.Gosched()
+			}
+			saw <- w.listenFDClosed.Load()
+		}()
+		notified := make(chan bool, 1)
+		go func() {
+			<-changed
+			notified <- w.listenFDClosed.Load()
+		}()
+		e.reportWorkerExit(w)
+		if !<-saw {
+			early++
+		}
+		if !<-notified {
+			early++
+		}
+		if fdOpen704(lfd) {
+			_ = unix.Close(lfd)
+			t.Fatalf("round %d: reportWorkerExit returned with the listener open", round)
+		}
+	}
+	if early > 0 {
+		t.Errorf("%d of %d observations (the flag, the notify) found the worker's listener not yet closed: the worker is marked exited and the pause notified before it has left the SO_REUSEPORT group", early, 2*rounds)
+	}
+}

From 85a1c29c52af1489a588cbf830edd0005849ed85 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:13:36 +0100
Subject: [PATCH 08/17] iouring: comments for the gate, and item 63's known
 limit (v1.6.0, Refs #884)

---
 internal/engine/iouring/fd_lifetime.go        | 10 +++-
 .../shutdown_drain_trials_884_linux_test.go   |  6 ++-
 internal/engine/iouring/worker.go             | 48 ++++++++++++++-----
 test/server/shutdown_h2_pool_linux_test.go    | 11 +++--
 4 files changed, 54 insertions(+), 21 deletions(-)

diff --git a/internal/engine/iouring/fd_lifetime.go b/internal/engine/iouring/fd_lifetime.go
index 9688c8bec..f0b1f408d 100644
--- a/internal/engine/iouring/fd_lifetime.go
+++ b/internal/engine/iouring/fd_lifetime.go
@@ -740,7 +740,15 @@ func (w *Worker) endOwedOpsAtShutdown() {
 					// (a pause, a resume and a shutdown at once; cancelAccept
 					// reaps inline, which keeps that window narrow) is not
 					// taken for it unless a re-created socket reused the
-					// number (appended to #884).
+					// number, which it usually does. Known limit (celeris#884
+					// item 63, decided 2026-10-12 as a comment, not a fix): the
+					// wait then ends early and the port stays bound for the
+					// ring teardown's ~2 ms. Counting terminal accept CQEs
+					// against the cancel's result (the number it cancelled,
+					// listenCancelAnswer's positive count) would close it, but
+					// needs the order of the cancel's CQE and the accept's
+					// terminal CQE pinned on every kernel CI runs. Narrow, not
+					// reproduced; cancelAccept has the same limit.
 					if !cqeHasMore(c.Flags) && decodeFD(ud) == listenFDAtCancel {
 						acceptOwed = false
 					}
diff --git a/internal/engine/iouring/shutdown_drain_trials_884_linux_test.go b/internal/engine/iouring/shutdown_drain_trials_884_linux_test.go
index 4fd9a4044..ac7c12b3d 100644
--- a/internal/engine/iouring/shutdown_drain_trials_884_linux_test.go
+++ b/internal/engine/iouring/shutdown_drain_trials_884_linux_test.go
@@ -12,8 +12,10 @@ import (
 // (endOwedOpsAtShutdown) ends the ops the kernel still owes on a live
 // connection's descriptor before shutdown closes it. TestShutdownDrainSurvivesAFullCQRing873
 // and the 880 tests pin it for a standalone recv; PR #1099 recorded two forms
-// they do not reach, and its open question, keep or remove the drain, is
-// answered by what these show: the drain is the only thing that ends them.
+// they do not reach. These close them. The open question of the item, keep or
+// remove the drain, is not decided here: what they show is that the drain is
+// the only thing in the engine that ends these two ops before the descriptor is
+// closed, not whether a kernel would issue work after that (c: not measured).
 //
 //   - (a) a recv linked behind a SEND whose CQE the loop has already read. The
 //     kernel consumed both SQEs at the submit; the recv is issued, and its
diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go
index 658cb296a..2f304f676 100644
--- a/internal/engine/iouring/worker.go
+++ b/internal/engine/iouring/worker.go
@@ -819,8 +819,10 @@ type Worker struct {
 
 	// refuseRecv is true once the send drain, or the wait for the HTTP/2
 	// pool handlers before it, is past its first shutdownSendDrainNanos:
-	// handleRecv then reads nothing more from an HTTP/1 connection
-	// (celeris#806, celeris#884). The drain lasts as long as the
+	// handleRecv then reads no new request from an HTTP/1 connection
+	// (celeris#806, celeris#884): a request in progress is read on
+	// (gateReadsOn), and the conns that hold nothing are closed
+	// (closeIdleH1, shutdown_gate_884.go). The drain lasts as long as the
 	// budget, and on keep-alive connections whose responses queue in the
 	// worker, new requests would keep the queues from ever emptying.
 	// Worker thread only; read in handleRecv.
@@ -1593,11 +1595,16 @@ func (w *Worker) run(ctx context.Context) {
 				// the walk over every conn is not paid on every pass); the
 				// HTTP/2 conns this wait is for are served on
 				// (celeris#884). h2PoolSettled has set h2DrainStart. A
-				// request whose bytes have not all arrived by then is not
-				// answered: the bytes after the gate are dropped (a slow
-				// upload, headers in pieces). The producers' OnError and
-				// OnDetachClose run on this thread, as in the send drain,
-				// so a slow callback delays the HTTP/2 conns served here.
+				// request in progress at the gate (a slow upload, headers in
+				// pieces) is read on and answered, with Connection: close, as
+				// std's Shutdown answers an active conn; the conns that hold
+				// nothing (idle keep-alive, no byte sent, a detached stream
+				// the gate ended) are closed (closeIdleH1), paced with the
+				// walk, so a client fails at once instead of waiting for the
+				// end of the wait with no answer (celeris#884 item 73). The
+				// producers' OnError and OnDetachClose run on this thread, as
+				// in the send drain, so a slow callback delays the HTTP/2
+				// conns served here.
 				if now := time.Now().UnixNano(); now > w.h2DrainStart+shutdownSendDrainNanos {
 					w.refuseRecv = true
 					if w.stopProducersPaced(now) {
@@ -2902,6 +2909,18 @@ func (w *Worker) cancelAccept(ctx context.Context, lfd int) {
 				}
 			default:
 				w.processCQE(ctx, c, now)
+				// The terminal CQE of an accept of THIS descriptor number ends
+				// the wait (celeris#896). Known limit (celeris#884 item 63,
+				// decided 2026-10-12 as a comment, not a fix): a terminal CQE
+				// of an earlier listen socket that is still unread, when a
+				// re-created socket shares its number (a pause, a resume and a
+				// shutdown together: the re-created socket usually gets the
+				// same number), is taken for this one and ends the wait early;
+				// the port then stays bound for the ring teardown's ~2 ms.
+				// Counting terminal accept CQEs against the cancel's result
+				// (the number it cancelled) would close it, but needs the
+				// order of the cancel's CQE and the accept's terminal CQE
+				// pinned on every kernel CI runs. Narrow, not reproduced.
 				if ud&udMask == udAccept && decodeFD(ud) == lfd && !cqeHasMore(c.Flags) {
 					acceptOwed = false
 				}
@@ -3969,12 +3988,15 @@ func (w *Worker) handleRecv(c *completionEntry, fd int, now int64) {
 	// more (celeris#806): the drain finishes the responses it has, and a
 	// request taken now could only be answered into a queue that the
 	// drain, which lasts as long as the budget, would never find empty
-	// under steady keep-alive traffic. Epoll's shutdown phases stop reading
-	// (its pool wait does not: celeris#885). The bytes are dropped with the
-	// connection, which shutdown closes; the recv is not re-armed, so a
-	// request whose bytes were still arriving at the gate (the body of an
-	// upload, headers in pieces) is never answered. A detached conn is read no
-	// more either (its writes were ended with the gate: stopDetachedProducers).
+	// under steady keep-alive traffic. The bytes of a NEW request are dropped
+	// and the recv is not re-armed: the conn holds nothing, and the gate's pass
+	// closes it (closeIdleH1). A request already IN PROGRESS (the body of an
+	// upload, headers in pieces, the bytes of a protocol decision begun) is read
+	// on, as std's Shutdown answers an active conn, and answered with
+	// Connection: close (gateReadsOn, celeris#884 item 73); that includes a
+	// conn whose dispatch goroutine owns the parse state, which epoll cannot
+	// yet look inside. A detached conn is read no more (its writes were ended
+	// with the gate: stopDetachedProducers).
 	// HTTP/2 conns are served on: they have had GOAWAY since the pool wait,
 	// and what their responses may still wait for was waited for there or
 	// is out of time.
diff --git a/test/server/shutdown_h2_pool_linux_test.go b/test/server/shutdown_h2_pool_linux_test.go
index 2a212847c..931a621d5 100644
--- a/test/server/shutdown_h2_pool_linux_test.go
+++ b/test/server/shutdown_h2_pool_linux_test.go
@@ -652,11 +652,12 @@ func TestShutdownWaitsForFlowControlledH2Response(t *testing.T) {
 // The gate comes shutdownSendDrainNanos (250 ms) into the wait, as the send
 // drain's does, so a request in flight when the shutdown begins is still
 // answered: the first assertion is that requests made as soon as every
-// listener has closed are. A request whose bytes have not all arrived by the
-// gate is another matter: the gate drops the bytes that come after it and the
-// recv is not re-armed (a slow upload, headers in pieces), which the pull
-// request's open questions record. The oracle looks 1.5 s in, well past the grace and the time a worker idle in its ring
-// wait takes to notice the cancel.
+// listener has closed are. A request in progress at the gate (a slow upload,
+// headers in pieces) is another matter, pinned by TestShutdownPoolWaitGateRequestsInProgressAndIdleConns885
+// on epoll and by TestIOUringShutdownGateConns884 on io_uring (both read it on
+// and answer it with Connection: close). The oracle looks 1.5 s in, well past
+// the grace and the time a worker idle in its ring wait takes to notice the
+// cancel.
 //
 // Both native engines have the gate: io_uring's (celeris#884) and epoll's
 // (celeris#885 item 49, poolWaitGate). Before epoll's, its pool wait did not

From 9dd2bc81720dda68ac0456d53398bbb301121b87 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:38:29 +0100
Subject: [PATCH 09/17] iouring: the stuck-ring engine arm asks a survivor only
 when a second worker exists (v1.6.0, Refs #884 item 72)

---
 internal/engine/iouring/ring_wait_ceiling_884_linux_test.go | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go b/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
index 8493c63c6..765e0370d 100644
--- a/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
+++ b/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
@@ -150,6 +150,10 @@ func TestRunStopsOnAStuckRingWait884(t *testing.T) {
 			t.Errorf("worker %d stopped though only worker %d met the failure", w.id, dead.id)
 		}
 	}
+	if len(e.workers) < 2 {
+		t.Log("one worker (the memlock cap): no survivor to ask")
+		return
+	}
 	c, err := net.DialTimeout("tcp", addr, time.Second)
 	if err != nil {
 		t.Fatalf("dial after the stuck worker stopped: %v", err)

From f532c024fbbe76d9f50e19a9b3f3defc201e3211 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:53:45 +0100
Subject: [PATCH 10/17] test: the download conn must end with a FIN, not a
 reset (v1.6.0, Refs #884)

---
 .../shutdown_iouring_gate_884_linux_test.go   | 25 +++++--------------
 1 file changed, 6 insertions(+), 19 deletions(-)

diff --git a/test/server/shutdown_iouring_gate_884_linux_test.go b/test/server/shutdown_iouring_gate_884_linux_test.go
index 563ed8ea9..ad082dd53 100644
--- a/test/server/shutdown_iouring_gate_884_linux_test.go
+++ b/test/server/shutdown_iouring_gate_884_linux_test.go
@@ -288,17 +288,6 @@ func runIOUringGateArm884(t *testing.T, eng celeris.EngineType, gate string, asy
 
 	// The rest of every request in progress arrives after the gate.
 	time.Sleep(time.Until(begin.Add(1200 * time.Millisecond)))
-	// The download's client sends more after the gate, twice, which the engine
-	// does not read (the conn is not idle, but the gate reads a request only
-	// when one is in progress): bytes that stay unread in the kernel, so a close
-	// of that conn while any of the response is still unacknowledged in the
-	// kernel's send queue would be a reset that throws the tail of it away.
-	for range 2 {
-		if _, err := io.WriteString(dl, "GET /ping HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
-			t.Errorf("a request written on the download's conn after the gate: %v", err)
-		}
-		time.Sleep(100 * time.Millisecond)
-	}
 	var pwg sync.WaitGroup
 	for _, p := range inProgress {
 		pwg.Add(1)
@@ -389,8 +378,9 @@ func runIOUringGateArm884(t *testing.T, eng celeris.EngineType, gate string, asy
 	if err != nil {
 		t.Fatalf("the download's head: %v", err)
 	}
-	// Read slowly (about 20 MiB/s), so that the kernel's send queue (a few MiB)
-	// is still draining when the engine's own queue has emptied.
+	// Read slowly (about 20 MiB/s): the conn's engine queue empties while the
+	// kernel's send queue (a few MiB) is still draining, and the conn must be
+	// closed only after that, with a FIN.
 	var dlN int64
 	var dlErr error
 	for chunk := make([]byte, 64<<10); dlErr == nil; {
@@ -402,13 +392,10 @@ func runIOUringGateArm884(t *testing.T, eng celeris.EngineType, gate string, asy
 	if dlN != int64(len(dlBody)) {
 		t.Errorf("the queued download delivered %d of %d bytes (%v): the gate cut a response that was still going out", dlN, len(dlBody), dlErr)
 	}
-	// Closed once drained: a FIN, or, because the client's two requests after
-	// the gate were never read and close(2) with unread bytes resets, an RST that
-	// comes after the whole response has been received (the length check above).
-	// Silence is what the gate must not leave.
+	// Closed once drained: a FIN (not a reset, not silence).
 	_ = dl.SetReadDeadline(time.Now().Add(3 * time.Second))
-	if _, err := dbr.ReadByte(); err != io.EOF && !errors.Is(err, syscall.ECONNRESET) {
-		t.Errorf("after the download the conn read %v, want a close (FIN, or a reset after the whole response): a drained conn was left open", err)
+	if _, err := dbr.ReadByte(); err != io.EOF {
+		t.Errorf("after the download the conn read %v, want a FIN (io.EOF): a drained conn was left open", err)
 	}
 
 	for _, c := range holders { // the holders go: the wait ends

From 556a5f231a419d79bdc372757051a152096b3946 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 22:56:49 +0100
Subject: [PATCH 11/17] epoll: the pool wait gate's comments no longer say
 io_uring has no idle close (v1.6.0, Refs #884)

---
 internal/engine/epoll/poolwait_gate_885.go | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/internal/engine/epoll/poolwait_gate_885.go b/internal/engine/epoll/poolwait_gate_885.go
index 03e59c0b1..c8d4ab2ff 100644
--- a/internal/engine/epoll/poolwait_gate_885.go
+++ b/internal/engine/epoll/poolwait_gate_885.go
@@ -43,7 +43,8 @@ import (
 // route every conn whose read leaves a partial request is promoted, whatever its
 // route, and a conn on an Async route is) is otherwise left to the end of the
 // wait, with a request in progress on it unread: the documented behaviour and
-// a follow-up shared with io_uring (which has no idle close at all).
+// a follow-up (io_uring's gate, shutdown_gate_884.go, looks inside such a conn
+// under the goroutine's own locks and closes it when idle).
 
 // poolWaitGrace is how long into the wait for the HTTP/2 pool handlers the
 // gate stays open: io_uring's shutdownSendDrainNanos, and this engine's
@@ -58,10 +59,10 @@ const gateClosesPerPass = 64
 
 // refusesRecv reports whether a loop past the pool wait's gate reads no more
 // from cs: an HTTP/1 conn (detached or not) or one whose protocol is not
-// detected yet, which io_uring's refuseRecv test mirrors, with one exception
-// 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
+// detected yet, which io_uring's gateReadsOn mirrors: 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.
 func refusesRecv(cs *connState) bool {

From e4ced28c23b9621d792da8ecc7242bb1dc73b310 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 23:31:45 +0100
Subject: [PATCH 12/17] test: failing regression tests for the gate's h1State
 read order and the ring-wait run across a long pass (v1.6.0, Refs #884)

---
 .../ring_wait_ceiling_884_linux_test.go       | 20 ++++++++++++
 .../iouring/shutdown_gate_884_linux_test.go   | 31 +++++++++++++++++++
 2 files changed, 51 insertions(+)

diff --git a/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go b/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
index 765e0370d..db691c2b3 100644
--- a/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
+++ b/internal/engine/iouring/ring_wait_ceiling_884_linux_test.go
@@ -167,3 +167,23 @@ func TestRunStopsOnAStuckRingWait884(t *testing.T) {
 		t.Errorf("the surviving worker did not answer: %v", err)
 	}
 }
+
+// TestRingWaitContinuesDoesNotCarryARunAcrossALongPass884 is the regression test
+// of the review's m1 (PR #1123, round 1): a retried wait, then one loop pass that
+// lasts longer than the ceiling (a park in SUSPENDED, or a slow callback in the
+// gate pass: the pass moves tickCounter by one and takes over a second), then one
+// more retried wait. The two retries are one pass apart, but they are not
+// consecutive: the run's allowed spacing is a few passes' time, and a gap longer
+// than that ends it. Before, the worker stopped on the second retry of its life.
+func TestRingWaitContinuesDoesNotCarryARunAcrossALongPass884(t *testing.T) {
+	ring := newTestRing(t)
+	w := &Worker{ring: ring, id: 3, logger: slog.New(&warnLog884{})}
+	if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
+		t.Fatal("the first retry gave up")
+	}
+	w.tickCounter++ // the pass that parks
+	time.Sleep(1100 * time.Millisecond)
+	if !w.ringWaitContinues(waitModeTimed, ceilingErr884()) {
+		t.Fatal("the worker gave up on its second retried wait, after one long pass between two isolated retries: a run is carried across a gap")
+	}
+}
diff --git a/internal/engine/iouring/shutdown_gate_884_linux_test.go b/internal/engine/iouring/shutdown_gate_884_linux_test.go
index c10d5bda3..60032c648 100644
--- a/internal/engine/iouring/shutdown_gate_884_linux_test.go
+++ b/internal/engine/iouring/shutdown_gate_884_linux_test.go
@@ -361,3 +361,34 @@ func TestGatePassWaitsForTheKernelSendQueue884(t *testing.T) {
 	}
 	_ = unix.Close(local)
 }
+
+// TestGateReadsOnNeverReadsH1StateOfAGoroutinesConn884 is the regression test of
+// the review's major finding (PR #1123, round 1): gateReadsOn read cs.h1State
+// and h.Detached before it checked asyncPromoted. On a promoted conn the
+// dispatch goroutine owns h1State: switchToH2Local sets it to nil under
+// detachMu during an h2c upgrade (celeris#722: the worker may not read it while
+// the goroutine can be running). The goroutine here is in that upgrade while the
+// worker's handleRecv, past the gate, asks gateReadsOn; it is a data race, so
+// the test only fails under -race (the repository's CI runs -race).
+func TestGateReadsOnNeverReadsH1StateOfAGoroutinesConn884(t *testing.T) {
+	for range 20 {
+		w := &Worker{}
+		cs := gateConn884(7)
+		cs.detachMu = &sync.Mutex{}
+		cs.asyncPromoted.Store(true)
+		cs.asyncRun = true // the goroutine is out of its park loop: dispatchHasWork says busy
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() { // the dispatch goroutine, inside switchToH2Local
+			defer wg.Done()
+			cs.detachMu.Lock()
+			cs.h1State = nil
+			cs.protocol.Store(int32(engine.H2C))
+			cs.detachMu.Unlock()
+		}()
+		if !w.gateReadsOn(cs) {
+			t.Error("a promoted conn whose goroutine is busy was not read on")
+		}
+		wg.Wait()
+	}
+}

From d5fcd86ebe327dd3c3802bf099946b37324d54b9 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 23:34:19 +0100
Subject: [PATCH 13/17] iouring: gateReadsOn checks asyncPromoted before it
 reads h1State, the ring-wait run ends at a gap, comments (v1.6.0, Refs #884)

---
 .../iouring/park_retracts_residue_test.go     |  7 +++-
 internal/engine/iouring/shutdown_gate_884.go  | 32 +++++++++++++++----
 internal/engine/iouring/shutdown_ring_wait.go |  7 ++++
 internal/engine/iouring/worker.go             | 23 +++++++++----
 .../worker_exit_order_884_linux_test.go       |  3 ++
 5 files changed, 58 insertions(+), 14 deletions(-)

diff --git a/internal/engine/iouring/park_retracts_residue_test.go b/internal/engine/iouring/park_retracts_residue_test.go
index 2b8867f46..dddf119f4 100644
--- a/internal/engine/iouring/park_retracts_residue_test.go
+++ b/internal/engine/iouring/park_retracts_residue_test.go
@@ -27,7 +27,12 @@ package iouring
 // that close (celeris#884 item 80): the conn's recv is ended and reaped inside
 // the sweep's own pass, so its close lands after sweep() with nothing owed to
 // the kernel, the gate does not hold the worker, and only the retraction at
-// the park clears the gauges (red with it removed).
+// the park clears the gauges (red with it removed). What this proves: the
+// retraction is needed for that state. What it does not: that production
+// reaches that state (no traffic path was found that leaves the last conn with
+// no op in flight after sweep(); the state is manufactured in the hook), so the
+// item's other outcome, calling the retraction dead behind the closeFDOwed gate,
+// is not shown either.
 
 import (
 	"context"
diff --git a/internal/engine/iouring/shutdown_gate_884.go b/internal/engine/iouring/shutdown_gate_884.go
index 49bae688a..303e8f42f 100644
--- a/internal/engine/iouring/shutdown_gate_884.go
+++ b/internal/engine/iouring/shutdown_gate_884.go
@@ -61,13 +61,16 @@ func (w *Worker) gateReadsOn(cs *connState) bool {
 		// bytes): a request that begins after the gate.
 		return len(cs.detectAccum) > 0
 	}
+	// Before any read of h1State: on a promoted conn the dispatch goroutine owns
+	// the field (switchToH2Local sets it to nil during an h2c upgrade, celeris#722),
+	// so it is read only under detachMu (promotedInProgress).
+	if cs.asyncPromoted.Load() {
+		return w.promotedInProgress(cs)
+	}
 	h := cs.h1State
 	if h == nil || h.Detached.Load() {
 		return false // a detached stream is read no more: its writes were ended with the gate
 	}
-	if cs.asyncPromoted.Load() {
-		return w.promotedInProgress(cs)
-	}
 	// A conn the worker parses itself: its h1State is the worker's to read.
 	if cs.recvIntoBody || h.HasPendingData() || !h.AtRequestBoundary() {
 		h.CloseAfter = true
@@ -82,6 +85,12 @@ func (w *Worker) gateReadsOn(cs *connState) bool {
 // request. It marks an idle conn close-after too: the worker does not know
 // that no request will follow before the pass closes the conn. Worker thread.
 func (w *Worker) promotedInProgress(cs *connState) bool {
+	// A detached stream whose writes were ended (by the gate or a close) is read
+	// no more; writeEnded is an atomic, so this holds even while a guarded write
+	// has detachMu.
+	if cs.writeEnded.Load() {
+		return false
+	}
 	if dispatchHasWork(cs) {
 		return true
 	}
@@ -117,7 +126,11 @@ func dispatchHasWork(cs *connState) bool {
 // hold nothing are closed (gateCloseable). A conn whose response is still going
 // out, or whose lock is busy, is met again by the next pass. closeConn removes
 // a conn from liveConns by swapping it with the last entry, so the walk goes
-// from the top down, as checkTimeouts does. Worker thread.
+// from the top down, as checkTimeouts does. The pass runs at the top of the loop
+// pass, before the pass reaps new completions: a request whose recv CQE was
+// posted after the last reap meets a conn that looks idle and is closed with
+// that request unread, the race std's own idle close has (microseconds wide;
+// not measured). Worker thread.
 func (w *Worker) closeIdleH1() {
 	closed := 0
 	for i := len(w.liveConns) - 1; i >= 0; i-- {
@@ -132,12 +145,19 @@ func (w *Worker) closeIdleH1() {
 		if !w.gateCloseable(cs) {
 			continue
 		}
+		// Past the cap the conn was still looked at, so it is marked close-after
+		// (gateCloseable); it is closed by a later pass.
 		if closed >= gateClosesPerPass {
 			continue
 		}
 		// The kernel's send queue: a conn whose response left user space but
-		// not the host is not idle either (a close with unsent bytes would race
-		// them with the FIN). A fixed-file conn's slot is an index into the
+		// has not all been acknowledged by the peer is not idle yet. close(2)
+		// keeps the send queue and sends the FIN after it, so nothing is lost by
+		// closing early unless the peer sends after the close (the unread
+		// bytes turn the close into an RST, and the peer discards what it had
+		// not read) or an SO_LINGER{1,0} path is in play; waiting for an empty
+		// queue (SIOCOUTQ counts unacknowledged bytes) keeps that window shut,
+		// and is what epoll does. A fixed-file conn's slot is an index into the
 		// ring's file table, not a descriptor, so the ioctl cannot ask about it:
 		// such a conn is judged by its user-space queues alone
 		// (ownsDescriptorNumber).
diff --git a/internal/engine/iouring/shutdown_ring_wait.go b/internal/engine/iouring/shutdown_ring_wait.go
index 2129e6a9e..cc12e89f2 100644
--- a/internal/engine/iouring/shutdown_ring_wait.go
+++ b/internal/engine/iouring/shutdown_ring_wait.go
@@ -37,3 +37,10 @@ func ringWaitRetryable(err error) bool {
 // item 72). A full CQ ring is cleared by the loop's own reap within a pass or
 // two, so a second of retries that reap nothing is a ring that will not recover.
 const ringWaitCeiling = time.Second
+
+// ringWaitRunGap is the longest time between two retried waits that still make
+// one run (ringWaitContinues): a failing ring answers its retries a pass apart,
+// microseconds to a few milliseconds, and an idle worker's timed wait takes up
+// to 100 ms (the sleep of a seam-injected EBUSY test is that wait). A longer gap
+// means something else ran in between, so the run starts again.
+const ringWaitRunGap = 250 * time.Millisecond
diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go
index 2f304f676..a43177fed 100644
--- a/internal/engine/iouring/worker.go
+++ b/internal/engine/iouring/worker.go
@@ -968,10 +968,12 @@ type Worker struct {
 	// conns again (stopProducersPaced, closeIdlePaced). Worker thread only.
 	producersStopNext int64
 	// ringWaitRunStart is when (UnixNano) the run of consecutive retried ring
-	// waits that reaped nothing began, and ringWaitRunTick the loop pass
+	// waits that reaped nothing began, ringWaitRunLast when the last of them
+	// was, and ringWaitRunTick the loop pass
 	// (tickCounter) of the last of them: ringWaitContinues' ceiling
 	// (celeris#884 item 72). Zero start: no run. Worker thread only.
 	ringWaitRunStart int64
+	ringWaitRunLast  int64
 	ringWaitRunTick  uint32
 }
 
@@ -8048,7 +8050,9 @@ func (w *Worker) stopDetachedProducers() {
 // and ends the run; so does a wait that succeeded in between (the loop pass
 // count, which moves once per pass, jumps by more than one: a retry is part
 // of the run only if it follows the last one at once), and a run is not
-// carried over a gap longer than a wait takes. Any other error is the ring
+// carried over a gap longer than ringWaitRunGap between two retries, which is
+// more than a pass of a ring that fails takes and than a timed wait of an
+// idle one (a park, a slow callback in the gate pass, a long wait end it). Any other error is the ring
 // failing: the worker stops, and closes what it owns.
 func (w *Worker) ringWaitContinues(mode ringWaitMode, err error) bool {
 	if h := ringWaitSeam; h != nil {
@@ -8062,9 +8066,10 @@ func (w *Worker) ringWaitContinues(mode ringWaitMode, err error) bool {
 	}
 	w.ringWaitRetries.Add(1)
 	if !w.ringWaitWarned {
-		// Once per worker: there is no ceiling on the retries and the counter
-		// is not a public metric, so this line is how an operator on a 5.19
-		// to 6.1 kernel sees a loop that keeps retrying.
+		// Once per worker: the counter is not a public metric, so this line is
+		// how an operator on a 5.19 to 6.1 kernel sees a loop that keeps
+		// retrying (it stops after ringWaitCeiling of retries that reap
+		// nothing).
 		w.ringWaitWarned = true
 		if w.logger != nil {
 			w.logger.Warn("io_uring: a ring wait answered a retryable error (the CQ ring has no room); the worker reaps and retries, and does not log this again",
@@ -8077,8 +8082,11 @@ func (w *Worker) ringWaitContinues(mode ringWaitMode, err error) bool {
 		return true
 	}
 	now := time.Now().UnixNano()
-	if w.ringWaitRunStart == 0 || w.tickCounter-w.ringWaitRunTick > 1 {
-		w.ringWaitRunStart = now // a new run: a wait succeeded since the last retry
+	if w.ringWaitRunStart == 0 || w.tickCounter-w.ringWaitRunTick > 1 || now-w.ringWaitRunLast > int64(ringWaitRunGap) {
+		// A new run: a wait succeeded since the last retry (the pass count
+		// moved by more than one), or a pass took longer than a retried
+		// pass can (a park, a slow callback): the retries are not consecutive.
+		w.ringWaitRunStart = now
 	} else if now-w.ringWaitRunStart >= int64(ringWaitCeiling) {
 		if w.logger != nil {
 			w.logger.Error("io_uring: a ring wait has answered a retryable error for the whole ceiling with nothing to reap; the ring is failing, the worker stops",
@@ -8088,6 +8096,7 @@ func (w *Worker) ringWaitContinues(mode ringWaitMode, err error) bool {
 		return false
 	}
 	w.ringWaitRunTick = w.tickCounter
+	w.ringWaitRunLast = now
 	time.Sleep(50 * time.Microsecond)
 	return true
 }
diff --git a/internal/engine/iouring/worker_exit_order_884_linux_test.go b/internal/engine/iouring/worker_exit_order_884_linux_test.go
index 7245cf5a4..3f04e789a 100644
--- a/internal/engine/iouring/worker_exit_order_884_linux_test.go
+++ b/internal/engine/iouring/worker_exit_order_884_linux_test.go
@@ -84,6 +84,9 @@ func TestLeaveGroupAfterExitClosesTheRingAndTheListener884(t *testing.T) {
 	}
 }
 
+// (Not a guard on its own: under the order mutant it was red 3 runs of 5, the
+// watcher racing the worker; TestReportWorkerExitLeavesTheGroupBeforeTheFlag884
+// below is the deterministic detector, red 5 of 5.)
 // TestGoexitedWorkerLeavesTheGroupBeforeItIsMarkedExited884 is the engine half,
 // in the child process every test that kills a worker runs in. A watcher per
 // worker spins on exited; at the instant it sees it set, the listener must

From 40a54b8e6a6255257b4fca200b8aa3ee713676ee Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 23:34:37 +0100
Subject: [PATCH 14/17] test: failing regression test for
 stopDetachedProducers' h1State read order (v1.6.0, Refs #884)

---
 .../iouring/shutdown_gate_884_linux_test.go   | 30 +++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/internal/engine/iouring/shutdown_gate_884_linux_test.go b/internal/engine/iouring/shutdown_gate_884_linux_test.go
index 60032c648..b73a3e02c 100644
--- a/internal/engine/iouring/shutdown_gate_884_linux_test.go
+++ b/internal/engine/iouring/shutdown_gate_884_linux_test.go
@@ -392,3 +392,33 @@ func TestGateReadsOnNeverReadsH1StateOfAGoroutinesConn884(t *testing.T) {
 		wg.Wait()
 	}
 }
+
+// TestStopDetachedProducersNeverReadsH1StateOfAGoroutinesConn884 is the same
+// race in the walk the gate pass runs before closeIdleH1 (stopDetachedProducers
+// read cs.h1State and Detached before its detachMu TryLock, found by the search
+// the review's finding asked for): a promoted conn whose goroutine is in an h2c
+// upgrade sets h1State to nil under detachMu. Fails only under -race.
+func TestStopDetachedProducersNeverReadsH1StateOfAGoroutinesConn884(t *testing.T) {
+	for range 20 {
+		w := &Worker{conns: make([]*connState, 16)}
+		cs := gateConn884(7)
+		cs.detachMu = &sync.Mutex{}
+		cs.asyncPromoted.Store(true)
+		w.conns[cs.fd] = cs
+		w.addLiveConn(cs)
+		started := make(chan struct{})
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() { // the dispatch goroutine, inside switchToH2Local
+			defer wg.Done()
+			cs.detachMu.Lock()
+			close(started)
+			cs.h1State = nil
+			cs.protocol.Store(int32(engine.H2C))
+			cs.detachMu.Unlock()
+		}()
+		<-started
+		w.stopDetachedProducers()
+		wg.Wait()
+	}
+}

From 5170f236fa748c7b7db2ced50e16d19ccc9fd2de Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 23:34:50 +0100
Subject: [PATCH 15/17] iouring: stopDetachedProducers reads a promoted conn's
 h1State under detachMu (v1.6.0, Refs #884)

---
 internal/engine/iouring/worker.go | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/internal/engine/iouring/worker.go b/internal/engine/iouring/worker.go
index a43177fed..89a98a60e 100644
--- a/internal/engine/iouring/worker.go
+++ b/internal/engine/iouring/worker.go
@@ -8011,12 +8011,23 @@ func endDetachedWrites(cs *connState) {
 func (w *Worker) stopDetachedProducers() {
 	for _, fd := range w.liveConns {
 		cs := w.conns[fd]
-		if cs == nil || cs.detachMu == nil || cs.h1State == nil || !cs.h1State.Detached.Load() {
+		if cs == nil || cs.detachMu == nil {
+			continue
+		}
+		// h1State is read before the lock only on a conn the worker owns; on a
+		// promoted conn the dispatch goroutine may set it to nil (an h2c
+		// upgrade, celeris#722), so it is read under detachMu.
+		promoted := cs.asyncPromoted.Load()
+		if !promoted && (cs.h1State == nil || !cs.h1State.Detached.Load()) {
 			continue
 		}
 		if !cs.detachMu.TryLock() {
 			continue
 		}
+		if promoted && (cs.h1State == nil || !cs.h1State.Detached.Load()) {
+			cs.detachMu.Unlock()
+			continue
+		}
 		if !cs.detachClosed && !cs.asyncDetachPending {
 			endDetachedWrites(cs)
 			if !cs.sending && !cs.zcNotifPending && connSendPending(cs) {

From 046fe0fc56e773403c9bd7112ec115a372abd76c Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 26 Oct 2026 23:35:32 +0100
Subject: [PATCH 16/17] test: recvIntoBody implies a request in progress
 (v1.6.0, Refs #884)

---
 .../iouring/shutdown_gate_884_linux_test.go     | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/internal/engine/iouring/shutdown_gate_884_linux_test.go b/internal/engine/iouring/shutdown_gate_884_linux_test.go
index b73a3e02c..c9e9ccf38 100644
--- a/internal/engine/iouring/shutdown_gate_884_linux_test.go
+++ b/internal/engine/iouring/shutdown_gate_884_linux_test.go
@@ -422,3 +422,20 @@ func TestStopDetachedProducersNeverReadsH1StateOfAGoroutinesConn884(t *testing.T
 		wg.Wait()
 	}
 }
+
+// TestRecvIntoBodyImpliesPendingData884 pins why no server-level case can tell
+// gateCloseable's recvIntoBody term from HasPendingData (review m3): the worker
+// sets recvIntoBody only when NextRecvBuf returned a buffer, and NextRecvBuf
+// returns one only while bodyNeeded > 0, which HasPendingData also reports. The
+// term is defence in depth (the unit table holds it); an equivalent mutant on
+// the wire.
+func TestRecvIntoBodyImpliesPendingData884(t *testing.T) {
+	cs := gateConn884(7)
+	feed884(t, cs, "POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 100000\r\n\r\nabc")
+	if cs.h1State.NextRecvBuf() == nil {
+		t.Fatal("setup: no direct body buffer offered")
+	}
+	if !cs.h1State.HasPendingData() || cs.h1State.AtRequestBoundary() {
+		t.Error("a conn the worker would recv into the body buffer reports no request in progress")
+	}
+}

From 273374deec4a8deffdc512cba8ba084d1eabf200 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Tue, 27 Oct 2026 00:03:59 +0100
Subject: [PATCH 17/17] iouring: a test row for the writeEnded check of
 promotedInProgress, the ring-wait gap comment (v1.6.0, Refs #884)

---
 .../iouring/shutdown_gate_884_linux_test.go       | 14 ++++++++++++++
 internal/engine/iouring/shutdown_ring_wait.go     | 15 +++++++++++----
 2 files changed, 25 insertions(+), 4 deletions(-)

diff --git a/internal/engine/iouring/shutdown_gate_884_linux_test.go b/internal/engine/iouring/shutdown_gate_884_linux_test.go
index c9e9ccf38..9e830aadb 100644
--- a/internal/engine/iouring/shutdown_gate_884_linux_test.go
+++ b/internal/engine/iouring/shutdown_gate_884_linux_test.go
@@ -163,6 +163,20 @@ func TestGateLookAtAConn884(t *testing.T) {
 			cs.asyncRun, cs.asyncParked = true, true
 			return cs
 		}, true, false, false},
+		{"promoted-detached-ended-lock-held", func(t *testing.T) *connState {
+			// A promoted stream the gate ended, whose detachMu a guarded write
+			// holds right now: not read on, whatever the lock says (writeEnded is
+			// an atomic; review r2 m5).
+			cs := gateConn884(7)
+			cs.detachMu = &sync.Mutex{}
+			cs.detachMu.Lock()
+			cs.asyncPromoted.Store(true)
+			cs.asyncRun, cs.asyncParked, cs.asyncDetachUnlocked = true, true, true
+			cs.h1State.Detached.Store(true)
+			cs.detachClosed = true
+			cs.writeEnded.Store(true)
+			return cs
+		}, false, false, false},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			cs := tc.build(t)
diff --git a/internal/engine/iouring/shutdown_ring_wait.go b/internal/engine/iouring/shutdown_ring_wait.go
index cc12e89f2..3558e1a39 100644
--- a/internal/engine/iouring/shutdown_ring_wait.go
+++ b/internal/engine/iouring/shutdown_ring_wait.go
@@ -39,8 +39,15 @@ func ringWaitRetryable(err error) bool {
 const ringWaitCeiling = time.Second
 
 // ringWaitRunGap is the longest time between two retried waits that still make
-// one run (ringWaitContinues): a failing ring answers its retries a pass apart,
-// microseconds to a few milliseconds, and an idle worker's timed wait takes up
-// to 100 ms (the sleep of a seam-injected EBUSY test is that wait). A longer gap
-// means something else ran in between, so the run starts again.
+// one run (ringWaitContinues). A ring that is failing returns its error at
+// once, so its retries come a pass apart, microseconds to a few milliseconds
+// (the 50 us sleep plus one pass). An idle worker's successful timed wait takes
+// up to 100 ms, or 1 s for a long-idle one (baseTimeout), which is longer than
+// this gap: such a worker's retries are never joined into a run, and that is
+// the point, since a wait that took that long did not fail. What the rule
+// relies on is only that a real failure is fast: passes of 200 ms between
+// retries (the seam-injected test, whose real wait precedes the injected error)
+// still stop at about 1.01 s, passes of 300 ms never do, by design. A longer gap
+// means something else ran in between (a park, a slow callback), so the run
+// starts again.
 const ringWaitRunGap = 250 * time.Millisecond