forgotten nodes cancel their own records

2 files changed, 53 insertions, 15 deletions

diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index 4247090..06a2f12 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -375,3 +375,42 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
     assert_eq!(a.local_epoch(), Epoch::GENESIS);
     assert!(a.is_ready().await);
 }
+
+/// A swarm-wide forget of a node that is alive: nobody forgets it, and it cancels the
+/// record itself.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn forgotten_node_that_is_alive_cancels_its_record() {
+    use crate::document::store;
+    use crate::swarm::peer::forgotten_path;
+
+    let sim = Sim::new(9);
+    let a = sim.start("a", &[], Start::Genesis);
+    let b = sim.start("b", &["a"], Start::Join);
+    let c = sim.start("c", &["a"], Start::Join);
+    let (a, b, c) = (&a, &b, &c);
+    sim.wait_ready(&[a, b, c], SOON).await;
+
+    let record = forgotten_path(b.local_id());
+    let written = record.clone();
+    a.mutate_doc(move |doc| store::write_path(doc, &written, &json!({"at": 0, "by": "a"})))
+        .await
+        .expect("write forget record");
+
+    // The forget check runs every 5 s.
+    for node in [a, b, c] {
+        let record = record.clone();
+        sim.wait_until("the record is cancelled", SOON + SOON, move || {
+            let record = record.clone();
+            async move {
+                store::read_path(&node.doc.read().await.doc, &record)
+                    .expect("read")
+                    .is_none()
+            }
+        })
+        .await;
+    }
+    for node in [a, c] {
+        assert!(node.get_peer_state(b.local_id()).await.is_some());
+    }
+    sim.wait_converged(&[a, b, c], SOON).await;
+}
diff --git a/src/swarm/peer/forget.rs b/src/swarm/peer/forget.rs
index df09c3c..ae49a5a 100644
--- a/src/swarm/peer/forget.rs
+++ b/src/swarm/peer/forget.rs
@@ -38,11 +38,11 @@ pub(super) async fn apply_forget_records(state: &AppState) {
     }
 
     let now = Instant::now();
-    let mut cancel = Vec::new();
+    let mut cancel = false;
 
     for pk in forgotten {
         if pk == state.local_id() {
-            cancel.push(pk); // we are evidently alive
+            cancel = true; // we are evidently alive
             continue;
         }
 
@@ -52,29 +52,28 @@ pub(super) async fn apply_forget_records(state: &AppState) {
 
         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() {
+        // 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 !alive && state.peer_states.write().await.remove(&pk).is_some() {
             info!(peer = %pk, "Forgetting peer per swarm-wide record");
         }
     }
 
-    if cancel.is_empty() {
+    if !cancel {
         return;
     }
 
-    // Individual failures are logged inside; NotFound means another node already
-    // cancelled the record.
+    // NotFound: gone since we looked.
+    let local_id = state.local_id().to_owned();
     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")
-                    }
+            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>(())
         })