From 1f6a11dba2ba66916dbc4141686490c770046e31 Mon Sep 17 00:00:00 2001 From: Jean Date: Thu, 16 Jul 2026 17:50:11 +0000 Subject: [PATCH] statserv: persist per-channel activity across restart too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The earlier StatsSet fix persisted only the shared namespaced counters. The per-channel BOTSTATS view (lines seen + top talkers, from /statserv #chan) lived on the LiveChannel struct, which is rebuilt from each burst — so it reset to zero on restart. Move it to a name-keyed Network.chan_activity map (independent of live membership, seedable before the burst) and carry it in the StatsSet snapshot (new serde-default `channels` field) alongside the counters. Reworded the notice from 'seen this session' now that it persists. --- modules/statserv/src/channel.rs | 2 +- src/engine/db/event.rs | 15 ++++++--- src/engine/db/mod.rs | 12 ++++++-- src/engine/db/network.rs | 20 ++++++++++-- src/engine/db/tests.rs | 4 ++- src/engine/mod.rs | 8 +++-- src/engine/state.rs | 54 ++++++++++++++++++++++++++------- src/engine/tests.rs | 2 +- 8 files changed, 91 insertions(+), 26 deletions(-) diff --git a/modules/statserv/src/channel.rs b/modules/statserv/src/channel.rs index 34b9949..5e6fce5 100644 --- a/modules/statserv/src/channel.rs +++ b/modules/statserv/src/channel.rs @@ -10,7 +10,7 @@ pub fn handle(me: &str, from: &Sender, chan: &str, ctx: &mut ServiceCtx, net: &d ctx.notice(me, from.uid, format!("No activity recorded for \x02{chan}\x02 yet.")); return; }; - ctx.notice(me, from.uid, format!("\x02{chan}\x02 — \x02{lines}\x02 line(s) seen this session.")); + ctx.notice(me, from.uid, format!("\x02{chan}\x02 — \x02{lines}\x02 line(s) seen.")); if top.is_empty() { return; } diff --git a/src/engine/db/event.rs b/src/engine/db/event.rs index 821a7f2..d742ad1 100644 --- a/src/engine/db/event.rs +++ b/src/engine/db/event.rs @@ -136,9 +136,15 @@ pub enum Event { // by minting a fake one on `sid`), so Local scope but still persisted. JupeAdded { name: String, sid: String, reason: String }, JupeRemoved { name: String }, - // A full snapshot of the shared stat counters (StatServ / gRPC Stats), - // written periodically and on shutdown so they survive a restart. - StatsSet { counters: Vec<(String, u64)> }, + // A full snapshot of the stats (StatServ / gRPC Stats), written periodically + // and on shutdown so they survive a restart: shared namespaced counters, plus + // per-channel activity (channel, line count, top talkers). `channels` defaults + // for log entries written before it existed. + StatsSet { + counters: Vec<(String, u64)>, + #[serde(default)] + channels: super::ChanStats, + }, } // Whether an event replicates across the federation. Account identity is Global @@ -627,9 +633,10 @@ pub(crate) fn apply(accounts: &mut HashMap, channels: &mut Hash Event::JupeRemoved { name } => { net.jupes.retain(|j| !j.name.eq_ignore_ascii_case(&name)); } - Event::StatsSet { counters } => { + Event::StatsSet { counters, channels } => { // A full snapshot: replace, so replaying the newest wins. net.stats = counters.into_iter().collect(); + net.chan_stats = channels; } Event::AccountExpiryWarned { account } => { if let Some(a) = accounts.get_mut(&key(&account)) { diff --git a/src/engine/db/mod.rs b/src/engine/db/mod.rs index 9a386e0..98c2a93 100644 --- a/src/engine/db/mod.rs +++ b/src/engine/db/mod.rs @@ -282,6 +282,9 @@ pub struct Report { // The network-wide lists that aren't accounts/channels/grouped/bots: the network // bans and the news items. Bundled so `apply` threads one parameter for them // (and stays within its argument budget as more are added). +// Persisted per-channel BOTSTATS activity: (channel, line count, top talkers). +pub type ChanStats = Vec<(String, u64, Vec<(String, u64)>)>; + #[derive(Debug, Clone, Default)] pub struct NetData { pub akills: Vec, @@ -310,6 +313,8 @@ pub struct NetData { // Persisted shared stat counters (StatServ / gRPC Stats), snapshotted so they // survive a restart (live gauges are re-derived, not stored here). pub stats: std::collections::BTreeMap, + // Persisted per-channel BOTSTATS activity: (channel, lines, top talkers). + pub chan_stats: ChanStats, } // A session-limit exception: an IP-mask glob and the session allowance for IPs @@ -1168,8 +1173,11 @@ impl Db { if self.net.default_bot.is_some() { snapshot.push(Event::DefaultBotSet { bot: self.net.default_bot.clone() }); } - if !self.net.stats.is_empty() { - snapshot.push(Event::StatsSet { counters: self.net.stats.iter().map(|(k, v)| (k.clone(), *v)).collect() }); + if !self.net.stats.is_empty() || !self.net.chan_stats.is_empty() { + snapshot.push(Event::StatsSet { + counters: self.net.stats.iter().map(|(k, v)| (k.clone(), *v)).collect(), + channels: self.net.chan_stats.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 50da87c..1a0465c 100644 --- a/src/engine/db/network.rs +++ b/src/engine/db/network.rs @@ -250,10 +250,24 @@ impl Db { self.net.stats.clone() } - /// Snapshot the live stat counters to the log so they survive a restart. - pub fn persist_stats(&mut self, counters: &std::collections::BTreeMap) -> std::io::Result<()> { - self.log.append(Event::StatsSet { counters: counters.iter().map(|(k, v)| (k.clone(), *v)).collect() })?; + /// The persisted per-channel activity, to seed BOTSTATS on startup. + pub fn persisted_chan_stats(&self) -> ChanStats { + self.net.chan_stats.clone() + } + + /// Snapshot the live stats (shared counters + per-channel activity) to the log + /// so they survive a restart. + pub fn persist_stats( + &mut self, + counters: &std::collections::BTreeMap, + chan_stats: ChanStats, + ) -> std::io::Result<()> { + self.log.append(Event::StatsSet { + counters: counters.iter().map(|(k, v)| (k.clone(), *v)).collect(), + channels: chan_stats.clone(), + })?; self.net.stats = counters.clone(); + self.net.chan_stats = chan_stats; Ok(()) } diff --git a/src/engine/db/tests.rs b/src/engine/db/tests.rs index 54142ad..1c170e1 100644 --- a/src/engine/db/tests.rs +++ b/src/engine/db/tests.rs @@ -532,13 +532,15 @@ let p = tmp("statspersist"); let counters: std::collections::BTreeMap = [("chanserv.register".to_string(), 7u64), ("nickserv.identify".to_string(), 42u64)].into_iter().collect(); + let chans = vec![("#devs".to_string(), 6u64, vec![("reverse".to_string(), 5u64), ("Keiko".to_string(), 1u64)])]; { let mut db = Db::open(&p, "N1"); - db.persist_stats(&counters).unwrap(); + db.persist_stats(&counters, chans.clone()).unwrap(); assert_eq!(db.persisted_stats(), counters, "stored while live"); } let db = Db::open(&p, "N1"); assert_eq!(db.persisted_stats(), counters, "counters replay from the log after a restart"); + assert_eq!(db.persisted_chan_stats(), chans, "per-channel activity replays too"); } // Fold parity: the live state after a broad sequence of writes must equal the diff --git a/src/engine/mod.rs b/src/engine/mod.rs index b179056..4d81fa9 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -213,8 +213,9 @@ impl Engine { (!topics.is_empty()).then(|| (s.nick().to_string(), blurb, topics)) }) .collect(); - // Restore the persisted stat counters so they survive a restart. + // Restore persisted stats so they survive a restart. network.seed_stats(db.persisted_stats()); + network.seed_chan_activity(db.persisted_chan_stats()); Self { services, network, @@ -275,10 +276,11 @@ impl Engine { // Only the real counters are persisted; the live gauges above are re-derived. pub fn persist_stats(&mut self) { let counters = self.network.stat_counters().clone(); - if counters.is_empty() { + let chan_stats = self.network.chan_activity_snapshot(); + if counters.is_empty() && chan_stats.is_empty() { return; } - if let Err(err) = self.db.persist_stats(&counters) { + if let Err(err) = self.db.persist_stats(&counters, chan_stats) { tracing::warn!(%err, "failed to persist stat counters"); } } diff --git a/src/engine/state.rs b/src/engine/state.rs index a54b95b..eb3dc71 100644 --- a/src/engine/state.rs +++ b/src/engine/state.rs @@ -23,9 +23,12 @@ pub struct Network { // Downstream server tree: SID -> the SID that introduced it. Lets a hub's // SQUIT cascade to every server (and user) behind it. Rebuilt each link. servers: HashMap, - // Shared, namespaced stat counters any service contributes to (ephemeral, - // ordered for a stable snapshot). Read by StatServ and the gRPC Stats API. + // Shared, namespaced stat counters any service contributes to (persisted via + // StatsSet). Read by StatServ and the gRPC Stats API. stats: BTreeMap, + // Per-channel BOTSTATS activity (lines + top talkers), name-keyed and + // persisted alongside `stats` so it survives a restart. + chan_activity: HashMap, // Live session count per connecting IP, for OperServ session limiting. sessions: HashMap, // Recent moderation/action incidents (bounded ring), for OperServ LOGSEARCH. @@ -85,7 +88,13 @@ pub struct Channel { pub ops: HashSet, // uids holding channel-operator status pub voices: HashSet, // uids holding +v pub key: Option, - // BOTSTATS: lines seen this session, and per-nick counts (top-talkers). Bounded. +} + +// Per-channel BOTSTATS activity: total lines and per-nick counts (top talkers). +// Held name-keyed (not on the live Channel, which is rebuilt from each burst) and +// persisted, so a channel's history survives a services restart. +#[derive(Default, Clone)] +pub struct ChanActivity { pub lines: u64, pub talkers: HashMap, } @@ -332,10 +341,10 @@ impl Network { // is capped so a busy channel can't grow it without bound. pub fn record_line(&mut self, channel: &str, nick: &str) { const TALKER_CAP: usize = 512; - let c = self.channels.entry(lc(channel)).or_default(); - c.lines = c.lines.saturating_add(1); - if c.talkers.len() < TALKER_CAP || c.talkers.contains_key(nick) { - *c.talkers.entry(nick.to_string()).or_insert(0) += 1; + let a = self.chan_activity.entry(lc(channel)).or_default(); + a.lines = a.lines.saturating_add(1); + if a.talkers.len() < TALKER_CAP || a.talkers.contains_key(nick) { + *a.talkers.entry(nick.to_string()).or_insert(0) += 1; } } @@ -356,11 +365,34 @@ impl Network { // BOTSTATS view: (total lines, top talkers by count, descending). pub fn channel_activity(&self, channel: &str) -> Option<(u64, Vec<(String, u64)>)> { - let c = self.channels.get(&lc(channel))?; - let mut top: Vec<(String, u64)> = c.talkers.iter().map(|(n, v)| (n.clone(), *v)).collect(); - top.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0))); + let a = self.chan_activity.get(&lc(channel))?; + let mut top: Vec<(String, u64)> = a.talkers.iter().map(|(n, v)| (n.clone(), *v)).collect(); + top.sort_by(|x, y| y.1.cmp(&x.1).then_with(|| x.0.cmp(&y.0))); top.truncate(10); - Some((c.lines, top)) + Some((a.lines, top)) + } + + // Seed per-channel activity from persisted state at startup. + pub fn seed_chan_activity(&mut self, data: crate::engine::db::ChanStats) { + self.chan_activity = data + .into_iter() + .map(|(c, lines, talkers)| (c, ChanActivity { lines, talkers: talkers.into_iter().collect() })) + .collect(); + } + + // Snapshot per-channel activity for persistence: each active channel's line + // count and its top talkers (capped, so a busy channel bounds the event). + pub fn chan_activity_snapshot(&self) -> crate::engine::db::ChanStats { + self.chan_activity + .iter() + .filter(|(_, a)| a.lines > 0) + .map(|(c, a)| { + let mut talkers: Vec<(String, u64)> = a.talkers.iter().map(|(n, v)| (n.clone(), *v)).collect(); + talkers.sort_by(|x, y| y.1.cmp(&x.1).then_with(|| x.0.cmp(&y.0))); + talkers.truncate(50); + (c.clone(), a.lines, talkers) + }) + .collect() } // Uids currently in `channel`. diff --git a/src/engine/tests.rs b/src/engine/tests.rs index 18668eb..f3d8e7a 100644 --- a/src/engine/tests.rs +++ b/src/engine/tests.rs @@ -2723,7 +2723,7 @@ // Per-channel view. let out = ss(&mut e, "#c"); - assert!(out.iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("line(s) seen this session"))), "channel lines: {out:?}"); + assert!(out.iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("line(s) seen"))), "channel lines: {out:?}"); assert!(out.iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("spammer") && text.contains("3"))), "top talker: {out:?}"); // Global view (oper) shows the shared counters. let out = ss(&mut e, "SERVER");