From 6427530f01db0c11ffccd537bf7d9d940002f2c9 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 21 Oct 2025 12:45:24 +1100 Subject: [PATCH] Try errors This propogates errors along streams. It removes the state transition and extensions helpers in favor of a simple `apply_transform` function. Sign and Proof now return their errors. Implements for `TryStream` instead of `Stream` --- crates/app/src/block_awaits_proofs.rs | 17 +- crates/app/src/block_awaits_proofs/tests.rs | 32 +- crates/app/src/block_height_parent.rs | 55 ++ crates/app/src/buffer.rs | 77 +++ crates/app/src/extensions.rs | 180 ------ crates/app/src/extensions/tests.rs | 227 ------- crates/app/src/lib.rs | 665 +++----------------- crates/app/src/prove.rs | 89 +++ crates/app/src/scan_owned.rs | 26 +- crates/app/src/sign.rs | 72 +++ crates/app/src/tests.rs | 74 ++- crates/app/src/try_interval_latest.rs | 87 +++ crates/app/src/try_push_notification.rs | 47 ++ crates/app/tests/app.rs | 48 +- crates/types/src/lib.rs | 7 + 15 files changed, 629 insertions(+), 1074 deletions(-) create mode 100644 crates/app/src/block_height_parent.rs create mode 100644 crates/app/src/buffer.rs delete mode 100644 crates/app/src/extensions.rs delete mode 100644 crates/app/src/extensions/tests.rs create mode 100644 crates/app/src/prove.rs create mode 100644 crates/app/src/sign.rs create mode 100644 crates/app/src/try_interval_latest.rs create mode 100644 crates/app/src/try_push_notification.rs diff --git a/crates/app/src/block_awaits_proofs.rs b/crates/app/src/block_awaits_proofs.rs index a8bc385..91488ee 100644 --- a/crates/app/src/block_awaits_proofs.rs +++ b/crates/app/src/block_awaits_proofs.rs @@ -1,5 +1,5 @@ use futures::Stream; -use futures::ready; +use futures::TryStream; use pin_project_lite::pin_project; use std::task::Poll; use void_types::Block; @@ -9,7 +9,7 @@ mod tests; pin_project! { /// Stream produced by calling `VoidStream::blocks_await_proofs`. - pub struct BlocksAwaitProofs { + pub struct BlocksAwaitProofs { #[pin] stream: St, #[pin] @@ -21,7 +21,7 @@ pin_project! { impl BlocksAwaitProofs where - St: Stream, + St: TryStream, P: Stream, { /// Create a new `BlocksAwaitProofs` stream. @@ -37,11 +37,10 @@ where impl Stream for BlocksAwaitProofs where - St: Stream, - St: Stream, + St: TryStream, P: Stream, { - type Item = Block; + type Item = Result; fn poll_next( self: std::pin::Pin<&mut Self>, @@ -55,7 +54,7 @@ where let return_block = this.block.as_ref().is_some_and(|b| latest >= b.height); *this.latest_proof_height = Some(latest); if return_block && let Some(block) = this.block.take() { - return Poll::Ready(Some(block)); + return Poll::Ready(Some(Ok(block))); } } Poll::Ready(None) => return Poll::Ready(None), @@ -65,7 +64,7 @@ where // After this block there is guaranteed to be one // because otherwise we would have returned Pending or None. if this.block.is_none() { - let Some(block) = ready!(this.stream.poll_next(cx)) else { + let Some(block) = ready_ok!(this.stream.try_poll_next(cx)) else { return Poll::Ready(None); }; *this.block = Some(block); @@ -78,7 +77,7 @@ where .as_ref() .is_some_and(|proof_height| *proof_height >= block.height) { - Poll::Ready(Some(block)) + Poll::Ready(Some(Ok(block))) } else { *this.block = Some(block); Poll::Pending diff --git a/crates/app/src/block_awaits_proofs/tests.rs b/crates/app/src/block_awaits_proofs/tests.rs index 031e358..affb454 100644 --- a/crates/app/src/block_awaits_proofs/tests.rs +++ b/crates/app/src/block_awaits_proofs/tests.rs @@ -1,3 +1,8 @@ +use std::convert::Infallible; + +use futures::Stream; +use futures::StreamExt; + use crate::VoidStream; use super::*; @@ -9,7 +14,8 @@ async fn test_block_awaits_proofs() { height: 1, parent_hash: [0u8; 32], events: vec![vec![1, 2, 3]], - }]); + }]) + .map(Ok::<_, Infallible>); let proof_heights_stream = futures::stream::pending(); let stream = blocks_stream.blocks_await_proofs(proof_heights_stream); futures::pin_mut!(stream); @@ -26,18 +32,18 @@ async fn test_block_awaits_proofs() { events: vec![vec![1, 2, 3]], }; let height = 1; - let blocks_stream = futures::stream::iter(vec![block]); + let blocks_stream = futures::stream::iter(vec![block]).map(Ok::<_, Infallible>); let proof_heights_stream = futures::stream::iter(vec![height]); let stream = blocks_stream.blocks_await_proofs(proof_heights_stream); futures::pin_mut!(stream); assert_eq!( stream.as_mut().poll_next(&mut cx), - Poll::Ready(Some(Block { + Poll::Ready(Some(Ok(Block { height: 1, parent_hash: [0u8; 32], events: vec![vec![1, 2, 3]], - })) + }))) ); // Proof @ block 2 @@ -47,18 +53,18 @@ async fn test_block_awaits_proofs() { events: vec![vec![1, 2, 3]], }; let height = 2; - let blocks_stream = futures::stream::iter(vec![block]); + let blocks_stream = futures::stream::iter(vec![block]).map(Ok::<_, Infallible>); let proof_heights_stream = futures::stream::iter(vec![height]); let stream = blocks_stream.blocks_await_proofs(proof_heights_stream); futures::pin_mut!(stream); assert_eq!( stream.as_mut().poll_next(&mut cx), - Poll::Ready(Some(Block { + Poll::Ready(Some(Ok(Block { height: 1, parent_hash: [0u8; 32], events: vec![vec![1, 2, 3]], - })) + }))) ); // Block 1 then proof @ block 2 @@ -69,7 +75,7 @@ async fn test_block_awaits_proofs() { }; let height = 2; let (tx, rx) = tokio::sync::mpsc::channel(10); - let blocks_stream = futures::stream::iter(vec![block]); + let blocks_stream = futures::stream::iter(vec![block]).map(Ok::<_, Infallible>); let proof_heights_stream = futures::stream::unfold(rx, |mut rx| async { rx.recv().await.map(|loc| (loc, rx)) }); let stream = blocks_stream.blocks_await_proofs(proof_heights_stream); @@ -81,11 +87,11 @@ async fn test_block_awaits_proofs() { assert_eq!( stream.as_mut().poll_next(&mut cx), - Poll::Ready(Some(Block { + Poll::Ready(Some(Ok(Block { height: 1, parent_hash: [0u8; 32], events: vec![vec![1, 2, 3]], - })) + }))) ); // Block 2 then proof @ block 1 @@ -96,7 +102,7 @@ async fn test_block_awaits_proofs() { }; let mut height = 1; let (tx, rx) = tokio::sync::mpsc::channel(10); - let blocks_stream = futures::stream::iter(vec![block]); + let blocks_stream = futures::stream::iter(vec![block]).map(Ok::<_, Infallible>); let proof_heights_stream = futures::stream::unfold(rx, |mut rx| async { rx.recv().await.map(|loc| (loc, rx)) }); let stream = blocks_stream.blocks_await_proofs(proof_heights_stream); @@ -113,10 +119,10 @@ async fn test_block_awaits_proofs() { assert_eq!( stream.as_mut().poll_next(&mut cx), - Poll::Ready(Some(Block { + Poll::Ready(Some(Ok(Block { height: 2, parent_hash: [0u8; 32], events: vec![vec![1, 2, 3]], - })) + }))) ); } diff --git a/crates/app/src/block_height_parent.rs b/crates/app/src/block_height_parent.rs new file mode 100644 index 0000000..02a28db --- /dev/null +++ b/crates/app/src/block_height_parent.rs @@ -0,0 +1,55 @@ +use futures::Stream; +use futures::TryStream; +use pin_project_lite::pin_project; +use std::task::Poll; +use void_types::Block; + +pin_project! { + /// Stream produced by calling `VoidStream::block_height_parent`. + pub struct BlockHeightParent { + #[pin] + stream: St, + prev_height: Option, + prev_parent_hash: [u8; 32], + } +} + +impl Stream for BlockHeightParent +where + St: TryStream, +{ + type Item = Result; + + fn poll_next( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> Poll> { + let mut this = self.project(); + let Some(mut block) = ready_ok!(this.stream.as_mut().try_poll_next(cx)) else { + return Poll::Ready(None); + }; + block.height = this.prev_height.map_or(0, |h| h.saturating_add(1)); + block.parent_hash = *this.prev_parent_hash; + *this.prev_height = Some(block.height); + *this.prev_parent_hash = void_hash::Hash::hash(&block); + Poll::Ready(Some(Ok(block))) + } +} + +impl BlockHeightParent +where + St: TryStream, +{ + /// Create a new `BlockHeightParent` stream. + pub fn new( + stream: St, + previous_parent_height: Option, + previous_parent_hash: [u8; 32], + ) -> Self { + Self { + stream, + prev_height: previous_parent_height, + prev_parent_hash: previous_parent_hash, + } + } +} diff --git a/crates/app/src/buffer.rs b/crates/app/src/buffer.rs new file mode 100644 index 0000000..6e20d9d --- /dev/null +++ b/crates/app/src/buffer.rs @@ -0,0 +1,77 @@ +use std::{collections::VecDeque, task::Poll}; + +use futures::{Stream, TryStream}; +use pin_project_lite::pin_project; +use tracing::warn; + +pin_project! { + /// Stream produced by calling `VoidStream::buffer`. + pub struct Buffer { + #[pin] + stream: St, + capacity: usize, + buffer: VecDeque, + closed: bool, + } +} + +impl Stream for Buffer +where + St: TryStream, +{ + type Item = Result; + + fn poll_next( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> Poll> { + let mut this = self.project(); + loop { + match this.stream.as_mut().try_poll_next(cx) { + Poll::Ready(Some(Ok(item))) => { + this.buffer.push_back(item); + if this.buffer.len() >= *this.capacity { + warn!( + capacity = this.capacity, + current = this.buffer.len(), + "Stream buffer at capacity" + ); + break; + } + } + Poll::Ready(Some(Err(e))) => { + return Poll::Ready(Some(Err(e))); + } + Poll::Ready(None) => { + *this.closed = true; + break; + } + Poll::Pending => break, + } + } + if this.buffer.is_empty() { + if *this.closed { + Poll::Ready(None) + } else { + Poll::Pending + } + } else { + Poll::Ready(this.buffer.pop_front().map(Ok)) + } + } +} + +impl Buffer +where + St: TryStream, +{ + /// Create a new `Buffer` stream wrapping the given stream with the specified capacity. + pub fn new(stream: St, capacity: usize) -> Self { + Self { + stream, + capacity, + buffer: VecDeque::with_capacity(capacity), + closed: false, + } + } +} diff --git a/crates/app/src/extensions.rs b/crates/app/src/extensions.rs deleted file mode 100644 index 2d535ee..0000000 --- a/crates/app/src/extensions.rs +++ /dev/null @@ -1,180 +0,0 @@ -use futures::{FutureExt, Stream, ready}; -use pin_project_lite::pin_project; -use std::{pin::Pin, task::Poll}; -use void_types::Block; - -#[cfg(test)] -mod tests; - -pin_project! { - /// Stream produced by calling `VoidStream::state_transition_extensions`. - pub struct Extensions - { - #[pin] - stream: St, - state: A, - state_transition_func: F, - pre_extension: Pre, - post_extension: Post, - _pre_marker: std::marker::PhantomData, - } -} - -pin_project! { - /// Stream produced by calling `VoidStream::state_transition_extensions_async`. - pub struct ExtensionsAsync - { - #[pin] - stream: St, - state: A, - #[pin] - access_fut: Option>>>, - state_transition_func: Option, - pre_extension: Option
,
-        post_extension: Option,
-    }
-}
-
-impl Stream for Extensions
-where
-    A: void_types::SyncAccess,
-    F: Fn(&Block, &mut A::Item<'_>),
-    Pre: Fn(&Block, &mut A::Item<'_>),
-    Post: Fn(&Block, &mut A::Item<'_>),
-    St: Stream,
-    T: for<'a, 'b> From<&'a mut ::Item<'b>>,
-{
-    type Item = (Block, T);
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let this = self.project();
-        let out = ready!(this.stream.poll_next(cx));
-        match out {
-            Some(block) => {
-                let derived = this.state.access(|item| {
-                    (this.pre_extension)(&block, item);
-                    (this.state_transition_func)(&block, item);
-                    (this.post_extension)(&block, item);
-                    item.into()
-                });
-                Poll::Ready(Some((block, derived)))
-            }
-            None => Poll::Ready(None),
-        }
-    }
-}
-
-impl Stream for ExtensionsAsync
-where
-    A: void_types::AsyncAccess + Clone + Send + 'static,
-    F: Fn(&Block, &mut A::Item<'_>) + Send + 'static,
-    Pre: for<'a> Fn(&Block, &mut A::Item<'a>) + Send + 'static,
-    Post: for<'a> Fn(&Block, &mut A::Item<'a>) + Send + 'static,
-    St: Stream,
-    T: for<'a, 'b> From<&'a mut A::Item<'b>> + Send + 'static,
-{
-    type Item = (Block, T);
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let mut this = self.project();
-
-        Poll::Ready(loop {
-            if let Some(fut) = this.access_fut.as_mut().as_pin_mut() {
-                let (block, state_transition_func, pre_extension, post_extension, derived) =
-                    ready!(fut.poll(cx));
-                this.access_fut.set(None);
-                *this.state_transition_func = Some(state_transition_func);
-                *this.pre_extension = Some(pre_extension);
-                *this.post_extension = Some(post_extension);
-                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().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 {
-                        state
-                            .access(move |item| {
-                                (pre_extension)(&block, item);
-                                (state_transition_func)(&block, item);
-                                (post_extension)(&block, item);
-                                let derived = item.into();
-                                (
-                                    block,
-                                    state_transition_func,
-                                    pre_extension,
-                                    post_extension,
-                                    derived,
-                                )
-                            })
-                            .await
-                    }
-                    .boxed(),
-                ));
-            } else {
-                // Stream ended.
-                break None;
-            }
-        })
-    }
-}
-
-impl Extensions
-where
-    A: void_types::SyncAccess,
-{
-    /// Create a new `Extensions` stream.
-    pub fn new(
-        stream: St,
-        state: A,
-        state_transition_func: F,
-        pre_extension: Pre,
-        post_extension: Post,
-    ) -> Self {
-        Self {
-            stream,
-            state,
-            state_transition_func,
-            pre_extension,
-            post_extension,
-            _pre_marker: std::marker::PhantomData,
-        }
-    }
-}
-
-impl ExtensionsAsync
-where
-    A: void_types::AsyncAccess + Clone + Send + 'static,
-    F: Fn(&Block, &mut A::Item<'_>) + Send + 'static,
-    Pre: for<'a> Fn(&Block, &mut A::Item<'a>) + Send + 'static,
-    Post: for<'a> Fn(&Block, &mut A::Item<'a>) + Send + 'static,
-    St: Stream,
-    T: for<'a, 'b> From<&'a mut A::Item<'b>>,
-{
-    /// Create a new `ExtensionsAsync` stream.
-    pub fn new(
-        stream: St,
-        state: A,
-        state_transition_func: F,
-        pre_extension: Pre,
-        post_extension: Post,
-    ) -> Self {
-        Self {
-            stream,
-            state,
-            access_fut: None,
-            state_transition_func: Some(state_transition_func),
-            pre_extension: Some(pre_extension),
-            post_extension: Some(post_extension),
-        }
-    }
-}
diff --git a/crates/app/src/extensions/tests.rs b/crates/app/src/extensions/tests.rs
deleted file mode 100644
index 10248df..0000000
--- a/crates/app/src/extensions/tests.rs
+++ /dev/null
@@ -1,227 +0,0 @@
-use std::{
-    borrow::{Borrow, BorrowMut},
-    sync::Arc,
-};
-
-use futures::StreamExt;
-use void_types::{AsyncAccess, Lock};
-
-use crate::VoidStream;
-
-use super::*;
-
-struct State(u64);
-
-impl From<&mut State> for () {
-    fn from(_: &mut State) -> Self {}
-}
-
-#[derive(Clone)]
-struct Db(Arc>);
-
-struct Inner {
-    value: u64,
-    api_state: bool,
-    block_cache: Vec,
-}
-
-struct Tx<'a> {
-    inner: tokio::sync::MutexGuard<'a, Inner>,
-    delta: Vec,
-}
-
-#[derive(Debug)]
-enum Delta {
-    SetValue(u64),
-    SetApiState(bool),
-    AddBlock(Block),
-}
-
-impl<'a, 'b> From<&'a mut Tx<'b>> for () {
-    fn from(_: &'a mut Tx<'b>) -> Self {}
-}
-
-impl AsyncAccess for Db {
-    type Item<'a>
-        = Tx<'a>
-    where
-        Self: 'a;
-
-    async fn access(&self, f: F) -> R
-    where
-        F: FnOnce(&mut Self::Item<'_>) -> R + Send,
-    {
-        let guard = self.0.lock().await;
-        let mut tx = Tx {
-            inner: guard,
-            delta: vec![],
-        };
-        let r = f(&mut tx);
-        tx.commit();
-        r
-    }
-}
-
-trait AppState {
-    fn read(&self) -> u64;
-    fn write(&mut self, value: u64);
-}
-
-trait ApiState {
-    fn read_app_state(&self) -> u64;
-    fn write(&mut self, value: bool);
-    fn add_block(&mut self, block: Block);
-}
-
-impl AppState for Tx<'_> {
-    fn write(&mut self, value: u64) {
-        self.delta.push(Delta::SetValue(value));
-    }
-    fn read(&self) -> u64 {
-        let mut v = self.inner.value;
-        for delta in &self.delta {
-            if let Delta::SetValue(val) = delta {
-                v = *val;
-            }
-        }
-        v
-    }
-}
-
-impl ApiState for Tx<'_> {
-    fn write(&mut self, value: bool) {
-        self.delta.push(Delta::SetApiState(value));
-    }
-    fn read_app_state(&self) -> u64 {
-        self.read()
-    }
-    fn add_block(&mut self, block: Block) {
-        self.delta.push(Delta::AddBlock(block));
-    }
-}
-
-impl Tx<'_> {
-    fn commit(mut self) {
-        for delta in self.delta {
-            match delta {
-                Delta::SetValue(v) => self.inner.value = v,
-                Delta::SetApiState(s) => self.inner.api_state = s,
-                Delta::AddBlock(b) => self.inner.block_cache.push(b),
-            }
-        }
-    }
-}
-
-#[tokio::test]
-async fn test_extensions() {
-    let blocks = vec![
-        Block {
-            height: 1,
-            parent_hash: [0u8; 32],
-            events: vec![vec![1, 2, 3]],
-        },
-        Block {
-            height: 2,
-            parent_hash: [0u8; 32],
-            events: vec![vec![4, 5, 6]],
-        },
-    ];
-    let block_cache = Lock::new(Vec::new());
-    let app_state = Lock::new(State(0));
-    let api_state = Lock::new(false);
-
-    fn state_transition_function(block: &Block, state: &mut impl BorrowMut) {
-        state.borrow_mut().0 += block.height;
-    }
-
-    fn post_state_extension(block: &Block, state: &impl Borrow, api_state: &mut bool) {
-        *api_state = !(state.borrow().0 % 2 == 0 && block.height.is_multiple_of(2));
-    }
-
-    futures::stream::iter(blocks)
-        .state_transition_extensions(
-            app_state.clone(),
-            state_transition_function,
-            {
-                let block_cache = block_cache.clone();
-                move |block, _| {
-                    block_cache.access(|cache| {
-                        cache.push(block.clone());
-                    });
-                }
-            },
-            {
-                let api_state = api_state.clone();
-                move |block, state: &mut State| {
-                    api_state.access(|api| {
-                        post_state_extension(block, state, api);
-                    });
-                }
-            },
-        )
-        .for_each(|_: (_, ())| async {})
-        .await;
-
-    app_state.access(|state| {
-        assert_eq!(state.0, 3);
-    });
-    block_cache.access(|cache| {
-        assert_eq!(cache.len(), 2);
-        assert_eq!(cache[0].height, 1);
-        assert_eq!(cache[1].height, 2);
-    });
-    api_state.access(|api| {
-        assert!(*api);
-    });
-}
-
-#[tokio::test]
-async fn test_extensions_async() {
-    let blocks = vec![
-        Block {
-            height: 1,
-            parent_hash: [0u8; 32],
-            events: vec![vec![1, 2, 3]],
-        },
-        Block {
-            height: 2,
-            parent_hash: [0u8; 32],
-            events: vec![vec![4, 5, 6]],
-        },
-    ];
-    let db = Db(Arc::new(tokio::sync::Mutex::new(Inner {
-        value: 0,
-        api_state: false,
-        block_cache: vec![],
-    })));
-
-    fn state_transition_function(block: &Block, state: &mut impl AppState) {
-        state.write(state.read() + block.height);
-    }
-
-    fn pre_state_extension(block: &Block, state: &mut A) {
-        state.add_block(block.clone());
-    }
-
-    fn post_state_extension(block: &Block, state: &mut A) {
-        state.write(!(state.read_app_state().is_multiple_of(2) && block.height.is_multiple_of(2)));
-    }
-
-    futures::stream::iter(blocks)
-        .state_transition_extensions_async(
-            db.clone(),
-            // Closures are required when the access item has a lifetime parameter
-            |block, state| state_transition_function(block, state),
-            |block, state| pre_state_extension(block, state),
-            |block, state| post_state_extension(block, state),
-        )
-        .for_each(|_: (_, ())| async {})
-        .await;
-
-    let guard = db.0.lock().await;
-    assert_eq!(guard.value, 3);
-    assert!(guard.api_state);
-    assert_eq!(guard.block_cache.len(), 2);
-    assert_eq!(guard.block_cache[0].height, 1);
-    assert_eq!(guard.block_cache[1].height, 2);
-}
diff --git a/crates/app/src/lib.rs b/crates/app/src/lib.rs
index e73003b..c8ab844 100644
--- a/crates/app/src/lib.rs
+++ b/crates/app/src/lib.rs
@@ -1,148 +1,46 @@
 #![deny(missing_docs)]
 //! This create provides helpers for building streams that turn blocks into state and state into proofs.
 
