summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--docs/openapi.yaml11
-rw-r--r--src/api/stats/prometheus.rs4
-rw-r--r--src/simtest/identity.rs10
-rw-r--r--src/simtest/timing.rs16
-rw-r--r--src/state/host.rs4
-rw-r--r--src/state/mod.rs2
-rw-r--r--src/state/peer.rs24
-rw-r--r--src/state/timing.rs8
-rw-r--r--src/swarm/pair.rs2
-rw-r--r--src/swarm/peer/inbound.rs34
-rw-r--r--src/swarm/peer/keepalive.rs36
-rw-r--r--src/swarm/peer/mod.rs4
-rw-r--r--src/swarm/peer/worker.rs6
-rw-r--r--src/swarm/relay.rs12
-rw-r--r--src/swarm/rtt.rs42
-rw-r--r--src/swarm/source.rs4
-rw-r--r--src/swarm/stats.rs2
-rw-r--r--src/transport/mod.rs20
-rw-r--r--src/transport/mycelium/mod.rs22
-rw-r--r--src/transport/p2p.rs18
-rw-r--r--src/transport/sim.rs22
-rw-r--r--src/transport/traffic.rs16
-rw-r--r--src/transport/wire.rs37
23 files changed, 181 insertions, 175 deletions
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index 05f47f7..be21300 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -1190,7 +1190,7 @@ components:
ByKind:
type: object
- required: [hello, helloReply, sync, ping, pong, blob]
+ required: [hello, helloReply, sync, status, statusReply, blob, identity]
properties:
hello:
$ref: '#/components/schemas/Counter'
@@ -1198,17 +1198,20 @@ components:
$ref: '#/components/schemas/Counter'
sync:
$ref: '#/components/schemas/Counter'
- ping:
+ status:
$ref: '#/components/schemas/Counter'
- pong:
+ statusReply:
$ref: '#/components/schemas/Counter'
blob:
$ref: '#/components/schemas/Counter'
description: Blob requests and answers (chunked downloads, e.g. of an epoch's snapshot).
+ identity:
+ $ref: '#/components/schemas/Counter'
+ description: WHO and ME, counted under the address they went to or came from (so only in the totals).
Traffic:
type: object
- description: Since start. Every PONG answers a PING, so received pongs over sent pings is the keepalive answer ratio.
+ description: Since start. Every STATUS_REPLY answers a STATUS, so received status replies over sent statuses is the keepalive answer ratio.
required: [sent, received, refused]
properties:
sent:
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 8c6e3e6..11645a3 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -173,8 +173,8 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou
(direction, "hello", by.hello),
(direction, "hello_reply", by.hello_reply),
(direction, "sync", by.sync),
- (direction, "ping", by.ping),
- (direction, "pong", by.pong),
+ (direction, "status", by.status),
+ (direction, "status_reply", by.status_reply),
(direction, "blob", by.blob),
(direction, "identity", by.identity),
]
diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs
index ec5d665..b789bda 100644
--- a/src/simtest/identity.rs
+++ b/src/simtest/identity.rs
@@ -15,7 +15,7 @@ use crate::swarm::members;
use crate::transport::identity::{WHO_LABEL, signed};
use crate::transport::sim::SimNet;
use crate::transport::{
- Destination, HeadsSummary, IdentityProof, Keepalive, ParsedMessage, Transport,
+ Destination, HeadsSummary, IdentityProof, ParsedMessage, Status, Transport,
};
const SOON: Duration = Duration::from_secs(20);
@@ -97,8 +97,8 @@ async fn members_find_each_other_through_the_member_list() {
}
}
-fn ping(node: &AppState) -> ParsedMessage {
- ParsedMessage::Ping(Keepalive {
+fn status(node: &AppState) -> ParsedMessage {
+ ParsedMessage::Status(Status {
heads: HeadsSummary::of(Vec::new()),
epoch: node.local_epoch(),
epoch_seq: 0,
@@ -129,7 +129,7 @@ async fn a_proof_holds_only_at_its_address() {
who(&ghost, "sim:ghost", "sim:b"),
] {
sim.net.inject("ghost", "a", &forged).await.unwrap();
- sim.net.inject("ghost", "a", &ping(&a)).await.unwrap();
+ sim.net.inject("ghost", "a", &status(&a)).await.unwrap();
assert!(a.get_peer_state(&ghost.id()).await.is_none());
}
let counts = a.transport.stats().identity.unwrap().counts;
@@ -140,7 +140,7 @@ async fn a_proof_holds_only_at_its_address() {
.inject("ghost", "a", &who(&ghost, "sim:ghost", "sim:a"))
.await
.unwrap();
- sim.net.inject("ghost", "a", &ping(&a)).await.unwrap();
+ sim.net.inject("ghost", "a", &status(&a)).await.unwrap();
assert!(a.get_peer_state(&ghost.id()).await.is_some());
// Known by its node ID, not by its address.
assert!(a.get_peer_state("ghost").await.is_none());
diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index ce1dfd5..20ad15c 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -10,7 +10,7 @@ use crate::document::replica::Epoch;
use crate::state::{AppState, Liveness};
use crate::swarm::bootstrap::Start;
use crate::transport::sim::LinkParams;
-use crate::transport::{HeadsSummary, HelloRequest, Keepalive, MessageKind, ParsedMessage, RunId};
+use crate::transport::{HeadsSummary, HelloRequest, MessageKind, ParsedMessage, RunId, Status};
/// Long enough for anything that should happen within a few keepalives.
const SOON: Duration = Duration::from_secs(10);
@@ -311,8 +311,8 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
handshake: 1,
})
};
- let ping = |epoch: Epoch, epoch_seq: u32| {
- ParsedMessage::Ping(Keepalive {
+ let status = |epoch: Epoch, epoch_seq: u32| {
+ ParsedMessage::Status(Status {
heads: HeadsSummary::of(Vec::new()),
epoch,
epoch_seq,
@@ -336,20 +336,20 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
// The ghost switched twice; we hear of the second switch first.
sim.net
- .inject("ghost", "a", &ping(epoch(2), 2))
+ .inject("ghost", "a", &status(epoch(2), 2))
.await
.unwrap();
assert_eq!(recorded().await, (Some(epoch(2)), Some(2)));
- // The PING from between the switches arrives late: ignored.
+ // The STATUS from between the switches arrives late: ignored.
sim.net
- .inject("ghost", "a", &ping(epoch(1), 1))
+ .inject("ghost", "a", &status(epoch(1), 1))
.await
.unwrap();
assert_eq!(recorded().await, (Some(epoch(2)), Some(2)));
// So is one from before both.
sim.net
- .inject("ghost", "a", &ping(Epoch::GENESIS, 0))
+ .inject("ghost", "a", &status(Epoch::GENESIS, 0))
.await
.unwrap();
assert_eq!(recorded().await, (Some(epoch(2)), Some(2)));
@@ -366,7 +366,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() {
Some(second_run)
);
sim.net
- .inject("ghost", "a", &ping(epoch(3), 1))
+ .inject("ghost", "a", &status(epoch(3), 1))
.await
.unwrap();
assert_eq!(recorded().await, (Some(epoch(3)), Some(1)));
diff --git a/src/state/host.rs b/src/state/host.rs
index f6c460d..441139e 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");
}
}
diff --git a/src/state/mod.rs b/src/state/mod.rs
index c8243a2..20f92c5 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -167,7 +167,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..c9356ea 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 keepalive STATUSes. Fixed for
/// the pair, so computed once.
pub(crate) leads: bool,
@@ -85,7 +85,7 @@ 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>,
+ 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 +133,7 @@ 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 STATUS_REPLY answers a STATUS.
#[derive(Debug, Default, Clone, Copy, serde::Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Incidents {
@@ -172,7 +172,7 @@ impl PeerState {
liveness: Liveness::Active,
added_at: Instant::now(),
last_seen: None,
- last_ping_at: None,
+ last_status_at: None,
last_sync_rx: None,
last_resync_at: None,
diverged_since: None,
@@ -287,15 +287,15 @@ impl PeerState {
true
}
- /// True (and records the ping) if an established, quiet peer needs a keepalive.
+ /// True (and records the time) if an established, quiet peer needs a keepalive.
///
- /// 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
+ /// One STATUS and its STATUS_REPLY tell both sides that the other is alive, so only one side of
+ /// a pair needs to send them: the one leading the pair (`leads`). Each of its STATUSes
/// 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
+ /// sends while it does. If they stop, the other side sends 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.
+ /// STATUSes would often cross.
pub(crate) fn take_keepalive_due(&mut self, timing: &Timing, now: Instant) -> bool {
let quiet_enough = if self.leads {
timing.keepalive
@@ -307,16 +307,16 @@ impl PeerState {
&& now.saturating_duration_since(self.added_at) >= timing.keepalive
&& self.quiet_for(now) >= quiet_enough
&& self
- .last_ping_at
+ .last_status_at
.is_none_or(|t| now.saturating_duration_since(t) >= timing.keepalive);
if due {
- self.last_ping_at = Some(now);
+ self.last_status_at = Some(now);
}
due
}
- /// Self-healing check on PING/PONG. Returns true if a re-handshake should start.
+ /// Self-healing check on STATUS/STATUS_REPLY. Returns true if a re-handshake should start.
pub(crate) fn needs_heal(
&mut self,
local: &transport::HeadsSummary,
diff --git a/src/state/timing.rs b/src/state/timing.rs
index 0dab3da..6ce5bef 100644
--- a/src/state/timing.rs
+++ b/src/state/timing.rs
@@ -21,7 +21,7 @@ pub(crate) struct Timing {
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,8 +42,8 @@ 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
+ /// and only one side of a pair sends them, so idle, a node sends about `k` messages and
+ /// receives about `k` (STATUSes and STATUS_REPLYs) per keepalive interval (more on a stable
/// member others prefer). 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 +68,7 @@ impl Timing {
) -> Self {
Self {
keepalive,
- dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost PINGs in a row
+ dormant_after: keepalive * 4, // dormant_after >= 4* keepalive to tolerate three lost STATUSes 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),
diff --git a/src/swarm/pair.rs b/src/swarm/pair.rs
index c1a196b..41fc219 100644
--- a/src/swarm/pair.rs
+++ b/src/swarm/pair.rs
@@ -1,5 +1,5 @@
//! Which side of a pair of nodes leads, for duties only one side needs to do (sending
-//! keepalive PINGs).
+//! keepalive STATUSes).
//!
//! Both sides decide on their own, from the two keys alone, and agree. The leader is
//! chosen per pair by one bit of a hash of both keys, not by comparing the keys: a
diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index b04aad9..4c7e7a8 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_status};
use super::lanes::SyncLanes;
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
@@ -143,20 +143,22 @@ 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) => {
+ 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) => {
+ // 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(_) | transport::ParsedMessage::Me(_) => {}
}
@@ -164,19 +166,19 @@ pub(crate) async fn handle_message(
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..810c377 100644
--- a/src/swarm/peer/keepalive.rs
+++ b/src/swarm/peer/keepalive.rs
@@ -1,4 +1,4 @@
-//! Keepalives (PING / PONG) and the self-healing re-handshake when they show our
+//! Keepalives (STATUS / STATUS_REPLY) and the self-healing re-handshake when they show our
//! documents differ while sync is idle.
use std::sync::atomic::Ordering;
@@ -14,37 +14,37 @@ 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.
-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());
+/// Sends `peer` a STATUS with our status, noting the time for its round trip.
+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());
state
.transport
- .send_message(peer, &ParsedMessage::Ping(keepalive))
+ .send_message(peer, &ParsedMessage::Status(status))
.await
}
-/// How long shutting down waits for the last PINGs (`farewell`).
+/// 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
+/// 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);
- let mut pings = tokio::task::JoinSet::new();
+ 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 { send_status(&state, &peer).await });
}
}
- let all_sent = async { while pings.join_next().await.is_some() {} };
+ let all_sent = async { while statuses.join_next().await.is_some() {} };
let _ = tokio::time::timeout(FAREWELL_TIMEOUT, all_sent).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 +64,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
}
None => false,
};
- transport::Keepalive {
+ transport::Status {
heads,
epoch,
epoch_seq,
@@ -73,8 +73,8 @@ 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) {
+/// 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 +103,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..415cddf 100644
--- a/src/swarm/peer/mod.rs
+++ b/src/swarm/peer/mod.rs
@@ -2,7 +2,7 @@
//! 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`: 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 +24,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::{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..c68e2ea 100644
--- a/src/swarm/peer/worker.rs
+++ b/src/swarm/peer/worker.rs
@@ -10,7 +10,7 @@ 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_status;
use crate::locks::MutexExt;
use crate::state::AppState;
use crate::swarm::sync::SyncEvent;
@@ -83,9 +83,9 @@ 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
+ && let Err(error) = send_status(&state, peer).await
{
- debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive PING failed");
+ debug!(peer = %peer, error = %format!("{error:#}"), "Keepalive STATUS failed");
}
if let Err(error) = retry_hello_if_due(&state, peer).await {
diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs
index 202ae40..6172f67 100644
--- a/src/swarm/relay.rs
+++ b/src/swarm/relay.rs
@@ -2,7 +2,7 @@
//!
//! A node pushes its own changes to its neighbours, but nothing it receives goes any
//! further by itself. So after applying changes from one neighbour, a node *announces*
-//! its new heads to its other neighbours with a PING (a keepalive: heads, about 100
+//! its new heads to its other neighbours with a STATUS (a keepalive: heads, about 100
//! bytes, not the changes). A neighbour that sees heads it lacks *pulls*: it enters the
//! announced heads as the peer's heads in its sync state, Automerge asks for what's
//! missing (`need`), and the announcer answers with exactly those changes. Each node
@@ -11,7 +11,7 @@
//!
//! Announcements are coalesced: the first applied change schedules one announcement
//! round `ANNOUNCE_DELAY` later, and every change applied until then rides along. A
-//! round costs one PING and one PONG per neighbour (plus the pulls), whatever the
+//! round costs one STATUS and one STATUS_REPLY per neighbour (plus the pulls), whatever the
//! write rate; the delay is the price per hop. Without coalescing, every applied sync
//! message would announce to every neighbour, and announcements would outnumber the
//! data they announce (about 15 to 1 at 250 nodes).
@@ -27,7 +27,7 @@ use tracing::debug;
use crate::locks::MutexExt;
use crate::state::{AppState, Standing};
-use crate::swarm::peer::send_ping;
+use crate::swarm::peer::send_status;
use crate::swarm::sync::SyncEvent;
use crate::swarm::topology;
use crate::transport::{self, HeadsSummary};
@@ -97,7 +97,7 @@ async fn announce_now(state: &AppState, senders: &HashSet<String>) {
.cloned()
.collect();
for peer in neighbours {
- if let Err(error) = send_ping(state, &peer).await {
+ if let Err(error) = send_status(state, &peer).await {
debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed");
}
}
@@ -111,7 +111,7 @@ pub(crate) struct Pulling {
/// The neighbour we are pulling from, and since when.
current: Option<(String, Instant)>,
/// The latest keepalive of each neighbour waiting to be pulled from.
- waiting: HashMap<String, transport::Keepalive>,
+ waiting: HashMap<String, transport::Status>,
}
/// A keepalive from `peer`: if it has heads we lack, pull them (one peer at a time; a
@@ -119,7 +119,7 @@ pub(crate) struct Pulling {
pub(crate) async fn pull_if_behind(
state: &AppState,
peer: &str,
- remote: &transport::Keepalive,
+ remote: &transport::Status,
) -> bool {
if remote.epoch != state.local_epoch() || !topology::is_neighbour(state, peer) {
return false;
diff --git a/src/swarm/rtt.rs b/src/swarm/rtt.rs
index ed6cc73..01d10ee 100644
--- a/src/swarm/rtt.rs
+++ b/src/swarm/rtt.rs
@@ -1,4 +1,4 @@
-//! Round-trip times to peers, measured from traffic we send anyway: PING → PONG and
+//! Round-trip times to peers, measured from traffic we send anyway: STATUS → STATUS_REPLY and
//! HELLO → HELLO_REPLY. No probes of our own.
//!
//! A sample only counts when exactly one request was outstanding: with two, we can't tell
@@ -18,7 +18,7 @@ use tokio::time::Instant;
#[derive(Debug, Default)]
pub(crate) struct RttTable {
- pings: HashMap<String, Outstanding>,
+ statuses: HashMap<String, Outstanding>,
hellos: HashMap<String, Outstanding>,
smoothed: HashMap<String, Sample>,
}
@@ -36,12 +36,12 @@ struct Outstanding {
}
impl RttTable {
- pub(crate) fn ping_sent(&mut self, peer: &str, now: Instant) {
- Self::sent(&mut self.pings, peer, now);
+ pub(crate) fn status_sent(&mut self, peer: &str, now: Instant) {
+ Self::sent(&mut self.statuses, peer, now);
}
- pub(crate) fn pong_received(&mut self, peer: &str, now: Instant) {
- if let Some(sample) = Self::answered(&mut self.pings, peer, now) {
+ 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);
}
}
@@ -68,7 +68,7 @@ impl RttTable {
.smoothed
.get(peer)
.filter(|sample| now.saturating_duration_since(sample.at) < fresh_for)?;
- let waiting = [self.pings.get(peer), self.hellos.get(peer)]
+ let waiting = [self.statuses.get(peer), self.hellos.get(peer)]
.into_iter()
.flatten()
.map(|outstanding| now.saturating_duration_since(outstanding.sent))
@@ -128,31 +128,31 @@ mod tests {
let mut table = RttTable::default();
let start = Instant::now();
let at = |ms| start + Duration::from_millis(ms);
- table.ping_sent("a", start);
- table.pong_received("a", at(100));
+ table.status_sent("a", start);
+ table.status_reply_received("a", at(100));
assert_eq!(
table.effective("a", at(100), FRESH),
Some(Duration::from_millis(100))
);
- table.ping_sent("a", at(100));
- table.pong_received("a", at(300));
+ table.status_sent("a", at(100));
+ table.status_reply_received("a", at(300));
assert_eq!(
table.effective("a", at(300), FRESH),
Some(Duration::from_millis(125))
);
- // Two PINGs out: the PONG can't be attributed.
- table.ping_sent("a", at(300));
- table.ping_sent("a", at(400));
- table.pong_received("a", at(450));
+ // 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));
assert_eq!(
table.effective("a", at(450), FRESH),
Some(Duration::from_millis(125))
);
- // An unsolicited PONG changes nothing.
- table.pong_received("b", start);
+ // An unsolicited STATUS_REPLY changes nothing.
+ table.status_reply_received("b", start);
assert_eq!(table.effective("b", start, FRESH), None);
}
@@ -160,12 +160,12 @@ mod tests {
fn unanswered_requests_raise_it_and_old_samples_expire() {
let mut table = RttTable::default();
let start = Instant::now();
- table.ping_sent("a", start);
- table.pong_received("a", start + Duration::from_millis(50));
+ table.status_sent("a", start);
+ table.status_reply_received("a", start + Duration::from_millis(50));
- // A PING unanswered for 40 s: at least 40 s.
+ // A STATUS unanswered for 40 s: at least 40 s.
let sent = start + Duration::from_secs(60);
- table.ping_sent("a", sent);
+ table.status_sent("a", sent);
let later = sent + Duration::from_secs(40);
assert_eq!(
table.effective("a", later, FRESH),
diff --git a/src/swarm/source.rs b/src/swarm/source.rs
index 699b03b..d80e4d7 100644
--- a/src/swarm/source.rs
+++ b/src/swarm/source.rs
@@ -12,7 +12,7 @@
//! A source must be ready and willing to sync with us, as far as we know (its sync
//! messages and keepalives say so). It counts as alive while any message from it
//! arrives, so a slow transfer of a large document never looks like a stall. When it
-//! is quiet for `Timing::catch_up_probe`, we PING it; after three unanswered probes, or
+//! is quiet for `Timing::catch_up_probe`, we send it a STATUS; after three unanswered probes, or
//! when it says it isn't ready, or twice in a row that it wouldn't sync with us (the
//! first time may be a handshake still under way), the next peer takes over.
//!
@@ -268,7 +268,7 @@ pub(crate) fn reported(state: &AppState, peer: &str, ready: bool, syncs_with_us:
}
}
-/// Whether to PING `peer` now: it is the source and has been quiet for a probe interval.
+/// Whether to send `peer` a STATUS now: it is the source and has been quiet for a probe interval.
pub(crate) fn probe_due(state: &AppState, peer: &str, now: Instant) -> bool {
if !catching_up(state) {
return false;
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index 308c2a5..b5548d5 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -96,7 +96,7 @@ pub(crate) struct SavedSize {
/// Passing changes on (`swarm::relay`), since start.
#[derive(Debug, Clone, Copy, Default)]
pub(crate) struct Relay {
- /// Announcement rounds (each a PING to the neighbours).
+ /// Announcement rounds (each a STATUS to the neighbours).
pub(crate) announce_rounds: u64,
/// Pulls started.
pub(crate) pulls: u64,
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 10a8ae5..b0d0723 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -116,9 +116,9 @@ fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] {
hasher.finalize().into()
}
-/// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient.
+/// STATUS and STATUS_REPLY. Addressed to one peer: `syncs_with_you` is about the recipient.
#[derive(Debug, Clone, Serialize, Deserialize)]
-pub(crate) struct Keepalive {
+pub(crate) struct Status {
/// Only comparable between nodes on the same epoch.
#[serde(with = "crate::transport::wire::codec::heads")]
pub(crate) heads: HeadsSummary,
@@ -138,8 +138,8 @@ pub(crate) enum ParsedMessage {
Hello(HelloRequest),
HelloReply(HelloReply),
Sync(SyncEnvelope),
- Ping(Keepalive),
- Pong(Keepalive),
+ Status(Status),
+ StatusReply(Status),
Blob(BlobMessage),
/// Asks the recipient who it is, saying who the sender is (`transport::identity`).
Who(IdentityProof),
@@ -263,8 +263,8 @@ pub(crate) enum MessageKind {
Hello,
HelloReply,
Sync,
- Ping,
- Pong,
+ Status,
+ StatusReply,
Blob,
Who,
Me,
@@ -276,8 +276,8 @@ impl ParsedMessage {
ParsedMessage::Hello(_) => MessageKind::Hello,
ParsedMessage::HelloReply(_) => MessageKind::HelloReply,
ParsedMessage::Sync(_) => MessageKind::Sync,
- ParsedMessage::Ping(_) => MessageKind::Ping,
- ParsedMessage::Pong(_) => MessageKind::Pong,
+ ParsedMessage::Status(_) => MessageKind::Status,
+ ParsedMessage::StatusReply(_) => MessageKind::StatusReply,
ParsedMessage::Blob(_) => MessageKind::Blob,
ParsedMessage::Who(_) => MessageKind::Who,
ParsedMessage::Me(_) => MessageKind::Me,
@@ -291,8 +291,8 @@ impl ParsedMessage {
ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)),
ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)),
ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)),
- ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => {
- Some((keepalive.epoch, keepalive.epoch_seq))
+ ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => {
+ Some((status.epoch, status.epoch_seq))
}
ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)),
ParsedMessage::Who(_) | ParsedMessage::Me(_) => None,
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 354aebc..cf979e2 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -28,7 +28,7 @@ const INBOX_POLLS: usize = 2;
/// (`post_timeout`).
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
/// The slowest our Mycelium is expected to take a message body, in bytes per second:
-/// a large sync message gets more time than a PING, instead of failing (and being
+/// a large sync message gets more time than a STATUS, instead of failing (and being
/// generated and sent whole again) at a fixed limit.
const POST_MIN_RATE: u64 = 1 << 20;
@@ -552,7 +552,7 @@ pub(crate) enum TransportError {
mod tests {
use super::*;
use crate::transport::{
- BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, Keepalive, MAX_LISTED_HEADS,
+ BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, MAX_LISTED_HEADS, Status,
};
use automerge::ChangeHash;
@@ -571,7 +571,7 @@ mod tests {
#[test]
fn largest_keepalive_fits_in_one_chunk() {
// A digest with a full sample: the most a keepalive carries.
- let keepalive = Keepalive {
+ let status = Status {
heads: HeadsSummary::of(
(0..=MAX_LISTED_HEADS as u8)
.map(|n| ChangeHash([n; 32]))
@@ -585,17 +585,17 @@ mod tests {
ready: true,
syncs_with_you: true,
};
- let payload = wire::encode(&ParsedMessage::Pong(keepalive.clone())).unwrap();
+ let payload = wire::encode(&ParsedMessage::StatusReply(status.clone())).unwrap();
// Mycelium's message data: a length byte, the topic, then our payload.
let size = 1 + TOPIC.len() + payload.len();
assert!(
size + 50 <= MYCELIUM_CHUNK,
- "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes"
+ "STATUS with {MAX_LISTED_HEADS} heads takes {size} bytes"
);
- let ParsedMessage::Pong(decoded) = wire::decode(&payload).unwrap() else {
- panic!("not a PONG");
+ let ParsedMessage::StatusReply(decoded) = wire::decode(&payload).unwrap() else {
+ panic!("not a STATUS_REPLY");
};
- assert_eq!(decoded.heads, keepalive.heads);
+ assert_eq!(decoded.heads, status.heads);
}
#[test]
@@ -648,17 +648,17 @@ mod tests {
#[test]
fn messages_round_trip_through_the_api_body() {
- let keepalive = ParsedMessage::Ping(Keepalive {
+ let status = ParsedMessage::Status(Status {
heads: HeadsSummary::of(vec![ChangeHash([1; 32])]),
epoch: Epoch::GENESIS,
epoch_seq: 0,
ready: true,
syncs_with_you: true,
});
- let encoded = encode_message("peer", &keepalive).unwrap();
+ let encoded = encode_message("peer", &status).unwrap();
assert!(matches!(
decode_message(&encoded.body).unwrap(),
- ParsedMessage::Ping(_)
+ ParsedMessage::Status(_)
));
}
diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs
index fcf961e..7a92407 100644
--- a/src/transport/p2p.rs
+++ b/src/transport/p2p.rs
@@ -1438,7 +1438,7 @@ mod tests {
use super::*;
use crate::document::replica::Epoch;
- use crate::transport::{HeadsSummary, Keepalive};
+ use crate::transport::{HeadsSummary, Status};
/// Memory ports are process-wide.
static NEXT_PORT: AtomicU64 = AtomicU64::new(40_000);
@@ -1481,8 +1481,8 @@ mod tests {
memory_node_with(true, Vec::new(), false)
}
- fn ping() -> ParsedMessage {
- ParsedMessage::Ping(Keepalive {
+ fn status() -> ParsedMessage {
+ ParsedMessage::Status(Status {
heads: HeadsSummary::of(Vec::new()),
epoch: Epoch::GENESIS,
epoch_seq: 3,
@@ -1522,13 +1522,13 @@ mod tests {
b.resolve_seed(&a_address.to_string()).await.unwrap(),
a.local_id()
);
- b.send_message(a.local_id(), &ping()).await.unwrap();
+ b.send_message(a.local_id(), &status()).await.unwrap();
let (from, message) = at_a.recv().await.unwrap();
assert_eq!(from, b.local_id());
- assert!(matches!(message, ParsedMessage::Ping(keepalive) if keepalive.epoch_seq == 3));
+ assert!(matches!(message, ParsedMessage::Status(status) if status.epoch_seq == 3));
// a knows no address of b, but b's connection is there.
- a.send_message(b.local_id(), &ping()).await.unwrap();
+ a.send_message(b.local_id(), &status()).await.unwrap();
let (from, _) = at_b.recv().await.unwrap();
assert_eq!(from, a.local_id());
@@ -1588,7 +1588,7 @@ mod tests {
let (a, b) = (Arc::new(a), Arc::new(b));
let _at_a = received(Arc::clone(&a));
b.resolve_seed(&a_address.to_string()).await.unwrap();
- b.send_message(a.local_id(), &ping()).await.unwrap();
+ b.send_message(a.local_id(), &status()).await.unwrap();
let b_id = b.local_id().to_owned();
assert!(
eventually(|| a.locations(&b_id) == [b_address.to_string()]).await,
@@ -1665,7 +1665,7 @@ mod tests {
let (a, _) = memory_node();
assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
- a.send_message(b.local_id(), &ping()).await.unwrap();
+ a.send_message(b.local_id(), &status()).await.unwrap();
let (from, _) = at_b.recv().await.unwrap();
assert_eq!(from, a.local_id());
assert!(a.connection(b.local_id()).is_some());
@@ -1808,7 +1808,7 @@ mod tests {
.public()
.to_peer_id()
.to_string();
- assert!(b.send_message(&stranger, &ping()).await.is_err());
+ assert!(b.send_message(&stranger, &status()).await.is_err());
assert!(b.resolve_seed(&stranger).await.is_err());
assert_eq!(b.connection(&stranger), None);
}
diff --git a/src/transport/sim.rs b/src/transport/sim.rs
index 149798b..500050b 100644
--- a/src/transport/sim.rs
+++ b/src/transport/sim.rs
@@ -319,10 +319,10 @@ impl Transport for SimTransport {
mod tests {
use super::*;
use crate::document::replica::Epoch;
- use crate::transport::{HeadsSummary, Keepalive};
+ use crate::transport::{HeadsSummary, Status};
- fn ping() -> ParsedMessage {
- ParsedMessage::Ping(Keepalive {
+ fn status() -> ParsedMessage {
+ ParsedMessage::Status(Status {
heads: HeadsSummary::of(Vec::new()),
epoch: Epoch::GENESIS,
epoch_seq: 0,
@@ -353,13 +353,13 @@ mod tests {
let received = record(net.transport("b"));
let settle = || tokio::time::sleep(Duration::from_millis(30));
- a.send_message("b", &ping()).await.unwrap();
+ a.send_message("b", &status()).await.unwrap();
settle().await;
net.take_down("b");
- a.send_message("b", &ping()).await.unwrap();
+ a.send_message("b", &status()).await.unwrap();
settle().await;
net.bring_up("b");
- a.send_message("b", &ping()).await.unwrap();
+ a.send_message("b", &status()).await.unwrap();
settle().await;
assert_eq!(*received.lock().unwrap(), vec!["a", "a"]);
@@ -381,7 +381,7 @@ mod tests {
let a = net.transport("a");
let _b = net.transport("b");
for _ in 0..200 {
- a.send_message("b", &ping()).await.unwrap();
+ a.send_message("b", &status()).await.unwrap();
}
tokio::time::sleep(Duration::from_millis(50)).await;
net.counts()
@@ -408,8 +408,8 @@ mod tests {
let at_a = record(Arc::clone(&a));
let at_b = record(Arc::clone(&b));
- a.send_message("b", &ping()).await.unwrap();
- b.send_message("a", &ping()).await.unwrap();
+ a.send_message("b", &status()).await.unwrap();
+ b.send_message("a", &status()).await.unwrap();
tokio::time::sleep(Duration::from_millis(30)).await;
assert!(at_b.lock().unwrap().is_empty());
@@ -420,8 +420,8 @@ mod tests {
async fn inject_returns_once_handled() {
let net = SimNet::new(1);
let received = record(net.transport("b"));
- net.inject("ghost", "b", &ping()).await.unwrap();
+ net.inject("ghost", "b", &status()).await.unwrap();
assert_eq!(*received.lock().unwrap(), vec!["ghost"]);
- assert!(net.inject("ghost", "nobody", &ping()).await.is_err());
+ assert!(net.inject("ghost", "nobody", &status()).await.is_err());
}
}
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index e912302..2134521 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -37,8 +37,8 @@ pub(crate) struct ByKind {
pub(crate) hello: Counter,
pub(crate) hello_reply: Counter,
pub(crate) sync: Counter,
- pub(crate) ping: Counter,
- pub(crate) pong: Counter,
+ pub(crate) status: Counter,
+ pub(crate) status_reply: Counter,
pub(crate) blob: Counter,
/// WHO and ME (`transport::identity`), counted by the identity layer under the
/// address they went to or came from.
@@ -51,8 +51,8 @@ impl ByKind {
MessageKind::Hello => &mut self.hello,
MessageKind::HelloReply => &mut self.hello_reply,
MessageKind::Sync => &mut self.sync,
- MessageKind::Ping => &mut self.ping,
- MessageKind::Pong => &mut self.pong,
+ MessageKind::Status => &mut self.status,
+ MessageKind::StatusReply => &mut self.status_reply,
MessageKind::Blob => &mut self.blob,
MessageKind::Who | MessageKind::Me => &mut self.identity,
}
@@ -227,20 +227,20 @@ mod tests {
#[test]
fn totals_survive_retain() {
let traffic = Traffic::default();
- traffic.sent("a", MessageKind::Ping, 100);
+ traffic.sent("a", MessageKind::Status, 100);
traffic.sent("b", MessageKind::Sync, 1000);
- traffic.received("b", MessageKind::Pong, 90);
+ traffic.received("b", MessageKind::StatusReply, 90);
traffic.refused("a", MessageKind::Sync, 500);
let b = traffic.peer("b").unwrap();
assert_eq!(b.sent.sync.messages, 1);
- assert_eq!(b.received.pong.bytes, 90);
+ assert_eq!(b.received.status_reply.bytes, 90);
traffic.retain(|peer| peer == "b");
assert!(traffic.peer("a").is_none());
let total = traffic.total();
- assert_eq!(total.sent.ping.bytes, 100);
+ assert_eq!(total.sent.status.bytes, 100);
assert_eq!(total.sent.sync.bytes, 1000);
assert_eq!(total.refused.sync.bytes, 500);
}
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index 5577f42..62aa3ca 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -22,8 +22,8 @@ fn tag(kind: MessageKind) -> u8 {
MessageKind::Hello => 0,
MessageKind::HelloReply => 1,
MessageKind::Sync => 2,
- MessageKind::Ping => 3,
- MessageKind::Pong => 4,
+ MessageKind::Status => 3,
+ MessageKind::StatusReply => 4,
MessageKind::Blob => 5,
MessageKind::Who => 6,
MessageKind::Me => 7,
@@ -55,8 +55,8 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
ParsedMessage::Hello(hello) => rmp_serde::encode::write(&mut payload, hello)?,
ParsedMessage::HelloReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
ParsedMessage::Sync(envelope) => rmp_serde::encode::write(&mut payload, envelope)?,
- ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => {
- rmp_serde::encode::write(&mut payload, keepalive)?
+ ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => {
+ rmp_serde::encode::write(&mut payload, status)?
}
ParsedMessage::Blob(blob) => rmp_serde::encode::write(&mut payload, blob)?,
ParsedMessage::Who(proof) | ParsedMessage::Me(proof) => {
@@ -80,10 +80,10 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
ParsedMessage::HelloReply(rmp_serde::from_slice(body)?)
} else if is(MessageKind::Sync) {
ParsedMessage::Sync(rmp_serde::from_slice(body)?)
- } else if is(MessageKind::Ping) {
- ParsedMessage::Ping(rmp_serde::from_slice(body)?)
- } else if is(MessageKind::Pong) {
- ParsedMessage::Pong(rmp_serde::from_slice(body)?)
+ } else if is(MessageKind::Status) {
+ ParsedMessage::Status(rmp_serde::from_slice(body)?)
+ } else if is(MessageKind::StatusReply) {
+ ParsedMessage::StatusReply(rmp_serde::from_slice(body)?)
} else if is(MessageKind::Blob) {
ParsedMessage::Blob(rmp_serde::from_slice(body)?)
} else if is(MessageKind::Who) {
@@ -322,7 +322,7 @@ mod tests {
use crate::document::replica::Epoch;
use crate::transport::{
BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, IdentityProof,
- Keepalive, ListedPeer, MAX_LISTED_HEADS,
+ ListedPeer, MAX_LISTED_HEADS, Status,
};
use automerge::ChangeHash;
@@ -380,7 +380,7 @@ mod tests {
#[test]
fn keepalives_keep_their_kind_and_heads() {
- let keepalive = Keepalive {
+ let status = Status {
heads: HeadsSummary::Digest {
digest: [8; 32],
sample: vec![ChangeHash([9; 32])],
@@ -390,17 +390,18 @@ mod tests {
ready: true,
syncs_with_you: false,
};
- let decoded = decode(&encode(&ParsedMessage::Pong(keepalive.clone())).unwrap()).unwrap();
- let ParsedMessage::Pong(decoded) = decoded else {
- panic!("not a PONG");
+ let decoded =
+ decode(&encode(&ParsedMessage::StatusReply(status.clone())).unwrap()).unwrap();
+ let ParsedMessage::StatusReply(decoded) = decoded else {
+ panic!("not a STATUS_REPLY");
};
- assert_eq!(decoded.heads, keepalive.heads);
+ assert_eq!(decoded.heads, status.heads);
assert_eq!(decoded.heads.listed(), [ChangeHash([9; 32])]);
}
#[test]
fn too_many_listed_heads_are_refused() {
- let keepalive = Keepalive {
+ let status = Status {
heads: HeadsSummary::Heads(
(0..=MAX_LISTED_HEADS as u8)
.map(|n| ChangeHash([n; 32]))
@@ -411,7 +412,7 @@ mod tests {
ready: true,
syncs_with_you: true,
};
- assert!(decode(&encode(&ParsedMessage::Ping(keepalive)).unwrap()).is_err());
+ assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err());
}
#[test]
@@ -495,7 +496,7 @@ mod tests {
)
);
- let ping = ParsedMessage::Ping(Keepalive {
+ let status = ParsedMessage::Status(Status {
heads: HeadsSummary::Heads(Vec::new()),
epoch,
epoch_seq: 1,
@@ -503,7 +504,7 @@ mod tests {
syncs_with_you: false,
});
assert_eq!(
- hex(&encode(&ping).unwrap()),
+ hex(&encode(&status).unwrap()),
format!("0503959290c400{epoch_hex}01c3c2")
);
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 18:46:09 +0000