summaryrefslogtreecommitdiffstats
path: root/src/transport/mycelium.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/mycelium.rs')
-rw-r--r--src/transport/mycelium.rs469
1 files changed, 53 insertions, 416 deletions
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index efab475..a7bd7d9 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -1,11 +1,8 @@
use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
-use automerge::ChangeHash;
use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
use reqwest::StatusCode;
-use serde::{Deserialize, Deserializer, Serialize, Serializer};
-use serde_bytes::{ByteBuf, Bytes};
-use std::borrow::Cow;
+use serde::{Deserialize, Serialize};
use std::boxed::Box;
use std::fs::File;
use std::io::Write;
@@ -18,21 +15,14 @@ use thiserror::Error;
use tokio::process::{Child, Command};
use url::Url;
-use crate::transport::EncodedMessage;
-use crate::transport::{BlobBody, BlobId, BlobMessage};
-use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope};
-use crate::transport::{HeadsSummary, Keepalive, MAX_LISTED_HEADS};
-use crate::transport::{MessageHandler, ParsedMessage, Transport};
+use crate::transport::wire;
+use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport};
const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
-const TOPIC_PREFIX: &str = "continuum.v1.";
-const TOPIC_HELLO: &str = "continuum.v1.hello";
-const TOPIC_HELLO_REPLY: &str = "continuum.v1.hello-reply";
-const TOPIC_SYNC: &str = "continuum.v1.sync";
-const TOPIC_PING: &str = "continuum.v1.ping";
-const TOPIC_PONG: &str = "continuum.v1.pong";
-const TOPIC_BLOB: &str = "continuum.v1.blob";
+/// The one Mycelium topic all of continuum's messages travel on; what a message is, its
+/// payload says (`wire`).
+const TOPIC: &str = "continuum.v1";
const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);
@@ -490,7 +480,7 @@ impl MyceliumTransport {
.client
.get(self.messages_endpoint.clone())
.query(&[("timeout", 0)])
- .query(&[("topic", encode_topic(TOPIC_PREFIX))])
+ .query(&[("topic", encode_topic(TOPIC))])
.send()
.await;
match response {
@@ -626,7 +616,7 @@ impl Transport for MyceliumTransport {
.poll_client
.get(self.messages_endpoint.clone())
.query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)])
- .query(&[("topic", encode_topic(TOPIC_PREFIX))])
+ .query(&[("topic", encode_topic(TOPIC))])
.send()
.await
{
@@ -746,26 +736,11 @@ fn base64_decoded_len(encoded: &str) -> usize {
}
fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
- let topic: &str = &decode_topic(topic_encoded)?;
- let bytes = BASE64.decode(payload)?;
- match topic {
- TOPIC_HELLO => {
- let msg = bytes_to_hello_request(&bytes)?;
- Ok(ParsedMessage::Hello(msg))
- }
- TOPIC_HELLO_REPLY => {
- let msg = bytes_to_hello_reply(&bytes)?;
- Ok(ParsedMessage::HelloReply(msg))
- }
- TOPIC_SYNC => {
- let msg = bytes_to_sync_envelope(&bytes)?;
- Ok(ParsedMessage::Sync(msg))
- }
- TOPIC_PING => Ok(ParsedMessage::Ping(bytes_to_keepalive(&bytes)?)),
- TOPIC_PONG => Ok(ParsedMessage::Pong(bytes_to_keepalive(&bytes)?)),
- TOPIC_BLOB => Ok(ParsedMessage::Blob(bytes_to_blob(&bytes)?)),
- t => Err(TransportError::UnknownTopic(t.to_string())),
+ let topic = decode_topic(topic_encoded)?;
+ if topic != TOPIC {
+ return Err(TransportError::UnknownTopic(topic));
}
+ Ok(wire::decode(&BASE64.decode(payload)?)?)
}
fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), TransportError> {
@@ -785,376 +760,17 @@ fn format_message(
recipient: &str,
message: &ParsedMessage,
) -> Result<(OutboundMessage, usize), TransportError> {
- let (topic, payload_bytes) = match message {
- ParsedMessage::Hello(req) => (TOPIC_HELLO, hello_request_to_bytes(req)?),
- ParsedMessage::HelloReply(res) => (TOPIC_HELLO_REPLY, hello_reply_to_bytes(res)?),
- ParsedMessage::Sync(env) => (TOPIC_SYNC, sync_envelope_to_bytes(env)?),
- ParsedMessage::Ping(k) => (TOPIC_PING, keepalive_to_bytes(k)?),
- ParsedMessage::Pong(k) => (TOPIC_PONG, keepalive_to_bytes(k)?),
- ParsedMessage::Blob(blob) => (TOPIC_BLOB, blob_to_bytes(blob)?),
- };
+ let payload_bytes = wire::encode(message)?;
let outbound = OutboundMessage {
dst: Destination {
pk: recipient.to_owned(),
},
- topic: encode_topic(topic),
+ topic: encode_topic(TOPIC),
payload: encode_payload(&payload_bytes),
};
Ok((outbound, payload_bytes.len()))
}
-// =====================================================================
-// 1. WIRE SHADOWS
-// =====================================================================
-//
-// rmp_serde encodes structs as arrays. Fields added later are appended at the end and
-// marked #[serde(default)], so a message from a node that predates them still decodes
-// (an absent epoch is the genesis epoch).
-
-/// Epoch on the wire. The default (generation 0, empty id) is the genesis epoch.
-#[derive(Serialize, Deserialize, Default)]
-struct EpochShadow {
- generation: u64,
- id: ByteBuf,
- /// The sender's epoch changes in its current run (`SyncEnvelope::epoch_seq`).
- #[serde(default)]
- seq: u32,
-}
-
-fn epoch_to_shadow(epoch: &Epoch, seq: u32) -> EpochShadow {
- EpochShadow {
- generation: epoch.generation,
- id: ByteBuf::from(epoch.id.to_vec()),
- seq,
- }
-}
-
-fn epoch_from_shadow(shadow: &EpochShadow) -> Result<Epoch, TransportError> {
- if shadow.generation == 0 && shadow.id.is_empty() {
- return Ok(Epoch::GENESIS);
- }
- let id = shadow
- .id
- .as_slice()
- .try_into()
- .map_err(|_| TransportError::InvalidPayload("epoch id must be 32 bytes".into()))?;
- Ok(Epoch {
- generation: shadow.generation,
- id,
- })
-}
-
-/// Either `heads` (listed) or `digest` (32 bytes, `heads` empty) is set.
-#[derive(Serialize, Deserialize)]
-struct KeepaliveShadow {
- heads: Vec<ByteBuf>,
- digest: ByteBuf,
- epoch: EpochShadow,
- ready: bool,
- syncs_with_you: bool,
-}
-
-fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> {
- let (heads, digest) = match &msg.heads {
- HeadsSummary::Heads(heads) => (
- heads
- .iter()
- .map(|head| ByteBuf::from(head.0.to_vec()))
- .collect(),
- ByteBuf::new(),
- ),
- HeadsSummary::Digest(digest) => (Vec::new(), ByteBuf::from(digest.to_vec())),
- };
- rmp_serde::to_vec(&KeepaliveShadow {
- heads,
- digest,
- epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
- ready: msg.ready,
- syncs_with_you: msg.syncs_with_you,
- })
-}
-
-fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
- let shadow: KeepaliveShadow = rmp_serde::from_slice(bytes)?;
- let heads = if shadow.digest.is_empty() {
- if shadow.heads.len() > MAX_LISTED_HEADS {
- return Err(TransportError::InvalidPayload(format!(
- "more than {MAX_LISTED_HEADS} heads listed"
- )));
- }
- let heads = shadow
- .heads
- .iter()
- .map(|head| ChangeHash::try_from(head.as_slice()))
- .collect::<Result<Vec<_>, _>>()
- .map_err(|_| TransportError::InvalidPayload("a head must be 32 bytes".into()))?;
- HeadsSummary::Heads(heads)
- } else {
- let digest =
- shadow.digest.as_slice().try_into().map_err(|_| {
- TransportError::InvalidPayload("heads digest must be 32 bytes".into())
- })?;
- HeadsSummary::Digest(digest)
- };
- Ok(Keepalive {
- heads,
- epoch: epoch_from_shadow(&shadow.epoch)?,
- epoch_seq: shadow.epoch.seq,
- ready: shadow.ready,
- syncs_with_you: shadow.syncs_with_you,
- })
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct HelloRequestShadow {
- swarm_id: Option<String>,
- epoch: EpochShadow,
- auth: ByteBuf,
- ready: bool,
- run_id: ByteBuf,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct HelloReplyShadow<'a> {
- // Cow allows zero-copy references when saving, but creates an owned String when loading
- pub swarm_id: Cow<'a, str>,
- #[serde(default)]
- pub peers: Vec<String>,
- #[serde(default)]
- epoch: EpochShadow,
- #[serde(default)]
- auth: ByteBuf,
- ready: bool,
- run_id: ByteBuf,
- away_ms: Option<u64>,
- sender_run_id: ByteBuf,
-}
-
-#[derive(Serialize, Deserialize)]
-pub struct SyncEnvelopeShadow<'a> {
- pub swarm_id: Cow<'a, str>,
- #[serde(with = "automerge_msg_codec")]
- pub message: automerge::sync::Message,
- #[serde(default)]
- epoch: EpochShadow,
- /// Absent from nodes that predate it: never counts as ready.
- #[serde(default)]
- ready: bool,
-}
-
-// =====================================================================
-// 2. CUSTOM CODEC
-// =====================================================================
-mod automerge_msg_codec {
- use super::*;
- use automerge::sync::Message;
-
- pub fn serialize<S>(msg: &Message, serializer: S) -> Result<S::Ok, S::Error>
- where
- S: Serializer,
- {
- let bytes = msg.clone().encode();
- Bytes::new(&bytes).serialize(serializer)
- }
-
- pub fn deserialize<'de, D>(deserializer: D) -> Result<Message, D::Error>
- where
- D: Deserializer<'de>,
- {
- let byte_buf = ByteBuf::deserialize(deserializer)?;
- Message::decode(&byte_buf)
- .map_err(|e| serde::de::Error::custom(format!("Automerge decode failed: {:?}", e)))
- }
-}
-
-// =====================================================================
-// 3. CLEAN CONVERSION INTERFACE
-// =====================================================================
-
-// --- HelloRequest ---
-pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
- rmp_serde::to_vec(&HelloRequestShadow {
- swarm_id: msg.swarm_id.clone(),
- epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
- auth: ByteBuf::from(msg.auth.clone()),
- ready: msg.ready,
- run_id: ByteBuf::from(msg.run_id.to_vec()),
- })
-}
-pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> {
- let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
- Ok(HelloRequest {
- swarm_id: shadow.swarm_id,
- epoch: epoch_from_shadow(&shadow.epoch)?,
- epoch_seq: shadow.epoch.seq,
- auth: shadow.auth.into_vec(),
- ready: shadow.ready,
- run_id: run_id_from_shadow(&shadow.run_id)?,
- })
-}
-
-fn run_id_from_shadow(bytes: &ByteBuf) -> Result<crate::transport::RunId, TransportError> {
- bytes
- .as_slice()
- .try_into()
- .map_err(|_| TransportError::InvalidPayload("run id must be 16 bytes".into()))
-}
-
-// --- HelloReply ---
-pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> {
- let shadow = HelloReplyShadow {
- swarm_id: Cow::Borrowed(&msg.swarm_id),
- peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
- epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
- auth: ByteBuf::from(msg.auth.clone()),
- ready: msg.ready,
- run_id: ByteBuf::from(msg.run_id.to_vec()),
- away_ms: msg.away_ms,
- sender_run_id: ByteBuf::from(msg.sender_run_id.to_vec()),
- };
- rmp_serde::to_vec(&shadow)
-}
-pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> {
- let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?;
- Ok(HelloReply {
- swarm_id: shadow.swarm_id.into_owned(),
- peers: shadow.peers,
- epoch: epoch_from_shadow(&shadow.epoch)?,
- epoch_seq: shadow.epoch.seq,
- auth: shadow.auth.into_vec(),
- ready: shadow.ready,
- run_id: run_id_from_shadow(&shadow.run_id)?,
- away_ms: shadow.away_ms,
- sender_run_id: run_id_from_shadow(&shadow.sender_run_id)?,
- })
-}
-
-// --- SyncEnvelope ---
-pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> {
- let shadow = SyncEnvelopeShadow {
- swarm_id: Cow::Borrowed(&msg.swarm_id),
- message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules
- epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
- ready: msg.ready,
- };
- rmp_serde::to_vec(&shadow)
-}
-pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> {
- let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
- Ok(SyncEnvelope {
- swarm_id: shadow.swarm_id.into_owned(),
- epoch: epoch_from_shadow(&shadow.epoch)?,
- epoch_seq: shadow.epoch.seq,
- message: shadow.message,
- ready: shadow.ready,
- })
-}
-
-// --- BlobMessage ---
-#[derive(Serialize, Deserialize)]
-enum BlobIdShadow {
- Change(ByteBuf),
- Content(ByteBuf),
-}
-
-#[derive(Serialize, Deserialize)]
-enum BlobBodyShadow {
- Request {
- chunks: Vec<u32>,
- },
- Manifest {
- size: u64,
- chunk_size: u32,
- hashes: Vec<ByteBuf>,
- },
- Chunk {
- index: u32,
- data: ByteBuf,
- },
- Missing,
-}
-
-#[derive(Serialize, Deserialize)]
-struct BlobShadow {
- epoch: EpochShadow,
- id: BlobIdShadow,
- body: BlobBodyShadow,
-}
-
-fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error> {
- let id = match msg.id {
- BlobId::Change(hash) => BlobIdShadow::Change(ByteBuf::from(hash.0.to_vec())),
- BlobId::Content(hash) => BlobIdShadow::Content(ByteBuf::from(hash.to_vec())),
- };
- let body = match &msg.body {
- BlobBody::Request { chunks } => BlobBodyShadow::Request {
- chunks: chunks.clone(),
- },
- BlobBody::Manifest {
- size,
- chunk_size,
- hashes,
- } => BlobBodyShadow::Manifest {
- size: *size,
- chunk_size: *chunk_size,
- hashes: hashes
- .iter()
- .map(|hash| ByteBuf::from(hash.to_vec()))
- .collect(),
- },
- BlobBody::Chunk { index, data } => BlobBodyShadow::Chunk {
- index: *index,
- data: ByteBuf::from(data.to_vec()),
- },
- BlobBody::Missing => BlobBodyShadow::Missing,
- };
- rmp_serde::to_vec(&BlobShadow {
- epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
- id,
- body,
- })
-}
-
-fn hash32(bytes: &[u8], what: &str) -> Result<[u8; 32], TransportError> {
- bytes
- .try_into()
- .map_err(|_| TransportError::InvalidPayload(format!("{what} must be 32 bytes")))
-}
-
-fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> {
- let shadow: BlobShadow = rmp_serde::from_slice(bytes)?;
- let id = match shadow.id {
- BlobIdShadow::Change(hash) => BlobId::Change(ChangeHash(hash32(&hash, "a change hash")?)),
- BlobIdShadow::Content(hash) => BlobId::Content(hash32(&hash, "a content hash")?),
- };
- let body = match shadow.body {
- BlobBodyShadow::Request { chunks } => BlobBody::Request { chunks },
- BlobBodyShadow::Manifest {
- size,
- chunk_size,
- hashes,
- } => BlobBody::Manifest {
- size,
- chunk_size,
- hashes: hashes
- .iter()
- .map(|hash| hash32(hash, "a chunk hash"))
- .collect::<Result<_, _>>()?,
- },
- BlobBodyShadow::Chunk { index, data } => BlobBody::Chunk {
- index,
- data: data.into_vec().into(),
- },
- BlobBodyShadow::Missing => BlobBody::Missing,
- };
- Ok(BlobMessage {
- epoch: epoch_from_shadow(&shadow.epoch)?,
- epoch_seq: shadow.epoch.seq,
- id,
- body,
- })
-}
-
#[derive(Error, Debug)]
pub enum TransportError {
#[error("Mycelium delivered a message without a sender public key")]
@@ -1166,18 +782,12 @@ pub enum TransportError {
#[error("UTF-8 decoding failed: {0}")]
Utf8Decode(#[from] std::string::FromUtf8Error),
- #[error("MessagePack serialization failed: {0}")]
- MessagePackEncode(#[from] rmp_serde::encode::Error),
-
- #[error("MessagePack deserialization failed: {0}")]
- MessagePackDecode(#[from] rmp_serde::decode::Error),
+ #[error(transparent)]
+ Wire(#[from] wire::WireError),
#[error("Unknown topic: {0}")]
UnknownTopic(String),
- #[error("Invalid payload: {0}")]
- InvalidPayload(String),
-
#[error("Network HTTP request failed: {0}")]
Http(#[from] reqwest::Error),
@@ -1191,6 +801,10 @@ pub enum TransportError {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::transport::{
+ BlobBody, BlobId, BlobMessage, Epoch, HeadsSummary, Keepalive, MAX_LISTED_HEADS,
+ };
+ use automerge::ChangeHash;
/// Mycelium splits messages into chunks of this size.
const MYCELIUM_CHUNK: usize = 1300;
@@ -1211,14 +825,17 @@ mod tests {
ready: true,
syncs_with_you: true,
};
- let payload = keepalive_to_bytes(&keepalive).unwrap();
+ let payload = wire::encode(&ParsedMessage::Pong(keepalive.clone())).unwrap();
// Mycelium's message data: a length byte, the topic, then our payload.
- let size = 1 + TOPIC_PONG.len().max(TOPIC_PING.len()) + payload.len();
+ let size = 1 + TOPIC.len() + payload.len();
assert!(
size + 50 <= MYCELIUM_CHUNK,
"keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes"
);
- assert_eq!(bytes_to_keepalive(&payload).unwrap().heads, keepalive.heads);
+ let ParsedMessage::Pong(decoded) = wire::decode(&payload).unwrap() else {
+ panic!("not a PONG");
+ };
+ assert_eq!(decoded.heads, keepalive.heads);
}
#[test]
@@ -1228,7 +845,11 @@ mod tests {
id: [7; 32],
};
let id = BlobId::Change(ChangeHash([9; 32]));
- let chunk = BlobMessage {
+ let decode_blob = |payload: &[u8]| match wire::decode(payload).unwrap() {
+ ParsedMessage::Blob(blob) => blob,
+ other => panic!("not a blob: {other:?}"),
+ };
+ let chunk = ParsedMessage::Blob(BlobMessage {
epoch,
epoch_seq: 4,
id,
@@ -1236,11 +857,11 @@ mod tests {
index: 5,
data: vec![1u8; 64 * 1024].into(),
},
- };
- let bytes = blob_to_bytes(&chunk).unwrap();
+ });
+ let bytes = wire::encode(&chunk).unwrap();
// A 64 KiB chunk costs only a few bytes of framing.
assert!(bytes.len() < 64 * 1024 + 128, "{} bytes", bytes.len());
- let decoded = bytes_to_blob(&bytes).unwrap();
+ let decoded = decode_blob(&bytes);
assert_eq!(decoded.id, id);
assert_eq!(decoded.epoch, epoch);
assert_eq!(decoded.epoch_seq, 4);
@@ -1248,7 +869,7 @@ mod tests {
matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024)
);
- let manifest = BlobMessage {
+ let manifest = ParsedMessage::Blob(BlobMessage {
epoch,
epoch_seq: 4,
id,
@@ -1257,8 +878,8 @@ mod tests {
chunk_size: 64,
hashes: vec![[1; 32], [2; 32]],
},
- };
- let decoded = bytes_to_blob(&blob_to_bytes(&manifest).unwrap()).unwrap();
+ });
+ let decoded = decode_blob(&wire::encode(&manifest).unwrap());
assert!(matches!(
decoded.body,
BlobBody::Manifest { size: 100, chunk_size: 64, ref hashes } if hashes.len() == 2
@@ -1266,6 +887,22 @@ mod tests {
}
#[test]
+ fn messages_round_trip_through_the_api_body() {
+ let keepalive = ParsedMessage::Ping(Keepalive {
+ 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();
+ assert!(matches!(
+ decode_message(&encoded.body).unwrap(),
+ ParsedMessage::Ping(_)
+ ));
+ }
+
+ #[test]
fn listed_heads_equal_their_digest() {
let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect();
let listed = HeadsSummary::of(heads.clone());
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 04:42:27 +0000