move keepalive and firewall to paths

1 files changed, 24 insertions, 11 deletions

diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs
index c68e2ea..f3d0e84 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -10,9 +10,9 @@ use tracing::{debug, info, warn};
 
 use super::forget::{AppliedForgets, apply_forget_records, forget_strangers};
 use super::handshake::retry_hello_if_due;
-use super::keepalive::send_status;
+use super::keepalive::{send_ping, send_status};
 use crate::locks::MutexExt;
-use crate::state::AppState;
+use crate::state::{AppState, Keepalive};
 use crate::swarm::sync::SyncEvent;
 use crate::swarm::topology;
 
@@ -50,16 +50,20 @@ pub(crate) async fn run_peer_worker(state: AppState) {
             // 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 (became_dormant, keepalive, 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);
+                let keepalive = if neighbour {
+                    peer_state.take_keepalive_due(&state.timing, now)
+                } else {
+                    None
+                };
                 if !neighbour {
                     peer_state.stop_hello_retries();
                 }
                 let unanswered = peer_state.take_overdue_sync_reply(now);
-                (became_dormant, ping_due, unanswered)
+                (became_dormant, keepalive, unanswered)
             };
 
             if let Some(changes) = unanswered {
@@ -80,12 +84,21 @@ pub(crate) async fn run_peer_worker(state: AppState) {
                 );
             }
 
-            let probe_due =
-                !ping_due && crate::swarm::source::probe_due(&state, peer, Instant::now());
-            if (ping_due || probe_due)
-                && let Err(error) = send_status(&state, peer).await
-            {
-                debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive STATUS failed");
+            // The catch-up source gets STATUSes: we need its readiness, and whether it
+            // syncs with us.
+            let keepalive = match keepalive {
+                None if crate::swarm::source::probe_due(&state, peer, Instant::now()) => {
+                    Some(Keepalive::Status)
+                }
+                keepalive => keepalive,
+            };
+            let sent = match keepalive {
+                Some(Keepalive::Ping) => send_ping(&state, peer).await,
+                Some(Keepalive::Status) => send_status(&state, peer).await,
+                None => Ok(()),
+            };
+            if let Err(error) = sent {
+                debug!(peer = %peer, kind = ?keepalive, error = %format!("{error:#}"), "Keepalive failed");
             }
 
             if let Err(error) = retry_hello_if_due(&state, peer).await {