[consensus/marshal] commonware-glue marshal changes

Author: clabby

consensus/src/marshal/ancestry.rs

@@ -2,6 +2,7 @@
 
 use crate::{types::Height, Block, Heightable};
 use commonware_cryptography::Digestible;
+use commonware_utils::sync::Mutex;
 use futures::{
     future::{BoxFuture, OptionFuture},
     FutureExt, Stream,
@@ -10,6 +11,7 @@ use pin_project::pin_project;
 use std::{
     future::Future,
     pin::Pin,
+    sync::Arc,
     task::{Context, Poll},
 };
 
@@ -34,6 +36,45 @@ pub trait BlockProvider: Clone + Send + 'static {
     ) -> impl Future<Output = Option<Self::Block>> + Send;
 }
 
+/// 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> {
+    fetch: Arc<dyn Fn(<B as Digestible>::Digest) -> BoxFuture<'static, Option<B>> + Send + Sync>,
+}
+
+impl<B: Block> ErasedBlockProvider<B> {
+    /// Erase a concrete [`BlockProvider`] into a type-erased provider.
+    ///
+    /// The provider is wrapped in a `Mutex` so that the resulting closure
+    /// is `Sync` (required by `Arc<dyn Fn + Send + Sync>`). The lock is
+    /// held only for the duration of `clone()`, never across an await.
+    pub fn new<M: BlockProvider<Block = B>>(provider: M) -> Self {
+        let provider = Arc::new(Mutex::new(provider));
+        Self {
+            fetch: Arc::new(move |digest| {
+                let p = provider.lock().clone();
+                Box::pin(async move { p.fetch_block(digest).await })
+            }),
+        }
+    }
+}
+
+impl<B: Block> BlockProvider for ErasedBlockProvider<B> {
+    type Block = B;
+
+    fn fetch_block(
+        self,
+        digest: <B as Digestible>::Digest,
+    ) -> impl Future<Output = Option<B>> + Send {
+        (self.fetch)(digest)
+    }
+}
+
 /// 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,
