split peer module

1 files changed, 79 insertions, 0 deletions

diff --git a/src/swarm/peer/forget.rs b/src/swarm/peer/forget.rs
new file mode 100644
index 0000000..624ca97
--- /dev/null
+++ b/src/swarm/peer/forget.rs
@@ -0,0 +1,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;
+}