diff options
Diffstat (limited to 'src/mycelium.rs')
| -rw-r--r-- | src/mycelium.rs | 964 |
1 files changed, 0 insertions, 964 deletions
diff --git a/src/mycelium.rs b/src/mycelium.rs deleted file mode 100644 index d2deebe..0000000 --- a/src/mycelium.rs +++ /dev/null @@ -1,964 +0,0 @@ -use anyhow::{anyhow, Context, Result}; -use async_trait::async_trait; -use base64::engine::{general_purpose::STANDARD as BASE64, Engine}; -use reqwest::StatusCode; -use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use serde_bytes::{ByteBuf, Bytes}; -use std::borrow::Cow; -use std::boxed::Box; -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 std::time::Duration; -use thiserror::Error; -use tokio::process::{Child, Command}; -use url::Url; - -use crate::transport::EncodedMessage; -use crate::transport::Keepalive; -use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope}; -use crate::transport::{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 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, -} - -/// 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<Url>, - pub key_file_opt: Option<PathBuf>, - pub network_key_file: Option<PathBuf>, - pub network_name: String, - pub tcp_port: Option<u16>, - pub api_port: Option<u16>, - pub rpc_port: Option<u16>, - pub discovery_port: Option<u16>, - pub peers: Vec<String>, -} - -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, - api_port: None, - rpc_port: None, - discovery_port: None, - peers: vec!["tls://127.0.0.1:9651".to_string()], - } - } -} - -pub enum MyceliumRuntime { - Supervised(Box<SupervisedMycelium>), - 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<Self> { - // 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.api_port, - config.rpc_port, - config.discovery_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<std::process::ExitStatus> { - match &mut self.runtime { - MyceliumRuntime::Supervised(node) => node.wait_for_exit().await, - MyceliumRuntime::External => { - std::future::pending::<std::io::Result<std::process::ExitStatus>>().await - } - } - } -} - -#[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)] -pub struct ClusterNodeConfig { - pub tcp_port: u16, - pub api_port: u16, - pub rpc_port: u16, - pub discovery_port: Option<u16>, - pub private_network: Option<String>, -} - -pub struct SupervisedMycelium { - child: Child, - pub config: ClusterNodeConfig, - _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 { - async fn spawn( - node_key: KeySource, - private_network: Option<PrivateNetworkConfig>, - explicit_tcp: Option<u16>, - explicit_api: Option<u16>, - explicit_rpc: Option<u16>, - explicit_discovery_port: Option<u16>, - peers: &[String], - ) -> Result<Self> { - let tcp_port = match explicit_tcp { - Some(p) => p, - None => Self::find_available_port().context("Failed to allocate TCP wire port")?, - }; - let api_port = match explicit_api { - Some(p) => p, - None => Self::find_available_port().context("Failed to allocate API port")?, - }; - let rpc_port = match explicit_rpc { - Some(p) => p, - None => Self::find_available_port().context("Failed to allocate JSON-RPC port")?, - }; - let discovery_port = match explicit_discovery_port { - Some(0) => Some( - Self::find_available_port().context("Failed to allocate peer discovery port")?, - ), - p => p, - }; - - let config = ClusterNodeConfig { - tcp_port, - api_port, - rpc_port, - discovery_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, is_ephemeral_key) = - 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 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<i32> = [node_fd, net_fd].iter().filter_map(|&x| x).collect(); - if !fds_to_inherit.is_empty() { - unsafe { - cmd.pre_exec(move || { - for &fd in &fds_to_inherit { - libc::fcntl(fd, libc::F_SETFD, 0); - } - 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"); - - if let Some(ref discovery_port) = config.discovery_port { - cmd.arg("--peer-discovery-port") - .arg(discovery_port.to_string()); - } else { - 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")?; - - // 1. FIXED: Pass-through streaming for child process STDOUT - 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; - }); - } - - // 2. FIXED: Pass-through streaming for child process STDERR - 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, - 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) - } - - fn resolve_key_strategy( - desc_tag: &str, - source: KeySource, - parent_pid: u32, - ) -> Result<(Option<File>, PathBuf, Option<i32>, bool)> { - 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)) - } - KeySource::Path(existing_path) => Ok((None, existing_path, None, false)), - } - } - - fn find_available_port() -> Result<u16> { - 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)) - } - - #[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)) - } - } - - //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 - } -} - -impl Drop for SupervisedMycelium { - fn drop(&mut self) { - 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); - } - } - } - } -} - -fn is_pubkey(s: &str) -> bool { - s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit()) -} - -pub struct MyceliumTransport { - // Wrap your zero-allocation lifecycle supervisor directly inside the driver plane - pub setup: tokio::sync::Mutex<MyceliumSetup>, - pub client: reqwest::Client, - pub poll_client: reqwest::Client, - pub messages_endpoint: Url, - pubkey_basepath: Url, - local_id: String, -} - -impl MyceliumTransport { - pub async fn connect(setup: MyceliumSetup) -> Result<Self> { - 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")?; - - Ok(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, - }) - } - - async fn lookup_pubkey(&self, ip: &std::net::Ipv6Addr) -> Result<String> { - #[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_peer(&self, hint: &str) -> Result<String> { - if is_pubkey(hint) { - return Ok(hint.to_owned()); - } - - if let Ok(ip) = hint.parse::<std::net::Ipv6Addr>() { - 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<EncodedMessage> { - let outbound = format_message(peer, message)?; - Ok(EncodedMessage { - peer: peer.to_owned(), - body: serde_json::to_vec(&outbound)?.into(), - }) - } - - 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::<serde_json::Value>(&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::<OutboundMessage>(&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_PREFIX))]) - .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::<InboundMessage>(&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" - ); - if let Err(error) = handler(sender, parsed).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<std::process::ExitStatus> { - let mut guard = self.setup.lock().await; - guard.wait_for_exit().await - } -} - -fn encode_topic(topic: &str) -> String { - BASE64.encode(topic.as_bytes()) -} - -fn decode_topic(topic: &str) -> Result<String, TransportError> { - let bytes = BASE64.decode(topic)?; - Ok(String::from_utf8(bytes)?) -} - -fn encode_payload(payload: &Vec<u8>) -> String { - BASE64.encode(payload) -} - -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)?)), - t => Err(TransportError::UnknownTopic(t.to_string())), - } -} - -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)) -} - -fn format_message( - recipient: &str, - message: &ParsedMessage, -) -> Result<OutboundMessage, 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)?), - }; - Ok(OutboundMessage { - dst: Destination { - pk: recipient.to_owned(), - }, - topic: encode_topic(topic), - payload: encode_payload(&payload_bytes), - }) -} - -// ===================================================================== -// 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, -} - -fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow { - EpochShadow { - generation: epoch.generation, - id: ByteBuf::from(epoch.id.to_vec()), - } -} - -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, - }) -} - -#[derive(Serialize, Deserialize)] -struct KeepaliveShadow { - heads_digest: ByteBuf, - #[serde(default)] - epoch: EpochShadow, -} - -fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> { - rmp_serde::to_vec(&KeepaliveShadow { - heads_digest: ByteBuf::from(msg.heads_digest.to_vec()), - epoch: epoch_to_shadow(&msg.epoch), - }) -} - -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()))?; - Ok(Keepalive { - heads_digest, - epoch: epoch_from_shadow(shadow.epoch)?, - }) -} - -#[derive(Serialize, Deserialize)] -pub struct HelloRequestShadow { - #[serde(default)] - epoch: EpochShadow, -} - -#[derive(Serialize, Deserialize)] -pub struct HelloReplyShadow<'a> { - // Cow allows zero-copy references when saving, but creates an owned String when loading - pub cluster_id: Cow<'a, str>, - #[serde(default)] - pub peers: Vec<String>, - #[serde(default)] - epoch: EpochShadow, -} - -#[derive(Serialize, Deserialize)] -pub struct SyncEnvelopeShadow<'a> { - pub cluster_id: Cow<'a, str>, - #[serde(with = "automerge_msg_codec")] - pub message: automerge::sync::Message, - #[serde(default)] - epoch: EpochShadow, -} - -// ===================================================================== -// 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 { - epoch: epoch_to_shadow(&msg.epoch), - }) -} -pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> { - let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; - Ok(HelloRequest { - epoch: epoch_from_shadow(shadow.epoch)?, - }) -} - -// --- HelloReply --- -pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> { - let shadow = HelloReplyShadow { - cluster_id: Cow::Borrowed(&msg.cluster_id), - peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching - epoch: epoch_to_shadow(&msg.epoch), - }; - 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 { - cluster_id: shadow.cluster_id.into_owned(), - peers: shadow.peers, - epoch: epoch_from_shadow(shadow.epoch)?, - }) -} - -// --- SyncEnvelope --- -pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::encode::Error> { - let shadow = SyncEnvelopeShadow { - cluster_id: Cow::Borrowed(&msg.cluster_id), - message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules - epoch: epoch_to_shadow(&msg.epoch), - }; - 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 { - cluster_id: shadow.cluster_id.into_owned(), - epoch: epoch_from_shadow(shadow.epoch)?, - message: shadow.message, - }) -} - -#[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("MessagePack serialization failed: {0}")] - MessagePackEncode(#[from] rmp_serde::encode::Error), - - #[error("MessagePack deserialization failed: {0}")] - MessagePackDecode(#[from] rmp_serde::decode::Error), - - #[error("Unknown topic: {0}")] - UnknownTopic(String), - - #[error("Invalid payload: {0}")] - InvalidPayload(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, - }, -} |