[consensus/marshal] commonware-glue marshal changes

Author: clabby

consensus/src/marshal/ancestry.rs

@@ -10,6 +10,7 @@ use pin_project::pin_project;
 use std::{
     future::Future,
     pin::Pin,
+    sync::Arc,
     task::{Context, Poll},
 };
 
@@ -54,6 +55,57 @@ pub trait BlockProvider: Clone + Send + 'static {
     }
 }
 
+/// A type-erased [`BlockProvider`] that wraps any concrete provider behind
+/// a shared closure.
+///
+/// Used by actor mailboxes and other channel-based patterns where a generic
+/// `BlockProvider` type parameter must be eliminated before sending across
+/// a channel.
+#[derive(Clone)]
+pub struct ErasedBlockProvider<B: Block> {
+    subscribe:
+        Arc<dyn Fn(<B as Digestible>::Digest) -> BoxFuture<'static, Option<B>> + Send + Sync>,
+    subscribe_parent: Arc<dyn Fn(B) -> BoxFuture<'static, Option<B>> + Send + Sync>,
+}
+
+impl<B: Block> ErasedBlockProvider<B> {
+    /// Erase a concrete [`BlockProvider`] into a type-erased provider.
+    pub fn new<M: BlockProvider<Block = B> + Sync>(provider: M) -> Self {
+        let provider = Arc::new(provider);
+        Self {
+            subscribe: Arc::new({
+                let provider = provider.clone();
+                move |digest| {
+                    let p = provider.as_ref().clone();
+                    Box::pin(p.subscribe(digest))
+                }
+            }),
+            subscribe_parent: Arc::new(move |block| {
+                let p = provider.as_ref().clone();
+                Box::pin(p.subscribe_parent(block))
+            }),
+        }
+    }
+}
+
+impl<B: Block> BlockProvider for ErasedBlockProvider<B> {
+    type Block = B;
+
+    fn subscribe(
+        self,
+        digest: <B as Digestible>::Digest,
+    ) -> impl Future<Output = Option<B>> + Send {
+        (self.subscribe)(digest)
+    }
+
+    fn subscribe_parent(
+        self,
+        block: Self::Block,
+    ) -> impl Future<Output = Option<Self::Block>> + Send {
+        (self.subscribe_parent)(block)
+    }
+}
+
 /// Yields the ancestors of a block while prefetching parents, _not_ including the genesis block.
 ///
 // TODO(<https://github.com/commonwarexyz/monorepo/issues/2982>): Once marshal can also yield the genesis block,
