summaryrefslogtreecommitdiffstats
path: root/src/state
diff options
context:
space:
mode:
Diffstat (limited to 'src/state')
-rw-r--r--src/state/host.rs30
-rw-r--r--src/state/mod.rs9
-rw-r--r--src/state/peer.rs168
-rw-r--r--src/state/timing.rs18
4 files changed, 186 insertions, 39 deletions
diff --git a/src/state/host.rs b/src/state/host.rs
index f6c460d..6eab803 100644
--- a/src/state/host.rs
+++ b/src/state/host.rs
@@ -29,7 +29,7 @@ impl AppState {
// Tell peers about the new epoch now instead of with the next keepalive (up to a
// keepalive interval later). Ready peers already on it learn it from our sync
- // opening, which goes out right away; everyone else gets a PING, so peers that
+ // opening, which goes out right away; everyone else gets a STATUS, so peers that
// never saw the commit start catching up (we don't sync with them, so nothing
// else would tell them).
let local = self.local_epoch();
@@ -52,7 +52,7 @@ impl AppState {
}
for id in announce {
- if let Err(error) = crate::swarm::peer::send_ping(self, id).await {
+ if let Err(error) = crate::swarm::peer::send_status(self, id).await {
debug!(peer = %id, error = %format!("{error:#}"), "Epoch announcement failed");
}
}
@@ -161,14 +161,24 @@ impl Host for AppState {
fn resume(&self) -> impl Future<Output = ()> + Send {
let state = self.clone();
async move {
- let mut lifecycle = state.lifecycle.write().await;
- if let NodeLifecycle::Joining {
- discovered_swarm_id: Some(swarm_id),
- } = &*lifecycle
- {
- let swarm_id = swarm_id.clone();
- debug!(swarm_id = %swarm_id, "Node resumed");
- *lifecycle = NodeLifecycle::Ready { swarm_id };
+ let resumed = {
+ let mut lifecycle = state.lifecycle.write().await;
+ let swarm_id = match &*lifecycle {
+ NodeLifecycle::Joining {
+ discovered_swarm_id: Some(swarm_id),
+ } => Some(swarm_id.clone()),
+ _ => None,
+ };
+ if let Some(swarm_id) = &swarm_id {
+ debug!(swarm_id = %swarm_id, "Node resumed");
+ *lifecycle = NodeLifecycle::Ready {
+ swarm_id: swarm_id.clone(),
+ };
+ }
+ swarm_id.is_some()
+ };
+ if resumed {
+ crate::swarm::peer::announce(&state);
}
}
}
diff --git a/src/state/mod.rs b/src/state/mod.rs
index c8243a2..fec164d 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -24,7 +24,7 @@ mod read;
mod session;
mod timing;
-pub(crate) use peer::{HelloState, Incidents, Liveness, PeerState, Standing};
+pub(crate) use peer::{HelloState, Incidents, Keepalive, Liveness, PeerState, Standing};
pub(crate) use read::{ReadSource, StaleView};
pub(crate) use session::SyncSession;
pub(crate) use timing::Timing;
@@ -40,7 +40,8 @@ use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
use tokio::time::Instant;
use tracing::{debug, error, warn};
-use crate::document::replica::{Epoch, Replica, Stamp};
+use crate::document::lock::DocumentLock;
+use crate::document::replica::{Epoch, Stamp};
use crate::locks::MutexExt;
use crate::remake::{Host, RemakeState};
use crate::swarm::sync::{ChangeSet, SyncEvent};
@@ -58,7 +59,7 @@ pub(crate) enum NodeLifecycle {
#[derive(Clone)]
pub(crate) struct AppState {
/// The document and its epoch. Derefs to `AutoCommit`.
- pub(crate) doc: Arc<RwLock<Replica>>,
+ pub(crate) doc: Arc<DocumentLock>,
/// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers).
/// Only ever changed while holding `doc`'s write lock, together with `doc.epoch`.
@@ -167,7 +168,7 @@ pub(crate) struct CatchUpSource {
pub(crate) reply_timeouts: u32,
/// When we last warned that catching up from it stalls.
pub(crate) last_warned: Option<Instant>,
- /// Our last probe PING to it.
+ /// Our last probe STATUS to it.
pub(crate) last_probe: Option<Instant>,
/// Its keepalives in a row saying it wouldn't sync with us.
pub(crate) refusals: u32,
diff --git a/src/state/peer.rs b/src/state/peer.rs
index 8086f63..b723ccd 100644
--- a/src/state/peer.rs
+++ b/src/state/peer.rs
@@ -67,7 +67,7 @@ const HELLO_RETRY_MAX: Duration = Duration::from_secs(30);
#[derive(Clone)]
pub(crate) struct PeerState {
- /// Whether we lead the pair (`swarm::pair`): we send the keepalive PINGs. Fixed for
+ /// Whether we lead the pair (`swarm::pair`): we send the first keepalive. Fixed for
/// the pair, so computed once.
pub(crate) leads: bool,
@@ -85,7 +85,12 @@ pub(crate) struct PeerState {
pub(crate) liveness: Liveness,
pub(crate) added_at: Instant,
pub(crate) last_seen: Option<Instant>,
- pub(crate) last_ping_at: Option<Instant>,
+ /// Our last keepalive to it (a PING or a STATUS).
+ pub(crate) last_keepalive_at: Option<Instant>,
+ /// Its last keepalive to us (a PING or a STATUS): after it, the next is ours.
+ pub(crate) keepalive_received_at: Option<Instant>,
+ /// The last STATUS between us, either way (`status_refresh`).
+ pub(crate) last_status_at: Option<Instant>,
/// Last time a sync message from this peer was applied (survives resets: the
/// keepalive heal looks at it).
@@ -133,7 +138,8 @@ pub(crate) struct PeerState {
}
/// Trouble with a peer, counted since it was first listed. Lost keepalives and refused
-/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING.
+/// sends are in the traffic counters (`transport::traffic`): every PONG answers a PING,
+/// every STATUS_REPLY a STATUS.
#[derive(Debug, Default, Clone, Copy, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Incidents {
@@ -157,6 +163,15 @@ impl Incidents {
}
}
+/// What a keepalive due is (`PeerState::take_keepalive_due`).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum Keepalive {
+ /// A PING: the peer's transport answers it with a PONG.
+ Ping,
+ /// A STATUS: our state, answered with the peer's.
+ Status,
+}
+
/// Unanswered HELLOs after which a peer counts as unreachable (1 + 2 + 4 s apart).
const UNREACHABLE_AFTER_HELLOS: u32 = 3;
@@ -172,7 +187,9 @@ impl PeerState {
liveness: Liveness::Active,
added_at: Instant::now(),
last_seen: None,
- last_ping_at: None,
+ last_keepalive_at: None,
+ keepalive_received_at: None,
+ last_status_at: None,
last_sync_rx: None,
last_resync_at: None,
diverged_since: None,
@@ -287,17 +304,38 @@ impl PeerState {
true
}
- /// True (and records the ping) if an established, quiet peer needs a keepalive.
+ /// The keepalive an established, quiet peer needs, if any (and records it): a PING,
+ /// or a STATUS if none went between us for `status_refresh`.
+ ///
+ /// A PING and its PONG (or a STATUS and its STATUS_REPLY) tell both sides that the
+ /// other is alive, so one per interval does for the pair. The two sides take turns:
+ /// after a keepalive from the peer, the next is ours (the one leading the pair,
+ /// `leads`, sends the first), so both measure the round trip and each sends every
+ /// other interval. On our turn we send once the pair was quiet for an interval; on its
+ /// turn only after half an interval more, in case it doesn't (a keepalive or its
+ /// answer was lost). Without turns, both sides would get due at about the same time
+ /// (each one's quiet time starts with the same exchange) and their keepalives would
+ /// often cross.
///
- /// One PING and its PONG tell both sides that the other is alive, so only one side of
- /// a pair needs to ping: the one leading the pair (`leads`). Each of its PINGs
- /// makes the other side's peer look recently heard from, so the other side never
- /// pings while it does. If the pings stop, the other side pings once it has heard
- /// nothing for half an interval more. Without that, both sides would get due at about
- /// the same time (each one's quiet time starts with the same exchange) and their
- /// PINGs would often cross.
- pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
- let quiet_enough = if self.leads {
+ /// When they do cross anyway (both sent before seeing the other's: the peer's came
+ /// within half an interval after ours), each side would take the next turn for its
+ /// own and they would keep crossing; then the next turn is the leader's.
+ pub(crate) fn take_keepalive_due(
+ &mut self,
+ timing: &Timing,
+ now: Instant,
+ ) -> Option<Keepalive> {
+ let our_turn = match (self.last_keepalive_at, self.keepalive_received_at) {
+ (None, None) => self.leads,
+ (None, Some(_)) => true,
+ (Some(_), None) => false,
+ (Some(sent), Some(received)) if received > sent => {
+ let crossed = received.saturating_duration_since(sent) < timing.keepalive / 2;
+ !crossed || self.leads
+ }
+ (Some(_), Some(_)) => false,
+ };
+ let quiet_enough = if our_turn {
timing.keepalive
} else {
timing.keepalive + timing.keepalive / 2
@@ -307,16 +345,24 @@ impl PeerState {
&& now.saturating_duration_since(self.added_at) >= timing.keepalive
&& self.quiet_for(now) >= quiet_enough
&& self
- .last_ping_at
+ .last_keepalive_at
.is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
-
- if due {
- self.last_ping_at = Some(now);
+ if !due {
+ return None;
}
- due
+ self.last_keepalive_at = Some(now);
+ let status = self
+ .last_status_at
+ .is_none_or(|t| now.saturating_duration_since(t) >= timing.status_refresh);
+ Some(if status {
+ Keepalive::Status
+ } else {
+ Keepalive::Ping
+ })
}
- /// Self-healing check on PING/PONG. Returns true if a re-handshake should start.
+ /// Self-healing check on a STATUS or STATUS_REPLY. Returns true if a re-handshake
+ /// should start.
pub(crate) fn needs_heal(
&mut self,
local: &transport::HeadsSummary,
@@ -378,6 +424,88 @@ impl PeerState {
mod tests {
use super::*;
+ /// The two sides of a pair take turns: each sends every other interval.
+ #[test]
+ fn keepalives_take_turns() {
+ let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
+ let mut a = PeerState::new(true, &timing);
+ let mut b = PeerState::new(false, &timing);
+ a.hello_state = HelloState::Established;
+ b.hello_state = HelloState::Established;
+ b.added_at = a.added_at;
+ let mut senders = Vec::new();
+ let mut now = a.added_at;
+ for _ in 0..40 {
+ now += Duration::from_secs(5);
+ // Whoever is due sends; the other one gets it at once, and both hear from
+ // each other (the keepalive, its answer).
+ if a.take_keepalive_due(&timing, now).is_some() {
+ senders.push("a");
+ b.keepalive_received_at = Some(now);
+ } else if b.take_keepalive_due(&timing, now).is_some() {
+ senders.push("b");
+ a.keepalive_received_at = Some(now);
+ } else {
+ continue;
+ }
+ a.mark_seen(now);
+ b.mark_seen(now);
+ }
+ // 200 s: one keepalive a minute, a first, then taking turns.
+ assert_eq!(senders, ["a", "b", "a"]);
+ }
+
+ /// Keepalives that crossed (both sides sent before seeing the other's) leave the next
+ /// turn to the leader, rather than both taking it and crossing again.
+ #[test]
+ fn crossed_keepalives_go_back_to_taking_turns() {
+ let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
+ let mut a = PeerState::new(true, &timing);
+ let mut b = PeerState::new(false, &timing);
+ let start = a.added_at;
+ b.added_at = start;
+ for peer in [&mut a, &mut b] {
+ peer.hello_state = HelloState::Established;
+ }
+ // Both sent at 100 s and got the other's a moment later.
+ let sent = start + Duration::from_secs(100);
+ for peer in [&mut a, &mut b] {
+ peer.last_keepalive_at = Some(sent);
+ peer.keepalive_received_at = Some(sent + Duration::from_millis(50));
+ peer.mark_seen(sent + Duration::from_millis(50));
+ }
+ let next = sent + timing.keepalive + Duration::from_secs(1);
+ assert!(a.take_keepalive_due(&timing, next).is_some()); // The leader's turn.
+ assert!(b.take_keepalive_due(&timing, next).is_none());
+ }
+
+ /// Keepalives are PINGs, every `status_refresh` a STATUS; only when quiet.
+ #[test]
+ fn keepalives_are_pings_and_now_and_then_a_status() {
+ let timing = Timing::from_keepalive(Duration::from_secs(60), Duration::from_secs(300));
+ let mut peer = PeerState::new(true, &timing);
+ peer.hello_state = HelloState::Established;
+ let start = peer.added_at;
+ assert_eq!(peer.take_keepalive_due(&timing, start), None); // Too new.
+ let mut kinds = Vec::new();
+ for interval in 1..=6 {
+ let now = start + timing.keepalive * interval;
+ let due = peer.take_keepalive_due(&timing, now);
+ if due == Some(Keepalive::Status) {
+ peer.last_status_at = Some(now);
+ }
+ kinds.extend(due);
+ // Not again within the interval.
+ assert_eq!(peer.take_keepalive_due(&timing, now), None);
+ }
+ use Keepalive::{Ping, Status};
+ assert_eq!(kinds, [Status, Ping, Ping, Status, Ping, Ping]);
+ // Heard from just now: nothing due.
+ let now = start + timing.keepalive * 7;
+ peer.mark_seen(now);
+ assert_eq!(peer.take_keepalive_due(&timing, now), None);
+ }
+
/// R41: unreachability lasts until the peer is heard from, not for a set time, and
/// probes back off meanwhile.
#[test]
diff --git a/src/state/timing.rs b/src/state/timing.rs
index 0dab3da..9055ed6 100644
--- a/src/state/timing.rs
+++ b/src/state/timing.rs
@@ -18,10 +18,14 @@ const PROBE_MAX_FACTOR: u32 = 6;
#[derive(Clone, Copy, Debug)]
pub(crate) struct Timing {
pub(crate) keepalive: Duration,
+ /// How often a quiet pair's keepalive is a STATUS rather than a PING
+ /// (`PeerState::take_keepalive_due`): its state (heads, readiness, epoch) for the heal
+ /// and against lost announcements. Changes go out as they happen anyway.
+ pub(crate) status_refresh: Duration,
pub(crate) dormant_after: Duration,
pub(crate) resync_after: Duration,
pub(crate) dormant_probe: Duration,
- /// While catching up: PING the source after this much silence (`swarm::source`).
+ /// While catching up: send the source a STATUS after this much silence (`swarm::source`).
pub(crate) catch_up_probe: Duration,
/// First wait for an answer to a sync message before sending again; doubles while
/// answers stay out (`SyncSession::reply_backoff`).
@@ -42,9 +46,10 @@ impl Timing {
/// The timings for `--keepalive`.
///
/// Each node keeps about `k` neighbours (`--neighbours`) whatever the swarm's size,
- /// and only one side of a pair pings, so idle, a node sends about `k` messages and
- /// receives about `k` (PINGs and PONGs) per keepalive interval (more on a stable
- /// member others prefer). The interval trades that against how soon a silent peer is
+ /// and only one side of a pair sends them, so idle, a node sends about `k` messages and
+ /// receives about `k` per keepalive interval (more on a stable member others prefer):
+ /// PINGs and PONGs (a few bytes), every third a STATUS and STATUS_REPLY (heads,
+ /// readiness, epoch) instead. The interval trades that against how soon a silent peer is
/// noticed (`dormant_after`, 4 intervals) and a gap closes (within an interval).
/// Recommended:
///
@@ -68,7 +73,10 @@ impl Timing {
) -> Self {
Self {
keepalive,
- dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row
+ // Every third keepalive; with `resync_after` at two keepalives, a heal needs
+ // two STATUSes that differ: within about two refreshes.
+ status_refresh: keepalive * 3,
+ dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost keepalives in a row
// >= 2 keepalives, so a second exchange confirms the difference; and >= twice
// the first sync reply timeout, so a lost message's sender resends it first.
resync_after: (keepalive * 2).max(sync_reply_timeout * 2),
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 08:03:32 +0000