db: persist jupes across restart (event-source them)

A fold-parity property test caught it: jupes were plain in-memory Db fields
that logged no event, so after any services restart every juped server was
silently un-juped and could relink. Moved jupes into NetData (like akills/
forbids/groups) and event-sourced them with JupeAdded/JupeRemoved (Local
scope), so a jupe now replays from the log. Adds the round-trip property test
(live state == replayed state) as a permanent guard against this bug class.

src/engine/db/event.rs

@@ -132,6 +132,10 @@ pub enum Event {
     // "NICK", "CHAN", or "EMAIL".
     ForbidAdded { kind: String, mask: String, setter: String, reason: String, ts: u64 },
     ForbidRemoved { kind: String, mask: String },
+    // Server jupes (OperServ JUPE). Node-local (each network holds a rogue server
+    // by minting a fake one on `sid`), so Local scope but still persisted.
+    JupeAdded { name: String, sid: String, reason: String },
+    JupeRemoved { name: String },
 }
 
 // Whether an event replicates across the federation. Account identity is Global
@@ -232,7 +236,9 @@ impl Event {
             | Event::ChannelUsed { .. }
             | Event::ChannelNoExpire { .. }
             | Event::ChannelExpiryWarned { .. }
-            | Event::ChannelOperNoteSet { .. } => Scope::Local,
+            | Event::ChannelOperNoteSet { .. }
+            | Event::JupeAdded { .. }
+            | Event::JupeRemoved { .. } => Scope::Local,
         }
     }
 }