@@ -67,12 +119,37 @@ pub struct AncestorStream<M: BlockProvider> {
 }
 
 impl<M: BlockProvider> AncestorStream<M> {
+    /// Returns a reference to the next block that will be yielded by the
+    /// stream, without consuming it.
+    ///
+    /// Returns `None` if the buffer is empty and the next block is being
+    /// fetched asynchronously.
+    pub fn peek(&self) -> Option<&M::Block> {
+        self.buffered.last()
+    }
+
+    /// Erase the block provider type, producing an
+    /// `AncestorStream<ErasedBlockProvider<M::Block>>`.
+    ///
+    /// The returned stream behaves identically but can be sent across
+    /// channels that require a concrete type.
+    pub fn erase(self) -> AncestorStream<ErasedBlockProvider<M::Block>>
+    where
+        M: Sync,
+    {
+        AncestorStream {
+            buffered: self.buffered,
+            marshal: ErasedBlockProvider::new(self.marshal),
+            pending: self.pending,
+        }
+    }
+
     /// Creates a new [AncestorStream] starting from the given ancestry.
     ///
     /// # Panics
     ///
     /// Panics if the initial blocks are not contiguous in height.
-    pub(crate) fn new(marshal: M, initial: impl IntoIterator<Item = M::Block>) -> Self {
+    pub fn new(marshal: M, initial: impl IntoIterator<Item = M::Block>) -> Self {
         let mut buffered = initial.into_iter().collect::<Vec<M::Block>>();
         buffered.sort_by_key(Heightable::height);
 

consensus/src/marshal/coding/mod.rs

@@ -744,6 +744,11 @@ mod tests {
         harness::prune_finalized_archives::<CodingHarness>();
     }
 
+    #[test_traced("WARN")]
+    fn test_coding_set_floor_without_pruning_preserves_archives() {
+        harness::set_floor_without_pruning_preserves_archives::<CodingHarness>();
+    }
+
     #[test_traced("WARN")]
     fn test_coding_rejects_block_delivery_below_floor() {
         harness::reject_stale_block_delivery_after_floor_update::<CodingHarness>();

consensus/src/marshal/core/actor.rs

@@ -748,17 +748,52 @@ where
                         let finalization = self.get_finalization_by_height(height).await;
                         response.send_lossy(finalization);
                     }
+                    Message::GetProcessedHeight { response } => {
+                        response.send_lossy(self.floor.height);
+                    }
                     Message::HintFinalized { height, targets } => {
-                        // Skip if finalization is already available locally
-                        if self.get_finalization_by_height(height).await.is_some() {
+                        // Skip if height is below the floor
+                        if height < self.floor.height {
                             continue;
                         }
 
-                        self.floor.fetch_targeted_if_permitted(
-                            &mut resolver,
-                            Request::finalized(height),
-                            targets,
-                        );
+                        // Skip if both the finalization and the block are
+                        // already available locally. Both are checked because
+                        // `set_floor` can advance the processed height to
+                        // a height whose block was never finalized locally.
+                        if self.get_finalization_by_height(height).await.is_some()
+                            && self.get_finalized_block(height).await.is_some()
+                        {
+                            continue;
+                        }
+
+                        match targets {
+                            Recipients::All => {
+                                self.floor.fetch_if_permitted(
+                                    &mut resolver,
+                                    Request::<V::Commitment>::finalized(height),
+                                );
+                            }
+                            Recipients::Some(peers) => {
+                                if peers.is_empty() {
+                                    continue;
+                                }
+                                let peers = NonEmptyVec::from_unchecked(peers);
+                                self.floor.fetch_targeted_if_permitted(
+                                    &mut resolver,
+                                    Request::<V::Commitment>::finalized(height),
+                                    peers,
+                                );
+                            }
+                            Recipients::One(peer) => {
+                                let peers = NonEmptyVec::new(peer);
+                                self.floor.fetch_targeted_if_permitted(
+                                    &mut resolver,
+                                    Request::<V::Commitment>::finalized(height),
+                                    peers,
+                                );
+                            }
+                        }
                     }
                     Message::SubscribeByDigest {
                         digest,
@@ -799,7 +834,10 @@ where
                         )
                         .await;
                     }
-                    Message::SetFloor { height } => {
+                    Message::SetFloor {
+                        height,
+                        prune_archives,
+                    } => {
                         if self.floor.height >= height {
                             warn!(
                                 %height,
@@ -822,10 +860,11 @@ where
                         // updating `floor.height`.
                         self.pending_acks.clear();
 
-                        // The floor is durable, so cache/finalized data below it can be pruned.
-                        if let Err(err) = self.prune_after_floor(height).await {
-                            error!(?err, %height, "failed to prune data below floor");
-                            return;
+                        if prune_archives {
+                            if let Err(err) = self.prune_after_floor(height).await {
+                                error!(?err, %height, "failed to prune data below floor");
+                                return;
+                            }
                         }
 
                         // Intentionally keep existing block subscriptions alive. Canceling
@@ -1638,15 +1677,19 @@ where
         finalization: Option<Finalization<P::Scheme, V::Commitment>>,
         application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
     ) -> bool {
-        // Blocks below the last processed height are not useful to us, so we ignore them (this
-        // has the nice byproduct of ensuring we don't call a backing store with a block below the
-        // pruning boundary)
-        if height <= self.floor.height {
+        // Blocks below the last processed height are not useful to us, so we
+        // ignore them (this has the nice byproduct of ensuring we don't call
+        // a backing store with a block below the pruning boundary).
+        //
+        // Blocks at exactly the processed height are allowed, however. After
+        // a `set_floor`, it may be possible that the floor was set to a height
+        // that marshal doesn't have.
+        if height < self.floor.height {
             debug!(
                 %height,
                 floor = %self.floor.height,
                 ?digest,
-                "dropping finalization at or below processed height floor"
+                "dropping finalization below processed height floor"
             );
             return false;
         }