forget records only once

2 files changed, 37 insertions, 9 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);
         }
     }
 
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 5161ae6..20e2542 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -8,7 +8,7 @@ use std::time::Duration;
 use tokio::time::{Instant, MissedTickBehavior};
 use tracing::{debug, info, warn};
 
-use super::forget::{apply_forget_records, forget_strangers};
+use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
 use super::handshake::retry_hello_if_due;
 use super::keepalive::send_ping;
 use crate::locks::MutexExt;
@@ -27,6 +27,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
     info!("Starting peer maintenance worker");
 
     let mut last_forget_check: Option<Instant> = None;
+    let mut forgets_applied = AppliedForgets::new();
     let mut was_ready = false;
 
     loop {
@@ -93,7 +94,7 @@ pub(crate) async fn run_peer_worker(state: AppState) {
         }
 
         if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
-            apply_forget_records(&state).await;
+            apply_forget_records(&state, &mut forgets_applied).await;
             forget_strangers(&state).await;
             crate::swarm::members::ensure_registered(&state).await;
             crate::swarm::stability::tick(&state).await;