add epoch catch-up / snapshots

1 files changed, 31 insertions, 0 deletions

diff --git a/src/transport.rs b/src/transport.rs
index 9cfb026..b7a4094 100644
--- a/src/transport.rs
+++ b/src/transport.rs
@@ -23,6 +23,9 @@ pub enum ParsedMessage {
     Sync(SyncEnvelope),
     Ping(Keepalive),
     Pong(Keepalive),
+    /// Asks a peer for its document of a newer epoch (late or new nodes).
+    SnapshotRequest(SnapshotRequest),
+    SnapshotReply(SnapshotReply),
 }
 
 impl ParsedMessage {
@@ -33,6 +36,8 @@ impl ParsedMessage {
             ParsedMessage::HelloReply(reply) => reply.epoch,
             ParsedMessage::Sync(envelope) => envelope.epoch,
             ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch,
+            ParsedMessage::SnapshotRequest(request) => request.epoch,
+            ParsedMessage::SnapshotReply(reply) => reply.epoch,
         }
     }
 }
@@ -57,6 +62,32 @@ pub struct SyncEnvelope {
     pub message: Message,
 }
 
+#[derive(Debug, Clone)]
+pub struct SnapshotRequest {
+    /// The requester's epoch.
+    pub epoch: Epoch,
+    /// The epoch whose document is wanted.
+    pub wanted: Epoch,
+}
+
+#[derive(Clone)]
+pub struct SnapshotReply {
+    /// The sender's epoch, which is the epoch of `document`.
+    pub epoch: Epoch,
+    /// `AutoCommit::save()` of the sender's document; empty if it isn't on the wanted
+    /// epoch (any more).
+    pub document: Vec<u8>,
+}
+
+impl std::fmt::Debug for SnapshotReply {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SnapshotReply")
+            .field("epoch", &self.epoch)
+            .field("document_bytes", &self.document.len())
+            .finish()
+    }
+}
+
 /// A message serialized for the wire. Cloning is cheap (shared buffer).
 #[derive(Clone, Debug)]
 pub struct EncodedMessage {