diff --git a/crates/app/src/extensions.rs b/crates/app/src/extensions.rs index 5863db5..c2ce1cf 100644 --- a/crates/app/src/extensions.rs +++ b/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; } }) diff --git a/crates/app/src/lib.rs b/crates/app/src/lib.rs index d745716..3dc00ab 100644 --- a/crates/app/src/lib.rs +++ b/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 } } diff --git a/crates/types/src/lib.rs b/crates/types/src/lib.rs index d97c5af..f901526 100644 --- a/crates/types/src/lib.rs +++ b/crates/types/src/lib.rs @@ -77,7 +77,8 @@ impl Lock { 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) } }