diff options
| -rw-r--r-- | docs/openapi.yaml | 2 | ||||
| -rw-r--r-- | docs/peers.md | 15 | ||||
| -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 |
7 files changed, 143 insertions, 24 deletions
diff --git a/docs/openapi.yaml b/docs/openapi.yaml index a15339a..fd0b855 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -1571,7 +1571,7 @@ components: type: array items: type: string - description: The members this node reaches (an active session with each), by node ID, as it last told its neighbours. + description: The members this node reaches (an active session with each, answering), by node ID, as it last told its neighbours. peers: type: object description: By node ID. diff --git a/docs/peers.md b/docs/peers.md index 127b2ad..c2f1ddb 100644 --- a/docs/peers.md +++ b/docs/peers.md @@ -298,10 +298,14 @@ NAT may reach a single one, one on libp2p alone none on Tor alone), and no node which from where it is. A neighbour that has the changes already only answers. Every node tells its neighbours whom it reaches ([`swarm/reach.rs`](../src/swarm/reach.rs)): the members it -has an active session with (handshake done, neither dormant nor unreachable), each by -an 8-byte key (the hash its pair scores come from), in its keepalive STATUSes and -their replies, and within 5 s when it changed; not in announcements of heads, which go -often. So every node knows its 2-hop neighbourhood as measured, not as the member list +has an active session with (handshake done, neither dormant nor unreachable) that +answer, each by an 8-byte key (the hash its pair scores come from), in its keepalive +STATUSes and their replies, and within 5 s when it changed; not in announcements of +heads, which go often. A member that leaves a keepalive or sync message unanswered, +with nothing else heard from it, is asked again (a PING) after the first sync reply +timeout, and after twice that no longer counts as reached: its neighbours are told at +once, and the members whose parent it was in a writer's tree take another, within +seconds rather than after `dormant_after`. Any message from it, and it counts again. So every node knows its 2-hop neighbourhood as measured, not as the member list says, for as long as each neighbour's session lasts (one that goes dormant or starts anew says it again), and nothing goes into the document. Not for a set time: while a pair exchanges changes it sends no keepalives, and so no STATUS, however long the @@ -403,5 +407,6 @@ Derived from `--keepalive` (`Timing::from_keepalive`, [`state/timing.rs`](../src | `dormant_probe` | 10 × keepalive | the first probe of an unreachable peer, doubling up to 6 × (or `--dormant-probe-max`) | | `catch_up_probe` | keepalive / 30, from 1 to 5 s | how soon a silent catch-up source is probed | | sync reply timeout | 10 s (30 s with Tor enabled: the longest of the transports'), doubling up to 30 × | resending a sync message nothing answered | +| `unreached_after` | 2 × the first sync reply timeout | a keepalive or sync message unanswered this long (asked again halfway), a peer no longer counts as reached | -The peer worker runs when the earliest of these is due for some peer (and when something may have made one due sooner: a HELLO sent, a sync message sent, a keepalive received, a new peer or new neighbours), at least every 5 s; the remake worker every second. +The peer worker runs when the earliest of these is due for some peer (and when something may have made one due sooner: a HELLO sent, a sync message sent, a keepalive received, a new peer or new neighbours), at least every 5 s, and tells the neighbours whom it reaches at once when a member it reached stops answering; the remake worker every second. 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>>, } |