From bc587b2a2b3cd2d66d3de7ac61030f7e24cdc1e8 Mon Sep 17 00:00:00 2001
From: IntellyCode <47359527+IntellyCode@users.noreply.github.com>
Date: Tue, 18 Aug 2026 06:04:26 +0200
Subject: [PATCH 01/13] Record an idle session's timeout at debug (#87)

A session that goes quiet for longer than the configured timeout is
closed, which is the configured behaviour rather than a fault. The
surrounding lifecycle events already record at debug: a session
beginning, a session dropped, a session closed after 2MSL. A proxy
holding many idle connections reports every one of these twice at
warn, which is where an application looks for problems it can act on.
---
 src/stream/tcp.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index c545978..296b84e 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -306,12 +306,12 @@ impl AsyncRead for IpStackTcpStream {
                 let mut tcb = self.tcb.lock().unwrap();
                 let (seq, ack) = (tcb.get_seq().0, tcb.get_ack().0);
                 let l_info = format!("local {{ seq: {seq}, ack: {ack} }}");
-                log::warn!("{network_tuple} {state:?}: [poll_read] {l_info}, session timeout reached, closing forcefully...");
+                log::debug!("{network_tuple} {state:?}: [poll_read] {l_info}, session timeout reached, closing forcefully...");
                 let sender = &self.up_packet_sender;
                 write_packet_to_device(sender, network_tuple, &tcb, None, ACK | RST, None, None)?;
                 tcb.change_state(TcpState::Closed);
                 let state = tcb.get_state();
-                log::warn!("{network_tuple} {state:?}: [poll_read] {l_info}, session notified to close");
+                log::debug!("{network_tuple} {state:?}: [poll_read] {l_info}, session notified to close");
             }
             self.shutdown.lock().unwrap().ready();
 

