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
130
|
//! 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::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, keepalive, unanswered) = {
let mut peer_state = peer_state_mutex.lock().await;
let became_dormant =
ready && neighbour && peer_state.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);
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"
);
}
// 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());
}
}
}
|
