From dca915986f33186109088e2a4036f56452167ce3 Mon Sep 17 00:00:00 2001
From: black-binary <blackbinary1024@gmail.com>
Date: Thu, 3 Sep 2026 19:11:22 +0800
Subject: [PATCH 1/2] fix: harden mux lifecycle and backpressure

Add regression coverage for shutdown, flushing, queue bounds, keepalive, frame validation, and stream lifecycle edge cases. Preserve accepted writes, enforce real flush/FIN ordering, bound receive work, and make concurrent close cancellation-safe.
---
 .github/workflows/ci.yml |   6 +-
 README.md                |   5 +-
 src/builder.rs           |  13 +-
 src/config.rs            |   4 +-
 src/error.rs             |   2 +
 src/frame.rs             |  27 +-
 src/lib.rs               | 846 +++++++++++++++++++++++++++++++++++----
 src/mux.rs               | 444 ++++++++++++++------
 8 files changed, 1137 insertions(+), 210 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 84869d9..06acd1c 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -17,7 +17,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable
         with:
-          components: clippy
+          components: clippy,rustfmt
       - uses: Swatinem/rust-cache@v2
       - name: Build
         run: cargo build --lib --tests --examples
@@ -25,5 +25,7 @@ jobs:
         run: cargo test --lib
       - name: Test (release)
         run: cargo test --lib --release
+      - name: Format
+        run: cargo fmt --all -- --check
       - name: Clippy
-        run: cargo clippy --lib --tests
+        run: cargo clippy --lib --tests --examples --all-features -- -D warnings
diff --git a/README.md b/README.md
index c7a1c4a..da8bb6d 100644
--- a/README.md
+++ b/README.md
@@ -87,8 +87,9 @@ let (connector, acceptor, worker) = MuxBuilder::server()
     // Per-stream idle timeout (seconds): close streams with no
     // recent traffic.
     .with_idle_timeout(NonZeroU64::new(60).unwrap())
-    // Backpressure thresholds: cap how many frames may sit in the
-    // tx/rx queues before poll_write / poll_read park.
+    // Backpressure thresholds: cap queued tx frames and the combined
+    // inbound-frame/unaccepted-stream backlog. Keep-alive expiry pauses
+    // while the RX budget deliberately prevents carrier reads.
     .with_max_tx_queue(NonZeroUsize::new(1024).unwrap())
     .with_max_rx_queue(NonZeroUsize::new(1024).unwrap())
     .with_connection(connection)
diff --git a/src/builder.rs b/src/builder.rs
index 07c1e0a..f00ea1c 100644
--- a/src/builder.rs
+++ b/src/builder.rs
@@ -83,23 +83,24 @@ impl MuxBuilder<WithConfig> {
     }
 
     /// Per-stream idle timeout: if a stream sees no traffic for this
-    /// many seconds, it is closed and its handle is reaped.
+    /// many seconds, it is closed.
     pub fn with_idle_timeout(&mut self, timeout_secs: NonZeroU64) -> &mut Self {
         self.state.config.idle_timeout = Some(timeout_secs);
         self
     }
 
-    /// Backpressure threshold for outbound frames. `poll_write` parks
-    /// once a stream's pending tx queue exceeds this value. Defaults
+    /// Backpressure threshold for outbound frames per stream. `poll_write` parks
+    /// once a stream's pending tx queue reaches this value. Defaults
     /// to 1024.
     pub fn with_max_tx_queue(&mut self, size: NonZeroUsize) -> &mut Self {
         self.state.config.max_tx_queue = size;
         self
     }
 