From 0f95edc89f23c6700e858eeb5120dd7f6dd1a1c7 Mon Sep 17 00:00:00 2001
From: IntellyCode <47359527+IntellyCode@users.noreply.github.com>
Date: Tue, 18 Aug 2026 06:04:40 +0200
Subject: [PATCH 02/13] Close a dropped TCP session without blocking on its
 task (#86)

`IpStackTcpStream::drop` waited for the session's task to finish, through
`block_in_place` and `block_on`. `block_in_place` works by handing the worker's
core to the blocking pool, and the pool stops accepting work once the runtime
begins shutting down. A session dropped from that point on blocks its worker
forever: the core is never handed off, the task it waits for has no thread left
to run on, and the runtime's own shutdown waits for that worker in turn.

The farewell packet reaches the device through an unbounded channel, so it is
sent from the drop itself. The task is told to exit through the channel it
already watches, and aborted rather than awaited. Aborting drops the task's
`destroy_messenger`, and the watcher that removes the session treats a dropped
sender the same as a sent one, so the session still leaves the stack.
---
 src/stream/tcp.rs | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index 296b84e..2be4bfe 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -434,11 +434,22 @@ impl Drop for IpStackTcpStream {
         log::trace!("{nt} {state:?}: [drop] session dropping, ========================= ");
         if let Some(task_handle) = self.task_handle.take() {
             if !task_handle.is_finished() {
+                // The farewell packet reaches the device through an unbounded channel, so it
+                // is sent here rather than left to the task.
+                {
+                    let mut tcb = self.tcb.lock().unwrap();
+                    if let Err(e) = send_fin_n_change_state_to_fin_wait1("[drop]", nt, &self.up_packet_sender, &mut tcb) {
+                        log::debug!("{nt} {state:?}: [drop] cannot send the farewell packet: {e}");
+                    }
+                }
                 if let Some(notifier) = self.exit_notifier.take() {
-                    _ = tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(notifier.send(())));
+                    // The channel holds ten slots and one signal ends the task, so the send
+                    // needs no runtime of its own.
+                    _ = notifier.try_send(());
                 }
-                // synchronously wait for the task to finish
-                _ = tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(task_handle));
+                // Dropping the task drops its `destroy_messenger`, which wakes the watcher
+                // that removes this session from the stack.
+                task_handle.abort();
             } else {
                 log::trace!("{nt} {state:?}: [drop] task already finished, no need to wait exiting");
             }

From e1d8506e39b661df01ed37a9e8d6a960d6b56af0 Mon Sep 17 00:00:00 2001
From: IntellyCode <47359527+IntellyCode@users.noreply.github.com>
Date: Sun, 30 Aug 2026 18:37:53 +0300
Subject: [PATCH 03/13] Enforce the TCP receive window to bound per-session
 memory (#90)

* Enforce the TCP receive window to bound per-session memory

The receive window ipstack advertises now holds. A segment beyond it is
dropped for the peer to resend, the head-of-line segment always admitted
so the stream advances. The handoff channel to the reader is bounded and
filled by reserving a slot before consuming, so buffered data leaves the
reassembly map only once it has a home; a reader that frees space wakes
the loop to flush more and the follow-up ACK carries the reopened window.
The window is advertised as zero below one segment, so a stalled reader
puts the peer into persist mode until space frees.

Consuming trims a stale head entry a re-segmented retransmission left
below the ack, and a FIN is accepted only once the data before it has
been consumed, so a full channel never strands the tail.

* Cover reserve-before-consume with an async test

The stream's first tokio test drives extract_data_n_write_upstream
against a full handoff channel: buffered data stays in the reassembly
map and the ack holds until the reader drains a slot, then the tail
flushes and the ack advances.
---
 src/stream/tcb.rs |  70 +++++++++++++++++++++++++++++++-
 src/stream/tcp.rs | 100 ++++++++++++++++++++++++++++++++++++++++------
 2 files changed, 157 insertions(+), 13 deletions(-)

diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs
index 1d496d3..9446aaa 100644
--- a/src/stream/tcb.rs
+++ b/src/stream/tcb.rs
@@ -4,6 +4,7 @@ use std::{collections::BTreeMap, time::Duration};
 
 pub(super) const MAX_UNACK: u32 = 1024 * 16; // 16KB
 pub(super) const READ_BUFFER_SIZE: usize = 1024 * 16; // 16KB
+pub(super) const READ_CHUNK: usize = 8192; // 8KB, bytes drained from the reassembly buffer per handoff
 pub(super) const MAX_COUNT_FOR_DUP_ACK: usize = 3; // Maximum number of duplicate ACKs before retransmission
 
 /// Retransmission timeout
@@ -121,6 +122,13 @@ impl Tcb {
             log::warn!("{:?}: Received packet seq {seq} < self ack {}, len = {}", self.state, self.ack, buf.len());
             return;
         }
+        // The head-of-line segment always advances the stream, so it is admitted even at the limit;
+        // any other segment beyond the receive window is dropped for the peer's RTO to resend.
+        if seq != self.ack && self.get_unordered_packets_total_len() >= self.read_buffer_size {
+            #[rustfmt::skip]
+            log::warn!("{:?}: Receive window full, dropping packet seq {seq}, len = {}", self.state, buf.len());
+            return;
+        }
         self.unordered_packets.insert(seq, buf);
     }
     pub(super) fn get_available_read_buffer_size(&self) -> usize {
@@ -137,10 +145,21 @@ impl Tcb {
 
         while remaining_bytes > 0 {
             if let Some(seq) = self.unordered_packets.keys().next().copied() {
-                if seq != self.ack {
+                if seq > self.ack {
                     break; // sequence number is not continuous, stop extracting
                 }
 
+                if seq < self.ack {
+                    // A retransmission re-segmented across `ack` left a stale head entry; trim the
+                    // part already delivered so consumption can continue from `ack`.
+                    let payload = self.unordered_packets.remove(&seq).unwrap();
+                    let consumed = self.ack.distance(seq) as usize;
+                    if consumed < payload.len() {
+                        self.unordered_packets.insert(self.ack, payload[consumed..].to_vec());
+                    }
+                    continue;
+                }
+
                 // remove and get the first packet
                 let mut payload = self.unordered_packets.remove(&seq).unwrap();
                 let payload_len = payload.len();
@@ -408,6 +427,55 @@ mod tests {
         assert!(data.is_none());
     }
 
+    #[test]
+    fn test_add_unordered_packet_enforces_read_buffer() {
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            RTO,
+            MAX_RETRANSMIT_COUNT,
+        );
+
+        // fill the receive buffer to its limit with an out-of-order gap held open
+        tcb.add_unordered_packet(SeqNum(1000 + READ_BUFFER_SIZE as u32), vec![7; READ_BUFFER_SIZE]);
+        assert_eq!(tcb.get_unordered_packets_total_len(), READ_BUFFER_SIZE);
+
+        // a further out-of-order segment is dropped, keeping the buffer bounded
+        tcb.add_unordered_packet(SeqNum(1000 + 2 * READ_BUFFER_SIZE as u32), vec![8; 500]);
+        assert_eq!(tcb.get_unordered_packets_total_len(), READ_BUFFER_SIZE);
+
+        // the head-of-line segment is admitted even at the limit, so the stream advances
+        tcb.add_unordered_packet(SeqNum(1000), vec![9; 500]);
+        assert_eq!(tcb.unordered_packets.get(&SeqNum(1000)).unwrap().len(), 500);
+    }
+
+    #[test]
+    fn test_consume_trims_overlapping_head_entry() {
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            RTO,
+            MAX_RETRANSMIT_COUNT,
+        );
+
+        // an out-of-order segment stored ahead of ack
+        tcb.add_unordered_packet(SeqNum(1200), vec![2; 300]);
+        // the gap-filler that a retransmission re-segmented to overlap the stored one
+        tcb.add_unordered_packet(SeqNum(1000), vec![1; 400]);
+
+        // consuming pulls [1000..1400), advancing ack into the stored entry keyed at 1200
+        let data = tcb.consume_unordered_packets(10_000).unwrap();
+        assert_eq!(data.len(), 500); // 400 + the 100 bytes of the stored entry past ack
+        assert_eq!(tcb.ack, SeqNum(1500));
+        assert_eq!(tcb.unordered_packets.len(), 0);
+    }
+
     #[test]
     fn test_update_inflight_packet_queue() {
         let mut tcb = Tcb::new(
diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index 2be4bfe..d766f16 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -7,7 +7,7 @@ use crate::{
         tcp_flags::{ACK, FIN, PSH, RST, SYN},
         tcp_header_flags, tcp_header_fmt,
     },
-    stream::tcb::{MAX_COUNT_FOR_DUP_ACK, MAX_RETRANSMIT_COUNT, MAX_UNACK, PacketType, READ_BUFFER_SIZE, RTO, Tcb, TcpState},
+    stream::tcb::{MAX_COUNT_FOR_DUP_ACK, MAX_RETRANSMIT_COUNT, MAX_UNACK, PacketType, READ_BUFFER_SIZE, READ_CHUNK, RTO, Tcb, TcpState},
 };
 use etherparse::{IpNumber, Ipv4Header, Ipv6FlowLabel, TcpHeader, TcpOptionElement};
 use std::{
@@ -165,9 +165,10 @@ pub struct IpStackTcpStream {
     write_notify: std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
     destroy_messenger: Option<::tokio::sync::oneshot::Sender<()>>,
     timeout: Pin<Box<tokio::time::Sleep>>,
-    data_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
-    data_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
+    data_tx: tokio::sync::mpsc::Sender<Vec<u8>>,
+    data_rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
     read_notify: std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
+    drain_notify: Arc<tokio::sync::Notify>,
     task_handle: Option<tokio::task::JoinHandle<std::io::Result<()>>>,
     exit_notifier: Option<tokio::sync::mpsc::Sender<()>>,
     temp_read_buffer: Vec<u8>,
@@ -205,7 +206,8 @@ impl IpStackTcpStream {
         }
 
         let (stream_sender, stream_receiver) = tokio::sync::mpsc::unbounded_channel::<NetworkPacket>();
-        let (data_tx, data_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+        let data_channel_len = config.read_buffer_size.div_ceil(READ_CHUNK).max(1);
+        let (data_tx, data_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(data_channel_len);
         let deadline = tokio::time::Instant::now() + config.timeout;
 
         let mut stream = IpStackTcpStream {
@@ -222,6 +224,7 @@ impl IpStackTcpStream {
             data_tx,
             data_rx,
             read_notify: std::sync::Arc::new(std::sync::Mutex::new(None)),
+            drain_notify: Arc::new(tokio::sync::Notify::new()),
             task_handle: None,
             exit_notifier: None,
             temp_read_buffer: Vec::new(),
@@ -330,6 +333,8 @@ impl AsyncRead for IpStackTcpStream {
                     buf.put_slice(&data[..capacity]);
                     self.temp_read_buffer.extend_from_slice(&data[capacity..]);
                 }
+                // A channel slot just freed, so wake the loop to flush more and reopen the window.
+                self.drain_notify.notify_one();
                 Poll::Ready(Ok(()))
             }
             Poll::Ready(None) => Poll::Ready(Ok(())),
@@ -471,6 +476,7 @@ impl IpStackTcpStream {
         let write_notify = self.write_notify.clone();
         let read_notify = self.read_notify.clone();
         let data_tx = self.data_tx.clone();
+        let drain_notify = self.drain_notify.clone();
         let destroy_messenger = self.destroy_messenger.take();
 
         let (exit_task_notifier, exit_monitor) = tokio::sync::mpsc::channel::<()>(10);
@@ -489,6 +495,7 @@ impl IpStackTcpStream {
                 write_notify,
                 read_notify,
                 data_tx,
+                drain_notify,
                 exit_monitor,
             )
             .await;
@@ -516,7 +523,8 @@ async fn tcp_main_logic_loop(
     network_tuple: NetworkTuple,
     write_notify: std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
     read_notify: std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
-    data_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    data_tx: tokio::sync::mpsc::Sender<Vec<u8>>,
+    drain_notify: Arc<tokio::sync::Notify>,
     mut exit_monitor: tokio::sync::mpsc::Receiver<()>,
 ) -> std::io::Result<()> {
     {
@@ -642,6 +650,13 @@ async fn tcp_main_logic_loop(
                 log::debug!("{network_tuple} task exited due to exit signal");
                 break;
             }
+            _ = drain_notify.notified() => {
+                // The upstream reader freed channel space, so flush whatever is buffered and
+                // let the follow-up ACK carry the reopened window.
+                let mut tcb = tcb.lock().unwrap();
+                extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
+                continue;
+            }
             network_packet = stream_receiver.recv() => network_packet,
         };
 
@@ -743,7 +758,7 @@ async fn tcp_main_logic_loop(
                         }
                         PacketType::Invalid => {}
                     }
-                } else if flags == (ACK | FIN) {
+                } else if flags == (ACK | FIN) && tcb.get_ack() == incoming_seq {
                     // The other side is closing the connection, we need to send an ACK and change state to CloseWait
                     tcb.increase_ack();
                     write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK, None, None)?;
@@ -838,7 +853,7 @@ async fn tcp_main_logic_loop(
                 log::trace!("{network_tuple} {state:?}: Received final ACK, transitioned to {new_state:?}");
             }
             TcpState::FinWait1 => {
-                if flags & (ACK | FIN) == (ACK | FIN) && len == 0 {
+                if flags & (ACK | FIN) == (ACK | FIN) && len == 0 && tcb.get_ack() == incoming_seq {
                     // If the received packet is an ACK with FIN, we need to send an ACK and change state to TimeWait directly, not to FinWait2
                     tcb.increase_ack();
                     write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK, None, None)?;
@@ -863,7 +878,7 @@ async fn tcp_main_logic_loop(
                 }
             }
             TcpState::FinWait2 => {
-                if flags & (ACK | FIN) == (ACK | FIN) && len == 0 {
+                if flags & (ACK | FIN) == (ACK | FIN) && len == 0 && tcb.get_ack() == incoming_seq {
                     tcb.increase_ack();
                     write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK, None, None)?;
                     tcb.change_state(TcpState::TimeWait);
@@ -914,7 +929,7 @@ fn extract_data_n_write_upstream(
     up_packet_sender: &PacketSender,
     tcb: &mut Tcb,
     network_tuple: NetworkTuple,
-    data_tx: &tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
+    data_tx: &tokio::sync::mpsc::Sender<Vec<u8>>,
     read_notify: &std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
 ) -> std::io::Result<()> {
     let (state, seq, ack) = (tcb.get_state(), tcb.get_seq(), tcb.get_ack());
@@ -924,10 +939,23 @@ fn extract_data_n_write_upstream(
         return Ok(());
     }
 
-    if let Some(data) = tcb.consume_unordered_packets(8192) {
+    // Reserve the handoff slot before consuming, so buffered data is removed only once it has a
+    // guaranteed home; the reserved permit shrinks the advertised window until the reader drains it.
+    let permit = match data_tx.try_reserve() {
+        Ok(permit) => permit,
+        Err(tokio::sync::mpsc::error::TrySendError::Full(())) => {
+            write_packet_to_device(up_packet_sender, network_tuple, tcb, None, ACK, None, None)?;
+            return Ok(());
+        }
+        Err(tokio::sync::mpsc::error::TrySendError::Closed(())) => {
+            return Err(std::io::Error::new(BrokenPipe, "data channel closed"));
+        }
+    };
+
+    if let Some(data) = tcb.consume_unordered_packets(READ_CHUNK) {
         let hint = if state == TcpState::Established { "normally" } else { "still" };
         log::trace!("{network_tuple} {state:?}: {l_info} {hint} receiving data, len = {}", data.len());
-        data_tx.send(data).map_err(|e| std::io::Error::new(BrokenPipe, e))?;
+        permit.send(data);
         read_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
         write_packet_to_device(up_packet_sender, network_tuple, tcb, None, ACK, None, None)?;
     }
@@ -947,7 +975,11 @@ pub(crate) fn write_packet_to_device(
 ) -> std::io::Result<usize> {
     use std::io::Error;
     let seq = seq.unwrap_or(tcb.get_seq()).0;
-    let (ack, window_size) = (tcb.get_ack().0, tcb.get_recv_window().max(tcb.get_mtu()));
+    // Silly-window-syndrome avoidance: advertise a real window only when a full segment fits,
+    // otherwise advertise zero so the peer enters persist mode until the reader frees space.
+    let recv_window = tcb.get_recv_window();
+    let window_size = if recv_window >= tcb.get_mtu() { recv_window } else { 0 };
+    let ack = tcb.get_ack().0;
     let (src, dst) = (tuple.dst, tuple.src); // Note: The address is reversed here
     let calc = |ip_header_len: usize, tcp_header_len: usize| tcb.calculate_payload_max_len(ip_header_len, tcp_header_len);
     let packet = create_raw_packet(
@@ -1048,3 +1080,47 @@ pub(crate) fn create_raw_packet(
         payload: Some(payload),
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::stream::tcb::{MAX_COUNT_FOR_DUP_ACK, MAX_RETRANSMIT_COUNT, MAX_UNACK, READ_BUFFER_SIZE, RTO};
+
+    #[tokio::test]
+    async fn extract_reserves_before_consuming() {
+        let (up_tx, _up_rx) = tokio::sync::mpsc::unbounded_channel::<NetworkPacket>();
+        let (data_tx, mut data_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(1);
+        let read_notify = std::sync::Arc::new(std::sync::Mutex::new(None));
+        let nt = NetworkTuple::new("1.1.1.1:1".parse().unwrap(), "2.2.2.2:2".parse().unwrap(), true);
+
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            RTO,
+            MAX_RETRANSMIT_COUNT,
+        );
+        tcb.change_state(TcpState::Established);
+        tcb.add_unordered_packet(SeqNum(1000), vec![1; 500]);
+        tcb.add_unordered_packet(SeqNum(1500), vec![2; 500]);
+
+        // first extract fills the single channel slot and advances ack over the first chunk
+        extract_data_n_write_upstream(&up_tx, &mut tcb, nt, &data_tx, &read_notify).unwrap();
+        assert_eq!(tcb.get_ack(), SeqNum(2000));
+
+        // channel is full: extract leaves the remaining data in the map and does not advance ack
+        tcb.add_unordered_packet(SeqNum(2000), vec![3; 500]);
+        extract_data_n_write_upstream(&up_tx, &mut tcb, nt, &data_tx, &read_notify).unwrap();
+        assert_eq!(tcb.get_ack(), SeqNum(2000));
+        assert_eq!(tcb.get_unordered_packets_total_len(), 500);
+
+        // draining the reader frees a slot, and the next extract flushes the tail
+        let first = data_rx.recv().await.unwrap();
+        assert_eq!(first.len(), 1000);
+        extract_data_n_write_upstream(&up_tx, &mut tcb, nt, &data_tx, &read_notify).unwrap();
+        assert_eq!(tcb.get_ack(), SeqNum(2500));
+        assert_eq!(tcb.get_unordered_packets_total_len(), 0);
+    }
+}

From f1444f7977e07ce78f43183a005ae8593f513ad1 Mon Sep 17 00:00:00 2001
From: IntellyCode <47359527+IntellyCode@users.noreply.github.com>
Date: Wed, 16 Sep 2026 05:24:49 +0200
Subject: [PATCH 04/13] Negotiate window scaling so a peer's window reaches
 past 65535 (#91)

* Negotiate window scaling so a peer's window reaches past 65535

The SYN exchange settles a shift for each direction. A peer whose SYN
carries a window scale option gets one back in the SYN-ACK, and every
window it advertises afterwards is honoured shifted left by its count,
limited to the 14 the RFC allows and the 1 GiB that implies. The stack's
own shift is the smallest that expresses its read buffer in the 16-bit
field, zero for the 16 KiB default, and every window it advertises is
shifted down by it. The window field of a segment carrying SYN travels
unscaled in both directions.

A peer that offers no scaling leaves both shifts off and sees the raw
16-bit window it saw before.

The options a segment carries are assembled where its flags are known,
so create_raw_packet writes the list it is handed.

* Wake a writer held by a closed peer window and pin the SYN window

A writer held by a closed peer window is woken by the segment that
completes the handshake and by an established connection's data segments,
as every other segment that can reopen the window already does.

The window tests assert the SYN's window before any later segment
replaces it, the handshake test asserts the shift the SYN-ACK carries, and
the zero-window test uses a recording waker so it fails when a reopening
segment leaves the writer asleep. A read buffer too large for any shift is
limited to the maximum, with a warning.
---
 src/stream/tcb.rs | 144 +++++++++++++++++++++++++++++++++++--
 src/stream/tcp.rs | 179 +++++++++++++++++++++++++++++++++++++++++-----
 2 files changed, 300 insertions(+), 23 deletions(-)

diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs
index 9446aaa..ec73e7c 100644
--- a/src/stream/tcb.rs
+++ b/src/stream/tcb.rs
@@ -13,6 +13,9 @@ pub(super) const RTO: std::time::Duration = std::time::Duration::from_secs(1);
 /// Maximum count of retransmissions before dropping the packet
 pub(super) const MAX_RETRANSMIT_COUNT: usize = 3;
 
+/// Maximum window scale shift count, which RFC 7323 §2.3 limits to 14 for a maximum window of 1 GiB
+const MAX_WINDOW_SHIFT: u8 = 14;
+
 #[derive(Debug, PartialEq, Clone, Copy)]
 pub(crate) enum TcpState {
     // Init, /* Since we always act as a server, it starts from `Listen`, so we don't use states Init & SynSent. */
@@ -44,13 +47,19 @@ pub(super) enum PacketType {
 /// - `unordered_packets` is the bytes stream received from the lower device,
 ///   which can be acknowledged and extracted by `consume_unordered_packets` method
 ///   then can be read by upstream application via `Tcp::poll_read` method.
+/// - `send_window_shift` is the peer's window scale, applied to every window the peer advertises,
+///   and `recv_window_shift` is this stack's own, applied to every window this stack advertises.
+///   Both are settled by the SYN exchange, and `recv_window_shift` is `None` when the peer's SYN
+///   carried no window scale option, which leaves both directions unscaled.
 #[derive(Debug, Clone)]
 pub(crate) struct Tcb {
     seq: SeqNum,
     ack: SeqNum,
     mtu: u16,
     last_received_ack: SeqNum,
-    send_window: u16,
+    send_window: u32,
+    send_window_shift: u8,
+    recv_window_shift: Option<u8>,
     state: TcpState,
     inflight_packets: BTreeMap<SeqNum, InflightPacket>,
     unordered_packets: BTreeMap<SeqNum, Vec<u8>>,
@@ -64,8 +73,11 @@ pub(crate) struct Tcb {
 }
 
 impl Tcb {
+    #[allow(clippy::too_many_arguments)]
     pub(super) fn new(
         ack: SeqNum,
+        peer_window: u16,
+        peer_window_shift: Option<u8>,
         mtu: u16,
         max_unacked_bytes: u32,
         read_buffer_size: usize,
@@ -77,12 +89,32 @@ impl Tcb {
         let seq = 100;
         #[cfg(not(debug_assertions))]
         let seq = rand::RngExt::random::<u32>(&mut rand::rng());
+        let send_window_shift = peer_window_shift.map_or(0, |shift| {
+            if shift > MAX_WINDOW_SHIFT {
+                log::warn!("Peer window scale shift count {shift} is too large, limiting it to {MAX_WINDOW_SHIFT}");
+                MAX_WINDOW_SHIFT
+            } else {
+                shift
+            }
+        });
+        // The stack scales its own receive window by the smallest shift that expresses the whole
+        // read buffer in the 16-bit window field.
+        let recv_window_shift = peer_window_shift.map(|_| {
+            (0..=MAX_WINDOW_SHIFT)
+                .find(|&shift| read_buffer_size >> shift <= u16::MAX as usize)
+                .unwrap_or_else(|| {
+                    log::warn!("Read buffer size {read_buffer_size} is too large to scale, limiting the shift count to {MAX_WINDOW_SHIFT}");
+                    MAX_WINDOW_SHIFT
+                })
+        });
         Tcb {
             seq: seq.into(),
             ack,
             mtu,
             last_received_ack: seq.into(),
-            send_window: u16::MAX,
+            send_window: peer_window as u32,
+            send_window_shift,
+            recv_window_shift,
             state: TcpState::Listen,
             inflight_packets: BTreeMap::new(),
             unordered_packets: BTreeMap::new(),
@@ -209,15 +241,29 @@ impl Tcb {
     pub(super) fn get_state(&self) -> TcpState {
         self.state
     }
-    pub(super) fn update_send_window(&mut self, window: u16) {
-        self.send_window = window;
+    fn honoured_window(&self, tcp_header: &TcpHeader) -> u32 {
+        // RFC 7323 §2.3: SND.WND = SEG.WND << Snd.Wind.Shift, except on a segment carrying SYN,
+        // whose window field is never scaled.
+        let window = tcp_header.window_size as u32;
+        if tcp_header.syn { window } else { window << self.send_window_shift }
     }
-    pub(super) fn get_send_window(&self) -> u16 {
+    pub(super) fn update_send_window(&mut self, tcp_header: &TcpHeader) {
+        self.send_window = self.honoured_window(tcp_header);
+    }
+    pub(super) fn get_send_window(&self) -> u32 {
         self.send_window
     }
     pub(super) fn get_recv_window(&self) -> u16 {
         self.get_available_read_buffer_size().try_into().unwrap_or(u16::MAX)
     }
+    pub(super) fn get_scaled_recv_window(&self) -> u16 {
+        // RFC 7323 §2.3: SEG.WND = RCV.WND >> Rcv.Wind.Shift
+        let window = self.get_available_read_buffer_size() >> self.recv_window_shift.unwrap_or(0);
+        window.try_into().unwrap_or(u16::MAX)
+    }
+    pub(super) fn get_recv_window_shift(&self) -> Option<u8> {
+        self.recv_window_shift
+    }
     // #[inline(always)]
     // pub(super) fn buffer_size(&self, payload_len: u16) -> u16 {
     //     match MAX_UNACK - self.inflight_packets.len() as u32 {
@@ -234,7 +280,7 @@ impl Tcb {
     pub(super) fn check_pkt_type(&self, tcp_header: &TcpHeader, payload: &[u8]) -> PacketType {
         let rcvd_ack = SeqNum(tcp_header.acknowledgment_number);
         let rcvd_seq = SeqNum(tcp_header.sequence_number);
-        let rcvd_window = tcp_header.window_size;
+        let rcvd_window = self.honoured_window(tcp_header);
         let len = payload.len();
         let res = if rcvd_ack > self.seq {
             PacketType::Invalid
@@ -339,7 +385,7 @@ impl Tcb {
     pub fn is_send_buffer_full(&self) -> bool {
         // To respect the receiver's window (remote_window) size and avoid sending too many unacknowledged packets, which may cause packet loss
         // Simplified version: min(cwnd, rwnd)
-        self.seq.distance(self.get_last_received_ack()) >= self.max_unacked_bytes.min(self.get_send_window() as u32)
+        self.seq.distance(self.get_last_received_ack()) >= self.max_unacked_bytes.min(self.get_send_window())
     }
 }
 
@@ -391,6 +437,8 @@ mod tests {
     fn test_get_unordered_packets_with_max_bytes() {
         let mut tcb = Tcb::new(
             SeqNum(1000),
+            u16::MAX,
+            None,
             1500,
             MAX_UNACK,
             READ_BUFFER_SIZE,
@@ -431,6 +479,8 @@ mod tests {
     fn test_add_unordered_packet_enforces_read_buffer() {
         let mut tcb = Tcb::new(
             SeqNum(1000),
+            u16::MAX,
+            None,
             1500,
             MAX_UNACK,
             READ_BUFFER_SIZE,
@@ -456,6 +506,8 @@ mod tests {
     fn test_consume_trims_overlapping_head_entry() {
         let mut tcb = Tcb::new(
             SeqNum(1000),
+            u16::MAX,
+            None,
             1500,
             MAX_UNACK,
             READ_BUFFER_SIZE,
@@ -480,6 +532,8 @@ mod tests {
     fn test_update_inflight_packet_queue() {
         let mut tcb = Tcb::new(
             SeqNum(1000),
+            u16::MAX,
+            None,
             1500,
             MAX_UNACK,
             READ_BUFFER_SIZE,
@@ -512,6 +566,8 @@ mod tests {
     fn test_update_inflight_packet_queue_cumulative_ack() {
         let mut tcb = Tcb::new(
             SeqNum(1000),
+            u16::MAX,
+            None,
             1500,
             MAX_UNACK,
             READ_BUFFER_SIZE,
@@ -535,6 +591,8 @@ mod tests {
     fn test_retransmit_with_exponential_backoff() {
         let mut tcb = Tcb::new(
             SeqNum(1000),
+            u16::MAX,
+            None,
             1500,
             MAX_UNACK,
             READ_BUFFER_SIZE,
@@ -563,4 +621,76 @@ mod tests {
         assert!(packets.is_empty());
         assert!(tcb.inflight_packets.is_empty());
     }
+
+    /// A peer shift of 7 is applied to every window the peer advertises after the handshake.
+    #[test]
+    fn test_peer_window_shift_is_taken_from_the_syn() {
+        let mut tcb = window_tcb(4000, Some(7), READ_BUFFER_SIZE);
+        assert_eq!(tcb.get_send_window(), 4000); // the SYN's own window is unscaled
+
+        tcb.update_send_window(&TcpHeader::new(1, 2, 1000, 40_000));
+        assert_eq!(tcb.get_send_window(), 40_000 << 7);
+    }
+
+    /// A peer that offers no window scale option has its windows honoured as they stand.
+    #[test]
+    fn test_no_window_scale_option_leaves_the_window_unshifted() {
+        let mut tcb = window_tcb(4000, None, READ_BUFFER_SIZE);
+        assert_eq!(tcb.get_send_window(), 4000);
+
+        tcb.update_send_window(&TcpHeader::new(1, 2, 1000, 40_000));
+        assert_eq!(tcb.get_send_window(), 40_000);
+        assert_eq!(tcb.get_recv_window_shift(), None);
+    }
+
+    /// RFC 7323 §2.3 limits the shift count to 14, so a larger one is used as 14.
+    #[test]
+    fn test_peer_window_shift_above_the_maximum_is_clamped() {
+        let mut tcb = window_tcb(4000, Some(15), READ_BUFFER_SIZE);
+
+        tcb.update_send_window(&TcpHeader::new(1, 2, 1000, 40_000));
+        assert_eq!(tcb.get_send_window(), 40_000 << MAX_WINDOW_SHIFT);
+    }
+
+    /// A peer opening with a closed window is held to it, whatever shift the same SYN offers.
+    #[test]
+    fn test_zero_peer_window_is_honoured() {
+        assert_eq!(window_tcb(0, None, READ_BUFFER_SIZE).get_send_window(), 0);
+        assert_eq!(window_tcb(0, Some(7), READ_BUFFER_SIZE).get_send_window(), 0);
+    }
+
+    /// The announced shift is the smallest expressing the read buffer in 16 bits, the advertised
+    /// window is the free space shifted down by it, and a peer shift of 0 still enables scaling.
+    #[test]
+    fn test_advertised_window_is_derived_from_the_read_buffer() {
+        assert_eq!(window_tcb(4000, Some(0), 16 * 1024).get_recv_window_shift(), Some(0));
+        assert_eq!(window_tcb(4000, Some(0), 64 * 1024).get_recv_window_shift(), Some(1));
+
+        // 1.2 MiB needs a shift of 5, which expresses the window only in multiples of 32, so the
+        // advertised value rounds down and withholds the remainder rather than overstating the room
+        let buffer = 1_258_291;
+        let mut tcb = window_tcb(4000, Some(0), buffer);
+        assert_eq!(tcb.get_recv_window_shift(), Some(5));
+        assert_eq!(usize::from(tcb.get_scaled_recv_window()), buffer >> 5);
+        assert!(usize::from(tcb.get_scaled_recv_window()) << 5 < buffer);
+
+        // the advertised window shrinks as the buffer fills
+        tcb.add_unordered_packet(SeqNum(1000), vec![0; buffer - 1000]);
+        assert_eq!(usize::from(tcb.get_scaled_recv_window()), 1000 >> 5);
+    }
+
+    // A `Tcb` opened by a SYN advertising `peer_window` under `peer_window_shift`.
+    fn window_tcb(peer_window: u16, peer_window_shift: Option<u8>, read_buffer_size: usize) -> Tcb {
+        Tcb::new(
+            SeqNum(1000),
+            peer_window,
+            peer_window_shift,
+            1500,
+            MAX_UNACK,
+            read_buffer_size,
+            MAX_COUNT_FOR_DUP_ACK,
+            RTO,
+            MAX_RETRANSMIT_COUNT,
+        )
+    }
 }
diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index d766f16..3e1c41d 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -185,8 +185,20 @@ impl IpStackTcpStream {
         destroy_messenger: Option<::tokio::sync::oneshot::Sender<()>>,
         config: Arc<TcpConfig>,
     ) -> Result<IpStackTcpStream, IpStackError> {
+        // RFC 7323 §2.2: a window scale option is read only from a segment carrying SYN.
+        let peer_window_shift = tcp
+            .syn
+            .then(|| {
+                tcp.options_iterator().flatten().find_map(|option| match option {
+                    TcpOptionElement::WindowScale(shift) => Some(shift),
+                    _ => None,
+                })
+            })
+            .flatten();
         let tcb = Tcb::new(
             SeqNum(tcp.sequence_number),
+            tcp.window_size,
+            peer_window_shift,
             mtu,
             config.max_unacked_bytes,
             config.read_buffer_size,
@@ -677,7 +689,6 @@ async fn tcp_main_logic_loop(
         let flags = tcp_header_flags(tcp_header);
         let incoming_ack: SeqNum = tcp_header.acknowledgment_number.into();
         let incoming_seq: SeqNum = tcp_header.sequence_number.into();
-        let incoming_win = tcp_header.window_size;
 
         let mut tcb = tcb.lock().unwrap();
 
@@ -727,6 +738,7 @@ async fn tcp_main_logic_loop(
                     extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
                 }
                 tcb.change_state(TcpState::Established);
+                write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
             }
             TcpState::Established => {
                 if flags == ACK {
@@ -813,6 +825,7 @@ async fn tcp_main_logic_loop(
                         tcb.add_unordered_packet(incoming_seq, payload);
                         extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
                     }
+                    write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
                 } else {
                     // unnormal case, we do nothing here
                     log::trace!("{network_tuple} {state:?}: {l_info}, {pkt_type:?}, unnormal case, we do nothing here");
@@ -920,7 +933,7 @@ async fn tcp_main_logic_loop(
         } // end of match state
 
         tcb.update_last_received_ack(incoming_ack);
-        tcb.update_send_window(incoming_win);
+        tcb.update_send_window(tcp_header);
     } // end of loop
     Ok::<(), std::io::Error>(())
 }
@@ -978,9 +991,29 @@ pub(crate) fn write_packet_to_device(
     // Silly-window-syndrome avoidance: advertise a real window only when a full segment fits,
     // otherwise advertise zero so the peer enters persist mode until the reader frees space.
     let recv_window = tcb.get_recv_window();
-    let window_size = if recv_window >= tcb.get_mtu() { recv_window } else { 0 };
+    // RFC 7323 §2.2: the window field of a segment carrying SYN is not scaled.
+    let advertised = if flags & SYN != 0 {
+        recv_window
+    } else {
+        tcb.get_scaled_recv_window()
+    };
+    let window_size = if recv_window >= tcb.get_mtu() { advertised } else { 0 };
     let ack = tcb.get_ack().0;
     let (src, dst) = (tuple.dst, tuple.src); // Note: The address is reversed here
+    let mut tcp_options = Vec::new();
+    // Note: Instead of an `if let Some ... ` a loop is used as the enum is #[non_exhaustive]
+    for option in options.into_iter().flatten() {
+        match option {
+            TcpOptions::MaximumSegmentSize(mss) => tcp_options.push(TcpOptionElement::MaximumSegmentSize(*mss)),
+        }
+    }
+    // RFC 7323 §2.2: the window scale option travels only on a segment carrying SYN, and only when
+    // the peer's SYN carried one.
+    if flags & SYN != 0
+        && let Some(shift) = tcb.get_recv_window_shift()
+    {
+        tcp_options.push(TcpOptionElement::WindowScale(shift));
+    }
     let calc = |ip_header_len: usize, tcp_header_len: usize| tcb.calculate_payload_max_len(ip_header_len, tcp_header_len);
     let packet = create_raw_packet(
         src,
@@ -992,7 +1025,7 @@ pub(crate) fn write_packet_to_device(
         ack,
         window_size,
         payload.unwrap_or_default(),
-        options,
+        &tcp_options,
     )?;
     let len = packet.payload.as_ref().map(|p| p.len()).unwrap_or(0);
     up_packet_sender.send(packet).map_err(|e| Error::new(UnexpectedEof, e))?;
@@ -1010,7 +1043,7 @@ pub(crate) fn create_raw_packet(
     ack: u32,
     win: u16,
     mut payload: Vec<u8>,
-    options: Option<&Vec<TcpOptions>>,
+    options: &[TcpOptionElement],
 ) -> std::io::Result<NetworkPacket> {
     let mut tcp_header = etherparse::TcpHeader::new(src_addr.port(), dst_addr.port(), seq, win);
     tcp_header.acknowledgment_number = ack;
@@ -1020,17 +1053,8 @@ pub(crate) fn create_raw_packet(
     tcp_header.fin = flags & FIN != 0;
     tcp_header.psh = flags & PSH != 0;
 
-    if let Some(opts) = options {
-        let mut tcp_options = Vec::new();
-        for opt in opts {
-            match opt {
-                TcpOptions::MaximumSegmentSize(mss) => tcp_options.push(TcpOptionElement::MaximumSegmentSize(*mss)),
-            }
-        }
-        tcp_header
-            .set_options(&tcp_options)
-            .map_err(|e| std::io::Error::new(InvalidInput, e))?;
-    }
+    tcp_header.set_options(options).map_err(|e| std::io::Error::new(InvalidInput, e))?;
+
     let ip_header = match (src_addr.ip(), dst_addr.ip()) {
         (std::net::IpAddr::V4(src), std::net::IpAddr::V4(dst)) => {
             let mut ip_h =
@@ -1086,6 +1110,18 @@ mod tests {
     use super::*;
     use crate::stream::tcb::{MAX_COUNT_FOR_DUP_ACK, MAX_RETRANSMIT_COUNT, MAX_UNACK, READ_BUFFER_SIZE, RTO};
 
+    /// A waker that records whether it was woken, and the flag it sets.
+    fn recording_waker() -> (Arc<std::sync::atomic::AtomicBool>, Waker) {
+        struct Recording(Arc<std::sync::atomic::AtomicBool>);
+        impl std::task::Wake for Recording {
+            fn wake(self: Arc<Self>) {
+                self.0.store(true, std::sync::atomic::Ordering::SeqCst);
+            }
+        }
+        let woken = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        (woken.clone(), Waker::from(Arc::new(Recording(woken))))
+    }
+
     #[tokio::test]
     async fn extract_reserves_before_consuming() {
         let (up_tx, _up_rx) = tokio::sync::mpsc::unbounded_channel::<NetworkPacket>();
@@ -1095,6 +1131,8 @@ mod tests {
 
         let mut tcb = Tcb::new(
             SeqNum(1000),
+            u16::MAX,
+            None,
             1500,
             MAX_UNACK,
             READ_BUFFER_SIZE,
@@ -1123,4 +1161,113 @@ mod tests {
         assert_eq!(tcb.get_ack(), SeqNum(2500));
         assert_eq!(tcb.get_unordered_packets_total_len(), 0);
     }
+
+    /// Opens a connection with a SYN advertising `window` and `syn_options`, returning the stream
+    /// and the SYN-ACK the stack emitted.
+    async fn open(window: u16, syn_options: &[TcpOptionElement]) -> (IpStackTcpStream, TcpHeader, PacketReceiver) {
+        let (up_tx, mut up_rx) = tokio::sync::mpsc::unbounded_channel::<NetworkPacket>();
+        let (src, dst) = (SocketAddr::from(([10, 0, 0, 2], 40000)), SocketAddr::from(([10, 0, 0, 1], 80)));
+
+        let mut syn = TcpHeader::new(src.port(), dst.port(), 1000, window);
+        syn.syn = true;
+        syn.set_options(syn_options).unwrap();
+
+        let stream = IpStackTcpStream::new(src, dst, syn, up_tx, 1500, None, Arc::new(TcpConfig::default())).unwrap();
+        let packet = up_rx.recv().await.expect("the stack emitted no SYN-ACK");
+        let TransportHeader::Tcp(syn_ack) = packet.transport_header() else {
+            panic!("the emitted SYN-ACK is not a TCP packet");
+        };
+        (stream, syn_ack.clone(), up_rx)
+    }
+
+    /// Feeds `stream` a segment advertising `window`, carrying `payload` at `seq`.
+    fn feed(stream: &IpStackTcpStream, syn_ack: &TcpHeader, seq: SeqNum, window: u16, payload: &[u8]) {
+        let builder = etherparse::PacketBuilder::ipv4([10, 0, 0, 2], [10, 0, 0, 1], TTL)
+            .tcp(40000, 80, seq.0, window)
+            .ack((SeqNum(syn_ack.sequence_number) + 1).0)
+            .psh();
+        let mut buf = Vec::new();
+        builder.write(&mut buf, payload).unwrap();
+        stream.stream_sender().send(NetworkPacket::parse(&buf).unwrap()).unwrap();
+    }
+
+    /// Returns the shift count of the window scale option in `header`, if it carries one.
+    fn window_scale_of(header: &TcpHeader) -> Option<u8> {
+        header.options_iterator().flatten().find_map(|option| match option {
+            TcpOptionElement::WindowScale(shift) => Some(shift),
+            _ => None,
+        })
+    }
+
+    /// A peer offering a shift of 7 gets this stack's shift of 0 back and has its windows honoured
+    /// shifted by 7.
+    #[tokio::test]
+    async fn syn_with_window_scale_is_answered_and_honoured() {
+        let (stream, syn_ack, mut up_rx) = open(4000, &[TcpOptionElement::WindowScale(7)]).await;
+        assert_eq!(
+            window_scale_of(&syn_ack),
+            Some(0),
+            "the SYN-ACK does not carry this stack's window scale"
+        );
+
+        let peer_seq = SeqNum(syn_ack.acknowledgment_number);
+        feed(&stream, &syn_ack, peer_seq, 40_000, &[1; 4]);
+        feed(&stream, &syn_ack, peer_seq + 4, 40_000, &[2; 4]);
+        up_rx.recv().await.unwrap();
+        up_rx.recv().await.unwrap();
+
+        assert_eq!(stream.tcb.lock().unwrap().get_send_window(), 40_000 << 7);
+    }
+
+    /// A peer offering a shift above the RFC 7323 §2.3 maximum has its windows honoured shifted by 14.
+    #[tokio::test]
+    async fn syn_with_a_malformed_window_scale_is_clamped() {
+        let (stream, syn_ack, mut up_rx) = open(4000, &[TcpOptionElement::WindowScale(20)]).await;
+
+        let peer_seq = SeqNum(syn_ack.acknowledgment_number);
+        feed(&stream, &syn_ack, peer_seq, 40_000, &[1; 4]);
+        feed(&stream, &syn_ack, peer_seq + 4, 40_000, &[2; 4]);
+        up_rx.recv().await.unwrap();
+        up_rx.recv().await.unwrap();
+
+        let tcb = stream.tcb.lock().unwrap();
+        assert_eq!(tcb.get_state(), TcpState::Established);
+        assert_eq!(tcb.get_send_window(), 40_000 << 14);
+    }
+
+    /// A closed peer window holds the writer, and the segment that reopens it wakes the writer, both
+    /// when it completes the handshake and when it arrives on an established connection.
+    #[tokio::test]
+    async fn zero_peer_window_holds_the_writer_until_it_reopens() {
+        let (mut stream, syn_ack, mut up_rx) = open(0, &[]).await;
+        let peer_seq = SeqNum(syn_ack.acknowledgment_number);
+
+        // the segment completing the handshake reopens the window
+        let (woken, waker) = recording_waker();
+        let mut cx = Context::from_waker(&waker);
+        assert!(Pin::new(&mut stream).poll_write(&mut cx, b"held").is_pending());
+        feed(&stream, &syn_ack, peer_seq, 40_000, &[1; 4]);
+        up_rx.recv().await.unwrap();
+        assert!(
+            woken.load(std::sync::atomic::Ordering::SeqCst),
+            "the handshake reopened the window without waking the writer"
+        );
+        assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, b"held"), Poll::Ready(Ok(4))));
+        assert_eq!(up_rx.recv().await.unwrap().payload.as_deref(), Some(&b"held"[..]));
+
+        // a data segment closes the window and a later one reopens it
+        feed(&stream, &syn_ack, peer_seq + 4, 0, &[2; 4]);
+        up_rx.recv().await.unwrap();
+        let (woken, waker) = recording_waker();
+        let mut cx = Context::from_waker(&waker);
+        assert!(Pin::new(&mut stream).poll_write(&mut cx, b"sent").is_pending());
+        feed(&stream, &syn_ack, peer_seq + 8, 40_000, &[3; 4]);
+        up_rx.recv().await.unwrap();
+        assert!(
+            woken.load(std::sync::atomic::Ordering::SeqCst),
+            "a data segment reopened the window without waking the writer"
+        );
+        assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, b"sent"), Poll::Ready(Ok(4))));
+        assert_eq!(up_rx.recv().await.unwrap().payload.as_deref(), Some(&b"sent"[..]));
+    }
 }

From 6489fcfb66036e3619f1f1d2c7b96b8cbe5f5930 Mon Sep 17 00:00:00 2001
From: Sajjad Pourali <sajjad@pourali.com>
Date: Tue, 15 Sep 2026 20:35:36 -0700
Subject: [PATCH 05/13] Handle device EOF and I/O errors in the main event loop
 #88 (#93)

---
 src/lib.rs | 35 ++++++++++++++++++++++++++++++++++-
 1 file changed, 34 insertions(+), 1 deletion(-)

diff --git a/src/lib.rs b/src/lib.rs
index e67072a..45b1018 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -319,7 +319,18 @@ fn run<Device: AsyncRead + AsyncWrite + Unpin + Send + 'static>(
     tokio::spawn(async move {
         loop {
             select! {
-                Ok(n) = device.read(&mut buffer) => {
+                read_result = device.read(&mut buffer) => {
+                    let n = match read_result {
+                        Ok(0) => {
+                            log::info!("Device EOF, stopping IP stack");
+                            return Ok(());
+                        }
+                        Ok(n) => n,
+                        Err(e) => {
+                            log::error!("Device read error: {e}");
+                            return Err(e.into());
+                        }
+                    };
                     if let Err(e) = process_device_read(&buffer[offset..n], &mut sessions, &session_remove_tx, &up_pkt_sender, &config, &accept_sender).await {
                         let io_err: std::io::Error = e.into();
                         if io_err.kind() == std::io::ErrorKind::ConnectionRefused {
@@ -439,3 +450,25 @@ async fn process_upstream_recv<Device: AsyncWrite + Unpin + 'static>(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use tokio::{io::duplex, time::timeout};
+
+    use super::*;
+
+    #[tokio::test]
+    async fn device_eof_closes_accept_channel() {
+        let (device, peer) = duplex(64);
+        let mut ip_stack = IpStack::new(IpStackConfig::default(), device);
+
+        drop(peer);
+
+        let accept_result = timeout(Duration::from_secs(1), ip_stack.accept())
+            .await
+            .expect("accept should not hang after device EOF");
+        assert!(matches!(accept_result, Err(IpStackError::AcceptError)));
+    }
+}

From 7cf41a682433d134c62a73e5c4c080d883ac040c Mon Sep 17 00:00:00 2001
From: IntellyCode <47359527+IntellyCode@users.noreply.github.com>
Date: Thu, 17 Sep 2026 17:24:49 +0200
Subject: [PATCH 06/13] Retransmit on a timer and close the connection at R2
 (#92)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* Retransmit on a timer and close the connection at R2

The session loop waits on the earliest deadline its inflight queue holds
and sweeps when it passes, so a lost segment with nothing behind it is
retransmitted on time rather than when the peer's own timeout recovers it.
A write that reopens an empty queue wakes the loop, since the deadline is
read on entry and a later segment always sits behind one already waited
on. The arrival path asks nothing about timeouts.

The timeout belongs to the connection. It doubles on every expiry up to a
minute, and an acknowledgement retiring a segment sent exactly once puts
it back to the configured value: only that retirement says which
transmission the acknowledgement answers.

A segment stays in the inflight queue until it is acknowledged. Reaching
R2 transmissions of the same segment closes the connection, per RFC 9293
§3.8.3, which at seven transmissions is 123 seconds against the at least
100 that SHLD-11 asks for.

* Restart the retransmission timer for every segment and keep the ceiling above the configured timeout

An expiry restarts the timer for every outstanding segment, so no segment
is retransmitted before the backed-off timeout has run from that expiry.
The backoff is bounded by the larger of MAX_RTO and the configured
timeout, so an expiry never shortens it.

* Close at R2 without resending, stop the timer after a reset, and floor the timeout at one second

* Reuse RTO as the timeout floor and read the deadline from the queue head
---
 Cargo.toml        |   1 +
 src/stream/tcb.rs | 264 +++++++++++++++++++++++++++++++++++++++-------
 src/stream/tcp.rs | 218 +++++++++++++++++++++++++++++++++++---
 3 files changed, 427 insertions(+), 56 deletions(-)

diff --git a/Cargo.toml b/Cargo.toml
index 3f39b01..9f6d55a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -31,6 +31,7 @@ dotenvy = "0.15"
 env_logger = "0.11"
 tokio = { version = "1.52", default-features = false, features = [
     "rt-multi-thread",
+    "test-util",
 ] }
 tun = { version = "0.8", default-features = false, features = ["async"] }
 udp-stream = { version = "0.0", default-features = false }
diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs
index ec73e7c..2453667 100644
--- a/src/stream/tcb.rs
+++ b/src/stream/tcb.rs
@@ -7,11 +7,15 @@ pub(super) const READ_BUFFER_SIZE: usize = 1024 * 16; // 16KB
 pub(super) const READ_CHUNK: usize = 8192; // 8KB, bytes drained from the reassembly buffer per handoff
 pub(super) const MAX_COUNT_FOR_DUP_ACK: usize = 3; // Maximum number of duplicate ACKs before retransmission
 
-/// Retransmission timeout
+/// Retransmission timeout, and the floor RFC 6298 §2.4 rounds a configured one up to
 pub(super) const RTO: std::time::Duration = std::time::Duration::from_secs(1);
 
-/// Maximum count of retransmissions before dropping the packet
-pub(super) const MAX_RETRANSMIT_COUNT: usize = 3;
+/// Ceiling on the backed-off retransmission timeout; RFC 6298 §2.5 permits one of at least 60 seconds
+const MAX_RTO: std::time::Duration = std::time::Duration::from_secs(60);
+
+/// R2, the transmission count at which the connection closes. Seven reaches 123 seconds against
+/// RFC 9293 §3.8.3's SHLD-11 of at least 100.
+pub(super) const MAX_RETRANSMIT_COUNT: usize = 7;
 
 /// Maximum window scale shift count, which RFC 7323 §2.3 limits to 14 for a maximum window of 1 GiB
 const MAX_WINDOW_SHIFT: u8 = 14;
@@ -68,7 +72,12 @@ pub(crate) struct Tcb {
     max_unacked_bytes: u32,
     read_buffer_size: usize,
     max_count_for_dup_ack: usize,
+    /// Configured retransmission timeout, the value `current_rto` collapses back to.
     rto: std::time::Duration,
+    /// Retransmission timeout in force, doubled on every expiry per RFC 6298 §5.5 up to `MAX_RTO` or `rto`, whichever is larger.
+    /// RFC 6298 §2's round-trip estimator is absent, so this starts at the configured timeout and
+    /// returns to it rather than being recomputed from a measurement.
+    current_rto: std::time::Duration,
     max_retransmit_count: usize,
 }
 
@@ -85,6 +94,7 @@ impl Tcb {
         rto: std::time::Duration,
         max_retransmit_count: usize,
     ) -> Tcb {
+        let rto = rto.max(RTO);
         #[cfg(debug_assertions)]
         let seq = 100;
         #[cfg(not(debug_assertions))]
@@ -124,6 +134,7 @@ impl Tcb {
             read_buffer_size,
             max_count_for_dup_ack,
             rto,
+            current_rto: rto,
             max_retransmit_count,
         }
     }
@@ -317,7 +328,7 @@ impl Tcb {
             return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "Empty payload"));
         }
         let buf_len = buf.len() as u32;
-        self.inflight_packets.insert(self.seq, InflightPacket::new(self.seq, buf, self.rto));
+        self.inflight_packets.insert(self.seq, InflightPacket::new(self.seq, buf));
         self.seq += buf_len;
         Ok(())
     }
@@ -332,6 +343,9 @@ impl Tcb {
             Some((&seq, _)) if ack < seq => return,
             _ => {}
         }
+        // RFC 6298 §3: a sample from a retransmitted segment is ambiguous, so only a segment sent
+        // once is the measurement §5's note collapses the backed-off timeout on.
+        let mut measured = false;
         if let Some(seq) = self
             .inflight_packets
             .iter()
@@ -344,39 +358,73 @@ impl Tcb {
                 inflight_packet.payload.drain(0..distance);
                 inflight_packet.seq = ack;
                 self.inflight_packets.insert(ack, inflight_packet);
+            } else {
+                measured |= inflight_packet.retransmit_count == 0;
             }
         }
-        self.inflight_packets.retain(|_, p| ack < p.seq + p.payload.len() as u32);
+        self.inflight_packets.retain(|_, p| {
+            if ack < p.seq + p.payload.len() as u32 {
+                return true; // keep the packet in the inflight_packets
+            }
+            measured |= p.retransmit_count == 0;
+            false // remove this packet
+        });
+        if measured {
+            // With no estimator, the computation RFC 6298 §5's note calls for gives the configured
+            // timeout back, collapsing whatever §5.5 backed it off to.
+            self.current_rto = self.rto;
+        }
     }
 
     pub(crate) fn find_inflight_packet(&self, seq: SeqNum) -> Option<&InflightPacket> {
         self.inflight_packets.get(&seq)
     }
 
+    /// The deadline the connection waits on: the earliest send time in the inflight queue plus the
+    /// current retransmission timeout, present while data is outstanding and the sum fits the clock.
+    pub(crate) fn get_retransmission_deadline(&self) -> Option<tokio::time::Instant> {
+        self.inflight_packets
+            .values()
+            .next()
+            .and_then(|p| p.send_time.checked_add(self.current_rto))
+    }
+
     #[must_use]
-    pub(crate) fn collect_timed_out_inflight_packets(&mut self) -> Vec<InflightPacket> {
-        let mut retransmit_list = Vec::new();
+    /// The segments whose deadline has passed, and whether one of them reached R2, the transmission
+    /// count RFC 9293 §3.8.3 closes the connection at. At R2 the list is empty.
+    pub(crate) fn collect_timed_out_inflight_packets(&mut self) -> (Vec<InflightPacket>, bool) {
+        let (rto, r2) = (self.current_rto, self.max_retransmit_count);
+        if self
+            .inflight_packets
+            .values()
+            .any(|p| p.is_timed_out(rto) && p.retransmit_count + 1 >= r2)
+        {
+            return (Vec::new(), true);
+        }
 
-        self.inflight_packets.retain(|_, packet| {
-            if packet.retransmit_count >= self.max_retransmit_count {
-                log::warn!("Packet with seq {:?} reached max retransmit count, dropping packet", packet.seq);
-                return false; // remove this packet
-            }
-            if packet.is_timed_out() {
+        let mut retransmit_list = Vec::new();
+        for packet in self.inflight_packets.values_mut() {
+            if packet.is_timed_out(rto) {
                 packet.retransmit_count += 1;
-                packet.retransmit_timeout *= 2; // increase timeout exponentially
-                packet.send_time = std::time::Instant::now();
                 retransmit_list.push(packet.clone());
             }
-            true // keep the packet in the inflight_packets
-        });
-        retransmit_list
+        }
+        if !retransmit_list.is_empty() {
+            let now = tokio::time::Instant::now();
+            self.inflight_packets.values_mut().for_each(|packet| packet.send_time = now); // restart the timer, per RFC 6298 §5.6
+            self.current_rto = self.current_rto.saturating_mul(2).min(MAX_RTO.max(self.rto)); // back off the timer, per RFC 6298 §5.5
+        }
+        (retransmit_list, false)
     }
 
     pub(crate) fn get_inflight_packets_total_len(&self) -> usize {
         self.inflight_packets.values().map(|p| p.payload.len()).sum()
     }
 
+    pub(crate) fn is_inflight_queue_empty(&self) -> bool {
+        self.inflight_packets.is_empty()
+    }
+
     #[allow(dead_code)]
     pub(crate) fn get_all_inflight_packets(&self) -> Vec<&InflightPacket> {
         self.inflight_packets.values().collect::<Vec<_>>()
@@ -393,26 +441,24 @@ impl Tcb {
 pub struct InflightPacket {
     pub seq: SeqNum,
     pub payload: Vec<u8>,
-    pub send_time: std::time::Instant,
+    pub send_time: tokio::time::Instant,
     pub retransmit_count: usize,
-    pub retransmit_timeout: std::time::Duration, // current retransmission timeout
 }
 
 impl InflightPacket {
-    fn new(seq: SeqNum, payload: Vec<u8>, rto: Duration) -> Self {
+    fn new(seq: SeqNum, payload: Vec<u8>) -> Self {
         Self {
             seq,
             payload,
-            send_time: std::time::Instant::now(),
+            send_time: tokio::time::Instant::now(),
             retransmit_count: 0,
-            retransmit_timeout: rto,
         }
     }
     pub(crate) fn contains_seq_num(&self, seq: SeqNum) -> bool {
         self.seq <= seq && seq < self.seq + self.payload.len() as u32
     }
-    pub(crate) fn is_timed_out(&self) -> bool {
-        self.send_time.elapsed() >= self.retransmit_timeout
+    pub(crate) fn is_timed_out(&self, rto: Duration) -> bool {
+        self.send_time.elapsed() >= rto
     }
 }
 
@@ -422,7 +468,7 @@ mod tests {
 
     #[test]
     fn test_in_flight_packet() {
-        let p = InflightPacket::new((u32::MAX - 1).into(), vec![10, 20, 30, 40, 50], RTO);
+        let p = InflightPacket::new((u32::MAX - 1).into(), vec![10, 20, 30, 40, 50]);
 
         assert!(p.contains_seq_num((u32::MAX - 1).into()));
         assert!(p.contains_seq_num(u32::MAX.into()));
@@ -587,8 +633,137 @@ mod tests {
         assert_eq!(tcb.inflight_packets.len(), 0); // all packets should be removed
     }
 
-    #[test]
-    fn test_retransmit_with_exponential_backoff() {
+    /// RFC 6298 §5's note: the backed-off timeout collapses once a segment sent exactly once is acknowledged
+    #[tokio::test(start_paused = true)]
+    async fn test_backoff_collapses_on_an_unretransmitted_segment() {
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            u16::MAX,
+            None,
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            RTO,
+            MAX_RETRANSMIT_COUNT,
+        );
+
+        tcb.add_inflight_packet(vec![1; 500]).unwrap();
+        tokio::time::advance(RTO).await;
+        assert_eq!(tcb.collect_timed_out_inflight_packets().0.len(), 1);
+        assert_eq!(tcb.current_rto, RTO * 2);
+
+        // the retransmitted segment cannot say which transmission the acknowledgement answers
+        tcb.update_inflight_packet_queue(tcb.get_seq());
+        assert!(tcb.is_inflight_queue_empty());
+        assert_eq!(tcb.current_rto, RTO * 2);
+
+        // a segment sent exactly once, whose acknowledgement is the measurement
+        tcb.add_inflight_packet(vec![2; 500]).unwrap();
+        tcb.update_inflight_packet_queue(tcb.get_seq());
+        assert!(tcb.is_inflight_queue_empty());
+        assert_eq!(tcb.current_rto, RTO);
+    }
+
+    /// RFC 6298 §2.4: a configured timeout below one second is raised to one second.
+    #[tokio::test(start_paused = true)]
+    async fn test_timeout_below_one_second_is_raised() {
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            u16::MAX,
+            None,
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            Duration::from_millis(100),
+            MAX_RETRANSMIT_COUNT,
+        );
+
+        tcb.add_inflight_packet(vec![1; 500]).unwrap();
+        assert_eq!(
+            tcb.get_retransmission_deadline(),
+            Some(tokio::time::Instant::now() + Duration::from_secs(1))
+        );
+    }
+
+    /// A configured timeout beyond the clock's range gives a deadline of `None`.
+    #[tokio::test(start_paused = true)]
+    async fn test_timeout_too_large_for_the_clock_has_no_deadline() {
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            u16::MAX,
+            None,
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            Duration::MAX,
+            MAX_RETRANSMIT_COUNT,
+        );
+
+        tcb.add_inflight_packet(vec![1; 500]).unwrap();
+        assert!(tcb.get_retransmission_deadline().is_none());
+    }
+
+    /// RFC 6298 §5.6: an expiry restarts the timer for every outstanding segment, so none is
+    /// retransmitted before the backed-off timeout has run from that expiry.
+    #[tokio::test(start_paused = true)]
+    async fn test_expiry_restarts_the_timer_for_every_segment() {
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            u16::MAX,
+            None,
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            RTO,
+            MAX_RETRANSMIT_COUNT,
+        );
+
+        tcb.add_inflight_packet(vec![1; 500]).unwrap();
+        tokio::time::advance(Duration::from_millis(900)).await;
+        tcb.add_inflight_packet(vec![2; 500]).unwrap();
+
+        // the first segment expires, doubling the timeout to two seconds
+        tokio::time::advance(Duration::from_millis(100)).await;
+        assert_eq!(tcb.collect_timed_out_inflight_packets().0.len(), 1);
+
+        // nothing expires until two seconds after that expiry, and then both segments do
+        tokio::time::advance(Duration::from_millis(1900)).await;
+        assert!(tcb.collect_timed_out_inflight_packets().0.is_empty());
+        tokio::time::advance(Duration::from_millis(100)).await;
+        assert_eq!(tcb.collect_timed_out_inflight_packets().0.len(), 2);
+    }
+
+    /// RFC 6298 §5.5: an expiry doubles the timeout, bounded above by the ceiling, so a configured
+    /// timeout larger than `MAX_RTO` is never shortened.
+    #[tokio::test(start_paused = true)]
+    async fn test_backoff_never_shortens_a_large_configured_timeout() {
+        let rto = Duration::from_secs(120);
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            u16::MAX,
+            None,
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            rto,
+            MAX_RETRANSMIT_COUNT,
+        );
+
+        tcb.add_inflight_packet(vec![1; 500]).unwrap();
+        tokio::time::advance(rto).await;
+        assert_eq!(tcb.collect_timed_out_inflight_packets().0.len(), 1);
+        assert!(tcb.current_rto >= rto, "the backoff shortened the timeout to {:?}", tcb.current_rto);
+    }
+
+    /// RFC 6298 §5.5: every expiry backs the connection's timer off, so the segment is retransmitted
+    /// after a longer wait each time, and it stays outstanding once R2 reports the connection closed.
+    #[tokio::test(start_paused = true)]
+    async fn test_retransmit_with_exponential_backoff() {
         let mut tcb = Tcb::new(
             SeqNum(1000),
             u16::MAX,
@@ -604,22 +779,29 @@ mod tests {
         tcb.add_inflight_packet(vec![1; 500]).unwrap();
 
         // Simulate retransmission timeouts
+        let mut previous_wait = Duration::ZERO;
         for i in 0..MAX_RETRANSMIT_COUNT {
-            // Simulate a timeout for the first packet
-            let timeout = tcb.inflight_packets.values().next().unwrap().retransmit_timeout + std::time::Duration::from_millis(100);
-            println!("timeout: {timeout:?}");
-            std::thread::sleep(timeout);
-
-            let packets = tcb.collect_timed_out_inflight_packets();
-            assert_eq!(packets.len(), 1);
-            let packet = &packets[0];
-            assert_eq!(packet.retransmit_count, i + 1);
-            assert!(packet.retransmit_timeout > RTO);
+            // Wait exactly as long as the connection asks, which is longer on every round
+            let wait = tcb.get_retransmission_deadline().unwrap() - tokio::time::Instant::now();
+            println!("timeout: {wait:?}");
+            assert!(wait >= previous_wait); // doubling, until the ceiling flattens it
+            previous_wait = wait;
+            tokio::time::advance(wait).await;
+
+            let (packets, exhausted) = tcb.collect_timed_out_inflight_packets();
+            if i + 1 < MAX_RETRANSMIT_COUNT {
+                assert_eq!(packets.len(), 1);
+                assert_eq!(packets[0].retransmit_count, i + 1);
+                assert!(!exhausted);
+            } else {
+                assert!(packets.is_empty());
+                assert!(exhausted);
+            }
+            assert!(tcb.current_rto > RTO);
         }
 
-        let packets = tcb.collect_timed_out_inflight_packets();
-        assert!(packets.is_empty());
-        assert!(tcb.inflight_packets.is_empty());
+        // the segment stays outstanding at R2; the connection closes rather than the queue losing it
+        assert!(!tcb.is_inflight_queue_empty());
     }
 
     /// A peer shift of 7 is applied to every window the peer advertises after the handshake.
diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index 3e1c41d..a1e0cac 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -51,7 +51,7 @@ pub struct TcpConfig {
     pub max_count_for_dup_ack: usize,
     /// Retransmission timeout duration.
     pub rto: std::time::Duration,
-    /// Maximum number of retransmissions before giving up.
+    /// R2, the transmission count of one segment at which the connection closes.
     pub max_retransmit_count: usize,
     /// TCP options
     pub options: Option<Vec<TcpOptions>>,
@@ -169,6 +169,7 @@ pub struct IpStackTcpStream {
     data_rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
     read_notify: std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
     drain_notify: Arc<tokio::sync::Notify>,
+    arm_notify: Arc<tokio::sync::Notify>,
     task_handle: Option<tokio::task::JoinHandle<std::io::Result<()>>>,
     exit_notifier: Option<tokio::sync::mpsc::Sender<()>>,
     temp_read_buffer: Vec<u8>,
@@ -237,6 +238,7 @@ impl IpStackTcpStream {
             data_rx,
             read_notify: std::sync::Arc::new(std::sync::Mutex::new(None)),
             drain_notify: Arc::new(tokio::sync::Notify::new()),
+            arm_notify: Arc::new(tokio::sync::Notify::new()),
             task_handle: None,
             exit_notifier: None,
             temp_read_buffer: Vec::new(),
@@ -383,7 +385,13 @@ impl AsyncWrite for IpStackTcpStream {
 
         let sender = &self.up_packet_sender;
         let payload_len = write_packet_to_device(sender, nt, &tcb, None, ACK | PSH, None, Some(buf.to_vec()))?;
+        let was_empty = tcb.is_inflight_queue_empty();
         tcb.add_inflight_packet(buf[..payload_len].to_vec())?;
+        if was_empty {
+            // The loop reads the deadline when it enters `select!`, so the segment that reopens an
+            // empty queue has to wake it; a later one sits behind a deadline it already waits on.
+            self.arm_notify.notify_one();
+        }
 
         let (state, seq, ack) = (tcb.get_state(), tcb.get_seq(), tcb.get_ack());
         let l_info = format!("local {{ seq: {seq}, ack: {ack} }}");
@@ -489,6 +497,7 @@ impl IpStackTcpStream {
         let read_notify = self.read_notify.clone();
         let data_tx = self.data_tx.clone();
         let drain_notify = self.drain_notify.clone();
+        let arm_notify = self.arm_notify.clone();
         let destroy_messenger = self.destroy_messenger.take();
 
         let (exit_task_notifier, exit_monitor) = tokio::sync::mpsc::channel::<()>(10);
@@ -508,6 +517,7 @@ impl IpStackTcpStream {
                 read_notify,
                 data_tx,
                 drain_notify,
+                arm_notify,
                 exit_monitor,
             )
             .await;
@@ -537,6 +547,7 @@ async fn tcp_main_logic_loop(
     read_notify: std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
     data_tx: tokio::sync::mpsc::Sender<Vec<u8>>,
     drain_notify: Arc<tokio::sync::Notify>,
+    arm_notify: Arc<tokio::sync::Notify>,
     mut exit_monitor: tokio::sync::mpsc::Receiver<()>,
 ) -> std::io::Result<()> {
     {
@@ -669,6 +680,39 @@ async fn tcp_main_logic_loop(
                 extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
                 continue;
             }
+            _ = wait_retransmission_deadline(&tcb, &arm_notify) => {
+                // Either a deadline passed or the queue just reopened, so ask once who is overdue.
+                let mut tcb = tcb.lock().unwrap();
+                let state = tcb.get_state();
+                if state == TcpState::Closed {
+                    log::debug!("{network_tuple} {state:?}: session finished, exiting task...");
+                    break;
+                }
+                let (packets, exhausted) = tcb.collect_timed_out_inflight_packets();
+                if exhausted {
+                    // RFC 9293 §3.8.3: R2 transmissions of the same segment close the connection.
+                    log::warn!("{network_tuple} segment reached R2 transmissions, closing session");
+                    write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK | RST, None, None)?;
+                    tcb.change_state(TcpState::Closed);
+                    write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
+                    read_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
+                    break;
+                }
+                for packet in packets {
+                    let (seq, count) = (packet.seq, packet.retransmit_count);
+                    log::debug!("{network_tuple} inflight packet retransmission timeout: {seq:?}, retransmit_count: {count}",);
+                    write_packet_to_device(
+                        &up_packet_sender,
+                        network_tuple,
+                        &tcb,
+                        None,
+                        ACK | PSH,
+                        Some(seq),
+                        Some(packet.payload),
+                    )?;
+                }
+                continue;
+            }
             network_packet = stream_receiver.recv() => network_packet,
         };
 
@@ -707,20 +751,6 @@ async fn tcp_main_logic_loop(
 
         tcb.update_inflight_packet_queue(incoming_ack);
 
-        for packet in tcb.collect_timed_out_inflight_packets() {
-            let (seq, count) = (packet.seq, packet.retransmit_count);
-            log::debug!("{network_tuple} inflight packet retransmission timeout: {seq:?}, retransmit_count: {count}",);
-            write_packet_to_device(
-                &up_packet_sender,
-                network_tuple,
-                &tcb,
-                None,
-                ACK | PSH,
-                Some(seq),
-                Some(packet.payload),
-            )?;
-        }
-
         let pkt_type = tcb.check_pkt_type(tcp_header, &payload);
 
         let (state, seq, ack) = { (tcb.get_state(), tcb.get_seq(), tcb.get_ack()) };
@@ -938,6 +968,18 @@ async fn tcp_main_logic_loop(
     Ok::<(), std::io::Error>(())
 }
 
+/// Wait for the connection's retransmission deadline, which RFC 6298 §5.6 sets after every
+/// backoff and §5.4 acts on by retransmitting the earliest unacknowledged segment. A connection with nothing
+/// outstanding has no deadline, so it waits for `arm_notify` to say a segment reopened the inflight
+/// queue; the caller re-enters `select!` afterwards and reads the deadline that segment created.
+async fn wait_retransmission_deadline(tcb: &TcbPtr, arm_notify: &tokio::sync::Notify) {
+    let deadline = { tcb.lock().unwrap().get_retransmission_deadline() };
+    match deadline {
+        Some(deadline) => tokio::time::sleep_until(deadline).await,
+        None => arm_notify.notified().await,
+    }
+}
+
 fn extract_data_n_write_upstream(
     up_packet_sender: &PacketSender,
     tcb: &mut Tcb,
@@ -1122,6 +1164,152 @@ mod tests {
         (woken.clone(), Waker::from(Arc::new(Recording(woken))))
     }
 
+    use tokio::io::AsyncWriteExt;
+
+    const PEER: &str = "10.0.0.2:12345";
+    const LOCAL: &str = "10.0.0.1:80";
+
+    /// Takes the TCP header of a packet the stack wrote to the device.
+    fn tcp_header_of(packet: &NetworkPacket) -> &TcpHeader {
+        match packet.transport_header() {
+            TransportHeader::Tcp(header) => header,
+            _ => panic!("the stack wrote a packet that is not TCP"),
+        }
+    }
+
+    /// Opens a connection and completes its handshake, returning the stream, the device receiver,
+    /// and the sequence number the stack sends its first data byte at.
+    async fn established() -> (IpStackTcpStream, PacketReceiver, u32) {
+        let (peer, local) = (PEER.parse().unwrap(), LOCAL.parse().unwrap());
+        let (up_tx, mut up_rx) = tokio::sync::mpsc::unbounded_channel::<NetworkPacket>();
+
+        let mut syn = TcpHeader::new(12345, 80, 1000, u16::MAX);
+        syn.syn = true;
+        let stream = IpStackTcpStream::new(peer, local, syn, up_tx, 1500, None, Arc::new(TcpConfig::default())).unwrap();
+
+        let syn_ack = up_rx.recv().await.unwrap();
+        let local_seq = tcp_header_of(&syn_ack).sequence_number + 1;
+        stream.stream_sender().send(peer_ack(1001, local_seq)).unwrap();
+        tokio::task::yield_now().await;
+        assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::Established);
+
+        (stream, up_rx, local_seq)
+    }
+
+    /// An acknowledgement from the peer, carrying no data.
+    fn peer_ack(seq: u32, ack: u32) -> NetworkPacket {
+        let (src, dst) = (PEER.parse().unwrap(), LOCAL.parse().unwrap());
+        create_raw_packet(src, dst, |_, _| usize::MAX, ACK, TTL, seq, ack, u16::MAX, Vec::new(), &[]).unwrap()
+    }
+
+    /// The intervals RFC 6298 §5.5 produces from a one-second timeout doubling to the §2.5 ceiling.
+    const BACKOFF: [u64; MAX_RETRANSMIT_COUNT] = [1, 2, 4, 8, 16, 32, 60];
+
+    /// Advances the clock by `by` and returns what the stack wrote to the device.
+    async fn advance(by: Duration, up_rx: &mut PacketReceiver) -> Option<NetworkPacket> {
+        tokio::time::advance(by).await;
+        tokio::task::yield_now().await;
+        up_rx.try_recv().ok()
+    }
+
+    /// RFC 6298 §5.4: a segment with nothing behind it is retransmitted once its deadline passes,
+    /// with no packet arriving to prompt it.
+    #[tokio::test(start_paused = true)]
+    async fn lost_segment_is_retransmitted_without_an_arrival() {
+        let (mut stream, mut up_rx, local_seq) = established().await;
+
+        stream.write_all(b"hello").await.unwrap();
+        let sent = up_rx.recv().await.unwrap();
+        assert_eq!(tcp_header_of(&sent).sequence_number, local_seq);
+
+        let again = advance(Duration::from_secs(BACKOFF[0]), &mut up_rx)
+            .await
+            .expect("no retransmission");
+        assert_eq!(tcp_header_of(&again).sequence_number, local_seq);
+        assert_eq!(again.payload.as_deref(), Some(&b"hello"[..]));
+    }
+
+    /// RFC 9293 §3.8.3: the connection closes when a segment's transmissions reach R2, one timeout
+    /// after the last retransmission and past the 100 seconds SHLD-11 asks for.
+    #[tokio::test(start_paused = true)]
+    async fn unacknowledged_segment_closes_the_connection_at_r2() {
+        let (mut stream, mut up_rx, _) = established().await;
+        let opened = tokio::time::Instant::now();
+        let just_before = |seconds: u64| Duration::from_secs(seconds) - Duration::from_millis(1);
+
+        stream.write_all(b"hello").await.unwrap();
+        up_rx.recv().await.unwrap();
+
+        for seconds in BACKOFF.iter().take(MAX_RETRANSMIT_COUNT - 1) {
+            assert!(advance(just_before(*seconds), &mut up_rx).await.is_none());
+            let packet = advance(Duration::from_millis(1), &mut up_rx).await.expect("no retransmission");
+            assert!(!tcp_header_of(&packet).rst);
+        }
+
+        assert!(advance(just_before(BACKOFF[MAX_RETRANSMIT_COUNT - 1]), &mut up_rx).await.is_none());
+        let reset = advance(Duration::from_millis(1), &mut up_rx).await.expect("no reset");
+        assert!(tcp_header_of(&reset).rst);
+        assert!(up_rx.try_recv().is_err());
+
+        assert!((tokio::time::Instant::now() - opened).as_secs() > 100);
+        assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::Closed);
+    }
+
+    /// RFC 9293 §3.10.7.4: a reset closes the connection and ends its retransmission.
+    #[tokio::test(start_paused = true)]
+    async fn a_reset_stops_retransmission() {
+        let (mut stream, mut up_rx, local_seq) = established().await;
+
+        stream.write_all(b"hello").await.unwrap();
+        up_rx.recv().await.unwrap();
+
+        let (src, dst) = (PEER.parse().unwrap(), LOCAL.parse().unwrap());
+        let reset = create_raw_packet(src, dst, |_, _| usize::MAX, RST, TTL, 1001, local_seq, u16::MAX, Vec::new(), &[]).unwrap();
+        stream.stream_sender().send(reset).unwrap();
+        tokio::task::yield_now().await;
+
+        assert!(advance(Duration::from_secs(BACKOFF[0]), &mut up_rx).await.is_none());
+    }
+
+    /// RFC 6298 §5.4 retransmits the earliest unacknowledged segment, so every attempt carries the
+    /// sequence number and payload of the first.
+    #[tokio::test(start_paused = true)]
+    async fn retransmissions_carry_the_original_segment() {
+        let (mut stream, mut up_rx, local_seq) = established().await;
+
+        stream.write_all(b"hello").await.unwrap();
+        up_rx.recv().await.unwrap();
+
+        for seconds in BACKOFF.iter().take(4) {
+            let packet = advance(Duration::from_secs(*seconds), &mut up_rx).await.expect("no retransmission");
+            assert_eq!(tcp_header_of(&packet).sequence_number, local_seq);
+            assert_eq!(packet.payload.as_deref(), Some(&b"hello"[..]));
+        }
+    }
+
+    /// An acknowledgement retires the segment, so no deadline remains, the connection stays open,
+    /// and a further segment flows.
+    #[tokio::test(start_paused = true)]
+    async fn an_acknowledgement_stops_retransmission() {
+        let (mut stream, mut up_rx, local_seq) = established().await;
+
+        stream.write_all(b"hello").await.unwrap();
+        up_rx.recv().await.unwrap();
+
+        for seconds in BACKOFF.iter().take(5) {
+            advance(Duration::from_secs(*seconds), &mut up_rx).await.expect("no retransmission");
+        }
+
+        stream.stream_sender().send(peer_ack(1001, local_seq + 5)).unwrap();
+        tokio::task::yield_now().await;
+        assert!(stream.tcb.lock().unwrap().get_retransmission_deadline().is_none());
+        assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::Established);
+
+        stream.write_all(b"more").await.unwrap();
+        let sent = up_rx.recv().await.unwrap();
+        assert_eq!(sent.payload.as_deref(), Some(&b"more"[..]));
+    }
+
     #[tokio::test]
     async fn extract_reserves_before_consuming() {
         let (up_tx, _up_rx) = tokio::sync::mpsc::unbounded_channel::<NetworkPacket>();

From 15751c253e73d7c047b88cdebb1d5f87e01872ee Mon Sep 17 00:00:00 2001
From: IntellyCode <47359527+IntellyCode@users.noreply.github.com>
Date: Wed, 23 Sep 2026 04:31:14 +0200
Subject: [PATCH 07/13] Acknowledge received data as it arrives (#95)

In-sequence data is acknowledged on arrival and held in a received
queue, which the reader drains through the bounded handoff channel. The
advertised window counts both the reassembly buffer and the queue, and a
segment is stored only up to the window's right edge. A FIN arriving in
sequence is acknowledged where it lands. A closing connection hands the
reader every stored byte before end-of-stream.
---
 src/stream/tcb.rs | 51 ++++++++++++++++++++++++++++++++---------------
 src/stream/tcp.rs | 28 ++++++++++++++++++--------
 2 files changed, 55 insertions(+), 24 deletions(-)

diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs
index 2453667..9d322a2 100644
--- a/src/stream/tcb.rs
+++ b/src/stream/tcb.rs
@@ -1,10 +1,13 @@
 use super::seqnum::SeqNum;
 use etherparse::TcpHeader;
-use std::{collections::BTreeMap, time::Duration};
+use std::{
+    collections::{BTreeMap, VecDeque},
+    time::Duration,
+};
 
 pub(super) const MAX_UNACK: u32 = 1024 * 16; // 16KB
 pub(super) const READ_BUFFER_SIZE: usize = 1024 * 16; // 16KB
-pub(super) const READ_CHUNK: usize = 8192; // 8KB, bytes drained from the reassembly buffer per handoff
+pub(super) const READ_CHUNK: usize = 8192; // 8KB, bytes drained from the received queue per handoff
 pub(super) const MAX_COUNT_FOR_DUP_ACK: usize = 3; // Maximum number of duplicate ACKs before retransmission
 
 /// Retransmission timeout, and the floor RFC 6298 §2.4 rounds a configured one up to
@@ -49,8 +52,8 @@ pub(super) enum PacketType {
 /// - `inflight_packets` is prerepresented bytes stream from upstream application,
 ///   which have been sent to the lower device but not yet acknowledged.
 /// - `unordered_packets` is the bytes stream received from the lower device,
-///   which can be acknowledged and extracted by `consume_unordered_packets` method
-///   then can be read by upstream application via `Tcp::poll_read` method.
+///   whose in-sequence part `acknowledge_unordered_packets` acknowledges and moves to `received`.
+/// - `received` is the acknowledged bytes stream, which `take_received` hands to the upstream reader.
 /// - `send_window_shift` is the peer's window scale, applied to every window the peer advertises,
 ///   and `recv_window_shift` is this stack's own, applied to every window this stack advertises.
 ///   Both are settled by the SYN exchange, and `recv_window_shift` is `None` when the peer's SYN
@@ -67,6 +70,7 @@ pub(crate) struct Tcb {
     state: TcpState,
     inflight_packets: BTreeMap<SeqNum, InflightPacket>,
     unordered_packets: BTreeMap<SeqNum, Vec<u8>>,
+    received: VecDeque<u8>,
     duplicate_ack_count: usize,
     duplicate_ack_count_helper: SeqNum,
     max_unacked_bytes: u32,
@@ -128,6 +132,7 @@ impl Tcb {
             state: TcpState::Listen,
             inflight_packets: BTreeMap::new(),
             unordered_packets: BTreeMap::new(),
+            received: VecDeque::new(),
             duplicate_ack_count: 0,
             duplicate_ack_count_helper: seq.into(),
             max_unacked_bytes,
@@ -159,23 +164,28 @@ impl Tcb {
         self.duplicate_ack_count >= self.max_count_for_dup_ack
     }
 
-    pub(super) fn add_unordered_packet(&mut self, seq: SeqNum, buf: Vec<u8>) {
+    pub(super) fn add_unordered_packet(&mut self, seq: SeqNum, mut buf: Vec<u8>) {
         if seq < self.ack {
             #[rustfmt::skip]
             log::warn!("{:?}: Received packet seq {seq} < self ack {}, len = {}", self.state, self.ack, buf.len());
             return;
         }
-        // The head-of-line segment always advances the stream, so it is admitted even at the limit;
-        // any other segment beyond the receive window is dropped for the peer's RTO to resend.
-        if seq != self.ack && self.get_unordered_packets_total_len() >= self.read_buffer_size {
+        // A segment starting beyond a closed receive window is dropped for the peer's RTO to resend.
+        if seq != self.ack && self.get_available_read_buffer_size() == 0 {
             #[rustfmt::skip]
             log::warn!("{:?}: Receive window full, dropping packet seq {seq}, len = {}", self.state, buf.len());
             return;
         }
-        self.unordered_packets.insert(seq, buf);
+        // Bytes past the right edge of the receive window are dropped for the peer to resend.
+        let window = self.read_buffer_size.saturating_sub(self.received.len());
+        buf.truncate(window.saturating_sub(seq.distance(self.ack) as usize));
+        if !buf.is_empty() {
+            self.unordered_packets.insert(seq, buf);
+        }
     }
     pub(super) fn get_available_read_buffer_size(&self) -> usize {
-        self.read_buffer_size.saturating_sub(self.get_unordered_packets_total_len())
+        self.read_buffer_size
+            .saturating_sub(self.get_unordered_packets_total_len() + self.received.len())
     }
     #[inline]
     pub(crate) fn get_unordered_packets_total_len(&self) -> usize {
@@ -228,6 +238,19 @@ impl Tcb {
         if data.is_empty() { None } else { Some(data) }
     }
 
+    /// Acknowledges the in-sequence data of the reassembly buffer, moving it to `received`.
+    pub(super) fn acknowledge_unordered_packets(&mut self) {
+        if let Some(data) = self.consume_unordered_packets(usize::MAX) {
+            self.received.extend(data);
+        }
+    }
+
+    /// Takes up to `max_bytes` of acknowledged data, in sequence order.
+    pub(super) fn take_received(&mut self, max_bytes: usize) -> Option<Vec<u8>> {
+        let len = max_bytes.min(self.received.len());
+        (len > 0).then(|| self.received.drain(..len).collect())
+    }
+
     pub(super) fn increase_seq(&mut self) {
         self.seq += 1;
     }
@@ -535,17 +558,13 @@ mod tests {
             MAX_RETRANSMIT_COUNT,
         );
 
-        // fill the receive buffer to its limit with an out-of-order gap held open
-        tcb.add_unordered_packet(SeqNum(1000 + READ_BUFFER_SIZE as u32), vec![7; READ_BUFFER_SIZE]);
+        // fill the receive buffer to its limit
+        tcb.add_unordered_packet(SeqNum(1000), vec![7; READ_BUFFER_SIZE]);
         assert_eq!(tcb.get_unordered_packets_total_len(), READ_BUFFER_SIZE);
 
         // a further out-of-order segment is dropped, keeping the buffer bounded
         tcb.add_unordered_packet(SeqNum(1000 + 2 * READ_BUFFER_SIZE as u32), vec![8; 500]);
         assert_eq!(tcb.get_unordered_packets_total_len(), READ_BUFFER_SIZE);
-
-        // the head-of-line segment is admitted even at the limit, so the stream advances
-        tcb.add_unordered_packet(SeqNum(1000), vec![9; 500]);
-        assert_eq!(tcb.unordered_packets.get(&SeqNum(1000)).unwrap().len(), 500);
     }
 
     #[test]
diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index a1e0cac..28e06b2 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -312,6 +312,14 @@ impl AsyncRead for IpStackTcpStream {
 
         let state = self.tcb.lock().unwrap().get_state();
         if state == TcpState::Closed {
+            // Data on hand reaches the reader before end-of-stream.
+            let data = self.data_rx.try_recv().ok();
+            if let Some(data) = data.or_else(|| self.tcb.lock().unwrap().take_received(usize::MAX)) {
+                let len = std::cmp::min(buf.remaining(), data.len());
+                buf.put_slice(&data[..len]);
+                self.temp_read_buffer.extend_from_slice(&data[len..]);
+                return Poll::Ready(Ok(()));
+            }
             self.shutdown.lock().unwrap().ready();
             self.write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
             return Poll::Ready(Ok(()));
@@ -671,6 +679,7 @@ async fn tcp_main_logic_loop(
         let network_packet = tokio::select! {
             _ = exit_monitor.recv() => {
                 log::debug!("{network_tuple} task exited due to exit signal");
+                read_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
                 break;
             }
             _ = drain_notify.notified() => {
@@ -943,7 +952,9 @@ async fn tcp_main_logic_loop(
                         extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
                         write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
                     }
-                    if flags & FIN == FIN {
+                    if flags & FIN == FIN && tcb.get_ack() == incoming_seq + len as u32 {
+                        tcb.increase_ack();
+                        write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK, None, None)?;
                         tcb.change_state(TcpState::TimeWait);
                         tokio::spawn(task_wait_to_close(tcb_clone.clone(), exit_notifier, network_tuple, config.two_msl));
                         let new_state = tcb.get_state();
@@ -994,8 +1005,9 @@ fn extract_data_n_write_upstream(
         return Ok(());
     }
 
-    // Reserve the handoff slot before consuming, so buffered data is removed only once it has a
-    // guaranteed home; the reserved permit shrinks the advertised window until the reader drains it.
+    tcb.acknowledge_unordered_packets();
+
+    // Reserve the handoff slot before taking acknowledged data.
     let permit = match data_tx.try_reserve() {
         Ok(permit) => permit,
         Err(tokio::sync::mpsc::error::TrySendError::Full(())) => {
@@ -1007,7 +1019,7 @@ fn extract_data_n_write_upstream(
         }
     };
 
-    if let Some(data) = tcb.consume_unordered_packets(READ_CHUNK) {
+    if let Some(data) = tcb.take_received(READ_CHUNK) {
         let hint = if state == TcpState::Established { "normally" } else { "still" };
         log::trace!("{network_tuple} {state:?}: {l_info} {hint} receiving data, len = {}", data.len());
         permit.send(data);
@@ -1336,18 +1348,18 @@ mod tests {
         extract_data_n_write_upstream(&up_tx, &mut tcb, nt, &data_tx, &read_notify).unwrap();
         assert_eq!(tcb.get_ack(), SeqNum(2000));
 
-        // channel is full: extract leaves the remaining data in the map and does not advance ack
+        // channel is full: extract advances ack over the tail and holds it for the reader
         tcb.add_unordered_packet(SeqNum(2000), vec![3; 500]);
         extract_data_n_write_upstream(&up_tx, &mut tcb, nt, &data_tx, &read_notify).unwrap();
-        assert_eq!(tcb.get_ack(), SeqNum(2000));
-        assert_eq!(tcb.get_unordered_packets_total_len(), 500);
+        assert_eq!(tcb.get_ack(), SeqNum(2500));
+        assert_eq!(tcb.get_available_read_buffer_size(), READ_BUFFER_SIZE - 500);
 
         // draining the reader frees a slot, and the next extract flushes the tail
         let first = data_rx.recv().await.unwrap();
         assert_eq!(first.len(), 1000);
         extract_data_n_write_upstream(&up_tx, &mut tcb, nt, &data_tx, &read_notify).unwrap();
         assert_eq!(tcb.get_ack(), SeqNum(2500));
-        assert_eq!(tcb.get_unordered_packets_total_len(), 0);
+        assert_eq!(tcb.get_available_read_buffer_size(), READ_BUFFER_SIZE);
     }
 
     /// Opens a connection with a SYN advertising `window` and `syn_options`, returning the stream

From f3c32d3c8fc246ae44df2ead42c945b387791678 Mon Sep 17 00:00:00 2001
From: chelovek <569636+hypnosis@users.noreply.github.com>
Date: Mon, 12 Oct 2026 04:20:21 +0300
Subject: [PATCH 08/13] Do not send past the right edge of the peer window
 (#99)

poll_write checked that the send buffer was not full, then sent the whole
buffer up to the MSS. The last segment could end past the right edge of the
peer window; the peer trims it and the tail has to be retransmitted.

Cut each write to the usable window: min(max_unacked_bytes, send_window)
minus the bytes in flight.

Co-authored-by: hypnosis <hypnosis@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/stream/tcb.rs | 11 ++++++++---
 src/stream/tcp.rs | 26 ++++++++++++++++++++++++++
 2 files changed, 34 insertions(+), 3 deletions(-)

diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs
index 9d322a2..208668d 100644
--- a/src/stream/tcb.rs
+++ b/src/stream/tcb.rs
@@ -453,10 +453,15 @@ impl Tcb {
         self.inflight_packets.values().collect::<Vec<_>>()
     }
 
+    /// Bytes that can still be sent before reaching the right edge of the peer window or the cap on
+    /// unacknowledged bytes, whichever is nearer. Simplified version: min(cwnd, rwnd) - in flight.
+    pub(super) fn get_usable_send_window(&self) -> u32 {
+        let in_flight = self.seq.distance(self.get_last_received_ack());
+        self.max_unacked_bytes.min(self.get_send_window()).saturating_sub(in_flight)
+    }
+
     pub fn is_send_buffer_full(&self) -> bool {
-        // To respect the receiver's window (remote_window) size and avoid sending too many unacknowledged packets, which may cause packet loss
-        // Simplified version: min(cwnd, rwnd)
-        self.seq.distance(self.get_last_received_ack()) >= self.max_unacked_bytes.min(self.get_send_window())
+        self.get_usable_send_window() == 0
     }
 }
 
diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index 28e06b2..4af9895 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -391,6 +391,8 @@ impl AsyncWrite for IpStackTcpStream {
             return Poll::Pending;
         }
 
+        // A segment past the right edge of the peer window is trimmed by the peer (RFC 9293 §3.10.7.4).
+        let buf = &buf[..buf.len().min(tcb.get_usable_send_window() as usize)];
         let sender = &self.up_packet_sender;
         let payload_len = write_packet_to_device(sender, nt, &tcb, None, ACK | PSH, None, Some(buf.to_vec()))?;
         let was_empty = tcb.is_inflight_queue_empty();
@@ -1470,4 +1472,28 @@ mod tests {
         assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, b"sent"), Poll::Ready(Ok(4))));
         assert_eq!(up_rx.recv().await.unwrap().payload.as_deref(), Some(&b"sent"[..]));
     }
+
+    /// A write that would pass the right edge of the peer window is cut at the edge, and the next
+    /// one waits for an ACK instead of sending past it.
+    #[tokio::test]
+    async fn writer_stops_at_the_right_edge_of_the_peer_window() {
+        let (mut stream, syn_ack, mut up_rx) = open(2000, &[]).await;
+        let peer_seq = SeqNum(syn_ack.acknowledgment_number);
+        feed(&stream, &syn_ack, peer_seq, 2000, &[1; 4]);
+        up_rx.recv().await.unwrap();
+
+        let (_, waker) = recording_waker();
+        let mut cx = Context::from_waker(&waker);
+        let chunk = [7; 1460];
+        assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, &chunk), Poll::Ready(Ok(1460))));
+        assert_eq!(up_rx.recv().await.unwrap().payload.map(|p| p.len()), Some(1460));
+
+        assert!(
+            matches!(Pin::new(&mut stream).poll_write(&mut cx, &chunk), Poll::Ready(Ok(540))),
+            "the second write is not cut at the right edge of the 2000-byte window"
+        );
+        assert_eq!(up_rx.recv().await.unwrap().payload.map(|p| p.len()), Some(540));
+
+        assert!(Pin::new(&mut stream).poll_write(&mut cx, &chunk).is_pending());
+    }
 }

From e15e90aa9a88a960ab8a80fca11754364d7e3c85 Mon Sep 17 00:00:00 2001
From: wycliff omondi odhiambo <115923623+WYCLIFF001@users.noreply.github.com>
Date: Mon, 12 Oct 2026 04:23:16 +0300
Subject: [PATCH 09/13] Format log context only when a line is logged (#98)

The session loop built the `local { seq, ack }` string and the header
description for every packet, and poll_write another string for every
held write, whatever the log level. At the default level none of them is
ever printed, so every packet paid for an allocation and a format that
went unused. The context is now a value that formats itself when a log
line is emitted, and the header description is built only when trace is
enabled.
---
 src/stream/tcp.rs | 34 ++++++++++++++++++++++++----------
 1 file changed, 24 insertions(+), 10 deletions(-)

diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index 4af9895..7fd3204 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -247,7 +247,7 @@ impl IpStackTcpStream {
 
         let sessions = SESSION_COUNTER.fetch_add(1, std::sync::atomic::Ordering::SeqCst).saturating_add(1);
         let (seq, ack, state) = { (tcb.get_seq().0, tcb.get_ack().0, tcb.get_state()) };
-        let l_info = format!("local {{ seq: {seq}, ack: {ack} }}");
+        let l_info = LocalInfo(seq, ack);
         log::debug!("{tuple} {state:?}: {l_info} session begins, total TCP sessions: {sessions}");
 
         stream.spawn_tasks()?;
@@ -330,7 +330,7 @@ impl AsyncRead for IpStackTcpStream {
             {
                 let mut tcb = self.tcb.lock().unwrap();
                 let (seq, ack) = (tcb.get_seq().0, tcb.get_ack().0);
-                let l_info = format!("local {{ seq: {seq}, ack: {ack} }}");
+                let l_info = LocalInfo(seq, ack);
                 log::debug!("{network_tuple} {state:?}: [poll_read] {l_info}, session timeout reached, closing forcefully...");
                 let sender = &self.up_packet_sender;
                 write_packet_to_device(sender, network_tuple, &tcb, None, ACK | RST, None, None)?;
@@ -386,8 +386,9 @@ impl AsyncWrite for IpStackTcpStream {
 
         if send_window == 0 || is_full {
             self.write_notify.lock().unwrap().replace(cx.waker().clone());
-            let info = format!("current send window: {send_window}, send buffer full: {is_full}");
-            log::trace!("{nt} {state:?}: [poll_write] {info}, waiting for the other side to send ACK...");
+            log::trace!(
+                "{nt} {state:?}: [poll_write] current send window: {send_window}, send buffer full: {is_full}, waiting for the other side to send ACK..."
+            );
             return Poll::Pending;
         }
 
@@ -404,7 +405,7 @@ impl AsyncWrite for IpStackTcpStream {
         }
 
         let (state, seq, ack) = (tcb.get_state(), tcb.get_seq(), tcb.get_ack());
-        let l_info = format!("local {{ seq: {seq}, ack: {ack} }}");
+        let l_info = LocalInfo(seq, ack);
         log::trace!("{nt} {state:?}: [poll_write] {l_info} upstream data written to device, len = {payload_len}");
 
         Poll::Ready(Ok(payload_len))
@@ -571,7 +572,7 @@ async fn tcp_main_logic_loop(
 
         tcb.increase_ack();
         let (seq, ack) = (tcb.get_seq().0, tcb.get_ack().0);
-        let l_info = format!("local {{ seq: {seq}, ack: {ack} }}");
+        let l_info = LocalInfo(seq, ack);
         log::trace!("{network_tuple} {state:?}: {l_info} session begins");
         write_packet_to_device(
             &up_packet_sender,
@@ -765,9 +766,12 @@ async fn tcp_main_logic_loop(
         let pkt_type = tcb.check_pkt_type(tcp_header, &payload);
 
         let (state, seq, ack) = { (tcb.get_state(), tcb.get_seq(), tcb.get_ack()) };
-        let (info, len) = (tcp_header_fmt(tcp_header), payload.len());
-        let l_info = format!("local {{ seq: {seq}, ack: {ack} }}");
-        log::trace!("{network_tuple} {state:?}: {l_info} {info}, {pkt_type:?}, len = {len}");
+        let len = payload.len();
+        let l_info = LocalInfo(seq, ack);
+        if log::log_enabled!(log::Level::Trace) {
+            let info = tcp_header_fmt(tcp_header);
+            log::trace!("{network_tuple} {state:?}: {l_info} {info}, {pkt_type:?}, len = {len}");
+        }
         if pkt_type == PacketType::Invalid {
             continue;
         }
@@ -1001,7 +1005,7 @@ fn extract_data_n_write_upstream(
     read_notify: &std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
 ) -> std::io::Result<()> {
     let (state, seq, ack) = (tcb.get_state(), tcb.get_seq(), tcb.get_ack());
-    let l_info = format!("local {{ seq: {seq}, ack: {ack} }}");
+    let l_info = LocalInfo(seq, ack);
     if state == TcpState::Closed {
         log::debug!("{network_tuple} {state:?}: {l_info} session closed, exiting \"data extraction task\"...");
         return Ok(());
@@ -1032,6 +1036,16 @@ fn extract_data_n_write_upstream(
 }
 
 /// Send a TCP packet to the downstream device, with the specified flags, sequence number, and payload.
+/// `local { seq, ack }` for log lines, formatted only when a line is actually emitted: the session
+/// loop builds one for every packet, whether or not anything is logged.
+struct LocalInfo<S, A>(S, A);
+
+impl<S: std::fmt::Display, A: std::fmt::Display> std::fmt::Display for LocalInfo<S, A> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "local {{ seq: {}, ack: {} }}", self.0, self.1)
+    }
+}
+
 /// The returned value is the length of the `payload` sent, it may be shorter than the length of the incoming parameter `payload`.
 pub(crate) fn write_packet_to_device(
     up_packet_sender: &PacketSender,

From b9c101fdd9ceae7a7d4aa268e2bc5d40fb3b00c1 Mon Sep 17 00:00:00 2001
From: wycliff omondi odhiambo <115923623+WYCLIFF001@users.noreply.github.com>
Date: Mon, 12 Oct 2026 04:38:40 +0300
Subject: [PATCH 10/13] Answer out-of-order segments with a duplicate ACK, keep
 straddling retransmissions, and keep segments inside the peer's window (#97)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* Answer an out-of-order segment with a duplicate ACK

A data segment is now always acknowledged. One that fills no gap, because
it arrived out of order or was already received, used to leave nothing
for the reader and so produced no ACK at all, which left the peer's
retransmission timer as the only way past a lost segment. RFC 5681 §4.2
asks for an immediate duplicate ACK instead, and three of them are what
drive the peer's fast retransmit.

A segment carrying PSH is handled like any other: out of order, it used
to be dropped without an ACK, so the peer resent it on its timer and
backed off with every loss.

* Keep the new part of a retransmission that straddles the ack

A segment starting below the ack was dropped whole, even when it ran past
it. That happens when the peer re-segments a retransmission after an ACK
it missed: the bytes past the ack were then lost with the rest, and the
peer had to send them yet again. The segment is now trimmed to the part
that is new (RFC 9293 §3.10.7.4); one lying entirely below the ack is
still dropped as a duplicate.

* Keep each segment inside the peer's window

A write was held only once the data in flight had reached the peer's
window, and was then sent as a whole segment, so the last segment before
the window filled could run up to a segment past it. RFC 9293 §3.8.6.1
keeps what is in flight within the window. A segment now carries no more
than the room the window has left.

The write also copies only the bytes the segment carries. It used to copy
the caller's whole buffer and send one segment of it, so a large write
was copied again for every segment it produced.

---------

Co-authored-by: Sajjad Pourali <sajjad@pourali.com>
---
 src/stream/tcb.rs |  43 +++++++++++--
 src/stream/tcp.rs | 156 +++++++++++++++++++++++++++++++++++++++++-----
 2 files changed, 179 insertions(+), 20 deletions(-)

diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs
index 208668d..abff5c1 100644
--- a/src/stream/tcb.rs
+++ b/src/stream/tcb.rs
@@ -164,11 +164,17 @@ impl Tcb {
         self.duplicate_ack_count >= self.max_count_for_dup_ack
     }
 
-    pub(super) fn add_unordered_packet(&mut self, seq: SeqNum, mut buf: Vec<u8>) {
+    pub(super) fn add_unordered_packet(&mut self, mut seq: SeqNum, mut buf: Vec<u8>) {
         if seq < self.ack {
-            #[rustfmt::skip]
-            log::warn!("{:?}: Received packet seq {seq} < self ack {}, len = {}", self.state, self.ack, buf.len());
-            return;
+            // A retransmission re-segmented across the ack: keep the bytes past it.
+            let received = self.ack.distance(seq) as usize;
+            if received >= buf.len() {
+                #[rustfmt::skip]
+                log::warn!("{:?}: Received packet seq {seq} < self ack {}, len = {}", self.state, self.ack, buf.len());
+                return;
+            }
+            buf.drain(..received);
+            seq = self.ack;
         }
         // A segment starting beyond a closed receive window is dropped for the peer's RTO to resend.
         if seq != self.ack && self.get_available_read_buffer_size() == 0 {
@@ -549,6 +555,35 @@ mod tests {
         assert!(data.is_none());
     }
 
+    /// A retransmission re-segmented across the ack keeps the bytes past it (RFC 9293 §3.10.7.4
+    /// trims a segment to the part that lies in the window), instead of being dropped whole.
+    #[test]
+    fn test_add_unordered_packet_keeps_the_new_part_of_a_straddling_segment() {
+        let mut tcb = Tcb::new(
+            SeqNum(1000),
+            u16::MAX,
+            None,
+            1500,
+            MAX_UNACK,
+            READ_BUFFER_SIZE,
+            MAX_COUNT_FOR_DUP_ACK,
+            RTO,
+            MAX_RETRANSMIT_COUNT,
+        );
+        tcb.add_unordered_packet(SeqNum(1000), vec![1; 100]);
+        tcb.consume_unordered_packets(usize::MAX).unwrap();
+        assert_eq!(tcb.ack, SeqNum(1100));
+
+        // 1050..1150: the first 50 bytes were already received, the last 50 are new.
+        tcb.add_unordered_packet(SeqNum(1050), vec![2; 100]);
+        assert_eq!(tcb.consume_unordered_packets(usize::MAX), Some(vec![2; 50]));
+        assert_eq!(tcb.ack, SeqNum(1150));
+
+        // Entirely below the ack: a pure duplicate, nothing is stored.
+        tcb.add_unordered_packet(SeqNum(1000), vec![3; 100]);
+        assert_eq!(tcb.consume_unordered_packets(usize::MAX), None);
+    }
+
     #[test]
     fn test_add_unordered_packet_enforces_read_buffer() {
         let mut tcb = Tcb::new(
diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index 7fd3204..d52c08c 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -395,7 +395,7 @@ impl AsyncWrite for IpStackTcpStream {
         // A segment past the right edge of the peer window is trimmed by the peer (RFC 9293 §3.10.7.4).
         let buf = &buf[..buf.len().min(tcb.get_usable_send_window() as usize)];
         let sender = &self.up_packet_sender;
-        let payload_len = write_packet_to_device(sender, nt, &tcb, None, ACK | PSH, None, Some(buf.to_vec()))?;
+        let payload_len = write_packet_to_device(sender, nt, &tcb, None, ACK | PSH, None, Some(buf[..take].to_vec()))?;
         let was_empty = tcb.is_inflight_queue_empty();
         tcb.add_inflight_packet(buf[..payload_len].to_vec())?;
         if was_empty {
@@ -779,8 +779,15 @@ async fn tcp_main_logic_loop(
         match state {
             TcpState::SynReceived if flags & ACK == ACK => {
                 if len > 0 {
-                    tcb.add_unordered_packet(incoming_seq, payload);
-                    extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
+                    receive_segment(
+                        &up_packet_sender,
+                        &mut tcb,
+                        network_tuple,
+                        incoming_seq,
+                        payload,
+                        &data_tx,
+                        &read_notify,
+                    )?;
                 }
                 tcb.change_state(TcpState::Established);
                 write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
@@ -805,9 +812,8 @@ async fn tcp_main_logic_loop(
                             }
                         }
                         PacketType::NewPacket => {
-                            tcb.add_unordered_packet(incoming_seq, payload);
                             let nt = network_tuple;
-                            extract_data_n_write_upstream(&up_packet_sender, &mut tcb, nt, &data_tx, &read_notify)?;
+                            receive_segment(&up_packet_sender, &mut tcb, nt, incoming_seq, payload, &data_tx, &read_notify)?;
                             write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
                         }
                         PacketType::Ack => {
@@ -866,9 +872,18 @@ async fn tcp_main_logic_loop(
                         ));
                     }
                 } else if flags == (ACK | PSH) && pkt_type == PacketType::NewPacket {
-                    if !payload.is_empty() && tcb.get_ack() == incoming_seq {
-                        tcb.add_unordered_packet(incoming_seq, payload);
-                        extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
+                    // PSH changes nothing about where a segment belongs: one that arrives out of order
+                    // is kept and answered like any other.
+                    if !payload.is_empty() {
+                        receive_segment(
+                            &up_packet_sender,
+                            &mut tcb,
+                            network_tuple,
+                            incoming_seq,
+                            payload,
+                            &data_tx,
+                            &read_notify,
+                        )?;
                     }
                     write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
                 } else {
@@ -924,8 +939,15 @@ async fn tcp_main_logic_loop(
                     tcb.change_state(TcpState::FinWait2);
                     if len > 0 {
                         // if the other side is still sending data, we need to deal with it like PacketStatus::NewPacket
-                        tcb.add_unordered_packet(incoming_seq, payload);
-                        extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
+                        receive_segment(
+                            &up_packet_sender,
+                            &mut tcb,
+                            network_tuple,
+                            incoming_seq,
+                            payload,
+                            &data_tx,
+                            &read_notify,
+                        )?;
                         write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
                     }
                     let new_state = tcb.get_state();
@@ -954,8 +976,15 @@ async fn tcp_main_logic_loop(
                         write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK, None, None)?;
                     } else {
                         // if the other side is still sending data, we need to deal with it like PacketStatus::NewPacket
-                        tcb.add_unordered_packet(incoming_seq, payload);
-                        extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?;
+                        receive_segment(
+                            &up_packet_sender,
+                            &mut tcb,
+                            network_tuple,
+                            incoming_seq,
+                            payload,
+                            &data_tx,
+                            &read_notify,
+                        )?;
                         write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
                     }
                     if flags & FIN == FIN && tcb.get_ack() == incoming_seq + len as u32 {
@@ -997,18 +1026,41 @@ async fn wait_retransmission_deadline(tcb: &TcbPtr, arm_notify: &tokio::sync::No
     }
 }
 
-fn extract_data_n_write_upstream(
+/// Stores a data segment that arrived from the peer and acknowledges it.
+///
+/// The ACK goes out even when no new data reaches the reader: a segment that arrives out of order,
+/// or one already received, is answered with a duplicate ACK at once (RFC 5681 §4.2), which is
+/// what lets the peer fast-retransmit the segment in the gap instead of waiting out its timer.
+fn receive_segment(
     up_packet_sender: &PacketSender,
     tcb: &mut Tcb,
     network_tuple: NetworkTuple,
+    seq: SeqNum,
+    payload: Vec<u8>,
     data_tx: &tokio::sync::mpsc::Sender<Vec<u8>>,
     read_notify: &std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
 ) -> std::io::Result<()> {
+    tcb.add_unordered_packet(seq, payload);
+    if !extract_data_n_write_upstream(up_packet_sender, tcb, network_tuple, data_tx, read_notify)? {
+        write_packet_to_device(up_packet_sender, network_tuple, tcb, None, ACK, None, None)?;
+    }
+    Ok(())
+}
+
+/// Hands acknowledged data to the reader. Returns whether an ACK was written, or none is due
+/// because the session is closed.
+fn extract_data_n_write_upstream(
+    up_packet_sender: &PacketSender,
+    tcb: &mut Tcb,
+    network_tuple: NetworkTuple,
+    data_tx: &tokio::sync::mpsc::Sender<Vec<u8>>,
+    read_notify: &std::sync::Arc<std::sync::Mutex<Option<Waker>>>,
+) -> std::io::Result<bool> {
     let (state, seq, ack) = (tcb.get_state(), tcb.get_seq(), tcb.get_ack());
     let l_info = LocalInfo(seq, ack);
     if state == TcpState::Closed {
         log::debug!("{network_tuple} {state:?}: {l_info} session closed, exiting \"data extraction task\"...");
-        return Ok(());
+        return Ok(true);
     }
 
     tcb.acknowledge_unordered_packets();
@@ -1018,7 +1070,7 @@ fn extract_data_n_write_upstream(
         Ok(permit) => permit,
         Err(tokio::sync::mpsc::error::TrySendError::Full(())) => {
             write_packet_to_device(up_packet_sender, network_tuple, tcb, None, ACK, None, None)?;
-            return Ok(());
+            return Ok(true);
         }
         Err(tokio::sync::mpsc::error::TrySendError::Closed(())) => {
             return Err(std::io::Error::new(BrokenPipe, "data channel closed"));
@@ -1031,8 +1083,9 @@ fn extract_data_n_write_upstream(
         permit.send(data);
         read_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(());
         write_packet_to_device(up_packet_sender, network_tuple, tcb, None, ACK, None, None)?;
+        return Ok(true);
     }
-    Ok(())
+    Ok(false)
 }
 
 /// Send a TCP packet to the downstream device, with the specified flags, sequence number, and payload.
@@ -1230,6 +1283,77 @@ mod tests {
         create_raw_packet(src, dst, |_, _| usize::MAX, ACK, TTL, seq, ack, u16::MAX, Vec::new(), &[]).unwrap()
     }
 
+    /// A segment from the peer carrying `payload` at `seq`, with `flags` and a raw `window`.
+    fn peer_segment(seq: u32, ack: u32, flags: u8, window: u16, payload: &[u8]) -> NetworkPacket {
+        let (src, dst) = (PEER.parse().unwrap(), LOCAL.parse().unwrap());
+        create_raw_packet(src, dst, |_, _| usize::MAX, flags, TTL, seq, ack, window, payload.to_vec(), &[]).unwrap()
+    }
+
+    /// Drains what the stack wrote to the device so far.
+    async fn written(up_rx: &mut PacketReceiver) -> Vec<NetworkPacket> {
+        tokio::task::yield_now().await;
+        let mut packets = Vec::new();
+        while let Ok(packet) = up_rx.try_recv() {
+            packets.push(packet);
+        }
+        packets
+    }
+
+    /// RFC 5681 §4.2: an out-of-order segment is answered at once with a duplicate ACK, whether or
+    /// not it carries PSH, and is kept for when the gap fills.
+    #[tokio::test(start_paused = true)]
+    async fn out_of_order_segment_is_answered_with_a_duplicate_ack() {
+        for flags in [ACK, ACK | PSH] {
+            let (mut stream, mut up_rx, local_seq) = established().await;
+
+            // Bytes 1001..1009 are missing; 1009..1013 arrive first.
+            stream
+                .stream_sender()
+                .send(peer_segment(1009, local_seq, flags, u16::MAX, &[3; 4]))
+                .unwrap();
+            let acks: Vec<u32> = written(&mut up_rx)
+                .await
+                .iter()
+                .map(|p| tcp_header_of(p).acknowledgment_number)
+                .collect();
+            assert_eq!(acks, [1001], "flags {flags:#x}: the out-of-order segment got no duplicate ACK");
+
+            // The gap fills: the ACK covers both segments and the reader sees them in order.
+            stream
+                .stream_sender()
+                .send(peer_segment(1001, local_seq, flags, u16::MAX, &[1; 8]))
+                .unwrap();
+            let acks: Vec<u32> = written(&mut up_rx)
+                .await
+                .iter()
+                .map(|p| tcp_header_of(p).acknowledgment_number)
+                .collect();
+            assert_eq!(acks.last(), Some(&1013), "flags {flags:#x}: the buffered segment was not kept");
+            let mut buf = [0u8; 12];
+            tokio::io::AsyncReadExt::read_exact(&mut stream, &mut buf).await.unwrap();
+            assert_eq!(buf[..8], [1; 8]);
+            assert_eq!(buf[8..], [3; 4]);
+        }
+    }
+
+    /// RFC 9293 §3.8.6.1: a sender keeps the data in flight within the peer's window, so the last
+    /// segment before the window fills is cut to the room left rather than sent whole.
+    #[tokio::test]
+    async fn segments_stay_inside_the_peers_window() {
+        let (mut stream, mut up_rx, local_seq) = established().await;
+        stream.stream_sender().send(peer_segment(1001, local_seq, ACK, 3000, &[])).unwrap();
+        written(&mut up_rx).await;
+
+        // More than the window: the write parks once the window is full.
+        let _ = tokio::time::timeout(Duration::from_millis(100), stream.write_all(&[7; 4000])).await;
+        let sent: Vec<usize> = written(&mut up_rx)
+            .await
+            .iter()
+            .filter_map(|p| p.payload.as_ref().map(Vec::len))
+            .collect();
+        assert_eq!(sent.iter().sum::<usize>(), 3000, "sent {sent:?} into a 3000-byte window");
+    }
+
     /// The intervals RFC 6298 §5.5 produces from a one-second timeout doubling to the §2.5 ceiling.
     const BACKOFF: [u64; MAX_RETRANSMIT_COUNT] = [1, 2, 4, 8, 16, 32, 60];
 

From 44bb34fe50bd9c04290b4b4f9bbeece1626a95d4 Mon Sep 17 00:00:00 2001
From: Sajjad Pourali <sajjad@pourali.com>
Date: Sun, 11 Oct 2026 18:41:30 -0700
Subject: [PATCH 11/13] fix(tcp): pass correct buffer slice to
 write_packet_to_device

---
 src/stream/tcp.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs
index d52c08c..212b7a0 100644
--- a/src/stream/tcp.rs
+++ b/src/stream/tcp.rs
@@ -395,7 +395,7 @@ impl AsyncWrite for IpStackTcpStream {
         // A segment past the right edge of the peer window is trimmed by the peer (RFC 9293 §3.10.7.4).
         let buf = &buf[..buf.len().min(tcb.get_usable_send_window() as usize)];
         let sender = &self.up_packet_sender;
-        let payload_len = write_packet_to_device(sender, nt, &tcb, None, ACK | PSH, None, Some(buf[..take].to_vec()))?;
+        let payload_len = write_packet_to_device(sender, nt, &tcb, None, ACK | PSH, None, Some(buf.to_vec()))?;
         let was_empty = tcb.is_inflight_queue_empty();
         tcb.add_inflight_packet(buf[..payload_len].to_vec())?;
         if was_empty {

From 51ee05f477e8a79bfcb884d818762c2cf06bff44 Mon Sep 17 00:00:00 2001
From: ackingliu <ackingliu@tencent.com>
Date: Tue, 27 Oct 2026 16:06:13 +0800
Subject: [PATCH 12/13] build: require tokio 1.54

Every other dependency requirement already admits its latest release
(etherparse 0.21, rand 0.10, thiserror 2, clap 4.6, criterion 0.8, tun 0.8).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 Cargo.toml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/Cargo.toml b/Cargo.toml
index 7935bed..792e5f4 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -15,7 +15,7 @@ etherparse = { version = "0.21", default-features = false, features = ["std"] }
 log = { version = "0.4", default-features = false }
 rand = { version = "0.10", default-features = false, features = ["thread_rng"] }
 thiserror = { version = "2.0", default-features = false }
-tokio = { version = "1.53", default-features = false, features = [
+tokio = { version = "1.54", default-features = false, features = [
     "sync",
     "rt",
     "time",
@@ -29,7 +29,7 @@ clap = { version = "4.6", default-features = false, features = ["derive"] }
 criterion = { version = "0.8" } # Benchmarks
 dotenvy = "0.15"
 env_logger = "0.11"
-tokio = { version = "1.53", default-features = false, features = [
+tokio = { version = "1.54", default-features = false, features = [
     "rt-multi-thread",
     "test-util",
 ] }

From 418ed41ed9358b5f4030c5fea1a5edc4b492ab54 Mon Sep 17 00:00:00 2001
From: ackingliu <ackingliu@tencent.com>
Date: Tue, 27 Oct 2026 16:22:06 +0800
Subject: [PATCH 13/13] build: create the examples directory before copying
 wintun.dll

Cargo creates `<profile>/examples` only when it builds an example. A
dependent crate, or any `cargo check`, never does, so on Windows targets
the build script failed with NotFound before the crate compiled at all.
Current nightly cargo no longer creates the directory up front.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 build.rs | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/build.rs b/build.rs
index 6ef1fc9..cca8814 100644
--- a/build.rs
+++ b/build.rs
@@ -16,9 +16,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         let dll_path = get_wintun_bin_relative_path()?;
         let src_path = crate_dir.join(dll_path);
 
-        let dst_path = cargo_target_dir.join("examples/wintun.dll");
+        let examples_dir = cargo_target_dir.join("examples");
+        let dst_path = examples_dir.join("wintun.dll");
 
-        // Copy to the target directory
+        // Cargo creates `examples/` only when it builds an example, which a
+        // dependent crate or a `cargo check` never does.
+        std::fs::create_dir_all(&examples_dir)?;
         std::fs::copy(src_path, &dst_path)?;
 
         // Set the modified time to the current time, or the publishing process will fail.