diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/state/peer.rs | 85 | ||||
| -rw-r--r-- | src/state/timing.rs | 6 | ||||
| -rw-r--r-- | src/swarm/peer/worker.rs | 25 | ||||
| -rw-r--r-- | src/swarm/reach.rs | 32 | ||||
| -rw-r--r-- | src/transport/mod.rs | 2 |
5 files changed, 132 insertions, 18 deletions
diff --git a/src/state/peer.rs b/src/state/peer.rs index ca721b5..a1d40ab 100644 --- a/src/state/peer.rs +++ b/src/state/peer.rs @@ -108,6 +108,13 @@ pub(crate) struct PeerState { pub(crate) keepalive_received_at: Option<Instant>, /// The last STATUS between us, either way (`status_refresh`). pub(crate) last_status_at: Option<Instant>, + /// Since when we wait for an answer from it (to a keepalive, or to a sync message + /// whose answer is overdue) with nothing heard from it since, and whether we asked + /// again: halfway to `Timing::unreached_after` a PING (`take_ask_again_due`; a sync + /// message whose answer is overdue goes again anyway), so one lost message or answer + /// doesn't do it; after it, we don't count it as reached (`swarm::reach`). Cleared by + /// any message from it (`mark_seen`). + awaiting_answer: Option<(Instant, bool)>, /// Last time a sync message from this peer was applied (survives resets: the /// keepalive heal looks at it). @@ -230,6 +237,7 @@ impl PeerState { last_keepalive_at: None, keepalive_received_at: None, last_status_at: None, + awaiting_answer: None, last_sync_rx: None, last_resync_at: None, diverged_since: None, @@ -270,9 +278,14 @@ impl PeerState { return None; } let largest = self.sync.reply.as_ref().map(|wait| wait.largest); + let started = self.sync.reply.as_ref().map(|wait| wait.started.0); let overdue = self.sync.take_overdue_reply(now); if overdue.is_some() { self.incidents.reply_timeouts += 1; + if let Some(started) = started { + // It goes again now: that is asking again. + self.awaiting_answer.get_or_insert((started, true)); + } if let Some(largest) = largest { self.whole_failed(largest, now); } @@ -303,6 +316,7 @@ impl PeerState { /// said farewell (`left`), only that it was heard from. pub(crate) fn mark_seen(&mut self, now: Instant) -> bool { self.last_seen = Some(now); + self.awaiting_answer = None; if self.left { return false; } @@ -342,6 +356,32 @@ impl PeerState { && self.epoch == Some(epoch) } + /// Whether what we sent it went unanswered, with nothing else heard from it, for + /// `Timing::unreached_after` (`awaiting_answer`): then we don't count it as reached + /// (`swarm::reach`), long before it would be dormant. + pub(crate) fn answer_overdue(&self, timing: &Timing, now: Instant) -> bool { + self.awaiting_answer.is_some_and(|(since, _)| { + now.saturating_duration_since(since) >= timing.unreached_after + }) + } + + /// Whether to ask an established, active peer again, with a PING, as nothing answered + /// us for half of `Timing::unreached_after` (`awaiting_answer`); records it. + pub(crate) fn take_ask_again_due(&mut self, timing: &Timing, now: Instant) -> bool { + if self.liveness.is_dormant() || self.hello_state != HelloState::Established { + return false; + } + match &mut self.awaiting_answer { + Some((since, asked_again)) + if !*asked_again && now >= *since + timing.unreached_after / 2 => + { + *asked_again = true; + true + } + _ => false, + } + } + /// Whether to leave the peer out when choosing neighbours (`swarm::topology`). pub(crate) fn is_unreachable(&self) -> bool { self.unreachable_since.is_some() @@ -452,6 +492,7 @@ impl PeerState { return None; } self.last_keepalive_at = Some(now); + self.awaiting_answer.get_or_insert((now, false)); let status = self .last_status_at .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh); @@ -493,8 +534,9 @@ impl PeerState { /// 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. + /// neighbour, once `ready`), a HELLO retry, an overdue sync answer, asking again for + /// an answer and its being overdue (`take_ask_again_due`, `answer_overdue`). None if nothing is due until something + /// changes. pub(crate) fn next_due( &self, timing: &Timing, @@ -515,7 +557,18 @@ impl PeerState { 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] + let unreached = self + .awaiting_answer + .map(|(since, asked_again)| { + if asked_again { + since + timing.unreached_after + } else { + since + timing.unreached_after / 2 + } + }) + // Once past, nothing more is due for it. + .filter(|at| !self.liveness.is_dormant() && *at > Instant::now()); + [dormancy, keepalive, hello, reply, unreached] .into_iter() .flatten() .min() @@ -674,6 +727,32 @@ mod tests { assert!(b.take_keepalive_due(&timing, next).is_none()); } + /// A keepalive nothing answers: asked again halfway, after `unreached_after` not + /// reached; any message from the peer, reached again. One lost answer doesn't do it. + #[test] + fn unanswered_keepalives_stop_counting_as_reached() { + let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300)); + let mut peer = PeerState::new(true, &timing); + peer.hello_state = HelloState::Established; + let sent = peer.added_at + timing.keepalive; + assert!(peer.take_keepalive_due(&timing, sent).is_some()); + let half = sent + timing.unreached_after / 2; + assert!(!peer.take_ask_again_due(&timing, half - Duration::from_millis(1))); + assert!(peer.take_ask_again_due(&timing, half)); + assert!(!peer.take_ask_again_due(&timing, half), "asked again once"); + assert!(!peer.answer_overdue(&timing, half)); + // The second ask was answered: reached all along. + peer.mark_seen(half + Duration::from_millis(50)); + assert!(!peer.answer_overdue(&timing, sent + timing.unreached_after)); + // The next keepalive, and its second ask, go unanswered. + let sent = half + timing.keepalive * 2; + assert!(peer.take_keepalive_due(&timing, sent).is_some()); + assert!(peer.take_ask_again_due(&timing, sent + timing.unreached_after / 2)); + assert!(peer.answer_overdue(&timing, sent + timing.unreached_after)); + peer.mark_seen(sent + timing.unreached_after * 2); + assert!(!peer.answer_overdue(&timing, sent + timing.unreached_after * 2)); + } + /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet. #[test] fn keepalives_are_pings_and_now_and_then_a_status() { diff --git a/src/state/timing.rs b/src/state/timing.rs index af5c0c0..fed48e9 100644 --- a/src/state/timing.rs +++ b/src/state/timing.rs @@ -35,6 +35,11 @@ pub(crate) struct Timing { /// First wait for an answer to a sync message before sending again; doubles while /// answers stay out (`SyncSession::reply_backoff`). pub(crate) sync_reply_timeout: Duration, + /// How long what we sent a peer may go unanswered, with nothing else heard from it, + /// before we don't count it as reached (`swarm::reach`, `PeerState::answer_overdue`): + /// twice the first sync reply timeout, so one lost message (or its answer) doesn't + /// do it. + pub(crate) unreached_after: Duration, /// How long the transport tries to deliver a message before dropping it /// (`Transport::delivery_limit`): a peer silent for longer has nothing on the way to /// us any more. @@ -93,6 +98,7 @@ impl Timing { // Only the source gets these, and only while catching up. catch_up_probe: (keepalive / 30).clamp(Duration::from_secs(1), Duration::from_secs(5)), sync_reply_timeout, + unreached_after: sync_reply_timeout * 2, delivery_limit, } } diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs index 45e8d8f..dad0870 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -1,8 +1,9 @@ -//! 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; -//! forget records also at once when they changed. +//! The peer worker: keepalives due, dormancy, HELLO retries, overdue sync replies and +//! answers (`PeerState::answer_overdue`, `swarm::reach`), 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; forget records +//! also at once when they changed. use std::time::Duration; @@ -80,13 +81,15 @@ pub(crate) async fn run_peer_worker(state: AppState) { was_ready = ready; let peers = state.peers_snapshot().await; + // A member we said we reach stopped answering: tell our neighbours now. + let mut reach_changed = false; for (peer, peer_state_mutex) in &peers { let now = Instant::now(); // Only neighbours are kept alive and judged; others we only answer. let neighbour = topology::is_neighbour(&state, peer); - let (became_dormant, keepalive, unanswered) = { + let (became_dormant, keepalive, unanswered, ask_again) = { let mut peer_state = peer_state_mutex.lock().await; let became_dormant = ready && neighbour @@ -102,7 +105,10 @@ pub(crate) async fn run_peer_worker(state: AppState) { peer_state.stop_hello_retries(topology::is_listed(&state, peer)); } let unanswered = peer_state.take_overdue_sync_reply(now); - (became_dormant, keepalive, unanswered) + let ask_again = peer_state.take_ask_again_due(&state.timing, now); + reach_changed |= peer_state.answer_overdue(&state.timing, now) + && crate::swarm::reach::reached_stops_answering(&state, peer); + (became_dormant, keepalive, unanswered, ask_again) }; if let Some(changes) = unanswered { @@ -128,6 +134,8 @@ pub(crate) async fn run_peer_worker(state: AppState) { } keepalive => keepalive, }; + // Nothing answered us for a while: once more before it counts as not reached. + let keepalive = keepalive.or(ask_again.then_some(Keepalive::Ping)); let sent = match keepalive { Some(Keepalive::Ping) => send_ping(&state, peer).await, Some(Keepalive::Status) => send_status(&state, peer).await, @@ -160,6 +168,9 @@ pub(crate) async fn run_peer_worker(state: AppState) { if records_changed || periodic { apply_forget_records(&state, &mut forgets_applied).await; } + if reach_changed && !periodic { + crate::swarm::reach::check(&state).await; + } if periodic { forget_strangers(&state).await; crate::swarm::members::ensure_registered(&state).await; diff --git a/src/swarm/reach.rs b/src/swarm/reach.rs index 901b47c..7bd0f73 100644 --- a/src/swarm/reach.rs +++ b/src/swarm/reach.rs @@ -2,13 +2,16 @@ //! member list says. //! //! A node knows its own links, and from them whom it reaches: the members it has an -//! active session with (handshake done, neither dormant nor unreachable). It tells its +//! active session with (handshake done, neither dormant nor unreachable) that answer +//! (`PeerState::answer_overdue`: none left a keepalive or sync message unanswered for +//! `Timing::unreached_after`, asked again halfway, with nothing else heard from it; a +//! member gone shows within seconds, not after `dormant_after`). It tells its //! neighbours so in its STATUS (`Status::reaches`, each member by its pair key, //! `topology::pair_key`, 8 bytes instead of a node ID): in every keepalive STATUS and -//! STATUS_REPLY, and soon (the peer worker looks every 5 s) to every neighbour when it -//! changed, -//! but not in announcements of heads, which go often. So every node knows whom its -//! neighbours reach, as of their latest STATUS, for as long as the session lasts (a +//! STATUS_REPLY, and soon to every neighbour when it changed (the peer worker looks +//! every 5 s, and at once when a member we reach stops answering), but not in +//! announcements of heads, which go often. So every node knows whom its neighbours +//! reach, as of their latest STATUS, for as long as the session lasts (a //! neighbour that goes dormant, or starts anew, says it again). Not for a set time: //! while a pair exchanges changes it sends no keepalives, and so no STATUS, for as //! long as the writes go on. Nothing goes into the document, and nobody knows more than @@ -37,14 +40,17 @@ pub(crate) struct Reach { published: Option<Vec<u64>>, } -/// The pair keys of the members we reach now, sorted: those with an active session. +/// The pair keys of the members we reach now, sorted: those with an active session +/// that answer. async fn measure(state: &AppState) -> Vec<u64> { + let now = Instant::now(); let mut reached = Vec::new(); for (peer, peer_state) in state.peers_snapshot().await { let peer_state = peer_state.lock().await; if peer_state.hello_state == HelloState::Established && !peer_state.liveness.is_dormant() && !peer_state.is_unreachable() + && !peer_state.answer_overdue(&state.timing, now) && matches!(peer_state.standing, Standing::Member { .. }) { reached.push(topology::pair_key(&peer)); @@ -64,7 +70,8 @@ pub(crate) fn published(state: &AppState) -> Vec<u64> { .unwrap_or_default() } -/// With the peer worker's periodic pass (every 5 s): whom we reach now; if it changed, +/// With the peer worker's periodic pass (every 5 s), and at once when a member we said +/// we reach stops answering (`reached_stops_answering`): whom we reach now; if it changed, /// our neighbours are told at once (a STATUS each, unanswered if the heads are the /// same). A change that is undone within a pass isn't told. pub(crate) async fn check(state: &AppState) { @@ -91,6 +98,17 @@ pub(crate) async fn check(state: &AppState) { } } +/// Whether `peer`, whose answer is overdue (`PeerState::answer_overdue`), is one we +/// said we reach: then the peer worker checks at once (`check`). +pub(crate) fn reached_stops_answering(state: &AppState, peer: &str) -> bool { + state + .reach + .lock_or_recover() + .published + .as_ref() + .is_some_and(|keys| keys.contains(&topology::pair_key(peer))) +} + /// A STATUS to `peer` with whom we reach: as an announcement (unanswered if the heads /// are the same), but with `reaches`. async fn tell(state: &AppState, peer: &str) -> anyhow::Result<()> { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index fb198ec..e27ce2d 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -207,7 +207,7 @@ pub(crate) struct Status { /// neighbour that has the changes already, which has nothing to tell. pub(crate) announcing: bool, /// Whom the sender reaches (`swarm::reach`): the pair keys (`topology::pair_key`) - /// of the members it has an active session with, sorted. In keepalive STATUSes and + /// of the members it has an active session with that answer, sorted. In keepalive STATUSes and /// their replies, and when it changed; not in announcements of heads (None). pub(crate) reaches: Option<Vec<u64>>, } |