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

Author: guy-starkware

Cargo.lock

@@ -16,6 +16,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_infra.workspace = true
 apollo_mempool_config.workspace = true
 apollo_metrics.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_infra::component_definitions::{ComponentRequestHandler, ComponentStarter};
 use apollo_mempool_config::config::MempoolDynamicConfig;
 use apollo_node_config::node_config::NodeDynamicConfig;
@@ -38,6 +39,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_mempool_dynamic_config(&self) -> ConfigManagerResult<MempoolDynamicConfig> {
         Ok(self.latest_node_dynamic_config.mempool_dynamic_config.as_ref().unwrap().clone())
     }
@@ -54,6 +59,9 @@ impl ComponentRequestHandler<ConfigManagerRequest, ConfigManagerResponse> for Co
                     self.get_consensus_dynamic_config(),
                 )
             }
+            ConfigManagerRequest::GetContextDynamicConfig => {
+                ConfigManagerResponse::GetContextDynamicConfig(self.get_context_dynamic_config())
+            }
             ConfigManagerRequest::GetMempoolDynamicConfig => {
                 ConfigManagerResponse::GetMempoolDynamicConfig(self.get_mempool_dynamic_config())
             }

crates/apollo_config_manager/src/config_manager.rs

@@ -10,6 +10,7 @@ workspace = true
 
 [dependencies]
 apollo_consensus_config.workspace = true
+apollo_consensus_orchestrator_config.workspace = true
 apollo_infra.workspace = true
 apollo_mempool_config.workspace = true
 apollo_metrics.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_infra::component_client::{ClientError, LocalComponentClient, RemoteComponentClient};
 use apollo_infra::component_definitions::{ComponentClient, PrioritizedRequest, RequestWrapper};
 use apollo_infra::{impl_debug_for_infra_requests_and_responses, impl_labeled_request};
@@ -32,6 +33,8 @@ pub trait ConfigManagerClient: Send + Sync {
         &self,
     ) -> ConfigManagerClientResult<ConsensusDynamicConfig>;
 
+    async fn get_context_dynamic_config(&self) -> ConfigManagerClientResult<ContextDynamicConfig>;
+
     async fn get_mempool_dynamic_config(&self) -> ConfigManagerClientResult<MempoolDynamicConfig>;
 
     async fn set_node_dynamic_config(
@@ -48,6 +51,7 @@ pub trait ConfigManagerClient: Send + Sync {
 )]
 pub enum ConfigManagerRequest {
     GetConsensusDynamicConfig,
+    GetContextDynamicConfig,
     GetMempoolDynamicConfig,
     SetNodeDynamicConfig(NodeDynamicConfig),
 }
@@ -65,6 +69,7 @@ generate_permutation_labels! {
 #[derive(Clone, Serialize, Deserialize, AsRefStr)]
 pub enum ConfigManagerResponse {
     GetConsensusDynamicConfig(ConfigManagerResult<ConsensusDynamicConfig>),
+    GetContextDynamicConfig(ConfigManagerResult<ContextDynamicConfig>),
     GetMempoolDynamicConfig(ConfigManagerResult<MempoolDynamicConfig>),
     SetNodeDynamicConfig(ConfigManagerResult<()>),
 }
@@ -96,6 +101,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_mempool_dynamic_config(&self) -> ConfigManagerClientResult<MempoolDynamicConfig> {
         let request = ConfigManagerRequest::GetMempoolDynamicConfig;
         handle_all_response_variants!(

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

@@ -11,6 +11,7 @@ apollo_batcher.workspace = true
 apollo_batcher_types.workspace = true
 apollo_class_manager_types.workspace = true
 apollo_config.workspace = true
+apollo_config_manager_types.workspace = true
 apollo_consensus.workspace = true
 apollo_consensus_orchestrator_config.workspace = true
 apollo_infra_utils.workspace = true

crates/apollo_consensus_orchestrator/Cargo.toml

@@ -13,7 +13,7 @@ use apollo_class_manager_types::transaction_converter::{
     TransactionConverterError,
 };
 use apollo_consensus::types::{ProposalCommitment as ConsensusProposalCommitment, Round};
-use apollo_consensus_orchestrator_config::config::{ContextConfig, ContextDynamicConfig};
+use apollo_consensus_orchestrator_config::config::ContextConfig;
 use apollo_infra::component_client::ClientError;
 use apollo_protobuf::consensus::{ConsensusBlockInfo, ProposalInit, ProposalPart};
 use apollo_state_sync_types::communication::StateSyncClientError;

crates/apollo_consensus_orchestrator/src/build_proposal_test.rs

@@ -18,6 +18,7 @@ use apollo_batcher_types::batcher_types::{
 };
 use apollo_batcher_types::communication::BatcherClient;
 use apollo_class_manager_types::transaction_converter::TransactionConverterTrait;
+use apollo_config_manager_types::communication::SharedConfigManagerClient;
 use apollo_consensus::types::{
     ConsensusContext,
     ConsensusError,
@@ -177,6 +178,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>,
 }
 
 impl SequencerConsensusContext {
@@ -726,6 +728,10 @@ impl ConsensusContext for SequencerConsensusContext {
         assert!(round > self.current_round);
         self.interrupt_active_proposal().await;
         self.current_round = round;
+        if let Some(config_manager_client) = self.deps.config_manager_client.clone() {
+            self.config.dynamic_config =
+                config_manager_client.get_context_dynamic_config().await.unwrap();
+        }
         let mut to_process = None;
         while let Some(entry) = self.queued_proposals.first_entry() {
             match self.current_round.cmp(entry.key()) {

crates/apollo_consensus_orchestrator/src/sequencer_consensus_context.rs

@@ -107,6 +107,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,
         }
     }
 }