apollo_node_config: batcher support for validation only node config. (#13625)

asaf-sw · web-flow · commit 53b6f0a6f678 · 2026-04-05T06:15:01.000Z
diff --git a/crates/apollo_batcher/src/batcher.rs b/crates/apollo_batcher/src/batcher.rs
@@ -157,7 +157,7 @@ pub struct Batcher {
     pub storage_writer: Box<dyn BatcherStorageWriter>,
     pub committer_client: SharedCommitterClient,
     pub l1_events_provider_client: SharedL1EventsProviderClient,
-    pub mempool_client: SharedMempoolClient,
+    pub mempool_client: Option<SharedMempoolClient>,
     pub config_manager_client: SharedConfigManagerClient,
 
     /// Used to create block builders.
@@ -212,13 +212,19 @@ impl Batcher {
         storage_writer: Box<dyn BatcherStorageWriter>,
         committer_client: SharedCommitterClient,
         l1_events_provider_client: SharedL1EventsProviderClient,
-        mempool_client: SharedMempoolClient,
+        mempool_client: Option<SharedMempoolClient>,
         config_manager_client: SharedConfigManagerClient,
         block_builder_factory: Box<dyn BlockBuilderFactoryTrait>,
         pre_confirmed_block_writer_factory: Box<dyn PreconfirmedBlockWriterFactoryTrait>,
         commitment_manager: ApolloCommitmentManager,
         storage_reader_server_handle: AbortHandle,
     ) -> Self {
+        assert!(
+            config.static_config.validation_only == mempool_client.is_none(),
+            "validation_only={} but mempool_client is {}",
+            config.static_config.validation_only,
+            if mempool_client.is_none() { "None" } else { "Some" }
+        );
         Self {
             config,
             storage_reader,
@@ -276,6 +282,9 @@ impl Batcher {
         &mut self,
         propose_block_input: ProposeBlockInput,
     ) -> BatcherResult<()> {
+        if self.config.static_config.validation_only {
+            return Err(BatcherError::ProposingNotSupported);
+        }
         let block_number = propose_block_input.block_info.block_number;
         let proposal_metrics_handle = ProposalMetricsHandle::new();
         let active_height = self.active_height.ok_or(BatcherError::NoActiveHeight)?;
@@ -285,12 +294,17 @@ impl Batcher {
             propose_block_input.retrospective_block_hash,
         )?;
 
+        let mempool_client = self
+            .mempool_client
+            .as_ref()
+            .expect("Mempool client must be present in non-validation-only mode.");
+
         // TODO(yair): extract function for the following calls, use join_all.
         info!(
             "Notifying the mempool we start to work on block {}, round {}.",
             block_number, propose_block_input.proposal_round
         );
-        self.mempool_client.commit_block(CommitBlockArgs::default()).await.map_err(|err| {
+        mempool_client.commit_block(CommitBlockArgs::default()).await.map_err(|err| {
             error!(
                 "Mempool is not ready to start proposal {}: {}.",
                 propose_block_input.proposal_id, err
@@ -301,7 +315,7 @@ impl Batcher {
             "Updating gas price for block {}, round {} in Mempool client",
             block_number, propose_block_input.proposal_round
         );
-        self.mempool_client
+        mempool_client
             .update_gas_price(
                 propose_block_input.block_info.gas_prices.strk_gas_prices.l2_gas_price.get(),
             )
@@ -333,7 +347,7 @@ impl Batcher {
             TxProviderPhase::Mempool
         };
         let tx_provider = ProposeTransactionProvider::new(
-            self.mempool_client.clone(),
+            mempool_client.clone(),
             self.l1_events_provider_client.clone(),
             self.config.static_config.max_l1_handler_txs_per_block_proposal,
             propose_block_input.block_info.block_number,
@@ -613,7 +627,11 @@ impl Batcher {
 
     #[instrument(skip(self), err)]
     pub async fn get_batch_timestamp(&self) -> BatcherResult<UnixTimestamp> {
-        self.mempool_client.resolve_batch_timestamp().await.map_err(|err| {
+        let mempool_client = self.mempool_client.as_ref().expect(
+            "Mempool client must be present in non-validation-only mode. Unreachable code when \
+             validation-only mode is enabled.",
+        );
+        mempool_client.resolve_batch_timestamp().await.map_err(|err| {
             error!("Failed to get timestamp from mempool: {err}");
             BatcherError::InternalError
         })
@@ -933,16 +951,16 @@ impl Batcher {
             BATCHER_L1_EVENTS_PROVIDER_ERRORS.increment(1);
         }
 
-        // Notify the mempool of the new block.
-        let mempool_result = self
-            .mempool_client
-            .commit_block(CommitBlockArgs { address_to_nonce, rejected_tx_hashes })
-            .await;
-
-        if let Err(mempool_err) = mempool_result {
-            // Recoverable error, mempool won't be updated with the new block.
-            error!("Failed to commit block to mempool: {}", mempool_err);
-        };
+        // Notify the mempool of the new block (skipped in validation-only mode).
+        if let Some(mempool_client) = &self.mempool_client {
+            let mempool_result = mempool_client
+                .commit_block(CommitBlockArgs { address_to_nonce, rejected_tx_hashes })
+                .await;
+            if let Err(mempool_err) = mempool_result {
+                // Recoverable error, mempool won't be updated with the new block.
+                error!("Failed to commit block to mempool: {}", mempool_err);
+            }
+        }
 
         BUILDING_HEIGHT.increment(1);
         Ok(())
@@ -1412,7 +1430,7 @@ impl DynamicConfigProvider for BatcherDynamicConfigProvider {
 pub async fn create_batcher(
     config: BatcherConfig,
     committer_client: SharedCommitterClient,
-    mempool_client: SharedMempoolClient,
+    mempool_client: Option<SharedMempoolClient>,
     l1_events_provider_client: SharedL1EventsProviderClient,
     class_manager_client: SharedClassManagerClient,
     pre_confirmed_cende_client: Arc<dyn PreconfirmedCendeClientTrait>,
diff --git a/crates/apollo_batcher/src/batcher_test.rs b/crates/apollo_batcher/src/batcher_test.rs
@@ -33,7 +33,11 @@ use apollo_infra::component_client::ClientError;
 use apollo_infra::component_definitions::ComponentStarter;
 use apollo_l1_events_types::errors::{L1EventsProviderClientError, L1EventsProviderError};
 use apollo_l1_events_types::{MockL1EventsProviderClient, SessionState};
-use apollo_mempool_types::communication::{MempoolClientError, MockMempoolClient};
+use apollo_mempool_types::communication::{
+    MempoolClientError,
+    MockMempoolClient,
+    SharedMempoolClient,
+};
 use apollo_mempool_types::mempool_types::CommitBlockArgs;
 use apollo_state_sync_types::state_sync_types::SyncBlock;
 use apollo_storage::db::DbError;
@@ -256,6 +260,11 @@ async fn create_batcher_impl<R: BatcherStorageReader + 'static>(
     clients: MockClients,
     config: BatcherConfig,
 ) -> Batcher {
+    let mempool_client: Option<SharedMempoolClient> = if config.static_config.validation_only {
+        None
+    } else {
+        Some(Arc::new(clients.mempool_client))
+    };
     let committer_client = Arc::new(clients.committer_client);
     let commitment_manager = CommitmentManager::create_commitment_manager(
         &config.static_config.commitment_manager_config,
@@ -275,7 +284,7 @@ async fn create_batcher_impl<R: BatcherStorageReader + 'static>(
         storage_writer,
         committer_client,
         Arc::new(clients.l1_provider_client),
-        Arc::new(clients.mempool_client),
+        mempool_client,
         Arc::new(mock_config_manager),
         Box::new(clients.block_builder_factory),
         Box::new(clients.pre_confirmed_block_writer_factory),
@@ -291,6 +300,39 @@ fn abort_signal_sender() -> AbortSignalSender {
     tokio::sync::oneshot::channel().0
 }
 
+/// Calls `Batcher::new` with an explicit `mempool_client`, bypassing the auto-derivation in
+/// `create_batcher_impl`. Used to test the consistency assert in `Batcher::new`.
+async fn new_batcher_with_mempool_override(
+    deps: MockDependencies,
+    mempool_client: Option<SharedMempoolClient>,
+) {
+    let storage_reader = Arc::new(deps.storage_reader);
+    let committer_client = Arc::new(deps.clients.committer_client);
+    let commitment_manager = CommitmentManager::create_commitment_manager(
+        &deps.batcher_config.static_config.commitment_manager_config,
+        storage_reader.clone(),
+        committer_client.clone(),
+    )
+    .await;
+    let mut mock_config_manager = MockConfigManagerClient::new();
+    mock_config_manager
+        .expect_get_batcher_dynamic_config()
+        .returning(|| Ok(BatcherDynamicConfig::default()));
+    Batcher::new(
+        deps.batcher_config,
+        storage_reader,
+        Box::new(deps.storage_writer),
+        committer_client,
+        Arc::new(deps.clients.l1_provider_client),
+        mempool_client,
+        Arc::new(mock_config_manager),
+        Box::new(deps.clients.block_builder_factory),
+        Box::new(deps.clients.pre_confirmed_block_writer_factory),
+        commitment_manager,
+        tokio::spawn(async {}).abort_handle(),
+    );
+}
+
 async fn batcher_propose_and_commit_block(
     mock_dependencies: MockDependencies,
 ) -> Result<DecisionReachedResponse, BatcherError> {
@@ -1840,3 +1882,90 @@ async fn test_reversed_state_diff() {
         reversed_state_diff.nonces,
     );
 }
+
+fn validation_only_mock_dependencies() -> MockDependencies {
+    let mut deps = MockDependencies::default();
+    deps.batcher_config.static_config.validation_only = true;
+    deps
+}
+
+#[tokio::test]
+async fn validation_only_propose_block_returns_not_supported() {
+    let mut batcher = create_batcher(validation_only_mock_dependencies()).await;
+    batcher.start_height(StartHeightInput { height: INITIAL_HEIGHT }).await.unwrap();
+
+    let result = batcher.propose_block(propose_block_input(PROPOSAL_ID)).await;
+
+    assert_eq!(result, Err(BatcherError::ProposingNotSupported));
+}
+
+#[tokio::test]
+#[should_panic(expected = "Mempool client must be present in non-validation-only mode.")]
+async fn validation_only_get_batch_timestamp_panics() {
+    let batcher = create_batcher(validation_only_mock_dependencies()).await;
+    batcher.get_batch_timestamp().await.unwrap();
+}
+
+#[tokio::test]
+async fn validation_only_validate_block_succeeds() {
+    let mut mock_deps = validation_only_mock_dependencies();
+    let mut block_builder_factory = MockBlockBuilderFactoryTrait::new();
+    mock_create_builder_for_validate_block(
+        &mut block_builder_factory,
+        Ok(BlockExecutionArtifacts::create_for_testing().await),
+    );
+    mock_deps.clients.block_builder_factory = block_builder_factory;
+    mock_deps.clients.l1_provider_client.expect_start_block().returning(|_, _| Ok(()));
+
+    let mut batcher = create_batcher(mock_deps).await;
+    batcher.start_height(StartHeightInput { height: INITIAL_HEIGHT }).await.unwrap();
+
+    batcher.validate_block(validate_block_input(PROPOSAL_ID)).await.unwrap();
+
+    let finish_proposal = FinishProposalInput {
+        proposal_id: PROPOSAL_ID,
+        final_n_executed_txs: DUMMY_FINAL_N_EXECUTED_TXS,
+    };
+    let result = batcher.finish_proposal(finish_proposal).await.unwrap();
+    assert_matches!(result, FinishProposalStatus::Finished(_));
+}
+
+#[tokio::test]
+async fn validation_only_decision_reached_skips_mempool_notification() {
+    let mut mock_deps = validation_only_mock_dependencies();
+
+    // The mempool_client on MockClients still exists but must not be called.
+    mock_deps.clients.mempool_client.checkpoint();
+
+    mock_deps.clients.l1_provider_client.expect_start_block().returning(|_, _| Ok(()));
+    mock_deps.clients.l1_provider_client.expect_commit_block().times(1).returning(|_, _, _| Ok(()));
+    mock_deps.storage_writer.expect_commit_proposal().returning(|_, _, _| Ok(()));
+
+    let mut block_builder_factory = MockBlockBuilderFactoryTrait::new();
+    mock_create_builder_for_validate_block(
+        &mut block_builder_factory,
+        Ok(BlockExecutionArtifacts::create_for_testing().await),
+    );
+    mock_deps.clients.block_builder_factory = block_builder_factory;
+
+    let mut batcher = create_batcher(mock_deps).await;
+    batcher.start_height(StartHeightInput { height: INITIAL_HEIGHT }).await.unwrap();
+    batcher.validate_block(validate_block_input(PROPOSAL_ID)).await.unwrap();
+    batcher.await_active_proposal(DUMMY_FINAL_N_EXECUTED_TXS).await.unwrap();
+
+    // decision_reached must succeed and not call mempool_client.commit_block.
+    batcher.decision_reached(DecisionReachedInput { proposal_id: PROPOSAL_ID }).await.unwrap();
+}
+
+#[tokio::test]
+#[should_panic(expected = "validation_only=false but mempool_client is None")]
+async fn validation_only_flag_false_with_no_mempool_panics() {
+    new_batcher_with_mempool_override(MockDependencies::default(), None).await;
+}
+
+#[tokio::test]
+#[should_panic(expected = "validation_only=true but mempool_client is Some")]
+async fn validation_only_flag_true_with_mempool_panics() {
+    let mempool: Option<SharedMempoolClient> = Some(Arc::new(MockMempoolClient::new()));
+    new_batcher_with_mempool_override(validation_only_mock_dependencies(), mempool).await;
+}
diff --git a/crates/apollo_batcher_types/src/errors.rs b/crates/apollo_batcher_types/src/errors.rs
@@ -45,6 +45,8 @@ pub enum BatcherError {
     ProposalAlreadyFinished { proposal_id: ProposalId },
     #[error("Proposal failed.")]
     ProposalFailed,
+    #[error("Proposing is not supported in validation-only mode.")]
+    ProposingNotSupported,
     #[error("Proposal with ID {proposal_id} not found.")]
     ProposalNotFound { proposal_id: ProposalId },
     #[error(
diff --git a/crates/apollo_integration_tests/src/utils.rs b/crates/apollo_integration_tests/src/utils.rs
@@ -363,6 +363,7 @@ pub fn create_node_config(
     let state_sync_config = wrap_if_component_config_expected!(state_sync, state_sync_config);
 
     let sequencer_node_config = SequencerNodeConfig {
+        // TODO(asaf): Add a validation-only node to the integration test.
         validation_only: false,
         base_layer_config,
         batcher_config,
diff --git a/crates/apollo_node/src/components.rs b/crates/apollo_node/src/components.rs
@@ -86,8 +86,15 @@ pub async fn create_node_components(
             let committer_client = clients
                 .get_committer_shared_client()
                 .expect("Committer client should be available");
-            let mempool_client =
-                clients.get_mempool_shared_client().expect("Mempool client should be available");
+            let mempool_client = if config.validation_only {
+                None
+            } else {
+                Some(
+                    clients
+                        .get_mempool_shared_client()
+                        .expect("Mempool client should be available in non-validation-only mode."),
+                )
+            };
             let l1_events_provider_client = clients
                 .get_l1_events_provider_shared_client()
                 .expect("L1 Provider client should be available");