-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 futures::TryStream;
+use tracing::info;
+use tracing::{instrument, warn};
 use void_types::Block;
 use void_types::Height;
-use void_types::Signed;
 
+pub use crate::try_push_notification::PushNotification;
 pub use block_awaits_proofs::BlocksAwaitProofs;
-pub use extensions::Extensions;
-pub use extensions::ExtensionsAsync;
+pub use block_height_parent::BlockHeightParent;
+pub use buffer::Buffer;
+pub use prove::Prove;
 pub use scan_owned::ScanOwned;
+pub use sign::Sign;
+pub use try_interval_latest::IntervalLatest;
+
+macro_rules! ready_ok {
+    ($e:expr $(,)?) => {
+        match futures::ready!($e) {
+            Some(Ok(v)) => Some(v),
+            None => None,
+            Some(Err(e)) => return Poll::Ready(Some(Err(e))),
+        }
+    };
+}
 
 #[cfg(test)]
 mod tests;
 
 mod block_awaits_proofs;
-mod extensions;
+mod block_height_parent;
+mod buffer;
+mod prove;
 mod scan_owned;
+mod sign;
+mod try_interval_latest;
+mod try_push_notification;
 
 /// Extension trait for streams that produces blocks, state or proofs.
-pub trait VoidStream: Stream {
-    /// Apply a state transition function to each block in the stream, to update state
-    /// that produces a stream of blocks and commitments.
-    ///
-    /// The state is accessed synchronously.
-    #[instrument(skip_all)]
-    fn state_transition_sync(
-        self,
-        state: A,
-        state_transition_func: F,
-    ) -> StateTransitionSync
-    where
-        A: void_types::SyncAccess,
-        T: for<'a, 'b> From<&'a mut ::Item<'b>>,
-        F: Fn(&Block, &mut A::Item<'_>),
-        Self: Sized,
-        Self: Stream,
-    {
-        info!("Sync state transition stream created");
-        StateTransitionSync {
-            stream: self,
-            state,
-            state_transition_func,
-            _pre_marker: std::marker::PhantomData,
-        }
-    }
-
-    /// Apply a state transition function to each block in the stream, to update state
-    /// that produces a stream of blocks and commitments.
-    ///
-    /// The state is accessed asynchronously.
-    #[instrument(skip(self, state_transition_func, state))]
-    fn state_transition_async(
-        self,
-        state: A,
-        state_transition_func: F,
-    ) -> StateTransitionAsync
-    where
-        A: void_types::AsyncAccess + Clone + Send + 'static,
-        F: for<'a> Fn(&Block, &mut A::Item<'a>) + Send + 'static,
-        Self: Sized,
-        T: for<'a> From<&'a mut ::Item<'a>>,
-        Self: Stream,
-    {
-        info!("Async state transition stream created");
-        StateTransitionAsync {
-            stream: self,
-            state,
-            access_fut: None,
-            state_transition_func: Some(state_transition_func),
-        }
-    }
-
-    /// Apply a state transition function to each block in the stream, to update state
-    /// that produces a stream of blocks and items derived from state.
-    ///
-    /// Apply side effect functions to the same block with pre or post state transition.
-    ///
-    /// The side effect functions are applied atomically so the state is only written after the side effects are complete.
-    ///
-    /// The state is accessed synchronously.
-    fn state_transition_extensions(
-        self,
-        state: A,
-        state_transition_func: F,
-        pre_extension: Pre,
-        post_extension: Post,
-    ) -> Extensions
-    where
-        A: void_types::SyncAccess,
-        F: Fn(&Block, &mut A::Item<'_>),
-        Pre: Fn(&Block, &mut A::Item<'_>),
-        Post: Fn(&Block, &mut A::Item<'_>),
-        Self: Sized,
-        Self: Stream,
-        T: for<'a, 'b> From<&'a mut ::Item<'b>>,
-    {
-        Extensions::new(
-            self,
-            state,
-            state_transition_func,
-            pre_extension,
-            post_extension,
-        )
-    }
-
-    /// Apply a state transition function to each block in the stream, to update state
-    /// that produces a stream of blocks and items derived from state.
-    ///
-    /// Apply side effect functions to the same block with pre or post state transition.
-    ///
-    /// The state is accessed asynchronously.
-    fn state_transition_extensions_async(
-        self,
-        state: A,
-        state_transition_func: F,
-        pre_extension: Pre,
-        post_extension: Post,
-    ) -> ExtensionsAsync
-    where
-        A: void_types::AsyncAccess + Clone + Send + 'static,
-        F: Fn(&Block, &mut A::Item<'_>) + Send + 'static,
-        Pre: Fn(&Block, &mut A::Item<'_>) + Send + 'static,
-        Post: Fn(&Block, &mut A::Item<'_>) + Send + 'static,
-        Self: Sized,
-        T: for<'a, 'b> From<&'a mut A::Item<'b>>,
-        Self: Stream,
-    {
-        ExtensionsAsync::new(
-            self,
-            state,
-            state_transition_func,
-            pre_extension,
-            post_extension,
-        )
-    }
-
+pub trait VoidStream: TryStream {
     /// Adds the correct height and parent to each block in the stream.
     #[instrument(skip(self), fields(starting_parent_height))]
     fn block_height_parent(
@@ -152,119 +50,97 @@ pub trait VoidStream: Stream {
     ) -> BlockHeightParent
     where
         Self: Sized,
-        Self: Stream,
+        Self: TryStream,
     {
         info!("Block height parent stream created");
-        BlockHeightParent {
-            stream: self,
-            prev_height: previous_parent_height,
-            prev_parent_hash: previous_parent_hash,
-        }
+        BlockHeightParent::new(self, previous_parent_height, previous_parent_hash)
     }
 
     /// Produce a stream of proofs from a stream of blocks and state commits.
     /// This runs the proof function in a blocking task.
+    ///
+    /// # Errors
+    /// If the proof function returns an error, it is propagated as a stream error.
+    /// Any errors upstream will also be propagated.
     #[instrument(skip(self, prove_func))]
-    fn prove(self, prove_func: F) -> Prove
+    fn prove(self, prove_func: F) -> Prove
     where
-        F: FnMut(Block, T) -> Vec + Send + 'static,
-        Self: Stream,
+        F: FnMut(Block, T) -> Result, Pe> + Send + 'static,
+        Self: TryStream)>,
         Self: Sized,
         T: Send + 'static,
+        Pe: Send + 'static,
+        Self::Error: From,
     {
         info!("Proof stream created");
-        Prove {
-            stream: self,
-            jh: None,
-            prove_func: Some(prove_func),
-        }
+        Prove::new(self, prove_func)
     }
 
     /// Produce a stream of signatures from a stream of state commitments.
+    ///
+    /// # Errors
+    /// If the sign function returns an error, it is propagated as a stream error.
+    /// Any errors upstream will also be propagated.
     #[instrument(skip(self, signer, sign))]
-    fn sign(self, signer: S, sign: F) -> Sign
+    fn sign(self, signer: S, sign: F) -> Sign
     where
         Self: Sized,
-        Self: Stream)>,
+        Self: TryStream)>,
         D: for<'a> From<&'a T> + AsRef<[u8]>,
-        F: FnMut(&S, Height) -> Option>,
+        F: FnMut(&S, Height) -> Result, Se>,
+        Self::Error: From,
     {
         info!("Signing stream created");
-        Sign {
-            stream: self,
-            signer,
-            sign,
-            _marker: std::marker::PhantomData,
-        }
+        Sign::new(self, signer, sign)
     }
 
     /// Yield items from the stream at a fixed interval.
     /// Only the latest item is returned at each interval.
     /// If the stream produces items faster than the interval, intermediate items are dropped.
     /// If the duration is zero, then the latest item is returned when polled.
+    ///
+    /// # Errors
+    /// Any errors upstream will be propagated immediately.
     #[instrument(
         skip(self),
         fields(interval_ms = duration.as_millis())
     )]
-    fn interval_latest(self, duration: std::time::Duration) -> Interval
+    fn interval_latest(self, duration: std::time::Duration) -> IntervalLatest
     where
         Self: Sized,
+        Self: TryStream,
     {
         info!("Interval stream created");
-        Interval {
-            stream: self,
-            interval_stream: if duration.is_zero() {
-                futures::StreamExt::boxed(futures::stream::repeat(()))
-            } else {
-                futures::StreamExt::boxed(futures::stream::unfold(
-                    tokio::time::interval(duration),
-                    |mut interval| async move {
-                        interval.tick().await;
-                        Some(((), interval))
-                    },
-                ))
-            },
-            latest_item: None,
-            done: false,
-        }
+        IntervalLatest::new(self, duration)
     }
 
     /// Buffer items from the stream up to the given capacity.
+    ///
+    /// # Errors
+    /// Any errors upstream will be propagated immediately.
     #[instrument(skip(self), fields(capacity))]
     fn buffer(self, capacity: usize) -> Buffer
     where
         Self: Sized,
+        Self: TryStream,
     {
         info!("Buffer stream created");
-        Buffer {
-            stream: self,
-            capacity,
-            buffer: VecDeque::with_capacity(capacity),
-            closed: false,
-        }
+        Buffer::new(self, capacity)
     }
 
     /// Push a notification when an item is received from the stream.
     /// Passes the item through unchanged.
+    ///
+    /// # Errors
+    /// Any errors upstream will be propagated without sending a notification.
     #[instrument(skip(self, notification))]
     fn push_notification(self, notification: Notification) -> PushNotification
     where
         Self: Sized,
+        Self: TryStream,
     {
         info!("Notification stream created");
-        PushNotification {
-            stream: self,
-            notification,
-        }
-    }
-
-    /// Attach the height to each item in the stream using the block
-    fn at_height(self) -> AtHeight
-    where
-        Self: Sized,
-        Self: Stream,
-    {
-        AtHeight { stream: self }
+        PushNotification::new(self, notification)
     }
 
     /// Waits for proofs at the same or future location before yielding blocks.
@@ -273,7 +149,7 @@ pub trait VoidStream: Stream {
     fn blocks_await_proofs
(self, proof_locations_stream: P) -> BlocksAwaitProofs
     where
         Self: Sized,
-        Self: Stream,
+        Self: TryStream,
         P: Stream,
     {
         BlocksAwaitProofs::new(self, proof_locations_stream)
@@ -281,12 +157,13 @@ pub trait VoidStream: Stream {
 
     /// A stream combinator that maintains state across stream items.
     /// Similar to `fold` but yields intermediate results.
-    fn scan_owned(self, initial_state: T, f: F) -> ScanOwned
+    fn scan_owned(self, initial_state: T, f: F) -> ScanOwned
     where
         Self: Sized,
-        Self: Stream,
-        F: FnMut(T, Self::Item) -> Fut,
-        Fut: std::future::Future,
+        Self: TryStream,
+        F: FnMut(T, Self::Ok) -> Fut,
+        Fut: std::future::Future)>,
+        Self::Error: From,
     {
         ScanOwned::new(self, initial_state, f)
     }
@@ -300,103 +177,7 @@ pub trait UpdateLatestBlock {
     fn update_latest_block(&mut self, height: u64, hash: [u8; 32]) -> Result<(), Self::Error>;
 }
 
-impl VoidStream for T where T: Stream {}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::prove`..
-    pub struct Prove {
-        #[pin]
-        stream: St,
-        #[pin]
-        jh: Option, F)>>>,
-        prove_func: Option,
-    }
-}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::state_transition_sync`.
-    pub struct StateTransitionSync {
-        #[pin]
-        stream: St,
-        state: A,
-        state_transition_func: F,
-        _pre_marker: std::marker::PhantomData,
-    }
-}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::state_transition_async`.
-    pub struct StateTransitionAsync
-    {
-        #[pin]
-        stream: St,
-        state: A,
-        #[pin]
-        access_fut: Option>>>,
-        state_transition_func: Option,
-    }
-}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::block_height_parent`.
-    pub struct BlockHeightParent {
-        #[pin]
-        stream: St,
-        prev_height: Option,
-        prev_parent_hash: [u8; 32],
-    }
-}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::buffer`.
-    pub struct Buffer {
-        #[pin]
-        stream: St,
-        capacity: usize,
-        buffer: VecDeque,
-        closed: bool,
-    }
-}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::push_notification`.
-    pub struct PushNotification {
-        #[pin]
-        stream: St,
-        notification: Notification,
-    }
-}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::sign`.
-    pub struct Sign {
-        #[pin]
-        stream: St,
-        signer: S,
-        sign: F,
-        _marker: std::marker::PhantomData,
-    }
-}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::interval_latest`.
-    pub struct Interval {
-        #[pin]
-        stream: St,
-        #[pin]
-        interval_stream: Pin>>,
-        latest_item: Option,
-        done: bool,
-    }
-}
-
-pin_project! {
-    /// Stream produced by calling `VoidStream::at_height`.
-    pub struct AtHeight {
-        #[pin]
-        stream: St,
-    }
-}
+impl VoidStream for T where T: TryStream {}
 
 #[derive(Clone)]
 /// Notification can be used to send and receive notifications.
@@ -405,312 +186,6 @@ pub struct Notification {
     rx: tokio::sync::watch::Receiver<()>,
 }
 
-impl Stream for StateTransitionSync
-where
-    A: void_types::SyncAccess,
-    F: Fn(&Block, &mut A::Item<'_>),
-    St: Stream,
-    T: for<'a, 'b> From<&'a mut ::Item<'b>>,
-{
-    type Item = (Block, T);
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let this = self.project();
-        let out = ready!(this.stream.poll_next(cx));
-        match out {
-            Some(block) => {
-                let derived = this.state.access(|item| {
-                    (this.state_transition_func)(&block, item);
-                    item.into()
-                });
-                Poll::Ready(Some((block, derived)))
-            }
-            None => Poll::Ready(None),
-        }
-    }
-}
-
-impl Stream for StateTransitionAsync
-where
-    A: void_types::AsyncAccess + Clone + Send + 'static,
-    F: for<'a> Fn(&Block, &mut A::Item<'a>) + Send + 'static,
-    St: Stream,
-    T: for<'a, 'b> From<&'a mut ::Item<'b>> + Send + 'static,
-{
-    type Item = (Block, T);
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let mut this = self.project();
-
-        Poll::Ready(loop {
-            if let Some(fut) = this.access_fut.as_mut().as_pin_mut() {
-                let (block, state_transition_func, derived) = ready!(fut.poll(cx));
-                this.access_fut.set(None);
-                *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(
-                    async move {
-                        state
-                            .access(move |item| {
-                                (state_transition_func)(&block, item);
-                                let derived = item.into();
-                                (block, state_transition_func, derived)
-                            })
-                            .await
-                    }
-                    .boxed(),
-                ));
-            } else {
-                // Stream ended.
-                break None;
-            }
-        })
-    }
-}
-
-impl Stream for Prove
-where
-    F: FnMut(Block, T) -> Vec + Send + 'static,
-    St: Stream,
-    T: Send + 'static,
-{
-    type Item = Vec;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let mut this = self.project();
-
-        Poll::Ready(loop {
-            if let Some(fut) = this.jh.as_mut().as_pin_mut() {
-                let item = ready!(fut.poll(cx));
-                this.jh.set(None);
-                match item {
-                    Ok((item, f)) => {
-                        *this.prove_func = Some(f);
-                        break Some(item);
-                    }
-                    Err(e) => {
-                        error!(error = %e, "Proof generation task failed");
-                        break None;
-                    }
-                }
-            } else if let Some((block, witness)) = ready!(this.stream.as_mut().poll_next(cx)) {
-                match this.prove_func.take() {
-                    Some(mut f) => this.jh.set(Some(
-                        tokio::task::spawn_blocking(move || {
-                            let r = (f)(block, witness);
-                            (r, f)
-                        })
-                        .in_current_span(),
-                    )),
-                    None => break None,
-                }
-            } else {
-                break None;
-            }
-        })
-    }
-}
-
-impl Stream for BlockHeightParent
-where
-    St: Stream,
-{
-    type Item = Block;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let mut this = self.project();
-        let Some(mut block) = ready!(this.stream.as_mut().poll_next(cx)) else {
-            return Poll::Ready(None);
-        };
-        block.height = this.prev_height.map_or(0, |h| h.saturating_add(1));
-        block.parent_hash = *this.prev_parent_hash;
-        *this.prev_height = Some(block.height);
-        *this.prev_parent_hash = void_hash::Hash::hash(&block);
-        Poll::Ready(Some(block))
-    }
-}
-
-impl Stream for Buffer
-where
-    St: Stream,
-{
-    type Item = St::Item;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let mut this = self.project();
-        loop {
-            match this.stream.as_mut().poll_next(cx) {
-                Poll::Ready(Some(item)) => {
-                    this.buffer.push_back(item);
-                    if this.buffer.len() >= *this.capacity {
-                        warn!(
-                            capacity = this.capacity,
-                            current = this.buffer.len(),
-                            "Stream buffer at capacity"
-                        );
-                        break;
-                    }
-                }
-                Poll::Ready(None) => {
-                    *this.closed = true;
-                    break;
-                }
-                Poll::Pending => break,
-            }
-        }
-        if this.buffer.is_empty() {
-            if *this.closed {
-                Poll::Ready(None)
-            } else {
-                Poll::Pending
-            }
-        } else {
-            Poll::Ready(this.buffer.pop_front())
-        }
-    }
-}
-
-impl Stream for PushNotification
-where
-    St: Stream,
-{
-    type Item = St::Item;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let this = self.project();
-        let item = ready!(this.stream.poll_next(cx));
-        if item.is_some() {
-            this.notification.notify();
-        }
-        Poll::Ready(item)
-    }
-}
-
-impl Stream for Sign
-where
-    St: Stream)>,
-    D: for<'a> From<&'a T> + AsRef<[u8]>,
-    F: FnMut(&S, Height) -> Option>,
-{
-    type Item = Signed>;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let mut this = self.project();
-        loop {
-            let item = ready!(this.stream.as_mut().poll_next(cx));
-            let Some((_, item)) = item else {
-                return Poll::Ready(None);
-            };
-            let Height { data, block_height } = item;
-            let summary = D::from(&data);
-            let summary = Height {
-                block_height,
-                data: summary,
-            };
-            let signature = (this.sign)(this.signer, summary);
-            let item = Height { block_height, data };
-            match signature {
-                None => continue,
-                Some(signature) => {
-                    return Poll::Ready(Some(Signed {
-                        signature,
-                        data: item,
-                    }));
-                }
-            }
-        }
-    }
-}
-
-impl Stream for Interval
-where
-    St: Stream,
-{
-    type Item = St::Item;
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let mut this = self.project();
-
-        if *this.done {
-            ready!(this.interval_stream.as_mut().poll_next(cx));
-            return Poll::Ready(this.latest_item.take());
-        }
-
-        loop {
-            match this.stream.as_mut().poll_next(cx) {
-                Poll::Ready(Some(item)) => {
-                    *this.latest_item = Some(item);
-                }
-                Poll::Ready(None) => {
-                    *this.done = true;
-                    ready!(this.interval_stream.as_mut().poll_next(cx));
-                    break Poll::Ready(this.latest_item.take());
-                }
-                Poll::Pending => {
-                    ready!(this.interval_stream.as_mut().poll_next(cx));
-                    if this.latest_item.is_some() {
-                        debug!("Emitting latest buffered item (intermediate items were dropped)");
-                        break Poll::Ready(this.latest_item.take());
-                    } else {
-                        break Poll::Pending;
-                    }
-                }
-            }
-        }
-    }
-}
-
-impl Stream for AtHeight
-where
-    St: Stream,
-    St: Stream,
-{
-    type Item = (Block, Height);
-
-    fn poll_next(
-        self: std::pin::Pin<&mut Self>,
-        cx: &mut std::task::Context<'_>,
-    ) -> Poll> {
-        let this = self.project();
-        let item = ready!(this.stream.poll_next(cx));
-        match item {
-            Some((block, data)) => {
-                let block_height = block.height;
-                Poll::Ready(Some((block, Height { block_height, data })))
-            }
-            None => Poll::Ready(None),
-        }
-    }
-}
-
 impl Notification {
     /// Create a new notification.
     pub fn new() -> Self {
diff --git a/crates/app/src/prove.rs b/crates/app/src/prove.rs
new file mode 100644
index 0000000..ea5c10f
--- /dev/null
+++ b/crates/app/src/prove.rs
@@ -0,0 +1,89 @@
+use futures::Stream;
+use futures::TryStream;
+use futures::ready;
+use pin_project_lite::pin_project;
+use std::task::Poll;
+use tracing::Instrument;
+use tracing::error;
+use tracing::instrument::Instrumented;
+use void_types::Block;
+use void_types::Height;
+
+pin_project! {
+    /// Stream produced by calling `VoidStream::prove`..
+    pub struct Prove {
+        #[pin]
+        stream: St,
+        #[pin]
+        jh: Option>, Pe>, F)>>>,
+        prove_func: Option,
+    }
+}
+
+impl Stream for Prove
+where
+    F: FnMut(Block, T) -> Result, Pe> + Send + 'static,
+    St: TryStream)>,
+    T: Send + 'static,
+    Pe: Send + 'static,
+    St::Error: From,
+{
+    type Item = Result>, St::Error>;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll> {
+        let mut this = self.project();
+
+        Poll::Ready(loop {
+            if let Some(fut) = this.jh.as_mut().as_pin_mut() {
+                let item = ready!(fut.poll(cx));
+                this.jh.set(None);
+                match item {
+                    Ok((item, f)) => {
+                        *this.prove_func = Some(f);
+                        break Some(item.map_err(|e| e.into()));
+                    }
+                    Err(e) => {
+                        error!(error = %e, "Proof generation task failed");
+                        break None;
+                    }
+                }
+            } else if let Some((block, witness)) = ready_ok!(this.stream.as_mut().try_poll_next(cx))
+            {
+                match this.prove_func.take() {
+                    Some(mut f) => this.jh.set(Some(
+                        tokio::task::spawn_blocking(move || {
+                            let r = (f)(block, witness.data);
+                            let r = r.map(|data| Height::new(witness.block_height, data));
+                            (r, f)
+                        })
+                        .in_current_span(),
+                    )),
+                    None => break None,
+                }
+            } else {
+                break None;
+            }
+        })
+    }
+}
+
+impl Prove
+where
+    F: FnMut(Block, T) -> Result, Pe> + Send + 'static,
+    St: TryStream)>,
+    T: Send + 'static,
+    Pe: Send + 'static,
+    St::Error: From,
+{
+    /// Create a new `Prove` stream.
+    pub fn new(stream: St, prove_func: F) -> Self {
+        Self {
+            stream,
+            jh: None,
+            prove_func: Some(prove_func),
+        }
+    }
+}
diff --git a/crates/app/src/scan_owned.rs b/crates/app/src/scan_owned.rs
index a6b4ec0..0e6efff 100644
--- a/crates/app/src/scan_owned.rs
+++ b/crates/app/src/scan_owned.rs
@@ -1,6 +1,6 @@
 use std::task::Poll;
 