-    /// Backpressure threshold for inbound frames. The dispatcher parks
-    /// once total pending rx exceeds this value, propagating
-    /// backpressure to the peer's tx side. Defaults to 1024.
+    /// Backpressure threshold for inbound frames and streams waiting to be
+    /// accepted. The dispatcher parks once the total reaches this value,
+    /// propagating backpressure to the peer's tx side. Keep-alive expiry is
+    /// suspended while reads are deliberately parked. Defaults to 1024.
     pub fn with_max_rx_queue(&mut self, size: NonZeroUsize) -> &mut Self {
         self.state.config.max_rx_queue = size;
         self
diff --git a/src/config.rs b/src/config.rs
index 39ecff4..3218cc2 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -30,6 +30,8 @@ pub struct MuxConfig {
     pub idle_timeout: Option<NonZeroU64>,
     /// Backpressure threshold for outbound frames per stream.
     pub max_tx_queue: NonZeroUsize,
-    /// Backpressure threshold for inbound frames across the session.
+    /// Backpressure threshold for inbound frames and unaccepted streams
+    /// across the session. Dead-peer detection is suspended while this
+    /// budget is exhausted because the carrier is intentionally not polled.
     pub max_rx_queue: NonZeroUsize,
 }
diff --git a/src/error.rs b/src/error.rs
index 228b4ee..4c39c5c 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -14,6 +14,8 @@ pub enum MuxError {
     InvalidControlFramePayload(u16),
     #[error("Reserved stream id {0}")]
     ReservedStreamId(u32),
+    #[error("NOP frame must use stream id 0, got {0}")]
+    InvalidNopStreamId(u32),
     #[error("Duplicated stream id {0}")]
     DuplicatedStreamId(u32),
 
diff --git a/src/frame.rs b/src/frame.rs
index 457590b..83cce34 100644
--- a/src/frame.rs
+++ b/src/frame.rs
@@ -95,25 +95,36 @@ impl Decoder for MuxCodec {
     type Error = MuxError;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        src.reserve(HEADER_SIZE + MAX_PAYLOAD_SIZE + HEADER_SIZE);
-
         if src.len() < HEADER_SIZE {
+            src.reserve(HEADER_SIZE - src.len());
             return Ok(None);
         }
         let header = MuxFrameHeader::decode(src)?;
-        let len = header.length as usize;
-        if src.len() < HEADER_SIZE + len {
-            return Ok(None);
-        }
         // Per smux v1, only PSH carries payload; SYN/FIN/NOP must be empty.
-        // Reject non-empty control frames so a peer cannot use them for
-        // bandwidth amplification or covert framing.
+        // These checks depend only on the header, so reject immediately rather
+        // than waiting for an invalid peer to deliver its claimed payload.
         match header.command {
             MuxCommand::Sync | MuxCommand::Finish | MuxCommand::Nop if header.length != 0 => {
                 return Err(MuxError::InvalidControlFramePayload(header.length));
             }
             _ => {}
         }
+        match header.command {
+            MuxCommand::Nop if header.stream_id != 0 => {
+                return Err(MuxError::InvalidNopStreamId(header.stream_id));
+            }
+            MuxCommand::Sync | MuxCommand::Finish | MuxCommand::Push if header.stream_id == 0 => {
+                return Err(MuxError::ReservedStreamId(0));
+            }
+            _ => {}
+        }
+
+        let len = header.length as usize;
+        let frame_len = HEADER_SIZE + len;
+        if src.len() < frame_len {
+            src.reserve(frame_len - src.len());
+            return Ok(None);
+        }
         src.advance(HEADER_SIZE);
         let payload = src.split_to(len).freeze();
 
diff --git a/src/lib.rs b/src/lib.rs
index 6d80e77..b265121 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -66,16 +66,20 @@ pub use mux::{mux_connection, MuxAcceptor, MuxConnector, MuxStream};
 #[cfg(test)]
 mod tests {
     use std::{
-        future::poll_fn,
+        future::{poll_fn, Future},
         num::{NonZeroU64, NonZeroUsize},
         pin::Pin,
-        task::Poll,
+        sync::{
+            atomic::{AtomicBool, AtomicUsize, Ordering},
+            Arc, Mutex as StdMutex,
+        },
+        task::{Context, Poll, Waker},
         time::Duration,
     };
 
     use rand::Rng;
     use tokio::{
-        io::{AsyncRead, AsyncReadExt, AsyncWriteExt, ReadBuf},
+        io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf},
         net::{TcpListener, TcpStream},
     };
 
@@ -94,6 +98,107 @@ mod tests {
         (a, b)
     }
 
+    async fn get_duplex_mux_pair() -> (
+        MuxStream<tokio::io::DuplexStream>,
+        MuxStream<tokio::io::DuplexStream>,
+    ) {
+        let (a, b) = tokio::io::duplex(4096);
+        let (connector_a, _acceptor_a, worker_a) = MuxBuilder::client().with_connection(a).build();
+        let (_connector_b, mut acceptor_b, worker_b) =
+            MuxBuilder::server().with_connection(b).build();
+        tokio::spawn(worker_a);
+        tokio::spawn(worker_b);
+
+        let stream_a = connector_a.connect().unwrap();
+        let stream_b = acceptor_b.accept().await.unwrap();
+        (stream_a, stream_b)
+    }
+
+    fn raw_frame(command: u8, stream_id: u32, payload: &[u8]) -> Vec<u8> {
+        let mut frame = Vec::with_capacity(8 + payload.len());
+        frame.push(1); // smux v1
+        frame.push(command);
+        frame.extend_from_slice(&(payload.len() as u16).to_le_bytes());
+        frame.extend_from_slice(&stream_id.to_le_bytes());
+        frame.extend_from_slice(payload);
+        frame
+    }
+
+    /// A transport whose read and shutdown paths deliberately share one waker
+    /// slot. This is legal for `AsyncRead + AsyncWrite` implementations and
+    /// lets the close tests catch actors that poll the same carrier with
+    /// different task wakers during shutdown.
+    #[derive(Default)]
+    struct ShutdownGateState {
+        shutdown_started: AtomicBool,
+        allow_shutdown: AtomicBool,
+        write_after_shutdown: AtomicBool,
+        read_polls: AtomicUsize,
+        io_waker: StdMutex<Option<Waker>>,
+    }
+
+    impl ShutdownGateState {
+        fn register(&self, waker: &Waker) {
+            *self.io_waker.lock().unwrap() = Some(waker.clone());
+        }
+
+        fn allow_shutdown(&self) {
+            self.allow_shutdown.store(true, Ordering::SeqCst);
+            if let Some(waker) = self.io_waker.lock().unwrap().take() {
+                waker.wake();
+            }
+        }
+    }
+
+    struct ShutdownGate {
+        state: Arc<ShutdownGateState>,
+    }
+
+    impl AsyncRead for ShutdownGate {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            _buf: &mut ReadBuf<'_>,
+        ) -> Poll<std::io::Result<()>> {
+            self.state.read_polls.fetch_add(1, Ordering::SeqCst);
+            self.state.register(cx.waker());
+            Poll::Pending
+        }
+    }
+
+    impl AsyncWrite for ShutdownGate {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            buf: &[u8],
+        ) -> Poll<std::io::Result<usize>> {
+            if self.state.shutdown_started.load(Ordering::SeqCst) {
+                self.state
+                    .write_after_shutdown
+                    .store(true, Ordering::SeqCst);
+                return Poll::Ready(Err(std::io::Error::new(
+                    std::io::ErrorKind::BrokenPipe,
+                    "write after shutdown started",
+                )));
+            }
+            Poll::Ready(Ok(buf.len()))
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            self.state.shutdown_started.store(true, Ordering::SeqCst);
+            if self.state.allow_shutdown.load(Ordering::SeqCst) {
+                Poll::Ready(Ok(()))
+            } else {
+                self.state.register(cx.waker());
+                Poll::Pending
+            }
+        }
+    }
+
     async fn test_stream<T: TokioConn>(mut a: MuxStream<T>, mut b: MuxStream<T>) {
         const LEN: usize = MAX_PAYLOAD_SIZE + 0x200;
         let mut data1 = vec![0; LEN];
@@ -262,8 +367,8 @@ mod tests {
             worker_b.await.unwrap();
         });
 
-        let mut stream1 = connector_a.connect().unwrap();
-        let mut stream2 = acceptor_b.accept().await.unwrap();
+        let stream1 = connector_a.connect().unwrap();
+        let stream2 = acceptor_b.accept().await.unwrap();
         tokio::time::sleep(Duration::from_secs(1)).await;
         assert!(!stream1.is_closed());
         assert!(!stream2.is_closed());
@@ -276,18 +381,14 @@ mod tests {
 
     #[tokio::test]
     async fn test_recv_block() {
-        let (a, b) = get_tcp_pair().await;
+        let (a, b) = tokio::io::duplex(4096);
         let (connector_a, _, worker_a) = MuxBuilder::client().with_connection(a).build();
         let (_, mut acceptor_b, worker_b) = MuxBuilder::server()
-            .with_max_rx_queue(12.try_into().unwrap())
+            .with_max_rx_queue(NonZeroUsize::new(1).unwrap())
             .with_connection(b)
             .build();
-        tokio::spawn(async move {
-            worker_a.await.unwrap();
-        });
-        tokio::spawn(async move {
-            worker_b.await.unwrap();
-        });
+        tokio::spawn(worker_a);
+        tokio::spawn(worker_b);
 
         let mut stream_x1 = connector_a.connect().unwrap();
         let mut stream_x2 = acceptor_b.accept().await.unwrap();
@@ -296,37 +397,31 @@ mod tests {
         let mut stream_y2 = acceptor_b.accept().await.unwrap();
 
         let data = &[1, 2, 3, 4];
-        for _ in 0..3 {
-            stream_x1.write_all(data).await.unwrap();
-        }
-        // stream_x is full now
+        stream_x1.write_all(data).await.unwrap();
+        stream_x1.flush().await.unwrap();
+        // The single x frame fills the receive budget. The y frame reaches
+        // the carrier but cannot be dispatched until x is consumed.
         stream_y1.write_all(data).await.unwrap();
+        stream_y1.flush().await.unwrap();
 
-        // stream_y should be blocked unless x incoming bytes is handled
-        poll_fn(|cx| {
+        let y_is_pending = poll_fn(|cx| {
             let mut buf = [0; 128];
             let mut buf = ReadBuf::new(&mut buf);
             let res = Pin::new(&mut stream_y2).poll_read(cx, &mut buf);
-            assert!(res.is_pending());
-            Poll::Ready(())
+            Poll::Ready(res.is_pending())
         })
         .await;
+        assert!(y_is_pending, "y was dispatched despite a full RX budget");
 
         let mut buf = [0; 4];
-        for _ in 0..3 {
-            stream_x2.read_exact(&mut buf).await.unwrap();
-            assert_eq!(&buf, data);
-        }
+        stream_x2.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf, data);
 
-        // stream_y is avaliable now
-        poll_fn(|cx| {
-            let mut buf_arr = [0; 128];
-            let mut buf = ReadBuf::new(&mut buf_arr);
-            let res = Pin::new(&mut stream_y2).poll_read(cx, &mut buf);
-            assert!(res.is_ready());
-            Poll::Ready(())
-        })
-        .await;
+        tokio::time::timeout(Duration::from_secs(1), stream_y2.read_exact(&mut buf))
+            .await
+            .expect("dispatcher did not resume after RX consumption")
+            .unwrap();
+        assert_eq!(&buf, data);
     }
 
     #[tokio::test]
@@ -435,7 +530,10 @@ mod tests {
             .await
             .unwrap()
             .unwrap();
-        assert!(res.is_err());
+        assert!(
+            res.is_ok(),
+            "dropping the last public handle is an orderly local shutdown"
+        );
     }
 
     // BUG: writing then dropping the stream (without explicit flush/shutdown)
@@ -499,7 +597,10 @@ mod tests {
 
         // EOF must remain EOF.
         let n = stream1.read(&mut buf).await.unwrap();
-        assert_eq!(n, 0, "subsequent read must remain EOF, not return late data");
+        assert_eq!(
+            n, 0,
+            "subsequent read must remain EOF, not return late data"
+        );
     }
 
     // BUG: poll_flush returns Err the moment the stream is locally closed,
@@ -539,7 +640,10 @@ mod tests {
         use crate::error::MuxError;
         let e = MuxError::InvalidPeerStreamIdType(7, StreamIdType::Even);
         let msg = format!("{}", e);
-        assert!(msg.contains("Even"), "Display should mention type, got: {msg}");
+        assert!(
+            msg.contains("Even"),
+            "Display should mention type, got: {msg}"
+        );
     }
 
     // BUG: MuxConnector::close used to call state.inner.poll_close_unpin
@@ -623,8 +727,7 @@ mod tests {
         // Use an in-memory duplex channel as the underlying transport, then
         // forget the b half: bytes go nowhere and nothing comes back, but
         // the socket is not closed - simulating a black-holed peer.
-        let (a, b) = tokio::io::duplex(1024);
-        std::mem::forget(b);
+        let (a, _peer_held_open) = tokio::io::duplex(1024);
 
         let (connector_a, _acceptor_a, worker_a) = MuxBuilder::client()
             .with_keep_alive_interval(NonZeroU64::new(1).unwrap())
@@ -689,41 +792,54 @@ mod tests {
     // interleave on the wire instead of being grouped per stream.
     #[tokio::test(flavor = "multi_thread")]
     async fn test_flush_frames_round_robin_across_streams() {
-        let (a, b) = get_tcp_pair().await;
-        let (connector_a, _, worker_a) = MuxBuilder::client().with_connection(a).build();
-        let (_, mut acceptor_b, worker_b) = MuxBuilder::server().with_connection(b).build();
-        tokio::spawn(worker_a);
-        tokio::spawn(worker_b);
+        let (a, mut peer) = tokio::io::duplex(4096);
+        let (connector, _acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+
+        // Queue all work before the worker starts so both streams are
+        // continuously eligible during the sender's scheduling pass.
+        let mut s1 = connector.connect().unwrap();
+        let mut s2 = connector.connect().unwrap();
+        let s1_id = s1.get_stream_id();
+        let s2_id = s2.get_stream_id();
+
+        const FRAMES_PER_STREAM: usize = 8;
+        for _ in 0..FRAMES_PER_STREAM {
+            s1.write_all(&[1]).await.unwrap();
+            s2.write_all(&[2]).await.unwrap();
+        }
+        tokio::spawn(worker);
 
-        // Open two streams and immediately enqueue many small writes on each
-        // before yielding to the worker. Use poll_fn so the writes happen
-        // back-to-back in a single tick of the runtime.
-        let mut s1_tx = connector_a.connect().unwrap();
-        let mut s2_tx = connector_a.connect().unwrap();
-        let mut s1_rx = acceptor_b.accept().await.unwrap();
-        let mut s2_rx = acceptor_b.accept().await.unwrap();
-
-        // Tag bytes so we know which stream they came from on a per-frame basis.
-        const FRAMES: usize = 16;
-        for i in 0..FRAMES {
-            s1_tx.write_all(&[1u8; 8]).await.unwrap();
-            s2_tx.write_all(&[2u8; 8]).await.unwrap();
-            // Periodically flush one of them to fight runtime scheduling
-            // luck without sequencing them strictly.
-            if i % 4 == 3 {
-                s1_tx.flush().await.unwrap();
-                s2_tx.flush().await.unwrap();
+        // Two SYNs followed by 16 one-byte PSHs.
+        let expected_len = 2 * 8 + 2 * FRAMES_PER_STREAM * 9;
+        let mut wire = vec![0u8; expected_len];
+        tokio::time::timeout(Duration::from_secs(1), peer.read_exact(&mut wire))
+            .await
+            .expect("sender did not drain queued streams")
+            .unwrap();
+
+        let mut push_ids = Vec::new();
+        let mut offset = 0;
+        while offset < wire.len() {
+            let command = wire[offset + 1];
+            let payload_len = u16::from_le_bytes([wire[offset + 2], wire[offset + 3]]) as usize;
+            let stream_id = u32::from_le_bytes([
+                wire[offset + 4],
+                wire[offset + 5],
+                wire[offset + 6],
+                wire[offset + 7],
+            ]);
+            if command == 2 {
+                push_ids.push(stream_id);
             }
+            offset += 8 + payload_len;
         }
-        s1_tx.flush().await.unwrap();
-        s2_tx.flush().await.unwrap();
 
-        let mut buf1 = vec![0u8; FRAMES * 8];
-        let mut buf2 = vec![0u8; FRAMES * 8];
-        s1_rx.read_exact(&mut buf1).await.unwrap();
-        s2_rx.read_exact(&mut buf2).await.unwrap();
-        assert!(buf1.iter().all(|b| *b == 1));
-        assert!(buf2.iter().all(|b| *b == 2));
+        assert_eq!(push_ids.len(), 2 * FRAMES_PER_STREAM);
+        assert!(push_ids.iter().all(|id| *id == s1_id || *id == s2_id));
+        assert!(
+            push_ids.windows(2).all(|pair| pair[0] != pair[1]),
+            "sender did not alternate ready streams: {push_ids:?}"
+        );
     }
 
     // BUG: control frames (SYN/FIN/NOP) are spec'd as length=0, but the
@@ -743,6 +859,56 @@ mod tests {
         assert!(res.is_err(), "NOP with non-zero length must be rejected");
     }
 
+    #[test]
+    fn test_decode_rejects_invalid_reserved_stream_ids() {
+        use crate::frame::MuxCodec;
+        use bytes::BytesMut;
+        use tokio_util::codec::Decoder;
+
+        let mut codec = MuxCodec {};
+
+        let mut nop_with_stream = BytesMut::from(&raw_frame(3, 2, &[])[..]);
+        assert!(
+            codec.decode(&mut nop_with_stream).is_err(),
+            "NOP must use reserved stream id 0"
+        );
+
+        let mut push_on_zero = BytesMut::from(&raw_frame(2, 0, b"x")[..]);
+        assert!(
+            codec.decode(&mut push_on_zero).is_err(),
+            "PSH must not use reserved stream id 0"
+        );
+    }
+
+    #[test]
+    fn test_invalid_control_header_is_rejected_without_waiting_for_payload() {
+        use crate::frame::MuxCodec;
+        use bytes::BytesMut;
+        use tokio_util::codec::Decoder;
+
+        // A control frame can never carry a payload, so this header is
+        // already invalid even though its claimed 64 KiB body has not arrived.
+        let mut wire = BytesMut::from(&[1, 0, 0xff, 0xff, 2, 0, 0, 0][..]);
+        let mut codec = MuxCodec {};
+        assert!(codec.decode(&mut wire).is_err());
+    }
+
+    #[test]
+    fn test_decoder_does_not_preallocate_max_frame_for_empty_input() {
+        use crate::frame::{MuxCodec, HEADER_SIZE};
+        use bytes::BytesMut;
+        use tokio_util::codec::Decoder;
+
+        let mut wire = BytesMut::new();
+        let mut codec = MuxCodec {};
+        assert!(codec.decode(&mut wire).unwrap().is_none());
+        assert!(
+            wire.capacity() <= HEADER_SIZE * 2,
+            "empty decode preallocated {} bytes",
+            wire.capacity()
+        );
+    }
+
     // BUG: close()'s `nothing_pending` shortcut to hard_close bypasses
     // Framed's BytesMut flush. With a slow underlying transport, the sender
     // may have moved bytes into Framed (start_send) but Framed::poll_flush
@@ -836,8 +1002,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread")]
     async fn test_close_flushes_in_flight_data_without_worker() {
         let (a, b) = get_tcp_pair().await;
-        let (mut connector_a, _acc, _worker_a) =
-            MuxBuilder::client().with_connection(a).build();
+        let (mut connector_a, _acc, _worker_a) = MuxBuilder::client().with_connection(a).build();
         // Worker NOT spawned. close() must drive its own shutdown so the
         // bytes the user already accepted via write_all reach the wire.
 
@@ -889,8 +1054,7 @@ mod tests {
         // Hold the peer half so the duplex never drains. close() will park
         // in poll_close_unpin and never finish under the timeout.
         let (a, _b_held) = tokio::io::duplex(32);
-        let (mut connector_a, _acc, _worker_a) =
-            MuxBuilder::client().with_connection(a).build();
+        let (mut connector_a, _acc, _worker_a) = MuxBuilder::client().with_connection(a).build();
 
         let mut s = connector_a.connect().unwrap();
         s.write_all(&vec![0u8; 1024]).await.unwrap();
@@ -907,4 +1071,538 @@ mod tests {
             "closing_inline must be reset after close() future is cancelled"
         );
     }
+
+    #[tokio::test]
+    async fn test_stream_drop_during_connection_close_does_not_write_after_shutdown() {
+        let gate_state = Arc::new(ShutdownGateState::default());
+        let transport = ShutdownGate {
+            state: gate_state.clone(),
+        };
+        let (mut connector, _acceptor, _worker) =
+            MuxBuilder::client().with_connection(transport).build();
+        let stream = connector.connect().unwrap();
+
+        let mut close = Box::pin(connector.close());
+        let first_poll = poll_fn(|cx| Poll::Ready(close.as_mut().poll(cx))).await;
+        assert!(first_poll.is_pending());
+        assert!(gate_state.shutdown_started.load(Ordering::SeqCst));
+
+        // Once the carrier's shutdown has started, a concurrently dropped
+        // stream must not append a FIN that forces a subsequent write.
+        drop(stream);
+        let second_poll = poll_fn(|cx| Poll::Ready(close.as_mut().poll(cx))).await;
+        assert!(
+            second_poll.is_pending(),
+            "dropping a stream queued a frame after carrier shutdown: {second_poll:?}"
+        );
+        assert!(
+            !gate_state.write_after_shutdown.load(Ordering::SeqCst),
+            "connection close attempted a write after poll_shutdown"
+        );
+
+        gate_state.allow_shutdown();
+        close.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_dispatcher_does_not_overwrite_connection_close_waker() {
+        let gate_state = Arc::new(ShutdownGateState::default());
+        let transport = ShutdownGate {
+            state: gate_state.clone(),
+        };
+        let (mut connector, _acceptor, worker) =
+            MuxBuilder::client().with_connection(transport).build();
+        let mut worker = Box::pin(worker);
+
+        let initial_worker_poll = poll_fn(|cx| Poll::Ready(worker.as_mut().poll(cx))).await;
+        assert!(initial_worker_poll.is_pending());
+        assert_eq!(gate_state.read_polls.load(Ordering::SeqCst), 1);
+
+        // close() installs its shutdown waker in the transport's shared slot.
+        let mut close = Box::pin(connector.close());
+        let close_poll = poll_fn(|cx| Poll::Ready(close.as_mut().poll(cx))).await;
+        assert!(close_poll.is_pending());
+        assert!(gate_state.shutdown_started.load(Ordering::SeqCst));
+
+        // A worker poll during shutdown must not reach AsyncRead. Otherwise a
+        // transport with one read/write waker slot can strand the close task.
+        let closing_worker_poll = poll_fn(|cx| Poll::Ready(worker.as_mut().poll(cx))).await;
+        assert!(closing_worker_poll.is_pending());
+        assert_eq!(
+            gate_state.read_polls.load(Ordering::SeqCst),
+            1,
+            "dispatcher polled the carrier after connection close began"
+        );
+
+        gate_state.allow_shutdown();
+        close.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_empty_write_does_not_create_false_eof() {
+        let (mut tx, mut rx) = get_duplex_mux_pair().await;
+
+        assert_eq!(tx.write(&[]).await.unwrap(), 0);
+        tx.write_all(b"x").await.unwrap();
+        tx.flush().await.unwrap();
+
+        let mut byte = [0u8; 1];
+        let n = tokio::time::timeout(Duration::from_secs(1), rx.read(&mut byte))
+            .await
+            .expect("read timed out")
+            .unwrap();
+        assert_eq!(n, 1, "an empty write must not surface as EOF");
+        assert_eq!(byte, *b"x");
+    }
+
+    #[tokio::test]
+    async fn test_peer_zero_length_push_does_not_create_false_eof() {
+        let (a, mut peer) = tokio::io::duplex(4096);
+        let (_connector, mut acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+        tokio::spawn(worker);
+
+        let stream_id = 2; // peer ids are even when the local side is a client
+        peer.write_all(&raw_frame(0, stream_id, &[])).await.unwrap();
+        peer.write_all(&raw_frame(2, stream_id, &[])).await.unwrap();
+        peer.write_all(&raw_frame(2, stream_id, b"x"))
+            .await
+            .unwrap();
+
+        let mut stream = acceptor.accept().await.unwrap();
+        let mut byte = [0u8; 1];
+        let n = tokio::time::timeout(Duration::from_secs(1), stream.read(&mut byte))
+            .await
+            .expect("read timed out")
+            .unwrap();
+        assert_eq!(n, 1, "a zero-length PSH must not surface as EOF");
+        assert_eq!(byte, *b"x");
+    }
+
+    #[tokio::test]
+    async fn test_zero_capacity_read_returns_immediately() {
+        let (_tx, mut rx) = get_duplex_mux_pair().await;
+        let mut empty = [];
+
+        let n = tokio::time::timeout(Duration::from_millis(100), rx.read(&mut empty))
+            .await
+            .expect("a zero-capacity read must not wait for network data")
+            .unwrap();
+        assert_eq!(n, 0);
+    }
+
+    #[tokio::test]
+    async fn test_close_without_pending_data_closes_carrier() {
+        let (a, mut peer) = tokio::io::duplex(1024);
+        let (mut connector, acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+
+        connector.close().await.unwrap();
+
+        let mut byte = [0u8; 1];
+        let n = tokio::time::timeout(Duration::from_millis(100), peer.read(&mut byte))
+            .await
+            .expect("peer did not observe carrier shutdown")
+            .unwrap();
+        assert_eq!(n, 0);
+
+        // Keep these alive through the assertion: close(), rather than Arc
+        // teardown, must be what closes the carrier.
+        drop((acceptor, worker));
+    }
+
+    #[tokio::test]
+    async fn test_num_streams_excludes_closed_handles() {
+        let (a, _peer) = tokio::io::duplex(1024);
+        let (mut connector, _acceptor, _worker) = MuxBuilder::client().with_connection(a).build();
+        let _stream = connector.connect().unwrap();
+        assert_eq!(connector.get_num_streams(), 1);
+
+        connector.close().await.unwrap();
+
+        assert_eq!(
+            connector.get_num_streams(),
+            0,
+            "closed handles must not be reported as open streams"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_stream_flush_waits_for_carrier_flush() {
+        let (a, peer_not_reading) = tokio::io::duplex(32);
+        let (connector, _acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+        tokio::spawn(worker);
+
+        let mut stream = connector.connect().unwrap();
+        stream.write_all(&vec![1u8; 1024]).await.unwrap();
+
+        let result = tokio::time::timeout(Duration::from_millis(100), stream.flush()).await;
+        assert!(
+            result.is_err(),
+            "flush completed while the carrier was full and the peer was not reading"
+        );
+
+        drop(peer_not_reading);
+    }
+
+    #[tokio::test]
+    async fn test_remote_fin_then_drop_preserves_accepted_writes() {
+        let (a, mut peer) = tokio::io::duplex(32);
+        let (connector, acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+        let mut stream = connector.connect().unwrap();
+        let stream_id = stream.get_stream_id();
+        let payload = vec![7u8; MAX_PAYLOAD_SIZE * 2 + 10];
+        stream.write_all(&payload).await.unwrap();
+
+        peer.write_all(&raw_frame(1, stream_id, &[])).await.unwrap();
+        tokio::spawn(worker);
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while !stream.is_closed() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("remote FIN was not dispatched");
+
+        drop(stream);
+        drop(connector);
+        drop(acceptor);
+
+        let mut wire = Vec::new();
+        tokio::time::timeout(Duration::from_secs(2), peer.read_to_end(&mut wire))
+            .await
+            .expect("carrier did not close")
+            .unwrap();
+
+        let payload_frames = payload.len().div_ceil(MAX_PAYLOAD_SIZE);
+        let expected = 8 + payload_frames * 8 + payload.len(); // SYN + PSHs
+        assert_eq!(
+            wire.len(),
+            expected,
+            "dropping after remote FIN lost writes previously accepted by poll_write"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_idle_timeout_sends_only_one_fin() {
+        let (a, mut peer) = tokio::io::duplex(4096);
+        let (connector, _acceptor, worker) = MuxBuilder::client()
+            .with_idle_timeout(NonZeroU64::new(1).unwrap())
+            .with_connection(a)
+            .build();
+        tokio::spawn(worker);
+        let _held_stream = connector.connect().unwrap();
+
+        let deadline = tokio::time::Instant::now() + Duration::from_millis(2200);
+        let mut wire = Vec::new();
+        let mut buf = [0u8; 128];
+        while let Some(remaining) = deadline.checked_duration_since(tokio::time::Instant::now()) {
+            match tokio::time::timeout(remaining, peer.read(&mut buf)).await {
+                Ok(Ok(n)) if n > 0 => wire.extend_from_slice(&buf[..n]),
+                _ => break,
+            }
+        }
+
+        let fin_count = (0..wire.len() / 8)
+            .filter(|frame_index| wire[frame_index * 8 + 1] == 1)
+            .count();
+        assert_eq!(fin_count, 1, "an idle stream must be finished only once");
+    }
+
+    #[tokio::test]
+    async fn test_max_tx_queue_is_a_hard_limit() {
+        let (a, _peer) = tokio::io::duplex(4096);
+        let (connector, _acceptor, _worker) = MuxBuilder::client()
+            .with_max_tx_queue(NonZeroUsize::new(1).unwrap())
+            .with_connection(a)
+            .build();
+        let mut stream = connector.connect().unwrap();
+
+        assert_eq!(stream.write(b"a").await.unwrap(), 1);
+        let second_is_pending =
+            poll_fn(|cx| Poll::Ready(Pin::new(&mut stream).poll_write(cx, b"b").is_pending()))
+                .await;
+        assert!(
+            second_is_pending,
+            "max_tx_queue=1 accepted a second queued frame"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_unaccepted_syns_are_bounded_by_rx_queue() {
+        let (a, mut peer) = tokio::io::duplex(4096);
+        let (connector, mut acceptor, worker) = MuxBuilder::client()
+            .with_max_rx_queue(NonZeroUsize::new(2).unwrap())
+            .with_connection(a)
+            .build();
+        tokio::spawn(worker);
+
+        let mut syns = Vec::new();
+        syns.extend_from_slice(&raw_frame(0, 2, &[]));
+        syns.extend_from_slice(&raw_frame(0, 4, &[]));
+        syns.extend_from_slice(&raw_frame(0, 6, &[]));
+        peer.write_all(&syns).await.unwrap();
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while connector.get_num_streams() < 2 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("first two SYNs were not dispatched");
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        assert_eq!(
+            connector.get_num_streams(),
+            2,
+            "unaccepted SYNs bypassed the configured receive bound"
+        );
+
+        drop(acceptor.accept().await.unwrap());
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while connector.get_num_streams() < 2 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("dispatcher did not resume after an accept slot was consumed");
+    }
+
+    #[tokio::test]
+    async fn test_rx_backpressure_suspends_keep_alive_timeout() {
+        let (a, mut peer) = tokio::io::duplex(1024);
+        let (connector, mut acceptor, worker) = MuxBuilder::client()
+            .with_keep_alive_interval(NonZeroU64::new(1).unwrap())
+            .with_keep_alive_timeout(NonZeroU64::new(2).unwrap())
+            .with_max_rx_queue(NonZeroUsize::new(1).unwrap())
+            .with_connection(a)
+            .build();
+        let worker_task = tokio::spawn(worker);
+
+        // One unaccepted SYN fills the configured receive budget, so the
+        // dispatcher intentionally stops polling the carrier.
+        peer.write_all(&raw_frame(0, 2, &[])).await.unwrap();
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while connector.get_num_streams() != 1 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("incoming SYN was not dispatched");
+
+        // Not observing frames while intentionally backpressured is not proof
+        // that the peer is dead. Keep the healthy carrier open beyond the
+        // configured liveness timeout and verify the session survives.
+        tokio::time::sleep(Duration::from_millis(2600)).await;
+        assert!(
+            connector.connect().is_ok(),
+            "receive backpressure caused a false keep-alive timeout"
+        );
+        assert!(!worker_task.is_finished());
+
+        drop(acceptor.accept().await.unwrap());
+        drop(peer);
+        let _ = worker_task.await;
+    }
+
+    #[tokio::test]
+    async fn test_stream_shutdown_waits_until_fin_is_driven() {
+        let (a, _peer) = tokio::io::duplex(1024);
+        let (connector, _acceptor, _worker) = MuxBuilder::client().with_connection(a).build();
+        let mut stream = connector.connect().unwrap();
+
+        let shutdown_is_pending =
+            poll_fn(|cx| Poll::Ready(Pin::new(&mut stream).poll_shutdown(cx).is_pending())).await;
+        assert!(
+            shutdown_is_pending,
+            "shutdown completed before an unpolled worker could send FIN"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_connection_close_rejects_new_streams_and_writes() {
+        let (a, peer_not_reading) = tokio::io::duplex(32);
+        let (mut closer, _acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+        let other_connector = closer.clone();
+        let mut stream = closer.connect().unwrap();
+        stream.write_all(&vec![1u8; 1024]).await.unwrap();
+
+        let mut close = Box::pin(closer.close());
+        let close_is_pending =
+            poll_fn(|cx| Poll::Ready(close.as_mut().poll(cx).is_pending())).await;
+        assert!(close_is_pending, "test carrier did not apply backpressure");
+
+        assert!(
+            other_connector.connect().is_err(),
+            "connect succeeded after connection shutdown began"
+        );
+        let write_was_accepted = poll_fn(|cx| {
+            let accepted = matches!(
+                Pin::new(&mut stream).poll_write(cx, b"late"),
+                Poll::Ready(Ok(_))
+            );
+            Poll::Ready(accepted)
+        })
+        .await;
+        assert!(
+            !write_was_accepted,
+            "stream write succeeded after connection shutdown began"
+        );
+
+        drop(close);
+        drop(peer_not_reading);
+        drop(worker);
+    }
+
+    #[tokio::test]
+    async fn test_connection_close_ends_acceptor_with_queued_streams() {
+        let (a, mut peer) = tokio::io::duplex(32);
+        let (mut connector, mut acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+        tokio::spawn(worker);
+
+        peer.write_all(&raw_frame(0, 2, &[])).await.unwrap();
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while connector.get_num_streams() != 1 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("incoming SYN was not queued");
+
+        // Fill the opposite direction so connection close remains in progress
+        // long enough to inspect the acceptor's soft-close behavior.
+        let mut outgoing = connector.connect().unwrap();
+        outgoing.write_all(&vec![1u8; 1024]).await.unwrap();
+        let mut close = Box::pin(connector.close());
+        let close_is_pending =
+            poll_fn(|cx| Poll::Ready(close.as_mut().poll(cx).is_pending())).await;
+        assert!(close_is_pending);
+
+        assert!(
+            acceptor.accept().await.is_none(),
+            "acceptor yielded queued work after connection shutdown began"
+        );
+
+        drop(close);
+        drop(peer);
+    }
+
+    #[tokio::test]
+    async fn test_idle_timeout_reaps_unaccepted_stream() {
+        let (a, mut peer) = tokio::io::duplex(1024);
+        let (connector, _acceptor, worker) = MuxBuilder::client()
+            .with_idle_timeout(NonZeroU64::new(1).unwrap())
+            .with_max_rx_queue(NonZeroUsize::new(1).unwrap())
+            .with_connection(a)
+            .build();
+        tokio::spawn(worker);
+
+        peer.write_all(&raw_frame(0, 2, &[])).await.unwrap();
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while connector.get_num_streams() != 1 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("incoming SYN was not dispatched");
+
+        tokio::time::sleep(Duration::from_millis(1600)).await;
+        assert_eq!(
+            connector.get_num_streams(),
+            0,
+            "an expired unaccepted stream kept its receive slot"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_idle_timeout_orders_fin_after_accepted_writes() {
+        let (a, mut peer) = tokio::io::duplex(32);
+        let (connector, _acceptor, worker) = MuxBuilder::client()
+            .with_idle_timeout(NonZeroU64::new(1).unwrap())
+            .with_connection(a)
+            .build();
+        tokio::spawn(worker);
+
+        let mut stream = connector.connect().unwrap();
+        let payload = vec![9u8; MAX_PAYLOAD_SIZE + 10];
+        stream.write_all(&payload).await.unwrap();
+
+        // Keep the tiny carrier full until idle timeout queues FIN while one
+        // PSH is still waiting in the stream queue.
+        tokio::time::sleep(Duration::from_millis(1600)).await;
+
+        let expected_len = 8 + 2 * 8 + payload.len() + 8; // SYN + 2 PSHs + FIN
+        let mut wire = vec![0u8; expected_len];
+        tokio::time::timeout(Duration::from_secs(2), peer.read_exact(&mut wire))
+            .await
+            .expect("timed out draining carrier")
+            .unwrap();
+
+        let mut commands = Vec::new();
+        let mut offset = 0;
+        while offset < wire.len() {
+            commands.push(wire[offset + 1]);
+            let payload_len = u16::from_le_bytes([wire[offset + 2], wire[offset + 3]]) as usize;
+            offset += 8 + payload_len;
+        }
+        assert_eq!(commands, vec![0, 2, 2, 1], "FIN overtook pending PSH");
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_concurrent_connector_close_wakes_every_waiter() {
+        use tokio::sync::oneshot;
+
+        let (a, mut peer) = tokio::io::duplex(32);
+        let (connector, acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+        let first_connector = connector.clone();
+        let second_connector = connector;
+
+        let mut stream = first_connector.connect().unwrap();
+        stream.write_all(&vec![1u8; 1024]).await.unwrap();
+        drop(stream);
+
+        let spawn_closer = |mut connector: crate::MuxConnector<tokio::io::DuplexStream>| {
+            let (started_tx, started_rx) = oneshot::channel();
+            let task = tokio::spawn(async move {
+                let mut close = Box::pin(connector.close());
+                let mut started_tx = Some(started_tx);
+                poll_fn(|cx| match close.as_mut().poll(cx) {
+                    Poll::Pending => {
+                        if let Some(tx) = started_tx.take() {
+                            let _ = tx.send(());
+                        }
+                        Poll::Ready(())
+                    }
+                    Poll::Ready(result) => {
+                        panic!("close unexpectedly completed before peer draining: {result:?}")
+                    }
+                })
+                .await;
+                close.await
+            });
+            (started_rx, task)
+        };
+
+        let (first_started, first_close) = spawn_closer(first_connector);
+        first_started.await.unwrap();
+        let (second_started, second_close) = spawn_closer(second_connector);
+        second_started.await.unwrap();
+
+        let reader = tokio::spawn(async move {
+            let mut wire = Vec::new();
+            peer.read_to_end(&mut wire).await.unwrap();
+            wire
+        });
+
+        let (first_result, second_result) = tokio::time::timeout(Duration::from_secs(2), async {
+            tokio::join!(first_close, second_close)
+        })
+        .await
+        .expect("one concurrent close waiter was never woken");
+        first_result.unwrap().unwrap();
+        second_result.unwrap().unwrap();
+        reader.await.unwrap();
+
+        // These handles deliberately remain alive until both close futures
+        // complete, so hard-close must wake all waiters itself.
+        drop((acceptor, worker));
+    }
 }
diff --git a/src/mux.rs b/src/mux.rs
index 434e6ef..3ae4b37 100644
--- a/src/mux.rs
+++ b/src/mux.rs
@@ -85,11 +85,10 @@ pub struct MuxConnector<T: TokioConn> {
     state: Arc<Mutex<MuxState<T>>>,
 }
 
-/// Resets `closing_inline` if `MuxConnector::close` is dropped before
+/// Releases one inline-close claim if `MuxConnector::close` is dropped before
 /// it finishes. On normal completion, hard_close has already set
-/// `state.closed=true` and the sender exits via `check_closed`, so the
-/// reset is a no-op. On cancellation, this hands control of Framed back
-/// to the sender so the worker can still make progress.
+/// `state.closed=true`. On cancellation, another close waiter is woken; when
+/// the last inline closer goes away, control of Framed returns to the sender.
 struct CloseGuard<'a, T: TokioConn> {
     state: &'a Arc<Mutex<MuxState<T>>>,
 }
@@ -97,9 +96,14 @@ struct CloseGuard<'a, T: TokioConn> {
 impl<T: TokioConn> Drop for CloseGuard<'_, T> {
     fn drop(&mut self) {
         let mut state = self.state.lock();
+        if state.inline_closers > 0 {
+            state.inline_closers -= 1;
+        }
         if !state.closed {
-            state.closing_inline = false;
-            state.notify_should_tx();
+            state.notify_close_waiters();
+            if state.inline_closers == 0 {
+                state.notify_should_tx();
+            }
         }
     }
 }
@@ -113,7 +117,7 @@ impl<T: TokioConn> MuxConnector<T> {
     /// available.
     pub fn connect(&self) -> MuxResult<MuxStream<T>> {
         let mut state = self.state.lock();
-        state.check_closed()?;
+        state.check_accepting_work()?;
 
         let stream_id = state.alloc_stream_id()?;
         state.process_sync(stream_id, Direction::Tx)?;
@@ -144,30 +148,24 @@ impl<T: TokioConn> MuxConnector<T> {
         {
             let mut state = self.state.lock();
             state.close();
-            // Take ownership of Framed for the rest of close(). The
-            // sender will see this flag at the top of its loop and bow
-            // out, leaving Framed's single waker slot to us. The mutex
-            // already serializes physical access; this flag avoids
-            // logical wakeup loss between the two actors.
-            state.closing_inline = true;
+            // Claim inline ownership of Framed for the rest of close(). The
+            // sender bows out while at least one closer is active, leaving
+            // Framed's sink waker slot to the close futures.
+            state.inline_closers += 1;
         }
         // RAII guard: if this future is dropped (e.g. select! with a
-        // timeout) before close() finishes, reset closing_inline so the
-        // sender takes over again. Without this the sender would stay
-        // permanently parked on closing_inline=true and the worker would
-        // hang.
-        let _guard = CloseGuard {
-            state: &self.state,
-        };
+        // timeout), release its inline claim and wake another closer or the
+        // sender so shutdown cannot become wedged.
+        let _guard = CloseGuard { state: &self.state };
         poll_fn(|cx| {
             let mut state = self.state.lock();
             // Save waker so an externally-driven hard_close (dispatcher
             // error, transport failure) can wake us if it happens to
             // race ahead of our own progress here.
-            state.close_waker = Some(cx.waker().clone());
             if state.closed {
                 return Poll::Ready(Ok(()));
             }
+            state.register_close_waker(cx);
             // Drain → flush → close, ourselves. This makes close() not
             // depend on the worker still being polled.
             match state.poll_flush_frames(cx) {
@@ -186,7 +184,11 @@ impl<T: TokioConn> MuxConnector<T> {
                 }
                 Poll::Pending => return Poll::Pending,
             }
-            match state.inner.poll_close_unpin(cx) {
+            let close_result = match state.inner.as_mut() {
+                Some(inner) => inner.poll_close_unpin(cx),
+                None => Poll::Ready(Err(MuxError::ConnectionClosed)),
+            };
+            match close_result {
                 Poll::Ready(Ok(())) => {
                     state.hard_close();
                     Poll::Ready(Ok(()))
@@ -204,12 +206,17 @@ impl<T: TokioConn> MuxConnector<T> {
     /// Number of streams currently open on this session (in either
     /// direction).
     pub fn get_num_streams(&self) -> usize {
-        self.state.lock().handles.len()
+        self.state
+            .lock()
+            .handles
+            .values()
+            .filter(|handle| !handle.closed)
+            .count()
     }
 
     #[cfg(test)]
     pub(crate) fn is_closing_inline(&self) -> bool {
-        self.state.lock().closing_inline
+        self.state.lock().inline_closers > 0
     }
 }
 
@@ -257,11 +264,12 @@ impl<T: TokioConn> Stream for MuxAcceptor<T> {
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut state = self.state.lock();
-        if state.check_closed().is_err() {
+        if state.closed || state.shutdown_requested {
             return Poll::Ready(None);
         }
 
         if let Some(stream_id) = state.accept_queue.pop_front() {
+            state.notify_rx_consumed();
             state.public_handles += 1;
             let stream = MuxStream {
                 stream_id,
@@ -313,6 +321,17 @@ impl<T: TokioConn> Future for MuxTimer<T> {
 
             let mut state = this.state.lock();
 
+            if state.closed {
+                return Poll::Ready(Err(MuxError::ConnectionClosed));
+            }
+            // An orderly close owns the carrier's write side and deliberately
+            // stops accepting new work. Do not append periodic frames while
+            // it is draining, or a timer tick could write after poll_shutdown
+            // has already started.
+            if state.shutdown_requested {
+                return Poll::Pending;
+            }
+
             // Ping send
             if is_ping_send_needs {
                 state.enqueue_frame_global(MuxFrame::new(MuxCommand::Nop, 0, Bytes::new()));
@@ -324,7 +343,13 @@ impl<T: TokioConn> Future for MuxTimer<T> {
             // the connection as dead and tear everything down so callers
             // unblock instead of hanging on a half-open socket.
             if let Some(timeout) = state.keep_alive_timeout {
-                if Instant::now().duration_since(state.last_rx) >= timeout {
+                // When the configured RX budget is exhausted, the dispatcher
+                // intentionally does not poll the carrier. Silence cannot be
+                // used as evidence of a dead peer until application reads or
+                // accepts enough work for dispatching to resume.
+                if !state.rx_backpressured
+                    && Instant::now().duration_since(state.last_rx) >= timeout
+                {
                     state.hard_close();
                     return Poll::Ready(Err(MuxError::ConnectionClosed));
                 }
@@ -337,7 +362,7 @@ impl<T: TokioConn> Future for MuxTimer<T> {
                     .handles
                     .iter()
                     .filter_map(|(id, h)| {
-                        if now.duration_since(h.last_active) >= timeout {
+                        if !h.closed && now.duration_since(h.last_active) >= timeout {
                             Some(*id)
                         } else {
                             None
@@ -346,9 +371,30 @@ impl<T: TokioConn> Future for MuxTimer<T> {
                     .collect::<Vec<_>>();
 
                 for stream_id in dead_ids {
-                    state.try_mark_finish(stream_id);
-                    state.send_finish(stream_id);
-                    state.notify_rx_consumed();
+                    if let Some(position) = state
+                        .accept_queue
+                        .iter()
+                        .position(|queued| *queued == stream_id)
+                    {
+                        // An unaccepted stream cannot have local writes, so
+                        // its FIN can go directly to the control queue before
+                        // the handle is reaped.
+                        state.try_mark_finish(stream_id);
+                        state.send_finish(stream_id);
+                        state.accept_queue.remove(position);
+                        state.remove_stream(stream_id);
+                    } else {
+                        // Keep FIN behind PSHs already accepted for an active
+                        // stream. Putting it in the global control queue would
+                        // let it overtake the stream's pending data.
+                        state.try_mark_finish(stream_id);
+                        state.enqueue_frame_stream(
+                            stream_id,
+                            MuxFrame::new(MuxCommand::Finish, stream_id, Bytes::new()),
+                        );
+                        state.notify_should_tx();
+                        state.notify_rx_consumed();
+                    }
                 }
             }
         }
@@ -373,16 +419,19 @@ impl<T: TokioConn> Future for MuxSender<T> {
             state.check_closed()?;
             // MuxConnector::close drives Framed inline; stay out of the
             // way. We'll be re-woken by hard_close via should_tx_waker.
-            if state.closing_inline {
+            if state.inline_closers > 0 {
                 return Poll::Pending;
             }
             ready!(state.poll_flush_frames(cx)).inspect_err(|_| state.hard_close())?;
             ready!(state.poll_flush_inner(cx)).inspect_err(|_| state.hard_close())?;
             // After draining, finalize an orderly shutdown if one was
-            // requested. Closing the inner sink lets the dispatcher's
-            // poll_next observe EOF and exit cleanly.
+            // requested. hard_close then wakes the parked dispatcher and all
+            // public handles.
             if state.shutdown_requested && !state.closed {
-                let res = state.inner.poll_close_unpin(cx);
+                let res = match state.inner.as_mut() {
+                    Some(inner) => inner.poll_close_unpin(cx),
+                    None => Poll::Ready(Err(MuxError::ConnectionClosed)),
+                };
                 if let Poll::Ready(r) = res {
                     state.hard_close();
                     return Poll::Ready(r);
@@ -420,8 +469,16 @@ impl<T: TokioConn> Future for MuxDispatcher<T> {
         loop {
             let mut state = self.state.lock();
             state.check_closed()?;
+            // Once orderly shutdown begins, the close driver is the only
+            // actor that may poll the carrier. Besides preventing late peer
+            // frames from generating new replies, this avoids replacing a
+            // close future's transport waker with the worker's read waker.
+            if state.shutdown_requested {
+                return Poll::Pending;
+            }
 
-            ready!(state.poll_ready_rx_consumed(cx)); // Can stuck here forever, be careful
+            // Apply session-wide backpressure before reading another frame.
+            ready!(state.poll_ready_rx_consumed(cx));
 
             let frame = ready!(state.poll_next_frame(cx)).inspect_err(|_| state.hard_close())?;
             // Refresh peer-liveness clock on every successfully-decoded
@@ -491,8 +548,8 @@ impl<T: TokioConn> Future for MuxWorker<T> {
 /// A bi-directional stream multiplexed over the session. Implements
 /// `AsyncRead + AsyncWrite + Unpin`, so it can be used anywhere a
 /// `TcpStream` would. Dropping it without `shutdown()` is fine — the
-/// stream's pending tx queue is moved to the global queue and a FIN
-/// is enqueued so the peer sees a clean close.
+/// stream's pending tx queue is moved to the global queue, and a FIN is
+/// enqueued when neither the peer nor the session has already closed it.
 pub struct MuxStream<T: TokioConn> {
     stream_id: u32,
     state: Arc<Mutex<MuxState<T>>>,
@@ -502,22 +559,32 @@ pub struct MuxStream<T: TokioConn> {
 impl<T: TokioConn> Drop for MuxStream<T> {
     fn drop(&mut self) {
         let mut state = self.state.lock();
-        if !state.is_closed(self.stream_id) {
-            // The user did not call `shutdown()`. Anything still queued in
-            // the per-stream tx_queue would otherwise be dropped together
-            // with the StreamHandle below, so move it onto the global
-            // tx_queue first, then enqueue FIN after, so the wire sees
-            // PSH... PSH FIN in order. Handle is guaranteed to exist here
-            // — is_closed just asserted it.
-            let h = state.handles.get_mut(&self.stream_id).unwrap();
-            let drained: VecDeque<MuxFrame> = std::mem::take(&mut h.tx_queue);
+        if !state.closed {
+            // Preserve every frame already accepted by poll_write even when a
+            // remote FIN raced with Drop. Whether we need to originate our own
+            // FIN is separate from whether accepted data must be drained.
+            let may_send_finish = !state.shutdown_requested;
+            let (drained, should_send_finish) = state
+                .handles
+                .get_mut(&self.stream_id)
+                .map(|h| {
+                    (
+                        std::mem::take(&mut h.tx_queue),
+                        may_send_finish && !h.closed,
+                    )
+                })
+                .unwrap_or_default();
             state.tx_queue.extend(drained);
-            state.enqueue_frame_global(MuxFrame::new(
-                MuxCommand::Finish,
-                self.stream_id,
-                Bytes::new(),
-            ));
-            state.notify_should_tx();
+            if should_send_finish {
+                state.enqueue_frame_global(MuxFrame::new(
+                    MuxCommand::Finish,
+                    self.stream_id,
+                    Bytes::new(),
+                ));
+            }
+            if should_send_finish || !state.tx_queue.is_empty() {
+                state.notify_should_tx();
+            }
         }
         state.remove_stream(self.stream_id);
         state.dec_public_handles();
@@ -530,6 +597,10 @@ impl<T: TokioConn> AsyncRead for MuxStream<T> {
         cx: &mut Context<'_>,
         buf: &mut io::ReadBuf<'_>,
     ) -> Poll<StdIo::Result<()>> {
+        if buf.remaining() == 0 {
+            return Poll::Ready(Ok(()));
+        }
+
         loop {
             if let Some(read_buffer) = &mut self.read_buffer {
                 if read_buffer.len() <= buf.remaining() {
@@ -548,7 +619,9 @@ impl<T: TokioConn> AsyncRead for MuxStream<T> {
 
             if let Some(frame) = frame {
                 debug_assert_eq!(frame.header.command, MuxCommand::Push);
-                self.read_buffer = Some(frame.payload);
+                if !frame.payload.is_empty() {
+                    self.read_buffer = Some(frame.payload);
+                }
             } else {
                 // EOF
                 return Poll::Ready(Ok(()));
@@ -559,7 +632,7 @@ impl<T: TokioConn> AsyncRead for MuxStream<T> {
 
 #[inline]
 fn mux_to_io_err(e: MuxError) -> StdIo::Error {
-    StdIo::Error::new(ErrorKind::Other, e)
+    StdIo::Error::other(e)
 }
 
 #[inline]
@@ -573,6 +646,10 @@ impl<T: TokioConn> AsyncWrite for MuxStream<T> {
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<Result<usize, StdIo::Error>> {
+        if buf.is_empty() {
+            return Poll::Ready(Ok(0));
+        }
+
         let mut state = self.state.lock();
         if state.is_closed(self.stream_id) {
             return Poll::Ready(Err(new_io_err(
@@ -629,15 +706,19 @@ impl<T: TokioConn> AsyncWrite for MuxStream<T> {
             }
 
             state.try_mark_finish(self.stream_id);
-            state.send_finish(self.stream_id);
+            state.enqueue_frame_stream(
+                self.stream_id,
+                MuxFrame::new(MuxCommand::Finish, self.stream_id, Bytes::new()),
+            );
+            state.notify_should_tx();
         }
     }
 }
 
 impl<T: TokioConn> MuxStream<T> {
     /// True once the stream has received FIN, the session has hard-
-    /// closed, or an idle timeout reaped the handle.
-    pub fn is_closed(&mut self) -> bool {
+    /// closed, or an idle timeout has closed the stream.
+    pub fn is_closed(&self) -> bool {
         self.state.lock().is_closed(self.stream_id)
     }
 
@@ -652,6 +733,13 @@ struct StreamHandle {
 
     tx_queue: VecDeque<MuxFrame>,
     tx_done_waker: Option<Waker>,
+    /// True after at least one frame for this stream has been handed to
+    /// `Framed`, but before the carrier sink has completed a flush.
+    unflushed: bool,
+    /// Set by `MuxStream::poll_flush`. Requested streams are drained and
+    /// carrier-flushed ahead of ordinary background traffic so a flush on
+    /// one stream cannot be held hostage by every other stream's backlog.
+    flush_requested: bool,
 
     rx_queue: VecDeque<MuxFrame>,
     rx_ready_waker: Option<Waker>,
@@ -663,9 +751,11 @@ impl StreamHandle {
     fn new() -> Self {
         Self {
             closed: false,
-            tx_queue: VecDeque::with_capacity(128),
+            tx_queue: VecDeque::new(),
             tx_done_waker: None,
-            rx_queue: VecDeque::with_capacity(128),
+            unflushed: false,
+            flush_requested: false,
+            rx_queue: VecDeque::new(),
             rx_ready_waker: None,
             last_active: Instant::now(),
         }
@@ -703,7 +793,9 @@ enum Direction {
 }
 
 struct MuxState<T: TokioConn> {
-    inner: Framed<T, MuxCodec>,
+    /// Kept in an Option so hard-close can release the carrier immediately,
+    /// even while public handles still retain the shared MuxState.
+    inner: Option<Framed<T, MuxCodec>>,
     handles: HashMap<u32, StreamHandle>,
 
     accept_queue: VecDeque<u32>,
@@ -712,9 +804,8 @@ struct MuxState<T: TokioConn> {
     tx_queue: VecDeque<MuxFrame>,
     should_tx_waker: Option<Waker>,
     rx_consumed_waker: Option<Waker>,
-    /// Wakes whoever is awaiting on hard_close completion (currently
-    /// MuxConnector::close).
-    close_waker: Option<Waker>,
+    /// Wakes every connector awaiting hard-close completion.
+    close_wakers: Vec<Waker>,
 
     closed: bool,
     /// Soft-close requested: drain everything that's already queued, then
@@ -722,14 +813,11 @@ struct MuxState<T: TokioConn> {
     /// teardown) so that user-initiated shutdown does not lose pending FIN
     /// / PSH frames already enqueued by Drop impls.
     shutdown_requested: bool,
-    /// Set while `MuxConnector::close` is driving Framed itself. The
-    /// sender steps out of the way on this flag so that close() can
-    /// finish even when the worker isn't being polled (e.g. the user
-    /// forgot to spawn it, or it has already exited). The mutex
-    /// serializes access to Framed; the flag prevents close()/sender
-    /// from clobbering each other's wakers on Framed's single waker
-    /// slot.
-    closing_inline: bool,
+    /// Number of `MuxConnector::close` futures driving Framed inline. The
+    /// sender steps out of the way while this is non-zero, so close() can
+    /// finish even when the worker is not being polled. Counting claims keeps
+    /// cancellation of one concurrent closer from releasing another's claim.
+    inline_closers: usize,
     accept_closed: bool,
     public_handles: usize,
 
@@ -743,6 +831,9 @@ struct MuxState<T: TokioConn> {
     keep_alive_timeout: Option<Duration>,
     /// Last time the dispatcher successfully decoded a frame from the peer.
     last_rx: Instant,
+    /// True while the dispatcher has intentionally stopped polling the
+    /// carrier because the session-wide receive budget is exhausted.
+    rx_backpressured: bool,
 
     max_tx_queue: usize,
     max_rx_queue: usize,
@@ -767,17 +858,17 @@ impl<T: TokioConn> MuxState<T> {
             Duration::from_secs(secs)
         });
         Self {
-            inner,
+            inner: Some(inner),
             handles: HashMap::new(),
             accept_queue: VecDeque::new(),
             accept_waker: None,
             tx_queue: VecDeque::with_capacity(config.max_tx_queue.get()),
             should_tx_waker: None,
             rx_consumed_waker: None,
-            close_waker: None,
+            close_wakers: Vec::new(),
             closed: false,
             shutdown_requested: false,
-            closing_inline: false,
+            inline_closers: 0,
             accept_closed: false,
             public_handles: 0,
             stream_id_hint: Wrapping(config.stream_id_type as u32),
@@ -785,6 +876,7 @@ impl<T: TokioConn> MuxState<T> {
             idle_timeout: config.idle_timeout.map(|n| Duration::from_secs(n.get())),
             keep_alive_timeout,
             last_rx: Instant::now(),
+            rx_backpressured: false,
             max_tx_queue: config.max_tx_queue.get(),
             max_rx_queue: config.max_rx_queue.get(),
         }
@@ -891,9 +983,13 @@ impl<T: TokioConn> MuxState<T> {
             if handle.closed {
                 return false;
             }
-            handle.rx_queue.push_back(frame);
-            handle.notify_rx_ready();
             handle.last_active = Instant::now();
+            // A zero-length PSH carries no data. Queuing it would make a
+            // non-empty AsyncRead return zero and falsely signal EOF.
+            if !frame.payload.is_empty() {
+                handle.rx_queue.push_back(frame);
+                handle.notify_rx_ready();
+            }
             true
         } else {
             false
@@ -902,19 +998,27 @@ impl<T: TokioConn> MuxState<T> {
 
     #[inline]
     fn get_rx_pending(&mut self) -> usize {
-        self.handles
-            .values()
-            .filter(|h| !h.closed)
-            .map(|h| h.rx_queue.len())
-            .sum()
+        self.accept_queue.len()
+            + self
+                .handles
+                .values()
+                .map(|h| h.rx_queue.len())
+                .sum::<usize>()
     }
 
     fn poll_ready_rx_consumed(&mut self, cx: &Context<'_>) -> Poll<()> {
         let pending = self.get_rx_pending();
-        if pending > self.max_rx_queue {
+        if pending >= self.max_rx_queue {
+            self.rx_backpressured = true;
             self.register_rx_consumed_waker(cx);
             Poll::Pending
         } else {
+            if std::mem::replace(&mut self.rx_backpressured, false) {
+                // No peer liveness observations were possible while reads
+                // were parked. Give the resumed dispatcher a fresh timeout
+                // window instead of immediately closing on stale last_rx.
+                self.last_rx = Instant::now();
+            }
             Poll::Ready(())
         }
     }
@@ -932,7 +1036,10 @@ impl<T: TokioConn> MuxState<T> {
     }
 
     fn poll_next_frame(&mut self, cx: &mut Context<'_>) -> Poll<MuxResult<MuxFrame>> {
-        if let Some(r) = ready!(self.inner.poll_next_unpin(cx)) {
+        let Some(inner) = self.inner.as_mut() else {
+            return Poll::Ready(Err(MuxError::ConnectionClosed));
+        };
+        if let Some(r) = ready!(inner.poll_next_unpin(cx)) {
             let frame = r?;
             Poll::Ready(Ok(frame))
         } else {
@@ -941,17 +1048,20 @@ impl<T: TokioConn> MuxState<T> {
     }
 
     #[inline]
-    fn pin_inner(&mut self) -> Pin<&mut Framed<T, MuxCodec>> {
-        Pin::new(&mut self.inner)
+    fn pin_inner(&mut self) -> MuxResult<Pin<&mut Framed<T, MuxCodec>>> {
+        self.inner
+            .as_mut()
+            .map(Pin::new)
+            .ok_or(MuxError::ConnectionClosed)
     }
 
     fn poll_write_ready(&mut self, cx: &mut Context<'_>) -> Poll<MuxResult<()>> {
-        ready!(self.pin_inner().poll_ready(cx))?;
+        ready!(self.pin_inner()?.poll_ready(cx))?;
         Poll::Ready(Ok(()))
     }
 
     fn write_frame(&mut self, frame: MuxFrame) -> MuxResult<()> {
-        self.pin_inner().start_send(frame)?;
+        self.pin_inner()?.start_send(frame)?;
         Ok(())
     }
 
@@ -984,7 +1094,7 @@ impl<T: TokioConn> MuxState<T> {
     }
 
     fn poll_stream_write_ready(&mut self, cx: &Context<'_>, stream_id: u32) -> Poll<MuxResult<()>> {
-        self.check_closed()?;
+        self.check_accepting_work()?;
         debug_assert!(
             self.handles.contains_key(&stream_id),
             "poll_stream_write_ready called with unknown stream id {stream_id}"
@@ -992,7 +1102,7 @@ impl<T: TokioConn> MuxState<T> {
         let Some(handle) = self.handles.get_mut(&stream_id) else {
             return Poll::Ready(Err(MuxError::StreamClosed(stream_id)));
         };
-        if handle.tx_queue.len() > self.max_tx_queue {
+        if handle.tx_queue.len() >= self.max_tx_queue {
             // A stream's tx queue is full
             handle.register_tx_done_waker(cx);
             // Notify the worker to transfer data now
@@ -1040,6 +1150,18 @@ impl<T: TokioConn> MuxState<T> {
         }
     }
 
+    fn register_close_waker(&mut self, cx: &Context<'_>) {
+        if !self.close_wakers.iter().any(|w| w.will_wake(cx.waker())) {
+            self.close_wakers.push(cx.waker().clone());
+        }
+    }
+
+    fn notify_close_waiters(&mut self) {
+        for waker in self.close_wakers.drain(..) {
+            waker.wake();
+        }
+    }
+
     fn poll_flush_stream_frames(
         &mut self,
         cx: &mut Context<'_>,
@@ -1053,9 +1175,10 @@ impl<T: TokioConn> MuxState<T> {
         let Some(handle) = self.handles.get_mut(&stream_id) else {
             return Poll::Ready(Ok(()));
         };
-        if handle.tx_queue.is_empty() {
+        if handle.tx_queue.is_empty() && !handle.unflushed {
             Poll::Ready(Ok(()))
         } else {
+            handle.flush_requested = true;
             handle.register_tx_done_waker(cx);
             self.notify_should_tx();
             Poll::Pending
@@ -1085,22 +1208,24 @@ impl<T: TokioConn> MuxState<T> {
         // requested; the dispatcher would otherwise queue late streams
         // that nobody can ever accept.
         self.accept_closed = true;
+        // No public stream exists for queued accepts. Release them now; the
+        // carrier close supersedes per-stream FIN replies.
+        while let Some(stream_id) = self.accept_queue.pop_front() {
+            self.handles.remove(&stream_id);
+        }
+        self.notify_rx_consumed();
+        // Close every exposed stream logically at the start of shutdown so
+        // reads/writes wake immediately. Their already-accepted tx queues are
+        // retained and will still be drained by poll_flush_frames.
+        for handle in self.handles.values_mut() {
+            handle.closed = true;
+            handle.notify_rx_ready();
+            handle.notify_tx_done();
+        }
         self.notify_accept_stream();
         // Tell the sender to wake up: it will drain remaining frames and
         // then call hard_close to actually tear things down.
         self.notify_should_tx();
-        // If there's truly nothing left to flush - including Framed's own
-        // BytesMut write buffer, which holds bytes that have been
-        // start_send'd but not yet written to the underlying transport -
-        // jump straight to hard close. Skipping the Framed check here is
-        // unsafe: hard_close drops Framed without flushing it, so any
-        // bytes still in the BytesMut are lost.
-        let nothing_pending = self.tx_queue.is_empty()
-            && self.handles.values().all(|h| h.tx_queue.is_empty())
-            && self.inner.write_buffer().is_empty();
-        if nothing_pending {
-            self.hard_close();
-        }
     }
 
     /// Immediate teardown: wake all wakers, mark every handle closed, and
@@ -1111,15 +1236,24 @@ impl<T: TokioConn> MuxState<T> {
             return;
         }
         self.closed = true;
+        // Releasing the carrier here is important: public handles may remain
+        // alive indefinitely after a timeout/error and must not keep the
+        // underlying socket open.
+        self.inner.take();
+        self.tx_queue = VecDeque::new();
+        // No MuxStream exists for these queued ids yet, so they can be
+        // released immediately rather than retained by a closed acceptor.
+        while let Some(stream_id) = self.accept_queue.pop_front() {
+            self.handles.remove(&stream_id);
+        }
         // Wake up everyone
         self.notify_accept_stream();
         self.notify_rx_consumed();
         self.notify_should_tx();
-        if let Some(w) = self.close_waker.take() {
-            w.wake();
-        }
-        for (_, h) in self.handles.iter_mut() {
+        self.notify_close_waiters();
+        for h in self.handles.values_mut() {
             h.closed = true;
+            h.tx_queue = VecDeque::new();
             h.notify_rx_ready();
             h.notify_tx_done();
         }
@@ -1133,6 +1267,14 @@ impl<T: TokioConn> MuxState<T> {
         }
     }
 
+    fn check_accepting_work(&self) -> MuxResult<()> {
+        if self.closed || self.shutdown_requested {
+            Err(MuxError::ConnectionClosed)
+        } else {
+            Ok(())
+        }
+    }
+
     fn poll_flush_frames(&mut self, cx: &mut Context<'_>) -> Poll<MuxResult<()>> {
         // Global queue first - control frames (SYN/FIN/NOP) take priority.
         while !self.tx_queue.is_empty() {
@@ -1141,12 +1283,44 @@ impl<T: TokioConn> MuxState<T> {
             self.write_frame(frame)?;
         }
 
+        // A stream actively waiting in poll_flush gets a real carrier flush
+        // boundary before unrelated background queues are drained. Besides
+        // matching AsyncWrite::flush semantics, this prevents an unrelated
+        // connection-wide transmit backlog from blocking this stream's flush.
+        loop {
+            let requested = self
+                .handles
+                .iter()
+                .find_map(|(id, h)| h.flush_requested.then_some(*id));
+            let Some(sid) = requested else { break };
+
+            while self
+                .handles
+                .get(&sid)
+                .is_some_and(|h| !h.tx_queue.is_empty())
+            {
+                ready!(self.pin_inner()?.poll_ready(cx))?;
+                let (frame, became_writable) = {
+                    let h = self.handles.get_mut(&sid).unwrap();
+                    let frame = h.tx_queue.pop_front().unwrap();
+                    (frame, h.tx_queue.len() < self.max_tx_queue)
+                };
+                self.pin_inner()?.start_send(frame)?;
+                if let Some(h) = self.handles.get_mut(&sid) {
+                    h.unflushed = true;
+                    if became_writable {
+                        h.notify_tx_done();
+                    }
+                }
+            }
+
+            ready!(self.poll_flush_inner(cx))?;
+        }
+
         // Round-robin across per-stream tx_queues. Each outer pass pops at
         // most one frame from each non-empty stream, so a single noisy
-        // stream can no longer starve smaller ones (head-of-line blocking
-        // inside the worker). We also defer notify_tx_done until a
-        // stream's queue actually drains, instead of waking the writer
-        // after every single frame.
+        // stream cannot monopolize the sender. A writer is woken as soon as
+        // its queue falls below the configured limit.
         loop {
             let ids: Vec<u32> = self
                 .handles
@@ -1161,18 +1335,19 @@ impl<T: TokioConn> MuxState<T> {
             for sid in ids {
                 // poll_ready must come *before* the pop; otherwise a Pending
                 // return would silently drop the frame.
-                ready!(Pin::new(&mut self.inner).poll_ready(cx))?;
-                let (frame, drained) = if let Some(h) = self.handles.get_mut(&sid) {
+                ready!(self.pin_inner()?.poll_ready(cx))?;
+                let (frame, became_writable) = if let Some(h) = self.handles.get_mut(&sid) {
                     let f = h.tx_queue.pop_front();
-                    (f, h.tx_queue.is_empty())
+                    (f, h.tx_queue.len() < self.max_tx_queue)
                 } else {
                     (None, false)
                 };
                 let Some(frame) = frame else { continue };
-                Pin::new(&mut self.inner).start_send(frame)?;
+                self.pin_inner()?.start_send(frame)?;
                 sent_any = true;
-                if drained {
-                    if let Some(h) = self.handles.get_mut(&sid) {
+                if let Some(h) = self.handles.get_mut(&sid) {
+                    h.unflushed = true;
+                    if became_writable {
                         h.notify_tx_done();
                     }
                 }
@@ -1186,7 +1361,22 @@ impl<T: TokioConn> MuxState<T> {
     }
 
     fn poll_flush_inner(&mut self, cx: &mut Context<'_>) -> Poll<MuxResult<()>> {
-        self.inner.poll_flush_unpin(cx)
+        let result = match self.inner.as_mut() {
+            Some(inner) => inner.poll_flush_unpin(cx),
+            None => Poll::Ready(Err(MuxError::ConnectionClosed)),
+        };
+        if matches!(result, Poll::Ready(Ok(()))) {
+            for handle in self.handles.values_mut() {
+                if handle.unflushed {
+                    handle.unflushed = false;
+                }
+                if handle.flush_requested && handle.tx_queue.is_empty() {
+                    handle.flush_requested = false;
+                }
+                handle.notify_tx_done();
+            }
+        }
+        result
     }
 
     fn has_pending_tx(&self) -> bool {
@@ -1235,4 +1425,24 @@ mod alloc_tests {
         let res = s.process_sync(0, Direction::Rx);
         assert!(res.is_err(), "SYN with stream id 0 must be rejected");
     }
+
+    #[test]
+    fn closed_stream_payload_still_counts_toward_rx_backpressure() {
+        let mut s = fresh_state(StreamIdType::Odd);
+        let stream_id = 2;
+        s.process_sync(stream_id, Direction::Rx).unwrap();
+        assert!(s.recv_push(MuxFrame::new(
+            MuxCommand::Push,
+            stream_id,
+            Bytes::from_static(b"queued")
+        )));
+
+        s.try_mark_finish(stream_id);
+
+        assert_eq!(
+            s.get_rx_pending(),
+            1,
+            "unread payload stopped counting as soon as FIN arrived"
+        );
+    }
 }

From 4830e4fe8fb365b913d58d44c5aed87577c4782c Mon Sep 17 00:00:00 2001
From: black-binary <blackbinary1024@gmail.com>
Date: Thu, 3 Sep 2026 20:00:26 +0800
Subject: [PATCH 2/2] test: add deterministic mux stress coverage

---
 .github/workflows/ci.yml |   2 +
 src/lib.rs               | 506 ++++++++++++++++++++++++++++++++++++++-
 src/mux.rs               |   5 +
 3 files changed, 512 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 06acd1c..cce4434 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -25,6 +25,8 @@ jobs:
         run: cargo test --lib
       - name: Test (release)
         run: cargo test --lib --release
+      - name: Stress (release)
+        run: cargo test --lib --release test_soak_randomized_stream_lifecycles -- --ignored
       - name: Format
         run: cargo fmt --all -- --check
       - name: Clippy
diff --git a/src/lib.rs b/src/lib.rs
index b265121..198d651 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -77,7 +77,7 @@ mod tests {
         time::Duration,
     };
 
-    use rand::Rng;
+    use rand::{rngs::StdRng, Rng, RngExt, SeedableRng};
     use tokio::{
         io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf},
         net::{TcpListener, TcpStream},
@@ -233,6 +233,79 @@ mod tests {
         b.shutdown().await.unwrap();
     }
 
+    fn next_stress_word(state: &mut u64) -> u64 {
+        // xorshift64*: cheap, deterministic scheduling and chunk variation.
+        // A fixed generator makes a failing stress run exactly reproducible.
+        *state ^= *state >> 12;
+        *state ^= *state << 25;
+        *state ^= *state >> 27;
+        state.wrapping_mul(0x2545_f491_4f6c_dd1d)
+    }
+
+    fn stress_payload(seed: u64, len: usize) -> Vec<u8> {
+        let mut state = seed.max(1);
+        (0..len)
+            .map(|_| next_stress_word(&mut state) as u8)
+            .collect()
+    }
+
+    async fn write_stress_chunks<W: AsyncWrite + Unpin>(writer: &mut W, data: &[u8], seed: u64) {
+        let mut state = seed.max(1);
+        let mut offset = 0;
+        while offset < data.len() {
+            let chunk_len = 1 + next_stress_word(&mut state) as usize % 4096;
+            let end = (offset + chunk_len).min(data.len());
+            writer.write_all(&data[offset..end]).await.unwrap();
+            offset = end;
+            if state & 3 == 0 {
+                tokio::task::yield_now().await;
+            }
+        }
+    }
+
+    fn stress_message(seed: u64, round: usize) -> Vec<u8> {
+        let mixed = seed
+            .wrapping_add((round as u64).wrapping_mul(0x9e37_79b9_7f4a_7c15))
+            .rotate_left((round % 63) as u32);
+        let len = 1 + mixed as usize % 2048;
+        stress_payload(mixed, len)
+    }
+
+    async fn run_full_duplex_stress(
+        stream: MuxStream<tokio::io::DuplexStream>,
+        write_seed: u64,
+        read_seed: u64,
+        rounds: usize,
+        before_shutdown: Arc<tokio::sync::Barrier>,
+    ) {
+        let (mut reader, mut writer) = tokio::io::split(stream);
+        let (write_result, ()) = tokio::join!(
+            async {
+                for round in 0..rounds {
+                    let message = stress_message(write_seed, round);
+                    write_stress_chunks(&mut writer, &message, write_seed ^ round as u64).await;
+                    if round % 7 == 0 {
+                        writer.flush().await.unwrap();
+                    }
+                }
+                writer.flush().await
+            },
+            async {
+                for round in 0..rounds {
+                    let expected = stress_message(read_seed, round);
+                    let mut actual = vec![0; expected.len()];
+                    reader.read_exact(&mut actual).await.unwrap();
+                    assert_eq!(actual, expected, "payload mismatch in round {round}");
+                }
+            }
+        );
+        write_result.unwrap();
+        before_shutdown.wait().await;
+
+        let mut stream = reader.unsplit(writer);
+        stream.shutdown().await.unwrap();
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn test_tcp() {
         let (a, b) = get_tcp_pair().await;
@@ -1605,4 +1678,435 @@ mod tests {
         // complete, so hard-close must wake all waiters itself.
         drop((acceptor, worker));
     }
+
+    async fn run_randomized_lifecycle_stress(batches: usize, streams_per_batch: usize) {
+        const SEED: u64 = 0x5eed_cafe_f00d_beef;
+
+        let (a, b) = tokio::io::duplex(128);
+        let (mut connector_a, _acceptor_a, worker_a) = MuxBuilder::client()
+            .with_max_tx_queue(NonZeroUsize::new(2).unwrap())
+            .with_max_rx_queue(NonZeroUsize::new(8).unwrap())
+            .with_connection(a)
+            .build();
+        let (connector_b, mut acceptor_b, worker_b) = MuxBuilder::server()
+            .with_max_tx_queue(NonZeroUsize::new(2).unwrap())
+            .with_max_rx_queue(NonZeroUsize::new(8).unwrap())
+            .with_connection(b)
+            .build();
+        let worker_a = tokio::spawn(worker_a);
+        let worker_b = tokio::spawn(worker_b);
+        let mut rng = StdRng::seed_from_u64(SEED);
+
+        for batch in 0..batches {
+            let mut clients = tokio::task::JoinSet::new();
+            for slot in 0..streams_per_batch {
+                let case_id = (batch * streams_per_batch + slot) as u64;
+                let payload_len = match case_id % 16 {
+                    0 => 0,
+                    1 => 1,
+                    2 => 7,
+                    3 => 8,
+                    4 => 1024,
+                    5 => 8191,
+                    6 => 8192,
+                    7 => MAX_PAYLOAD_SIZE - 1,
+                    8 => MAX_PAYLOAD_SIZE,
+                    9 => MAX_PAYLOAD_SIZE + 1,
+                    _ => rng.random_range(1..=16 * 1024),
+                };
+                let chunk_seed = rng.random::<u64>();
+                let drop_without_flush = case_id.is_multiple_of(5);
+                let connector = connector_a.clone();
+
+                clients.spawn(async move {
+                    if case_id.is_multiple_of(3) {
+                        tokio::task::yield_now().await;
+                    }
+                    let mut stream = connector.connect().unwrap();
+                    let mut header = [0u8; 16];
+                    header[..8].copy_from_slice(&case_id.to_le_bytes());
+                    header[8..12].copy_from_slice(&(payload_len as u32).to_le_bytes());
+                    header[12] = u8::from(drop_without_flush);
+                    header[13..].copy_from_slice(b"smx");
+                    write_stress_chunks(&mut stream, &header, chunk_seed).await;
+
+                    let payload_seed = SEED ^ case_id;
+                    let payload = stress_payload(payload_seed, payload_len);
+                    if case_id.is_multiple_of(7) {
+                        stream.write_all(&payload).await.unwrap();
+                    } else {
+                        write_stress_chunks(&mut stream, &payload, chunk_seed ^ payload_seed).await;
+                    }
+
+                    if drop_without_flush {
+                        // Exercise the path that transfers accepted frames to
+                        // the global queue and puts FIN behind them.
+                        drop(stream);
+                        return;
+                    }
+
+                    if case_id.is_multiple_of(2) {
+                        stream.flush().await.unwrap();
+                    }
+                    let response_len = payload_len / 2 + case_id as usize % 257;
+                    let expected = stress_payload(!payload_seed, response_len);
+                    let mut actual = vec![0; response_len];
+                    stream.read_exact(&mut actual).await.unwrap();
+                    assert_eq!(actual, expected, "response mismatch for case {case_id}");
+                    // Tell the responder it may send FIN. Without this
+                    // application-level handshake, its valid early FIN can
+                    // race our final request flush and make the stress test
+                    // assert a stronger half-close contract than smux has.
+                    stream.write_all(&[0xac]).await.unwrap();
+                    stream.shutdown().await.unwrap();
+                });
+            }
+
+            let mut servers = tokio::task::JoinSet::new();
+            for _ in 0..streams_per_batch {
+                let mut stream = acceptor_b
+                    .accept()
+                    .await
+                    .expect("session closed during stress");
+                servers.spawn(async move {
+                    let mut header = [0u8; 16];
+                    stream.read_exact(&mut header).await.unwrap();
+                    assert_eq!(&header[13..], b"smx", "corrupted stress header");
+                    let case_id = u64::from_le_bytes(header[..8].try_into().unwrap());
+                    let payload_len =
+                        u32::from_le_bytes(header[8..12].try_into().unwrap()) as usize;
+                    let drop_without_flush = header[12] != 0;
+                    assert!(
+                        payload_len <= MAX_PAYLOAD_SIZE + 1,
+                        "invalid payload length in case {case_id}: {payload_len}"
+                    );
+
+                    let payload_seed = SEED ^ case_id;
+                    let expected = stress_payload(payload_seed, payload_len);
+                    let mut actual = vec![0; payload_len];
+                    stream.read_exact(&mut actual).await.unwrap();
+                    assert_eq!(actual, expected, "request mismatch for case {case_id}");
+
+                    if drop_without_flush {
+                        let mut byte = [0u8; 1];
+                        assert_eq!(
+                            stream.read(&mut byte).await.unwrap(),
+                            0,
+                            "case {case_id} delivered data after FIN"
+                        );
+                    } else {
+                        let response_len = payload_len / 2 + case_id as usize % 257;
+                        let response = stress_payload(!payload_seed, response_len);
+                        write_stress_chunks(&mut stream, &response, !case_id).await;
+                        if let Err(error) = stream.flush().await {
+                            assert_eq!(
+                                error.kind(),
+                                std::io::ErrorKind::ConnectionReset,
+                                "unexpected response flush error for case {case_id}: {error}"
+                            );
+                            // The client only sends FIN after verifying the
+                            // complete response, so this race still proves
+                            // that every accepted byte reached the peer.
+                            return;
+                        }
+                        let mut ack = [0u8; 1];
+                        stream.read_exact(&mut ack).await.unwrap();
+                        assert_eq!(ack, [0xac], "invalid completion ack for case {case_id}");
+                        stream.shutdown().await.unwrap();
+                    }
+                });
+            }
+
+            while let Some(result) = clients.join_next().await {
+                result.unwrap();
+            }
+            while let Some(result) = servers.join_next().await {
+                result.unwrap();
+            }
+
+            assert_eq!(connector_a.get_num_streams(), 0);
+            assert_eq!(connector_b.get_num_streams(), 0);
+            assert_eq!(
+                connector_a.get_num_tracked_streams(),
+                0,
+                "client retained stream handles after batch {batch}"
+            );
+            assert_eq!(
+                connector_b.get_num_tracked_streams(),
+                0,
+                "server retained stream handles after batch {batch}"
+            );
+            assert!(!worker_a.is_finished());
+            assert!(!worker_b.is_finished());
+        }
+
+        connector_a.close().await.unwrap();
+        let local_worker_result = tokio::time::timeout(Duration::from_secs(2), worker_a)
+            .await
+            .expect("local worker did not stop after stress close")
+            .unwrap();
+        assert!(matches!(
+            local_worker_result,
+            Ok(()) | Err(crate::error::MuxError::ConnectionClosed)
+        ));
+        let _ = tokio::time::timeout(Duration::from_secs(2), worker_b)
+            .await
+            .expect("peer worker did not observe stress close")
+            .unwrap();
+    }
+
+    #[test]
+    fn test_codec_randomized_fragmentation_round_trip() {
+        use bytes::{Bytes, BytesMut};
+        use tokio_util::codec::{Decoder, Encoder};
+
+        use crate::frame::{MuxCodec, MuxCommand, MuxFrame};
+
+        const SEED: u64 = 0xd15c_a11e_5eed_1234;
+        const FRAMES: usize = 256;
+        let mut rng = StdRng::seed_from_u64(SEED);
+        let mut encoder = MuxCodec {};
+        let mut wire = BytesMut::new();
+        let mut expected = Vec::with_capacity(FRAMES);
+
+        for index in 0..FRAMES {
+            let command = match rng.random_range(0..4) {
+                0 => MuxCommand::Sync,
+                1 => MuxCommand::Finish,
+                2 => MuxCommand::Push,
+                _ => MuxCommand::Nop,
+            };
+            let stream_id = if command == MuxCommand::Nop {
+                0
+            } else {
+                rng.random_range(1..=u32::MAX)
+            };
+            let payload_len = if command == MuxCommand::Push {
+                match index % 16 {
+                    0 => 0,
+                    1 => 1,
+                    2 => MAX_PAYLOAD_SIZE - 1,
+                    3 => MAX_PAYLOAD_SIZE,
+                    _ => rng.random_range(0..=4096),
+                }
+            } else {
+                0
+            };
+            let payload = stress_payload(rng.random(), payload_len);
+            encoder
+                .encode(
+                    MuxFrame::new(command, stream_id, Bytes::from(payload.clone())),
+                    &mut wire,
+                )
+                .unwrap();
+            expected.push((command, stream_id, payload));
+        }
+
+        let wire = wire.freeze();
+        let mut decoder = MuxCodec {};
+        let mut buffered = BytesMut::new();
+        let mut actual = Vec::with_capacity(FRAMES);
+        let mut offset = 0;
+        while offset < wire.len() {
+            let chunk_len = rng.random_range(1..=257).min(wire.len() - offset);
+            buffered.extend_from_slice(&wire[offset..offset + chunk_len]);
+            offset += chunk_len;
+            while let Some(frame) = decoder.decode(&mut buffered).unwrap() {
+                actual.push((
+                    frame.header.command,
+                    frame.header.stream_id,
+                    frame.payload.to_vec(),
+                ));
+            }
+        }
+        while let Some(frame) = decoder.decode(&mut buffered).unwrap() {
+            actual.push((
+                frame.header.command,
+                frame.header.stream_id,
+                frame.payload.to_vec(),
+            ));
+        }
+
+        assert!(buffered.is_empty());
+        assert_eq!(actual, expected);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn test_stress_randomized_stream_lifecycles() {
+        tokio::time::timeout(
+            Duration::from_secs(20),
+            run_randomized_lifecycle_stress(8, 48),
+        )
+        .await
+        .expect("randomized lifecycle stress test deadlocked");
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn test_stress_sustained_bidirectional_backpressure() {
+        const STREAMS: usize = 24;
+        const ROUNDS: usize = 96;
+        const SEED_A: u64 = 0xa11c_e001_1234_5678;
+        const SEED_B: u64 = 0xb0b0_0002_8765_4321;
+
+        tokio::time::timeout(Duration::from_secs(20), async {
+            let (a, b) = tokio::io::duplex(64);
+            let (connector_a, _acceptor_a, worker_a) = MuxBuilder::client()
+                .with_max_tx_queue(NonZeroUsize::new(1).unwrap())
+                .with_max_rx_queue(NonZeroUsize::new(1).unwrap())
+                .with_connection(a)
+                .build();
+            let (connector_b, mut acceptor_b, worker_b) = MuxBuilder::server()
+                .with_max_tx_queue(NonZeroUsize::new(1).unwrap())
+                .with_max_rx_queue(NonZeroUsize::new(1).unwrap())
+                .with_connection(b)
+                .build();
+            tokio::spawn(worker_a);
+            tokio::spawn(worker_b);
+
+            let mut pairs = Vec::with_capacity(STREAMS);
+            for _ in 0..STREAMS {
+                let local = connector_a.connect().unwrap();
+                let peer = acceptor_b.accept().await.unwrap();
+                pairs.push((local, peer));
+            }
+
+            let mut tasks = tokio::task::JoinSet::new();
+            for (index, (local, peer)) in pairs.into_iter().enumerate() {
+                tasks.spawn(async move {
+                    let a_seed = SEED_A ^ index as u64;
+                    let b_seed = SEED_B ^ index as u64;
+                    let before_shutdown = Arc::new(tokio::sync::Barrier::new(2));
+                    tokio::join!(
+                        run_full_duplex_stress(
+                            local,
+                            a_seed,
+                            b_seed,
+                            ROUNDS,
+                            before_shutdown.clone()
+                        ),
+                        run_full_duplex_stress(peer, b_seed, a_seed, ROUNDS, before_shutdown)
+                    );
+                });
+            }
+            while let Some(result) = tasks.join_next().await {
+                result.unwrap();
+            }
+
+            assert_eq!(connector_a.get_num_tracked_streams(), 0);
+            assert_eq!(connector_b.get_num_tracked_streams(), 0);
+        })
+        .await
+        .expect("sustained bidirectional backpressure test deadlocked");
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn test_stress_concurrent_close_cancellation_and_stream_drop() {
+        use tokio::sync::oneshot;
+
+        const STREAMS: usize = 128;
+        const CLOSERS: usize = 24;
+        const PAYLOAD_LEN: usize = 257;
+
+        tokio::time::timeout(Duration::from_secs(10), async {
+            let (a, mut peer) = tokio::io::duplex(64);
+            let (connector, acceptor, worker) = MuxBuilder::client().with_connection(a).build();
+            let worker = tokio::spawn(worker);
+
+            let mut streams = Vec::with_capacity(STREAMS);
+            for index in 0..STREAMS {
+                let mut stream = connector.connect().unwrap();
+                let payload = stress_payload(index as u64 + 1, PAYLOAD_LEN);
+                stream.write_all(&payload).await.unwrap();
+                streams.push(stream);
+            }
+            drop(streams);
+
+            let mut closers = Vec::with_capacity(CLOSERS);
+            for _ in 0..CLOSERS {
+                let mut connector = connector.clone();
+                let (started_tx, started_rx) = oneshot::channel();
+                let task = tokio::spawn(async move {
+                    let mut close = Box::pin(connector.close());
+                    let mut started_tx = Some(started_tx);
+                    poll_fn(|cx| match close.as_mut().poll(cx) {
+                        Poll::Pending => {
+                            if let Some(tx) = started_tx.take() {
+                                let _ = tx.send(());
+                            }
+                            Poll::Ready(())
+                        }
+                        Poll::Ready(result) => {
+                            panic!("close completed before the blocked carrier drained: {result:?}")
+                        }
+                    })
+                    .await;
+                    close.await
+                });
+                started_rx.await.unwrap();
+                closers.push(task);
+            }
+
+            for (index, closer) in closers.iter().enumerate() {
+                if index % 3 == 0 {
+                    closer.abort();
+                }
+            }
+
+            let reader = tokio::spawn(async move {
+                let mut wire = Vec::new();
+                peer.read_to_end(&mut wire).await.unwrap();
+                wire
+            });
+
+            for (index, closer) in closers.into_iter().enumerate() {
+                match closer.await {
+                    Err(error) if index % 3 == 0 => assert!(error.is_cancelled()),
+                    Ok(result) => result.unwrap(),
+                    Err(error) => panic!("active close task failed: {error}"),
+                }
+            }
+            let worker_result = worker.await.unwrap();
+            assert!(matches!(
+                worker_result,
+                Ok(()) | Err(crate::error::MuxError::ConnectionClosed)
+            ));
+            let wire = reader.await.unwrap();
+
+            let mut offset = 0;
+            let mut push_bytes = 0;
+            while offset < wire.len() {
+                assert!(wire.len() - offset >= 8, "truncated frame header");
+                let command = wire[offset + 1];
+                let payload_len = u16::from_le_bytes([wire[offset + 2], wire[offset + 3]]) as usize;
+                assert!(
+                    wire.len() - offset >= 8 + payload_len,
+                    "truncated frame payload"
+                );
+                if command == 2 {
+                    push_bytes += payload_len;
+                }
+                offset += 8 + payload_len;
+            }
+            assert_eq!(
+                push_bytes,
+                STREAMS * PAYLOAD_LEN,
+                "orderly close lost bytes accepted before shutdown"
+            );
+
+            drop((connector, acceptor));
+        })
+        .await
+        .expect("concurrent close/cancellation stress test deadlocked");
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[ignore = "long-running soak test; run explicitly in release mode"]
+    async fn test_soak_randomized_stream_lifecycles() {
+        tokio::time::timeout(
+            Duration::from_secs(120),
+            run_randomized_lifecycle_stress(64, 64),
+        )
+        .await
+        .expect("randomized lifecycle soak test deadlocked");
+    }
 }
diff --git a/src/mux.rs b/src/mux.rs
index 3ae4b37..12f1b6c 100644
--- a/src/mux.rs
+++ b/src/mux.rs
@@ -218,6 +218,11 @@ impl<T: TokioConn> MuxConnector<T> {
     pub(crate) fn is_closing_inline(&self) -> bool {
         self.state.lock().inline_closers > 0
     }
+
+    #[cfg(test)]
+    pub(crate) fn get_num_tracked_streams(&self) -> usize {
+        self.state.lock().handles.len()
+    }
 }
 
 impl<T: TokioConn> Clone for MuxConnector<T> {