From c066f1072f021edbcb0c8f0ed2cf16568f5123b7 Mon Sep 17 00:00:00 2001 From: Jean Date: Sun, 12 Jul 2026 06:08:27 +0000 Subject: [PATCH] Add EventLog with origin/seq metadata for replication --- src/engine/db.rs | 99 +++++++++++++++++++++++++++++++++-------------- src/engine/mod.rs | 19 ++++++++- src/main.rs | 2 +- 3 files changed, 90 insertions(+), 30 deletions(-) diff --git a/src/engine/db.rs b/src/engine/db.rs index e086fab..2834a4e 100644 --- a/src/engine/db.rs +++ b/src/engine/db.rs @@ -39,6 +39,57 @@ pub enum Event { CertRemoved { account: String, fp: String }, } +// A durable record: the event plus the metadata a gossip layer needs to address +// and order it — the origin node and its per-node sequence. Lines written before +// these existed default them in. +#[derive(Debug, Clone, Serialize, Deserialize)] +struct LogEntry { + #[serde(default)] + origin: String, + #[serde(default)] + seq: u64, + #[serde(flatten)] + event: Event, +} + +// Append-only log, the sole persistent source of truth: `open` replays it, `append` +// stamps and writes one entry. Owning origin + next_seq is the seam a future gossip +// layer ships entries over, without the services knowing. +pub struct EventLog { + path: PathBuf, + origin: String, + next_seq: u64, +} + +impl EventLog { + fn open(path: PathBuf, origin: String) -> (Self, Vec) { + let mut events = Vec::new(); + let mut next_seq = 0; + if let Ok(data) = std::fs::read_to_string(&path) { + for line in data.lines().filter(|l| !l.trim().is_empty()) { + match serde_json::from_str::(line) { + Ok(entry) => { + if entry.origin == origin { + next_seq = next_seq.max(entry.seq + 1); + } + events.push(entry.event); + } + Err(e) => tracing::warn!(%e, "skipping malformed event log line"), + } + } + } + (Self { path, origin, next_seq }, events) + } + + fn append(&mut self, event: Event) -> std::io::Result<()> { + let entry = LogEntry { origin: self.origin.clone(), seq: self.next_seq, event }; + let mut f = std::fs::OpenOptions::new().create(true).append(true).open(&self.path)?; + writeln!(f, "{}", serde_json::to_string(&entry).unwrap_or_default())?; + self.next_seq += 1; + Ok(()) + } +} + #[derive(Debug)] pub enum RegError { Exists, @@ -71,7 +122,7 @@ pub struct Credentials { pub struct Db { accounts: HashMap, // keyed by casefolded name - path: PathBuf, + log: EventLog, // PBKDF2 cost baked into new SCRAM verifiers; lowered by tests. pub(crate) scram_iterations: u32, } @@ -85,31 +136,28 @@ fn now() -> u64 { } impl Db { - pub fn open(path: impl Into) -> Self { - let path = path.into(); + pub fn open(path: impl Into, origin: impl Into) -> Self { + let (log, events) = EventLog::open(path.into(), origin.into()); let mut accounts = HashMap::new(); - if let Ok(data) = std::fs::read_to_string(&path) { - for line in data.lines().filter(|l| !l.trim().is_empty()) { - match serde_json::from_str::(line) { - Ok(Event::AccountRegistered(a)) => { - accounts.insert(key(&a.name), a); + for event in events { + match event { + Event::AccountRegistered(a) => { + accounts.insert(key(&a.name), a); + } + Event::CertAdded { account, fp } => { + if let Some(a) = accounts.get_mut(&key(&account)) { + a.certfps.push(fp); } - Ok(Event::CertAdded { account, fp }) => { - if let Some(a) = accounts.get_mut(&key(&account)) { - a.certfps.push(fp); - } + } + Event::CertRemoved { account, fp } => { + if let Some(a) = accounts.get_mut(&key(&account)) { + a.certfps.retain(|c| *c != fp); } - Ok(Event::CertRemoved { account, fp }) => { - if let Some(a) = accounts.get_mut(&key(&account)) { - a.certfps.retain(|c| *c != fp); - } - } - Err(e) => tracing::warn!(%e, "skipping malformed event log line"), } } } - tracing::info!(accounts = accounts.len(), ?path, "account store loaded"); - Self { accounts, path, scram_iterations: scram::DEFAULT_ITERATIONS } + tracing::info!(accounts = accounts.len(), "account store loaded"); + Self { accounts, log, scram_iterations: scram::DEFAULT_ITERATIONS } } pub fn exists(&self, name: &str) -> bool { @@ -141,7 +189,7 @@ impl Db { scram512: Some(creds.scram512), certfps: Vec::new(), }; - self.append(&Event::AccountRegistered(account.clone())).map_err(|_| RegError::Internal)?; + self.log.append(Event::AccountRegistered(account.clone())).map_err(|_| RegError::Internal)?; self.accounts.insert(key(name), account); Ok(()) } @@ -197,7 +245,7 @@ impl Db { if !self.accounts.contains_key(&k) { return Err(CertError::NoAccount); } - self.append(&Event::CertAdded { account: account.to_string(), fp: fp.clone() }).map_err(|_| CertError::Internal)?; + self.log.append(Event::CertAdded { account: account.to_string(), fp: fp.clone() }).map_err(|_| CertError::Internal)?; self.accounts.get_mut(&k).unwrap().certfps.push(fp); Ok(()) } @@ -211,15 +259,10 @@ impl Db { Some(a) if !a.certfps.iter().any(|c| *c == fp) => return Ok(false), Some(_) => {} } - self.append(&Event::CertRemoved { account: account.to_string(), fp: fp.clone() }).map_err(|_| CertError::Internal)?; + self.log.append(Event::CertRemoved { account: account.to_string(), fp: fp.clone() }).map_err(|_| CertError::Internal)?; self.accounts.get_mut(&k).unwrap().certfps.retain(|c| *c != fp); Ok(true) } - - fn append(&self, event: &Event) -> std::io::Result<()> { - let mut f = std::fs::OpenOptions::new().create(true).append(true).open(&self.path)?; - writeln!(f, "{}", serde_json::to_string(event).unwrap_or_default()) - } } fn hash_password(password: &str) -> Option { diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 47c5c5f..ae4a34d 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -491,7 +491,7 @@ mod tests { fn engine_with(name: &str, account: &str, password: &str) -> Engine { let path = std::env::temp_dir().join(format!("fedserv-sasl-{name}.jsonl")); let _ = std::fs::remove_file(&path); - let mut db = Db::open(&path); + let mut db = Db::open(&path, "test"); db.scram_iterations = 4096; // keep the debug-build verifier cheap in tests assert!(db.register(account, password, None).is_ok()); Engine::new(vec![Box::new(NickServ { uid: "42SAAAAAA".to_string(), guest_nick: "Guest".to_string(), guest_seq: 12345 })], db) @@ -833,6 +833,23 @@ mod tests { assert!(second.iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("already identified"))), "{second:?}"); } + // The event log is the source of truth: state survives a reopen, rebuilt by + // folding the log (with its origin/seq metadata) rather than a snapshot. + #[test] + fn account_store_survives_reopen() { + let path = std::env::temp_dir().join("fedserv-reopen.jsonl"); + let _ = std::fs::remove_file(&path); + { + let mut db = Db::open(&path, "n1"); + db.scram_iterations = 4096; + db.register("foo", "sesame", None).unwrap(); + db.certfp_add("foo", "aabbccddeeff00112233445566778899").unwrap(); + } + let db = Db::open(&path, "n1"); + assert_eq!(db.certfps("foo").len(), 1, "cert replayed from log"); + assert!(db.authenticate("foo", "sesame").is_some(), "account replayed from log"); + } + // Regression: after a nick change (e.g. a guest rename, then back to your // real nick), IDENTIFY must authenticate the CURRENT nick, not the one held // at burst time — otherwise you log into the wrong account. diff --git a/src/main.rs b/src/main.rs index 13c5239..08faa71 100644 --- a/src/main.rs +++ b/src/main.rs @@ -37,7 +37,7 @@ async fn main() -> Result<()> { guest_nick: cfg.server.guest_nick.clone(), guest_seq: (ts % 100_000) as u32, })]; - let mut db = engine::db::Db::open("fedserv.db.jsonl"); + let mut db = engine::db::Db::open("fedserv.db.jsonl", &cfg.server.sid); db.scram_iterations = cfg.server.scram_iterations; let engine = Engine::new(services, db);