From c71d564bbf857c7f6ae4ee814cb466696134813f Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 28 Sep 2026 08:17:37 +0200
Subject: [PATCH 1/7] fix(epoll, iouring): send a response larger than the
 write cap whole, close on a refused write, keep pipelined responses in order
 (celeris#761, celeris#802)

The per-connection write back-pressure cap (4 MiB: epoll maxPendingBytes,
io_uring maxSendQueueBytes) was held against a single response. The
write hook that stages a zero-copy body counted the body itself, so an
HTTP/1.1 response whose headers and body passed the cap went out as its
headers only: the body was dropped without an error and the connection
left open, and a keep-alive client waited for the declared
Content-Length until its own timeout. The check after the handler closed
any connection whose backlog was over the cap, which cut off whatever
that path let through (a copied body, a sendfile body, an HTTP/2
stream's flow-control window), and every refused write was silent.

Now the cap bounds the backlog a write finds, not the write: a write is
refused only when the bytes still queued before it are over the cap (a
peer that stopped reading while it keeps sending requests), and a
refused write sets writeRefused, on which the site that ran the handler
closes the connection. An HTTP/2 connection gets a 64 MiB cap, as a
detached one has: its DATA is already bounded by the windows the peer
grants, and net/http's client keeps 4 MiB of frames queued per stream as
a matter of course.

epoll closes a connection once what it has queued has gone out
(closeWhenFlushed: EPOLLIN off, EPOLLOUT armed, the close deferred as
for EPOLLRDHUP), where it used to close at once after Connection: close,
a request error or a refused write, cutting off the tail of a response
larger than the socket buffers. The zero-copy body receive path and the
HTTP/2 write-queue flush now resync pendingBytes as every other flush
point does; left alone it grew by each response until the hooks refused
the writes of a connection with nothing queued (the 64th 64 KiB response
to split-body POSTs on one connection).

celeris#802: the write buffer is sent before a staged zero-copy body (and,
on epoll, a staged sendfile), so a response pipelined behind one went out
ahead of it. A write that follows staged output first moves it into the
write buffer (epoll unstage; io_uring copies bodyBuf), a copy paid only
by the next pipelined response.

Tests (large_response_linux_test.go, every engine): bodies of 4 MiB - 4
KiB, 4 MiB - 1, 4 MiB, 4 MiB + 1 and 64 MiB, sync, async-loop and
async-route handlers, keep-alive (then a next request on the connection)
and Connection: close; c.File of 4 MiB + 1 and 64 MiB; HTTP/2 bodies;
96 split-body POSTs on one connection; pipelined requests mixing large
and small bodies and files, compared in order; and a peer that pipelines
four 3 MiB requests without reading, which must get whole responses and
then the close. BenchmarkWriteHooks (both engines) measures the hooks.

Fixes #761
Fixes #802
---
 engine/epoll/backpressure_test.go             |  24 +-
 engine/epoll/conn.go                          |  32 +-
 engine/epoll/loop.go                          | 162 ++++-
 engine/epoll/write_hooks_bench_linux_test.go  |  42 ++
 engine/epoll/writer.go                        |  40 ++
 engine/iouring/conn.go                        |  22 +-
 engine/iouring/worker.go                      |  68 +-
 .../iouring/write_hooks_bench_linux_test.go   |  40 ++
 large_response_linux_test.go                  | 590 ++++++++++++++++++
 9 files changed, 969 insertions(+), 51 deletions(-)
 create mode 100644 engine/epoll/write_hooks_bench_linux_test.go
 create mode 100644 engine/iouring/write_hooks_bench_linux_test.go
 create mode 100644 large_response_linux_test.go

diff --git a/engine/epoll/backpressure_test.go b/engine/epoll/backpressure_test.go
index 7eec777e..2ae89adb 100644
--- a/engine/epoll/backpressure_test.go
+++ b/engine/epoll/backpressure_test.go
@@ -37,9 +37,12 @@ func (h *bigResponseHandler) HandleStream(_ context.Context, s *stream.Stream) e
 }
 
 // TestWriteBufBackpressureClosesSlowConsumer is a best-effort
-// end-to-end assertion of the maxPendingBytes / WriteTimeout close
-// path: a slow consumer must not cause unbounded server-side
-// buffering.
+// end-to-end assertion of the timeout close path: a slow consumer must
+// not hold server-side buffering for longer than the timeouts allow.
+// Since celeris#761 a single response larger than maxPendingBytes is
+// staged whole rather than closed at once (the cap bounds the backlog a
+// write finds, not the write), so it is the timeout sweep, here
+// ReadTimeout, that closes a consumer that never reads it.
 //
 // The test is marked t.Skip by default because reliably triggering
 // the backpressure path on loopback TCP is hard — Linux's
