botserv: op assigned bots on join; statserv: persist counters across restart
Some checks failed
CI / check (push) Failing after 3m48s
Some checks failed
CI / check (push) Failing after 3m48s
Problem 1: an assigned bot joined via IJOIN with no status mode, so it had no +o. Join it with the 4-param IJOIN form (ts 1, mode o) so it's opped, matching Anope's default bot modes (+o), and exempt bots from SECUREOPS/RESTRICTED so the op is never stripped. Problem 2: the shared stat counters lived only in memory, so a restart wiped them. Snapshot them to the log (StatsSet event, folded into NetData + compaction) periodically and on shutdown, and reseed the live registry from the log at startup. Live gauges (online counts) stay re-derived, not stored.
This commit is contained in:
parent
fd455922a5
commit
8a9b6a37ca
10 changed files with 85 additions and 10 deletions
|
|
@ -136,6 +136,9 @@ 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)> },
|
||||
}
|
||||
|
||||
// Whether an event replicates across the federation. Account identity is Global
|
||||
|
|
@ -238,7 +241,8 @@ impl Event {
|
|||
| Event::ChannelExpiryWarned { .. }
|
||||
| Event::ChannelOperNoteSet { .. }
|
||||
| Event::JupeAdded { .. }
|
||||
| Event::JupeRemoved { .. } => Scope::Local,
|
||||
| Event::JupeRemoved { .. }
|
||||
| Event::StatsSet { .. } => Scope::Local,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -623,6 +627,10 @@ pub(crate) fn apply(accounts: &mut HashMap<String, Account>, channels: &mut Hash
|
|||
Event::JupeRemoved { name } => {
|
||||
net.jupes.retain(|j| !j.name.eq_ignore_ascii_case(&name));
|
||||
}
|
||||
Event::StatsSet { counters } => {
|
||||
// A full snapshot: replace, so replaying the newest wins.
|
||||
net.stats = counters.into_iter().collect();
|
||||
}
|
||||
Event::AccountExpiryWarned { account } => {
|
||||
if let Some(a) = accounts.get_mut(&key(&account)) {
|
||||
a.expiry_warned = true;
|
||||
|
|
|
|||
|
|
@ -307,6 +307,9 @@ pub struct NetData {
|
|||
// The bot auto-assigned to newly registered channels (BotServ AUTOASSIGN),
|
||||
// if any. Casefolded nick; None disables auto-assignment.
|
||||
pub default_bot: Option<String>,
|
||||
// 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<String, u64>,
|
||||
}
|
||||
|
||||
// A session-limit exception: an IP-mask glob and the session allowance for IPs
|
||||
|
|
@ -1165,6 +1168,9 @@ 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() });
|
||||
}
|
||||
for host in &self.host_cfg.offers {
|
||||
snapshot.push(Event::VhostOfferAdded { host: host.clone() });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -245,6 +245,18 @@ impl Db {
|
|||
self.net.jupes.iter().map(|j| (j.name.clone(), j.sid.clone(), j.reason.clone())).collect()
|
||||
}
|
||||
|
||||
/// The persisted stat counters, to seed the live registry on startup.
|
||||
pub fn persisted_stats(&self) -> std::collections::BTreeMap<String, u64> {
|
||||
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<String, u64>) -> std::io::Result<()> {
|
||||
self.log.append(Event::StatsSet { counters: counters.iter().map(|(k, v)| (k.clone(), *v)).collect() })?;
|
||||
self.net.stats = counters.clone();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 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<u64> {
|
||||
|
|
|
|||
|
|
@ -525,6 +525,22 @@
|
|||
assert_eq!(db.jupes(), vec![("rogue.example".to_string(), sid, "abuse".to_string())], "the jupe replays from the log");
|
||||
}
|
||||
|
||||
// StatServ / Stats-API counters are snapshotted to the log, so they survive a
|
||||
// restart instead of resetting to zero.
|
||||
#[test]
|
||||
fn stat_counters_survive_a_restart() {
|
||||
let p = tmp("statspersist");
|
||||
let counters: std::collections::BTreeMap<String, u64> =
|
||||
[("chanserv.register".to_string(), 7u64), ("nickserv.identify".to_string(), 42u64)].into_iter().collect();
|
||||
{
|
||||
let mut db = Db::open(&p, "N1");
|
||||
db.persist_stats(&counters).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");
|
||||
}
|
||||
|
||||
// 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.
|
||||
|
|
|
|||
|
|
@ -213,6 +213,8 @@ impl Engine {
|
|||
(!topics.is_empty()).then(|| (s.nick().to_string(), blurb, topics))
|
||||
})
|
||||
.collect();
|
||||
// Restore the persisted stat counters so they survive a restart.
|
||||
network.seed_stats(db.persisted_stats());
|
||||
Self {
|
||||
services,
|
||||
network,
|
||||
|
|
@ -268,6 +270,19 @@ impl Engine {
|
|||
m
|
||||
}
|
||||
|
||||
// Snapshot the accumulated stat counters to the log (periodically and on
|
||||
// shutdown) so StatServ / the Stats API keep their history across a restart.
|
||||
// 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() {
|
||||
return;
|
||||
}
|
||||
if let Err(err) = self.db.persist_stats(&counters) {
|
||||
tracing::warn!(%err, "failed to persist stat counters");
|
||||
}
|
||||
}
|
||||
|
||||
// Wall-clock seconds, overridable in tests so the time-based FLOOD kicker is
|
||||
// deterministic.
|
||||
fn now_secs(&self) -> u64 {
|
||||
|
|
@ -930,13 +945,16 @@ impl Engine {
|
|||
// Computed before the match below moves `channel` into its actions.
|
||||
let greet = self.greet_on_join(&channel, account.as_deref());
|
||||
let mut out = Vec::new();
|
||||
// A services bot assigned here is opped on join and is never subject
|
||||
// to secureops/restricted — it's staff, not a member.
|
||||
let is_bot = self.bot_uids.values().any(|b| b == &uid);
|
||||
// SECUREOPS: a user who arrives opped (FJOIN prefix) but lacks op-level
|
||||
// access loses it, unless we're about to grant it to them anyway.
|
||||
if op && mode != Some("+o") && self.db.channel(&channel).is_some_and(|c| c.settings.secureops) {
|
||||
if op && mode != Some("+o") && !is_bot && self.db.channel(&channel).is_some_and(|c| c.settings.secureops) {
|
||||
out.push(NetAction::ChannelMode { from: from.clone(), channel: channel.clone(), modes: format!("-o {uid}") });
|
||||
}
|
||||
// RESTRICTED: only users with access (or opers) may be in the channel.
|
||||
if mode.is_none() && self.db.channel(&channel).is_some_and(|c| c.settings.restricted) {
|
||||
if mode.is_none() && !is_bot && self.db.channel(&channel).is_some_and(|c| c.settings.restricted) {
|
||||
let is_oper = account.as_deref().is_some_and(|a| self.oper_privs(a).any());
|
||||
if !is_oper {
|
||||
out.push(NetAction::Kick { from, channel, uid, reason: "This channel is restricted to users with access.".to_string() });
|
||||
|
|
@ -1348,7 +1366,7 @@ fn audit_summary(event: &db::Event) -> Option<String> {
|
|||
| ChannelMlock { .. } | ChannelDescSet { .. } | ChannelEntryMsgSet { .. } | ChannelUrlSet { .. } | ChannelEmailSet { .. } | ChannelSettingsSet { .. }
|
||||
| ChannelKickerSet { .. } | ChannelBadwordsSet { .. } | ChannelTriggersSet { .. }
|
||||
| ChannelTopicSet { .. } | AccountSeen { .. } | ChannelUsed { .. }
|
||||
| AccountExpiryWarned { .. } | ChannelExpiryWarned { .. } => return None,
|
||||
| AccountExpiryWarned { .. } | ChannelExpiryWarned { .. } | StatsSet { .. } => return None,
|
||||
};
|
||||
Some(s)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -349,6 +349,11 @@ impl Network {
|
|||
&self.stats
|
||||
}
|
||||
|
||||
// Seed the counters from persisted state at startup, so they survive restarts.
|
||||
pub fn seed_stats(&mut self, stats: BTreeMap<String, u64>) {
|
||||
self.stats = stats;
|
||||
}
|
||||
|
||||
// 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))?;
|
||||
|
|
|
|||
|
|
@ -181,6 +181,7 @@ fn to_wire(entry: &LogEntry) -> Option<ReplicationEvent> {
|
|||
| Event::ForbidRemoved { .. }
|
||||
| Event::JupeAdded { .. }
|
||||
| Event::JupeRemoved { .. }
|
||||
| Event::StatsSet { .. }
|
||||
| Event::AccountExpiryWarned { .. }
|
||||
| Event::ChannelExpiryWarned { .. }
|
||||
| Event::AccountOperNoteSet { .. }
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@ fn redact(line: &str) -> Cow<'_, str> {
|
|||
let is_set_password = cmd == "set"
|
||||
&& words.next().map(|w| w.eq_ignore_ascii_case("password")).unwrap_or(false);
|
||||
if (SECRET_CMDS.contains(&cmd.as_str()) || is_set_password) && argstart < line.len() {
|
||||
return format!("{} :[REDACTED]", &line[..tstart].trim_end_matches(" :")).into();
|
||||
return format!("{} :[REDACTED]", line[..tstart].trim_end_matches(" :")).into();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -130,6 +130,7 @@ pub async fn run(mut proto: Box<dyn Protocol>, engine: Arc<Mutex<Engine>>, addr:
|
|||
continue;
|
||||
}
|
||||
if let NetAction::Shutdown { restart, reason } = act {
|
||||
engine.lock().await.persist_stats();
|
||||
let _ = write.flush().await;
|
||||
shutdown(restart, &reason);
|
||||
}
|
||||
|
|
@ -146,6 +147,7 @@ pub async fn run(mut proto: Box<dyn Protocol>, engine: Arc<Mutex<Engine>>, 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 _ = write.flush().await;
|
||||
shutdown(restart, &reason);
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -208,8 +208,9 @@ async fn main() -> Result<()> {
|
|||
let engine = engine.clone();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
tokio::time::sleep(std::time::Duration::from_secs(1800)).await;
|
||||
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.expire_sweep();
|
||||
if let Err(err) = e.maybe_compact() {
|
||||
tracing::warn!(%err, "compaction failed");
|
||||
|
|
@ -246,9 +247,12 @@ async fn main() -> Result<()> {
|
|||
// Run until the uplink loop ends or the process is asked to stop. Every
|
||||
// committed change is already fsync'd, so a clean stop loses nothing; this
|
||||
// just lets systemd stop us without waiting out the kill timeout.
|
||||
let shutdown_engine = engine.clone();
|
||||
tokio::select! {
|
||||
res = link::run(proto, engine, &addr, irc_rx, cfg.email.clone()) => res,
|
||||
_ = shutdown_signal() => {
|
||||
// Flush stat counters so a clean stop/restart keeps StatServ history.
|
||||
shutdown_engine.lock().await.persist_stats();
|
||||
tracing::info!("received shutdown signal, exiting");
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue