Address Feedback

consensus/src/marshal/coding/mod.rs

@@ -259,6 +259,112 @@ mod tests {
         harness::finalize_same_height_different_views::<CodingHarness>();
     }
 
+    /// Regression test for issue #167: finalizing a descendant must not
+    /// height-prune the shard-engine buffer before `try_repair_gaps` has
+    /// consumed buffer-only ancestors.
+    ///
+    /// Places parent (height 1) and descendant (height 2) in the shard engine's
+    /// reconstructed-block cache via `proposed()`, then reports a finalization
+    /// for the descendant only. With the fix, the gap repair pass pulls the
+    /// parent from the buffer and archives it before any height-prune runs. The
+    /// pre-fix behavior would prune the parent from the buffer inside
+    /// `store_finalization(descendant)` and leave the finalized chain waiting
+    /// on a network fetch that never arrives.
+    #[test_traced("WARN")]
+    fn test_coding_store_finalization_does_not_prune_buffer_before_repair() {
+        let runner = deterministic::Runner::timed(Duration::from_secs(60));
+        runner.start(|mut context| async move {
+            let Fixture {
+                participants,
+                schemes,
+                ..
+            } = bls12381_threshold_vrf::fixture::<V, _>(
+                &mut context,
+                NAMESPACE,
+                NUM_VALIDATORS,
+            );
+            let mut oracle = setup_network_with_participants(
+                context.clone(),
+                NZUsize!(1),
+                participants.clone(),
+            )
+            .await;
+
+            let setup = CodingHarness::setup_validator(
+                context.with_label("validator_0"),
+                &mut oracle,
+                participants[0].clone(),
+                ConstantProvider::new(schemes[0].clone()),
+            )
+            .await;
+            let mut handle = harness::ValidatorHandle::<CodingHarness> {
+                mailbox: setup.mailbox,
+                extra: setup.extra,
+            };
+
+            // Build a 2-block chain: parent at height 1, descendant at height 2.
+            let parent_block = CodingHarness::make_test_block(
+                Sha256::hash(b""),
+                CodingHarness::genesis_parent_commitment(NUM_VALIDATORS as u16),
+                Height::new(1),
+                1,
+                NUM_VALIDATORS as u16,
+            );
+            let parent_digest = CodingHarness::digest(&parent_block);
+            let parent_commitment = CodingHarness::commitment(&parent_block);
+
+            let descendant_block = CodingHarness::make_test_block(
+                parent_digest,
+                parent_commitment,
+                Height::new(2),
+                2,
+                NUM_VALIDATORS as u16,
+            );
+            let descendant_commitment = CodingHarness::commitment(&descendant_block);
+
+            // Seed the shard engine's reconstructed-block cache with both blocks.
+            CodingHarness::propose(
+                &mut handle,
+                Round::new(Epoch::new(0), View::new(1)),
+                &parent_block,
+            )
+            .await;
+            CodingHarness::propose(
+                &mut handle,
+                Round::new(Epoch::new(0), View::new(2)),
+                &descendant_block,
+            )
+            .await;
+
+            // Report finalization for the descendant only. The parent has no
+            // finalization certificate: it must be archived by walking the
+            // parent link from the descendant and sourcing the block from the
+            // shard-engine buffer.
+            let descendant_proposal = Proposal {
+                round: Round::new(Epoch::new(0), View::new(2)),
+                parent: View::new(1),
+                payload: descendant_commitment,
+            };
+            let descendant_finalization =
+                CodingHarness::make_finalization(descendant_proposal, &schemes, QUORUM);
+            CodingHarness::report_finalization(&mut handle.mailbox, descendant_finalization)
+                .await;
+
+            context.sleep(Duration::from_millis(200)).await;
+
+            let parent = handle.mailbox.get_block(Height::new(1)).await;
+            assert!(
+                parent.is_some(),
+                "parent must be archived from shard buffer before height-prune evicts it"
+            );
+            let descendant = handle.mailbox.get_block(Height::new(2)).await;
+            assert!(
+                descendant.is_some(),
+                "descendant must be archived after finalization"
+            );
+        });
+    }
+
     #[test_traced("WARN")]
     fn test_coding_init_processed_height() {
         harness::init_processed_height::<CodingHarness>();

consensus/src/marshal/core/actor.rs

@@ -243,6 +243,11 @@ where
     pending_acks: PendingAcks<V, A>,
     // Highest known finalized height
     tip: Height,
+    // Commitment of the highest finalized block awaiting buffer cleanup.
+    // Deferred from `store_finalization` so that `try_repair_gaps` can consume
+    // buffer-only ancestors before variant cleanup (e.g. the coding shard
+    // engine's height-based prune) evicts them.
+    pending_buffer_finalize: Option<V::Commitment>,
     // Outstanding subscriptions for blocks
     block_subscriptions: BTreeMap<BlockSubscriptionKeyFor<V>, BlockSubscription<V>>,
 
@@ -347,6 +352,7 @@ where
                 last_processed_height,
                 pending_acks: PendingAcks::new(config.max_pending_acks.get()),
                 tip: Height::zero(),
+                pending_buffer_finalize: None,
                 block_subscriptions: BTreeMap::new(),
                 cache,
                 application_metadata,
@@ -412,6 +418,7 @@ where
         {
             self.sync_finalized().await;
         }
+        self.flush_buffer_finalize(&mut buffer).await;
 
         // Attempt to dispatch the next finalized block to the application, if it is ready.
         self.try_dispatch_blocks(&mut application).await;
@@ -588,13 +595,13 @@ where
                                     block,
                                     Some(finalization),
                                     &mut application,
-                                    &mut buffer,
                                 )
                                 .await
                             {
                                 self.try_repair_gaps(&mut buffer, &mut resolver, &mut application)
                                     .await;
                                 self.sync_finalized().await;
+                                self.flush_buffer_finalize(&mut buffer).await;
                                 self.try_dispatch_blocks(&mut application).await;
                                 debug!(?round, %height, "finalized block stored");
                             }
