From 2646e19cf29b410f542097d75e620116c96da96b Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 26 Oct 2026 04:43:09 +0100 Subject: cleanup mycelium daemon --- src/transport/mycelium/daemon.rs | 337 ++++++++++++++++++++++++ src/transport/mycelium/mod.rs | 552 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 889 insertions(+) create mode 100644 src/transport/mycelium/daemon.rs create mode 100644 src/transport/mycelium/mod.rs (limited to 'src/transport/mycelium') diff --git a/src/transport/mycelium/daemon.rs b/src/transport/mycelium/daemon.rs new file mode 100644 index 0000000..80b6be5 --- /dev/null +++ b/src/transport/mycelium/daemon.rs @@ -0,0 +1,337 @@ +//! Running Mycelium for this node: either a daemon the operator runs (its REST URL), or +//! one we start and supervise (`SupervisedMycelium`): its keys handed over in memory, +//! never on disk, and it dies with us. + +use anyhow::{Context, Result, anyhow, bail}; +use std::fs::File; +use std::io::Write; +use std::net::TcpListener; +use std::os::unix::io::FromRawFd; +use std::path::PathBuf; +use std::process::Stdio; +use tokio::process::{Child, Command}; +use url::Url; + +/// Represents where a 32-byte cryptographic key originates. +#[derive(Debug, Clone)] +pub enum KeySource { + Raw([u8; 32]), + Path(PathBuf), +} + +/// Atomic structure ensuring private network credentials are always handled together. +#[derive(Debug, Clone)] +pub struct PrivateNetworkConfig { + pub name: String, + pub key: KeySource, +} + +/// Comprehensive parameters required to spawn or target an instance. +#[derive(Debug, Clone)] +pub struct MyceliumSpawnConfig { + pub rest_url_opt: Option, + pub key_file_opt: Option, + pub network_key_file: Option, + pub network_name: String, + pub tcp_port: Option, + pub peers: Vec, +} + +impl Default for MyceliumSpawnConfig { + fn default() -> Self { + Self { + rest_url_opt: None, + key_file_opt: None, + network_key_file: None, + network_name: "something".to_string(), + tcp_port: None, + peers: vec!["tls://127.0.0.1:9651".to_string()], + } + } +} + +pub enum MyceliumRuntime { + Supervised(Box), + External, +} + +pub struct MyceliumSetup { + pub runtime: MyceliumRuntime, + pub rest_url: Url, +} + +impl MyceliumSetup { + /// Evaluates the configuration payload and initializes the chosen execution strategy. + pub async fn bootstrap(config: MyceliumSpawnConfig) -> Result { + // TODO warn for all ignored options if rest_url_opt is set + if let Some(url) = config.rest_url_opt { + tracing::info!( + "Configured to interface with an external standalone Mycelium instance at: {}", + url + ); + return Ok(Self { + runtime: MyceliumRuntime::External, + rest_url: url, + }); + } + + tracing::info!( + "Configured to initialize an embedded user-space Mycelium background worker" + ); + + let key = match config.key_file_opt { + Some(key_file) => KeySource::Path(key_file), + None => { + let mut bytes = [0u8; 32]; + getrandom::fill(&mut bytes)?; + KeySource::Raw(bytes) + } + }; + + let private_net_config = config + .network_key_file + .map(|net_key_path| PrivateNetworkConfig { + name: config.network_name.clone(), + key: KeySource::Path(net_key_path), + }); + + let supervised_node = + SupervisedMycelium::spawn(key, private_net_config, config.tcp_port, &config.peers) + .await + .context("Failed to securely bootstrap embedded Mycelium background worker")?; + + let resolved_url = Url::parse(&format!( + "http://127.0.0.1:{}", + supervised_node.config.api_port + ))?; + + Ok(Self { + runtime: MyceliumRuntime::Supervised(Box::new(supervised_node)), + rest_url: resolved_url, + }) + } + + /// Blocks until the background daemon drops out. If running externally, hangs open indefinitely. + pub async fn wait_for_exit(&mut self) -> std::io::Result { + match &mut self.runtime { + MyceliumRuntime::Supervised(node) => node.wait_for_exit().await, + MyceliumRuntime::External => { + std::future::pending::>().await + } + } + } +} + +#[derive(Debug, Clone)] +pub struct MyceliumNodeConfig { + pub tcp_port: u16, + pub api_port: u16, + pub rpc_port: u16, + pub private_network: Option, +} + +pub struct SupervisedMycelium { + child: Child, + pub config: MyceliumNodeConfig, + _key_file_handle: Option, + _net_key_file_handle: Option, +} + +impl SupervisedMycelium { + async fn spawn( + node_key: KeySource, + private_network: Option, + explicit_tcp: Option, + peers: &[String], + ) -> Result { + // 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")?, + }; + // Local only, so any free port will do. + let api_port = Self::find_available_port().context("Failed to allocate API port")?; + let rpc_port = Self::find_available_port().context("Failed to allocate JSON-RPC port")?; + + let config = MyceliumNodeConfig { + tcp_port, + api_port, + rpc_port, + private_network: private_network.as_ref().map(|pn| pn.name.clone()), + }; + + let parent_pid = std::process::id(); + + 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) = 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); + + let mut cmd = Command::new("mycelium"); + + let fds_to_inherit: Vec = [node_fd, net_fd].iter().filter_map(|&x| x).collect(); + // The child must never outlive us. An orphaned Mycelium keeps running with our + // key; the next one we start then shares its address, and replies (including + // Mycelium's own message ACKs) go to whichever instance the routers picked, + // leaving our messages pending forever. Drop only runs on a clean shutdown, so + // let the kernel signal the child when we die, however we die. + let parent_pid = std::process::id() as libc::pid_t; + unsafe { + cmd.pre_exec(move || { + for &fd in &fds_to_inherit { + libc::fcntl(fd, libc::F_SETFD, 0); + } + // Own process group: a Ctrl-C in the terminal goes to continuum only, + // which then stops Mycelium itself, instead of both exiting at once (which + // shows up as "Mycelium runtime process exited"). + if libc::setpgid(0, 0) != 0 { + return Err(std::io::Error::last_os_error()); + } + #[cfg(target_os = "linux")] + { + if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM) != 0 { + return Err(std::io::Error::last_os_error()); + } + // We may have died between fork and prctl. + if libc::getppid() != parent_pid { + libc::_exit(1); + } + } + Ok(()) + }); + } + + cmd.arg("--key-file") + .arg(&key_file_path) + .arg("--tcp-listen-port") + .arg(config.tcp_port.to_string()) + .arg("--disable-quic") + .arg("--api-addr") + .arg(&api_addr) + .arg("--jsonrpc-addr") + .arg(&rpc_addr) + .arg("--no-tun"); + + // Peers come from `--mycelium-peers`; no local multicast discovery. + cmd.arg("--disable-peer-discovery"); + + if let Some(ref net_name) = config.private_network { + cmd.arg("--network-name").arg(net_name); + } + if let Some(ref net_key_path) = net_key_file_path { + cmd.arg("--network-key-file").arg(net_key_path); + } + + if !peers.is_empty() { + cmd.arg("--peers").arg(peers.join(" ")); + } + + cmd.env("RUST_LOG", "info"); + + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + + let mut child = cmd + .spawn() + .context("Failed to execute mycelium process daemon")?; + + // Its output goes where ours does. + if let Some(mut stdout) = child.stdout.take() { + tokio::spawn(async move { + let mut host_stdout = tokio::io::stdout(); + let _ = tokio::io::copy(&mut stdout, &mut host_stdout).await; + }); + } + + if let Some(mut stderr) = child.stderr.take() { + tokio::spawn(async move { + let mut host_stderr = tokio::io::stderr(); + let _ = tokio::io::copy(&mut stderr, &mut host_stderr).await; + }); + } + + let node = Self { + child, + config, + _key_file_handle: key_file_handle, + _net_key_file_handle: net_key_file_handle, + }; + + Ok(node) + } + + fn resolve_key_strategy( + desc_tag: &str, + source: KeySource, + parent_pid: u32, + ) -> Result<(Option, PathBuf, Option)> { + 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))) + } + KeySource::Path(existing_path) => Ok((None, existing_path, None)), + } + } + + fn find_available_port() -> Result { + let listener = TcpListener::bind("127.0.0.1:0")?; + let addr = listener.local_addr()?; + Ok(addr.port()) + } + + fn create_ephemeral_file(desc_tag: &str, raw_key: [u8; 32]) -> Result<(File, i32)> { + #[cfg(target_os = "linux")] + { + let memfd_name = std::ffi::CString::new(desc_tag)?; + let fd = unsafe { libc::memfd_create(memfd_name.as_ptr(), 0) }; + if fd < 0 { + return Err(anyhow!( + "memfd_create failed natively inside Linux kernel for tag: {}", + desc_tag + )); + } + let mut file = unsafe { File::from_raw_fd(fd) }; + file.write_all(&raw_key)?; + file.flush()?; + Ok((file, fd)) + } + + // Unreachable: `spawn` refuses to run elsewhere. + #[cfg(not(target_os = "linux"))] + { + let _ = (desc_tag, raw_key); + bail!("In-memory key files need Linux") + } + } + + pub async fn wait_for_exit(&mut self) -> std::io::Result { + self.child.wait().await + } +} + +impl Drop for SupervisedMycelium { + fn drop(&mut self) { + // The memfd key files go away with their handles. + let _ = self.child.start_kill(); + } +} diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs new file mode 100644 index 0000000..cbae567 --- /dev/null +++ b/src/transport/mycelium/mod.rs @@ -0,0 +1,552 @@ +use anyhow::{Context, Result, anyhow}; +use async_trait::async_trait; +use base64::engine::{Engine, general_purpose::STANDARD as BASE64}; +use reqwest::StatusCode; +use serde::{Deserialize, Serialize}; +use std::time::Duration; +use thiserror::Error; +use url::Url; + +use crate::transport::wire; +use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport}; + +pub mod daemon; + +pub use daemon::{MyceliumSetup, MyceliumSpawnConfig}; + +const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300; + +/// 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); + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct InboundMessage { + id: String, + src_pk: String, + topic: String, + payload: String, +} + +// Deserialize only for debugging +#[derive(Debug, Serialize, Deserialize)] +struct OutboundMessage { + dst: Destination, + topic: String, + payload: String, +} + +// Deserialize only for debugging +#[derive(Debug, Serialize, Deserialize)] +struct Destination { + pk: String, +} + +/// The part of Mycelium's `/api/v1/admin` answer we use. +#[derive(Deserialize, Debug, Clone)] +#[serde(rename_all = "camelCase")] +pub struct MyceliumAdminInfo { + pub node_pubkey: String, +} + +fn is_pubkey(s: &str) -> bool { + s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit()) +} + +pub struct MyceliumTransport { + /// The daemon, kept to wait for its exit. + setup: tokio::sync::Mutex, + client: reqwest::Client, + /// For the long-polling inbox (its timeout is longer). + poll_client: reqwest::Client, + messages_endpoint: Url, + pubkey_basepath: Url, + local_id: String, +} + +impl MyceliumTransport { + pub async fn connect(setup: MyceliumSetup) -> Result { + let rest_url = setup.rest_url.clone(); + + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build()?; + let poll_client = reqwest::Client::builder() + .timeout(Duration::from_secs(MYCELIUM_POLL_TIMEOUT_SECS + 30)) + .build()?; + + let admin: MyceliumAdminInfo = client + .get(rest_url.join("/api/v1/admin")?) + .send() + .await? + .error_for_status()? + .json() + .await + .context("Querying local Mycelium identity")?; + + 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))]) + .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 { + #[derive(Deserialize)] + struct PublicKeyResponse { + #[serde(rename = "publicKey")] + public_key: String, + } + + let response: PublicKeyResponse = self + .client + .get(self.pubkey_basepath.join(&ip.to_string())?) + .send() + .await? + .error_for_status()? + .json() + .await?; + Ok(response.public_key) + } +} + +#[async_trait] +impl Transport for MyceliumTransport { + fn local_id(&self) -> &str { + &self.local_id + } + + async fn resolve_seed(&self, hint: &str) -> Result { + if is_pubkey(hint) { + return Ok(hint.to_owned()); + } + + if let Ok(ip) = hint.parse::() { + return self.lookup_pubkey(&ip).await; + } + + let ip = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((hint, 0))) + .await + .context("DNS lookup timed out")?? + .find_map(|addr| match addr.ip() { + std::net::IpAddr::V6(ip) => Some(ip), + _ => None, + }) + .ok_or_else(|| anyhow!("{hint}: no AAAA record found"))?; + + self.lookup_pubkey(&ip).await + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result { + encode_message(peer, message) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + tracing::trace!(payload_bytes = message.body.len(), "Sending message"); + let response = self + .client + .post(self.messages_endpoint.clone()) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body(message.body.clone()) + .send() + .await?; + + let status = response.status(); + if !status.is_success() { + tracing::error!(status = %status, "Mycelium rejected outbound message"); + return Err(TransportError::MyceliumRejected { + peer: message.peer.clone(), + status, + } + .into()); + } + + // Mycelium's message ID, for GET /api/v1/messages/status/{id} when chasing + // a message that never arrived. + if tracing::enabled!(tracing::Level::DEBUG) { + let Ok(body) = response + .bytes() + .await + .map_err(anyhow::Error::from) + .and_then(|bytes| Ok(serde_json::from_slice::(&bytes)?)) + else { + tracing::debug!(to = %message.peer, "Sent message (couldn't decode response from Mycelium REST API)"); + return Ok(()); + }; + let Ok(outbound) = serde_json::from_slice::(&message.body) else { + tracing::debug!(to = %message.peer, body_len = %message.body.len(), "Sent message (couldn't parse outbound message)"); + return Ok(()); + }; + let Ok(parsed) = decode_payload(&outbound.topic, &outbound.payload) else { + tracing::debug!(to = %message.peer, message = ?outbound, "Sent message (couldn't decode outbound message)"); + return Ok(()); + }; + let Some(id) = body.get("id").and_then(|id| id.as_str()) else { + tracing::debug!(to = %message.peer, message = ?parsed, "Sent message (couldn't read id from Mycelium REST API response)"); + return Ok(()); + }; + tracing::debug!(to = %message.peer, message = ?parsed, id = %id, "Sent message"); + } + Ok(()) + } + + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + tracing::info!( + endpoint = self.messages_endpoint.as_str(), + "Starting Mycelium message consumer" + ); + + loop { + tracing::trace!("Polling Mycelium inbox"); + + let response = match self + .poll_client + .get(self.messages_endpoint.clone()) + .query(&[("timeout", MYCELIUM_POLL_TIMEOUT_SECS)]) + .query(&[("topic", encode_topic(TOPIC))]) + .send() + .await + { + Ok(resp) => resp, + Err(error) => { + tracing::warn!(error = %error, "Mycelium inbox poll failed"); + tokio::time::sleep(Duration::from_millis(250)).await; + continue; + } + }; + + let status = response.status(); + + if status == StatusCode::NO_CONTENT { + tracing::trace!("Mycelium inbox is empty"); + tokio::time::sleep(EMPTY_QUEUE_DELAY).await; + continue; + } + + if !status.is_success() { + tracing::warn!(status = %status, "Mycelium inbox returned unsuccessful HTTP status"); + tokio::time::sleep(Duration::from_millis(500)).await; + continue; + } + + let raw_bytes = match response.bytes().await { + Ok(bytes) => bytes, + Err(error) => { + tracing::warn!(error = %error, "Failed to read raw Mycelium response stream"); + tokio::time::sleep(Duration::from_millis(100)).await; + continue; + } + }; + + let inbound_message = match serde_json::from_slice::(&raw_bytes) { + Ok(msg) => msg, + Err(error) => { + tracing::error!( + error = %error, + raw_payload = %String::from_utf8_lossy(&raw_bytes), + "Failed to decode inbound Mycelium message structure" + ); + tokio::time::sleep(Duration::from_millis(500)).await; + continue; + } + }; + + match parse_message(&inbound_message) { + Ok((sender, parsed)) => { + tracing::debug!( + from = %inbound_message.src_pk, + message = ?parsed, + id = %inbound_message.id, + //payload_bytes = raw_bytes.len(), + "Received message" + ); + 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"); + } + } + Err(transport_error) => { + tracing::error!( + error = %transport_error, + topic = %inbound_message.topic, + "Dropping poisoned packet: parsing failed gracefully" + ); + // No sleep required: we successfully popped a malformed packet and want to process the next one immediately + } + } + } + } + + async fn wait_for_exit(&self) -> std::io::Result { + let mut guard = self.setup.lock().await; + guard.wait_for_exit().await + } +} + +/// A message as the Mycelium REST API takes it (the JSON body of an outbound message). +/// Also the simulator's wire format (`transport::sim`), so tests go through the same +/// encoding. +pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result { + 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, + }) +} + +/// The message in a body made by [`encode_message`]. +#[cfg(test)] +pub(crate) fn decode_message(body: &[u8]) -> Result { + let outbound: OutboundMessage = serde_json::from_slice(body)?; + Ok(decode_payload(&outbound.topic, &outbound.payload)?) +} + +fn encode_topic(topic: &str) -> String { + BASE64.encode(topic.as_bytes()) +} + +fn decode_topic(topic: &str) -> Result { + let bytes = BASE64.decode(topic)?; + Ok(String::from_utf8(bytes)?) +} + +fn encode_payload(payload: &Vec) -> 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 { + 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> { + let sender = message.src_pk.clone(); + + if sender.is_empty() { + return Err(TransportError::NoSender); + } + + let parsed = decode_payload(&message.topic, &message.payload)?; + + Ok((sender, parsed)) +} + +/// The message for Mycelium's API, and the length of its payload. +fn format_message( + recipient: &str, + message: &ParsedMessage, +) -> Result<(OutboundMessage, usize), TransportError> { + let payload_bytes = wire::encode(message)?; + let outbound = OutboundMessage { + dst: Destination { + pk: recipient.to_owned(), + }, + topic: encode_topic(TOPIC), + payload: encode_payload(&payload_bytes), + }; + Ok((outbound, payload_bytes.len())) +} + +#[derive(Error, Debug)] +pub enum TransportError { + #[error("Mycelium delivered a message without a sender public key")] + NoSender, + + #[error("Base64 decoding failed: {0}")] + Base64Decode(#[from] base64::DecodeError), + + #[error("UTF-8 decoding failed: {0}")] + Utf8Decode(#[from] std::string::FromUtf8Error), + + #[error(transparent)] + Wire(#[from] wire::WireError), + + #[error("Unknown topic: {0}")] + UnknownTopic(String), + + #[error("Network HTTP request failed: {0}")] + Http(#[from] reqwest::Error), + + #[error("Mycelium server rejected outbound message to {peer} (HTTP {status})")] + MyceliumRejected { + peer: String, + status: reqwest::StatusCode, + }, +} + +#[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; + + #[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], + }, + epoch_seq: u32::MAX, + ready: true, + syncs_with_you: true, + }; + 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.len() + payload.len(); + assert!( + size + 50 <= MYCELIUM_CHUNK, + "keepalive with {MAX_LISTED_HEADS} heads takes {size} bytes" + ); + let ParsedMessage::Pong(decoded) = wire::decode(&payload).unwrap() else { + panic!("not a PONG"); + }; + assert_eq!(decoded.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 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, + body: BlobBody::Chunk { + index: 5, + data: vec![1u8; 64 * 1024].into(), + }, + }); + 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 = decode_blob(&bytes); + assert_eq!(decoded.id, id); + assert_eq!(decoded.epoch, epoch); + assert_eq!(decoded.epoch_seq, 4); + assert!( + matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024) + ); + + let manifest = ParsedMessage::Blob(BlobMessage { + epoch, + epoch_seq: 4, + id, + body: BlobBody::Manifest { + size: 100, + chunk_size: 64, + hashes: vec![[1; 32], [2; 32]], + }, + }); + 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 + )); + } + + #[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()); + let digest = HeadsSummary::Digest(listed.digest()); + assert_eq!(listed, digest); + assert_ne!(listed, HeadsSummary::of(heads[..2].to_vec())); + } +} -- cgit v1.3.1