diff --git a/common/splicetcp/src/direct_rx.rs b/common/splicetcp/src/direct_rx.rs index 532d14787..65c611d0a 100644 --- a/common/splicetcp/src/direct_rx.rs +++ b/common/splicetcp/src/direct_rx.rs @@ -52,6 +52,25 @@ impl FrameSink for ChannelFrameSink { } } +/// Shared retransmission ring between the bridge's `FastPathConn` and the +/// inline owner that sends on its behalf: `(seq of buf[0], unACKed sent +/// bytes, FIN seq once sent)`. +/// +/// The owner tees every payload it sends into `buf` (and records the FIN +/// position); `TcpBridge::poll_fast_path` drains the ACKed prefix as +/// `guest_acked` advances and re-emits from the ring on dup-ACK/RTO — +/// giving inline flows the same sender-side loss recovery as the polled +/// path (the link beyond guest eth0 drops under burst, so window gating +/// alone cannot prevent a wedge; see `tcp_bridge`'s module docs). +/// +/// Deliberately a tuple of std types only, so `arcbox-net-inject` can hold +/// the same ring without a splicetcp dependency (the same pattern as the +/// shared seq atomics). Bounded in practice by `HONORED_WINDOW_CAP`: the +/// owner never sends beyond that budget, and the ring holds only unACKed +/// bytes plus at most one poll tick of drain lag. +pub type RetxRing = + std::sync::Arc, Option)>>; + /// A TCP connection promoted to the inline inject path. /// /// Carries everything the inject thread needs to construct @@ -106,6 +125,10 @@ pub struct PromotedConn { /// writes still reach `try_fast_path_intercept` (parity with the /// non-inline path). pub dead: std::sync::Arc, + /// Retransmission ring shared with the bridge — the owner tees every + /// sent payload (and the FIN position) into it; `poll_fast_path` + /// drains and re-emits. See [`RetxRing`]. + pub retx: RetxRing, } /// Accepts promoted fast-path connections and delivers them to the RX @@ -195,6 +218,7 @@ impl ConnSink for TokioFrameConnSink { gw_mac, guest_mac, dead, + retx, } = conn; // A `false` return means "not accepted": the flow stays on the // bridge's slow path, so `dead` must NOT be set here. @@ -212,6 +236,7 @@ impl ConnSink for TokioFrameConnSink { guest_window, down_bytes, dead, + retx, gw_mac, guest_mac, // Bound this sink's own MTU-derived segmentation by the peer's MSS so @@ -237,6 +262,7 @@ struct AsyncPromotedConn { guest_window: Arc, down_bytes: Option>, dead: Arc, + retx: RetxRing, gw_mac: [u8; 6], guest_mac: [u8; 6], guest_mss: usize, @@ -295,6 +321,9 @@ async fn read_promoted_conn( src_mac: conn.gw_mac, dst_mac: conn.guest_mac, }); + // Record the FIN position so the bridge retransmits a lost + // FIN like any other in-flight byte. + conn.retx.lock().unwrap().2 = Some(seq_now); conn.our_seq.fetch_add(1, Ordering::Relaxed); let _ = frames.send(fin).await; return; @@ -325,6 +354,14 @@ async fn read_promoted_conn( ); conn.our_seq .fetch_add(chunk.len() as u32, Ordering::Relaxed); + // Tee each chunk as its sequence space is committed — + // never the whole read up front: `frames.send` can park + // this task on backpressure, and a ring running ahead + // of SND.NXT would let the bridge retransmit bytes the + // guest was never sent (whose ACKs the intercept then + // rejects as beyond `our_seq`). The bridge drains the + // ring as the guest ACKs and re-emits on dup-ACK/RTO. + conn.retx.lock().unwrap().1.extend(chunk); if frames.send(frame).await.is_err() { return; } @@ -391,6 +428,11 @@ mod tests { let our_seq = Arc::new(AtomicU32::new(1000)); let last_ack = Arc::new(AtomicU32::new(2000)); let down = Arc::new(AtomicU64::new(0)); + let retx: RetxRing = Arc::new(std::sync::Mutex::new(( + 1000, + std::collections::VecDeque::new(), + None, + ))); let std_stream = accepted.into_std().unwrap(); let accepted_conn = PromotedConn { stream: std_stream, @@ -407,6 +449,7 @@ mod tests { gw_mac: [0x02, 0, 0, 0, 0, 1], guest_mac: [0x02, 0, 0, 0, 0, 2], dead: Arc::new(std::sync::atomic::AtomicBool::new(false)), + retx: Arc::clone(&retx), }; assert!(sink.send_conn(accepted_conn)); @@ -429,6 +472,13 @@ mod tests { assert_eq!(u16::from_be_bytes([frame[tcp], frame[tcp + 1]]), 443); assert_eq!(u16::from_be_bytes([frame[tcp + 2], frame[tcp + 3]]), 50000); assert_eq!(&frame[tcp + 20..tcp + 24], b"pong"); + // The sent bytes are teed into the shared retransmission ring so + // the bridge can re-emit them on dup-ACK/RTO. + assert_eq!( + retx.lock().unwrap().1.iter().copied().collect::>(), + b"pong", + "sent payload teed into the ring" + ); } #[tokio::test] @@ -457,6 +507,11 @@ mod tests { gw_mac: [0x02, 0, 0, 0, 0, 1], guest_mac: [0x02, 0, 0, 0, 0, 2], dead: Arc::new(std::sync::atomic::AtomicBool::new(false)), + retx: Arc::new(std::sync::Mutex::new(( + 0, + std::collections::VecDeque::new(), + None, + ))), }; assert!(sink.send_conn(accepted_conn)); @@ -509,6 +564,11 @@ mod tests { gw_mac: [0x02, 0, 0, 0, 0, 1], guest_mac: [0x02, 0, 0, 0, 0, 2], dead: Arc::new(std::sync::atomic::AtomicBool::new(false)), + retx: Arc::new(std::sync::Mutex::new(( + 0, + std::collections::VecDeque::new(), + None, + ))), }; assert!(sink.send_conn(accepted_conn)); @@ -555,6 +615,11 @@ mod tests { let (sink, mut rx) = TokioFrameConnSink::channel(4); let dead = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let retx: RetxRing = Arc::new(std::sync::Mutex::new(( + 1000, + std::collections::VecDeque::new(), + None, + ))); let conn = PromotedConn { stream: accepted.into_std().unwrap(), remote_ip: std::net::Ipv4Addr::new(203, 0, 113, 10), @@ -570,6 +635,7 @@ mod tests { gw_mac: [0x02, 0, 0, 0, 0, 1], guest_mac: [0x02, 0, 0, 0, 0, 2], dead: Arc::clone(&dead), + retx: Arc::clone(&retx), }; assert!(sink.send_conn(conn)); @@ -582,6 +648,11 @@ mod tests { let flags = frame[ETH_HEADER_LEN + 20 + 13]; assert_ne!(flags & 0x01, 0, "must be a FIN (flags {flags:#04x})"); assert_eq!(flags & 0x04, 0, "must not be a RST (flags {flags:#04x})"); + assert_eq!( + retx.lock().unwrap().2, + Some(1000), + "the FIN position is recorded for bridge retransmission" + ); // Give the read task time to exit, then confirm it left `dead` unset. tokio::time::sleep(std::time::Duration::from_millis(50)).await; @@ -618,6 +689,11 @@ mod tests { gw_mac: [0x02, 0, 0, 0, 0, 1], guest_mac: [0x02, 0, 0, 0, 0, 2], dead: Arc::clone(&dead), + retx: Arc::new(std::sync::Mutex::new(( + 0, + std::collections::VecDeque::new(), + None, + ))), }; assert!(sink.send_conn(conn)); diff --git a/common/splicetcp/src/tcp_bridge/fast_path.rs b/common/splicetcp/src/tcp_bridge/fast_path.rs index 1764f318d..e33c4172d 100644 --- a/common/splicetcp/src/tcp_bridge/fast_path.rs +++ b/common/splicetcp/src/tcp_bridge/fast_path.rs @@ -460,10 +460,10 @@ impl TcpBridge { // zero-window guest must answer with a window-bearing ACK // (RFC 793 §3.9). It consumes no sequence space and touches no // shared state, so a lost probe is harmlessly re-sent next - // interval with no stream gap (the inline path has no - // retransmit). Gated on nothing-in-flight: the guest has ACKed - // up to `our_seq`, so `our_seq - 1` is always an old byte, and - // any in-flight data would elicit its own window-bearing ACKs. + // interval with no stream gap. Gated on nothing-in-flight: the + // guest has ACKed up to `our_seq`, so `our_seq - 1` is always + // an old byte, and any in-flight data would elicit its own + // window-bearing ACKs. let sent = conn.our_seq.load(std::sync::atomic::Ordering::Relaxed); let acked = conn.guest_acked.load(std::sync::atomic::Ordering::Relaxed); let window = conn @@ -517,6 +517,96 @@ impl TcpBridge { to_remove.push(*key); continue; } + // Sender-side loss recovery for inline flows, from the shared + // ring the owner tees every sent payload into (the inline + // counterpart of `retransmit_buf`): drain the ACKed prefix, + // then re-emit everything past the guest's ack point on + // triple-dup-ACK (counted by the intercept, same as + // non-inline) or RTO. The path beyond guest eth0 (bridge → + // veth → container netns backlog) drops under burst, so + // window gating alone cannot prevent a wedge — without this + // a single lost frame stalled the flow forever. + // Retransmissions travel the ordinary polled frame path + // (guest_tx), not the zero-copy inject path — correct and + // rare, so the cost is irrelevant. + if let Some(retx) = conn.retx.as_ref().map(std::sync::Arc::clone) { + let mut ring = retx.lock().unwrap(); + let (ref mut base, ref mut ring_buf, fin_seq) = *ring; + let drained = acked.wrapping_sub(*base); + if drained > 0 && drained < 0x8000_0000 { + // The guest ACKed more of our stream — release the + // buffered prefix and reset the loss-recovery clocks. + let n = (drained as usize).min(ring_buf.len()); + ring_buf.drain(..n); + *base = acked; + conn.last_progress = now; + conn.rto = super::INITIAL_RTO; + conn.dup_acks = 0; + } + if !nothing_in_flight { + let timed_out = now.duration_since(conn.last_progress) >= conn.rto; + if conn.fast_retransmit || timed_out { + let offset = acked.wrapping_sub(*base) as usize; + if offset < ring_buf.len() { + // Clamp to in-flight bytes: the owner tees + // after advancing SND.NXT, but never emit + // past `sent` even if a racing tee briefly + // runs ahead — the guest would ACK bytes + // the intercept then rejects as beyond + // `our_seq`. + let data: Vec = ring_buf + .iter() + .skip(offset) + .take(in_flight as usize) + .copied() + .collect(); + emit_data_frames(&ctx, conn, acked, &data, &mut frames); + } + if let Some(fin_seq) = fin_seq + && sent.wrapping_sub(fin_seq) < 0x8000_0000 + && fin_seq.wrapping_sub(acked) < 0x8000_0000 + { + frames.push(crate::ethernet::build_tcp_fin_frame( + &crate::ethernet::TcpFrameParams { + src_ip: conn.remote_ip, + dst_ip: conn.guest_ip, + src_port: conn.remote_port, + dst_port: conn.guest_port, + seq: fin_seq, + ack: conn.last_ack, + window: 65535, + src_mac: gw_mac, + dst_mac: guest_mac, + }, + )); + } + tracing::debug!( + "Fast path inline retransmit {}:{} → {}:{} ({} bytes from seq={acked}, cause={}, rto={:?})", + conn.remote_ip, + conn.remote_port, + conn.guest_ip, + conn.guest_port, + in_flight, + if conn.fast_retransmit { + "dup-acks" + } else { + "rto" + }, + conn.rto, + ); + conn.fast_retransmit = false; + conn.dup_acks = 0; + conn.retransmits += 1; + conn.last_progress = now; + conn.rto = (conn.rto * 2).min(super::MAX_RTO); + } + } else { + // Nothing in flight — keep the RTO clock parked so + // the next send starts a fresh timeout window. + conn.last_progress = now; + conn.fast_retransmit = false; + } + } if super::send_budget(sent, acked, window) == 0 && nothing_in_flight { match conn.window_stalled_at { None => conn.window_stalled_at = Some(now), @@ -946,12 +1036,19 @@ impl TcpBridge { // polling path below, where each segment matches its advertised MSS. let inline_eligible = peer_mss >= GSO_SEGMENT_MSS; let mut dead: Option> = None; + let mut retx: Option = None; if let Some(sink) = self.conn_sink.as_ref().filter(|_| inline_eligible) { match stream.try_clone() { Ok(cloned) => { let gw_mac = self.fast_path_gateway_mac; let guest_mac = self.fast_path_guest_mac.unwrap_or([0xFF; 6]); let dead_flag = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + // Retransmission ring: base seq starts at our current + // send cursor; the owner tees sent bytes, the bridge + // drains and re-emits (see RetxRing). + let retx_ring: crate::direct_rx::RetxRing = std::sync::Arc::new( + std::sync::Mutex::new((our_seq, std::collections::VecDeque::new(), None)), + ); let promoted = crate::direct_rx::PromotedConn { stream: cloned, remote_ip: key.dst_ip, @@ -967,10 +1064,12 @@ impl TcpBridge { gw_mac, guest_mac, dead: std::sync::Arc::clone(&dead_flag), + retx: std::sync::Arc::clone(&retx_ring), }; if sink.send_conn(promoted) { inline_owned = true; dead = Some(dead_flag); + retx = Some(retx_ring); tracing::info!( "Fast path: promoted INLINE {}:{} → {}:{} (seq={our_seq}, ack={last_ack})", key.src_ip, @@ -1043,6 +1142,7 @@ impl TcpBridge { dead, last_guest_activity: std::time::Instant::now(), soliciting_since: None, + retx, }, ); } diff --git a/common/splicetcp/src/tcp_bridge/mod.rs b/common/splicetcp/src/tcp_bridge/mod.rs index 60f3a3c47..5d041e67a 100644 --- a/common/splicetcp/src/tcp_bridge/mod.rs +++ b/common/splicetcp/src/tcp_bridge/mod.rs @@ -440,6 +440,12 @@ pub(super) struct FastPathConn { /// long-idle flow whose upstream wakes up is measured from the wake-up — /// not from its (legitimately ancient) last guest frame. soliciting_since: Option, + /// Retransmission ring shared with the inline owner (`Some` iff + /// `inline_owned`): the owner tees every sent payload and the FIN + /// position into it; `poll_fast_path` drains the ACKed prefix and + /// re-emits on dup-ACK/RTO — the inline counterpart of + /// `retransmit_buf`. See [`crate::direct_rx::RetxRing`]. + retx: Option, } impl FastPathConn { diff --git a/common/splicetcp/src/tcp_bridge/tests.rs b/common/splicetcp/src/tcp_bridge/tests.rs index e81df3961..0e4615e19 100644 --- a/common/splicetcp/src/tcp_bridge/tests.rs +++ b/common/splicetcp/src/tcp_bridge/tests.rs @@ -1772,8 +1772,12 @@ async fn poll_fast_path_respects_guest_window() { accepted.write_all(&vec![0xDD; 100_000]).await.unwrap(); - async fn drain(bridge: &mut TcpBridge) -> usize { - let mut sent = 0usize; + /// Highest sequence-space coverage the bridge has emitted past `base`. + /// Coverage, not a byte total: on a slow machine the 200 ms RTO can fire + /// mid-drain and retransmit in-window bytes — which does not violate the + /// window and must not fail the assertion. + async fn drain(bridge: &mut TcpBridge, base: u32) -> usize { + let mut covered = 0usize; let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); let mut idle = 0; while std::time::Instant::now() < deadline && idle < 10 { @@ -1783,16 +1787,19 @@ async fn poll_fast_path_respects_guest_window() { tokio::time::sleep(std::time::Duration::from_millis(10)).await; } else { idle = 0; - sent += batch - .iter() - .map(|f| f.len().saturating_sub(ETH_HEADER_LEN + 40)) - .sum::(); + for f in &batch { + let payload = f.len().saturating_sub(ETH_HEADER_LEN + 40); + let end = tcp_seq_of(f) + .wrapping_add(payload as u32) + .wrapping_sub(base); + covered = covered.max(end as usize); + } } } - sent + covered } - let first = drain(&mut bridge).await; + let first = drain(&mut bridge, 1).await; assert!( first <= 65535, "no guest ACK yet: at most one unscaled window may be sent, got {first}" @@ -1806,7 +1813,7 @@ async fn poll_fast_path_respects_guest_window() { let ack = make_guest_segment((40024, dst_ip, 443), 1, 1 + first as u32, 65535, 0x10, &[]); bridge.try_fast_path_intercept(&ack).expect("intercepted"); - let second = drain(&mut bridge).await; + let second = drain(&mut bridge, 1 + first as u32).await; assert!(second > 0, "an opening ACK must release more data"); assert!( second <= 65535, @@ -2613,3 +2620,204 @@ async fn long_idle_flow_survives_upstream_wakeup() { "a freshly woken flow must survive until the guest had its full deadline" ); } + +// -------- Inline sender-side loss recovery (shared retransmission ring) ---- +// The inline owners tee every sent payload into a ring shared with the +// bridge; poll_fast_path drains it on ACK progress and re-emits on +// dup-ACK/RTO — without this a single frame lost past guest eth0 wedged +// an inline flow forever (issue #486). + +struct CaptureSink(std::sync::Mutex>); +impl crate::direct_rx::ConnSink for CaptureSink { + fn send_conn(&self, conn: crate::direct_rx::PromotedConn) -> bool { + *self.0.lock().unwrap() = Some(conn); + true + } +} + +/// Promotes an inline flow and returns the bridge plus the captured owner +/// half (whose ring/atomics the test drives in the owner's stead). +async fn inline_fixture( + src_port: u16, + dst_ip: Ipv4Addr, +) -> (TcpBridge, crate::direct_rx::PromotedConn, SynFlowKey) { + let sink = std::sync::Arc::new(CaptureSink(std::sync::Mutex::new(None))); + let mut bridge = TcpBridge::new(GW_IP); + bridge.set_fast_path_macs(GW_MAC, GUEST_MAC); + bridge.set_conn_sink(std::sync::Arc::clone(&sink) as _); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let connect = tokio::net::TcpStream::connect(addr); + let (client, accepted) = tokio::join!(connect, listener.accept()); + let _accepted = accepted.unwrap(); + + let key = SynFlowKey { + src_ip: GUEST_IP, + src_port, + dst_ip, + dst_port: 443, + }; + // peer_mss ≥ GSO_SEGMENT_MSS → inline-owned. + bridge.promote_to_fast_path( + key, + client.unwrap().into_std().unwrap(), + 1000, + 2000, + 9000, + None, + ); + let promoted = sink.0.lock().unwrap().take().expect("inline conn"); + (bridge, promoted, key) +} + +/// Simulates the inline owner sending `payload`: tee into the ring and +/// advance the shared send cursor, exactly as inject.rs / direct_rx.rs do. +fn owner_sends(promoted: &crate::direct_rx::PromotedConn, payload: &[u8]) { + promoted.retx.lock().unwrap().1.extend(payload); + promoted + .our_seq + .fetch_add(payload.len() as u32, std::sync::atomic::Ordering::Relaxed); +} + +fn backdate_rto_clock(bridge: &mut TcpBridge, key: &SynFlowKey) { + bridge.fast_path_conns.get_mut(key).unwrap().last_progress = std::time::Instant::now() + .checked_sub(super::MAX_RTO) + .expect("test clock underflow"); +} + +fn tcp_payload_len(frame: &[u8]) -> usize { + frame.len() - (ETH_HEADER_LEN + 20 + 20) +} + +/// An inline flow whose guest never ACKs must have its teed bytes re-emitted +/// from the shared ring after the RTO — the wedge #486 describes. +#[tokio::test] +async fn inline_rto_retransmits_from_shared_ring() { + let (mut bridge, promoted, key) = inline_fixture(40110, Ipv4Addr::new(198, 18, 30, 150)).await; + + owner_sends(&promoted, b"lost-inline-payload"); + assert!( + bridge.poll_fast_path().is_empty(), + "no retransmission before the RTO fires" + ); + + backdate_rto_clock(&mut bridge, &key); + let frames = bridge.poll_fast_path(); + assert_eq!(frames.len(), 1, "one retransmission after the RTO"); + assert_eq!(tcp_seq_of(&frames[0]), 1000, "resent from the unACKed base"); + assert_eq!( + tcp_payload_len(&frames[0]), + b"lost-inline-payload".len(), + "the full unACKed payload is resent" + ); +} + +/// Three duplicate ACKs from the guest must trigger an immediate inline +/// fast retransmit, without waiting out the RTO. +#[tokio::test] +async fn inline_triple_dup_ack_fast_retransmits() { + let dst_ip = Ipv4Addr::new(198, 18, 30, 151); + let (mut bridge, promoted, _key) = inline_fixture(40111, dst_ip).await; + + owner_sends(&promoted, b"hole-behind-this-data"); + + // Three dup-ACKs at the promoted cursor (ack=1000, unchanged window). + for _ in 0..3 { + let dup = make_guest_segment((40111, dst_ip, 443), 2000, 1000, 65535, 0x10, &[]); + bridge.try_fast_path_intercept(&dup); + } + + let frames = bridge.poll_fast_path(); + assert_eq!(frames.len(), 1, "fast retransmit fires without an RTO wait"); + assert_eq!(tcp_seq_of(&frames[0]), 1000); +} + +/// A guest ACK must drain the ring; a fully ACKed flow retransmits nothing +/// even after an RTO's worth of silence. +#[tokio::test] +async fn inline_ack_drains_ring_and_stops_retransmit() { + let dst_ip = Ipv4Addr::new(198, 18, 30, 152); + let (mut bridge, promoted, key) = inline_fixture(40112, dst_ip).await; + + owner_sends(&promoted, b"delivered"); + let acked_to = 1000 + b"delivered".len() as u32; + let ack = make_guest_segment((40112, dst_ip, 443), 2000, acked_to, 65535, 0x10, &[]); + bridge.try_fast_path_intercept(&ack); + + backdate_rto_clock(&mut bridge, &key); + assert!( + bridge.poll_fast_path().is_empty(), + "a fully ACKed inline flow must not retransmit" + ); + let ring = promoted.retx.lock().unwrap(); + assert!(ring.1.is_empty(), "the ACKed prefix must be drained"); + assert_eq!(ring.0, acked_to, "ring base advances to the ack point"); +} + +/// A lost inline FIN must be retransmitted from its recorded position, and +/// stop once the guest ACKs past it. +#[tokio::test] +async fn inline_lost_fin_is_retransmitted() { + let dst_ip = Ipv4Addr::new(198, 18, 30, 153); + let (mut bridge, promoted, key) = inline_fixture(40113, dst_ip).await; + + // Owner sends data + FIN (as inject.rs does at EOF). + owner_sends(&promoted, b"tail"); + let fin_seq = 1000 + b"tail".len() as u32; + promoted.retx.lock().unwrap().2 = Some(fin_seq); + promoted + .our_seq + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + + // Guest ACKs the data but never sees the FIN. + let ack = make_guest_segment((40113, dst_ip, 443), 2000, fin_seq, 65535, 0x10, &[]); + bridge.try_fast_path_intercept(&ack); + // First poll drains the ACKed data and restarts the RTO clock for the + // still-unACKed FIN; only then can the timeout be simulated. + assert!(bridge.poll_fast_path().is_empty()); + + backdate_rto_clock(&mut bridge, &key); + let frames = bridge.poll_fast_path(); + assert_eq!(frames.len(), 1, "the lost FIN is re-emitted"); + assert_ne!( + tcp_flags_of(&frames[0]) & 0x01, + 0, + "the retransmission is a FIN" + ); + assert_eq!(tcp_seq_of(&frames[0]), fin_seq); + + // ACK past the FIN ends the recovery. + let fin_ack = make_guest_segment((40113, dst_ip, 443), 2000, fin_seq + 1, 65535, 0x10, &[]); + bridge.try_fast_path_intercept(&fin_ack); + backdate_rto_clock(&mut bridge, &key); + assert!( + bridge.poll_fast_path().is_empty(), + "an ACKed FIN must not be retransmitted" + ); +} + +/// A ring that momentarily runs ahead of SND.NXT (the owner tees while a +/// racing bridge poll fires) must never cause bytes beyond `sent` to be +/// retransmitted — the guest would ACK them and the intercept would reject +/// that ACK as beyond `our_seq`. +#[tokio::test] +async fn inline_retransmit_never_exceeds_sent() { + let (mut bridge, promoted, key) = inline_fixture(40114, Ipv4Addr::new(198, 18, 30, 154)).await; + + // Simulate the transient: 100 bytes teed, only 40 of sequence space + // committed so far. + promoted.retx.lock().unwrap().1.extend([0xAA; 100]); + promoted + .our_seq + .fetch_add(40, std::sync::atomic::Ordering::Relaxed); + + backdate_rto_clock(&mut bridge, &key); + let frames = bridge.poll_fast_path(); + assert_eq!(frames.len(), 1); + assert_eq!( + tcp_payload_len(&frames[0]), + 40, + "retransmission is clamped to the committed sequence space" + ); +} diff --git a/tests/e2e/src/docker.rs b/tests/e2e/src/docker.rs index 177e1b8f9..3a5ad5f23 100644 --- a/tests/e2e/src/docker.rs +++ b/tests/e2e/src/docker.rs @@ -109,13 +109,20 @@ pub fn ensure_image(data_dir: &Path, image: &str) -> Result<()> { image.replace(['/', ':'], "_").replace('.', "-") )); if tar.exists() { - docker_output( + // A corrupt cache must not poison every subsequent run (an HV + // `docker save` can truncate the streamed response — issue #256): + // discard it and fall through to a fresh pull. + match docker_output( data_dir, &["load", "-i", &tar.display().to_string()], Duration::from_secs(60), - ) - .context("docker load from cache")?; - return Ok(()); + ) { + Ok(_) => return Ok(()), + Err(e) => { + tracing::warn!("cached image load failed ({e:#}); discarding cache, re-pulling"); + let _ = std::fs::remove_file(&tar); + } + } } let mut last_err = None; @@ -129,6 +136,16 @@ pub fn ensure_image(data_dir: &Path, image: &str) -> Result<()> { Duration::from_secs(60), ) .context("docker save to cache")?; + // Validate before trusting: a truncated save (#256) only + // surfaces as "invalid byte in chunk length" on the NEXT + // run's load — catch it now instead. + let listing = std::process::Command::new("tar") + .args(["-tf", &tar.display().to_string()]) + .output(); + if !listing.is_ok_and(|o| o.status.success()) { + tracing::warn!("saved image cache fails tar validation; discarding"); + let _ = std::fs::remove_file(&tar); + } return Ok(()); } Err(e) => { diff --git a/tests/e2e/src/scenario.rs b/tests/e2e/src/scenario.rs index 0458161ff..1c9aa7833 100644 --- a/tests/e2e/src/scenario.rs +++ b/tests/e2e/src/scenario.rs @@ -50,6 +50,10 @@ pub fn run_vz_scenario_with_log( // Diagnostic override for the daemon under test, e.g. // ARCBOX_E2E_DAEMON_LOG="info,splicetcp::tcp_bridge=debug". let rust_log = std::env::var("ARCBOX_E2E_DAEMON_LOG").unwrap_or_else(|_| rust_log.to_owned()); + // Backend override: ARCBOX_E2E_BACKEND=hv reruns any scenario against + // the HV backend (default vz, the name notwithstanding) — e.g. the + // network-workload suite as the acceptance for HV datapath changes. + let backend = std::env::var("ARCBOX_E2E_BACKEND").unwrap_or_else(|_| "vz".to_owned()); let root = crate::repo_root(); if !crate::env_flag("SKIP_BUILD") { @@ -79,13 +83,13 @@ pub fn run_vz_scenario_with_log( args: vec![], env: vec![ ("ARCBOX_BOOT_ASSET_VERSION".to_owned(), version), - ("ARCBOX_VM_BACKEND".to_owned(), "vz".to_owned()), + ("ARCBOX_VM_BACKEND".to_owned(), backend.clone()), ("ARCBOX_DNS_PORT".to_owned(), dns_port.to_string()), ("RUST_LOG".to_owned(), rust_log), ], })?; - let mut metrics = RunMetrics::new(name, Some("vz")); + let mut metrics = RunMetrics::new(name, Some(&backend)); let result = scenario(&mut daemon, data_dir.path(), &mut metrics); metrics.passed = result.is_ok(); if let Err(error) = metrics.write(Some(data_dir.path())) { diff --git a/virt/arcbox-net-inject/src/inject.rs b/virt/arcbox-net-inject/src/inject.rs index bb28df97c..282bd3edf 100644 --- a/virt/arcbox-net-inject/src/inject.rs +++ b/virt/arcbox-net-inject/src/inject.rs @@ -230,6 +230,15 @@ impl RxInjectThread { // budget for other conns. const PER_CONN_READS: u16 = 16; + // Batch headroom reserved for the channel drain that follows this + // pass: `guest_tx` carries the intercept's ACK frames and the + // bridge's retransmissions for these very flows. One continuously + // readable conn can otherwise consume the entire batch + // (PER_CONN_READS × MAX_MERGE == BATCH_SIZE) every cycle and + // starve a stalled sibling's recovery frames indefinitely. + const CHANNEL_RESERVE: usize = 64; + let inline_cap = BATCH_SIZE - CHANNEL_RESERVE; + // Max descriptors per `readv` (upper bound on num_buffers stamped // into the first descriptor's virtio-net header). const MAX_MERGE: usize = 16; @@ -248,7 +257,7 @@ impl RxInjectThread { for conn in inline_conns.iter_mut() { let mut per_conn = 0u16; loop { - if (*batch as usize) >= BATCH_SIZE { + if (*batch as usize) >= inline_cap { break; } if per_conn >= PER_CONN_READS { @@ -256,9 +265,8 @@ impl RxInjectThread { } // Never send beyond the guest's advertised receive window: - // outside that budget the guest kernel drops what it can't - // buffer, and this path has no retransmission to repair the - // gap — the flow would wedge permanently (2026-07-19). A + // the budget bounds the burst toward the guest-internal + // backlog and the shared retransmission ring alike. A // window-limited conn is revisited next pass, once the // guest's ACKs (via the datapath intercept) reopen it. let budget = conn.send_budget() as usize; @@ -400,6 +408,9 @@ impl RxInjectThread { let first_buf = unsafe { std::slice::from_raw_parts_mut(desc_ptrs[0], desc_lens[0]) }; inline_conn::write_fin_headers(first_buf, conn); + // Record the FIN position so the bridge retransmits + // a lost FIN like any other in-flight byte. + conn.retx.lock().unwrap().2 = Some(conn.our_seq.load(Ordering::Relaxed)); conn.our_seq.fetch_add(1, Ordering::Relaxed); *fire |= @@ -435,6 +446,39 @@ impl RxInjectThread { // the datapath carry the correct seq value. conn.our_seq.fetch_add(n as u32, Ordering::Relaxed); + // Tee the sent bytes into the shared retransmission + // ring: the bridge drains it as the guest ACKs and + // re-emits from it on dup-ACK/RTO — the path beyond + // guest eth0 drops under burst, and without this a + // single lost frame wedged the flow forever. The tee + // comes AFTER the seq advance so the ring never runs + // ahead of SND.NXT: a bridge poll racing this window + // may retransmit short (repaired by the next RTO), + // never bytes the guest hasn't been sent. + { + let mut ring = conn.retx.lock().unwrap(); + let mut copied = 0usize; + for i in 0..num_used { + let start = if i == 0 { + inline_conn::TOTAL_HDR_LEN + } else { + 0 + }; + let take = per_desc_len[i].min(n - copied); + if take == 0 { + break; + } + // SAFETY: same device-owned descriptor + // buffers readv just filled; still exclusive + // to us until the used publish below. + let payload = unsafe { + std::slice::from_raw_parts(desc_ptrs[i].add(start), take) + }; + ring.1.extend(payload); + copied += take; + } + } + // Return the gathered-but-unfilled buffers, then // publish one used entry per consumed buffer. The // first entry also accounts for the 66-byte header @@ -631,6 +675,11 @@ mod tests { guest_mac: [0x02, 0, 0, 0, 0, 2], host_eof: false, dead: Arc::new(std::sync::atomic::AtomicBool::new(false)), + retx: Arc::new(std::sync::Mutex::new(( + 1000, + std::collections::VecDeque::new(), + None, + ))), } } @@ -723,6 +772,20 @@ mod tests { 1000 + payload.len() as u32 ); + // Every sent byte is teed into the shared retransmission ring, in + // order across the descriptor spans, so the bridge can re-emit on + // dup-ACK/RTO. + { + let ring = conns[0].retx.lock().unwrap(); + assert_eq!(ring.0, 1000, "ring base is the promoted seq"); + assert_eq!( + ring.1.iter().copied().collect::>(), + payload, + "teed bytes match the sent payload" + ); + assert_eq!(ring.2, None, "no FIN yet"); + } + // Nothing more buffered: a second poll consumes nothing. let (mut batch2, mut fire2) = (0u16, false); thread.poll_inline_conns(&mut queue, &mut conns, &mut batch2, &mut fire2); @@ -781,6 +844,9 @@ mod tests { // TCP flags byte in the injected frame: FIN | ACK. let first = ram.buffer(0, inline_conn::TOTAL_HDR_LEN); assert_eq!(first[12 + 14 + 20 + 13], 0x11); + // The FIN position is recorded in the shared ring so the bridge can + // retransmit a lost FIN. + assert_eq!(conns[0].retx.lock().unwrap().2, Some(1000)); // Clean EOF must NOT mark the flow dead: the bridge entry stays // alive so the guest's ACK/FIN and half-close writes still reach diff --git a/virt/arcbox-net-inject/src/inline_conn.rs b/virt/arcbox-net-inject/src/inline_conn.rs index dd803f5f3..eb6c90a2b 100644 --- a/virt/arcbox-net-inject/src/inline_conn.rs +++ b/virt/arcbox-net-inject/src/inline_conn.rs @@ -22,6 +22,17 @@ const ETH_IP_TCP_HDR_LEN: usize = 54; /// Virtio-net header (12 bytes) + Ethernet+IP+TCP headers (54 bytes). pub const TOTAL_HDR_LEN: usize = 12 + ETH_IP_TCP_HDR_LEN; +/// Retransmission ring shared with the TCP bridge: `(seq of buf[0], unACKed +/// sent bytes, FIN seq once sent)`. +/// +/// The inject thread tees every payload it sends into `buf` (and records the +/// FIN position); the bridge's `poll_fast_path` drains the ACKed prefix and +/// re-emits on dup-ACK/RTO. A tuple of std types only so this crate and +/// `splicetcp` can share the same ring without depending on each other (the +/// same pattern as the shared seq atomics). Must stay layout-identical to +/// `splicetcp::direct_rx::RetxRing`. +pub type RetxRing = Arc, Option)>>; + /// A promoted fast-path TCP connection owned by the inject thread. /// The socket lives here — reads go directly to guest memory. pub struct InlineConn { @@ -60,6 +71,10 @@ pub struct InlineConn { /// (ABX-431). After a clean EOF the flag stays unset — the entry /// survives for the guest's close handshake. pub dead: Arc, + /// Retransmission ring shared with the bridge — this thread tees every + /// sent payload (and the FIN position) into it; the bridge drains and + /// re-emits. See [`RetxRing`]. + pub retx: RetxRing, } // SAFETY: TcpStream is Send, all other fields are Send+Sync. @@ -74,9 +89,10 @@ const HONORED_WINDOW_CAP: u32 = 256 * 1024; impl InlineConn { /// Bytes this thread may still send without exceeding the guest's /// advertised receive window: `window − (sent − acked)`, wrap-safe. - /// Within the budget the guest kernel guarantees buffering, and the - /// inject path is lossless — so staying inside it means no gap can - /// form that this retransmission-free path could never repair. + /// The window bounds the burst and the shared retransmission ring + /// ([`RetxRing`]) — the inject path is lossless only to guest eth0; + /// the guest-internal bridge → veth → container backlog drops under + /// burst, and the bridge repairs those from the ring. /// (Mirrors `splicetcp::tcp_bridge::send_budget`.) pub fn send_budget(&self) -> u32 { let sent = self.our_seq.load(Ordering::Relaxed); diff --git a/virt/arcbox-vmm/src/vmm/darwin_hv/inline_sink.rs b/virt/arcbox-vmm/src/vmm/darwin_hv/inline_sink.rs index 42c777ea8..89d703ad2 100644 --- a/virt/arcbox-vmm/src/vmm/darwin_hv/inline_sink.rs +++ b/virt/arcbox-vmm/src/vmm/darwin_hv/inline_sink.rs @@ -26,6 +26,7 @@ impl arcbox_net::direct_rx::ConnSink for InlineConnSinkAdapter { guest_mac: conn.guest_mac, host_eof: false, dead: conn.dead, + retx: conn.retx, }; self.tx.try_send(inline).is_ok() }