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.rs462
1 files changed, 358 insertions, 104 deletions
diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 057916f..82e6016 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -1,6 +1,7 @@
-use anyhow::{anyhow, Context, Result};
+use anyhow::{Context, Result, anyhow, bail};
use async_trait::async_trait;
-use base64::engine::{general_purpose::STANDARD as BASE64, Engine};
+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};
@@ -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);
@@ -179,27 +182,11 @@ impl MyceliumSetup {
}
}
+/// The part of Mycelium's `/api/v1/admin` answer we use.
#[derive(Deserialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct MyceliumAdminInfo {
- pub node_ip: String,
pub node_pubkey: String,
- pub node_start_time: u64,
- pub node_subnet: String,
-}
-
-pub async fn fetch_admin_info(rest_url: &Url) -> Result<MyceliumAdminInfo> {
- let url = rest_url.join("/api/v1/admin")?;
- let info = reqwest::Client::builder()
- .timeout(Duration::from_secs(10))
- .build()?
- .get(url)
- .send()
- .await?
- .error_for_status()?
- .json()
- .await?;
- Ok(info)
}
#[derive(Debug, Clone)]
@@ -216,10 +203,6 @@ pub struct SupervisedMycelium {
pub config: MyceliumNodeConfig,
_key_file_handle: Option<File>,
_net_key_file_handle: Option<File>,
- key_file_path: PathBuf,
- net_key_file_path: Option<PathBuf>,
- is_ephemeral_key: bool,
- is_ephemeral_net_key: bool,
}
impl SupervisedMycelium {
@@ -232,6 +215,15 @@ impl SupervisedMycelium {
explicit_discovery_port: Option<u16>,
peers: &[String],
) -> Result<Self> {
+ // Linux only: keys are handed over in memory (memfd, never on disk), and the
+ // child dies with us (PR_SET_PDEATHSIG). Elsewhere, run myceliumd yourself.
+ if cfg!(not(target_os = "linux")) {
+ bail!(
+ "Supervising Mycelium is only supported on Linux; run myceliumd yourself \
+ and pass --mycelium-rest-url"
+ );
+ }
+
let tcp_port = match explicit_tcp {
Some(p) => p,
None => Self::find_available_port().context("Failed to allocate TCP wire port")?,
@@ -261,18 +253,17 @@ impl SupervisedMycelium {
let parent_pid = std::process::id();
- let (key_file_handle, key_file_path, node_fd, is_ephemeral_key) =
+ let (key_file_handle, key_file_path, node_fd) =
Self::resolve_key_strategy("mycelium_node_key", node_key, parent_pid)?;
- let (net_key_file_handle, net_key_file_path, net_fd, is_ephemeral_net_key) =
- match private_network {
- Some(net_cfg) => {
- let (h, p, fd, eph) =
- Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?;
- (h, Some(p), fd, eph)
- }
- None => (None, None, None, false),
- };
+ let (net_key_file_handle, net_key_file_path, net_fd) = match private_network {
+ Some(net_cfg) => {
+ let (h, p, fd) =
+ Self::resolve_key_strategy("mycelium_net_key", net_cfg.key, parent_pid)?;
+ (h, Some(p), fd)
+ }
+ None => (None, None, None),
+ };
let api_addr = format!("127.0.0.1:{}", config.api_port);
let rpc_addr = format!("127.0.0.1:{}", config.rpc_port);
@@ -370,14 +361,8 @@ impl SupervisedMycelium {
config,
_key_file_handle: key_file_handle,
_net_key_file_handle: net_key_file_handle,
- key_file_path,
- net_key_file_path,
- is_ephemeral_key,
- is_ephemeral_net_key,
};
- //let admin_info = node.poll_admin_endpoint().await?;
- //Ok((node, admin_info))
Ok(node)
}
@@ -385,14 +370,14 @@ impl SupervisedMycelium {
desc_tag: &str,
source: KeySource,
parent_pid: u32,
- ) -> Result<(Option<File>, PathBuf, Option<i32>, bool)> {
+ ) -> Result<(Option<File>, PathBuf, Option<i32>)> {
match source {
KeySource::Raw(raw_bytes) => {
let (handle, fd) = Self::create_ephemeral_file(desc_tag, raw_bytes)?;
let pseudo_path = PathBuf::from(format!("/proc/{}/fd/{}", parent_pid, fd));
- Ok((Some(handle), pseudo_path, Some(fd), true))
+ Ok((Some(handle), pseudo_path, Some(fd)))
}
- KeySource::Path(existing_path) => Ok((None, existing_path, None, false)),
+ KeySource::Path(existing_path) => Ok((None, existing_path, None)),
}
}
@@ -419,41 +404,14 @@ impl SupervisedMycelium {
Ok((file, fd))
}
+ // Unreachable: `spawn` refuses to run elsewhere.
#[cfg(not(target_os = "linux"))]
{
- let mut path = std::env::tempdir();
- path.push(format!("{}_{}.key", desc_tag, uuid::Uuid::new_v4()));
- let mut file = File::create(&path)?;
- file.write_all(&raw_key)?;
- file.flush()?;
- Ok((file, -1))
+ let _ = (desc_tag, raw_key);
+ bail!("In-memory key files need Linux")
}
}
- //async fn poll_admin_endpoint(&mut self) -> Result<MyceliumAdminInfo> {
- // let client = reqwest::Client::new();
- // let url = format!("http://127.0.0.1:{}/api/v1/admin", self.config.api_port);
- //
- // for _ in 0..30 {
- // if let Some(status) = self.child.try_wait()? {
- // return Err(anyhow!(
- // "myceliumd process closed prematurely with status: {}",
- // status
- // ));
- // }
- // if let Ok(resp) = client.get(&url).send().await {
- // if resp.status().is_success() {
- // let info: MyceliumAdminInfo = resp.json().await?;
- // return Ok(info);
- // }
- // }
- // tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
- // }
- // Err(anyhow!(
- // "Timeout waiting for local mycelium administration socket to respond"
- // ))
- //}
-
pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
self.child.wait().await
}
@@ -461,19 +419,8 @@ impl SupervisedMycelium {
impl Drop for SupervisedMycelium {
fn drop(&mut self) {
+ // The memfd key files go away with their handles.
let _ = self.child.start_kill();
-
- #[cfg(not(target_os = "linux"))]
- {
- if self.is_ephemeral_key {
- let _ = std::fs::remove_file(&self.key_file_path);
- }
- if self.is_ephemeral_net_key {
- if let Some(ref path) = self.net_key_file_path {
- let _ = std::fs::remove_file(path);
- }
- }
- }
}
}
@@ -511,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> {
@@ -546,7 +539,7 @@ impl Transport for MyceliumTransport {
&self.local_id
}
- async fn resolve_peer(&self, hint: &str) -> Result<String> {
+ async fn resolve_seed(&self, hint: &str) -> Result<String> {
if is_pubkey(hint) {
return Ok(hint.to_owned());
}
@@ -568,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,
})
}
@@ -694,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");
}
}
@@ -729,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)?;
@@ -747,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())),
}
}
@@ -763,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()))
}
// =====================================================================
@@ -820,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)]
@@ -865,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)]
@@ -874,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,
}
// =====================================================================
@@ -908,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 {
@@ -927,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)
}
@@ -937,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,
})
}
@@ -946,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)
}
@@ -955,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,
})
}
@@ -990,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 19:28:42 +0000