diff --git a/crates/turnloop-io/README.md b/crates/turnloop-io/README.md
index 58b1bc6..2dc9d52 100644
--- a/crates/turnloop-io/README.md
+++ b/crates/turnloop-io/README.md
@@ -25,6 +25,23 @@ one absolute backend-clock deadline for a whole exchange.
 5. Allocate retained buffers at connection creation/warm-up, never per I/O call.
    Application-owned headers/messages and crypto state have separate accounting.
 
+## Half-close and lingering close
+
+`close` releases the whole transport. `shutdown` (the `HalfClose` trait, implemented
+by `AsyncIo`, `TlsStream` and `Transport`) flushes accepted writes and ends only the
+write direction: TLS sends close_notify, then TCP `shutdown(SHUT_WR)` (IOCP
+`SD_SEND`, WASI `shutdown(send)`). Reads continue until the peer's EOF and the handle
+stays open until `close` or drop. UDP, Windows named pipes and non-socket Unix
+descriptors return `Unsupported`.
+
+A server should not close a socket while peer bytes are unread: that sends RST, and
+on macOS and Windows a received RST discards data the peer has not read yet (the
+tail of the final response). `linger_close(stream, scratch, deadline)` flushes and
+half-closes, reads and discards into caller-retained scratch until the peer's EOF
+or the deadline, then closes. It waits only on one read and one executor timer (no
+spin) and allocates nothing per read; `Lingered` reports how it ended and how many
+bytes were discarded.
+
 Native TCP/pipes and WASI sockets use the identical generic stream code. Browser
 raw sockets/listeners return Unsupported; browser HTTP uses host fetch. Windows
 uses the production Backend implementation when integrated.
diff --git a/crates/turnloop-io/src/lib.rs b/crates/turnloop-io/src/lib.rs
index ce2a738..d226c6a 100644
--- a/crates/turnloop-io/src/lib.rs
+++ b/crates/turnloop-io/src/lib.rs
@@ -6,6 +6,7 @@ use std::{
     io,
     net::SocketAddr,
     pin::Pin,
+    task::{Context, Poll},
 };
 pub use turnloop::{self, AsyncIo, ExecutorHandle, Instant, backend::Backend};
 pub mod dns;
@@ -53,6 +54,233 @@ pub async fn flush<S: AsyncWrite + Unpin>(stream: &mut S) -> io::Result<()> {
 pub async fn close<S: AsyncWrite + Unpin>(stream: &mut S) -> io::Result<()> {
     poll_fn(|cx| Pin::new(&mut *stream).poll_close(cx)).await
 }
+/// A stream whose write direction can end while its read direction stays open.
+///
+/// futures-io's `poll_close` releases the whole transport. A half-close instead
+/// sends the peer EOF and keeps reading, which a server needs for a lingering
+/// close: closing a socket with unread peer bytes sends RST, and a received RST
+/// discards data the peer has not read yet on macOS and Windows. `AsyncIo`,
+/// `turnloop_tls::asynchronous::TlsStream` and `Transport` implement it.
+pub trait HalfClose: Stream {
+    /// Backend of the executor that owns the native handle.
+    type Backend: Backend;
+    /// The executor owning the native handle, whose timers bound a lingering close.
+    /// `None` once the transport has been released.
+    fn executor(&self) -> Option<ExecutorHandle<Self::Backend>>;
+    /// Flush accepted writes, then end the write direction without closing the
+    /// handle (TLS sends close_notify first). Reads continue until the peer's EOF;
+    /// later writes fail. Completed shutdowns stay complete on repeated polls.
+    /// Transports without a separate write direction return
+    /// `io::ErrorKind::Unsupported` (see `AsyncIo::poll_shutdown`).
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
+}
+impl<B: Backend> HalfClose for AsyncIo<B> {
+    type Backend = B;
+    fn executor(&self) -> Option<ExecutorHandle<B>> {
+        Some(AsyncIo::executor(self))
+    }
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.get_mut().poll_shutdown(cx)
+    }
+}
+/// Half-close: flush accepted writes and end the write direction, keeping the
+/// transport open for reads. [`close`] keeps its full-close semantics.
+pub async fn shutdown<S: HalfClose>(stream: &mut S) -> io::Result<()> {
+    poll_fn(|cx| Pin::new(&mut *stream).poll_shutdown(cx)).await
+}
+/// How the discard phase of a lingering close ended.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum LingerEnd {
+    /// The peer ended its stream; all its remaining input was read and discarded.
+    Eof,
+    /// The linger deadline passed before the peer's EOF.
+    Deadline,
+    /// The transport cannot half-close, so it closed without lingering.
+    Unsupported,
+    /// The half-close, a discard read or the deadline timer failed (for example
+    /// the peer reset the connection); lingering stopped and the stream closed.
+    Failed(io::ErrorKind),
+}
+/// Outcome of a lingering close, including proof of the discarded input.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct Lingered {
+    /// Why discarding stopped.
+    pub end: LingerEnd,
+    /// Non-empty reads performed after the half-close.
+    pub reads: u64,
+    /// Peer bytes read and discarded after the half-close.
+    pub discarded: u64,
+}
+enum LingerPhase {
+    Shutdown,
+    Discard,
+    Close,
+    Done,
+}
+/// Future for [`linger_close`]. It borrows the stream and caller-retained scratch,
+/// allocates nothing, and waits only on the stream's own read and one executor
+/// timer, so an idle peer costs no turns before the deadline.
+pub struct LingerClose<'a, S: HalfClose> {
+    stream: &'a mut S,
+    scratch: &'a mut [u8],
+    executor: Option<ExecutorHandle<S::Backend>>,
+    deadline: Option<Instant>,
+    sleep: Option<(Instant, turnloop::Sleep<S::Backend>)>,
+    phase: LingerPhase,
+    lingered: Lingered,
+}
+/// Lingering close, as nginx `lingering_close`: flush and half-close, read and
+/// discard peer input until its EOF or `deadline`, then close.
+///
+/// Flush the final response before calling this. Closing with unread peer bytes
+/// (a pipelined request, HTTP/2 WINDOW_UPDATE or SETTINGS acknowledgements) sends
+/// RST, which can discard the peer's unread copy of that response. `scratch` must
+/// be non-empty; reuse a connection's retained input buffer. `None` lingers until
+/// the peer's EOF. The deadline bounds the half-close too. Transports that cannot
+/// half-close close at once ([`LingerEnd::Unsupported`]); half-close or read
+/// failures stop lingering and close ([`LingerEnd::Failed`]).
+///
+/// The result is `Err` when scratch is empty, when the half-close has not finished
+/// by the deadline (`TimedOut`: the peer stopped reading, so pending output cannot
+/// flush) or when the final `poll_close` fails. The stream is then not closed: drop
+/// it, which releases the handle. Dropping the future mid-way keeps in-flight
+/// operations owned by the stream, so a new lingering close resumes them.
+pub fn linger_close<'a, S: HalfClose>(
+    stream: &'a mut S,
+    scratch: &'a mut [u8],
+    deadline: Option<Instant>,
+) -> LingerClose<'a, S> {
+    LingerClose {
+        executor: stream.executor(),
+        stream,
+        scratch,
+        deadline,
+        sleep: None,
+        phase: LingerPhase::Shutdown,
+        lingered: Lingered {
+            end: LingerEnd::Deadline,
+            reads: 0,
+            discarded: 0,
+        },
+    }
+}
+impl<S: HalfClose> LingerClose<'_, S> {
+    /// Move the deadline, for example to honour an earlier server shutdown
+    /// deadline. The timer is re-armed on the next poll only if it changed.
+    pub fn set_deadline(&mut self, deadline: Option<Instant>) {
+        self.deadline = deadline;
+    }
+    /// Discard progress so far; `end` is meaningful once the future completes.
+    pub fn lingered(&self) -> Lingered {
+        self.lingered
+    }
+    /// Whether the deadline has passed. Otherwise its timer is armed (re-armed
+    /// only when the deadline moved) and registered with `cx`.
+    fn poll_expired(&mut self, cx: &mut Context<'_>) -> Result<bool, io::ErrorKind> {
+        let Some(deadline) = self.deadline else {
+            self.sleep = None;
+            return Ok(false);
+        };
+        if self.sleep.as_ref().is_none_or(|(at, _)| *at != deadline) {
+            let Some(executor) = &self.executor else {
+                return Ok(true);
+            };
+            if executor.now() >= deadline {
+                self.sleep = None;
+                return Ok(true);
+            }
+            // Replacing the timer drops (and closes) the previous one.
+            self.sleep = Some((deadline, executor.sleep_until(deadline)));
+        }
+        let Some((_, sleep)) = self.sleep.as_mut() else {
+            return Ok(false);
+        };
+        match Pin::new(sleep).poll(cx) {
+            Poll::Ready(Ok(())) => Ok(true),
+            Poll::Ready(Err(e)) => Err(error(e).kind()),
+            Poll::Pending => Ok(false),
+        }
+    }
+    fn poll_discard(&mut self, cx: &mut Context<'_>) -> Option<LingerEnd> {
+        loop {
+            match Pin::new(&mut *self.stream).poll_read(cx, self.scratch) {
+                Poll::Ready(Ok(0)) => return Some(LingerEnd::Eof),
+                Poll::Ready(Ok(n)) => {
+                    self.lingered.reads += 1;
+                    self.lingered.discarded += n as u64;
+                }
+                Poll::Ready(Err(e)) => return Some(LingerEnd::Failed(e.kind())),
+                Poll::Pending => break,
+            }
+        }
+        match self.poll_expired(cx) {
+            Ok(false) => None,
+            Ok(true) => Some(LingerEnd::Deadline),
+            Err(kind) => Some(LingerEnd::Failed(kind)),
+        }
+    }
+}
+impl<S: HalfClose> Future for LingerClose<'_, S> {
+    type Output = io::Result<Lingered>;
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+        loop {
+            match this.phase {
+                LingerPhase::Shutdown => {
+                    if this.scratch.is_empty() {
+                        return Poll::Ready(Err(io::Error::new(
+                            io::ErrorKind::InvalidInput,
+                            "lingering close needs non-empty scratch storage",
+                        )));
+                    }
+                    match Pin::new(&mut *this.stream).poll_shutdown(cx) {
+                        Poll::Pending => {
+                            let expired = match this.poll_expired(cx) {
+                                Ok(false) => return Poll::Pending,
+                                Ok(true) => io::ErrorKind::TimedOut,
+                                Err(kind) => kind,
+                            };
+                            this.lingered.end = if expired == io::ErrorKind::TimedOut {
+                                LingerEnd::Deadline
+                            } else {
+                                LingerEnd::Failed(expired)
+                            };
+                            this.sleep = None;
+                            this.phase = LingerPhase::Done;
+                            return Poll::Ready(Err(io::Error::new(
+                                expired,
+                                "half-close did not complete before the linger deadline",
+                            )));
+                        }
+                        Poll::Ready(Ok(())) => this.phase = LingerPhase::Discard,
+                        Poll::Ready(Err(e)) => {
+                            this.lingered.end = if e.kind() == io::ErrorKind::Unsupported {
+                                LingerEnd::Unsupported
+                            } else {
+                                LingerEnd::Failed(e.kind())
+                            };
+                            this.phase = LingerPhase::Close;
+                        }
+                    }
+                }
+                LingerPhase::Discard => {
+                    let Some(end) = this.poll_discard(cx) else {
+                        return Poll::Pending;
+                    };
+                    this.lingered.end = end;
+                    this.sleep = None;
+                    this.phase = LingerPhase::Close;
+                }
+                LingerPhase::Close => {
+                    let result = std::task::ready!(Pin::new(&mut *this.stream).poll_close(cx));
+                    this.phase = LingerPhase::Done;
+                    return Poll::Ready(result.map(|()| this.lingered));
+                }
+                LingerPhase::Done => return Poll::Ready(Ok(this.lingered)),
+            }
+        }
+    }
+}
 /// One deadline shared across every phase of an exchange; uses the executor timer.
 pub async fn deadline<B: Backend, F: Future<Output = io::Result<T>>, T>(
     executor: &ExecutorHandle<B>,
diff --git a/crates/turnloop-io/tests/allocations.rs b/crates/turnloop-io/tests/allocations.rs
index 13a2c4d..4dc5944 100644
--- a/crates/turnloop-io/tests/allocations.rs
+++ b/crates/turnloop-io/tests/allocations.rs
@@ -171,12 +171,199 @@ mod native {
             _ => panic!("client incomplete"),
         }
     }
