more

Author: patrick-ogrady

consensus/src/marshal/core/actor.rs

@@ -18,7 +18,7 @@ use crate::{
         scheme::Scheme,
         types::{verify_certificates, Finalization, Notarization, Subject},
     },
-    types::{Epoch, Epocher, Height, Round, ViewDelta},
+    types::{Epoch, Epocher, Height, Round, View, ViewDelta},
     Block, Epochable, Heightable, Reporter,
 };
 use bytes::Bytes;
@@ -54,7 +54,11 @@ use std::{collections::BTreeMap, future::Future, num::NonZeroUsize, sync::Arc};
 use tracing::{debug, warn};
 
 /// The key used to store the last application-acknowledged height.
-const LATEST_KEY: U64 = U64::new(0xFF);
+const LATEST_HEIGHT_KEY: U64 = U64::new(0xFF);
+/// The key used to store the epoch component of the processed round floor.
+const LATEST_ROUND_EPOCH_KEY: U64 = U64::new(0xFE);
+/// The key used to store the view component of the processed round floor.
+const LATEST_ROUND_VIEW_KEY: U64 = U64::new(0xFD);
 
 // Resolver request keys are expressed in the variant commitment type, which
 // may differ from the block digest for coded variants.
@@ -195,7 +199,7 @@ where
         )
         .await
         .expect("failed to initialize application metadata");
-        let last_applied_height = application_metadata.get(&LATEST_KEY).copied();
+        let last_applied_height = application_metadata.get(&LATEST_HEIGHT_KEY).copied();
         let processed_floor_height = last_applied_height.unwrap_or(Height::zero());
 
         // Genesis is seeded as a local anchor. If the application has not
@@ -215,8 +219,11 @@ where
             }
             Start::Floor(finalization) => Some(finalization),
         };
-        let last_processed_round =
+        let height_derived_processed_round =
             Self::latest_processed_round(&finalizations_by_height, processed_floor_height).await;
+        let last_processed_round = Self::metadata_round_floor(&application_metadata)
+            .unwrap_or(Round::zero())
+            .max(height_derived_processed_round);
 
         // Create metrics
         let finalized_height = context.gauge("finalized_height", "Finalized height of application");
@@ -485,7 +492,7 @@ where
             }
         };
 
-        // Persist buffered processed-height updates once after draining all ready acks.
+        // Persist buffered progress updates once after draining all ready acks.
         self.application_metadata
             .sync()
             .await
@@ -2042,7 +2049,7 @@ where
         >,
     ) {
         self.last_applied_height = Some(height);
-        self.application_metadata.put(LATEST_KEY, height);
+        self.application_metadata.put(LATEST_HEIGHT_KEY, height);
         self.floor.set_processed_height(height);
         let _ = self
             .processed_height
@@ -2052,6 +2059,13 @@ where
         resolver.retain(handler::above_height_floor::<V::Commitment>(height));
     }
 
+    /// Returns the durably stored processed round floor, if present.
+    fn metadata_round_floor(metadata: &Metadata<E, U64, Height>) -> Option<Round> {
+        let epoch = metadata.get(&LATEST_ROUND_EPOCH_KEY)?;
+        let view = metadata.get(&LATEST_ROUND_VIEW_KEY)?;
+        Some(Round::new(Epoch::new(epoch.get()), View::new(view.get())))
+    }
+
     /// Returns the latest known finalization round at or below the processed height.
     async fn latest_processed_round(finalizations_by_height: &FC, height: Height) -> Round {
         let Some(finalization_height) = finalizations_by_height
@@ -2106,6 +2120,10 @@ where
 
         let previous = self.floor.processed_round();
         self.floor.set_processed_round(round);
+        self.application_metadata
+            .put(LATEST_ROUND_EPOCH_KEY, Height::new(round.epoch().get()));
+        self.application_metadata
+            .put(LATEST_ROUND_VIEW_KEY, Height::new(round.view().get()));
 
         // Retain view-indexed cache data for a window behind the previously
         // processed finalized block.

consensus/src/marshal/standard/mod.rs

@@ -4851,6 +4851,101 @@ mod tests {
         });
     }
 
