summaryrefslogtreecommitdiffstats
path: root/src/swarm/peer
diff options
context:
space:
mode:
Diffstat (limited to 'src/swarm/peer')
-rw-r--r--src/swarm/peer/inbound.rs41
-rw-r--r--src/swarm/peer/keepalive.rs140
-rw-r--r--src/swarm/peer/mod.rs5
-rw-r--r--src/swarm/peer/worker.rs35
4 files changed, 167 insertions, 54 deletions
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index b04aad9..2e89739 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_keepalive, note_keepalive};
+use super::keepalive::{heal_if_diverged, local_status, note_alive, note_status, status_received};
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(());
@@ -143,40 +148,42 @@ pub(crate) async fn handle_message(
transport::ParsedMessage::Blob(blob) => {
crate::swarm::blob::handle(state, &peer, blob).await;
}
- transport::ParsedMessage::Ping(remote) => {
- let local = on_keepalive(state, &peer, &remote, reactivated).await;
+ transport::ParsedMessage::Status(remote) => {
+ status_received(state, &peer).await;
+ let local = on_status(state, &peer, &remote, reactivated).await;
if let Err(error) = state
.transport
- .send_message(&peer, &ParsedMessage::Pong(local))
+ .send_message(&peer, &ParsedMessage::StatusReply(local))
.await
{
- debug!(peer = %peer, error = %format!("{error:#}"), "PONG failed");
+ debug!(peer = %peer, error = %format!("{error:#}"), "STATUS_REPLY failed");
}
}
- transport::ParsedMessage::Pong(remote) => {
- // It answers our PING: a round-trip sample.
- state.lock_rtt().pong_received(&peer, Instant::now());
- on_keepalive(state, &peer, &remote, reactivated).await;
+ transport::ParsedMessage::StatusReply(remote) => {
+ on_status(state, &peer, &remote, reactivated).await;
}
- transport::ParsedMessage::Who(_) | transport::ParsedMessage::Me(_) => {}
+ transport::ParsedMessage::Who(_)
+ | transport::ParsedMessage::Me(_)
+ | transport::ParsedMessage::Ping(_)
+ | transport::ParsedMessage::Pong(_) => {}
}
Ok(())
}
-/// What a PING and a PONG have in common: a handshake if needed, the peer's keepalive
-/// noted, and the heal if our documents differ. Returns our keepalive for the peer.
-async fn on_keepalive(
+/// What a STATUS and a STATUS_REPLY have in common: a handshake if needed, the peer's status
+/// noted, and the heal if our documents differ. Returns our status for the peer.
+async fn on_status(
state: &AppState,
peer: &str,
- remote: &transport::Keepalive,
+ remote: &transport::Status,
reactivated: bool,
-) -> transport::Keepalive {
+) -> transport::Status {
if reactivated || peer_swarm(state, peer).await.is_none() {
ensure_handshake(state, peer).await;
}
- note_keepalive(state, peer, remote).await;
- let local = local_keepalive(state, peer).await;
+ note_status(state, peer, remote).await;
+ let local = local_status(state, peer).await;
heal_if_diverged(state, peer, &local, remote).await;
local
}
diff --git a/src/swarm/peer/keepalive.rs b/src/swarm/peer/keepalive.rs
index ce2951b..c8a93eb 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -1,12 +1,19 @@
-//! Keepalives (PING / PONG) 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 measures the path (`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,37 +21,111 @@ use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, ParsedMessage};
-/// Sends `peer` a PING with our keepalive, 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;
+ 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 (the paths layer measures the path with it).
pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
- let keepalive = local_keepalive(state, peer).await;
- state.lock_rtt().ping_sent(peer, Instant::now());
+ let nonce = crate::transport::paths::random_nonce();
state
.transport
- .send_message(peer, &ParsedMessage::Ping(keepalive))
+ .send_message(peer, &ParsedMessage::Ping(nonce))
.await
}
-/// How long shutting down waits for the last PINGs (`farewell`).
+/// A PING or PONG from `peer`: it is alive, and after a PONG the paths layer has a
+/// new round-trip time to it (`swarm::rtt`). 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(_) => {
+ if let Some(rtt) = state.transport.rtt(peer) {
+ state.lock_rtt().measured(peer, rtt, now);
+ }
+ }
+ _ => {}
+ }
+ let Some(peer_state) = state.get_peer_state(peer).await else {
+ return;
+ };
+ let reactivated = {
+ let mut peer_state = peer_state.lock().await;
+ if matches!(message, ParsedMessage::Ping(_)) {
+ // Its keepalive (or a path probe): the next is ours.
+ peer_state.keepalive_received_at = Some(now);
+ }
+ 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 PING 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);
- let mut pings = tokio::task::JoinSet::new();
+/// 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();
- pings.spawn(async move { send_ping(&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 pings.join_next().await.is_some() {} };
- let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).await;
+ let all_sent = async { while statuses.join_next().await.is_some() {} };
+ 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 keepalive for `peer`.
-pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
+/// Our status for `peer`.
+pub(crate) async fn local_status(state: &AppState, peer: &str) -> transport::Status {
// Heads and epoch are published together (`Replica::stamps`), so they always match,
// and reading them doesn't wait for a sync message being applied.
let (heads, epoch, epoch_seq) = {
@@ -64,7 +145,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
}
None => false,
};
- transport::Keepalive {
+ transport::Status {
heads,
epoch,
epoch_seq,
@@ -73,8 +154,19 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
}
}
-/// What a peer's keepalive says about it.
-pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transport::Keepalive) {
+/// A STATUS (not a reply) from `peer`: its keepalive, and a STATUS between us
+/// (`Timing::status_refresh`).
+pub(super) async fn status_received(state: &AppState, peer: &str) {
+ if let Some(peer_state) = state.get_peer_state(peer).await {
+ let mut peer_state = peer_state.lock().await;
+ let now = Instant::now();
+ peer_state.keepalive_received_at = Some(now);
+ peer_state.last_status_at = Some(now);
+ }
+}
+
+/// What a peer's status says about it.
+pub(super) async fn note_status(state: &AppState, peer: &str, remote: &transport::Status) {
if let Some(peer_state) = state.get_peer_state(peer).await {
let mut peer_state = peer_state.lock().await;
let became_ready = remote.ready && peer_state.reports_ready != Some(true);
@@ -103,8 +195,8 @@ pub(super) async fn note_keepalive(state: &AppState, peer: &str, remote: &transp
pub(super) async fn heal_if_diverged(
state: &AppState,
peer: &str,
- local: &transport::Keepalive,
- remote: &transport::Keepalive,
+ local: &transport::Status,
+ remote: &transport::Status,
) {
// Heads of documents from different epochs are never equal and never converge.
// And a peer that isn't our neighbour isn't ours to keep in step.
diff --git a/src/swarm/peer/mod.rs b/src/swarm/peer/mod.rs
index 0249f3b..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`: PING / PONG 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_ping};
+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 057e4bd..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_ping;
+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_ping(&state, peer).await
- {
- debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING 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 {
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 06:39:56 +0000