1 files changed, 115 insertions, 0 deletions
diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
new file mode 100644
index 0000000..a27f144
--- /dev/null
+++ b/src/swarm/peer/worker.rs
@@ -0,0 +1,115 @@
+//! The peer worker: every `PEER_TICK`, keepalives due, dormancy, HELLO retries and
+//! overdue sync replies; every `FORGET_CHECK_INTERVAL` (and at once when the node
+//! becomes ready), forget records, the member registry, stability, measurements and
+//! the state file.
+
+use std::time::Duration;
+
+use tokio::time::{Instant, MissedTickBehavior};
+use tracing::{debug, info, warn};
+
+use super::forget::apply_forget_records;
+use super::handshake::retry_hello_if_due;
+use super::keepalive::send_ping;
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::swarm::sync::SyncEvent;
+use crate::swarm::topology;
+
+const PEER_TICK: Duration = Duration::from_millis(500);
+
+const FORGET_CHECK_INTERVAL: Duration = Duration::from_secs(5);
+
+pub(crate) async fn run_peer_worker(state: AppState) {
+    let mut tick = tokio::time::interval(PEER_TICK);
+    tick.set_missed_tick_behavior(MissedTickBehavior::Delay);
+
+    info!("Starting peer maintenance worker");
+
+    let mut last_forget_check: Option<Instant> = None;
+    let mut was_ready = false;
+
+    loop {
+        tick.tick().await;
+
+        // While joining, keep handshakes fast; dormancy would slow probes to minutes.
+        let ready = state.is_ready().await;
+        if ready && !was_ready {
+            // Just became ready: register in the member list (and do the other periodic
+            // checks) now rather than at the next check. Until it is registered, other
+            // members don't count this node as a neighbour.
+            last_forget_check = None;
+        }
+        was_ready = ready;
+
+        let peers = state.peers_snapshot().await;
+
+        for (peer, peer_state_mutex) in &peers {
+            let now = Instant::now();
+            // Only neighbours are kept alive and judged; others we only answer.
+            let neighbour = topology::is_neighbour(&state, peer);
+
+            let (became_dormant, ping_due, unanswered) = {
+                let mut peer_state = peer_state_mutex.lock().await;
+                let became_dormant =
+                    ready && neighbour && peer_state.check_dormancy(&state.timing, now);
+                let ping_due = neighbour && peer_state.take_keepalive_due(&state.timing, now);
+                if !neighbour {
+                    peer_state.stop_hello_retries();
+                }
+                let unanswered = peer_state.take_overdue_sync_reply(now);
+                (became_dormant, ping_due, unanswered)
+            };
+
+            if let Some(changes) = unanswered {
+                debug!(peer = %peer, changes, "No answer to our sync message; sending again");
+                crate::swarm::source::reply_timed_out(&state, peer);
+                let _ = state
+                    .sync_events
+                    .send(SyncEvent::PeerNeedsSync(peer.clone()));
+            }
+
+            if became_dormant {
+                state.topology_changed.notify_one();
+                warn!(
+                    peer = %peer,
+                    silent_secs = state.timing.dormant_after.as_secs(),
+                    probe_secs = state.timing.dormant_probe.as_secs(),
+                    "Peer went dormant"
+                );
+            }
+
+            let probe_due =
+                !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now());
+            if ping_due || probe_due {
+                if let Err(error) = send_ping(&state, peer).await {
+                    debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed");
+                }
+            }
+
+            if let Err(error) = retry_hello_if_due(&state, peer).await {
+                warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed");
+            }
+        }
+
+        if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) {
+            apply_forget_records(&state).await;
+            crate::swarm::members::ensure_registered(&state).await;
+            crate::swarm::stability::tick(&state).await;
+            crate::swarm::stats::measure_saved_size(&state);
+            crate::swarm::source::warn_if_stalled(&state);
+            crate::swarm::state_file::update(&state, false).await;
+            {
+                let known = state.peer_states.read().await;
+                state.traffic.retain(|peer| known.contains_key(peer));
+                state
+                    .throughput
+                    .lock_or_recover()
+                    .retain(|peer| known.contains_key(peer));
+            }
+            // Fallback for what doesn't notify (`topology::run_worker`).
+            state.topology_changed.notify_one();
+            last_forget_check = Some(Instant::now());
+        }
+    }
+}