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.rs348
1 files changed, 329 insertions, 19 deletions
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 84a34ea..82e6016 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -1,5 +1,6 @@
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};
@@ -18,8 +19,9 @@ use tokio::process::{Child, Command};
use url::Url;
use crate::transport::EncodedMessage;
-use crate::transport::Keepalive;
+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};
const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
@@ -30,6 +32,7 @@ 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";
const EMPTY_QUEUE_DELAY: Duration = Duration::from_millis(25);
@@ -455,14 +458,60 @@ impl MyceliumTransport {
.await
.context("Querying local Mycelium identity")?;
- Ok(Self {
+ let transport = Self {
setup: tokio::sync::Mutex::new(setup),
client,
poll_client,
messages_endpoint: rest_url.join("/api/v1/messages")?,
pubkey_basepath: rest_url.join("/api/v1/pubkey/")?,
local_id: admin.node_pubkey,
- })
+ };
+ transport.discard_backlog().await;
+ Ok(transport)
+ }
+
+ /// Drops what Mycelium queued for us before we started. A Mycelium that outlives us
+ /// (an external one, or across our restart) keeps accepting messages meant for us:
+ /// HELLO retries, keepalives, sync messages of sessions that ended with our previous
+ /// run. Processing them would only replay stale handshakes. Everything that still
+ /// matters is resent: peers retry their HELLOs, and sync starts over after the
+ /// handshake. Runs before we send anything, so no answer to us is dropped.
+ ///
+ /// Bounded in time, not count: the backlog is whatever piled up while we were away,
+ /// which a local loop drains in well under a second even for large swarms. Whatever
+ /// is still arriving after that is current traffic (peers that don't know we're
+ /// back), and it is processed normally, exactly as without this step.
+ async fn discard_backlog(&self) {
+ const DRAIN_FOR: Duration = Duration::from_secs(2);
+ let deadline = std::time::Instant::now() + DRAIN_FOR;
+ let mut discarded = 0u64;
+ while std::time::Instant::now() < deadline {
+ let response = self
+ .client
+ .get(self.messages_endpoint.clone())
+ .query(&[("timeout", 0)])
+ .query(&[("topic", encode_topic(TOPIC_PREFIX))])
+ .send()
+ .await;
+ match response {
+ Ok(response) if response.status() == StatusCode::NO_CONTENT => break,
+ Ok(response) if response.status().is_success() => discarded += 1,
+ Ok(response) => {
+ tracing::warn!(status = %response.status(), "Couldn't drain the Mycelium inbox");
+ break;
+ }
+ Err(error) => {
+ tracing::warn!(error = %error, "Couldn't drain the Mycelium inbox");
+ break;
+ }
+ }
+ }
+ if discarded > 0 {
+ tracing::info!(
+ discarded,
+ "Discarded messages Mycelium queued for us before we started"
+ );
+ }
}
async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> {
@@ -512,10 +561,12 @@ impl Transport for MyceliumTransport {
}
fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
- let outbound = format_message(peer, message)?;
+ let (outbound, payload_len) = format_message(peer, message)?;
Ok(EncodedMessage {
peer: peer.to_owned(),
body: serde_json::to_vec(&outbound)?.into(),
+ kind: message.kind(),
+ payload_len,
})
}
@@ -638,7 +689,8 @@ impl Transport for MyceliumTransport {
//payload_bytes = raw_bytes.len(),
"Received message"
);
- if let Err(error) = handler(sender, parsed).await {
+ let payload_len = base64_decoded_len(&inbound_message.payload);
+ if let Err(error) = handler(sender, parsed, payload_len).await {
tracing::warn!(error = %format!("{error:#}"), "Failed to process inbound message");
}
}
@@ -673,6 +725,12 @@ fn encode_payload(payload: &Vec<u8>) -> String {
BASE64.encode(payload)
}
+/// Length of what padded base64 `encoded` decodes to.
+fn base64_decoded_len(encoded: &str) -> usize {
+ let padding = encoded.bytes().rev().take_while(|&b| b == b'=').count();
+ (encoded.len() / 4 * 3).saturating_sub(padding)
+}
+
fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, TransportError> {
let topic: &str = &decode_topic(topic_encoded)?;
let bytes = BASE64.decode(payload)?;
@@ -691,6 +749,7 @@ fn decode_payload(topic_encoded: &str, payload: &str) -> Result<ParsedMessage, T
}
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())),
}
}
@@ -707,24 +766,27 @@ fn parse_message(message: &InboundMessage) -> Result<(String, ParsedMessage), Tr
Ok((sender, parsed))
}
+/// The message for Mycelium's API, and the length of its payload.
fn format_message(
recipient: &str,
message: &ParsedMessage,
-) -> Result<OutboundMessage, TransportError> {
+) -> 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)?),
};
- Ok(OutboundMessage {
+ let outbound = OutboundMessage {
dst: Destination {
pk: recipient.to_owned(),
},
topic: encode_topic(topic),
payload: encode_payload(&payload_bytes),
- })
+ };
+ Ok((outbound, payload_bytes.len()))
}
// =====================================================================
@@ -764,39 +826,73 @@ fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> {
})
}
+/// Either `heads` (listed) or `digest` (32 bytes, `heads` empty) is set.
#[derive(Serialize, Deserialize)]
struct KeepaliveShadow {
- heads_digest: ByteBuf,
- #[serde(default)]
+ 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: ByteBuf::from(msg.heads_digest.to_vec()),
+ heads,
+ digest,
epoch: epoch_to_shadow(&msg.epoch),
+ 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_digest = shadow
- .heads_digest
- .as_slice()
- .try_into()
- .map_err(|_| TransportError::InvalidPayload("heads digest must be 32 bytes".into()))?;
+ 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_digest,
+ heads,
epoch: epoch_from_shadow(shadow.epoch)?,
+ ready: shadow.ready,
+ syncs_with_you: shadow.syncs_with_you,
})
}
#[derive(Serialize, Deserialize)]
pub struct HelloRequestShadow {
- #[serde(default)]
+ swarm_id: Option<String>,
epoch: EpochShadow,
- #[serde(default)]
auth: ByteBuf,
+ ready: bool,
+ run_id: ByteBuf,
}
#[derive(Serialize, Deserialize)]
@@ -809,6 +905,9 @@ pub struct HelloReplyShadow<'a> {
epoch: EpochShadow,
#[serde(default)]
auth: ByteBuf,
+ ready: bool,
+ run_id: ByteBuf,
+ away_ms: Option<u64>,
}
#[derive(Serialize, Deserialize)]
@@ -818,6 +917,9 @@ pub struct SyncEnvelopeShadow<'a> {
pub message: automerge::sync::Message,
#[serde(default)]
epoch: EpochShadow,
+ /// Absent from nodes that predate it: never counts as ready.
+ #[serde(default)]
+ ready: bool,
}
// =====================================================================
@@ -852,18 +954,31 @@ mod automerge_msg_codec {
// --- 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),
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)?,
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 {
@@ -871,6 +986,9 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco
peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
epoch: epoch_to_shadow(&msg.epoch),
auth: ByteBuf::from(msg.auth.clone()),
+ ready: msg.ready,
+ run_id: ByteBuf::from(msg.run_id.to_vec()),
+ away_ms: msg.away_ms,
};
rmp_serde::to_vec(&shadow)
}
@@ -881,6 +999,9 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError>
peers: shadow.peers,
epoch: epoch_from_shadow(shadow.epoch)?,
auth: shadow.auth.into_vec(),
+ ready: shadow.ready,
+ run_id: run_id_from_shadow(&shadow.run_id)?,
+ away_ms: shadow.away_ms,
})
}
@@ -890,6 +1011,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::
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),
+ ready: msg.ready,
};
rmp_serde::to_vec(&shadow)
}
@@ -899,6 +1021,111 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
swarm_id: shadow.swarm_id.into_owned(),
epoch: epoch_from_shadow(shadow.epoch)?,
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),
+ 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)?,
+ id,
+ body,
})
}
@@ -934,3 +1161,86 @@ pub enum TransportError {
status: reqwest::StatusCode,
},
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ /// Mycelium splits messages into chunks of this size.
+ const MYCELIUM_CHUNK: usize = 1300;
+
+ #[test]
+ fn largest_keepalive_fits_in_one_chunk() {
+ let keepalive = Keepalive {
+ heads: HeadsSummary::Heads(
+ (0..MAX_LISTED_HEADS as u8)
+ .map(|n| ChangeHash([n; 32]))
+ .collect(),
+ ),
+ epoch: Epoch {
+ generation: u64::MAX,
+ id: [0xff; 32],
+ },
+ ready: true,
+ syncs_with_you: true,
+ };
+ let payload = keepalive_to_bytes(&keepalive).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();
+ assert!(
+ size + 50 <= MYCELIUM_CHUNK,
+ "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes"
+ );
+ assert_eq!(bytes_to_keepalive(&payload).unwrap().heads, keepalive.heads);
+ }
+
+ #[test]
+ fn blob_messages_round_trip() {
+ let epoch = Epoch {
+ generation: 3,
+ id: [7; 32],
+ };
+ let id = BlobId::Change(ChangeHash([9; 32]));
+ let chunk = BlobMessage {
+ epoch,
+ id,
+ body: BlobBody::Chunk {
+ index: 5,
+ data: vec![1u8; 64 * 1024].into(),
+ },
+ };
+ let bytes = blob_to_bytes(&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();
+ assert_eq!(decoded.id, id);
+ assert_eq!(decoded.epoch, epoch);
+ assert!(
+ matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024)
+ );
+
+ let manifest = BlobMessage {
+ epoch,
+ id,
+ body: BlobBody::Manifest {
+ size: 100,
+ chunk_size: 64,
+ hashes: vec![[1; 32], [2; 32]],
+ },
+ };
+ let decoded = bytes_to_blob(&blob_to_bytes(&manifest).unwrap()).unwrap();
+ assert!(matches!(
+ decoded.body,
+ BlobBody::Manifest { size: 100, chunk_size: 64, ref hashes } if hashes.len() == 2
+ ));
+ }
+
+ #[test]
+ fn listed_heads_equal_their_digest() {
+ let heads: Vec<_> = (0..3u8).map(|n| ChangeHash([n; 32])).collect();
+ let listed = HeadsSummary::of(heads.clone());
+ let digest = HeadsSummary::Digest(listed.digest());
+ assert_eq!(listed, digest);
+ assert_ne!(listed, HeadsSummary::of(heads[..2].to_vec()));
+ }
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 22:40:05 +0000