Max/lwd stream patch

sdk/rust/nym-sdk/src/error.rs

@@ -2,6 +2,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use nym_ip_packet_requests::ConnectFailureReason;
+use nym_sphinx::addressing::clients::Recipient;
+use nym_topology::NymTopologyError;
 use nym_validator_client::nym_api::error::NymAPIError;
 use nym_validator_client::nyxd::error::NyxdError;
 use std::path::PathBuf;
@@ -108,6 +110,13 @@ pub enum Error {
     #[error("Stream subsystem failed to initialise: reconstructed_receiver unavailable")]
     StreamInitFailure,
 
+    #[error("cannot route to {recipient}: {source}")]
+    UnroutableRecipient {
+        recipient: Box<Recipient>,
+        #[source]
+        source: NymTopologyError,
+    },
+
     #[error("client not connected")]
     IprStreamClientNotConnected,
 

sdk/rust/nym-sdk/src/ipr_wrapper/ip_mix_stream.rs

@@ -248,7 +248,13 @@ impl IpMixStream {
                     return Err(Error::IPRConnectResponseTimeout);
                 }
                 result = stream.recv() => {
-                    let data = result.ok_or(Error::IPRClientStreamClosed)?;
+                    let data = match result.ok_or(Error::IPRClientStreamClosed)? {
+                        Ok(data) => data,
+                        Err(e) => {
+                            debug!("ignoring stream loss during connect: {e}");
+                            continue;
+                        }
+                    };
 
                     // Ignore stragglers from an earlier version; we selected v10
                     // from the node's directory version.
@@ -292,7 +298,13 @@ impl IpMixStream {
                     return Err(Error::IPRConnectResponseTimeout);
                 }
                 result = stream.recv() => {
-                    let data = result.ok_or(Error::IPRClientStreamClosed)?;
+                    let data = match result.ok_or(Error::IPRClientStreamClosed)? {
+                        Ok(data) => data,
+                        Err(e) => {
+                            debug!("ignoring stream loss during connect: {e}");
+                            continue;
+                        }
+                    };
 
                     // Skip frames from another version rather than aborting: in the
                     // v10-to-v9 fallback a late v10 response can land here, and it
@@ -335,15 +347,21 @@ impl IpMixStream {
     /// Handle incoming messages from the mixnet.
     ///
     /// Reads from the underlying `MixnetStream`, parses IPR responses, and
-    /// extracts IP packets. Returns an empty vec on timeout (10 s).
+    /// extracts IP packets. Returns an empty vec on timeout (10 s), on a
+    /// lost or version-mismatched frame, or on an unrecognised response.
     pub async fn handle_incoming(&mut self) -> Result<Vec<Bytes>, Error> {
         let data = match tokio::time::timeout(Duration::from_secs(10), self.stream.recv()).await {
             Err(_) => return Ok(Vec::new()),
             Ok(None) => {
                 self.connected = false;
                 return Err(Error::IPRClientStreamClosed);
             }
-            Ok(Some(data)) => data,
+            Ok(Some(Err(err))) => {
+                // Lost in transit: drop them, TCP inside the tunnel retransmits.
+                warn!("dropping lost mixnet messages: {err}");
+                return Ok(Vec::new());
+            }
+            Ok(Some(Ok(data))) => data,
         };
 
         // The IPR mirrors the connect-time version on all traffic (data included),

sdk/rust/nym-sdk/src/mixnet/native_client.rs

@@ -412,6 +412,13 @@ impl MixnetClient {
     /// # }
     /// ```
     ///
+    /// # Errors
+    ///
+    /// [`Error::UnroutableRecipient`](crate::Error::UnroutableRecipient) if
+    /// the recipient's gateway is not in the current topology. The source
+    /// distinguishes an empty local view (retry shortly) from an unknown
+    /// gateway (the address may be stale).
+    ///
     /// # Cancel safety
     ///
     /// This method is **not** cancel safe. Cancelling after the `Open`

sdk/rust/nym-sdk/src/mixnet/stream/ARCHITECTURE.md

@@ -113,6 +113,10 @@ streams, and listener. Methods: `register_stream`, `remove`,
 - **No `Close` message** — there is no explicit stream-close signal.
   Streams clean up locally via `Drop` and idle timeout. A proper
   close/EOF mechanism requires further protocol work.
-- **Reorder buffer cap** — out-of-order messages are buffered up to
-  `MAX_REORDER_BUFFER` (256) per stream. If a sequence number is
-  permanently lost, the buffer skips ahead once full.
+- **Reorder buffer cap** - out-of-order messages are buffered up to
+  `MAX_REORDER_BUFFER_BYTES` (8 MiB) per stream. A full buffer skips the
+  missing range and reports the loss in-band as `InvalidData`. `recv()`
+  surfaces it once and later messages keep flowing; `AsyncRead` fails
+  the stream permanently. The cap is generous relative to per-tunnel
+  throughput, so a late frame with a retransmit in flight does not trip
+  it.

sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs

@@ -22,7 +22,7 @@ use tokio_util::sync::PollSender;
 use nym_lp_data::packet::frame::SphinxStreamMsgType;
 
 use super::protocol::{encode_stream_message, StreamId};
-use super::StreamMap;
+use super::{StreamFailure, StreamMap};
 
 /// How to address outbound messages on this stream.
 enum Destination {
@@ -47,10 +47,15 @@ pub struct MixnetStream {
     packet_type: Option<PacketType>,
     streams: StreamMap,
 
-    inbound_rx: mpsc::UnboundedReceiver<Vec<u8>>,
+    inbound_rx: mpsc::UnboundedReceiver<Result<Vec<u8>, StreamFailure>>,
     read_buf: BytesMut,
     deregistered: bool,
     next_seq: u32,
+
+    /// Set when `poll_read` hits a lost-data marker. Once set, all reads
+    /// and writes fail. `recv()` never sets this: it returns the error
+    /// once and keeps going.
+    failure: Option<StreamFailure>,
 }
 
 impl MixnetStream {
@@ -62,7 +67,7 @@ impl MixnetStream {
         client_input: ClientInput,
         packet_type: Option<PacketType>,
         streams: StreamMap,
-        inbound_rx: mpsc::UnboundedReceiver<Vec<u8>>,
+        inbound_rx: mpsc::UnboundedReceiver<Result<Vec<u8>, StreamFailure>>,
     ) -> Self {
         let sender = PollSender::new(client_input.input_sender.clone());
         Self {
@@ -78,6 +83,7 @@ impl MixnetStream {
             read_buf: BytesMut::new(),
             deregistered: false,
             next_seq: 0,
+            failure: None,
         }
     }
 
@@ -88,7 +94,7 @@ impl MixnetStream {
         client_input: ClientInput,
         packet_type: Option<PacketType>,
         streams: StreamMap,
-        inbound_rx: mpsc::UnboundedReceiver<Vec<u8>>,
+        inbound_rx: mpsc::UnboundedReceiver<Result<Vec<u8>, StreamFailure>>,
         initial_data: Vec<u8>,
     ) -> Self {
         let mut read_buf = BytesMut::new();
@@ -106,6 +112,7 @@ impl MixnetStream {
             read_buf,
             deregistered: false,
             next_seq: 0,
+            failure: None,
         }
     }
 
@@ -116,13 +123,23 @@ impl MixnetStream {
 
     /// Receive a single message payload directly from the stream channel.
     ///
-    /// Returns `None` on EOF (channel closed). Drains any leftover from
-    /// a prior `AsyncRead` call first.
-    pub async fn recv(&mut self) -> Option<Vec<u8>> {
+    /// Returns `None` on EOF (channel closed). Returns `Some(Err(_))`
+    /// when messages were lost at this point in the sequence; later calls
+    /// return the messages that follow. Drains any leftover from a prior
+    /// `AsyncRead` call first.
+    pub async fn recv(&mut self) -> Option<std::io::Result<Vec<u8>>> {
+        if let Some(failure) = &self.failure {
+            return Some(Err(failure.as_io_error()));
+        }
         if !self.read_buf.is_empty() {
-            return Some(self.read_buf.split().to_vec());
+            return Some(Ok(self.read_buf.split().to_vec()));
         }
-        self.inbound_rx.recv().await
+        Some(
+            self.inbound_rx
+                .recv()
+                .await?
+                .map_err(StreamFailure::as_io_error),
+        )
     }
 
     /// Wrap `data` in the appropriate `InputMessage` for this stream's destination.
@@ -162,6 +179,10 @@ impl AsyncRead for MixnetStream {
         cx: &mut Context<'_>,
         buf: &mut ReadBuf,
     ) -> Poll<std::io::Result<()>> {
+        if let Some(failure) = &self.failure {
+            return Poll::Ready(Err(failure.as_io_error()));
+        }
+
         // Drain spillover first
         if !self.read_buf.is_empty() {
             let n = std::cmp::min(buf.remaining(), self.read_buf.len());
@@ -170,14 +191,18 @@ impl AsyncRead for MixnetStream {
         }
 
         match ready!(self.inbound_rx.poll_recv(cx)) {
-            Some(data) => {
+            Some(Ok(data)) => {
                 let n = std::cmp::min(buf.remaining(), data.len());
                 buf.put_slice(&data[..n]);
                 if n < data.len() {
                     self.read_buf.extend_from_slice(&data[n..]);
                 }
                 Poll::Ready(Ok(()))
             }
+            Some(Err(failure)) => {
+                self.failure = Some(failure);
+                Poll::Ready(Err(failure.as_io_error()))
+            }
             None => Poll::Ready(Ok(())), // EOF
         }
     }
@@ -189,6 +214,10 @@ impl AsyncWrite for MixnetStream {
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<std::io::Result<usize>> {
+        if let Some(failure) = &self.failure {
+            return Poll::Ready(Err(failure.as_io_error()));
+        }
+
         if buf.is_empty() {
             return Poll::Ready(Ok(0));
         }