path: root/src/transport/mod.rs

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;

pub use crate::document::replica::Epoch;

pub mod firewall;
pub mod mycelium;
pub mod traffic;

/// 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 {
    /// 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)]
pub enum ParsedMessage {
    Hello(HelloRequest),
    HelloReply(HelloReply),
    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),
}

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}"),
        }
    }
}

/// 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 {
            ParsedMessage::Hello(hello) => hello.epoch,
            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)]
pub struct HelloReply {
    pub swarm_id: String,
    pub peers: Vec<String>,
    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)]
pub struct SyncEnvelope {
    pub swarm_id: String,
    /// 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).
#[derive(Clone, Debug)]
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,
}

/// 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;
pub type MessageHandler = Box<HandlerFn>;

#[async_trait]
pub trait Transport: Send + Sync {
    /// This node's identity on the transport (for Mycelium: its public key).
    fn local_id(&self) -> &str;

    /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's
    /// identity, as `local_id` gives it on that node.
    async fn resolve_seed(&self, seed: &str) -> Result<String>;

    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage>;
    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()>;

    /// Sends an outbound message payload to a targeted peer on a specific topic.
    async fn send_message(&self, peer: &str, message: &ParsedMessage) -> Result<()> {
        let encoded = self.encode(peer, message)?;
        self.send_encoded(&encoded).await
    }

    /// Starts the long-polling inbound event loops, driving messages to the provided handler closure.
    async fn start_consumer(&self, handler: MessageHandler) -> Result<()>;

    /// Asynchronously monitors for fatal network client or sub-process failures.
    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus>;
}