@@ -603,6 +609,17 @@ pub(crate) fn apply(accounts: &mut HashMap<String, Account>, channels: &mut Hash
         Event::ForbidRemoved { kind, mask } => {
             net.forbids.retain(|f| !(f.kind == kind && f.mask.eq_ignore_ascii_case(&mask)));
         }
+        Event::JupeAdded { name, sid, reason } => {
+            // Idempotent over a snapshot; keep jupe_seq ahead so a fresh add can't
+            // reuse a sid.
+            if !net.jupes.iter().any(|j| j.name.eq_ignore_ascii_case(&name)) {
+                net.jupes.push(Jupe { name, sid, reason });
+                net.jupe_seq += 1;
+            }
+        }
+        Event::JupeRemoved { name } => {
+            net.jupes.retain(|j| !j.name.eq_ignore_ascii_case(&name));
+        }
         Event::AccountExpiryWarned { account } => {
             if let Some(a) = accounts.get_mut(&key(&account)) {
                 a.expiry_warned = true;

src/engine/db/mod.rs

@@ -301,6 +301,9 @@ pub struct NetData {
     pub opers: HashMap<String, OperGrant>,
     // Session-limit exceptions (OperServ SESSION): per-IP-mask allowances.
     pub sess_exceptions: Vec<SessionException>,
+    // Server jupes (OperServ JUPE), and the counter for minting their fake SIDs.
+    pub jupes: Vec<Jupe>,
+    pub jupe_seq: u32,
     // The bot auto-assigned to newly registered channels (BotServ AUTOASSIGN),
     // if any. Casefolded nick; None disables auto-assignment.
     pub default_bot: Option<String>,
@@ -947,10 +950,6 @@ pub struct Db {
     net: NetData,
     // Services ignores (OperServ IGNORE), node-local and in-memory.
     ignores: Vec<Ignore>,
-    // Juped servers (OperServ JUPE), node-local: the introducing node owns them,
-    // so they aren't federated (that would collide SIDs across nodes).
-    jupes: Vec<Jupe>,
-    jupe_seq: u32,
     // Network defence level (OperServ DEFCON), 5 = normal down to 1 = lockdown.
     // Node-local; the derived restrictions are read by the engine and services.
     defcon: u8,
@@ -1026,7 +1025,7 @@ impl Db {
             apply(&mut accounts, &mut channels, &mut grouped, &mut bots, &mut host_cfg, &mut net, event);
         }
         tracing::info!(accounts = accounts.len(), channels = channels.len(), "account store loaded");
-        Self { accounts, channels, grouped, log, scram_iterations: scram::DEFAULT_ITERATIONS, email_enabled: false, email_brand: "Network Services".to_string(), email_accent: "#4f46e5".to_string(), email_logo: String::new(), codes: HashMap::new(), auth_fails: HashMap::new(), vhost_req_times: HashMap::new(), report_times: HashMap::new(), bots, host_cfg, net, ignores: Vec::new(), jupes: Vec::new(), jupe_seq: 0, defcon: 5, external_accounts: false }
+        Self { accounts, channels, grouped, log, scram_iterations: scram::DEFAULT_ITERATIONS, email_enabled: false, email_brand: "Network Services".to_string(), email_accent: "#4f46e5".to_string(), email_logo: String::new(), codes: HashMap::new(), auth_fails: HashMap::new(), vhost_req_times: HashMap::new(), report_times: HashMap::new(), bots, host_cfg, net, ignores: Vec::new(), defcon: 5, external_accounts: false }
     }
 
     /// Fold an entry authored by another node into the store — the services-side

src/engine/db/network.rs

@@ -221,25 +221,28 @@ impl Db {
     /// Jupe a server name: allocate a fake sid, store it, return the sid to
     /// introduce (or the existing sid if the name is already juped).
     pub fn jupe_add(&mut self, name: &str, reason: &str) -> String {
-        if let Some(j) = self.jupes.iter().find(|j| j.name.eq_ignore_ascii_case(name)) {
+        if let Some(j) = self.net.jupes.iter().find(|j| j.name.eq_ignore_ascii_case(name)) {
             return j.sid.clone();
         }
-        let sid = jupe_sid(self.jupe_seq);
+        let sid = jupe_sid(self.net.jupe_seq);
-        self.jupe_seq += 1;
+        // Persisted (Local scope) so a juped server stays juped across a restart.
-        self.jupes.push(Jupe { name: name.to_string(), sid: sid.clone(), reason: reason.to_string() });
+        let _ = self.log.append(Event::JupeAdded { name: name.to_string(), sid: sid.clone(), reason: reason.to_string() });
+        self.net.jupe_seq += 1;
+        self.net.jupes.push(Jupe { name: name.to_string(), sid: sid.clone(), reason: reason.to_string() });
         sid
     }
 
     /// Lift a jupe. Returns the sid to squit if it existed.
     pub fn jupe_del(&mut self, name: &str) -> Option<String> {
-        let sid = self.jupes.iter().find(|j| j.name.eq_ignore_ascii_case(name)).map(|j| j.sid.clone())?;
+        let sid = self.net.jupes.iter().find(|j| j.name.eq_ignore_ascii_case(name)).map(|j| j.sid.clone())?;
-        self.jupes.retain(|j| !j.name.eq_ignore_ascii_case(name));
+        let _ = self.log.append(Event::JupeRemoved { name: name.to_string() });
+        self.net.jupes.retain(|j| !j.name.eq_ignore_ascii_case(name));
         Some(sid)
     }
 
     /// The juped servers, as (name, sid, reason).
     pub fn jupes(&self) -> Vec<(String, String, String)> {
-        self.jupes.iter().map(|j| (j.name.clone(), j.sid.clone(), j.reason.clone())).collect()
+        self.net.jupes.iter().map(|j| (j.name.clone(), j.sid.clone(), j.reason.clone())).collect()
     }
 
     /// File an abuse report, rate-limited per reporter. Returns the new report's

src/engine/db/tests.rs

@@ -462,6 +462,89 @@
         assert_eq!(db.channel("#c").unwrap().join_mode("alice"), None, "removal survives reopen");
     }
 
+    // A jupe must survive a services restart — otherwise every juped server is
+    // silently un-juped on restart and can relink.
+    #[test]
+    fn jupes_survive_a_restart() {
+        let p = tmp("jupepersist");
+        let sid = {
+            let mut db = Db::open(&p, "N1");
+            let sid = db.jupe_add("rogue.example", "abuse");
+            assert_eq!(db.jupes().len(), 1, "juped while live");
+            sid
+        };
+        let db = Db::open(&p, "N1");
+        assert_eq!(db.jupes(), vec![("rogue.example".to_string(), sid, "abuse".to_string())], "the jupe replays from the log");
+    }
+
+    // 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.
+    #[test]
+    fn live_state_matches_replay_after_a_broad_sequence() {
+        let p = tmp("foldparity");
+        let snapshot = |db: &Db| -> String {
+            let mut a: Vec<_> = db.accounts().cloned().collect();
+            a.sort_by(|x, y| x.name.cmp(&y.name));
+            let mut c: Vec<_> = db.channels().cloned().collect();
+            c.sort_by(|x, y| x.name.cmp(&y.name));
+            let mut gn = db.groups();
+            gn.sort();
+            let gv: Vec<_> = gn.iter().filter_map(|n| db.group(n)).collect();
+            let mut ak = db.akills();
+            ak.sort_by(|x, y| x.mask.cmp(&y.mask));
+            let mut fb = db.forbids();
+            fb.sort_by(|x, y| x.mask.cmp(&y.mask));
+            let mut of = db.vhost_offers().to_vec();
+            of.sort();
+            let mut jp = db.jupes();
+            jp.sort();
+            format!("{a:?}\n{c:?}\n{gv:?}\n{ak:?}\n{fb:?}\n{of:?}\n{jp:?}")
+        };
+
+        let live = {
+            let mut db = Db::open(&p, "N1");
+            db.scram_iterations = 4096;
+            db.register("alice", "pw", Some("a@x.io".into())).unwrap();
+            db.register("bob", "pw", None).unwrap();
+            db.register("carol", "pw", None).unwrap();
+            db.register_channel("#chan", "bob").unwrap();
+            db.register_channel("#other", "carol").unwrap();
+            db.access_add("#chan", "alice", "op").unwrap();
+            db.akick_add("#chan", "*!*@bad.host", "spam").unwrap();
+            db.set_desc("#chan", "a place").unwrap();
+            db.set_url("#chan", "https://x.io").unwrap();
+            db.set_channel_email("#chan", "s@x.io").unwrap();
+            db.set_mlock("#chan", "nt", "s").unwrap();
+            db.set_channel_setting("#chan", ChanSetting::Private, true).unwrap();
+            db.set_successor("#other", Some("alice")).unwrap();
+            db.set_greet("alice", "hi there").unwrap();
+            db.set_account_autoop("alice", false).unwrap();
+            db.set_account_kill("alice", false).unwrap();
+            db.set_account_hide_status("alice", true).unwrap();
+            db.ajoin_add("alice", "#chan", "").unwrap();
+            db.set_vhost("alice", "alice.vhost", "oper", None).unwrap();
+            db.certfp_add("alice", "aabbccddeeff00112233445566778899").unwrap();
+            db.suspend_account("bob", "oper", "bad", None).unwrap();
+            db.memo_send("carol", "alice", "hello there", false).unwrap();
+            db.group_register("!grp", "alice").unwrap();
+            db.group_set_flags("!grp", "carol", "").unwrap();
+            db.group_register("!other", "bob").unwrap();
+            db.group_set_flags("!other", "alice", "").unwrap();
+            db.akill_add("G", "*@evil", "oper", "spam", None).unwrap();
+            db.forbid_add("NICK", "bad*", "oper", "squat").unwrap();
+            db.vhost_offer_add("cool.vhost").unwrap();
+            db.jupe_add("rogue.server", "abuse");
+            // The compound one: dropping alice must purge her access, successorship
+            // and group standing — identically in the live path and on replay.
+            db.drop_account("alice").unwrap();
+            db.drop_channel("#other").unwrap();
+            snapshot(&db)
+        };
+        let replayed = snapshot(&Db::open(&p, "N1"));
+        assert_eq!(live, replayed, "live state must equal the state replayed from the log");
+    }
+
     // Dropping an account erases every reference to it — channel access and
     // successorship, group membership and foundership — so a later re-registration
     // of the same name can't inherit its standing. Holds live and after replay.

src/engine/mod.rs

@@ -1300,6 +1300,8 @@ fn audit_summary(event: &db::Event) -> Option<String> {
         AkillRemoved { kind, mask } => format!("lifted the {} on \x02{mask}\x02", ban_kind_label(kind)),
         ForbidAdded { kind, mask, reason, .. } => format!("forbade {} \x02{mask}\x02 ({reason})", kind.to_ascii_lowercase()),
         ForbidRemoved { kind, mask } => format!("un-forbade {} \x02{mask}\x02", kind.to_ascii_lowercase()),
+        JupeAdded { name, reason, .. } => format!("juped server \x02{name}\x02 ({reason})"),
+        JupeRemoved { name } => format!("lifted the jupe on \x02{name}\x02"),
         AccountOperNoteSet { account, note } => match note {
             Some(_) => format!("set a staff note on \x02{account}\x02"),
             None => format!("cleared the staff note on \x02{account}\x02"),

src/grpc.rs

@@ -179,6 +179,8 @@ fn to_wire(entry: &LogEntry) -> Option<ReplicationEvent> {
         | Event::AkillRemoved { .. }
         | Event::ForbidAdded { .. }
         | Event::ForbidRemoved { .. }
+        | Event::JupeAdded { .. }
+        | Event::JupeRemoved { .. }
         | Event::AccountExpiryWarned { .. }
         | Event::ChannelExpiryWarned { .. }
         | Event::AccountOperNoteSet { .. }