peer worker sleeps until deadline, not fixed time

2 files changed, 60 insertions, 18 deletions

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(