diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/firewall.rs | 4 | ||||
| -rw-r--r-- | src/transport/mod.rs | 208 | ||||
| -rw-r--r-- | src/transport/mycelium.rs | 369 | ||||
| -rw-r--r-- | src/transport/mycelium.rs.orig | 975 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 201 |
5 files changed, 1721 insertions, 36 deletions
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs index 84fdc59..b8e05cd 100644 --- a/src/transport/firewall.rs +++ b/src/transport/firewall.rs @@ -89,12 +89,12 @@ impl Transport for FirewallTransport { let firewall = Arc::clone(&self.firewall); self.inner - .start_consumer(Box::new(move |sender, message| { + .start_consumer(Box::new(move |sender, message, payload_len| { if firewall.blocks_inbound(&sender) { tracing::trace!(peer = %sender, "Firewall dropped inbound message"); return Box::pin(async { Ok(()) }); } - handler(sender, message) + handler(sender, message, payload_len) })) .await } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 69d9e70..c5769ca 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -1,6 +1,8 @@ use anyhow::Result; use async_trait::async_trait; +use automerge::ChangeHash; use automerge::sync::Message; +use sha2::{Digest, Sha256}; use std::boxed::Box; use std::future::Future; use std::pin::Pin; @@ -9,14 +11,86 @@ pub use crate::document::replica::Epoch; pub mod firewall; pub mod mycelium; +pub mod traffic; -pub type HeadsDigest = [u8; 32]; +/// Random identifier of one run of a node (from start to exit). +pub type RunId = [u8; 16]; +/// Most heads listed in a keepalive; more are sent as a digest. A head takes 34 bytes on +/// the wire, so 32 of them keep a keepalive within one Mycelium chunk (1300 bytes; see +/// the size test in `mycelium`). There are rarely more heads than nodes that wrote +/// concurrently since the last merging write. +pub const MAX_LISTED_HEADS: usize = 32; + +/// A document's heads, compactly: the heads themselves while there are few, otherwise a +/// digest of them. Two summaries are equal exactly when the heads are, whatever form +/// each took (nodes with different limits still agree). +#[derive(Debug, Clone)] +pub enum HeadsSummary { + /// Sorted. + Heads(Vec<ChangeHash>), + /// SHA-256 over the sorted heads. + Digest([u8; 32]), +} + +impl HeadsSummary { + pub fn of(mut heads: Vec<ChangeHash>) -> Self { + heads.sort_unstable(); + if heads.len() <= MAX_LISTED_HEADS { + Self::Heads(heads) + } else { + Self::Digest(digest_of(&heads)) + } + } + + /// The digest of the heads, computed if they are listed. + pub fn digest(&self) -> [u8; 32] { + match self { + Self::Heads(heads) => digest_of(heads), + Self::Digest(digest) => *digest, + } + } + + /// The heads, if listed. + pub fn heads(&self) -> Option<&[ChangeHash]> { + match self { + Self::Heads(heads) => Some(heads), + Self::Digest(_) => None, + } + } +} + +impl PartialEq for HeadsSummary { + fn eq(&self, other: &Self) -> bool { + match (self, other) { + (Self::Heads(a), Self::Heads(b)) => a == b, + _ => self.digest() == other.digest(), + } + } +} + +impl Eq for HeadsSummary {} + +/// SHA-256 over sorted heads. +fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] { + let mut hasher = Sha256::new(); + for head in sorted_heads { + hasher.update(head.0); + } + hasher.finalize().into() +} + +/// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient. #[derive(Debug, Clone)] pub struct Keepalive { - pub heads_digest: HeadsDigest, - /// The digest is only comparable between nodes on the same epoch. + /// Only comparable between nodes on the same epoch. + pub heads: HeadsSummary, pub epoch: Epoch, + /// The sender is ready (same meaning as `SyncEnvelope::ready`). + pub ready: bool, + /// The sender would sync with the recipient: counts it as an active member on its + /// own epoch. + pub syncs_with_you: bool, } #[derive(Debug)] @@ -26,9 +100,111 @@ pub enum ParsedMessage { Sync(SyncEnvelope), Ping(Keepalive), Pong(Keepalive), + Blob(BlobMessage), +} + +/// Content-addressed bytes that any node holding them can serve, in chunks +/// (`swarm::blob`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum BlobId { + /// An Automerge change, by its hash: e.g. an epoch's snapshot change. + Change(ChangeHash), + /// Bytes by their BLAKE3 hash: e.g. the content of a remake. + Content([u8; 32]), +} + +impl std::fmt::Display for BlobId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BlobId::Change(hash) => write!(f, "change:{hash}"), + BlobId::Content(hash) => { + write!(f, "content:")?; + hash.iter().try_for_each(|byte| write!(f, "{byte:02x}")) + } + } + } +} + +/// A blob request or one of its answers. Each is a message of its own, so no message +/// is larger than one chunk. +#[derive(Debug, Clone)] +pub struct BlobMessage { + /// The sender's epoch (every message carries it). + pub epoch: Epoch, + pub id: BlobId, + pub body: BlobBody, +} + +#[derive(Clone)] +pub enum BlobBody { + /// Asks for these chunks; for the manifest if empty. + Request { + chunks: Vec<u32>, + }, + /// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter. + Manifest { + size: u64, + chunk_size: u32, + hashes: Vec<[u8; 32]>, + }, + Chunk { + index: u32, + data: bytes::Bytes, + }, + /// The sender doesn't have the blob. + Missing, +} + +/// Sizes instead of contents: logged messages stay short. +impl std::fmt::Debug for BlobBody { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BlobBody::Request { chunks } => { + f.debug_struct("Request").field("chunks", chunks).finish() + } + BlobBody::Manifest { + size, + chunk_size, + hashes, + } => f + .debug_struct("Manifest") + .field("size", size) + .field("chunk_size", chunk_size) + .field("chunks", &hashes.len()) + .finish(), + BlobBody::Chunk { index, data } => f + .debug_struct("Chunk") + .field("index", index) + .field("bytes", &data.len()) + .finish(), + BlobBody::Missing => f.write_str("Missing"), + } + } +} + +/// What a message is, without its contents. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MessageKind { + Hello, + HelloReply, + Sync, + Ping, + Pong, + Blob, } impl ParsedMessage { + pub fn kind(&self) -> MessageKind { + match self { + ParsedMessage::Hello(_) => MessageKind::Hello, + ParsedMessage::HelloReply(_) => MessageKind::HelloReply, + ParsedMessage::Sync(_) => MessageKind::Sync, + ParsedMessage::Ping(_) => MessageKind::Ping, + ParsedMessage::Pong(_) => MessageKind::Pong, + ParsedMessage::Blob(_) => MessageKind::Blob, + } + } + /// Every message carries the sender's epoch. pub fn epoch(&self) -> Epoch { match self { @@ -36,15 +212,23 @@ impl ParsedMessage { ParsedMessage::HelloReply(reply) => reply.epoch, ParsedMessage::Sync(envelope) => envelope.epoch, ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch, + ParsedMessage::Blob(blob) => blob.epoch, } } } #[derive(Debug, Clone)] pub struct HelloRequest { + /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in + /// another swarm doesn't answer. + pub swarm_id: Option<String>, pub epoch: Epoch, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, + /// The sender is ready (same meaning as `SyncEnvelope::ready`). + pub ready: bool, + /// Random per run of the sender; the reply echoes it. + pub run_id: RunId, } #[derive(Debug, Clone)] @@ -54,6 +238,14 @@ pub struct HelloReply { pub epoch: Epoch, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, + /// The sender is ready. + pub ready: bool, + /// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the + /// recipient (delivered late, or queued while it was down) is ignored. + pub run_id: RunId, + /// How long before this HELLO the sender had last heard from the recipient, if ever + /// (in milliseconds): tells a restarted node how long it was away. + pub away_ms: Option<u64>, } #[derive(Debug, Clone)] @@ -62,6 +254,10 @@ pub struct SyncEnvelope { /// Epoch of the document the message was generated from. pub epoch: Epoch, pub message: Message, + /// The sender was ready when it generated the message: its document is the swarm's, + /// not one it is still filling (joining, rejoining). Only such a peer can tell a + /// catching-up node that it has everything; an empty document says nothing else. + pub ready: bool, } /// A message serialized for the wire. Cloning is cheap (shared buffer). @@ -69,9 +265,13 @@ pub struct SyncEnvelope { pub struct EncodedMessage { pub peer: String, pub body: bytes::Bytes, + pub kind: MessageKind, + /// Bytes the message takes on the wire (without the transport's own framing). + pub payload_len: usize, } -pub type HandlerFn = dyn Fn(String, ParsedMessage) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> +/// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`). +pub type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> + Send + Sync + 'static; diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index a41d509..82e6016 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -1,6 +1,7 @@ -use anyhow::{anyhow, bail, 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); @@ -253,15 +256,14 @@ impl SupervisedMycelium { 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 (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); @@ -456,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> { @@ -513,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, }) } @@ -639,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"); } } @@ -674,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)?; @@ -692,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())), } } @@ -708,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())) } // ===================================================================== @@ -765,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)] @@ -810,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)] @@ -819,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, } // ===================================================================== @@ -853,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 { @@ -872,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) } @@ -882,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, }) } @@ -891,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) } @@ -900,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, }) } @@ -935,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())); + } +} diff --git a/src/transport/mycelium.rs.orig b/src/transport/mycelium.rs.orig new file mode 100644 index 0000000..9d1fb90 --- /dev/null +++ b/src/transport/mycelium.rs.orig @@ -0,0 +1,975 @@ +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 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::{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; + +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 + } + } + } +} + +/// 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, +} + +#[derive(Debug, Clone)] +pub struct MyceliumNodeConfig { + 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: MyceliumNodeConfig, + _key_file_handle: Option<File>, + _net_key_file_handle: Option<File>, +} + +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> { + // 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")?, + }; + 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 = MyceliumNodeConfig { + 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) = + 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<i32> = [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"); + + 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, + }; + + Ok(node) + } + + fn resolve_key_strategy( + desc_tag: &str, + source: KeySource, + parent_pid: u32, + ) -> 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))) + } + KeySource::Path(existing_path) => Ok((None, existing_path, None)), + } + } + + 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)) + } + + // 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<std::process::ExitStatus> { + 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(); + } +} + +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_seed(&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, + }) +} + +/// 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), + 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)?, + ready: shadow.ready, + syncs_with_you: shadow.syncs_with_you, + }) +} + +#[derive(Serialize, Deserialize)] +pub struct HelloRequestShadow { + #[serde(default)] + epoch: EpochShadow, + #[serde(default)] + auth: 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, +} + +#[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 { + epoch: epoch_to_shadow(&msg.epoch), + auth: ByteBuf::from(msg.auth.clone()), + }) +} +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)?, + auth: shadow.auth.into_vec(), + }) +} + +// --- 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), + auth: ByteBuf::from(msg.auth.clone()), + }; + 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)?, + auth: shadow.auth.into_vec(), + }) +} + +// --- 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), + 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)?, + message: shadow.message, + ready: shadow.ready, + }) +} + +#[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, + }, +} diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs new file mode 100644 index 0000000..9dcd91a --- /dev/null +++ b/src/transport/traffic.rs @@ -0,0 +1,201 @@ +//! Messages and bytes exchanged with each peer, by kind: counted where every message +//! passes anyway, at the transport. No messages of our own. +//! +//! Bytes are payload bytes, what the Mycelium message carries, without its framing, +//! the topic, or the base64 of its HTTP API. Counters only grow (from the start of the +//! process); rates are for whoever reads them. +//! +//! A message dropped by the debug firewall still counts as sent (the firewall plays a +//! lossy network) but not as received. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, MutexGuard}; + +use anyhow::Result; +use async_trait::async_trait; +use serde::Serialize; +use utoipa::ToSchema; + +use crate::transport::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport}; + +#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)] +pub struct Counter { + pub messages: u64, + pub bytes: u64, +} + +impl Counter { + fn add(&mut self, bytes: usize) { + self.messages += 1; + self.bytes += bytes as u64; + } +} + +#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct ByKind { + pub hello: Counter, + pub hello_reply: Counter, + pub sync: Counter, + pub ping: Counter, + pub pong: Counter, + pub blob: Counter, +} + +impl ByKind { + fn get_mut(&mut self, kind: MessageKind) -> &mut Counter { + match kind { + MessageKind::Hello => &mut self.hello, + MessageKind::HelloReply => &mut self.hello_reply, + MessageKind::Sync => &mut self.sync, + MessageKind::Ping => &mut self.ping, + MessageKind::Pong => &mut self.pong, + MessageKind::Blob => &mut self.blob, + } + } +} + +#[derive(Debug, Default, Clone, Copy, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub struct PeerTraffic { + /// Handed to Mycelium (it accepted them). + pub sent: ByKind, + pub received: ByKind, + /// Sends Mycelium refused (e.g. no route to the peer); not counted as sent. + pub refused: ByKind, +} + +#[derive(Debug, Default)] +struct Table { + peers: HashMap<String, PeerTraffic>, + /// All peers, including ones no longer listed. + total: PeerTraffic, +} + +#[derive(Debug, Default)] +pub struct Traffic { + table: Mutex<Table>, +} + +impl Traffic { + fn lock(&self) -> MutexGuard<'_, Table> { + self.table + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn count(&self, peer: &str, bytes: usize, pick: impl Fn(&mut PeerTraffic) -> &mut Counter) { + let mut guard = self.lock(); + let table = &mut *guard; + pick(&mut table.total).add(bytes); + match table.peers.get_mut(peer) { + Some(traffic) => pick(traffic).add(bytes), + None => pick(table.peers.entry(peer.to_owned()).or_default()).add(bytes), + } + } + + fn sent(&self, peer: &str, kind: MessageKind, bytes: usize) { + self.count(peer, bytes, |t| t.sent.get_mut(kind)); + } + + fn refused(&self, peer: &str, kind: MessageKind, bytes: usize) { + self.count(peer, bytes, |t| t.refused.get_mut(kind)); + } + + fn received(&self, peer: &str, kind: MessageKind, bytes: usize) { + self.count(peer, bytes, |t| t.received.get_mut(kind)); + } + + pub fn peer(&self, peer: &str) -> Option<PeerTraffic> { + self.lock().peers.get(peer).copied() + } + + pub fn total(&self) -> PeerTraffic { + self.lock().total + } + + /// Drops the counters of peers not kept (anyone can send us messages; the totals + /// keep their share). + pub fn retain(&self, mut keep: impl FnMut(&str) -> bool) { + self.lock().peers.retain(|peer, _| keep(peer)); + } +} + +/// Counts what passes through `inner`. +pub struct CountingTransport { + inner: Arc<dyn Transport>, + traffic: Arc<Traffic>, +} + +impl CountingTransport { + pub fn new(inner: Arc<dyn Transport>, traffic: Arc<Traffic>) -> Self { + Self { inner, traffic } + } +} + +#[async_trait] +impl Transport for CountingTransport { + fn local_id(&self) -> &str { + self.inner.local_id() + } + + async fn resolve_seed(&self, seed: &str) -> Result<String> { + self.inner.resolve_seed(seed).await + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + self.inner.encode(peer, message) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + let result = self.inner.send_encoded(message).await; + if result.is_ok() { + self.traffic + .sent(&message.peer, message.kind, message.payload_len); + } else { + self.traffic + .refused(&message.peer, message.kind, message.payload_len); + } + result + } + + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let traffic = Arc::clone(&self.traffic); + self.inner + .start_consumer(Box::new(move |sender, message, payload_len| { + traffic.received(&sender, message.kind(), payload_len); + handler(sender, message, payload_len) + })) + .await + } + + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + self.inner.wait_for_exit().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn totals_survive_retain() { + let traffic = Traffic::default(); + traffic.sent("a", MessageKind::Ping, 100); + traffic.sent("b", MessageKind::Sync, 1000); + traffic.received("b", MessageKind::Pong, 90); + traffic.refused("a", MessageKind::Sync, 500); + + let b = traffic.peer("b").unwrap(); + assert_eq!(b.sent.sync.messages, 1); + assert_eq!(b.received.pong.bytes, 90); + + traffic.retain(|peer| peer == "b"); + assert!(traffic.peer("a").is_none()); + + let total = traffic.total(); + assert_eq!(total.sent.ping.bytes, 100); + assert_eq!(total.sent.sync.bytes, 1000); + assert_eq!(total.refused.sync.bytes, 500); + } +} |