Apply gossip entries in strict per-origin order and cap ingested memos
This commit is contained in:
parent
a774dfedb1
commit
01da702704
3 changed files with 51 additions and 4 deletions
|
|
@ -439,9 +439,16 @@ pub(crate) fn apply(accounts: &mut HashMap<String, Account>, channels: &mut Hash
|
||||||
}
|
}
|
||||||
Event::MemoSent { account, from, text, ts, receipt } => {
|
Event::MemoSent { account, from, text, ts, receipt } => {
|
||||||
if let Some(a) = accounts.get_mut(&key(&account)) {
|
if let Some(a) = accounts.get_mut(&key(&account)) {
|
||||||
|
// 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 });
|
a.memos.push(Memo { from, text, ts, read: false, receipt });
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
Event::MemoRead { account, index } => {
|
Event::MemoRead { account, index } => {
|
||||||
if let Some(m) = accounts.get_mut(&key(&account)).and_then(|a| a.memos.get_mut(index)) {
|
if let Some(m) = accounts.get_mut(&key(&account)).and_then(|a| a.memos.get_mut(index)) {
|
||||||
m.read = true;
|
m.read = true;
|
||||||
|
|
|
||||||
|
|
@ -874,8 +874,20 @@ impl EventLog {
|
||||||
if entry.event.scope() != Scope::Global {
|
if entry.event.scope() != Scope::Global {
|
||||||
return Ok(None);
|
return Ok(None);
|
||||||
}
|
}
|
||||||
if self.versions.get(&entry.origin).is_some_and(|&s| entry.seq <= s) {
|
// Apply strictly in per-origin sequence. The FIRST entry ever seen from an
|
||||||
return Ok(None); // already have it
|
// 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.persist(&entry)?;
|
||||||
self.lamport = self.lamport.max(entry.lamport).saturating_add(1); // Lamport receive rule
|
self.lamport = self.lamport.max(entry.lamport).saturating_add(1); // Lamport receive rule
|
||||||
|
|
|
||||||
|
|
@ -557,7 +557,7 @@
|
||||||
// genuine removal, releases her founderless channel.
|
// genuine removal, releases her founderless channel.
|
||||||
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
|
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
|
||||||
e.set_irc_out(tx);
|
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");
|
assert!(e.db.channel("#hers").is_none(), "the dropped account's channel is released");
|
||||||
let parted = std::iter::from_fn(|| rx.try_recv().ok())
|
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");
|
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
|
// Losing a registration conflict logs out the local session on that name and
|
||||||
// notifies them over the services-initiated outbound path.
|
// notifies them over the services-initiated outbound path.
|
||||||
#[test]
|
#[test]
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue