peer worker forget records on change

6 files changed, 32 insertions, 4 deletions

diff --git a/src/node.rs b/src/node.rs
index 6de8f0c..eca0e40 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -129,6 +129,7 @@ impl Node {
             neighbours: Default::default(),
             topology_changed: Default::default(),
             join_progress: Default::default(),
+            forget_records_changed: Default::default(),
             peers_due: Default::default(),
             pulling: Default::default(),
             announce_pending: Default::default(),
diff --git a/src/state/mod.rs b/src/state/mod.rs
index ba90269..a4ea553 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -154,6 +154,9 @@ pub(crate) struct AppState {
     /// Something a joining node waits for may have happened: the swarm ID learned, the
     /// initial sync done, a rejoin ended, a seed found (`swarm::bootstrap`).
     pub(crate) join_progress: Arc<tokio::sync::Notify>,
+    /// The forget records may have changed (`swarm::changes`): the peer worker acts on
+    /// them at once instead of at its next periodic check.
+    pub(crate) forget_records_changed: Arc<AtomicBool>,
     /// When the peer worker is next due (`swarm::peer::Due`).
     pub(crate) peers_due: Arc<crate::swarm::peer::Due>,
 
diff --git a/src/swarm/changes.rs b/src/swarm/changes.rs
index 7f3c2e9..8565cb6 100644
--- a/src/swarm/changes.rs
+++ b/src/swarm/changes.rs
@@ -1,6 +1,7 @@
 //! The change publisher: after document changes, which store paths they touched
 //! (`AppState::change_events`, for the API's streams, the dashboard and remakes), and
-//! a nudge to topology when the member list changed.
+//! a nudge to topology when the member list changed, and to the peer worker when the
+//! forget records did.
 
 use std::sync::Arc;
 
@@ -62,6 +63,15 @@ pub(crate) async fn run_change_publisher(state: AppState) {
         {
             state.topology_changed.notify_one();
         }
+        if paths
+            .iter()
+            .any(|path| crate::swarm::peer::affects_forget_records(path))
+        {
+            state
+                .forget_records_changed
+                .store(true, std::sync::atomic::Ordering::Release);
+            state.peers_due.now();
+        }
 
         // Fails only when nobody is subscribed.
         let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
diff --git a/src/swarm/peer/forget.rs b/src/swarm/peer/forget.rs
index 3a97677..68266d5 100644
--- a/src/swarm/peer/forget.rs
+++ b/src/swarm/peer/forget.rs
@@ -24,6 +24,13 @@ pub(crate) fn forgotten_path(pk: &str) -> Vec<String> {
     path
 }
 
+/// Whether a change at `path` may change the forget records: one among them, or above
+/// them (the root, after an epoch switch).
+pub(crate) fn affects_forget_records(path: &[String]) -> bool {
+    let dir = forgotten_dir();
+    path.starts_with(&dir) || dir.starts_with(path)
+}
+
 fn forgotten_peers(doc: &Document) -> Vec<String> {
     store::keys_at(doc, &forgotten_dir()).unwrap_or_default()
 }
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index 589f48c..d1dea9b 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -20,7 +20,7 @@ mod keepalive;
 mod lanes;
 mod worker;
 
-pub(crate) use forget::{forget_unless_alive, forgotten_path};
+pub(crate) use forget::{affects_forget_records, forget_unless_alive, forgotten_path};
 pub(crate) use handshake::{
     apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once,
 };
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index 546f867..e8e6f8b 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -1,7 +1,8 @@
 //! The peer worker: keepalives due, dormancy, HELLO retries and overdue sync replies,
 //! when the earliest of them is due (or something may have made one due sooner,
 //! `Due`); every `FORGET_CHECK_INTERVAL` (and at once when the node becomes ready),
-//! forget records, the member registry, stability, measurements and the state file.
+//! forget records, the member registry, stability, measurements and the state file;
+//! forget records also at once when they changed.
 
 use std::time::Duration;
 
@@ -152,8 +153,14 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                 .unwrap_or(next);
         }
 
-        if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
+        let records_changed = state
+            .forget_records_changed
+            .swap(false, std::sync::atomic::Ordering::AcqRel);
+        let periodic = last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL);
+        if records_changed || periodic {
             apply_forget_records(&state, &mut forgets_applied).await;
+        }
+        if periodic {
             forget_strangers(&state).await;
             crate::swarm::members::ensure_registered(&state).await;
             crate::swarm::stability::tick(&state).await;