aboutsummaryrefslogtreecommitdiffstats
path: root/src/transport/wire.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/wire.rs')
-rw-r--r--src/transport/wire.rs193
1 files changed, 171 insertions, 22 deletions
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index fcbee85..7717200 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -10,11 +10,11 @@
use thiserror::Error;
-use super::{MessageKind, ParsedMessage};
+use super::{EncodedMessage, MessageKind, ParsedMessage};
/// The wire format's version, the first byte of every payload. Bump it with every change
/// to a message type or a codec (`golden_bytes` fails then).
-pub(crate) const VERSION: u8 = 6;
+pub(crate) const VERSION: u8 = 14;
/// What the second byte of a payload says.
fn tag(kind: MessageKind) -> u8 {
@@ -29,6 +29,8 @@ fn tag(kind: MessageKind) -> u8 {
MessageKind::Me => 7,
MessageKind::Ping => 8,
MessageKind::Pong => 9,
+ MessageKind::Links => 10,
+ MessageKind::LinksReply => 11,
}
}
@@ -50,6 +52,21 @@ pub(crate) enum WireError {
Decode(#[from] rmp_serde::decode::Error),
}
+/// `message` to `peer`, encoded as it goes on the wire alone (no envelope: libp2p,
+/// Tor, the simulator); `peer` is its address too.
+#[cfg_attr(not(any(feature = "libp2p", feature = "tor", test)), allow(dead_code))]
+pub(crate) fn encoded(peer: &str, message: &ParsedMessage) -> anyhow::Result<EncodedMessage> {
+ let body = encode(message)?;
+ Ok(EncodedMessage {
+ peer: peer.to_owned(),
+ address: peer.to_owned(),
+ payload_len: body.len(),
+ body: body.into(),
+ kind: message.kind(),
+ message: None,
+ })
+}
+
/// The payload of `message`.
pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
let mut payload = vec![VERSION, tag(message.kind())];
@@ -64,9 +81,10 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
ParsedMessage::Who(proof) | ParsedMessage::Me(proof) => {
rmp_serde::encode::write(&mut payload, proof)?
}
- ParsedMessage::Ping(nonce) | ParsedMessage::Pong(nonce) => {
- rmp_serde::encode::write(&mut payload, nonce)?
- }
+ ParsedMessage::Ping(nonce) => rmp_serde::encode::write(&mut payload, nonce)?,
+ ParsedMessage::Pong { nonce, at } => rmp_serde::encode::write(&mut payload, &(nonce, at))?,
+ ParsedMessage::Links(request) => rmp_serde::encode::write(&mut payload, request)?,
+ ParsedMessage::LinksReply(reply) => rmp_serde::encode::write(&mut payload, reply)?,
}
Ok(payload)
}
@@ -98,7 +116,12 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
} else if is(MessageKind::Ping) {
ParsedMessage::Ping(rmp_serde::from_slice(body)?)
} else if is(MessageKind::Pong) {
- ParsedMessage::Pong(rmp_serde::from_slice(body)?)
+ let (nonce, at) = rmp_serde::from_slice(body)?;
+ ParsedMessage::Pong { nonce, at }
+ } else if is(MessageKind::Links) {
+ ParsedMessage::Links(rmp_serde::from_slice(body)?)
+ } else if is(MessageKind::LinksReply) {
+ ParsedMessage::LinksReply(rmp_serde::from_slice(body)?)
} else {
return Err(WireError::UnknownKind(kind));
})
@@ -180,7 +203,7 @@ pub(crate) mod codec {
/// An Automerge change hash.
pub(crate) mod change_hash {
use super::*;
- use automerge::ChangeHash;
+ use crate::document::ChangeHash;
pub(crate) fn serialize<S: Serializer>(
value: &ChangeHash,
@@ -219,8 +242,8 @@ pub(crate) mod codec {
/// empty), or as a digest (none listed).
pub(crate) mod heads {
use super::*;
+ use crate::document::ChangeHash;
use crate::transport::{HeadsSummary, MAX_LISTED_HEADS};
- use automerge::ChangeHash;
pub(crate) fn serialize<S: Serializer>(
value: &HeadsSummary,
@@ -267,7 +290,7 @@ pub(crate) mod codec {
/// Change hashes, each a 32-byte binary.
pub(crate) mod change_hashes {
use super::*;
- use automerge::ChangeHash;
+ use crate::document::ChangeHash;
pub(crate) fn serialize<S: Serializer>(
value: &[ChangeHash],
@@ -325,15 +348,134 @@ pub(crate) mod codec {
}
}
+/// One message of every kind, for the codecs' tests (this one's, and each transport's
+/// envelope around it).
+#[cfg(test)]
+pub(crate) fn samples() -> Vec<ParsedMessage> {
+ use crate::document::ChangeHash;
+ use crate::document::replica::Epoch;
+ use crate::transport::{
+ BlobBody, BlobId, BlobMessage, DeltaSpec, HeadsSummary, HelloReply, HelloRequest,
+ IdentityProof, ListedPeer, Status, SyncBody, SyncEnvelope,
+ };
+ let epoch = Epoch {
+ generation: 2,
+ id: [3; 32],
+ };
+ let status = Status {
+ heads: HeadsSummary::of(vec![ChangeHash([1; 32])]),
+ epoch,
+ epoch_seq: 1,
+ ready: true,
+ syncs_with_you: true,
+ leaving: false,
+ };
+ let proof = IdentityProof {
+ key: [8; 32],
+ signature: [9; 64],
+ };
+ vec![
+ ParsedMessage::Hello(HelloRequest {
+ swarm_id: Some("swarm".into()),
+ epoch,
+ epoch_seq: 1,
+ auth: vec![1, 2, 3],
+ ready: true,
+ run_id: [4; 16],
+ handshake: 7,
+ }),
+ ParsedMessage::HelloReply(HelloReply {
+ swarm_id: "swarm".into(),
+ peers: vec![ListedPeer {
+ id: "p".into(),
+ addresses: vec!["a".into()],
+ }],
+ epoch,
+ epoch_seq: 1,
+ auth: Vec::new(),
+ ready: false,
+ run_id: [4; 16],
+ away_ms: Some(5),
+ sender_run_id: [6; 16],
+ }),
+ ParsedMessage::Sync(SyncEnvelope {
+ swarm_id: "swarm".into(),
+ epoch,
+ epoch_seq: 1,
+ body: SyncBody::Delta(DeltaSpec {
+ from: Vec::new(),
+ to: vec![ChangeHash([2; 32])],
+ }),
+ ready: true,
+ }),
+ ParsedMessage::Status(status.clone()),
+ ParsedMessage::StatusReply(status),
+ ParsedMessage::Blob(BlobMessage {
+ epoch,
+ epoch_seq: 1,
+ id: BlobId::Content([5; 32]),
+ body: BlobBody::Missing,
+ }),
+ ParsedMessage::Who(proof.clone()),
+ ParsedMessage::Me(proof),
+ ParsedMessage::Ping(7),
+ ParsedMessage::Pong {
+ nonce: u64::MAX,
+ at: crate::clock::UnixMillis(5),
+ },
+ ParsedMessage::Links(crate::transport::LinksRequest {
+ epoch,
+ epoch_seq: 1,
+ nonce: 3,
+ subscribe_secs: 60,
+ }),
+ ParsedMessage::LinksReply(crate::transport::LinksReply {
+ epoch,
+ epoch_seq: 1,
+ nonce: 3,
+ links: vec![crate::transport::Link {
+ peer: "p".into(),
+ transport: Some("libp2p".into()),
+ relayed: true,
+ down: false,
+ rtt_us: Some(1500),
+ dormant: false,
+ others: vec![crate::transport::OtherPath {
+ transport: "mycelium".into(),
+ rtt_us: None,
+ relayed: false,
+ }],
+ clock: Some(crate::transport::paths::ClockOffset {
+ offset_ms: -2_500,
+ within_ms: 40,
+ }),
+ }],
+ }),
+ ]
+}
+
#[cfg(test)]
mod tests {
use super::*;
+ use crate::document::ChangeHash;
use crate::document::replica::Epoch;
use crate::transport::{
BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, IdentityProof,
ListedPeer, MAX_LISTED_HEADS, Status,
};
- use automerge::ChangeHash;
+
+ /// Every kind decodes to what was encoded: re-encoded, the same bytes.
+ #[test]
+ fn every_kind_round_trips() {
+ let samples = samples();
+ assert_eq!(samples.len(), 12, "a kind is missing from `samples`");
+ for message in samples {
+ let bytes = encode(&message).unwrap();
+ let decoded = decode(&bytes).unwrap();
+ assert_eq!(decoded.kind(), message.kind());
+ assert_eq!(encode(&decoded).unwrap(), bytes, "{message:?}");
+ }
+ }
#[test]
fn handshakes_round_trip() {
@@ -398,6 +540,7 @@ mod tests {
epoch_seq: 0,
ready: true,
syncs_with_you: false,
+ leaving: false,
};
let decoded =
decode(&encode(&ParsedMessage::StatusReply(status.clone())).unwrap()).unwrap();
@@ -420,6 +563,7 @@ mod tests {
epoch_seq: 0,
ready: true,
syncs_with_you: true,
+ leaving: false,
};
assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err());
}
@@ -454,13 +598,15 @@ mod tests {
}
fn hex(bytes: &[u8]) -> String {
- bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+ crate::transport::Hex(bytes).to_string()
}
/// The exact bytes of a few messages. If this fails after a change to a message type
/// or a codec, the change is a new wire format: bump `VERSION` and take the new bytes.
#[test]
fn golden_bytes() {
+ // The version byte follows `VERSION`: bumped for a change to other messages.
+ let v = format!("{VERSION:02x}");
let epoch = Epoch {
generation: 2,
id: [3; 32],
@@ -479,7 +625,7 @@ mod tests {
});
assert_eq!(
hex(&encode(&hello).unwrap()),
- format!("060097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+ format!("{v}0097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
);
let reply = ParsedMessage::HelloReply(HelloReply {
@@ -499,7 +645,7 @@ mod tests {
assert_eq!(
hex(&encode(&reply).unwrap()),
format!(
- "060199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
+ "{v}0199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
run("04"),
run("06")
)
@@ -511,10 +657,11 @@ mod tests {
epoch_seq: 1,
ready: true,
syncs_with_you: false,
+ leaving: false,
});
assert_eq!(
hex(&encode(&status).unwrap()),
- format!("0603959290c400{epoch_hex}01c3c2")
+ format!("{v}03969290c400{epoch_hex}01c3c2c2")
);
let who = ParsedMessage::Who(IdentityProof {
@@ -523,7 +670,7 @@ mod tests {
});
assert_eq!(
hex(&encode(&who).unwrap()),
- format!("060692c420{}c440{}", "08".repeat(32), "09".repeat(64))
+ format!("{v}0692c420{}c440{}", "08".repeat(32), "09".repeat(64))
);
let ParsedMessage::Me(proof) = decode(
&encode(&ParsedMessage::Me(IdentityProof {
@@ -537,14 +684,16 @@ mod tests {
};
assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64]));
- assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), "060807");
- assert_eq!(
- hex(&encode(&ParsedMessage::Pong(u64::MAX)).unwrap()),
- "0609cfffffffffffffffff"
- );
- let ParsedMessage::Pong(nonce) = decode(&[VERSION, 9, 0x2a]).unwrap() else {
+ assert_eq!(hex(&encode(&ParsedMessage::Ping(7)).unwrap()), format!("{v}0807"));
+ let pong = ParsedMessage::Pong {
+ nonce: u64::MAX,
+ at: crate::clock::UnixMillis(5),
+ };
+ assert_eq!(hex(&encode(&pong).unwrap()), format!("{v}0992cfffffffffffffffff05"));
+ let ParsedMessage::Pong { nonce, at } = decode(&[VERSION, 9, 0x92, 0x2a, 0x07]).unwrap()
+ else {
panic!("not a PONG");
};
- assert_eq!(nonce, 42);
+ assert_eq!((nonce, at.as_u64()), (42, 7));
}
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-11-13 02:12:22 +0000