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
|
//! 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 = false;
for pk in forgotten {
if pk == state.local_id() {
cancel = true; // we are evidently alive
continue;
}
// An alive peer is kept. Only the peer itself cancels its record: it gets the
// record from whoever it is alive to, and one cancel is one change instead of
// one from every node that hears from it.
if forget_unless_alive(state, &pk, now).await {
info!(peer = %pk, "Forgetting peer per swarm-wide record");
}
}
if !cancel {
return;
}
// NotFound: gone since we looked.
let local_id = state.local_id().to_owned();
let _ = state
.mutate_doc(move |doc| {
match store::delete_path(&mut *doc, &forgotten_path(&local_id), false) {
Ok(()) => {
info!("We were forgotten swarm-wide but are alive; cancelling the record")
}
Err(StoreError::NotFound) => {}
Err(error) => warn!(error = %error, "Failed to cancel forget record"),
}
Ok::<(), StoreError>(())
})
.await;
}
/// Forgets `pk` here unless it is alive, and returns whether it did. Judged and removed
/// under the same locks, so a message that arrives meanwhile can't be lost to a peer
/// state that is already gone. A peer whose state is in use is left for the next check.
pub(crate) async fn forget_unless_alive(state: &AppState, pk: &str, now: Instant) -> bool {
let mut peer_states = state.peer_states.write().await;
let Some(peer_state) = peer_states.get(pk) else {
return false;
};
// Not waiting: holding the map, we must not wait for a peer's lock (its holder may
// be waiting for the map).
let Ok(peer_state) = peer_state.try_lock() else {
return false;
};
if peer_state.is_alive(&state.timing, now) {
return false;
}
drop(peer_state);
peer_states.remove(pk);
true
}
/// 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"
);
}
|
