sync virtual genesis id between consensus and execution

Bo Wu · areshand · commit 62b89109de0d · 2025-07-25T07:10:52.000-07:00
diff --git a/config/src/config/consensus_config.rs b/config/src/config/consensus_config.rs
@@ -363,7 +363,7 @@ impl Default for ConsensusConfig {
             max_pending_rounds_in_commit_vote_cache: 100,
             optimistic_sig_verification: true,
             enable_round_timeout_msg: true,
-            enable_pipeline: false,
+            enable_pipeline: true,
         }
     }
 }
diff --git a/consensus/consensus-types/src/pipelined_block.rs b/consensus/consensus-types/src/pipelined_block.rs
@@ -580,7 +580,7 @@ impl PipelinedBlock {
             .await
             .map(|(compute_result, execution_time, _)| (compute_result, execution_time))
             .map_err(|e| ExecutorError::InternalError {
-                error: e.to_string()
+                error: e.to_string(),
             })
     }
 
diff --git a/consensus/src/consensus_observer/observer/consensus_observer.rs b/consensus/src/consensus_observer/observer/consensus_observer.rs
@@ -1077,16 +1077,17 @@ impl ConsensusObserver {
             .start_epoch(
                 sk,
                 epoch_state.clone(),
-                dummy_signer.clone(),
+                dummy_signer,
                 payload_manager,
                 &consensus_config,
                 &execution_config,
                 &randomness_config,
-                None,
-                None,
+                None, // rand_config
+                None, // fast_rand_config
                 rand_msg_rx,
-                0,
-                self.pipeline_enabled(),
+                0, // highest_committed_round
+                self.observer_epoch_state.pipeline_enabled(),
+                None, // Consensus observer doesn't use virtual genesis
             )
             .await;
         if self.pipeline_enabled() {
diff --git a/consensus/src/epoch_manager.rs b/consensus/src/epoch_manager.rs
@@ -844,6 +844,13 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
 
         let safety_rules_container = Arc::new(Mutex::new(safety_rules));
 
+        // Extract virtual genesis block ID if consensus created one
+        let virtual_genesis_block_id = if recovery_data.commit_root_block().is_genesis_block() {
+            Some(recovery_data.commit_root_block().id())
+        } else {
+            None
+        };
+
         self.execution_client
             .start_epoch(
                 consensus_key.clone(),
@@ -858,6 +865,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
                 rand_msg_rx,
                 recovery_data.commit_root_block().round(),
                 self.config.enable_pipeline,
+                virtual_genesis_block_id,
             )
             .await;
         let consensus_sk = consensus_key;
@@ -1422,6 +1430,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
                 rand_msg_rx,
                 highest_committed_round,
                 self.config.enable_pipeline,
+                None, // DAG doesn't use virtual genesis
             )
             .await;
 
@@ -1479,7 +1488,7 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
     //TODO(l1-migration) turn on quorum store
     fn enable_quorum_store(&mut self, onchain_config: &OnChainConsensusConfig) -> bool {
         fail_point!("consensus::start_new_epoch::disable_qs", |_| false);
-        onchain_config.quorum_store_enabled() && false
+        onchain_config.quorum_store_enabled()
     }
 
     async fn process_message(
diff --git a/consensus/src/persistent_liveness_storage.rs b/consensus/src/persistent_liveness_storage.rs
@@ -110,6 +110,8 @@ impl LedgerRecoveryData {
         // We start from the block that storage's latest ledger info, if storage has end-epoch
         // LedgerInfo, we generate the virtual genesis block
         let ends_epoch = self.storage_ledger.ledger_info().ends_epoch();
+        //TODO(l1-migration): when blocks is empty and we have to create virtual genesis from storage to start consensus.
+        // This is only required for migration when we only have a single validator network
         let blocks_empty = blocks.is_empty();
         let (latest_commit_id, latest_ledger_info_sig) = if ends_epoch || blocks_empty {
             let genesis =
@@ -211,6 +213,7 @@ impl LedgerRecoveryData {
         // We start from the block that storage's latest ledger info, if storage has end-epoch
         // LedgerInfo, we generate the virtual genesis block
         let ends_epoch = self.storage_ledger.ledger_info().ends_epoch();
+        // TODO(l1-migration): This is for single validator network boostrap. We created virtual genesis to start the consensus
         let blocks_empty = blocks.is_empty();
         let (root_id, latest_ledger_info_sig) = if ends_epoch || blocks_empty {
             let genesis =
diff --git a/consensus/src/pipeline/execution_client.rs b/consensus/src/pipeline/execution_client.rs
@@ -72,6 +72,7 @@ pub trait TExecutionClient: Send + Sync {
         rand_msg_rx: aptos_channel::Receiver<AccountAddress, IncomingRandGenRequest>,
         highest_committed_round: Round,
         new_pipeline_enabled: bool,
+        virtual_genesis_block_id: Option<aptos_crypto::HashValue>,
     );
 
     /// This is needed for some DAG tests. Clean this up as a TODO.
@@ -325,8 +326,9 @@ impl TExecutionClient for ExecutionProxyClient {
         rand_msg_rx: aptos_channel::Receiver<AccountAddress, IncomingRandGenRequest>,
         highest_committed_round: Round,
         new_pipeline_enabled: bool,
+        virtual_genesis_block_id: Option<aptos_crypto::HashValue>,
     ) {
-        let maybe_rand_msg_tx = self.spawn_decoupled_execution(
+        self.spawn_decoupled_execution(
             maybe_consensus_key,
             commit_signer_provider,
             epoch_state.clone(),
@@ -357,9 +359,8 @@ impl TExecutionClient for ExecutionProxyClient {
             transaction_deduper,
             randomness_enabled,
             onchain_consensus_config.order_vote_enabled(),
+            virtual_genesis_block_id,
         );
-
-        maybe_rand_msg_tx
     }
 
     fn get_execution_channel(&self) -> Option<UnboundedSender<OrderedBlocks>> {
@@ -546,6 +547,7 @@ impl TExecutionClient for DummyExecutionClient {
         _rand_msg_rx: aptos_channel::Receiver<AccountAddress, IncomingRandGenRequest>,
         _highest_committed_round: Round,
         _new_pipeline_enabled: bool,
+        _virtual_genesis_block_id: Option<aptos_crypto::HashValue>,
     ) {
     }
 
diff --git a/consensus/src/state_computer.rs b/consensus/src/state_computer.rs
@@ -492,7 +492,17 @@ impl StateComputer for ExecutionProxy {
         transaction_deduper: Arc<dyn TransactionDeduper>,
         randomness_enabled: bool,
         order_vote_enabled: bool,
+        virtual_genesis_block_id: Option<HashValue>,
     ) {
+        // Reset the executor with the virtual genesis block ID if provided
+        if let Some(virtual_genesis_id) = virtual_genesis_block_id {
+            self.executor
+                .reset_with_virtual_genesis(Some(virtual_genesis_id))
+                .expect("Failed to reset executor with virtual genesis");
+        } else {
+            self.executor.reset().expect("Failed to reset executor");
+        }
+
         *self.state.write() = Some(MutableState {
             validators: epoch_state
                 .verifier
@@ -668,6 +678,7 @@ async fn test_commit_sync_race() {
         create_transaction_deduper(TransactionDeduperType::NoDedup),
         false,
         false,
+        None,
     );
     executor
         .commit(vec![], generate_li(1, 1), callback.clone())
diff --git a/consensus/src/state_computer_tests.rs b/consensus/src/state_computer_tests.rs
@@ -195,6 +195,7 @@ async fn should_see_and_notify_validator_txns() {
         Arc::new(NoOpDeduper {}),
         false,
         false,
+        None,
     );
 
     // Ensure the dummy executor has received the txns.
diff --git a/consensus/src/state_replication.rs b/consensus/src/state_replication.rs
@@ -73,6 +73,7 @@ pub trait StateComputer: Send + Sync {
         transaction_deduper: Arc<dyn TransactionDeduper>,
         randomness_enabled: bool,
         order_vote_enabled: bool,
+        virtual_genesis_block_id: Option<aptos_crypto::HashValue>,
     );
 
     // Reconfigure to clear epoch state at end of epoch.
diff --git a/consensus/src/test_utils/mock_execution_client.rs b/consensus/src/test_utils/mock_execution_client.rs
@@ -107,6 +107,7 @@ impl TExecutionClient for MockExecutionClient {
         _rand_msg_rx: aptos_channel::Receiver<AccountAddress, IncomingRandGenRequest>,
         _highest_committed_round: Round,
         _new_pipeline_enabled: bool,
+        _virtual_genesis_block_id: Option<aptos_crypto::HashValue>,
     ) {
     }
 
diff --git a/consensus/src/test_utils/mock_state_computer.rs b/consensus/src/test_utils/mock_state_computer.rs
@@ -91,6 +91,7 @@ impl StateComputer for EmptyStateComputer {
         _: Arc<dyn TransactionDeduper>,
         _: bool,
         _: bool,
+        _: Option<HashValue>,
     ) {
     }
 
@@ -179,6 +180,7 @@ impl StateComputer for RandomComputeResultStateComputer {
         _: Arc<dyn TransactionDeduper>,
         _: bool,
         _: bool,
+        _: Option<HashValue>,
     ) {
     }
 
diff --git a/execution/executor-types/src/lib.rs b/execution/executor-types/src/lib.rs
@@ -124,6 +124,15 @@ pub trait BlockExecutorTrait: Send + Sync {
     /// Reset the internal state including cache with newly fetched latest committed block from storage.
     fn reset(&self) -> Result<()>;
 
+    /// Reset with a virtual genesis block ID (used for consensus recovery)
+    fn reset_with_virtual_genesis(
+        &self,
+        _virtual_genesis_block_id: Option<HashValue>,
+    ) -> Result<()> {
+        // Default implementation just calls reset() for backward compatibility
+        self.reset()
+    }
+
     /// Executes a block - TBD, this API will be removed in favor of `execute_and_state_checkpoint`, followed
     /// by `ledger_update` once we have ledger update as a separate pipeline phase.
     #[cfg(any(test, feature = "fuzzing"))]
diff --git a/execution/executor/src/block_executor/block_tree/mod.rs b/execution/executor/src/block_executor/block_tree/mod.rs
@@ -18,13 +18,8 @@ use aptos_drop_helper::DEFAULT_DROPPER;
 use aptos_executor_types::ExecutorError;
 use aptos_infallible::Mutex;
 use aptos_logger::{debug, info};
-use aptos_storage_interface::{
-    state_store::state_view::db_state_view::LatestDbStateCheckpointView, DbReader,
-};
-use aptos_types::{
-    account_config::ChainIdResource, ledger_info::LedgerInfo, on_chain_config::OnChainConfig,
-    proof::definition::LeafCount,
-};
+use aptos_storage_interface::DbReader;
+use aptos_types::{ledger_info::LedgerInfo, proof::definition::LeafCount};
 use std::{
     collections::{hash_map::Entry, HashMap},
     sync::{mpsc::Receiver, Arc, Weak},
@@ -182,14 +177,25 @@ pub struct BlockTree {
 
 impl BlockTree {
     pub fn new(db: &Arc<dyn DbReader>) -> Result<Self> {
+        Self::new_with_virtual_genesis(db, None)
+    }
+
+    pub fn new_with_virtual_genesis(
+        db: &Arc<dyn DbReader>,
+        virtual_genesis_block_id: Option<HashValue>,
+    ) -> Result<Self> {
         let block_lookup = Arc::new(BlockLookup::new());
-        let root = Mutex::new(Self::root_from_db(&block_lookup, db)?);
+        let root = Mutex::new(Self::root_from_db(
+            &block_lookup,
+            db,
+            virtual_genesis_block_id,
+        )?);
 
         Ok(Self { root, block_lookup })
     }
 
     pub fn reset(&self, db: &Arc<dyn DbReader>) -> Result<()> {
-        *self.root.lock() = Self::root_from_db(&self.block_lookup, db)?;
+        *self.root.lock() = Self::root_from_db(&self.block_lookup, db, None)?;
         Ok(())
     }
 
@@ -209,22 +215,11 @@ impl BlockTree {
         self.block_lookup.multi_get(ids)
     }
 
-    fn is_new_virtual_genesis(db: &Arc<dyn DbReader>) -> Result<bool> {
-        let db_state_view = db
-            .latest_state_checkpoint_view()
-            .map_err(|err| anyhow!("[aptos-node] failed to create db state view {}", err))?;
-        let chain_id = ChainIdResource::fetch_config(&db_state_view)
-            .expect("[aptos-node] missing chain ID resource")
-            .chain_id();
-        let ledger_info_with_sigs = db.get_latest_ledger_info()?;
-        let cur_epoch = ledger_info_with_sigs.ledger_info().epoch();
-        if chain_id.is_movement_mainnet() && cur_epoch == 1848695 {
-            return Ok(true);
-        }
-        Ok(false)
-    }
-
-    fn root_from_db(block_lookup: &Arc<BlockLookup>, db: &Arc<dyn DbReader>) -> Result<Arc<Block>> {
+    fn root_from_db(
+        block_lookup: &Arc<BlockLookup>,
+        db: &Arc<dyn DbReader>,
+        virtual_genesis_block_id: Option<HashValue>,
+    ) -> Result<Arc<Block>> {
         let ledger_info_with_sigs = db.get_latest_ledger_info()?;
         let ledger_info = ledger_info_with_sigs.ledger_info();
         let ledger_summary = db.get_pre_committed_ledger_summary()?;
@@ -236,7 +231,14 @@ impl BlockTree {
             ledger_info.version(),
         );
 
-        let id = if ledger_info.ends_epoch() || Self::is_new_virtual_genesis(db)? {
+        let id = if let Some(virtual_genesis_id) = virtual_genesis_block_id {
+            // Use the virtual genesis block ID provided by consensus
+            info!(
+                "Using virtual genesis block ID from consensus: {:x}",
+                virtual_genesis_id
+            );
+            virtual_genesis_id
+        } else if ledger_info.ends_epoch() {
             epoch_genesis_block_id(ledger_info)
         } else {
             ledger_info.consensus_block_id()
diff --git a/execution/executor/src/block_executor/mod.rs b/execution/executor/src/block_executor/mod.rs
@@ -93,6 +93,19 @@ where
         Ok(())
     }
 
+    fn reset_with_virtual_genesis(
+        &self,
+        virtual_genesis_block_id: Option<HashValue>,
+    ) -> Result<()> {
+        let _guard = CONCURRENCY_GAUGE.concurrency_with(&["block", "reset"]);
+
+        *self.inner.write() = Some(BlockExecutorInner::new_with_virtual_genesis(
+            self.db.clone(),
+            virtual_genesis_block_id,
+        )?);
+        Ok(())
+    }
+
     fn execute_and_update_state(
         &self,
         block: ExecutableBlock,
@@ -162,7 +175,14 @@ where
     V: VMBlockExecutor,
 {
     pub fn new(db: DbReaderWriter) -> Result<Self> {
-        let block_tree = BlockTree::new(&db.reader)?;
+        Self::new_with_virtual_genesis(db, None)
+    }
+
+    pub fn new_with_virtual_genesis(
+        db: DbReaderWriter,
+        virtual_genesis_block_id: Option<HashValue>,
+    ) -> Result<Self> {
+        let block_tree = BlockTree::new_with_virtual_genesis(&db.reader, virtual_genesis_block_id)?;
         Ok(Self {
             db,
             block_tree,
diff --git a/tools/l1-migration/local-node-configs/validator_node.yaml b/tools/l1-migration/local-node-configs/validator_node.yaml
@@ -16,7 +16,7 @@ consensus:
       from_file:
         waypoint:
           from_file: /Users/bowu/data/.maptos/waypoint.txt # update to your path
-        identity_blob_path: /Users/bowu/projects/aptos-core/tools/l1-migration/operator_keys/mainnet/validator-identity.yaml # update to your path
+        identity_blob_path: /Users/bowu/data/.maptos/operator_keys/mainnet/validator-identity.yaml # update to your path
     
 execution:
   genesis_file_location: /Users/bowu/data/.maptos/genesis.blob # update to your path
@@ -29,7 +29,7 @@ validator_network:
   mutual_authentication: true
   identity:
     type: from_file
-    path: /Users/bowu/projects/aptos-core/tools/l1-migration/operator_keys/mainnet/validator-identity.yaml # update to your path
+    path: /Users/bowu/data/.maptos/operator_keys/mainnet/validator-identity.yaml # update to your path
   listen_address: /ip4/0.0.0.0/tcp/6180
 api:
   enabled: true

Original file line number	Diff line number	Diff line change
`@@ -363,7 +363,7 @@ impl Default for ConsensusConfig {`
`363`	`363`	`max_pending_rounds_in_commit_vote_cache: 100,`
`364`	`364`	`optimistic_sig_verification: true,`
`365`	`365`	`enable_round_timeout_msg: true,`
`366`		`- enable_pipeline: false,`
	`366`	`+ enable_pipeline: true,`
`367`	`367`	`}`
`368`	`368`	`}`
`369`	`369`	`}`
Original file line number	Diff line number	Diff line change
`@@ -580,7 +580,7 @@ impl PipelinedBlock {`
`580`	`580`	`.await`
`581`	`581`	`.map(\|(compute_result, execution_time, _)\| (compute_result, execution_time))`
`582`	`582`	`.map_err(\|e\| ExecutorError::InternalError {`
`583`		`- error: e.to_string()`
	`583`	`+ error: e.to_string(),`
`584`	`584`	`})`
`585`	`585`	`}`
`586`	`586`
Original file line number	Diff line number	Diff line change
`@@ -107,6 +107,7 @@ impl TExecutionClient for MockExecutionClient {`
`107`	`107`	`_rand_msg_rx: aptos_channel::Receiver<AccountAddress, IncomingRandGenRequest>,`
`108`	`108`	`_highest_committed_round: Round,`
`109`	`109`	`_new_pipeline_enabled: bool,`
	`110`	`+ _virtual_genesis_block_id: Option<aptos_crypto::HashValue>,`
`110`	`111`	`) {`
`111`	`112`	`}`
`112`	`113`