summaryrefslogtreecommitdiffstats
path: root/src/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm')
-rw-r--r--src/swarm/bootstrap.rs1
-rw-r--r--src/swarm/peer/inbound.rs11
-rw-r--r--src/swarm/peer/keepalive.rs98
-rw-r--r--src/swarm/peer/mod.rs5
-rw-r--r--src/swarm/peer/worker.rs35
-rw-r--r--src/swarm/rtt.rs108
6 files changed, 196 insertions, 62 deletions
diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 7c4c444..a9ab1b2 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -119,6 +119,7 @@ pub(crate) async fn bootstrap(state: &AppState, seeds: &[String], start: Start)
if state.initial_sync_done.load(Ordering::Acquire) && !state.remake.is_rejoining() {
state.mark_ready(swarm_id.clone()).await?;
info!(swarm_id = %swarm_id, "Node is ready");
+ crate::swarm::peer::announce(state);
return Ok(());
}
}
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 92a1a5e..0eaefb9 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -11,7 +11,7 @@ use super::epoch::observe_peer_epoch;
use super::handshake::{
discover_peers, ensure_handshake, establish_peer, peer_swarm, remember_peer, send_hello_reply,
};
-use super::keepalive::{heal_if_diverged, local_status, note_status};
+use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status};
use super::lanes::SyncLanes;
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
@@ -30,6 +30,11 @@ pub(crate) async fn handle_message(
debug!("Ignoring message from self");
return Ok(());
}
+ // PING and PONG only say the peer is alive (`keepalive::note_alive`).
+ if matches!(message, ParsedMessage::Ping(_) | ParsedMessage::Pong(_)) {
+ note_alive(state, &peer, &message).await;
+ return Ok(());
+ }
// WHO and ME end at the identity layer (`transport::identity`).
let Some((epoch, epoch_seq)) = message.epoch_stamp() else {
return Ok(());
@@ -154,10 +159,6 @@ pub(crate) async fn handle_message(
}
}
transport::ParsedMessage::StatusReply(remote) => {
- // It answers our STATUS: a round-trip sample.
- state
- .lock_rtt()
- .status_reply_received(&peer, Instant::now());
on_status(state, &peer, &remote, reactivated).await;
}
transport::ParsedMessage::Who(_)
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index 810c377..460b71e 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -1,12 +1,19 @@
-//! Keepalives (STATUS / STATUS_REPLY) and the self-healing re-handshake when they show our
-//! documents differ while sync is idle.
+//! Keepalives and the self-healing re-handshake.
+//!
+//! A quiet pair's keepalive (`PeerState::take_keepalive_due`) is a PING, answered with a
+//! PONG by the peer's paths layer (`transport::paths`), over the same path: it tells both
+//! sides the other is alive, and gives the round trip (`swarm::rtt`). Every `Timing::status_refresh` it is a
+//! STATUS instead, answered with a STATUS_REPLY: each side's state (heads, readiness,
+//! epoch), which also go out as they change (relay announcements, the epoch after a
+//! switch, readiness: `announce`). When a STATUS shows our documents differ while sync
+//! is idle, the pair re-handshakes (`heal_if_diverged`).
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::Result;
use tokio::time::Instant;
-use tracing::warn;
+use tracing::{debug, info, warn};
use super::handshake::ensure_handshake;
use crate::state::{AppState, Liveness, Standing};
@@ -14,33 +21,100 @@ use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, ParsedMessage};
-/// Sends `peer` a STATUS with our status, noting the time for its round trip.
+/// Sends `peer` a STATUS with our status.
pub(crate) async fn send_status(state: &AppState, peer: &str) -> Result<()> {
let status = local_status(state, peer).await;
- state.lock_rtt().status_sent(peer, Instant::now());
+ if let Some(peer_state) = state.get_peer_state(peer).await {
+ peer_state.lock().await.last_status_at = Some(Instant::now());
+ }
state
.transport
.send_message(peer, &ParsedMessage::Status(status))
.await
}
+/// Sends `peer` a PING, noting it for its round trip.
+pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
+ let nonce = crate::transport::paths::random_nonce();
+ state.lock_rtt().ping_sent(peer, nonce, Instant::now());
+ state
+ .transport
+ .send_message(peer, &ParsedMessage::Ping(nonce))
+ .await
+}
+
+/// A PING or PONG from `peer`: it is alive, and a PONG may answer our PING (its round
+/// trip). A transport without the paths layer (`Transport::answers_pings`: in tests)
+/// leaves answering a PING to us. Only peers we know already: a PING makes none.
+pub(super) async fn note_alive(state: &AppState, peer: &str, message: &ParsedMessage) {
+ let now = Instant::now();
+ match message {
+ ParsedMessage::Ping(nonce) if !state.transport.answers_pings() => {
+ let pong = ParsedMessage::Pong(*nonce);
+ if let Err(error) = state.transport.send_message(peer, &pong).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
+ }
+ }
+ ParsedMessage::Pong(nonce) => state.lock_rtt().pong_received(peer, *nonce, now),
+ _ => {}
+ }
+ let Some(peer_state) = state.get_peer_state(peer).await else {
+ return;
+ };
+ let reactivated = {
+ let mut peer_state = peer_state.lock().await;
+ if peer_state.unreachable_since.is_some() {
+ // Reachable again (`mark_seen` clears it).
+ state.topology_changed.notify_one();
+ }
+ peer_state.mark_seen(now)
+ };
+ crate::swarm::source::heard_from(state, peer);
+ if reactivated {
+ info!(peer = %peer, "Dormant peer is active again; re-handshaking");
+ ensure_handshake(state, peer).await;
+ }
+}
+
/// How long shutting down waits for the last STATUSes (`farewell`).
const FAREWELL_TIMEOUT: Duration = Duration::from_secs(2);
-/// At shutdown: one last STATUS to each neighbour, saying we aren't ready. A neighbour
-/// catching up from us picks another source now, instead of once we go dormant. Bounded
-/// in time: our transport may be gone already.
-pub(crate) async fn farewell(state: &AppState) {
- state.stopping.store(true, Ordering::Release);
+/// A STATUS to each neighbour, at most `within` (all sent: none).
+async fn status_to_neighbours(state: &AppState, within: Option<Duration>) {
let mut statuses = tokio::task::JoinSet::new();
for (peer, _) in state.peers_snapshot().await {
if topology::is_neighbour(state, &peer) {
let state = state.clone();
- statuses.spawn(async move { send_status(&state, &peer).await });
+ statuses.spawn(async move {
+ if let Err(error) = send_status(&state, &peer).await {
+ debug!(peer = %peer, error = %format!("{error:#}"), "STATUS failed");
+ }
+ });
}
}
let all_sent = async { while statuses.join_next().await.is_some() {} };
- let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).await;
+ match within {
+ Some(within) => {
+ let _ = tokio::time::timeout(within, all_sent).await;
+ }
+ None => all_sent.await,
+ }
+}
+
+/// Our readiness changed (we became ready, or ready again after a rejoin): a STATUS
+/// to each neighbour now, rather than with the next refresh. A peer opens sync towards
+/// us only once it knows we are ready.
+pub(crate) fn announce(state: &AppState) {
+ let state = state.clone();
+ tokio::spawn(async move { status_to_neighbours(&state, None).await });
+}
+
+/// At shutdown: one last STATUS to each neighbour, saying we aren't ready. A neighbour
+/// catching up from us picks another source now, instead of once we go dormant. Bounded
+/// in time: our transport may be gone already.
+pub(crate) async fn farewell(state: &AppState) {
+ state.stopping.store(true, Ordering::Release);
+ status_to_neighbours(state, Some(FAREWELL_TIMEOUT)).await;
}
/// Our status for `peer`.
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index 415cddf..326f3a9 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -2,7 +2,8 @@
//! swarm-wide forget records, and dispatch of inbound messages.
//!
//! * `handshake`: discovering peers, HELLO / HELLO_REPLY, membership.
-//! * `keepalive`: STATUS / STATUS_REPLY and the self-healing re-handshake.
+//! * `keepalive`: PING / PONG and STATUS / STATUS_REPLY, and the self-healing
+//! re-handshake.
//! * `epoch`: the epochs peers advertise.
//! * `admit`: admission with a swarm key.
//! * `forget`: swarm-wide forget records.
@@ -24,6 +25,6 @@ pub(crate) use handshake::{
apply_neighbour_change, peer_swarm, rehandshake, remember_peer, send_hello_once,
};
pub(crate) use inbound::handle_message;
-pub(crate) use keepalive::{farewell, send_status};
+pub(crate) use keepalive::{announce, farewell, send_status};
pub(crate) use lanes::SyncLanes;
pub(crate) use worker::run_peer_worker;
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 {
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs
index 01d10ee..466fc98 100644
--- a/src/swarm/rtt.rs
+++ b/src/swarm/rtt.rs
@@ -1,9 +1,11 @@
-//! Round-trip times to peers, measured from traffic we send anyway: STATUS → STATUS_REPLY and
-//! HELLO → HELLO_REPLY. No probes of our own.
+//! Round-trip times to peers, measured from traffic we send anyway: the keepalive PING
+//! → PONG (`swarm::peer::keepalive`), and HELLO → HELLO_REPLY.
//!
-//! A sample only counts when exactly one request was outstanding: with two, we can't tell
-//! which one the answer belongs to (an answer to the older one would look too fast).
-//! Samples are smoothed (exponential moving average, weight 1/4) to ride out jitter.
+//! A PONG answers the PING with its nonce, so it is attributed exactly; a PONG to an
+//! older PING (one a newer one replaced) is no sample. A HELLO_REPLY only counts when
+//! exactly one HELLO was outstanding: with two, we can't tell which one the answer
+//! belongs to (an answer to the older one would look too fast). Samples are smoothed
+//! (exponential moving average, weight 1/4) to ride out jitter.
//!
//! What we use is the *effective* round-trip time: the smoothed value, raised to the age
//! of the oldest request still unanswered (the round trip takes at least that long), so
@@ -18,11 +20,20 @@ use tokio::time::Instant;
#[derive(Debug, Default)]
pub(crate) struct RttTable {
- statuses: HashMap<String, Outstanding>,
+ pings: HashMap<String, Ping>,
hellos: HashMap<String, Outstanding>,
smoothed: HashMap<String, Sample>,
}
+/// The PING out to a peer.
+#[derive(Debug, Clone, Copy)]
+struct Ping {
+ nonce: u64,
+ sent: Instant,
+ /// When the oldest PING still unanswered went out (an earlier one, replaced).
+ waiting_since: Instant,
+}
+
#[derive(Debug, Clone, Copy)]
struct Sample {
rtt: Duration,
@@ -36,13 +47,26 @@ struct Outstanding {
}
impl RttTable {
- pub(crate) fn status_sent(&mut self, peer: &str, now: Instant) {
- Self::sent(&mut self.statuses, peer, now);
+ pub(crate) fn ping_sent(&mut self, peer: &str, nonce: u64, now: Instant) {
+ let waiting_since = self
+ .pings
+ .get(peer)
+ .map_or(now, |earlier| earlier.waiting_since);
+ self.pings.insert(
+ peer.to_owned(),
+ Ping {
+ nonce,
+ sent: now,
+ waiting_since,
+ },
+ );
}
- pub(crate) fn status_reply_received(&mut self, peer: &str, now: Instant) {
- if let Some(sample) = Self::answered(&mut self.statuses, peer, now) {
- self.record(peer, sample, now);
+ pub(crate) fn pong_received(&mut self, peer: &str, nonce: u64, now: Instant) {
+ if self.pings.get(peer).is_some_and(|ping| ping.nonce == nonce)
+ && let Some(ping) = self.pings.remove(peer)
+ {
+ self.record(peer, now.saturating_duration_since(ping.sent), now);
}
}
@@ -68,12 +92,15 @@ impl RttTable {
.smoothed
.get(peer)
.filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?;
- let waiting = [self.statuses.get(peer), self.hellos.get(peer)]
- .into_iter()
- .flatten()
- .map(|outstanding| now.saturating_duration_since(outstanding.sent))
- .max()
- .unwrap_or_default();
+ let waiting = [
+ self.pings.get(peer).map(|ping| ping.waiting_since),
+ self.hellos.get(peer).map(|outstanding| outstanding.sent),
+ ]
+ .into_iter()
+ .flatten()
+ .map(|sent| now.saturating_duration_since(sent))
+ .max()
+ .unwrap_or_default();
Some(sample.rtt.max(waiting))
}
@@ -128,31 +155,47 @@ mod tests {
let mut table = RttTable::default();
let start = Instant::now();
let at = |ms| start + Duration::from_millis(ms);
- table.status_sent("a", start);
- table.status_reply_received("a", at(100));
+ table.ping_sent("a", 1, start);
+ table.pong_received("a", 1, at(100));
assert_eq!(
table.effective("a", at(100), FRESH),
Some(Duration::from_millis(100))
);
- table.status_sent("a", at(100));
- table.status_reply_received("a", at(300));
+ table.ping_sent("a", 2, at(100));
+ table.pong_received("a", 2, at(300));
assert_eq!(
table.effective("a", at(300), FRESH),
Some(Duration::from_millis(125))
);
- // Two STATUSes out: the STATUS_REPLY can't be attributed.
- table.status_sent("a", at(300));
- table.status_sent("a", at(400));
- table.status_reply_received("a", at(450));
+ // A newer PING replaced it: a PONG to the older one is no sample.
+ table.ping_sent("a", 3, at(300));
+ table.ping_sent("a", 4, at(400));
+ table.pong_received("a", 3, at(410));
assert_eq!(
- table.effective("a", at(450), FRESH),
+ table.effective("a", at(410), FRESH),
Some(Duration::from_millis(125))
);
+ table.pong_received("a", 4, at(500));
+ assert_eq!(
+ table.effective("a", at(500), FRESH),
+ Some(Duration::from_micros(118_750)) // (125 * 3 + 100) / 4
+ );
+
+ // Two HELLOs out: the HELLO_REPLY can't be attributed.
+ table.hello_sent("c", start);
+ table.hello_reply_received("c", at(100));
+ table.hello_sent("c", at(100));
+ table.hello_sent("c", at(200));
+ table.hello_reply_received("c", at(210));
+ assert_eq!(
+ table.effective("c", at(210), FRESH),
+ Some(Duration::from_millis(100))
+ );
- // An unsolicited STATUS_REPLY changes nothing.
- table.status_reply_received("b", start);
+ // An unsolicited PONG changes nothing.
+ table.pong_received("b", 9, start);
assert_eq!(table.effective("b", start, FRESH), None);
}
@@ -160,12 +203,13 @@ mod tests {
fn unanswered_requests_raise_it_and_old_samples_expire() {
let mut table = RttTable::default();
let start = Instant::now();
- table.status_sent("a", start);
- table.status_reply_received("a", start + Duration::from_millis(50));
+ table.ping_sent("a", 1, start);
+ table.pong_received("a", 1, start + Duration::from_millis(50));
- // A STATUS unanswered for 40 s: at least 40 s.
+ // PINGs unanswered for 40 s: at least 40 s, from the oldest.
let sent = start + Duration::from_secs(60);
- table.status_sent("a", sent);
+ table.ping_sent("a", 2, sent);
+ table.ping_sent("a", 3, sent + Duration::from_secs(30));
let later = sent + Duration::from_secs(40);
assert_eq!(
table.effective("a", later, FRESH),
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 07:36:10 +0000