From 2a85e21ab81d4497dcc5680b4348bbf802866944 Mon Sep 17 00:00:00 2001 From: tv Date: Tue, 3 Nov 2026 08:34:09 +0100 Subject: unanswered members not reached --- src/swarm/peer/worker.rs | 25 ++++++++++++++++++------- src/swarm/reach.rs | 32 +++++++++++++++++++++++++------- 2 files changed, 43 insertions(+), 14 deletions(-) (limited to 'src/swarm') 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>, } -/// 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 { + 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 { .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<()> { -- cgit v1.3.1