summaryrefslogtreecommitdiffstats
path: root/src/transport
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport')
-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
6 files changed, 68 insertions, 67 deletions
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 10:03:32 +0000