-use futures::{Stream, ready};
+use futures::{Stream, TryStream, ready};
 use pin_project_lite::pin_project;
 
 pin_project! {
@@ -17,13 +17,14 @@ pin_project! {
     }
 }
 
-impl Stream for ScanOwned
+impl Stream for ScanOwned
 where
-    St: Stream,
-    F: FnMut(T, St::Item) -> Fut,
-    Fut: std::future::Future,
+    St: TryStream,
+    F: FnMut(T, St::Ok) -> Fut,
+    Fut: std::future::Future)>,
+    St::Error: From,
 {
-    type Item = R;
+    type Item = Result;
 
     fn poll_next(
         self: std::pin::Pin<&mut Self>,
@@ -39,8 +40,8 @@ where
                 *this.state = Some(state);
                 // Clear the future slot for the next operation
                 this.fut.set(None);
-                break Some(item);
-            } else if let Some(item) = ready!(this.stream.as_mut().poll_next(cx)) {
+                break Some(item.map_err(|e| e.into()));
+            } else if let Some(item) = ready_ok!(this.stream.as_mut().try_poll_next(cx)) {
                 // We got a new item from the underlying stream
                 match this.state.take() {
                     Some(state) => {
@@ -62,11 +63,12 @@ where
     }
 }
 
-impl ScanOwned
+impl ScanOwned
 where
-    St: Stream,
-    F: FnMut(T, St::Item) -> Fut,
-    Fut: std::future::Future,
+    St: TryStream,
+    F: FnMut(T, St::Ok) -> Fut,
+    Fut: std::future::Future)>,
+    St::Error: From,
 {
     /// Create a new `ScanOwned` combinator.
     pub fn new(stream: St, init: T, f: F) -> Self {
diff --git a/crates/app/src/sign.rs b/crates/app/src/sign.rs
new file mode 100644
index 0000000..1c916ca
--- /dev/null
+++ b/crates/app/src/sign.rs
@@ -0,0 +1,72 @@
+use futures::Stream;
+use futures::TryStream;
+use pin_project_lite::pin_project;
+use std::task::Poll;
+use void_types::Block;
+use void_types::Height;
+use void_types::Signed;
+
+pin_project! {
+    /// Stream produced by calling `VoidStream::sign`.
+    pub struct Sign {
+        #[pin]
+        stream: St,
+        signer: S,
+        sign: F,
+        _marker: std::marker::PhantomData,
+    }
+}
+
+impl Stream for Sign
+where
+    St: TryStream)>,
+    D: for<'a> From<&'a T> + AsRef<[u8]>,
+    F: FnMut(&S, Height) -> Result, Se>,
+    St::Error: From,
+{
+    type Item = Result>, St::Error>;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll> {
+        let mut this = self.project();
+        let item = ready_ok!(this.stream.as_mut().try_poll_next(cx));
+        let Some((_, item)) = item else {
+            return Poll::Ready(None);
+        };
+        let Height { data, block_height } = item;
+        let summary = D::from(&data);
+        let summary = Height {
+            block_height,
+            data: summary,
+        };
+        let signature = (this.sign)(this.signer, summary);
+        let item = Height { block_height, data };
+        match signature {
+            Err(e) => Poll::Ready(Some(Err(e.into()))),
+            Ok(signature) => Poll::Ready(Some(Ok(Signed {
+                signature,
+                data: item,
+            }))),
+        }
+    }
+}
+
+impl Sign
+where
+    St: TryStream)>,
+    D: for<'a> From<&'a T> + AsRef<[u8]>,
+    F: FnMut(&S, Height) -> Result, Se>,
+    St::Error: From,
+{
+    /// Create a new `Sign` stream.
+    pub fn new(stream: St, signer: S, sign: F) -> Self {
+        Self {
+            stream,
+            signer,
+            sign,
+            _marker: std::marker::PhantomData,
+        }
+    }
+}
diff --git a/crates/app/src/tests.rs b/crates/app/src/tests.rs
index 38294e0..fb45c6e 100644
--- a/crates/app/src/tests.rs
+++ b/crates/app/src/tests.rs
@@ -1,6 +1,6 @@
-use std::sync::Arc;
+use std::{convert::Infallible, sync::Arc};
 
-use futures::StreamExt;
+use futures::{StreamExt, TryStreamExt};
 use void_types::{AsyncAccess, Lock};
 
 use super::*;
@@ -11,14 +11,17 @@ struct Permits {
     state: Lock,
 }
 
-struct State(Vec);
+struct State(Vec, u64, [u8; 32]);
 
 impl AsyncAccess for Permits {
-    type Item<'a> = State;
+    type Item<'a>
+        = State
+    where
+        Self: 'a;
 
     async fn access(&self, f: F) -> R
     where
-        F: for<'a> FnOnce(&mut Self::Item<'a>) -> R + Send,
+        F: FnOnce(&mut Self::Item<'_>) -> R + Send,
     {
         let _permit = self.permits.acquire().await.unwrap();
         self.state.access(f)
@@ -31,9 +34,19 @@ impl From<&mut State> for u64 {
     }
 }
 
+impl UpdateLatestBlock for State {
+    type Error = Infallible;
+
+    fn update_latest_block(&mut self, height: u64, hash: [u8; 32]) -> Result<(), Self::Error> {
+        self.1 = height;
+        self.2 = hash;
+        Ok(())
+    }
+}
+
 #[tokio::test]
 async fn test_state_transition_async() {
-    let state = Lock::new(State(Vec::new()));
+    let state = Lock::new(State(Vec::new(), 0, [0; 32]));
     let permits = Permits {
         permits: Arc::new(tokio::sync::Semaphore::new(3)),
         state: state.clone(),
@@ -49,16 +62,31 @@ async fn test_state_transition_async() {
             parent_hash: [0u8; 32],
             events: vec![vec![4, 5]],
         },
-    ]);
+    ])
+    .map(Ok::<_, Infallible>);
+
+    fn state_transition(block: &Block, state: &mut State) -> Result<(), Infallible> {
+        state
+            .0
+            .push(block.events.iter().map(|e| e.len() as u64).sum());
+        Ok(())
+    }
+
+    async fn access(
+        permits: Permits,
+        block: Block,
+    ) -> (Permits, Result<(Block, Height), Infallible>) {
+        let res = permits
+            .access(|state| apply_transition(block, state, state_transition))
+            .await;
+        (permits, res)
+    }
 
     stream
