event_graph: Return errors from history ordering

bin/darkirc/src/irc/command.rs

@@ -180,7 +180,7 @@ impl Client {
                 self.reg_paused.store(false, SeqCst);
                 if self.registered.load(SeqCst) && !self.is_cap_end.load(SeqCst) {
                     self.is_cap_end.store(true, SeqCst);
-                    return Ok(self.welcome().await)
+                    return self.welcome().await
                 }
 
                 return Ok(vec![])
@@ -305,7 +305,7 @@ impl Client {
 
         if hist {
             // Potentially extend the replies with channel history
-            replies.extend(self.get_history(&channels).await.unwrap());
+            replies.extend(self.get_history(&channels).await?);
         }
 
         Ok(replies)
@@ -500,7 +500,7 @@ impl Client {
             if self.reg_paused.load(SeqCst) {
                 return Ok(vec![])
             } else {
-                return Ok(self.welcome().await)
+                return self.welcome().await
             }
         }
 
@@ -820,7 +820,7 @@ impl Client {
             if self.reg_paused.load(SeqCst) {
                 return Ok(vec![])
             } else {
-                return Ok(self.welcome().await)
+                return self.welcome().await
             }
         }
 
@@ -850,7 +850,7 @@ impl Client {
     }
 
     /// Internal function that constructs the welcome message.
-    async fn welcome(&self) -> Vec<ReplyType> {
+    async fn welcome(&self) -> Result<Vec<ReplyType>> {
         let nick = self.nickname.read().await.to_string();
 
         let mut replies = vec![
@@ -865,7 +865,7 @@ impl Client {
         ];
 
         // Append the MOTD
-        replies.append(&mut self.handle_cmd_motd("").await.unwrap());
+        replies.append(&mut self.handle_cmd_motd("").await?);
 
         let mut channels = HashSet::new();
 
@@ -873,13 +873,13 @@ impl Client {
         // and set their topics, if any.
         if !*self.caps.read().await.get("no-autojoin").unwrap() {
             for channel in self.server.autojoin.read().await.iter() {
-                replies.extend(self.handle_cmd_join(channel, false).await.unwrap());
+                replies.extend(self.handle_cmd_join(channel, false).await?);
                 channels.insert(channel.to_string());
             }
         }
 
         // Potentially extend the replies with channel history
-        replies.extend(self.get_history(&channels).await.unwrap());
+        replies.extend(self.get_history(&channels).await?);
 
         // And request NAMES list.
         if !*self.caps.read().await.get("no-autojoin").unwrap() {
@@ -900,7 +900,7 @@ impl Client {
             }
         }
 
-        replies
+        Ok(replies)
     }
 
     /// Internal function that scans the DAG and returns events for
@@ -914,7 +914,7 @@ impl Client {
         }
 
         // Fetch and order all the events from the DAG
-        let dag_events = self.server.darkirc.event_graph.order_events().await;
+        let dag_events = self.server.darkirc.event_graph.order_events().await?;
 
         // Here we'll hold the events in order we'll push to the client
         let mut replies = vec![];

src/event_graph/mod.rs

@@ -1944,12 +1944,12 @@ impl EventGraph {
         Ok(select_parents_from_tips(&tips))
     }
 
-    pub async fn order_events(&self) -> Vec<Event> {
+    pub async fn order_events(&self) -> Result<Vec<Event>> {
         let mut all = vec![];
         for (_, slot) in self.dag_store.read().await.dags.iter() {
             for item in slot.main_tree.iter() {
-                let (_, b) = item.unwrap();
-                let ev: Event = deserialize_async(&b).await.unwrap();
+                let (_, b) = item?;
+                let ev: Event = deserialize_async(&b).await?;
                 if ev.header.parents != NULL_PARENTS {
                     all.push(ev);
                 }
@@ -1957,7 +1957,7 @@ impl EventGraph {
         }
 
         all.sort_unstable_by(display_order);
-        all
+        Ok(all)
     }
 
     pub async fn fetch_headers_with_tips(

src/event_graph/tests.rs

@@ -857,6 +857,23 @@ fn evgr_fetch_page_both_directions() {
     })
 }
 
+#[test]
+fn evgr_order_events_rejects_corrupt_event_record() {
+    smol::block_on(async {
+        let eg = make_eg().await;
+        let dag_ts = eg.current_genesis.read().await.header.timestamp;
+        let bad_id = [0u8; 32];
+        {
+            let store = eg.dag_store.read().await;
+            let slot = store.get_slot(&dag_ts).unwrap();
+            slot.main_tree.insert(bad_id.as_slice(), b"not-an-event".as_slice()).unwrap();
+        }
+
+        let result = eg.order_events().await;
+        assert!(result.is_err(), "corrupt event records should fail history ordering");
+    })
+}
+
 async fn build_graph() -> Result<(EventGraphPtr, std::collections::HashMap<&'static str, Event>)> {
     let eg = make_eg().await;
     let dag_name = eg.current_genesis.read().await.header.timestamp.to_string();