+    /// Server-side progress the peer lock-steps on, so the gate never depends on
+    /// how a kernel coalesces paced writes into reads.
+    #[derive(Default)]
+    struct Progress {
+        shutdowns: std::cell::Cell<u32>,
+        read: std::cell::Cell<u64>,
+        peer: std::cell::RefCell<Option<Waker>>,
+    }
+    impl Progress {
+        fn wake_peer(&self) {
+            if let Some(waker) = self.peer.borrow_mut().take() {
+                waker.wake();
+            }
+        }
+        async fn until(&self, ready: impl Fn(&Self) -> bool) {
+            std::future::poll_fn(|cx| {
+                if ready(self) {
+                    Poll::Ready(())
+                } else {
+                    *self.peer.borrow_mut() = Some(cx.waker().clone());
+                    Poll::Pending
+                }
+            })
+            .await
+        }
+    }
+    /// Transparent server transport that publishes half-close and read progress.
+    struct Observed {
+        inner: AsyncIo<Platform>,
+        progress: std::rc::Rc<Progress>,
+        shut: bool,
+    }
+    impl AsyncRead for Observed {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            bytes: &mut [u8],
+        ) -> Poll<io::Result<usize>> {
+            let this = self.get_mut();
+            let result = Pin::new(&mut this.inner).poll_read(cx, bytes);
+            if let Poll::Ready(Ok(n @ 1..)) = &result {
+                this.progress.read.set(this.progress.read.get() + *n as u64);
+                this.progress.wake_peer();
+            }
+            result
+        }
+    }
+    impl AsyncWrite for Observed {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            bytes: &[u8],
+        ) -> Poll<io::Result<usize>> {
+            Pin::new(&mut self.get_mut().inner).poll_write(cx, bytes)
+        }
+        fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Pin::new(&mut self.get_mut().inner).poll_flush(cx)
+        }
+        fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Pin::new(&mut self.get_mut().inner).poll_close(cx)
+        }
+    }
+    impl HalfClose for Observed {
+        type Backend = Platform;
+        fn executor(&self) -> Option<ExecutorHandle<Platform>> {
+            HalfClose::executor(&self.inner)
+        }
+        fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            let this = self.get_mut();
+            std::task::ready!(HalfClose::poll_shutdown(Pin::new(&mut this.inner), cx))?;
+            if !std::mem::replace(&mut this.shut, true) {
+                // Once per connection: its discard count starts here.
+                this.progress.read.set(0);
+                this.progress
+                    .shutdowns
+                    .set(this.progress.shutdowns.get() + 1);
+                this.progress.wake_peer();
+            }
+            Poll::Ready(Ok(()))
+        }
+    }
+    /// Lingering close over real TCP: the half-close, every discard read of late
+    /// peer bytes, the deadline timer and the final close allocate nothing once a
+    /// first identical connection has warmed the executor and driver tables.
+    pub fn lingering_close_discards_without_allocating() {
+        const LATE: u64 = 256;
+        let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+        let h = executor.handle();
+        let client_h = h.clone();
+        let listener =
+            Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listener");
+        let address = listener.local_addr().expect("address");
+        let progress = std::rc::Rc::new(Progress::default());
+        let server_progress = progress.clone();
+        let server = executor
+            .spawn_local(async move {
+                let mut scratch = [0; 64];
+                let mut measured = None;
+                for round in 0..2 {
+                    let mut stream = Observed {
+                        inner: listener.accept().await.expect("accept"),
+                        progress: server_progress.clone(),
+                        shut: false,
+                    };
+                    write_all(&mut stream, b"final")
+                        .await
+                        .expect("final response");
+                    if round == 1 {
+                        COUNT.store(0, Ordering::Relaxed);
+                        ACTIVE.store(true, Ordering::Relaxed);
+                    }
+                    let lingered = linger_close(
+                        &mut stream,
+                        &mut scratch,
+                        Some(h.now() + Duration::from_secs(10)),
+                    )
+                    .await
+                    .expect("lingering close");
+                    if round == 1 {
+                        ACTIVE.store(false, Ordering::Relaxed);
+                        measured = Some((lingered, COUNT.load(Ordering::Relaxed)));
+                    }
+                    assert_eq!(lingered.end, LingerEnd::Eof);
+                    assert_eq!(lingered.discarded, 4 * LATE);
+                }
+                measured.expect("measured round")
+            })
+            .expect("spawn");
+        let client = executor
+            .spawn_local(async move {
+                for round in 0..2 {
+                    let mut stream = client_h
+                        .connect(address, Default::default())
+                        .await
+                        .expect("connect");
+                    let mut bytes = [0; 8];
+                    let mut n = 0;
+                    loop {
+                        let got = read(&mut stream, &mut bytes[n..]).await.expect("read");
+                        if got == 0 {
+                            break;
+                        }
+                        n += got;
+                    }
+                    assert_eq!(&bytes[..n], b"final");
+                    // The server may observe its half-close after our EOF.
+                    progress.until(|p| p.shutdowns.get() > round).await;
+                    // Lock-step: the next chunk goes out only after the server
+                    // discarded the previous one, so each chunk is its own read.
+                    for chunk in 1..=LATE {
+                        write_all(&mut stream, b"late").await.expect("late bytes");
+                        progress.until(|p| p.read.get() >= 4 * chunk).await;
+                    }
+                    close(&mut stream).await.expect("close");
+                }
+            })
+            .expect("spawn");
+        let end = executor.driver().now() + Duration::from_secs(20);
+        let mut server = server;
+        let mut client = client;
+        while !client.is_finished() || !server.is_finished() {
+            assert!(executor.driver().now() < end);
+            executor.turn(Timeout::Until(end)).expect("turn");
+        }
+        match Pin::new(&mut client).poll(&mut Context::from_waker(Waker::noop())) {
+            Poll::Ready(Ok(())) => {}
+            _ => panic!("client incomplete"),
+        }
+        let (lingered, allocations) =
+            match Pin::new(&mut server).poll(&mut Context::from_waker(Waker::noop())) {
+                Poll::Ready(Ok(v)) => v,
+                _ => panic!("server incomplete"),
+            };
+        // The subject ran: at least one discard read per lock-stepped chunk, every
+        // byte accounted for, all inside the measured window.
+        assert!(
+            lingered.reads >= LATE,
+            "one discard read per chunk: {lingered:?}"
+        );
+        assert_eq!(allocations, 0, "lingering close allocates: {lingered:?}");
+        println!(
+            "lingering close: {} discard reads, {} bytes, {allocations} allocations",
+            lingered.reads, lingered.discarded
+        );
+    }
 }
 fn main() {
     #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
     {
         native::real_tcp_shared_driver_zero_allocations_after_warmup();
         println!("test real_tcp_shared_driver_zero_allocations_after_warmup ... ok");
-        println!("test result: ok. 1 passed; 0 failed; 0 ignored;");
+        native::lingering_close_discards_without_allocating();
+        println!("test lingering_close_discards_without_allocating ... ok");
+        println!("test result: ok. 2 passed; 0 failed; 0 ignored;");
     }
 }
diff --git a/crates/turnloop-io/tests/streams.rs b/crates/turnloop-io/tests/streams.rs
index 1587d29..0eb05ef 100644
--- a/crates/turnloop-io/tests/streams.rs
+++ b/crates/turnloop-io/tests/streams.rs
@@ -78,6 +78,271 @@ fn tcp_roundtrip_and_deadline() {
     finish(&mut timeout);
 }
 
