fix(net): give inline fast-path flows sender-side retransmission

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<std::sync::Mutex<(u32, std::collections::VecDeque<u8>, Option<u32>)>>;
+
 /// 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<std::sync::atomic::AtomicBool>,
+    /// 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<std::sync::atomic::AtomicU32>,
     down_bytes: Option<Arc<std::sync::atomic::AtomicU64>>,
     dead: Arc<std::sync::atomic::AtomicBool>,
+    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::<Vec<u8>>(),
+            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));
 

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<u8> = 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<std::sync::Arc<std::sync::atomic::AtomicBool>> = None;
+        let mut retx: Option<crate::direct_rx::RetxRing> = 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,
             },
         );
     }

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<StdInstant>,
+    /// 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<crate::direct_rx::RetxRing>,
 }
 
 impl FastPathConn {