event_graph: Finish panic-free runtime paths

src/event_graph/event.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{cmp::Ordering, collections::HashSet, time::UNIX_EPOCH};
+use std::{cmp::Ordering, collections::HashSet};
 
 use darkfi_serial::{async_trait, deserialize_async, Encodable, SerialDecodable, SerialEncodable};
 use sled_overlay::{sled, SledTreeOverlay};
@@ -76,9 +76,9 @@ impl Header {
     /// Blake3 hash of `(timestamp, parents, layer, content_hash)`.
     pub fn id(&self) -> blake3::Hash {
         let mut h = blake3::Hasher::new();
-        self.timestamp.encode(&mut h).unwrap();
-        self.parents.encode(&mut h).unwrap();
-        self.layer.encode(&mut h).unwrap();
+        let _ = self.timestamp.encode(&mut h);
+        let _ = self.parents.encode(&mut h);
+        let _ = self.layer.encode(&mut h);
         h.update(self.content_hash.as_bytes());
         h.finalize()
     }
@@ -135,7 +135,7 @@ impl Header {
         }
 
         if config.hours_rotation == 0 {
-            let now = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
+            let Ok(now) = unix_timestamp_millis() else { return false };
             return self.timestamp <= now.saturating_add(EVENT_TIME_DRIFT)
         }
 
@@ -210,10 +210,10 @@ impl Event {
             return false
         }
 
-        let now = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
+        let Ok(now) = unix_timestamp_millis() else { return false };
 
-        if self.header.timestamp < now - EVENT_TIME_DRIFT ||
-            self.header.timestamp > now + EVENT_TIME_DRIFT
+        if self.header.timestamp < now.saturating_sub(EVENT_TIME_DRIFT) ||
+            self.header.timestamp > now.saturating_add(EVENT_TIME_DRIFT)
         {
             return false
         }

src/event_graph/mod.rs

@@ -366,7 +366,7 @@ impl DagStore {
         let mut dags = BTreeMap::new();
 
         if config.hours_rotation == 0 {
-            let genesis = generate_genesis(config);
+            let genesis = generate_genesis(config)?;
             dags.insert(genesis.header.timestamp, Self::create_slot(&sled_db, &genesis).await?);
             return Ok(Self { db: sled_db, dags })
         }
@@ -402,7 +402,7 @@ impl DagStore {
         // Ensure the recent window of DAGs exists.
         // Creates them if they're not already loaded from the discovery step.
         for i in 1..=window {
-            let ts = next_hour_timestamp((i as i64) - (window as i64));
+            let ts = next_hour_timestamp((i as i64) - (window as i64))?;
             if dags.contains_key(&ts) {
                 // Already loaded from sled discovery
                 continue
@@ -682,7 +682,7 @@ impl EventGraph {
         config.validate()?;
         let identity_state = IdentityState::new(&sled_db)?;
         let rln_app_id = rln::RlnAppId::from_genesis(&config.genesis_contents);
-        let current_genesis = generate_genesis(&config);
+        let current_genesis = generate_genesis(&config)?;
         let (pregenerated_identity_commitments, pregenerated_identity_commitment_reprs) =
             validate_pregenerated_identity_commitments(&config)?;
         let dag_store = DagStore::new(sled_db.clone(), &config).await?;

src/event_graph/proto.rs

@@ -1024,7 +1024,10 @@ impl ProtocolEventGraph {
     async fn broadcast_rate_limiter(self: Arc<Self>) -> Result<()> {
         let mut rl = MovingWindow::new(RATELIMIT_EXPIRY_TIME);
         loop {
-            let ep = self.broadcaster_pull.recv().await.expect("broadcaster closed");
+            let Ok(ep) = self.broadcaster_pull.recv().await else {
+                warn!(target: "event_graph::protocol", "broadcaster channel closed");
+                return Ok(())
+            };
             rl.ticktock();
             if rl.count() > RATELIMIT_MIN_COUNT {
                 let ms = ((rl.count() - RATELIMIT_MIN_COUNT) * RATELIMIT_SAMPLE_SLEEP /

src/event_graph/rln.rs

@@ -679,7 +679,10 @@ pub fn sss_recover(shares: &[(pallas::Base, pallas::Base)]) -> Result<pallas::Ba
         let mut basis = pallas::Base::one();
         for (i, si) in shares.iter().enumerate() {
             if i != j {
-                basis *= si.0 * (si.0 - sj.0).invert().unwrap();
+                let Some(denominator) = Option::<pallas::Base>::from((si.0 - sj.0).invert()) else {
+                    return Err(Error::Custom("Duplicate x-coordinates in SSS shares".into()))
+                };
+                basis *= si.0 * denominator;
             }
         }
         secret += basis * sj.1;

src/event_graph/tests.rs

@@ -280,7 +280,7 @@ fn evgr_rotation_helpers_are_total() {
     let future_start = now + HOUR_MS;
     assert_eq!(next_rotation_timestamp(future_start, 1).unwrap(), future_start);
     assert!(millis_until_next_rotation(now.saturating_sub(1)).is_err());
-    assert_eq!(super::util::hours_since(future_start), 0);
+    assert_eq!(super::util::hours_since(future_start).unwrap(), 0);
 }
 
 #[test]
@@ -326,7 +326,7 @@ fn evgr_dag_store_eviction_policy() {
         // (a) bounded
         let mut store = make_dag_store().await.unwrap();
         let oldest_ts = store.dag_timestamps()[0];
-        let new_ts = next_hour_timestamp(1);
+        let new_ts = next_hour_timestamp(1).unwrap();
         let hdr = Header {
             timestamp: new_ts,
             parents: NULL_PARENTS,
@@ -344,7 +344,7 @@ fn evgr_dag_store_eviction_policy() {
         let mut archive = DagStore::new(sled_db, &archive_config()).await.unwrap();
         let initial = archive.dag_timestamps().len();
         for i in 1..=30i64 {
-            let ts = next_hour_timestamp(i);
+            let ts = next_hour_timestamp(i).unwrap();
             let hdr = Header {
                 timestamp: ts,
                 parents: NULL_PARENTS,
@@ -362,7 +362,7 @@ fn evgr_dag_store_eviction_policy() {
 fn evgr_dag_store_archive_mode_discovers_existing_trees() {
     smol::block_on(async {
         let sled_db = sled::Config::new().temporary(true).open().unwrap();
-        let historical_ts = next_hour_timestamp(-100);
+        let historical_ts = next_hour_timestamp(-100).unwrap();
         {
             let mut store = DagStore::new(sled_db.clone(), &archive_config()).await.unwrap();
             let hdr = Header {
@@ -766,7 +766,7 @@ fn evgr_header_insert_rejects_unloaded_dag_slot() {
     smol::block_on(async {
         let config = EventGraphConfig { hours_rotation: 1, max_dags: Some(2), ..test_config() };
         let eg = make_eg_with_config(config).await;
-        let dag_ts = next_hour_timestamp(-100);
+        let dag_ts = next_hour_timestamp(-100).unwrap();
         let dag_name = dag_ts.to_string();
         let genesis = Header {
             timestamp: dag_ts,

src/event_graph/tests_rln.rs

@@ -366,7 +366,7 @@ fn rln_bootstrapped_identities_parent_static_genesis() {
         };
         let eg = make_eg_with_config(config).await;
         let static_genesis =
-            generate_genesis(&EventGraphConfig { hours_rotation: 0, ..eg.config.clone() });
+            generate_genesis(&EventGraphConfig { hours_rotation: 0, ..eg.config.clone() }).unwrap();
         let static_genesis_id = static_genesis.id();
         let rotating_genesis_id = eg.current_genesis.read().await.id();
 
@@ -1045,7 +1045,8 @@ async fn old_static_event_propagates(ex: Arc<Executor<'static>>) {
     let content = serialize_async(&rln_node).await;
 
     let static_genesis =
-        generate_genesis(&EventGraphConfig { hours_rotation: 0, ..nodes[0].config.clone() });
+        generate_genesis(&EventGraphConfig { hours_rotation: 0, ..nodes[0].config.clone() })
+            .unwrap();
     let mut parents = NULL_PARENTS;
     parents[0] = static_genesis.id();
     let header = crate::event_graph::event::Header {

src/event_graph/util.rs

@@ -58,22 +58,22 @@ pub(super) fn unix_timestamp_millis() -> Result<u64> {
 }
 
 /// Timestamp (millis) for the start of the hour `hours` offsets from now.
-pub(super) fn next_hour_timestamp(hours: i64) -> u64 {
-    let now = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
+pub(super) fn next_hour_timestamp(hours: i64) -> Result<u64> {
+    let now = unix_timestamp_millis()?;
     let base = (now / HOUR_MS) * HOUR_MS;
     let offset = hours.unsigned_abs().saturating_mul(HOUR_MS);
 
     if hours.is_negative() {
-        base.saturating_sub(offset)
+        Ok(base.saturating_sub(offset))
     } else {
-        base.saturating_add(offset)
+        Ok(base.saturating_add(offset))
     }
 }
 
 /// Whole hours elapsed since `ts`.
-pub(super) fn hours_since(ts: u64) -> u64 {
-    let now = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
-    now.saturating_sub(ts) / HOUR_MS
+pub(super) fn hours_since(ts: u64) -> Result<u64> {
+    let now = unix_timestamp_millis()?;
+    Ok(now.saturating_sub(ts) / HOUR_MS)
 }
 
 /// Timestamp of the next DAG rotation.
@@ -85,7 +85,7 @@ pub fn next_rotation_timestamp(starting_timestamp: u64, rotation_period: u64) ->
     let rotation_ms = rotation_period.checked_mul(HOUR_MS).ok_or_else(|| {
         Error::Custom("event graph rotation period overflows milliseconds".into())
     })?;
-    let now = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
+    let now = unix_timestamp_millis()?;
 
     if now < starting_timestamp {
         return Ok(starting_timestamp)
@@ -107,7 +107,7 @@ pub fn next_rotation_timestamp(starting_timestamp: u64, rotation_period: u64) ->
 
 /// Milliseconds remaining until `next_rotation`.
 pub fn millis_until_next_rotation(next_rotation: u64) -> Result<u64> {
-    let now = UNIX_EPOCH.elapsed().unwrap().as_millis() as u64;
+    let now = unix_timestamp_millis()?;
     next_rotation
         .checked_sub(now)
         .ok_or_else(|| Error::Custom("event graph next rotation is in the past".into()))
@@ -119,11 +119,11 @@ pub fn millis_until_next_rotation(next_rotation: u64) -> Result<u64> {
 /// * `hours_rotation == 0` -> timestamp is `initial_genesis`.
 /// * `hours_rotation > 0`  -> timestamp is the most recent
 ///   multiple-of-N boundary since `initial_genesis`.
-pub fn generate_genesis(config: &EventGraphConfig) -> Event {
+pub fn generate_genesis(config: &EventGraphConfig) -> Result<Event> {
     let timestamp = if config.hours_rotation == 0 {
         config.initial_genesis
     } else {
-        let passed = hours_since(config.initial_genesis);
+        let passed = hours_since(config.initial_genesis)?;
         let rotations = passed / config.hours_rotation;
         let offset_hours = rotations.saturating_mul(config.hours_rotation);
         let offset_ms = offset_hours.saturating_mul(HOUR_MS);
@@ -131,7 +131,7 @@ pub fn generate_genesis(config: &EventGraphConfig) -> Event {
     };
     let content_hash = blake3::hash(&config.genesis_contents);
     let header = Header { timestamp, parents: [NULL_ID; N_EVENT_PARENTS], layer: 0, content_hash };
-    Event { header, content: config.genesis_contents.clone() }
+    Ok(Event { header, content: config.genesis_contents.clone() })
 }
 
 /// Append a replayer log entry for DAG state recreation.
@@ -156,21 +156,53 @@ pub async fn recreate_from_replayer_log(datastore: &Path) -> JsonResult {
             1,
         ))
     }
-    let reader = load_file(&log_path).unwrap();
-    let sled_db = sled::open(datastore.join("replayed_db")).unwrap();
-    let dag = sled_db.open_tree("replayer").unwrap();
+    let replay_error =
+        |e: String| JsonResult::Error(JsonError::new(ErrorCode::ParseError, Some(e), 1));
+    let reader = match load_file(&log_path) {
+        Ok(reader) => reader,
+        Err(e) => return replay_error(e.to_string()),
+    };
+    let sled_db = match sled::open(datastore.join("replayed_db")) {
+        Ok(db) => db,
+        Err(e) => return replay_error(e.to_string()),
+    };
+    let dag = match sled_db.open_tree("replayer") {
+        Ok(tree) => tree,
+        Err(e) => return replay_error(e.to_string()),
+    };
     for line in reader.lines() {
         let parts = line.split(' ').collect::<Vec<&str>>();
-        if parts[0] == "insert" {
-            let v: Event = deserialize(&base64::decode(parts[1]).unwrap()).unwrap();
-            dag.insert(v.header.id().as_bytes(), serialize(&v)).unwrap();
+        if parts.first() == Some(&"insert") {
+            let Some(encoded) = parts.get(1) else {
+                return replay_error("malformed event graph replay insert entry".into())
+            };
+            let Some(bytes) = base64::decode(encoded) else {
+                return replay_error("invalid base64 in event graph replay log".into())
+            };
+            let v: Event = match deserialize(&bytes) {
+                Ok(event) => event,
+                Err(e) => return replay_error(e.to_string()),
+            };
+            if let Err(e) = dag.insert(v.header.id().as_bytes(), serialize(&v)) {
+                return replay_error(e.to_string())
+            }
         }
     }
     let mut graph = HashMap::new();
     for item in dag.iter() {
-        let (id, val) = item.unwrap();
-        let id = blake3::Hash::from_bytes((&id as &[u8]).try_into().unwrap());
-        graph.insert(id, deserialize_async::<Event>(&val).await.unwrap());
+        let (id, val) = match item {
+            Ok(item) => item,
+            Err(e) => return replay_error(e.to_string()),
+        };
+        let id = match <[u8; 32]>::try_from(&id as &[u8]) {
+            Ok(bytes) => blake3::Hash::from_bytes(bytes),
+            Err(e) => return replay_error(e.to_string()),
+        };
+        let event = match deserialize_async::<Event>(&val).await {
+            Ok(event) => event,
+            Err(e) => return replay_error(e.to_string()),
+        };
+        graph.insert(id, event);
     }
     let json_graph = graph.into_iter().map(|(k, v)| (k.to_string(), JsonValue::from(v))).collect();
     let values = json_map([("dag", JsonValue::Object(json_graph))]);