1 files changed, 68 insertions, 10 deletions

diff --git a/src/mycelium.rs b/src/mycelium.rs
index 94c5a3d..3471f71 100644
--- a/src/mycelium.rs
+++ b/src/mycelium.rs
@@ -19,7 +19,7 @@ use url::Url;
 
 use crate::transport::EncodedMessage;
 use crate::transport::Keepalive;
-use crate::transport::{HelloReply, HelloRequest, SyncEnvelope};
+use crate::transport::{Epoch, HelloReply, HelloRequest, SyncEnvelope};
 use crate::transport::{MessageHandler, ParsedMessage, Transport};
 
 const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
@@ -739,14 +739,54 @@ fn format_message(
     })
 }
 
+// =====================================================================
+// 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,
+    })
+}
+
 #[derive(Serialize, Deserialize)]
 struct KeepaliveShadow {
     heads_digest: ByteBuf,
+    #[serde(default)]
+    epoch: EpochShadow,
 }
 
 fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Error> {
     rmp_serde::to_vec(&KeepaliveShadow {
         heads_digest: ByteBuf::from(msg.heads_digest.to_vec()),
+        epoch: epoch_to_shadow(&msg.epoch),
     })
 }
 
@@ -757,11 +797,17 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
         .as_slice()
         .try_into()
         .map_err(|_| TransportError::InvalidPayload("heads digest must be 32 bytes".into()))?;
-    Ok(Keepalive { heads_digest })
+    Ok(Keepalive {
+        heads_digest,
+        epoch: epoch_from_shadow(shadow.epoch)?,
+    })
 }
 
 #[derive(Serialize, Deserialize)]
-pub struct HelloRequestShadow {}
+pub struct HelloRequestShadow {
+    #[serde(default)]
+    epoch: EpochShadow,
+}
 
 #[derive(Serialize, Deserialize)]
 pub struct HelloReplyShadow<'a> {
@@ -769,6 +815,8 @@ pub struct HelloReplyShadow<'a> {
     pub cluster_id: Cow<'a, str>,
     #[serde(default)]
     pub peers: Vec<String>,
+    #[serde(default)]
+    epoch: EpochShadow,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -776,6 +824,8 @@ pub struct SyncEnvelopeShadow<'a> {
     pub cluster_id: Cow<'a, str>,
     #[serde(with = "automerge_msg_codec")]
     pub message: automerge::sync::Message,
+    #[serde(default)]
+    epoch: EpochShadow,
 }
 
 // =====================================================================
@@ -808,12 +858,16 @@ 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 {})
+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),
+    })
 }
-pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, rmp_serde::decode::Error> {
-    let _: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
-    Ok(HelloRequest {})
+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)?,
+    })
 }
 
 // --- HelloReply ---
@@ -821,14 +875,16 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco
     let shadow = HelloReplyShadow {
         cluster_id: Cow::Borrowed(&msg.cluster_id),
         peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
+        epoch: epoch_to_shadow(&msg.epoch),
     };
     rmp_serde::to_vec(&shadow)
 }
-pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, rmp_serde::decode::Error> {
+pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> {
     let shadow: HelloReplyShadow = rmp_serde::from_slice(bytes)?;
     Ok(HelloReply {
         cluster_id: shadow.cluster_id.into_owned(),
         peers: shadow.peers,
+        epoch: epoch_from_shadow(shadow.epoch)?,
     })
 }
 
@@ -837,13 +893,15 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::
     let shadow = SyncEnvelopeShadow {
         cluster_id: Cow::Borrowed(&msg.cluster_id),
         message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules
+        epoch: epoch_to_shadow(&msg.epoch),
     };
     rmp_serde::to_vec(&shadow)
 }
-pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, rmp_serde::decode::Error> {
+pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportError> {
     let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
     Ok(SyncEnvelope {
         cluster_id: shadow.cluster_id.into_owned(),
+        epoch: epoch_from_shadow(shadow.epoch)?,
         message: shadow.message,
     })
 }