add chunked download and use it during rejoin

1 files changed, 77 insertions, 0 deletions

diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index b2f5c9d..c96b053 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -100,6 +100,80 @@ 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.
@@ -110,6 +184,7 @@ pub enum MessageKind {
     Sync,
     Ping,
     Pong,
+    Blob,
 }
 
 impl ParsedMessage {
@@ -120,6 +195,7 @@ impl ParsedMessage {
             ParsedMessage::Sync(_) => MessageKind::Sync,
             ParsedMessage::Ping(_) => MessageKind::Ping,
             ParsedMessage::Pong(_) => MessageKind::Pong,
+            ParsedMessage::Blob(_) => MessageKind::Blob,
         }
     }
 
@@ -130,6 +206,7 @@ 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,
         }
     }
 }