late messages don't flip peer's epoch back

2 files changed, 53 insertions, 12 deletions

diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index c5769ca..564aada 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -86,6 +86,8 @@ pub struct Keepalive {
     /// Only comparable between nodes on the same epoch.
     pub heads: HeadsSummary,
     pub epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub epoch_seq: u32,
     /// 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
@@ -131,6 +133,8 @@ impl std::fmt::Display for BlobId {
 pub struct BlobMessage {
     /// The sender's epoch (every message carries it).
     pub epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub epoch_seq: u32,
     pub id: BlobId,
     pub body: BlobBody,
 }
@@ -215,6 +219,17 @@ impl ParsedMessage {
             ParsedMessage::Blob(blob) => blob.epoch,
         }
     }
+
+    /// `SyncEnvelope::epoch_seq`, as every message carries it.
+    pub fn epoch_seq(&self) -> u32 {
+        match self {
+            ParsedMessage::Hello(hello) => hello.epoch_seq,
+            ParsedMessage::HelloReply(reply) => reply.epoch_seq,
+            ParsedMessage::Sync(envelope) => envelope.epoch_seq,
+            ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch_seq,
+            ParsedMessage::Blob(blob) => blob.epoch_seq,
+        }
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -223,6 +238,8 @@ pub struct HelloRequest {
     /// another swarm doesn't answer.
     pub swarm_id: Option<String>,
     pub epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub epoch_seq: u32,
     /// Proof of the swarm key (`swarm::auth`); empty without one.
     pub auth: Vec<u8>,
     /// The sender is ready (same meaning as `SyncEnvelope::ready`).
@@ -236,6 +253,8 @@ pub struct HelloReply {
     pub swarm_id: String,
     pub peers: Vec<String>,
     pub epoch: Epoch,
+    /// See `SyncEnvelope::epoch_seq`.
+    pub epoch_seq: u32,
     /// Proof of the swarm key (`swarm::auth`); empty without one.
     pub auth: Vec<u8>,
     /// The sender is ready.
@@ -246,6 +265,8 @@ pub struct HelloReply {
     /// 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>,
+    /// The sender's own run (as its HELLOs carry it): a new one means it restarted.
+    pub sender_run_id: RunId,
 }
 
 #[derive(Debug, Clone)]
@@ -253,6 +274,10 @@ pub struct SyncEnvelope {
     pub swarm_id: String,
     /// Epoch of the document the message was generated from.
     pub epoch: Epoch,
+    /// How many times the sender changed epoch in its current run, when it sent this
+    /// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver
+    /// ignores an epoch older than the newest it has seen in the same session.
+    pub epoch_seq: u32,
     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
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,