From 9f31932b51310e35e521ee3365cf268a5708cfe7 Mon Sep 17 00:00:00 2001 From: clabby Date: Mon, 13 Jul 2026 11:05:53 -0400 Subject: [PATCH 1/8] engine: select restart floor through SyncPlan --- bin/validator/src/run.rs | 12 +- crates/engine/src/engine.rs | 193 +++++++++++++------------- crates/engine/src/tests/common.rs | 124 +++++++++++++---- crates/engine/src/tests/mod.rs | 112 +++++++++++---- crates/engine/src/tests/properties.rs | 92 +++++++++++- 5 files changed, 375 insertions(+), 158 deletions(-) diff --git a/bin/validator/src/run.rs b/bin/validator/src/run.rs index de6fc1b5..32ded9c4 100644 --- a/bin/validator/src/run.rs +++ b/bin/validator/src/run.rs @@ -852,19 +852,13 @@ fn run_with_config(config: LoadedConfig, config_path: PathBuf) { let startup = match startup { StartupModeConfig::MarshalSync => StartupMode::MarshalSync, - StartupModeConfig::StateSync => { - let finalization = probe_mailbox - .subscribe() - .await - .expect("probe actor exited before selecting a state-sync floor"); - StartupMode::StateSync { finalization } - } + StartupModeConfig::StateSync => StartupMode::StateSync, }; let startup_mode = match &startup { StartupMode::MarshalSync => "marshal_sync", - StartupMode::StateSync { .. } => "state_sync", + StartupMode::StateSync => "state_sync", }; - info!(startup_mode, "selected validator startup mode"); + info!(startup_mode, "requested validator startup mode"); // Build the indexer wiring up-front. This consumes `indexer` from the // loaded config and returns `None` for primaries or validators that diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index 0cd30138..bcbb2efb 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -20,7 +20,7 @@ use commonware_consensus::{ coding::{Marshaled, MarshaledConfig, shards, types::coding_config_for_participants}, core::{Actor as MarshalActor, Variant as MarshalVariant}, resolver::p2p as marshal_resolver, - store::{Blocks, Certificates}, + store::Blocks, }, simplex::{ self, config::Floor as SimplexFloor, elector::Config as Elector, types::Finalization, @@ -76,6 +76,7 @@ pub type ThresholdScheme = simplex::scheme::bls12381_threshold::standard:: const FIXED_EPOCH_LENGTH: NonZero = NZU64!(u64::MAX); const MAILBOX_SIZE: NonZero = NZUsize!(1024); const ACTIVITY_TIMEOUT: ViewDelta = ViewDelta::new(256); +#[cfg(not(all(test, feature = "test-utils")))] const PRUNABLE_ITEMS_PER_SECTION: NonZero = NZU64!(4_096); const FREEZER_VALUE_COMPRESSION: Option = None; const REPLAY_BUFFER: NonZero = NZUsize!(8 * 1024 * 1024); @@ -139,14 +140,15 @@ where pub transaction_resolver: (S, R), } -/// Engine initialization parameters. +/// Requested engine startup behavior. /// -/// `O` is the type of an optional simplex activity observer (e.g. the -/// indexer's certificate publisher). Pass `None::>` -/// when no external observer is wired in. -pub enum StartupMode { +/// The engine resolves this request against its durable [`SyncPlan`]. A state-sync +/// request only probes for a floor when the plan determines state sync is needed. +pub enum StartupMode { + /// Recover consensus and application state from local storage. MarshalSync, - StateSync { finalization: F }, + /// Request state sync from peers when required by the durable sync plan. + StateSync, } pub struct Config @@ -170,12 +172,14 @@ where pub partition_prefix: String, pub strategy: St, pub public_key_cache: PublicKeyCache, - pub startup: StartupMode>, + pub startup: StartupMode, pub sync_config: SyncEngineConfig, pub prune_config: Option, pub genesis_leader: C::PublicKey, pub transaction_namespace: &'static [u8], pub block_codec: BlockCfg, + #[cfg(all(test, feature = "test-utils"))] + pub prunable_items_per_section: NonZero, pub probe: Option>, /// Optional external observer of the simplex activity stream. The marshal /// reporter is always wired up; this slot is fanned out via @@ -232,6 +236,8 @@ where >, #[cfg(all(test, feature = "test-utils"))] marshal_mailbox: EngineMarshalMailbox, + #[cfg(all(test, feature = "test-utils"))] + startup_sync_floor: Option>, simplex: SimplexEngine, } @@ -254,6 +260,11 @@ where self.marshal_mailbox.clone() } + #[cfg(all(test, feature = "test-utils"))] + pub(crate) fn startup_sync_floor(&self) -> Option> { + self.startup_sync_floor.clone() + } + /// Returns the state database once the stateful actor has initialized it. /// Blocks until the database is ready. pub async fn subscribe_databases(&self) -> StateSyncDb { @@ -336,28 +347,25 @@ where coding_config, )); + #[cfg(all(test, feature = "test-utils"))] + let prunable_items_per_section = config.prunable_items_per_section; + #[cfg(not(all(test, feature = "test-utils")))] + let prunable_items_per_section = PRUNABLE_ITEMS_PER_SECTION; let (finalizations_by_height, finalized_blocks) = futures::join!( init_finalizations_archive::( &context, &page_cache, &config.partition_prefix, + prunable_items_per_section, ), init_finalized_blocks_archive::( &context, &page_cache, &config.partition_prefix, &config.block_codec, + prunable_items_per_section, ), ); - let recovered_floor = - if let Some(height) = Certificates::last_index(&finalizations_by_height) { - finalizations_by_height - .get(ArchiveIdentifier::Index(height.get())) - .await - .expect("failed to read recovered finalization floor") - } else { - None - }; let transaction_db_config = transaction_db_config( &config.partition_prefix, &page_cache, @@ -371,79 +379,72 @@ where stateful_partition_prefix.clone(), ) .await; - let (genesis_state_target, genesis_transactions_target, marshal_start, simplex_floor) = - if let Some(finalization) = recovered_floor { - let block_digest = - as MarshalVariant>::commitment_to_inner( - finalization.proposal.payload, - ); - let stored_block = finalized_blocks - .get(ArchiveIdentifier::Key(&block_digest)) - .await - .expect("failed to read recovered finalization floor block") - .expect("recovered finalization floor block must exist"); - let floor_block = as MarshalVariant>::into_inner( - stored_block.into(), - ); - let (state_target, transaction_target) = block_targets(&floor_block); - - ( - state_target, - transaction_target, - marshal::Start::Floor(finalization.clone()), - SimplexFloor::Finalized(finalization), - ) - } else { - let genesis_state_db = StateDb::::init( - context.child("genesis_state"), - state_db_config( - &config.partition_prefix, - &storage_page_cache, - config.strategy.clone(), - ), - ) - .await - .expect("state db must initialize for genesis target"); - let genesis_state_target = - as ManagedDb>::sync_target(&genesis_state_db).await; - let genesis_transaction_db = TransactionDb::::init( - context.child("genesis_transactions"), - transaction_db_config.clone(), - ) + let state_sync_requested = matches!(&config.startup, StartupMode::StateSync); + if startup_plan.should_state_sync(state_sync_requested) { + let finalization = config + .probe + .as_ref() + .expect("state sync requires a probe mailbox") + .subscribe() .await - .expect("transaction history db must initialize for genesis target"); - let genesis_transactions_target = - as ManagedDb>::sync_target(&genesis_transaction_db) - .await; - let genesis_block = - constantinople_application::consensus::genesis_block_with_parent( - &mut H::default(), - config.genesis_leader.clone(), - (commonware_consensus::types::View::zero(), genesis_parent), - 0, - genesis_state_target.clone(), - genesis_transactions_target.clone(), - ); - let coded_block = - EngineCodedBlock::new(genesis_block, coding_config, &config.strategy); - let commitment = coded_block.commitment(); - let simplex_floor = match &config.startup { - StartupMode::StateSync { finalization } if startup_plan.may_state_sync() => { - startup_plan = startup_plan.with_floor(finalization.clone()); - SimplexFloor::Finalized(finalization.clone()) - } - StartupMode::MarshalSync | StartupMode::StateSync { .. } => { - SimplexFloor::Genesis(commitment) - } - }; - - ( - genesis_state_target, - genesis_transactions_target, - startup_plan.marshal_start(coded_block), - simplex_floor, - ) - }; + .expect("probe actor exited before selecting a state-sync floor"); + startup_plan = startup_plan.with_floor(finalization); + } + + let stored_genesis = finalized_blocks + .get(ArchiveIdentifier::Index(0)) + .await + .expect("failed to read canonical genesis block"); + let canonical_genesis = if let Some(stored_genesis) = stored_genesis { + stored_genesis.into() + } else { + let state_db = StateDb::::init( + context.child("genesis_state"), + state_db_config( + &config.partition_prefix, + &storage_page_cache, + config.strategy.clone(), + ), + ) + .await + .expect("state db must initialize for genesis target"); + let database_state_target = + as ManagedDb>::sync_target(&state_db).await; + let transaction_db = TransactionDb::::init( + context.child("genesis_transactions"), + transaction_db_config.clone(), + ) + .await + .expect("transaction history db must initialize for genesis target"); + let database_transactions_target = + as ManagedDb>::sync_target(&transaction_db).await; + drop(state_db); + drop(transaction_db); + + let genesis_block = constantinople_application::consensus::genesis_block_with_parent( + &mut H::default(), + config.genesis_leader.clone(), + (commonware_consensus::types::View::zero(), genesis_parent), + 0, + database_state_target, + database_transactions_target, + ); + EngineCodedBlock::new(genesis_block, coding_config, &config.strategy) + }; + let application_genesis = as MarshalVariant>::into_inner( + canonical_genesis.clone(), + ); + let (application_state_target, application_transactions_target) = + block_targets(&application_genesis); + + #[cfg(all(test, feature = "test-utils"))] + let startup_sync_floor = startup_plan.floor().cloned(); + let genesis_commitment = canonical_genesis.commitment(); + let simplex_floor = startup_plan.floor().map_or_else( + || SimplexFloor::Genesis(genesis_commitment), + |finalization| SimplexFloor::Finalized(finalization.clone()), + ); + let marshal_start = startup_plan.marshal_start(canonical_genesis); let (marshal, marshal_mailbox, _) = MarshalActor::init( context.child("marshal"), @@ -456,7 +457,7 @@ where partition_prefix: format!("{}_marshal", config.partition_prefix), mailbox_size: MAILBOX_SIZE, view_retention_timeout: ACTIVITY_TIMEOUT, - prunable_items_per_section: PRUNABLE_ITEMS_PER_SECTION, + prunable_items_per_section, page_cache: page_cache.clone(), replay_buffer: REPLAY_BUFFER, key_write_buffer: WRITE_BUFFER, @@ -495,8 +496,8 @@ where genesis_parent, config.transaction_namespace, config.public_key_cache, - genesis_state_target, - genesis_transactions_target, + application_state_target, + application_transactions_target, config.finalized_hook, ); let (stateful, stateful_mailbox) = Stateful::init( @@ -586,6 +587,8 @@ where marshal, #[cfg(all(test, feature = "test-utils"))] marshal_mailbox, + #[cfg(all(test, feature = "test-utils"))] + startup_sync_floor, simplex, } } @@ -708,6 +711,7 @@ async fn init_finalizations_archive( context: &E, page_cache: &CacheRef, partition_prefix: &str, + items_per_section: NonZero, ) -> PrunableArchive, Commitment>> where E: BufferPooler + Spawner + Metrics + CryptoRngCore + Clock + Storage + Network, @@ -724,7 +728,7 @@ where key_page_cache: page_cache.clone(), value_partition: format!("{partition_prefix}-finalizations-by-height-value"), compression: FREEZER_VALUE_COMPRESSION, - items_per_section: PRUNABLE_ITEMS_PER_SECTION, + items_per_section, codec_config: ThresholdScheme::::certificate_codec_config_unbounded(), replay_buffer: REPLAY_BUFFER, key_write_buffer: WRITE_BUFFER, @@ -742,6 +746,7 @@ async fn init_finalized_blocks_archive( page_cache: &CacheRef, partition_prefix: &str, block_codec: &BlockCfg, + items_per_section: NonZero, ) -> PrunableArchive> where E: BufferPooler + Spawner + Metrics + CryptoRngCore + Clock + Storage + Network, @@ -757,7 +762,7 @@ where key_page_cache: page_cache.clone(), value_partition: format!("{partition_prefix}-finalized-blocks-value"), compression: FREEZER_VALUE_COMPRESSION, - items_per_section: PRUNABLE_ITEMS_PER_SECTION, + items_per_section, codec_config: block_codec.clone(), replay_buffer: REPLAY_BUFFER, key_write_buffer: WRITE_BUFFER, diff --git a/crates/engine/src/tests/common.rs b/crates/engine/src/tests/common.rs index 7aa08a7f..89ee3fae 100644 --- a/crates/engine/src/tests/common.rs +++ b/crates/engine/src/tests/common.rs @@ -4,7 +4,7 @@ use crate::{ }; use commonware_actor::Feedback; use commonware_consensus::{ - Reporter, + Heightable, Reporter, marshal::{self, Identifier}, types::{Height, View}, }; @@ -18,10 +18,17 @@ use commonware_cryptography::{ sha256::Sha256, }; use commonware_glue::simulate::{processed::ProcessedHeight, tracker::FinalizationUpdate}; -use commonware_runtime::{Clock, Metrics, Quota, Storage}; -use commonware_storage::metadata::{Config as MetadataConfig, Metadata}; -use commonware_utils::{Acknowledgement, N3f1, TryCollect, channel::mpsc, sequence::U64, test_rng}; -use std::collections::BTreeMap; +use commonware_runtime::Quota; +use commonware_utils::{ + Acknowledgement, N3f1, TryCollect, acknowledgement::Exact, channel::mpsc, sync::Mutex, test_rng, +}; +use std::{ + collections::BTreeMap, + sync::{ + Arc, + atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, + }, +}; pub(crate) type TestHasher = Sha256; pub(crate) type TestPrivateKey = ed25519::PrivateKey; @@ -30,19 +37,95 @@ pub(crate) type TestScheme = ThresholdScheme; pub(crate) type TestBlock = EngineBlock; pub(crate) type TestMarshalMailbox = EngineMarshalMailbox; pub(crate) const TRANSACTION_NAMESPACE: &[u8] = b"constantinople-engine-test-transactions"; -const STATE_SYNC_METADATA_SUFFIX: &str = "_state_sync_metadata"; -const SYNC_DONE_KEY: U64 = U64::new(0); pub(crate) const TEST_QUOTA: Quota = Quota::per_second(std::num::NonZeroU32::MAX); -#[derive(Clone, Copy, Debug, Default)] -pub(crate) struct NoopReporter; +#[derive(Clone, Default)] +pub(crate) struct RestartBarrier { + held: Arc>>, + released: Arc, + starts: Arc, + latest_finalized: Arc, + recovered_finalized: Arc, + observed_processed: Arc>>, +} + +impl RestartBarrier { + pub(crate) fn begin_start(&self) -> bool { + let restarting = self.starts.fetch_add(1, Ordering::SeqCst) > 0; + if !restarting { + return false; + } + + self.recovered_finalized.store( + self.latest_finalized.load(Ordering::SeqCst), + Ordering::SeqCst, + ); + self.held.lock().clear(); + true + } + + fn observe_finalized(&self, height: u64) { + self.latest_finalized.fetch_max(height, Ordering::SeqCst); + } + + fn acknowledge(&self, block_height: u64, acknowledgement: Exact) { + if block_height == 0 || self.released.load(Ordering::SeqCst) { + acknowledgement.acknowledge(); + return; + } + + self.held.lock().push(acknowledgement); + } + + pub(crate) fn observe_processed(&self, height: u64) { + *self.observed_processed.lock() = Some(height); + } -impl Reporter for NoopReporter { + pub(crate) fn release(&self) { + self.released.store(true, Ordering::SeqCst); + let acknowledgements = std::mem::take(&mut *self.held.lock()); + for acknowledgement in acknowledgements { + acknowledgement.acknowledge(); + } + } + + pub(crate) fn recovered_finalized(&self) -> u64 { + self.recovered_finalized.load(Ordering::SeqCst) + } + + pub(crate) fn observed_processed(&self) -> Option { + *self.observed_processed.lock() + } +} + +#[derive(Clone, Default)] +pub(crate) struct TestReporter { + restart_barrier: Option, +} + +impl TestReporter { + pub(crate) const fn new(restart_barrier: Option) -> Self { + Self { restart_barrier } + } +} + +impl Reporter for TestReporter { type Activity = marshal::Update; fn report(&mut self, activity: Self::Activity) -> Feedback { - if let marshal::Update::Block(_, response) = activity { - Acknowledgement::acknowledge(response); + match activity { + marshal::Update::Tip(_, height, _) => { + if let Some(barrier) = &self.restart_barrier { + barrier.observe_finalized(height.get()); + } + } + marshal::Update::Block(block, response) => { + if let Some(barrier) = &self.restart_barrier { + barrier.acknowledge(block.height().get(), response); + } else { + response.acknowledge(); + } + } } Feedback::Ok } @@ -152,20 +235,3 @@ pub(crate) fn validator_fixture(validators: u32) -> Fixture { (signers, output, shares) } - -pub(crate) async fn state_sync_done( - context: &(impl Storage + Clock + Metrics), - partition_prefix: &str, -) -> bool { - let metadata = Metadata::<_, U64, bool>::init( - context.child("state_sync_done"), - MetadataConfig { - partition: format!("{partition_prefix}{STATE_SYNC_METADATA_SUFFIX}"), - codec_config: (), - }, - ) - .await - .expect("failed to read state sync metadata"); - - metadata.get(&SYNC_DONE_KEY).copied().unwrap_or(false) -} diff --git a/crates/engine/src/tests/mod.rs b/crates/engine/src/tests/mod.rs index a37d6df1..9128d2c5 100644 --- a/crates/engine/src/tests/mod.rs +++ b/crates/engine/src/tests/mod.rs @@ -9,10 +9,12 @@ use crate::{ TRANSACTION_RESOLVER_CHANNEL, VOTE_CHANNEL, }; use common::{ - HeightMonitorReporter, NoopReporter, TEST_QUOTA, TRANSACTION_NAMESPACE, TestHasher, - TestPrivateKey, TestPublicKey, TestScheme, ValidatorState, state_sync_done, validator_fixture, + HeightMonitorReporter, RestartBarrier, TEST_QUOTA, TRANSACTION_NAMESPACE, TestHasher, + TestPrivateKey, TestPublicKey, TestReporter, TestScheme, ValidatorState, validator_fixture, }; use commonware_consensus::{ + Heightable, + marshal::core::CommitmentFallback, simplex::elector::RoundRobin, types::{Epoch, coding::Commitment}, }; @@ -48,7 +50,7 @@ use constantinople_mempool::mocks::StaticTransactionSource; use constantinople_primitives::PublicKeyCache; use properties::{ BlockAgreementAtHeight, FinalizedHeightAtLeast, LateJoinerStateSyncHandoff, - StateSyncReadyAtHeight, + RestartPreservesProcessedHeight, RestartRecoveryComplete, StateSyncReadyAtHeight, }; use std::{collections::BTreeMap, sync::Arc, time::Duration}; use tracing::{info, warn}; @@ -86,6 +88,7 @@ struct TestEngineDefinition { /// configured by `PlanBuilder`. use_discovery_split: bool, sync_heights: Arc>>, + restart_barrier: Option, } impl TestEngineDefinition { @@ -99,6 +102,7 @@ impl TestEngineDefinition { enable_state_sync: false, use_discovery_split: false, sync_heights: Arc::new(Mutex::new(BTreeMap::new())), + restart_barrier: None, } } @@ -132,6 +136,11 @@ impl TestEngineDefinition { self.enable_state_sync = true; self } + + fn with_restart_barrier(mut self, barrier: RestartBarrier) -> Self { + self.restart_barrier = Some(barrier); + self + } } impl EngineDefinition for TestEngineDefinition { @@ -174,11 +183,14 @@ impl EngineDefinition for TestEngineDefinition { let signer = self.signers[index].clone(); let share = self.shares.get(&public_key).cloned().flatten(); let partition_prefix = format!("validator-{index}"); - let stateful_partition_prefix = format!("{partition_prefix}_stateful"); let output = self.output.clone(); let sync_heights = self.sync_heights.clone(); let enable_state_sync = self.enable_state_sync; let uses_state_sync = enable_state_sync && index == 0; + let restart_barrier = (index == 0).then(|| self.restart_barrier.clone()).flatten(); + let is_restart = restart_barrier + .as_ref() + .is_some_and(RestartBarrier::begin_start); let genesis_leader = self.signers[0].public_key(); let mut manager = oracle.manager(); let blocker = oracle.control(public_key.clone()); @@ -233,33 +245,19 @@ impl EngineDefinition for TestEngineDefinition { (None, None) }; - let (startup, startup_sync_height) = if uses_state_sync - && !state_sync_done(&context, &stateful_partition_prefix).await - { - probe_mailbox - .as_ref() - .expect("state-sync scenario requires probe") - .subscribe() - .await - .map(|finalization| { - let height = finalization.proposal.round.view().get(); - sync_heights.lock().insert(public_key.clone(), height); - (StartupMode::StateSync { finalization }, Some(height)) - }) - .expect("probe actor exited before selecting a state-sync floor") + let startup = if uses_state_sync { + StartupMode::StateSync } else { - let prior = sync_heights.lock().get(&public_key).copied(); - (StartupMode::MarshalSync, prior) + StartupMode::MarshalSync }; let startup_mode = match &startup { StartupMode::MarshalSync => "marshal_sync", - StartupMode::StateSync { .. } => "state_sync", + StartupMode::StateSync => "state_sync", }; info!( validator = %public_key, %startup_mode, - startup_sync_height, - "initialized validator startup mode", + "requested validator startup mode", ); let channels = Channels { @@ -274,7 +272,11 @@ impl EngineDefinition for TestEngineDefinition { let input = StaticTransactionSource::::new(Vec::new()); - let reporter = HeightMonitorReporter::new(public_key.clone(), monitor, NoopReporter); + let reporter = HeightMonitorReporter::new( + public_key.clone(), + monitor, + TestReporter::new(restart_barrier.clone()), + ); let engine = Engine::< _, _, @@ -320,6 +322,11 @@ impl EngineDefinition for TestEngineDefinition { genesis_leader, transaction_namespace: TRANSACTION_NAMESPACE, block_codec: Default::default(), + prunable_items_per_section: if restart_barrier.is_some() { + NZU64!(1) + } else { + NZU64!(4_096) + }, probe: probe_mailbox.clone(), simplex_observer: None, finalized_hook: None, @@ -327,7 +334,25 @@ impl EngineDefinition for TestEngineDefinition { ) .await; + let selected_sync_floor = engine.startup_sync_floor(); let marshal = engine.marshal_mailbox(); + let restart_marshal = marshal.clone(); + let engine_handle = engine.start(channels, Some(reporter)); + let startup_sync_height = if let Some(finalization) = selected_sync_floor { + let block = marshal + .subscribe_by_commitment( + finalization.proposal.payload, + CommitmentFallback::Wait, + ) + .await + .expect("state-sync floor block must be available"); + let height = block.height().get(); + sync_heights.lock().insert(public_key.clone(), height); + info!(validator = %public_key, height, "resolved state-sync floor block"); + Some(height) + } else { + sync_heights.lock().get(&public_key).copied() + }; if state_sender .send(ValidatorState { marshal, @@ -339,7 +364,15 @@ impl EngineDefinition for TestEngineDefinition { return; } - let engine_handle = engine.start(channels, Some(reporter)); + if is_restart { + let processed = restart_marshal + .get_processed_height() + .await + .map_or(0, |height| height.get()); + let barrier = restart_barrier.expect("restart barrier must exist"); + barrier.observe_processed(processed); + barrier.release(); + } let engine_result = if let Some(probe_handle) = probe_handle { let (probe_result, engine_result) = futures::join!(probe_handle, engine_handle); if let Err(error) = probe_result { @@ -420,6 +453,29 @@ fn run_crash_restart(engine: TestEngineDefinition) { .unwrap(); } +fn run_restart_with_archived_finalizations() { + let barrier = RestartBarrier::default(); + let engine = TestEngineDefinition::new(NUM_VALIDATORS).with_restart_barrier(barrier.clone()); + let validator = engine.participants()[0].clone(); + + PlanBuilder::new(engine) + .link(default_link()) + .seed(0) + .crash(Crash::Schedule( + Schedule::new() + .at( + Duration::from_millis(2_500), + Action::Crash(validator.clone()), + ) + .at(Duration::from_millis(5_000), Action::Restart(validator)), + )) + .timeout(Duration::from_secs(30)) + .exit_condition(RestartRecoveryComplete::new(barrier.clone())) + .property(RestartPreservesProcessedHeight::new(barrier)) + .run() + .unwrap(); +} + fn run_delayed_start(engine: TestEngineDefinition) { PlanBuilder::new(engine) .link(default_link()) @@ -690,6 +746,12 @@ fn crash_and_restart_one_validator() { run_crash_restart(TestEngineDefinition::new(NUM_VALIDATORS)); } +#[test_group("slow")] +#[test_traced("DEBUG")] +fn restart_replays_finalizations_archived_before_acknowledgement() { + run_restart_with_archived_finalizations(); +} + #[test_group("slow")] #[test_traced("DEBUG")] fn delayed_start_one_validator() { diff --git a/crates/engine/src/tests/properties.rs b/crates/engine/src/tests/properties.rs index 810d7cb7..646de307 100644 --- a/crates/engine/src/tests/properties.rs +++ b/crates/engine/src/tests/properties.rs @@ -1,4 +1,4 @@ -use crate::tests::common::ValidatorState; +use crate::tests::common::{RestartBarrier, ValidatorState}; use commonware_cryptography::PublicKey; use commonware_glue::simulate::{ exit::ExitCondition, property::Property, tracker::ProgressTracker, @@ -85,6 +85,96 @@ pub(crate) struct FinalizedHeightAtLeast { height: u64, } +#[derive(Clone)] +pub(crate) struct RestartRecoveryComplete { + barrier: RestartBarrier, +} + +impl RestartRecoveryComplete { + pub(crate) const fn new(barrier: RestartBarrier) -> Self { + Self { barrier } + } +} + +impl ExitCondition for RestartRecoveryComplete { + fn name(&self) -> &str { + "restart_recovery_complete" + } + + fn requires_polling(&self) -> bool { + true + } + + fn reached<'a>( + &'a self, + _tracker: &'a ProgressTracker

, + states: &'a [&'a ValidatorState], + target_count: usize, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + let recovered_finalized = self.barrier.recovered_finalized(); + if recovered_finalized == 0 || self.barrier.observed_processed().is_none() { + return Ok(false); + } + + let mut recovered = 0; + for state in states { + if state.processed_height().await >= recovered_finalized { + recovered += 1; + } + } + + Ok(recovered >= target_count) + }) + } +} + +#[derive(Clone)] +pub(crate) struct RestartPreservesProcessedHeight { + barrier: RestartBarrier, +} + +impl RestartPreservesProcessedHeight { + pub(crate) const fn new(barrier: RestartBarrier) -> Self { + Self { barrier } + } +} + +impl Property + for RestartPreservesProcessedHeight +{ + fn name(&self) -> &str { + "restart_preserves_processed_height" + } + + fn check<'a>( + &'a self, + _tracker: &'a ProgressTracker, + _states: &'a [&'a ValidatorState], + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + let recovered_finalized = self.barrier.recovered_finalized(); + if recovered_finalized <= 1 { + return Err(format!( + "restart recovered finalization {recovered_finalized}, expected a height above the held processed floor" + )); + } + + let observed_processed = self + .barrier + .observed_processed() + .ok_or_else(|| "restart processed height was not observed".to_string())?; + if observed_processed != 0 { + return Err(format!( + "restart moved processed height from 0 to {observed_processed} before acknowledgement; recovered finalization was {recovered_finalized}" + )); + } + + Ok(()) + }) + } +} + impl FinalizedHeightAtLeast { pub(crate) const fn new(height: u64) -> Self { Self { height } From 872ec5139afb137f4abcb6e0f738b8ca6cb75baf Mon Sep 17 00:00:00 2001 From: clabby Date: Mon, 13 Jul 2026 12:57:22 -0400 Subject: [PATCH 2/8] engine: Keep `items_per_section` config --- bin/validator/src/run.rs | 2 ++ crates/engine/src/engine.rs | 6 ------ 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/bin/validator/src/run.rs b/bin/validator/src/run.rs index 32ded9c4..fc4f762a 100644 --- a/bin/validator/src/run.rs +++ b/bin/validator/src/run.rs @@ -82,6 +82,7 @@ const PRUNE_CONFIG: PruneConfig = PruneConfig { retained_marshal_blocks: 1024, retained_qmdb_blocks: 32, }; +const PRUNABLE_ITEMS_PER_SECTION: NonZeroU64 = NZU64!(4_096); const FINALIZED_QUEUE_ITEMS_PER_SECTION: NonZeroU64 = NZU64!(128); const FINALIZED_QUEUE_PAGE_SIZE: NonZeroU16 = NZU16!(4_096); const FINALIZED_QUEUE_PAGE_CACHE_CAPACITY: NonZeroUsize = NZUsize!(8_192); @@ -905,6 +906,7 @@ fn run_with_config(config: LoadedConfig, config_path: PathBuf) { genesis_leader: decoded.genesis_leader, transaction_namespace: constantinople_primitives::TRANSACTION_NAMESPACE, block_codec: Default::default(), + prunable_items_per_section: PRUNABLE_ITEMS_PER_SECTION, probe: Some(probe_mailbox.clone()), simplex_observer: relayer_observer.map(SimplexObserver::Relayer).or_else(|| { indexer_handle diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index bcbb2efb..560321e6 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -76,8 +76,6 @@ pub type ThresholdScheme = simplex::scheme::bls12381_threshold::standard:: const FIXED_EPOCH_LENGTH: NonZero = NZU64!(u64::MAX); const MAILBOX_SIZE: NonZero = NZUsize!(1024); const ACTIVITY_TIMEOUT: ViewDelta = ViewDelta::new(256); -#[cfg(not(all(test, feature = "test-utils")))] -const PRUNABLE_ITEMS_PER_SECTION: NonZero = NZU64!(4_096); const FREEZER_VALUE_COMPRESSION: Option = None; const REPLAY_BUFFER: NonZero = NZUsize!(8 * 1024 * 1024); const WRITE_BUFFER: NonZero = NZUsize!(1024 * 1024); @@ -178,7 +176,6 @@ where pub genesis_leader: C::PublicKey, pub transaction_namespace: &'static [u8], pub block_codec: BlockCfg, - #[cfg(all(test, feature = "test-utils"))] pub prunable_items_per_section: NonZero, pub probe: Option>, /// Optional external observer of the simplex activity stream. The marshal @@ -347,10 +344,7 @@ where coding_config, )); - #[cfg(all(test, feature = "test-utils"))] let prunable_items_per_section = config.prunable_items_per_section; - #[cfg(not(all(test, feature = "test-utils")))] - let prunable_items_per_section = PRUNABLE_ITEMS_PER_SECTION; let (finalizations_by_height, finalized_blocks) = futures::join!( init_finalizations_archive::( &context, From 2f92aa02e0c2b20dee4aebc765d1e60e3908bb49 Mon Sep 17 00:00:00 2001 From: clabby Date: Mon, 13 Jul 2026 13:00:42 -0400 Subject: [PATCH 3/8] engine: document SyncPlan startup invariants --- crates/engine/src/engine.rs | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index 560321e6..1d0a399c 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -373,6 +373,9 @@ where stateful_partition_prefix.clone(), ) .await; + // The durable plan distinguishes normal recovery from peer state sync. Normal recovery + // stays floorless so marshal restores its acknowledged progress; only a requested or + // interrupted state sync discovers a new floor. let state_sync_requested = matches!(&config.startup, StartupMode::StateSync); if startup_plan.should_state_sync(state_sync_requested) { let finalization = config @@ -385,6 +388,8 @@ where startup_plan = startup_plan.with_floor(finalization); } + // A floorless plan requires marshal's canonical height-zero anchor. Reuse the archived + // block on restart so current database targets cannot change the genesis commitment. let stored_genesis = finalized_blocks .get(ArchiveIdentifier::Index(0)) .await @@ -433,6 +438,8 @@ where #[cfg(all(test, feature = "test-utils"))] let startup_sync_floor = startup_plan.floor().cloned(); + // Simplex adopts the peer floor only for state sync. Otherwise it independently replays + // its journal from canonical genesis while marshal restores application progress. let genesis_commitment = canonical_genesis.commitment(); let simplex_floor = startup_plan.floor().map_or_else( || SimplexFloor::Genesis(genesis_commitment), From c64757980d6e2bae52b92253900f17d4c767ede0 Mon Sep 17 00:00:00 2001 From: clabby Date: Mon, 13 Jul 2026 13:02:56 -0400 Subject: [PATCH 4/8] engine: explain temporary genesis databases --- crates/engine/src/engine.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index 1d0a399c..d1398785 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -417,6 +417,8 @@ where .expect("transaction history db must initialize for genesis target"); let database_transactions_target = as ManagedDb>::sync_target(&transaction_db).await; + // Stateful owns the long-lived database initialization after selecting its startup + // path. These temporary handles exist only to derive the first-boot genesis targets. drop(state_db); drop(transaction_db); From ac6ec3beefeae8c74d7ae5f3837e670bf4d7a5f9 Mon Sep 17 00:00:00 2001 From: clabby Date: Mon, 13 Jul 2026 13:12:43 -0400 Subject: [PATCH 5/8] engine: persist canonical genesis outside prunable archive --- crates/engine/src/engine.rs | 56 +++++++++++++++++++++++++++++----- crates/engine/src/tests/mod.rs | 38 +++++++++++++++++------ 2 files changed, 77 insertions(+), 17 deletions(-) diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index d1398785..183387f6 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -52,11 +52,12 @@ use commonware_storage::{ fixed::Config as FixedJournalConfig, variable::Config as VariableJournalConfig, }, merkle::full::Config as MmrConfig, + metadata::{Config as MetadataConfig, Metadata}, mmr, qmdb::{any::FixedConfig, keyless::fixed as keyless_fixed}, translator::EightCap, }; -use commonware_utils::{NZU16, NZU64, NZUsize, non_empty_range, union}; +use commonware_utils::{NZU16, NZU64, NZUsize, non_empty_range, sequence::U64, union}; use constantinople_application::consensus::{ Application, FinalizedHookFn, StateSyncTarget, TransactionHistoryTarget, }; @@ -85,6 +86,7 @@ const ITEMS_PER_BLOB: NonZero = NZU64!(1_048_576 * 25); // ~1gb const MAX_REPAIR: NonZero = NZUsize!(200); pub const MAX_PENDING_ACKS: NonZero = NZUsize!(4); const WITNESS_ITEMS_PER_SECTION: NonZero = NZU64!(64); +const CANONICAL_GENESIS_KEY: U64 = U64::new(0); const SHARD_BACKGROUND_CHANNEL_CAPACITY: NonZero = NZUsize!(1024); const SHARD_PEER_BUFFER_SIZE: NonZero = NZUsize!(64); const DB_WRITE_BUFFER: NonZero = NZUsize!(8 * 1024 * 1024); @@ -235,6 +237,8 @@ where marshal_mailbox: EngineMarshalMailbox, #[cfg(all(test, feature = "test-utils"))] startup_sync_floor: Option>, + #[cfg(all(test, feature = "test-utils"))] + genesis_commitment: Commitment, simplex: SimplexEngine, } @@ -262,6 +266,11 @@ where self.startup_sync_floor.clone() } + #[cfg(all(test, feature = "test-utils"))] + pub(crate) const fn genesis_commitment(&self) -> Commitment { + self.genesis_commitment + } + /// Returns the state database once the stateful actor has initialized it. /// Blocks until the database is ready. pub async fn subscribe_databases(&self) -> StateSyncDb { @@ -388,12 +397,26 @@ where startup_plan = startup_plan.with_floor(finalization); } - // A floorless plan requires marshal's canonical height-zero anchor. Reuse the archived - // block on restart so current database targets cannot change the genesis commitment. - let stored_genesis = finalized_blocks - .get(ArchiveIdentifier::Index(0)) - .await - .expect("failed to read canonical genesis block"); + // Height zero must outlive the prunable block archive. The archive fallback migrates + // existing storage; the database fallback is only reachable on first boot. + let mut genesis_metadata = Metadata::<_, U64, CodingBlock>::init( + context.child("canonical_genesis"), + MetadataConfig { + partition: format!("{}-canonical-genesis", config.partition_prefix), + codec_config: config.block_codec.clone(), + }, + ) + .await + .expect("failed to initialize canonical genesis metadata"); + let stored_genesis = genesis_metadata.get(&CANONICAL_GENESIS_KEY).cloned(); + let persist_genesis = stored_genesis.is_none(); + let stored_genesis = match stored_genesis { + Some(stored_genesis) => Some(stored_genesis), + None => finalized_blocks + .get(ArchiveIdentifier::Index(0)) + .await + .expect("failed to read canonical genesis block"), + }; let canonical_genesis = if let Some(stored_genesis) = stored_genesis { stored_genesis.into() } else { @@ -417,8 +440,16 @@ where .expect("transaction history db must initialize for genesis target"); let database_transactions_target = as ManagedDb>::sync_target(&transaction_db).await; + let state_is_empty = *database_state_target.range.start() == mmr::Location::new(0) + && *database_state_target.range.end() == mmr::Location::new(1); + let transactions_are_empty = + database_transactions_target.leaf_count == mmr::Location::new(1); + assert!( + state_is_empty && transactions_are_empty, + "canonical genesis is missing but application databases are not empty", + ); // Stateful owns the long-lived database initialization after selecting its startup - // path. These temporary handles exist only to derive the first-boot genesis targets. + // path. These handles only derive the first-boot genesis targets. drop(state_db); drop(transaction_db); @@ -432,6 +463,13 @@ where ); EngineCodedBlock::new(genesis_block, coding_config, &config.strategy) }; + if persist_genesis { + genesis_metadata.put(CANONICAL_GENESIS_KEY, canonical_genesis.clone().into()); + genesis_metadata + .sync() + .await + .expect("failed to persist canonical genesis block"); + } let application_genesis = as MarshalVariant>::into_inner( canonical_genesis.clone(), ); @@ -592,6 +630,8 @@ where marshal_mailbox, #[cfg(all(test, feature = "test-utils"))] startup_sync_floor, + #[cfg(all(test, feature = "test-utils"))] + genesis_commitment, simplex, } } diff --git a/crates/engine/src/tests/mod.rs b/crates/engine/src/tests/mod.rs index 9128d2c5..aef938c7 100644 --- a/crates/engine/src/tests/mod.rs +++ b/crates/engine/src/tests/mod.rs @@ -52,7 +52,7 @@ use properties::{ BlockAgreementAtHeight, FinalizedHeightAtLeast, LateJoinerStateSyncHandoff, RestartPreservesProcessedHeight, RestartRecoveryComplete, StateSyncReadyAtHeight, }; -use std::{collections::BTreeMap, sync::Arc, time::Duration}; +use std::{collections::BTreeMap, num::NonZeroU64, sync::Arc, time::Duration}; use tracing::{info, warn}; const NUM_VALIDATORS: u32 = 4; @@ -88,7 +88,10 @@ struct TestEngineDefinition { /// configured by `PlanBuilder`. use_discovery_split: bool, sync_heights: Arc>>, + genesis_commitments: Arc>>, restart_barrier: Option, + prunable_items_per_section: NonZeroU64, + retained_marshal_blocks: usize, } impl TestEngineDefinition { @@ -102,7 +105,10 @@ impl TestEngineDefinition { enable_state_sync: false, use_discovery_split: false, sync_heights: Arc::new(Mutex::new(BTreeMap::new())), + genesis_commitments: Arc::new(Mutex::new(BTreeMap::new())), restart_barrier: None, + prunable_items_per_section: NZU64!(4_096), + retained_marshal_blocks: 16, } } @@ -139,6 +145,13 @@ impl TestEngineDefinition { fn with_restart_barrier(mut self, barrier: RestartBarrier) -> Self { self.restart_barrier = Some(barrier); + self.prunable_items_per_section = NZU64!(1); + self + } + + const fn with_aggressive_pruning(mut self) -> Self { + self.prunable_items_per_section = NZU64!(1); + self.retained_marshal_blocks = 0; self } } @@ -185,6 +198,9 @@ impl EngineDefinition for TestEngineDefinition { let partition_prefix = format!("validator-{index}"); let output = self.output.clone(); let sync_heights = self.sync_heights.clone(); + let genesis_commitments = self.genesis_commitments.clone(); + let prunable_items_per_section = self.prunable_items_per_section; + let retained_marshal_blocks = self.retained_marshal_blocks; let enable_state_sync = self.enable_state_sync; let uses_state_sync = enable_state_sync && index == 0; let restart_barrier = (index == 0).then(|| self.restart_barrier.clone()).flatten(); @@ -316,17 +332,13 @@ impl EngineDefinition for TestEngineDefinition { prune_config: Some(PruneConfig { max_pending_acks: MAX_PENDING_ACKS, maintenance_interval: NZUsize!(16), - retained_marshal_blocks: 16, + retained_marshal_blocks, retained_qmdb_blocks: 0, }), genesis_leader, transaction_namespace: TRANSACTION_NAMESPACE, block_codec: Default::default(), - prunable_items_per_section: if restart_barrier.is_some() { - NZU64!(1) - } else { - NZU64!(4_096) - }, + prunable_items_per_section, probe: probe_mailbox.clone(), simplex_observer: None, finalized_hook: None, @@ -334,6 +346,14 @@ impl EngineDefinition for TestEngineDefinition { ) .await; + let genesis_commitment = engine.genesis_commitment(); + if let Some(expected) = genesis_commitments + .lock() + .insert(public_key.clone(), genesis_commitment) + { + assert_eq!(genesis_commitment, expected, "genesis changed on restart"); + } + let selected_sync_floor = engine.startup_sync_floor(); let marshal = engine.marshal_mailbox(); let restart_marshal = marshal.clone(); @@ -742,8 +762,8 @@ fn deterministic_across_seeds() { #[test_group("slow")] #[test_traced("DEBUG")] -fn crash_and_restart_one_validator() { - run_crash_restart(TestEngineDefinition::new(NUM_VALIDATORS)); +fn restart_preserves_genesis_after_pruning() { + run_crash_restart(TestEngineDefinition::new(NUM_VALIDATORS).with_aggressive_pruning()); } #[test_group("slow")] From 3eaec2df501fc4880951e6422ab911a1f66a7073 Mon Sep 17 00:00:00 2001 From: Patrick O'Grady Date: Thu, 16 Jul 2026 14:32:06 -0700 Subject: [PATCH 6/8] nits --- crates/engine/src/engine.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index de7f5c5d..840b89a3 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -395,6 +395,7 @@ where stateful_partition_prefix.clone(), ) .await; + // The durable plan distinguishes normal recovery from peer state sync. Normal recovery // stays floorless so marshal restores its acknowledged progress; only a requested or // interrupted state sync discovers a new floor. From d15e3df06d1e0e341d24a797d639d2a83e5e4451 Mon Sep 17 00:00:00 2001 From: Patrick O'Grady Date: Thu, 16 Jul 2026 14:36:21 -0700 Subject: [PATCH 7/8] initial target --- crates/engine/src/engine.rs | 39 ++++--------------------------------- 1 file changed, 4 insertions(+), 35 deletions(-) diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index 840b89a3..b756fe29 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -434,46 +434,15 @@ where let canonical_genesis = if let Some(stored_genesis) = stored_genesis { stored_genesis.into() } else { - let state_db = StateDb::::init( - context.child("genesis_state"), - state_db_config( - &config.partition_prefix, - &storage_page_cache, - config.strategy.clone(), - ), - ) - .await - .expect("state db must initialize for genesis target"); - let database_state_target = - as ManagedDb>::sync_target(&state_db); - let transaction_db = TransactionDb::::init( - context.child("genesis_transactions"), - transaction_db_config.clone(), - ) - .await - .expect("transaction history db must initialize for genesis target"); - let database_transactions_target = - as ManagedDb>::sync_target(&transaction_db); - let state_is_empty = *database_state_target.range.start() == mmr::Location::new(0) - && *database_state_target.range.end() == mmr::Location::new(1); - let transactions_are_empty = - database_transactions_target.leaf_count == mmr::Location::new(1); - assert!( - state_is_empty && transactions_are_empty, - "canonical genesis is missing but application databases are not empty", - ); - // Stateful owns the long-lived database initialization after selecting its startup - // path. These handles only derive the first-boot genesis targets. - drop(state_db); - drop(transaction_db); - + // First boot: the genesis targets are the canonical empty-database roots, leaving + // the long-lived databases untouched for Stateful to open. let genesis_block = constantinople_application::consensus::genesis_block_with_parent( &mut H::default(), config.genesis_leader.clone(), (commonware_consensus::types::View::zero(), genesis_parent), 0, - database_state_target, - database_transactions_target, + as ManagedDb>::initial_sync_target(), + as ManagedDb>::initial_sync_target(), ); EngineCodedBlock::new(genesis_block, coding_config, &config.strategy) }; From a6412877652de20c1eee412fb52362f18f20f88e Mon Sep 17 00:00:00 2001 From: Patrick O'Grady Date: Thu, 16 Jul 2026 14:46:03 -0700 Subject: [PATCH 8/8] fmt --- crates/engine/src/engine.rs | 67 +++++++++---------------------------- 1 file changed, 16 insertions(+), 51 deletions(-) diff --git a/crates/engine/src/engine.rs b/crates/engine/src/engine.rs index b756fe29..8b16d5dd 100644 --- a/crates/engine/src/engine.rs +++ b/crates/engine/src/engine.rs @@ -20,7 +20,6 @@ use commonware_consensus::{ coding::{Marshaled, MarshaledConfig, shards, types::coding_config_for_participants}, core::{Actor as MarshalActor, Variant as MarshalVariant}, resolver::p2p as marshal_resolver, - store::Blocks, }, simplex::{ self, config::Floor as SimplexFloor, elector::Config as Elector, types::Finalization, @@ -47,17 +46,16 @@ use commonware_runtime::{ buffer::paged::CacheRef, spawn_cell, }; use commonware_storage::{ - archive::{Identifier as ArchiveIdentifier, prunable, prunable::Archive as PrunableArchive}, + archive::{prunable, prunable::Archive as PrunableArchive}, journal::contiguous::{ fixed::Config as FixedJournalConfig, variable::Config as VariableJournalConfig, }, merkle::full::Config as MmrConfig, - metadata::{Config as MetadataConfig, Metadata}, mmr, qmdb::{any::FixedConfig, keyless::fixed as keyless_fixed}, translator::EightCap, }; -use commonware_utils::{NZU16, NZU64, NZUsize, non_empty_range, sequence::U64, union}; +use commonware_utils::{NZU16, NZU64, NZUsize, non_empty_range, union}; use constantinople_application::consensus::{ Application, FinalizedHookFn, StateSyncTarget, TransactionHistoryTarget, }; @@ -85,7 +83,6 @@ const ITEMS_PER_BLOB: NonZero = NZU64!(1_048_576 * 25); // ~1gb const MAX_REPAIR: NonZero = NZUsize!(200); pub const MAX_PENDING_ACKS: NonZero = NZUsize!(4); const WITNESS_ITEMS_PER_SECTION: NonZero = NZU64!(64); -const CANONICAL_GENESIS_KEY: U64 = U64::new(0); const SHARD_BACKGROUND_CHANNEL_CAPACITY: NonZero = NZUsize!(1024); const SHARD_PEER_BUFFER_SIZE: NonZero = NZUsize!(64); const DB_WRITE_BUFFER: NonZero = NZUsize!(8 * 1024 * 1024); @@ -411,51 +408,19 @@ where startup_plan = startup_plan.with_floor(finalization); } - // Height zero must outlive the prunable block archive. The archive fallback migrates - // existing storage; the database fallback is only reachable on first boot. - let mut genesis_metadata = Metadata::<_, U64, CodingBlock>::init( - context.child("canonical_genesis"), - MetadataConfig { - partition: format!("{}-canonical-genesis", config.partition_prefix), - codec_config: config.block_codec.clone(), - }, - ) - .await - .expect("failed to initialize canonical genesis metadata"); - let stored_genesis = genesis_metadata.get(&CANONICAL_GENESIS_KEY).cloned(); - let persist_genesis = stored_genesis.is_none(); - let stored_genesis = match stored_genesis { - Some(stored_genesis) => Some(stored_genesis), - None => finalized_blocks - .get(ArchiveIdentifier::Index(0)) - .await - .expect("failed to read canonical genesis block"), - }; - let canonical_genesis = if let Some(stored_genesis) = stored_genesis { - stored_genesis.into() - } else { - // First boot: the genesis targets are the canonical empty-database roots, leaving - // the long-lived databases untouched for Stateful to open. - let genesis_block = constantinople_application::consensus::genesis_block_with_parent( - &mut H::default(), - config.genesis_leader.clone(), - (commonware_consensus::types::View::zero(), genesis_parent), - 0, - as ManagedDb>::initial_sync_target(), - as ManagedDb>::initial_sync_target(), - ); - EngineCodedBlock::new(genesis_block, coding_config, &config.strategy) - }; - if persist_genesis { - genesis_metadata.put(CANONICAL_GENESIS_KEY, canonical_genesis.clone().into()); - genesis_metadata - .sync() - .await - .expect("failed to persist canonical genesis block"); - } - let application_genesis = as MarshalVariant>::into_inner( - canonical_genesis.clone(), + // The canonical genesis is a pure function of configuration: the leader, the + // participant-derived coding config, and the canonical empty-database roots. + let genesis_block = constantinople_application::consensus::genesis_block_with_parent( + &mut H::default(), + config.genesis_leader.clone(), + (commonware_consensus::types::View::zero(), genesis_parent), + 0, + as ManagedDb>::initial_sync_target(), + as ManagedDb>::initial_sync_target(), ); + let coded_genesis = EngineCodedBlock::new(genesis_block, coding_config, &config.strategy); + let application_genesis = + as MarshalVariant>::into_inner(coded_genesis.clone()); let (application_state_target, application_transactions_target) = block_targets(&application_genesis); @@ -463,12 +428,12 @@ where let startup_sync_floor = startup_plan.floor().cloned(); // Simplex adopts the peer floor only for state sync. Otherwise it independently replays // its journal from canonical genesis while marshal restores application progress. - let genesis_commitment = canonical_genesis.commitment(); + let genesis_commitment = coded_genesis.commitment(); let simplex_floor = startup_plan.floor().map_or_else( || SimplexFloor::Genesis(genesis_commitment), |finalization| SimplexFloor::Finalized(finalization.clone()), ); - let marshal_start = startup_plan.marshal_start(canonical_genesis); + let marshal_start = startup_plan.marshal_start(coded_genesis); let (marshal, marshal_mailbox, _) = MarshalActor::init( context.child("marshal"),