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
118
119
120
121
122
123
124
125
126
127
128
129
|
//! 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_ping, send_status};
use crate::locks::MutexExt;
use crate::state::{AppState, Keepalive};
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, keepalive, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
let became_dormant = ready
&& neighbour
&& topology::change_peer(&state, &mut peer_state, |p| {
p.check_dormancy(&state.timing, now)
});
let keepalive = if neighbour {
peer_state.take_keepalive_due(&state.timing, now)
} else {
None
};
if !neighbour {
peer_state.stop_hello_retries();
}
let unanswered = peer_state.take_overdue_sync_reply(now);
(became_dormant, keepalive, 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);
state.request_sync(peer);
}
if became_dormant {
warn!(
peer = %peer,
silent_secs = state.timing.dormant_after.as_secs(),
probe_secs = state.timing.dormant_probe.as_secs(),
"Peer went dormant"
);
}
// The catch-up source gets STATUSes: we need its readiness, and whether it
// syncs with us.
let keepalive = match keepalive {
None if crate::swarm::source::probe_due(&state, peer, Instant::now()) => {
Some(Keepalive::Status)
}
keepalive => keepalive,
};
let sent = match keepalive {
Some(Keepalive::Ping) => send_ping(&state, peer).await,
Some(Keepalive::Status) => send_status(&state, peer).await,
None => Ok(()),
};
if let Err(error) = sent {
debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive 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());
}
}
}
|