@@ -72,6 +75,7 @@ func TestWriteBufBackpressureClosesSlowConsumer(t *testing.T) {
 		Addr:         addr,
 		Protocol:     engine.HTTP1,
 		WriteTimeout: 200 * time.Millisecond,
+		ReadTimeout:  200 * time.Millisecond,
 		Resources: resource.Resources{
 			Workers: 2,
 		},
@@ -110,11 +114,9 @@ func TestWriteBufBackpressureClosesSlowConsumer(t *testing.T) {
 	}
 
 	// Send GET. Deliberately DO NOT read the response — we want the
-	// server's writeBuf to fill and either maxPendingBytes trip or
-	// WriteTimeout fire. lastActivity on the server does not
-	// advance after the GET (no further client→server data), and
-	// pendingBytes climbs past writeCap (4 MiB) as soon as the
-	// handler writes the 50 MiB body.
+	// server's writeBuf to fill and the timeout sweep to fire.
+	// lastActivity on the server does not advance after the GET (no
+	// further client→server data).
 	req := "GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"
 	if _, err := fmt.Fprint(tcp, req); err != nil {
 		t.Fatalf("write: %v", err)
@@ -122,9 +124,9 @@ func TestWriteBufBackpressureClosesSlowConsumer(t *testing.T) {
 
 	// Hold — don't read at all. The inline flush after the handler
 	// runs attempts unix.Write of the full 50 MiB body; kernel
-	// absorbs at most a few hundred KiB before EAGAIN. That leaves
-	// pendingBytes >> writeCap, which drainRead's post-flush check
-	// or checkTimeouts converts into closeConn.
+	// absorbs at most a few hundred KiB before EAGAIN. The rest waits
+	// on EPOLLOUT until checkTimeouts converts the stalled conn into
+	// closeConn.
 	time.Sleep(500 * time.Millisecond)
 
 	// Now the server should have closed us. Read should drain
diff --git a/engine/epoll/conn.go b/engine/epoll/conn.go
index 00244095..5832b6df 100644
--- a/engine/epoll/conn.go
+++ b/engine/epoll/conn.go
@@ -22,8 +22,17 @@ import (
 // echo payloads larger than 4 MiB (RFC 6455 allows frames up to 2^63,
 // Autobahn 9.1.6 sends 16 MiB). 64 MiB matches the WS default ReadLimit.
 const (
-	maxPendingBytes         = 4 << 20  // 4 MiB (H1/H2)
+	maxPendingBytes         = 4 << 20  // 4 MiB (H1)
 	maxPendingBytesDetached = 64 << 20 // 64 MiB (WS/SSE)
+	// maxPendingBytesH2 is the limit for an HTTP/2 connection. Its DATA is
+	// already bounded by the flow-control windows the peer grants, and a
+	// peer that reads keeps up to a window of frames queued behind the
+	// socket as a matter of course (net/http's client grants 4 MiB per
+	// stream, browsers more per connection), so the H1 limit refused, and
+	// closed, healthy HTTP/2 connections (celeris#761). 64 MiB, as for a
+	// detached connection, still bounds a peer that grants large windows
+	// and stops reading.
+	maxPendingBytesH2 = 64 << 20
 	// maxPendingInputBytes caps the async dispatch input buffer
 	// (cs.asyncInBuf) so a client pipelining requests faster than
 	// the dispatch goroutine drains them cannot balloon per-conn
@@ -61,7 +70,16 @@ func trimPooledBuf(b []byte) []byte {
 // for whether the connection is detached. Async-mode HTTP1 conns set
 // detachMu up front without being truly detached; they keep the H1/H2
 // limit so a stalled peer cannot balloon per-conn memory to 64 MiB.
+//
+// The limit bounds the backlog a write finds, not the write: a write is
+// refused only when the bytes still queued before it (cs.pendingBytes) are
+// over the limit, which is a peer that stopped reading while it keeps
+// sending requests. One response larger than the limit is not such a
+// backlog and is staged whole; before celeris#761 its body was dropped.
 func (cs *connState) writeCap() int {
+	if cs.h2State != nil {
+		return maxPendingBytesH2
+	}
 	if cs.detachMu != nil && cs.h1State != nil && cs.h1State.Detached.Load() {
 		return maxPendingBytesDetached
 	}
@@ -112,8 +130,19 @@ type connState struct {
 	// peerClosed is set when EPOLLRDHUP reports the peer half-closed (FIN) while
 	// a response was still flushing (write backpressure). The conn is closed once
 	// its pending write drains, so the response is not truncated. Reset on release.
+	// closeWhenFlushed sets it too, when the engine itself ends a conn whose
+	// response the kernel has not all taken (Connection: close, a request
+	// error, a refused write; celeris#761): the close it defers is the same.
 	peerClosed bool
 
+	// writeRefused records that a write hook refused bytes because the
+	// conn's backlog was already over writeCap (celeris#761). A refused
+	// write is never silent: the site that ran the handler closes the conn
+	// (closeWhenFlushed) instead of leaving the client waiting for bytes that
+	// will not come. Sticky until release. Written under detachMu when the
+	// conn has one, like the buffers it guards.
+	writeRefused bool
+
 	// drainDeadline bounds how long checkTimeouts defers the idle-deadline
 	// reap of a truly-detached conn whose terminal bytes (SSE last event, WS
 	// close echo) are still queued: stamped now+detachDrainGrace on the first
@@ -319,6 +348,7 @@ func releaseConnState(cs *connState) {
 	cs.asyncOutBuf = trimPooledBuf(cs.asyncOutBuf)
 	cs.writeBuf = trimPooledBuf(cs.writeBuf)
 	cs.peerClosed = false
+	cs.writeRefused = false
 	cs.drainDeadline = 0
 	cs.asyncRun = false
 	cs.asyncParked = false
diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go
index 7b239f06..0a28c1c5 100644
--- a/engine/epoll/loop.go
+++ b/engine/epoll/loop.go
@@ -679,20 +679,36 @@ func (l *Loop) run(ctx context.Context) {
 
 		// Drain H2 async write queues. Handler goroutines enqueue response
 		// frame bytes; we drain them into writeBuf and flush to the wire.
-		for _, fd := range l.h2Conns {
+		// By index, from the end: a close below swap-removes the conn from
+		// h2Conns, which a range would then skip one entry past.
+		for i := len(l.h2Conns) - 1; i >= 0; i-- {
+			fd := l.h2Conns[i]
 			cs := l.conns[fd]
 			if cs != nil && cs.h2State != nil && cs.h2State.WriteQueuePending() {
 				cs.h2State.DrainWriteQueue(cs.writeFn)
+				if cs.writeRefused {
+					// A frame refused on back-pressure is lost, and the
+					// connection's framing with it: close (celeris#761).
+					l.closeWhenFlushed(cs)
+					continue
+				}
 				if cs.writePos < len(cs.writeBuf) {
 					if fErr := l.flushWrites(cs, true); fErr != nil {
 						l.removeDirty(cs)
 						l.closeConn(fd)
+						continue
 					} else if cs.writePos < len(cs.writeBuf) {
 						// Send buffer full: arm EPOLLOUT instead of the dirty
 						// list so we don't busy-poll the backpressured H2 conn.
 						l.armEpollOut(cs)
 					}
 				}
+				// Resync pendingBytes, as every other flush point does: the
+				// write hook added every frame drained above to it, and left
+				// alone it outgrew what is really queued until the hook
+				// refused frames of a conn that had nothing queued
+				// (celeris#761).
+				cs.pendingBytes = csPendingBytes(cs)
 			}
 		}
 
@@ -1244,7 +1260,7 @@ func (l *Loop) drainRead(fd int, now int64) {
 				if mu := cs.detachMu; mu != nil {
 					mu.Unlock()
 				}
-				l.closeConn(fd)
+				l.closeWhenFlushed(cs) // celeris#761: the response's tail too
 				return
 			}
 			if len(rest) > 0 {
@@ -1261,7 +1277,7 @@ func (l *Loop) drainRead(fd int, now int64) {
 					if mu := cs.detachMu; mu != nil {
 						mu.Unlock()
 					}
-					l.closeConn(fd)
+					l.closeWhenFlushed(cs) // celeris#761: the response's tail too
 					return
 				}
 			}
@@ -1275,10 +1291,25 @@ func (l *Loop) drainRead(fd int, now int64) {
 				l.closeConn(fd)
 				return
 			}
-			dirty := len(cs.writeBuf)-cs.writePos > 0 || len(cs.bodyBuf) > 0
+			// Resync pendingBytes, as the inline flush below does: the write
+			// hooks add every response to it, and only a flush point brings
+			// it back to what is still queued. Left alone here it grew by
+			// each response answered on this path, until the hooks refused
+			// the writes of a conn that had nothing queued (celeris#761).
+			dirty := csWritePending(cs)
+			if dirty {
+				cs.pendingBytes = csPendingBytes(cs)
+			} else {
+				cs.pendingBytes = 0
+			}
+			refused := cs.writeRefused
 			if mu := cs.detachMu; mu != nil {
 				mu.Unlock()
 			}
+			if refused {
+				l.closeWhenFlushed(cs)
+				return
+			}
 			if dirty {
 				// Send buffer full after the body-recv response flush. This
 				// path is sync-only (intoBody is gated on !l.async), so the
@@ -1511,7 +1542,10 @@ func (l *Loop) drainRead(fd int, now int64) {
 			if mu := cs.detachMu; mu != nil {
 				mu.Unlock()
 			}
-			l.closeConn(fd)
+			// What that one write could not send (a response larger than
+			// the socket buffers, answering Connection: close) still goes
+			// out before the close (celeris#761).
+			l.closeWhenFlushed(cs)
 			return
 		}
 
@@ -1562,15 +1596,21 @@ func (l *Loop) drainRead(fd int, now int64) {
 				}
 			}
 		}
-		// Capture pendingBytes inside the lock so the check below is safe
+		// Read writeRefused inside the lock so the check below is safe
 		// against concurrent goroutine writes via the guarded writeFn.
-		pending := cs.pendingBytes
+		refused := cs.writeRefused
 		if mu := cs.detachMu; mu != nil {
 			mu.Unlock()
 		}
 
-		if pending > cs.writeCap() {
-			l.closeConn(fd)
+		// Back-pressure close: a write hook refused bytes because the
+		// backlog before them was over writeCap, a peer that stopped
+		// reading while it kept sending requests. A backlog over the cap
+		// by itself is not a reason to close: one response larger than the
+		// cap puts it there, and so does a sendfile body. Closing on it cut
+		// such a response off mid-body (celeris#761).
+		if refused {
+			l.closeWhenFlushed(cs)
 			return
 		}
 
@@ -1600,6 +1640,38 @@ func (l *Loop) drainRead(fd int, now int64) {
 	}
 }
 
+// closeWhenFlushed closes cs as closeConn does, but only once the response
+// bytes it still has queued have reached the kernel: closeConn's SHUT_WR
+// commits only what the kernel has already taken, so a response larger than
+// the socket buffers lost its tail when the conn was closed right after it
+// (Connection: close, a request error, a refused write; celeris#761). It
+// stops reading cs, so no request is parsed or answered on a conn that is
+// closing, arms EPOLLOUT and marks the close deferred (peerClosed), which
+// handleWritable and the dirty pass carry out once the conn has drained. A
+// peer that never reads again is reaped by checkTimeouts, like any conn
+// stalled on write back-pressure. A conn with nothing left to send, and a
+// truly-detached one, whose middleware owns its close, are closed at once.
+//
+// Loop thread, with no handler of cs's running: the inline path after its
+// handler returned, or drainDetachQueue after the dispatch goroutine exited.
+func (l *Loop) closeWhenFlushed(cs *connState) {
+	if (cs.h1State != nil && cs.h1State.Detached.Load()) || !csWritePending(cs) {
+		l.closeConn(cs.fd)
+		return
+	}
+	cs.peerClosed = true
+	l.removeDirty(cs)
+	issued, err := l.modEpollOut(cs, unix.EPOLLOUT|unix.EPOLLET|unix.EPOLLRDHUP)
+	if !issued {
+		return // hijacked: drainDetachQueue settles the conn
+	}
+	if err != nil {
+		l.closeConn(cs.fd)
+		return
+	}
+	cs.epollOut = true
+}
+
 // closeOnReadEnd is drainRead's read-error and EOF branch: flush what is
 // queued, surface err to a detached middleware (OnError, under detachMu, as
 // every OnError site in this file runs), and close.
@@ -2105,7 +2177,19 @@ func (l *Loop) switchToH2Local(cs *connState, writeFn func([]byte)) error {
 
 func (l *Loop) makeWriteFn(cs *connState) func([]byte) {
 	return func(data []byte) {
-		if cs.pendingBytes+len(cs.bodyBuf) > cs.writeCap() {
+		// Back-pressure: refuse the write only when the backlog before it is
+		// over the cap (see writeCap), and say so, so the conn is closed
+		// rather than left waiting for bytes that will not come (celeris#761).
+		// pendingBytes already counts a staged bodyBuf.
+		if cs.pendingBytes > cs.writeCap() {
+			cs.writeRefused = true
+			return
+		}
+		// A zero-copy body or sendfile is staged, and the flush sends
+		// writeBuf before them: bytes written after them (the next
+		// pipelined response) would go out ahead (celeris#802).
+		if (cs.bodyBuf != nil || cs.sendfile != nil) && !unstage(cs) {
+			cs.writeRefused = true
 			return
 		}
 		cs.writeBuf = append(cs.writeBuf, data...)
@@ -2122,14 +2206,20 @@ func (l *Loop) makeWriteFn(cs *connState) func([]byte) {
 // adapter for bodies ≥ 8 KiB to skip the respBuf → writeBuf memcpy.
 func (l *Loop) makeWriteBodyFn(cs *connState) func([]byte) {
 	return func(body []byte) {
-		if cs.pendingBytes+len(cs.bodyBuf)+len(body) > cs.writeCap() {
+		// The body itself is not held against the cap: a response larger
+		// than the cap is staged whole, and only a backlog already over it
+		// refuses the write (celeris#761; see writeCap and makeWriteFn).
+		if cs.pendingBytes > cs.writeCap() {
+			cs.writeRefused = true
 			return
 		}
-		if cs.bodyBuf != nil {
-			// A second large-body write in the same request: fall back
-			// to copying (we only carry one writev body slot per flush).
-			cs.writeBuf = append(cs.writeBuf, body...)
-			cs.pendingBytes += len(body)
+		// A second large body in the same flush (a pipelined response), or
+		// one behind a staged sendfile: there is one writev body slot, so
+		// what is staged moves into writeBuf, where it stays ahead of this
+		// body, and this body takes the slot. Copying this one into writeBuf
+		// instead sent it before the staged body (celeris#802).
+		if (cs.bodyBuf != nil || cs.sendfile != nil) && !unstage(cs) {
+			cs.writeRefused = true
 			return
 		}
 		cs.bodyBuf = body
@@ -2153,14 +2243,16 @@ func (l *Loop) makeWriteBodyFn(cs *connState) func([]byte) {
 // closed by flushSendfile on completion or by closeConn / releaseConnState
 // on teardown.
 //
-// Pipelined file requests in a single recv (cs.sendfile already set) and
-// rare dup failures fall back to a buffered copy of this response into
-// writeBuf, so the engine never needs two concurrent sendfile slots and
-// response ordering is always preserved.
+// There is one sendfile slot. A file response pipelined behind another one
+// in a single recv (cs.sendfile already set) first moves the staged one into
+// writeBuf as a buffered copy (unstage), so it stays ahead, and then takes
+// the slot; handing the new one the buffered copy instead sent it before the
+// staged file (celeris#802). A rare dup failure falls back to a buffered
+// copy of this response.
 func (l *Loop) makeSendFileFn(cs *connState) func(header []byte, file *os.File, offset, length int64) error {
 	return func(header []byte, file *os.File, offset, length int64) error {
-		if cs.sendfile != nil {
-			return bufferedFileFallback(cs, header, file, offset, length)
+		if cs.sendfile != nil && !unstage(cs) {
+			return errUnstageSendfile
 		}
 		dupfd, err := unix.Dup(int(file.Fd()))
 		if err != nil {
@@ -2185,6 +2277,11 @@ func (l *Loop) makeSendFileFn(cs *connState) func(header []byte, file *os.File,
 // caller still owns and closes the *os.File. Correctness over zero-copy:
 // the response is delivered byte-exact, just without the syscall savings.
 func bufferedFileFallback(cs *connState, header []byte, file *os.File, offset, length int64) error {
+	// This response is appended to writeBuf, which goes out before a
+	// staged body or sendfile (celeris#802).
+	if (cs.bodyBuf != nil || cs.sendfile != nil) && !unstage(cs) {
+		return errUnstageSendfile
+	}
 	if length <= 0 {
 		fi, err := file.Stat()
 		if err != nil {
@@ -2464,13 +2561,17 @@ func (l *Loop) runAsyncHandler(cs *connState) {
 			}
 		}
 		partial := flushErr == nil && (cs.writePos < len(cs.writeBuf) || len(cs.bodyBuf) > 0)
+		// A refused write (celeris#761) ends the conn as a request error
+		// does: this goroutine exits and the loop closes the conn once what
+		// was staged has gone out (drainDetachQueue, closeWhenFlushed).
+		refused := cs.writeRefused
 		cs.detachMu.Unlock()
 
 		if partial {
 			l.enqueueDetach(cs)
 		}
 
-		if processErr != nil || flushErr != nil {
+		if processErr != nil || flushErr != nil || refused {
 			// Signal the worker to tear down the conn from its own
 			// goroutine. Never close cs.fd directly here — the worker
 			// goroutine still has l.conns[fd] pointing at cs, and a
@@ -2590,9 +2691,20 @@ func (l *Loop) drainDetachQueue() {
 		}
 		// Dispatch goroutine signaled close via asyncClosed. Only the
 		// worker can safely touch l.conns / dirty list, so we handle
-		// the teardown here.
+		// the teardown here. Once the goroutine has exited, the response
+		// it left queued goes out before the close (celeris#761): closing
+		// at once cut off, after Connection: close, any response larger
+		// than the socket buffers. While it still runs, closeConn leaves
+		// the close to it (celeris#669), as before.
 		if cs.asyncClosed.Load() {
-			l.closeConn(cs.fd)
+			cs.asyncInMu.Lock()
+			running := cs.asyncRun
+			cs.asyncInMu.Unlock()
+			if running {
+				l.closeConn(cs.fd)
+			} else {
+				l.closeWhenFlushed(cs)
+			}
 			continue
 		}
 		// Dispatch goroutine promoted the conn to H2 via switchToH2Local.
diff --git a/engine/epoll/write_hooks_bench_linux_test.go b/engine/epoll/write_hooks_bench_linux_test.go
new file mode 100644
index 00000000..914f3922
--- /dev/null
+++ b/engine/epoll/write_hooks_bench_linux_test.go
@@ -0,0 +1,42 @@
+//go:build linux
+
+package epoll
+
+import "testing"
+
+// BenchmarkWriteHooks measures the per-response cost of the write hooks the
+// H1 response adapter calls (makeWriteFn, makeWriteBodyFn), whose
+// back-pressure check celeris#761 changed: a header block and a small body
+// through writeFn, and a header block and a 16 KiB zero-copy body through
+// writeFn + writeBodyFn. The buffers are reset as a completed flush leaves
+// them.
+func BenchmarkWriteHooks(b *testing.B) {
+	hdr := make([]byte, 121)
+	small := make([]byte, 64)
+	large := make([]byte, 16<<10)
+	b.Run("small", func(b *testing.B) {
+		l := &Loop{}
+		cs := &connState{}
+		w := l.makeWriteFn(cs)
+		b.ReportAllocs()
+		for b.Loop() {
+			w(hdr)
+			w(small)
+			cs.writeBuf = cs.writeBuf[:0]
+			cs.pendingBytes = 0
+		}
+	})
+	b.Run("zero-copy-body", func(b *testing.B) {
+		l := &Loop{}
+		cs := &connState{}
+		w, wb := l.makeWriteFn(cs), l.makeWriteBodyFn(cs)
+		b.ReportAllocs()
+		for b.Loop() {
+			w(hdr)
+			wb(large)
+			cs.writeBuf = cs.writeBuf[:0]
+			cs.bodyBuf = nil
+			cs.pendingBytes = 0
+		}
+	})
+}
diff --git a/engine/epoll/writer.go b/engine/epoll/writer.go
index 7f4526f4..dbf7489e 100644
--- a/engine/epoll/writer.go
+++ b/engine/epoll/writer.go
@@ -3,6 +3,8 @@
 package epoll
 
 import (
+	"errors"
+
 	"golang.org/x/sys/unix"
 )
 
@@ -108,6 +110,44 @@ func (l *Loop) flushSendfile(cs *connState, onLoopThread bool) error {
 	return nil
 }
 
+// errUnstageSendfile is the error a file response gets when the file of the
+// sendfile staged ahead of it could not be read into writeBuf (unstage); the
+// conn is closed.
+var errUnstageSendfile = errors.New("celeris: epoll: reading a staged sendfile response into the write buffer failed")
+
+// unstage moves what the conn has staged outside writeBuf, a zero-copy body
+// (bodyBuf) and then a sendfile response, into writeBuf, in the order the
+// flush sends them (writeBuf, bodyBuf, sendfile), so that bytes written next
+// go out after them: every writer appends to writeBuf, which the flush sends
+// first (celeris#802). The body is copied; the sendfile's unsent header and
+// file bytes are read in and its descriptor closed. Only a write that
+// follows a large body or a file response before the flush pays for it: the
+// next pipelined response. Reports false if the file could not be read; the
+// conn must then be closed, its response being lost either way.
+func unstage(cs *connState) bool {
+	if cs.bodyBuf != nil {
+		cs.writeBuf = append(cs.writeBuf, cs.bodyBuf...)
+		cs.bodyBuf = nil
+	}
+	st := cs.sendfile
+	if st == nil {
+		return true
+	}
+	cs.sendfile = nil
+	defer st.close()
+	cs.writeBuf = append(cs.writeBuf, st.headers[st.headerOff:]...)
+	if st.remaining <= 0 {
+		return true
+	}
+	start := len(cs.writeBuf)
+	cs.writeBuf = append(cs.writeBuf, make([]byte, st.remaining)...)
+	if err := readFullAt(st.file, cs.writeBuf[start:], st.off); err != nil {
+		cs.writeBuf = cs.writeBuf[:start]
+		return false
+	}
+	return true
+}
+
 // csWritePending reports whether cs has any unsent output: buffered bytes
 // in writeBuf, a staged scatter-gather bodyBuf, or an in-progress
 // sendfile response. The flush paths use it to decide whether the
diff --git a/engine/iouring/conn.go b/engine/iouring/conn.go
index 732d3142..449235c3 100644
--- a/engine/iouring/conn.go
+++ b/engine/iouring/conn.go
@@ -21,8 +21,17 @@ import (
 // echo payloads larger than 4 MiB (Autobahn 9.1.6 sends 16 MiB).
 // 64 MiB matches the WS default ReadLimit.
 const (
-	maxSendQueueBytes         = 4 << 20  // 4 MiB (H1/H2)
+	maxSendQueueBytes         = 4 << 20  // 4 MiB (H1)
 	maxSendQueueBytesDetached = 64 << 20 // 64 MiB (WS/SSE)
+	// maxSendQueueBytesH2 is the limit for an HTTP/2 connection. Its DATA
+	// is already bounded by the flow-control windows the peer grants, and a
+	// peer that reads keeps up to a window of frames queued, or in a SEND
+	// in flight, as a matter of course (net/http's client grants 4 MiB per
+	// stream, browsers more per connection), so the H1 limit closed healthy
+	// HTTP/2 connections at the window's edge (celeris#761). 64 MiB, as for
+	// a detached connection, still bounds a peer that grants large windows
+	// and stops reading.
+	maxSendQueueBytesH2 = 64 << 20
 	// maxPendingInputBytes caps the async dispatch input buffer
 	// (cs.asyncInBuf) so a client pipelining requests faster than
 	// the dispatch goroutine drains them cannot balloon per-conn
@@ -34,7 +43,16 @@ const (
 // for whether the connection is detached. Async-mode HTTP1 conns set
 // detachMu up front without being truly detached; they keep the H1/H2
 // limit so a stalled peer cannot balloon per-conn memory to 64 MiB.
+//
+// The limit bounds the backlog a write finds, not the write: a write is
+// refused only when the bytes still queued before it are over the limit,
+// which is a peer that stopped reading while it keeps sending requests. One
+// response larger than the limit is not such a backlog and is staged whole;
+// before celeris#761 its body was dropped.
 func (cs *connState) sendCap() int {
+	if cs.h2State != nil {
+		return maxSendQueueBytesH2
+	}
 	if cs.detachMu != nil && cs.h1State != nil && cs.h1State.Detached.Load() {
 		return maxSendQueueBytesDetached
 	}
@@ -84,6 +102,7 @@ type connState struct {
 	detected       bool         // 1
 	sending        bool         // 1: true when a SEND SQE is in-flight
 	closing        bool         // 1: defers close until sends complete
+	writeRefused   bool         // 1: a write hook refused bytes on back-pressure; the conn is closed (celeris#761, makeWriteFn)
 	dirty          bool         // 1: true when data needs flushing
 	fixedFile      bool         // 1: true when fd is fixed file index
 	recvLinked     bool         // 1: RECV was linked to SEND (skip standalone prepareRecv)
@@ -498,6 +517,7 @@ func releaseConnState(cs *connState) {
 	cs.detected = false
 	cs.sending = false
 	cs.closing = false
+	cs.writeRefused = false
 	cs.dirty = false
 	cs.fixedFile = false
 	cs.recvLinked = false
diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go
index b734d332..3df93661 100644
--- a/engine/iouring/worker.go
+++ b/engine/iouring/worker.go
@@ -49,6 +49,10 @@ const fixedFileTableSize = 65536
 // reports the same condition through its own errPeerClosed.
 var errPeerClosed = fmt.Errorf("celeris: peer closed connection: %w", io.EOF)
 
+// errWriteRefused ends an async dispatch goroutine whose handler's writes
+// the back-pressure cap refused (celeris#761): the worker closes the conn.
+var errWriteRefused = errors.New("celeris: write refused: send backlog over the limit")
+
 // errIORingRecv wraps a negative io_uring recv result as a syscall.Errno.
 // Used to surface concrete errors to detached middleware via H1State.OnError.
 // Callers MUST handle res == 0 before reaching here: a zero result is an
@@ -1340,10 +1344,20 @@ func (w *Worker) run(ctx context.Context) {
 		// response frame bytes; draining them before the dirty list ensures
 		// SEND SQEs are queued as early as possible after CQE processing,
 		// reducing pipeline stalls for H2 multiplexed streams.
-		for _, fd := range w.h2Conns {
+		// By index, from the end: a close below swap-removes the conn from
+		// h2Conns, which a range would then skip one entry past.
+		for i := len(w.h2Conns) - 1; i >= 0; i-- {
+			fd := w.h2Conns[i]
 			cs := w.conns[fd]
 			if cs != nil && cs.h2State != nil && cs.h2State.WriteQueuePending() {
 				cs.h2State.DrainWriteQueue(cs.writeFn)
+				if cs.writeRefused {
+					// A frame refused on back-pressure is lost, and the
+					// connection's framing with it: close, sending what
+					// was staged first (celeris#761).
+					w.closeConn(fd)
+					continue
+				}
 				if w.flushSend(cs) {
 					w.markDirty(cs)
 				}
@@ -2783,8 +2797,9 @@ func (w *Worker) handleRecv(c *completionEntry, fd int, now int64) {
 		}
 		// The direct-body tail: flushed unlinked (the next recv may target
 		// the body buffer again), then a standalone recv — or none, held for
-		// a hand-off (celeris#657).
-		w.respondAndArm(cs, fd, c, false, false)
+		// a hand-off (celeris#657). A refused write closes here too
+		// (celeris#761).
+		w.respondAndArm(cs, fd, c, false, true)
 		return
 	}
 
@@ -3193,7 +3208,11 @@ func (w *Worker) handleRecv(c *completionEntry, fd int, now int64) {
 // response, then arm the connection's next recv, chained behind the SEND
 // (link, single-shot recv only; flushSendLink falls back to an unlinked send
 // itself when it must) or standalone. capCheck applies the back-pressure
-// close of the sync tail.
+// close of the sync tail: a conn whose write hooks refused bytes is closed,
+// and its close sends what was staged first. A backlog over the cap by
+// itself is not a reason to close: one response larger than the cap puts it
+// there, and so does an HTTP/2 stream's flow-control window; closing on it
+// cut such a response off (celeris#761).
 //
 // With a drain set and a conn the hand-off would take, the response is
 // flushed UNLINKED and no recv is armed at all (HOLD): the conn is handed off
@@ -3208,9 +3227,9 @@ func (w *Worker) respondAndArm(cs *connState, fd int, c *completionEntry, link,
 	if mu != nil {
 		mu.Lock()
 	}
-	// Back-pressure: capture pending size inside the lock so concurrent
+	// Back-pressure: read writeRefused inside the lock so concurrent
 	// goroutine writes via the guarded writeFn don't race the read.
-	if capCheck && len(cs.writeBuf)+len(cs.sendBuf) > cs.sendCap() {
+	if capCheck && cs.writeRefused {
 		if mu != nil {
 			mu.Unlock()
 		}
@@ -4520,6 +4539,11 @@ func (w *Worker) runAsyncHandler(cs *connState) {
 		// and closes the 3× integrated-Redis regression observed on
 		// iouring (95 µs/op → target ~30 µs/op, matching epoll and
 		// go-redis + stdlib).
+		// A refused write (celeris#761) ends the conn as a request error
+		// does: the worker's closeConn sends what was staged, then closes.
+		if processErr == nil && cs.writeRefused {
+			processErr = errWriteRefused
+		}
 		var partial bool
 		if processErr == nil && cs.fixedFile && len(cs.writeBuf) > 0 {
 			// Fixed-file conn: cs.fd is a registered-file TABLE INDEX, not a
@@ -4581,11 +4605,23 @@ func (w *Worker) makeWriteFn(cs *connState) func([]byte) {
 		if cs.closing {
 			return
 		}
-		// Back-pressure: drop writes when total pending data exceeds limit.
-		// The connection will be closed after processing completes.
+		// Back-pressure: refuse the write only when the backlog before it
+		// is over the cap (see sendCap), and say so: the site that ran the
+		// handler then closes the conn (celeris#761), where it used to be
+		// left waiting for the refused bytes.
 		if len(cs.writeBuf)+len(cs.sendBuf)+len(cs.bodyBuf) > cs.sendCap() {
+			cs.writeRefused = true
 			return
 		}
+		if cs.bodyBuf != nil {
+			// A zero-copy body is staged, and flushSend sends writeBuf
+			// before it: bytes written after the body (the next pipelined
+			// response) would go out ahead of it. Move the body into
+			// writeBuf first, a copy paid only when a write follows a large
+			// body before the flush (celeris#802).
+			cs.writeBuf = append(cs.writeBuf, cs.bodyBuf...)
+			cs.bodyBuf = nil
+		}
 		// Append to writeBuf — no per-write allocation. The kernel holds
 		// sendBuf (not writeBuf), so appending here is safe.
 		// Don't markDirty here — handleRecv calls flushSend after the
@@ -4614,14 +4650,20 @@ func (w *Worker) makeWriteBodyFn(cs *connState) func([]byte) {
 		if cs.closing {
 			return
 		}
-		if len(cs.writeBuf)+len(cs.sendBuf)+len(cs.bodyBuf)+len(body) > cs.sendCap() {
+		// The body itself is not held against the cap: a response larger
+		// than the cap is staged whole, and only a backlog already over it
+		// refuses the write (celeris#761; see sendCap and makeWriteFn).
+		if len(cs.writeBuf)+len(cs.sendBuf)+len(cs.bodyBuf) > cs.sendCap() {
+			cs.writeRefused = true
 			return
 		}
 		if cs.bodyBuf != nil {
-			// A second large-body write in the same request: fall back
-			// to copying (we only carry one iovec entry for the body).
-			cs.writeBuf = append(cs.writeBuf, body...)
-			return
+			// A second large body before the flush (a pipelined response):
+			// there is one iovec entry for a body, so the staged body moves
+			// into writeBuf, where it stays ahead of this one, and this one
+			// takes the entry. Copying this one into writeBuf instead sent
+			// it before the staged body (celeris#802).
+			cs.writeBuf = append(cs.writeBuf, cs.bodyBuf...)
 		}
 		cs.bodyBuf = body
 	}
diff --git a/engine/iouring/write_hooks_bench_linux_test.go b/engine/iouring/write_hooks_bench_linux_test.go
new file mode 100644
index 00000000..cd2ad567
--- /dev/null
+++ b/engine/iouring/write_hooks_bench_linux_test.go
@@ -0,0 +1,40 @@
+//go:build linux
+
+package iouring
+
+import "testing"
+
+// BenchmarkWriteHooks measures the per-response cost of the write hooks the
+// H1 response adapter calls (makeWriteFn, makeWriteBodyFn), whose
+// back-pressure check celeris#761 changed: a header block and a small body
+// through writeFn, and a header block and a 16 KiB zero-copy body through
+// writeFn + writeBodyFn. The buffers are reset as a completed send leaves
+// them.
+func BenchmarkWriteHooks(b *testing.B) {
+	hdr := make([]byte, 121)
+	small := make([]byte, 64)
+	large := make([]byte, 16<<10)
+	b.Run("small", func(b *testing.B) {
+		w := &Worker{}
+		cs := &connState{}
+		wf := w.makeWriteFn(cs)
+		b.ReportAllocs()
+		for b.Loop() {
+			wf(hdr)
+			wf(small)
+			cs.writeBuf = cs.writeBuf[:0]
+		}
+	})
+	b.Run("zero-copy-body", func(b *testing.B) {
+		w := &Worker{}
+		cs := &connState{}
+		wf, wb := w.makeWriteFn(cs), w.makeWriteBodyFn(cs)
+		b.ReportAllocs()
+		for b.Loop() {
+			wf(hdr)
+			wb(large)
+			cs.writeBuf = cs.writeBuf[:0]
+			cs.bodyBuf = nil
+		}
+	})
+}
diff --git a/large_response_linux_test.go b/large_response_linux_test.go
new file mode 100644
index 00000000..ef87c8bc
--- /dev/null
+++ b/large_response_linux_test.go
@@ -0,0 +1,590 @@
+//go:build linux
+
+package celeris_test
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/goceleris/celeris"
+)
+
+// Tests for celeris#761. On epoll and io_uring the per-connection write
+// back-pressure cap (4 MiB: maxPendingBytes, maxSendQueueBytes) was held
+// against a single response: the write hook that stages a large body counted
+// the body itself, so an HTTP/1.1 response whose headers and body passed the
+// cap went out as its headers only, the body dropped without an error and the
+// connection left open, and a keep-alive client waited for the declared
+// Content-Length until its own timeout. The check after the handler closed
+// any connection whose backlog was over the cap, which cut off every response
+// that path let through (a copied body, a sendfile body, an HTTP/2 stream's
+// flow-control window), and epoll closed a connection answering Connection:
+// close with whatever the socket had not taken yet unsent. The cap is there
+// to stop a peer that does not read from piling up responses; a single
+// response is not such a backlog.
+//
+// A client that stops receiving bytes for idleCap761 fails its case, so a
+// dropped body fails in seconds, not at a timeout.
+
+// idleCap761 is how long a client waits for the next byte before it calls the
+// response lost: loopback delivers a 64 MiB body in well under a second even
+// under -race, and a dropped body never sends one more byte.
+const idleCap761 = 5 * time.Second
+
+var engines761 = []struct {
+	name string
+	eng  celeris.EngineType
+}{{"std", celeris.Std}, {"epoll", celeris.Epoll}, {"io_uring", celeris.IOUring}, {"adaptive", celeris.Adaptive}}
+
+// bodies761 builds one patterned body per size, so a body that arrives out of
+// order or shifted does not compare equal.
+func bodies761(sizes ...int) map[int][]byte {
+	m := make(map[int][]byte, len(sizes))
+	for _, n := range sizes {
+		b := make([]byte, n)
+		for i := range b {
+			b[i] = byte(i*7 + i>>13)
+		}
+		m[n] = b
+	}
+	return m
+}
+
+// TestLargeResponseIsDelivered asks for one body of each size on a keep-alive
+// connection and then for /ping on the same connection (the framing must
+// still be intact), and once more with Connection: close, where the body must
+// be followed by EOF. The sizes straddle the old threshold, which was the cap
+// minus the header block, so 4 MiB - 1 failed too; 4 MiB - 4 KiB was
+// delivered before the fix on a keep-alive connection.
+//
+// "sync" runs the handler on the connection's worker, where epoll and
+// io_uring hand a large body to the engine as a zero-copy slice (the write
+// hook that dropped it). "async-loop" is a server with AsyncHandlers and a
+// route that is not async, so the handler runs on the worker but the body is
+// copied into the write buffer. "async-route" runs the handler on the
+// connection's dispatch goroutine.
+func TestLargeResponseIsDelivered(t *testing.T) {
+	sizes := []int{4<<20 - 4096, 4<<20 - 1, 4 << 20, 4<<20 + 1, 64 << 20}
+	bodies := bodies761(sizes...)
+	shapes := []struct {
+		name                    string
+		asyncServer, asyncRoute bool
+	}{{"sync", false, false}, {"async-loop", true, false}, {"async-route", false, true}}
+
+	for _, e := range engines761 {
+		for _, sh := range shapes {
+			t.Run(e.name+"/"+sh.name, func(t *testing.T) {
+				addr := startServer761(t, e.eng, sh.asyncServer, func(s *celeris.Server) {
+					big := s.GET("/big", func(c *celeris.Context) error {
+						n, err := strconv.Atoi(c.Query("n"))
+						if err != nil {
+							return err
+						}
+						body, ok := bodies[n]
+						if !ok {
+							return fmt.Errorf("no body of %d bytes", n)
+						}
+						return c.Blob(http.StatusOK, "application/octet-stream", body)
+					})
+					if sh.asyncRoute {
+						big.Async()
+					}
+				})
+				for _, n := range sizes {
+					for _, keepAlive := range []bool{true, false} {
+						mode := "keep-alive"
+						if !keepAlive {
+							mode = "close"
+						}
+						t.Run(strconv.Itoa(n)+"/"+mode, func(t *testing.T) {
+							desc := fmt.Sprintf("%s/%s body %d %s", e.name, sh.name, n, mode)
+							checkH1Response761(t, addr, "/big?n="+strconv.Itoa(n), desc, bodies[n], keepAlive)
+						})
+					}
+				}
+			})
+		}
+	}
+}
+
+// TestLargeFileResponseIsDelivered serves a file with c.File. On epoll's
+// worker that is sendfile(2), whose backlog the check after the handler held
+// against the cap and closed mid-file; elsewhere it is read into memory and
+// written like a Blob.
+func TestLargeFileResponseIsDelivered(t *testing.T) {
+	sizes := []int{4<<20 + 1, 64 << 20}
+	bodies := bodies761(sizes...)
+	dir := t.TempDir()
+	for _, n := range sizes {
+		if err := os.WriteFile(filepath.Join(dir, strconv.Itoa(n)), bodies[n], 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for _, e := range engines761 {
+		for _, route := range []string{"sync", "async-route"} {
+			t.Run(e.name+"/"+route, func(t *testing.T) {
+				addr := startServer761(t, e.eng, false, func(s *celeris.Server) {
+					r := s.GET("/file/:n", func(c *celeris.Context) error {
+						return c.File(filepath.Join(dir, c.Param("n")))
+					})
+					if route == "async-route" {
+						r.Async()
+					}
+				})
+				for _, n := range sizes {
+					t.Run(strconv.Itoa(n), func(t *testing.T) {
+						desc := fmt.Sprintf("%s/%s file %d", e.name, route, n)
+						checkH1Response761(t, addr, "/file/"+strconv.Itoa(n), desc, bodies[n], true)
+					})
+				}
+			})
+		}
+	}
+}
+
+// TestLargeResponseIsDeliveredH2 asks for each body over HTTP/2 (h2c, prior
+// knowledge). net/http's client opens a 4 MiB stream window, so a larger body
+// leaves up to 4 MiB of frames queued behind the socket: io_uring's check
+// after the handler closed the connection on that backlog, at exactly 4 MiB.
+func TestLargeResponseIsDeliveredH2(t *testing.T) {
+	sizes := []int{4<<20 - 4096, 4<<20 + 1, 64 << 20}
+	bodies := bodies761(sizes...)
+	for _, e := range engines761 {
+		for _, route := range []string{"sync", "async-route"} {
+			t.Run(e.name+"/"+route, func(t *testing.T) {
+				addr := startServer761(t, e.eng, false, func(s *celeris.Server) {
+					r := s.GET("/big/:n", func(c *celeris.Context) error {
+						n, _ := strconv.Atoi(c.Param("n"))
+						body, ok := bodies[n]
+						if !ok {
+							return fmt.Errorf("no body of %d bytes", n)
+						}
+						return c.Blob(http.StatusOK, "application/octet-stream", body)
+					})
+					if route == "async-route" {
+						r.Async()
+					}
+				})
+				p := new(http.Protocols)
+				p.SetUnencryptedHTTP2(true)
+				cl := &http.Client{Timeout: 60 * time.Second, Transport: &http.Transport{
+					Protocols: p,
+					DialContext: func(ctx context.Context, network, a string) (net.Conn, error) {
+						var d net.Dialer
+						c, err := d.DialContext(ctx, network, a)
+						if err != nil {
+							return nil, err
+						}
+						return idleConn761{c}, nil
+					},
+				}}
+				defer cl.CloseIdleConnections()
+				for _, n := range sizes {
+					t.Run(strconv.Itoa(n), func(t *testing.T) {
+						desc := fmt.Sprintf("%s/%s h2 body %d", e.name, route, n)
+						resp, err := cl.Get("http://" + addr + "/big/" + strconv.Itoa(n))
+						if err != nil {
+							t.Fatalf("%s: %v", desc, err)
+						}
+						defer func() { _ = resp.Body.Close() }()
+						if resp.ProtoMajor != 2 || resp.StatusCode != http.StatusOK {
+							t.Fatalf("%s: %s %d", desc, resp.Proto, resp.StatusCode)
+						}
+						got, err := io.ReadAll(resp.Body)
+						if err != nil || !bytes.Equal(got, bodies[n]) {
+							t.Fatalf("%s: received %d of %d body bytes, equal=%v, then %s",
+								desc, len(got), n, bytes.Equal(got, bodies[n]), describeReadEnd761(err))
+						}
+					})
+				}
+			})
+		}
+	}
+}
+
+// TestSplitBodyResponsesKeepTheConnection sends one request after another on
+// one connection, each with a body that arrives in two writes, so the rest of
+// the body is read straight into the request's body buffer (epoll's
+// zero-copy body receive). epoll's flush on that path never brought the
+// pending-byte count back down, so it grew by every response, and once the
+// responses on the connection added up to the cap the write hook dropped the
+// next one and the client waited for it.
+func TestSplitBodyResponsesKeepTheConnection(t *testing.T) {
+	const (
+		requests = 96       // 96 x 64 KiB = 6 MiB of responses, past the 4 MiB cap
+		respSize = 64 << 10 // over the 8 KiB zero-copy body threshold
+		bodySize = 32 << 10
+	)
+	resp := bodies761(respSize)[respSize]
+	for _, e := range engines761 {
+		t.Run(e.name, func(t *testing.T) {
+			addr := startServer761(t, e.eng, false, func(s *celeris.Server) {
+				s.POST("/upload", func(c *celeris.Context) error {
+					if len(c.Body()) != bodySize {
+						return fmt.Errorf("body %d bytes, want %d", len(c.Body()), bodySize)
+					}
+					return c.Blob(http.StatusOK, "application/octet-stream", resp)
+				})
+			})
+			raw, err := net.Dial("tcp", addr)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer func() { _ = raw.Close() }()
+			br := bufio.NewReaderSize(idleConn761{raw}, 64<<10)
+			body := bytes.Repeat([]byte("b"), bodySize)
+			for i := range requests {
+				head := fmt.Sprintf("POST /upload HTTP/1.1\r\nHost: x\r\nContent-Length: %d\r\n\r\n", bodySize)
+				if _, err := io.WriteString(raw, head+string(body[:1024])); err != nil {
+					t.Fatalf("%s request %d: %v", e.name, i+1, err)
+				}
+				time.Sleep(2 * time.Millisecond) // the rest in a later read
+				if _, err := raw.Write(body[1024:]); err != nil {
+					t.Fatalf("%s request %d: %v", e.name, i+1, err)
+				}
+				r, err := http.ReadResponse(br, nil)
+				if err != nil {
+					t.Fatalf("%s: no answer to request %d of %d on the connection (%s)", e.name, i+1, requests, describeReadEnd761(err))
+				}
+				got, err := io.ReadAll(r.Body)
+				_ = r.Body.Close()
+				if err != nil || r.StatusCode != http.StatusOK || !bytes.Equal(got, resp) {
+					t.Fatalf("%s: request %d answered %d with %d of %d bytes (%s)", e.name, i+1, r.StatusCode, len(got), respSize, describeReadEnd761(err))
+				}
+			}
+		})
+	}
+}
+
+// TestPipelinedResponsesKeepTheirOrder pins celeris#802: a client that
+// pipelines requests must get the responses in request order. The H1 response
+// adapter hands a body of 8 KiB or more to epoll and io_uring as a zero-copy
+// slice (and a file of 16 KiB or more to epoll as a sendfile), which the
+// engine stages outside its write buffer and sends after it; a response
+// written behind it in the same flush was appended to the write buffer and
+// went out first. The requests mix large bodies, small bodies and files, in
+// one packet, and every response is compared byte for byte, in order. The
+// total stays under the write cap, so the back-pressure refusal of
+// celeris#761 does not apply.
+func TestPipelinedResponsesKeepTheirOrder(t *testing.T) {
+	bodies := bodies761(64, 16<<10, 1<<20)
+	dir := t.TempDir()
+	fileBody := bodies761(256 << 10)[256<<10]
+	if err := os.WriteFile(filepath.Join(dir, "f"), fileBody, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	type req struct {
+		target string
+		want   []byte
+	}
+	seq := []req{
+		{"/big?n=1048576", bodies[1<<20]},
+		{"/big?n=64", bodies[64]},
+		{"/file", fileBody},
+		{"/big?n=16384", bodies[16<<10]},
+		{"/file", fileBody},
+		{"/big?n=1048576", bodies[1<<20]},
+		{"/ping", []byte("ok")},
+	}
+	shapes := []struct {
+		name                    string
+		asyncServer, asyncRoute bool
+	}{{"sync", false, false}, {"async-loop", true, false}, {"async-route", false, true}}
+	for _, e := range engines761 {
+		for _, sh := range shapes {
+			t.Run(e.name+"/"+sh.name, func(t *testing.T) {
+				addr := startServer761(t, e.eng, sh.asyncServer, func(s *celeris.Server) {
+					big := s.GET("/big", func(c *celeris.Context) error {
+						n, _ := strconv.Atoi(c.Query("n"))
+						return c.Blob(http.StatusOK, "application/octet-stream", bodies[n])
+					})
+					file := s.GET("/file", func(c *celeris.Context) error {
+						return c.File(filepath.Join(dir, "f"))
+					})
+					if sh.asyncRoute {
+						big.Async()
+						file.Async()
+					}
+				})
+				raw, err := net.Dial("tcp", addr)
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer func() { _ = raw.Close() }()
+				var batch bytes.Buffer
+				for _, r := range seq {
+					fmt.Fprintf(&batch, "GET %s HTTP/1.1\r\nHost: x\r\n\r\n", r.target)
+				}
+				if _, err := raw.Write(batch.Bytes()); err != nil {
+					t.Fatal(err)
+				}
+				br := bufio.NewReaderSize(idleConn761{raw}, 64<<10)
+				for i, r := range seq {
+					resp, err := http.ReadResponse(br, nil)
+					if err != nil {
+						t.Fatalf("%s/%s: response %d of %d (%s): %s", e.name, sh.name, i+1, len(seq), r.target, describeReadEnd761(err))
+					}
+					got, err := io.ReadAll(resp.Body)
+					_ = resp.Body.Close()
+					if err != nil || resp.StatusCode != http.StatusOK || !bytes.Equal(got, r.want) {
+						t.Fatalf("%s/%s: response %d of %d (%s): status %d, %d bytes (want %d), equal=%v, %s",
+							e.name, sh.name, i+1, len(seq), r.target, resp.StatusCode, len(got), len(r.want), bytes.Equal(got, r.want), describeReadEnd761(err))
+					}
+				}
+			})
+		}
+	}
+}
+
+// TestBackloggedPeerIsClosed is the other side of the cap: a peer that sends
+// requests without reading the responses must not make the engine buffer
+// without bound, nor be left waiting. Four requests for 3 MiB each, in one
+// packet, to a client that reads nothing for a while: the native engines stage
+// responses until the backlog they find is over the cap, then refuse the
+// next write and close the connection once what was staged has gone out. The
+// client must get whole responses, in order, followed by either the rest or
+// EOF; before celeris#761 the refused response was dropped and the client
+// waited for it.
+func TestBackloggedPeerIsClosed(t *testing.T) {
+	const n = 3 << 20
+	const requests = 4
+	body := bodies761(n)[n]
+	shapes := []struct {
+		name                    string
+		asyncServer, asyncRoute bool
+	}{{"sync", false, false}, {"async-loop", true, false}, {"async-route", false, true}}
+	for _, e := range engines761 {
+		for _, sh := range shapes {
+			t.Run(e.name+"/"+sh.name, func(t *testing.T) {
+				addr := startServer761(t, e.eng, sh.asyncServer, func(s *celeris.Server) {
+					r := s.GET("/big", func(c *celeris.Context) error {
+						return c.Blob(http.StatusOK, "application/octet-stream", body)
+					})
+					if sh.asyncRoute {
+						r.Async()
+					}
+				})
+				raw, err := net.Dial("tcp", addr)
+				if err != nil {
+					t.Fatal(err)
+				}
+				defer func() { _ = raw.Close() }()
+				_ = raw.(*net.TCPConn).SetReadBuffer(64 << 10)
+				if _, err := io.WriteString(raw, strings.Repeat("GET /big HTTP/1.1\r\nHost: x\r\n\r\n", requests)); err != nil {
+					t.Fatal(err)
+				}
+				time.Sleep(300 * time.Millisecond) // the server stages what it will before the client reads
+				br := bufio.NewReaderSize(idleConn761{raw}, 64<<10)
+				whole := 0
+				for whole < requests {
+					resp, err := http.ReadResponse(br, nil)
+					if err != nil {
+						if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+							break
+						}
+						t.Fatalf("%s/%s: after %d whole responses: %s", e.name, sh.name, whole, describeReadEnd761(err))
+					}
+					got, err := io.ReadAll(resp.Body)
+					_ = resp.Body.Close()
+					if err != nil || !bytes.Equal(got, body) {
+						t.Fatalf("%s/%s: response %d: %d of %d bytes, equal=%v, then %s", e.name, sh.name, whole+1, len(got), n, bytes.Equal(got, body), describeReadEnd761(err))
+					}
+					whole++
+				}
+				t.Logf("%s/%s: %d of %d responses, then the close", e.name, sh.name, whole, requests)
+				if whole == 0 {
+					t.Fatalf("%s/%s: no response at all", e.name, sh.name)
+				}
+			})
+		}
+	}
+}
+
+// startServer761 starts a server with routes and waits until it answers
+// /ping. An io_uring start that fails only with ENOMEM is retried, with a new
+// server, for up to 30 s: the kernel charges ring memory to RLIMIT_MEMLOCK
+// per UID and gives it back some milliseconds after a ring closes, so at the
+// CI runner's 8 MiB a start made right after the previous server stopped, or
+// while another package's test binary holds rings, can fail although nothing
+// leaked (see startC714DetachServer).
+func startServer761(t *testing.T, eng celeris.EngineType, asyncServer bool, routes func(*celeris.Server)) string {
+	t.Helper()
+	retryUntil := time.Now().Add(30 * time.Second)
+	for tries := 1; ; tries++ {
+		ln, err := net.Listen("tcp", "127.0.0.1:0")
+		if err != nil {
+			t.Fatal(err)
+		}
+		addr := ln.Addr().String()
+		_ = ln.Close()
+		s := celeris.New(celeris.Config{Engine: eng, Addr: addr, AsyncHandlers: asyncServer})
+		s.GET("/ping", func(c *celeris.Context) error { return c.String(http.StatusOK, "ok") })
+		routes(s)
+		startDone := make(chan error, 1)
+		go func() { startDone <- s.Start() }()
+		err = waitReady761(addr, startDone)
+		if err == nil {
+			if tries > 1 {
+				t.Logf("server start retried on ring ENOMEM: %d tries", tries)
+			}
+			t.Cleanup(func() {
+				ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+				defer cancel()
+				_ = s.Shutdown(ctx)
+				select {
+				case <-startDone:
+				case <-time.After(15 * time.Second):
+					t.Errorf("Start did not return within 15s of Shutdown")
+				}
+			})
+			return addr
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		_ = s.Shutdown(ctx)
+		cancel()
+		if strings.Contains(err.Error(), "cannot allocate memory") && time.Now().Before(retryUntil) {
+			time.Sleep(20 * time.Millisecond)
+			continue
+		}
+		t.Fatalf("server did not start: %v", err)
+	}
+}
+
+// waitReady761 polls /ping until it answers 200, or returns the error the
+// start returned first.
+func waitReady761(addr string, startDone <-chan error) error {
+	probe := &http.Client{Timeout: 300 * time.Millisecond}
+	for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); {
+		select {
+		case err := <-startDone:
+			if err == nil {
+				err = errors.New("Start returned nil before the server was ready")
+			}
+			return err
+		default:
+		}
+		if resp, err := probe.Get("http://" + addr + "/ping"); err == nil {
+			_, _ = io.Copy(io.Discard, resp.Body)
+			_ = resp.Body.Close()
+			if resp.StatusCode == http.StatusOK {
+				return nil
+			}
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	return fmt.Errorf("no answer on /ping at %s within 15s", addr)
+}
+
+// idleConn761 gives every Read a fresh deadline, so a response that keeps
+// arriving is never cut off and one that stops is given up after idleCap761.
+type idleConn761 struct{ net.Conn }
+
+func (c idleConn761) Read(p []byte) (int, error) {
+	_ = c.SetReadDeadline(time.Now().Add(idleCap761))
+	return c.Conn.Read(p)
+}
+
+// countingReader761 counts the bytes the client has received, headers
+// included, so a failure says how far the response got.
+type countingReader761 struct {
+	r io.Reader
+	n int
+}
+
+func (c *countingReader761) Read(p []byte) (int, error) {
+	n, err := c.r.Read(p)
+	c.n += n
+	return n, err
+}
+
+func describeReadEnd761(err error) string {
+	var ne net.Error
+	switch {
+	case err == nil:
+		return "no error"
+	case errors.As(err, &ne) && ne.Timeout():
+		return fmt.Sprintf("no byte for %v, connection still open", idleCap761)
+	case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF):
+		return "EOF"
+	}
+	return err.Error()
+}
+
+// checkH1Response761 sends one GET for target on a new connection and checks
+// that the whole of want arrives. With keepAlive the connection must then
+// answer /ping; without it, the request says Connection: close and the body
+// must be followed by EOF.
+func checkH1Response761(t *testing.T, addr, target, desc string, want []byte, keepAlive bool) {
+	t.Helper()
+	n := len(want)
+	raw, err := net.Dial("tcp", addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = raw.Close() }()
+	cr := &countingReader761{r: idleConn761{raw}}
+	br := bufio.NewReaderSize(cr, 64<<10)
+	connHdr := ""
+	if !keepAlive {
+		connHdr = "Connection: close\r\n"
+	}
+	if _, err := fmt.Fprintf(raw, "GET %s HTTP/1.1\r\nHost: x\r\n%s\r\n", target, connHdr); err != nil {
+		t.Fatal(err)
+	}
+	resp, err := http.ReadResponse(br, nil)
+	if err != nil {
+		t.Fatalf("%s: no response head (%d bytes received, %s)", desc, cr.n, describeReadEnd761(err))
+	}
+	if resp.StatusCode != http.StatusOK || resp.ContentLength != int64(n) {
+		t.Fatalf("%s: status %d, Content-Length %d", desc, resp.StatusCode, resp.ContentLength)
+	}
+	got := make([]byte, n)
+	m, err := io.ReadFull(resp.Body, got)
+	if err != nil {
+		t.Fatalf("%s: received %d of %d body bytes (%d bytes in all), then %s",
+			desc, m, n, cr.n, describeReadEnd761(err))
+	}
+	if !bytes.Equal(got, want) {
+		i := 0
+		for i < n && got[i] == want[i] {
+			i++
+		}
+		t.Fatalf("%s: all bytes arrived but differ from byte %d on", desc, i)
+	}
+	_ = resp.Body.Close()
+
+	if !keepAlive {
+		// The body must be followed by the close, not by more bytes and not
+		// by an open connection.
+		extra, err := io.Copy(io.Discard, br)
+		if err != nil || extra != 0 {
+			t.Fatalf("%s: %d bytes after the body, then %s", desc, extra, describeReadEnd761(err))
+		}
+		return
+	}
+	// The connection must still carry the next request.
+	if _, err := io.WriteString(raw, "GET /ping HTTP/1.1\r\nHost: x\r\n\r\n"); err != nil {
+		t.Fatalf("%s: the connection took no next request: %v", desc, err)
+	}
+	next, err := http.ReadResponse(br, nil)
+	if err != nil {
+		t.Fatalf("%s: no answer to the next request on the connection (%s)", desc, describeReadEnd761(err))
+	}
+	b, err := io.ReadAll(next.Body)
+	_ = next.Body.Close()
+	if err != nil || next.StatusCode != http.StatusOK || string(b) != "ok" {
+		t.Fatalf("%s: next request on the connection answered %d %q (%v)", desc, next.StatusCode, b, err)
+	}
+}

From 98ba46d38878ae4c5db4232a0bff02e31e6d90ff Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 28 Sep 2026 11:43:36 +0200
Subject: [PATCH 2/7] fix(epoll, iouring): send a zero-copy body before the
 handler can reuse it, hold the H1 cap per request, give a closing io_uring
 conn WriteTimeout (celeris#817, celeris#761)

Round 2 of #805, from its review.

- The H1 zero-copy body writer kept a reference to the handler's slice past
  the write: c.JSON puts its buffer back in a pool at once, and a buffered
  response's body lives on a reused Context, so a response went out with the
  next pipelined request's body or another connection's (celeris#817).
  epoll's writer now makes the writev in the call and copies what the kernel
  did not take; io_uring installs none (its kernel reads a WRITEV's iovec at
  the next submit), so the adapter copies the body. SetWriteBodyFn's
  contract says so.
- The 4 MiB H1 cap is held per request, not per write
  (H1State.WriteBacklogged): a request that finds the unsent responses over
  it is not served (ErrWriteBacklog) and the conn is closed once they have
  gone out; a response, however large and in however many writes (a
  StreamWriter's chunks), is staged whole. writeCap/sendCap keep a per-write
  limit for HTTP/2 and detached conns only.
- io_uring's closing drain gives a conn the longer of 5 s and WriteTimeout
  without progress, restamped on every send that makes some: a response
  larger than the socket buffers answering Connection: close was cut 5 s
  after the close.
- A failed read of a staged file closes the conn instead of leaving a
  header block with no body.
- Tests: pipelined and concurrent c.JSON bodies (their own, not another
  connection's), an 8 MiB StreamWriter response, the closing drain's bound
  and restamp; TestBackloggedPeerIsClosed now fails if the cap is gone; the
  HTTP/2 and c.File sizes shrink to 16 MiB for CI time; the gated
  backpressure test asserts the server's close.
---
 engine/epoll/backpressure_test.go             |  32 +--
 engine/epoll/conn.go                          |  52 ++--
 engine/epoll/loop.go                          |  72 ++---
 engine/epoll/write_hooks_bench_linux_test.go  |  33 ++-
 engine/epoll/writer.go                        |  12 +-
 engine/iouring/closing_drain_test.go          |  55 ++++
 engine/iouring/conn.go                        |  38 ++-
 engine/iouring/worker.go                      | 116 ++++----
 .../iouring/write_hooks_bench_linux_test.go   |  19 +-
 internal/conn/h1.go                           |  37 ++-
 large_response_linux_test.go                  |  78 +++++-
 response_body_ownership_linux_test.go         | 251 ++++++++++++++++++
 12 files changed, 605 insertions(+), 190 deletions(-)
 create mode 100644 response_body_ownership_linux_test.go

diff --git a/engine/epoll/backpressure_test.go b/engine/epoll/backpressure_test.go
index 2ae89adb..99c44ed6 100644
--- a/engine/epoll/backpressure_test.go
+++ b/engine/epoll/backpressure_test.go
@@ -5,7 +5,6 @@ package epoll
 import (
 	"context"
 	"fmt"
-	"io"
 	"net"
 	"os"
 	"testing"
@@ -40,9 +39,13 @@ func (h *bigResponseHandler) HandleStream(_ context.Context, s *stream.Stream) e
 // end-to-end assertion of the timeout close path: a slow consumer must
 // not hold server-side buffering for longer than the timeouts allow.
 // Since celeris#761 a single response larger than maxPendingBytes is
-// staged whole rather than closed at once (the cap bounds the backlog a
-// write finds, not the write), so it is the timeout sweep, here
-// ReadTimeout, that closes a consumer that never reads it.
+// staged whole rather than closed at once (the cap is held per request,
+// not per write), so it is the timeout sweep, here ReadTimeout, that
+// closes a consumer that never reads it. The assertion is the server's:
+// its conn count goes back to zero. What the client sees after the close is
+// the kernel sending the send queue it still holds, into a 4 KiB window
+// that the zero-window probes' backoff opens seconds apart, so a client
+// deadline measured that backoff, not the engine.
 //
 // The test is marked t.Skip by default because reliably triggering
 // the backpressure path on loopback TCP is hard — Linux's
@@ -129,21 +132,12 @@ func TestWriteBufBackpressureClosesSlowConsumer(t *testing.T) {
 	// closeConn.
 	time.Sleep(500 * time.Millisecond)
 
-	// Now the server should have closed us. Read should drain
-	// whatever the kernel absorbed + hit EOF.
-	_ = tcp.SetReadDeadline(time.Now().Add(3 * time.Second))
-	drain := make([]byte, 64<<10)
-	for {
-		_, rerr := tcp.Read(drain)
-		if rerr != nil {
-			if rerr == io.EOF || rerr == io.ErrUnexpectedEOF {
-				return
-			}
-			// ECONNRESET / broken pipe count as server-terminated.
-			if ne, ok := rerr.(net.Error); ok && ne.Timeout() {
-				t.Fatalf("conn stayed open past deadline; backpressure/timeout did not fire")
-			}
-			return
+	// Now the server should have closed the conn (the sweep runs at
+	// least every few hundred ms under these timeouts).
+	for end := time.Now().Add(3 * time.Second); e.Metrics().ActiveConnections != 0; {
+		if time.Now().After(end) {
+			t.Fatalf("the server still holds %d conn(s) 3.5s after the request; the timeout close did not fire", e.Metrics().ActiveConnections)
 		}
+		time.Sleep(20 * time.Millisecond)
 	}
 }
diff --git a/engine/epoll/conn.go b/engine/epoll/conn.go
index 5832b6df..cb11d0ac 100644
--- a/engine/epoll/conn.go
+++ b/engine/epoll/conn.go
@@ -5,6 +5,7 @@ package epoll
 
 import (
 	"context"
+	"math"
 	"sync"
 	"sync/atomic"
 
@@ -13,8 +14,11 @@ import (
 )
 
 // maxPendingBytes is the per-connection back-pressure limit for pending
-// writes on H1/H2 connections. Intentionally small (4 MiB) so a stalled
-// peer cannot fill server memory with un-ACKed responses.
+// writes on H1 connections. Intentionally small (4 MiB) so a stalled
+// peer cannot fill server memory with un-ACKed responses. It is held per
+// request, not per write (celeris#761): a request that finds more than this
+// unsent is not served and the conn is closed (conn.H1State.WriteBacklogged),
+// while a response, however large, is staged whole.
 //
 // maxPendingBytesDetached is the per-connection limit once the
 // connection is detached (WebSocket / SSE). Detached middleware owns
@@ -66,16 +70,15 @@ func trimPooledBuf(b []byte) []byte {
 	return b[:0]
 }
 
-// writeCap returns the effective back-pressure limit for cs, accounting
-// for whether the connection is detached. Async-mode HTTP1 conns set
-// detachMu up front without being truly detached; they keep the H1/H2
-// limit so a stalled peer cannot balloon per-conn memory to 64 MiB.
-//
-// The limit bounds the backlog a write finds, not the write: a write is
-// refused only when the bytes still queued before it (cs.pendingBytes) are
-// over the limit, which is a peer that stopped reading while it keeps
-// sending requests. One response larger than the limit is not such a
-// backlog and is staged whole; before celeris#761 its body was dropped.
+// writeCap returns the per-write back-pressure limit for cs: a write is
+// refused when the bytes still queued before it (cs.pendingBytes) are over
+// it. An HTTP/2 conn has maxPendingBytesH2 and a truly-detached one (WS/SSE)
+// maxPendingBytesDetached. An HTTP/1 conn has none: its writes are its
+// handlers' responses, and a limit per write cut a response off mid-body, a
+// large body or the chunks of a StreamWriter (celeris#761). Its limit,
+// maxPendingBytes, is held per request instead (overBacklogH1), before the
+// handler runs. Async-mode HTTP1 conns set detachMu up front without being
+// truly detached; they are HTTP/1 conns here.
 func (cs *connState) writeCap() int {
 	if cs.h2State != nil {
 		return maxPendingBytesH2
@@ -83,7 +86,16 @@ func (cs *connState) writeCap() int {
 	if cs.detachMu != nil && cs.h1State != nil && cs.h1State.Detached.Load() {
 		return maxPendingBytesDetached
 	}
-	return maxPendingBytes
+	return math.MaxInt
+}
+
+// overBacklogH1 is the HTTP/1 back-pressure limit (conn.H1State.WriteBacklogged,
+// celeris#761): whether the responses cs still has unsent, which the write
+// hooks count in pendingBytes, are over maxPendingBytes, i.e. its client
+// stopped reading while it kept sending requests. The next request is then
+// not served, and the conn is closed once what is queued has gone out.
+func (cs *connState) overBacklogH1() bool {
+	return cs.pendingBytes > maxPendingBytes
 }
 
 // connState holds per-connection state for the epoll engine.
@@ -135,12 +147,14 @@ type connState struct {
 	// error, a refused write; celeris#761): the close it defers is the same.
 	peerClosed bool
 
-	// writeRefused records that a write hook refused bytes because the
-	// conn's backlog was already over writeCap (celeris#761). A refused
-	// write is never silent: the site that ran the handler closes the conn
-	// (closeWhenFlushed) instead of leaving the client waiting for bytes that
-	// will not come. Sticky until release. Written under detachMu when the
-	// conn has one, like the buffers it guards.
+	// writeRefused records that response bytes were lost (celeris#761): a
+	// write hook refused them because the conn's backlog was already over
+	// writeCap, or the write the zero-copy body hook made failed, or a
+	// staged file could not be read. It is never silent: the site that ran
+	// the handler closes the conn (closeWhenFlushed) instead of leaving the
+	// client waiting for bytes that will not come. Sticky until release.
+	// Written under detachMu when the conn has one, like the buffers it
+	// guards.
 	writeRefused bool
 
 	// drainDeadline bounds how long checkTimeouts defers the idle-deadline
diff --git a/engine/epoll/loop.go b/engine/epoll/loop.go
index 70a1d8c3..b1fe4f07 100644
--- a/engine/epoll/loop.go
+++ b/engine/epoll/loop.go
@@ -1895,11 +1895,16 @@ func (l *Loop) initProtocol(cs *connState) {
 		if !l.cfg.EnableH2Upgrade {
 			cs.h1State.DisableH2CDetect()
 		}
+		// Back-pressure for HTTP/1 is held per request (celeris#761): a
+		// request that finds the conn's unsent responses over the limit is
+		// not served, and the conn is closed once they have gone out.
+		cs.h1State.WriteBacklogged = cs.overBacklogH1
 		// Scatter-gather body writer: handler hands large bodies to the
-		// engine as a zero-copy slice; flushWrites emits writev(2) with
-		// [headers, body] so we save the respBuf → writeBuf memcpy.
-		// Disabled in async mode because cs.bodyBuf access would race
-		// with the dispatch goroutine without a mutex.
+		// engine as a zero-copy slice, which it writes at once with
+		// writev(2) of [headers, body], saving the respBuf → writeBuf
+		// memcpy (see makeWriteBodyFn). Disabled in async mode because
+		// cs.bodyBuf access would race with the dispatch goroutine
+		// without a mutex.
 		if !l.async {
 			cs.h1State.SetWriteBodyFn(l.makeWriteBodyFn(cs))
 			// Zero-copy sendfile(2) for large static-file responses: the
@@ -2191,18 +2196,18 @@ func (l *Loop) switchToH2Local(cs *connState, writeFn func([]byte)) error {
 
 func (l *Loop) makeWriteFn(cs *connState) func([]byte) {
 	return func(data []byte) {
-		// Back-pressure: refuse the write only when the backlog before it is
-		// over the cap (see writeCap), and say so, so the conn is closed
-		// rather than left waiting for bytes that will not come (celeris#761).
-		// pendingBytes already counts a staged bodyBuf.
+		// Back-pressure (an HTTP/2 or detached conn; see writeCap): refuse
+		// the write only when the backlog before it is over the cap, and
+		// say so, so the conn is closed rather than left waiting for bytes
+		// that will not come (celeris#761).
 		if cs.pendingBytes > cs.writeCap() {
 			cs.writeRefused = true
 			return
 		}
-		// A zero-copy body or sendfile is staged, and the flush sends
-		// writeBuf before them: bytes written after them (the next
-		// pipelined response) would go out ahead (celeris#802).
-		if (cs.bodyBuf != nil || cs.sendfile != nil) && !unstage(cs) {
+		// A sendfile is staged, and the flush sends writeBuf before it:
+		// bytes written after it (the next pipelined response) would go
+		// out ahead (celeris#802).
+		if cs.sendfile != nil && !unstage(cs) {
 			cs.writeRefused = true
 			return
 		}
@@ -2214,30 +2219,33 @@ func (l *Loop) makeWriteFn(cs *connState) func([]byte) {
 	}
 }
 
-// makeWriteBodyFn stages a zero-copy body slice for scatter-gather
-// writev at flush time. The body is NOT copied — it must remain valid
-// and unmutated until flushWrites drains it. Used by the H1 response
-// adapter for bodies ≥ 8 KiB to skip the respBuf → writeBuf memcpy.
+// makeWriteBodyFn returns the zero-copy body writer of the H1 response
+// adapter, for bodies of 8 KiB or more: the body goes to the kernel straight
+// from the handler's memory, one writev(2) of [writeBuf, body], skipping the
+// respBuf → writeBuf memcpy. The write is made here, in the call, and what
+// the kernel does not take is copied into writeBuf before it returns: the
+// body belongs to the handler, which may reuse it as soon as its write
+// returns. c.JSON puts its buffer back in a pool at once, where the next
+// handler to encode takes it, and a buffered response's body lives on a
+// pooled Context; a body staged for a flush after the handler went out as
+// the next request's bytes, or another connection's (celeris#817). The
+// syscall is the one the flush after the handler would have made. Runs on
+// the loop thread: the hook is installed only in sync mode.
 func (l *Loop) makeWriteBodyFn(cs *connState) func([]byte) {
 	return func(body []byte) {
-		// The body itself is not held against the cap: a response larger
-		// than the cap is staged whole, and only a backlog already over it
-		// refuses the write (celeris#761; see writeCap and makeWriteFn).
-		if cs.pendingBytes > cs.writeCap() {
+		// A staged sendfile goes out before this body (celeris#802).
+		if cs.sendfile != nil && !unstage(cs) {
 			cs.writeRefused = true
 			return
 		}
-		// A second large body in the same flush (a pipelined response), or
-		// one behind a staged sendfile: there is one writev body slot, so
-		// what is staged moves into writeBuf, where it stays ahead of this
-		// body, and this body takes the slot. Copying this one into writeBuf
-		// instead sent it before the staged body (celeris#802).
-		if (cs.bodyBuf != nil || cs.sendfile != nil) && !unstage(cs) {
+		cs.bodyBuf = body
+		err := l.flushWrites(cs, true)
+		unstage(cs) // what the kernel did not take; no sendfile is staged
+		cs.pendingBytes = csPendingBytes(cs)
+		if err != nil {
+			// The response is lost with the conn: close it.
 			cs.writeRefused = true
-			return
 		}
-		cs.bodyBuf = body
-		cs.pendingBytes += len(body)
 	}
 }
 
@@ -2266,6 +2274,7 @@ func (l *Loop) makeWriteBodyFn(cs *connState) func([]byte) {
 func (l *Loop) makeSendFileFn(cs *connState) func(header []byte, file *os.File, offset, length int64) error {
 	return func(header []byte, file *os.File, offset, length int64) error {
 		if cs.sendfile != nil && !unstage(cs) {
+			cs.writeRefused = true // the staged response is cut: close
 			return errUnstageSendfile
 		}
 		dupfd, err := unix.Dup(int(file.Fd()))
@@ -2292,8 +2301,9 @@ func (l *Loop) makeSendFileFn(cs *connState) func(header []byte, file *os.File,
 // the response is delivered byte-exact, just without the syscall savings.
 func bufferedFileFallback(cs *connState, header []byte, file *os.File, offset, length int64) error {
 	// This response is appended to writeBuf, which goes out before a
-	// staged body or sendfile (celeris#802).
-	if (cs.bodyBuf != nil || cs.sendfile != nil) && !unstage(cs) {
+	// staged sendfile (celeris#802).
+	if cs.sendfile != nil && !unstage(cs) {
+		cs.writeRefused = true // the staged response is cut: close
 		return errUnstageSendfile
 	}
 	if length <= 0 {
diff --git a/engine/epoll/write_hooks_bench_linux_test.go b/engine/epoll/write_hooks_bench_linux_test.go
index 914f3922..066082de 100644
--- a/engine/epoll/write_hooks_bench_linux_test.go
+++ b/engine/epoll/write_hooks_bench_linux_test.go
@@ -2,40 +2,53 @@
 
 package epoll
 
-import "testing"
+import (
+	"os"
+	"testing"
+)
 
 // BenchmarkWriteHooks measures the per-response cost of the write hooks the
 // H1 response adapter calls (makeWriteFn, makeWriteBodyFn), whose
-// back-pressure check celeris#761 changed: a header block and a small body
-// through writeFn, and a header block and a 16 KiB zero-copy body through
-// writeFn + writeBodyFn. The buffers are reset as a completed flush leaves
-// them.
+// back-pressure check celeris#761 changed, and of the flush that sends the
+// response: a header block and a small body through writeFn, and a header
+// block and a 16 KiB zero-copy body through writeFn + writeBodyFn, which
+// since celeris#817 makes the writev(2) itself. The conn writes to /dev/null,
+// so the syscall is in every arm and the flush is complete.
 func BenchmarkWriteHooks(b *testing.B) {
 	hdr := make([]byte, 121)
 	small := make([]byte, 64)
 	large := make([]byte, 16<<10)
+	devNull, err := os.OpenFile(os.DevNull, os.O_WRONLY, 0)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer func() { _ = devNull.Close() }()
+	fd := int(devNull.Fd())
 	b.Run("small", func(b *testing.B) {
 		l := &Loop{}
-		cs := &connState{}
+		cs := &connState{fd: fd}
 		w := l.makeWriteFn(cs)
 		b.ReportAllocs()
 		for b.Loop() {
 			w(hdr)
 			w(small)
-			cs.writeBuf = cs.writeBuf[:0]
+			if err := l.flushWrites(cs, true); err != nil {
+				b.Fatal(err)
+			}
 			cs.pendingBytes = 0
 		}
 	})
 	b.Run("zero-copy-body", func(b *testing.B) {
 		l := &Loop{}
-		cs := &connState{}
+		cs := &connState{fd: fd}
 		w, wb := l.makeWriteFn(cs), l.makeWriteBodyFn(cs)
 		b.ReportAllocs()
 		for b.Loop() {
 			w(hdr)
 			wb(large)
-			cs.writeBuf = cs.writeBuf[:0]
-			cs.bodyBuf = nil
+			if err := l.flushWrites(cs, true); err != nil {
+				b.Fatal(err)
+			}
 			cs.pendingBytes = 0
 		}
 	})
diff --git a/engine/epoll/writer.go b/engine/epoll/writer.go
index dbf7489e..4ec70bc8 100644
--- a/engine/epoll/writer.go
+++ b/engine/epoll/writer.go
@@ -119,11 +119,13 @@ var errUnstageSendfile = errors.New("celeris: epoll: reading a staged sendfile r
 // (bodyBuf) and then a sendfile response, into writeBuf, in the order the
 // flush sends them (writeBuf, bodyBuf, sendfile), so that bytes written next
 // go out after them: every writer appends to writeBuf, which the flush sends
-// first (celeris#802). The body is copied; the sendfile's unsent header and
-// file bytes are read in and its descriptor closed. Only a write that
-// follows a large body or a file response before the flush pays for it: the
-// next pipelined response. Reports false if the file could not be read; the
-// conn must then be closed, its response being lost either way.
+// first (celeris#802). The body is copied: that is what the kernel did not
+// take of the writev makeWriteBodyFn makes, which never leaves a body staged
+// past its call (celeris#817). The sendfile's unsent header and file bytes
+// are read in and its descriptor closed; only a write that follows a file
+// response before the flush pays for that: the next pipelined response.
+// Reports false if the file could not be read; the conn must then be closed,
+// its response being lost either way.
 func unstage(cs *connState) bool {
 	if cs.bodyBuf != nil {
 		cs.writeBuf = append(cs.writeBuf, cs.bodyBuf...)
diff --git a/engine/iouring/closing_drain_test.go b/engine/iouring/closing_drain_test.go
index a272b0f8..7edf509b 100644
--- a/engine/iouring/closing_drain_test.go
+++ b/engine/iouring/closing_drain_test.go
@@ -129,3 +129,58 @@ func TestCheckTimeoutsLetsFreshClosingConnDrain(t *testing.T) {
 		t.Errorf("liveConns = %d, want 1", got)
 	}
 }
+
+// TestCheckTimeoutsGivesClosingConnItsWriteTimeout guards celeris#761 on the
+// closing drain: a closing conn can carry a whole response (one larger than
+// the socket buffers, answering Connection: close), and the sweep reaped it
+// closingDrainTimeoutNanos (5 s) after the close, where the same response on
+// a keep-alive conn gets WriteTimeout: a client that paused reading it for 6
+// s lost its tail. The bound is now the longer of the two.
+func TestCheckTimeoutsGivesClosingConnItsWriteTimeout(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		idle  time.Duration // how long the peer has taken nothing
+		wt    time.Duration // cfg.WriteTimeout (0: disabled)
+		reaps bool
+	}{
+		{"6s-idle-writetimeout-60s", 6 * time.Second, 60 * time.Second, false},
+		{"61s-idle-writetimeout-60s", 61 * time.Second, 60 * time.Second, true},
+		{"6s-idle-writetimeout-off", 6 * time.Second, 0, true},
+		{"4s-idle-writetimeout-200ms", 4 * time.Second, 200 * time.Millisecond, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			ring := newTestRing(t)
+			w, cs := newClosingDrainWorker(t, ring)
+			w.cfg.WriteTimeout = tc.wt
+			fd := cs.fd
+			cs.closing = true
+			cs.lastActivity = time.Now().Add(-tc.idle).UnixNano()
+			w.checkTimeouts()
+			if reaped := w.conns[fd] == nil; reaped != tc.reaps {
+				t.Fatalf("closing conn idle %v with WriteTimeout %v: reaped=%v, want %v", tc.idle, tc.wt, reaped, tc.reaps)
+			}
+		})
+	}
+}
+
+// TestCompleteSendRestampsClosingDrain guards the other half: the closing
+// drain's clock measures how long the peer has taken nothing, so a send that
+// makes progress restarts it. Without that, a client that reads a large
+// response steadily, but for longer than the bound, is cut off.
+func TestCompleteSendRestampsClosingDrain(t *testing.T) {
+	ring := newTestRing(t)
+	w, cs := newClosingDrainWorker(t, ring)
+	fd := cs.fd
+	cs.closing = true
+	stale := time.Now().Add(-time.Hour).UnixNano()
+	cs.lastActivity = stale
+	cs.sendBuf = make([]byte, 64<<10) // a response tail, 1 KiB of it sent below
+	now := time.Now().UnixNano()
+	w.completeSend(cs, fd, 1<<10, now, false)
+	if w.conns[fd] != cs {
+		t.Fatalf("a partial send closed the closing conn")
+	}
+	if cs.lastActivity != now {
+		t.Fatalf("a send that made progress left the closing drain's clock %v stale", time.Duration(now-cs.lastActivity))
+	}
+}
diff --git a/engine/iouring/conn.go b/engine/iouring/conn.go
index 5490a78c..7bd5f9b0 100644
--- a/engine/iouring/conn.go
+++ b/engine/iouring/conn.go
@@ -4,6 +4,7 @@ package iouring
 
 import (
 	"context"
+	"math"
 	"sync"
 	"sync/atomic"
 
@@ -12,9 +13,12 @@ import (
 )
 
 // maxSendQueueBytes is the per-connection back-pressure limit for
-// H1/H2 connections. When pending send data exceeds this, the
+// H1 connections. When pending send data exceeds this, the
 // connection is closed to prevent unbounded memory growth while a
-// slow peer stalls with un-ACKed responses.
+// slow peer stalls with un-ACKed responses. It is held per request, not
+// per write (celeris#761): a request that finds more than this unsent is not
+// served (conn.H1State.WriteBacklogged), while a response, however large, is
+// staged whole.
 //
 // maxSendQueueBytesDetached is the corresponding limit once a
 // connection is detached (WebSocket / SSE). Detached middleware owns
@@ -40,16 +44,15 @@ const (
 	maxPendingInputBytes = 4 << 20
 )
 
-// sendCap returns the effective back-pressure limit for cs, accounting
-// for whether the connection is detached. Async-mode HTTP1 conns set
-// detachMu up front without being truly detached; they keep the H1/H2
-// limit so a stalled peer cannot balloon per-conn memory to 64 MiB.
-//
-// The limit bounds the backlog a write finds, not the write: a write is
-// refused only when the bytes still queued before it are over the limit,
-// which is a peer that stopped reading while it keeps sending requests. One
-// response larger than the limit is not such a backlog and is staged whole;
-// before celeris#761 its body was dropped.
+// sendCap returns the per-write back-pressure limit for cs: a write is
+// refused when the bytes still queued before it are over it. An HTTP/2 conn
+// has maxSendQueueBytesH2 and a truly-detached one (WS/SSE)
+// maxSendQueueBytesDetached. An HTTP/1 conn has none: its writes are its
+// handlers' responses, and a limit per write cut a response off mid-body, a
+// large body or the chunks of a StreamWriter (celeris#761). Its limit,
+// maxSendQueueBytes, is held per request instead (overBacklogH1), before the
+// handler runs. Async-mode HTTP1 conns set detachMu up front without being
+// truly detached; they are HTTP/1 conns here.
 func (cs *connState) sendCap() int {
 	if cs.h2State != nil {
 		return maxSendQueueBytesH2
@@ -57,7 +60,16 @@ func (cs *connState) sendCap() int {
 	if cs.detachMu != nil && cs.h1State != nil && cs.h1State.Detached.Load() {
 		return maxSendQueueBytesDetached
 	}
-	return maxSendQueueBytes
+	return math.MaxInt
+}
+
+// overBacklogH1 is the HTTP/1 back-pressure limit (conn.H1State.WriteBacklogged,
+// celeris#761): whether the responses cs still has unsent, queued or in a
+// SEND in flight, are over maxSendQueueBytes, i.e. its client stopped
+// reading while it kept sending requests. The next request is then not
+// served, and the conn is closed once what is queued has gone out.
+func (cs *connState) overBacklogH1() bool {
+	return len(cs.writeBuf)+len(cs.sendBuf) > maxSendQueueBytes
 }
 
 // iovec mirrors Linux struct iovec (16 bytes on 64-bit platforms).
diff --git a/engine/iouring/worker.go b/engine/iouring/worker.go
index b1e7ccc3..10f7ba15 100644
--- a/engine/iouring/worker.go
+++ b/engine/iouring/worker.go
@@ -280,12 +280,14 @@ const shutdownSendDrainNanos int64 = int64(250 * time.Millisecond)
 // backstops for an operation the kernel may never complete, and keeping them
 // on one scale means a wedged conn is fully reclaimed — fd here, connState
 // when the close-path ASYNC_CANCEL's terminal CQE lands — inside one such
-// window rather than two unrelated ones. It is deliberately NOT derived from
-// cfg.WriteTimeout: that governs a LIVE conn whose handler is still producing
-// bytes at its own pace, whereas a closing conn's queue is final (writeFn
-// no-ops once detachClosed is set, drainDetachQueue skips it, and handleRecv
-// drops incoming data on a closing conn), so this is not a throughput budget
-// but the point at which we conclude the peer will never take the bytes.
+// window rather than two unrelated ones. It is the floor of the bound, not
+// the bound: closingDrainBound extends it to cfg.WriteTimeout, since the
+// queue of a closing conn can be a whole response (celeris#761). Either way
+// it is not a throughput budget but the point at which we conclude the peer
+// will never take the bytes: the clock restarts whenever it takes some. A
+// closing conn's queue is final (writeFn no-ops once detachClosed is set,
+// drainDetachQueue skips it, and handleRecv drops incoming data on a closing
+// conn).
 //
 // Prompt completions are untouched: the SEND CQE normally lands within a loop
 // pass or two — microseconds on localhost, four orders of magnitude inside
@@ -293,6 +295,19 @@ const shutdownSendDrainNanos int64 = int64(250 * time.Millisecond)
 // path and never reaches the sweep.
 const closingDrainTimeoutNanos int64 = int64(5 * time.Second)
 
+// closingDrainBound is how long a closing conn may go without the peer
+// taking a byte of what is queued before checkTimeouts tears it down: the
+// longer of closingDrainTimeoutNanos and cfg.WriteTimeout, the bound a live
+// conn stalled on a write gets. completeSend restamps the clock on every send
+// that makes progress. A closing conn may carry a whole response, not only
+// its last bytes: a response larger than the socket buffers answering
+// Connection: close, or followed by a request error. A client that paused
+// reading it for more than 5 s, where it had WriteTimeout on a keep-alive
+// conn, lost its tail (celeris#761).
+func (w *Worker) closingDrainBound() int64 {
+	return max(closingDrainTimeoutNanos, int64(w.cfg.WriteTimeout))
+}
+
 // Worker is an io_uring event-loop worker pinned to a single OS thread.
 type Worker struct {
 	id         int
@@ -2524,17 +2539,19 @@ func (w *Worker) initProtocol(cs *connState) {
 		if !w.cfg.EnableH2Upgrade {
 			cs.h1State.DisableH2CDetect()
 		}
-		// Wire the scatter-gather body writer so the H1 response adapter
-		// can hand large bodies straight to the WRITEV path without the
-		// intermediate respBuf → cs.writeBuf memcpy. writeBodyFn stores
-		// the body slice on the connState; flushSend emits an iovec SQE.
-		// Only enabled on the synchronous inline-handler path: async
-		// mode handlers run on goroutines and need detachMu-guarded
-		// access to cs.bodyBuf, which the current writer does not
-		// provide. Async mode falls back to the copy path.
-		if !w.async {
-			cs.h1State.SetWriteBodyFn(w.makeWriteBodyFn(cs))
-		}
+		// Back-pressure for HTTP/1 is held per request (celeris#761): a
+		// request that finds the conn's unsent responses over the limit is
+		// not served, and the conn is closed once they have gone out.
+		cs.h1State.WriteBacklogged = cs.overBacklogH1
+		// No zero-copy body writer (SetWriteBodyFn): the H1 response
+		// adapter copies every body into writeBuf. A body the WRITEV path
+		// left in place was read by the kernel only at the next
+		// io_uring_enter, after the handlers of the other conns in the same
+		// completion batch had run, and the body belongs to the handler,
+		// which may reuse it as soon as its write returns: c.JSON puts its
+		// buffer back in a pool at once, where the next handler to encode
+		// takes it, so one conn's response went out with another's bytes
+		// (celeris#817).
 		cs.h1State.OnDetach = func() {
 			// Async mode may have already allocated detachMu in
 			// acquireConnState; reuse it so the async goroutine and the
@@ -3772,6 +3789,13 @@ func (w *Worker) completeSend(cs *connState, fd int, sent int, now int64, fromZC
 	// (covers regular SEND and the SEND_ZC NOTIF path, both of which
 	// reach completeSend with the byte count).
 	w.bytesWrittenBatch += uint64(sent)
+	if cs.closing && sent > 0 {
+		// The closing drain's clock (closingDrainBound) measures how long
+		// the peer has taken nothing, not how long the drain has run: a
+		// response larger than the socket buffers to a client that reads
+		// it steadily is not cut off (celeris#761).
+		cs.lastActivity = now
+	}
 	// celeris#591: the ring-send share of those bytes. Plain local add on
 	// the per-request send path — it is published with one atomic per
 	// event-loop iteration next to bytesWrittenBatch, never per request.
@@ -4901,23 +4925,15 @@ func (w *Worker) makeWriteFn(cs *connState) func([]byte) {
 		if cs.closing {
 			return
 		}
-		// Back-pressure: refuse the write only when the backlog before it
-		// is over the cap (see sendCap), and say so: the site that ran the
-		// handler then closes the conn (celeris#761), where it used to be
-		// left waiting for the refused bytes.
+		// Back-pressure (an HTTP/2 or detached conn; see sendCap): refuse
+		// the write only when the backlog before it is over the cap, and
+		// say so: the site that ran the handler then closes the conn
+		// (celeris#761), where it used to be left waiting for the refused
+		// bytes.
 		if len(cs.writeBuf)+len(cs.sendBuf)+len(cs.bodyBuf) > cs.sendCap() {
 			cs.writeRefused = true
 			return
 		}
-		if cs.bodyBuf != nil {
-			// A zero-copy body is staged, and flushSend sends writeBuf
-			// before it: bytes written after the body (the next pipelined
-			// response) would go out ahead of it. Move the body into
-			// writeBuf first, a copy paid only when a write follows a large
-			// body before the flush (celeris#802).
-			cs.writeBuf = append(cs.writeBuf, cs.bodyBuf...)
-			cs.bodyBuf = nil
-		}
 		// Append to writeBuf — no per-write allocation. The kernel holds
 		// sendBuf (not writeBuf), so appending here is safe.
 		// Don't markDirty here — handleRecv calls flushSend after the
@@ -4927,44 +4943,6 @@ func (w *Worker) makeWriteFn(cs *connState) func([]byte) {
 	}
 }
 
-// makeWriteBodyFn returns a closure that stores a zero-copy body reference
-// for scatter-gather send via IORING_OP_WRITEV. The body slice is NOT
-// copied — it must remain valid and unmutated until completeSend clears
-// cs.sendBody. For HTTP/1 response writers calling
-// writeBody(pre-computed-response) this is always safe; handlers that
-// generate bodies per-request should only call writeBody once with a
-// slice they do not mutate further.
-//
-// Saves one full body-sized memcpy per request: the traditional path
-// appends body into a.respBuf, then writeFn appends respBuf into
-// cs.writeBuf (two userspace copies of body bytes). With writeBody, the
-// body stays in the handler's memory; the engine issues a single WRITEV
-// SQE with iovec = [sendBuf (headers), body (alias)] and the kernel
-// does one copy directly to the socket buffer.
-func (w *Worker) makeWriteBodyFn(cs *connState) func([]byte) {
-	return func(body []byte) {
-		if cs.closing {
-			return
-		}
-		// The body itself is not held against the cap: a response larger
-		// than the cap is staged whole, and only a backlog already over it
-		// refuses the write (celeris#761; see sendCap and makeWriteFn).
-		if len(cs.writeBuf)+len(cs.sendBuf)+len(cs.bodyBuf) > cs.sendCap() {
-			cs.writeRefused = true
-			return
-		}
-		if cs.bodyBuf != nil {
-			// A second large body before the flush (a pipelined response):
-			// there is one iovec entry for a body, so the staged body moves
-			// into writeBuf, where it stays ahead of this one, and this one
-			// takes the entry. Copying this one into writeBuf instead sent
-			// it before the staged body (celeris#802).
-			cs.writeBuf = append(cs.writeBuf, cs.bodyBuf...)
-		}
-		cs.bodyBuf = body
-	}
-}
-
 // prepareH2Poll submits a single-shot POLL_ADD SQE on the H2 eventfd.
 // When handler goroutines write the eventfd, the CQE wakes the ring
 // event-driven, replacing the 100μs polling timeout.
@@ -5870,7 +5848,7 @@ func (w *Worker) checkTimeouts() {
 			w.rescueHold(cs)
 		}
 		if cs.closing {
-			if now-cs.lastActivity > closingDrainTimeoutNanos {
+			if now-cs.lastActivity > w.closingDrainBound() {
 				// Everything closeConn does before deferring (detach
 				// signalling, CloseH1, detachedCount) has already run, so
 				// finish exactly where completeSend would have. removeDirty
diff --git a/engine/iouring/write_hooks_bench_linux_test.go b/engine/iouring/write_hooks_bench_linux_test.go
index cd2ad567..383742a5 100644
--- a/engine/iouring/write_hooks_bench_linux_test.go
+++ b/engine/iouring/write_hooks_bench_linux_test.go
@@ -4,12 +4,12 @@ package iouring
 
 import "testing"
 
-// BenchmarkWriteHooks measures the per-response cost of the write hooks the
-// H1 response adapter calls (makeWriteFn, makeWriteBodyFn), whose
-// back-pressure check celeris#761 changed: a header block and a small body
-// through writeFn, and a header block and a 16 KiB zero-copy body through
-// writeFn + writeBodyFn. The buffers are reset as a completed send leaves
-// them.
+// BenchmarkWriteHooks measures the per-response cost of the write hook the
+// H1 response adapter calls (makeWriteFn), whose back-pressure check
+// celeris#761 changed: a header block and a small body, and a header block
+// and a 16 KiB body. The adapter copies every body through makeWriteFn since
+// io_uring has no zero-copy body writer (celeris#817), so the 16 KiB case
+// includes that copy. The buffers are reset as a completed send leaves them.
 func BenchmarkWriteHooks(b *testing.B) {
 	hdr := make([]byte, 121)
 	small := make([]byte, 64)
@@ -25,16 +25,15 @@ func BenchmarkWriteHooks(b *testing.B) {
 			cs.writeBuf = cs.writeBuf[:0]
 		}
 	})
-	b.Run("zero-copy-body", func(b *testing.B) {
+	b.Run("large-body", func(b *testing.B) {
 		w := &Worker{}
 		cs := &connState{}
-		wf, wb := w.makeWriteFn(cs), w.makeWriteBodyFn(cs)
+		wf := w.makeWriteFn(cs)
 		b.ReportAllocs()
 		for b.Loop() {
 			wf(hdr)
-			wb(large)
+			wf(large)
 			cs.writeBuf = cs.writeBuf[:0]
-			cs.bodyBuf = nil
 		}
 	})
 }
diff --git a/internal/conn/h1.go b/internal/conn/h1.go
index 2a91df51..c3f20cc3 100644
--- a/internal/conn/h1.go
+++ b/internal/conn/h1.go
@@ -39,6 +39,12 @@ var ErrAsyncDispatch = errors.New("celeris: route requires async dispatch")
 // The engine must not close or reuse the FD after receiving this error.
 var ErrHijacked = errors.New("celeris: connection hijacked")
 
+// ErrWriteBacklog is returned by ProcessH1 when a request finds the
+// connection's unsent responses over the engine's back-pressure limit
+// (H1State.WriteBacklogged). Its handler has NOT run. The engine closes the
+// connection once what is queued has gone out (celeris#761).
+var ErrWriteBacklog = errors.New("celeris: unsent responses over the back-pressure limit: request not served")
+
 // ClearHeaderDeadline drops the slowloris-defence read-header deadline
 // — called after a successful ParseRequest signals that the next state
 // is request handling, not header reading. The engine's checkTimeouts
@@ -254,6 +260,19 @@ type H1State struct {
 	// path and on pure-sync / pure-async servers (no behavior change).
 	InlineMode bool
 	RouteAsync func(method, path string) bool
+
+	// WriteBacklogged, set by an engine that queues responses (epoll,
+	// io_uring), reports whether the connection's responses still unsent
+	// are over the engine's back-pressure limit: a client that stopped
+	// reading while it kept sending requests. It is asked before each
+	// request's handler runs (handleH1Request); a request that finds such
+	// a backlog is not served, and ProcessH1 returns ErrWriteBacklog. The
+	// limit is held per request, not per write (celeris#761): a response
+	// is staged whole however large it is and in however many writes it
+	// comes (a StreamWriter's chunks), where a limit per write cut it off
+	// mid-body. Called on the goroutine running ProcessH1, under the
+	// engine's lock for the connection's buffers if it has one.
+	WriteBacklogged func() bool
 }
 
 // TakeBufferedBytes returns a copy of any bytes ProcessH1 stashed in the
@@ -326,11 +345,15 @@ func (s *H1State) UpdateWriteFn(fn func([]byte)) {
 // SetWriteBodyFn installs a scatter-gather body writer on the response
 // adapter. When non-nil, WriteResponse for large bodies bypasses the
 // respBuf → cs.writeBuf copy and hands the body slice straight to the
-// engine for a single WRITEV/sendmsg submission. Callers must not mutate
-// the body slice after a writeBody call until the response returns (the
-// engine keeps a reference until the SEND CQE fires). The std engine and
-// adapters that cannot express scatter-gather may leave this unset; the
-// adapter then falls back to the single-buffer path.
+// engine. The engine must not keep a reference to the body once fn returns
+// (celeris#817): the body belongs to the handler, which may reuse it as soon
+// as its write returns, as net/http allows. c.JSON puts its encode buffer
+// back in a pool at once, and a buffered response's body lives on a pooled
+// Context. So fn writes the body to the socket itself and copies what the
+// socket did not take (epoll); an engine that cannot finish its write
+// before returning (io_uring, whose kernel reads a WRITEV's iovec when the
+// ring is next entered) leaves this unset, and the adapter then copies the
+// body into the write buffer.
 func (s *H1State) SetWriteBodyFn(fn func([]byte)) {
 	s.rw.writeBody = fn
 }
@@ -935,6 +958,10 @@ func expectHeaders(req *h1.Request) [][2]string {
 func handleH1Request(ctx context.Context, state *H1State, body []byte,
 	handler stream.Handler, write func([]byte)) error {
 
+	if state.WriteBacklogged != nil && state.WriteBacklogged() {
+		return ErrWriteBacklog
+	}
+
 	req := &state.req
 	s := populateCachedStream(state, req, body)
 
diff --git a/large_response_linux_test.go b/large_response_linux_test.go
index ef87c8bc..2b24ae59 100644
--- a/large_response_linux_test.go
+++ b/large_response_linux_test.go
@@ -124,7 +124,7 @@ func TestLargeResponseIsDelivered(t *testing.T) {
 // against the cap and closed mid-file; elsewhere it is read into memory and
 // written like a Blob.
 func TestLargeFileResponseIsDelivered(t *testing.T) {
-	sizes := []int{4<<20 + 1, 64 << 20}
+	sizes := []int{4<<20 + 1, 16 << 20}
 	bodies := bodies761(sizes...)
 	dir := t.TempDir()
 	for _, n := range sizes {
@@ -158,8 +158,12 @@ func TestLargeFileResponseIsDelivered(t *testing.T) {
 // knowledge). net/http's client opens a 4 MiB stream window, so a larger body
 // leaves up to 4 MiB of frames queued behind the socket: io_uring's check
 // after the handler closed the connection on that backlog, at exactly 4 MiB.
+// The largest body is 16 MiB, not 64 MiB: an HTTP/2 transfer on the native
+// engines is about ten times slower than on std under -race (celeris#809),
+// and the four windows it spans cover what the 4 MiB cut and a 4 MiB cap on
+// HTTP/2 conns could do to it.
 func TestLargeResponseIsDeliveredH2(t *testing.T) {
-	sizes := []int{4<<20 - 4096, 4<<20 + 1, 64 << 20}
+	sizes := []int{4<<20 - 4096, 4<<20 + 1, 16 << 20}
 	bodies := bodies761(sizes...)
 	for _, e := range engines761 {
 		for _, route := range []string{"sync", "async-route"} {
@@ -351,12 +355,14 @@ func TestPipelinedResponsesKeepTheirOrder(t *testing.T) {
 // TestBackloggedPeerIsClosed is the other side of the cap: a peer that sends
 // requests without reading the responses must not make the engine buffer
 // without bound, nor be left waiting. Four requests for 3 MiB each, in one
-// packet, to a client that reads nothing for a while: the native engines stage
-// responses until the backlog they find is over the cap, then refuse the
-// next write and close the connection once what was staged has gone out. The
-// client must get whole responses, in order, followed by either the rest or
-// EOF; before celeris#761 the refused response was dropped and the client
-// waited for it.
+// packet, to a client that reads nothing for a while: the native engines
+// serve requests until one finds the unsent responses over the 4 MiB cap,
+// which is not served, and close the connection once what was staged has
+// gone out. The client must get whole responses, in order, then EOF: fewer
+// than it asked for from the native engines (the cap bounds what they
+// buffer), all of them from std, whose handler blocks on the socket instead.
+// Before celeris#761 the refused response was dropped and the client waited
+// for it.
 func TestBackloggedPeerIsClosed(t *testing.T) {
 	const n = 3 << 20
 	const requests = 4
@@ -407,6 +413,56 @@ func TestBackloggedPeerIsClosed(t *testing.T) {
 				if whole == 0 {
 					t.Fatalf("%s/%s: no response at all", e.name, sh.name)
 				}
+				if e.eng != celeris.Std && whole == requests {
+					t.Fatalf("%s/%s: all %d responses (%d MiB) were staged for a client that read nothing: the 4 MiB back-pressure cap did not apply", e.name, sh.name, requests, requests*n>>20)
+				}
+			})
+		}
+	}
+}
+
+// TestStreamedResponseIsDelivered streams 8 MiB through c.StreamWriter, in
+// 1 MiB chunks, without detaching. The native engines buffer every chunk
+// until the handler returns, and the cap, when it was held per write, took a
+// response's own earlier chunks for a backlog: everything after 4 MiB was
+// dropped, and the connection left open (before celeris#761's first fix) or
+// closed (after it). The limit is now held per request.
+func TestStreamedResponseIsDelivered(t *testing.T) {
+	const chunks, chunk = 8, 1 << 20
+	want := bodies761(chunks * chunk)[chunks*chunk]
+	for _, e := range engines761 {
+		for _, route := range []string{"sync", "async-route"} {
+			t.Run(e.name+"/"+route, func(t *testing.T) {
+				addr := startServer761(t, e.eng, false, func(s *celeris.Server) {
+					r := s.GET("/stream", func(c *celeris.Context) error {
+						sw := c.StreamWriter()
+						if sw == nil {
+							return errors.New("no StreamWriter")
+						}
+						if err := sw.WriteHeader(http.StatusOK, [][2]string{{"content-type", "application/octet-stream"}}); err != nil {
+							return err
+						}
+						for i := range chunks {
+							if _, err := sw.Write(want[i*chunk : (i+1)*chunk]); err != nil {
+								return err
+							}
+						}
+						return sw.Close()
+					})
+					if route == "async-route" {
+						r.Async()
+					}
+				})
+				for _, keepAlive := range []bool{true, false} {
+					mode := "keep-alive"
+					if !keepAlive {
+						mode = "close"
+					}
+					t.Run(mode, func(t *testing.T) {
+						desc := fmt.Sprintf("%s/%s streamed %d MiB %s", e.name, route, chunks*chunk>>20, mode)
+						checkH1Response761(t, addr, "/stream", desc, want, keepAlive)
+					})
+				}
 			})
 		}
 	}
@@ -547,7 +603,8 @@ func checkH1Response761(t *testing.T, addr, target, desc string, want []byte, ke
 	if err != nil {
 		t.Fatalf("%s: no response head (%d bytes received, %s)", desc, cr.n, describeReadEnd761(err))
 	}
-	if resp.StatusCode != http.StatusOK || resp.ContentLength != int64(n) {
+	chunked := len(resp.TransferEncoding) > 0 && resp.ContentLength == -1
+	if resp.StatusCode != http.StatusOK || (resp.ContentLength != int64(n) && !chunked) {
 		t.Fatalf("%s: status %d, Content-Length %d", desc, resp.StatusCode, resp.ContentLength)
 	}
 	got := make([]byte, n)
@@ -563,6 +620,9 @@ func checkH1Response761(t *testing.T, addr, target, desc string, want []byte, ke
 		}
 		t.Fatalf("%s: all bytes arrived but differ from byte %d on", desc, i)
 	}
+	if k, err := resp.Body.Read(make([]byte, 1)); k != 0 || !errors.Is(err, io.EOF) {
+		t.Fatalf("%s: the body did not end after %d bytes (%d more, %v)", desc, n, k, err)
+	}
 	_ = resp.Body.Close()
 
 	if !keepAlive {
diff --git a/response_body_ownership_linux_test.go b/response_body_ownership_linux_test.go
new file mode 100644
index 00000000..bb5549c7
--- /dev/null
+++ b/response_body_ownership_linux_test.go
@@ -0,0 +1,251 @@
+//go:build linux
+
+package celeris_test
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/goceleris/celeris"
+)
+
+// Tests for celeris#817. The H1 response adapter hands a body of 8 KiB or
+// more to epoll and io_uring through their zero-copy body writer, which kept
+// a reference to the caller's slice past the write: epoll sent it at the
+// flush after the handler (or later, on EPOLLOUT), io_uring when the ring was
+// next entered. The body belongs to the handler, which may reuse it as soon
+// as its write returns: c.JSON puts its encode buffer back in a pool at
+// once, where the next handler to encode takes it, and a buffered response's
+// body lives on a Context that is reused. So a response went out carrying
+// the next pipelined request's body, or another connection's. std copies
+// every body and never did.
+
+// ownedList761 is a []string (the reflection-free c.JSON fast path, whose
+// buffer comes from a pool and goes back when c.JSON returns) of at least
+// size bytes of JSON, every 256-byte element naming id and its index, so a
+// body that is not its own does not compare equal.
+func ownedList761(id, size int) []string {
+	tag := strconv.Itoa(10000000 + id)
+	fill := strings.Repeat(tag, 256/len(tag))
+	var out []string
+	for n, i := 0, 0; n < size; i++ {
+		s := strconv.Itoa(1000+i) + "-" + fill
+		out = append(out, s)
+		n += len(s) + 3
+	}
+	return out
+}
+
+type ownedStruct761 struct {
+	ID    int      `json:"id"`
+	Items []string `json:"items"`
+}
+
+// ownedBody761 is the response the /j route sends for id in the given
+// kind: "fast" (c.JSON of a []string), "encjson" (c.JSON of a struct, the
+// encoding/json path and its pooled encoder) or "buffered" (the same through
+// BufferResponse and FlushResponse, whose body is the Context's own).
+func ownedBody761(kind string, id, size int) any {
+	if kind == "encjson" {
+		return ownedStruct761{ID: id, Items: ownedList761(id, size)}
+	}
+	return ownedList761(id, size)
+}
+
+func ownedRoute761(s *celeris.Server, kind string, size int, async bool) {
+	r := s.GET("/j", func(c *celeris.Context) error {
+		id, err := strconv.Atoi(c.Query("id"))
+		if err != nil {
+			return err
+		}
+		if kind == "buffered" {
+			c.BufferResponse()
+			if err := c.JSON(http.StatusOK, ownedBody761(kind, id, size)); err != nil {
+				return err
+			}
+			return c.FlushResponse()
+		}
+		return c.JSON(http.StatusOK, ownedBody761(kind, id, size))
+	})
+	if async {
+		r.Async()
+	}
+}
+
+// firstDiff761 describes where got departs from want.
+func firstDiff761(got, want []byte) string {
+	i := 0
+	for i < len(got) && i < len(want) && got[i] == want[i] {
+		i++
+	}
+	end := min(i+32, len(got))
+	return fmt.Sprintf("%d bytes (want %d), differ from byte %d: %q", len(got), len(want), i, got[i:end])
+}
+
+// TestPipelinedResponsesOwnTheirBodies pipelines three requests, whose
+// responses are 16 KiB c.JSON bodies, in one packet, rounds times on new
+// connections, and compares every body with its own. The handler of request
+// 2 took the pool buffer request 1's body still pointed into.
+func TestPipelinedResponsesOwnTheirBodies(t *testing.T) {
+	const (
+		size   = 16 << 10
+		rounds = 10
+	)
+	for _, e := range engines761 {
+		for _, kind := range []string{"fast", "encjson", "buffered"} {
+			t.Run(e.name+"/"+kind, func(t *testing.T) {
+				addr := startServer761(t, e.eng, false, func(s *celeris.Server) {
+					ownedRoute761(s, kind, size, false)
+				})
+				bad := 0
+				for r := range rounds {
+					ids := []int{r*10 + 1, r*10 + 2, r*10 + 3}
+					raw, err := net.Dial("tcp", addr)
+					if err != nil {
+						t.Fatal(err)
+					}
+					var batch bytes.Buffer
+					for _, id := range ids {
+						fmt.Fprintf(&batch, "GET /j?id=%d HTTP/1.1\r\nHost: x\r\n\r\n", id)
+					}
+					if _, err := raw.Write(batch.Bytes()); err != nil {
+						t.Fatal(err)
+					}
+					br := bufio.NewReaderSize(idleConn761{raw}, 64<<10)
+					for i, id := range ids {
+						want, _ := json.Marshal(ownedBody761(kind, id, size))
+						resp, err := http.ReadResponse(br, nil)
+						if err != nil {
+							t.Errorf("%s/%s round %d: response %d: %s", e.name, kind, r, i+1, describeReadEnd761(err))
+							bad++
+							break
+						}
+						got, err := io.ReadAll(resp.Body)
+						_ = resp.Body.Close()
+						if err != nil || !bytes.Equal(got, want) {
+							t.Errorf("%s/%s round %d: response %d (id %d): %s", e.name, kind, r, i+1, id, firstDiff761(got, want))
+							bad++
+							break
+						}
+					}
+					_ = raw.Close()
+				}
+				if bad > 0 {
+					t.Errorf("%s/%s: %d of %d rounds had a response that was not its own", e.name, kind, bad, rounds)
+				}
+			})
+		}
+	}
+}
+
+// TestConcurrentResponsesOwnTheirBodies: conns connections at once, each
+// asking rounds times for its own 16 KiB c.JSON body, while the other
+// connections' handlers run and take pool buffers; every JSON body is
+// compared with its own. "direct" asks for the JSON body alone and reads it
+// at once: io_uring's kernel read the body at the next submit, after other
+// handlers in the same completion batch had run. "behind-a-large-body"
+// pipelines it behind a 2 MiB static body and reads nothing for 100 ms,
+// through a 64 KiB receive buffer, so the JSON body waits for the socket:
+// epoll left it staged until the socket took it. The other request on a
+// connection has no id, so a body carrying another id came from another
+// connection's handler.
+func TestConcurrentResponsesOwnTheirBodies(t *testing.T) {
+	const (
+		size   = 16 << 10
+		conns  = 16
+		rounds = 4
+	)
+	fill := bytes.Repeat([]byte("f"), 2<<20)
+	for _, e := range engines761 {
+		for _, mode := range []string{"direct", "behind-a-large-body"} {
+			t.Run(e.name+"/"+mode, func(t *testing.T) {
+				addr := startServer761(t, e.eng, false, func(s *celeris.Server) {
+					ownedRoute761(s, "fast", size, false)
+					s.GET("/fill", func(c *celeris.Context) error {
+						return c.Blob(http.StatusOK, "application/octet-stream", fill)
+					})
+				})
+				var bad, foreign, total, errs atomic.Int64
+				var first atomic.Value
+				var wg sync.WaitGroup
+				for k := range conns {
+					wg.Go(func() {
+						raw, err := net.Dial("tcp", addr)
+						if err != nil {
+							errs.Add(1)
+							return
+						}
+						defer func() { _ = raw.Close() }()
+						responses := 1
+						if mode == "behind-a-large-body" {
+							responses = 2
+							_ = raw.(*net.TCPConn).SetReadBuffer(64 << 10)
+						}
+						br := bufio.NewReaderSize(idleConn761{raw}, 64<<10)
+						for r := range rounds {
+							id := (k+1)*100000 + r
+							req := fmt.Sprintf("GET /j?id=%d HTTP/1.1\r\nHost: x\r\n\r\n", id)
+							if responses == 2 {
+								req = "GET /fill HTTP/1.1\r\nHost: x\r\n\r\n" + req
+							}
+							if _, err := io.WriteString(raw, req); err != nil {
+								errs.Add(1)
+								return
+							}
+							if responses == 2 {
+								time.Sleep(100 * time.Millisecond)
+							}
+							var got []byte
+							for i := range responses {
+								resp, err := http.ReadResponse(br, nil)
+								if err != nil {
+									errs.Add(1)
+									first.CompareAndSwap(nil, fmt.Sprintf("id %d, response %d: %s", id, i+1, describeReadEnd761(err)))
+									return
+								}
+								got, err = io.ReadAll(resp.Body)
+								_ = resp.Body.Close()
+								if err != nil {
+									errs.Add(1)
+									return
+								}
+							}
+							total.Add(1)
+							want, _ := json.Marshal(ownedList761(id, size))
+							if !bytes.Equal(got, want) {
+								bad.Add(1)
+								carried := ""
+								if len(got) >= 15 && got[0] == '[' {
+									if n, err := strconv.Atoi(string(got[7:15])); err == nil && n-10000000 != id {
+										carried = fmt.Sprintf(", carrying id %d", n-10000000)
+										if (n-10000000)/100000 != k+1 {
+											foreign.Add(1)
+										}
+									}
+								}
+								first.CompareAndSwap(nil, fmt.Sprintf("id %d: %s%s", id, firstDiff761(got, want), carried))
+							}
+						}
+					})
+				}
+				wg.Wait()
+				f, _ := first.Load().(string)
+				if bad.Load() > 0 || errs.Load() > 0 {
+					t.Errorf("%s/%s: %d of %d JSON bodies were not their own (%d carrying another connection's id), %d connections failed; first: %s",
+						e.name, mode, bad.Load(), total.Load(), foreign.Load(), errs.Load(), f)
+				}
+			})
+		}
+	}
+}

From 01375a05fb20b9901b35eae0684ed791afcf66d7 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 28 Sep 2026 11:53:53 +0200
Subject: [PATCH 3/7] test(iouring): the no-drain SQE witness expects a large
 body's tail to be the small one's, now that the body is copied (celeris#817)

---
 engine/iouring/fd_lifetime_fixture_test.go | 4 ++--
 engine/iouring/fd_lifetime_test.go         | 8 ++++++--
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/engine/iouring/fd_lifetime_fixture_test.go b/engine/iouring/fd_lifetime_fixture_test.go
index af3de4b2..d356d7c3 100644
--- a/engine/iouring/fd_lifetime_fixture_test.go
+++ b/engine/iouring/fd_lifetime_fixture_test.go
@@ -80,8 +80,8 @@ func takeSQEs(r *Ring) []sqeRec {
 	return out
 }
 
-// fdlHandler answers "ok", or a body large enough for the scatter-gather
-// (WRITEV) path on /large.
+// fdlHandler answers "ok", or a 16 KiB body on /large (over the H1 adapter's
+// 8 KiB zero-copy threshold, which io_uring no longer uses: celeris#817).
 type fdlHandler struct{}
 
 var fdlLargeBody = make([]byte, 16<<10)
diff --git a/engine/iouring/fd_lifetime_test.go b/engine/iouring/fd_lifetime_test.go
index b1a97516..fc323b6a 100644
--- a/engine/iouring/fd_lifetime_test.go
+++ b/engine/iouring/fd_lifetime_test.go
@@ -681,11 +681,15 @@ func TestNoDrainSQESequenceIsUnchanged(t *testing.T) {
 		}
 	})
 
-	t.Run("sync_tail_writev", func(t *testing.T) {
+	// A 16 KiB body took the WRITEV path (an unlinked WRITEV, then a
+	// standalone RECV) until celeris#817: io_uring has no zero-copy body
+	// writer now, the body is copied into the write buffer, and the tail is
+	// the small response's.
+	t.Run("sync_tail_large_body", func(t *testing.T) {
 		f := newFDLFixture(t, false)
 		f.armFirstRecv()
 		f.deliver("GET /large HTTP/1.1\r\nHost: x\r\n\r\n")
-		check(t, f, takeSQEs(f.w.ring), []want{{opWRITEV, 0, udSend}, {opRECV, 0, udRecv}})
+		check(t, f, takeSQEs(f.w.ring), []want{{opSEND, sqeIOLink, udSend}, {opRECV, 0, udRecv}})
 		f.process(f.sendCQE())
 		check(t, f, takeSQEs(f.w.ring), nil)
 	})

From 1f13ad8462bfebc12ca9e26fc293101b3bc97088 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 28 Sep 2026 11:59:23 +0200
Subject: [PATCH 4/7] fix(h2): read a stream's OutboundBuffer under its lock
 after the handler returns (celeris#822)

---
 protocol/h2/stream/processor.go | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/protocol/h2/stream/processor.go b/protocol/h2/stream/processor.go
index f81241e5..28d4ee9a 100644
--- a/protocol/h2/stream/processor.go
+++ b/protocol/h2/stream/processor.go
@@ -784,7 +784,12 @@ func (p *Processor) executeHandler(stream *Stream) {
 	}
 
 	if state == StateOpen || state == StateHalfClosedRemote {
-		if stream.OutboundBuffer != nil && stream.OutboundBuffer.Len() > 0 {
+		// Under the stream's lock, as every access of the buffer is: the
+		// event loop sends and resets it on a WINDOW_UPDATE (celeris#822).
+		stream.mu.RLock()
+		pending := stream.OutboundBuffer != nil && stream.OutboundBuffer.Len() > 0
+		stream.mu.RUnlock()
+		if pending {
 			return
 		}
 

From 621372ae8f4877d22be8b49dee3c5c196d1c1b10 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 28 Sep 2026 12:00:33 +0200
Subject: [PATCH 5/7] test: fewer large transfers for CI time: 64 MiB on
 keep-alive only, no below-threshold sizes (celeris#761)

---
 large_response_linux_test.go | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/large_response_linux_test.go b/large_response_linux_test.go
index 2b24ae59..51385341 100644
--- a/large_response_linux_test.go
+++ b/large_response_linux_test.go
@@ -66,8 +66,10 @@ func bodies761(sizes ...int) map[int][]byte {
 // connection and then for /ping on the same connection (the framing must
 // still be intact), and once more with Connection: close, where the body must
 // be followed by EOF. The sizes straddle the old threshold, which was the cap
-// minus the header block, so 4 MiB - 1 failed too; 4 MiB - 4 KiB was
-// delivered before the fix on a keep-alive connection.
+// minus the header block, so 4 MiB - 1 failed too. The 64 MiB body is asked
+// for on a keep-alive connection only: the close path is the same as for the
+// sizes around 4 MiB, which already overrun the socket buffers, and CI's
+// -race runner pays for every 64 MiB transfer.
 //
 // "sync" runs the handler on the connection's worker, where epoll and
 // io_uring hand a large body to the engine as a zero-copy slice (the write
@@ -76,7 +78,7 @@ func bodies761(sizes ...int) map[int][]byte {
 // copied into the write buffer. "async-route" runs the handler on the
 // connection's dispatch goroutine.
 func TestLargeResponseIsDelivered(t *testing.T) {
-	sizes := []int{4<<20 - 4096, 4<<20 - 1, 4 << 20, 4<<20 + 1, 64 << 20}
+	sizes := []int{4<<20 - 1, 4 << 20, 4<<20 + 1, 64 << 20}
 	bodies := bodies761(sizes...)
 	shapes := []struct {
 		name                    string
@@ -104,6 +106,9 @@ func TestLargeResponseIsDelivered(t *testing.T) {
 				})
 				for _, n := range sizes {
 					for _, keepAlive := range []bool{true, false} {
+						if n == 64<<20 && !keepAlive {
+							continue
+						}
 						mode := "keep-alive"
 						if !keepAlive {
 							mode = "close"
@@ -163,7 +168,7 @@ func TestLargeFileResponseIsDelivered(t *testing.T) {
 // and the four windows it spans cover what the 4 MiB cut and a 4 MiB cap on
 // HTTP/2 conns could do to it.
 func TestLargeResponseIsDeliveredH2(t *testing.T) {
-	sizes := []int{4<<20 - 4096, 4<<20 + 1, 16 << 20}
+	sizes := []int{4<<20 + 1, 16 << 20}
 	bodies := bodies761(sizes...)
 	for _, e := range engines761 {
 		for _, route := range []string{"sync", "async-route"} {

From 9b84a69527beb06599550021620dd9d0732e9c30 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 28 Sep 2026 12:13:47 +0200
Subject: [PATCH 6/7] test: under -race or coverage the largest bodies are 16
 MiB (H1), 8 MiB (file, HTTP/2), and the concurrent test runs 2 rounds, for CI
 time (celeris#761)

---
 large_response_linux_test.go          | 37 +++++++++++++++++++--------
 race761_off_test.go                   |  6 +++++
 race761_on_test.go                    |  9 +++++++
 response_body_ownership_linux_test.go |  9 ++++---
 4 files changed, 48 insertions(+), 13 deletions(-)
 create mode 100644 race761_off_test.go
 create mode 100644 race761_on_test.go

diff --git a/large_response_linux_test.go b/large_response_linux_test.go
index 51385341..4864d152 100644
--- a/large_response_linux_test.go
+++ b/large_response_linux_test.go
@@ -38,6 +38,13 @@ import (
 // A client that stops receiving bytes for idleCap761 fails its case, so a
 // dropped body fails in seconds, not at a timeout.
 
+// lean761 reports a run under the race detector or with coverage, where the
+// large-response tests move their largest bodies at a quarter or half the
+// size: CI runs the root package both ways, within one -timeout, and pays for
+// every byte several times over there. The sizes around the 4 MiB threshold
+// are the same in every run; the full sizes run without -race.
+func lean761() bool { return raceOn761 || testing.CoverMode() != "" }
+
 // idleCap761 is how long a client waits for the next byte before it calls the
 // response lost: loopback delivers a 64 MiB body in well under a second even
 // under -race, and a dropped body never sends one more byte.
@@ -66,10 +73,10 @@ func bodies761(sizes ...int) map[int][]byte {
 // connection and then for /ping on the same connection (the framing must
 // still be intact), and once more with Connection: close, where the body must
 // be followed by EOF. The sizes straddle the old threshold, which was the cap
-// minus the header block, so 4 MiB - 1 failed too. The 64 MiB body is asked
-// for on a keep-alive connection only: the close path is the same as for the
-// sizes around 4 MiB, which already overrun the socket buffers, and CI's
-// -race runner pays for every 64 MiB transfer.
+// minus the header block, so 4 MiB - 1 failed too. The largest body, 64 MiB
+// (16 MiB under lean761), is asked for on a keep-alive connection only: the
+// close path is the same as for the sizes around 4 MiB, which already overrun
+// the socket buffers.
 //
 // "sync" runs the handler on the connection's worker, where epoll and
 // io_uring hand a large body to the engine as a zero-copy slice (the write
@@ -78,7 +85,11 @@ func bodies761(sizes ...int) map[int][]byte {
 // copied into the write buffer. "async-route" runs the handler on the
 // connection's dispatch goroutine.
 func TestLargeResponseIsDelivered(t *testing.T) {
-	sizes := []int{4<<20 - 1, 4 << 20, 4<<20 + 1, 64 << 20}
+	largest := 64 << 20
+	if lean761() {
+		largest = 16 << 20
+	}
+	sizes := []int{4<<20 - 1, 4 << 20, 4<<20 + 1, largest}
 	bodies := bodies761(sizes...)
 	shapes := []struct {
 		name                    string
@@ -106,7 +117,7 @@ func TestLargeResponseIsDelivered(t *testing.T) {
 				})
 				for _, n := range sizes {
 					for _, keepAlive := range []bool{true, false} {
-						if n == 64<<20 && !keepAlive {
+						if n == largest && !keepAlive {
 							continue
 						}
 						mode := "keep-alive"
@@ -130,6 +141,9 @@ func TestLargeResponseIsDelivered(t *testing.T) {
 // written like a Blob.
 func TestLargeFileResponseIsDelivered(t *testing.T) {
 	sizes := []int{4<<20 + 1, 16 << 20}
+	if lean761() {
+		sizes[1] = 8 << 20
+	}
 	bodies := bodies761(sizes...)
 	dir := t.TempDir()
 	for _, n := range sizes {
@@ -163,12 +177,15 @@ func TestLargeFileResponseIsDelivered(t *testing.T) {
 // knowledge). net/http's client opens a 4 MiB stream window, so a larger body
 // leaves up to 4 MiB of frames queued behind the socket: io_uring's check
 // after the handler closed the connection on that backlog, at exactly 4 MiB.
-// The largest body is 16 MiB, not 64 MiB: an HTTP/2 transfer on the native
-// engines is about ten times slower than on std under -race (celeris#809),
-// and the four windows it spans cover what the 4 MiB cut and a 4 MiB cap on
-// HTTP/2 conns could do to it.
+// The largest body is 16 MiB (8 MiB under lean761), not 64 MiB: an HTTP/2
+// transfer on the native engines is about ten times slower than on std under
+// -race (celeris#809), and the windows it spans cover what the 4 MiB cut and
+// a 4 MiB cap on HTTP/2 conns could do to it.
 func TestLargeResponseIsDeliveredH2(t *testing.T) {
 	sizes := []int{4<<20 + 1, 16 << 20}
+	if lean761() {
+		sizes[1] = 8 << 20
+	}
 	bodies := bodies761(sizes...)
 	for _, e := range engines761 {
 		for _, route := range []string{"sync", "async-route"} {
diff --git a/race761_off_test.go b/race761_off_test.go
new file mode 100644
index 00000000..cc0910fe
--- /dev/null
+++ b/race761_off_test.go
@@ -0,0 +1,6 @@
+//go:build !race
+
+package celeris_test
+
+// raceOn761: see race761_on_test.go.
+const raceOn761 = false
diff --git a/race761_on_test.go b/race761_on_test.go
new file mode 100644
index 00000000..59af4d01
--- /dev/null
+++ b/race761_on_test.go
@@ -0,0 +1,9 @@
+//go:build race
+
+package celeris_test
+
+// raceOn761 reports a -race build. The large-response tests (celeris#761,
+// celeris#817) move a few GiB over loopback; under the race detector on CI's
+// runners that costs several times what it does elsewhere, so they move less
+// there (lean761).
+const raceOn761 = true
diff --git a/response_body_ownership_linux_test.go b/response_body_ownership_linux_test.go
index bb5549c7..d15379e2 100644
--- a/response_body_ownership_linux_test.go
+++ b/response_body_ownership_linux_test.go
@@ -162,10 +162,13 @@ func TestPipelinedResponsesOwnTheirBodies(t *testing.T) {
 // connection's handler.
 func TestConcurrentResponsesOwnTheirBodies(t *testing.T) {
 	const (
-		size   = 16 << 10
-		conns  = 16
-		rounds = 4
+		size  = 16 << 10
+		conns = 16
 	)
+	rounds := 4
+	if lean761() {
+		rounds = 2
+	}
 	fill := bytes.Repeat([]byte("f"), 2<<20)
 	for _, e := range engines761 {
 		for _, mode := range []string{"direct", "behind-a-large-body"} {

From 052b5cc7b6f6fe2dd1a216115e62bb3ab2e8d7b6 Mon Sep 17 00:00:00 2001
From: Albert Bausili <albert.bausili@gmail.com>
Date: Mon, 28 Sep 2026 12:20:34 +0200
Subject: [PATCH 7/7] test: skip the io_uring async-route pipelined cases while
 celeris#751 (PR #800) is open

---
 large_response_linux_test.go | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/large_response_linux_test.go b/large_response_linux_test.go
index 4864d152..1cef03ac 100644
--- a/large_response_linux_test.go
+++ b/large_response_linux_test.go
@@ -331,6 +331,7 @@ func TestPipelinedResponsesKeepTheirOrder(t *testing.T) {
 	for _, e := range engines761 {
 		for _, sh := range shapes {
 			t.Run(e.name+"/"+sh.name, func(t *testing.T) {
+				skipIouringAsyncPipelined751(t, e.eng, sh.asyncRoute)
 				addr := startServer761(t, e.eng, sh.asyncServer, func(s *celeris.Server) {
 					big := s.GET("/big", func(c *celeris.Context) error {
 						n, _ := strconv.Atoi(c.Query("n"))
@@ -396,6 +397,7 @@ func TestBackloggedPeerIsClosed(t *testing.T) {
 	for _, e := range engines761 {
 		for _, sh := range shapes {
 			t.Run(e.name+"/"+sh.name, func(t *testing.T) {
+				skipIouringAsyncPipelined751(t, e.eng, sh.asyncRoute)
 				addr := startServer761(t, e.eng, sh.asyncServer, func(s *celeris.Server) {
 					r := s.GET("/big", func(c *celeris.Context) error {
 						return c.Blob(http.StatusOK, "application/octet-stream", body)
@@ -490,6 +492,19 @@ func TestStreamedResponseIsDelivered(t *testing.T) {
 	}
 }
 
+// skipIouringAsyncPipelined751 skips a pipelined case on io_uring with an
+// async route: the dispatch goroutine's direct write can go out while a ring
+// SEND of the connection's earlier bytes is still in flight, so pipelined
+// responses interleave (celeris#751, open PR #800), which these tests would
+// report as a wrong or out-of-order body. That route writes no zero-copy
+// body, so it is outside what these tests pin. Remove with #800.
+func skipIouringAsyncPipelined751(t *testing.T, eng celeris.EngineType, asyncRoute bool) {
+	t.Helper()
+	if eng == celeris.IOUring && asyncRoute {
+		t.Skip("celeris#751 (PR #800): an io_uring async handler's direct write can overtake a ring SEND of the conn's earlier bytes")
+	}
+}
+
 // startServer761 starts a server with routes and waits until it answers
 // /ping. An io_uring start that fails only with ENOMEM is retried, with a new
 // server, for up to 30 s: the kernel charges ring memory to RLIMIT_MEMLOCK