feat(reachability): gate relay publication with canary quorum

src/dht_network_manager.rs

@@ -27,6 +27,12 @@ use crate::{
     dht::{AdmissionResult, DhtCoreEngine, DhtKey, Key, RoutingTableEvent},
     error::{DhtError, IdentityError, NetworkError},
     network::{NodeConfig, NodeMode},
+    rate_limit::{Engine, SharedEngine},
+    reachability::canary::{
+        RELAY_CANARY_HANDLER_TIMEOUT, RELAY_CANARY_PROTOCOL, RELAY_CANARY_WIRE_TOPIC,
+        RelayCanaryProbeResult, RelayCanaryRequest, RelayCanaryResponse,
+        answer_relay_canary_request, relay_canary_rate_limit_config, validate_relay_canary_request,
+    },
     security::canonicalize_ip,
     self_address::build_self_address_set,
 };
@@ -728,6 +734,12 @@ pub struct DhtNetworkManager {
     /// Self-lookups and foreground/payment/user lookup calls do not use this
     /// semaphore.
     bucket_refresh_lookup_semaphore: Arc<Semaphore>,
+    /// Per-source relay canary rate limiter.
+    ///
+    /// Answering a canary request triggers a cold relay dial, so each
+    /// authenticated source is throttled (see [`relay_canary_rate_limit_config`])
+    /// to stop a peer using this node as a reflection/amplification dialer.
+    relay_canary_rate_limiter: SharedEngine<PeerId>,
     /// Shutdown token for background tasks
     shutdown: CancellationToken,
     /// Handle for the network event handler task
@@ -1633,6 +1645,7 @@ impl DhtNetworkManager {
             identity_failure_cache: Arc::new(IdentityFailureCache::new()),
             pending_peer_dials: Arc::new(DashMap::new()),
             lookup_failures: Arc::new(LookupFailureCoordinator::new()),
+            relay_canary_rate_limiter: Arc::new(Engine::new(relay_canary_rate_limit_config())),
         })
     }
 
@@ -4237,6 +4250,116 @@ impl DhtNetworkManager {
         self.send_dht_request(peer_id, operation, None).await
     }
 
+    /// Ask a peer to cold-dial a freshly acquired relay address.
+    ///
+    /// This intentionally uses the generic request/response transport rather
+    /// than adding a DHT operation: the canary is a reachability proof for the
+    /// acquisition driver, not routing-table state.
+    pub(crate) async fn send_relay_canary_request(
+        &self,
+        peer_id: &PeerId,
+        candidates: &[(MultiAddr, AddressType)],
+        request: RelayCanaryRequest,
+        timeout: Duration,
+    ) -> Result<RelayCanaryResponse> {
+        self.ensure_peer_channel(peer_id, candidates).await?;
+
+        let request_bytes = postcard::to_stdvec(&request)
+            .map_err(|e| P2PError::Serialization(e.to_string().into()))?;
+        let response = self
+            .transport
+            .send_request(peer_id, RELAY_CANARY_PROTOCOL, request_bytes, timeout)
+            .await?;
+        postcard::from_bytes(&response.data)
+            .map_err(|e| P2PError::Serialization(e.to_string().into()))
+    }
+
+    async fn handle_relay_canary_message(&self, source_peer: PeerId, data: Vec<u8>) -> Result<()> {
+        if data.len() > MAX_MESSAGE_SIZE {
+            debug!(
+                "Ignoring oversized relay canary message from {source_peer}: {} bytes (max: {MAX_MESSAGE_SIZE})",
+                data.len()
+            );
+            return Ok(());
+        }
+
+        let Some((message_id, is_response, payload)) =
+            crate::transport_handle::TransportHandle::parse_request_envelope(&data)
+        else {
+            debug!(
+                peer = %source_peer.to_hex(),
+                "Ignoring malformed relay canary request envelope"
+            );
+            return Ok(());
+        };
+
+        if is_response {
+            trace!(
+                message_id = %message_id,
+                peer = %source_peer.to_hex(),
+                "Ignoring relay canary response in request handler"
+            );
+            return Ok(());
+        }
+
+        let request: RelayCanaryRequest = match postcard::from_bytes(&payload) {
+            Ok(request) => request,
+            Err(e) => {
+                debug!(
+                    peer = %source_peer.to_hex(),
+                    error = %e,
+                    "Ignoring malformed relay canary request payload"
+                );
+                return Ok(());
+            }
+        };
+        if let Err(reason) = validate_relay_canary_request(&source_peer, &request) {
+            debug!(
+                peer = %source_peer.to_hex(),
+                reason = %reason.summary(),
+                "Rejecting relay canary request"
+            );
+            return Ok(());
+        }
+        // Answering this request makes us cold-dial `relay_addr`, so throttle
+        // each authenticated source to stop a peer using this node as a
+        // reflection/amplification dialer toward an address of its choosing.
+        if !self.relay_canary_rate_limiter.try_consume_key(&source_peer) {
+            debug!(
+                peer = %source_peer.to_hex(),
+                "Throttling relay canary request from source"
+            );
+            let response = RelayCanaryResponse {
+                result: RelayCanaryProbeResult::WitnessRateLimited,
+            };
+            return self
+                .send_relay_canary_response(&source_peer, &message_id, response)
+                .await;
+        }
+
+        let response = answer_relay_canary_request(self.transport.as_ref(), request).await;
+        self.send_relay_canary_response(&source_peer, &message_id, response)
+            .await
+    }
+
+    async fn send_relay_canary_response(
+        &self,
+        source_peer: &PeerId,
+        message_id: &str,
+        response: RelayCanaryResponse,
+    ) -> Result<()> {
+        let response_bytes = postcard::to_stdvec(&response)
+            .map_err(|e| P2PError::Serialization(e.to_string().into()))?;
+        self.transport
+            .send_response(
+                source_peer,
+                RELAY_CANARY_PROTOCOL,
+                message_id,
+                response_bytes,
+            )
+            .await
+    }
+
     /// Handle DHT response message
     ///
     /// Delivers the response via oneshot channel to the waiting request coroutine.
