peer worker sleeps until deadline, not fixed time

10 files changed, 170 insertions, 32 deletions

diff --git a/src/node.rs b/src/node.rs
index 6f55ed5..6de8f0c 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -129,6 +129,7 @@ impl Node {
             neighbours: Default::default(),
             topology_changed: Default::default(),
             join_progress: Default::default(),
+            peers_due: Default::default(),
             pulling: Default::default(),
             announce_pending: Default::default(),
             rtt: Default::default(),
diff --git a/src/state/mod.rs b/src/state/mod.rs
index 2902234..ba90269 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -154,6 +154,8 @@ pub(crate) struct AppState {
     /// 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>,
+    /// When the peer worker is next due (`swarm::peer::Due`).
+    pub(crate) peers_due: Arc<crate::swarm::peer::Due>,
 
     /// While rejoining a newer epoch, the previous document, served read-only (and
     /// marked stale) instead of answering 503.
@@ -319,6 +321,8 @@ impl AppState {
                 let old_lifecycle = lifecycle.clone();
                 *lifecycle = NodeLifecycle::Ready { swarm_id };
                 debug!("Node transitioned from {old_lifecycle:?} {lifecycle:?}");
+                // Dormancy is judged once ready, and the periodic checks run at once.
+                self.peers_due.now();
             }
             NodeLifecycle::Ready { swarm_id: existing } if existing == &swarm_id => {
                 debug!("Node already ready with the same identity");
@@ -366,6 +370,7 @@ impl AppState {
     /// State for a peer we haven't known so far (not inserted).
     pub(crate) fn new_peer_state(&self, pk: &str) -> Arc<Mutex<PeerState>> {
         let leads = crate::swarm::pair::leads(self.local_id(), pk);
+        self.peers_due.now();
         Arc::new(Mutex::new(PeerState::new(leads, &self.timing)))
     }
 
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 3e1332d..cb42dc1 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -392,6 +392,26 @@ impl PeerState {
         timing: &Timing,
         now: Instant,
     ) -> Option<Keepalive> {
+        if self.keepalive_due_at(timing).is_none_or(|at| now < at) {
+            return None;
+        }
+        self.last_keepalive_at = Some(now);
+        let status = self
+            .last_status_at
+            .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
+        Some(if status {
+            Keepalive::Status
+        } else {
+            Keepalive::Ping
+        })
+    }
+
+    /// When the next keepalive is due (`take_keepalive_due`), if any is: none to a
+    /// dormant peer, nor before the handshake.
+    pub(crate) fn keepalive_due_at(&self, timing: &Timing) -> Option<Instant> {
+        if self.liveness.is_dormant() || self.hello_state != HelloState::Established {
+            return None;
+        }
         let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) {
             (None, None) => self.leads,
             (None, Some(_)) => true,
@@ -407,27 +427,44 @@ impl PeerState {
         } else {
             timing.keepalive + timing.keepalive / 2
         };
-        let due = !self.liveness.is_dormant()
-            && self.hello_state == HelloState::Established
-            && now.saturating_duration_since(self.added_at) >= timing.keepalive
-            && self.quiet_for(now) >= quiet_enough
-            && self
-                .last_keepalive_at
-                .is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
-        if !due {
-            return None;
-        }
-        self.last_keepalive_at = Some(now);
-        let status = self
-            .last_status_at
-            .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
-        Some(if status {
-            Keepalive::Status
-        } else {
-            Keepalive::Ping
+        let quiet_since = self.last_seen.unwrap_or(self.added_at);
+        let at = (self.added_at + timing.keepalive).max(quiet_since + quiet_enough);
+        Some(match self.last_keepalive_at {
+            Some(sent) => at.max(sent + timing.keepalive),
+            None => at,
         })
     }
 
+    /// When the peer worker next has something to do for this peer
+    /// (`swarm::peer::run_peer_worker`): a keepalive (to a `neighbour`), dormancy (of a
+    /// neighbour, once `ready`), a HELLO retry, an overdue sync answer. None if nothing
+    /// is due until something changes.
+    pub(crate) fn next_due(
+        &self,
+        timing: &Timing,
+        neighbour: bool,
+        ready: bool,
+    ) -> Option<Instant> {
+        let dormancy = (ready && neighbour && !self.liveness.is_dormant())
+            .then(|| self.last_seen.unwrap_or(self.added_at) + timing.dormant_after);
+        let keepalive = if neighbour {
+            self.keepalive_due_at(timing)
+        } else {
+            None
+        };
+        let hello = match self.hello_state {
+            HelloState::AwaitingReply { next_attempt } => Some(next_attempt),
+            _ => None,
+        };
+        let reply = (!self.liveness.is_dormant() && self.hello_state == HelloState::Established)
+            .then(|| self.sync.reply.as_ref().map(|wait| wait.deadline))
+            .flatten();
+        [dormancy, keepalive, hello, reply]
+            .into_iter()
+            .flatten()
+            .min()
+    }
+
     /// Self-healing check on a STATUS or STATUS_REPLY. Returns true if a re-handshake
     /// should start.
     pub(crate) fn needs_heal(
diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs
index 13a1264..896ffdf 100644
--- a/src/swarm/peer/handshake.rs
+++ b/src/swarm/peer/handshake.rs
@@ -149,9 +149,9 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
         if peer_state.hello_state == HelloState::Sending {
             topology::change_peer(state, &mut peer_state, |p| p.hello_sent(Instant::now()));
             let wait = peer_state.next_hello_wait();
-            peer_state.hello_state = HelloState::AwaitingReply {
-                next_attempt: Instant::now() + wait,
-            };
+            let next_attempt = Instant::now() + wait;
+            peer_state.hello_state = HelloState::AwaitingReply { next_attempt };
+            state.peers_due.by(next_attempt);
         }
     }
 
@@ -413,6 +413,8 @@ pub(super) async fn establish_peer(
     if from_hello || peer_state.hello_state != HelloState::Established {
         peer_state.hello_state = HelloState::Established;
         peer_state.reset_sync();
+        // Keepalives start.
+        state.peers_due.now();
         info!(peer = %peer, from_hello, "Handshake complete; sync state reset");
     }
 
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 11817cc..8597dcd 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -71,6 +71,9 @@ pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMes
         if ping {
             // Its keepalive (or a path probe): the next is ours.
             peer_state.keepalive_received_at = Some(now);
+            if let Some(at) = peer_state.keepalive_due_at(&state.timing) {
+                state.peers_due.by(at);
+            }
         }
     })
     .await;
@@ -181,6 +184,9 @@ pub(super) async fn status_received(state: &AppState, peer: &str) {
         let now = Instant::now();
         peer_state.keepalive_received_at = Some(now);
         peer_state.last_status_at = Some(now);
+        if let Some(at) = peer_state.keepalive_due_at(&state.timing) {
+            state.peers_due.by(at);
+        }
     }
 }
 
@@ -195,6 +201,10 @@ pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport
             // its HELLO when it's back, makes it active again).
             remote.leaving && p.say_farewell(Instant::now())
         });
+        if left {
+            // Probed from now on.
+            state.peers_due.now();
+        }
         let became_ready = remote.ready && peer_state.reports_ready != Some(true);
         peer_state.reports_ready = Some(remote.ready);
         peer_state.reported_heads = remote.heads.heads().map(<[_]>::len);
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index 326f3a9..589f48c 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -27,4 +27,4 @@ pub(crate) use handshake::{
 pub(crate) use inbound::handle_message;
 pub(crate) use keepalive::{announce, farewell, send_status};
 pub(crate) use lanes::SyncLanes;
-pub(crate) use worker::run_peer_worker;
+pub(crate) use worker::{Due, run_peer_worker};
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 9b62d24..546f867 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -1,11 +1,11 @@
-//! The peer worker: every `PEER_TICK`, keepalives due, dormancy, HELLO retries and
-//! overdue sync replies; every `FORGET_CHECK_INTERVAL` (and at once when the node
-//! becomes ready), forget records, the member registry, stability, measurements and
-//! the state file.
+//! The peer worker: keepalives due, dormancy, HELLO retries and overdue sync replies,
+//! when the earliest of them is due (or something may have made one due sooner,
+//! `Due`); every `FORGET_CHECK_INTERVAL` (and at once when the node becomes ready),
+//! forget records, the member registry, stability, measurements and the state file.
 
 use std::time::Duration;
 
-use tokio::time::{Instant, MissedTickBehavior};
+use tokio::time::Instant;
 use tracing::{debug, info, warn};
 
 use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
@@ -15,14 +15,49 @@ use crate::locks::MutexExt;
 use crate::state::{AppState, Keepalive};
 use crate::swarm::topology;
 
-const PEER_TICK: Duration = Duration::from_millis(500);
+/// The least time between two passes, however often something is due.
+const MIN_PASS_GAP: Duration = Duration::from_millis(20);
 
 const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5);
 
-pub(crate) async fn run_peer_worker(state: AppState) {
-    let mut tick = tokio::time::interval(PEER_TICK);
-    tick.set_missed_tick_behavior(MissedTickBehavior::Delay);
+/// When the peer worker is next due. What makes something of its due sooner than it
+/// knows (a HELLO sent, a sync message waiting for its answer, a keepalive received, a
+/// new peer, new neighbours) says so (`by`); what only puts things off needn't.
+#[derive(Default)]
+pub(crate) struct Due {
+    /// Its next pass while it waits; None while it runs.
+    next: std::sync::Mutex<Option<Instant>>,
+    wake: tokio::sync::Notify,
+}
+
+impl Due {
+    /// Something is due at `at`: a pass then at the latest.
+    pub(crate) fn by(&self, at: Instant) {
+        if self.next.lock_or_recover().is_none_or(|next| at < next) {
+            self.wake.notify_one();
+        }
+    }
+
+    /// A pass now.
+    pub(crate) fn now(&self) {
+        self.wake.notify_one();
+    }
+
+    fn passing(&self) {
+        *self.next.lock_or_recover() = None;
+    }
 
+    /// Until `at`, or until something is due sooner.
+    async fn until(&self, at: Instant) {
+        *self.next.lock_or_recover() = Some(at);
+        tokio::select! {
+            () = self.wake.notified() => {}
+            () = tokio::time::sleep_until(at) => {}
+        }
+    }
+}
+
+pub(crate) async fn run_peer_worker(state: AppState) {
     info!("Starting peer maintenance worker");
 
     let mut last_forget_check: Option<Instant> = None;
@@ -30,7 +65,8 @@ pub(crate) async fn run_peer_worker(state: AppState) {
     let mut was_ready = false;
 
     loop {
-        tick.tick().await;
+        state.peers_due.passing();
+        let mut next = Instant::now() + FORGET_CHECK_INTERVAL;
 
         // While joining, keep handshakes fast; dormancy would slow probes to minutes.
         let ready = state.is_ready().await;
@@ -103,6 +139,17 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             if let Err(error) = retry_hello_if_due(&state, peer).await {
                 warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
             }
+
+            let due = peer_state_mutex
+                .lock()
+                .await
+                .next_due(&state.timing, neighbour, ready);
+            let probe = crate::swarm::source::probe_at(&state, peer);
+            next = [Some(next), due, probe]
+                .into_iter()
+                .flatten()
+                .min()
+                .unwrap_or(next);
         }
 
         if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
@@ -126,5 +173,12 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             state.topology_changed.notify_one();
             last_forget_check = Some(Instant::now());
         }
+        if let Some(checked) = last_forget_check {
+            next = next.min(checked + FORGET_CHECK_INTERVAL);
+        }
+
+        let passed = Instant::now();
+        state.peers_due.until(next).await;
+        tokio::time::sleep_until(passed + MIN_PASS_GAP).await;
     }
 }
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index e098dc1..1288d86 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -146,6 +146,7 @@ pub(crate) fn defer(state: &AppState, peer: &str, peer_state: &PeerState) -> Opt
         last_probe: None,
         refusals: 0,
     });
