1 files changed, 201 insertions, 6 deletions

diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 58db611..c96b053 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,105 @@ 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),
+}
+
+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 {
@@ -36,15 +206,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 +232,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 +248,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 +259,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;
@@ -82,8 +276,9 @@ pub trait Transport: Send + Sync {
     /// This node's identity on the transport (for Mycelium: its public key).
     fn local_id(&self) -> &str;
 
-    /// Resolves a user-supplied peer hint (e.g. a seed) to a peer identity.
-    async fn resolve_peer(&self, hint: &str) -> Result<String>;
+    /// 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<()>;