@@ -47,12 +88,37 @@ pub struct AncestorStream<M, B: Block> {
 }
 
 impl<M, B: Block> AncestorStream<M, B> {
+    /// 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<&B> {
+        self.buffered.last()
+    }
+
+    /// Erase the block provider type, producing an
+    /// `AncestorStream<ErasedBlockProvider<B>, B>`.
+    ///
+    /// The returned stream behaves identically but can be sent across
+    /// channels that require a concrete type.
+    pub fn erase(self) -> AncestorStream<ErasedBlockProvider<B>, B>
+    where
+        M: BlockProvider<Block = B>,
+    {
+        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 = B>) -> Self {
+    pub fn new(marshal: M, initial: impl IntoIterator<Item = B>) -> Self {
         let mut buffered = initial.into_iter().collect::<Vec<B>>();
         buffered.sort_by_key(Heightable::height);
 

consensus/src/marshal/coding/mod.rs

@@ -197,6 +197,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

@@ -40,6 +40,7 @@ use commonware_utils::{
     channel::{fallible::OneshotExt, mpsc, oneshot},
     futures::{AbortablePool, Aborter, OptionFuture},
     sequence::U64,
+    vec::NonEmptyVec,
     Acknowledgement, BoxedError,
 };
 use futures::{future::join_all, try_join, FutureExt};
@@ -659,20 +660,41 @@ where
                         let finalization = self.get_finalization_by_height(height).await;
                         response.send_lossy(finalization);
                     }
+                    Message::GetProcessedHeight { response } => {
+                        response.send_lossy(self.last_processed_height);
+                    }
                     Message::HintFinalized { height, targets } => {
-                        // Skip if height is at or below the floor
-                        if height <= self.last_processed_height {
+                        // Skip if height is below the floor
+                        if height < self.last_processed_height {
                             continue;
                         }
 
-                        // Skip if finalization is already available locally
-                        if self.get_finalization_by_height(height).await.is_some() {
+                        // Skip if both the finalization and the block are
+                        // already available locally. Both are checked because
+                        // `set_floor` can advance `last_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;
                         }
 
-                        // Trigger a targeted fetch via the resolver
+                        // Trigger a fetch via the resolver
                         let request = Request::<V::Commitment>::Finalized { height };
-                        resolver.fetch_targeted(request, targets).await;
+                        match targets {
+                            Recipients::All => resolver.fetch(request).await,
+                            Recipients::Some(peers) => {
+                                if peers.is_empty() {
+                                    continue;
+                                }
+                                let peers = NonEmptyVec::from_unchecked(peers);
+                                resolver.fetch_targeted(request, peers).await;
+                            },
+                            Recipients::One(peer) => {
+                                let peers = NonEmptyVec::new(peer);
+                                resolver.fetch_targeted(request, peers).await;
+                            }
+                        }
                     }
                     Message::SubscribeByDigest {
                         round,
@@ -704,7 +726,10 @@ where
                         )
                         .await;
                     }
-                    Message::SetFloor { height } => {
+                    Message::SetFloor {
+                        height,
+                        prune_archives,
+                    } => {
                         if self.last_processed_height >= height {
                             warn!(
                                 %height,
@@ -726,10 +751,12 @@ where
                         // updating `last_processed_height`.
                         self.pending_acks.clear();
 
-                        // Prune data in the finalized archives below the new floor.
-                        if let Err(err) = self.prune_finalized_archives(height).await {
-                            error!(?err, %height, "failed to prune finalized archives");
-                            return;
+                        if prune_archives {
+                            // Prune data in the finalized archives below the new floor.
+                            if let Err(err) = self.prune_finalized_archives(height).await {
+                                error!(?err, %height, "failed to prune finalized archives");
+                                return;
+                            }
                         }
 
                         // Intentionally keep existing block subscriptions alive. Canceling
@@ -1455,15 +1482,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.last_processed_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.last_processed_height {
             debug!(
                 %height,
                 floor = %self.last_processed_height,
                 ?digest,
-                "dropping finalization at or below processed height floor"
+                "dropping finalization below processed height floor"
             );
             return false;
         }

consensus/src/marshal/core/mailbox.rs

@@ -10,10 +10,7 @@ use crate::{
 };
 use commonware_cryptography::{certificate::Scheme, Digestible};
 use commonware_p2p::Recipients;
-use commonware_utils::{
-    channel::{fallible::AsyncFallibleExt, mpsc, oneshot},
-    vec::NonEmptyVec,
-};
+use commonware_utils::channel::{fallible::AsyncFallibleExt, mpsc, oneshot};
 
 /// Messages sent to the marshal [Actor](super::Actor).
 ///
@@ -46,6 +43,11 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
         /// A channel to send the retrieved finalization.
         response: oneshot::Sender<Option<Finalization<S, V::Commitment>>>,
     },
+    /// A request to retrieve the latest processed height acknowledged by the application.
+    GetProcessedHeight {
+        /// A channel to send the latest processed height.
+        response: oneshot::Sender<Height>,
+    },
     /// A hint that a finalized block may be available at a given height.
     ///
     /// This triggers a network fetch if the finalization is not available locally.
@@ -63,8 +65,9 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
     HintFinalized {
         /// The height of the finalization to fetch.
         height: Height,
-        /// Target peers to fetch from. Added to any existing targets for this height.
-        targets: NonEmptyVec<S::PublicKey>,
+        /// Target peers to fetch from. Added to any existing targets for this
+        /// height.
+        targets: Recipients<S::PublicKey>,
     },
     /// A request to subscribe to a block by its digest.
     SubscribeByDigest {
@@ -132,7 +135,7 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
     /// Sets the sync starting point (advances if higher than current).
     ///
     /// Marshal will sync and deliver blocks starting at `floor + 1`. Data below
-    /// the floor is pruned.
+    /// the floor is pruned when `prune_archives` is `true`.
     ///
     /// To prune data without affecting the sync starting point (say at some trailing depth
     /// from tip), use [Message::Prune] instead.
@@ -141,6 +144,9 @@ pub(crate) enum Message<S: Scheme, V: Variant> {
     SetFloor {
         /// The candidate floor height.
         height: Height,
+
+        /// Whether to prune finalized archives below the new floor.
+        prune_archives: bool,
     },
     /// Prunes finalized blocks and certificates below the given height.
     ///
@@ -215,16 +221,18 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
             .flatten()
     }
 
+    /// Retrieve the latest processed height acknowledged by the application.
+    pub async fn get_processed_height(&self) -> Option<Height> {
+        self.sender
+            .request(|response| Message::GetProcessedHeight { response })
+            .await
+    }
+
     /// Hints that a finalized block may be available at the given height.
     ///
     /// This method will request the finalization from the network via the resolver
     /// if it is not available locally.
     ///
-    /// Targets are required because this is typically called when a peer claims to be
-    /// ahead. By targeting only those peers, we limit who we ask. If a target returns
-    /// invalid data, they will be blocked by the resolver. If targets don't respond
-    /// or return "no data", they effectively rate-limit themselves.
-    ///
     /// Calling this multiple times for the same height with different targets will
     /// add to the target set if there is an ongoing fetch, allowing more peers to be tried.
     ///
@@ -234,7 +242,7 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
     /// The height must be covered by both the epocher and the provider. If the
     /// epocher cannot map the height to an epoch, or the provider cannot supply
     /// a scheme for that epoch, the hint is silently dropped.
-    pub async fn hint_finalized(&self, height: Height, targets: NonEmptyVec<S::PublicKey>) {
+    pub async fn hint_finalized(&self, height: Height, targets: Recipients<S::PublicKey>) {
         self.sender
             .send_lossy(Message::HintFinalized { height, targets })
             .await;
@@ -346,14 +354,19 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
     /// Sets the sync starting point (advances if higher than current).
     ///
     /// Marshal will sync and deliver blocks starting at `floor + 1`. Data below
-    /// the floor is pruned.
+    /// the floor is pruned when `prune_archives` is `true`.
     ///
     /// To prune data without affecting the sync starting point (say at some trailing depth
     /// from tip), use [Self::prune] instead.
     ///
     /// The default floor is 0.
-    pub async fn set_floor(&self, height: Height) {
-        self.sender.send_lossy(Message::SetFloor { height }).await;
+    pub async fn set_floor(&self, height: Height, prune_archives: bool) {
+        self.sender
+            .send_lossy(Message::SetFloor {
+                height,
+                prune_archives,
+            })
+            .await;
     }
 
     /// Prunes finalized blocks and certificates below the given height.