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
|
//! 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.
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;
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;
}
|