@@ -4751,6 +4874,48 @@ impl DhtNetworkManager {
                                             }
                                             // _permit dropped here, releasing semaphore slot
                                         });
+                                    } else if topic == RELAY_CANARY_WIRE_TOPIC {
+                                        // Relay canary requests must be authenticated so the
+                                        // response can be routed back through request/response.
+                                        let Some(source_peer) = source else {
+                                            warn!("Ignoring unsigned relay canary request");
+                                            continue;
+                                        };
+                                        let manager_clone = Arc::clone(&self_arc);
+                                        let semaphore = Arc::clone(&self_arc.message_handler_semaphore);
+                                        tokio::spawn(async move {
+                                            let _permit = match semaphore.acquire().await {
+                                                Ok(permit) => permit,
+                                                Err(_) => {
+                                                    warn!("Message handler semaphore closed");
+                                                    return;
+                                                }
+                                            };
+
+                                            match tokio::time::timeout(
+                                                RELAY_CANARY_HANDLER_TIMEOUT,
+                                                manager_clone
+                                                    .handle_relay_canary_message(source_peer, data),
+                                            )
+                                            .await
+                                            {
+                                                Ok(Ok(())) => {}
+                                                Ok(Err(e)) => {
+                                                    warn!(
+                                                        peer = %source_peer.to_hex(),
+                                                        error = %e,
+                                                        "Failed to handle relay canary request"
+                                                    );
+                                                }
+                                                Err(_) => {
+                                                    warn!(
+                                                        timeout = ?RELAY_CANARY_HANDLER_TIMEOUT,
+                                                        peer = %source_peer.to_hex(),
+                                                        "Relay canary request handler timed out"
+                                                    );
+                                                }
+                                            }
+                                        });
                                     }
                                 }
                             },

src/network.rs

@@ -1200,49 +1200,16 @@ impl P2PNode {
             info!("client mode — skipping relay acquisition driver");
         }
 
-        // Spawn background task to forward peer address updates to the DHT.
+        // Drain peer-advertised ADD_ADDRESS updates from the transport layer.
         //
