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 { + 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 { - 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, 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::>() @@ -393,26 +441,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 +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>, @@ -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,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::(); + + 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 { + 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::();