aboutsummaryrefslogtreecommitdiffstats
path: root/src/swarm/peer
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/peer')
-rw-r--r--src/swarm/peer/worker.rs25
1 files changed, 18 insertions, 7 deletions
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;