apollo_http_server: add dynamic config poller service

Author: Itay-Tsabary-Starkware

crates/apollo_http_server/src/http_server.rs

@@ -1,6 +1,7 @@
 use std::clone::Clone;
 use std::net::SocketAddr;
 use std::string::String;
+use std::time::Duration;
 
 use apollo_config_manager_types::communication::SharedConfigManagerClient;
 use apollo_gateway_types::communication::{GatewayClientError, SharedGatewayClient};
@@ -15,19 +16,20 @@ use apollo_gateway_types::gateway_types::{
     GatewayOutput,
     SUPPORTED_TRANSACTION_VERSIONS,
 };
-use apollo_http_server_config::config::HttpServerConfig;
+use apollo_http_server_config::config::{HttpServerConfig, HttpServerDynamicConfig};
 use apollo_infra::component_definitions::ComponentStarter;
 use apollo_infra_utils::type_name::short_type_name;
 use apollo_proc_macros::sequencer_latency_histogram;
-use axum::extract::State;
 use axum::http::HeaderMap;
 use axum::routing::{get, post};
-use axum::{async_trait, Json, Router};
+use axum::{async_trait, Extension, Json, Router};
 use blockifier_reexecution::serde_utils::deserialize_transaction_json_to_starknet_api_tx;
 use serde::de::Error;
 use starknet_api::rpc_transaction::RpcTransaction;
 use starknet_api::serde_utils::bytes_from_hex_str;
 use starknet_api::transaction::fields::ValidResourceBounds;
+use tokio::sync::watch::{channel, Receiver, Sender};
+use tokio::time;
 use tracing::{debug, info, instrument, warn};
 
 use crate::deprecated_gateway_transaction::DeprecatedGatewayTransactionV3;
