diff --git a/src/engine/db/event.rs b/src/engine/db/event.rs index 4500b90..8028432 100644 --- a/src/engine/db/event.rs +++ b/src/engine/db/event.rs @@ -439,7 +439,14 @@ pub(crate) fn apply(accounts: &mut HashMap, channels: &mut Hash } Event::MemoSent { account, from, text, ts, receipt } => { if let Some(a) = accounts.get_mut(&key(&account)) { - a.memos.push(Memo { from, text, ts, read: false, receipt }); + // Enforce the mailbox cap in the fold too, so a gossip peer can't push + // memos past the limit the local MEMO SEND path enforces. Local sends + // are already capped before the event is logged, so this is a no-op on + // replay and only bounds an ingested flood. Default mirrors MemoServ's. + let limit = a.memo_limit.unwrap_or(30) as usize; + if a.memos.len() < limit { + a.memos.push(Memo { from, text, ts, read: false, receipt }); + } } } Event::MemoRead { account, index } => { diff --git a/src/engine/db/mod.rs b/src/engine/db/mod.rs index e330b6b..94e106a 100644 --- a/src/engine/db/mod.rs +++ b/src/engine/db/mod.rs @@ -874,8 +874,20 @@ impl EventLog { if entry.event.scope() != Scope::Global { return Ok(None); } - if self.versions.get(&entry.origin).is_some_and(|&s| entry.seq <= s) { - return Ok(None); // already have it + // Apply strictly in per-origin sequence. The FIRST entry ever seen from an + // origin sets the baseline — a peer syncing from a COMPACTED node starts + // mid-stream (its seqs restart at next_seq(), not 0), so we can't demand 0. + // After that, only the next contiguous seq is accepted; a non-contiguous one + // (an older duplicate, or a future entry a relay delivered ahead of the direct + // path in a 3+-node mesh) is dropped rather than applied, so it can't advance + // the version vector past a gap — the digest then re-requests the gap in order + // instead of losing it forever. + let expected = match self.versions.get(&entry.origin) { + None => entry.seq, + Some(&s) => s + 1, + }; + if entry.seq != expected { + return Ok(None); } self.persist(&entry)?; self.lamport = self.lamport.max(entry.lamport).saturating_add(1); // Lamport receive rule diff --git a/src/engine/tests.rs b/src/engine/tests.rs index 167e8a1..f226ecb 100644 --- a/src/engine/tests.rs +++ b/src/engine/tests.rs @@ -557,7 +557,7 @@ // genuine removal, releases her founderless channel. let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); e.set_irc_out(tx); - e.gossip_ingest(LogEntry::for_test("peer", 1, 1, db::Event::AccountDropped { account: "alice".into() })).unwrap(); + e.gossip_ingest(LogEntry::for_test("peer", 0, 1, db::Event::AccountDropped { account: "alice".into() })).unwrap(); assert!(e.db.channel("#hers").is_none(), "the dropped account's channel is released"); let parted = std::iter::from_fn(|| rx.try_recv().ok()) @@ -878,6 +878,34 @@ assert!(lifted, "gossiped ban lift removes the line from the ircd"); } + // A gossiped entry that arrives out of sequence (a relay racing the direct path + // in a 3+-node mesh) must NOT advance the version vector past the gap, or the + // skipped entries are lost forever; the digest re-requests them in order. + #[test] + fn gossip_drops_an_out_of_order_entry_until_the_gap_fills() { + use echo_nickserv::NickServ; + let path = std::env::temp_dir().join("echo-gossip-ooo.jsonl"); + let _ = std::fs::remove_file(&path); + let mut e = Engine::new(vec![Box::new(NickServ { uid: "42SAAAAAA".into(), guest_nick: "Guest".into(), guest_seq: 0 })], Db::open(&path, "test")); + e.set_sid("42S".into()); + let (tx, _rx) = tokio::sync::mpsc::unbounded_channel(); + e.set_irc_out(tx); + let akill = |seq: u64, mask: &str| LogEntry::for_test("peer", seq, seq + 1, db::Event::AkillAdded { + kind: "G".into(), mask: mask.into(), setter: "op".into(), reason: "x".into(), ts: 0, expires: None, + }); + + e.gossip_ingest(akill(0, "*@a")).unwrap(); + assert_eq!(e.gossip_digest().get("peer"), Some(&0), "first entry applies"); + // Seq 2 arrives before seq 1: dropped, version stays at 0. + e.gossip_ingest(akill(2, "*@c")).unwrap(); + assert_eq!(e.gossip_digest().get("peer"), Some(&0), "an out-of-order entry doesn't advance the version"); + // Seq 1 fills the gap, then the re-delivered seq 2 is contiguous. + e.gossip_ingest(akill(1, "*@b")).unwrap(); + assert_eq!(e.gossip_digest().get("peer"), Some(&1)); + e.gossip_ingest(akill(2, "*@c")).unwrap(); + assert_eq!(e.gossip_digest().get("peer"), Some(&2), "the gap filled, seq 2 now applies"); + } + // Losing a registration conflict logs out the local session on that name and // notifies them over the services-initiated outbound path. #[test]