diff options
| author | tv <tv@krebsco.de> | 2026-10-26 06:13:05 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-10-26 06:13:05 +0100 |
| commit | 03807a3c7b33c67293f496b4f1222719113f80bb (patch) | |
| tree | c84a55d278cb74d08acb5b9f882aa9a71d5bde1e /src/swarm/peer/forget.rs | |
| parent | 013bf10824210a6270e4b9d349a22c8afbdadc76 (diff) | |
forget records only once
Diffstat (limited to 'src/swarm/peer/forget.rs')
| -rw-r--r-- | src/swarm/peer/forget.rs | 41 |
1 files changed, 34 insertions, 7 deletions
diff --git a/src/swarm/peer/forget.rs b/src/swarm/peer/forget.rs index 10a676d..6386e3e 100644 --- a/src/swarm/peer/forget.rs +++ b/src/swarm/peer/forget.rs @@ -2,9 +2,11 @@ //! 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}; @@ -26,13 +28,31 @@ 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) { +/// 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 = forgotten_peers(&state.doc.read().await.doc); + 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; } @@ -40,17 +60,24 @@ pub(super) async fn apply_forget_records(state: &AppState) { let now = Instant::now(); let mut cancel = false; - for pk in forgotten { + 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. 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 { + // 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); } } |