+    #[test_traced("WARN")]
+    fn test_standard_floor_round_restored_before_anchor_ack() {
+        let runner = deterministic::Runner::timed(Duration::from_secs(30));
+        runner.start(|mut context| async move {
+            let Fixture { schemes, .. } =
+                bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
+            let partition_prefix = "floor-round-before-anchor-ack";
+
+            let floor_round = Round::new(Epoch::zero(), View::new(5));
+            let floor_block = make_raw_block(Sha256::hash(b"floor-parent"), Height::new(5), 500);
+            let floor_finalization = StandardHarness::make_finalization(
+                Proposal::new(
+                    floor_round,
+                    View::new(4),
+                    StandardHarness::commitment(&floor_block),
+                ),
+                &schemes,
+                QUORUM,
+            );
+
+            let (application, started) = HoldingBlockReporter::new_after(Height::zero());
+            let (mailbox, buffer, resolver, actor_handle) = start_standard_actor(
+                context.child("validator").with_attribute("index", 0),
+                partition_prefix,
+                ConstantProvider::new(schemes[0].clone()),
+                application,
+                Some(RecordingBuffer::default()),
+                Start::Genesis(StandardHarness::genesis_block(NUM_VALIDATORS as u16)),
+            )
+            .await;
+
+            assert!(mailbox.verified(floor_round, floor_block.clone()).await);
+            mailbox.set_floor(floor_finalization);
+            select! {
+                height = started => {
+                    assert_eq!(
+                        height.expect("floor anchor delivery signal missing"),
+                        Height::new(5),
+                        "floor anchor should be delivered before the simulated crash"
+                    );
+                },
+                _ = context.sleep(Duration::from_secs(5)) => {
+                    panic!("floor anchor was not delivered before restart");
+                },
+            }
+
+            actor_handle.abort();
+            drop(mailbox);
+            drop(buffer);
+            drop(resolver);
+            context.sleep(Duration::from_millis(1)).await;
+
+            let (mailbox, _buffer, resolver, _actor_handle) = start_standard_actor(
+                context
+                    .child("validator_restart")
+                    .with_attribute("index", 0),
+                partition_prefix,
+                ConstantProvider::new(schemes[0].clone()),
+                Application::<B>::manual_ack(),
+                Some(RecordingBuffer::default()),
+                Start::Genesis(StandardHarness::genesis_block(NUM_VALIDATORS as u16)),
+            )
+            .await;
+
+            let fetches_before = resolver.fetches().len();
+            let subscription = mailbox.subscribe_by_commitment(
+                Sha256::hash(b"missing-at-restored-floor-round"),
+                CommitmentFallback::FetchByRound { round: floor_round },
+            );
+            let barrier = make_raw_block(floor_block.digest(), Height::new(6), 600);
+            assert!(
+                mailbox
+                    .verified(Round::new(Epoch::zero(), View::new(6)), barrier)
+                    .await,
+                "barrier verification should be processed"
+            );
+            assert_eq!(
+                resolver.fetches().len(),
+                fetches_before,
+                "restart must restore the floor round before the anchor is acknowledged"
+            );
+            select! {
+                result = subscription => {
+                    assert!(
+                        result.is_err(),
+                        "floor-round subscription should be canceled after restart"
+                    );
+                },
+                _ = context.sleep(Duration::from_secs(5)) => {
+                    panic!("floor-round subscription remained open after restart");
+                },
+            }
+        });
+    }
+
     #[test_traced("WARN")]
     fn test_standard_set_floor_prunes_round_bound_fetches() {
         let runner = deterministic::Runner::timed(Duration::from_secs(30));