aboutsummaryrefslogtreecommitdiffstats
path: root/src/transport/mod.rs
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-11-05 03:50:38 +0100
committertv <tv@krebsco.de>2026-11-05 03:50:38 +0100
commita02ed87abf6a0dc9d3f36bafcf333272586093aa (patch)
treee22e2f58703d3eac74df8e0eccc51c47b2f45133 /src/transport/mod.rs
parenta9fb4e6890bfaef1e25cdceb4eea9c997e3c7cdf (diff)
epoch tag in sync
Diffstat (limited to 'src/transport/mod.rs')
-rw-r--r--src/transport/mod.rs69
1 files changed, 64 insertions, 5 deletions
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 3fd4490..32be29b 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -389,12 +389,14 @@ impl ParsedMessage {
}
/// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them
- /// but WHO, ME, PING and PONG (which the node never sees).
+ /// but WHO, ME, PING and PONG (which the node never sees), and SYNC, which carries a
+ /// tag of it only (`EpochTag`).
pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> {
match self {
ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)),
ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)),
- ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)),
+ // Only a tag (`EpochTag`): resolved by the receiver, which knows the epochs.
+ ParsedMessage::Sync(_) => None,
ParsedMessage::Status(status) | ParsedMessage::StatusReply(status) => {
Some((status.epoch, status.epoch_seq))
}
@@ -656,9 +658,9 @@ impl std::fmt::Debug for HelloReply {
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct SyncEnvelope {
- /// Epoch of the document the message was generated from.
- #[serde(with = "crate::transport::wire::codec::epoch")]
- pub(crate) epoch: Epoch,
+ /// Epoch of the document the message was generated from, by a tag (`EpochTag`): the
+ /// receiver knows the epoch it names, as ours or as the sender's.
+ pub(crate) epoch: EpochTag,
/// 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.
@@ -682,6 +684,46 @@ pub(crate) struct SyncEnvelope {
pub(crate) sent: crate::clock::UnixMillis,
}
+/// An epoch, as sync messages name it: its generation and the first bytes of its ID,
+/// about a third of the whole. Sync messages go often, and only between peers on one
+/// epoch, which each knows from the other's handshake and keepalives: the tag tells which
+/// of the epochs the receiver knows it is, and one it knows of no other is dropped (an
+/// epoch of the same generation agreeing in 8 bytes, but not the rest, would take
+/// competing remakes that also collide there).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct EpochTag {
+ pub(crate) generation: u64,
+ #[serde(with = "crate::transport::wire::codec::bytes8")]
+ pub(crate) prefix: [u8; 8],
+}
+
+impl EpochTag {
+ pub(crate) fn of(epoch: Epoch) -> Self {
+ let mut prefix = [0; 8];
+ prefix.copy_from_slice(&epoch.id[..8]);
+ Self {
+ generation: epoch.generation,
+ prefix,
+ }
+ }
+
+ /// Whether this tags `epoch`.
+ pub(crate) fn names(&self, epoch: &Epoch) -> bool {
+ *self == Self::of(*epoch)
+ }
+
+ /// The first of `known` it names, if any.
+ pub(crate) fn resolve(&self, known: impl IntoIterator<Item = Epoch>) -> Option<Epoch> {
+ known.into_iter().find(|epoch| self.names(epoch))
+ }
+}
+
+impl std::fmt::Display for EpochTag {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ write!(f, "{}:{}…", self.generation, Hex(&self.prefix))
+ }
+}
+
/// What a sync envelope carries.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) enum SyncBody {
@@ -1042,4 +1084,21 @@ mod tests {
}
assert_eq!(seen.len(), heads.len());
}
+
+ /// A tag names its epoch, not one of another generation or another ID, and resolves
+ /// to the epoch it names among those known.
+ #[test]
+ fn epoch_tags_name_their_epoch() {
+ let epoch = |generation, byte| Epoch {
+ generation,
+ id: [byte; 32],
+ };
+ let ours = epoch(3, 1);
+ let tag = EpochTag::of(ours);
+ assert!(tag.names(&ours));
+ assert!(!tag.names(&epoch(4, 1)), "another generation");
+ assert!(!tag.names(&epoch(3, 2)), "another remake of it");
+ assert_eq!(tag.resolve([epoch(3, 2), ours]), Some(ours));
+ assert_eq!(tag.resolve([epoch(2, 1)]), None);
+ }
}