diff options
Diffstat (limited to 'src/state')
| -rw-r--r-- | src/state/mod.rs | 5 | ||||
| -rw-r--r-- | src/state/peer.rs | 73 |
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( |
