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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
//! 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::collections::HashMap;
use std::time::Duration;
use automerge::AutoCommit;
use serde_json::Value as JsonValue;
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()
}
/// The forget records this node acted on: peer → the record as it read then.
pub(super) type AppliedForgets = HashMap<String, JsonValue>;
/// Acts on each forget record once (`applied`). A peer forgotten here may be heard of
/// again before its cancel arrives (named in a peer list, or alive again), and is kept
/// then; forgetting it again every check would only make it flap. A new record (another
/// `DELETE`) is acted on again.
pub(super) async fn apply_forget_records(state: &AppState, applied: &mut AppliedForgets) {
// Before the initial sync we can't judge the records, and we have nothing to delete.
if !state.is_ready().await {
return;
}
let forgotten: Vec<(String, JsonValue)> = {
let replica = state.doc.read().await;
forgotten_peers(&replica.doc)
.into_iter()
.filter_map(|pk| {
let record = store::read_path(&replica.doc, &forgotten_path(&pk)).ok()??;
Some((pk, record))
})
.collect()
};
// Cancelled records are done with.
applied.retain(|pk, _| forgotten.iter().any(|(forgotten, _)| forgotten == pk));
if forgotten.is_empty() {
return;
}
let now = Instant::now();
let mut cancel = false;
for (pk, record) in forgotten {
if pk == state.local_id() {
cancel = true; // we are evidently alive
continue;
}
if applied.get(&pk) == Some(&record) {
continue;
}
// An alive peer is kept, and the record looked at again next time. 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. Unknown here counts as acted on: one heard of later is kept.
if state.get_peer_state(&pk).await.is_none() {
applied.insert(pk, record);
} else if forget_unless_alive(state, &pk, now).await {
info!(peer = %pk, "Forgetting peer per swarm-wide record");
applied.insert(pk, 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"
);
}
|