@@ -51,14 +53,27 @@ pub type HttpServerResult<T> = Result<T, HttpServerError>;
 const CLIENT_REGION_HEADER: &str = "X-Client-Region";
 
 pub struct HttpServer {
-    pub config: HttpServerConfig,
+    config: HttpServerConfig,
     app_state: AppState,
+    config_manager_client: SharedConfigManagerClient,
+    dynamic_config_tx: Sender<HttpServerDynamicConfig>,
 }
 
 #[derive(Clone)]
 pub struct AppState {
-    config_manager_client: SharedConfigManagerClient,
-    pub gateway_client: SharedGatewayClient,
+    gateway_client: SharedGatewayClient,
+    dynamic_config_rx: Receiver<HttpServerDynamicConfig>,
+}
+
+impl AppState {
+    fn get_dynamic_config(&self) -> HttpServerDynamicConfig {
+        // `borrow()` returns a reference to the value owned by the channel, hence we clone it.
+        let config = {
+            let config = self.dynamic_config_rx.borrow();
+            config.clone()
+        };
+        config
+    }
 }
 
 impl HttpServer {
@@ -67,8 +82,10 @@ impl HttpServer {
         config_manager_client: SharedConfigManagerClient,
         gateway_client: SharedGatewayClient,
     ) -> Self {
-        let app_state = AppState { config_manager_client, gateway_client };
-        HttpServer { config, app_state }
+        let (dynamic_config_tx, dynamic_config_rx) =
+            channel::<HttpServerDynamicConfig>(config.dynamic_config.clone());
+        let app_state = AppState { gateway_client, dynamic_config_rx };
+        HttpServer { config, app_state, config_manager_client, dynamic_config_tx }
     }
 
     pub async fn run(&mut self) -> Result<(), HttpServerRunError> {
@@ -80,6 +97,18 @@ impl HttpServer {
         let app = self.app();
         info!("HttpServer running using socket: {}", addr);
 
+        // TODO(Tsabary): make the poll interval part of the config.
+        let poll_interval = Duration::from_millis(1_000);
+        tokio::spawn(dynamic_config_poll(
+            self.dynamic_config_tx.clone(),
+            self.config_manager_client.clone(),
+            poll_interval,
+        ));
+
+        // TODO(update the http server struct to hold optional fields of the dynamic_config_tx,
+        // config_manager_client, and a JoinHandle for the polling task. Then, use `set` and `take`
+        // to move these around as needed.
+
         // Create a server that runs forever.
         Ok(axum::Server::bind(&addr).serve(app.into_make_service()).await?)
     }
@@ -89,12 +118,8 @@ impl HttpServer {
         Router::new()
             // Json Rpc endpoint
             .route("/gateway/add_rpc_transaction", post(add_rpc_tx))
-            .with_state(self.app_state.clone())
             // Rest api endpoint
-            .route("/gateway/add_transaction", post({
-                move |app_state: State<AppState>, headers: HeaderMap, tx: String| add_tx(app_state, headers, tx)
-            }))
-            .with_state(self.app_state.clone())
+            .route("/gateway/add_transaction", post(add_tx))
             // TODO(shahak): Remove this once we fix the centralized simulator to not use is_alive
             // and is_ready.
             .route(
@@ -105,21 +130,22 @@ impl HttpServer {
                 "/gateway/is_ready",
                 get(|| futures::future::ready("Gateway is ready".to_owned()))
             )
+            .layer(Extension(self.app_state.clone()))
     }
 }
 
 // HttpServer handlers.
 
 #[instrument(skip(app_state))]
 async fn add_rpc_tx(
-    State(app_state): State<AppState>,
+    Extension(app_state): Extension<AppState>,
     headers: HeaderMap,
     Json(tx): Json<RpcTransaction>,
 ) -> HttpServerResult<Json<GatewayOutput>> {
     debug!("ADD_TX_START: Http server received a new transaction.");
 
-    let dynamic_config = app_state.config_manager_client.get_http_server_dynamic_config().await?;
-    check_new_transactions_are_allowed(dynamic_config.accept_new_txs)?;
+    let HttpServerDynamicConfig { accept_new_txs, .. } = app_state.get_dynamic_config();
+    check_new_transactions_are_allowed(accept_new_txs)?;
 
     ADDED_TRANSACTIONS_TOTAL.increment(1);
     add_tx_inner(app_state, headers, tx).await
@@ -128,14 +154,15 @@ async fn add_rpc_tx(
 #[instrument(skip(app_state))]
 #[sequencer_latency_histogram(HTTP_SERVER_ADD_TX_LATENCY, true)]
 async fn add_tx(
-    State(app_state): State<AppState>,
+    Extension(app_state): Extension<AppState>,
     headers: HeaderMap,
     tx: String,
 ) -> HttpServerResult<Json<GatewayOutput>> {
     debug!("ADD_TX_START: Http server received a new transaction.");
 
-    let dynamic_config = app_state.config_manager_client.get_http_server_dynamic_config().await?;
-    check_new_transactions_are_allowed(dynamic_config.accept_new_txs)?;
+    let HttpServerDynamicConfig { accept_new_txs, max_sierra_program_size } =
+        app_state.get_dynamic_config();
+    check_new_transactions_are_allowed(accept_new_txs)?;
 
     ADDED_TRANSACTIONS_TOTAL.increment(1);
     let tx: DeprecatedGatewayTransactionV3 = match serde_json::from_str(&tx) {
@@ -152,7 +179,7 @@ async fn add_tx(
         }
     };
 
-    let rpc_tx = tx.convert_to_rpc_tx(dynamic_config.max_sierra_program_size).inspect_err(|e| {
+    let rpc_tx = tx.convert_to_rpc_tx(max_sierra_program_size).inspect_err(|e| {
         debug!("Error while converting deprecated gateway transaction into RPC transaction: {}", e);
     })?;
 
@@ -304,3 +331,19 @@ fn increment_failure_metrics(err: &HttpServerError) {
         ADDED_TRANSACTIONS_INTERNAL_ERROR.increment(1);
     }
 }
+
+async fn dynamic_config_poll(
+    tx: Sender<HttpServerDynamicConfig>,
+    config_manager_client: SharedConfigManagerClient,
+    poll_interval: Duration,
+) {
+    let mut interval = time::interval(poll_interval);
+    loop {
+        interval.tick().await;
+        let dynamic_config_result = config_manager_client.get_http_server_dynamic_config().await;
+        // Make the config available if it was successfully updated.
+        if let Ok(dynamic_config) = dynamic_config_result {
+            let _ = tx.send(dynamic_config);
+        }
+    }
+}