1 files changed, 64 insertions, 0 deletions

diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index c5769ca..35e3499 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -11,6 +11,8 @@ pub use crate::document::replica::Epoch;
 
 pub mod firewall;
 pub mod mycelium;
+#[cfg(test)]
+pub(crate) mod sim;
 pub mod traffic;
 
 /// Random identifier of one run of a node (from start to exit).
@@ -86,6 +88,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 +135,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 +221,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 +240,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 +255,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 +267,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 +276,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
@@ -301,3 +328,40 @@ pub trait Transport: Send + Sync {
     /// Asynchronously monitors for fatal network client or sub-process failures.
     async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus>;
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::testrand::for_seeds;
+
+    /// Two summaries are equal exactly when the heads are, in any order, whichever form
+    /// (listed or digest) each took.
+    #[test]
+    fn heads_summary_equality_is_set_equality() {
+        for_seeds(300, |rng| {
+            let n = rng.between(0, 2 * MAX_LISTED_HEADS + 2);
+            let heads: Vec<ChangeHash> = (0..n).map(|_| ChangeHash(rng.bytes32())).collect();
+            let summary = HeadsSummary::of(heads.clone());
+            assert_eq!(summary.heads().is_some(), n <= MAX_LISTED_HEADS);
+
+            let mut shuffled = heads.clone();
+            rng.shuffle(&mut shuffled);
+            assert_eq!(summary, HeadsSummary::of(shuffled));
+            assert_eq!(summary, HeadsSummary::Digest(summary.digest()));
+
+            let mut other = heads.clone();
+            match rng.below(3) {
+                1 if !other.is_empty() => {
+                    let i = rng.below(other.len());
+                    other.remove(i);
+                }
+                2 if !other.is_empty() => {
+                    let i = rng.below(other.len());
+                    other[i] = ChangeHash(rng.bytes32());
+                }
+                _ => other.push(ChangeHash(rng.bytes32())),
+            }
+            assert_ne!(summary, HeadsSummary::of(other));
+        });
+    }
+}