From 3cd94b10c0836bcaf091736a6d23059b0072e7e7 Mon Sep 17 00:00:00 2001 From: frontierlabs-admin <314691291+frontierlabs-admin@users.noreply.github.com> Date: Sun, 1 Nov 2026 11:42:25 +0000 Subject: [PATCH 1/3] Keep the payload of a segment that carries FIN MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A peer that half-closes with data still queued, as Linux does on shutdown(SHUT_WR), puts its FIN on the last data segment. In Established such a segment was handled by its flags alone: - ACK|FIN in sequence took the FIN without its payload, so the last bytes of the stream never reached the reader and the ACK number fell short of them. - ACK|PSH|FIN matched no arm and was dropped without an ACK, every retransmission of it too. - A FIN segment arriving ahead of a gap was dropped without an ACK. RFC 9293 §3.10.7.4 processes the segment text before the FIN bit. The payload is now received like that of any other data segment, PSH or not, and the FIN is taken only once every byte before it has arrived. A FIN that is not yet in sequence leaves its payload queued and is taken when the peer retransmits it. Refs #111 --- src/stream/tcp.rs | 195 +++++++++++++++++++++++++++++++++++----------- 1 file changed, 148 insertions(+), 47 deletions(-) diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index 212b7a0..0b7e198 100644 --- a/src/stream/tcp.rs +++ b/src/stream/tcp.rs @@ -821,55 +821,64 @@ async fn tcp_main_logic_loop( } PacketType::Invalid => {} } - } 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)?; - tcb.change_state(TcpState::CloseWait); - - let s = tcb.get_state(); - let len = tcb.get_inflight_packets_total_len(); - if len == 0 { - // All upstream data sent, proceed to LastAck - log::trace!("{network_tuple} {s:?}: {l_info}, {pkt_type:?}, closed by the other side, no upstream data"); - - // Here we don't wait, just send FIN to the other side and change state to LastAck directly, - write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK | FIN, None, None)?; - tcb.increase_seq(); - tcb.change_state(TcpState::LastAck); + } else if flags & !PSH == (ACK | FIN) { + // RFC 9293 §3.10.7.4: the text of the segment is processed before its FIN, and the FIN is + // taken only once every byte before it has arrived; until then the peer retransmits it. + if len > 0 { + let nt = network_tuple; + 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(()); + } + if tcb.get_ack() == incoming_seq + len as u32 { + // 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)?; + tcb.change_state(TcpState::CloseWait); let s = tcb.get_state(); - log::trace!("{network_tuple} {s:?}: {l_info}, {pkt_type:?}, wait the last ack from the other side"); - - // Here we set a timer to wait for the last ACK from the other side. - // If the timer expires, we send an ACK|FIN packet to the other side again and wait anthoer timeout - // till the retries reach the limit, and then close the session forcibly. - let up = up_packet_sender.clone(); - tokio::spawn(task_last_ack( - tcb_clone.clone(), - exit_notifier, - network_tuple, - up, - config.last_ack_timeout, - config.last_ack_max_retries, - )); - } else { - // Upstream data pending, wake write_notify and wait - write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); - log::debug!("{network_tuple} {state:?}: Waiting for upstream data to complete, inflight packets: {len}",); - - // Spawn a timeout task to force FIN if upstream is unresponsive - let tcb = tcb_clone.clone(); - let up = up_packet_sender.clone(); - tokio::spawn(task_timed_out_for_close_wait( - tcb, - exit_notifier, - network_tuple, - up, - config.close_wait_timeout, - config.last_ack_timeout, - config.last_ack_max_retries, - )); + let len = tcb.get_inflight_packets_total_len(); + if len == 0 { + // All upstream data sent, proceed to LastAck + log::trace!("{network_tuple} {s:?}: {l_info}, {pkt_type:?}, closed by the other side, no upstream data"); + + // Here we don't wait, just send FIN to the other side and change state to LastAck directly, + write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK | FIN, None, None)?; + tcb.increase_seq(); + tcb.change_state(TcpState::LastAck); + + let s = tcb.get_state(); + log::trace!("{network_tuple} {s:?}: {l_info}, {pkt_type:?}, wait the last ack from the other side"); + + // Here we set a timer to wait for the last ACK from the other side. + // If the timer expires, we send an ACK|FIN packet to the other side again and wait anthoer timeout + // till the retries reach the limit, and then close the session forcibly. + let up = up_packet_sender.clone(); + tokio::spawn(task_last_ack( + tcb_clone.clone(), + exit_notifier, + network_tuple, + up, + config.last_ack_timeout, + config.last_ack_max_retries, + )); + } else { + // Upstream data pending, wake write_notify and wait + write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); + log::debug!("{network_tuple} {state:?}: Waiting for upstream data to complete, inflight packets: {len}",); + + // Spawn a timeout task to force FIN if upstream is unresponsive + let tcb = tcb_clone.clone(); + let up = up_packet_sender.clone(); + tokio::spawn(task_timed_out_for_close_wait( + tcb, + exit_notifier, + network_tuple, + up, + config.close_wait_timeout, + config.last_ack_timeout, + config.last_ack_max_retries, + )); + } } } else if flags == (ACK | PSH) && pkt_type == PacketType::NewPacket { // PSH changes nothing about where a segment belongs: one that arrives out of order @@ -1336,6 +1345,98 @@ mod tests { } } + /// The ACK numbers of what the stack wrote to the device so far. + async fn acks_written(up_rx: &mut PacketReceiver) -> Vec { + written(up_rx) + .await + .iter() + .map(|p| tcp_header_of(p).acknowledgment_number) + .collect() + } + + /// RFC 9293 §3.10.7.4: the text of a segment carrying FIN is processed before its FIN, whether + /// or not it carries PSH, so the payload reaches the reader and the ACK covers both. + #[tokio::test(start_paused = true)] + async fn payload_of_a_fin_segment_reaches_the_reader() { + for flags in [ACK | FIN, ACK | PSH | FIN] { + let (mut stream, mut up_rx, local_seq) = established().await; + + stream + .stream_sender() + .send(peer_segment(1001, local_seq, flags, u16::MAX, b"tail")) + .unwrap(); + let acks = acks_written(&mut up_rx).await; + assert_eq!( + acks.last(), + Some(&1006), + "flags {flags:#x}: the ACK does not cover the payload and the FIN" + ); + + let mut buf = [0u8; 4]; + tokio::io::AsyncReadExt::read_exact(&mut stream, &mut buf).await.unwrap(); + assert_eq!(&buf, b"tail", "flags {flags:#x}"); + } + } + + /// A FIN segment that arrives ahead of a gap has its payload kept, but the FIN is taken only + /// once every byte before it has arrived; until then the peer retransmits it. + #[tokio::test(start_paused = true)] + async fn fin_ahead_of_a_gap_waits_for_the_gap_to_fill() { + for flags in [ACK | FIN, ACK | PSH | FIN] { + let (mut stream, mut up_rx, local_seq) = established().await; + + // Bytes 1001..1005 are missing; the FIN segment carrying 1005..1009 arrives first. + stream + .stream_sender() + .send(peer_segment(1005, local_seq, flags, u16::MAX, b"tail")) + .unwrap(); + assert_eq!(acks_written(&mut up_rx).await, [1001], "flags {flags:#x}: no duplicate ACK"); + assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::Established); + + // The gap fills: the ACK covers the kept payload, but not the FIN that arrived out of order. + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | PSH, u16::MAX, b"head")) + .unwrap(); + assert_eq!(acks_written(&mut up_rx).await.last(), Some(&1009), "flags {flags:#x}"); + assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::Established); + + // The peer retransmits the FIN segment, and its FIN is now in sequence. + stream + .stream_sender() + .send(peer_segment(1005, local_seq, flags, u16::MAX, b"tail")) + .unwrap(); + assert_eq!(acks_written(&mut up_rx).await.last(), Some(&1010), "flags {flags:#x}"); + + let mut buf = [0u8; 8]; + tokio::io::AsyncReadExt::read_exact(&mut stream, &mut buf).await.unwrap(); + assert_eq!(&buf, b"headtail", "flags {flags:#x}"); + } + } + + /// A FIN segment carrying data is answered like any other data segment, so one that reopens + /// the peer window wakes the writer it held, even when its FIN is not yet in sequence. + #[tokio::test(start_paused = true)] + async fn fin_segment_reopening_the_window_wakes_the_writer() { + let (mut stream, mut up_rx, local_seq) = established().await; + stream.stream_sender().send(peer_segment(1001, local_seq, ACK, 0, &[])).unwrap(); + written(&mut up_rx).await; + + let (woken, waker) = recording_waker(); + let mut cx = Context::from_waker(&waker); + assert!(Pin::new(&mut stream).poll_write(&mut cx, b"reply").is_pending()); + + stream + .stream_sender() + .send(peer_segment(1005, local_seq, ACK | FIN, u16::MAX, b"tail")) + .unwrap(); + written(&mut up_rx).await; + assert!( + woken.load(std::sync::atomic::Ordering::SeqCst), + "the FIN segment reopened the window without waking the writer" + ); + } + /// 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] From f1f76d9438a58ec2baefdf239820daf74da1a79f Mon Sep 17 00:00:00 2001 From: frontierlabs-admin <314691291+frontierlabs-admin@users.noreply.github.com> Date: Sun, 1 Nov 2026 11:58:55 +0000 Subject: [PATCH 2/3] Keep the write side open after the peer's FIN MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A FIN closes only the peer's direction (RFC 9293 §3.6). On a FIN in Established with nothing in flight, the stack answered at once with its own FIN and moved to LastAck, and in CloseWait any ACK with nothing in flight did the same. A client that sends a request and then calls shutdown(SHUT_WR) therefore had the reply cut off as soon as the data already written was acknowledged. A reader in CloseWait also got Pending rather than end of stream, and waited for a timer. After the peer's FIN the stack now: - acknowledges it, wakes the reader, which gets every byte sent before the FIN and then end of stream, and keeps the write side open; - in CloseWait, acknowledges a retransmitted FIN, or segment carrying it, and answers duplicate ACKs with a fast retransmission, as in Established; - sends its own FIN when the application shuts down, once everything written is acknowledged, or when it drops the stream. The FIN sent on shutdown is retransmitted by the LastAck timer, which now lives at module level so poll_shutdown can start it. A write after it fails with BrokenPipe. close_wait_timeout remains the backstop for an application that never shuts down, and it is now armed on every FIN rather than only when data was in flight. It counts time without progress, so a reply that is still being written and acknowledged is not cut after a fixed delay. A writer still waiting for window when the connection ends is woken. Refs #111, related to #89. --- src/stream/tcp.rs | 623 ++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 516 insertions(+), 107 deletions(-) diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index 0b7e198..79c398b 100644 --- a/src/stream/tcp.rs +++ b/src/stream/tcp.rs @@ -37,7 +37,8 @@ pub struct TcpConfig { pub last_ack_max_retries: usize, /// Timeout for the last ACK in the LAST_ACK state. Default is 500ms. pub last_ack_timeout: Duration, - /// Timeout for the CLOSE_WAIT state. Default is 5 seconds. + /// Timeout for the CLOSE_WAIT state: once this long passes with nothing sent or acknowledged, our FIN is + /// sent for an application that has not shut down. Default is 5 seconds. pub close_wait_timeout: Duration, /// Timeout for TCP connections. Default is 60 seconds. pub timeout: Duration, @@ -324,6 +325,18 @@ impl AsyncRead for IpStackTcpStream { self.write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); return Poll::Ready(Ok(())); } + if matches!(state, TcpState::CloseWait | TcpState::LastAck) { + // The peer's FIN came after every byte it sent, so what is on hand is the rest of the + // stream. The lock keeps the main loop from moving bytes into the channel in between. + let tcb = self.tcb.clone(); + let mut tcb = tcb.lock().unwrap(); + if let Some(data) = self.data_rx.try_recv().ok().or_else(|| tcb.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(())); + } // handle timeout if matches!(Pin::new(&mut self.timeout).poll(cx), Poll::Ready(_)) { @@ -362,6 +375,10 @@ impl AsyncRead for IpStackTcpStream { Poll::Ready(None) => Poll::Ready(Ok(())), Poll::Pending => { self.read_notify.lock().unwrap().replace(cx.waker().clone()); + // The peer's FIN may have landed after the state was read, and found no waker to wake. + if matches!(self.tcb.lock().unwrap().get_state(), TcpState::CloseWait | TcpState::LastAck) { + cx.waker().wake_by_ref(); + } Poll::Pending } } @@ -383,6 +400,10 @@ impl AsyncWrite for IpStackTcpStream { self.read_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); return Poll::Ready(Err(std::io::Error::new(BrokenPipe, "TCP connection closed"))); } + if state == TcpState::LastAck { + // Our FIN is out, and nothing may follow it. + return Poll::Ready(Err(std::io::Error::new(BrokenPipe, "TCP connection closing"))); + } if send_window == 0 || is_full { self.write_notify.lock().unwrap().replace(cx.waker().clone()); @@ -426,6 +447,21 @@ impl AsyncWrite for IpStackTcpStream { if state == TcpState::Closed { return Poll::Ready(Ok(())); } + if state == TcpState::CloseWait { + // The peer has closed its direction, so our FIN closes the connection, once every byte + // written before it is acknowledged; each ACK in CloseWait wakes `write_notify`. + let mut tcb = self.tcb.lock().unwrap(); + if tcb.get_inflight_packets_total_len() > 0 { + self.write_notify.lock().unwrap().replace(cx.waker().clone()); + } else if tcb.get_state() == TcpState::CloseWait { + send_fin_n_change_state_to_last_ack("[poll_shutdown]", nt, &self.up_packet_sender, &mut tcb)?; + if let Some(exit_notifier) = self.exit_notifier.clone() { + let (timeout, retries) = (self.config.last_ack_timeout, self.config.last_ack_max_retries); + let sender = self.up_packet_sender.clone(); + tokio::spawn(task_last_ack(self.tcb.clone(), exit_notifier, nt, sender, timeout, retries)); + } + } + } match shutdown { Shutdown::None => { if is_ready && state == TcpState::Established { @@ -464,6 +500,53 @@ fn send_fin_n_change_state_to_fin_wait1(hint: &str, nt: NetworkTuple, sender: &P Ok(()) } +/// Sends our FIN after the peer's, which closes the connection in both directions. +fn send_fin_n_change_state_to_last_ack(hint: &str, nt: NetworkTuple, sender: &PacketSender, tcb: &mut Tcb) -> std::io::Result<()> { + log::debug!("{nt} CloseWait: {hint} the peer has closed, send our FIN"); + write_packet_to_device(sender, nt, tcb, None, ACK | FIN, None, None)?; + tcb.increase_seq(); + tcb.change_state(TcpState::LastAck); + Ok(()) +} + +async fn task_last_ack( + tcb: TcbPtr, + exit_notifier: tokio::sync::mpsc::Sender<()>, + nt: NetworkTuple, + pkt_sdr: PacketSender, + last_ack_timeout: Duration, + last_ack_max_retries: usize, +) { + let hint = "[task_last_ack]"; + for idx in 1..=last_ack_max_retries { + let state = { tcb.lock().unwrap().get_state() }; + if state == TcpState::Closed { + log::debug!("{nt} {state:?}: {hint} session closed, exiting 1..."); + return; + } + + tokio::time::sleep(last_ack_timeout).await; + + { + let tcb = tcb.lock().unwrap(); + let state = tcb.get_state(); + if state == TcpState::Closed { + log::debug!("{nt} {state:?}: {hint} session closed, exiting 2..."); + return; + } + log::debug!("{nt} {state:?}: {hint} timer expired, resending ACK|FIN (retry {idx}/{last_ack_max_retries})"); + _ = write_packet_to_device(&pkt_sdr, nt, &tcb, None, ACK | FIN, None, None); + } + } + { + let mut tcb = tcb.lock().unwrap(); + tcb.change_state(TcpState::Closed); + let state = tcb.get_state(); + log::warn!("{nt} {state:?}: {hint} max retries reached, forcibly closing session"); + } + exit_notifier.send(()).await.unwrap_or(()); +} + impl Drop for IpStackTcpStream { fn drop(&mut self) { let (nt, state) = (self.network_tuple(), self.tcb.lock().unwrap().get_state()); @@ -474,7 +557,12 @@ impl Drop for IpStackTcpStream { // 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) { + let sent = if tcb.get_state() == TcpState::CloseWait { + send_fin_n_change_state_to_last_ack("[drop]", nt, &self.up_packet_sender, &mut tcb) + } else { + send_fin_n_change_state_to_fin_wait1("[drop]", nt, &self.up_packet_sender, &mut tcb) + }; + if let Err(e) = sent { log::debug!("{nt} {state:?}: [drop] cannot send the farewell packet: {e}"); } } @@ -602,44 +690,6 @@ async fn tcp_main_logic_loop( exit_notifier.send(()).await.unwrap_or(()); } - async fn task_last_ack( - tcb: TcbPtr, - exit_notifier: tokio::sync::mpsc::Sender<()>, - nt: NetworkTuple, - pkt_sdr: PacketSender, - last_ack_timeout: Duration, - last_ack_max_retries: usize, - ) { - let hint = "[task_last_ack]"; - for idx in 1..=last_ack_max_retries { - let state = { tcb.lock().unwrap().get_state() }; - if state == TcpState::Closed { - log::debug!("{nt} {state:?}: {hint} session closed, exiting 1..."); - return; - } - - tokio::time::sleep(last_ack_timeout).await; - - { - let tcb = tcb.lock().unwrap(); - let state = tcb.get_state(); - if state == TcpState::Closed { - log::debug!("{nt} {state:?}: {hint} session closed, exiting 2..."); - return; - } - log::debug!("{nt} {state:?}: {hint} timer expired, resending ACK|FIN (retry {idx}/{last_ack_max_retries})"); - _ = write_packet_to_device(&pkt_sdr, nt, &tcb, None, ACK | FIN, None, None); - } - } - { - let mut tcb = tcb.lock().unwrap(); - tcb.change_state(TcpState::Closed); - let state = tcb.get_state(); - log::warn!("{nt} {state:?}: {hint} max retries reached, forcibly closing session"); - } - exit_notifier.send(()).await.unwrap_or(()); - } - async fn task_timed_out_for_close_wait( tcb: TcbPtr, exit_notifier: tokio::sync::mpsc::Sender<()>, @@ -649,7 +699,19 @@ async fn tcp_main_logic_loop( last_ack_timeout: Duration, last_ack_max_retries: usize, ) -> std::io::Result<()> { - tokio::time::sleep(close_wait_timeout).await; // Wait CLOSE_WAIT_TIMEOUT for upstream + // Wait until a whole CLOSE_WAIT_TIMEOUT passes with nothing new sent or acknowledged, so a + // reply still on its way is not cut. + let progress = |tcb: &Tcb| (tcb.get_seq(), tcb.get_last_received_ack()); + let mut last = progress(&tcb.lock().unwrap()); + loop { + tokio::time::sleep(close_wait_timeout).await; + let tcb = tcb.lock().unwrap(); + let now = progress(&tcb); + if now == last || tcb.get_state() != TcpState::CloseWait { + break; + } + last = now; + } let tcb_clone = tcb.clone(); let mut tcb = tcb.lock().unwrap(); let state = tcb.get_state(); @@ -683,6 +745,7 @@ async fn tcp_main_logic_loop( _ = 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(()); + write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); break; } _ = drain_notify.notified() => { @@ -830,55 +893,29 @@ async fn tcp_main_logic_loop( write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); } if tcb.get_ack() == incoming_seq + len as u32 { - // The other side is closing the connection, we need to send an ACK and change state to CloseWait + // RFC 9293 §3.6: the peer has closed its direction only. Acknowledge its FIN and + // let the reader reach end of stream, but keep sending: our FIN waits for the + // application to shut down or drop the stream. tcb.increase_ack(); write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK, None, None)?; tcb.change_state(TcpState::CloseWait); - - let s = tcb.get_state(); + read_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); + write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); let len = tcb.get_inflight_packets_total_len(); - if len == 0 { - // All upstream data sent, proceed to LastAck - log::trace!("{network_tuple} {s:?}: {l_info}, {pkt_type:?}, closed by the other side, no upstream data"); - - // Here we don't wait, just send FIN to the other side and change state to LastAck directly, - write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK | FIN, None, None)?; - tcb.increase_seq(); - tcb.change_state(TcpState::LastAck); - - let s = tcb.get_state(); - log::trace!("{network_tuple} {s:?}: {l_info}, {pkt_type:?}, wait the last ack from the other side"); - - // Here we set a timer to wait for the last ACK from the other side. - // If the timer expires, we send an ACK|FIN packet to the other side again and wait anthoer timeout - // till the retries reach the limit, and then close the session forcibly. - let up = up_packet_sender.clone(); - tokio::spawn(task_last_ack( - tcb_clone.clone(), - exit_notifier, - network_tuple, - up, - config.last_ack_timeout, - config.last_ack_max_retries, - )); - } else { - // Upstream data pending, wake write_notify and wait - write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); - log::debug!("{network_tuple} {state:?}: Waiting for upstream data to complete, inflight packets: {len}",); - - // Spawn a timeout task to force FIN if upstream is unresponsive - let tcb = tcb_clone.clone(); - let up = up_packet_sender.clone(); - tokio::spawn(task_timed_out_for_close_wait( - tcb, - exit_notifier, - network_tuple, - up, - config.close_wait_timeout, - config.last_ack_timeout, - config.last_ack_max_retries, - )); - } + log::trace!("{network_tuple} CloseWait: {l_info}, closed by the other side, inflight: {len}"); + + // Spawn a timeout task to force FIN if upstream stays idle without closing + let tcb = tcb_clone.clone(); + let up = up_packet_sender.clone(); + tokio::spawn(task_timed_out_for_close_wait( + tcb, + exit_notifier, + network_tuple, + up, + config.close_wait_timeout, + config.last_ack_timeout, + config.last_ack_max_retries, + )); } } else if flags == (ACK | PSH) && pkt_type == PacketType::NewPacket { // PSH changes nothing about where a segment belongs: one that arrives out of order @@ -901,28 +938,25 @@ async fn tcp_main_logic_loop( } } TcpState::CloseWait => { - if flags & ACK == ACK && tcb.get_inflight_packets_total_len() == 0 { - write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK | FIN, None, None)?; - tcb.increase_seq(); - tcb.change_state(TcpState::LastAck); - let new_state = tcb.get_state(); - log::trace!("{network_tuple} {state:?}: Received ACK|FIN, transitioned to {new_state:?}"); - - // Here we set a timer to wait for the last ACK from the other side. - // If the timer expires, we send an ACK|FIN packet to the other side again and wait anthoer timeout - // till the retries reach the limit, and then close the session forcibly. - let up = up_packet_sender.clone(); - tokio::spawn(task_last_ack( - tcb_clone.clone(), - exit_notifier, - network_tuple, - up, - config.last_ack_timeout, - config.last_ack_max_retries, - )); - } else { - write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); + // Our direction is still open: the peer acknowledges what we send, asks for what it + // lost, and retransmits its FIN, or the segment carrying it, when our ACK was lost. + match pkt_type { + PacketType::KeepAlive | PacketType::NewPacket => { + write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK, None, None)?; + } + PacketType::RetransmissionRequest => { + if let Some(packet) = tcb.find_inflight_packet(incoming_ack) { + let (s, p) = (packet.seq, packet.payload.clone()); + log::debug!( + "{network_tuple} {state:?}: {l_info}, retransmission request, seq = {s}, len = {}", + p.len() + ); + write_packet_to_device(&up_packet_sender, network_tuple, &tcb, None, ACK | PSH, Some(s), Some(p))?; + } + } + _ => {} } + write_notify.lock().unwrap().take().map(|w| w.wake_by_ref()).unwrap_or(()); } TcpState::LastAck if flags & ACK == ACK => { tcb.change_state(TcpState::Closed); @@ -1437,6 +1471,381 @@ mod tests { ); } + /// The sequence numbers of the segments carrying FIN that the stack wrote to the device so far. + async fn fins_written(up_rx: &mut PacketReceiver) -> Vec { + written(up_rx) + .await + .iter() + .map(tcp_header_of) + .filter(|h| h.fin) + .map(|h| h.sequence_number) + .collect() + } + + /// Lets the session task and the timers it spawned run until they wait again. + async fn settle() { + for _ in 0..4 { + tokio::task::yield_now().await; + } + } + + /// Advances the clock by `by` in steps, letting each timer that falls due run in its turn. + async fn pass(by: Duration) { + let step = Duration::from_millis(100); + let mut left = by; + while !left.is_zero() { + let by = step.min(left); + tokio::time::advance(by).await; + settle().await; + left -= by; + } + } + + /// Polls `stream` for a read once, giving the bytes it read, none at end of stream. + fn read_once(stream: &mut IpStackTcpStream, cx: &mut Context<'_>) -> Poll>> { + let mut storage = [0u8; 64]; + let mut buf = tokio::io::ReadBuf::new(&mut storage); + Pin::new(stream).poll_read(cx, &mut buf).map_ok(|()| buf.filled().to_vec()) + } + + /// Reads `stream` to its end, which a half-closed connection reaches without waiting on a timer. + async fn read_to_end_now(stream: &mut IpStackTcpStream) -> Vec { + let mut data = Vec::new(); + let read = tokio::io::AsyncReadExt::read_to_end(stream, &mut data); + tokio::time::timeout(Duration::from_millis(100), read) + .await + .expect("no end of stream after the peer's FIN") + .unwrap(); + data + } + + /// RFC 9293 §3.6: a FIN closes only the peer's direction. The stack acknowledges it and keeps + /// sending, and its own FIN waits for the application to shut down, which takes it to LastAck. + #[tokio::test(start_paused = true)] + async fn peer_fin_leaves_the_write_side_open() { + let (mut stream, mut up_rx, local_seq) = established().await; + let (woken, waker) = recording_waker(); + let mut cx = Context::from_waker(&waker); + + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + let packets = written(&mut up_rx).await; + assert!( + packets.iter().all(|p| !tcp_header_of(p).fin), + "the peer's FIN was answered with ours" + ); + assert_eq!(packets.last().map(|p| tcp_header_of(p).acknowledgment_number), Some(1002)); + assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::CloseWait); + + // The reply still flows, and its acknowledgement does not close the connection. + stream.write_all(b"reply").await.unwrap(); + let sent = up_rx.recv().await.unwrap(); + assert_eq!(sent.payload.as_deref(), Some(&b"reply"[..])); + stream.stream_sender().send(peer_ack(1002, local_seq + 5)).unwrap(); + assert!( + fins_written(&mut up_rx).await.is_empty(), + "the stack closed with the application still open" + ); + + // Shutting down sends our FIN; nothing may be written after it, and reading stays at its end. + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending()); + assert_eq!(fins_written(&mut up_rx).await, [local_seq + 5]); + assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::LastAck); + let write = Pin::new(&mut stream).poll_write(&mut cx, b"late"); + assert!(matches!(write, Poll::Ready(Err(ref e)) if e.kind() == BrokenPipe), "{write:?}"); + assert!(matches!(read_once(&mut stream, &mut cx), Poll::Ready(Ok(ref d)) if d.is_empty())); + + // The peer acknowledges our FIN, and the shutdown completes. + stream.stream_sender().send(peer_ack(1002, local_seq + 6)).unwrap(); + written(&mut up_rx).await; + assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::Closed); + settle().await; + assert!(woken.load(std::sync::atomic::Ordering::SeqCst), "the shutdown was not woken"); + assert!(matches!(Pin::new(&mut stream).poll_shutdown(&mut cx), Poll::Ready(Ok(())))); + } + + /// After the peer's FIN the reader gets every byte sent before it, then end of stream. + #[tokio::test(start_paused = true)] + async fn peer_fin_ends_the_stream_after_its_data() { + let (mut stream, mut up_rx, local_seq) = established().await; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | PSH, u16::MAX, b"abc")) + .unwrap(); + stream + .stream_sender() + .send(peer_segment(1004, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + written(&mut up_rx).await; + + assert_eq!(read_to_end_now(&mut stream).await, b"abc"); + // End of stream holds on every later read. + let mut cx = Context::from_waker(Waker::noop()); + assert!(matches!(read_once(&mut stream, &mut cx), Poll::Ready(Ok(ref d)) if d.is_empty())); + } + + /// A writer held by a closed peer window is woken by the peer's FIN, which can reopen it. + #[tokio::test(start_paused = true)] + async fn peer_fin_reopening_the_window_wakes_the_writer() { + let (mut stream, mut up_rx, local_seq) = established().await; + stream.stream_sender().send(peer_segment(1001, local_seq, ACK, 0, &[])).unwrap(); + written(&mut up_rx).await; + + let (woken, waker) = recording_waker(); + let mut cx = Context::from_waker(&waker); + assert!(Pin::new(&mut stream).poll_write(&mut cx, b"reply").is_pending()); + + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + written(&mut up_rx).await; + assert!( + woken.load(std::sync::atomic::Ordering::SeqCst), + "the peer's FIN did not wake the writer" + ); + assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, b"reply"), Poll::Ready(Ok(5)))); + } + + /// A reader waiting for data is woken by the peer's FIN and gets end of stream. + #[tokio::test(start_paused = true)] + async fn peer_fin_wakes_a_waiting_reader() { + let (mut stream, mut up_rx, local_seq) = established().await; + let (woken, waker) = recording_waker(); + let mut cx = Context::from_waker(&waker); + assert!(read_once(&mut stream, &mut cx).is_pending()); + + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + written(&mut up_rx).await; + assert!( + woken.load(std::sync::atomic::Ordering::SeqCst), + "the peer's FIN did not wake the reader" + ); + assert!(matches!(read_once(&mut stream, &mut cx), Poll::Ready(Ok(ref d)) if d.is_empty())); + } + + /// Data acknowledged but not yet handed to a reader that fell behind still reaches it, in order, + /// before end of stream. + #[tokio::test(start_paused = true)] + async fn peer_fin_ends_the_stream_after_data_the_reader_fell_behind_on() { + let (mut stream, mut up_rx, local_seq) = established().await; + // Three segments, one more than the handoff channel holds, then the FIN. + for (i, byte) in [1u8, 2, 3].into_iter().enumerate() { + let seq = 1001 + 4 * i as u32; + stream + .stream_sender() + .send(peer_segment(seq, local_seq, ACK | PSH, u16::MAX, &[byte; 4])) + .unwrap(); + } + stream + .stream_sender() + .send(peer_segment(1013, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + assert_eq!(acks_written(&mut up_rx).await.last(), Some(&1014)); + + assert_eq!(read_to_end_now(&mut stream).await, [[1; 4], [2; 4], [3; 4]].concat()); + } + + /// The peer retransmits its FIN, or the segment carrying it, when our ACK was lost; each copy is + /// acknowledged again, and none of them closes our direction. + #[tokio::test(start_paused = true)] + async fn retransmitted_peer_fin_is_acknowledged_again() { + let (stream, mut up_rx, local_seq) = established().await; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, b"tail")) + .unwrap(); + assert_eq!(acks_written(&mut up_rx).await.last(), Some(&1006)); + + // The whole segment again, then the FIN alone at its own sequence number. + for (seq, payload) in [(1001, &b"tail"[..]), (1005, &b""[..])] { + stream + .stream_sender() + .send(peer_segment(seq, local_seq, ACK | FIN, u16::MAX, payload)) + .unwrap(); + let packets = written(&mut up_rx).await; + let acks: Vec<_> = packets.iter().map(|p| tcp_header_of(p).acknowledgment_number).collect(); + assert_eq!(acks, [1006], "seq {seq}: the retransmitted FIN was not acknowledged"); + assert!(packets.iter().all(|p| !tcp_header_of(p).fin), "seq {seq}: answered with our FIN"); + } + assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::CloseWait); + } + + /// After the peer's FIN, duplicate ACKs still ask for a fast retransmission of what we sent. + #[tokio::test(start_paused = true)] + async fn duplicate_acks_after_the_peer_fin_retransmit() { + let (mut stream, mut up_rx, local_seq) = established().await; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + written(&mut up_rx).await; + stream.write_all(b"reply").await.unwrap(); + written(&mut up_rx).await; + + let mut payloads = Vec::new(); + for _ in 0..MAX_COUNT_FOR_DUP_ACK { + stream.stream_sender().send(peer_ack(1002, local_seq)).unwrap(); + payloads.extend( + written(&mut up_rx) + .await + .into_iter() + .filter_map(|p| p.payload.filter(|p| !p.is_empty())), + ); + } + assert_eq!(payloads, [b"reply".to_vec()], "the duplicate ACKs got no retransmission"); + } + + /// A shutdown with data still in flight sends our FIN once that data is acknowledged. + #[tokio::test(start_paused = true)] + async fn shutdown_after_the_peer_fin_waits_for_data_in_flight() { + let (mut stream, mut up_rx, local_seq) = established().await; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + written(&mut up_rx).await; + stream.write_all(b"reply").await.unwrap(); + written(&mut up_rx).await; + + let (woken, waker) = recording_waker(); + let mut cx = Context::from_waker(&waker); + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending()); + assert!( + fins_written(&mut up_rx).await.is_empty(), + "our FIN went out ahead of unacknowledged data" + ); + + stream.stream_sender().send(peer_ack(1002, local_seq + 5)).unwrap(); + written(&mut up_rx).await; + assert!( + woken.load(std::sync::atomic::Ordering::SeqCst), + "the acknowledgement did not wake the shutdown" + ); + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending()); + assert_eq!(fins_written(&mut up_rx).await, [local_seq + 5]); + } + + /// Our FIN, sent by a shutdown after the peer's, is retransmitted until the peer acknowledges it. + #[tokio::test(start_paused = true)] + async fn our_fin_after_the_peer_fin_is_retransmitted() { + let (mut stream, mut up_rx, local_seq) = established().await; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + written(&mut up_rx).await; + + let mut cx = Context::from_waker(Waker::noop()); + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending()); + assert_eq!(fins_written(&mut up_rx).await.len(), 1); + + tokio::time::advance(TcpConfig::default().last_ack_timeout).await; + assert_eq!(fins_written(&mut up_rx).await.len(), 1, "our FIN was not retransmitted"); + } + + /// A stream dropped after the peer's FIN sends ours. + #[tokio::test(start_paused = true)] + async fn dropping_the_stream_after_the_peer_fin_sends_ours() { + let (stream, mut up_rx, local_seq) = established().await; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + assert!(fins_written(&mut up_rx).await.is_empty()); + + drop(stream); + assert_eq!(fins_written(&mut up_rx).await, [local_seq]); + } + + /// An application that never shuts down after the peer's FIN gets our FIN sent for it once + /// `close_wait_timeout` passes with nothing written and nothing acknowledged. + #[tokio::test(start_paused = true)] + async fn idle_close_wait_sends_our_fin_at_the_timeout() { + let (stream, mut up_rx, local_seq) = established().await; + let timeout = TcpConfig::default().close_wait_timeout; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + settle().await; + written(&mut up_rx).await; + + pass(timeout - Duration::from_millis(1)).await; + assert!(fins_written(&mut up_rx).await.is_empty()); + pass(Duration::from_millis(1)).await; + assert_eq!( + fins_written(&mut up_rx).await, + [local_seq], + "no FIN after an idle close_wait_timeout" + ); + } + + /// A reply still moving after the peer's FIN is not cut by `close_wait_timeout`, which counts only + /// time without progress. + #[tokio::test(start_paused = true)] + async fn close_wait_timeout_does_not_cut_a_reply_in_progress() { + let (mut stream, mut up_rx, local_seq) = established().await; + let timeout = TcpConfig::default().close_wait_timeout; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + settle().await; + written(&mut up_rx).await; + + // Within every `close_wait_timeout` a byte is written, and then the peer acknowledges one. + let step = timeout - Duration::from_secs(1); + for i in 1..=2 { + pass(step).await; + stream.write_all(b"x").await.unwrap(); + assert!( + fins_written(&mut up_rx).await.is_empty(), + "write {i}: the reply was cut while written" + ); + } + for i in 1..=2 { + pass(step).await; + stream.stream_sender().send(peer_ack(1002, local_seq + i)).unwrap(); + assert!( + fins_written(&mut up_rx).await.is_empty(), + "ack {i}: the reply was cut while acknowledged" + ); + } + + // Once it stops, our FIN follows within two periods. + pass(timeout * 2).await; + assert_eq!(fins_written(&mut up_rx).await.first(), Some(&(local_seq + 2))); + } + + /// A writer waiting for window when the connection closes under it is woken to see the close. + #[tokio::test(start_paused = true)] + async fn writer_waiting_for_window_is_woken_when_the_connection_ends() { + let (mut stream, mut up_rx, local_seq) = established().await; + stream + .stream_sender() + .send(peer_segment(1001, local_seq, ACK | FIN, 0, &[])) + .unwrap(); + settle().await; + written(&mut up_rx).await; + + let (woken, waker) = recording_waker(); + let mut cx = Context::from_waker(&waker); + assert!(Pin::new(&mut stream).poll_write(&mut cx, b"reply").is_pending()); + + // The backstop sends our FIN, which is never acknowledged, and the connection ends. + let config = TcpConfig::default(); + pass(config.close_wait_timeout + config.last_ack_timeout * (config.last_ack_max_retries as u32 + 1)).await; + assert_eq!(stream.tcb.lock().unwrap().get_state(), TcpState::Closed); + assert!(woken.load(std::sync::atomic::Ordering::SeqCst), "the writer was left waiting"); + assert!(matches!(Pin::new(&mut stream).poll_write(&mut cx, b"reply"), Poll::Ready(Err(_)))); + } + /// 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] From 1968cbb89aa71c507d576ca00470e0e6c4859baa Mon Sep 17 00:00:00 2001 From: Musyg <223545376+Musyg@users.noreply.github.com> Date: Tue, 3 Nov 2026 23:02:54 +0100 Subject: [PATCH 3/3] Avoid late window ACKs and extend Linux close qualification --- docs/tcp-close-qualification.md | 74 ++++ scripts/qualify-linux-tcp-extended.sh | 27 ++ src/stream/tcp.rs | 240 ++++++++++++- tests/linux_tcp_close_extended.rs | 494 ++++++++++++++++++++++++++ 4 files changed, 833 insertions(+), 2 deletions(-) create mode 100644 docs/tcp-close-qualification.md create mode 100755 scripts/qualify-linux-tcp-extended.sh create mode 100644 tests/linux_tcp_close_extended.rs diff --git a/docs/tcp-close-qualification.md b/docs/tcp-close-qualification.md new file mode 100644 index 0000000..3ed981e --- /dev/null +++ b/docs/tcp-close-qualification.md @@ -0,0 +1,74 @@ +# TCP close qualification + +This work integrates frontierlabs-admin's half-close changes from PR #112 and preserves SajjadPourali's shutdown wakeup fix from PR #114, following Musyg's report in issue #89. + +## Running the tests + +Build and run the ordinary tests first. This also generates Cargo.lock in a fresh checkout: + +```sh +cargo test +cargo fmt --check +cargo clippy --locked --all-targets --all-features -- -D warnings +bash scripts/qualify-linux-tcp.sh +bash scripts/qualify-linux-tcp-extended.sh +``` + +The network tests require Linux, TUN, iproute2, Python 3, util-linux, and permission to run `sudo -n unshare --net`. Builds run as the ordinary user. Every network profile runs in a separate network namespace. No host routes or host qdiscs are changed. The integration tests are ignored during ordinary `cargo test` and run explicitly by these scripts. + +## Coverage + +The original matrix has 32 cases: four application close paths, two payload sizes, with and without losing the first ipstack FIN, under clean and delay/loss profiles. + +The extended matrix has 264 cases, 88 per profile: clean, 10 ms delay with 1% random loss, and 10 ms delay with 25% reordering and 50% correlation. Impairments apply in the Linux-to-ipstack direction. Each profile tests IPv4 and IPv6, with 5-byte and 128-KiB payloads in both directions. Payload bytes vary by position so misordering or duplication cannot be hidden by constant data. + +| Scenario | Assertion beyond complete payload delivery | +| --- | --- | +| Client-first, server-first, application overlap, and drop | Bounded completion with no unexpected RST | +| Lose one, two, or three ipstack FIN attempts | Every requested loss occurs, a later FIN arrives at the same sequence, and its ACK is observed | +| Lose the kernel's final ACK of ipstack FIN | ipstack retransmits FIN and a subsequent kernel ACK is observed | +| Lose ipstack's final ACK of kernel FIN | Linux retransmits FIN and ipstack answers it | +| Swap two incoming data segments | The swap occurs and the complete varied payload is delivered in order | +| Force crossed FINs | ipstack FIN is retained until Linux sends its own FIN without having seen it; release waits for ipstack's ACK of the kernel FIN | + +ACK-loss cases are applied to the close paths where that direction carries the final ACK. Deterministic data swaps use 128-KiB transfers. FIN loss applies to the connection's client port, not to late packets from earlier cases. FIN retransmissions preserve the FIN sequence even when the original FIN carries data. ACK observations include ACKs carried by data or FIN; deterministic ACK loss targets a pure ACK. The crossed-FIN gate also retains retransmissions until release. + +Every extended case waits for both FINs and their acknowledgements, including cases without injected loss. The fault oracle also waits for recovery packets after application EOF. `shutdown()` or EOF alone cannot establish that a lost final ACK was recovered. Every case has a 45-second deadline; each extended profile also has a 15-minute process deadline. + +A reordered duplicate FIN can reach ipstack after Linux has consumed the final ACK and closed. Linux may then reset ipstack's required ACK of that duplicate. These late resets are counted separately only when both FINs and ACKs have been observed, the reset is incoming and has the exact post-FIN sequence, the kernel FIN was seen more than once, and exactly one final-ACK attempt was made per observed kernel FIN. Other resets fail the case. This retains the distinction from an unsolicited extra window ACK. The TimeWait reset behavior follows [RFC 9293 section 3.10.7.4](https://www.rfc-editor.org/rfc/rfc9293.html#section-3.10.7.4); a dedicated unit test checks duplicate-FIN acknowledgement followed by silent closure on RST. + +For a targeted diagnostic case: + +```sh +IPSTACK_EXT_CASE='v6/server-first/131072/AckToKernel' \ + bash scripts/qualify-linux-tcp-extended.sh clean +``` + +A targeted run does not qualify the full matrix. Qualification evidence must contain all 264 unique extended case lines and all 32 original case lines. + +## Lifecycle and timer tests + +Unit tests additionally cover sequence wraparound, invalid ACKs, partial ACKs, independent shutdown waiters, CloseWait data preservation, TimeWait restart, detached cleanup, and FinWait2 expiry. New boundary tests cover: + +- Drop with no entered Tokio runtime: immediate RST, closed state, and worker cancellation. +- Active detached close: wait for data ACKs, local FIN ACK, peer FIN, and TimeWait before destruction. +- Receive progress renewing the FinWait2 inactivity deadline. +- FIN retry counts of zero, one, and three in FinWait1, Closing, and LastAck, with configured floors and an RTO above the ordinary cap. This is 27 configuration/state combinations within one unit test. +- A delayed reader notification after the final ACK, while unread data remains buffered. +- 256 independent streams established simultaneously with data outstanding, released by a barrier to race shutdown against peer FIN, followed by explicit worker-destruction checks on a four-thread runtime. This differs from the older test, which races two tasks but processes sessions sequentially. + +## Defect found by the extended matrix + +The server-first 128-KiB scenario with repeated FIN loss exposed an extra window-update ACK after ipstack had acknowledged Linux's FIN. Linux had already closed and replied with RST. The application had received all bytes, so an EOF-only test missed this failure. + +A delayed reader-drain notification could call the data extraction helper in TimeWait and send that extra ACK. The worker now processes window-update notifications only while the receive direction remains open. After the peer FIN, `poll_read` drains the remaining bytes without advertising another window. A deterministic unit regression fails before this fix, and the Linux packet-level oracle checks for the resulting RST. + +## Scope of the result + +On Phoenix5-Orion (Ubuntu 24.04.5, Linux 7.0.0-42-generic, Rust 1.95.0), the final run passed all 296 Linux cases, 101 unit tests, 23 doctests, formatting, Clippy with warnings denied, and diff checks. The extended matrix contains 132 IPv4 and 132 IPv6 cases. The loss profile recorded 56 actual packet drops. No unexpected resets occurred; one late closed-peer reset met the strict classification above. + +Twenty repetitions of the simultaneous 256-stream test passed, totalling 5,120 streams. Separately, twenty repetitions of the older per-session task-race test passed, totalling another 5,120 sessions. These stress runs use the in-process packet harness; the 296 integration cases exercise Linux kernel TCP through TUN. The evidence bundle retains the failing pre-fix regression, intermediate diagnostics, final logs, exact dependency lockfile, and a matrix auditor. + +The qualification concerns these close paths, faults, payload sizes, and configuration boundaries. It is not exhaustive TCP conformance. The implementation uses a configured RTO and exponential backoff; it does not implement a new adaptive RTT estimator. Packet parsing in the test fault injector covers the IPv4 and IPv6 TCP traffic generated here, not arbitrary IPv6 extension-header chains. + +The lifetime after drop, the use of `config.timeout`, and FIN retry limits remain API decisions for maintainer review even when their implemented behavior passes the tests. diff --git a/scripts/qualify-linux-tcp-extended.sh b/scripts/qualify-linux-tcp-extended.sh new file mode 100755 index 0000000..6c03611 --- /dev/null +++ b/scripts/qualify-linux-tcp-extended.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash +set -euo pipefail +cd -- "$(dirname -- "$0")/.." +build_log=$(mktemp) +trap 'rm -f -- "$build_log"' EXIT +cargo test --locked --test linux_tcp_close_extended --no-run --message-format=json > "$build_log" +binary=$(python3 - "$build_log" <<'PY' +import json, sys +for line in open(sys.argv[1]): + item = json.loads(line) + if item.get('reason') == 'compiler-artifact' and item.get('target', {}).get('name') == 'linux_tcp_close_extended' and item.get('executable'): + print(item['executable']) +PY +) +test -n "$binary" +profiles=(clean netem reorder) +if [[ $# -gt 0 ]]; then profiles=("$@"); fi +for profile in "${profiles[@]}"; do + echo "Running extended Linux TCP matrix: $profile" + timeout 15m sudo -n unshare --net -- bash -euo pipefail -c ' + ip link set lo up + export IPSTACK_TEST_NETNS=1 IPSTACK_EXT_PROFILE="$2" + export IPSTACK_EXT_CASE="$3" RUST_LOG="$4" + exec "$1" --ignored --nocapture + ' bash "$binary" "$profile" "${IPSTACK_EXT_CASE:-}" "${RUST_LOG:-}" +done +echo EXTENDED_NETWORK_MATRIX=PASS diff --git a/src/stream/tcp.rs b/src/stream/tcp.rs index a4541ef..e5b0435 100644 --- a/src/stream/tcp.rs +++ b/src/stream/tcp.rs @@ -814,7 +814,12 @@ async fn tcp_main_logic_loop( // 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)?; + // After the peer's FIN, poll_read drains the remaining bytes itself. + // A delayed notification must not send a window update after our final + // ACK: a peer already in Closed would answer it with RST. + if matches!(tcb.get_state(), TcpState::Established | TcpState::FinWait1 | TcpState::FinWait2) { + extract_data_n_write_upstream(&up_packet_sender, &mut tcb, network_tuple, &data_tx, &read_notify)?; + } continue; } _ = wait_retransmission_deadline(&tcb, &arm_notify) => { @@ -1364,12 +1369,19 @@ mod tests { } async fn established_with_destroy(destroy: Option>) -> (IpStackTcpStream, PacketReceiver, u32) { + established_with_config(destroy, TcpConfig::default()).await + } + + async fn established_with_config( + destroy: Option>, + config: TcpConfig, + ) -> (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, destroy, Arc::new(TcpConfig::default())).unwrap(); + let stream = IpStackTcpStream::new(peer, local, syn, up_tx, 1500, destroy, Arc::new(config)).unwrap(); let syn_ack = tokio::time::timeout(Duration::from_secs(1), up_rx.recv()) .await @@ -3016,4 +3028,228 @@ mod tests { assert_eq!(review_state(&stream), TcpState::Closed); assert!(written(&mut rx).await.iter().map(tcp_header_of).any(|h| h.rst)); } + + #[test] + fn complete_drop_outside_runtime_resets_and_removes_session() { + let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + let (destroy_tx, destroyed) = tokio::sync::oneshot::channel(); + let (stream, mut rx, _) = runtime.block_on(established_with_destroy(Some(destroy_tx))); + let tcb = stream.tcb.clone(); + assert!(tokio::runtime::Handle::try_current().is_err()); + drop(stream); + assert_eq!(tcb.lock().unwrap().get_state(), TcpState::Closed); + let packet = rx.try_recv().expect("outside-runtime drop must enqueue RST immediately"); + assert!(tcp_header_of(&packet).rst); + assert!(!tcp_header_of(&packet).fin); + runtime.block_on(async { + assert!(tokio::time::timeout(Duration::from_secs(1), destroyed).await.unwrap().is_err()); + }); + } + + #[tokio::test(start_paused = true)] + async fn complete_delayed_reader_notification_does_not_ack_after_final_ack() { + let (mut stream, mut rx, seq) = established().await; + let mut cx = Context::from_waker(Waker::noop()); + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending()); + written(&mut rx).await; + stream.stream_sender().send(peer_ack(1001, seq + 1)).unwrap(); + settle().await; + for i in 0..3u32 { + stream + .stream_sender() + .send(peer_segment(1001 + i * 4, seq + 1, ACK | PSH, u16::MAX, &[i as u8; 4])) + .unwrap(); + } + stream + .stream_sender() + .send(peer_segment(1013, seq + 1, ACK | FIN, u16::MAX, &[])) + .unwrap(); + assert_eq!(acks_written(&mut rx).await.last(), Some(&1014)); + assert_eq!(review_state(&stream), TcpState::TimeWait); + // A notification from an earlier read can be selected after processing FIN. + stream.drain_notify.notify_one(); + assert!( + written(&mut rx).await.is_empty(), + "reader drain sent an ACK after the final FIN ACK" + ); + assert_eq!(read_to_end_now(&mut stream).await, [[0; 4], [1; 4], [2; 4]].concat()); + } + + #[tokio::test(start_paused = true)] + async fn complete_time_wait_duplicate_fin_then_rst_closes_without_reply() { + let (mut stream, mut rx, seq) = established().await; + let mut cx = Context::from_waker(Waker::noop()); + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending()); + written(&mut rx).await; + for _ in 0..2 { + stream + .stream_sender() + .send(peer_segment(1001, seq + 1, ACK | FIN, u16::MAX, &[])) + .unwrap(); + assert_eq!(acks_written(&mut rx).await, [1002]); + assert_eq!(review_state(&stream), TcpState::TimeWait); + } + // A peer already in Closed may reset our ACK of its delayed duplicate FIN. + // RFC 9293 section 3.10.7.4 closes TimeWait on receipt of RST. + stream.stream_sender().send(peer_segment(1002, 0, RST, 0, &[])).unwrap(); + assert!(written(&mut rx).await.is_empty()); + assert_eq!(review_state(&stream), TcpState::Closed); + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_ready()); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn complete_parallel_sessions_close_and_release_workers() { + let start = Arc::new(tokio::sync::Barrier::new(257)); + let mut sessions = tokio::task::JoinSet::new(); + for iteration in 0..256 { + let start = start.clone(); + sessions.spawn(async move { + let (destroy_tx, destroyed) = tokio::sync::oneshot::channel(); + let (mut stream, mut rx, seq) = established_with_destroy(Some(destroy_tx)).await; + stream.write_all(b"hello").await.unwrap(); + assert_eq!(rx.recv().await.unwrap().payload.as_deref(), Some(&b"hello"[..])); + let sender = stream.stream_sender(); + // All 256 streams exist with outstanding data before any closes. + start.wait().await; + let peer = tokio::spawn(async move { + if iteration % 2 == 0 { + tokio::task::yield_now().await; + } + sender + .send(peer_segment(1001, seq + 5, ACK | FIN | PSH, u16::MAX, b"tail")) + .unwrap(); + loop { + let packet = rx.recv().await.expect("stream closed before FIN"); + if tcp_header_of(&packet).fin { + assert_eq!(tcp_header_of(&packet).sequence_number, seq + 5); + assert_eq!(tcp_header_of(&packet).acknowledgment_number, 1006); + sender.send(peer_ack(1006, seq + 6)).unwrap(); + break; + } + } + }); + if iteration % 2 != 0 { + tokio::task::yield_now().await; + } + stream.shutdown().await.unwrap(); + peer.await.unwrap(); + assert_eq!(read_to_end_now(&mut stream).await, b"tail"); + destroyed.await.expect("worker aborted instead of completing"); + }); + } + tokio::time::timeout(Duration::from_secs(10), async { + start.wait().await; + let mut completed = 0; + while let Some(result) = sessions.join_next().await { + result.unwrap(); + completed += 1; + } + assert_eq!(completed, 256); + }) + .await + .expect("parallel close or worker cleanup hung"); + } + + #[tokio::test(start_paused = true)] + async fn complete_drop_active_close_waits_for_both_fin_directions() { + let (destroy_tx, mut destroyed) = tokio::sync::oneshot::channel(); + let (mut stream, mut rx, seq) = established_with_destroy(Some(destroy_tx)).await; + let sender = stream.stream_sender(); + let tcb = stream.tcb.clone(); + stream.write_all(b"reply").await.unwrap(); + written(&mut rx).await; + drop(stream); + settle().await; + assert!(fins_written(&mut rx).await.is_empty()); + sender.send(peer_ack(1001, seq + 5)).unwrap(); + assert_eq!(fins_written(&mut rx).await, [seq + 5]); + sender.send(peer_ack(1001, seq + 6)).unwrap(); + settle().await; + assert_eq!(tcb.lock().unwrap().get_state(), TcpState::FinWait2); + assert!(destroyed.try_recv().is_err()); + sender.send(peer_segment(1001, seq + 6, ACK | FIN, u16::MAX, &[])).unwrap(); + written(&mut rx).await; + assert_eq!(tcb.lock().unwrap().get_state(), TcpState::TimeWait); + pass(TcpConfig::default().two_msl - Duration::from_millis(1)).await; + assert!(destroyed.try_recv().is_err()); + pass(Duration::from_millis(1)).await; + assert_eq!(destroyed.try_recv(), Ok(())); + assert_eq!(tcb.lock().unwrap().get_state(), TcpState::Closed); + } + + #[tokio::test(start_paused = true)] + async fn complete_fin_wait2_data_renews_inactivity_deadline() { + let config = TcpConfig { + timeout: Duration::from_millis(100), + ..TcpConfig::default() + }; + let (mut stream, mut rx, seq) = established_with_config(None, config).await; + let mut cx = Context::from_waker(Waker::noop()); + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending()); + written(&mut rx).await; + stream.stream_sender().send(peer_ack(1001, seq + 1)).unwrap(); + settle().await; + pass(Duration::from_millis(75)).await; + stream + .stream_sender() + .send(peer_segment(1001, seq + 1, ACK | PSH, u16::MAX, b"x")) + .unwrap(); + written(&mut rx).await; + pass(Duration::from_millis(26)).await; + assert_eq!(review_state(&stream), TcpState::FinWait2, "original deadline was not renewed"); + pass(Duration::from_millis(73)).await; + assert_eq!(review_state(&stream), TcpState::FinWait2); + pass(Duration::from_millis(1)).await; + assert_eq!(review_state(&stream), TcpState::Closed); + assert!(written(&mut rx).await.iter().map(tcp_header_of).any(|h| h.rst)); + } + + #[tokio::test(start_paused = true)] + async fn complete_fin_retry_limits_and_floors_in_all_closing_states() { + for state in [TcpState::FinWait1, TcpState::Closing, TcpState::LastAck] { + for retries in [0, 1, 3] { + // Explicit expected intervals include a configured floor above the + // data RTO and a data RTO above the ordinary 60-second ceiling. + for (rto, floor_ms, intervals) in [(1, 500, [1, 2, 4, 8]), (2, 3000, [3, 6, 12, 24]), (120, 500, [120; 4])] { + let config = TcpConfig { + rto: Duration::from_secs(rto), + last_ack_timeout: Duration::from_millis(floor_ms), + last_ack_max_retries: retries, + ..TcpConfig::default() + }; + let (mut stream, mut rx, seq) = established_with_config(None, config).await; + if state == TcpState::LastAck { + stream + .stream_sender() + .send(peer_segment(1001, seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + written(&mut rx).await; + } + let mut cx = Context::from_waker(Waker::noop()); + assert!(Pin::new(&mut stream).poll_shutdown(&mut cx).is_pending()); + assert_eq!(fins_written(&mut rx).await, [seq]); + if state == TcpState::Closing { + stream + .stream_sender() + .send(peer_segment(1001, seq, ACK | FIN, u16::MAX, &[])) + .unwrap(); + written(&mut rx).await; + } + assert_eq!(review_state(&stream), state); + for seconds in intervals.iter().take(retries) { + pass(Duration::from_secs(*seconds) - Duration::from_millis(1)).await; + assert!(fins_written(&mut rx).await.is_empty()); + assert_eq!(review_state(&stream), state); + pass(Duration::from_millis(1)).await; + assert_eq!(fins_written(&mut rx).await, [seq]); + } + pass(Duration::from_secs(intervals[retries]) - Duration::from_millis(1)).await; + assert_eq!(review_state(&stream), state); + pass(Duration::from_millis(1)).await; + assert_eq!(review_state(&stream), TcpState::Closed); + assert!(fins_written(&mut rx).await.is_empty()); + } + } + } + } } diff --git a/tests/linux_tcp_close_extended.rs b/tests/linux_tcp_close_extended.rs new file mode 100644 index 0000000..2d783e3 --- /dev/null +++ b/tests/linux_tcp_close_extended.rs @@ -0,0 +1,494 @@ +#![cfg(target_os = "linux")] +//! Deterministic close-handshake faults against real IPv4 and IPv6 Linux sockets. +use ipstack::{IpStack, IpStackConfig, IpStackStream}; +use std::{ + io, + pin::Pin, + sync::{Arc, Mutex}, + task::{Context, Poll}, + time::Duration, +}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf}; + +#[derive(Clone, Copy, Debug, PartialEq)] +enum Fault { + None, + Fins(usize), + AckToStack, + AckToKernel, + ReorderData, + CrossFins, +} + +#[derive(Clone, Copy)] +struct Packet { + version: u8, + src: u16, + dst: u16, + seq: u32, + ack: u32, + flags: u8, + len: usize, +} +impl Packet { + fn parse(bytes: &[u8]) -> Option { + let version = *bytes.first()? >> 4; + let offset = match version { + 4 if bytes.len() >= 20 && bytes[9] == 6 => usize::from(bytes[0] & 15) * 4, + 6 if bytes.len() >= 40 && bytes[6] == 6 => 40, + _ => return None, + }; + let tcp = bytes.get(offset..)?; + let header = usize::from(*tcp.get(12)? >> 4) * 4; + if header < 20 || tcp.len() < header { + return None; + } + Some(Self { + version, + src: u16::from_be_bytes(tcp[0..2].try_into().ok()?), + dst: u16::from_be_bytes(tcp[2..4].try_into().ok()?), + seq: u32::from_be_bytes(tcp[4..8].try_into().ok()?), + ack: u32::from_be_bytes(tcp[8..12].try_into().ok()?), + flags: tcp[13], + len: tcp.len() - header, + }) + } + fn fin(self) -> bool { + self.flags & 1 != 0 + } + fn pure_ack(self) -> bool { + self.flags == 0x10 && self.len == 0 + } +} + +struct Trace { + version: u8, + port: u16, + fault: Fault, + stack_fins: Vec, + kernel_fins: Vec, + dropped_fins: usize, + dropped_ack: usize, + stack_ack_passed: usize, + kernel_ack_passed: usize, + reordered_pairs: usize, + resets: usize, + late_resets: usize, + peer_fin_reply_attempts: usize, + crossed_peer_fin: bool, + released_cross_fin: bool, +} +impl Trace { + fn new(version: u8, port: u16, fault: Fault) -> Self { + Self { + version, + port, + fault, + stack_fins: vec![], + kernel_fins: vec![], + dropped_fins: 0, + dropped_ack: 0, + stack_ack_passed: 0, + kernel_ack_passed: 0, + reordered_pairs: 0, + resets: 0, + late_resets: 0, + peer_fin_reply_attempts: 0, + crossed_peer_fin: false, + released_cross_fin: false, + } + } + fn matches(&self, p: Packet, outgoing: bool) -> bool { + p.version == self.version + && if outgoing { + p.src == 18089 && p.dst == self.port + } else { + p.src == self.port && p.dst == 18089 + } + } + fn fin_ack(&self, p: Packet, outgoing: bool) -> bool { + let fins = if outgoing { &self.kernel_fins } else { &self.stack_fins }; + p.flags & 0x10 != 0 && fins.first().is_some_and(|seq| p.ack == seq.wrapping_add(1)) + } + fn should_drop(&self, p: Packet, outgoing: bool) -> bool { + if !self.matches(p, outgoing) { + return false; + } + match self.fault { + Fault::Fins(count) => outgoing && p.fin() && self.dropped_fins < count, + Fault::AckToStack => !outgoing && p.pure_ack() && self.dropped_ack == 0 && self.fin_ack(p, outgoing), + Fault::AckToKernel => outgoing && p.pure_ack() && self.dropped_ack == 0 && self.fin_ack(p, outgoing), + _ => false, + } + } + fn record(&mut self, p: Packet, outgoing: bool, dropped: bool) { + if !self.matches(p, outgoing) { + return; + } + if p.flags & 4 != 0 { + // A delayed duplicate FIN can arrive after Linux has consumed our + // final ACK and closed. Its RST in reply to the required duplicate + // FIN ACK is distinct from an unsolicited extra window ACK. + let late = !outgoing + && p.flags == 4 + && p.ack == 0 + && p.len == 0 + && self.kernel_fins.len() > 1 + && self.kernel_fins.first().is_some_and(|fin| p.seq == fin.wrapping_add(1)) + && !self.stack_fins.is_empty() + && self.stack_ack_passed > 0 + && self.kernel_ack_passed > 0 + && self.peer_fin_reply_attempts == self.kernel_fins.len(); + if late { + self.late_resets += 1; + } else { + self.resets += 1; + } + let label = if late { "LATE_CLOSED_PEER_RST" } else { "WIRE_RESET" }; + eprintln!( + "{label} outgoing={outgoing} v={} src={} dst={} seq={} ack={} flags={} len={} stack_fins={:?} kernel_fins={:?}", + p.version, p.src, p.dst, p.seq, p.ack, p.flags, p.len, self.stack_fins, self.kernel_fins + ); + } + if p.fin() { + if !outgoing && self.fault == Fault::CrossFins && !self.stack_fins.is_empty() && !self.released_cross_fin { + // The kernel sent its FIN before it could have seen ours. + self.crossed_peer_fin = true; + } + let seq = p.seq.wrapping_add(p.len as u32); + let fins = if outgoing { &mut self.stack_fins } else { &mut self.kernel_fins }; + if let Some(first) = fins.first() { + assert_eq!(*first, seq, "FIN retransmission changed sequence"); + } + fins.push(seq); + if dropped { + self.dropped_fins += 1; + } + } + if self.fin_ack(p, outgoing) { + if outgoing && p.pure_ack() { + self.peer_fin_reply_attempts += 1; + } + if dropped && !p.fin() { + self.dropped_ack += 1; + } else if !dropped && outgoing { + self.stack_ack_passed += 1; + } else if !dropped { + self.kernel_ack_passed += 1; + } + } + } + fn fault_recovered(&self) -> bool { + // EOF alone is insufficient, including in cases without injected loss. + if self.stack_fins.is_empty() || self.kernel_fins.is_empty() || self.stack_ack_passed == 0 || self.kernel_ack_passed == 0 { + return false; + } + match self.fault { + Fault::None => true, + Fault::Fins(n) => self.dropped_fins == n && self.stack_fins.len() > n && self.kernel_ack_passed > 0, + Fault::AckToStack => self.dropped_ack == 1 && self.stack_fins.len() >= 2 && self.kernel_ack_passed > 0, + Fault::AckToKernel => self.dropped_ack == 1 && self.kernel_fins.len() >= 2 && self.stack_ack_passed > 0, + Fault::ReorderData => self.reordered_pairs == 1, + Fault::CrossFins => self.crossed_peer_fin && self.released_cross_fin && self.kernel_ack_passed > 0, + } + } +} + +struct FaultTun { + device: tun::AsyncDevice, + trace: Arc>, + buffer: Vec, + held: Option>, + release_held: bool, + held_fin: Option>, +} +impl AsyncRead for FaultTun { + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, output: &mut ReadBuf<'_>) -> Poll> { + let this = self.get_mut(); + if this.release_held { + let bytes = this.held.take().unwrap(); + this.release_held = false; + output.put_slice(&bytes); + return Poll::Ready(Ok(())); + } + loop { + let mut buffer = ReadBuf::new(&mut this.buffer); + match Pin::new(&mut this.device).poll_read(cx, &mut buffer) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), + Poll::Ready(Ok(())) => {} + } + let bytes = buffer.filled(); + if let Some(packet) = Packet::parse(bytes) { + let mut trace = this.trace.lock().unwrap(); + let dropped = trace.should_drop(packet, false); + trace.record(packet, false, dropped); + if dropped { + continue; + } + if trace.fault == Fault::ReorderData + && trace.matches(packet, false) + && packet.len > 0 + && !packet.fin() + && trace.reordered_pairs == 0 + { + if this.held.is_none() { + this.held = Some(bytes.to_vec()); + continue; + } + trace.reordered_pairs += 1; + this.release_held = true; + cx.waker().wake_by_ref(); + } + } + output.put_slice(bytes); + return Poll::Ready(Ok(())); + } + } +} +impl AsyncWrite for FaultTun { + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bytes: &[u8]) -> Poll> { + let packet = Packet::parse(bytes); + if let Some(packet) = packet { + let hold = { + let trace = self.trace.lock().unwrap(); + trace.matches(packet, true) && trace.fault == Fault::CrossFins && packet.fin() && !trace.released_cross_fin + }; + if hold { + self.trace.lock().unwrap().record(packet, true, false); + self.held_fin = Some(bytes.to_vec()); + return Poll::Ready(Ok(bytes.len())); + } + let release = { + let trace = self.trace.lock().unwrap(); + self.held_fin.is_some() && trace.matches(packet, true) && trace.fin_ack(packet, true) && !packet.fin() + }; + if release { + let this = self.as_mut().get_mut(); + let held = this.held_fin.as_ref().unwrap(); + match Pin::new(&mut this.device).poll_write(cx, held) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), + Poll::Ready(Ok(size)) => { + assert_eq!(size, held.len()); + this.held_fin = None; + this.trace.lock().unwrap().released_cross_fin = true; + } + } + } + } + if let Some(packet) = packet { + let mut trace = self.trace.lock().unwrap(); + if trace.should_drop(packet, true) { + trace.record(packet, true, true); + return Poll::Ready(Ok(bytes.len())); + } + } + let result = Pin::new(&mut self.device).poll_write(cx, bytes); + if matches!(&result, Poll::Ready(Ok(n)) if *n == bytes.len()) + && let Some(packet) = packet + { + self.trace.lock().unwrap().record(packet, true, false); + } + result + } + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.device).poll_flush(cx) + } + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.device).poll_shutdown(cx) + } +} + +fn command(program: &str, args: &[&str]) { + assert!( + std::process::Command::new(program).args(args).status().unwrap().success(), + "{program} {args:?}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "requires TUN and an isolated Linux network namespace"] +async fn linux_kernel_extended_close_matrix() { + assert_eq!(std::env::var("IPSTACK_TEST_NETNS").as_deref(), Ok("1")); + let _ = env_logger::try_init(); + let profile = std::env::var("IPSTACK_EXT_PROFILE").unwrap_or_else(|_| "clean".into()); + let mut tun_config = tun::Configuration::default(); + tun_config + .tun_name("ipstack-ext") + .address((198, 18, 1, 1)) + .destination((198, 18, 1, 2)) + .netmask((255, 255, 255, 0)) + .mtu(1500) + .up(); + let trace = Arc::new(Mutex::new(Trace::new(4, 0, Fault::None))); + let device = FaultTun { + device: tun::create_as_async(&tun_config).unwrap(), + trace: trace.clone(), + buffer: vec![0; 65536], + held: None, + release_held: false, + held_fin: None, + }; + command("ip", &["-6", "addr", "add", "fd42:89:112::1/64", "dev", "ipstack-ext", "nodad"]); + let mut config = IpStackConfig::default(); + config.mtu(1500).unwrap(); + let mut stack = IpStack::new(config, device); + match profile.as_str() { + "clean" => {} + "netem" => command( + "tc", + &["qdisc", "add", "dev", "ipstack-ext", "root", "netem", "delay", "10ms", "loss", "1%"], + ), + "reorder" => command( + "tc", + &[ + "qdisc", + "add", + "dev", + "ipstack-ext", + "root", + "netem", + "delay", + "10ms", + "reorder", + "25%", + "50%", + ], + ), + _ => panic!("unknown profile {profile}"), + } + for version in [4, 6] { + for size in [5usize, 131_072] { + for mode in ["client-first", "server-first", "simultaneous", "drop", "crossed"] { + let mut faults = vec![Fault::None, Fault::Fins(1), Fault::Fins(2), Fault::Fins(3)]; + if matches!(mode, "client-first" | "drop") { + faults.push(Fault::AckToStack); + } + if mode == "server-first" { + faults.push(Fault::AckToKernel); + } + if size > 5 { + faults.push(Fault::ReorderData); + } + if mode == "crossed" { + faults = vec![Fault::CrossFins]; + } + for fault in faults { + let case = format!("v{version}/{mode}/{size}/{fault:?}"); + if let Ok(filter) = std::env::var("IPSTACK_EXT_CASE") + && !filter.is_empty() + && case != filter + { + continue; + } + let socket = if version == 4 { + tokio::net::TcpSocket::new_v4() + } else { + tokio::net::TcpSocket::new_v6() + } + .unwrap(); + let (local, remote) = if version == 4 { + ("198.18.1.1:0", "198.18.1.2:18089") + } else { + ("[fd42:89:112::1]:0", "[fd42:89:112::2]:18089") + }; + socket.bind(local.parse().unwrap()).unwrap(); + *trace.lock().unwrap() = Trace::new(version, socket.local_addr().unwrap().port(), fault); + let request: Vec = (0..size).map(|i| (i.wrapping_mul(31) ^ (i >> 8)) as u8).collect(); + let response: Vec = (0..size).map(|i| (i.wrapping_mul(67) ^ (i >> 9) ^ 0xa5) as u8).collect(); + let client_request = request.clone(); + let client_response = response.clone(); + let client_trace = trace.clone(); + let client = tokio::spawn(async move { + let mut socket = socket.connect(remote.parse().unwrap()).await.unwrap(); + socket.set_nodelay(true).unwrap(); + let mut received = Vec::new(); + if mode == "server-first" { + socket.read_to_end(&mut received).await.unwrap(); + socket.write_all(&client_request).await.unwrap(); + socket.shutdown().await.unwrap(); + } else if mode == "crossed" { + socket.write_all(&client_request).await.unwrap(); + received.resize(size, 0); + socket.read_exact(&mut received).await.unwrap(); + while client_trace.lock().unwrap().stack_fins.is_empty() { + tokio::time::sleep(Duration::from_millis(1)).await; + } + socket.shutdown().await.unwrap(); + socket.read_to_end(&mut received).await.unwrap(); + } else { + socket.write_all(&client_request).await.unwrap(); + socket.shutdown().await.unwrap(); + socket.read_to_end(&mut received).await.unwrap(); + } + assert_eq!(received, client_response); + }); + let result = tokio::time::timeout(Duration::from_secs(45), async { + let mut stream = loop { + match stack.accept().await.unwrap() { + IpStackStream::Tcp(stream) => break stream, + IpStackStream::UnknownTransport(other) + if other.ip_protocol().0 == 58 && other.dst_addr().is_multicast() => + { + eprintln!("CONTROL icmpv6 dst={} type={:?}", other.dst_addr(), other.payload().first()); + } + _ => panic!("unexpected non-TCP stream"), + } + }; + let mut received = Vec::new(); + if mode == "server-first" { + stream.write_all(&response).await.unwrap(); + let (mut reader, mut writer) = tokio::io::split(stream); + tokio::try_join!(writer.shutdown(), reader.read_to_end(&mut received)).unwrap(); + } else if matches!(mode, "simultaneous" | "crossed") { + received.resize(size, 0); + stream.read_exact(&mut received).await.unwrap(); + stream.write_all(&response).await.unwrap(); + stream.shutdown().await.unwrap(); + } else { + stream.read_to_end(&mut received).await.unwrap(); + stream.write_all(&response).await.unwrap(); + if mode == "drop" { + drop(stream); + } else { + stream.shutdown().await.unwrap(); + } + } + assert_eq!(received, request); + client.await.unwrap(); + while !trace.lock().unwrap().fault_recovered() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + let state = trace.lock().unwrap(); + assert!( + result.is_ok(), + "timeout profile={profile} v={version} mode={mode} bytes={size} fault={fault:?} stack_fin={} kernel_fin={} drop_fin={} drop_ack={} stack_ack={} kernel_ack={}", + state.stack_fins.len(), + state.kernel_fins.len(), + state.dropped_fins, + state.dropped_ack, + state.stack_ack_passed, + state.kernel_ack_passed + ); + assert_eq!(state.resets, 0, "unexpected RST in {case}"); + eprintln!( + "PASS_EXT profile={profile} v={version} mode={mode} bytes={size} fault={fault:?} stack_fin={} kernel_fin={} drop_fin={} drop_ack={} reorder={} crossed={} stack_ack={} kernel_ack={} late_rst={}", + state.stack_fins.len(), + state.kernel_fins.len(), + state.dropped_fins, + state.dropped_ack, + state.reordered_pairs, + state.crossed_peer_fin && state.released_cross_fin, + state.stack_ack_passed, + state.kernel_ack_passed, + state.late_resets + ); + } + } + } + } + command("tc", &["-s", "qdisc", "show", "dev", "ipstack-ext"]); +}