event_graph: Rebuild stale RLN state on startup

src/event_graph/mod.rs

@@ -737,11 +737,6 @@ impl EventGraph {
             rln_app_id,
         });
 
-        // Init genesis registration events
-        if config.hours_rotation > 0 {
-            self_.bootstrap_genesis_identities().await?;
-        }
-
         if need_prune {
             info!(
                 target: "event_graph::new",
@@ -750,17 +745,18 @@ impl EventGraph {
             self_.dag_prune(current_genesis).await?;
         }
 
-        // Consistency check: if the static DAG has events but the
-        // historical-roots tables are empty, rebuild them by
-        // replaying the static DAG in canonical order. This handles
-        // the case where the operator manually deleted the
-        // historical-roots trees, or where this is the first startup
-        // after upgrading from a version that didn't track them.
-        //
-        // Without this, signal verification would fail for any root
-        // beyond the in-memory `recent_roots` window.
+        // Reconcile persisted RLN state before bootstrapping. If an
+        // earlier process crashed after writing identity leaves but before
+        // inserting the corresponding static event, bootstrapping must see
+        // the corrected leaf set rather than skip the configured identity.
         self_.rebuild_historical_roots_if_needed().await?;
 
+        // Init genesis registration events after recovery has made the
+        // static DAG authoritative for the current identity tree.
+        if config.hours_rotation > 0 {
+            self_.bootstrap_genesis_identities().await?;
+        }
+
         if config.hours_rotation > 0 {
             let task = StoppableTask::new();
             let _ = self_.prune_task.set(task.clone()).await;
@@ -780,126 +776,147 @@ impl EventGraph {
         Ok(self_)
     }
 
-    /// Rebuild the historical-roots side-tables from the static DAG.
+    /// Rebuild the RLN state side-tables from the static DAG.
     ///
-    /// Called once at startup. No-op if the historical-roots tables
-    /// already match the static-DAG event count. Otherwise replays
-    /// every static-DAG event in canonical `(layer, event_id)` order
-    /// and re-records the post-mutation root for each one.
+    /// Called once at startup. No-op if the historical-root indexes match
+    /// the canonical static-DAG event sequence and the persisted identity
+    /// leaves match the commitment set obtained by replaying that sequence.
+    /// Otherwise resets the identity SMT and root indexes, then replays every
+    /// parseable static-DAG event in canonical `(layer, event_id)` order.
     ///
-    /// **Side effect.** Resets the in-memory SMT to empty, then
-    /// rebuilds it leaf-by-leaf in canonical order, so the SMT and
-    /// the historical-roots tables come out consistent. The
-    /// `rln-identity-leaves` tree (which `IdentityState::new`
-    /// originally read) is implicitly re-derived; we don't read it
-    /// during rebuild because we want to honor any slashes in the
-    /// static DAG even if the leaves tree is stale.
+    /// **Side effect.** The static DAG is authoritative. The in-memory SMT,
+    /// the persistent `rln-identity-leaves` tree, and both historical-root
+    /// indexes are derived from it so crashes between the old split write
+    /// steps cannot leave stale leaves or unusable root indexes behind.
     async fn rebuild_historical_roots_if_needed(self: &Arc<Self>) -> Result<()> {
-        // Walk the static DAG once, computing both:
-        //   * static_count: total non-genesis events
-        //   * expected_leaves: registrations - slashes (the number
-        //     of identities that should currently be in the SMT)
-        // We need the second one to detect a state where leaves and
-        // historical-roots happen to share counts but the leaves
-        // don't actually correspond to the static-DAG events. That
-        // can happen across schema changes or when older code paths
-        // wrote to leaves without going through `apply_rln_static_event`.
-        let mut static_count: u64 = 0;
-        let mut registrations: i64 = 0;
-        let mut slashes: i64 = 0;
+        let mut events: Vec<(Event, rln::RLNNode)> = vec![];
+
         for item in self.static_dag.iter() {
             let (_, val) = item?;
             let ev: Event = deserialize_async(&val).await?;
             if ev.header.parents == NULL_PARENTS {
                 continue
             }
-            static_count += 1;
-            // Try to classify this event. We tolerate failed parses
-            // here because the rebuild path is best-effort: if an
-            // event's content is unparseable, we just don't count it
-            // toward expected_leaves. The replay loop below skips
-            // it for the same reason.
-            if let Ok((node, _)) = deserialize_async_partial::<rln::RLNNode>(ev.content()).await {
-                match node {
-                    rln::RLNNode::Registration(_) => registrations += 1,
-                    rln::RLNNode::Slashing(_) => slashes += 1,
+
+            let Ok((node, _)) = deserialize_async_partial::<rln::RLNNode>(ev.content()).await
+            else {
+                continue
+            };
+            events.push((ev, node));
+        }
+
+        events.sort_by(|(a, _), (b, _)| {
+            a.header
+                .layer
+                .cmp(&b.header.layer)
+                .then_with(|| a.id().as_bytes().cmp(b.id().as_bytes()))
+        });
+
+        let mut expected_commitments = BTreeSet::new();
+        for (_, node) in &events {
+            match node {
+                rln::RLNNode::Registration(commitment) => {
+                    expected_commitments.insert(commitment.to_repr());
+                }
+                rln::RLNNode::Slashing(commitment) => {
+                    expected_commitments.remove(&commitment.to_repr());
                 }
             }
         }
-        let expected_leaves = (registrations - slashes).max(0) as usize;
 
-        let recorded_count = self.rln_historical_roots_ordered.len() as u64;
-        let actual_leaves = self.identity_state.read().await.leaves_count();
+        let expected_leaves = expected_commitments.len();
+        let actual_commitments = self.identity_state.read().await.commitment_reprs();
+        let (actual_leaves, leaves_consistent) = match actual_commitments {
+            Ok(commitments) => {
+                let len = commitments.len();
+                (len, commitments == expected_commitments)
+            }
+            Err(e) => {
+                warn!(
+                    target: "event_graph::new",
+                    "[EVENTGRAPH] RLN identity leaf audit failed: {e}; rebuilding",
+                );
+                (0, false)
+            }
+        };
 
-        let counts_consistent = recorded_count == static_count;
-        let leaves_consistent = actual_leaves == expected_leaves;
+        let static_count = events.len();
+        let historical_roots_consistent = self.historical_roots_index_consistent(static_count)?;
+        let recorded_count = self.rln_historical_roots_ordered.len();
+        let by_value_count = self.rln_historical_roots_by_value.len();
+        let consistent = historical_roots_consistent && leaves_consistent;
 
         info!(
             target: "event_graph::new",
-            "[EVENTGRAPH] RLN state audit: static_count={} recorded_count={} \
-             actual_leaves={} expected_leaves={} consistent={}",
-            static_count, recorded_count, actual_leaves, expected_leaves,
-            counts_consistent && leaves_consistent,
+            concat!(
+                "[EVENTGRAPH] RLN state audit: static_count={} recorded_count={} ",
+                "by_value_count={} actual_leaves={} expected_leaves={} consistent={}",
+            ),
+            static_count, recorded_count, by_value_count, actual_leaves, expected_leaves,
+            consistent,
         );
 
-        if counts_consistent && leaves_consistent {
-            // Already consistent across all three sources (static
-            // DAG, historical-roots table, leaves tree).
+        if consistent {
             return Ok(())
         }
 
         info!(
             target: "event_graph::new",
-            "[EVENTGRAPH] Rebuilding historical-roots: {} static events, \
-             {} recorded roots, {} leaves (expected {})",
-            static_count, recorded_count, actual_leaves, expected_leaves,
+            concat!(
+                "[EVENTGRAPH] Rebuilding RLN state: {} static events, {} recorded roots, ",
+                "{} by-value roots, {} leaves (expected {})",
+            ),
+            static_count, recorded_count, by_value_count, actual_leaves, expected_leaves,
         );
 
-        // Reset the historical-roots tables to a known-empty state.
         self.rln_historical_roots_ordered.clear()?;
         self.rln_historical_roots_by_value.clear()?;
 
-        // Reset the in-memory SMT and the leaves tree so the replay
-        // below builds it correctly from the canonical static-DAG
-        // sequence (including any slashes).
         {
             let mut state = self.identity_state.write().await;
             state.clear_for_rebuild()?;
         }
 
-        // Collect static-DAG events and sort canonically.
-        let mut events: Vec<Event> = vec![];
-        for item in self.static_dag.iter() {
-            let (_, val) = item?;
-            let ev: Event = deserialize_async(&val).await?;
-            if ev.header.parents != NULL_PARENTS {
-                events.push(ev);
-            }
-        }
-        events.sort_by(|a, b| {
-            a.header
-                .layer
-                .cmp(&b.header.layer)
-                .then_with(|| a.id().as_bytes().cmp(b.id().as_bytes()))
-        });
-
-        // Replay each event through the canonical apply path.
-        for ev in events {
-            let rln_node: rln::RLNNode = match deserialize_async_partial(ev.content()).await {
-                Ok((v, _)) => v,
-                Err(_) => continue,
-            };
+        for (ev, rln_node) in events {
             let _ = self.apply_rln_static_event(&ev, &rln_node).await?;
         }
 
         info!(
             target: "event_graph::new",
-            "[EVENTGRAPH] Historical-roots rebuild complete",
+            "[EVENTGRAPH] RLN state rebuild complete",
         );
 
         Ok(())
     }
 
+    fn historical_roots_index_consistent(&self, expected_count: usize) -> Result<bool> {
+        if self.rln_historical_roots_ordered.len() != expected_count {
+            return Ok(false)
+        }
+        if self.rln_historical_roots_by_value.len() != expected_count {
+            return Ok(false)
+        }
+
+        for item in self.rln_historical_roots_ordered.iter() {
+            let (ordered_key_bytes, value_bytes) = item?;
+            if ordered_key_bytes.len() != 40 {
+                return Ok(false)
+            }
+            let Ok((root, _)) = decode_historical_root_value(&value_bytes) else {
+                return Ok(false)
+            };
+
+            let mut ordered_key = [0u8; 40];
+            ordered_key.copy_from_slice(&ordered_key_bytes);
+            let by_value_key = encode_historical_root_by_value_key(&root, &ordered_key);
+            if !self.rln_historical_roots_by_value.contains_key(by_value_key)? {
+                return Ok(false)
+            }
+        }
+
+        Ok(true)
+    }
+
     /// After header sync, event content can be fetched lazily via
     /// [`fetch_page`] or peer [`RangeReq`] - the application pulls
     /// the events it actually wants to display or process, without

src/event_graph/rln.rs

@@ -26,7 +26,7 @@
 //! proof to remove them from the identity tree.
 
 use std::{
-    collections::{BTreeMap, VecDeque},
+    collections::{BTreeMap, BTreeSet, VecDeque},
     io::Cursor,
 };
 
@@ -336,7 +336,10 @@ impl IdentityState {
 
         let mut batch = vec![];
         for item in leaves.iter() {
-            let (_, val) = item?;
+            let (key, val) = item?;
+            if key.len() != 32 || val.len() != 32 || key.as_ref() != val.as_ref() {
+                continue
+            }
             let mut repr = [0u8; 32];
             repr.copy_from_slice(&val);
             if let Some(c) = pallas::Base::from_repr(repr).into() {
@@ -398,13 +401,31 @@ impl IdentityState {
         self.smt.root()
     }
 
-    /// Number of leaves currently in the persistent identity tree.
-    /// Used by [`EventGraph::rebuild_historical_roots_if_needed`] to
-    /// detect a leaves-vs-events mismatch that bypasses the simpler
-    /// recorded-count check (e.g. stale leaves left over from an
-    /// older code path that bypassed `apply_rln_static_event`).
-    pub(crate) fn leaves_count(&self) -> usize {
-        self.leaves.len()
+    /// Return the commitments currently persisted in the identity leaf tree.
+    ///
+    /// Startup recovery compares this set against the set derived by replaying
+    /// the static DAG. Counting leaves is not enough because a crash or old
+    /// code path can leave the right number of leaves with the wrong
+    /// commitments.
+    pub(crate) fn commitment_reprs(&self) -> Result<BTreeSet<[u8; 32]>> {
+        let mut commitments = BTreeSet::new();
+        for item in self.leaves.iter() {
+            let (key, val) = item?;
+            if key.len() != 32 || val.len() != 32 {
+                return Err(Error::Custom(format!(
+                    "RLN identity leaf key/value must be 32 bytes, got {}/{}",
+                    key.len(),
+                    val.len()
+                )))
+            }
+            if key.as_ref() != val.as_ref() {
+                return Err(Error::Custom("RLN identity leaf key/value mismatch".into()))
+            }
+            let mut repr = [0u8; 32];
+            repr.copy_from_slice(&val);
+            commitments.insert(repr);
+        }
+        Ok(commitments)
     }
 
     /// Check whether `root` matches the current root or any recent

src/event_graph/test_helpers.rs

@@ -120,9 +120,17 @@ pub async fn make_eg() -> EventGraphPtr {
 
 /// Construct an [`EventGraph`] with a caller-provided test config.
 pub async fn make_eg_with_config(config: EventGraphConfig) -> EventGraphPtr {
+    let sled_db = sled::Config::new().temporary(true).open().unwrap();
+    make_eg_with_config_and_db(config, sled_db).await
+}
+
+/// Construct an [`EventGraph`] with a caller-provided config and sled DB.
+pub async fn make_eg_with_config_and_db(
+    config: EventGraphConfig,
+    sled_db: sled::Db,
+) -> EventGraphPtr {
     let ex = Arc::new(Executor::new());
     let p2p = P2p::new(Settings::default(), ex.clone()).await.unwrap();
-    let sled_db = sled::Config::new().temporary(true).open().unwrap();
     EventGraph::with_zk_keys(p2p, sled_db, "/tmp".into(), false, config, shared_zk_keys(), ex)
         .await
         .unwrap()

src/event_graph/tests_rln.rs

@@ -34,8 +34,8 @@ use crate::{
             GENESIS_BLOB_GUARD, MAX_MSG_LIMIT, RLN_EPOCH_LEN, RLN_GENESIS,
         },
         test_helpers::{
-            make_eg, make_eg_with_config, make_network, run_multi_node_test, shutdown_network,
-            TestIdentity,
+            make_eg, make_eg_with_config, make_eg_with_config_and_db, make_network,
+            run_multi_node_test, shutdown_network, test_config, TestIdentity,
         },
         util::generate_genesis,
         Event, EventGraphConfig, EventGraphPtr, NULL_ID, NULL_PARENTS,
@@ -392,6 +392,68 @@ fn rln_bootstrapped_identities_parent_static_genesis() {
     })
 }
 
+#[test]
+fn rln_startup_rebuild_before_bootstrap_restores_configured_identity() {
+    smol::block_on(async {
+        let config = EventGraphConfig { hours_rotation: 1, ..test_config() };
+        let commitment = pallas::Base::from_repr(config.pregenerated_identity_commitments[0])
+            .into_option()
+            .unwrap();
+        let db = sled::Config::new().temporary(true).open().unwrap();
+
+        // Simulate a crash after the identity leaf was written but before the
+        // corresponding pregenerated static event reached the static DAG.
+        let leaves = db.open_tree("rln-identity-leaves").unwrap();
+        leaves.insert(commitment.to_repr(), commitment.to_repr().as_ref()).unwrap();
+        leaves.insert(b"bad-leaf", b"bad-value").unwrap();
+
+        let eg = make_eg_with_config_and_db(config, db).await;
+        assert!(eg.rln_contains(&commitment).await);
+
+        let mut matching_static_events = 0usize;
+        for item in eg.static_dag.iter() {
+            let (_, bytes) = item.unwrap();
+            let ev: Event = deserialize_async(&bytes).await.unwrap();
+            if ev.header.parents == NULL_PARENTS {
+                continue
+            }
+            let node: RLNNode = deserialize_async(ev.content()).await.unwrap();
+            if matches!(node, RLNNode::Registration(c) if c == commitment) {
+                matching_static_events += 1;
+                assert_eq!(eg.static_blob_fetch(&ev.id()).unwrap().unwrap(), GENESIS_BLOB_GUARD,);
+            }
+        }
+        assert_eq!(matching_static_events, 1);
+    })
+}
+
+#[test]
+fn rln_rebuild_detects_stale_leaf_with_same_count() {
+    smol::block_on(async {
+        let config = EventGraphConfig { hours_rotation: 1, ..test_config() };
+        let eg = make_eg_with_config(config).await;
+        let commitment = genesis_commitment_at(&eg, 0);
+        let stale = pallas::Base::from(0x51a1e_u64);
+
+        {
+            let mut state = eg.identity_state.write().await;
+            state.clear_for_rebuild().unwrap();
+            state.register(stale).unwrap();
+        }
+
+        assert!(!eg.rln_contains(&commitment).await);
+        assert!(eg.rln_contains(&stale).await);
+        assert_eq!(eg.rln_historical_roots_ordered.len(), 1);
+
+        eg.rebuild_historical_roots_if_needed().await.unwrap();
+
+        assert!(eg.rln_contains(&commitment).await);
+        assert!(!eg.rln_contains(&stale).await);
+        assert_eq!(eg.rln_historical_roots_ordered.len(), 1);
+        assert_eq!(eg.rln_historical_roots_by_value.len(), 1);
+    })
+}
+
 async fn make_static_event(content: &[u8], eg: &EventGraphPtr) -> Event {
     use crate::event_graph::event::Header;
     let timestamp = eg.current_genesis.read().await.header.timestamp;
@@ -1687,6 +1749,42 @@ fn rln_rebuild_historical_roots() {
     })
 }
 
+#[test]
+fn rln_rebuild_repairs_mismatched_historical_root_by_value_index() {
+    smol::block_on(async {
+        let eg = make_eg().await;
+
+        let c1 = pallas::Base::from(0x5555_u64);
+        let c2 = pallas::Base::from(0x6666_u64);
+        let n1 = RLNNode::Registration(c1);
+        let n2 = RLNNode::Registration(c2);
+
+        let ev1 = synth_static_event(1, 200_000, &n1).await;
+        let ev2 = synth_static_event(2, 200_001, &n2).await;
+        let _ = eg.apply_rln_static_event(&ev1, &n1).await.unwrap();
+        eg.static_insert(&ev1).await.unwrap();
+        let r2 = eg.apply_rln_static_event(&ev2, &n2).await.unwrap();
+        eg.static_insert(&ev2).await.unwrap();
+
+        assert!(eg.is_root_valid_at(&r2, 200_001).unwrap());
+
+        eg.rln_historical_roots_by_value.clear().unwrap();
+        let bogus_a = [0u8; 72];
+        let mut bogus_b = [0u8; 72];
+        bogus_b[71] = 1;
+        eg.rln_historical_roots_by_value.insert(bogus_a, &[]).unwrap();
+        eg.rln_historical_roots_by_value.insert(bogus_b, &[]).unwrap();
+        assert_eq!(eg.rln_historical_roots_by_value.len(), 2);
+        assert!(!eg.is_root_valid_at(&r2, 200_001).unwrap());
+
+        eg.rebuild_historical_roots_if_needed().await.unwrap();
+
+        assert_eq!(eg.rln_historical_roots_ordered.len(), 2);
+        assert_eq!(eg.rln_historical_roots_by_value.len(), 2);
+        assert!(eg.is_root_valid_at(&r2, 200_001).unwrap());
+    })
+}
+
 #[test]
 fn rln_perf_signal_verify() {
     use std::time::Instant;