event_graph: Return errors from DAG startup scans

bin/darkirc/src/irc/services/nickserv.rs

@@ -821,7 +821,7 @@ impl NickServ {
         let blob_bytes = serialize_async(&slash_blob).await;
 
         let rln_node = RLNNode::Slashing(identity.commitment());
-        let event = Event::new_static(serialize_async(&rln_node).await, evgr).await;
+        let event = Event::new_static(serialize_async(&rln_node).await, evgr).await?;
 
         // Commit through the verified static-event pipeline so durable event
         // storage stays ahead of RLN side tables, while subscribers still see

src/event_graph/event.rs

@@ -56,14 +56,14 @@ impl Header {
         }
     }
 
-    pub async fn new_static(content: &[u8], eg: &EventGraph) -> Self {
-        let (layer, parents) = eg.get_next_layer_with_parents_static().await;
-        Self {
+    pub async fn new_static(content: &[u8], eg: &EventGraph) -> Result<Self> {
+        let (layer, parents) = eg.get_next_layer_with_parents_static().await?;
+        Ok(Self {
             timestamp: UNIX_EPOCH.elapsed().unwrap().as_millis() as u64,
             parents,
             layer,
             content_hash: blake3::hash(content),
-        }
+        })
     }
 
     pub async fn with_timestamp(timestamp: u64, content: &[u8], eg: &EventGraph) -> Self {
@@ -160,9 +160,9 @@ impl Event {
         Self { header, content: data }
     }
 
-    pub async fn new_static(data: Vec<u8>, eg: &EventGraph) -> Self {
-        let header = Header::new_static(&data, eg).await;
-        Self { header, content: data }
+    pub async fn new_static(data: Vec<u8>, eg: &EventGraph) -> Result<Self> {
+        let header = Header::new_static(&data, eg).await?;
+        Ok(Self { header, content: data })
     }
 
     pub fn id(&self) -> blake3::Hash {

src/event_graph/mod.rs

@@ -212,15 +212,15 @@ impl TimeIndex {
         Self::default()
     }
 
-    pub async fn from_header_dag(tree: &sled::Tree) -> Self {
+    pub async fn from_header_dag(tree: &sled::Tree) -> Result<Self> {
         let mut idx = Self::new();
         for item in tree.iter() {
-            let (id, hdr) = item.unwrap();
-            let id = blake3::Hash::from_bytes((&id as &[u8]).try_into().unwrap());
-            let hdr: Header = deserialize_async(&hdr).await.unwrap();
+            let (id, hdr) = item?;
+            let id = blake3::Hash::from_bytes((&id as &[u8]).try_into()?);
+            let hdr: Header = deserialize_async(&hdr).await?;
             idx.insert(hdr.timestamp, id);
         }
-        idx
+        Ok(idx)
     }
 
     pub fn insert(&mut self, ts: u64, id: blake3::Hash) {
@@ -295,14 +295,14 @@ pub struct DagSlot {
 /// Full-scan tip computation.
 /// Compute unreferenced tips - events that exist in the DAG but are
 /// not referenced as a parent by any other event - grouped by layer.
-pub(crate) async fn compute_unreferenced_tips(dag: &sled::Tree) -> LayerUTips {
+pub(crate) async fn compute_unreferenced_tips(dag: &sled::Tree) -> Result<LayerUTips> {
     let mut candidates: HashMap<blake3::Hash, u64> = HashMap::new();
     let mut referenced: HashSet<blake3::Hash> = HashSet::new();
 
     for item in dag.iter() {
-        let (id_bytes, val_bytes) = item.unwrap();
-        let id = blake3::Hash::from_bytes((&id_bytes as &[u8]).try_into().unwrap());
-        let ev: Event = deserialize_async(&val_bytes).await.unwrap();
+        let (id_bytes, val_bytes) = item?;
+        let id = blake3::Hash::from_bytes((&id_bytes as &[u8]).try_into()?);
+        let ev: Event = deserialize_async(&val_bytes).await?;
 
         candidates.insert(id, ev.header.layer);
         for p in ev.header.parents.iter() {
@@ -319,7 +319,7 @@ pub(crate) async fn compute_unreferenced_tips(dag: &sled::Tree) -> LayerUTips {
             map.entry(layer).or_default().insert(id);
         }
     }
-    map
+    Ok(map)
 }
 
 /// Pick up to N_EVENT_PARENTS tips from the highest layers.
@@ -367,7 +367,7 @@ impl DagStore {
 
         if config.hours_rotation == 0 {
             let genesis = generate_genesis(config);
-            dags.insert(genesis.header.timestamp, Self::create_slot(&sled_db, &genesis).await);
+            dags.insert(genesis.header.timestamp, Self::create_slot(&sled_db, &genesis).await?);
             return Ok(Self { db: sled_db, dags })
         }
 
@@ -393,7 +393,7 @@ impl DagStore {
                         content_hash: blake3::hash(&config.genesis_contents),
                     };
                     let genesis = Event { header: hdr, content: config.genesis_contents.clone() };
-                    let slot = Self::create_slot(&sled_db, &genesis).await;
+                    let slot = Self::create_slot(&sled_db, &genesis).await?;
                     dags.insert(ts, slot);
                 }
             }
@@ -414,33 +414,33 @@ impl DagStore {
                 content_hash: blake3::hash(&config.genesis_contents),
             };
             let genesis = Event { header: hdr, content: config.genesis_contents.clone() };
-            dags.insert(ts, Self::create_slot(&sled_db, &genesis).await);
+            dags.insert(ts, Self::create_slot(&sled_db, &genesis).await?);
         }
 
         Ok(Self { db: sled_db, dags })
     }
 
-    async fn create_slot(db: &sled::Db, genesis: &Event) -> DagSlot {
+    async fn create_slot(db: &sled::Db, genesis: &Event) -> Result<DagSlot> {
         let name = genesis.header.timestamp.to_string();
-        let ht = db.open_tree(format!("headers_{name}")).unwrap();
-        let mt = db.open_tree(&name).unwrap();
+        let ht = db.open_tree(format!("headers_{name}"))?;
+        let mt = db.open_tree(&name)?;
         for (tree, data) in
             [(&ht, serialize_async(&genesis.header).await), (&mt, serialize_async(genesis).await)]
         {
             if tree.is_empty() {
                 let mut ov = SledTreeOverlay::new(tree);
-                ov.insert(genesis.id().as_bytes(), &data).unwrap();
+                ov.insert(genesis.id().as_bytes(), &data)?;
                 if let Some(b) = ov.aggregate() {
-                    tree.apply_batch(b).unwrap();
+                    tree.apply_batch(b)?;
                 }
             }
         }
-        DagSlot {
-            tips: compute_unreferenced_tips(&mt).await,
-            time_index: TimeIndex::from_header_dag(&ht).await,
+        Ok(DagSlot {
+            tips: compute_unreferenced_tips(&mt).await?,
+            time_index: TimeIndex::from_header_dag(&ht).await?,
             header_tree: ht,
             main_tree: mt,
-        }
+        })
     }
 
     /// Add a new DAG on rotation. In bounded mode, drops the oldest DAG
@@ -455,11 +455,11 @@ impl DagStore {
                 let Some((_, old)) = self.dags.pop_first() else {
                     return Err(Error::Custom("event graph DAG store is empty".into()))
                 };
-                self.db.drop_tree(old.header_tree.name()).unwrap();
-                self.db.drop_tree(old.main_tree.name()).unwrap();
+                self.db.drop_tree(old.header_tree.name())?;
+                self.db.drop_tree(old.main_tree.name())?;
             }
         }
-        let slot = Self::create_slot(&self.db, genesis).await;
+        let slot = Self::create_slot(&self.db, genesis).await?;
         self.dags.insert(genesis.header.timestamp, slot);
         Ok(())
     }
@@ -1935,8 +1935,9 @@ impl EventGraph {
 
     pub(crate) async fn get_next_layer_with_parents_static(
         &self,
-    ) -> (u64, [blake3::Hash; N_EVENT_PARENTS]) {
-        select_parents_from_tips(&compute_unreferenced_tips(&self.static_dag).await)
+    ) -> Result<(u64, [blake3::Hash; N_EVENT_PARENTS])> {
+        let tips = compute_unreferenced_tips(&self.static_dag).await?;
+        Ok(select_parents_from_tips(&tips))
     }
 
     pub async fn order_events(&self) -> Vec<Event> {
@@ -2098,7 +2099,7 @@ impl EventGraph {
             None => None,
         })
     }
-    pub async fn static_unreferenced_tips(&self) -> LayerUTips {
+    pub async fn static_unreferenced_tips(&self) -> Result<LayerUTips> {
         compute_unreferenced_tips(&self.static_dag).await
     }
 

src/event_graph/proto.rs

@@ -738,7 +738,13 @@ impl ProtocolEventGraph {
         let slash_blob = SlashBlob { proof, identity_secret_hash, merkle_root: root };
         let blob = serialize_async(&slash_blob).await;
         let node = RLNNode::Slashing(commitment);
-        let ev = Event::new_static(serialize_async(&node).await, &self.event_graph).await;
+        let ev = match Event::new_static(serialize_async(&node).await, &self.event_graph).await {
+            Ok(ev) => ev,
+            Err(e) => {
+                error!(target: "event_graph::protocol", "[RLN] Slash event creation failed: {e}");
+                return
+            }
+        };
         if let Err(e) = self.event_graph.commit_verified_static_event(&ev, &blob, &node).await {
             error!(target: "event_graph::protocol", "[RLN] Slash static commit failed: {e}");
             return
@@ -963,7 +969,13 @@ impl ProtocolEventGraph {
             }
 
             let layers = match dag_name.as_str() {
-                "static-dag" => self.event_graph.static_unreferenced_tips().await,
+                "static-dag" => match self.event_graph.static_unreferenced_tips().await {
+                    Ok(tips) => tips,
+                    Err(e) => {
+                        warn!(target: "event_graph::protocol", "failed to scan static DAG tips: {e}");
+                        continue
+                    }
+                },
                 _ => {
                     let ts = match u64::from_str(&dag_name) {
                         Ok(v) => v,

src/event_graph/tests.rs

@@ -384,6 +384,54 @@ fn evgr_dag_store_archive_mode_discovers_existing_trees() {
     })
 }
 
+#[test]
+fn evgr_dag_store_rejects_corrupt_header_index_on_open() {
+    smol::block_on(async {
+        let sled_db = sled::Config::new().temporary(true).open().unwrap();
+        let config = bounded_dag_store_config();
+        let store = DagStore::new(sled_db.clone(), &config).await.unwrap();
+        let ts = *store.dag_timestamps().last().unwrap();
+        drop(store);
+
+        let bad_id = [0u8; 32];
+        let headers = sled_db.open_tree(format!("headers_{ts}")).unwrap();
+        headers.insert(bad_id.as_slice(), b"not-a-header".as_slice()).unwrap();
+
+        let result = DagStore::new(sled_db, &config).await;
+        assert!(result.is_err(), "corrupt header bytes should fail DAG startup");
+    })
+}
+
+#[test]
+fn evgr_dag_store_rejects_corrupt_event_tree_on_open() {
+    smol::block_on(async {
+        let sled_db = sled::Config::new().temporary(true).open().unwrap();
+        let config = bounded_dag_store_config();
+        let store = DagStore::new(sled_db.clone(), &config).await.unwrap();
+        let ts = *store.dag_timestamps().last().unwrap();
+        drop(store);
+
+        let bad_id = [0u8; 32];
+        let events = sled_db.open_tree(ts.to_string()).unwrap();
+        events.insert(bad_id.as_slice(), b"not-an-event".as_slice()).unwrap();
+
+        let result = DagStore::new(sled_db, &config).await;
+        assert!(result.is_err(), "corrupt event bytes should fail DAG startup");
+    })
+}
+
+#[test]
+fn evgr_static_event_creation_rejects_corrupt_static_dag() {
+    smol::block_on(async {
+        let eg = make_eg().await;
+        let bad_id = [0u8; 32];
+        eg.static_dag.insert(bad_id.as_slice(), b"not-an-event".as_slice()).unwrap();
+
+        let result = Event::new_static(b"static-after-corruption".to_vec(), &eg).await;
+        assert!(result.is_err(), "corrupt static DAG should fail static event creation");
+    })
+}
+
 #[test]
 fn evgr_compute_unreferenced_tips_single_pass() {
     smol::block_on(async {
@@ -435,7 +483,7 @@ fn evgr_compute_unreferenced_tips_single_pass() {
 
         assert_ne!(e2.id(), e4.id(), "e2 and e4 must have distinct IDs");
 
-        let tips = compute_unreferenced_tips(&slot.main_tree).await;
+        let tips = compute_unreferenced_tips(&slot.main_tree).await.unwrap();
 
         assert!(tips.get(&2).unwrap().contains(&e3.id()));
         assert!(tips.get(&1).unwrap().contains(&e4.id()));

src/event_graph/tests_rln.rs

@@ -460,7 +460,7 @@ fn rln_rebuild_detects_stale_leaf_with_same_count() {
 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;
-    let (layer, parents) = eg.get_next_layer_with_parents_static().await;
+    let (layer, parents) = eg.get_next_layer_with_parents_static().await.unwrap();
     let header = Header { timestamp, parents, layer, content_hash: blake3::hash(content) };
     Event { header, content: content.to_vec() }
 }
@@ -946,8 +946,9 @@ async fn concurrent_slashes(ex: Arc<Executor<'static>>) {
         )
         .expect("proof");
         let blob = SlashBlob { proof, identity_secret_hash: ish, merkle_root: root };
-        let event =
-            Event::new_static(serialize_async(&RLNNode::Slashing(commitment)).await, eg).await;
+        let event = Event::new_static(serialize_async(&RLNNode::Slashing(commitment)).await, eg)
+            .await
+            .unwrap();
         (event, serialize_async(&blob).await)
     }
 
@@ -1103,7 +1104,7 @@ async fn static_sync_registration(ex: Arc<Executor<'static>>) {
 
     let rln_node = RLNNode::Registration(commitment);
     let content = serialize_async(&rln_node).await;
-    let event = Event::new_static(content, &nodes[0]).await;
+    let event = Event::new_static(content, &nodes[0]).await.unwrap();
 
     // Seed nodes 0..=3 the same way a real verified static event is
     // committed: blob and event become durable before RLN side tables,
@@ -1176,7 +1177,7 @@ async fn static_sync_blob_propagation(ex: Arc<Executor<'static>>) {
     let commitment = id.commitment();
 
     let content = serialize_async(&RLNNode::Registration(commitment)).await;
-    let event = Event::new_static(content.clone(), &nodes[0]).await;
+    let event = Event::new_static(content.clone(), &nodes[0]).await.unwrap();
     // Synthetic blob - content doesn't matter for propagation
     // testing, only that it's non-empty so static_sync's
     // verification path takes the "blob present" branch.