1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
//! The peer worker: every `PEER_TICK`, keepalives due, dormancy, HELLO retries and
//! overdue sync replies; every `FORGET_CHECK_INTERVAL` (and at once when the node
//! becomes ready), forget records, the member registry, stability, measurements and
//! the state file.
use std::time::Duration;
use tokio::time::{Instant, MissedTickBehavior};
use tracing::{debug, info, warn};
use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
use super::handshake::retry_hello_if_due;
use super::keepalive::send_status;
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
const PEER_TICK: Duration = Duration::from_millis(500);
const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5);
pub(crate) async fn run_peer_worker(state: AppState) {
let mut tick = tokio::time::interval(PEER_TICK);
tick.set_missed_tick_behavior(MissedTickBehavior::Delay);
info!("Starting peer maintenance worker");
let mut last_forget_check: Option<Instant> = None;
let mut forgets_applied = AppliedForgets::new();
let mut was_ready = false;
loop {
tick.tick().await;
// While joining, keep handshakes fast; dormancy would slow probes to minutes.
let ready = state.is_ready().await;
if ready && !was_ready {
// Just became ready: register in the member list (and do the other periodic
// checks) now rather than at the next check. Until it is registered, other
// members don't count this node as a neighbour.
last_forget_check = None;
}
was_ready = ready;
let peers = state.peers_snapshot().await;
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, ping_due, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
let became_dormant =
ready && neighbour && peer_state.check_dormancy(&state.timing, now);
let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now);
if !neighbour {
peer_state.stop_hello_retries();
}
let unanswered = peer_state.take_overdue_sync_reply(now);
(became_dormant, ping_due, unanswered)
};
if let Some(changes) = unanswered {
debug!(peer = %peer, changes, "No answer to our sync message; sending again");
crate::swarm::source::reply_timed_out(&state, peer);
let _ = state
.sync_events
.send(SyncEvent::PeerNeedsSync(peer.clone()));
}
if became_dormant {
state.topology_changed.notify_one();
warn!(
peer = %peer,
silent_secs = state.timing.dormant_after.as_secs(),
probe_secs = state.timing.dormant_probe.as_secs(),
"Peer went dormant"
);
}
let probe_due =
!ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now());
if (ping_due || probe_due)
&& let Err(error) = send_status(&state, peer).await
{
debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive STATUS failed");
}
if let Err(error) = retry_hello_if_due(&state, peer).await {
warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
}
}
if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
apply_forget_records(&state, &mut forgets_applied).await;
forget_strangers(&state).await;
crate::swarm::members::ensure_registered(&state).await;
crate::swarm::stability::tick(&state).await;
crate::swarm::stats::measure_saved_size(&state);
crate::swarm::source::warn_if_stalled(&state);
crate::swarm::state_file::update(&state, false).await;
{
let known = state.peer_states.read().await;
state.traffic.retain(|peer| known.contains_key(peer));
state
.throughput
.lock_or_recover()
.retain(|peer| known.contains_key(peer));
}
// Fallback for what doesn't notify (`topology::run_worker`).
state.topology_changed.notify_one();
last_forget_check = Some(Instant::now());
}
}
}
|
