From f20e1d4cae84fb2a36d7b3f0343094f6a30ff775 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 15:46:16 +0100
Subject: [PATCH 1/9] test(epoll): an async route's closing response is cut
 short when the loop looks between asyncClosed and endDispatch (failing first;
 Refs #1068, Refs #890)

serveAsync declares a conn closed (asyncClosed) after a request that ends it,
and only then takes asyncInMu to mark its goroutine gone (endDispatch). A
drainDetachQueue that reads the conn between the two sees a closed conn whose
goroutine still counts as running, calls closeConn, and the handler's detachMu
is free, so closeConn closes the descriptor with the rest of the response
unsent. The entry that makes the loop look is the one the goroutine enqueued
for the remainder of its partial flush.

Two test hooks hold the goroutine in that gap. With them the test fails every
time on this tree, with the byte counts of the CI failures: 2634119 of
3145728 body bytes (TestBackloggedPeerIsClosed, refused write) and 2634100 of
8388608 (TestStreamedResponseIsDelivered, Connection: close; CI saw 2681685
and 2634069, the bytes the runner's socket buffers took).
---
 .../async_close_flush_1068_linux_test.go      | 137 ++++++++++++++++++
 internal/engine/epoll/loop.go                 |  18 +++
 2 files changed, 155 insertions(+)
 create mode 100644 internal/engine/epoll/async_close_flush_1068_linux_test.go

diff --git a/internal/engine/epoll/async_close_flush_1068_linux_test.go b/internal/engine/epoll/async_close_flush_1068_linux_test.go
new file mode 100644
index 000000000..6522c375e
--- /dev/null
+++ b/internal/engine/epoll/async_close_flush_1068_linux_test.go
@@ -0,0 +1,137 @@
+//go:build linux
+
+package epoll
+
+import (
+	"bufio"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/goceleris/celeris/internal/resource"
+)
+
+// celeris#1068 (and the TestStreamedResponseIsDelivered/*/async-route/close
+// item of #890): an async-route response that ends its connection while its
+// remainder is still queued was cut short, at the same byte count each time
+// (the bytes the socket buffers took at the first flush: 2634119 of a
+// 3 MiB body, 2681685 of an 8 MiB one), and the client saw EOF.
+//
+// serveAsync, after a request that ends the conn (Connection: close, an error,
+// a refused write), declared the conn closed (asyncClosed) and only then took
+// asyncInMu to mark its goroutine gone (endDispatch). A drainDetachQueue that
+// read the conn between the two found a conn declared closed whose goroutine
+// was still counted as running, took that for a close requested from outside
+// while a handler runs, and called closeConn: the handler's detachMu was free
+// (the goroutine had released it after its flush), so closeConn went through
+// and closed the descriptor with the rest of the response unsent. The entry
+// that made the loop look was the one the goroutine enqueued for the remainder
+// of its partial flush, before it declared the close.
+//
+// The test holds the goroutine in that gap, with the response's remainder
+// queued, and waits for the loop to look. With the gap closed the loop cannot
+// look until the goroutine has finished (the wait ends on its timeout), and
+// sends the remainder before it closes.
+
+func TestAsyncRouteCloseSendsTheQueuedRemainder1068(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		// reqs requests are sent in one packet, header the lines after Host.
+		reqs   int
+		header string
+		body   int
+		// closesAtBoundary: the conn ends after whole responses only (the
+		// back-pressure refusal of celeris#761 drops the responses past the
+		// cap), so any number of them is right, but each one that is read
+		// must be whole, and then the stream ends (TestBackloggedPeerIsClosed
+		// takes an EOF in the next response's head as that end too).
+		closesAtBoundary bool
+	}{
+		// TestStreamedResponseIsDelivered/*/async-route/close: one response
+		// larger than the socket buffers, then Connection: close.
+		{"connection-close", 1, "Connection: close\r\n", 8 << 20, false},
+		// TestBackloggedPeerIsClosed/*/async-*: four pipelined 3 MiB responses
+		// for a client that reads nothing; the cap refuses one past 4 MiB.
+		{"refused-write", 4, "", 3 << 20, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var (
+				looked   = make(chan struct{})
+				lookOnce sync.Once
+				held     = make(chan struct{})
+				heldOnce sync.Once
+			)
+			exitHook := func(*connState) {
+				select {
+				case <-looked:
+				case <-time.After(500 * time.Millisecond):
+				}
+				heldOnce.Do(func() { close(held) })
+			}
+			drainHook := func(*connState) { lookOnce.Do(func() { close(looked) }) }
+			testHookAsyncExitClosed.Store(&exitHook)
+			testHookDrainAsyncClosed.Store(&drainHook)
+			t.Cleanup(func() {
+				testHookAsyncExitClosed.Store(nil)
+				testHookDrainAsyncClosed.Store(nil)
+			})
+
+			_, addr := startE2E885(t, true, newPathHandler885(tc.body), func(*resource.Config) {})
+			c, err := net.DialTimeout("tcp", addr, 3*time.Second)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() { _ = c.Close() }()
+			if _, err := io.WriteString(c, strings.Repeat("GET /big HTTP/1.1\r\nHost: x\r\n"+tc.header+"\r\n", tc.reqs)); err != nil {
+				t.Fatal(err)
+			}
+			// The client reads nothing until the goroutine has been held and
+			// let go, so the response cannot fit the socket buffers and its
+			// remainder is queued at every point of the hold.
+			select {
+			case <-held:
+			case <-time.After(10 * time.Second):
+				t.Fatal("the dispatch goroutine never reached the exit of a request that ends its conn")
+			}
+			_ = c.SetReadDeadline(time.Now().Add(20 * time.Second))
+			br := bufio.NewReaderSize(c, 64<<10)
+			whole := 0
+			for {
+				resp, err := http.ReadResponse(br, nil)
+				if err != nil {
+					if whole > 0 && tc.closesAtBoundary && (errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF)) {
+						break
+					}
+					t.Fatalf("response %d: %v (looked=%v)", whole+1, err, isClosed1068(looked))
+				}
+				n, err := io.Copy(io.Discard, resp.Body)
+				_ = resp.Body.Close()
+				if err != nil || n != int64(tc.body) {
+					t.Fatalf("response %d: received %d of %d body bytes (%v): the loop closed the conn with the rest of the response unsent while the goroutine was still on its way out (looked=%v)",
+						whole+1, n, tc.body, err, isClosed1068(looked))
+				}
+				whole++
+				if !tc.closesAtBoundary || whole == tc.reqs {
+					break
+				}
+			}
+			if tc.closesAtBoundary && whole == tc.reqs {
+				t.Fatalf("all %d responses were sent to a client that read nothing: the cap did not refuse one", tc.reqs)
+			}
+		})
+	}
+}
+
+func isClosed1068(ch chan struct{}) bool {
+	select {
+	case <-ch:
+		return true
+	default:
+		return false
+	}
+}
diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index 0b7f6fdf1..273d13978 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -1951,6 +1951,18 @@ func (l *Loop) drainH2Writes() {
 // and a nil test per close request, which is never a hot path.
 var testHookCloseWhenFlushed atomic.Pointer[func(*connState)]
 
+// testHookAsyncExitClosed, when set, runs in serveAsync's exit for a request
+// that ends its connection (Connection: close, an error, a refused write),
+// right after the goroutine has declared the conn closed (asyncClosed) and
+// before it has marked itself gone (endDispatch). testHookDrainAsyncClosed
+// runs on the loop's goroutine in drainDetachQueue when an entry names a conn
+// declared closed whose goroutine is still counted as running. Together they
+// let a test hold the goroutine in the one place the loop can see that
+// combination (celeris#1068). Atomic pointers, as testHookCloseWhenFlushed is;
+// a load and a nil test per close, never a hot path.
+var testHookAsyncExitClosed atomic.Pointer[func(*connState)]
+var testHookDrainAsyncClosed atomic.Pointer[func(*connState)]
+
 // discardWrite is the write function the run loop drains the H2 write queue of
 // a conn that is closing on a refused write into: the frames go nowhere.
 func discardWrite([]byte) {}
@@ -3399,6 +3411,9 @@ func (l *Loop) serveAsync(cs *connState, holds *dispatchHolds) {
 			// drainDetachQueue pass will observe asyncClosed and run
 			// closeConn thread-safely.
 			cs.asyncClosed.Store(true)
+			if h := testHookAsyncExitClosed.Load(); h != nil {
+				(*h)(cs)
+			}
 			cs.asyncInMu.Lock()
 			cs.asyncInBuf = cs.asyncInBuf[:0]
 			cs.endDispatch() // enqueued below: that is the hand-back
@@ -3519,6 +3534,9 @@ func (l *Loop) drainDetachQueue() {
 			running := cs.asyncRun
 			cs.asyncInMu.Unlock()
 			if running {
+				if h := testHookDrainAsyncClosed.Load(); h != nil {
+					(*h)(cs)
+				}
 				l.closeConn(cs.fd)
 			} else {
 				l.closeWhenFlushed(cs)

From 1488d0b6f2567625de1253f8d91d76cc842cf1d7 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 15:46:35 +0100
Subject: [PATCH 2/9] fix(epoll): the dispatch goroutine declares its conn
 closed and gone under one lock, so the loop cannot close it over a queued
 remainder (Refs #1068, Refs #890)

drainDetachQueue reads asyncClosed and then asyncRun under asyncInMu, and
treats a conn that is closed with its goroutine still running as a close asked
from outside while a handler runs: closeConn, which goes through at once when
detachMu is free. serveAsync stored asyncClosed before it took asyncInMu to
clear asyncRun, so the entry the goroutine enqueued for the remainder of its
partial flush could be drained in between, and the descriptor was closed with
that remainder unsent: the client read EOF at the bytes the socket buffers had
taken (2634119 of 3145728 in TestBackloggedPeerIsClosed, 2681685 of 8388608 in
TestStreamedResponseIsDelivered/*/async-route/close).

asyncClosed is now stored inside the asyncInMu section that clears asyncRun,
at the end-of-conn exit and at the two exits (a failed H2 upgrade flush, an
error after Detach) that stored it the same way. The loop reads asyncClosed
first, so a true it sees was stored inside the section, and the asyncInMu it
then takes waits for the section to end: it finds the goroutine gone and
closes through closeWhenFlushed, which sends the remainder first.

io_uring has no such gap: its async exit does not flush into the socket before
it declares the close (a request that ends the conn hands its staged bytes to
the worker, whose closeConn defers the close behind them).
---
 internal/engine/epoll/loop.go | 33 ++++++++++++++++++++++++---------
 1 file changed, 24 insertions(+), 9 deletions(-)

diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index 273d13978..5f2340168 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -1954,12 +1954,13 @@ var testHookCloseWhenFlushed atomic.Pointer[func(*connState)]
 // testHookAsyncExitClosed, when set, runs in serveAsync's exit for a request
 // that ends its connection (Connection: close, an error, a refused write),
 // right after the goroutine has declared the conn closed (asyncClosed) and
-// before it has marked itself gone (endDispatch). testHookDrainAsyncClosed
-// runs on the loop's goroutine in drainDetachQueue when an entry names a conn
-// declared closed whose goroutine is still counted as running. Together they
-// let a test hold the goroutine in the one place the loop can see that
-// combination (celeris#1068). Atomic pointers, as testHookCloseWhenFlushed is;
-// a load and a nil test per close, never a hot path.
+// before it has marked itself gone (endDispatch), with asyncInMu held for both:
+// the loop cannot read the two apart. testHookDrainAsyncClosed runs on the
+// loop's goroutine in drainDetachQueue when an entry names a conn declared
+// closed whose goroutine is still counted as running. A test that holds the
+// first and waits on the second sees the loop in the gap the lock closes
+// (celeris#1068). Atomic pointers, as testHookCloseWhenFlushed is; a load and a
+// nil test per close, never a hot path.
 var testHookAsyncExitClosed atomic.Pointer[func(*connState)]
 var testHookDrainAsyncClosed atomic.Pointer[func(*connState)]
 
@@ -3315,8 +3316,10 @@ func (l *Loop) serveAsync(cs *connState, holds *dispatchHolds) {
 			holds.detachMu = false
 			cs.detachMu.Unlock()
 			if promoteErr != nil {
-				cs.asyncClosed.Store(true)
+				// asyncClosed under asyncInMu, with endDispatch (see the exit
+				// at the end of this loop, celeris#1068).
 				cs.asyncInMu.Lock()
+				cs.asyncClosed.Store(true)
 				cs.asyncInBuf = cs.asyncInBuf[:0]
 				cs.endDispatch() // enqueued below: that is the hand-back
 				cs.asyncInMu.Unlock()
@@ -3342,8 +3345,10 @@ func (l *Loop) serveAsync(cs *connState, holds *dispatchHolds) {
 			// considers a hijacked conn "done with the request". Treat
 			// it like nil here — the middleware now owns the conn.
 			if processErr != nil && !errors.Is(processErr, conn.ErrHijacked) {
-				cs.asyncClosed.Store(true)
+				// asyncClosed under asyncInMu, with endDispatch (see the exit
+				// at the end of this loop, celeris#1068).
 				cs.asyncInMu.Lock()
+				cs.asyncClosed.Store(true)
 				cs.asyncInBuf = cs.asyncInBuf[:0]
 				cs.endDispatch() // enqueued below: that is the hand-back
 				cs.asyncInMu.Unlock()
@@ -3410,11 +3415,21 @@ func (l *Loop) serveAsync(cs *connState, holds *dispatchHolds) {
 			// on detachQueue, and write eventfd; the worker's
 			// drainDetachQueue pass will observe asyncClosed and run
 			// closeConn thread-safely.
+			//
+			// asyncClosed is stored under asyncInMu, with endDispatch: the loop
+			// reads asyncClosed and asyncRun together under that lock
+			// (drainDetachQueue), and a conn declared closed whose goroutine
+			// still counts as running is, to it, a close requested from outside
+			// while a handler runs, which closeConn carries out at once when
+			// detachMu is free, as it is here (the flush above is done). The
+			// entry enqueued for a partial flush's remainder could make it
+			// look in the gap between the two stores, and the conn was closed
+			// with that remainder unsent (celeris#1068).
+			cs.asyncInMu.Lock()
 			cs.asyncClosed.Store(true)
 			if h := testHookAsyncExitClosed.Load(); h != nil {
 				(*h)(cs)
 			}
-			cs.asyncInMu.Lock()
 			cs.asyncInBuf = cs.asyncInBuf[:0]
 			cs.endDispatch() // enqueued below: that is the hand-back
 			cs.asyncInMu.Unlock()

From 041616966d260f313726b1d52a1bf12dfbec0914 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 15:47:14 +0100
Subject: [PATCH 3/9] test(iouring): TestRunSurvivesARingWaitEBUSY884's
 timed-wait arm paces its clients so the loop reaches that wait (Refs #890)

The arm injects EBUSY into the loop's timed wait (mode 3b), which the loop
takes only when it finds no completion ready and no send to wait for. 32
keep-alive clients that send the next request the moment they are answered keep
that from happening for the whole 3 s window, and the test's apparatus guard
("the injected EBUSY never reached a submit-and-wait-timeout wait") failed:
9 times in the Coverage workflow's -race -covermode=atomic build since the
test landed (3 of them on main), never in the Unit job. On a worker with a CPU
to itself the arm ran 79 000 requests and met the timed wait 0 times
(--cpus 4, the CI runner's 4 vCPUs: 31 of 40 runs failed).

The arm's clients now pause 20 ms after each answer (runKeepAliveLoadPaced; the
unpaced runKeepAliveLoad is unchanged), which leaves the worker idle between
requests, where the timed wait is the one it takes. The other two arms keep
the unpaced load, which reaches their waits at every pass. What the test
asserts is unchanged.
---
 .../engine/iouring/fd_lifetime_engine_test.go | 18 +++++++++++++++
 .../iouring/run_ebusy_884_linux_test.go       | 23 +++++++++++++------
 2 files changed, 34 insertions(+), 7 deletions(-)

diff --git a/internal/engine/iouring/fd_lifetime_engine_test.go b/internal/engine/iouring/fd_lifetime_engine_test.go
index 8df27b8e3..a4fdc2aae 100644
--- a/internal/engine/iouring/fd_lifetime_engine_test.go
+++ b/internal/engine/iouring/fd_lifetime_engine_test.go
@@ -163,6 +163,17 @@ type loadResult struct {
 // the server never answers surfaces as that client's read timeout; the client
 // then stops, so each lost request is exactly one error.
 func runKeepAliveLoad(t *testing.T, addr string, n int, stop <-chan struct{}) func() loadResult {
+	t.Helper()
+	return runKeepAliveLoadPaced(t, addr, n, stop, 0)
+}
+
+// runKeepAliveLoadPaced is runKeepAliveLoad with think between a client's
+// answered request and its next (0: none, back to back). A worker serving
+// clients that never wait finds a completion ready at every pass of its loop
+// and takes no wait at all, or only the one that has sends to wait for; with
+// think the clients leave it idle between requests, and the wait it takes
+// then is the timed one (celeris#884 item 49, the test of that wait).
+func runKeepAliveLoadPaced(t *testing.T, addr string, n int, stop <-chan struct{}, think time.Duration) func() loadResult {
 	t.Helper()
 	var ok, errsN atomic.Int64
 	var mu sync.Mutex
@@ -216,6 +227,13 @@ func runKeepAliveLoad(t *testing.T, addr string, n int, stop <-chan struct{}) fu
 					return
 				}
 				ok.Add(1)
+				if think > 0 {
+					select {
+					case <-stop:
+						return
+					case <-time.After(think):
+					}
+				}
 			}
 		}()
 	}
diff --git a/internal/engine/iouring/run_ebusy_884_linux_test.go b/internal/engine/iouring/run_ebusy_884_linux_test.go
index ae2486af4..a6513d3d9 100644
--- a/internal/engine/iouring/run_ebusy_884_linux_test.go
+++ b/internal/engine/iouring/run_ebusy_884_linux_test.go
@@ -134,10 +134,10 @@ func setRingWaitSeam884(t *testing.T, fn func(*Worker, ringWaitMode, error) erro
 // warmLoad884 dials n keep-alive connections, each answered once, so every
 // worker that will own one does; the returned stop ends the load the
 // connections then carry and reports what the clients saw.
-func warmLoad884(t *testing.T, addr string, n int) (stop func() loadResult) {
+func warmLoad884(t *testing.T, addr string, n int, think time.Duration) (stop func() loadResult) {
 	t.Helper()
 	ch := make(chan struct{})
-	wait := runKeepAliveLoad(t, addr, n, ch)
+	wait := runKeepAliveLoadPaced(t, addr, n, ch, think)
 	return func() loadResult {
 		close(ch)
 		return wait()
@@ -151,10 +151,19 @@ func TestRunSurvivesARingWaitEBUSY884(t *testing.T) {
 	for _, tc := range []struct {
 		name string
 		mode int32 // -1 any
+		// think is the clients' pause between requests. The timed wait is
+		// taken only when the loop finds no completion ready and no send to
+		// wait for, which 32 clients that send the next request at once keep
+		// from happening for the whole window: on a worker with a CPU to
+		// itself it did not happen once in 3 s of 79 000 requests, and the
+		// apparatus guard below failed 10 times in Coverage's -race
+		// -covermode=atomic build (celeris#890). The other two modes are the
+		// ones that load always reaches.
+		think time.Duration
 	}{
-		{"any-wait", -1},
-		{"submit-and-wait-timeout", int32(waitModeTimed)},
-		{"submit-and-wait-sends-pending", int32(waitModeSubmitSends)},
+		{"any-wait", -1, 0},
+		{"submit-and-wait-timeout", int32(waitModeTimed), 20 * time.Millisecond},
+		{"submit-and-wait-sends-pending", int32(waitModeSubmitSends), 0},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			inj := &ebusyInject884{err: ebusyErr884(), n: 40}
@@ -162,7 +171,7 @@ func TestRunSurvivesARingWaitEBUSY884(t *testing.T) {
 			setRingWaitSeam884(t, inj.seam)
 			e, addr := startFDLEngine(t, transplantTestHandler{}, nil)
 
-			stop := warmLoad884(t, addr, 32)
+			stop := warmLoad884(t, addr, 32, tc.think)
 			time.Sleep(200 * time.Millisecond) // every connection answers, un-injected
 			inj.armed.Store(true)
 			deadline := time.Now().Add(3 * time.Second)
@@ -211,7 +220,7 @@ func TestRunStopsOnARingWaitFailure884(t *testing.T) {
 	setRingWaitSeam884(t, inj.seam)
 	e, addr := startFDLEngine(t, transplantTestHandler{}, nil)
 
-	stop := warmLoad884(t, addr, 32)
+	stop := warmLoad884(t, addr, 32, 0)
 	time.Sleep(200 * time.Millisecond)
 	inj.armed.Store(true)
 	deadline := time.Now().Add(3 * time.Second)

From 07d136d820befebb2d2d9818cb1503b432453860 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 15:47:14 +0100
Subject: [PATCH 4/9] test(epoll, iouring): the port 0 hold tests name every
 socket they find bound to the port (diagnostic only; Refs #1104)

TestPort0HoldIsReleasedWhenListenFails895 failed once on main's CI ("1 sockets
still bound to port 37483 after Listen failed") and passed on the re-run of the
same tree; the failure message does not say whose socket it counted, and the
two explanations in the issue (the engine's own descriptor released late, or a
socket of something else in the test binary) differ exactly in that. The
census now keeps, for every port, each socket's descriptor, whether it
listens, is connected (and to whom) or is only bound, and its SO_REUSEPORT,
and the hold tests (the Listen-failed one and the Listen-returned one, in both
engines) print them with the count.

No assertion changes. This is not a fix: the failure has not been reproduced
(0 of 35 000 runs of the test alone under load), so the next occurrence says
what the extra socket is.
---
 .../epoll/listen_port0_895_linux_test.go      | 32 ++++++++++++++++++-
 .../epoll/listen_port0_hold_895_linux_test.go |  8 ++---
 .../iouring/listen_port0_895_linux_test.go    | 32 ++++++++++++++++++-
 .../listen_port0_hold_895_linux_test.go       |  8 ++---
 4 files changed, 70 insertions(+), 10 deletions(-)

diff --git a/internal/engine/epoll/listen_port0_895_linux_test.go b/internal/engine/epoll/listen_port0_895_linux_test.go
index d91821432..1fb323055 100644
--- a/internal/engine/epoll/listen_port0_895_linux_test.go
+++ b/internal/engine/epoll/listen_port0_895_linux_test.go
@@ -5,6 +5,7 @@ package epoll
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"net"
@@ -37,6 +38,19 @@ import (
 // local address has a port are counted.
 type sockets895 struct {
 	bound, listening map[int]int
+	// owners says, for every port, what each socket counted in bound is: its
+	// descriptor, whether it listens, is connected (and to whom) or is only
+	// bound, and its SO_REUSEPORT. A census that finds a socket nobody expects
+	// (celeris#1104) then names it in the failure.
+	owners map[int][]string
+}
+
+// owned895 is the owners of port, for a failure message.
+func (s sockets895) owned895(port int) string {
+	if len(s.owners[port]) == 0 {
+		return "none"
+	}
+	return strings.Join(s.owners[port], "; ")
 }
 
 func census895(t *testing.T) sockets895 {
@@ -45,7 +59,7 @@ func census895(t *testing.T) sockets895 {
 	if err != nil {
 		t.Fatalf("read /proc/self/fd: %v", err)
 	}
-	s := sockets895{bound: map[int]int{}, listening: map[int]int{}}
+	s := sockets895{bound: map[int]int{}, listening: map[int]int{}, owners: map[int][]string{}}
 	for _, ent := range ents {
 		fd, err := strconv.Atoi(ent.Name())
 		if err != nil {
@@ -73,13 +87,29 @@ func census895(t *testing.T) sockets895 {
 			continue
 		}
 		s.bound[port]++
+		state := "bound only"
 		if v, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_ACCEPTCONN); err == nil && v == 1 {
 			s.listening[port]++
+			state = "listening"
+		} else if peer, err := unix.Getpeername(fd); err == nil {
+			state = fmt.Sprintf("connected to %s", sockaddrString895(peer))
 		}
+		reuse, _ := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT)
+		s.owners[port] = append(s.owners[port], fmt.Sprintf("fd %d %s (SO_REUSEPORT=%d)", fd, state, reuse))
 	}
 	return s
 }
 
+func sockaddrString895(sa unix.Sockaddr) string {
+	switch a := sa.(type) {
+	case *unix.SockaddrInet4:
+		return net.JoinHostPort(net.IP(a.Addr[:]).String(), strconv.Itoa(a.Port))
+	case *unix.SockaddrInet6:
+		return net.JoinHostPort(net.IP(a.Addr[:]).String(), strconv.Itoa(a.Port))
+	}
+	return fmt.Sprintf("%T", sa)
+}
+
 func ports895(m map[int]int) []int {
 	var ps []int
 	for p := range m {
diff --git a/internal/engine/epoll/listen_port0_hold_895_linux_test.go b/internal/engine/epoll/listen_port0_hold_895_linux_test.go
index 403bf7a5b..cc53f2914 100644
--- a/internal/engine/epoll/listen_port0_hold_895_linux_test.go
+++ b/internal/engine/epoll/listen_port0_hold_895_linux_test.go
@@ -89,8 +89,8 @@ func TestPort0IsHeldBetweenBeingChosenAndBeingBound895(t *testing.T) {
 	if err := r.stop(t); err != nil {
 		t.Fatalf("Listen: %v", err)
 	}
-	if n := census895(t).bound[port]; n != 0 {
-		t.Fatalf("%d sockets still bound to port %d after Listen returned, want 0 (the hold included)", n, port)
+	if c := census895(t); c.bound[port] != 0 {
+		t.Fatalf("%d sockets still bound to port %d after Listen returned, want 0 (the hold included): %s", c.bound[port], port, c.owned895(port))
 	}
 }
 
@@ -122,8 +122,8 @@ func TestPort0HoldIsReleasedWhenListenFails895(t *testing.T) {
 		t.Fatal("the hook never ran")
 	}
 	port := portOfString895(t, addr)
-	if n := census895(t).bound[port]; n != 0 {
-		t.Fatalf("%d sockets still bound to port %d after Listen failed, want 0", n, port)
+	if c := census895(t); c.bound[port] != 0 {
+		t.Fatalf("%d sockets still bound to port %d after Listen failed, want 0: %s", c.bound[port], port, c.owned895(port))
 	}
 	ln, err := net.Listen("tcp", addr)
 	if err != nil {
diff --git a/internal/engine/iouring/listen_port0_895_linux_test.go b/internal/engine/iouring/listen_port0_895_linux_test.go
index 31110c85b..38cce788f 100644
--- a/internal/engine/iouring/listen_port0_895_linux_test.go
+++ b/internal/engine/iouring/listen_port0_895_linux_test.go
@@ -5,6 +5,7 @@ package iouring
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"net"
@@ -43,6 +44,19 @@ import (
 // local address has a port are counted.
 type sockets895 struct {
 	bound, listening map[int]int
+	// owners says, for every port, what each socket counted in bound is: its
+	// descriptor, whether it listens, is connected (and to whom) or is only
+	// bound, and its SO_REUSEPORT. A census that finds a socket nobody expects
+	// (celeris#1104) then names it in the failure.
+	owners map[int][]string
+}
+
+// owned895 is the owners of port, for a failure message.
+func (s sockets895) owned895(port int) string {
+	if len(s.owners[port]) == 0 {
+		return "none"
+	}
+	return strings.Join(s.owners[port], "; ")
 }
 
 func census895(t *testing.T) sockets895 {
@@ -51,7 +65,7 @@ func census895(t *testing.T) sockets895 {
 	if err != nil {
 		t.Fatalf("read /proc/self/fd: %v", err)
 	}
-	s := sockets895{bound: map[int]int{}, listening: map[int]int{}}
+	s := sockets895{bound: map[int]int{}, listening: map[int]int{}, owners: map[int][]string{}}
 	for _, ent := range ents {
 		fd, err := strconv.Atoi(ent.Name())
 		if err != nil {
@@ -79,13 +93,29 @@ func census895(t *testing.T) sockets895 {
 			continue
 		}
 		s.bound[port]++
+		state := "bound only"
 		if v, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_ACCEPTCONN); err == nil && v == 1 {
 			s.listening[port]++
+			state = "listening"
+		} else if peer, err := unix.Getpeername(fd); err == nil {
+			state = fmt.Sprintf("connected to %s", sockaddrString895(peer))
 		}
+		reuse, _ := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT)
+		s.owners[port] = append(s.owners[port], fmt.Sprintf("fd %d %s (SO_REUSEPORT=%d)", fd, state, reuse))
 	}
 	return s
 }
 
+func sockaddrString895(sa unix.Sockaddr) string {
+	switch a := sa.(type) {
+	case *unix.SockaddrInet4:
+		return net.JoinHostPort(net.IP(a.Addr[:]).String(), strconv.Itoa(a.Port))
+	case *unix.SockaddrInet6:
+		return net.JoinHostPort(net.IP(a.Addr[:]).String(), strconv.Itoa(a.Port))
+	}
+	return fmt.Sprintf("%T", sa)
+}
+
 func ports895(m map[int]int) []int {
 	var ps []int
 	for p := range m {
diff --git a/internal/engine/iouring/listen_port0_hold_895_linux_test.go b/internal/engine/iouring/listen_port0_hold_895_linux_test.go
index 5c787181d..f38eb2d9a 100644
--- a/internal/engine/iouring/listen_port0_hold_895_linux_test.go
+++ b/internal/engine/iouring/listen_port0_hold_895_linux_test.go
@@ -107,8 +107,8 @@ func TestPort0IsHeldBetweenBeingChosenAndBeingBound895(t *testing.T) {
 	if err := r.stop(t); err != nil {
 		t.Fatalf("Listen: %v", err)
 	}
-	if n := census895(t).bound[port]; n != 0 {
-		t.Fatalf("%d sockets still bound to port %d after Listen returned, want 0 (the hold included)", n, port)
+	if c := census895(t); c.bound[port] != 0 {
+		t.Fatalf("%d sockets still bound to port %d after Listen returned, want 0 (the hold included): %s", c.bound[port], port, c.owned895(port))
 	}
 }
 
@@ -144,8 +144,8 @@ func TestPort0HoldIsReleasedWhenListenFails895(t *testing.T) {
 		t.Fatal("the hook never ran")
 	}
 	port := portOfString895(t, addr)
-	if n := census895(t).bound[port]; n != 0 {
-		t.Fatalf("%d sockets still bound to port %d after Listen failed, want 0", n, port)
+	if c := census895(t); c.bound[port] != 0 {
+		t.Fatalf("%d sockets still bound to port %d after Listen failed, want 0: %s", c.bound[port], port, c.owned895(port))
 	}
 	// bindWithin656, not an immediate bind: the sockets of a worker that
 	// was cancelled are let go of asynchronously on a main without the

From b2605e4bc0fbb187c896f26dc1dd2ebafaf3641a Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 16:49:21 +0100
Subject: [PATCH 5/9] test(epoll): start909 returns before the engine has
 logged its listening line (failing first; Refs #890)

TestNICIRQReportAddsNoLoop1036 failed in CI ("default: the listening line does
not say loops=2", a log with the pinning lines and no listening line) on
PR #1105's Coverage run and on this PR's Unit run. Listen publishes the address
(Engine.Addr) and only then logs "epoll engine listening", and start909, which
the NIC interrupt, loop count and pinning tests start their engines with,
returned as soon as the address was there, so a test that read the log on
return could read it before the line was written.

The test holds the line's write until it has looked at whether start909 has
returned: it has, every time, on this tree.
---
 .../epoll/pin_allowed_909_linux_test.go       | 58 ++++++++++++++++++-
 1 file changed, 57 insertions(+), 1 deletion(-)

diff --git a/internal/engine/epoll/pin_allowed_909_linux_test.go b/internal/engine/epoll/pin_allowed_909_linux_test.go
index 11b4350c0..d1954a5e2 100644
--- a/internal/engine/epoll/pin_allowed_909_linux_test.go
+++ b/internal/engine/epoll/pin_allowed_909_linux_test.go
@@ -38,9 +38,15 @@ func (l *lockedBuf) String() string {
 	return l.b.String()
 }
 
+// logSink is where start909's engine logs and its caller reads back.
+type logSink interface {
+	Write(p []byte) (int, error)
+	String() string
+}
+
 // start909 starts an epoll engine with the given worker count, logging into
 // log, and returns it with its stop function.
-func start909(t *testing.T, workers int, log *lockedBuf) (*Engine, func()) {
+func start909(t *testing.T, workers int, log logSink) (*Engine, func()) {
 	t.Helper()
 	ln, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
@@ -190,3 +196,53 @@ func TestFailedPinIsLoggedOnceAndNotPretended909(t *testing.T) {
 	}
 	_ = conn.Close()
 }
+
+// gatedLog holds the write of the engine's listening line until gate is
+// closed.
+type gatedLog struct {
+	lockedBuf
+	gate chan struct{}
+}
+
+func (g *gatedLog) Write(p []byte) (int, error) {
+	if bytes.Contains(p, []byte("epoll engine listening")) {
+		<-g.gate
+	}
+	return g.lockedBuf.Write(p)
+}
+
+// celeris#890 (TestNICIRQReportAddsNoLoop1036 failed in CI with a log that had
+// the pinning lines and no "epoll engine listening" line): Listen publishes
+// the address and only then logs that line, and start909 returned as soon as
+// the address was there, so a test that read the log on return could read it
+// before the line was written. start909 returns once the line is in the log.
+func TestStart909ReturnsOnlyOnceTheListeningLineIsLogged(t *testing.T) {
+	g := &gatedLog{gate: make(chan struct{})}
+	type started struct {
+		eng  *Engine
+		stop func()
+	}
+	ret := make(chan started, 1)
+	go func() {
+		eng, stop := start909(t, 2, g)
+		ret <- started{eng, stop}
+	}()
+	select {
+	case r := <-ret:
+		logged := g.String()
+		close(g.gate)
+		r.stop()
+		t.Fatalf("start909 returned (address %v) before the listening line was in the log: %q", r.eng.Addr(), logged)
+	case <-time.After(300 * time.Millisecond):
+	}
+	close(g.gate)
+	select {
+	case r := <-ret:
+		defer r.stop()
+		if !strings.Contains(g.String(), "epoll engine listening") {
+			t.Fatalf("start909 returned without the listening line in the log: %q", g.String())
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("start909 did not return once the listening line was logged")
+	}
+}

From 02cc5887882225b43f2bd9640d56efabc9902909 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 16:50:16 +0100
Subject: [PATCH 6/9] test(epoll): start909 returns once the engine has logged
 its listening line (Refs #890)

Waits, within the 10 s it already allows the address, for "epoll engine
listening" to be in the log. The NIC interrupt, loop count and pinning tests
read the log on return; they no longer race the line Listen logs after it
publishes the address. A test helper only; the engine is unchanged.
---
 internal/engine/epoll/pin_allowed_909_linux_test.go | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/internal/engine/epoll/pin_allowed_909_linux_test.go b/internal/engine/epoll/pin_allowed_909_linux_test.go
index d1954a5e2..0cb2f96bd 100644
--- a/internal/engine/epoll/pin_allowed_909_linux_test.go
+++ b/internal/engine/epoll/pin_allowed_909_linux_test.go
@@ -79,6 +79,15 @@ func start909(t *testing.T, workers int, log logSink) (*Engine, func()) {
 		}
 		time.Sleep(time.Millisecond)
 	}
+	// Listen publishes the address and only then logs its listening line
+	// (celeris#890): a caller that reads the log on return must find it.
+	for deadline := time.Now().Add(10 * time.Second); !strings.Contains(log.String(), "epoll engine listening"); {
+		if time.Now().After(deadline) {
+			cancel()
+			t.Fatal("the epoll engine did not log its listening line within 10s")
+		}
+		time.Sleep(time.Millisecond)
+	}
 	return eng, func() {
 		cancel()
 		select {

From 32cb57e1ac6b586a0bf7ed1d5f1431a6b4e8826d Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 18:26:43 +0100
Subject: [PATCH 7/9] fix(epoll): the loop leaves the close of a still-running
 dispatch goroutine to it; test hygiene from the review (Refs #1068, Refs
 #1107, Refs #1108)

abortAsyncHandler stores asyncClosed first, lets go of detachMu, formats and
logs the panic with its stack, and only then calls endDispatch. A pipelining
client whose first response was partly sent (the remainder queued, an entry
for it in the detach queue) and whose second request's handler panics, or
calls runtime.Goexit, past every recovery, had that first response cut when
the loop drained an entry in the window: a closed conn whose goroutine is
running, and detachMu free, so closeConn went through. The same defect class
as the serveAsync exits.

drainDetachQueue now asks dispatchBusy before closeConn for a closed conn
whose goroutine still runs: a goroutine that is running a request (not parked,
not detached) owes the loop a hand-back, and dispatchBusy records the close
as owed (closeOwed), as closeConn does for a handler that holds detachMu. The
goroutine's exit enqueues the conn again, and that entry finds it gone and
closes through closeWhenFlushed. Detached conns (WS/SSE) still close at once.
The lock moves of the serveAsync exits stay: they keep a goroutine's own
closing exit from showing as running at all.

- test: TestAbortedGoroutineDoesNotLoseTheQueuedRemainder1068 (new hook
  testHookAbortAsyncLogged). Without the dispatchBusy line: 40 of 40 FAIL,
  received 2383751 of 5242880; with it: 60 of 60 PASS.
- test: the hooks of both tests assert that bytes were still queued while the
  goroutine was held, so an arm cannot pass on a host whose socket buffers take
  the whole response.
- test: start909's engine start no longer calls t.Fatal from the helper's
  goroutine in TestStart909ReturnsOnlyOnceTheListeningLineIsLogged
  (launch909 returns errors).
- test: the EBUSY arm's comment counts 9 failures, as the PR does.
---
 .../async_close_flush_1068_linux_test.go      | 160 +++++++++++++++++-
 internal/engine/epoll/loop.go                 |  27 ++-
 .../epoll/pin_allowed_909_linux_test.go       |  64 +++++--
 .../iouring/run_ebusy_884_linux_test.go       |   2 +-
 4 files changed, 230 insertions(+), 23 deletions(-)

diff --git a/internal/engine/epoll/async_close_flush_1068_linux_test.go b/internal/engine/epoll/async_close_flush_1068_linux_test.go
index 6522c375e..98cb455fc 100644
--- a/internal/engine/epoll/async_close_flush_1068_linux_test.go
+++ b/internal/engine/epoll/async_close_flush_1068_linux_test.go
@@ -4,15 +4,19 @@ package epoll
 
 import (
 	"bufio"
+	"context"
 	"errors"
 	"io"
 	"net"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/goceleris/celeris/internal/protocol/h2/stream"
 	"github.com/goceleris/celeris/internal/resource"
 )
 
@@ -66,7 +70,9 @@ func TestAsyncRouteCloseSendsTheQueuedRemainder1068(t *testing.T) {
 				held     = make(chan struct{})
 				heldOnce sync.Once
 			)
-			exitHook := func(*connState) {
+			var queued atomic.Int64
+			exitHook := func(cs *connState) {
+				queued.Store(int64(queued1068(cs)))
 				select {
 				case <-looked:
 				case <-time.After(500 * time.Millisecond):
@@ -98,6 +104,12 @@ func TestAsyncRouteCloseSendsTheQueuedRemainder1068(t *testing.T) {
 			case <-time.After(10 * time.Second):
 				t.Fatal("the dispatch goroutine never reached the exit of a request that ends its conn")
 			}
+			// The arm proves nothing unless the remainder was queued while the
+			// goroutine was held (an 8 MiB response must outgrow the socket
+			// buffers of the host it runs on).
+			if q := queued.Load(); q <= 0 {
+				t.Fatalf("%d bytes were queued while the goroutine was held (-1: not readable): the arm would prove nothing", q)
+			}
 			_ = c.SetReadDeadline(time.Now().Add(20 * time.Second))
 			br := bufio.NewReaderSize(c, 64<<10)
 			whole := 0
@@ -135,3 +147,149 @@ func isClosed1068(ch chan struct{}) bool {
 		return false
 	}
 }
+
+// queued1068 is the bytes cs has queued for its socket, read under detachMu
+// with TryLock only: the caller may hold asyncInMu, and the loop may hold
+// detachMu for a flush. -1 if the lock stayed busy for 100 ms.
+func queued1068(cs *connState) int {
+	for dl := time.Now().Add(100 * time.Millisecond); time.Now().Before(dl); time.Sleep(time.Millisecond) {
+		if cs.detachMu.TryLock() {
+			n := csPendingBytes(cs)
+			cs.detachMu.Unlock()
+			return n
+		}
+	}
+	return -1
+}
+
+// abortHandler1068: /big waits for proceed, then answers with a body larger
+// than the socket buffers; /die panics, past every recovery (this handler is
+// the stream.Handler itself).
+type abortHandler1068 struct {
+	big     []byte
+	entered chan struct{}
+	proceed chan struct{}
+	once    sync.Once
+}
+
+func (h *abortHandler1068) HandleStream(_ context.Context, s *stream.Stream) error {
+	if s.ResponseWriter == nil {
+		return nil
+	}
+	if s.Path == "/die" {
+		panic("celeris1068: handler panic")
+	}
+	h.once.Do(func() { close(h.entered) })
+	<-h.proceed
+	return s.ResponseWriter.WriteResponse(s, 200,
+		[][2]string{{"content-type", "application/octet-stream"}, {"content-length", strconv.Itoa(len(h.big))}}, h.big)
+}
+func (*abortHandler1068) RouteAsync(_, _ string) bool { return true }
+func (*abortHandler1068) HasAsyncRoutes() bool        { return true }
+
+var _ stream.AsyncRouteResolver = (*abortHandler1068)(nil)
+
+// TestAbortedGoroutineDoesNotLoseTheQueuedRemainder1068: the same gap as above
+// on the other way a dispatch goroutine ends a conn. abortAsyncHandler stores
+// asyncClosed first, then lets go of detachMu and logs the panic with its
+// stack, and only then calls endDispatch. A pipelining client's first response
+// is partly sent and its remainder queued (the entry the goroutine enqueued
+// for it is still in the detach queue) when the second request's handler
+// panics; the loop that drains that entry in the window found a closed conn
+// whose goroutine was running and closed it over the remainder.
+//
+// The second request is on the wire while the first handler still runs, so
+// the goroutine takes it straight after the first flush without parking. The
+// test holds the abort after its log and waits for the loop to look.
+func TestAbortedGoroutineDoesNotLoseTheQueuedRemainder1068(t *testing.T) {
+	// What stays queued (the body less what the socket takes, about 2.4 MB
+	// here) must be under the 4 MiB cap, or the second request is refused
+	// before its handler runs (ErrWriteBacklog) and the abort is never reached;
+	// and more than the kernel takes from the loop while it decides.
+	const body = 5 << 20
+	var (
+		looked   = make(chan struct{})
+		lookOnce sync.Once
+		held     = make(chan struct{})
+		heldOnce sync.Once
+		queued   atomic.Int64
+	)
+	// The loop's look counts only once the goroutine is held: an earlier
+	// drain of the entry for the remainder finds the goroutine still holding
+	// detachMu (it has not released it before it logs), and leaves the close
+	// to it, as it always did.
+	var inWindow atomic.Bool
+	drainHook := func(*connState) {
+		if inWindow.Load() {
+			lookOnce.Do(func() { close(looked) })
+		}
+	}
+	abortHook := func(l *Loop, cs *connState) {
+		queued.Store(int64(queued1068(cs)))
+		inWindow.Store(true)
+		// An entry naming cs reaches the loop while the goroutine is in the
+		// window. (The one enqueued for the remainder is normally drained
+		// before the handler of the second request has panicked; one that is
+		// drained late, or another, is all the loop needs to look.)
+		l.enqueueDetach(cs)
+		select {
+		case <-looked:
+		case <-time.After(500 * time.Millisecond):
+		}
+		heldOnce.Do(func() { close(held) })
+	}
+	testHookAbortAsyncLogged.Store(&abortHook)
+	testHookDrainAsyncClosed.Store(&drainHook)
+	t.Cleanup(func() {
+		testHookAbortAsyncLogged.Store(nil)
+		testHookDrainAsyncClosed.Store(nil)
+	})
+
+	h := &abortHandler1068{big: newPathHandler885(body).big, entered: make(chan struct{}), proceed: make(chan struct{})}
+	e, addr := startE2E885(t, true, h, func(*resource.Config) {})
+	// A small receive buffer keeps the kernel from taking the remainder on
+	// its own (the send buffer grows with the peer's window) while the loop
+	// is deciding.
+	c := dialRcvBuf885(t, addr, 16<<10)
+	req1 := "GET /big HTTP/1.1\r\nHost: x\r\n\r\n"
+	req2 := "GET /die HTTP/1.1\r\nHost: x\r\n\r\n"
+	if _, err := io.WriteString(c, req1); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-h.entered:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the first handler never ran")
+	}
+	if _, err := io.WriteString(c, req2); err != nil {
+		t.Fatal(err)
+	}
+	// The loop has read the second request (into the goroutine's input) once
+	// the engine has counted both.
+	for dl := time.Now().Add(10 * time.Second); e.Metrics().BytesRead < uint64(len(req1)+len(req2)); time.Sleep(time.Millisecond) {
+		if time.Now().After(dl) {
+			t.Fatalf("the engine read %d of %d request bytes", e.Metrics().BytesRead, len(req1)+len(req2))
+		}
+	}
+	close(h.proceed)
+	select {
+	case <-held:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the dispatch goroutine never aborted")
+	}
+	if q := queued.Load(); q <= 0 {
+		t.Fatalf("%d bytes were queued while the aborted goroutine was held (-1: not readable): the test would prove nothing", q)
+	}
+	_ = c.SetReadDeadline(time.Now().Add(20 * time.Second))
+	br := bufio.NewReaderSize(c, 64<<10)
+	resp, err := http.ReadResponse(br, nil)
+	if err != nil {
+		t.Fatalf("response 1: %v (looked=%v)", err, isClosed1068(looked))
+	}
+	n, err := io.Copy(io.Discard, resp.Body)
+	_ = resp.Body.Close()
+	if err != nil || n != body {
+		t.Fatalf("response 1: received %d of %d body bytes (%v): the loop closed the conn over the remainder while the aborted goroutine was still on its way out (looked=%v)",
+			n, body, err, isClosed1068(looked))
+	}
+}
diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index 5f2340168..52f99a6ec 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -1957,13 +1957,19 @@ var testHookCloseWhenFlushed atomic.Pointer[func(*connState)]
 // before it has marked itself gone (endDispatch), with asyncInMu held for both:
 // the loop cannot read the two apart. testHookDrainAsyncClosed runs on the
 // loop's goroutine in drainDetachQueue when an entry names a conn declared
-// closed whose goroutine is still counted as running. A test that holds the
-// first and waits on the second sees the loop in the gap the lock closes
-// (celeris#1068). Atomic pointers, as testHookCloseWhenFlushed is; a load and a
+// closed whose goroutine is still counted as running, before the loop decides
+// whether to close it. A test that holds the first and waits on the second
+// sees the loop in the gap the lock closes (celeris#1068). Atomic pointers, as testHookCloseWhenFlushed is; a load and a
 // nil test per close, never a hot path.
 var testHookAsyncExitClosed atomic.Pointer[func(*connState)]
 var testHookDrainAsyncClosed atomic.Pointer[func(*connState)]
 
+// testHookAbortAsyncLogged, when set, runs in abortAsyncHandler after its log
+// record and before endDispatch, holding no lock: the one place an aborted
+// goroutine is closed and still counted as running (celeris#1068). It gets the
+// loop so that a test can queue an entry for the loop to drain in that place.
+var testHookAbortAsyncLogged atomic.Pointer[func(*Loop, *connState)]
+
 // discardWrite is the write function the run loop drains the H2 write queue of
 // a conn that is closing on a refused write into: the frames go nowhere.
 func discardWrite([]byte) {}
@@ -3150,6 +3156,9 @@ func (l *Loop) abortAsyncHandler(cs *connState, r any, holds *dispatchHolds) {
 			)
 		}
 	}
+	if h := testHookAbortAsyncLogged.Load(); h != nil {
+		(*h)(l, cs)
+	}
 	cs.asyncClosed.Store(true)
 	cs.asyncInMu.Lock()
 	cs.asyncInBuf = cs.asyncInBuf[:0]
@@ -3552,6 +3561,18 @@ func (l *Loop) drainDetachQueue() {
 				if h := testHookDrainAsyncClosed.Load(); h != nil {
 					(*h)(cs)
 				}
+				// A goroutine that is still running a request is not done
+				// with the conn: the close is its to hand back (closeOwed),
+				// as closeConn does for a handler that holds detachMu. A
+				// goroutine that declared the conn closed itself and is on
+				// its way out (abortAsyncHandler logs between its store of
+				// asyncClosed and endDispatch) would otherwise be closed
+				// over, and the response it left queued cut short
+				// (celeris#1068). Its exit always enqueues cs again, and
+				// that entry finds it gone.
+				if dispatchBusy(cs, &cs.closeOwed) {
+					continue
+				}
 				l.closeConn(cs.fd)
 			} else {
 				l.closeWhenFlushed(cs)
diff --git a/internal/engine/epoll/pin_allowed_909_linux_test.go b/internal/engine/epoll/pin_allowed_909_linux_test.go
index 0cb2f96bd..8ef77c338 100644
--- a/internal/engine/epoll/pin_allowed_909_linux_test.go
+++ b/internal/engine/epoll/pin_allowed_909_linux_test.go
@@ -5,6 +5,8 @@ package epoll
 import (
 	"bytes"
 	"context"
+	"errors"
+	"fmt"
 	"log/slog"
 	"net"
 	"strings"
@@ -48,9 +50,23 @@ type logSink interface {
 // log, and returns it with its stop function.
 func start909(t *testing.T, workers int, log logSink) (*Engine, func()) {
 	t.Helper()
+	eng, stop, err := launch909(workers, log)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return eng, func() {
+		if err := stop(); err != nil {
+			t.Error(err)
+		}
+	}
+}
+
+// launch909 is start909 without a *testing.T, so that a goroutine other than
+// the test's may start an engine and report why it could not.
+func launch909(workers int, log logSink) (*Engine, func() error, error) {
 	ln, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
-		t.Fatalf("pick port: %v", err)
+		return nil, nil, fmt.Errorf("pick port: %w", err)
 	}
 	addr := ln.Addr().String()
 	_ = ln.Close()
@@ -61,7 +77,7 @@ func start909(t *testing.T, workers int, log logSink) (*Engine, func()) {
 		Logger:    slog.New(slog.NewTextHandler(log, nil)),
 	}, respondingHandler{})
 	if err != nil {
-		t.Fatalf("New: %v", err)
+		return nil, nil, fmt.Errorf("New: %w", err)
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan error, 1)
@@ -70,12 +86,12 @@ func start909(t *testing.T, workers int, log logSink) (*Engine, func()) {
 		select {
 		case err := <-done:
 			cancel()
-			t.Fatalf("Listen returned before it bound: %v", err)
+			return nil, nil, fmt.Errorf("Listen returned before it bound: %w", err)
 		default:
 		}
 		if time.Now().After(deadline) {
 			cancel()
-			t.Fatal("the epoll engine did not bind within 10s")
+			return nil, nil, errors.New("the epoll engine did not bind within 10s")
 		}
 		time.Sleep(time.Millisecond)
 	}
@@ -84,21 +100,22 @@ func start909(t *testing.T, workers int, log logSink) (*Engine, func()) {
 	for deadline := time.Now().Add(10 * time.Second); !strings.Contains(log.String(), "epoll engine listening"); {
 		if time.Now().After(deadline) {
 			cancel()
-			t.Fatal("the epoll engine did not log its listening line within 10s")
+			return nil, nil, errors.New("the epoll engine did not log its listening line within 10s")
 		}
 		time.Sleep(time.Millisecond)
 	}
-	return eng, func() {
+	return eng, func() error {
 		cancel()
 		select {
 		case err := <-done:
 			if err != nil {
-				t.Errorf("Listen: %v", err)
+				return fmt.Errorf("Listen: %w", err)
 			}
+			return nil
 		case <-time.After(15 * time.Second):
-			t.Fatal("the epoll engine did not stop within 15s")
+			return errors.New("the epoll engine did not stop within 15s")
 		}
-	}
+	}, nil
 }
 
 // celeris#909: with the process restricted to a mask that leaves out its lowest
@@ -229,29 +246,40 @@ func TestStart909ReturnsOnlyOnceTheListeningLineIsLogged(t *testing.T) {
 	g := &gatedLog{gate: make(chan struct{})}
 	type started struct {
 		eng  *Engine
-		stop func()
+		stop func() error
+		err  error
 	}
 	ret := make(chan started, 1)
 	go func() {
-		eng, stop := start909(t, 2, g)
-		ret <- started{eng, stop}
+		eng, stop, err := launch909(2, g)
+		ret <- started{eng, stop, err}
 	}()
 	select {
 	case r := <-ret:
 		logged := g.String()
 		close(g.gate)
-		r.stop()
-		t.Fatalf("start909 returned (address %v) before the listening line was in the log: %q", r.eng.Addr(), logged)
+		if r.err != nil {
+			t.Fatalf("launch909: %v", r.err)
+		}
+		_ = r.stop()
+		t.Fatalf("launch909 returned (address %v) before the listening line was in the log: %q", r.eng.Addr(), logged)
 	case <-time.After(300 * time.Millisecond):
 	}
 	close(g.gate)
 	select {
 	case r := <-ret:
-		defer r.stop()
+		if r.err != nil {
+			t.Fatalf("launch909: %v", r.err)
+		}
+		defer func() {
+			if err := r.stop(); err != nil {
+				t.Error(err)
+			}
+		}()
 		if !strings.Contains(g.String(), "epoll engine listening") {
-			t.Fatalf("start909 returned without the listening line in the log: %q", g.String())
+			t.Fatalf("launch909 returned without the listening line in the log: %q", g.String())
 		}
-	case <-time.After(10 * time.Second):
-		t.Fatal("start909 did not return once the listening line was logged")
+	case <-time.After(20 * time.Second):
+		t.Fatal("launch909 did not return once the listening line was logged")
 	}
 }
diff --git a/internal/engine/iouring/run_ebusy_884_linux_test.go b/internal/engine/iouring/run_ebusy_884_linux_test.go
index a6513d3d9..801c2d573 100644
--- a/internal/engine/iouring/run_ebusy_884_linux_test.go
+++ b/internal/engine/iouring/run_ebusy_884_linux_test.go
@@ -156,7 +156,7 @@ func TestRunSurvivesARingWaitEBUSY884(t *testing.T) {
 		// wait for, which 32 clients that send the next request at once keep
 		// from happening for the whole window: on a worker with a CPU to
 		// itself it did not happen once in 3 s of 79 000 requests, and the
-		// apparatus guard below failed 10 times in Coverage's -race
+		// apparatus guard below failed 9 times in Coverage's -race
 		// -covermode=atomic build (celeris#890). The other two modes are the
 		// ones that load always reaches.
 		think time.Duration

From 66824483d694d7c9600d4822d98661eb5dc05666 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 19:09:50 +0100
Subject: [PATCH 8/9] fix(epoll): the drain decides "running", "busy" and the
 owed close in one asyncInMu section (Refs #1068, Refs #890, Refs #1107, Refs
 #1108, Refs #1104)

drainDetachQueue read asyncRun under asyncInMu and then asked dispatchBusy,
which took the lock again. An aborted goroutine that reached endDispatch
between the two reads was taken for gone by the second, closeConn (a direct
close, not closeWhenFlushed) cut the queued remainder, and the goroutine's own
entry found the conn closed.

The branch now reads asyncRun and the busy test (dispatchBusyLocked, the body
of dispatchBusy) and records closeOwed in one section; what it does with the
answer is unchanged. The second store of asyncClosed in abortAsyncHandler moves
under the lock with endDispatch, as at the serveAsync exits (the first store
stays early on purpose).

TestAbortedGoroutineFinishingBetweenTheDrainsReads1068 holds the loop, after
its decision, until the goroutine has finished: with the two reads put back it
fails 10 of 10 (received 2383751 of 5242880), with the guard removed both abort
tests fail 10 of 10, with the fix all pass (20 of 20 each, -race). The
abort scenario is a helper of the two tests.
---
 .../async_close_flush_1068_linux_test.go      | 39 +++++++++++++++++--
 internal/engine/epoll/loop.go                 | 38 ++++++++++++------
 2 files changed, 62 insertions(+), 15 deletions(-)

diff --git a/internal/engine/epoll/async_close_flush_1068_linux_test.go b/internal/engine/epoll/async_close_flush_1068_linux_test.go
index 98cb455fc..d778abbf4 100644
--- a/internal/engine/epoll/async_close_flush_1068_linux_test.go
+++ b/internal/engine/epoll/async_close_flush_1068_linux_test.go
@@ -202,6 +202,27 @@ var _ stream.AsyncRouteResolver = (*abortHandler1068)(nil)
 // the goroutine takes it straight after the first flush without parking. The
 // test holds the abort after its log and waits for the loop to look.
 func TestAbortedGoroutineDoesNotLoseTheQueuedRemainder1068(t *testing.T) {
+	abortedGoroutine1068(t, false)
+}
+
+// TestAbortedGoroutineFinishingBetweenTheDrainsReads1068: the loop decides
+// whether the goroutine still runs, and whether to leave the close to it, in
+// one asyncInMu section. Read in two (asyncRun, then dispatchBusy), a goroutine
+// that reached endDispatch between the reads was taken for gone by the second
+// and the conn was closed with closeConn, over the queued remainder, and the
+// goroutine's own entry found it closed. The drain hook here holds the loop
+// after its decision until the goroutine has finished.
+func TestAbortedGoroutineFinishingBetweenTheDrainsReads1068(t *testing.T) {
+	abortedGoroutine1068(t, true)
+}
+
+// abortedGoroutine1068 runs the scenario of the two tests above. The abort
+// hook queues an entry for the loop itself, to force the window: the entry
+// for the remainder is normally drained before the second request's handler
+// has panicked. waitGone holds the loop's drain, after its decision, until the
+// aborted goroutine has left (asyncRun false).
+func abortedGoroutine1068(t *testing.T, waitGone bool) {
+	t.Helper()
 	// What stays queued (the body less what the socket takes, about 2.4 MB
 	// here) must be under the 4 MiB cap, or the second request is refused
 	// before its handler runs (ErrWriteBacklog) and the abort is never reached;
@@ -219,9 +240,21 @@ func TestAbortedGoroutineDoesNotLoseTheQueuedRemainder1068(t *testing.T) {
 	// detachMu (it has not released it before it logs), and leaves the close
 	// to it, as it always did.
 	var inWindow atomic.Bool
-	drainHook := func(*connState) {
-		if inWindow.Load() {
-			lookOnce.Do(func() { close(looked) })
+	drainHook := func(cs *connState) {
+		if !inWindow.Load() {
+			return
+		}
+		lookOnce.Do(func() { close(looked) })
+		if !waitGone {
+			return
+		}
+		for dl := time.Now().Add(2 * time.Second); time.Now().Before(dl); time.Sleep(time.Millisecond) {
+			cs.asyncInMu.Lock()
+			running := cs.asyncRun
+			cs.asyncInMu.Unlock()
+			if !running {
+				return
+			}
 		}
 	}
 	abortHook := func(l *Loop, cs *connState) {
diff --git a/internal/engine/epoll/loop.go b/internal/engine/epoll/loop.go
index 52f99a6ec..5d24ab4f9 100644
--- a/internal/engine/epoll/loop.go
+++ b/internal/engine/epoll/loop.go
@@ -2266,11 +2266,19 @@ func dispatchBusy(cs *connState, owe *bool) bool {
 		return false // no async machinery: sync mode, no dispatch goroutine
 	}
 	cs.asyncInMu.Lock()
+	busy := dispatchBusyLocked(cs, owe)
+	cs.asyncInMu.Unlock()
+	return busy
+}
+
+// dispatchBusyLocked is dispatchBusy's test, for a caller that holds
+// cs.asyncInMu and needs it in the same section as another read of cs
+// (drainDetachQueue's asyncClosed branch).
+func dispatchBusyLocked(cs *connState, owe *bool) bool {
 	busy := cs.asyncRun && !cs.asyncParked && !cs.asyncDetachUnlocked
 	if busy && owe != nil {
 		*owe = true
 	}
-	cs.asyncInMu.Unlock()
 	return busy
 }
 
@@ -3159,8 +3167,11 @@ func (l *Loop) abortAsyncHandler(cs *connState, r any, holds *dispatchHolds) {
 	if h := testHookAbortAsyncLogged.Load(); h != nil {
 		(*h)(l, cs)
 	}
-	cs.asyncClosed.Store(true)
+	// Stored again, under the lock with endDispatch as at serveAsync's exits
+	// (the first store, above, is the one that matters here: it is early on
+	// purpose).
 	cs.asyncInMu.Lock()
+	cs.asyncClosed.Store(true)
 	cs.asyncInBuf = cs.asyncInBuf[:0]
 	cs.endDispatch() // enqueued below: that is the hand-back
 	cs.asyncInMu.Unlock()
@@ -3554,23 +3565,26 @@ func (l *Loop) drainDetachQueue() {
 		// than the socket buffers. While it still runs, closeConn leaves
 		// the close to it (celeris#669), as before.
 		if cs.asyncClosed.Load() {
+			// A goroutine that is still running a request is not done with
+			// the conn: the close is its to hand back (closeOwed), as
+			// closeConn does for a handler that holds detachMu. A goroutine
+			// that declared the conn closed itself and is on its way out
+			// (abortAsyncHandler logs between its store of asyncClosed and
+			// endDispatch) would otherwise be closed over, and the response
+			// it left queued cut short (celeris#1068). Its exit always
+			// enqueues cs again, and that entry finds it gone. Whether it
+			// runs, whether it is busy and the owed record are one section:
+			// read apart, a goroutine that finished between the reads was
+			// taken for gone by the second and closed over by closeConn.
 			cs.asyncInMu.Lock()
 			running := cs.asyncRun
+			busy := dispatchBusyLocked(cs, &cs.closeOwed)
 			cs.asyncInMu.Unlock()
 			if running {
 				if h := testHookDrainAsyncClosed.Load(); h != nil {
 					(*h)(cs)
 				}
-				// A goroutine that is still running a request is not done
-				// with the conn: the close is its to hand back (closeOwed),
-				// as closeConn does for a handler that holds detachMu. A
-				// goroutine that declared the conn closed itself and is on
-				// its way out (abortAsyncHandler logs between its store of
-				// asyncClosed and endDispatch) would otherwise be closed
-				// over, and the response it left queued cut short
-				// (celeris#1068). Its exit always enqueues cs again, and
-				// that entry finds it gone.
-				if dispatchBusy(cs, &cs.closeOwed) {
+				if busy {
 					continue
 				}
 				l.closeConn(cs.fd)

From cda1b485048aec021e65e9e2b2df380f52138a54 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Sun, 25 Oct 2026 19:42:22 +0100
Subject: [PATCH 9/9] test(epoll): the abort tests fail if the loop never
 looked while the goroutine was held (Refs #1068)

The abort hook waits only 500 ms for the loop's drain and nothing asserted
that it came, so a loop that did not look left a passing test that tested
nothing. Both abort tests now fail when the drain hook has not run by the time
the hold ends. With the two reads put back only the between-reads test fails
(10 of 10); with the drain guard removed both do (10 of 10); with the fix all
three 1068 tests pass 20 of 20 under -race.
---
 internal/engine/epoll/async_close_flush_1068_linux_test.go | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/internal/engine/epoll/async_close_flush_1068_linux_test.go b/internal/engine/epoll/async_close_flush_1068_linux_test.go
index d778abbf4..010bd4c4a 100644
--- a/internal/engine/epoll/async_close_flush_1068_linux_test.go
+++ b/internal/engine/epoll/async_close_flush_1068_linux_test.go
@@ -310,6 +310,11 @@ func abortedGoroutine1068(t *testing.T, waitGone bool) {
 	case <-time.After(10 * time.Second):
 		t.Fatal("the dispatch goroutine never aborted")
 	}
+	// The abort hook waits only so long for the loop; a loop that never looked
+	// leaves nothing tested.
+	if !isClosed1068(looked) {
+		t.Fatal("the loop never drained the closed conn while the aborted goroutine was held: the test would prove nothing")
+	}
 	if q := queued.Load(); q <= 0 {
 		t.Fatalf("%d bytes were queued while the aborted goroutine was held (-1: not readable): the test would prove nothing", q)
 	}