+async fn read_to_eof<S: AsyncRead + Unpin>(stream: &mut S) -> Vec<u8> {
+    let mut received = Vec::new();
+    let mut bytes = [0; 256];
+    loop {
+        // A reset here (instead of EOF) is the failure lingering close prevents.
+        let n = read(stream, &mut bytes)
+            .await
+            .expect("clean EOF, not a reset");
+        if n == 0 {
+            return received;
+        }
+        received.extend_from_slice(&bytes[..n]);
+    }
+}
+#[test]
+fn half_close_keeps_reading_until_peer_eof() {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen");
+    let address = listener.local_addr().expect("address");
+    let mut server = executor
+        .spawn_local(async move {
+            let mut s = listener.accept().await.expect("accept");
+            write_all(&mut s, b"response").await.expect("write");
+            assert!(!s.is_write_shut());
+            shutdown(&mut s).await.expect("half-close");
+            assert!(s.is_write_shut());
+            shutdown(&mut s)
+                .await
+                .expect("a completed half-close stays complete");
+            assert!(
+                write_all(&mut s, b"x").await.is_err(),
+                "writes end at the half-close"
+            );
+            let late = read_to_eof(&mut s).await;
+            close(&mut s).await.expect("close");
+            late
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            let mut s = h
+                .connect(address, Default::default())
+                .await
+                .expect("connect");
+            let response = read_to_eof(&mut s).await;
+            // The peer ended only its write direction: this one still delivers.
+            write_all(&mut s, b"late request")
+                .await
+                .expect("write after the peer's half-close");
+            shutdown(&mut s).await.expect("client half-close");
+            response
+        })
+        .expect("spawn");
+    let end = executor.driver().now() + Duration::from_secs(5);
+    while !server.is_finished() || !client.is_finished() {
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    assert_eq!(finish(&mut client), b"response");
+    assert_eq!(finish(&mut server), b"late request");
+}
+
+#[test]
+fn lingering_close_discards_peer_input_until_eof() {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let client_h = h.clone();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen");
+    let address = listener.local_addr().expect("address");
+    let mut server = executor
+        .spawn_local(async move {
+            let mut s = listener.accept().await.expect("accept");
+            write_all(&mut s, b"final response").await.expect("write");
+            let mut scratch = [0; 512];
+            linger_close(&mut s, &mut scratch, Some(h.now() + Duration::from_secs(5)))
+                .await
+                .expect("lingering close")
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            let mut s = client_h
+                .connect(address, Default::default())
+                .await
+                .expect("connect");
+            let response = read_to_eof(&mut s).await;
+            // Bytes sent after the server's final response stay unread by its
+            // protocol; lingering close must drain them instead of resetting.
+            for _ in 0..3 {
+                write_all(&mut s, &[7; 1024]).await.expect("late bytes");
+            }
+            close(&mut s).await.expect("client close");
+            response
+        })
+        .expect("spawn");
+    let end = executor.driver().now() + Duration::from_secs(5);
+    while !server.is_finished() || !client.is_finished() {
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    assert_eq!(finish(&mut client), b"final response");
+    let lingered = finish(&mut server);
+    assert_eq!(lingered.end, LingerEnd::Eof);
+    assert_eq!(lingered.discarded, 3 * 1024);
+    assert!(lingered.reads > 0);
+}
+
+#[test]
+fn lingering_close_deadline_closes_silent_peer_without_spinning() {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let client_h = h.clone();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen");
+    let address = listener.local_addr().expect("address");
+    const LINGER: Duration = Duration::from_millis(500);
+    let half_closed = std::rc::Rc::new(std::cell::Cell::new(false));
+    let server_half_closed = half_closed.clone();
+    let mut server = executor
+        .spawn_local(async move {
+            let mut s = listener.accept().await.expect("accept");
+            write_all(&mut s, b"final").await.expect("write");
+            let at = h.now() + LINGER;
+            // Half-close first so the host can tell when the server itself has
+            // observed it; linger_close then finds the half-close complete.
+            shutdown(&mut s).await.expect("half-close");
+            server_half_closed.set(true);
+            let mut scratch = [0; 64];
+            let lingered = linger_close(&mut s, &mut scratch, Some(at))
+                .await
+                .expect("lingering close");
+            assert!(h.now() >= at, "closed before the linger deadline");
+            lingered
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            let mut s = client_h
+                .connect(address, Default::default())
+                .await
+                .expect("connect");
+            assert_eq!(read_to_eof(&mut s).await, b"final");
+            // Never close: the server's deadline alone must end its linger.
+            s
+        })
+        .expect("spawn");
+    let end = executor.driver().now() + Duration::from_secs(5);
+    while !client.is_finished() {
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    let _silent_peer = finish(&mut client);
+    // The client saw the half-close. The server may observe its acknowledgement
+    // later (WASI 0.3 completes Shutdown after the send stream's result), so
+    // deliver it; the server then waits on its idle read and the linger timer.
+    while !half_closed.get() {
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    // Settle residual discovery from the exchange, as the keep-alive no-spin
+    // tests do; the remaining wait must block until the deadline: at most two
+    // turns and one zero-event wait for the one expiry.
+    for _ in 0..3 {
+        executor.turn(Timeout::Now).expect("settle queued events");
+    }
+    assert!(!server.is_finished(), "linger ended before its deadline");
+    let (mut turns, mut empty) = (0, 0);
+    while !server.is_finished() {
+        assert!(turns < 2, "lingering close spun");
+        let info = executor.turn(Timeout::Until(end)).expect("turn");
+        turns += 1;
+        empty += info.zero_event_waits;
+    }
+    assert!(empty <= 1, "zero-event waits while lingering: {empty}");
+    let lingered = finish(&mut server);
+    assert_eq!(lingered.end, LingerEnd::Deadline);
+    assert_eq!((lingered.reads, lingered.discarded), (0, 0));
+}
+
+#[test]
+fn lingering_close_deadline_bounds_a_stalled_half_close() {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let client_h = h.clone();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen");
+    let address = listener.local_addr().expect("address");
+    let mut server = executor
+        .spawn_local(async move {
+            let mut s = listener.accept().await.expect("accept");
+            // Fill the peer's receive window and our send buffer: the peer never
+            // reads, so the last accepted write can never be flushed.
+            let chunk = [9; 16384];
+            let mut written = 0;
+            loop {
+                let write = std::future::poll_fn(|cx| Pin::new(&mut s).poll_write(cx, &chunk));
+                match h.timeout(Duration::from_millis(200), write).await {
+                    Ok(result) => written += result.expect("write"),
+                    Err(e) => {
+                        assert_eq!(e.kind, turnloop::ErrorKind::TimedOut);
+                        break;
+                    }
+                }
+                assert!(written < 1 << 30, "socket buffers never filled");
+            }
+            let at = h.now() + Duration::from_millis(50);
+            let mut scratch = [0; 64];
+            let error = linger_close(&mut s, &mut scratch, Some(at))
+                .await
+                .expect_err("a half-close that cannot flush must not wait forever");
+            assert_eq!(error.kind(), std::io::ErrorKind::TimedOut);
+            assert!(h.now() >= at);
+            // Dropping the unclosed stream releases its handle.
+            drop(s);
+            written
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            client_h
+                .connect(address, Default::default())
+                .await
+                .expect("connect")
+        })
+        .expect("spawn");
+    let end = executor.driver().now() + Duration::from_secs(20);
+    while !server.is_finished() || !client.is_finished() {
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    assert!(finish(&mut server) > 0);
+    drop(finish(&mut client));
+}
+
+#[test]
+fn half_close_is_unsupported_on_datagram_adapters() {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let mut task = executor
+        .spawn_local(async move {
+            let handle = h
+                .driver()
+                .udp_bind("127.0.0.1:0".parse().expect("address"), &Default::default())
+                .expect("udp");
+            let peer = h.driver().local_addr(handle).expect("address");
+            let mut udp = h.udp(handle, peer);
+            let error = shutdown(&mut udp)
+                .await
+                .expect_err("datagrams cannot half-close");
+            assert_eq!(error.kind(), std::io::ErrorKind::Unsupported);
+            let mut scratch = [0; 16];
+            let lingered = linger_close(&mut udp, &mut scratch, None)
+                .await
+                .expect("closes without lingering");
+            assert_eq!(lingered.end, LingerEnd::Unsupported);
+            1
+        })
+        .expect("spawn");
+    let end = executor.driver().now() + Duration::from_secs(5);
+    while !task.is_finished() {
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    assert_eq!(finish(&mut task), 1);
+}
+
 #[cfg(any(
     not(target_arch = "wasm32"),
     all(target_os = "wasi", target_env = "p2")
diff --git a/crates/turnloop/README.md b/crates/turnloop/README.md
index 3bffea5..8aee0a5 100644
--- a/crates/turnloop/README.md
+++ b/crates/turnloop/README.md
@@ -39,7 +39,8 @@ driver.turn(Timeout::Until(deadline), &mut completions)?;
 
 Enable `executor` for `LocalExecutor<B>` and its futures-io adapters. Read/write
 staging and task tables are reserved at construction. Writes are buffered; flush
-or close before dropping an adapter to confirm underlying completion. The crate's
+or close before dropping an adapter to confirm underlying completion.
+`AsyncIo::poll_shutdown` half-closes a stream without releasing its handle. The crate's
 rustdoc includes runnable loop and executor examples. See the
 [revision 2 handoff](https://github.com/PerryTS/turnloop/blob/main/docs/BACKEND_REVISION_2.md)
 for native ownership, process teardown, platform integration and contract details.
diff --git a/crates/turnloop/src/backend/unix.rs b/crates/turnloop/src/backend/unix.rs
index 824f69a..b9285f6 100644
--- a/crates/turnloop/src/backend/unix.rs
+++ b/crates/turnloop/src/backend/unix.rs
@@ -489,6 +489,13 @@ unsafe impl Backend for Unix {
             return self.services.submit(&request);
         }
         let r = self.get(h)?;
+        if matches!(request.operation, Operation::Shutdown)
+            && matches!(r.transport.kind, Kind::Stream | Kind::File)
+        {
+            // Only sockets have an independent write direction. A tty, FIFO or
+            // regular file cannot half-close; closing it is the caller's choice.
+            return Err(Error::new(ErrorKind::Unsupported));
+        }
         if r.transport.kind == Kind::File {
             let fd = r.transport.fd.try_clone().map_err(Error::from)?;
             return self.files.submit(request, fd);
@@ -835,7 +842,8 @@ fn execute(
             Ok(Some((Outcome::Wrote(n as usize), true)))
         }
         Operation::Shutdown => {
-            // SAFETY: fd is a live TCP socket; SHUT_WR is a valid shutdown direction.
+            // SAFETY: fd is a live TCP or Unix-domain stream socket (submit rejects
+            // other kinds); SHUT_WR is a valid shutdown direction.
             if unsafe { libc::shutdown(fd, libc::SHUT_WR) } < 0 {
                 return Err(last_error());
             }
diff --git a/crates/turnloop/src/driver.rs b/crates/turnloop/src/driver.rs
index ee9feac..04572ae 100644
--- a/crates/turnloop/src/driver.rs
+++ b/crates/turnloop/src/driver.rs
@@ -717,7 +717,15 @@ impl<B: Backend> Driver<B> {
     pub fn recv(&mut self, h: Handle, buf: ReadBuf, token: Token) -> Result<OpId> {
         self.submit(h, Operation::RecvFrom(buf), token)
     }
-    /// Queue stream write-side shutdown after earlier writes.
+    /// Queue stream write-side shutdown after earlier writes. The handle and its
+    /// read direction stay open: reads continue until the peer's EOF.
+    ///
+    /// TCP half-closes on every native backend (`SHUT_WR`, `SD_SEND`, WASI
+    /// `shutdown(send)` / send-stream close) and Unix-domain stream sockets on
+    /// epoll/kqueue. Transports without an independent write direction fail with
+    /// [`ErrorKind::Unsupported`]: Windows named pipes and synchronous handles,
+    /// Unix ttys, FIFOs and regular files, and web fetches. WASI stdout/stderr
+    /// close their output stream; a web WebSocket starts its close handshake.
     pub fn shutdown(&mut self, h: Handle, token: Token) -> Result<OpId> {
         self.submit(h, Operation::Shutdown, token)
     }
diff --git a/crates/turnloop/src/executor.rs b/crates/turnloop/src/executor.rs
index f4f7e7b..f83532d 100644
--- a/crates/turnloop/src/executor.rs
+++ b/crates/turnloop/src/executor.rs
@@ -429,6 +429,8 @@ impl<B: Backend> ExecutorHandle<B> {
             peer: None,
             read: None,
             write: None,
+            shutdown: None,
+            write_shut: false,
             closed: false,
         }
     }
@@ -595,12 +597,18 @@ impl<T> Drop for JoinHandle<T> {
 /// A Pending poll never consumes the current write slice, so it may be replaced.
 /// UDP writes accept one complete datagram or reject it if staging is too small;
 /// UDP reads expose payload bytes without the sender address.
+///
+/// [`AsyncIo::poll_shutdown`] half-closes a stream: the write direction ends while
+/// reads continue until the peer's EOF, and the handle stays open until
+/// `poll_close` or drop. `poll_close` still fully closes the handle.
 pub struct AsyncIo<B: Backend> {
     shared: Rc<Shared<B>>,
     handle: Handle,
     peer: Option<SocketAddr>,
     read: Option<Key>,
     write: Option<Key>,
+    shutdown: Option<Key>,
+    write_shut: bool,
     closed: bool,
 }
 impl<B: Backend> Unpin for AsyncIo<B> {}
@@ -609,6 +617,85 @@ impl<B: Backend> AsyncIo<B> {
     pub fn handle(&self) -> Handle {
         self.handle
     }
+    /// The executor that owns this stream's handle, for deadlines and timers on
+    /// the same loop (for example a lingering close). Cloning allocates nothing.
+    pub fn executor(&self) -> ExecutorHandle<B> {
+        ExecutorHandle {
+            shared: self.shared.clone(),
+        }
+    }
+    /// Whether [`AsyncIo::poll_shutdown`] has completed: writes now fail with
+    /// `BrokenPipe`, while reads continue until the peer's EOF.
+    pub fn is_write_shut(&self) -> bool {
+        self.write_shut
+    }
+    /// Gracefully end the write direction without closing the handle.
+    ///
+    /// Flushes accepted writes first, then submits the backend's stream shutdown
+    /// after them: TCP `shutdown(SHUT_WR)` on epoll/kqueue, `shutdown(SD_SEND)` on
+    /// IOCP, `shutdown(send)` on WASI 0.2 and closing the send stream on WASI 0.3.
+    /// The peer reads EOF after every byte written before it. Reads on this adapter
+    /// continue until the peer's own EOF, so a server can drain unread peer input
+    /// instead of closing with bytes unread (which sends RST, and on macOS and
+    /// Windows discards data the peer has not read yet). Close or drop afterwards.
+    ///
+    /// Repeated calls after completion return `Ok`. Transports that cannot
+    /// half-close return `io::ErrorKind::Unsupported`: UDP adapters, Windows named
+    /// pipes and console/synchronous handles, and non-socket Unix descriptors (ttys,
+    /// FIFOs, regular files). WASI stdout and stderr close their output stream; a
+    /// web WebSocket starts its close handshake. A pending poll never loses
+    /// progress: the submitted shutdown stays owned by this adapter until it
+    /// completes.
+    pub fn poll_shutdown(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        if self.write_shut {
+            return Poll::Ready(Ok(()));
+        }
+        if self.closed {
+            return Poll::Ready(Err(io::ErrorKind::NotConnected.into()));
+        }
+        if self.peer.is_some() {
+            return Poll::Ready(Err(io::Error::new(
+                io::ErrorKind::Unsupported,
+                "a datagram adapter has no stream write direction to shut down",
+            )));
+        }
+        std::task::ready!(Pin::new(&mut *self).poll_flush(cx))?;
+        let key = match self.shutdown {
+            Some(key) => key,
+            None => {
+                let key = self.shared.reserve(cx).map_err(io_error)?;
+                let result = self
+                    .shared
+                    .driver
+                    .borrow_mut()
+                    .shutdown(self.handle, key.token());
+                match result {
+                    Ok(op) => {
+                        self.shared.slots.borrow_mut()[key.index].op = Some(op);
+                        self.shutdown = Some(key);
+                        key
+                    }
+                    Err(e) => {
+                        self.shared.free(key);
+                        return Poll::Ready(Err(shutdown_error(e)));
+                    }
+                }
+            }
+        };
+        let Some(result) = self.shared.result(key, cx) else {
+            return Poll::Pending;
+        };
+        self.shared.free(key);
+        self.shutdown = None;
+        Poll::Ready(match result {
+            OpResult::Shutdown => {
+                self.write_shut = true;
+                Ok(())
+            }
+            OpResult::Err(e) => Err(shutdown_error(e)),
+            _ => Err(io_error(Error::new(ErrorKind::Cancelled))),
+        })
+    }
     fn start_read(&mut self, cx: &Context<'_>, length: usize) -> Result<Key> {
         let key = self.shared.reserve(cx)?;
         let buf = {
@@ -649,6 +736,16 @@ fn io_error(error: Error) -> io::Error {
         io::Error::other(error)
     }
 }
+/// Keep the portable categories a lingering close branches on.
+fn shutdown_error(error: Error) -> io::Error {
+    match (error.os, error.kind) {
+        (None, ErrorKind::Unsupported) => io::Error::new(io::ErrorKind::Unsupported, error),
+        (None, ErrorKind::InvalidInput) => io::Error::new(io::ErrorKind::InvalidInput, error),
+        (None, ErrorKind::BrokenPipe) => io::Error::new(io::ErrorKind::BrokenPipe, error),
+        (None, ErrorKind::ConnectionReset) => io::Error::new(io::ErrorKind::ConnectionReset, error),
+        _ => io_error(error),
+    }
+}
 impl<B: Backend> AsyncRead for AsyncIo<B> {
     fn poll_read(
         self: Pin<&mut Self>,
@@ -710,7 +807,7 @@ impl<B: Backend> AsyncWrite for AsyncIo<B> {
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        if self.closed {
+        if self.closed || self.write_shut || self.shutdown.is_some() {
             return Poll::Ready(Err(io_error(Error::new(ErrorKind::BrokenPipe))));
         }
         // A Pending call must not consume the current caller's bytes. Drain the
@@ -813,6 +910,9 @@ impl<B: Backend> Drop for AsyncIo<B> {
         if let Some(key) = self.write.take() {
             self.shared.abandon(key);
         }
+        if let Some(key) = self.shutdown.take() {
+            self.shared.abandon(key);
+        }
         if !self.closed {
             let _ = self.shared.driver.borrow_mut().close(self.handle, Token(0));
         }
diff --git a/docs/lanes/linger-close.md b/docs/lanes/linger-close.md
new file mode 100644
index 0000000..059e61a
--- /dev/null
+++ b/docs/lanes/linger-close.md
@@ -0,0 +1,227 @@
+# linger-close — issue #21
+
+Base `f6fc128`, branch `lane/linger-close`, macOS 26 arm64 (Darwin 25.5), Node
+26.5.1, 2026-09-15. Implementation, tests and local verification complete.
+Linux, Windows and WASI 0.3 **runtime UNRUN here** (integrator CI).
+
+## Root cause
+
+`server::http2` closed the socket as soon as `core.is_drained()` held, and
+`server::http1` right after its final response (and dropped idle keep-alive
+connections outright at shutdown). A peer's last bytes could still be unread in
+the server's receive buffer: HTTP/2 SETTINGS/WINDOW_UPDATE acknowledgements, a
+PING, a pipelined HTTP/1 request. `close()` on a TCP socket with unread receive
+data sends RST instead of FIN. The peer then reads `ECONNRESET` instead of EOF,
+and a reset can discard data it has not read yet (probe on this Mac: a 1 MB
+response closed with unread peer bytes delivered 875,452 bytes before
+`ECONNRESET`; small responses arrived intact but ended in `ECONNRESET`, which is
+what Node's `ClientHttp2Stream` reported in PR #20's macOS run). The same close was
+used by `WebSocketStream::close`.
+
+## API
+
+**turnloop (`executor.rs`)**, `AsyncIo<B>`:
+
+- `poll_shutdown(&mut self, cx) -> Poll<io::Result<()>>`: flush accepted writes,
+  then submit the backend's `Operation::Shutdown` (queued after earlier writes)
+  **without closing the handle**. Reads continue until the peer's EOF. Idempotent
+  after completion; writes after (or during) a half-close fail with `BrokenPipe`;
+  a pending shutdown is owned by the adapter and abandoned on drop. UDP adapters
+  return `io::ErrorKind::Unsupported` without submitting.
+- `executor() -> ExecutorHandle<B>` (Rc clone, no allocation) and `is_write_shut()`.
+- `poll_close` is unchanged: it still fully closes the handle.
+
+**turnloop-io**:
+
+- `trait HalfClose: Stream { type Backend; fn executor(&self) -> Option<ExecutorHandle<_>>; fn poll_shutdown(...) }`,
+  implemented for `AsyncIo`, `turnloop_tls::asynchronous::TlsStream` and `Transport`.
+- `shutdown(&mut stream)` next to the unchanged `close(&mut stream)`.
+- `linger_close(&mut stream, &mut scratch, Option<Instant>) -> LingerClose` (future
+  with `set_deadline`), output `io::Result<Lingered { end, reads, discarded }>`,
+  `LingerEnd::{Eof, Deadline, Unsupported, Failed(kind)}`. Phases: half-close →
+  read-and-discard into caller scratch until EOF/error/deadline → `poll_close`.
+  The deadline bounds the half-close too. `Err` only for empty scratch, a half-close
+  that cannot finish by the deadline (`TimedOut`: peer stopped reading) or a failed
+  final close; the caller then drops the stream. It waits on the stream's read and
+  one executor timer, re-arming the timer only when the deadline moves.
+
+**turnloop-tls**: `TlsStream::poll_shutdown` = drive close_notify, flush it, then
+the transport's `poll_shutdown`. Decryption continues until the peer's
+close_notify/EOF. `poll_write` now refuses once close_notify was sent (previously
+it encrypted a record the transport then rejected, which blocked later reads).
+
+**turnloop-http** (`asynchronous::server`):
+
+- `Options { linger_timeout: Duration }`, default **5 s** (nginx
+  `lingering_timeout`); `Server::bind_with(executor, addr, options)`,
+  `Shutdown::with_options`, `Shutdown::options`.
+- `Shutdown::stop_by(deadline)`: stop and close every lingering connection no
+  later than `deadline` (earlier wins; connections already lingering are woken and
+  re-arm). `Shutdown::deadline()`. In-flight requests still finish; dropping the
+  `Server` cancels them, as before.
+- `server::http1`/`http2` now require `S: HalfClose` (all in-tree transports
+  implement it).
+
+**turnloop-websocket**: `WebSocketStream::close` (now on `S: HalfClose`) runs the
+closing handshake under its deadline, then a lingering close under the same deadline.
+
+## Server behaviour
+
+- HTTP/1: after a final response that ends the connection (`connection: close`,
+  non-reusable request, or shutdown), and for a connection idle between requests
+  when shutdown arrives: half-close, discard until EOF or
+  `min(now + linger_timeout, stop_by deadline)`, close. The idle wait now reads
+  outside the codec, so a shutdown no longer cancels a codec read (which dropped the
+  socket); a stop that interrupts a partially received head still releases the
+  transport immediately, as before. A buffered pipelined request is not served
+  after stop (unchanged) and is discarded.
+- HTTP/2: when drained (GOAWAY sent, streams done, output flushed): flush, then the
+  same lingering close. Protocol-error closes are unchanged (see follow-ups).
+- A completed exchange never becomes an error during lingering: half-close/read
+  failures end lingering (`Failed`), a failed final close drops the transport.
+- Discard uses the connection's retained input buffer (`Vec` resized within its
+  capacity), so no allocation per read; the linger future waits on one read and one
+  timer (no spin, DESIGN §10 rule 4a). A `stop_by` deadline change costs one
+  nonblocking turn that delivers the replaced timer's `Cancelled`/`Closed`
+  (rule 3: at most one zero-timeout discovery poll beside the pending read).
+
+## Per platform
+
+| Backend | `AsyncIo::poll_shutdown` | Notes |
+|---|---|---|
+| epoll / kqueue (`unix.rs`) | TCP and Unix-domain stream sockets: `shutdown(SHUT_WR)` | ttys, FIFOs and regular files (`Kind::Stream`/`File`) now fail at submit with `Unsupported` (was an async `ENOTSOCK` completion / `Unsupported` from the file pool) |
+| IOCP | TCP: `shutdown(SD_SEND)` | named pipes, sync/console handles: `Unsupported` at submit (existing validation, now documented) |
+| WASI 0.2 | TCP: `tcp-socket.shutdown(send)`; stdout/stderr close their output stream | stdin/UDP/listeners: `InvalidInput` (existing) |
+| WASI 0.3 | TCP: closes the send stream and awaits its result | as 0.2 |
+| web | host WebSocket: starts its close; others `Unsupported` | no listeners on the web backend |
+
+UDP `AsyncIo` adapters return `Unsupported` before submitting. When the transport
+cannot half-close, `linger_close` closes immediately (`LingerEnd::Unsupported`).
+
+## Tests
+
+All deterministic: the server transport (`Gated` test wrapper) holds its first
+`poll_close`/`poll_shutdown` until the peer has written late bytes after the
+server's last read, and the peer reads nothing until the server's close/half-close
+completed. Unread bytes are therefore in the server's receive buffer at close time
+(or arrive at a closed socket, which also resets).
+
+| Test | Proves |
+|---|---|
+| `turnloop-http/asynchronous::http1_lingering_close_delivers_response_before_pipelined_request` (+ `https1_…`) | 32 KiB response then clean EOF with a pipelined request unread; server discarded exactly the request |
+| `…::http2_lingering_close_delivers_response_despite_unread_frames` (+ `https2_…`) | client sends only preface+request, then SETTINGS ACK + WINDOW_UPDATE + PING after the server drained; full 32 KiB DATA, END_STREAM, GOAWAY(0), clean EOF |
+| `…::http1_idle_keepalive_lingers_at_shutdown` | idle keep-alive connection at `stop()` half-closes and drains a request racing the stop |
+| `…::http_linger_timeout_closes_silent_peer_without_spinning` | peer never closes: server closes at `sent + 500 ms`, ≤ 1 turn and ≤ 1 zero-event wait for the expiry |
+| `…::http_stop_by_deadline_ends_lingering_connection` | 60 s linger ended by `stop_by(now + 100 ms)`; `run_ready()` shows exactly the lingering task woken; ≤ 2 turns, ≤ 2 zero-event waits |
+| `turnloop-http/server_allocations` (new harness=false gate, `integration-tests`) | HTTP/1 and HTTP/2 discard phase (half-close done → close): 0 allocations, 1,024 bytes, ≥ 256 reads (the peer lock-steps each chunk on the server's discard progress) |
+| `turnloop-websocket/asynchronous::server_close_lingers_until_client_eof` | server `close()` after the closing handshake drains late client bytes; client reads EOF |
+| `turnloop-io/streams::half_close_keeps_reading_until_peer_eof` | EOF delivered, idempotent shutdown, writes rejected, reads continue |
+| `…::lingering_close_discards_peer_input_until_eof` | `Lingered { end: Eof, discarded: 3072 }` |
+| `…::lingering_close_deadline_closes_silent_peer_without_spinning` | `end: Deadline` at the deadline, ≤ 1 turn / ≤ 1 zero-event wait |
+| `…::lingering_close_deadline_bounds_a_stalled_half_close` | a peer that never reads: `TimedOut` at the deadline instead of hanging |
+| `…::half_close_is_unsupported_on_datagram_adapters` | UDP: `Unsupported`, `LingerEnd::Unsupported` |
+| `turnloop-io/allocations::lingering_close_discards_without_allocating` | whole `linger_close` (half-close, ≥ 256 lock-stepped reads, timer, close) after a warm-up connection: 0 allocations |
+| `turnloop-tls/asynchronous::tls_half_close_sends_close_notify_and_keeps_decrypting` | close_notify reaches the peer as EOF, server still decrypts the peer's later data |
+
+Existing Node/curl interop tests are unchanged and pass.
+
+### Proof the tests fail without the fix
+
+Temporarily restoring the old `turnloop_io::close` in `server.rs` (both
+post-response sites), 3 runs each, every run failed:
+
+- `http1_…`, `http2_…` (plain): `reset before EOF after 32855 bytes: Connection reset by peer (os error 54)`.
+- `https1_…`, `https2_…`: client's close_notify on the reset socket: `Broken pipe (os error 32)`.
+- Old idle path (`shutdown.until(conn.head())` → return): `http1_idle_keepalive_lingers_at_shutdown` hangs (no graceful close ever) → `lingering exchange hung` after 10 s.
+- Old `WebSocketStream::close`: `server_close_lingers_until_client_eof` → `ConnectionReset (os error 54)`.
+
+Gate mutations (then restored, checksums verified):
+
+- `cx.waker().wake_by_ref()` before returning Pending from the discard step: io and
+  both HTTP deadline tests fail with `lingering close spun`.
+- `Box::new(n)` per discard read: io gate reports 256 allocations; one `Box` per
+  HTTP linger poll: server gate reports 257.
+
+## Verification
+
+Toolchain: pinned `nightly-2026-08-20`, `CARGO_BUILD_JOBS=4`, one heavy command at
+a time. WASI/web C for `ring`: Homebrew LLVM clang/llvm-ar via `CC_<target>`/`AR_<target>`
+(CI uses the pinned wasi-sdk). Wasmtime 46.0.0 installed by `scripts/ci/install-wasmtime.sh`.
+
+| Command | Result |
+|---|---|
+| `cargo fmt --all -- --check` | **PASS** |
+| `cargo clippy --locked --workspace --all-targets -- -D warnings -D clippy::undocumented_unsafe_blocks` | **PASS** |
+| same with `--all-features` | **PASS** |
+| `cargo clippy --locked -p turnloop -p turnloop-contract -p turnloop-io --all-targets --target x86_64-unknown-linux-gnu [--all-features] -- -D warnings -D clippy::undocumented_unsafe_blocks` | **PASS** (default and all-features) |
+| same for `--target x86_64-pc-windows-msvc` | **PASS** (default and all-features) |
+| same for `--target wasm32-wasip2 --all-features` (core crates) | **PASS** |
+| `cargo clippy --locked --workspace --all-targets --target wasm32-wasip2 --all-features -- -D warnings -D clippy::undocumented_unsafe_blocks` | **PASS** |
+| `cargo clippy --locked --workspace --all-targets --target wasm32-unknown-unknown --all-features -- …` | **PASS** |
+| `cargo test --workspace -- --test-threads=1` | **PASS**: 266 passed, 0 failed, 13 ignored |
+| `cargo test --workspace --all-features --no-fail-fast -- --test-threads=1` | **PASS**: 333 passed, 0 failed, 20 ignored (all new tests included) |
+| `python3 scripts/test-servers.py --services http run python3 scripts/ci/run-tests.py interop --mode default` | **PASS**: http interop 9, asynchronous 20 (incl. ignored Node fixture test), server_allocations 2, tls tls 5 / asynchronous 3 / async_allocations 1, websocket websocket 3 / asynchronous 3 / async_allocations 3 |
+| same `--mode executor` | **PASS** (same 9 suites and counts) |
+| same `--mode all-features` | **PASS** (same 9 suites and counts) |
+| `python3 scripts/ci/run-tests.py protocol-wasi --target wasm32-wasip2 --package turnloop-io --package turnloop-tls --package turnloop-http --package turnloop-websocket` | **PASS**: io streams 7, io allocations 2 (discard gate: 238 reads, 0 allocations), http asynchronous 16 (all 7 new), codecs 16, allocations 3, tls asynchronous 3, async_allocations 1, portable 3, channel_binding 3, websocket asynchronous 2, async_allocations 3 |
+| `bash scripts/ci/no-tokio.sh` | **PASS** (every policy triple, default and all-features) |
+| `python3 scripts/ci/soak.py` | **PASS**: 251 locked registry versions, 1 active security exception (inherited rustls 0.23.45) |
+| `python3 scripts/ci/check-paths.py`, `python3 scripts/ci/feature_modes.py`, `python3 -m unittest discover -s scripts/ci -p 'test_*.py'` | **PASS** (98 unit tests) |
+| No-fix and mutation probes above (run again on the final tree) | **FAIL as intended**, sources restored and checksummed |
+| Linux / Windows / WASI 0.3 / web runtime, h2spec, instruction baselines | **UNRUN** (integrator CI) |
+
+The linger-timeout tests were widened from 200 ms to 500 ms after the runs above
+for slow-runner headroom (the settle turns must not overrun the deadline); the
+affected io/http/tls/websocket suites were re-run afterwards: **PASS**.
+
+## CI iteration 1 (PR #23, run 34940794647)
+
+Two failures; everything else (macOS, Windows, WASI 0.2, web, h2spec) was green.
+
+1. **`test-native (ubuntu-24.04[-arm], all-features)`: `server_allocations`
+   "http1 discard reads coalesced".** Root cause: the gate paced late bytes with
+   200 µs sleeps and assumed the kernel would deliver most chunks as separate reads
+   (≥ 128 of 256). Linux coalesced more (the passing default-mode run already showed
+   146). Not a product defect: bytes and allocations were right. Fix: both
+   allocation gates now lock-step. A transparent server-transport wrapper publishes
+   the completed half-close and discarded bytes; the peer sends each 4-byte chunk
+   only after the previous one was discarded. That forces one read per chunk on
+   every kernel, so the bound is now `reads ≥ 256` (stronger), `discarded == 1024`
+   and `allocations == 0` (unchanged). Re-checked with an injected per-read `Box`:
+   both gates report 256 allocations.
+2. **`protocol-wasi (wasm32-wasip3)`: `http_linger_timeout_closes_silent_peer_without_spinning`.**
+   Reproduced locally with the pinned p3 toolchain, wasi-sdk 34 and Wasmtime 46:
+   11 of 12 runs panicked at `assert!(gate.shut.get(), "the peer's EOF came from the
+   server's half-close")`, about 6 ms into the test, so not a timer bound. Root cause:
+   the WASI 0.3 backend closes the send stream (the peer reads EOF at once) and
+   completes `Shutdown` only when that stream's result future resolves in a later
+   wait-set step, so the peer can finish before the server observes its own
+   half-close. That is correct half-close behaviour; the test assumed ordering.
+   Fix: the host turns until the server's half-close is acknowledged before settling
+   and measuring. The turnloop-io silent-peer test had the same latent assumption
+   (it passed CI by luck): its server now half-closes explicitly and publishes that,
+   and the host waits for it. The no-spin bounds (`turns < 2`, `≤ 1` zero-event wait;
+   `stop_by`: `< 3`, `≤ 2`) are unchanged on every target.
+
+Local re-verification after the fix:
+
+| Command | Result |
+|---|---|
+| p3 `cargo +nightly-2026-09-07 test -p turnloop-http --test asynchronous --all-features --target wasm32-wasip3 -- linger stop_by` ×15 | **PASS** 15/15 (7 tests each) |
+| `python3 scripts/ci/run-tests.py protocol-wasi --target wasm32-wasip3 --package turnloop-io --package turnloop-tls --package turnloop-http --package turnloop-websocket` | **PASS**, same 11 suites as p2; io gate 256 reads, 0 allocations |
+| `python3 scripts/ci/run-tests.py protocol-wasi --target wasm32-wasip2 --package turnloop-io --package turnloop-http` | **PASS** |
+| `cargo +nightly-2026-09-07 clippy --locked --workspace --all-targets --all-features --target wasm32-wasip3 -- -D warnings -D clippy::undocumented_unsafe_blocks` | **PASS** |
+| native `cargo fmt --check`, strict clippy (default, all-features), io streams/allocations, http asynchronous/server_allocations | **PASS** |
+
+## Follow-ups / not covered
+
+- HTTP/2 protocol-error closes (`Http2::event` drops the transport after flushing
+  GOAWAY) still close immediately; lingering there needs `Http2::event` to keep the
+  transport on error, which changes client semantics too.
+- `Shutdown::stop_by` bounds lingering only, not in-flight requests.
+- `server_allocations` is registered as a native `integration-tests` suite only, not
+  in `wasi-tests`; the turnloop-io discard gate already runs on WASI 0.2 and 0.3.
+- Linux, Windows (IOCP `SD_SEND`, named-pipe `Unsupported`), WASI 0.3 and web runtime
+  behaviour is covered only by CI; the macOS RST reproduction is the local evidence.
+  Without the fix the reset should surface on Linux and Windows too (as `ECONNRESET`
+  or lost bytes), but that failure mode was only reproduced on macOS.
diff --git a/protocols/turnloop-http/Cargo.toml b/protocols/turnloop-http/Cargo.toml
index b6a70ee..1981cd9 100644
--- a/protocols/turnloop-http/Cargo.toml
+++ b/protocols/turnloop-http/Cargo.toml
@@ -45,7 +45,7 @@ workspace = true
 [package.metadata.turnloop-ci]
 role = "protocol"
 service-group = "http"
-integration-tests = ["interop", "asynchronous"]
+integration-tests = ["interop", "asynchronous", "server_allocations"]
 wasi-tests = ["asynchronous", "codecs", "allocations"]
 
 [[test]]
@@ -68,5 +68,10 @@ required-features = ["turnloop"]
 name = "asynchronous"
 required-features = ["turnloop"]
 
+[[test]]
+name = "server_allocations"
+harness = false
+required-features = ["turnloop"]
+
 [target.'cfg(all(target_os = "wasi", target_env = "p3"))'.dependencies]
 turnloop-io = { workspace = true, optional = true, features = ["wasi-p3-experimental"] }
diff --git a/protocols/turnloop-http/README.md b/protocols/turnloop-http/README.md
index cee8945..ac5f54a 100644
--- a/protocols/turnloop-http/README.md
+++ b/protocols/turnloop-http/README.md
@@ -41,6 +41,16 @@ The HTTP/1 callback streams request events into a reusable Response encoder.
 HTTP/2 GOAWAY while existing streams drain. Wrap drain in an application deadline;
 dropping Server cancels its remaining service tasks.
 
+Connections close with a lingering close, as nginx `lingering_close` does: after the
+final response (and GOAWAY) is flushed, the server half-closes, reads and discards
+peer input until the peer's EOF, then closes. Closing with unread peer bytes (a
+pipelined request, HTTP/2 SETTINGS or WINDOW_UPDATE acknowledgements) would send
+RST, which can discard the peer's unread copy of the response on macOS and Windows.
+`server::Options::linger_timeout` (default 5 s, `Server::bind_with` or
+`Shutdown::with_options`) bounds the discard phase, and `Shutdown::stop_by(deadline)`
+closes every lingering connection by an overall shutdown deadline. Streams passed to
+`server::http1`/`http2` implement `turnloop_io::HalfClose`.
+
 ```sh
 cargo run -p turnloop-http --features turnloop --example https_get -- https://example.com/
 cargo run -p turnloop-http --features turnloop --example tls_echo -- cert.pem key.pem 127.0.0.1:8443
diff --git a/protocols/turnloop-http/src/asynchronous/server.rs b/protocols/turnloop-http/src/asynchronous/server.rs
index cbebb41..a2447ae 100644
--- a/protocols/turnloop-http/src/asynchronous/server.rs
+++ b/protocols/turnloop-http/src/asynchronous/server.rs
@@ -1,13 +1,40 @@
 //! Owned accept loop and explicit graceful shutdown. Services own each stream.
+//!
+//! A connection that ends after its final response closes with a lingering close:
+//! the response (and HTTP/2 GOAWAY) is flushed, the write side is shut down, peer
+//! input is read and discarded until the peer's EOF or the linger deadline, and
+//! only then is the socket closed. Closing with unread peer bytes (a pipelined
+//! request, HTTP/2 SETTINGS or WINDOW_UPDATE acknowledgements) would send RST,
+//! which on macOS and Windows discards the peer's unread copy of that response.
 use std::{
     cell::{Cell, RefCell},
     future::{Future, poll_fn},
     io,
-    pin::pin,
+    pin::{Pin, pin},
     rc::Rc,
     task::{Poll, Waker},
+    time::Duration,
 };
-use turnloop_io::{AsyncIo, Backend, ExecutorHandle, Listener, turnloop::executor::JoinHandle};
+use turnloop_io::{
+    AsyncIo, Backend, ExecutorHandle, HalfClose, Instant, Listener, turnloop::executor::JoinHandle,
+};
+/// Connection policy shared with every service through its [`Shutdown`] signal.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct Options {
+    /// Longest time a closing connection keeps reading and discarding peer input
+    /// after shutting down its write side, like nginx `lingering_timeout`. The
+    /// connection closes earlier at the peer's EOF or at the [`Shutdown::stop_by`]
+    /// deadline. Zero closes right after the half-close; `Duration::MAX` waits
+    /// for the peer's EOF. Default: five seconds.
+    pub linger_timeout: Duration,
+}
+impl Default for Options {
+    fn default() -> Self {
+        Self {
+            linger_timeout: Duration::from_secs(5),
+        }
+    }
+}
 /// Cloneable shutdown signal. Waiting futures register their own waker.
 #[derive(Clone, Default)]
 pub struct Shutdown {
@@ -16,54 +43,133 @@ pub struct Shutdown {
 #[derive(Default)]
 struct State {
     stopped: Cell<bool>,
+    options: Cell<Options>,
+    deadline: Cell<Option<Instant>>,
     waiters: RefCell<Vec<Option<Waker>>>,
 }
+/// One retained waiter slot; the slot is reused after the registration drops.
+struct Registration<'a> {
+    signal: &'a Shutdown,
+    index: usize,
+}
+impl Registration<'_> {
+    fn arm(&self, waker: &Waker) {
+        let mut waiters = self.signal.inner.waiters.borrow_mut();
+        if waiters[self.index]
+            .as_ref()
+            .is_none_or(|w| !w.will_wake(waker))
+        {
+            waiters[self.index] = Some(waker.clone());
+        }
+    }
+}
+impl Drop for Registration<'_> {
+    fn drop(&mut self) {
+        self.signal.inner.waiters.borrow_mut()[self.index] = None;
+    }
+}
 impl Shutdown {
+    /// A signal whose services use `options` (for example the linger timeout).
+    pub fn with_options(options: Options) -> Self {
+        let signal = Self::default();
+        signal.inner.options.set(options);
+        signal
+    }
+    /// The connection policy services read when they close.
+    pub fn options(&self) -> Options {
+        self.inner.options.get()
+    }
     pub fn stop(&self) {
         self.inner.stopped.set(true);
+        self.wake();
+    }
+    /// Stop, and close every lingering connection no later than `deadline`,
+    /// including connections that are already lingering. An earlier deadline
+    /// replaces a later one. In-flight requests still finish; drop the `Server`
+    /// to cancel them.
+    pub fn stop_by(&self, deadline: Instant) {
+        let deadline = self.deadline().map_or(deadline, |d| d.min(deadline));
+        self.inner.deadline.set(Some(deadline));
+        self.stop();
+    }
+    /// The overall shutdown deadline set by [`Shutdown::stop_by`].
+    pub fn deadline(&self) -> Option<Instant> {
+        self.inner.deadline.get()
+    }
+    pub fn is_stopped(&self) -> bool {
+        self.inner.stopped.get()
+    }
+    fn wake(&self) {
         let mut waiters = self.inner.waiters.borrow_mut();
         for w in waiters.iter_mut().filter_map(Option::take) {
             w.wake();
         }
     }
-    pub fn is_stopped(&self) -> bool {
-        self.inner.stopped.get()
-    }
-    /// Race one operation against shutdown. Cancellation drops the operation.
-    pub async fn until<F: Future>(&self, future: F) -> Option<F::Output> {
-        let index = {
-            let mut waiters = self.inner.waiters.borrow_mut();
-            if let Some(i) = waiters.iter().position(Option::is_none) {
-                i
-            } else {
-                waiters.push(None);
-                waiters.len() - 1
-            }
+    fn register(&self) -> Registration<'_> {
+        let mut waiters = self.inner.waiters.borrow_mut();
+        let index = if let Some(i) = waiters.iter().position(Option::is_none) {
+            i
+        } else {
+            waiters.push(None);
+            waiters.len() - 1
         };
-        struct Registration<'a> {
-            signal: &'a Shutdown,
-            index: usize,
-        }
-        impl Drop for Registration<'_> {
-            fn drop(&mut self) {
-                self.signal.inner.waiters.borrow_mut()[self.index] = None;
-            }
-        }
-        let _registration = Registration {
+        Registration {
             signal: self,
             index,
-        };
+        }
+    }
+    /// Race one operation against shutdown. Cancellation drops the operation.
+    pub async fn until<F: Future>(&self, future: F) -> Option<F::Output> {
+        let registration = self.register();
         let mut future = pin!(future);
         poll_fn(|cx| {
             if self.is_stopped() {
                 return Poll::Ready(None);
             }
-            self.inner.waiters.borrow_mut()[index] = Some(cx.waker().clone());
+            registration.arm(cx.waker());
             future.as_mut().poll(cx).map(Some)
         })
         .await
     }
 }
+/// Lingering close after the final output is flushed: half-close, discard peer
+/// input until its EOF or the linger/shutdown deadline, then close. A finished
+/// exchange never becomes an error here; a transport whose close fails is dropped,
+/// which releases its handle. `scratch` is the connection's retained input buffer.
+async fn linger<S: HalfClose>(slot: &mut Option<S>, scratch: &mut Vec<u8>, signal: &Shutdown) {
+    let Some(stream) = slot.as_mut() else {
+        return;
+    };
+    let Some(executor) = stream.executor() else {
+        if turnloop_io::close(stream).await.is_err() {
+            slot.take();
+        }
+        return;
+    };
+    let own = executor.now().checked_add(signal.options().linger_timeout);
+    let deadline = || match (own, signal.deadline()) {
+        (Some(a), Some(b)) => Some(a.min(b)),
+        (a, b) => a.or(b),
+    };
+    // Unread request bytes are discarded with the rest of the peer's input.
+    scratch.clear();
+    scratch.resize(scratch.capacity(), 0);
+    let failed = {
+        let registration = signal.register();
+        let mut close = turnloop_io::linger_close(stream, scratch, deadline());
+        poll_fn(|cx| {
+            // A later stop_by wakes this registration and moves the deadline.
+            registration.arm(cx.waker());
+            close.set_deadline(deadline());
+            Pin::new(&mut close).poll(cx)
+        })
+        .await
+        .is_err()
+    };
+    if failed {
+        slot.take();
+    }
+}
 /// TCP accept loop. Dropping the server cancels all in-flight services. Calling
 /// `shutdown().stop()` stops accepts and asks services to drain; `run` joins them.
 pub struct Server<B: Backend> {
@@ -74,10 +180,18 @@ pub struct Server<B: Backend> {
 }
 impl<B: Backend + 'static> Server<B> {
     pub fn bind(executor: ExecutorHandle<B>, address: std::net::SocketAddr) -> io::Result<Self> {
+        Self::bind_with(executor, address, Options::default())
+    }
+    /// Bind with an explicit connection policy, shared through [`Server::shutdown`].
+    pub fn bind_with(
+        executor: ExecutorHandle<B>,
+        address: std::net::SocketAddr,
+        options: Options,
+    ) -> io::Result<Self> {
         Ok(Self {
             listener: Some(Listener::bind(&executor, address)?),
             executor,
-            shutdown: Shutdown::default(),
+            shutdown: Shutdown::with_options(options),
             tasks: Vec::new(),
         })
     }
@@ -182,7 +296,10 @@ impl Response {
 }
 /// Serve HTTP/1 keep-alive with streaming request/response callbacks. Idle reads
 /// are cancelled on shutdown; an active request finishes before the connection closes.
-pub async fn http1<S: turnloop_io::Stream>(
+/// A connection that ends after a response (shutdown, `connection: close` or a
+/// non-reusable request) and an idle connection at shutdown close with a lingering
+/// close bounded by [`Options::linger_timeout`] and [`Shutdown::stop_by`].
+pub async fn http1<S: HalfClose>(
     stream: S,
     shutdown: Shutdown,
     mut service: impl FnMut(crate::http1::Event<'_>, &mut Response) -> io::Result<()>,
@@ -190,7 +307,22 @@ pub async fn http1<S: turnloop_io::Stream>(
     let mut conn = super::Http1::new(stream, crate::http1::Mode::Request);
     let mut response = Response::default();
     loop {
+        // Wait for the next request outside the codec: a cancelled codec read
+        // releases the transport, while an idle connection at shutdown lingers.
+        while conn.input.is_empty() && !shutdown.is_stopped() {
+            let stream = conn.stream.as_mut().ok_or_else(super::closed)?;
+            match shutdown.until(super::append(stream, &mut conn.input)).await {
+                // EOF: the codec below reports a clean or truncated end as before.
+                None | Some(Ok(0)) => break,
+                Some(read) => {
+                    read?;
+                }
+            }
+        }
         let Some(head) = shutdown.until(conn.head()).await else {
+            // Stopped before this request started. A stop that interrupted a
+            // partial head has already released the transport.
+            linger(&mut conn.stream, &mut conn.input, &shutdown).await;
             return Ok(());
         };
         let head = head?;
@@ -226,9 +358,7 @@ pub async fn http1<S: turnloop_io::Stream>(
             return Err(io::Error::other("service did not finish response"));
         }
         if shutdown.is_stopped() || !conn.reusable() || !response.keep_alive {
-            if let Some(stream) = conn.stream.as_mut() {
-                turnloop_io::close(stream).await?;
-            }
+            linger(&mut conn.stream, &mut conn.input, &shutdown).await;
             return Ok(());
         }
         conn.reset()?;
@@ -238,19 +368,19 @@ pub async fn http1<S: turnloop_io::Stream>(
 }
 /// Serve HTTP/2 and flush control frames even on protocol failure. Services own
 /// stream-level application state and release capacity after consuming DATA.
-/// Shutdown sends GOAWAY and continues processing existing streams until drained.
-pub async fn http2<S: turnloop_io::Stream>(
+/// Shutdown sends GOAWAY and continues processing existing streams until drained;
+/// a drained connection then closes with a lingering close (see [`http1`]).
+pub async fn http2<S: HalfClose>(
     stream: S,
     shutdown: Shutdown,
     mut service: impl FnMut(&mut crate::http2::Connection, crate::http2::Event<'_>) -> io::Result<()>,
 ) -> io::Result<()> {
     let mut conn = super::Http2::new(stream, crate::http2::Role::Server)?;
-    conn.shutdown = Some(shutdown);
+    conn.shutdown = Some(shutdown.clone());
     loop {
         if conn.core.is_drained() {
-            if let Some(stream) = conn.stream.as_mut() {
-                turnloop_io::close(stream).await?;
-            }
+            conn.flush().await?;
+            linger(&mut conn.stream, &mut conn.input, &shutdown).await;
             return Ok(());
         }
         if !conn.event(&mut service).await? {
diff --git a/protocols/turnloop-http/tests/asynchronous.rs b/protocols/turnloop-http/tests/asynchronous.rs
index 15f919a..9e47111 100644
--- a/protocols/turnloop-http/tests/asynchronous.rs
+++ b/protocols/turnloop-http/tests/asynchronous.rs
@@ -1158,3 +1158,677 @@ fn graceful_goaway_drains_existing_pooled_request() {
     assert_eq!(finish(&mut server), 11);
     assert_eq!(finish(&mut client), 11);
 }
+
+// Lingering close. A server that closes with unread peer bytes sends RST instead
+// of FIN; the peer then reads ECONNRESET instead of EOF, and on macOS and Windows
+// the reset can discard response bytes it has not read yet (issue #21). These
+// tests hold every peer read until the server has finished its final close or
+// half-close, and make the peer's late bytes unread at that moment, so they
+// observe the reset deterministically rather than by timing luck.
+const LINGER_BODY: usize = 32 * 1024;
+const TLS_NOW: u64 = 1_789_344_000;
+type Wire = turnloop_tls::asynchronous::Transport<AsyncIo<Platform>>;
+fn linger_body() -> Vec<u8> {
+    (0..LINGER_BODY).map(|i| (i % 251) as u8).collect()
+}
+/// Rendezvous between a server transport and its test peer. The server holds its
+/// first close or half-close until the peer has sent bytes the server's protocol
+/// never reads, so they sit unread in the server's receive buffer when it closes.
+#[derive(Default)]
+struct CloseGate {
+    closing: std::cell::Cell<bool>,
+    late_sent: std::cell::Cell<bool>,
+    shut: std::cell::Cell<bool>,
+    released: std::cell::Cell<bool>,
+    discarded: std::cell::Cell<usize>,
+    server: std::cell::RefCell<Option<Waker>>,
+    peer: std::cell::RefCell<Option<Waker>>,
+}
+impl CloseGate {
+    fn wake(slot: &std::cell::RefCell<Option<Waker>>) {
+        if let Some(waker) = slot.borrow_mut().take() {
+            waker.wake();
+        }
+    }
+    async fn peer_until(&self, ready: impl Fn(&Self) -> bool) {
+        std::future::poll_fn(|cx| {
+            if ready(self) {
+                Poll::Ready(())
+            } else {
+                *self.peer.borrow_mut() = Some(cx.waker().clone());
+                Poll::Pending
+            }
+        })
+        .await
+    }
+    fn late_bytes_sent(&self) {
+        self.late_sent.set(true);
+        Self::wake(&self.server);
+    }
+}
+/// Server-side transport that applies the gate to both `poll_close` (a server
+/// without lingering) and `poll_shutdown` (a lingering server), and counts bytes
+/// read after the server's write side ended.
+struct Gated<S> {
+    inner: S,
+    gate: std::rc::Rc<CloseGate>,
+}
+impl<S> Gated<S> {
+    fn poll_gate(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        if !self.gate.closing.replace(true) {
+            CloseGate::wake(&self.gate.peer);
+        }
+        if self.gate.late_sent.get() {
+            Poll::Ready(())
+        } else {
+            *self.gate.server.borrow_mut() = Some(cx.waker().clone());
+            Poll::Pending
+        }
+    }
+    fn ended_writes(&self, released: bool) {
+        self.gate.shut.set(true);
+        self.gate.released.set(self.gate.released.get() || released);
+        CloseGate::wake(&self.gate.peer);
+    }
+}
+impl<S: Stream> AsyncRead for Gated<S> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bytes: &mut [u8],
+    ) -> Poll<std::io::Result<usize>> {
+        let this = self.get_mut();
+        let result = Pin::new(&mut this.inner).poll_read(cx, bytes);
+        if let (true, Poll::Ready(Ok(n))) = (this.gate.shut.get(), &result) {
+            this.gate.discarded.set(this.gate.discarded.get() + n);
+        }
+        result
+    }
+}
+impl<S: Stream> AsyncWrite for Gated<S> {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bytes: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        Pin::new(&mut self.get_mut().inner).poll_write(cx, bytes)
+    }
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.get_mut().inner).poll_flush(cx)
+    }
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        let this = self.get_mut();
+        std::task::ready!(this.poll_gate(cx));
+        std::task::ready!(Pin::new(&mut this.inner).poll_close(cx))?;
+        this.ended_writes(true);
+        Poll::Ready(Ok(()))
+    }
+}
+impl<S: HalfClose> HalfClose for Gated<S> {
+    type Backend = S::Backend;
+    fn executor(&self) -> Option<ExecutorHandle<S::Backend>> {
+        self.inner.executor()
+    }
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        let this = self.get_mut();
+        std::task::ready!(this.poll_gate(cx));
+        std::task::ready!(Pin::new(&mut this.inner).poll_shutdown(cx))?;
+        this.ended_writes(false);
+        Poll::Ready(Ok(()))
+    }
+}
+struct LingerPeer {
+    server: Option<turnloop_tls::ServerConfig>,
+    client: Option<turnloop_tls::ClientConfig>,
+}
+impl LingerPeer {
+    fn new(tls: bool, alpn: &[u8]) -> Self {
+        if !tls {
+            return Self {
+                server: None,
+                client: None,
+            };
+        }
+        use turnloop_tls::rustls::pki_types::PrivatePkcs8KeyDer;
+        let cert =
+            rcgen::generate_simple_self_signed(vec!["localhost".into()]).expect("certificate");
+        Self {
+            server: Some(
+                turnloop_tls::ServerConfig::new(
+                    vec![cert.cert.der().clone()],
+                    PrivatePkcs8KeyDer::from(cert.key_pair.serialize_der()).into(),
+                    vec![alpn.to_vec()],
+                    TLS_NOW,
+                )
+                .expect("server TLS config"),
+            ),
+            client: Some(
+                turnloop_tls::ClientConfig::new(
+                    turnloop_tls::ClientOptions {
+                        ca: Some(vec![cert.cert.der().clone()]),
+                        alpn: vec![alpn.to_vec()],
+                        ..Default::default()
+                    },
+                    TLS_NOW,
+                )
+                .expect("client TLS config"),
+            ),
+        }
+    }
+}
+async fn accept_wire(
+    listener: &Listener<Platform>,
+    tls: Option<turnloop_tls::ServerConfig>,
+    h: &ExecutorHandle<Platform>,
+    end: Instant,
+) -> Wire {
+    let stream = listener.accept().await.expect("accept");
+    match tls {
+        Some(config) => turnloop_tls::asynchronous::Transport::Tls(Box::new(
+            turnloop_tls::TlsStream::accept(stream, &config, h, end, TLS_NOW)
+                .await
+                .expect("server handshake"),
+        )),
+        None => turnloop_tls::asynchronous::Transport::Plain(stream),
+    }
+}
+async fn connect_wire(
+    address: std::net::SocketAddr,
+    tls: Option<turnloop_tls::ClientConfig>,
+    h: &ExecutorHandle<Platform>,
+    end: Instant,
+) -> Wire {
+    let stream = h
+        .connect(address, Default::default())
+        .await
+        .expect("connect");
+    match tls {
+        Some(config) => turnloop_tls::asynchronous::Transport::Tls(Box::new(
+            turnloop_tls::TlsStream::connect(
+                stream,
+                &config,
+                "localhost".try_into().expect("server name"),
+                h,
+                end,
+                TLS_NOW,
+            )
+            .await
+            .expect("client handshake"),
+        )),
+        None => turnloop_tls::asynchronous::Transport::Plain(stream),
+    }
+}
+/// Every byte until a clean EOF. `ECONNRESET` here is the bug under test.
+async fn read_until_eof<S: Stream>(stream: &mut S) -> Vec<u8> {
+    let mut received = Vec::new();
+    let mut bytes = [0; 4096];
+    loop {
+        let n = read(stream, &mut bytes)
+            .await
+            .unwrap_or_else(|e| panic!("reset before EOF after {} bytes: {e}", received.len()));
+        if n == 0 {
+            return received;
+        }
+        received.extend_from_slice(&bytes[..n]);
+    }
+}
+fn run_until<T, U>(
+    executor: &mut LocalExecutor<Platform>,
+    a: &mut turnloop::executor::JoinHandle<T>,
+    b: &mut turnloop::executor::JoinHandle<U>,
+) {
+    let end = executor.driver().now() + Duration::from_secs(10);
+    while !a.is_finished() || !b.is_finished() {
+        assert!(executor.driver().now() < end, "lingering exchange hung");
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+}
+
+const PIPELINED: &[u8] = b"GET /pipelined HTTP/1.1\r\nhost: localhost\r\n\r\n";
+fn http1_final_response_survives_pipelined_request(tls: bool) {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let sh = h.clone();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen");
+    let address = listener.local_addr().expect("address");
+    let peers = LingerPeer::new(tls, b"http/1.1");
+    let (server_tls, client_tls) = (peers.server, peers.client);
+    let gate = std::rc::Rc::new(CloseGate::default());
+    let server_gate = gate.clone();
+    let end = h.now() + Duration::from_secs(10);
+    let mut server = executor
+        .spawn_local(async move {
+            let wire = accept_wire(&listener, server_tls, &sh, end).await;
+            let body = linger_body();
+            let mut heads = 0;
+            let gated = Gated {
+                inner: wire,
+                gate: server_gate,
+            };
+            server::http1(gated, server::Shutdown::default(), |event, out| {
+                match event {
+                    Event::Head(head) => {
+                        assert_eq!(head.target, "/first", "pipelined request must stay unread");
+                        heads += 1;
+                        let head = Head {
+                            keep_alive: false,
+                            ..response()
+                        };
+                        out.start(&head, BodyLength::Known(LINGER_BODY as u64))?;
+                        out.body(&body)?;
+                    }
+                    Event::End => out.finish(&[])?,
+                    _ => {}
+                }
+                Ok(())
+            })
+            .await
+            .expect("final response and lingering close");
+            heads
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            let mut wire = connect_wire(address, client_tls, &h, end).await;
+            write_all(&mut wire, b"GET /first HTTP/1.1\r\nhost: localhost\r\n\r\n")
+                .await
+                .expect("request");
+            // Read nothing until the server finished writing and began to close.
+            gate.peer_until(|g| g.closing.get()).await;
+            write_all(&mut wire, PIPELINED)
+                .await
+                .expect("pipelined request");
+            gate.late_bytes_sent();
+            gate.peer_until(|g| g.shut.get()).await;
+            let response = read_until_eof(&mut wire).await;
+            // TLS: close_notify is a write, which fails on a connection already reset.
+            close(&mut wire)
+                .await
+                .expect("the server is still open for our close");
+            (response, gate)
+        })
+        .expect("spawn");
+    run_until(&mut executor, &mut server, &mut client);
+    let (response, gate) = finish(&mut client);
+    assert_eq!(finish(&mut server), 1);
+    let head_end = response
+        .windows(4)
+        .position(|w| w == b"\r\n\r\n")
+        .expect("response head")
+        + 4;
+    let head = std::str::from_utf8(&response[..head_end]).expect("ASCII head");
+    assert!(head.starts_with("HTTP/1.1 200"), "{head}");
+    assert!(head.contains("connection: close"), "{head}");
+    assert_eq!(
+        &response[head_end..],
+        linger_body(),
+        "every body byte arrives"
+    );
+    assert_eq!(
+        gate.discarded.get(),
+        PIPELINED.len(),
+        "server drained the pipelined request"
+    );
+    assert!(gate.released.get(), "server closed after its peer's EOF");
+}
+#[test]
+fn http1_lingering_close_delivers_response_before_pipelined_request() {
+    http1_final_response_survives_pipelined_request(false);
+}
+#[test]
+fn https1_lingering_close_delivers_response_before_pipelined_request() {
+    http1_final_response_survives_pipelined_request(true);
+}
+
+/// SETTINGS acknowledgement, connection WINDOW_UPDATE and PING: frames an HTTP/2
+/// client commonly sends after the server's final response and GOAWAY.
+const LATE_H2_FRAMES: &[u8] = &[
+    0, 0, 0, 4, 1, 0, 0, 0, 0, // SETTINGS ACK
+    0, 0, 4, 8, 0, 0, 0, 0, 0, 0, 1, 0, 0, // WINDOW_UPDATE stream 0, +65536
+    0, 0, 8, 6, 0, 0, 0, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8, // PING
+];
+fn http2_final_response_survives_unread_frames(tls: bool) {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let sh = h.clone();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen");
+    let address = listener.local_addr().expect("address");
+    let peers = LingerPeer::new(tls, b"h2");
+    let (server_tls, client_tls) = (peers.server, peers.client);
+    let gate = std::rc::Rc::new(CloseGate::default());
+    let server_gate = gate.clone();
+    let end = h.now() + Duration::from_secs(10);
+    let mut server = executor
+        .spawn_local(async move {
+            let wire = accept_wire(&listener, server_tls, &sh, end).await;
+            let body = linger_body();
+            let signal = server::Shutdown::default();
+            let stop = signal.clone();
+            let mut responses = 0;
+            let gated = Gated {
+                inner: wire,
+                gate: server_gate,
+            };
+            server::http2(gated, signal, |core, event| {
+                if let http2::Event::Headers {
+                    stream, end_stream, ..
+                } = event
+                {
+                    assert!(end_stream);
+                    let length = LINGER_BODY.to_string();
+                    core.send_headers(
+                        stream,
+                        &[
+                            Header::new(":status", "200"),
+                            Header::new("content-length", length),
+                        ],
+                        false,
+                    )
+                    .map_err(std::io::Error::other)?;
+                    let mut sent = 0;
+                    while sent < body.len() {
+                        let n = core
+                            .send_data(stream, &body[sent..], true)
+                            .map_err(std::io::Error::other)?;
+                        assert!(n > 0, "initial flow-control window covers the body");
+                        sent += n;
+                    }
+                    responses += 1;
+                    // GOAWAY, drain, then close: the final response of this connection.
+                    stop.stop();
+                }
+                Ok(())
+            })
+            .await
+            .expect("final response and lingering close");
+            responses
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            let mut wire = connect_wire(address, client_tls, &h, end).await;
+            let mut core =
+                http2::Connection::new(http2::Role::Client, Default::default()).expect("h2");
+            core.open(
+                &[
+                    Header::new(":method", "GET"),
+                    Header::new(":scheme", if tls { "https" } else { "http" }),
+                    Header::new(":authority", "localhost"),
+                    Header::new(":path", "/"),
+                ],
+                true,
+            )
+            .expect("request");
+            drain(&mut wire, &mut core)
+                .await
+                .expect("preface and request");
+            // Read nothing (not even SETTINGS) until the server drained and closes.
+            gate.peer_until(|g| g.closing.get()).await;
+            write_all(&mut wire, LATE_H2_FRAMES)
+                .await
+                .expect("late control frames");
+            gate.late_bytes_sent();
+            gate.peer_until(|g| g.shut.get()).await;
+            let wire_bytes = read_until_eof(&mut wire).await;
+            close(&mut wire)
+                .await
+                .expect("the server is still open for our close");
+            let (mut status, mut body, mut goaway, mut ended) = (None, Vec::new(), None, false);
+            let mut pos = 0;
+            while pos < wire_bytes.len() {
+                let step = core.receive(&wire_bytes[pos..]).expect("server frames");
+                assert!(step.consumed > 0 || step.event.is_some(), "truncated frame");
+                pos += step.consumed;
+                match step.event {
+                    Some(http2::Event::Headers { headers, .. }) => {
+                        status = headers
+                            .into_iter()
+                            .find(|h| h.name == ":status")
+                            .map(|h| h.value);
+                    }
+                    Some(http2::Event::Data {
+                        bytes, end_stream, ..
+                    }) => {
+                        body.extend_from_slice(bytes);
+                        ended |= end_stream;
+                    }
+                    Some(http2::Event::Goaway { code, .. }) => goaway = Some(code),
+                    _ => {}
+                }
+            }
+            assert_eq!(status.as_deref(), Some(b"200".as_slice()));
+            assert_eq!(goaway, Some(0), "graceful GOAWAY precedes EOF");
+            assert!(ended, "END_STREAM arrives");
+            (body, gate)
+        })
+        .expect("spawn");
+    run_until(&mut executor, &mut server, &mut client);
+    let (body, gate) = finish(&mut client);
+    assert_eq!(finish(&mut server), 1);
+    assert_eq!(body, linger_body(), "every DATA byte arrives");
+    assert_eq!(
+        gate.discarded.get(),
+        LATE_H2_FRAMES.len(),
+        "server drained late frames"
+    );
+    assert!(gate.released.get(), "server closed after its peer's EOF");
+}
+#[test]
+fn http2_lingering_close_delivers_response_despite_unread_frames() {
+    http2_final_response_survives_unread_frames(false);
+}
+#[test]
+fn https2_lingering_close_delivers_response_despite_unread_frames() {
+    http2_final_response_survives_unread_frames(true);
+}
+
+/// A peer that reads the final response but never closes: the server closes at
+/// its linger deadline (or an earlier `stop_by` deadline) and never spins.
+fn linger_deadline_with_silent_peer(options: server::Options, stop_by: Option<Duration>) {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let sh = h.clone();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen");
+    let address = listener.local_addr().expect("address");
+    let gate = std::rc::Rc::new(CloseGate::default());
+    // Nothing late to send: the half-close proceeds at once.
+    gate.late_sent.set(true);
+    let server_gate = gate.clone();
+    let signal = server::Shutdown::with_options(options);
+    assert_eq!(signal.options(), options);
+    let service_signal = signal.clone();
+    let mut server = executor
+        .spawn_local(async move {
+            let stream = listener.accept().await.expect("accept");
+            let gated = Gated {
+                inner: stream,
+                gate: server_gate,
+            };
+            server::http1(gated, service_signal, |event, out| {
+                match event {
+                    Event::Head(_) => out.start(
+                        &Head {
+                            keep_alive: false,
+                            ..response()
+                        },
+                        BodyLength::Known(4),
+                    )?,
+                    Event::End => {
+                        out.body(b"done")?;
+                        out.finish(&[])?;
+                    }
+                    _ => {}
+                }
+                Ok(())
+            })
+            .await
+            .expect("lingering close");
+            sh.now()
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            let mut stream = h
+                .connect(address, Default::default())
+                .await
+                .expect("connect");
+            let sent = h.now();
+            write_all(&mut stream, b"GET / HTTP/1.1\r\nhost: localhost\r\n\r\n")
+                .await
+                .expect("request");
+            let response = read_until_eof(&mut stream).await;
+            assert!(response.ends_with(b"done"));
+            // Keep the connection open and silent.
+            (stream, sent)
+        })
+        .expect("spawn");
+    let end = executor.driver().now() + Duration::from_secs(10);
+    while !client.is_finished() {
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    let (_silent_peer, sent) = finish(&mut client);
+    // The peer's EOF is the server's half-close, but the server may observe its
+    // own shutdown acknowledgement later: WASI 0.3 closes the send stream (the
+    // peer reads EOF) and completes Shutdown only once that stream's result
+    // future resolves in a later wait-set step. Deliver it before measuring.
+    while !gate.shut.get() {
+        assert!(!server.is_finished(), "server ended before its half-close");
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    for _ in 0..3 {
+        executor.turn(Timeout::Now).expect("settle queued events");
+    }
+    assert!(!server.is_finished(), "closed before the linger deadline");
+    let mut floor = sent + options.linger_timeout;
+    // One turn waits for the single expiry, with at most one zero-event wait.
+    let (mut turn_limit, mut empty_limit) = (2, 1);
+    if let Some(delay) = stop_by {
+        let at = executor.driver().now() + delay;
+        signal.stop_by(at);
+        assert_eq!(signal.deadline(), Some(at));
+        // A later deadline never extends an earlier one.
+        signal.stop_by(at + Duration::from_secs(60));
+        assert_eq!(signal.deadline(), Some(at));
+        floor = at;
+        // The deadline change woke exactly the lingering service; polling it
+        // re-arms its timer. The wait below is then the same single expiry.
+        assert_eq!(executor.run_ready(), 1, "stop_by must wake the linger");
+        assert!(!server.is_finished());
+        // Replacing the timer queues its Cancelled/Closed completions beside the
+        // pending read: one nonblocking turn with at most one empty discovery poll
+        // (DESIGN section 10, rule 3) delivers them before the single expiry wait.
+        (turn_limit, empty_limit) = (3, 2);
+    }
+    let (mut turns, mut empty) = (0, 0);
+    while !server.is_finished() {
+        assert!(turns < turn_limit, "lingering close spun");
+        let info = executor.turn(Timeout::Until(end)).expect("turn");
+        turns += 1;
+        empty += info.zero_event_waits;
+    }
+    assert!(
+        empty <= empty_limit,
+        "zero-event waits while lingering: {empty}"
+    );
+    let closed_at = finish(&mut server);
+    assert!(closed_at >= floor, "closed before the deadline");
+    assert!(gate.released.get(), "deadline closed the socket");
+    assert_eq!(gate.discarded.get(), 0);
+}
+#[test]
+fn http_linger_timeout_closes_silent_peer_without_spinning() {
+    linger_deadline_with_silent_peer(
+        server::Options {
+            linger_timeout: Duration::from_millis(500),
+        },
+        None,
+    );
+}
+#[test]
+fn http_stop_by_deadline_ends_lingering_connection() {
+    // The linger timeout alone would hold the connection for a minute.
+    linger_deadline_with_silent_peer(
+        server::Options {
+            linger_timeout: Duration::from_secs(60),
+        },
+        Some(Duration::from_millis(100)),
+    );
+}
+
+#[test]
+fn http1_idle_keepalive_lingers_at_shutdown() {
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listen");
+    let address = listener.local_addr().expect("address");
+    let gate = std::rc::Rc::new(CloseGate::default());
+    let server_gate = gate.clone();
+    let signal = server::Shutdown::default();
+    let stop = signal.clone();
+    let mut server = executor
+        .spawn_local(async move {
+            let stream = listener.accept().await.expect("accept");
+            let gated = Gated {
+                inner: stream,
+                gate: server_gate,
+            };
+            let mut heads = 0;
+            server::http1(gated, signal, |event, out| {
+                match event {
+                    Event::Head(_) => {
+                        heads += 1;
+                        out.start(&response(), BodyLength::Known(4))?;
+                    }
+                    Event::End => {
+                        out.body(b"idle")?;
+                        out.finish(&[])?;
+                    }
+                    _ => {}
+                }
+                Ok(())
+            })
+            .await
+            .expect("idle connection closes gracefully at shutdown");
+            heads
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            let mut stream = h
+                .connect(address, Default::default())
+                .await
+                .expect("connect");
+            write_all(&mut stream, b"GET / HTTP/1.1\r\nhost: localhost\r\n\r\n")
+                .await
+                .expect("request");
+            let mut response = Vec::new();
+            let mut bytes = [0; 256];
+            while !response.ends_with(b"\r\n\r\nidle") {
+                let n = read(&mut stream, &mut bytes).await.expect("response");
+                assert!(n > 0, "keep-alive response must not close");
+                response.extend_from_slice(&bytes[..n]);
+            }
+            // The connection is idle between requests when shutdown arrives.
+            stop.stop();
+            gate.peer_until(|g| g.closing.get()).await;
+            write_all(&mut stream, PIPELINED)
+                .await
+                .expect("request racing the shutdown");
+            gate.late_bytes_sent();
+            gate.peer_until(|g| g.shut.get()).await;
+            assert!(
+                read_until_eof(&mut stream).await.is_empty(),
+                "a request that raced shutdown is never answered"
+            );
+            close(&mut stream).await.expect("close");
+            gate
+        })
+        .expect("spawn");
+    run_until(&mut executor, &mut server, &mut client);
+    let gate = finish(&mut client);
+    assert_eq!(finish(&mut server), 1);
+    assert_eq!(gate.discarded.get(), PIPELINED.len());
+    assert!(gate.released.get());
+}
diff --git a/protocols/turnloop-http/tests/server_allocations.rs b/protocols/turnloop-http/tests/server_allocations.rs
new file mode 100644
index 0000000..d051e65
--- /dev/null
+++ b/protocols/turnloop-http/tests/server_allocations.rs
@@ -0,0 +1,322 @@
+//! Allocation gate for the HTTP servers' lingering close over real TCP. After a
+//! warm-up connection, the discard phase (from the completed half-close to the
+//! final close) must allocate nothing while the peer sends many late bytes.
+#![deny(unsafe_op_in_unsafe_fn)]
+#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
+mod native {
+    use std::{
+        alloc::{GlobalAlloc, Layout, System},
+        cell::{Cell, RefCell},
+        future::Future,
+        io,
+        pin::Pin,
+        rc::Rc,
+        sync::atomic::{AtomicBool, AtomicUsize, Ordering},
+        task::{Context, Poll, Waker},
+        time::Duration,
+    };
+    use turnloop_http::{
+        asynchronous::server,
+        http1::{BodyLength, Event, Head, Header},
+        http2,
+    };
+    use turnloop_io::{
+        turnloop::{Config, LocalExecutor, Timeout, backend::Platform},
+        *,
+    };
+    static ACTIVE: AtomicBool = AtomicBool::new(false);
+    static COUNT: AtomicUsize = AtomicUsize::new(0);
+    struct Counter;
+    fn record() {
+        if ACTIVE.load(Ordering::Relaxed) {
+            COUNT.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+    // SAFETY: the allocator forwards every layout and pointer unchanged to System.
+    unsafe impl GlobalAlloc for Counter {
+        unsafe fn alloc(&self, l: Layout) -> *mut u8 {
+            record();
+            // SAFETY: caller-provided allocation layout is forwarded unchanged.
+            unsafe { System.alloc(l) }
+        }
+        unsafe fn alloc_zeroed(&self, l: Layout) -> *mut u8 {
+            record();
+            // SAFETY: caller-provided allocation layout is forwarded unchanged.
+            unsafe { System.alloc_zeroed(l) }
+        }
+        unsafe fn realloc(&self, p: *mut u8, l: Layout, n: usize) -> *mut u8 {
+            record();
+            // SAFETY: pointer, original layout and new size satisfy GlobalAlloc.
+            unsafe { System.realloc(p, l, n) }
+        }
+        unsafe fn dealloc(&self, p: *mut u8, l: Layout) {
+            // SAFETY: pointer/layout belong to a prior System allocation.
+            unsafe { System.dealloc(p, l) }
+        }
+    }
+    #[global_allocator]
+    static ALLOCATOR: Counter = Counter;
+
+    const LATE: usize = 256;
+    /// Discard-phase accounting shared by the server transport and its peer.
+    #[derive(Default)]
+    pub struct Phase {
+        measure: Cell<bool>,
+        lingering: Cell<bool>,
+        /// Completed half-closes; each resets the discard counters.
+        shutdowns: Cell<u32>,
+        peer: RefCell<Option<Waker>>,
+        pub reads: Cell<u64>,
+        pub discarded: Cell<u64>,
+        pub allocations: Cell<usize>,
+        pub measured: Cell<bool>,
+    }
+    impl Phase {
+        fn wake_peer(&self) {
+            if let Some(waker) = self.peer.borrow_mut().take() {
+                waker.wake();
+            }
+        }
+        /// Peer side: wait until the server-side accounting satisfies `ready`.
+        async fn until(&self, ready: impl Fn(&Self) -> bool) {
+            std::future::poll_fn(|cx| {
+                if ready(self) {
+                    Poll::Ready(())
+                } else {
+                    *self.peer.borrow_mut() = Some(cx.waker().clone());
+                    Poll::Pending
+                }
+            })
+            .await
+        }
+    }
+    /// Starts counting when the server's half-close completes and stops when its
+    /// lingering close reaches `poll_close`, so exactly the discard phase counts.
+    struct Metered {
+        inner: AsyncIo<Platform>,
+        phase: Rc<Phase>,
+    }
+    impl AsyncRead for Metered {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            bytes: &mut [u8],
+        ) -> Poll<io::Result<usize>> {
+            let this = self.get_mut();
+            let result = Pin::new(&mut this.inner).poll_read(cx, bytes);
+            if let (true, Poll::Ready(Ok(n @ 1..))) = (this.phase.lingering.get(), &result) {
+                this.phase.reads.set(this.phase.reads.get() + 1);
+                this.phase
+                    .discarded
+                    .set(this.phase.discarded.get() + *n as u64);
+                this.phase.wake_peer();
+            }
+            result
+        }
+    }
+    impl AsyncWrite for Metered {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            bytes: &[u8],
+        ) -> Poll<io::Result<usize>> {
+            Pin::new(&mut self.get_mut().inner).poll_write(cx, bytes)
+        }
+        fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Pin::new(&mut self.get_mut().inner).poll_flush(cx)
+        }
+        fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            let this = self.get_mut();
+            if this.phase.lingering.replace(false) && this.phase.measure.get() {
+                ACTIVE.store(false, Ordering::Relaxed);
+                this.phase.allocations.set(COUNT.load(Ordering::Relaxed));
+                this.phase.measured.set(true);
+            }
+            Pin::new(&mut this.inner).poll_close(cx)
+        }
+    }
+    impl HalfClose for Metered {
+        type Backend = Platform;
+        fn executor(&self) -> Option<ExecutorHandle<Platform>> {
+            HalfClose::executor(&self.inner)
+        }
+        fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            let this = self.get_mut();
+            std::task::ready!(HalfClose::poll_shutdown(Pin::new(&mut this.inner), cx))?;
+            if !this.phase.lingering.replace(true) {
+                this.phase.reads.set(0);
+                this.phase.discarded.set(0);
+                this.phase.shutdowns.set(this.phase.shutdowns.get() + 1);
+                this.phase.wake_peer();
+                if this.phase.measure.get() {
+                    COUNT.store(0, Ordering::Relaxed);
+                    ACTIVE.store(true, Ordering::Relaxed);
+                }
+            }
+            Poll::Ready(Ok(()))
+        }
+    }
+    fn finish<T>(task: &mut turnloop::executor::JoinHandle<T>) -> T {
+        match Pin::new(task).poll(&mut Context::from_waker(Waker::noop())) {
+            Poll::Ready(Ok(v)) => v,
+            _ => panic!("task incomplete"),
+        }
+    }
+    /// Read the whole response (through the server's half-close), then send late
+    /// bytes the server only discards, then close. Lock-step: each chunk is sent
+    /// only after the server discarded the previous one, so every chunk is its own
+    /// discard read regardless of how the kernel would coalesce paced writes.
+    async fn peer(
+        h: &ExecutorHandle<Platform>,
+        address: std::net::SocketAddr,
+        request: &[u8],
+        phase: &Phase,
+        round: u32,
+    ) {
+        let mut stream = h
+            .connect(address, Default::default())
+            .await
+            .expect("connect");
+        write_all(&mut stream, request).await.expect("request");
+        let mut bytes = [0; 1024];
+        let mut total = 0;
+        loop {
+            let n = read(&mut stream, &mut bytes).await.expect("clean EOF");
+            if n == 0 {
+                break;
+            }
+            total += n;
+        }
+        assert!(total > 0, "final response");
+        // The server may observe its shutdown acknowledgement after our EOF.
+        phase.until(|p| p.shutdowns.get() > round).await;
+        for chunk in 1..=LATE as u64 {
+            write_all(&mut stream, b"late").await.expect("late bytes");
+            phase.until(|p| p.discarded.get() >= 4 * chunk).await;
+        }
+        close(&mut stream).await.expect("close");
+    }
+    fn h2_request() -> Vec<u8> {
+        let mut core = http2::Connection::new(http2::Role::Client, Default::default()).expect("h2");
+        core.open(
+            &[
+                Header::new(":method", "GET"),
+                Header::new(":scheme", "http"),
+                Header::new(":authority", "localhost"),
+                Header::new(":path", "/"),
+            ],
+            true,
+        )
+        .expect("request");
+        core.output().to_vec()
+    }
+    pub fn lingering_discard_allocates_zero(h2: bool) -> Rc<Phase> {
+        let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+        let h = executor.handle();
+        let client_h = h.clone();
+        let listener =
+            Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listener");
+        let address = listener.local_addr().expect("address");
+        let phase = Rc::new(Phase::default());
+        let server_phase = phase.clone();
+        let mut server = executor
+            .spawn_local(async move {
+                for round in 0..2 {
+                    let stream = listener.accept().await.expect("accept");
+                    server_phase.measure.set(round == 1);
+                    let metered = Metered {
+                        inner: stream,
+                        phase: server_phase.clone(),
+                    };
+                    // A fresh signal per connection: stopping ends only this one.
+                    let signal = server::Shutdown::default();
+                    if h2 {
+                        let stop = signal.clone();
+                        server::http2(metered, signal, |core, event| {
+                            if let http2::Event::Headers { stream, .. } = event {
+                                core.send_headers(stream, &[Header::new(":status", "204")], true)
+                                    .map_err(io::Error::other)?;
+                                stop.stop();
+                            }
+                            Ok(())
+                        })
+                        .await
+                        .expect("h2 lingering close");
+                    } else {
+                        server::http1(metered, signal, |event, out| {
+                            match event {
+                                Event::Head(_) => out.start(
+                                    &Head {
+                                        method: String::new(),
+                                        target: String::new(),
+                                        status: 200,
+                                        version: 1,
+                                        headers: vec![],
+                                        keep_alive: false,
+                                    },
+                                    BodyLength::Known(0),
+                                )?,
+                                Event::End => out.finish(&[])?,
+                                _ => {}
+                            }
+                            Ok(())
+                        })
+                        .await
+                        .expect("h1 lingering close");
+                    }
+                    assert!(
+                        !server_phase.lingering.get(),
+                        "lingering close reached close"
+                    );
+                }
+            })
+            .expect("spawn");
+        let request = if h2 {
+            h2_request()
+        } else {
+            b"GET / HTTP/1.1\r\nhost: localhost\r\n\r\n".to_vec()
+        };
+        let peer_phase = phase.clone();
+        let mut client = executor
+            .spawn_local(async move {
+                for round in 0..2 {
+                    peer(&client_h, address, &request, &peer_phase, round).await;
+                }
+            })
+            .expect("spawn");
+        let end = executor.driver().now() + Duration::from_secs(20);
+        while !client.is_finished() || !server.is_finished() {
+            assert!(executor.driver().now() < end);
+            executor.turn(Timeout::Until(end)).expect("turn");
+        }
+        finish(&mut client);
+        finish(&mut server);
+        phase
+    }
+}
+fn main() {
+    #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
+    for (name, h2) in [("http1", false), ("http2", true)] {
+        let phase = native::lingering_discard_allocates_zero(h2);
+        // The subject ran: at least one discard read per lock-stepped chunk,
+        // covering every late byte, all inside the measured window.
+        assert!(phase.measured.get(), "{name} discard phase was measured");
+        assert_eq!(phase.discarded.get(), 4 * 256, "{name} discarded bytes");
+        assert!(
+            phase.reads.get() >= 256,
+            "{name} one discard read per chunk"
+        );
+        assert_eq!(
+            phase.allocations.get(),
+            0,
+            "{name} lingering discard allocates"
+        );
+        println!(
+            "{name}: {} discard reads, {} bytes, 0 allocations",
+            phase.reads.get(),
+            phase.discarded.get()
+        );
+        println!("test {name}_lingering_discard_allocates_zero ... ok");
+    }
+    println!("test result: ok. 2 passed; 0 failed; 0 ignored;");
+}
diff --git a/protocols/turnloop-tls/README.md b/protocols/turnloop-tls/README.md
index 3bc3200..19627e8 100644
--- a/protocols/turnloop-tls/README.md
+++ b/protocols/turnloop-tls/README.md
@@ -21,7 +21,9 @@ an owned stream, shared TLS configuration, executor handle, absolute handshake
 deadline and host wall time. Await the handshake, inspect `alpn_protocol()`, then
 use `turnloop_io::{read, write_all, close}`. These helpers work on TCP and pipes
 and compose with HTTP/WebSocket adapters. `close` flushes close_notify; dropping
-aborts the transport. Refresh host wall time with `set_unix_seconds` when needed.
+aborts the transport. `turnloop_io::shutdown` flushes close_notify and half-closes
+the transport while decryption continues until the peer's close_notify or EOF, which
+is what a lingering server close uses. Refresh host wall time with `set_unix_seconds` when needed.
 The host continues calling `LocalExecutor::turn` throughout the connection.
 
 After a successful handshake, `peer_certificates()` borrows the peer chain,
diff --git a/protocols/turnloop-tls/src/asynchronous.rs b/protocols/turnloop-tls/src/asynchronous.rs
index 55ac029..3f3f59b 100644
--- a/protocols/turnloop-tls/src/asynchronous.rs
+++ b/protocols/turnloop-tls/src/asynchronous.rs
@@ -8,7 +8,7 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
-use turnloop_io::{AsyncRead, AsyncWrite, Backend, ExecutorHandle, Instant, Stream};
+use turnloop_io::{AsyncRead, AsyncWrite, Backend, ExecutorHandle, HalfClose, Instant, Stream};
 
 /// Certificate verification and wall time supplied by the embedding host.
 #[derive(Clone)]
@@ -108,6 +108,24 @@ impl<S: Stream> AsyncWrite for Transport<S> {
         }
     }
 }
+impl<S: HalfClose> HalfClose for Transport<S> {
+    type Backend = S::Backend;
+    fn executor(&self) -> Option<ExecutorHandle<S::Backend>> {
+        match self {
+            Self::Plain(s) => s.executor(),
+            Self::Tls(s) => s.executor(),
+            Self::Closed => None,
+        }
+    }
+    /// Half-close the active transport: TLS sends close_notify first.
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        match self.get_mut() {
+            Self::Plain(s) => Pin::new(s).poll_shutdown(cx),
+            Self::Tls(s) => Pin::new(&mut **s).poll_shutdown(cx),
+            Self::Closed => Poll::Ready(Err(io::ErrorKind::NotConnected.into())),
+        }
+    }
+}
 
 trait Endpoint {
     type Data;
@@ -267,7 +285,9 @@ fn step<E: Endpoint>(
 }
 /// An async client or server TLS connection. Construction completes the handshake
 /// under a single absolute deadline. Drop aborts the owned transport; `close`
-/// flushes close_notify first. Retained storage is bounded per connection.
+/// flushes close_notify first. [`HalfClose::poll_shutdown`] flushes close_notify
+/// and then half-closes the transport while decrypting reads continue until the
+/// peer's close_notify or EOF. Retained storage is bounded per connection.
 pub struct TlsStream<S> {
     stream: S,
     session: Session,
@@ -449,6 +469,14 @@ impl<S: Stream> AsyncWrite for TlsStream<S> {
         if bytes.is_empty() {
             return Poll::Ready(Ok(0));
         }
+        if self.buffers.close_sent {
+            // No application data may follow close_notify. Refuse before
+            // encrypting, so a rejected record never blocks later reads.
+            return Poll::Ready(Err(io::Error::new(
+                io::ErrorKind::BrokenPipe,
+                "TLS close_notify already sent",
+            )));
+        }
         self.get_mut().poll_drive(cx, &Intent::Write(bytes))
     }
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
@@ -461,3 +489,18 @@ impl<S: Stream> AsyncWrite for TlsStream<S> {
         Pin::new(&mut this.stream).poll_close(cx)
     }
 }
+impl<S: HalfClose> HalfClose for TlsStream<S> {
+    type Backend = S::Backend;
+    fn executor(&self) -> Option<ExecutorHandle<S::Backend>> {
+        self.stream.executor()
+    }
+    /// Send and flush close_notify, then half-close the transport (TCP `SHUT_WR`).
+    /// The handle stays open: reads keep decrypting until the peer's close_notify
+    /// or EOF. Repeated polls after completion return `Ok`.
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = self.get_mut();
+        std::task::ready!(this.poll_drive(cx, &Intent::Close))?;
+        std::task::ready!(this.poll_output(cx))?;
+        Pin::new(&mut this.stream).poll_shutdown(cx)
+    }
+}
diff --git a/protocols/turnloop-tls/tests/asynchronous.rs b/protocols/turnloop-tls/tests/asynchronous.rs
index 731947f..e1395f2 100644
--- a/protocols/turnloop-tls/tests/asynchronous.rs
+++ b/protocols/turnloop-tls/tests/asynchronous.rs
@@ -106,6 +106,91 @@ fn async_tls_alpn_fragmented_plaintext_and_close_notify() {
     assert_eq!(finish(&mut s), finish(&mut c));
 }
 #[test]
+fn tls_half_close_sends_close_notify_and_keeps_decrypting() {
+    let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).expect("certificate");
+    let server = ServerConfig::new(
+        vec![cert.cert.der().clone()],
+        PrivatePkcs8KeyDer::from(cert.key_pair.serialize_der()).into(),
+        vec![b"h2".to_vec()],
+        NOW,
+    )
+    .expect("server config");
+    let client = ClientConfig::new(
+        ClientOptions {
+            ca: Some(vec![cert.cert.der().clone()]),
+            alpn: vec![b"h2".to_vec()],
+            ..Default::default()
+        },
+        NOW,
+    )
+    .expect("client config");
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let h2 = h.clone();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listener");
+    let address = listener.local_addr().expect("address");
+    let end = h.now() + Duration::from_secs(5);
+    async fn read_to_end<S: Stream>(stream: &mut S) -> Vec<u8> {
+        let mut received = Vec::new();
+        let mut bytes = [0; 256];
+        loop {
+            let n = read(stream, &mut bytes).await.expect("clean TLS EOF");
+            if n == 0 {
+                return received;
+            }
+            received.extend_from_slice(&bytes[..n]);
+        }
+    }
+    let mut s = executor
+        .spawn_local(async move {
+            let stream = listener.accept().await.expect("accept");
+            let mut tls = TlsStream::accept(stream, &server, &h, end, NOW)
+                .await
+                .expect("server handshake");
+            write_all(&mut tls, b"final").await.expect("write");
+            // close_notify, then TCP SHUT_WR; the handle and decryption stay open.
+            shutdown(&mut tls).await.expect("TLS half-close");
+            shutdown(&mut tls).await.expect("repeated half-close");
+            assert!(tls.get_ref().is_write_shut());
+            assert!(write_all(&mut tls, b"x").await.is_err());
+            let late = read_to_end(&mut tls).await;
+            close(&mut tls).await.expect("close");
+            late
+        })
+        .expect("spawn");
+    let mut c = executor
+        .spawn_local(async move {
+            let stream = h2
+                .connect(address, Default::default())
+                .await
+                .expect("connect");
+            let mut tls = TlsStream::connect(
+                stream,
+                &client,
+                ServerName::try_from("localhost").expect("name"),
+                &h2,
+                end,
+                NOW,
+            )
+            .await
+            .expect("client handshake");
+            // EOF here is the server's close_notify.
+            let response = read_to_end(&mut tls).await;
+            write_all(&mut tls, b"late")
+                .await
+                .expect("write to a half-closed peer");
+            close(&mut tls).await.expect("close notify");
+            response
+        })
+        .expect("spawn");
+    while !s.is_finished() || !c.is_finished() {
+        assert!(executor.driver().now() < end);
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    assert_eq!(finish(&mut c), b"final");
+    assert_eq!(finish(&mut s), b"late");
+}
+#[test]
 fn silent_peer_handshake_deadline() {
     let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
     let h = executor.handle();
diff --git a/protocols/turnloop-websocket/README.md b/protocols/turnloop-websocket/README.md
index e9ee806..1fb6191 100644
--- a/protocols/turnloop-websocket/README.md
+++ b/protocols/turnloop-websocket/README.md
@@ -19,8 +19,10 @@ an HTTP upgrade on a connected TCP, pipe or TLS stream; supply a fresh random
 nonce, subprotocol list, executor handle and absolute deadline. `accept` performs
 the server upgrade. Both preserve coalesced first-frame bytes. Await `send` and
 `receive`; ping/pong and close replies flush before received messages are returned.
-`close` waits for the peer under a deadline. Dropping a pending frame operation
-closes the stream. The host keeps turning its LocalExecutor.
+`close` waits for the peer under a deadline, then closes the transport with a
+lingering close under the same deadline (half-close, discard until the peer's EOF),
+so a late peer frame cannot turn the close into a reset. Dropping a pending frame
+operation closes the stream. The host keeps turning its LocalExecutor.
 
 ```sh
 cargo run -p turnloop-websocket --features turnloop --example echo_server -- 127.0.0.1:8080
diff --git a/protocols/turnloop-websocket/src/asynchronous.rs b/protocols/turnloop-websocket/src/asynchronous.rs
index 4291cfe..6d3a1ad 100644
--- a/protocols/turnloop-websocket/src/asynchronous.rs
+++ b/protocols/turnloop-websocket/src/asynchronous.rs
@@ -5,7 +5,7 @@ use turnloop_http::{
     asynchronous::Http1,
     http1::{BodyLength, Event, Mode},
 };
-use turnloop_io::{Backend, CloseOnDrop, ExecutorHandle, Instant, Stream};
+use turnloop_io::{Backend, CloseOnDrop, ExecutorHandle, HalfClose, Instant, Stream};
 /// A framed connection with one retained input/output buffer. A cancelled frame
 /// operation closes the stream, preventing partial frames from being replayed.
 pub struct WebSocketStream<S> {
@@ -150,8 +150,14 @@ impl<S: Stream> WebSocketStream<S> {
             self.input.extend_from_slice(&bytes[..n]);
         }
     }
-    /// Send close, await its peer acknowledgement under the supplied deadline,
-    /// then close the underlying stream (including TLS close_notify when present).
+}
+impl<S: HalfClose> WebSocketStream<S> {
+    /// Send close and await the peer's close under the supplied deadline, then
+    /// close the underlying stream with a lingering close bounded by the same
+    /// deadline: half-close (TLS close_notify first), discard peer bytes until its
+    /// EOF, then close. A server's TCP close therefore never resets a peer that
+    /// still has unread frames; a client waits for the server's close, as RFC 6455
+    /// section 7.1.1 recommends. Only the close handshake can time out.
     pub async fn close<B: Backend>(
         &mut self,
         executor: &ExecutorHandle<B>,
@@ -164,16 +170,26 @@ impl<S: Stream> WebSocketStream<S> {
                     break;
                 }
             }
-            if let Some(stream) = &mut self.stream {
-                turnloop_io::close(stream).await?;
-            }
-            self.closed = true;
             Ok(())
         })
         .await;
         if result.is_err() {
             self.stream.take();
+            return result;
         }
-        result
+        self.closed = true;
+        if let Some(stream) = &mut self.stream {
+            // The retained frame buffer is the discard scratch; no allocation.
+            self.input.clear();
+            self.input.resize(self.input.capacity(), 0);
+            if turnloop_io::linger_close(stream, &mut self.input, Some(deadline))
+                .await
+                .is_err()
+            {
+                self.stream.take();
+            }
+            self.input.clear();
+        }
+        Ok(())
     }
 }
diff --git a/protocols/turnloop-websocket/tests/asynchronous.rs b/protocols/turnloop-websocket/tests/asynchronous.rs
index 487e241..40ac50c 100644
--- a/protocols/turnloop-websocket/tests/asynchronous.rs
+++ b/protocols/turnloop-websocket/tests/asynchronous.rs
@@ -130,3 +130,208 @@ fn node_websocket_against_async_server() {
     let status = node.wait().expect("Node exit");
     assert!(status.success(), "Node status: {status}");
 }
+
+/// Server-side gate: its TCP close (or half-close) waits until the peer has sent
+/// bytes the server never reads, so they are unread when it closes. Without a
+/// lingering close that close sends RST and the peer reads ECONNRESET, not EOF.
+#[derive(Default)]
+struct CloseGate {
+    closing: std::cell::Cell<bool>,
+    late_sent: std::cell::Cell<bool>,
+    shut: std::cell::Cell<bool>,
+    discarded: std::cell::Cell<usize>,
+    server: std::cell::RefCell<Option<Waker>>,
+    peer: std::cell::RefCell<Option<Waker>>,
+}
+fn wake(slot: &std::cell::RefCell<Option<Waker>>) {
+    if let Some(waker) = slot.borrow_mut().take() {
+        waker.wake();
+    }
+}
+async fn peer_until(gate: &CloseGate, ready: impl Fn(&CloseGate) -> bool) {
+    std::future::poll_fn(|cx| {
+        if ready(gate) {
+            Poll::Ready(())
+        } else {
+            *gate.peer.borrow_mut() = Some(cx.waker().clone());
+            Poll::Pending
+        }
+    })
+    .await
+}
+struct Gated {
+    inner: AsyncIo<turnloop::backend::Platform>,
+    gate: std::rc::Rc<CloseGate>,
+}
+impl Gated {
+    fn poll_gate(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        if !self.gate.closing.replace(true) {
+            wake(&self.gate.peer);
+        }
+        if self.gate.late_sent.get() {
+            Poll::Ready(())
+        } else {
+            *self.gate.server.borrow_mut() = Some(cx.waker().clone());
+            Poll::Pending
+        }
+    }
+    fn ended_writes(&self) {
+        self.gate.shut.set(true);
+        wake(&self.gate.peer);
+    }
+}
+impl AsyncRead for Gated {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bytes: &mut [u8],
+    ) -> Poll<std::io::Result<usize>> {
+        let this = self.get_mut();
+        let result = Pin::new(&mut this.inner).poll_read(cx, bytes);
+        if let (true, Poll::Ready(Ok(n))) = (this.gate.shut.get(), &result) {
+            this.gate.discarded.set(this.gate.discarded.get() + n);
+        }
+        result
+    }
+}
+impl AsyncWrite for Gated {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bytes: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        Pin::new(&mut self.get_mut().inner).poll_write(cx, bytes)
+    }
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.get_mut().inner).poll_flush(cx)
+    }
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        let this = self.get_mut();
+        std::task::ready!(this.poll_gate(cx));
+        std::task::ready!(Pin::new(&mut this.inner).poll_close(cx))?;
+        this.ended_writes();
+        Poll::Ready(Ok(()))
+    }
+}
+impl HalfClose for Gated {
+    type Backend = turnloop::backend::Platform;
+    fn executor(&self) -> Option<ExecutorHandle<Self::Backend>> {
+        HalfClose::executor(&self.inner)
+    }
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        let this = self.get_mut();
+        std::task::ready!(this.poll_gate(cx));
+        std::task::ready!(HalfClose::poll_shutdown(Pin::new(&mut this.inner), cx))?;
+        this.ended_writes();
+        Poll::Ready(Ok(()))
+    }
+}
+/// Client transport shared with the test body, so raw bytes can follow the
+/// WebSocket closing handshake on the same socket.
+#[derive(Clone)]
+struct Shared(std::rc::Rc<std::cell::RefCell<AsyncIo<turnloop::backend::Platform>>>);
+impl AsyncRead for Shared {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bytes: &mut [u8],
+    ) -> Poll<std::io::Result<usize>> {
+        Pin::new(&mut *self.0.borrow_mut()).poll_read(cx, bytes)
+    }
+}
+impl AsyncWrite for Shared {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bytes: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        Pin::new(&mut *self.0.borrow_mut()).poll_write(cx, bytes)
+    }
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut *self.0.borrow_mut()).poll_flush(cx)
+    }
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut *self.0.borrow_mut()).poll_close(cx)
+    }
+}
+#[test]
+fn server_close_lingers_until_client_eof() {
+    const LATE: &[u8] = b"bytes after the closing handshake";
+    let mut executor = LocalExecutor::<Platform>::new(Config::default()).expect("executor");
+    let h = executor.handle();
+    let h2 = h.clone();
+    let listener = Listener::bind(&h, "127.0.0.1:0".parse().expect("address")).expect("listener");
+    let address = listener.local_addr().expect("address");
+    let end = h.now() + Duration::from_secs(5);
+    let gate = std::rc::Rc::new(CloseGate::default());
+    let server_gate = gate.clone();
+    let mut server = executor
+        .spawn_local(async move {
+            let stream = Gated {
+                inner: listener.accept().await.expect("accept"),
+                gate: server_gate,
+            };
+            let (mut ws, _) = WebSocketStream::accept(stream, &[], &h, end)
+                .await
+                .expect("upgrade");
+            ws.send(Message::Text("bye".into())).await.expect("send");
+            ws.close(&h, end)
+                .await
+                .expect("close handshake and lingering close");
+            1
+        })
+        .expect("spawn");
+    let mut client = executor
+        .spawn_local(async move {
+            let io = h2
+                .connect(address, Default::default())
+                .await
+                .expect("connect");
+            let raw = Shared(std::rc::Rc::new(std::cell::RefCell::new(io)));
+            let (mut ws, _) = WebSocketStream::connect(
+                raw.clone(),
+                &address.to_string(),
+                "/",
+                [3; 16],
+                vec![],
+                &h2,
+                end,
+            )
+            .await
+            .expect("upgrade");
+            assert_eq!(
+                ws.receive().await.expect("text"),
+                Some(Message::Text("bye".into()))
+            );
+            // Receiving the server's Close also flushes our Close reply.
+            assert!(matches!(
+                ws.receive().await.expect("close"),
+                Some(Message::Close(_))
+            ));
+            let mut raw = raw;
+            peer_until(&gate, |g| g.closing.get()).await;
+            write_all(&mut raw, LATE).await.expect("late bytes");
+            gate.late_sent.set(true);
+            wake(&gate.server);
+            peer_until(&gate, |g| g.shut.get()).await;
+            let mut bytes = [0; 64];
+            let n = read(&mut raw, &mut bytes)
+                .await
+                .expect("clean EOF after the server's close, not a reset");
+            assert_eq!(n, 0);
+            close(&mut raw).await.expect("client close");
+            gate
+        })
+        .expect("spawn");
+    while !server.is_finished() || !client.is_finished() {
+        assert!(executor.driver().now() < end, "close hung");
+        executor.turn(Timeout::Until(end)).expect("turn");
+    }
+    assert_eq!(finish(&mut server), 1);
+    let gate = finish(&mut client);
+    assert_eq!(
+        gate.discarded.get(),
+        LATE.len(),
+        "server drained late bytes"
+    );
+}