summaryrefslogtreecommitdiffstats
path: root/src/swarm/peer/worker.rs
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-28 02:54:55 +0100
committertv <tv@krebsco.de>2026-10-28 02:54:55 +0100
commitb88be3ea1b475b1a656298b98708026388eaf2af (patch)
tree9ffb3c42800c348ae4d67d1d9a43daa608c2269d /src/swarm/peer/worker.rs
parent24327be7e09c20a9820eac61ba3ac60fb7bdb424 (diff)
move keepalive and firewall to paths
Diffstat (limited to 'src/swarm/peer/worker.rs')
-rw-r--r--src/swarm/peer/worker.rs35
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 {