From 8fc0bb43e15e9ffa99777e478162c101557d505b Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 23:08:48 +0100 Subject: probe when reachable --- src/node.rs | 4 ++++ src/swarm/peer/handshake.rs | 24 ++++++++++++++++++++++++ src/swarm/peer/mod.rs | 3 ++- src/transport/identity.rs | 4 ++++ src/transport/libp2p/book.rs | 3 +++ src/transport/libp2p/driver.rs | 10 ++++++++++ src/transport/libp2p/mod.rs | 4 ++++ src/transport/mod.rs | 7 +++++++ src/transport/paths.rs | 7 +++++++ src/transport/traffic.rs | 4 ++++ 10 files changed, 69 insertions(+), 1 deletion(-) diff --git a/src/node.rs b/src/node.rs index b334882..4597e66 100644 --- a/src/node.rs +++ b/src/node.rs @@ -191,6 +191,10 @@ impl Node { }); trace!("Peer worker task spawned"); + let reachable = Arc::new(tokio::sync::Notify::new()); + state.transport.on_reachable(Arc::clone(&reachable)); + tokio::spawn(peer::probe_when_reachable(state.clone(), reachable)); + tokio::spawn(remake::run_worker(state.clone())); trace!("Remake worker task spawned"); diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs index 6bd4784..072545b 100644 --- a/src/swarm/peer/handshake.rs +++ b/src/swarm/peer/handshake.rs @@ -304,6 +304,30 @@ pub(super) async fn discover_peers( } } +/// Whenever others may reach this node where they couldn't before +/// (`Transport::on_reachable`: a relay granted it a slot), the members it can't reach +/// are probed at once (`probe_now`): they may have given up on it meanwhile, waiting +/// for their next probe, up to `dormant-probe-max` away. Two nodes behind NATs that +/// restart together each fail to reach the other until it holds a slot; the second to +/// get one reaches the first this way. Runs for the node's life. +pub(crate) async fn probe_when_reachable(state: AppState, reachable: Arc) { + loop { + reachable.notified().await; + let mut probed = 0; + for (peer, _) in state.peers_snapshot().await { + if probe_now(&state, &peer).await { + probed += 1; + } + } + if probed > 0 { + debug!( + probed, + "Reachable anew; probing the members we can't reach now" + ); + } + } +} + /// Brings `peer`'s next probe forward to now, if it is a member we can't reach (dormant, /// or unanswered HELLOs) whose probe is pending: the peer worker sends it (woken for /// it). Whether it was. diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs index d1dea9b..cf427a2 100644 --- a/src/swarm/peer/mod.rs +++ b/src/swarm/peer/mod.rs @@ -22,7 +22,8 @@ mod worker; pub(crate) use forget::{affects_forget_records, forget_unless_alive, forgotten_path}; pub(crate) use handshake::{ - apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once, + apply_neighbour_change, peer_swarm, probe_when_reachable, rehandshake, remember_peer, + send_hello_once, }; pub(crate) use inbound::handle_message; pub(crate) use keepalive::{announce, farewell, send_status}; diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 7b4a8c7..1df84ec 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -642,6 +642,10 @@ impl Transport for IdentityTransport { self.shared.paths.on_seeds_found(found) } + fn on_reachable(&self, reachable: std::sync::Arc) { + self.shared.paths.on_reachable(reachable) + } + /// A node ID resolves to itself if this node knows where it is; anything else is /// an address (`:...`) or what one of the transports resolves to one (the /// first that does), asked WHO. diff --git a/src/transport/libp2p/book.rs b/src/transport/libp2p/book.rs index 3de8cef..0c61714 100644 --- a/src/transport/libp2p/book.rs +++ b/src/transport/libp2p/book.rs @@ -113,6 +113,9 @@ pub(super) struct Book { pub(super) discovered: Vec, /// Told when a seed is found (`Transport::on_seeds_found`). pub(super) seeds_found: Option>, + /// Told when others may reach this node where they couldn't before + /// (`Transport::on_reachable`). + pub(super) reachable: Option>, } /// What `Libp2pStats` counts since start. diff --git a/src/transport/libp2p/driver.rs b/src/transport/libp2p/driver.rs index 39a2a5d..6ef04e7 100644 --- a/src/transport/libp2p/driver.rs +++ b/src/transport/libp2p/driver.rs @@ -420,6 +420,10 @@ pub(super) async fn run( book.failed_relays.remove(&relay_peer_id); if !renewal { book.counts.slots_granted += 1; + // Peers that couldn't reach us may now, through it. + if let Some(reachable) = &book.reachable { + reachable.notify_one(); + } } } if !renewal { @@ -440,6 +444,12 @@ pub(super) async fn run( (Reachability::Public, Some(address.clone())) } }; + if matches!(new, NatStatus::Public(_)) + && !matches!(old, NatStatus::Public(_)) + && let Some(reachable) = &book.reachable + { + reachable.notify_one(); + } } nat = new; } diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs index 3ebe7f6..056b1b2 100644 --- a/src/transport/libp2p/mod.rs +++ b/src/transport/libp2p/mod.rs @@ -500,6 +500,10 @@ impl Transport for Libp2pTransport { lock(&self.book).seeds_found = Some(found); } + fn on_reachable(&self, reachable: Arc) { + lock(&self.book).reachable = Some(reachable); + } + /// Nodes found on the LAN (mDNS) that speak continuum. fn discovered_seeds(&self) -> Vec { std::mem::take(&mut lock(&self.book).discovered) diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 1de0633..96f7651 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -831,6 +831,13 @@ pub(crate) trait Transport: Send + Sync { let _ = found; } + /// Has the transport tell `reachable` whenever others may reach this node where they + /// couldn't before (libp2p: a relay granted it a slot, or AutoNAT found it public). + /// By default it never does. + fn on_reachable(&self, reachable: std::sync::Arc) { + let _ = reachable; + } + /// Whether this node has a way to reach `peer` at all: an address of it on one of /// its transports. Members it can't reach aren't its neighbours (`swarm::topology`). /// By default yes, as for `local_addresses`. diff --git a/src/transport/paths.rs b/src/transport/paths.rs index ba8581f..164574d 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -708,6 +708,13 @@ impl Paths { } } + /// Each transport's (`Transport::on_reachable`). + pub(crate) fn on_reachable(&self, reachable: std::sync::Arc) { + for link in &self.links { + link.inner.on_reachable(reachable.clone()); + } + } + /// The address of the node a seed leads to, as paths keep it: an address of ours /// (`:...`) whose transport reads it at once; with a scheme of ours, what /// that transport resolves the rest to (an address the transport can't read as an diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 21acca6..bf9438f 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -351,6 +351,10 @@ impl Transport for CountingTransport { self.inner.on_seeds_found(found) } + fn on_reachable(&self, reachable: std::sync::Arc) { + self.inner.on_reachable(reachable) + } + fn transports(&self) -> Vec { self.inner.transports() } -- cgit v1.3.1