summaryrefslogtreecommitdiffstats
path: root/src/transport/wire.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/wire.rs')
-rw-r--r--src/transport/wire.rs550
1 files changed, 550 insertions, 0 deletions
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
new file mode 100644
index 0000000..fcbee85
--- /dev/null
+++ b/src/transport/wire.rs
@@ -0,0 +1,550 @@
+//! The wire format of a message: one byte for the format's version (`VERSION`), one
+//! saying what the message is (`MessageKind`), then the message itself in MessagePack, as derived from its type in `transport` (structs as
+//! arrays, fields in declaration order). The codecs below make hashes, IDs and bytes
+//! compact (MessagePack binaries, not arrays of numbers) and check their lengths.
+//!
+//! The wire format may change freely until the first release: there is no compatibility
+//! with older nodes to keep. But every change bumps `VERSION`, so nodes on different
+//! builds tell each other apart instead of misreading each other's messages. The
+//! transport's topic says which application a message is for, not which version.
+
+use thiserror::Error;
+
+use super::{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;
+
+/// What the second byte of a payload says.
+fn tag(kind: MessageKind) -> u8 {
+ match kind {
+ MessageKind::Hello => 0,
+ MessageKind::HelloReply => 1,
+ MessageKind::Sync => 2,
+ MessageKind::Status => 3,
+ MessageKind::StatusReply => 4,
+ MessageKind::Blob => 5,
+ MessageKind::Who => 6,
+ MessageKind::Me => 7,
+ MessageKind::Ping => 8,
+ MessageKind::Pong => 9,
+ }
+}
+
+#[derive(Error, Debug)]
+pub(crate) enum WireError {
+ #[error("empty payload")]
+ Empty,
+
+ #[error("wire format version {0} (we speak version {VERSION})")]
+ Version(u8),
+
+ #[error("unknown message kind {0}")]
+ UnknownKind(u8),
+
+ #[error("MessagePack serialization failed: {0}")]
+ Encode(#[from] rmp_serde::encode::Error),
+
+ #[error("MessagePack deserialization failed: {0}")]
+ Decode(#[from] rmp_serde::decode::Error),
+}
+
+/// The payload of `message`.
+pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> {
+ let mut payload = vec![VERSION, tag(message.kind())];
+ match message {
+ 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::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) => {
+ rmp_serde::encode::write(&mut payload, proof)?
+ }
+ ParsedMessage::Ping(nonce) | ParsedMessage::Pong(nonce) => {
+ rmp_serde::encode::write(&mut payload, nonce)?
+ }
+ }
+ Ok(payload)
+}
+
+/// The message in `payload` (as made by [`encode`]).
+pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> {
+ let (&version, payload) = payload.split_first().ok_or(WireError::Empty)?;
+ if version != VERSION {
+ return Err(WireError::Version(version));
+ }
+ let (&kind, body) = payload.split_first().ok_or(WireError::Empty)?;
+ let is = |wanted: MessageKind| kind == tag(wanted);
+ Ok(if is(MessageKind::Hello) {
+ ParsedMessage::Hello(rmp_serde::from_slice(body)?)
+ } else if is(MessageKind::HelloReply) {
+ ParsedMessage::HelloReply(rmp_serde::from_slice(body)?)
+ } else if is(MessageKind::Sync) {
+ ParsedMessage::Sync(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) {
+ ParsedMessage::Who(rmp_serde::from_slice(body)?)
+ } else if is(MessageKind::Me) {
+ ParsedMessage::Me(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 {
+ return Err(WireError::UnknownKind(kind));
+ })
+}
+
+/// Field codecs (`#[serde(with = "...")]`) for the message types.
+pub(crate) mod codec {
+ use serde::de::Error as _;
+ use serde::{Deserialize, Deserializer, Serialize, Serializer};
+ use serde_bytes::{ByteBuf, Bytes};
+
+ /// Exactly `N` bytes, or a decoding error naming `what`.
+ fn fixed<const N: usize, E: serde::de::Error>(bytes: &[u8], what: &str) -> Result<[u8; N], E> {
+ bytes
+ .try_into()
+ .map_err(|_| E::custom(format!("{what} must be {N} bytes")))
+ }
+
+ /// A run ID (`RunId`).
+ pub(crate) mod run_id {
+ use super::*;
+
+ pub(crate) fn serialize<S: Serializer>(value: &[u8; 16], s: S) -> Result<S::Ok, S::Error> {
+ Bytes::new(value).serialize(s)
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 16], D::Error> {
+ fixed(&ByteBuf::deserialize(d)?, "a run id")
+ }
+ }
+
+ /// A 32-byte hash.
+ pub(crate) mod bytes32 {
+ use super::*;
+
+ pub(crate) fn serialize<S: Serializer>(value: &[u8; 32], s: S) -> Result<S::Ok, S::Error> {
+ Bytes::new(value).serialize(s)
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 32], D::Error> {
+ fixed(&ByteBuf::deserialize(d)?, "a hash")
+ }
+ }
+
+ /// A 64-byte signature.
+ pub(crate) mod bytes64 {
+ use super::*;
+
+ pub(crate) fn serialize<S: Serializer>(value: &[u8; 64], s: S) -> Result<S::Ok, S::Error> {
+ Bytes::new(value).serialize(s)
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 64], D::Error> {
+ fixed(&ByteBuf::deserialize(d)?, "a signature")
+ }
+ }
+
+ /// 32-byte hashes.
+ pub(crate) mod hashes32 {
+ use super::*;
+
+ pub(crate) fn serialize<S: Serializer>(
+ value: &[[u8; 32]],
+ s: S,
+ ) -> Result<S::Ok, S::Error> {
+ s.collect_seq(value.iter().map(|hash| Bytes::new(hash)))
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+ d: D,
+ ) -> Result<Vec<[u8; 32]>, D::Error> {
+ Vec::<ByteBuf>::deserialize(d)?
+ .iter()
+ .map(|hash| fixed(hash, "a hash"))
+ .collect()
+ }
+ }
+
+ /// An Automerge change hash.
+ pub(crate) mod change_hash {
+ use super::*;
+ use automerge::ChangeHash;
+
+ pub(crate) fn serialize<S: Serializer>(
+ value: &ChangeHash,
+ s: S,
+ ) -> Result<S::Ok, S::Error> {
+ Bytes::new(&value.0).serialize(s)
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<ChangeHash, D::Error> {
+ Ok(ChangeHash(fixed(
+ &ByteBuf::deserialize(d)?,
+ "a change hash",
+ )?))
+ }
+ }
+
+ /// An epoch: generation and ID.
+ pub(crate) mod epoch {
+ use super::*;
+ use crate::document::replica::Epoch;
+
+ pub(crate) fn serialize<S: Serializer>(value: &Epoch, s: S) -> Result<S::Ok, S::Error> {
+ (value.generation, Bytes::new(&value.id)).serialize(s)
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Epoch, D::Error> {
+ let (generation, id): (u64, ByteBuf) = Deserialize::deserialize(d)?;
+ Ok(Epoch {
+ generation,
+ id: fixed(&id, "an epoch id")?,
+ })
+ }
+ }
+
+ /// Heads as a keepalive carries them: listed (at most `MAX_LISTED_HEADS`, digest
+ /// empty), or as a digest (none listed).
+ pub(crate) mod heads {
+ use super::*;
+ use crate::transport::{HeadsSummary, MAX_LISTED_HEADS};
+ use automerge::ChangeHash;
+
+ pub(crate) fn serialize<S: Serializer>(
+ value: &HeadsSummary,
+ s: S,
+ ) -> Result<S::Ok, S::Error> {
+ let (listed, digest): (Vec<&Bytes>, &Bytes) = match value {
+ HeadsSummary::Heads(heads) => (
+ heads.iter().map(|head| Bytes::new(&head.0)).collect(),
+ Bytes::new(&[]),
+ ),
+ HeadsSummary::Digest { digest, sample } => (
+ sample.iter().map(|head| Bytes::new(&head.0)).collect(),
+ Bytes::new(digest),
+ ),
+ };
+ (listed, digest).serialize(s)
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+ d: D,
+ ) -> Result<HeadsSummary, D::Error> {
+ let (listed, digest): (Vec<ByteBuf>, ByteBuf) = Deserialize::deserialize(d)?;
+ if listed.len() > MAX_LISTED_HEADS {
+ return Err(D::Error::custom(format!(
+ "more than {MAX_LISTED_HEADS} heads listed"
+ )));
+ }
+ let listed = listed
+ .iter()
+ .map(|head| -> Result<ChangeHash, D::Error> {
+ Ok(ChangeHash(fixed(head, "a head")?))
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+ if digest.is_empty() {
+ return Ok(HeadsSummary::Heads(listed));
+ }
+ Ok(HeadsSummary::Digest {
+ digest: fixed(&digest, "a heads digest")?,
+ sample: listed,
+ })
+ }
+ }
+
+ /// Change hashes, each a 32-byte binary.
+ pub(crate) mod change_hashes {
+ use super::*;
+ use automerge::ChangeHash;
+
+ pub(crate) fn serialize<S: Serializer>(
+ value: &[ChangeHash],
+ s: S,
+ ) -> Result<S::Ok, S::Error> {
+ s.collect_seq(value.iter().map(|hash| Bytes::new(&hash.0)))
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+ d: D,
+ ) -> Result<Vec<ChangeHash>, D::Error> {
+ Vec::<ByteBuf>::deserialize(d)?
+ .iter()
+ .map(|hash| Ok(ChangeHash(fixed(hash, "a change hash")?)))
+ .collect()
+ }
+ }
+
+ /// An Automerge sync message, in Automerge's own encoding (left encoded:
+ /// `SyncPayload`).
+ pub(crate) mod sync_message {
+ use super::*;
+ use crate::transport::SyncPayload;
+
+ pub(crate) fn serialize<S: Serializer>(
+ value: &SyncPayload,
+ s: S,
+ ) -> Result<S::Ok, S::Error> {
+ Bytes::new(&value.0).serialize(s)
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+ d: D,
+ ) -> Result<SyncPayload, D::Error> {
+ Ok(SyncPayload(ByteBuf::deserialize(d)?.into_vec().into()))
+ }
+ }
+
+ /// Shared bytes (`bytes::Bytes`).
+ pub(crate) mod shared_bytes {
+ use super::*;
+
+ pub(crate) fn serialize<S: Serializer>(
+ value: &bytes::Bytes,
+ s: S,
+ ) -> Result<S::Ok, S::Error> {
+ Bytes::new(value).serialize(s)
+ }
+
+ pub(crate) fn deserialize<'de, D: Deserializer<'de>>(
+ d: D,
+ ) -> Result<bytes::Bytes, D::Error> {
+ Ok(ByteBuf::deserialize(d)?.into_vec().into())
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::document::replica::Epoch;
+ use crate::transport::{
+ BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, IdentityProof,
+ ListedPeer, MAX_LISTED_HEADS, Status,
+ };
+ use automerge::ChangeHash;
+
+ #[test]
+ fn handshakes_round_trip() {
+ let epoch = Epoch {
+ generation: 2,
+ id: [3; 32],
+ };
+ let hello = 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,
+ });
+ let ParsedMessage::Hello(decoded) = decode(&encode(&hello).unwrap()).unwrap() else {
+ panic!("not a HELLO");
+ };
+ assert_eq!(decoded.swarm_id.as_deref(), Some("swarm"));
+ assert_eq!((decoded.epoch, decoded.epoch_seq), (epoch, 1));
+ assert_eq!(decoded.auth, vec![1, 2, 3]);
+ assert_eq!(decoded.run_id, [4; 16]);
+ assert_eq!(decoded.handshake, 7);
+
+ let reply = 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],
+ });
+ let ParsedMessage::HelloReply(decoded) = decode(&encode(&reply).unwrap()).unwrap() else {
+ panic!("not a HELLO_REPLY");
+ };
+ assert_eq!(
+ decoded.peers,
+ vec![ListedPeer {
+ id: "p".into(),
+ addresses: vec!["a".into()],
+ }]
+ );
+ assert_eq!(decoded.away_ms, Some(5));
+ assert_eq!(decoded.sender_run_id, [6; 16]);
+ }
+
+ #[test]
+ fn keepalives_keep_their_kind_and_heads() {
+ let status = Status {
+ heads: HeadsSummary::Digest {
+ digest: [8; 32],
+ sample: vec![ChangeHash([9; 32])],
+ },
+ epoch: Epoch::GENESIS,
+ epoch_seq: 0,
+ ready: true,
+ syncs_with_you: false,
+ };
+ 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, status.heads);
+ assert_eq!(decoded.heads.listed(), [ChangeHash([9; 32])]);
+ }
+
+ #[test]
+ fn too_many_listed_heads_are_refused() {
+ let status = Status {
+ heads: HeadsSummary::Heads(
+ (0..=MAX_LISTED_HEADS as u8)
+ .map(|n| ChangeHash([n; 32]))
+ .collect(),
+ ),
+ epoch: Epoch::GENESIS,
+ epoch_seq: 0,
+ ready: true,
+ syncs_with_you: true,
+ };
+ assert!(decode(&encode(&ParsedMessage::Status(status)).unwrap()).is_err());
+ }
+
+ #[test]
+ fn unknown_kinds_and_bad_payloads_are_refused() {
+ assert!(matches!(decode(&[]), Err(WireError::Empty)));
+ assert!(matches!(decode(&[VERSION]), Err(WireError::Empty)));
+ assert!(matches!(
+ decode(&[VERSION, 99]),
+ Err(WireError::UnknownKind(99))
+ ));
+ assert!(matches!(
+ decode(&[VERSION + 1, 0]),
+ Err(WireError::Version(version)) if version == VERSION + 1
+ ));
+
+ let blob = ParsedMessage::Blob(BlobMessage {
+ epoch: Epoch::GENESIS,
+ epoch_seq: 0,
+ id: BlobId::Content([1; 32]),
+ body: BlobBody::Missing,
+ });
+ let payload = encode(&blob).unwrap();
+ assert!(matches!(decode(&payload).unwrap(), ParsedMessage::Blob(_)));
+ // A payload cut short doesn't decode.
+ assert!(decode(&payload[..payload.len() - 1]).is_err());
+ // Nor one of another kind.
+ let mut relabelled = payload.clone();
+ relabelled[1] = tag(MessageKind::Hello);
+ assert!(decode(&relabelled).is_err());
+ }
+
+ fn hex(bytes: &[u8]) -> String {
+ bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+ }
+
+ /// 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() {
+ let epoch = Epoch {
+ generation: 2,
+ id: [3; 32],
+ };
+ let epoch_hex = format!("9202c420{}", "03".repeat(32));
+ let run = |byte: &str| format!("c410{}", byte.repeat(16));
+
+ let hello = 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,
+ });
+ assert_eq!(
+ hex(&encode(&hello).unwrap()),
+ format!("060097a5737761726d{epoch_hex}01c403010203c3{}07", run("04"))
+ );
+
+ let reply = 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],
+ });
+ assert_eq!(
+ hex(&encode(&reply).unwrap()),
+ format!(
+ "060199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}",
+ run("04"),
+ run("06")
+ )
+ );
+
+ let status = ParsedMessage::Status(Status {
+ heads: HeadsSummary::Heads(Vec::new()),
+ epoch,
+ epoch_seq: 1,
+ ready: true,
+ syncs_with_you: false,
+ });
+ assert_eq!(
+ hex(&encode(&status).unwrap()),
+ format!("0603959290c400{epoch_hex}01c3c2")
+ );
+
+ let who = ParsedMessage::Who(IdentityProof {
+ key: [8; 32],
+ signature: [9; 64],
+ });
+ assert_eq!(
+ hex(&encode(&who).unwrap()),
+ format!("060692c420{}c440{}", "08".repeat(32), "09".repeat(64))
+ );
+ let ParsedMessage::Me(proof) = decode(
+ &encode(&ParsedMessage::Me(IdentityProof {
+ key: [8; 32],
+ signature: [9; 64],
+ }))
+ .unwrap(),
+ )
+ .unwrap() else {
+ panic!("not a ME");
+ };
+ 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 {
+ panic!("not a PONG");
+ };
+ assert_eq!(nonce, 42);
+ }
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 08:04:11 +0000