[consensus/marshal] commonware-glue marshal changes

Author: clabby

consensus/src/lib.rs

@@ -233,8 +233,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

@@ -86,12 +86,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.
-    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

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

@@ -353,9 +353,10 @@ where
 
         // A configured floor follows the same path as `SetFloor`: verify it,
         // then apply a local anchor or fetch the anchor block.
-        if let Some(finalization) = self.floor.take_pending_anchor() {
+        if let Some((finalization, prune_archives)) = self.floor.take_pending_anchor() {
             self.install_floor(
                 finalization,
+                prune_archives,
                 false,
                 &mut resolver,
                 &mut buffer,
@@ -692,6 +693,9 @@ where
                 let finalization = self.get_finalization_by_height(height).await;
                 response.send_lossy(finalization);
             }
+            Message::GetProcessedHeight { response } => {
+                response.send_lossy(self.floor.processed_height());
+            }
             Message::HintFinalized { height, targets } => {
                 // Skip if finalization is already available locally.
                 if self.get_finalization_by_height(height).await.is_some() {
@@ -743,9 +747,19 @@ where
                         .ignore();
                 }
             }
-            Message::SetFloor { finalization } => {
-                self.install_floor(finalization, true, resolver, buffer, application)
-                    .await;
+            Message::SetFloor {
+                finalization,
+                prune_archives,
+            } => {
+                self.install_floor(
+                    finalization,
+                    prune_archives,
+                    true,
+                    resolver,
+                    buffer,
+                    application,
+                )
+                .await;
             }
             Message::Prune { height } => {
                 // Only allow pruning at or below the current floor.
@@ -991,6 +1005,7 @@ where
     async fn install_floor<Buf, R>(
         &mut self,
         finalization: Finalization<P::Scheme, V::Commitment>,
+        prune_archives: bool,
         skip_if_superseded: bool,
         resolver: &mut R,
         buffer: &mut OptionalBuffer<V, Buf>,
@@ -1034,7 +1049,7 @@ where
         }
 
         if let Some(block) = self.find_block_by_commitment(buffer, commitment).await {
-            self.floor.await_anchor(finalization);
+            self.floor.await_anchor(finalization, prune_archives);
             assert!(
                 self.apply_floor_anchor(&block, buffer, application, resolver)
                     .await
@@ -1047,7 +1062,7 @@ where
         self.pending_acks.clear();
 
         debug!(?round, ?commitment, "starting fetch for floor block");
-        self.floor.await_anchor(finalization);
+        self.floor.await_anchor(finalization, prune_archives);
         self.floor
             .fetch_if_permitted(
                 resolver,
@@ -1094,7 +1109,7 @@ where
                 existing = %self.floor.processed_height(),
                 "floor not updated, at or below existing"
             );
-            let finalization = self
+            let (finalization, _prune_archives) = self
                 .floor
                 .take_pending_anchor()
                 .expect("pending floor anchor missing");
@@ -1108,7 +1123,7 @@ where
         }
 
         let digest = block.digest();
-        let finalization = self
+        let (finalization, prune_archives) = self
             .floor
             .take_pending_anchor()
             .expect("pending floor anchor missing");
@@ -1152,10 +1167,11 @@ where
         // acks for blocks below the new floor cannot rewrite the processed floor.
         self.pending_acks.clear();
 
-        // The floor is durable, so cache/finalized data below it can be pruned.
-        self.prune_after_floor(height)
-            .await
-            .expect("failed to prune data below floor");
+        if prune_archives {
+            self.prune_after_floor(height)
+                .await
+                .expect("failed to prune data below floor");
+        }
 
         // Intentionally keep existing block subscriptions alive. Canceling
         // waiters can have catastrophic consequences (nodes can get stuck in
@@ -1726,9 +1742,9 @@ 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)
+        // Blocks at or 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.processed_height() {
             debug!(
                 %height,

consensus/src/marshal/core/floor.rs

@@ -42,7 +42,7 @@ impl FetchAdmission {
 /// The processed floor plus any pending floor update awaiting its anchor block.
 pub(super) struct Floor<S: CertificateScheme, C: Digest> {
     processed: ProcessedFloor,
-    pending: Option<Finalization<S, C>>,
+    pending: Option<(Finalization<S, C>, bool)>,
 }
 
 impl<S: CertificateScheme, C: Digest> Floor<S, C> {
@@ -60,7 +60,7 @@ impl<S: CertificateScheme, C: Digest> Floor<S, C> {
     ) -> Self {
         Self {
             processed: ProcessedFloor { height, round },
-            pending: Some(finalization),
+            pending: Some((finalization, true)),
         }
     }
 
@@ -87,22 +87,22 @@ impl<S: CertificateScheme, C: Digest> Floor<S, C> {
 
     /// Returns true if a pending floor already supersedes the candidate floor round.
     pub(super) fn has_pending_anchor_at_or_after(&self, round: Round) -> bool {
-        matches!(&self.pending, Some(pending) if pending.round() >= round)
+        matches!(&self.pending, Some((pending, _)) if pending.round() >= round)
     }
 
     /// Returns true when `commitment` is the awaited anchor.
     pub(super) fn matches_pending_anchor(&self, commitment: C) -> bool {
-        matches!(&self.pending, Some(pending) if pending.proposal.payload == commitment)
+        matches!(&self.pending, Some((pending, _)) if pending.proposal.payload == commitment)
     }
 
     /// Records a verified floor finalization whose block anchor still needs to arrive.
-    pub(super) fn await_anchor(&mut self, finalization: Finalization<S, C>) {
-        self.pending = Some(finalization);
+    pub(super) fn await_anchor(&mut self, finalization: Finalization<S, C>, prune_archives: bool) {
+        self.pending = Some((finalization, prune_archives));
     }
 
     /// Takes the pending anchor finalization, if any.
     #[must_use]
-    pub(super) const fn take_pending_anchor(&mut self) -> Option<Finalization<S, C>> {
+    pub(super) const fn take_pending_anchor(&mut self) -> Option<(Finalization<S, C>, bool)> {
         self.pending.take()
     }