block · brocoppler · Jul 24, 2026
@@ -28,10 +28,22 @@ impl MeshEndpoint {
         bind_addr: SocketAddr,
     ) -> Result<Self, MeshError> {
         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,

@@ -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<u8>,
+}
+
+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<bool, MeshError> {
+        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,27 +209,28 @@ impl StreamSendHalf for IrohSendHalf {
 impl StreamRecvHalf for IrohRecvHalf {
     fn recv_frame(&mut self) -> crate::BoxFuture<'_, Result<Option<MeshStreamFrame>, 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,
                     max: wire::MAX_STREAM_FRAME as usize,
                 });
             }
 
-            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::<MeshStreamFrame>(&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::<MeshStreamFrame>(&self.buf[4..total]).map(Some);
+            self.buf.clear();
+            frame
         })
     }
 }

@@ -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<MeshStreamFrame>);
+    struct ChanRecvHalf(tokio::sync::mpsc::UnboundedReceiver<MeshStreamFrame>);
+
+    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<Option<MeshStreamFrame>, 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<Option<MeshStream>>,
+    }
+
+    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<dyn std::future::Future<Output = Result<MeshStream, MeshError>> + 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<dyn InboundHandler>) {}
+    }
+
     struct NoopTransport;
 
     impl RelayPeerTransport for NoopTransport {
@@ -258,15 +333,106 @@ 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 {
             return;
         };
         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();

@@ -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))) => {