diff --git a/docs/federation.md b/docs/federation.md index f05a670..3e3ae22 100644 --- a/docs/federation.md +++ b/docs/federation.md @@ -23,16 +23,19 @@ Each node owns an append-only event log. The log is split into two scopes: Nodes sync via anti-entropy: each advertises a per-origin version vector, and peers send whatever Global entries the other is missing. Delivery is idempotent, -so a re-sent entry is dropped, and each account/channel event is an idempotent -upsert. A peer converges by accepting any entry newer than its version vector -(including the forward jump a compacted node's log begins at). +so a re-sent entry is dropped, and each account event is an idempotent upsert. -> **Known limitation (3+ nodes).** In a relay mesh, an entry can reach a node -> ahead of its predecessors via a longer path, and the current ingest advances the -> version vector past that gap. With two nodes (direct delivery) this never -> happens. Closing it for larger meshes needs a snapshot-epoch marker so a -> compaction jump is distinguishable from a live gap; until then, prefer 2-node -> topologies. This is dormant today (no peers). +Entries are applied **strictly in order per origin**, tracked by a `(epoch, seq)` +cursor. Each node's seq increments per event; a **compaction** bumps that node's +`epoch` and resets its seq to 0. A receiver therefore distinguishes the two forward +jumps cleanly: + +- **A relay gap** (same epoch, a seq ahead of the receiver's) is *deferred* — the + cursor doesn't advance, so the digest re-requests the missing seqs in order. This + keeps a 3+-node relay mesh from silently dropping updates. +- **A compaction** (a higher epoch beginning at seq 0) is *adopted* — the receiver + takes the reset as the new authoritative baseline, so a peer converges across a + compaction even if it was mid-sync. ## The trust model @@ -80,9 +83,9 @@ Rules: all account data are on that wire. Echo warns loudly at startup if TLS is off. - **Firewall** the `bind` port to your peer nodes' IPs only. - Use a strong random `secret` and rotate it periodically. -- **Prefer 2 nodes.** Direct delivery keeps replication in order; a 3+-node relay - mesh has a known out-of-order edge (see "Known limitation" above) that isn't - closed yet. +- 3+-node relay meshes converge correctly (out-of-order delivery is deferred and + re-requested; compaction is handled by the epoch cursor — see "How gossip + works"). The 2-node path is still the most exercised, so test your topology. ## Tier C — cross-operator federation (per-origin signing) diff --git a/src/engine/db/mod.rs b/src/engine/db/mod.rs index 7a6b156..0b60e88 100644 --- a/src/engine/db/mod.rs +++ b/src/engine/db/mod.rs @@ -740,6 +740,16 @@ impl ChannelInfo { // and order it — the origin node, its per-node sequence (`origin:seq` is the // cluster-unique id), and a Lamport clock for causal ordering across nodes. // Lines written before these existed default them in. +fn is_zero(n: &u64) -> bool { + *n == 0 +} + +// Order two per-origin cursors: is `incoming` strictly newer than `have`? Epoch +// dominates (a compaction bumps it), then seq within an epoch. +fn cursor_newer(have: (u64, u64), incoming: (u64, u64)) -> bool { + incoming.0 > have.0 || (incoming.0 == have.0 && incoming.1 > have.1) +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct LogEntry { #[serde(default)] @@ -748,6 +758,12 @@ pub struct LogEntry { seq: u64, #[serde(default)] lamport: u64, + // Compaction generation for `origin`. Bumped each time that node compacts, which + // resets its seq to 0; lets a peer tell a compaction's forward jump (higher epoch, + // seq 0) from a live relay gap (same epoch, seq ahead). Omitted while 0, so a + // never-compacted / single-node log is byte-identical to before this field existed. + #[serde(default, skip_serializing_if = "is_zero")] + epoch: u64, // Optional Ed25519 signature (base64) over the entry, for Tier C federation. When // signing is off it's None and `skip_serializing_if` omits it entirely, so the log // format is byte-identical to an unsigned deployment. Must precede the flattened @@ -761,7 +777,10 @@ pub struct LogEntry { #[cfg(test)] impl LogEntry { pub(crate) fn for_test(origin: &str, seq: u64, lamport: u64, event: Event) -> Self { - LogEntry { origin: origin.to_string(), seq, lamport, sig: None, event } + LogEntry { origin: origin.to_string(), seq, lamport, epoch: 0, sig: None, event } + } + pub(crate) fn for_test_epoch(origin: &str, epoch: u64, seq: u64, lamport: u64, event: Event) -> Self { + LogEntry { origin: origin.to_string(), seq, lamport, epoch, sig: None, event } } } @@ -792,7 +811,8 @@ pub struct EventLog { path: PathBuf, origin: String, lamport: u64, // logical clock, ticked on every event - versions: HashMap, // per-origin highest seq applied (version vector) + epoch: u64, // our origin's current compaction generation + versions: HashMap, // per-origin (epoch, seq) cursor (version vector) entries: Vec, // full log, kept so peers can pull what they lack outbound: Option>, // push newly committed entries to peers // Operator lockdown (OperServ SET READONLY): while set, locally-authored @@ -810,7 +830,7 @@ pub struct EventLog { impl EventLog { fn open(path: PathBuf, origin: String) -> (Self, Vec) { - let mut log = Self { path, origin, lamport: 0, versions: HashMap::new(), entries: Vec::new(), outbound: None, readonly: false, file: None, signing: None }; + let mut log = Self { path, origin, lamport: 0, epoch: 0, versions: HashMap::new(), entries: Vec::new(), outbound: None, readonly: false, file: None, signing: None }; if let Ok(data) = std::fs::read_to_string(&log.path) { for line in data.lines().filter(|l| !l.trim().is_empty()) { match serde_json::from_str::(line) { @@ -829,16 +849,35 @@ impl EventLog { // Roll the clock and version vector forward over an entry. Local (channel) // entries carry no gossip identity, so they never touch the vector or clock. fn absorb(&mut self, entry: &LogEntry) { + // Recover our compaction epoch from ANY of our own entries, including Local + // ones — a compaction whose snapshot carried no Global events (a node with + // only channels, or with external accounts) would otherwise lose the epoch on + // restart, reset our seqs, and permanently diverge subsequent writes from peers. + if entry.origin == self.origin { + self.epoch = self.epoch.max(entry.epoch); + } if entry.event.scope() != Scope::Global { return; } self.lamport = self.lamport.max(entry.lamport); - self.versions.entry(entry.origin.clone()).and_modify(|s| *s = (*s).max(entry.seq)).or_insert(entry.seq); + let cur = (entry.epoch, entry.seq); + self.versions + .entry(entry.origin.clone()) + .and_modify(|c| { + if cursor_newer(*c, cur) { + *c = cur; + } + }) + .or_insert(cur); } - // Seq the next locally-authored event will carry (0-based, per our origin). + // Seq the next locally-authored event will carry (0-based within the current + // epoch, per our origin). Compaction resets this to 0 by bumping the epoch. fn next_seq(&self) -> u64 { - self.versions.get(&self.origin).map_or(0, |s| s + 1) + match self.versions.get(&self.origin) { + Some(&(e, s)) if e == self.epoch => s + 1, + _ => 0, + } } // Persist a locally-authored event. Global events get the next seq + a ticked @@ -853,10 +892,10 @@ impl EventLog { let entry = if global { self.lamport = self.lamport.saturating_add(1); let seq = self.next_seq(); - let sig = self.signing.as_ref().map(|s| s.sign(&self.origin, seq, self.lamport, &event)); - LogEntry { origin: self.origin.clone(), seq, lamport: self.lamport, sig, event } + let sig = self.signing.as_ref().map(|s| s.sign(&self.origin, self.epoch, seq, self.lamport, &event)); + LogEntry { origin: self.origin.clone(), seq, lamport: self.lamport, epoch: self.epoch, sig, event } } else { - LogEntry { origin: self.origin.clone(), seq: 0, lamport: 0, sig: None, event } + LogEntry { origin: self.origin.clone(), seq: 0, lamport: 0, epoch: 0, sig: None, event } }; if let Err(e) = self.persist(&entry) { // Surface the real cause (disk full, permissions, ...) here at the @@ -865,7 +904,7 @@ impl EventLog { return Err(e); } if global { - self.versions.insert(entry.origin.clone(), entry.seq); + self.versions.insert(entry.origin.clone(), (entry.epoch, entry.seq)); } // Push every committed entry to subscribers: the gossip forwarder filters to // Global before sending to a peer, but the gRPC directory stream wants the @@ -890,24 +929,28 @@ impl EventLog { // signature from a trusted origin key, or it's rejected before it can touch // the version vector. Inert when signing is off (flat-trust / Tier B). if let Some(s) = &self.signing { - if !s.verify(&entry.origin, entry.seq, entry.lamport, &entry.event, entry.sig.as_deref()) { + if !s.verify(&entry.origin, entry.epoch, entry.seq, entry.lamport, &entry.event, entry.sig.as_deref()) { return Ok(None); } } - // Accept any entry newer than what we hold for this origin. A forward jump is - // legitimate here: a peer syncing from a COMPACTED node receives seqs that - // start well above 0 (the intermediate ones were folded into the snapshot), - // and the fold is an idempotent upsert. NOTE: this cannot detect an out-of- - // order relay in a 3+-node mesh (an entry delivered ahead of its predecessors - // advances the version past the gap). Fixing that safely needs a snapshot-epoch - // marker so a compaction jump is distinguishable from a live gap — deferred; - // see docs/federation.md. Not reachable today (no gossip peers). - if self.versions.get(&entry.origin).is_some_and(|&s| entry.seq <= s) { - return Ok(None); // already have it + // Epoch-aware, strictly in-order acceptance. This distinguishes a COMPACTION's + // forward jump (a higher epoch beginning at seq 0 — accept and adopt the reset) + // from a live RELAY GAP in a 3+-node mesh (same epoch, a seq ahead of ours — + // drop, so the version vector never advances past the gap and the digest + // re-requests it in order). An entry from a superseded older epoch is dropped. + let (e, s) = (entry.epoch, entry.seq); + let accept = match self.versions.get(&entry.origin).copied() { + None => s == 0, // a new origin: start only from seq 0 + Some((eb, sb)) if e == eb => s == sb + 1, // contiguous within the epoch + Some((eb, _)) if e > eb => s == 0, // a compaction: adopt its seq-0 reset + Some(_) => false, // older epoch, already superseded + }; + if !accept { + return Ok(None); } self.persist(&entry)?; self.lamport = self.lamport.max(entry.lamport).saturating_add(1); // Lamport receive rule - self.versions.insert(entry.origin.clone(), entry.seq); + self.versions.insert(entry.origin.clone(), (e, s)); let event = entry.event.clone(); self.notify(&entry); self.entries.push(entry); @@ -938,17 +981,20 @@ impl EventLog { } // Our version vector: highest seq applied per origin. - fn version_vector(&self) -> HashMap { + fn version_vector(&self) -> HashMap { self.versions.clone() } - // Global entries a peer is missing, given the version vector it advertised. + // Global entries a peer is missing, given the (epoch, seq) cursor it advertised. // Local (channel) entries are never offered — they don't leave the node. - fn missing_for(&self, peer: &HashMap) -> Vec { + fn missing_for(&self, peer: &HashMap) -> Vec { self.entries .iter() .filter(|e| e.event.scope() == Scope::Global) - .filter(|e| peer.get(&e.origin).is_none_or(|&s| e.seq > s)) + .filter(|e| match peer.get(&e.origin) { + None => true, + Some(&h) => cursor_newer(h, (e.epoch, e.seq)), + }) .cloned() .collect() } @@ -987,20 +1033,32 @@ impl EventLog { // theirs on the next sync. Written to a temp file and renamed, so a crash // leaves the old log. fn compact(&mut self, events: Vec) -> std::io::Result<()> { - let mut seq = self.next_seq(); + // An empty snapshot means no state to preserve; skip so the bumped epoch is + // never written onto an empty file (where a restart couldn't recover it). + if events.is_empty() { + return Ok(()); + } + // A new compaction generation: bump the epoch and restart seqs at 0. A peer + // sees the higher epoch beginning at seq 0 and adopts it (accepting the jump) + // rather than mistaking it for a live gap. + self.epoch = self.epoch.saturating_add(1); + let mut seq = 0u64; let mut last_global = None; let mut snapshot = Vec::with_capacity(events.len()); for event in events { let entry = if event.scope() == Scope::Global { self.lamport = self.lamport.saturating_add(1); - // Compaction re-seqs entries, so any signature must be recomputed. - let sig = self.signing.as_ref().map(|s| s.sign(&self.origin, seq, self.lamport, &event)); - let e = LogEntry { origin: self.origin.clone(), seq, lamport: self.lamport, sig, event }; + // Compaction re-seqs (and re-epochs) entries, so any signature must be recomputed. + let sig = self.signing.as_ref().map(|s| s.sign(&self.origin, self.epoch, seq, self.lamport, &event)); + let e = LogEntry { origin: self.origin.clone(), seq, lamport: self.lamport, epoch: self.epoch, sig, event }; last_global = Some(seq); seq += 1; e } else { - LogEntry { origin: self.origin.clone(), seq: 0, lamport: 0, sig: None, event } + // Stamp Local entries with the epoch too, so it survives a restart even + // when the snapshot carries no Global events (Local entries never gossip, + // so this only feeds `absorb`'s epoch recovery). + LogEntry { origin: self.origin.clone(), seq: 0, lamport: 0, epoch: self.epoch, sig: None, event } }; snapshot.push(entry); } @@ -1018,7 +1076,7 @@ impl EventLog { self.file = None; self.versions = HashMap::new(); if let Some(last) = last_global { - self.versions.insert(self.origin.clone(), last); + self.versions.insert(self.origin.clone(), (self.epoch, last)); } self.entries = snapshot; Ok(()) @@ -1453,13 +1511,13 @@ impl Db { self.log.events_from(mark) } - /// Our version vector, advertised to peers so they can send what we lack. - pub fn version_vector(&self) -> HashMap { + /// Our version vector — a per-origin (epoch, seq) cursor — advertised to peers. + pub fn version_vector(&self) -> HashMap { self.log.version_vector() } /// The log entries a peer is missing, given the version vector it sent. - pub fn missing_for(&self, peer: &HashMap) -> Vec { + pub fn missing_for(&self, peer: &HashMap) -> Vec { self.log.missing_for(peer) } diff --git a/src/engine/db/sign.rs b/src/engine/db/sign.rs index 7085aad..85edfba 100644 --- a/src/engine/db/sign.rs +++ b/src/engine/db/sign.rs @@ -15,10 +15,11 @@ use super::Event; // The bytes a signature covers: everything that fixes the entry's identity and // content. `serde_json` is deterministic for the Global event set (all `Vec`/scalar // fields — no unordered maps), so sender and receiver derive the same bytes. -fn payload(origin: &str, seq: u64, lamport: u64, event: &Event) -> Vec { +fn payload(origin: &str, epoch: u64, seq: u64, lamport: u64, event: &Event) -> Vec { let mut v = Vec::new(); v.extend_from_slice(origin.as_bytes()); v.push(0); + v.extend_from_slice(&epoch.to_le_bytes()); v.extend_from_slice(&seq.to_le_bytes()); v.extend_from_slice(&lamport.to_le_bytes()); v.push(0); @@ -51,19 +52,19 @@ impl Signing { } // Sign an entry we're authoring; base64 of the 64-byte signature. - pub fn sign(&self, origin: &str, seq: u64, lamport: u64, event: &Event) -> String { - B64.encode(self.signer.sign(&payload(origin, seq, lamport, event)).to_bytes()) + pub fn sign(&self, origin: &str, epoch: u64, seq: u64, lamport: u64, event: &Event) -> String { + B64.encode(self.signer.sign(&payload(origin, epoch, seq, lamport, event)).to_bytes()) } // True if `sig` is a valid signature over the entry by the trusted key for its // origin. False if the origin isn't trusted, the signature is missing, or it // doesn't verify (`verify_strict` also rejects malleable/degenerate signatures). - pub fn verify(&self, origin: &str, seq: u64, lamport: u64, event: &Event, sig: Option<&str>) -> bool { + pub fn verify(&self, origin: &str, epoch: u64, seq: u64, lamport: u64, event: &Event, sig: Option<&str>) -> bool { let Some(vk) = self.trust.get(origin) else { return false }; let Some(sig) = sig else { return false }; let Ok(raw) = B64.decode(sig.trim()) else { return false }; let Ok(bytes) = <[u8; 64]>::try_from(raw.as_slice()) else { return false }; - vk.verify_strict(&payload(origin, seq, lamport, event), &Signature::from_bytes(&bytes)).is_ok() + vk.verify_strict(&payload(origin, epoch, seq, lamport, event), &Signature::from_bytes(&bytes)).is_ok() } } @@ -88,13 +89,14 @@ mod tests { let trust = HashMap::from([("A".to_string(), pubk)]); let s = Signing::new(&sec, &trust).unwrap(); - let sig = s.sign("A", 3, 4, &ev()); - assert!(s.verify("A", 3, 4, &ev(), Some(&sig)), "a genuine signature verifies"); + let sig = s.sign("A", 5, 3, 4, &ev()); + assert!(s.verify("A", 5, 3, 4, &ev(), Some(&sig)), "a genuine signature verifies"); // Any change to the covered fields invalidates it. - assert!(!s.verify("A", 4, 4, &ev(), Some(&sig)), "a different seq is rejected"); - assert!(!s.verify("A", 3, 4, &Event::AccountDropped { account: "bob".into() }, Some(&sig)), "a different event is rejected"); - assert!(!s.verify("A", 3, 4, &ev(), None), "a missing signature is rejected"); - assert!(!s.verify("B", 3, 4, &ev(), Some(&sig)), "an untrusted origin is rejected"); + assert!(!s.verify("A", 5, 4, 4, &ev(), Some(&sig)), "a different seq is rejected"); + assert!(!s.verify("A", 6, 3, 4, &ev(), Some(&sig)), "a different epoch is rejected"); + assert!(!s.verify("A", 5, 3, 4, &Event::AccountDropped { account: "bob".into() }, Some(&sig)), "a different event is rejected"); + assert!(!s.verify("A", 5, 3, 4, &ev(), None), "a missing signature is rejected"); + assert!(!s.verify("B", 5, 3, 4, &ev(), Some(&sig)), "an untrusted origin is rejected"); } #[test] @@ -104,7 +106,7 @@ mod tests { // We trust A's key; B tries to forge an entry claiming origin A. let s = Signing::new(&sec_a, &HashMap::from([("A".to_string(), pub_a)])).unwrap(); let forger = Signing::new(&sec_b, &HashMap::new()).unwrap(); - let forged = forger.sign("A", 1, 1, &ev()); - assert!(!s.verify("A", 1, 1, &ev(), Some(&forged)), "a signature from the wrong key is rejected"); + let forged = forger.sign("A", 0, 1, 1, &ev()); + assert!(!s.verify("A", 0, 1, 1, &ev(), Some(&forged)), "a signature from the wrong key is rejected"); } } diff --git a/src/engine/db/tests.rs b/src/engine/db/tests.rs index f47f3ec..8dce71f 100644 --- a/src/engine/db/tests.rs +++ b/src/engine/db/tests.rs @@ -84,7 +84,7 @@ hide_status: false, snotice: false, vhost: None, vhost_request: None, last_seen: ts, noexpire: false, expiry_warned: false, oper_note: None, }; - let reg = |origin: &str, a: Account| LogEntry::for_test(origin, 1, 1, Event::AccountRegistered(Box::new(a))); + let reg = |origin: &str, a: Account| LogEntry::for_test(origin, 0, 1, Event::AccountRegistered(Box::new(a))); // Same account, re-homed to B with the SAME verifier — not a takeover. let mut db = Db::open(tmp("rehome-same"), "A"); @@ -108,7 +108,7 @@ db.register_channel("#c", "alice").unwrap(); // local db.set_mlock("#c", "nt", "").unwrap(); // local // Only the account advances the version vector. - assert_eq!(db.version_vector().get("N1"), Some(&0), "channels don't advance the vector"); + assert_eq!(db.version_vector().get("N1"), Some(&(0, 0)), "channels don't advance the vector"); // A fresh peer is offered the account, never the channel. let missing = db.missing_for(&HashMap::new()); assert_eq!(missing.len(), 1, "only the global account entry is offered: {missing:?}"); @@ -118,7 +118,7 @@ let db = Db::open(&path, "N1"); assert!(db.exists("alice")); assert_eq!(db.channel("#c").map(|c| c.lock_on.clone()), Some("nt".to_string()), "channel state persists locally"); - assert_eq!(db.version_vector().get("N1"), Some(&0), "vector unchanged after replay"); + assert_eq!(db.version_vector().get("N1"), Some(&(0, 0)), "vector unchanged after replay"); } #[test] @@ -399,7 +399,7 @@ assert!(ev.is_empty()); log.append(cert("x", "f1")).unwrap(); log.append(cert("x", "f2")).unwrap(); - assert_eq!(log.versions.get("nodeA"), Some(&1)); // seqs 0 then 1 + assert_eq!(log.versions.get("nodeA"), Some(&(0, 1))); // epoch 0, seqs 0 then 1 assert_eq!(log.lamport, 2); assert_eq!(log.next_seq(), 2); } @@ -424,9 +424,9 @@ #[test] fn ingest_is_idempotent() { let (mut log, _) = EventLog::open(tmp("ingest"), "local".into()); - let entry = LogEntry { origin: "peer".into(), seq: 0, lamport: 5, sig: None, event: cert("x", "f") }; + let entry = LogEntry { origin: "peer".into(), seq: 0, lamport: 5, epoch: 0, sig: None, event: cert("x", "f") }; assert!(log.ingest(entry.clone()).unwrap().is_some(), "first ingest applies"); - assert_eq!(log.versions.get("peer"), Some(&0)); + assert_eq!(log.versions.get("peer"), Some(&(0, 0))); assert_eq!(log.lamport, 6, "receive rule: max(local, remote) + 1"); assert!(log.ingest(entry).unwrap().is_none(), "duplicate ingest is a no-op"); } @@ -441,7 +441,7 @@ name: "bob".into(), email: None, ts: 0, home: "peer".into(), scram256: None, scram512: None, certfps: vec![], verified: true, ajoin: vec![], suspension: None, memos: vec![], memo_ignore: vec![], memo_notify: true, memo_limit: None, greet: String::new(), no_autoop: false, no_protect: false, hide_status: false, snotice: false, vhost: None, vhost_request: None, last_seen: 0, noexpire: false, expiry_warned: false, oper_note: None, }; - let entry = LogEntry { origin: "peer".into(), seq: 0, lamport: 1, sig: None, event: Event::AccountRegistered(Box::new(bob)) }; + let entry = LogEntry { origin: "peer".into(), seq: 0, lamport: 1, epoch: 0, sig: None, event: Event::AccountRegistered(Box::new(bob)) }; db.ingest(entry).unwrap(); assert!(db.exists("bob"), "peer's account is present locally"); assert!(db.exists("alice"), "local account still there"); @@ -582,6 +582,112 @@ assert_eq!(b.certfps("alice"), &[fp][..], "certs arrive folded into the snapshot"); } + fn akill(origin: &str, epoch: u64, seq: u64, mask: &str) -> LogEntry { + LogEntry::for_test_epoch(origin, epoch, seq, seq + 1, Event::AkillAdded { + kind: "G".into(), mask: mask.into(), setter: "op".into(), reason: "x".into(), ts: 0, expires: None, + }) + } + + // The snapshot-epoch fix, part 1: a peer holding an origin's PARTIAL state still + // converges across that origin's compaction (the forward epoch jump is accepted). + // This is exactly the case the earlier strict-ordering attempt broke. + #[test] + fn lagging_peer_converges_across_a_compaction() { + let mut a = Db::open(tmp("epochA"), "AAA"); + a.scram_iterations = 4096; + a.register("u0", "pw", None).unwrap(); + a.register("u1", "pw", None).unwrap(); + + // B syncs A's initial (epoch 0) state, then lags. + let mut b = Db::open(tmp("epochB"), "BBB"); + b.scram_iterations = 4096; + for e in a.missing_for(&b.version_vector()) { + b.ingest(e).unwrap(); + } + assert!(b.exists("u0") && b.exists("u1"), "B has A's initial state"); + + // A advances and COMPACTS (epoch 0 -> 1, seqs reset to 0). + a.register("u2", "pw", None).unwrap(); + a.compact().unwrap(); + + // B, still at epoch 0, must converge across the epoch jump. + for e in a.missing_for(&b.version_vector()) { + b.ingest(e).unwrap(); + } + assert!(b.exists("u2"), "B converges across A's compaction (epoch jump adopted)"); + } + + // Live-critical: after a compaction bumps the epoch and resets seqs, further local + // appends continue correctly, and the epoch/next-seq reconstruct across a reopen. + #[test] + fn append_after_compaction_continues_and_survives_reopen() { + let path = tmp("epochAppend"); + { + let mut db = Db::open(&path, "AAA"); + db.scram_iterations = 4096; + db.register("u0", "pw", None).unwrap(); + db.compact().unwrap(); // epoch 0 -> 1, seqs reset + db.register("u1", "pw", None).unwrap(); + assert!(db.exists("u0") && db.exists("u1")); + } + let mut db = Db::open(&path, "AAA"); + db.scram_iterations = 4096; + db.register("u2", "pw", None).unwrap(); + assert!(db.exists("u0") && db.exists("u1") && db.exists("u2"), "state survives compaction + reopen + append"); + assert_eq!(db.version_vector().get("AAA").map(|c| c.0), Some(1), "epoch stays 1 after one compaction (reopen doesn't re-bump)"); + } + + // Durability: a compaction whose snapshot has NO Global events (only channels) + // must still persist the bumped epoch, or a restart resets it to 0 and later + // writes diverge from peers. + #[test] + fn epoch_survives_restart_after_an_all_local_compaction() { + let path = tmp("epochLocal"); + { + let mut db = Db::open(&path, "AAA"); + db.scram_iterations = 4096; + db.register_channel("#c", "founder").unwrap(); // Local only — no Global entry + db.compact().unwrap(); // epoch 0 -> 1, all-Local snapshot + } + let mut db = Db::open(&path, "AAA"); + db.scram_iterations = 4096; + db.register("alice", "pw", None).unwrap(); // first Global entry after restart + assert_eq!(db.version_vector().get("AAA").map(|c| c.0), Some(1), "epoch recovered from Local entries, not reset to 0"); + } + + // The snapshot-epoch fix, part 2: an out-of-order relay WITHIN an epoch (a future + // seq ahead of ours) is deferred — the version vector doesn't advance past the + // gap — and the entries apply once the gap fills in order. + #[test] + fn out_of_order_within_an_epoch_is_deferred_then_fills() { + let mut b = Db::open(tmp("epochOOO"), "BBB"); + b.ingest(akill("CCC", 0, 0, "*@a")).unwrap(); + assert_eq!(b.version_vector().get("CCC"), Some(&(0, 0))); + // seq 2 before seq 1: dropped, version stays at (0,0). + b.ingest(akill("CCC", 0, 2, "*@c")).unwrap(); + assert_eq!(b.version_vector().get("CCC"), Some(&(0, 0)), "a gap within the epoch is not applied"); + // seq 1 fills the gap, then the re-delivered seq 2 is contiguous. + b.ingest(akill("CCC", 0, 1, "*@b")).unwrap(); + assert_eq!(b.version_vector().get("CCC"), Some(&(0, 1))); + b.ingest(akill("CCC", 0, 2, "*@c")).unwrap(); + assert_eq!(b.version_vector().get("CCC"), Some(&(0, 2)), "the gap filled, seq 2 applies"); + } + + // The snapshot-epoch fix, part 3: a higher epoch is adopted ONLY from its seq 0, + // so a relayed higher-epoch entry arriving before the snapshot's start is deferred. + #[test] + fn a_new_epoch_is_adopted_only_from_seq_zero() { + let mut b = Db::open(tmp("epochNew"), "BBB"); + b.ingest(akill("CCC", 0, 0, "*@a")).unwrap(); + assert_eq!(b.version_vector().get("CCC"), Some(&(0, 0))); + // Epoch 1 seq 3 arrives (relay ahead of the new snapshot's seq 0): deferred. + b.ingest(akill("CCC", 1, 3, "*@x")).unwrap(); + assert_eq!(b.version_vector().get("CCC"), Some(&(0, 0)), "new epoch not adopted before its seq 0"); + // Its seq 0 arrives: adopt the new epoch (the compaction reset). + b.ingest(akill("CCC", 1, 0, "*@y")).unwrap(); + assert_eq!(b.version_vector().get("CCC"), Some(&(1, 0)), "new epoch adopted from seq 0"); + } + // Tier C: a node accepts a signed entry from a trusted origin, and rejects one // from an origin it doesn't trust — even if that origin signed it validly. #[test] diff --git a/src/engine/mod.rs b/src/engine/mod.rs index e720f47..28393fd 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -639,11 +639,11 @@ impl Engine { } // Gossip pass-throughs to the account store, used by the replication layer. - pub fn gossip_digest(&self) -> HashMap { + pub fn gossip_digest(&self) -> HashMap { self.db.version_vector() } - pub fn gossip_missing(&self, peer: &HashMap) -> Vec { + pub fn gossip_missing(&self, peer: &HashMap) -> Vec { self.db.missing_for(peer) } diff --git a/src/gossip.rs b/src/gossip.rs index cf1a469..c7a8160 100644 --- a/src/gossip.rs +++ b/src/gossip.rs @@ -65,7 +65,7 @@ enum Msg { // A version vector. `reply` asks the peer to answer with its own digest, so a // node that dropped a push can pull back exactly what the peer is missing. Digest { - versions: HashMap, + versions: HashMap, #[serde(default)] reply: bool, },