Max/lwd stream patch

common/nym-lp-data/src/packet/frame.rs

@@ -135,14 +135,17 @@ pub enum SphinxStreamMsgType {
     Open = 0,
     /// Data on an existing stream.
     Data = 1,
+    /// Acknowledges an accepted Open. Sent by the listener side; peers
+    /// without establishment support drop it during parsing.
+    OpenAck = 2,
 }
 
 /// Parsed form of the 14-byte `frame_attributes` for `LpFrameKind::SphinxStream`.
 ///
 /// Wire layout (big-endian):
 /// ```text
 /// [0..8 ) stream_id    : u64
-/// [8    ) msg_type      : u8   (0 = Open, 1 = Data)
+/// [8    ) msg_type      : u8   (0 = Open, 1 = Data, 2 = OpenAck)
 /// [9..13) sequence_num  : u32
 /// [13   ) reserved      : u8
 /// ```
@@ -173,6 +176,7 @@ impl SphinxStreamFrameAttributes {
         let msg_type = match attrs[8] {
             0 => SphinxStreamMsgType::Open,
             1 => SphinxStreamMsgType::Data,
+            2 => SphinxStreamMsgType::OpenAck,
             other => {
                 return Err(MalformedLpPacketError::DeserialisationFailure(format!(
                     "invalid stream msg_type: {other}"
@@ -326,3 +330,54 @@ impl ForwardPacketData {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn stream_attributes_roundtrip_all_msg_types() {
+        for msg_type in [
+            SphinxStreamMsgType::Open,
+            SphinxStreamMsgType::Data,
+            SphinxStreamMsgType::OpenAck,
+        ] {
+            let attrs = SphinxStreamFrameAttributes {
+                stream_id: 0xDEAD_BEEF_CAFE_F00D,
+                msg_type,
+                sequence_num: 0x0102_0304,
+            };
+            let parsed = SphinxStreamFrameAttributes::parse(&attrs.encode()).unwrap();
+            assert_eq!(parsed, attrs);
+        }
+    }
+
+    #[test]
+    fn stream_attributes_reject_unknown_msg_type() {
+        // A peer without a given extension must fail parsing cleanly so the
+        // frame is dropped rather than misinterpreted.
+        let attrs = SphinxStreamFrameAttributes {
+            stream_id: 1,
+            msg_type: SphinxStreamMsgType::Open,
+            sequence_num: 0,
+        };
+        let mut encoded = attrs.encode();
+        encoded[8] = 3; // first unassigned discriminant
+        assert!(SphinxStreamFrameAttributes::parse(&encoded).is_err());
+        encoded[8] = 0xFF;
+        assert!(SphinxStreamFrameAttributes::parse(&encoded).is_err());
+    }
+
+    #[test]
+    fn sequence_num_survives_roundtrip_at_bounds() {
+        for sequence_num in [0, 1, u32::MAX] {
+            let attrs = SphinxStreamFrameAttributes {
+                stream_id: 42,
+                msg_type: SphinxStreamMsgType::Data,
+                sequence_num,
+            };
+            let parsed = SphinxStreamFrameAttributes::parse(&attrs.encode()).unwrap();
+            assert_eq!(parsed.sequence_num, sequence_num);
+        }
+    }
+}

documentation/docs/pages/developers/rust/stream/tutorial.mdx

@@ -146,15 +146,23 @@ async fn main() {
     println!("Client address: {}", client.nym_address());
 
     // Open a stream to the server.
-    // The second argument (None) uses the default number of reply SURBs.
+    // The second argument is the reply SURB count; None uses the default.
+    // The server acknowledges the stream over these SURBs, so at least one
+    // is required. With zero, the acknowledgement has no way back.
     let mut stream = client.open_stream(server_addr, None).await.unwrap();
     println!("Stream opened: {}", stream.id());
 
-    // Give the Open message time to traverse the mixnet and reach the server.
-    // open_stream() returns immediately after sending; it doesn't wait for
-    // the server to accept. Writing too soon risks the data arriving before
-    // the Open, which the server would drop.
-    tokio::time::sleep(Duration::from_secs(5)).await;
+    // Wait for the server to acknowledge before writing. open_stream() returns
+    // as soon as the Open is sent; it does not wait for the server to accept.
+    // wait_established resolves when the acknowledgement arrives, or when the
+    // server's first data does, so we avoid writing before the server is ready.
+    //
+    // A timeout leaves the peer's state unknown: an older SDK without
+    // acknowledgement support, or a slow or absent peer. wait_established is
+    // best-effort, so log the timeout and proceed.
+    if let Err(e) = stream.wait_established(Duration::from_secs(30)).await {
+        println!("Establishment not confirmed: {e}; proceeding anyway");
+    }
 
     // Send three payloads of different sizes and verify the echo.
     // Random bytes show that streams are binary-safe, not just text.
@@ -187,6 +195,10 @@ async fn main() {
 }
 ```
 
+<Callout type="info">
+`wait_established(timeout)` confirms the server accepted the stream. The server sends its acknowledgement and the echo data over reply SURBs, so `open_stream` needs at least one; the `None` default provides them. With zero reply SURBs the server cannot answer at all: the acknowledgement never arrives and no echo returns. When SURBs are available, a timeout instead means the peer state is unknown, for example an older SDK, so do not discard the stream on a timeout alone.
+</Callout>
+
 ## Step 4: Run it
 
 In one terminal, start the server:
@@ -254,6 +266,7 @@ See the [Architecture](./architecture) page for the full technical details.
 - `client.listener()` activates stream mode and returns a `MixnetListener`
 - `listener.accept()` blocks until a remote peer opens a stream
 - `client.open_stream(recipient, surbs)` opens an outbound stream to a Nym address
+- `stream.wait_established(timeout)` confirms the server accepted the stream; it needs at least one reply SURB, and a timeout means the peer's state is unknown, not that the stream failed
 - `MixnetStream` implements `AsyncRead + AsyncWrite`, so standard tokio I/O works unchanged
 - Multiple streams are multiplexed over a single client
 - Streams deregister on `drop`; no close handshake is needed
@@ -339,11 +352,15 @@ async fn main() {
     let mut client = mixnet::MixnetClient::connect_new().await.unwrap();
     println!("Client address: {}", client.nym_address());
 
+    // None uses the default reply SURB count; the server needs at least one
+    // to acknowledge the stream (see Step 3).
     let mut stream = client.open_stream(server_addr, None).await.unwrap();
     println!("Stream opened: {}", stream.id());
 
-    // Wait for the Open message to reach the server through the mixnet
-    tokio::time::sleep(Duration::from_secs(5)).await;
+    // Wait for the server to acknowledge before writing (see Step 3).
+    if let Err(e) = stream.wait_established(Duration::from_secs(30)).await {
+        println!("Establishment not confirmed: {e}; proceeding anyway");
+    }
 
     let sizes = [320, 25_000, 1280];
 

documentation/docs/pages/developers/swizzle.mdx

@@ -55,7 +55,7 @@ Add `nym-swizzle` to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-nym-swizzle = "1.21.5"
+nym-swizzle = "1.21.6"
 ```
 
 **Minimum Rust version:** {RUST_MSRV}+

documentation/docs/pages/developers/swizzle/zcash.mdx

@@ -136,7 +136,7 @@ Add `nym-swizzle-zcash` to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-nym-swizzle-zcash = "1.21.5"
+nym-swizzle-zcash = "1.21.6"
 ```
 
 No network stack and no build script. It compiles for

openspec/changes/add-stream-establishment-ack/design.md

@@ -0,0 +1,61 @@
+# Design: Add Stream Establishment Acknowledgement
+
+## Context
+
+The stream module multiplexes byte streams over a `MixnetClient`: a router task demuxes reconstructed mixnet messages by `StreamId` into per-stream channels, with a per-stream reorder buffer (sequence-numbered `Data` frames), an orphan buffer for `Data` overtaking its `Open`, and a stale-stream reaper driven by a cleanup tick. Failures are reported in-band: `StreamFailure` values travel through the same channel as data, so `recv()` sees them in order and `AsyncRead` fails the stream.
+
+Facts established during exploration:
+
+- **The SURB-ack layer cannot provide liveness, by design.** Each fragment carries a prepaid SURB-ack that the recipient's gateway forwards regardless of whether the payload was pushed to a connected client, stored to disk, or dropped for an unregistered recipient (`nym-node/src/node/mixnet/handler.rs`, `handle_final_hop`; the comments state that a conditional ack would leak the recipient's state). It attests "reached the destination gateway", is consumed entirely by the retransmission machinery in `common/client-core/.../acknowledgement_control/`, and is not surfaced to the SDK. Any liveness signal gated on recipient state at this layer would be an online-status oracle usable by anyone who knows the address. Establishment confirmation must come from a layer where the recipient consents by responding.
+- **Frame parsing rejects unknown msg types cleanly.** An unknown `SphinxStreamMsgType` discriminant fails `SphinxStreamFrameAttributes::parse`, so `decode_stream_message` returns `None` and the router drops the frame as a non-stream message. `OpenAck` sent to an old peer degrades to silence, never to an error.
+- The acceptor's reply path (`InputMessage::new_reply(sender_tag)`) already exists and is reused for the `OpenAck` send.
+
+## Goals / Non-Goals
+
+**Goals:**
+
+- Consumers can distinguish "established" from "unknown" at dial time, without reading logs.
+- Old and new SDKs interoperate in both directions with no coordinated upgrade: an old peer simply never emits the frame this change introduces.
+- The privacy rationale (why acknowledgement is stream-layer) is recorded in the module documentation.
+
+**Non-Goals:**
+
+- Mid-stream liveness (ping/pong, unresponsive-peer failure, arming): deferred to `add-stream-keepalive`.
+- Open retransmission and idempotent re-ack (deferred).
+- A `Gone` frame for fast dead-stream signalling (deferred; silence suffices).
+- Any change to `DEFAULT_NUMBER_OF_SURBS` or non-stream send paths.
+
+## Decisions
+
+All settled interactively on 2026-08-27:
+
+1. **Ack from `accept()`, not from router receipt.** The signal means "a listener took your stream", which is the question consumers ask. Best-effort: a failed ack send (for example SURB starvation from a `reply_surbs = 0` dialer) never fails the accept. The send uses non-blocking `try_send` on the shared input channel, so a channel that is momentarily full cannot stall `accept()`.
+2. **`wait_established` is opt-in; `open_stream` stays non-blocking.** One method, `wait_established(timeout)`, with the caller supplying the budget. No default constant and no builder option: a sensible timeout depends on the application, and the SDK has no basis for choosing one. Timeout is inconclusive by definition (old peer, SURB starvation, or loss) and the stream remains usable after it.
+3. **`established` is a distinct signal from keepalive arming.** Inbound `Data` also resolves `wait_established`, because the peer provably accepted the stream, but this does not arm anything; arming is introduced in `add-stream-keepalive`.
+4. **`OpenAck` stays a distinct frame type** rather than being folded into a later liveness frame; one meaning per variant.
+5. **Passive introspection only**: `last_peer_activity()` exposes the last inbound instant; no active `is_alive()` ping method exists at this layer.
+
+## Backwards compatibility
+
+An old peer does not recognise `OpenAck` and drops it during frame parsing: `wait_established` on the new dialer times out, is reported as inconclusive, and the stream stays usable. An old dialer against a new acceptor receives and silently drops the `OpenAck` the same way, at the cost of one SURB from its pool. Neither direction requires a coordinated upgrade or produces a spurious failure.
+
+## Risks / Trade-offs
+
+- **Multi-fragment `Open`** raises establishment loss-variance slightly (all fragments needed for reconstruction). Nothing times out underneath this `Open` (unlike smoltcp SYN handling over the mixnet), so the cost is latency variance in `wait_established`, not correctness.
+- **The ack consumes one reply SURB per stream.** A dialer attaching the default 10 is unaffected; one attaching zero gets no acknowledgement and `wait_established` times out, which the API documents as inconclusive rather than as failure.
+
+## Deliberately out of scope
+
+- **SURB budgeting.** The acknowledgement is paid for out of the reply SURBs
+  the dialer already attaches to the `Open`, so no SURB-count change is needed
+  to make it work. Sizing `Open` and `Data` SURB counts to the Sphinx packet
+  boundary is a behaviour change for every stream caller and belongs in its own
+  change, kept on `wip/stream-surb-budgeting` with the measurement and its
+  pinning test.
+- **Anything in `common/nymsphinx`.** Reply-SURB serialisation is too
+  fundamental to modify in service of a stream-layer feature.
+- **Mid-stream liveness.** Keepalive is `add-stream-keepalive`.
+- **The IPR path.** `IpMixStream::connect_tunnel` already performs a connect
+  request and response carrying the allocated IPs, which is an establishment
+  handshake one layer up. `OpenAck` would duplicate it. This change targets the
+  client-to-client case, where no application protocol supplies one.

openspec/changes/add-stream-establishment-ack/proposal.md

@@ -0,0 +1,37 @@
+# Add Stream Establishment Acknowledgement
+
+## Why
+
+The SDK stream module (`sdk/rust/nym-sdk/src/mixnet/stream/`) is fire-and-forget. `open_stream` verifies at dial time that the recipient's gateway is routable, sends `Open`, and returns a usable `MixnetStream`; nothing ever confirms that the peer's client is running or that anyone called `accept()`.
+
+This cannot be fixed at a lower layer. Every Sphinx fragment already carries a prepaid SURB-ack, but the recipient's gateway forwards that ack whether the message was delivered to a connected client, stored to disk for an offline one, or dropped because the recipient never registered (`nym-node/src/node/mixnet/handler.rs`, `handle_final_hop`). The code comments state the reason: an ack conditional on recipient state would let anyone who knows an address probe whether that client is online, without its participation. The ack layer is deliberately liveness-blind. Establishment confirmation must therefore be built at the stream layer, where the recipient actively consents by responding.
+
+## What Changes
+
+- Add `OpenAck = 2` to `SphinxStreamMsgType` in `common/nym-lp-data`, with a parse arm and wire tests. Wire-compatible: a peer running current code drops the unknown discriminant cleanly during frame parsing.
+- `MixnetListener::accept()` sends a best-effort `OpenAck` to the dialer through the existing anonymous reply path (the dialer's supplied SURBs), immediately after registering the inbound stream. A failed send never fails the accept.
+- An `established` watch channel per stream. `MixnetStream::wait_established(timeout)` waits for it; the caller supplies the timeout rather than the SDK imposing a default. `open_stream` itself stays non-blocking. A timeout is reported as inconclusive (older peer, SURB starvation, or an unreachable peer) and the stream stays usable afterwards.
+- Inbound `Data` also resolves `wait_established`: data on the stream proves the peer accepted it, covering a lost ack.
+- A passive `MixnetStream::last_peer_activity()` getter reporting the time of the most recent inbound frame; calling it generates no traffic.
+- Documentation requirement: the SURB-ack privacy rationale (the recipient's gateway forwards acks regardless of recipient state precisely so they cannot become an online-status oracle, so liveness must live at the stream layer where the recipient consents by responding) is recorded in the stream module documentation. It motivates the whole design, including the keepalive follow-up in `add-stream-keepalive`.
+
+## Capabilities
+
+### New Capabilities
+
+- `sdk-mixnet-stream`: establishment acknowledgement for the SDK stream module: an `OpenAck` frame type, an opt-in wait for establishment, and the documentation recording why acknowledgement lives at the stream layer.
+
+### Modified Capabilities
+
+<!-- none: no existing spec covers the stream module -->
+
+## Impact
+
+- `common/nym-lp-data/src/packet/frame.rs`: one enum variant and its parse arm; existing variants and layout untouched.
+- `sdk/rust/nym-sdk/src/mixnet/stream/mod.rs`: `OpenAck` router arm, `established` watch state on `StreamEntry`, `accept()` ack send.
+- `sdk/rust/nym-sdk/src/mixnet/stream/mixnet_stream.rs`: `wait_established(timeout)`, `last_peer_activity`.
+- `sdk/rust/nym-sdk/src/mixnet/native_client.rs`: one doc-comment line. No configuration, no SURB-count changes, and `common/nymsphinx` is untouched.
+- `sdk/rust/nym-sdk/src/mixnet/stream/ARCHITECTURE.md`: establishment section with the SURB-ack privacy rationale.
+- Outstanding: stream tutorial updates (unchecked in tasks.md).
+- Wire compatibility: an old peer never sends the ack; `wait_established` times out and the stream stays usable. An old dialer against a new acceptor drops the incoming `OpenAck` silently, at the cost of one SURB from its pool.
+- Follow-up: `add-stream-keepalive` builds on this change to add mid-stream liveness (ping/pong, in-band unresponsive-peer failure).