From 65ea5f6f3a1ecc8c930af7429f96c3dad7271fae Mon Sep 17 00:00:00 2001 From: IntellyCode <47359527+IntellyCode@users.noreply.github.com> Date: Mon, 14 Sep 2026 13:06:54 +0200 Subject: [PATCH 1/4] Retransmit on a timer and close the connection at R2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- Cargo.toml | 1 + src/stream/tcb.rs | 149 ++++++++++++++++++++++++++--------- src/stream/tcp.rs | 193 ++++++++++++++++++++++++++++++++++++++++++---- 3 files changed, 292 insertions(+), 51 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..9a4ff5c 100644 --- a/src/stream/tcb.rs +++ b/src/stream/tcb.rs @@ -10,8 +10,12 @@ pub(super) const MAX_COUNT_FOR_DUP_ACK: usize = 3; // Maximum number of duplicat /// Retransmission timeout 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 up to `MAX_RTO` on every expiry per RFC 6298 §5.5. + /// 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, } @@ -124,6 +133,7 @@ impl Tcb { read_buffer_size, max_count_for_dup_ack, rto, + current_rto: rto, max_retransmit_count, } } @@ -317,7 +327,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 +342,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 +357,68 @@ 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, and nothing while no data is outstanding. + pub(crate) fn get_retransmission_deadline(&self) -> Option { + self.inflight_packets + .values() + .map(|p| p.send_time) + .min() + .map(|t| t + self.current_rto) + } + #[must_use] - pub(crate) fn collect_timed_out_inflight_packets(&mut self) -> Vec { + /// 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. + pub(crate) fn collect_timed_out_inflight_packets(&mut self) -> (Vec, bool) { let mut retransmit_list = Vec::new(); + let (rto, r2) = (self.current_rto, self.max_retransmit_count); + let mut exhausted = false; - 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() { + 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(); + packet.send_time = tokio::time::Instant::now(); + exhausted |= packet.retransmit_count >= r2; retransmit_list.push(packet.clone()); } - true // keep the packet in the inflight_packets - }); - retransmit_list + } + if !retransmit_list.is_empty() { + self.current_rto = (self.current_rto * 2).min(MAX_RTO); // back off the timer, per RFC 6298 §5.5 + } + (retransmit_list, exhausted) } 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::>() @@ -393,26 +435,24 @@ impl Tcb { pub struct InflightPacket { pub seq: SeqNum, pub payload: Vec, - 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, rto: Duration) -> Self { + fn new(seq: SeqNum, payload: Vec) -> 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 +462,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 +627,42 @@ 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 §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 +678,25 @@ 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(); + // 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(); assert_eq!(packets.len(), 1); let packet = &packets[0]; assert_eq!(packet.retransmit_count, i + 1); - assert!(packet.retransmit_timeout > RTO); + assert!(tcb.current_rto > RTO); + assert_eq!(exhausted, i + 1 >= MAX_RETRANSMIT_COUNT); } - 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..7918757 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>, @@ -169,6 +169,7 @@ pub struct IpStackTcpStream { data_rx: tokio::sync::mpsc::Receiver>, read_notify: std::sync::Arc>>, drain_notify: Arc, + arm_notify: Arc, task_handle: Option>>, exit_notifier: Option>, temp_read_buffer: Vec, @@ -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>>, data_tx: tokio::sync::mpsc::Sender>, drain_notify: Arc, + arm_notify: Arc, mut exit_monitor: tokio::sync::mpsc::Receiver<()>, ) -> std::io::Result<()> { { @@ -669,6 +680,34 @@ 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 (packets, exhausted) = tcb.collect_timed_out_inflight_packets(); + 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), + )?; + } + 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; + } + continue; + } network_packet = stream_receiver.recv() => network_packet, }; @@ -707,20 +746,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 +963,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 +1159,132 @@ 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::(); + + 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 `seconds` and returns what the stack wrote to the device. + async fn advance(seconds: u64, up_rx: &mut PacketReceiver) -> Option { + tokio::time::advance(Duration::from_secs(seconds)).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(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: R2 transmissions of the same segment close the connection, after the at + /// least 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(); + + stream.write_all(b"hello").await.unwrap(); + up_rx.recv().await.unwrap(); + + // every interval but the last carries a retransmission; the last carries the reset + for seconds in BACKOFF.iter().take(MAX_RETRANSMIT_COUNT - 1) { + let packet = advance(*seconds, &mut up_rx).await.expect("no retransmission"); + assert!(!tcp_header_of(&packet).rst); + } + // the last expiry retransmits once more and then resets + let last = advance(BACKOFF[MAX_RETRANSMIT_COUNT - 1], &mut up_rx).await; + assert!(!tcp_header_of(&last.expect("no retransmission")).rst); + assert!(tcp_header_of(&up_rx.try_recv().expect("no reset")).rst); + + assert!((tokio::time::Instant::now() - opened).as_secs() > 100); + assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::Closed); + } + + /// 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(*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(*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::(); From cb333e84069056a8879aa94e923761d429c8b569 Mon Sep 17 00:00:00 2001 From: IntellyCode <47359527+IntellyCode@users.noreply.github.com> Date: Tue, 15 Sep 2026 10:50:56 +0200 Subject: [PATCH 2/4] 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. --- src/stream/tcb.rs | 57 ++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 54 insertions(+), 3 deletions(-) diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index 9a4ff5c..9296c85 100644 --- a/src/stream/tcb.rs +++ b/src/stream/tcb.rs @@ -74,7 +74,7 @@ pub(crate) struct Tcb { 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 up to `MAX_RTO` on every expiry per RFC 6298 §5.5. + /// 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, @@ -400,13 +400,14 @@ impl Tcb { for packet in self.inflight_packets.values_mut() { if packet.is_timed_out(rto) { packet.retransmit_count += 1; - packet.send_time = tokio::time::Instant::now(); exhausted |= packet.retransmit_count >= r2; retransmit_list.push(packet.clone()); } } if !retransmit_list.is_empty() { - self.current_rto = (self.current_rto * 2).min(MAX_RTO); // back off the timer, per RFC 6298 §5.5 + 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 * 2).min(MAX_RTO.max(self.rto)); // back off the timer, per RFC 6298 §5.5 } (retransmit_list, exhausted) } @@ -659,6 +660,56 @@ mod tests { assert_eq!(tcb.current_rto, RTO); } + /// 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), + 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), + 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)] From 41abf3aa7f076b3f573ae997350a80cec2fff4bd Mon Sep 17 00:00:00 2001 From: IntellyCode <47359527+IntellyCode@users.noreply.github.com> Date: Tue, 15 Sep 2026 11:20:57 +0200 Subject: [PATCH 3/4] Close at R2 without resending, stop the timer after a reset, and floor the timeout at one second --- src/stream/tcb.rs | 82 ++++++++++++++++++++++++++++++++++++++++------- src/stream/tcp.rs | 71 +++++++++++++++++++++++++++------------- 2 files changed, 118 insertions(+), 35 deletions(-) diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index 9296c85..2c06d1b 100644 --- a/src/stream/tcb.rs +++ b/src/stream/tcb.rs @@ -10,6 +10,9 @@ pub(super) const MAX_COUNT_FOR_DUP_ACK: usize = 3; // Maximum number of duplicat /// Retransmission timeout pub(super) const RTO: std::time::Duration = std::time::Duration::from_secs(1); +/// Floor on the configured retransmission timeout, per RFC 6298 §2.4 +const MIN_RTO: std::time::Duration = std::time::Duration::from_secs(1); + /// 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); @@ -94,6 +97,7 @@ impl Tcb { rto: std::time::Duration, max_retransmit_count: usize, ) -> Tcb { + let rto = rto.max(MIN_RTO); #[cfg(debug_assertions)] let seq = 100; #[cfg(not(debug_assertions))] @@ -380,36 +384,41 @@ impl Tcb { } /// The deadline the connection waits on: the earliest send time in the inflight queue plus the - /// current retransmission timeout, and nothing while no data is outstanding. + /// current retransmission timeout, present while data is outstanding and the sum fits the clock. pub(crate) fn get_retransmission_deadline(&self) -> Option { self.inflight_packets .values() .map(|p| p.send_time) .min() - .map(|t| t + self.current_rto) + .and_then(|t| t.checked_add(self.current_rto)) } #[must_use] /// 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. + /// 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, bool) { - let mut retransmit_list = Vec::new(); let (rto, r2) = (self.current_rto, self.max_retransmit_count); - let mut exhausted = false; + if self + .inflight_packets + .values() + .any(|p| p.is_timed_out(rto) && p.retransmit_count + 1 >= r2) + { + return (Vec::new(), true); + } + let mut retransmit_list = Vec::new(); for packet in self.inflight_packets.values_mut() { if packet.is_timed_out(rto) { packet.retransmit_count += 1; - exhausted |= packet.retransmit_count >= r2; retransmit_list.push(packet.clone()); } } 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 * 2).min(MAX_RTO.max(self.rto)); // back off the timer, per RFC 6298 §5.5 + 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, exhausted) + (retransmit_list, false) } pub(crate) fn get_inflight_packets_total_len(&self) -> usize { @@ -660,12 +669,55 @@ mod tests { 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, @@ -696,6 +748,8 @@ mod tests { let rto = Duration::from_secs(120); let mut tcb = Tcb::new( SeqNum(1000), + u16::MAX, + None, 1500, MAX_UNACK, READ_BUFFER_SIZE, @@ -739,11 +793,15 @@ mod tests { tokio::time::advance(wait).await; let (packets, exhausted) = tcb.collect_timed_out_inflight_packets(); - assert_eq!(packets.len(), 1); - let packet = &packets[0]; - assert_eq!(packet.retransmit_count, i + 1); + 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); - assert_eq!(exhausted, i + 1 >= MAX_RETRANSMIT_COUNT); } // the segment stays outstanding at R2; the connection closes rather than the queue losing it diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index 7918757..a1e0cac 100644 --- a/src/stream/tcp.rs +++ b/src/stream/tcp.rs @@ -683,7 +683,21 @@ async fn tcp_main_logic_loop( _ = 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}",); @@ -697,15 +711,6 @@ async fn tcp_main_logic_loop( Some(packet.payload), )?; } - 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; - } continue; } network_packet = stream_receiver.recv() => network_packet, @@ -1200,9 +1205,9 @@ mod tests { /// 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 `seconds` and returns what the stack wrote to the device. - async fn advance(seconds: u64, up_rx: &mut PacketReceiver) -> Option { - tokio::time::advance(Duration::from_secs(seconds)).await; + /// Advances the clock by `by` and returns what the stack wrote to the device. + async fn advance(by: Duration, up_rx: &mut PacketReceiver) -> Option { + tokio::time::advance(by).await; tokio::task::yield_now().await; up_rx.try_recv().ok() } @@ -1217,35 +1222,55 @@ mod tests { let sent = up_rx.recv().await.unwrap(); assert_eq!(tcp_header_of(&sent).sequence_number, local_seq); - let again = advance(BACKOFF[0], &mut up_rx).await.expect("no retransmission"); + 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: R2 transmissions of the same segment close the connection, after the at - /// least 100 seconds SHLD-11 asks for. + /// 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(); - // every interval but the last carries a retransmission; the last carries the reset for seconds in BACKOFF.iter().take(MAX_RETRANSMIT_COUNT - 1) { - let packet = advance(*seconds, &mut up_rx).await.expect("no retransmission"); + 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); } - // the last expiry retransmits once more and then resets - let last = advance(BACKOFF[MAX_RETRANSMIT_COUNT - 1], &mut up_rx).await; - assert!(!tcp_header_of(&last.expect("no retransmission")).rst); - assert!(tcp_header_of(&up_rx.try_recv().expect("no reset")).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)] @@ -1256,7 +1281,7 @@ mod tests { up_rx.recv().await.unwrap(); for seconds in BACKOFF.iter().take(4) { - let packet = advance(*seconds, &mut up_rx).await.expect("no retransmission"); + 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"[..])); } @@ -1272,7 +1297,7 @@ mod tests { up_rx.recv().await.unwrap(); for seconds in BACKOFF.iter().take(5) { - advance(*seconds, &mut up_rx).await.expect("no retransmission"); + advance(Duration::from_secs(*seconds), &mut up_rx).await.expect("no retransmission"); } stream.stream_sender().send(peer_ack(1001, local_seq + 5)).unwrap(); From 5029d9cd16ae4cd71d79f40d03c8be298247dada Mon Sep 17 00:00:00 2001 From: IntellyCode <47359527+IntellyCode@users.noreply.github.com> Date: Thu, 17 Sep 2026 10:57:04 +0200 Subject: [PATCH 4/4] Reuse RTO as the timeout floor and read the deadline from the queue head --- src/stream/tcb.rs | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/src/stream/tcb.rs b/src/stream/tcb.rs index 2c06d1b..2453667 100644 --- a/src/stream/tcb.rs +++ b/src/stream/tcb.rs @@ -7,12 +7,9 @@ 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); -/// Floor on the configured retransmission timeout, per RFC 6298 §2.4 -const MIN_RTO: std::time::Duration = std::time::Duration::from_secs(1); - /// 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); @@ -97,7 +94,7 @@ impl Tcb { rto: std::time::Duration, max_retransmit_count: usize, ) -> Tcb { - let rto = rto.max(MIN_RTO); + let rto = rto.max(RTO); #[cfg(debug_assertions)] let seq = 100; #[cfg(not(debug_assertions))] @@ -388,9 +385,8 @@ impl Tcb { pub(crate) fn get_retransmission_deadline(&self) -> Option { self.inflight_packets .values() - .map(|p| p.send_time) - .min() - .and_then(|t| t.checked_add(self.current_rto)) + .next() + .and_then(|p| p.send_time.checked_add(self.current_rto)) } #[must_use]