Error handling (#32)

* better error handling

* fix several error handling mechanisms

Author: bengybade

crates/app/src/extensions.rs

@@ -1,7 +1,6 @@
-use std::{pin::Pin, task::Poll};
-
 use futures::{FutureExt, Stream, ready};
 use pin_project_lite::pin_project;
+use std::{pin::Pin, task::Poll};
 use void_types::Block;
 
 #[cfg(test)]
@@ -95,9 +94,11 @@ where
                 *this.post_extension = Some(post_extension);
                 break Some((block, derived));
             } else if let Some(block) = ready!(this.stream.as_mut().poll_next(cx)) {
-                let state_transition_func = this.state_transition_func.take().unwrap();
-                let pre_extension = this.pre_extension.take().unwrap();
-                let post_extension = this.post_extension.take().unwrap();
+                // Process block.
+                let state_transition_func =
+                    this.state_transition_func.take().expect("is always Some");
+                let pre_extension = this.pre_extension.take().expect("is always Some");
+                let post_extension = this.post_extension.take().expect("is always Some");
                 let state = this.state.clone();
                 this.access_fut.set(Some(
                     async move {
@@ -120,6 +121,7 @@ where
                     .boxed(),
                 ));
             } else {
+                // Stream ended.
                 break None;
             }
         })

crates/app/src/lib.rs

@@ -1,15 +1,14 @@
 #![deny(missing_docs)]
 //! This create provides helpers for building streams that turn blocks into state and state into proofs.
 
-use std::collections::VecDeque;
-use std::pin::Pin;
-use std::task::Poll;
-use tracing::{Instrument, debug, error, info, instrument, instrument::Instrumented, warn};
-
 use futures::FutureExt;
 use futures::Stream;
 use futures::ready;
 use pin_project_lite::pin_project;
+use std::collections::VecDeque;
+use std::pin::Pin;
+use std::task::Poll;
+use tracing::{Instrument, debug, error, info, instrument, instrument::Instrumented, warn};
 use void_types::Block;
 use void_types::Height;
 use void_types::Signed;
@@ -434,6 +433,7 @@ where
                 *this.state_transition_func = Some(state_transition_func);
                 break Some((block, derived));
             } else if let Some(block) = ready!(this.stream.as_mut().poll_next(cx)) {
+                // Process block.
                 let state_transition_func = this.state_transition_func.take().unwrap();
                 let state = this.state.clone();
                 this.access_fut.set(Some(
@@ -449,6 +449,7 @@ where
                     .boxed(),
                 ));
             } else {
+                // Stream ended.
                 break None;
             }
         })
@@ -697,12 +698,14 @@ impl Notification {
 
     /// Notify all receivers.
     pub fn notify(&self) {
+        // Ignore send error - if no receivers are listening, notification can be dropped
         let _ = self.tx.send(());
     }
 
     /// Wait for a notification.
-    pub async fn wait(&mut self) {
-        let _ = self.rx.changed().await;
+    /// Returns an error if the sender has been dropped.
+    pub async fn wait(&mut self) -> Result<(), tokio::sync::watch::error::RecvError> {
+        self.rx.changed().await
     }
 }
 

crates/types/src/lib.rs

@@ -78,7 +78,8 @@ impl<T> Lock<T> {
     where
         F: FnOnce(&mut T) -> R,
     {
-        let mut lock = self.lock.lock().unwrap();
+        // Not attempting poison recovery.
+        let mut lock = self.lock.lock().expect("poisoned lock");
         f(&mut lock)
     }
 }