Answer out-of-order segments with a duplicate ACK, keep straddling retransmissions, and keep segments inside the peer's window

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 {
@@ -453,6 +459,13 @@ impl Tcb {
         self.inflight_packets.values().collect::<Vec<_>>()
     }
 
+    /// Bytes that may still be sent: the peer's window, capped by the unacknowledged limit, less
+    /// what is already in flight.
+    pub(super) fn send_room(&self) -> usize {
+        let limit = self.max_unacked_bytes.min(self.get_send_window());
+        limit.saturating_sub(self.seq.distance(self.get_last_received_ack())) as usize
+    }
+
     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)
@@ -544,6 +557,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(

src/stream/tcp.rs

@@ -391,8 +391,11 @@ impl AsyncWrite for IpStackTcpStream {
             return Poll::Pending;
         }
 
+        // A segment carries no more than the peer's window has room for (RFC 9293 §3.8.6.1), and only
+        // that much of `buf` is copied: the rest is the caller's to offer again.
+        let take = buf.len().min(tcb.send_room()).min(tcb.get_mtu() 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 {
@@ -773,8 +776,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(());
@@ -799,9 +809,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 => {
@@ -860,9 +869,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 {
@@ -918,8 +936,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();
@@ -948,8 +973,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 {
@@ -991,18 +1023,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 = format!("local {{ seq: {seq}, ack: {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();
@@ -1012,7 +1067,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"));
@@ -1025,8 +1080,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.
@@ -1214,6 +1270,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];