-        .state_transition_async(permits.clone(), |block, state: &mut State| {
-            state
-                .0
-                .push(block.events.iter().map(|e| e.len() as u64).sum());
-        })
-        .for_each(|_: (_, u64)| futures::future::ready(()))
-        .await;
+        .scan_owned(permits, access)
+        .try_for_each(|_: (_, Height)| futures::future::ready(Ok(())))
+        .await
+        .unwrap();
 
     state.access(|s| {
         assert_eq!(s.0, vec![3, 2]);
@@ -79,12 +107,13 @@ async fn test_interval_latest() {
             interval.tick().await;
             Some((count, (count + 1, interval)))
         },
-    );
+    )
+    .map(Ok::<_, Infallible>);
     let interval_stream = stream.interval_latest(std::time::Duration::from_millis(50));
 
     let mut results = Vec::new();
     futures::pin_mut!(interval_stream);
-    while let Some(item) = interval_stream.next().await {
+    while let Some(Ok(item)) = interval_stream.next().await {
         results.push(item);
     }
 
@@ -102,12 +131,13 @@ async fn test_interval_latest() {
             interval.tick().await;
             Some((count, (count + 1, interval)))
         },
-    );
+    )
+    .map(Ok::<_, Infallible>);
     let interval_stream = stream.interval_latest(std::time::Duration::from_millis(0));
 
     let mut results = Vec::new();
     futures::pin_mut!(interval_stream);
