feat(security): add clear_and_install_from for in-place registry repair

All in-memory security and event registries (permissions, roles, API keys,
blacklist, credentials, RLS policies, change streams, consumer groups,
scheduler, streaming MVs, alerts, triggers, retention policies) gain a
clear_and_install_from method used by the catalog recovery sanity checker.

When the checker detects divergence between the in-memory registry and the
redb system catalog, it loads a fresh store from redb and calls
clear_and_install_from to repair in place, keeping all existing Arc
references stable so listeners need not be restarted.

Author: farhan-syah

nodedb/src/control/security/apikey.rs

@@ -137,6 +137,19 @@ impl ApiKeyStore {
         Ok(())
     }
 
+    /// Clear the in-memory key map and re-run `load_from`.
+    /// Used by the catalog recovery sanity checker to repair
+    /// a divergent registry.
+    pub(crate) fn clear_and_reload(&self, catalog: &SystemCatalog) -> crate::Result<()> {
+        {
+            let mut keys = self.keys.write().map_err(|e| crate::Error::Internal {
+                detail: format!("api key lock poisoned during repair: {e}"),
+            })?;
+            keys.clear();
+        }
+        self.load_from(catalog)
+    }
+
     /// Persist a single key record to the catalog.
     fn persist_to(&self, catalog: &SystemCatalog, record: &ApiKeyRecord) -> crate::Result<()> {
         catalog.put_api_key(&record.to_stored())

nodedb/src/control/security/blacklist/store.rs

@@ -281,6 +281,30 @@ impl BlacklistStore {
             .collect()
     }
 
+    /// All in-memory entries (including potentially expired ones that
+    /// haven't been lazily evicted yet). Used by the recovery verifier
+    /// for exact redb↔memory comparison.
+    pub fn list_all_entries(&self) -> Vec<BlacklistEntry> {
+        let entries = self.entries.read().unwrap_or_else(|p| p.into_inner());
+        entries.values().cloned().collect()
+    }
+
+    /// Clear all in-memory entries and reload from catalog.
+    /// Used by the recovery verifier repair path.
+    pub fn clear_and_reload(&self, catalog: &SystemCatalog) -> crate::Result<()> {
+        // Reload by clearing first then re-applying — load_from only appends.
+        let stored = catalog.load_all_blacklist_entries()?;
+        let mut entries = self.entries.write().unwrap_or_else(|p| p.into_inner());
+        entries.clear();
+        for s in stored {
+            let entry = BlacklistEntry::from_stored(&s);
+            if !entry.is_expired() {
+                entries.insert(entry.key.clone(), entry);
+            }
+        }
+        Ok(())
+    }
+
     /// Total active entries.
     pub fn count(&self) -> usize {
         let entries = self.entries.read().unwrap_or_else(|p| p.into_inner());

nodedb/src/control/security/catalog/collection_constraints.rs

@@ -88,23 +88,23 @@ pub struct LegalHold {
 }
 
 /// State transition constraint: column value can only change along declared paths.
-#[derive(Serialize, Deserialize, ToMessagePack, FromMessagePack, Debug, Clone)]
+#[derive(Serialize, Deserialize, ToMessagePack, FromMessagePack, Debug, Clone, PartialEq)]
 pub struct StateTransitionDef {
     pub name: String,
     pub column: String,
     pub transitions: Vec<TransitionRule>,
 }
 
 /// A single allowed state transition, optionally guarded by a role.
-#[derive(Serialize, Deserialize, ToMessagePack, FromMessagePack, Debug, Clone)]
+#[derive(Serialize, Deserialize, ToMessagePack, FromMessagePack, Debug, Clone, PartialEq)]
 pub struct TransitionRule {
     pub from: String,
     pub to: String,
     pub required_role: Option<String>,
 }
 
 /// Transition check predicate: evaluated on UPDATE with OLD and NEW access.
-#[derive(Serialize, Deserialize, ToMessagePack, FromMessagePack, Debug, Clone)]
+#[derive(Serialize, Deserialize, ToMessagePack, FromMessagePack, Debug, Clone, PartialEq)]
 pub struct TransitionCheckDef {
     pub name: String,
     pub predicate: SqlExpr,

nodedb/src/control/security/credential/store/list.rs

@@ -15,6 +15,37 @@ impl CredentialStore {
         users.values().filter(|u| u.is_active).cloned().collect()
     }
 
+    /// List ALL user records (active and inactive). Used by the
+    /// recovery verifier for a complete redb↔memory comparison.
+    pub fn list_all_user_details(&self) -> Vec<UserRecord> {
+        let users = match read_lock(&self.users) {
+            Ok(u) => u,
+            Err(_) => return Vec::new(),
+        };
+        users.values().cloned().collect()
+    }
+
+    /// Reload all users from the given catalog into the in-memory cache.
+    /// Used by the recovery verifier repair path.
+    pub fn reload_from_catalog(&self, catalog: &SystemCatalog) -> crate::Result<()> {
+        use super::super::record::UserRecord;
+        let stored_users = catalog.load_all_users()?;
+        let mut users = match self.users.write() {
+            Ok(u) => u,
+            Err(_) => {
+                return Err(crate::Error::Internal {
+                    detail: "credential store write lock poisoned in reload_from_catalog".into(),
+                });
+            }
+        };
+        users.clear();
+        for stored in stored_users {
+            let record = UserRecord::from_stored(stored);
+            users.insert(record.username.clone(), record);
+        }
+        Ok(())
+    }
+
     /// List all active usernames.
     pub fn list_users(&self) -> Vec<String> {
         let users = match read_lock(&self.users) {

nodedb/src/control/security/permission/store.rs

@@ -157,6 +157,83 @@ impl PermissionStore {
             .collect()
     }
 
+    /// Replace the entire in-memory grants + owners state
+    /// with the contents of `other`. Used by the catalog
+    /// recovery sanity checker to repair a divergent registry
+    /// by loading a fresh `PermissionStore` from redb and then
+    /// swapping its contents into `self`. Callers keep their
+    /// existing `Arc<PermissionStore>` reference stable.
+    pub(crate) fn clear_and_install_from(&self, other: &Self) {
+        let fresh_grants = other.snapshot_grants();
+        let fresh_owners = other.snapshot_owners();
+        let mut grants = match self.grants.write() {
+            Ok(g) => g,
+            Err(p) => {
+                tracing::error!("permission grants lock poisoned during repair — recovering");
+                p.into_inner()
+            }
+        };
+        grants.clear();
+        for g in fresh_grants {
+            grants.insert(g);
+        }
+        drop(grants);
+        let mut owners = match self.owners.write() {
+            Ok(o) => o,
+            Err(p) => {
+                tracing::error!("owner store lock poisoned during repair — recovering");
+                p.into_inner()
+            }
+        };
+        owners.clear();
+        for (k, v) in fresh_owners {
+            owners.insert(k, v);
+        }
+    }
+
+    /// Deterministic snapshot of every grant held in memory,
+    /// sorted by `(target, grantee, permission)` so diff-based
+    /// callers (the recovery sanity checker) can compare
+    /// against a catalog load without caring about HashSet
+    /// iteration order.
+    pub fn snapshot_grants(&self) -> Vec<Grant> {
+        let grants = match self.grants.read() {
+            Ok(g) => g,
+            Err(p) => p.into_inner(),
+        };
+        let mut out: Vec<Grant> = grants.iter().cloned().collect();
+        out.sort_by(|a, b| {
+            let a_key = (
+                a.target.clone(),
+                a.grantee.clone(),
+                format_permission(a.permission),
+            );
+            let b_key = (
+                b.target.clone(),
+                b.grantee.clone(),
+                format_permission(b.permission),
+            );
+            a_key.cmp(&b_key)
+        });
+        out
+    }
+
+    /// Deterministic snapshot of every owner held in memory as
+    /// `(owner_key, username)` pairs, sorted by key.
+    /// `owner_key` is the internal `"collection:{tenant}:{name}"`
+    /// composite — used by the sanity checker to cross-check
+    /// against `catalog.load_all_owners()`.
+    pub fn snapshot_owners(&self) -> Vec<(String, String)> {
+        let owners = match self.owners.read() {
+            Ok(o) => o,
+            Err(p) => p.into_inner(),
+        };
+        let mut out: Vec<(String, String)> =
+            owners.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
+        out.sort_by(|a, b| a.0.cmp(&b.0));
+        out
+    }
+
     /// List all grants on a target.
     pub fn grants_on(&self, target: &str) -> Vec<Grant> {
         let grants = match self.grants.read() {

nodedb/src/control/security/rls/store.rs

@@ -101,6 +101,36 @@ impl RlsPolicyStore {
             .unwrap_or_default()
     }
 
+    /// Flat list of all policies (all tenants, all collections).
+    /// Used by the recovery verifier.
+    pub fn list_all_flat(&self) -> Vec<RlsPolicy> {
+        let policies = self.lock_read();
+        policies.values().flat_map(|v| v.iter().cloned()).collect()
+    }
+
+    /// Clear all in-memory policies and reload from the catalog.
+    /// Used by the recovery verifier repair path.
+    pub fn clear_and_reload(
+        &self,
+        catalog: &crate::control::security::catalog::SystemCatalog,
+    ) -> crate::Result<()> {
+        let stored = catalog.load_all_rls_policies()?;
+        let mut policies = self.lock_write();
+        policies.clear();
+        for s in stored {
+            match s.to_runtime() {
+                Ok(rp) => {
+                    let key = super::types::policy_key(rp.tenant_id, &rp.collection);
+                    policies.entry(key).or_default().push(rp);
+                }
+                Err(e) => {
+                    tracing::warn!(error = %e, "rls_store.clear_and_reload: skipping unparseable policy");
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// Total policies across all collections.
     pub fn policy_count(&self) -> usize {
         self.policies

nodedb/src/control/security/role.rs

@@ -64,6 +64,20 @@ impl RoleStore {
         Ok(())
     }
 
+    /// Clear the in-memory role map and re-run `load_from`.
+    /// Used by the catalog recovery sanity checker to repair
+    /// a divergent registry. Callers keep their existing
+    /// `&RoleStore` reference.
+    pub(crate) fn clear_and_reload(&self, catalog: &SystemCatalog) -> crate::Result<()> {
+        {
+            let mut roles = self.roles.write().map_err(|e| crate::Error::Internal {
+                detail: format!("role store lock poisoned during repair: {e}"),
+            })?;
+            roles.clear();
+        }
+        self.load_from(catalog)
+    }
+
     // ── Cluster replication hooks ──────────────────────────────
     //
     // Symmetric partners to `CredentialStore::install_replicated_user`:

nodedb/src/control/trigger/registry.rs

@@ -152,6 +152,51 @@ impl TriggerRegistry {
         }
     }
 
+    /// Replace the entire in-memory trigger map with `rows`.
+    /// Used by the catalog recovery sanity checker to repair
+    /// a divergent registry by re-loading from redb. Callers
+    /// keep their existing `&TriggerRegistry` reference.
+    pub(crate) fn clear_and_install_all(&self, rows: Vec<StoredTrigger>) {
+        let mut map = match self.by_collection.write() {
+            Ok(m) => m,
+            Err(p) => p.into_inner(),
+        };
+        map.clear();
+        for trigger in rows {
+            let key = (trigger.tenant_id, trigger.collection.clone());
+            map.entry(key).or_default().push(trigger);
+        }
+        for list in map.values_mut() {
+            list.sort_by(|a, b| a.sort_key().cmp(&b.sort_key()));
+        }
+    }
+
+    /// Deterministic snapshot of every trigger across every
+    /// tenant, sorted by `(tenant_id, collection, name)` so the
+    /// recovery sanity checker can diff against
+    /// `catalog.load_all_triggers()` without caring about
+    /// HashMap iteration order.
+    pub fn snapshot_all(&self) -> Vec<StoredTrigger> {
+        let map = match self.by_collection.read() {
+            Ok(m) => m,
+            Err(p) => p.into_inner(),
+        };
+        let mut result: Vec<StoredTrigger> = Vec::new();
+        for list in map.values() {
+            for t in list {
+                result.push(t.clone());
+            }
+        }
+        result.sort_by(|a, b| {
+            (a.tenant_id, a.collection.clone(), a.name.clone()).cmp(&(
+                b.tenant_id,
+                b.collection.clone(),
+                b.name.clone(),
+            ))
+        });
+        result
+    }
+
     /// List all triggers for a tenant (for SHOW TRIGGERS).
     pub fn list_for_tenant(&self, tenant_id: u32) -> Vec<StoredTrigger> {
         let map = match self.by_collection.read() {

nodedb/src/engine/timeseries/retention_policy/registry.rs

@@ -84,6 +84,32 @@ impl RetentionPolicyRegistry {
             .collect()
     }
 
+    /// List all policies (all tenants, enabled and disabled).
+    /// Used by the recovery verifier.
+    pub fn list_all(&self) -> Vec<RetentionPolicyDef> {
+        self.policies
+            .read()
+            .expect("registry lock poisoned")
+            .values()
+            .cloned()
+            .collect()
+    }
+
+    /// Clear and reload from catalog. Used by the recovery verifier repair path.
+    pub fn clear_and_reload(
+        &self,
+        catalog: &crate::control::security::catalog::types::SystemCatalog,
+    ) -> crate::Result<()> {
+        let fresh = catalog.load_all_retention_policies()?;
+        let mut map = self.policies.write().expect("registry lock poisoned");
+        map.clear();
+        for p in fresh {
+            let key = (p.tenant_id, p.name.clone());
+            map.insert(key, p);
+        }
+        Ok(())
+    }
+
     /// List all policies for a tenant.
     pub fn list_for_tenant(&self, tenant_id: u32) -> Vec<RetentionPolicyDef> {
         self.policies

nodedb/src/event/alert/registry.rs

@@ -57,6 +57,27 @@ impl AlertRegistry {
             .collect()
     }
 
+    /// List all alerts (all tenants, enabled and disabled).
+    /// Used by the recovery verifier.
+    pub fn list_all(&self) -> Vec<AlertDef> {
+        self.read_map().values().cloned().collect()
+    }
+
+    /// Clear and reload from catalog. Used by the recovery verifier repair path.
+    pub fn clear_and_reload(
+        &self,
+        catalog: &crate::control::security::catalog::types::SystemCatalog,
+    ) -> crate::Result<()> {
+        let fresh = catalog.load_all_alert_rules()?;
+        let mut map = self.write_map();
+        map.clear();
+        for alert in fresh {
+            let key = (alert.tenant_id, alert.name.clone());
+            map.insert(key, alert);
+        }
+        Ok(())
+    }
+
     /// List all alerts for a tenant.
     pub fn list_for_tenant(&self, tenant_id: u32) -> Vec<AlertDef> {
         self.read_map()