late messages don't flip peer's epoch back

1 files changed, 28 insertions, 12 deletions

diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs
index 82e6016..f519928 100644
--- a/src/transport/mycelium.rs
+++ b/src/transport/mycelium.rs
@@ -802,16 +802,20 @@ fn format_message(
 struct EpochShadow {
     generation: u64,
     id: ByteBuf,
+    /// The sender's epoch changes in its current run (`SyncEnvelope::epoch_seq`).
+    #[serde(default)]
+    seq: u32,
 }
 
-fn epoch_to_shadow(epoch: &Epoch) -> EpochShadow {
+fn epoch_to_shadow(epoch: &Epoch, seq: u32) -> EpochShadow {
     EpochShadow {
         generation: epoch.generation,
         id: ByteBuf::from(epoch.id.to_vec()),
+        seq,
     }
 }
 
-fn epoch_from_shadow(shadow: EpochShadow) -> Result<Epoch, TransportError> {
+fn epoch_from_shadow(shadow: &EpochShadow) -> Result<Epoch, TransportError> {
     if shadow.generation == 0 && shadow.id.is_empty() {
         return Ok(Epoch::GENESIS);
     }
@@ -850,7 +854,7 @@ fn keepalive_to_bytes(msg: &Keepalive) -> Result<Vec<u8>, rmp_serde::encode::Err
     rmp_serde::to_vec(&KeepaliveShadow {
         heads,
         digest,
-        epoch: epoch_to_shadow(&msg.epoch),
+        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
         ready: msg.ready,
         syncs_with_you: msg.syncs_with_you,
     })
@@ -880,7 +884,8 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> {
     };
     Ok(Keepalive {
         heads,
-        epoch: epoch_from_shadow(shadow.epoch)?,
+        epoch: epoch_from_shadow(&shadow.epoch)?,
+        epoch_seq: shadow.epoch.seq,
         ready: shadow.ready,
         syncs_with_you: shadow.syncs_with_you,
     })
@@ -908,6 +913,7 @@ pub struct HelloReplyShadow<'a> {
     ready: bool,
     run_id: ByteBuf,
     away_ms: Option<u64>,
+    sender_run_id: ByteBuf,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -955,7 +961,7 @@ mod automerge_msg_codec {
 pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> {
     rmp_serde::to_vec(&HelloRequestShadow {
         swarm_id: msg.swarm_id.clone(),
-        epoch: epoch_to_shadow(&msg.epoch),
+        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
         auth: ByteBuf::from(msg.auth.clone()),
         ready: msg.ready,
         run_id: ByteBuf::from(msg.run_id.to_vec()),
@@ -965,7 +971,8 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportErr
     let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?;
     Ok(HelloRequest {
         swarm_id: shadow.swarm_id,
-        epoch: epoch_from_shadow(shadow.epoch)?,
+        epoch: epoch_from_shadow(&shadow.epoch)?,
+        epoch_seq: shadow.epoch.seq,
         auth: shadow.auth.into_vec(),
         ready: shadow.ready,
         run_id: run_id_from_shadow(&shadow.run_id)?,
@@ -984,11 +991,12 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco
     let shadow = HelloReplyShadow {
         swarm_id: Cow::Borrowed(&msg.swarm_id),
         peers: msg.peers.clone(), // Vec must be cloned for serialization wrapper matching
-        epoch: epoch_to_shadow(&msg.epoch),
+        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
         auth: ByteBuf::from(msg.auth.clone()),
         ready: msg.ready,
         run_id: ByteBuf::from(msg.run_id.to_vec()),
         away_ms: msg.away_ms,
+        sender_run_id: ByteBuf::from(msg.sender_run_id.to_vec()),
     };
     rmp_serde::to_vec(&shadow)
 }
@@ -997,11 +1005,13 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError>
     Ok(HelloReply {
         swarm_id: shadow.swarm_id.into_owned(),
         peers: shadow.peers,
-        epoch: epoch_from_shadow(shadow.epoch)?,
+        epoch: epoch_from_shadow(&shadow.epoch)?,
+        epoch_seq: shadow.epoch.seq,
         auth: shadow.auth.into_vec(),
         ready: shadow.ready,
         run_id: run_id_from_shadow(&shadow.run_id)?,
         away_ms: shadow.away_ms,
+        sender_run_id: run_id_from_shadow(&shadow.sender_run_id)?,
     })
 }
 
@@ -1010,7 +1020,7 @@ pub fn sync_envelope_to_bytes(msg: &SyncEnvelope) -> Result<Vec<u8>, rmp_serde::
     let shadow = SyncEnvelopeShadow {
         swarm_id: Cow::Borrowed(&msg.swarm_id),
         message: msg.message.clone(), // Clone required by automerge_msg_codec ownership rules
-        epoch: epoch_to_shadow(&msg.epoch),
+        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
         ready: msg.ready,
     };
     rmp_serde::to_vec(&shadow)
@@ -1019,7 +1029,8 @@ pub fn bytes_to_sync_envelope(bytes: &[u8]) -> Result<SyncEnvelope, TransportErr
     let shadow: SyncEnvelopeShadow = rmp_serde::from_slice(bytes)?;
     Ok(SyncEnvelope {
         swarm_id: shadow.swarm_id.into_owned(),
-        epoch: epoch_from_shadow(shadow.epoch)?,
+        epoch: epoch_from_shadow(&shadow.epoch)?,
+        epoch_seq: shadow.epoch.seq,
         message: shadow.message,
         ready: shadow.ready,
     })
@@ -1084,7 +1095,7 @@ fn blob_to_bytes(msg: &BlobMessage) -> Result<Vec<u8>, rmp_serde::encode::Error>
         BlobBody::Missing => BlobBodyShadow::Missing,
     };
     rmp_serde::to_vec(&BlobShadow {
-        epoch: epoch_to_shadow(&msg.epoch),
+        epoch: epoch_to_shadow(&msg.epoch, msg.epoch_seq),
         id,
         body,
     })
@@ -1123,7 +1134,8 @@ fn bytes_to_blob(bytes: &[u8]) -> Result<BlobMessage, TransportError> {
         BlobBodyShadow::Missing => BlobBody::Missing,
     };
     Ok(BlobMessage {
-        epoch: epoch_from_shadow(shadow.epoch)?,
+        epoch: epoch_from_shadow(&shadow.epoch)?,
+        epoch_seq: shadow.epoch.seq,
         id,
         body,
     })
@@ -1181,6 +1193,7 @@ mod tests {
                 generation: u64::MAX,
                 id: [0xff; 32],
             },
+            epoch_seq: u32::MAX,
             ready: true,
             syncs_with_you: true,
         };
@@ -1203,6 +1216,7 @@ mod tests {
         let id = BlobId::Change(ChangeHash([9; 32]));
         let chunk = BlobMessage {
             epoch,
+            epoch_seq: 4,
             id,
             body: BlobBody::Chunk {
                 index: 5,
@@ -1215,12 +1229,14 @@ mod tests {
         let decoded = bytes_to_blob(&bytes).unwrap();
         assert_eq!(decoded.id, id);
         assert_eq!(decoded.epoch, epoch);
+        assert_eq!(decoded.epoch_seq, 4);
         assert!(
             matches!(decoded.body, BlobBody::Chunk { index: 5, ref data } if data.len() == 64 * 1024)
         );
 
         let manifest = BlobMessage {
             epoch,
+            epoch_seq: 4,
             id,
             body: BlobBody::Manifest {
                 size: 100,