diff --git a/src/engine/db/event.rs b/src/engine/db/event.rs index 4fe75e6..d4bbb83 100644 --- a/src/engine/db/event.rs +++ b/src/engine/db/event.rs @@ -158,6 +158,14 @@ pub enum Event { #[serde(default)] channels: super::ChanStats, }, + // A snapshot of the recent-incident ring (OperServ LOGSEARCH), flushed + // periodically and on shutdown so the moderation trail survives a restart. + // `seq` restores the id counter; incidents are (id, ts, summary), oldest first. + IncidentsSet { + // NOT `seq`: the event flattens into LogEntry, which already has a `seq`. + incident_seq: u64, + incidents: Vec<(String, u64, String)>, + }, } // Whether an event replicates across the federation. Account identity is Global @@ -268,7 +276,8 @@ impl Event { | Event::ChannelOperNoteSet { .. } | Event::JupeAdded { .. } | Event::JupeRemoved { .. } - | Event::StatsSet { .. } => Scope::Local, + | Event::StatsSet { .. } + | Event::IncidentsSet { .. } => Scope::Local, } } } @@ -689,6 +698,10 @@ pub(crate) fn apply(accounts: &mut HashMap, channels: &mut Hash net.stats = counters.into_iter().collect(); net.chan_stats = channels; } + Event::IncidentsSet { incident_seq, incidents } => { + net.incident_seq = incident_seq; + net.incidents = incidents; + } Event::AccountExpiryWarned { account } => { if let Some(a) = accounts.get_mut(&key(&account)) { a.expiry_warned = true; diff --git a/src/engine/db/mod.rs b/src/engine/db/mod.rs index 5346f4c..f9d4c91 100644 --- a/src/engine/db/mod.rs +++ b/src/engine/db/mod.rs @@ -350,6 +350,10 @@ pub struct NetData { pub stats: std::collections::BTreeMap, // Persisted per-channel BOTSTATS activity: (channel, lines, top talkers). pub chan_stats: ChanStats, + // Persisted snapshot of the recent-incident ring (OperServ LOGSEARCH) and its + // id counter, so the moderation trail survives a restart. + pub incidents: Vec<(String, u64, String)>, + pub incident_seq: u64, } // A session-limit exception: an IP-mask glob and the session allowance for IPs @@ -1273,6 +1277,12 @@ impl Db { channels: self.net.chan_stats.clone(), }); } + if !self.net.incidents.is_empty() { + snapshot.push(Event::IncidentsSet { + incident_seq: self.net.incident_seq, + incidents: self.net.incidents.clone(), + }); + } for host in &self.host_cfg.offers { snapshot.push(Event::VhostOfferAdded { host: host.clone() }); } diff --git a/src/engine/db/network.rs b/src/engine/db/network.rs index 398d862..ad4f5ad 100644 --- a/src/engine/db/network.rs +++ b/src/engine/db/network.rs @@ -427,6 +427,21 @@ impl Db { Ok(()) } + /// Snapshot the recent-incident ring to the log so OperServ LOGSEARCH survives + /// a restart. `seq` is the id counter; `incidents` are (id, ts, summary). + pub fn persist_incidents(&mut self, seq: u64, incidents: Vec<(String, u64, String)>) -> std::io::Result<()> { + self.log.append(Event::IncidentsSet { incident_seq: seq, incidents: incidents.clone() })?; + self.net.incident_seq = seq; + self.net.incidents = incidents; + Ok(()) + } + + /// The persisted incident snapshot (id counter, incidents), for reseeding the + /// live ring at startup. + pub fn persisted_incidents(&self) -> (u64, Vec<(String, u64, String)>) { + (self.net.incident_seq, self.net.incidents.clone()) + } + /// File an abuse report, rate-limited per reporter. Returns the new report's /// id, or None if the reporter filed one too recently. pub fn report_file(&mut self, reporter: &str, target: &str, reason: &str) -> Option { diff --git a/src/engine/db/tests.rs b/src/engine/db/tests.rs index 1b11aa6..2ee29c9 100644 --- a/src/engine/db/tests.rs +++ b/src/engine/db/tests.rs @@ -636,6 +636,22 @@ assert_eq!(db.persisted_chan_stats(), chans, "per-channel activity replays too"); } + #[test] + fn incidents_survive_a_restart() { + let p = tmp("incpersist"); + let incidents = vec![ + ("i1".to_string(), 1000u64, "[REGISTER] baddie tried a look-alike nick".to_string()), + ("i2".to_string(), 1001u64, "kicked spammer from #chan".to_string()), + ]; + { + let mut db = Db::open(&p, "N1"); + db.persist_incidents(9, incidents.clone()).unwrap(); + assert_eq!(db.persisted_incidents(), (9, incidents.clone()), "stored while live"); + } + let db = Db::open(&p, "N1"); + assert_eq!(db.persisted_incidents(), (9, incidents), "incidents replay from the log after a restart"); + } + // Fold parity: the live state after a broad sequence of writes must equal the // state rebuilt by replaying the same log — every event's manual live mutation // has to match its `apply()` arm, or a restart silently changes the data. diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 3e39528..53f15a4 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -229,6 +229,7 @@ impl Engine { // Restore persisted stats so they survive a restart. network.seed_stats(db.persisted_stats()); network.seed_chan_activity(db.persisted_chan_stats()); + network.seed_incidents(db.persisted_incidents()); Self { services, network, @@ -308,6 +309,18 @@ impl Engine { } } + // Snapshot the recent-incident ring (OperServ LOGSEARCH) so it survives a + // restart. Flushed on the same cadence as the stat counters. + pub fn persist_incidents(&mut self) { + let (seq, incidents) = self.network.incidents_snapshot(); + if incidents.is_empty() { + return; + } + if let Err(err) = self.db.persist_incidents(seq, incidents) { + tracing::warn!(%err, "failed to persist incidents"); + } + } + // Wall-clock seconds, overridable in tests so the time-based FLOOD kicker is // deterministic. fn now_secs(&self) -> u64 { @@ -1875,7 +1888,7 @@ fn audit_summary(event: &db::Event) -> Option { | ChannelMlock { .. } | ChannelDescSet { .. } | ChannelEntryMsgSet { .. } | ChannelUrlSet { .. } | ChannelEmailSet { .. } | ChannelSettingsSet { .. } | ChannelKickerSet { .. } | ChannelBadwordsSet { .. } | ChannelTriggersSet { .. } | ChannelTopicSet { .. } | AccountSeen { .. } | ChannelUsed { .. } - | AccountExpiryWarned { .. } | ChannelExpiryWarned { .. } | StatsSet { .. } => return None, + | AccountExpiryWarned { .. } | ChannelExpiryWarned { .. } | StatsSet { .. } | IncidentsSet { .. } => return None, }; Some(s) } diff --git a/src/engine/state.rs b/src/engine/state.rs index 28b46af..f0eb150 100644 --- a/src/engine/state.rs +++ b/src/engine/state.rs @@ -456,6 +456,20 @@ impl Network { self.stats = stats; } + // Restore the recent-incident ring (and its id counter) from a persisted + // snapshot at startup, so OperServ LOGSEARCH survives a restart. + pub fn seed_incidents(&mut self, data: (u64, Vec<(String, u64, String)>)) { + let (seq, incidents) = data; + self.incident_seq = seq; + self.incidents = incidents.into_iter().map(|(id, ts, summary)| Incident { id, ts, summary }).collect(); + } + + // Snapshot the live incident ring (id counter, incidents oldest-first) for + // persistence. + pub fn incidents_snapshot(&self) -> (u64, Vec<(String, u64, String)>) { + (self.incident_seq, self.incidents.iter().map(|i| (i.id.clone(), i.ts, i.summary.clone())).collect()) + } + // BOTSTATS view: (total lines, top talkers by count, descending). pub fn channel_activity(&self, channel: &str) -> Option<(u64, Vec<(String, u64)>)> { let a = self.chan_activity.get(&lc(channel))?; diff --git a/src/grpc.rs b/src/grpc.rs index 8d3dc55..f97812a 100644 --- a/src/grpc.rs +++ b/src/grpc.rs @@ -189,6 +189,7 @@ fn to_wire(entry: &LogEntry) -> Option { | Event::JupeAdded { .. } | Event::JupeRemoved { .. } | Event::StatsSet { .. } + | Event::IncidentsSet { .. } | Event::AccountExpiryWarned { .. } | Event::ChannelExpiryWarned { .. } | Event::AccountOperNoteSet { .. } diff --git a/src/link.rs b/src/link.rs index bc212d0..4ceb943 100644 --- a/src/link.rs +++ b/src/link.rs @@ -165,7 +165,11 @@ pub async fn run(mut proto: Box, engine: Arc>, addr: continue; } if let NetAction::Shutdown { restart, reason } = act { - engine.lock().await.persist_stats(); + { + let mut e = engine.lock().await; + e.persist_stats(); + e.persist_incidents(); + } let _ = write.flush().await; shutdown(restart, &reason); } @@ -182,7 +186,11 @@ pub async fn run(mut proto: Box, engine: Arc>, addr: if let NetAction::SendEmail { to, subject, text, html } = action { dispatch_email(&email, to, subject, text, html); } else if let NetAction::Shutdown { restart, reason } = action { - engine.lock().await.persist_stats(); + { + let mut e = engine.lock().await; + e.persist_stats(); + e.persist_incidents(); + } let _ = write.flush().await; shutdown(restart, &reason); } else { diff --git a/src/main.rs b/src/main.rs index 5cc7e2e..86a4d11 100644 --- a/src/main.rs +++ b/src/main.rs @@ -260,6 +260,7 @@ async fn main() -> Result<()> { tokio::time::sleep(std::time::Duration::from_secs(300)).await; let mut e = engine.lock().await; e.persist_stats(); // snapshot counters so a crash loses at most ~5 min + e.persist_incidents(); // and the LOGSEARCH incident ring e.expire_sweep(); if let Err(err) = e.maybe_compact() { tracing::warn!(%err, "compaction failed"); @@ -300,8 +301,13 @@ async fn main() -> Result<()> { tokio::select! { res = link::run(proto, engine, &addr, irc_rx, irc_tx, cfg.email.clone(), cfg.keycard.clone(), cfg.dictserv.as_ref().map(|d| d.server.clone())) => res, _ = shutdown_signal() => { - // Flush stat counters so a clean stop/restart keeps StatServ history. - shutdown_engine.lock().await.persist_stats(); + // Flush stat counters + the incident ring so a clean stop/restart keeps + // StatServ history and OperServ LOGSEARCH. + { + let mut e = shutdown_engine.lock().await; + e.persist_stats(); + e.persist_incidents(); + } tracing::info!("received shutdown signal, exiting"); Ok(()) }