From ae2a573e2753e1e7a11b94e2a721cb4d638079b3 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 20:38:27 +0100 Subject: peer worker forget records on change --- src/swarm/changes.rs | 12 +++++++++++- src/swarm/peer/forget.rs | 7 +++++++ src/swarm/peer/mod.rs | 2 +- src/swarm/peer/worker.rs | 11 +++++++++-- 4 files changed, 28 insertions(+), 4 deletions(-) (limited to 'src/swarm') 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 { 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 { 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; -- cgit v1.3.1