apollo_consensus_orchestrator: use the config manager client to update consensus context config

Author: guy-starkware

Cargo.lock

@@ -17,6 +17,7 @@ apollo_config.workspace = true
 apollo_config_manager_config.workspace = true
 apollo_config_manager_types.workspace = true
 apollo_consensus_config.workspace = true
+apollo_consensus_orchestrator_config.workspace = true
 apollo_http_server_config.workspace = true
 apollo_infra.workspace = true
 apollo_mempool_config.workspace = true

crates/apollo_config_manager/Cargo.toml

@@ -2,6 +2,7 @@ use apollo_config_manager_config::config::ConfigManagerConfig;
 use apollo_config_manager_types::communication::{ConfigManagerRequest, ConfigManagerResponse};
 use apollo_config_manager_types::config_manager_types::ConfigManagerResult;
 use apollo_consensus_config::config::ConsensusDynamicConfig;
+use apollo_consensus_orchestrator_config::config::ContextDynamicConfig;
 use apollo_http_server_config::config::HttpServerDynamicConfig;
 use apollo_infra::component_definitions::{ComponentRequestHandler, ComponentStarter};
 use apollo_mempool_config::config::MempoolDynamicConfig;
@@ -39,6 +40,10 @@ impl ConfigManager {
         Ok(self.latest_node_dynamic_config.consensus_dynamic_config.as_ref().unwrap().clone())
     }
 
+    pub(crate) fn get_context_dynamic_config(&self) -> ConfigManagerResult<ContextDynamicConfig> {
+        Ok(self.latest_node_dynamic_config.context_dynamic_config.as_ref().unwrap().clone())
+    }
+
     pub(crate) fn get_http_server_dynamic_config(
         &self,
     ) -> ConfigManagerResult<HttpServerDynamicConfig> {
@@ -61,6 +66,9 @@ impl ComponentRequestHandler<ConfigManagerRequest, ConfigManagerResponse> for Co
                     self.get_consensus_dynamic_config(),
                 )
             }