-        // Two event streams are bridged from the transport layer onto DHT
-        // routing-table mutations:
-        //
-        //  - **Relay established**: when THIS node sets up a MASQUE relay,
-        //    perform a DHT self-lookup so the transport's re-advertisement
-        //    loop can ADD_ADDRESS the new relay address to the K closest
-        //    peers — propagating it beyond peers we already happen to be
-        //    connected to.
-        //  - **Peer address update**: when a connected peer advertises a new
-        //    reachable address via ADD_ADDRESS (typically its relay), update
-        //    the DHT routing table so future lookups return that address.
-        //
-        // Both are handled in a `tokio::select!` against the receiver
-        // futures so updates propagate immediately. The previous
-        // implementation polled both queues on a 1-second interval, which
-        // opened a race window in which a freshly-established relay was
-        // invisible to outbound DHT queries until the next tick — causing
-        // the first peers to dial direct (and fail) before learning about
-        // the relay.
-        //
-        // **Slow work isolation**: the relay-propagation path runs an
-        // iterative DHT lookup (`find_closest_nodes_network`) which can
-        // take many seconds. Doing it inline in the select loop would
-        // starve the peer-address-update branch and back up the bounded
-        // forwarder mpsc into drop territory. Instead, the lookup +
-        // publish is detached into its own task per relay event, so the
-        // select loop keeps polling both branches.
-        // DHT_BRIDGE: forward peer-advertised address updates from the
-        // transport layer onto DHT routing table mutations. When a connected
-        // peer's ADD_ADDRESS notification carries a different IP than the
-        // connection's source (i.e., the peer is behind a relay or has
-        // migrated), merge the advertised address into the peer's DHT entry.
-        //
-        // This node's OWN relay state changes are NOT handled here — the
-        // relay acquisition driver (see `reachability::driver`) owns them
-        // directly, so the "relay established" branch no longer belongs to
-        // the bridge. The driver knows the full typed address set for the
-        // self-record; the bridge did not.
+        // Relay-looking ADD_ADDRESS frames are intentionally not merged into
+        // DHT records. Relay reachability must arrive through the sequenced
+        // self-record path owned by `reachability::driver`, after the relay
+        // canary quorum has passed. Accepting transport hints here would let
+        // an unverified relay allocation leak into routing-table gossip before
+        // the canary gate has made a verdict.
         {
             let transport = Arc::clone(&self.transport);
-            let dht = self.adaptive_dht.dht_manager().clone();
             let shutdown = self.shutdown.clone();
             tokio::spawn(async move {
                 loop {
@@ -1255,10 +1222,10 @@ impl P2PNode {
                                 saorsa_transport::shared::normalize_socket_addr(peer_addr);
                             let normalized_adv =
                                 saorsa_transport::shared::normalize_socket_addr(advertised_addr);
-                            // Only update DHT when the advertised IP differs
-                            // from the peer's connection IP. Same-IP updates
-                            // are just different NATted ports (useless for
-                            // symmetric NAT); different-IP means a relay.
+                            // Same-IP updates are just different NATted ports,
+                            // which are useless for symmetric NAT. Different
+                            // IPs are relay-like hints and must not bypass the
+                            // canary-gated DHT publish path.
                             if normalized_peer.ip() == normalized_adv.ip() {
                                 debug!(
                                     "DHT_BRIDGE: dropping same-IP update peer={} addr={}",
@@ -1267,31 +1234,11 @@ impl P2PNode {
                                 );
                                 continue;
                             }
-                            info!(
-                                "DHT_BRIDGE: processing relay update peer={} addr={}",
+                            debug!(
+                                "DHT_BRIDGE: ignoring relay ADD_ADDRESS update pending sequenced canary publish peer={} addr={}",
                                 normalized_peer,
                                 normalized_adv
                             );
-                            // Look up peer ID by address (tries both IPv4 and
-                            // IPv4-mapped IPv6 forms via dual_stack_alternate).
-                            // For symmetric NAT, this may fail because the
-                            // connection's channel key uses a different NATted port.
-                            if let Some(peer_id) = transport.peer_id_for_addr(&normalized_peer).await {
-                                let multi_addr = MultiAddr::quic(normalized_adv);
-                                info!(
-                                    "Updating DHT: peer {} relay address {} (connection was {})",
-                                    peer_id, advertised_addr, peer_addr
-                                );
-                                if !dht
-                                    .touch_legacy_relay_hint_if_unsequenced(&peer_id, &multi_addr)
-                                    .await
-                                {
-                                    debug!(
-                                        "DHT_BRIDGE: ignored legacy relay hint for sequenced peer {} addr {}",
-                                        peer_id, advertised_addr
-                                    );
-                                }
-                            }
                         }
                     }
                 }

src/reachability/acquisition.rs

@@ -15,10 +15,9 @@
 //!
 //! Every non-client node calls [`RelayAcquisition::acquire`] after bootstrap
 //! to establish a relay from a close-group peer. The walker is unaware of
-//! whether the local node is public or private — if a candidate's Direct
-//! address is unreachable (private peer), the QUIC dial fails and the walk
-//! advances to the next-closest peer. "Is this candidate public?" is
-//! inferred ambiently from the dial attempt.
+//! whether the accepted relay is externally useful for third parties; it
+//! only establishes the MASQUE session. The reachability driver runs relay
+//! canaries before publishing the allocated address.
 //!
 //! 1. The caller supplies a pre-filtered list of [`RelayCandidate`]s sorted
 //!    by XOR distance (closest first). Filtering — selecting peers whose