Skip to content

fix: advertise dialable addresses to DHT and harden NAT traversal #71

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mickvandijke wants to merge 1 commit into from
Closed

fix: advertise dialable addresses to DHT and harden NAT traversal #71

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion src/dht/security_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,6 @@ async fn test_ipv4_ip_override_raises_limit() -> anyhow::Result<()> {
engine.set_ip_diversity_config(IPDiversityConfig {
max_per_ip: Some(3),
max_per_subnet: Some(usize::MAX),
..IPDiversityConfig::default()
});

for i in 1..=3u8 {
Expand Down
217 changes: 167 additions & 50 deletions src/dht_network_manager.rs
View file
Open in desktop

Large diffs are not rendered by default.

160 changes: 93 additions & 67 deletions src/network.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,16 @@ const DEFAULT_CONNECTION_TIMEOUT_SECS: u64 = 25;
const BOOTSTRAP_PEER_BATCH_SIZE: usize = 20;

/// Timeout in seconds for waiting on a bootstrap peer's identity exchange.
const BOOTSTRAP_IDENTITY_TIMEOUT_SECS: u64 = 5;
///
/// Identity exchange is two RTTs over a freshly-handshaken QUIC connection
/// plus an ML-DSA-65 signature verification. On a LAN this completes in
/// well under a second; on congested cellular or cross-region links it can
/// blow past 5s with retransmits. The previous 5s default fired
/// spuriously on slow networks during testnet validation, forcing
/// reconnect loops that masqueraded as NAT traversal failures, so we
/// budget enough headroom for two QUIC handshake retries on a high-latency
/// link.
const BOOTSTRAP_IDENTITY_TIMEOUT_SECS: u64 = 15;

/// Serde helper — returns `true`.
const fn default_true() -> bool {
Expand Down Expand Up @@ -1120,51 +1129,71 @@ impl P2PNode {
self.connect_bootstrap_peers(close_group_cache.as_ref())
.await?;

// Spawn background task to forward peer address updates to DHT.
// When a connected peer advertises a new address (e.g., relay), update
// the DHT routing table so future lookups return the new address.
// Spawn background task to forward peer address updates to the DHT.
//
// 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.
//
// Also handles RelayEstablished events: when THIS node sets up a relay,
// it does a DHT self-lookup to connect to K closest peers. The transport
// layer's re-advertisement loop then sends ADD_ADDRESS to those peers,
// propagating the relay address beyond the initial direct connections.
// 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.
{
let transport = Arc::clone(&self.transport);
let dht = self.adaptive_dht.dht_manager().clone();
let shutdown = self.shutdown.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
loop {
tokio::select! {
biased;
_ = shutdown.cancelled() => break,
_ = interval.tick() => {
// Check for relay established events.
// When this node sets up a relay, do a self-lookup to
// connect to K closest peers for relay address propagation.
if let Some(relay_addr) = transport.drain_relay_established().await {
// Normalize IPv6-mapped addresses to IPv4 so the
// published address is dialable by IPv4-only clients.
let normalized = saorsa_transport::shared::normalize_socket_addr(relay_addr);
let relay_multi =
crate::MultiAddr::quic(normalized);
info!(
"DHT_BRIDGE: relay established at {} — self-lookup then PublishAddress to K closest",
relay_addr
);
let own_key = *dht.peer_id().to_bytes();
// Self-lookup to discover K closest peers, then
// send PublishAddress to each. This is O(K) messages
// sent once, not a field on every DHT message.
match dht.find_closest_nodes_network(&own_key, dht.k_value()).await {
relay = transport.recv_relay_established() => {
let Some(relay_addr) = relay else { break };
// Normalize IPv6-mapped addresses to IPv4 so the
// published address is dialable by IPv4-only clients.
let normalized = saorsa_transport::shared::normalize_socket_addr(relay_addr);
let relay_multi = crate::MultiAddr::quic(normalized);
info!(
"DHT_BRIDGE: relay established at {} — spawning self-lookup + PublishAddress",
relay_addr
);
// Detach the slow work so the select loop is
// free to keep polling peer-address updates.
let dht_for_propagation = dht.clone();
tokio::spawn(async move {
let own_key = *dht_for_propagation.peer_id().to_bytes();
match dht_for_propagation
.find_closest_nodes_network(&own_key, dht_for_propagation.k_value())
.await
{
Ok(nodes) => {
info!(
"DHT_BRIDGE: self-lookup found {} nodes — sending PublishAddress",
nodes.len()
);
dht.publish_address_to_peers(
vec![relay_multi],
&nodes,
).await;
dht_for_propagation
.publish_address_to_peers(vec![relay_multi], &nodes)
.await;
}
Err(e) => {
warn!(
Expand All @@ -1173,44 +1202,41 @@ impl P2PNode {
);
}
}
});
}
update = transport.recv_peer_address_update() => {
let Some((peer_addr, advertised_addr)) = update else { break };
info!(
"DHT_BRIDGE: processing update peer={} addr={} same_ip={}",
peer_addr,
advertised_addr,
peer_addr.ip() == advertised_addr.ip()
);
// 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.
if peer_addr.ip() == advertised_addr.ip() {
continue;
}

let updates = transport.drain_peer_address_updates().await;
if !updates.is_empty() {
info!("DHT_BRIDGE: drained {} address updates", updates.len());
}
for (peer_addr, advertised_addr) in updates {
// 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(&peer_addr).await {
let normalized_adv =
saorsa_transport::shared::normalize_socket_addr(advertised_addr);
let multi_addr = crate::MultiAddr::quic(normalized_adv);
info!(
"DHT_BRIDGE: processing update peer={} addr={} same_ip={}",
peer_addr, advertised_addr, peer_addr.ip() == advertised_addr.ip()
"Updating DHT: peer {} relay address {} (connection was {})",
peer_id, advertised_addr, peer_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.
if peer_addr.ip() == advertised_addr.ip() {
continue;
}

// 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.
let peer_id = transport.peer_id_for_addr(&peer_addr).await;
if let Some(peer_id) = peer_id {
let normalized_adv = saorsa_transport::shared::normalize_socket_addr(advertised_addr);
let multi_addr = crate::MultiAddr::quic(normalized_adv);
info!(
"Updating DHT: peer {} relay address {} (connection was {})",
peer_id, advertised_addr, peer_addr
);
dht.touch_node_typed(
&peer_id,
Some(&multi_addr),
crate::dht::AddressType::Relay,
)
.await;
}
dht.touch_node_typed(
&peer_id,
Some(&multi_addr),
crate::dht::AddressType::Relay,
)
.await;
}
}
}
Expand Down
5 changes: 5 additions & 0 deletions src/transport.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,3 +22,8 @@ pub mod saorsa_transport_adapter;

// DHT protocol handler for SharedTransport integration
pub mod dht_handler;

// Observed-address cache: records `ExternalAddressDiscovered` events from the
// transport layer and serves as a frequency- and recency-aware fallback when
// no live connection has an observation.
pub(crate) mod observed_address_cache;
Loading
Loading