@@ -758,7 +765,6 @@ where
                                     response,
                                     &mut delivers,
                                     &mut application,
-                                    &mut buffer,
                                 )
                                 .await;
                         }
@@ -767,7 +773,7 @@ where
 
                 // Batch verify and process all delivers.
                 needs_sync |= self
-                    .verify_delivered(delivers, &mut application, &mut buffer)
+                    .verify_delivered(delivers, &mut application)
                     .await;
 
                 // Attempt to fill gaps before handling produce requests (so we
@@ -780,6 +786,7 @@ where
                 // durability).
                 if needs_sync {
                     self.sync_finalized().await;
+                    self.flush_buffer_finalize(&mut buffer).await;
                     self.try_dispatch_blocks(&mut application).await;
                 }
 
@@ -923,14 +930,13 @@ where
     /// immediately. Finalized/Notarized delivers are parsed and structurally
     /// validated, then collected into `delivers` for batch certificate verification.
     /// Returns true if finalization archives were written and need syncing.
-    async fn handle_deliver<Buf: Buffer<V>>(
+    async fn handle_deliver(
         &mut self,
         key: Request<V::Commitment>,
         value: Bytes,
         response: oneshot::Sender<bool>,
         delivers: &mut Vec<PendingVerification<P::Scheme, V>>,
         application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
-        buffer: &mut Buf,
     ) -> bool {
         match key {
             Request::Block(commitment) => {
@@ -949,7 +955,7 @@ where
                 let digest = block.digest();
                 let finalization = self.cache.get_finalization_for(digest).await;
                 let wrote = self
-                    .store_finalization(height, digest, block, finalization, application, buffer)
+                    .store_finalization(height, digest, block, finalization, application)
                     .await;
                 debug!(?digest, %height, "received block");
                 response.send_lossy(true); // if a valid block is received, we should still send true (even if it was stale)
@@ -1045,11 +1051,10 @@ where
 
     /// Batch verify pending certificates and process valid items. Returns true
     /// if finalization archives were written and need syncing.
-    async fn verify_delivered<Buf: Buffer<V>>(
+    async fn verify_delivered(
         &mut self,
         mut delivers: Vec<PendingVerification<P::Scheme, V>>,
         application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
-        buffer: &mut Buf,
     ) -> bool {
         if delivers.is_empty() {
             return false;
@@ -1138,7 +1143,6 @@ where
                             block,
                             Some(finalization),
                             application,
-                            buffer,
                         )
                         .await;
                 }
@@ -1169,7 +1173,6 @@ where
                                 block.clone(),
                                 Some(finalization),
                                 application,
-                                buffer,
                             )
                             .await;
                     }
@@ -1414,14 +1417,13 @@ where
     /// `select_loop!` so that archive data is durable before the ack handler
     /// advances `last_processed_height`. See [`Self::try_dispatch_blocks`] for the
     /// crash safety invariant.
-    async fn store_finalization<Buf: Buffer<V>>(
+    async fn store_finalization(
         &mut self,
         height: Height,
         digest: <V::Block as Digestible>::Digest,
         block: V::Block,
         finalization: Option<Finalization<P::Scheme, V::Commitment>>,
         application: &mut impl Reporter<Activity = Update<V::ApplicationBlock, A>>,
-        buffer: &mut Buf,
     ) -> 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
@@ -1463,17 +1465,27 @@ where
             panic!("failed to finalize: {e}");
         }
 
-        // Update metrics, buffer, and application
+        // Update metrics and application
         if let Some(round) = round.filter(|_| height > self.tip) {
             application.report(Update::Tip(round, height, digest)).await;
             self.tip = height;
             let _ = self.finalized_height.try_set(height.get());
+            self.pending_buffer_finalize = Some(commitment);
         }
-        buffer.finalized(commitment).await;
 
         true
     }
 
+    /// Flush any deferred `buffer.finalized` notification accumulated by
+    /// `store_finalization`. Must be invoked after `try_repair_gaps` and
+    /// `sync_finalized` so that variant cleanup (e.g. coding height-prune)
+    /// only runs once buffer-only ancestors have been archived.
+    async fn flush_buffer_finalize<Buf: Buffer<V>>(&mut self, buffer: &mut Buf) {
+        if let Some(commitment) = self.pending_buffer_finalize.take() {
+            buffer.finalized(commitment).await;
+        }
+    }
+
     /// Get the latest finalized block information (height and digest tuple).
     ///
     /// Blocks are only finalized directly with a finalization or indirectly via a descendant
@@ -1592,7 +1604,6 @@ where
                             block,
                             Some(finalization),
                             application,
-                            buffer,
                         )
                         .await;
                 } else {
@@ -1638,7 +1649,6 @@ where
                             block.clone(),
                             finalization,
                             application,
-                            buffer,
                         )
                         .await;
                     debug!(height = %block.height(), "repaired block");

consensus/src/marshal/standard/inline.rs

@@ -224,8 +224,11 @@ where
     /// Proposes a new block or re-proposes an epoch boundary block.
     ///
     /// Proposal runs in a spawned task and returns a receiver for the resulting digest.
-    /// Blocks are persisted and broadcast via `marshal.proposed()` before the digest
-    /// is returned to consensus.
+    /// The built block is cached in memory (`last_built`) for the subsequent
+    /// `Relay::broadcast(Plan::Propose)` call, which invokes `marshal.proposed()`
+    /// as the durable persistence boundary before consensus continues. Receiving
+    /// a digest from `propose()` alone does not mean the block is recoverable
+    /// after restart.
     async fn propose(
         &mut self,
         consensus_context: Context<Self::Digest, S::PublicKey>,

consensus/src/marshal/standard/mod.rs

@@ -201,6 +201,7 @@ mod tests {
         harness::verified_success_implies_recoverable_after_restart::<DeferredHarness>(0..16);
     }
 
+
     #[test_traced("WARN")]
     fn test_standard_delivery_visibility_implies_recoverable_after_restart() {
         harness::delivery_visibility_implies_recoverable_after_restart::<InlineHarness>(0..16);