fix(swarm): let Node.js clients connect to a firewalled peeroxide server

TESTING.md

@@ -163,12 +163,22 @@ Passing result: Protomux framing and blind-relay control messages are byte-ident
 
 | Suite | Command | Type | Count |
 |-------|---------|------|-------|
-| Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 21 |
-| Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 1 |
-
-**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test)
-
-Passing result: Rust Hyperswarm client connects to a Node.js `hyperswarm` server over the DHT. Full flow: DHT bootstrap → topic announce/lookup → PEER_HANDSHAKE → skip-holepunching (same-host) → UDX stream → SecretStream encrypted channel → receive "hello from node". Verifies end-to-end interoperability of the complete Hyperswarm stack.
+| Unit tests (peeroxide) | `cargo test -p peeroxide --lib` | Unit | 29 |
+| Live interop | `cargo test -p peeroxide --test hyperswarm_interop` | Live | 4 |
+| Server handshake | `cargo test -p peeroxide --test server_handshake_dedup` | Integration | 5 |
+
+**Node helper**: `tests/node/hyperswarm-server.js` (spawned automatically by live interop test). Flags: `--remote-client` (the Node client skips the same-host LAN shortcut and reports `FIREWALL_UNKNOWN`, as a client behind a NAT on another network would; one more DHT node is started so it can classify its NAT) `--open-client` (the Node client's DHT node is reachable, like a peer with a public address, but ephemeral, so it reaches the server through relays) and `--direct-client` (persistent, reporting `FIREWALL_UNKNOWN` and skipping the LAN shortcut: on this small testnet it stores the server's announce and sends its handshake straight to the server; through a relay it would have to holepunch). A `stats` command returns the Node DHT's punch counters (`punches.open` / `consistent` / `random`), which the tests use to check which side dialled.
+
+Passing result:
+- `hyperswarm_cross_language_connect`: Rust Hyperswarm client connects to a Node.js `hyperswarm` server over the DHT. Full flow: DHT bootstrap → topic announce/lookup → PEER_HANDSHAKE → skip-holepunching (same-host) → UDX stream → SecretStream encrypted channel → receive "hello from node". Verifies end-to-end interoperability of the complete Hyperswarm stack.
+- `hyperswarm_node_client_holepunches_to_rust_server`: a firewalled Node.js client connects to a firewalled Rust server. The test first waits for the server to classify itself as firewalled (`CONSISTENT`/`RANDOM`). The client sends one PEER_HANDSHAKE through two relays in parallel, then holepunches through the relays the server advertised (Node's punch counters confirm a holepunch and no direct dial); data flows both ways.
+- `hyperswarm_open_node_client_connects_to_rust_server`: a reachable Node.js client reaches the firewalled Rust server through relays, sees an advertised holepunch, and waits for the server to dial it, as a Node server would. Node's punch counters confirm the client neither dialled nor holepunched; on loopback a client dial would also succeed, so this is what shows the server dialled.
+- `hyperswarm_direct_node_client_connects_to_rust_server`: a persistent Node.js client that reports `FIREWALL_UNKNOWN` and skips the LAN shortcut sends its handshake straight to the Rust server (no relay); both sides dial and data flows both ways. Node's punch counters confirm no holepunch, which a relayed attempt would have needed.
+- `server_handshake_dedup::duplicate_peer_handshake_gets_the_same_reply`: a Rust client sends the same PEER_HANDSHAKE twice straight to a swarm server and gets byte-identical replies.
+- `server_handshake_dedup::duplicate_through_two_relays_keeps_the_staged_holepunch`: the same PEER_HANDSHAKE reaches a firewalled swarm server through two of its relays. Both copies get the same reply, and a PEER_HOLEPUNCH for the first reply's id is still answered, so the second copy did not replace the staged holepunch session.
+- `server_handshake_dedup::handshake_at_max_peers_is_not_cached_or_staged`: a firewalled server with `max_peers = 0` answers each relayed copy with a fresh reply (nothing cached) and stages no holepunch.
+- `server_handshake_dedup::handshake_from_connected_peer_releases_its_staged_holepunch`: a new relayed handshake from an already-connected peer is staged (the server only learns the peer's key from Noise), then released: its reply is not cached and a PEER_HOLEPUNCH for its id gets no answer.
+- `server_handshake_dedup::rust_client_connects_directly_to_firewalled_server`: a Rust client sends its handshake straight to a firewalled swarm server; the server dials it and data flows both ways (guards Rust to Rust direct connections against the server-side dial).
 
 ### Live Network Validation
 

libudx/src/native/stream.rs

@@ -1585,7 +1585,7 @@ async fn process_incoming(
                     // Fail all pending writes with timeout error
                     let pending: Vec<_> = {
                         let mut guard = inner.lock().unwrap_or_else(|e| e.into_inner());
-                        guard.pending_writes.drain(..).collect()
+                        std::mem::take(&mut guard.pending_writes)
                     };
                     for mut pw in pending {
                         if let Some(tx) = pw.tx.take() {

peeroxide-cli/src/cmd/chat/tui/status.rs

@@ -7,6 +7,17 @@
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 
+/// Decrement `counter`, saturating at zero.
+///
+/// `fetch_update` is deprecated in favour of `try_update` (Rust 1.99), which
+/// does not exist at our MSRV (1.85).
+#[allow(deprecated)]
+fn saturating_dec(counter: &AtomicUsize) {
+    let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
+        Some(v.saturating_sub(1))
+    });
+}
+
 use arc_swap::ArcSwap;
 use tokio::sync::Notify;
 
@@ -97,11 +108,7 @@ impl StatusState {
     /// Decrement `send_pending` (saturating) and notify.
     pub fn dec_send_pending(&self) {
         // saturating: don't wrap if mismatched inc/dec ever sneak in.
-        let _ = self.send_pending.fetch_update(
-            Ordering::Relaxed,
-            Ordering::Relaxed,
-            |v| Some(v.saturating_sub(1)),
-        );
+        saturating_dec(&self.send_pending);
         self.dirty.notify_one();
     }
 
@@ -124,11 +131,7 @@ impl StatusState {
 
     /// Decrement the in-flight counter (saturating at zero).
     pub fn dec_recv_in_flight(&self) {
-        let _ = self
-            .recv_pending
-            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
-                Some(v.saturating_sub(1))
-            });
+        saturating_dec(&self.recv_pending);
         self.dirty.notify_one();
     }
 
@@ -141,11 +144,7 @@ impl StatusState {
 
     /// Decrement the any-DHT-op counter (saturating at zero).
     pub fn dec_dht_active(&self) {
-        let _ = self
-            .dht_active
-            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
-                Some(v.saturating_sub(1))
-            });
+        saturating_dec(&self.dht_active);
         self.dirty.notify_one();
     }
 

peeroxide-dht/CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `HyperDhtHandle::current_relays()`: the relays of the last self-announce, each paired with our own address as that relay saw it (`RelayInfo { relay_address, peer_address }`), mirroring Node hyperdht's `Announcer.relays`. A server advertises these in the `holepunch.relays` of its handshake reply; the relay forwards a client's PEER_HOLEPUNCH to `peer_address`. `current_relay_addresses()` is unchanged.
+- `QueryReply::to`: our own address as the responding node saw it (the `to` field of its reply).
+
 ## [1.7.3](https://github.com/Rightbracket/peeroxide/compare/peeroxide-dht-v1.7.2...peeroxide-dht-v1.7.3) - 2026-07-12
 
 ### Other

peeroxide-dht/src/hyperdht.rs

@@ -24,7 +24,7 @@ use crate::hyperdht_messages::{
     decode_lookup_raw_reply_from_bytes, decode_mutable_get_response_from_bytes,
     encode_announce_to_bytes, encode_hyper_peer_to_bytes,
     encode_mutable_put_request_to_bytes, AnnounceMessage, HandshakeMessage, HolepunchMessage,
-    HolepunchPayload, HyperPeer, MutablePutRequest, NoisePayload, RelayThroughInfo,
+    HolepunchPayload, HyperPeer, MutablePutRequest, NoisePayload, RelayInfo, RelayThroughInfo,
     SecretStreamInfo, UdxInfo, ANNOUNCE, ERROR_ABORTED, ERROR_NONE, ERROR_TRY_LATER, FIND_PEER,
     FIREWALL_OPEN, FIREWALL_UNKNOWN, IMMUTABLE_GET, IMMUTABLE_PUT, LOOKUP, MUTABLE_GET,
     MUTABLE_PUT, PEER_HANDSHAKE, PEER_HOLEPUNCH, UNANNOUNCE,
@@ -518,7 +518,7 @@ pub struct HyperDhtHandle {
     router: Arc<Mutex<Router>>,
     server_tx: mpsc::UnboundedSender<ServerEvent>,
     admin_tx: mpsc::UnboundedSender<AdminRequest>,
-    current_relay_addresses: Arc<Mutex<Vec<Ipv4Peer>>>,
+    current_relays: Arc<Mutex<Vec<RelayInfo>>>,
 }
 
 impl HyperDhtHandle {
@@ -595,7 +595,7 @@ impl HyperDhtHandle {
             .await?;
 
         let mut closest_nodes = Vec::new();
-        let mut accepted_relays: Vec<Ipv4Peer> = Vec::new();
+        let mut accepted_relays: Vec<RelayInfo> = Vec::new();
 
         for reply in &replies {
             closest_nodes.push(reply.from.clone());
@@ -655,11 +655,15 @@ impl HyperDhtHandle {
 
             // Accumulate this acker into handle-state relay list (Node
             // `Announcer._commit`: `if (relayAddresses.length < 3)
-            // relayAddresses.push({ host: msg.from.host, port: msg.from.port })`).
+            // relayAddresses.push({ host: msg.from.host, port: msg.from.port })`
+            // and `relays.push({ relayAddress: msg.from, peerAddress: msg.to })`).
             // We push only once per accepted reply and cap at 3. Dedup is
             // implicit — each `reply.from` is a distinct DHT node.
             if accepted_relays.len() < 3 {
-                accepted_relays.push(reply.from.clone());
+                accepted_relays.push(RelayInfo {
+                    relay_address: reply.from.clone(),
+                    peer_address: reply.to.clone(),
+                });
             }
         }
 
@@ -671,7 +675,7 @@ impl HyperDhtHandle {
         // the relay list the topic-announce reads on a subsequent refresh.
         // See peer_discovery::do_refresh for the parallel-join order.
         if target == hash(&key_pair.public_key) {
-            if let Ok(mut guard) = self.current_relay_addresses.lock() {
+            if let Ok(mut guard) = self.current_relays.lock() {
                 *guard = accepted_relays;
             }
         }
@@ -1073,7 +1077,25 @@ impl HyperDhtHandle {
     ///
     /// Mirrors Node Hyperswarm's `Server.relayAddresses` / `Announcer.relayAddresses`.
     pub fn current_relay_addresses(&self) -> Vec<crate::messages::Ipv4Peer> {
-        self.current_relay_addresses
+        self.current_relays()
+            .into_iter()
+            .map(|relay| relay.relay_address)
+            .collect()
+    }
+
+    /// The same relays as [`Self::current_relay_addresses`], each paired with
+    /// our own address as that relay saw it when it accepted the announce.
+    ///
+    /// This is what a server advertises in the `holepunch.relays` of its
+    /// handshake reply: a client sends its PEER_HOLEPUNCH to `relay_address`,
+    /// and the relay forwards it to `peer_address`, so `peer_address` must be
+    /// how that relay reaches *us*. Empty until the first `announce()` on
+    /// `hash(public_key)` completes.
+    ///
+    /// Mirrors Node Hyperswarm's `Announcer.relays`
+    /// (`{ relayAddress: msg.from, peerAddress: msg.to }`).
+    pub fn current_relays(&self) -> Vec<RelayInfo> {
+        self.current_relays
             .lock()
             .map(|g| g.clone())
             .unwrap_or_default()
@@ -2493,7 +2515,7 @@ pub async fn spawn(
         router,
         server_tx,
         admin_tx,
-        current_relay_addresses: Arc::new(Mutex::new(Vec::new())),
+        current_relays: Arc::new(Mutex::new(Vec::new())),
     };
     Ok((join, handle, server_rx))
 }
@@ -3371,6 +3393,10 @@ mod tests {
             handle.current_relay_addresses().is_empty(),
             "current_relay_addresses should be empty before any announce()"
         );
+        assert!(
+            handle.current_relays().is_empty(),
+            "current_relays should be empty before any announce()"
+        );
 
         let _ = handle.destroy().await;
     }
@@ -3393,13 +3419,16 @@ mod tests {
         let clone = handle.clone();
 
         {
-            let mut guard = handle
-                .current_relay_addresses
-                .lock()
-                .expect("lock current_relay_addresses");
-            *guard = vec![Ipv4Peer {
-                host: "1.2.3.4".to_string(),
-                port: 49737,
+            let mut guard = handle.current_relays.lock().expect("lock current_relays");
+            *guard = vec![RelayInfo {
+                relay_address: Ipv4Peer {
+                    host: "1.2.3.4".to_string(),
+                    port: 49737,
+                },
+                peer_address: Ipv4Peer {
+                    host: "5.6.7.8".to_string(),
+                    port: 61000,
+                },
             }];
         }
 
@@ -3408,7 +3437,81 @@ mod tests {
         assert_eq!(via_clone[0].host, "1.2.3.4");
         assert_eq!(via_clone[0].port, 49737);
 
+        let relays = clone.current_relays();
+        assert_eq!(relays.len(), 1);
+        assert_eq!(relays[0].relay_address, via_clone[0]);
+        assert_eq!(relays[0].peer_address.host, "5.6.7.8");
+        assert_eq!(relays[0].peer_address.port, 61000);
+
+        let _ = handle.destroy().await;
+    }
+
+    /// Each relay is paired with our own address as that relay saw it (Node
+    /// `Announcer.relays`), not with anyone else's: that is where it forwards
+    /// a client's PEER_HOLEPUNCH.
+    #[tokio::test]
+    async fn current_relays_pair_each_relay_with_our_observed_address() {
+        use libudx::UdxRuntime;
+
+        // A reachable, persistent bootstrap node that accepts announces, and a
+        // firewalled ephemeral node that announces through it.
+        let local = |bootstrap: Vec<String>, firewalled: bool| HyperDhtConfig {
+            dht: crate::rpc::DhtConfig {
+                bootstrap,
+                port: 0,
+                host: "127.0.0.1".to_string(),
+                firewalled,
+                ephemeral: Some(firewalled),
+                ..crate::rpc::DhtConfig::default()
+            },
+            ..HyperDhtConfig::default()
+        };
+
+        let runtime = UdxRuntime::new().expect("runtime");
+        let (_bs_task, bs, _bs_rx) = spawn(&runtime, local(vec![], false))
+            .await
+            .expect("spawn bootstrap");
+        let bs_port = bs.local_port().await.expect("bootstrap port");
+
+        let bootstrap = vec![format!("127.0.0.1:{bs_port}")];
+        let (_task, handle, _rx) = spawn(&runtime, local(bootstrap, true))
+            .await
+            .expect("spawn announcer");
+        handle.bootstrapped().await.expect("bootstrapped");
+
+        let key_pair = KeyPair::generate();
+        handle
+            .announce(hash(&key_pair.public_key), &key_pair, &[])
+            .await
+            .expect("announce");
+
+        let relays = handle.current_relays();
+        assert!(!relays.is_empty(), "the bootstrap node should accept the announce");
+
+        let ours = handle
+            .server_socket()
+            .await
+            .expect("server socket")
+            .expect("bound socket")
+            .local_addr()
+            .await
+            .expect("local addr");
+        for relay in &relays {
+            assert_eq!(relay.relay_address.port, bs_port);
+            assert_eq!(relay.peer_address.host, "127.0.0.1");
+            assert_eq!(
+                relay.peer_address.port,
+                ours.port(),
+                "peer_address must be our own address as the relay saw it"
+            );
+        }
+        assert_eq!(
+            handle.current_relay_addresses(),
+            relays.iter().map(|r| r.relay_address.clone()).collect::<Vec<_>>()
+        );
+
         let _ = handle.destroy().await;
+        let _ = bs.destroy().await;
     }
 }
 

peeroxide-dht/src/protomux.rs

@@ -780,7 +780,7 @@ fn handle_command(cmd: MuxCommand, state: &mut MuxState, write_queue: &mut VecDe
             if state.cork_depth > 0 {
                 state.cork_depth -= 1;
                 if state.cork_depth == 0 && !state.batch.is_empty() {
-                    let entries: Vec<(u64, Vec<u8>)> = state.batch.drain(..).collect();
+                    let entries: Vec<(u64, Vec<u8>)> = std::mem::take(&mut state.batch);
                     let frame = encode_batch(&entries);
                     if !frame.is_empty() {
                         write_queue.push_back(frame);
@@ -860,7 +860,7 @@ fn create_channel_internal(
 
         // Drain any buffered messages
         if let Some(Some(rc)) = state.remote.get_mut(remote_id as usize - 1) {
-            let pending: Vec<_> = rc.pending.drain(..).collect();
+            let pending: Vec<_> = std::mem::take(&mut rc.pending);
             for (msg_type, data) in pending {
                 let _ = event_tx.send(ChannelEvent::Message {
                     message_type: msg_type as u32,

peeroxide-dht/src/query.rs

@@ -52,6 +52,9 @@ pub struct QueryReply {
     pub value: Option<Vec<u8>>,
     /// Measured round-trip time for this reply.
     pub rtt: Duration,
+    /// Our own address as the responder saw it (the `to` field of its reply).
+    /// Behind a NAT this is the public mapping that node can reach us on.
+    pub to: Ipv4Peer,
 }
 
 /// The final result of a completed query.
@@ -76,6 +79,7 @@ pub(crate) struct IoResponseData {
     pub error: u64,
     pub value: Option<Vec<u8>>,
     pub rtt: Duration,
+    pub to: Ipv4Peer,
 }
 
 /// A request produced by the query engine for the [`crate::rpc::DhtNode`] to send.
@@ -270,6 +274,7 @@ impl Query {
                         error: data.error,
                         value: data.value,
                         rtt: data.rtt,
+                        to: data.to,
                     };
                     self.push_closest(reply);
                 }
@@ -734,6 +739,7 @@ mod tests {
             error: 0,
             value: None,
             rtt: Duration::from_millis(5),
+            to: peer("127.0.0.1", 1),
         }
     }
 

peeroxide-dht/src/rpc.rs

@@ -818,6 +818,7 @@ impl DhtNode {
                     error,
                     value: value.clone(),
                     rtt,
+                    to: to.clone(),
                 };
 
                 if let Some((query_id, is_commit)) = self.tid_to_query.remove(&tid) {