-    while let Some(item) = interval_stream.next().await {
+    while let Some(Ok(item)) = interval_stream.next().await {
         results.push(item);
     }
 
@@ -119,11 +149,12 @@ async fn test_interval_latest() {
         }
         Some((count, count + 1))
     })
+    .map(Ok::<_, Infallible>)
     .interval_latest(std::time::Duration::from_millis(0));
 
     let mut results = Vec::new();
     futures::pin_mut!(interval_stream);
-    while let Some(item) = interval_stream.next().await {
+    while let Some(Ok(item)) = interval_stream.next().await {
         results.push(item);
     }
 
@@ -141,12 +172,13 @@ async fn test_interval_latest() {
             interval.tick().await;
             Some((count, (count + 1, interval)))
         },
-    );
+    )
+    .map(Ok::<_, Infallible>);
     let interval_stream = stream.interval_latest(std::time::Duration::from_millis(0));
 
     let mut results = Vec::new();
     futures::pin_mut!(interval_stream);
-    while let Some(item) = interval_stream.next().await {
+    while let Some(Ok(item)) = interval_stream.next().await {
         results.push(item);
         tokio::time::sleep(std::time::Duration::from_millis(35)).await;
     }
diff --git a/crates/app/src/try_interval_latest.rs b/crates/app/src/try_interval_latest.rs
new file mode 100644
index 0000000..d3bff55
--- /dev/null
+++ b/crates/app/src/try_interval_latest.rs
@@ -0,0 +1,87 @@
+use std::{pin::Pin, task::Poll};
+
+use futures::{Stream, TryStream, ready};
+use pin_project_lite::pin_project;
+use tracing::debug;
+
+pin_project! {
+    /// Stream produced by calling `VoidStream::interval_latest`.
+    pub struct IntervalLatest {
+        #[pin]
+        stream: St,
+        #[pin]
+        interval_stream: Pin>>,
+        latest_item: Option>,
+        done: bool,
+    }
+}
+
+impl Stream for IntervalLatest
+where
+    St: TryStream,
+{
+    type Item = Result;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll> {
+        let mut this = self.project();
+
+        if *this.done {
+            ready!(this.interval_stream.as_mut().poll_next(cx));
+            return Poll::Ready(this.latest_item.take());
+        }
+
+        loop {
+            match this.stream.as_mut().try_poll_next(cx) {
+                Poll::Ready(Some(Ok(item))) => {
+                    *this.latest_item = Some(Ok(item));
+                }
+                Poll::Ready(Some(Err(e))) => {
+                    *this.done = true;
+                    return Poll::Ready(Some(Err(e)));
+                }
+                Poll::Ready(None) => {
+                    *this.done = true;
+                    ready!(this.interval_stream.as_mut().poll_next(cx));
+                    break Poll::Ready(this.latest_item.take());
+                }
+                Poll::Pending => {
+                    ready!(this.interval_stream.as_mut().poll_next(cx));
+                    if this.latest_item.is_some() {
+                        debug!("Emitting latest buffered item (intermediate items were dropped)");
+                        break Poll::Ready(this.latest_item.take());
+                    } else {
+                        break Poll::Pending;
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl IntervalLatest
+where
+    St: TryStream,
+{
+    /// Create a new `IntervalLatest` stream.
+    pub fn new(stream: St, duration: std::time::Duration) -> Self {
+        Self {
+            stream,
+            interval_stream: if duration.is_zero() {
+                futures::StreamExt::boxed(futures::stream::repeat(()))
+            } else {
+                futures::StreamExt::boxed(futures::stream::unfold(
+                    tokio::time::interval(duration),
+                    |mut interval| async move {
+                        interval.tick().await;
+                        Some(((), interval))
+                    },
+                ))
+            },
+            latest_item: None,
+            done: false,
+        }
+    }
+}
diff --git a/crates/app/src/try_push_notification.rs b/crates/app/src/try_push_notification.rs
new file mode 100644
index 0000000..f1080f2
--- /dev/null
+++ b/crates/app/src/try_push_notification.rs
@@ -0,0 +1,47 @@
+use std::task::Poll;
+
+use futures::{Stream, TryStream};
+use pin_project_lite::pin_project;
+
+use crate::Notification;
+
+pin_project! {
+    /// Stream produced by calling `VoidStream::push_notification`.
+    pub struct PushNotification {
+        #[pin]
+        stream: St,
+        notification: Notification,
+    }
+}
+
+impl Stream for PushNotification
+where
+    St: TryStream,
+{
+    type Item = Result;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll> {
+        let this = self.project();
+        let item = ready_ok!(this.stream.try_poll_next(cx));
+        if item.is_some() {
+            this.notification.notify();
+        }
+        Poll::Ready(item.map(Ok))
+    }
+}
+
+impl PushNotification
+where
+    St: TryStream,
+{
+    /// Create a new `TryPushNotification` stream.
+    pub fn new(stream: St, notification: Notification) -> Self {
+        Self {
+            stream,
+            notification,
+        }
+    }
+}
diff --git a/crates/app/tests/app.rs b/crates/app/tests/app.rs
index 77a3a9c..22c1f49 100644
--- a/crates/app/tests/app.rs
+++ b/crates/app/tests/app.rs
@@ -1,11 +1,12 @@
 use std::{
     collections::HashMap,
+    convert::Infallible,
     sync::{Arc, atomic::AtomicU64},
 };
 
 use alloy::primitives::Address;
-use futures::{Stream, StreamExt};
-use void_app::VoidStream;
+use futures::{Stream, StreamExt, TryStreamExt};
+use void_app::{UpdateLatestBlock, VoidStream, apply_transition};
 use void_types::{Block, Height, Lock, Signed};
 
 enum Message {
@@ -70,10 +71,11 @@ impl BalanceState for Commit {
     }
 }
 
-fn block_state_transition(block: &Block, state: &mut S) {
+fn block_state_transition(block: &Block, state: &mut S) -> Result<(), Infallible> {
     for event in &block.events {
         state_transition(event.clone(), state);
     }
+    Ok(())
 }
 
 fn state_transition(event: Vec, state: &mut S) {
@@ -130,22 +132,22 @@ fn commit(state: &mut State) -> Commit {
     Commit(w)
 }
 
-fn prove(block: Block, pre: Commit) -> Vec {
+fn prove(block: Block, pre: Commit) -> Result, Infallible> {
     write_to_zk(pre);
     write_to_zk(block);
     std::thread::sleep(std::time::Duration::from_millis(300));
-    vec![]
+    Ok(vec![])
 }
 
-fn sign(_sk: &(), data: Height<[u8; 8]>) -> Option> {
+fn sign(_sk: &(), data: Height<[u8; 8]>) -> Result, Infallible> {
     let mut bytes = data.data.as_ref().to_vec();
     bytes.extend(&data.block_height.to_be_bytes());
-    Some(bytes)
+    Ok(bytes)
 }
 
 fn write_to_zk(_data: T) {}
 
-async fn fake_server(state: Lock, proofs: Lock>>) {
+async fn fake_server(state: Lock, proofs: Lock>>>) {
     loop {
         state.access(|s| println!("State: {:?}", s));
         proofs.access(|p| println!("Proofs: {:?}", p));
@@ -153,6 +155,14 @@ async fn fake_server(state: Lock, proofs: Lock>>) {
     }
 }
 
+impl UpdateLatestBlock for State {
+    type Error = Infallible;
+
+    fn update_latest_block(&mut self, _height: u64, _hash: [u8; 32]) -> Result<(), Self::Error> {
+        Ok(())
+    }
+}
+
 #[tokio::test]
 async fn app_stream() {
     let state = Lock::new(State::new());
@@ -165,17 +175,19 @@ async fn app_stream() {
         let sync = sync.clone();
         fake_oracle_stream(sync.load(std::sync::atomic::Ordering::SeqCst))
             .take(10)
-            .state_transition_sync(state, block_state_transition)
+            .map(move |block| {
+                state.access(|state| apply_transition(block, state, block_state_transition))
+            })
             .prove(prove)
-            .for_each(move |proof| {
+            .try_for_each(move |proof| {
                 proofs.access(|proofs| proofs.push(proof));
                 sync.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-                futures::future::ready(())
+                futures::future::ready(Ok::<_, Infallible>(()))
             })
     });
 
     tokio::spawn(fake_server(state, proofs));
-    jh.await.unwrap();
+    jh.await.unwrap().unwrap();
 }
 
 #[tokio::test]
@@ -185,14 +197,16 @@ async fn app_stream_sign() {
 
     fake_oracle_stream(0)
         .take(10)
-        .state_transition_sync(state, block_state_transition)
-        .at_height()
+        .map(move |block| {
+            state.access(|state| apply_transition(block, state, block_state_transition))
+        })
         .sign((), sign)
-        .for_each(move |proof| {
+        .try_for_each(move |proof| {
             proofs.access(|proofs| proofs.push(proof));
-            futures::future::ready(())
+            futures::future::ready(Ok::<_, Infallible>(()))
         })
-        .await;
+        .await
+        .unwrap();
 }
 
 impl State {
diff --git a/crates/types/src/lib.rs b/crates/types/src/lib.rs
index 03e09ab..fd79e49 100644
--- a/crates/types/src/lib.rs
+++ b/crates/types/src/lib.rs
@@ -65,6 +65,13 @@ pub struct Height {
     pub data: T,
 }
 
+impl Height {
+    /// Create a new `Height` instance.
+    pub fn new(block_height: u64, data: T) -> Self {
+        Self { block_height, data }
+    }
+}
+
 impl Lock {
     /// Create a new lock wrapping the given value.
     pub fn new(value: T) -> Self {