use super::StateSyncMetadata; use commonware_consensus::{ marshal::{core::Variant, Start}, simplex::types::Finalization, types::Height, }; use commonware_cryptography::certificate::Scheme; use commonware_storage::Context; use tracing::warn; /// Startup plan that determines whether one-time peer state sync may still run. /// /// Construction is two-phase so the caller can avoid fetching a finalized /// floor from peers when state sync has already completed: /// /// 1. [`SyncPlan::init`] reads the durable state sync state. /// 2. If [`SyncPlan::may_state_sync`] returns `true`, the caller may fetch a /// finalized floor and attach it via [`SyncPlan::with_floor`]. An interrupted /// sync already has a persisted floor, while a fresh sync needs one from the /// caller. Otherwise the caller skips floor selection entirely. /// /// The plan owns the opened metadata store and is later consumed by /// [`Stateful`](crate::stateful::Stateful), so startup does not reopen the same /// metadata partition from multiple places. /// /// Once state sync completes, this node never performs peer state sync /// again. Future startups must recover from the later of that synced height /// and marshal's processed height instead. pub struct SyncPlan where E: Context, S: Scheme, V: Variant, { sync_metadata: StateSyncMetadata, floor: Option>, } impl SyncPlan where E: Context, S: Scheme, V: Variant, { /// Load the durable state sync metadata for this partition prefix. /// /// # Panics /// /// Panics if the metadata store cannot be opened. A node that cannot /// determine whether state sync already completed cannot safely choose a /// startup path. pub async fn init(context: &E, partition_prefix: impl AsRef) -> Self { let sync_metadata = StateSyncMetadata::::init(context, partition_prefix).await; let floor = sync_metadata.in_progress_floor().cloned(); Self { sync_metadata, floor, } } /// Returns whether state sync can still run on this node. /// /// When `false`, the caller should skip floor selection: any floor passed /// to [`SyncPlan::with_floor`] would be ignored. The node already has a /// durable completed state sync height, so future boots must recover from that /// height or marshal's processed height instead of running peer state sync again. /// /// When `true`, the caller can optionally attach a finalized floor via /// [`SyncPlan::with_floor`]. If a floor is not attached, the node will /// attempt to sync from genesis via marshal unless it is resuming an /// interrupted state sync. pub fn may_state_sync(&self) -> bool { self.sync_metadata.sync_height().is_none() } /// Returns the durable completed state sync height, if one has been stored. pub fn sync_height(&self) -> Option { self.sync_metadata.sync_height() } /// Returns the partition prefix to use for state sync metadata storage. pub const fn partition_prefix(&self) -> &str { self.sync_metadata.partition_prefix() } /// Returns the selected or persisted in-progress state sync floor. pub const fn floor(&self) -> Option<&Finalization> { self.floor.as_ref() } /// Attach a finalized floor to state sync from. /// /// Has no effect if state sync has already completed. When resuming an /// interrupted sync, a lagging selection is ignored in favor of the /// persisted floor. #[must_use] pub fn with_floor(mut self, floor: Finalization) -> Self { if !self.may_state_sync() { return self; } if let Some(selected) = &self.floor { if floor.round() <= selected.round() { warn!( candidate = ?floor.round(), selected = ?selected.round(), "state sync floor not updated, candidate is not newer", ); return self; } } self.floor = Some(floor); self } /// Returns marshal's startup anchor for this plan. /// /// If a finalized floor was attached or persisted by an interrupted sync, /// marshal starts from that floor. Otherwise marshal starts from genesis /// and relies on its own durable progress to override that anchor when /// available. pub fn marshal_start(&self, genesis: B) -> Start { self.floor .as_ref() .cloned() .map_or_else(|| Start::Genesis(genesis), Start::Floor) } /// Returns whether startup must resume an interrupted state sync. /// /// This is `true` after a previous process crashed while state sync was /// in progress. In that case [`Self::may_state_sync`] is also `true`, and /// the persisted floor keeps partially synced database state on the same /// recovery path. pub fn requires_state_sync_floor(&self) -> bool { self.sync_metadata.in_progress() } /// Returns whether this startup should run peer state sync. /// /// A caller can request peer state sync for a fresh node. An interrupted /// state sync always requires peer state sync, even if the caller did not /// explicitly request it. pub fn should_state_sync(&self, requested: bool) -> bool { self.may_state_sync() && (requested || self.requires_state_sync_floor()) } /// Consumes this plan and returns its durable state-sync metadata handle. pub(crate) fn into_sync_metadata(self) -> StateSyncMetadata { self.sync_metadata } } #[cfg(test)] mod tests { use super::SyncPlan; use crate::stateful::{ actor::syncer::StateSyncMetadata, tests::mocks::{TestScheme, TestVariant}, }; use commonware_consensus::{ marshal::Start, simplex::{ mocks::scheme as scheme_mocks, types::{Finalization, Finalize, Proposal}, }, types::{Epoch, Height, Round, View}, }; use commonware_cryptography::sha256::{Digest as Sha256Digest, Sha256}; use commonware_parallel::Sequential; use commonware_runtime::{deterministic, Runner as _}; fn finalization( schemes: &[TestScheme], view: u64, digest_byte: u8, ) -> Finalization { let proposal = Proposal { round: Round::new(Epoch::zero(), View::new(view)), parent: View::new(view.saturating_sub(1)), payload: Sha256::fill(digest_byte), }; let finalizes = schemes .iter() .map(|scheme| Finalize::sign(scheme, proposal.clone()).expect("sign finalize")) .collect::>(); Finalization::from_finalizes(&schemes[0], &finalizes, &Sequential) .expect("recover finalization") } #[test] fn stored_sync_height_disables_state_sync() { deterministic::Runner::default().start(|context| async move { let partition_prefix = "stored_sync_height"; let plan = SyncPlan::<_, TestScheme, TestVariant>::init(&context, partition_prefix).await; assert!(plan.may_state_sync()); assert!(plan.should_state_sync(true)); assert!(!plan.should_state_sync(false)); assert_eq!(plan.sync_height(), None); drop(plan); let mut metadata = StateSyncMetadata::<_, TestScheme, Sha256Digest>::init(&context, partition_prefix) .await; metadata.set_complete(Height::new(7)).await; drop(metadata); let plan = SyncPlan::<_, TestScheme, TestVariant>::init(&context, partition_prefix).await; assert!(!plan.may_state_sync()); assert!(!plan.should_state_sync(true)); assert_eq!(plan.sync_height(), Some(Height::new(7))); assert!(plan.floor().is_none()); }); } #[test] #[should_panic(expected = "completed state sync cannot be marked in-progress")] fn completed_sync_cannot_be_marked_in_progress() { deterministic::Runner::default().start(|mut context| async move { let partition_prefix = "completed_sync_cannot_be_marked_in_progress"; let fixture = scheme_mocks::fixture(&mut context, b"_COMMONWARE_GLUE_SYNC_PLAN", 1); let mut metadata = StateSyncMetadata::<_, TestScheme, Sha256Digest>::init(&context, partition_prefix) .await; metadata.set_complete(Height::new(7)).await; metadata .begin_sync(finalization(&fixture.schemes, 8, 8)) .await; }); } #[test] #[should_panic(expected = "completed state sync height cannot move backward")] fn complete_height_cannot_move_backward() { deterministic::Runner::default().start(|context| async move { let partition_prefix = "complete_height_cannot_move_backward"; let mut metadata = StateSyncMetadata::<_, TestScheme, Sha256Digest>::init(&context, partition_prefix) .await; metadata.set_complete(Height::new(7)).await; metadata.set_complete(Height::new(6)).await; }); } #[test] fn in_progress_sync_requires_compatible_floor() { deterministic::Runner::default().start(|mut context| async move { let partition_prefix = "in_progress_sync_requires_compatible_floor"; let fixture = scheme_mocks::fixture(&mut context, b"_COMMONWARE_GLUE_SYNC_PLAN", 1); let stored = finalization(&fixture.schemes, 7, 7); let mut metadata = StateSyncMetadata::<_, TestScheme, Sha256Digest>::init(&context, partition_prefix) .await; metadata.begin_sync(stored.clone()).await; drop(metadata); let mut plan = SyncPlan::<_, TestScheme, TestVariant>::init(&context, partition_prefix).await; assert!(plan.may_state_sync()); assert!(plan.requires_state_sync_floor()); assert!(plan.should_state_sync(false)); plan.sync_metadata.begin_sync(stored).await; plan.sync_metadata .begin_sync(finalization(&fixture.schemes, 9, 9)) .await; }); } #[test] fn interrupted_sync_reuses_persisted_floor_when_probe_lags() { deterministic::Runner::default().start(|mut context| async move { let partition_prefix = "interrupted_sync_reuses_persisted_floor"; let fixture = scheme_mocks::fixture(&mut context, b"_COMMONWARE_GLUE_SYNC_PLAN", 1); let stored = finalization(&fixture.schemes, 7, 7); let mut metadata = StateSyncMetadata::<_, TestScheme, Sha256Digest>::init(&context, partition_prefix) .await; metadata.begin_sync(stored.clone()).await; drop(metadata); let plan = SyncPlan::<_, TestScheme, TestVariant>::init(&context, partition_prefix).await; assert!(plan.should_state_sync(false)); assert_eq!( plan.floor().expect("interrupted sync must have a floor"), &stored, ); assert!(matches!( plan.marshal_start(()), Start::Floor(ref floor) if floor == &stored )); let plan = SyncPlan::<_, TestScheme, TestVariant>::init(&context, partition_prefix).await; let plan = plan.with_floor(finalization(&fixture.schemes, 6, 6)); assert_eq!( plan.floor().expect("interrupted sync must have a floor"), &stored, "a lagging probe must not replace the persisted in-progress floor", ); let newer = finalization(&fixture.schemes, 9, 9); let plan = SyncPlan::<_, TestScheme, TestVariant>::init(&context, partition_prefix).await; let plan = plan.with_floor(newer.clone()); assert_eq!(plan.floor(), Some(&newer)); }); } #[test] fn with_floor_does_not_replace_newer_selection() { deterministic::Runner::default().start(|mut context| async move { let fixture = scheme_mocks::fixture(&mut context, b"_COMMONWARE_GLUE_SYNC_PLAN", 1); let newer = finalization(&fixture.schemes, 9, 9); let plan = SyncPlan::<_, TestScheme, TestVariant>::init( &context, "with_floor_does_not_replace_newer_selection", ) .await; let plan = plan.with_floor(newer.clone()) .with_floor(finalization(&fixture.schemes, 8, 8)); assert_eq!(plan.floor(), Some(&newer)); }); } #[test] #[should_panic( expected = "selected state sync floor cannot move behind the persisted in-progress floor" )] fn in_progress_sync_panics_for_backward_floor() { deterministic::Runner::default().start(|mut context| async move { let fixture = scheme_mocks::fixture(&mut context, b"_COMMONWARE_GLUE_SYNC_PLAN", 1); let mut metadata = StateSyncMetadata::<_, TestScheme, Sha256Digest>::init( &context, "in_progress_sync_panics_for_backward_floor", ) .await; metadata .begin_sync(finalization(&fixture.schemes, 7, 7)) .await; metadata .begin_sync(finalization(&fixture.schemes, 6, 6)) .await; }); } #[test] #[should_panic( expected = "selected state sync floor conflicts with the persisted in-progress round" )] fn in_progress_sync_panics_for_conflicting_round() { deterministic::Runner::default().start(|mut context| async move { let fixture = scheme_mocks::fixture(&mut context, b"_COMMONWARE_GLUE_SYNC_PLAN", 1); let mut metadata = StateSyncMetadata::<_, TestScheme, Sha256Digest>::init( &context, "in_progress_sync_panics_for_conflicting_round", ) .await; metadata .begin_sync(finalization(&fixture.schemes, 7, 7)) .await; metadata .begin_sync(finalization(&fixture.schemes, 7, 8)) .await; }); } }