[consensus/marshal] commonware-glue marshal changes

Author: clabby

consensus/src/lib.rs

@@ -236,8 +236,8 @@ stability_scope!(ALPHA {
     pub mod ordered_broadcast;
 });
 stability_scope!(ALPHA, cfg(not(target_arch = "wasm32")) {
-    use commonware_cryptography::certificate::Scheme;
     use crate::marshal::ancestry::Ancestry;
+    use commonware_cryptography::certificate::Scheme;
     use commonware_runtime::{Clock, Metrics, Spawner};
     use rand::Rng;
 

consensus/src/marshal/ancestry.rs

@@ -59,12 +59,21 @@ 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()
+    }
+
     /// 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

@@ -796,6 +796,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;
         }