diff --git a/crates/buzz-relay-mesh/src/endpoint.rs b/crates/buzz-relay-mesh/src/endpoint.rs index f7bedd8789..d1529c324b 100644 --- a/crates/buzz-relay-mesh/src/endpoint.rs +++ b/crates/buzz-relay-mesh/src/endpoint.rs @@ -28,10 +28,22 @@ impl MeshEndpoint { bind_addr: SocketAddr, ) -> Result { let runtime_id = runtime_id_from_public_key(secret_key.public()); - let endpoint = Endpoint::builder(iroh::endpoint::presets::Minimal) + let mut builder = Endpoint::builder(iroh::endpoint::presets::Minimal) .secret_key(secret_key) .alpns(vec![ALPN.to_vec()]) - .relay_mode(RelayMode::Disabled) + .relay_mode(RelayMode::Disabled); + // A loopback-bound socket can never transmit to a WAN address, but the + // portmapper still probes the gateway (UPnP/PCP/NAT-PMP) and, on + // networks where a mapping succeeds, advertises the external address + // as a `Portmapped` direct-addr candidate. Path selection can then + // steer the connection onto that unsendable candidate mid-stream + // (`sendmsg` → EADDRNOTAVAIL) and frame delivery silently stops. + // Loopback reachability is loopback-only by definition — skip gateway + // probing for loopback binds. + if bind_addr.ip().is_loopback() { + builder = builder.portmapper_config(iroh::endpoint::PortmapperConfig::Disabled); + } + let endpoint = builder .bind_addr(bind_addr) .map_err(|err| MeshError::Transport(err.to_string()))? .bind() @@ -148,6 +160,26 @@ mod tests { (a, b) } + /// A loopback-bound endpoint must never advertise non-loopback direct + /// addrs. With the portmapper enabled, a NAT-PMP/UPnP-capable gateway + /// hands out a WAN mapping that is advertised as a `Portmapped` + /// candidate; path selection can then steer onto an address the + /// loopback-bound socket cannot send to (`sendmsg` → EADDRNOTAVAIL) and + /// frame delivery stalls (#2458). Trivially green on hosts without a + /// portmap-capable gateway; fails pre-fix on hosts with one. + #[tokio::test] + async fn loopback_bind_advertises_only_loopback_addrs() { + let ep = MeshEndpoint::bind(loopback_any()).await.unwrap(); + // The portmap probe (when enabled) completes within ~a second of bind. + tokio::time::sleep(std::time::Duration::from_millis(1200)).await; + for addr in ep.ip_addrs() { + assert!( + addr.ip().is_loopback(), + "non-loopback direct addr advertised from loopback bind: {addr}" + ); + } + } + async fn connected_pair() -> ( crate::peer::MeshPeer, crate::peer::MeshPeer, diff --git a/crates/buzz-relay-mesh/src/peer.rs b/crates/buzz-relay-mesh/src/peer.rs index 59ee8308fc..7748155eb0 100644 --- a/crates/buzz-relay-mesh/src/peer.rs +++ b/crates/buzz-relay-mesh/src/peer.rs @@ -83,7 +83,7 @@ impl MeshPeer { self.counters.streams_opened.fetch_add(1, Ordering::Relaxed); Ok(MeshStream::new( Box::new(IrohSendHalf(send)), - Box::new(IrohRecvHalf(recv)), + Box::new(IrohRecvHalf::new(recv)), )) } @@ -98,7 +98,7 @@ impl MeshPeer { .fetch_add(1, Ordering::Relaxed); Ok(MeshStream::new( Box::new(IrohSendHalf(send)), - Box::new(IrohRecvHalf(recv)), + Box::new(IrohRecvHalf::new(recv)), )) } @@ -130,7 +130,49 @@ impl MeshPeer { } struct IrohSendHalf(iroh::endpoint::SendStream); -struct IrohRecvHalf(iroh::endpoint::RecvStream); + +/// Receiving half with resumable framing state. +/// +/// `recv_frame` futures get raced against drain ticks and shutdown signals +/// (e.g. `run_demo_echo`), so a dropped future must not lose stream position. +/// Partially-read frame bytes accumulate in `buf` across cancellations; each +/// `RecvStream::read` await is itself cancel-safe, so the only state that +/// matters lives here rather than in the future. +struct IrohRecvHalf { + stream: iroh::endpoint::RecvStream, + /// Bytes of the in-progress frame (length prefix + body) read so far. + buf: Vec, +} + +impl IrohRecvHalf { + fn new(stream: iroh::endpoint::RecvStream) -> Self { + Self { + stream, + buf: Vec::new(), + } + } + + /// Grow `self.buf` to `target` bytes, returning `false` on clean EOF + /// before any byte of the current frame arrived. + async fn fill(&mut self, target: usize) -> Result { + while self.buf.len() < target { + let mut chunk = vec![0u8; target - self.buf.len()]; + match self.stream.read(&mut chunk).await { + Ok(Some(n)) => self.buf.extend_from_slice(&chunk[..n]), + Ok(None) => { + if self.buf.is_empty() { + return Ok(false); + } + return Err(MeshError::Transport( + "stream finished mid-frame".to_string(), + )); + } + Err(err) => return Err(MeshError::Transport(err.to_string())), + } + } + Ok(true) + } +} impl StreamSendHalf for IrohSendHalf { fn send_frame( @@ -167,14 +209,12 @@ impl StreamSendHalf for IrohSendHalf { impl StreamRecvHalf for IrohRecvHalf { fn recv_frame(&mut self) -> crate::BoxFuture<'_, Result, MeshError>> { Box::pin(async move { - let mut len = [0u8; 4]; - match self.0.read_exact(&mut len).await { - Ok(_) => {} - Err(iroh::endpoint::ReadExactError::FinishedEarly(0)) => return Ok(None), - Err(err) => return Err(MeshError::Transport(err.to_string())), + // Length prefix. `fill` resumes from `self.buf` if a previous + // `recv_frame` future was dropped mid-read. + if !self.fill(4).await? { + return Ok(None); } - - let len = u32::from_le_bytes(len); + let len = u32::from_le_bytes([self.buf[0], self.buf[1], self.buf[2], self.buf[3]]); if len > wire::MAX_STREAM_FRAME { return Err(MeshError::FrameTooLarge { size: len as usize, @@ -182,12 +222,15 @@ impl StreamRecvHalf for IrohRecvHalf { }); } - let mut bytes = vec![0u8; len as usize]; - self.0 - .read_exact(&mut bytes) - .await - .map_err(|err| MeshError::Transport(err.to_string()))?; - wire::decode::(&bytes).map(Some) + let total = 4 + len as usize; + if !self.fill(total).await? { + return Err(MeshError::Transport( + "stream finished mid-frame".to_string(), + )); + } + let frame = wire::decode::(&self.buf[4..total]).map(Some); + self.buf.clear(); + frame }) } } diff --git a/crates/buzz-relay/src/api/mesh_demo.rs b/crates/buzz-relay/src/api/mesh_demo.rs index 8649b97671..3c5f581596 100644 --- a/crates/buzz-relay/src/api/mesh_demo.rs +++ b/crates/buzz-relay/src/api/mesh_demo.rs @@ -151,8 +151,8 @@ mod tests { use axum::body::to_bytes; use buzz_relay_mesh::endpoint::MeshEndpoint; use buzz_relay_mesh::{ - InboundHandler, MeshDatagram, MeshError, MeshStream, MeshStreamFrame, RuntimeId, - StreamHello, + BoxFuture, InboundHandler, MeshDatagram, MeshError, MeshStream, MeshStreamFrame, RuntimeId, + StreamHello, StreamRecvHalf, StreamSendHalf, }; use uuid::Uuid; @@ -179,6 +179,81 @@ mod tests { )) } + // ── In-memory mesh stream pair ──────────────────────────────────────── + // + // The forwarded-arm round trip exercises buzz's session logic: join → + // `Forwarded`, Hello handoff, `accept_inbound`, Redis fence validation, + // and the echo consumer. None of that depends on iroh's UDP/QUIC layer, + // so the default test runs over this in-memory pair — deterministic and + // fast, and it reproduced the #2458 frame loss (run_demo_echo cancelling + // `recv_validated` mid-validation) just as readily as the live + // transport, making it the regression test for that fix. The + // `_live_mesh` variant keeps the real transport under `#[ignore]` for + // explicit evidence runs. + + struct ChanSendHalf(tokio::sync::mpsc::UnboundedSender); + struct ChanRecvHalf(tokio::sync::mpsc::UnboundedReceiver); + + impl StreamSendHalf for ChanSendHalf { + fn send_frame(&mut self, frame: MeshStreamFrame) -> BoxFuture<'_, Result<(), MeshError>> { + Box::pin(async move { + self.0 + .send(frame) + .map_err(|_| MeshError::Transport("in-memory peer closed".into())) + }) + } + + fn finish(&mut self) -> Result<(), MeshError> { + Ok(()) + } + } + + impl StreamRecvHalf for ChanRecvHalf { + fn recv_frame(&mut self) -> BoxFuture<'_, Result, MeshError>> { + Box::pin(async move { Ok(self.0.recv().await) }) + } + } + + fn in_memory_stream_pair() -> (MeshStream, MeshStream) { + let (a_tx, b_rx) = tokio::sync::mpsc::unbounded_channel(); + let (b_tx, a_rx) = tokio::sync::mpsc::unbounded_channel(); + ( + MeshStream::new(Box::new(ChanSendHalf(a_tx)), Box::new(ChanRecvHalf(a_rx))), + MeshStream::new(Box::new(ChanSendHalf(b_tx)), Box::new(ChanRecvHalf(b_rx))), + ) + } + + /// Transport handing out a pre-built in-memory stream, sending the Hello + /// on it first — the same contract as the live transport's + /// `open_session_stream`. + struct InMemoryTransport { + stream: std::sync::Mutex>, + } + + impl RelayPeerTransport for InMemoryTransport { + fn send_datagram(&self, _to: RuntimeId, _dgram: MeshDatagram) -> Result<(), MeshError> { + unreachable!("forwarded-arm test never sends datagrams") + } + + fn open_session_stream( + &self, + _to: RuntimeId, + hello: StreamHello, + ) -> std::pin::Pin< + Box> + Send + '_>, + > { + Box::pin(async move { + let taken = self.stream.lock().expect("stream mutex").take(); + let mut stream = + taken.ok_or_else(|| MeshError::Transport("stream already taken".into()))?; + stream.send_frame(MeshStreamFrame::Hello(hello)).await?; + Ok(stream) + }) + } + + fn set_inbound(&self, _handler: Box) {} + } + struct NoopTransport; impl RelayPeerTransport for NoopTransport { @@ -258,8 +333,13 @@ mod tests { } /// Second runtime forwards to the owner and round-trips the payload - /// through the owner-side echo consumer (`recv_validated` + `send_bytes`), - /// end to end over a real mesh stream pair. + /// through the owner-side echo consumer over an in-memory stream pair. + /// + /// Regression test for #2458: pre-fix, `run_demo_echo` raced + /// `recv_validated` against its drain tick, and a tick firing during + /// fence validation dropped the future *with the frame it had already + /// read* — the echo then never happened and this test 504'd, in-memory + /// pair and live transport alike. #[tokio::test] async fn demo_join_forwarded_arm_round_trips_echo() { let Some(directory) = redis_directory_if_available().await else { @@ -267,6 +347,92 @@ mod tests { }; let community_id = Uuid::new_v4(); let session_id = Uuid::new_v4(); + let owner_runtime = RuntimeId([7; 32]); + let local_runtime = RuntimeId([9; 32]); + + // Owner acquires the lease first. + let owner_router = ReliableStreamRouter::new( + directory.clone(), + std::sync::Arc::new(NoopTransport), + owner_runtime, + ); + let owned = run_demo_join( + &owner_router, + &directory, + DemoEchoRequest { + community_id, + session_id, + payload: "unused".into(), + }, + ) + .await; + assert_eq!(owned.status(), StatusCode::OK); + assert_eq!(body_json(owned).await["outcome"], "owned"); + + let (local_stream, owner_stream) = in_memory_stream_pair(); + + // Owner side: receive the Hello and run the real demo echo consumer. + let echo_directory = directory.clone(); + let owner_task = tokio::spawn(async move { + let mut stream = owner_stream; + let hello = match stream.recv_frame().await.unwrap().unwrap() { + MeshStreamFrame::Hello(h) => h, + other => panic!("expected hello, got {other:?}"), + }; + let router = ReliableStreamRouter::new( + echo_directory.clone(), + std::sync::Arc::new(NoopTransport), + owner_runtime, + ); + let from = hello.sender; + let inbound = router.accept_inbound(from, hello, stream).await.unwrap(); + crate::mesh_boot::run_demo_echo( + echo_directory, + inbound, + std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + ) + .await; + }); + + // Forwarding side: join the same session through the demo core. + let local_router = ReliableStreamRouter::new( + directory.clone(), + std::sync::Arc::new(InMemoryTransport { + stream: std::sync::Mutex::new(Some(local_stream)), + }), + local_runtime, + ); + let resp = run_demo_join( + &local_router, + &directory, + DemoEchoRequest { + community_id, + session_id, + payload: "mesh echo evidence".into(), + }, + ) + .await; + assert_eq!(resp.status(), StatusCode::OK); + let body = body_json(resp).await; + assert_eq!(body["outcome"], "forwarded"); + assert_eq!(body["echoed_payload"], "mesh echo evidence"); + owner_task.abort(); + } + + /// Same round trip over live iroh endpoints — the transport-inclusive + /// smoke. `#[ignore]`: the in-memory variant covers the session logic + /// deterministically, and this one stays exposed to real-network timing + /// (UDP scheduling, suite-load stalls vs. the test lease TTL), so it + /// runs on demand rather than in the default suite: + /// `cargo test -p buzz-relay --lib live_mesh -- --ignored`. + #[tokio::test] + #[ignore = "live-transport evidence smoke — run explicitly; default-suite coverage is the in-memory variant"] + async fn demo_join_forwarded_arm_round_trips_echo_live_mesh() { + let Some(directory) = redis_directory_if_available().await else { + return; + }; + let community_id = Uuid::new_v4(); + let session_id = Uuid::new_v4(); let bind = || "127.0.0.1:0".parse().unwrap(); let local_endpoint = MeshEndpoint::bind(bind()).await.unwrap(); diff --git a/crates/buzz-relay/src/mesh_boot.rs b/crates/buzz-relay/src/mesh_boot.rs index 20e550aa08..c82a33f8fe 100644 --- a/crates/buzz-relay/src/mesh_boot.rs +++ b/crates/buzz-relay/src/mesh_boot.rs @@ -315,7 +315,13 @@ pub(crate) async fn run_demo_echo( tracing::info!(%session_id, %peer, "mesh demo echo: session open"); let mut drain_tick = tokio::time::interval(std::time::Duration::from_millis(100)); loop { - let frame = tokio::select! { + // Race only the cancel-safe frame *extraction* against the drain + // tick. `recv_validated` must not be raced here: it holds the frame + // it just read across the Redis fence await, so a drain tick firing + // mid-validation would drop the future and silently destroy the + // frame — the stream then looks idle forever and the peer times out + // waiting for an echo that can no longer happen (#2458). + let wire_frame = tokio::select! { _ = drain_tick.tick() => { if shutting_down.load(Ordering::Relaxed) { if let Some(community_id) = stream.community_id() { @@ -332,7 +338,13 @@ pub(crate) async fn run_demo_echo( } continue; } - frame = stream.recv_validated(&directory) => frame, + frame = stream.recv_frame() => frame, + }; + // Validation (fence + Redis) runs outside the cancellable region. + let frame = match wire_frame { + Ok(Some(frame)) => stream.validate_received(&directory, frame).await, + Ok(None) => Ok(None), + Err(e) => Err(e.into()), }; match frame { Ok(Some(ReliableFrame::Data(payload))) => { diff --git a/crates/buzz-relay/src/tunnel/reliable.rs b/crates/buzz-relay/src/tunnel/reliable.rs index 5153e104df..fee41cb997 100644 --- a/crates/buzz-relay/src/tunnel/reliable.rs +++ b/crates/buzz-relay/src/tunnel/reliable.rs @@ -270,6 +270,16 @@ impl ReliableMeshStream { self.fenced } + /// Receive the next raw wire frame without validation. + /// + /// Cancel-safe when the transport half is (the iroh half keeps + /// partial-read state across drops; in-memory halves are trivially + /// safe), so it may be raced in a `select!`. Pair with + /// [`Self::validate_received`] run outside the cancellable region. + pub async fn recv_frame(&mut self) -> Result, MeshError> { + self.stream.recv_frame().await + } + /// Send bytes as one or more ordered mesh `Data` frames. pub async fn send_bytes( &mut self, @@ -329,6 +339,13 @@ impl ReliableMeshStream { /// pinned fenced tuple and the Redis directory. This is the reliable-stream /// equivalent of Dawn's hot-path media floor, but authoritative: stale or /// mismatched frames fail the session rather than being dropped silently. + /// + /// NOT cancel-safe: a frame already read off the stream is lost if this + /// future is dropped during fence validation (the Redis await). Callers + /// that race receiving against another branch (drain ticks, shutdown) + /// must select over [`MeshStream::recv_frame`]-level extraction and run + /// [`Self::validate_received`] outside the cancellable region — see + /// `mesh_boot::run_demo_echo`. pub async fn recv_validated( &mut self, directory: &SessionDirectory, @@ -336,7 +353,20 @@ impl ReliableMeshStream { let Some(frame) = self.stream.recv_frame().await? else { return Ok(None); }; + self.validate_received(directory, frame).await + } + /// Validate an already-received wire frame against the stream's pinned + /// fenced tuple and the Redis directory. + /// + /// Split from [`Self::recv_validated`] so callers that must remain + /// cancel-safe can extract the frame first and validate it outside any + /// `select!`/`timeout` region. + pub async fn validate_received( + &mut self, + directory: &SessionDirectory, + frame: MeshStreamFrame, + ) -> Result, ReliableStreamError> { match frame { MeshStreamFrame::Data { fenced, payload } => { let frame = ReliableWireFrame::decode(&payload)?;