summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/document/replica.rs4
-rw-r--r--src/state.rs17
-rw-r--r--src/swarm/blob.rs8
-rw-r--r--src/swarm/peer.rs53
-rw-r--r--src/swarm/sync.rs7
-rw-r--r--src/transport/mod.rs25
-rw-r--r--src/transport/mycelium.rs40
7 files changed, 129 insertions, 25 deletions
diff --git a/src/document/replica.rs b/src/document/replica.rs
index e24f93c..db47223 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -86,6 +86,9 @@ impl fmt::Debug for Epoch {
/// [`Replica::heads`].
pub struct Replica {
pub epoch: Epoch,
+ /// How many times this node changed epoch (switch, rejoin, abandoned rejoin) in this
+ /// run: messages carry it, so a late one can't flip a peer's view of our epoch back.
+ pub epoch_seq: u32,
pub doc: AutoCommit,
heads: Vec<ChangeHash>,
}
@@ -94,6 +97,7 @@ impl Replica {
pub fn new(epoch: Epoch, doc: AutoCommit) -> Self {
let mut replica = Self {
epoch,
+ epoch_seq: 0,
doc,
heads: Vec::new(),
};
diff --git a/src/state.rs b/src/state.rs
index ce47673..d14c1ea 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -127,8 +127,14 @@ pub(crate) struct PeerState {
/// from it in between (None while they match).
pub(crate) diverged_since: Option<Instant>,
- /// Epoch the peer advertised in its most recent message.
+ /// Epoch the peer advertised in its most recent message (late ones aside: see
+ /// `epoch_seq`).
pub(crate) epoch: Option<Epoch>,
+ /// The peer's `epoch_seq` with `epoch`: how many times it changed epoch in its run.
+ /// An epoch with a lower one is older than what we have, delivered late.
+ pub(crate) epoch_seq: Option<u32>,
+ /// The peer's run, from its HELLO or HELLO_REPLY: a new one starts `epoch_seq` over.
+ pub(crate) run_id: Option<crate::transport::RunId>,
/// Whether the peer was ready according to its most recent sync message or keepalive.
pub(crate) reports_ready: Option<bool>,
@@ -330,6 +336,8 @@ impl PeerState {
last_resync_at: None,
diverged_since: None,
epoch: None,
+ epoch_seq: None,
+ run_id: None,
reports_ready: None,
syncs_with_us: None,
reported_heads: None,
@@ -676,6 +684,12 @@ impl AppState {
*self.epoch_tx.borrow()
}
+ /// The local epoch and `Replica::epoch_seq`, read together.
+ pub(crate) async fn epoch_stamp(&self) -> (Epoch, u32) {
+ let replica = self.doc.read().await;
+ (replica.epoch, replica.epoch_seq)
+ }
+
pub(crate) async fn is_ready(&self) -> bool {
matches!(*self.lifecycle.read().await, NodeLifecycle::Ready { .. })
}
@@ -1213,6 +1227,7 @@ impl Host for AppState {
let new_doc = f(&mut replica.doc)?;
let old_doc = std::mem::replace(&mut replica.doc, new_doc);
replica.epoch = epoch;
+ replica.epoch_seq += 1;
replica.refresh_heads();
// Updated under the same lock, so the cached epoch never disagrees
// with the document for anyone holding the lock.
diff --git a/src/swarm/blob.rs b/src/swarm/blob.rs
index 8c891ae..42cfbf3 100644
--- a/src/swarm/blob.rs
+++ b/src/swarm/blob.rs
@@ -171,8 +171,10 @@ async fn prepare(state: &AppState, id: BlobId) -> Option<Arc<Prepared>> {
}
async fn reply(state: &AppState, peer: &str, id: BlobId, body: BlobBody) {
+ let (epoch, epoch_seq) = state.epoch_stamp().await;
let message = ParsedMessage::Blob(BlobMessage {
- epoch: state.local_epoch(),
+ epoch,
+ epoch_seq,
id,
body,
});
@@ -223,8 +225,10 @@ impl Drop for Registration<'_> {
}
async fn request(state: &AppState, peer: &str, id: BlobId, chunks: Vec<u32>) {
+ let (epoch, epoch_seq) = state.epoch_stamp().await;
let message = ParsedMessage::Blob(BlobMessage {
- epoch: state.local_epoch(),
+ epoch,
+ epoch_seq,
id,
body: BlobBody::Request { chunks },
});
diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs
index daf1af4..8c1e7eb 100644
--- a/src/swarm/peer.rs
+++ b/src/swarm/peer.rs
@@ -110,11 +110,12 @@ pub(crate) async fn send_ping(state: &AppState, peer: &str) -> Result<()> {
pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::Keepalive {
// Heads and epoch come from the same locked replica, so they always match. The heads
// are cached (`Replica`), so the read lock is enough.
- let (heads, epoch) = {
+ let (heads, epoch, epoch_seq) = {
let replica = state.doc.read().await;
(
transport::HeadsSummary::of(replica.heads().to_vec()),
replica.epoch,
+ replica.epoch_seq,
)
};
let syncs_with_you = match state.get_peer_state(peer).await {
@@ -129,6 +130,7 @@ pub(crate) async fn local_keepalive(state: &AppState, peer: &str) -> transport::
transport::Keepalive {
heads,
epoch,
+ epoch_seq,
ready: state.is_ready().await,
syncs_with_you,
}
@@ -276,6 +278,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
.as_ref()
.map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer))
.unwrap_or_default();
+ let (epoch, epoch_seq) = state.epoch_stamp().await;
state.lock_rtt().hello_sent(peer, Instant::now());
let result = state
.transport
@@ -283,7 +286,8 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> {
peer,
&ParsedMessage::Hello(transport::HelloRequest {
swarm_id,
- epoch: state.local_epoch(),
+ epoch,
+ epoch_seq,
auth,
ready: state.is_ready().await,
run_id: state.run_id,
@@ -354,14 +358,17 @@ async fn send_hello_reply(
.map(|key| key.hello_reply_tag(&swarm_id, state.local_id(), peer))
.unwrap_or_default();
let ready = state.is_ready().await;
+ let (epoch, epoch_seq) = state.epoch_stamp().await;
let reply = transport::HelloReply {
swarm_id,
peers,
- epoch: state.local_epoch(),
+ epoch,
+ epoch_seq,
auth,
ready,
run_id,
away_ms: away.map(|away| away.as_millis() as u64),
+ sender_run_id: state.run_id,
};
debug!(to = %peer, ready, "Sending HELLO_REPLY");
@@ -582,9 +589,20 @@ pub(crate) async fn peer_swarm(state: &AppState, peer: &str) -> Option<String> {
// --- epochs ------------------------------------------------------------------
-/// Records the epoch a peer advertised and compares it with ours. Returns true when both
-/// are on the same epoch, the only case in which the documents may be synced.
-async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool {
+/// Records the epoch a peer advertised and compares it with ours. Returns true when the
+/// message's epoch is ours, the only case in which its documents may be synced.
+///
+/// `seq` is the peer's `epoch_seq` with it. A message with a lower one than we have was
+/// sent before the peer's last epoch change and delivered late: its epoch isn't
+/// recorded, so it can't flip our view of the peer back. The count starts over when the
+/// peer restarts, which its HELLO or HELLO_REPLY tells by a new run (`run`).
+async fn observe_peer_epoch(
+ state: &AppState,
+ peer: &str,
+ remote: Epoch,
+ seq: u32,
+ run: Option<transport::RunId>,
+) -> bool {
let local = state.local_epoch();
let relation = local.relation_to(&remote);
@@ -594,6 +612,21 @@ async fn observe_peer_epoch(state: &AppState, peer: &str, remote: Epoch) -> bool
let previous = {
let mut peer_state = peer_state_mutex.lock().await;
+ if let Some(run) = run.filter(|run| peer_state.run_id != Some(*run)) {
+ peer_state.run_id = Some(run);
+ peer_state.epoch_seq = None;
+ }
+ if peer_state.epoch_seq.is_some_and(|newest| seq < newest) {
+ debug!(
+ peer = %peer,
+ epoch = %remote,
+ seq,
+ newest = ?peer_state.epoch_seq,
+ "Ignoring a late message's epoch"
+ );
+ return relation == EpochRelation::Same;
+ }
+ peer_state.epoch_seq = Some(seq);
let previous = peer_state.epoch.replace(remote);
if previous.is_some_and(|previous| previous != remote) {
// The peer switched documents; our sync state describes its old one. Count
@@ -758,7 +791,13 @@ pub(crate) async fn handle_message(
info!(peer = %peer, "Dormant peer is active again; re-handshaking");
}
- let same_epoch = observe_peer_epoch(state, &peer, message.epoch()).await;
+ let run = match &message {
+ transport::ParsedMessage::Hello(hello) => Some(hello.run_id),
+ transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id),
+ _ => None,
+ };
+ let same_epoch =
+ observe_peer_epoch(state, &peer, message.epoch(), message.epoch_seq(), run).await;
//debug!(from = %peer, message = ?message, "Received message");
diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs
index 80d6d56..93af6f7 100644
--- a/src/swarm/sync.rs
+++ b/src/swarm/sync.rs
@@ -141,10 +141,10 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
tokio::task::spawn_blocking(move || -> Result<Option<transport::EncodedMessage>> {
// Label the message with the epoch of the document it was generated
// from, read under the same lock.
- let (epoch, message) = {
+ let (epoch, epoch_seq, message) = {
let mut replica = doc.blocking_write();
let mut peer_state = psm.blocking_lock();
- let epoch = replica.epoch;
+ let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
let sent_before = peer_state.sync.state.sent_hashes.clone();
let message = {
let sync_doc = replica.doc.sync();
@@ -178,7 +178,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
regenerate: false,
});
}
- (epoch, message)
+ (epoch, epoch_seq, message)
};
let Some(message) = message else {
return Ok(None);
@@ -186,6 +186,7 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str) -> Result<()> {
let envelope = transport::SyncEnvelope {
swarm_id,
epoch,
+ epoch_seq,
message,
ready,
};
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,
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-26 12:40:24 +0000