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
|
//! Swarm-wide forget records (`_continuum/forgotten/<pk>`): a peer forgotten on one
//! node is forgotten by all; a forgotten node that is alive cancels its record. And
//! locally, strangers that went silent.
use std::time::Duration;
use automerge::AutoCommit;
use tokio::time::Instant;
use tracing::{info, warn};
use crate::document::meta;
use crate::document::store::{self, StoreError};
use crate::state::{AppState, Standing};
fn forgotten_dir() -> Vec<String> {
meta::path(&[meta::FORGOTTEN])
}
pub(crate) fn forgotten_path(pk: &str) -> Vec<String> {
let mut path = forgotten_dir();
path.push(pk.to_owned());
path
}
fn forgotten_peers(doc: &AutoCommit) -> Vec<String> {
store::keys_at(doc, &forgotten_dir()).unwrap_or_default()
}
pub(super) async fn apply_forget_records(state: &AppState) {
// Before the initial sync we can't judge the records, and we have nothing to delete.
if !state.is_ready().await {
return;
}
let forgotten = forgotten_peers(&state.doc.read().await.doc);
if forgotten.is_empty() {
return;
}
let now = Instant::now();
let mut cancel = Vec::new();
for pk in forgotten {
if pk == state.local_id() {
cancel.push(pk); // we are evidently alive
continue;
}
let Some(peer_state_mutex) = state.get_peer_state(&pk).await else {
continue; // already forgotten here
};
let alive = peer_state_mutex.lock().await.is_alive(&state.timing, now);
if alive {
cancel.push(pk);
} else if state.peer_states.write().await.remove(&pk).is_some() {
info!(peer = %pk, "Forgetting peer per swarm-wide record");
}
}
if cancel.is_empty() {
return;
}
// Individual failures are logged inside; NotFound means another node already
// cancelled the record.
let _ = state
.mutate_doc(move |doc| {
for pk in &cancel {
match store::delete_path(&mut *doc, &forgotten_path(pk), false) {
Ok(()) => info!(peer = %pk, "Peer is alive; cancelling swarm-wide forget"),
Err(StoreError::NotFound) => {}
Err(error) => {
warn!(peer = %pk, error = %error, "Failed to cancel forget record")
}
}
}
Ok::<(), StoreError>(())
})
.await;
}
/// How long a peer that never became a member is kept once nothing more is heard from
/// it: one listed to us that never answered, or one that keeps failing the handshake
/// (another swarm, a wrong key). Anyone can name peers to a node, so without a limit
/// the map of peers would only grow.
const STRANGER_KEPT: Duration = Duration::from_secs(24 * 3600);
/// Forgets peers that never became members and were silent for `STRANGER_KEPT`. Only
/// here, not swarm-wide; one that comes back is a stranger again.
pub(super) async fn forget_strangers(state: &AppState) {
let now = Instant::now();
let mut strangers = Vec::new();
for (peer, peer_state) in state.peers_snapshot().await {
let peer_state = peer_state.lock().await;
let quiet_since = peer_state.last_seen.unwrap_or(peer_state.added_at);
if !matches!(peer_state.standing, Standing::Member { .. })
&& now.saturating_duration_since(quiet_since) >= STRANGER_KEPT
{
strangers.push(peer);
}
}
if strangers.is_empty() {
return;
}
let mut peer_states = state.peer_states.write().await;
for peer in &strangers {
peer_states.remove(peer);
}
info!(
forgotten = strangers.len(),
"Forgot peers that never became members and were silent for a day"
);
}
|
