Try errors

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<St: Stream, P: Stream> {
+    pub struct BlocksAwaitProofs<St: TryStream, P: Stream> {
         #[pin]
         stream: St,
         #[pin]
@@ -21,7 +21,7 @@ pin_project! {
 
 impl<St, P> BlocksAwaitProofs<St, P>
 where
-    St: Stream,
+    St: TryStream,
     P: Stream,
 {
     /// Create a new `BlocksAwaitProofs` stream.
@@ -37,11 +37,10 @@ where
 
 impl<St, P> Stream for BlocksAwaitProofs<St, P>
 where
-    St: Stream,
-    St: Stream<Item = Block>,
+    St: TryStream<Ok = Block>,
     P: Stream<Item = u64>,
 {
-    type Item = Block;
+    type Item = Result<Block, St::Error>;
 
     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

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]],
-        }))
+        })))
     );
 }

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<St: TryStream> {
+        #[pin]
+        stream: St,
+        prev_height: Option<u64>,
+        prev_parent_hash: [u8; 32],
+    }
+}
+
+impl<St> Stream for BlockHeightParent<St>
+where
+    St: TryStream<Ok = Block>,
+{
+    type Item = Result<Block, St::Error>;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        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<St> BlockHeightParent<St>
+where
+    St: TryStream<Ok = Block>,
+{
+    /// Create a new `BlockHeightParent` stream.
+    pub fn new(
+        stream: St,
+        previous_parent_height: Option<u64>,
+        previous_parent_hash: [u8; 32],
+    ) -> Self {
+        Self {
+            stream,
+            prev_height: previous_parent_height,
+            prev_parent_hash: previous_parent_hash,
+        }
+    }
+}

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<St: TryStream> {
+        #[pin]
+        stream: St,
+        capacity: usize,
+        buffer: VecDeque<St::Ok>,
+        closed: bool,
+    }
+}
+
+impl<St> Stream for Buffer<St>
+where
+    St: TryStream,
+{
+    type Item = Result<St::Ok, St::Error>;
+
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> Poll<Option<Self::Item>> {
+        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<St> Buffer<St>
+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,
+        }
+    }
+}