bootstrap: replace pull by progress notify

11 files changed, 52 insertions, 2 deletions

diff --git a/src/node.rs b/src/node.rs
index e3a832a..6f55ed5 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -128,6 +128,7 @@ impl Node {
             config: Arc::new(config.config_reports),
             neighbours: Default::default(),
             topology_changed: Default::default(),
+            join_progress: Default::default(),
             pulling: Default::default(),
             announce_pending: Default::default(),
             rtt: Default::default(),
diff --git a/src/state/host.rs b/src/state/host.rs
index e7921e4..5751998 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -330,6 +330,7 @@ impl Host for AppState {
                 .expect("rejoin task panicked")
             };
             state.publish_change();
+            state.join_progress.notify_one();
             if resumed {
                 crate::swarm::peer::announce(&state);
             }
@@ -360,6 +361,7 @@ impl Host for AppState {
                 .expect("rejoin task panicked")
             };
             state.after_epoch_switch().await;
+            state.join_progress.notify_one();
             if resumed {
                 crate::swarm::peer::announce(&state);
             }
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 7c1299f..2902234 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -151,6 +151,9 @@ pub(crate) struct AppState {
     /// Something the neighbours are computed from may have changed: the member list or a
     /// peer's state (`swarm::topology::run_worker`, `swarm::topology::change_peer`).
     pub(crate) topology_changed: Arc<tokio::sync::Notify>,
+    /// Something a joining node waits for may have happened: the swarm ID learned, the
+    /// initial sync done, a rejoin ended, a seed found (`swarm::bootstrap`).
+    pub(crate) join_progress: Arc<tokio::sync::Notify>,
 
     /// While rejoining a newer epoch, the previous document, served read-only (and
     /// marked stale) instead of answering 503.
@@ -276,6 +279,7 @@ impl AppState {
                     discovered_swarm_id: Some(swarm_id),
                 };
                 debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
+                self.join_progress.notify_one();
                 Ok(())
             }
             NodeLifecycle::Joining {
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 848201e..4f6c072 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -34,7 +34,6 @@ use crate::remake::snapshot;
 use crate::state::AppState;
 use crate::swarm::peer::{remember_peer, send_hello_once};
 
-const BOOTSTRAP_POLL: Duration = Duration::from_millis(500);
 /// How often a node that is still joining says why.
 const BOOTSTRAP_REPORT: Duration = Duration::from_secs(300);
 /// Retrying seeds that didn't resolve: first after this, doubling up to the maximum.
@@ -112,6 +111,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
     let mut resolve_retry = RESOLVE_RETRY;
     let mut next_resolve = Instant::now();
     let mut next_report = Instant::now() + BOOTSTRAP_REPORT;
+    state.transport.on_seeds_found(state.join_progress.clone());
     loop {
         if let Some(swarm_id) = state.swarm_id().await {
             // A node joining a compacted swarm rejoins the newer epoch first; it is
@@ -147,7 +147,19 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
             next_report = now + BOOTSTRAP_REPORT;
             report_joining(state, seeds, &unresolved).await;
         }
-        tokio::time::sleep(BOOTSTRAP_POLL).await;
+        // Until something changes (`join_progress`), or it's time to resolve or report.
+        let resolving = async {
+            if unresolved.is_empty() {
+                std::future::pending().await
+            } else {
+                tokio::time::sleep_until(next_resolve).await
+            }
+        };
+        tokio::select! {
+            () = state.join_progress.notified() => {}
+            () = resolving => {}
+            () = tokio::time::sleep_until(next_report) => {}
+        }
     }
 }
 
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 46bdef6..c643004 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -425,6 +425,7 @@ pub(crate) async fn handle_sync(
             .stats
             .caught_up(state.traffic.total().received.sync.bytes);
         source::finished(state, peer);
+        state.join_progress.notify_one();
     }
 
     state.notify_store_changed();
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index fb1600b..3ac06cb 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -632,6 +632,10 @@ impl Transport for IdentityTransport {
         self.shared.paths.discovered_seeds()
     }
 
+    fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) {
+        self.shared.paths.on_seeds_found(found)
+    }
+
     /// A node ID resolves to itself if this node knows where it is; anything else is
     /// an address (`<scheme>:...`) 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 8620895..3de8cef 100644
--- a/src/transport/libp2p/book.rs
+++ b/src/transport/libp2p/book.rs
@@ -111,6 +111,8 @@ pub(super) struct Book {
     /// Peers found on the LAN that speak continuum, with their address: seeds not
     /// handed out yet (`discovered_seeds`).
     pub(super) discovered: Vec<Multiaddr>,
+    /// Told when a seed is found (`Transport::on_seeds_found`).
+    pub(super) seeds_found: Option<Arc<tokio::sync::Notify>>,
 }
 
 /// What `Libp2pStats` counts since start.
@@ -253,6 +255,9 @@ impl Book {
             && self.discovered.len() < MAX_PEERS
         {
             self.discovered.push(address);
+            if let Some(found) = &self.seeds_found {
+                found.notify_one();
+            }
         }
     }
 
diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs
index 0bbabc7..a159280 100644
--- a/src/transport/libp2p/mod.rs
+++ b/src/transport/libp2p/mod.rs
@@ -488,6 +488,10 @@ impl Transport for Libp2pTransport {
         }
     }
 
+    fn on_seeds_found(&self, found: Arc<tokio::sync::Notify>) {
+        lock(&self.book).seeds_found = Some(found);
+    }
+
     /// Nodes found on the LAN (mDNS) that speak continuum.
     fn discovered_seeds(&self) -> Vec<String> {
         std::mem::take(&mut lock(&self.book).discovered)
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 5ad2427..a802269 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -818,6 +818,12 @@ pub(crate) trait Transport: Send + Sync {
         Vec::new()
     }
 
+    /// Has the transport tell `found` whenever it finds a seed by itself
+    /// (`discovered_seeds`). By default it never does.
+    fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) {
+        let _ = found;
+    }
+
     /// 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 ce19fd4..8149781 100644
--- a/src/transport/paths.rs
+++ b/src/transport/paths.rs
@@ -667,6 +667,13 @@ impl Paths {
             .collect()
     }
 
+    /// Each transport's (`Transport::on_seeds_found`).
+    pub(crate) fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) {
+        for link in &self.links {
+            link.inner.on_seeds_found(found.clone());
+        }
+    }
+
     /// The address of the node a seed leads to, as paths keep it: an address of ours
     /// (`<scheme>:...`) 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 e6d4a0c..6385f2e 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -330,6 +330,10 @@ impl Transport for CountingTransport {
         self.inner.discovered_seeds()
     }
 
+    fn on_seeds_found(&self, found: std::sync::Arc<tokio::sync::Notify>) {
+        self.inner.on_seeds_found(found)
+    }
+
     fn transports(&self) -> Vec<String> {
         self.inner.transports()
     }