+    state.peers_due.by(now + state.timing.catch_up_probe);
     None
 }
 
@@ -285,6 +286,21 @@ pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool {
     due
 }
 
+/// When a STATUS to `peer` is next due (`probe_due`), if it is the source.
+pub(crate) fn probe_at(state: &AppState, peer: &str) -> Option<Instant> {
+    if !catching_up(state) {
+        return None;
+    }
+    let probe = state.timing.catch_up_probe;
+    let source = lock(state);
+    let current = source.as_ref().filter(|current| current.peer == peer)?;
+    let heard = current.last_heard + probe;
+    Some(match current.last_probe {
+        Some(last) => heard.max(last + probe),
+        None => heard,
+    })
+}
+
 /// No catch-up going on: no source, no head start, nobody avoided. When one ends, and
 /// before one starts (a rejoin: `Host::reset_initial_sync`).
 pub(crate) fn reset(state: &AppState) {
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index c643004..c8564c2 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -198,6 +198,17 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
         let Some(encoded) = generated else {
             return Ok(());
         };
+        // Its answer is due.
+        let reply_due = peer_state_mutex
+            .lock()
+            .await
+            .sync
+            .reply
+            .as_ref()
+            .map(|wait| wait.deadline);
+        if let Some(at) = reply_due {
+            state.peers_due.by(at);
+        }
 
         (session, encoded)
     };
diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs
index e494f34..30fa56e 100644
--- a/src/swarm/topology/mod.rs
+++ b/src/swarm/topology/mod.rs
@@ -368,6 +368,8 @@ pub(crate) async fn refresh(state: &AppState) -> Option<NeighbourChange> {
         outbound,
         neighbours: computed,
     });
+    // New neighbours are kept alive and judged.
+    state.peers_due.now();
     Some(change)
 }