+            ConfigManagerRequest::GetContextDynamicConfig => {
+                ConfigManagerResponse::GetContextDynamicConfig(self.get_context_dynamic_config())
+            }
             ConfigManagerRequest::GetHttpServerDynamicConfig => {
                 ConfigManagerResponse::GetHttpServerDynamicConfig(
                     self.get_http_server_dynamic_config(),

crates/apollo_config_manager/src/config_manager.rs

@@ -11,6 +11,7 @@ workspace = true
 
 [dependencies]
 apollo_consensus_config.workspace = true
+apollo_consensus_orchestrator_config.workspace = true
 apollo_http_server_config.workspace = true
 apollo_infra.workspace = true
 apollo_mempool_config.workspace = true

crates/apollo_config_manager_types/Cargo.toml

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use apollo_consensus_config::config::ConsensusDynamicConfig;
+use apollo_consensus_orchestrator_config::config::ContextDynamicConfig;
 use apollo_http_server_config::config::HttpServerDynamicConfig;
 use apollo_infra::component_client::{ClientError, LocalComponentClient, RemoteComponentClient};
 use apollo_infra::component_definitions::{ComponentClient, PrioritizedRequest, RequestWrapper};
@@ -33,6 +34,7 @@ pub trait ConfigManagerClient: Send + Sync {
         &self,
     ) -> ConfigManagerClientResult<ConsensusDynamicConfig>;
 
+    async fn get_context_dynamic_config(&self) -> ConfigManagerClientResult<ContextDynamicConfig>;
     async fn get_http_server_dynamic_config(
         &self,
     ) -> ConfigManagerClientResult<HttpServerDynamicConfig>;
@@ -53,6 +55,7 @@ pub trait ConfigManagerClient: Send + Sync {
 )]
 pub enum ConfigManagerRequest {
     GetConsensusDynamicConfig,
+    GetContextDynamicConfig,
     GetHttpServerDynamicConfig,
     GetMempoolDynamicConfig,
     SetNodeDynamicConfig(NodeDynamicConfig),
@@ -71,6 +74,7 @@ generate_permutation_labels! {
 #[derive(Clone, Serialize, Deserialize, AsRefStr)]
 pub enum ConfigManagerResponse {
     GetConsensusDynamicConfig(ConfigManagerResult<ConsensusDynamicConfig>),
+    GetContextDynamicConfig(ConfigManagerResult<ContextDynamicConfig>),
     GetHttpServerDynamicConfig(ConfigManagerResult<HttpServerDynamicConfig>),
     GetMempoolDynamicConfig(ConfigManagerResult<MempoolDynamicConfig>),
     SetNodeDynamicConfig(ConfigManagerResult<()>),
@@ -103,6 +107,17 @@ where
         )
     }
 
+    async fn get_context_dynamic_config(&self) -> ConfigManagerClientResult<ContextDynamicConfig> {
+        let request = ConfigManagerRequest::GetContextDynamicConfig;
+        handle_all_response_variants!(
+            ConfigManagerResponse,
+            GetContextDynamicConfig,
+            ConfigManagerClientError,
+            ConfigManagerError,
+            Direct
+        )
+    }
+
     async fn get_http_server_dynamic_config(
         &self,
     ) -> ConfigManagerClientResult<HttpServerDynamicConfig> {

crates/apollo_config_manager_types/src/communication.rs

@@ -144,6 +144,7 @@ impl ConsensusManager {
         let consensus_context = self.create_sequencer_consensus_context(
             &votes_broadcast_channels,
             outbound_internal_sender,
+            self.config_manager_client.clone(),
         );
 
         let current_height =
@@ -264,6 +265,7 @@ impl ConsensusManager {
         &self,
         votes_broadcast_channels: &BroadcastTopicChannels<Vote>,
         outbound_internal_sender: mpsc::Sender<(HeightAndRound, mpsc::Receiver<ProposalPart>)>,
+        config_manager_client: SharedConfigManagerClient,
     ) -> SequencerConsensusContext {
         SequencerConsensusContext::new(
             self.config.context_config.clone(),
@@ -282,6 +284,7 @@ impl ConsensusManager {
                 clock: Arc::new(DefaultClock),
                 outbound_proposal_sender: outbound_internal_sender,
                 vote_broadcast_client: votes_broadcast_channels.broadcast_topic_client.clone(),
+                config_manager_client: Some(Arc::clone(&config_manager_client)),
             },
         )
     }

crates/apollo_consensus_manager/src/consensus_manager.rs

@@ -10,6 +10,7 @@ description = "Implements the consensus context and orchestrates the node's comp
 apollo_batcher.workspace = true
 apollo_batcher_types.workspace = true
 apollo_class_manager_types.workspace = true
+apollo_config_manager_types.workspace = true
 apollo_consensus.workspace = true
 apollo_consensus_orchestrator_config.workspace = true
 apollo_infra_utils.workspace = true
@@ -50,6 +51,7 @@ url = { workspace = true, features = ["serde"] }
 apollo_batcher.workspace = true
 apollo_batcher_types = { workspace = true, features = ["testing"] }
 apollo_class_manager_types = { workspace = true, features = ["testing"] }
+apollo_config_manager_types = { workspace = true, features = ["testing"] }
 apollo_infra = { workspace = true, features = ["testing"] }
 apollo_infra_utils = { workspace = true, features = ["testing"] }
 apollo_l1_gas_price_types = { workspace = true, features = ["testing"] }

crates/apollo_consensus_orchestrator/Cargo.toml

@@ -21,6 +21,7 @@ use apollo_class_manager_types::transaction_converter::{
     TransactionConverterError,
     TransactionConverterTrait,
 };
+use apollo_config_manager_types::communication::SharedConfigManagerClient;
 use apollo_consensus::types::{
     ConsensusContext,
     ConsensusError,
@@ -185,6 +186,7 @@ pub struct SequencerConsensusContextDeps {
     pub outbound_proposal_sender: mpsc::Sender<(HeightAndRound, mpsc::Receiver<ProposalPart>)>,
     // Used to broadcast votes to other consensus nodes.
     pub vote_broadcast_client: BroadcastTopicClient<Vote>,
+    pub config_manager_client: Option<SharedConfigManagerClient>,
 }
 
 #[derive(thiserror::Error, PartialEq, Debug)]
@@ -785,6 +787,7 @@ impl ConsensusContext for SequencerConsensusContext {
         round: Round,
     ) -> Result<(), ConsensusError> {
         if self.current_height.map(|h| height > h).unwrap_or(true) {
+            self.update_dynamic_config().await;
             self.current_height = Some(height);
             assert_eq!(round, 0);
             self.current_round = round;
@@ -795,13 +798,26 @@ impl ConsensusContext for SequencerConsensusContext {
             // return to the old height.
             return Ok(self.deps.batcher.start_height(StartHeightInput { height }).await?);
         }
-        assert_eq!(Some(height), self.current_height);
+        assert_eq!(
+            Some(height),
+            self.current_height,
+            "height {} is not equal to current height {:?}",
+            height,
+            self.current_height
+        );
         if round == self.current_round {
             return Ok(());
         }
-        assert!(round > self.current_round);
+        assert!(
+            round > self.current_round,
+            "round {} is not greater than current round {}",
+            round,
+            self.current_round
+        );
         self.interrupt_active_proposal().await;
         self.current_round = round;
+        self.update_dynamic_config().await;
+
         let mut to_process = None;
         while let Some(entry) = self.queued_proposals.first_entry() {
             match self.current_round.cmp(entry.key()) {
@@ -903,6 +919,23 @@ impl SequencerConsensusContext {
             }
         }
     }
+
+    async fn update_dynamic_config(&mut self) {
+        if let Some(config_manager_client) = self.deps.config_manager_client.clone() {
+            let config_result = config_manager_client.get_context_dynamic_config().await;
+            match config_result {
+                Ok(config) => {
+                    self.config.dynamic_config = config;
+                }
+                Err(e) => {
+                    error!(
+                        "Failed to get dynamic config for consensus context. Config not updated. \
+                         Error: {e:?}"
+                    );
+                }
+            }
+        }
+    }
 }
 
 async fn validate_and_send(

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use apollo_batcher_types::batcher_types::{CentralObjects, DecisionReachedResponse};
 use apollo_batcher_types::communication::BatcherClientError;
 use apollo_batcher_types::errors::BatcherError;
+use apollo_config_manager_types::communication::MockConfigManagerClient;
 use apollo_consensus::types::{ConsensusContext, Round};
 use apollo_consensus_orchestrator_config::config::{ContextConfig, ContextDynamicConfig};
 use apollo_infra::component_client::ClientError;
@@ -1042,3 +1043,128 @@ async fn override_prices_behavior(
         );
     }
 }
+
+#[tokio::test]
+async fn change_gas_price_overrides() {
+    let (mut deps, mut network) = create_test_and_network_deps();
+
+    // Validate two blocks, between the first and the second we will change the gas price overrides.
+    // After the second block we will do another round with another dynamic config change.
+    deps.setup_deps_for_validate(SetupDepsArgs { number_of_times: 2, ..Default::default() });
+    deps.setup_deps_for_validate(SetupDepsArgs {
+        number_of_times: 1,
+        expect_start_height: false,
+        start_block_number: BlockNumber(2),
+        ..Default::default()
+    });
+    deps.setup_deps_for_build(SetupDepsArgs {
+        start_block_number: BlockNumber(2),
+        ..Default::default()
+    });
+
+    // set up batcher decision_reached
+    deps.batcher.expect_decision_reached().times(2).returning(|_| {
+        Ok(DecisionReachedResponse {
+            state_diff: ThinStateDiff::default(),
+            l2_gas_used: GasAmount::default(),
+            central_objects: CentralObjects::default(),
+            block_header_commitments: BlockHeaderCommitments::default(),
+        })
+    });
+
+    // required for decision reached flow
+    deps.state_sync_client.expect_add_new_block().times(2).returning(|_| Ok(()));
+    // Mock sync to never provide any blocks in this test.
+    deps.state_sync_client.expect_get_latest_block_number().returning(|| Ok(None));
+    deps.cende_ambassador.expect_prepare_blob_for_next_height().times(2).returning(|_| Ok(()));
+
+    let mut context = deps.build_context();
+
+    // Validate block number 0.
+    context.set_height_and_round(BlockNumber(0), 0).await.unwrap();
+
+    let content_receiver =
+        send_proposal_to_validator_context(&mut context, block_info(BlockNumber(0))).await;
+    let fin_receiver =
+        context.validate_proposal(ProposalInit::default(), TIMEOUT, content_receiver).await;
+
+    let proposal_commitment = fin_receiver.await.unwrap();
+    assert_eq!(proposal_commitment.0, STATE_DIFF_COMMITMENT.0.0);
+
+    context.decision_reached(BlockNumber(0), proposal_commitment).await.unwrap();
+
+    let new_dynamic_config = ContextDynamicConfig {
+        override_l2_gas_price_fri: Some(ODDLY_SPECIFIC_L2_GAS_PRICE),
+        ..Default::default()
+    };
+    let config_manager_client = make_config_manager_client(new_dynamic_config);
+    context.deps.config_manager_client = Some(Arc::new(config_manager_client));
+
+    // Modify the incoming block info to make sure it matches the overrides. If we don't, the
+    // proposal will fail validation.
+    let mut modified_block_info = block_info(BlockNumber(1));
+    modified_block_info.l2_gas_price_fri = GasPrice(ODDLY_SPECIFIC_L2_GAS_PRICE);
+
+    // Validate block number 1, round 0.
+    context.set_height_and_round(BlockNumber(1), 0).await.unwrap();
+    let init = ProposalInit { height: BlockNumber(1), ..Default::default() };
+
+    let content_receiver =
+        send_proposal_to_validator_context(&mut context, modified_block_info).await;
+    let fin_receiver = context.validate_proposal(init, TIMEOUT, content_receiver).await;
+    let proposal_commitment = fin_receiver.await.unwrap();
+    assert_eq!(proposal_commitment.0, STATE_DIFF_COMMITMENT.0.0);
+
+    // Validate block number 1, round 1.
+    let new_dynamic_config = ContextDynamicConfig {
+        min_l1_data_gas_price_wei: ODDLY_SPECIFIC_L1_DATA_GAS_PRICE,
+        ..Default::default()
+    };
+    let config_manager_client = make_config_manager_client(new_dynamic_config);
+    context.deps.config_manager_client = Some(Arc::new(config_manager_client));
+
+    let mut modified_block_info = block_info(BlockNumber(1));
+    modified_block_info.l1_data_gas_price_wei = GasPrice(ODDLY_SPECIFIC_L1_DATA_GAS_PRICE);
+
+    context.set_height_and_round(BlockNumber(1), 1).await.unwrap();
+    let init = ProposalInit { height: BlockNumber(1), round: 1, ..Default::default() };
+
+    let content_receiver =
+        send_proposal_to_validator_context(&mut context, modified_block_info).await;
+    let fin_receiver = context.validate_proposal(init, TIMEOUT, content_receiver).await;
+    let proposal_commitment = fin_receiver.await.unwrap();
+    assert_eq!(proposal_commitment.0, STATE_DIFF_COMMITMENT.0.0);
+
+    context.decision_reached(BlockNumber(1), proposal_commitment).await.unwrap();
+
+    // Now build a proposal for height 2.
+    let new_dynamic_config = ContextDynamicConfig {
+        override_eth_to_fri_rate: Some(ODDLY_SPECIFIC_CONVERSION_RATE),
+        ..Default::default()
+    };
+    let config_manager_client = make_config_manager_client(new_dynamic_config);
+    context.deps.config_manager_client = Some(Arc::new(config_manager_client));
+
+    let init = ProposalInit { height: BlockNumber(2), ..Default::default() };
+
+    let fin_receiver = context.build_proposal(init, TIMEOUT).await.unwrap().await.unwrap();
+
+    assert_eq!(fin_receiver.0, STATE_DIFF_COMMITMENT.0.0);
+    let (_, mut receiver) = network.outbound_proposal_receiver.next().await.unwrap();
+
+    assert_eq!(receiver.next().await.unwrap(), init.into());
+    let info = receiver.next().await.unwrap();
+    let ProposalPart::BlockInfo(info) = info else {
+        panic!("Expected ProposalPart::BlockInfo");
+    };
+    assert_eq!(info.eth_to_fri_rate, ODDLY_SPECIFIC_CONVERSION_RATE);
+}
+
+fn make_config_manager_client(provider_config: ContextDynamicConfig) -> MockConfigManagerClient {
+    let mut config_manager_client = MockConfigManagerClient::new();
+    config_manager_client
+        .expect_get_context_dynamic_config()
+        .returning(move || Ok(provider_config.clone()));
+    config_manager_client.expect_set_node_dynamic_config().returning(|_| Ok(()));
+    config_manager_client
+}

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context_test.rs

@@ -117,6 +117,7 @@ impl From<TestDeps> for SequencerConsensusContextDeps {
             clock: deps.clock,
             outbound_proposal_sender: deps.outbound_proposal_sender,
             vote_broadcast_client: deps.vote_broadcast_client,
+            config_manager_client: None,
         }
     }
 }