apollo_http_server: use allow new tx flag

Author: Itay-Tsabary-Starkware

crates/apollo_http_server/src/errors.rs

@@ -29,6 +29,8 @@ pub enum HttpServerError {
     DecompressionError(#[from] CompressionError),
     #[error(transparent)]
     DeserializationError(#[from] serde_json::Error),
+    #[error("Server is rejecting new transactions.")]
+    DisabledError(),
     #[error(transparent)]
     GatewayClientError(#[from] Box<GatewayClientError>),
 }
@@ -41,6 +43,7 @@ impl IntoResponse for HttpServerError {
             }
             HttpServerError::DecompressionError(e) => compression_error_into_response(e),
             HttpServerError::DeserializationError(e) => serde_error_into_response(e),
+            HttpServerError::DisabledError() => disabled_error_into_response(),
             HttpServerError::GatewayClientError(e) => gw_client_err_into_response(*e),
         }
     }
@@ -74,6 +77,19 @@ fn serde_error_into_response(err: serde_json::Error) -> Response {
     (response_code, response_body).into_response()
 }
 
+fn disabled_error_into_response() -> Response {
+    debug!("Server is configured to reject transactions.");
+    let (response_code, starknet_error) = (
+        StatusCode::SERVICE_UNAVAILABLE,
+        StarknetError {
+            code: StarknetErrorCode::UnknownErrorCode("Server unavailable.".to_string()),
+            message: "Server unavailable.".to_string(),
+        },
+    );
+    let response_body = serialize_error(&starknet_error);
+    (response_code, response_body).into_response()
+}
+
 fn gw_client_err_into_response(err: GatewayClientError) -> Response {
     let (response_code, deprecated_gateway_error) = match err {
         GatewayClientError::ClientError(e) => (

crates/apollo_http_server/src/http_server.rs

@@ -117,6 +117,10 @@ async fn add_rpc_tx(
     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)?;
+
     ADDED_TRANSACTIONS_TOTAL.increment(1);
     add_tx_inner(app_state, headers, tx).await
 }
@@ -128,11 +132,12 @@ async fn add_tx(
     headers: HeaderMap,
     tx: String,
 ) -> HttpServerResult<Json<GatewayOutput>> {
-    ADDED_TRANSACTIONS_TOTAL.increment(1);
     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)?;
 
+    ADDED_TRANSACTIONS_TOTAL.increment(1);
     let tx: DeprecatedGatewayTransactionV3 = match serde_json::from_str(&tx) {
         Ok(value) => value,
         Err(e) => {
@@ -154,6 +159,13 @@ async fn add_tx(
     add_tx_inner(app_state, headers, rpc_tx).await
 }
 
+fn check_new_transactions_are_allowed(accept_new_txs: bool) -> HttpServerResult<()> {
+    match accept_new_txs {
+        true => Ok(()),
+        false => Err(HttpServerError::DisabledError()),
+    }
+}
+
 #[allow(clippy::result_large_err)]
 fn validate_supported_tx_version_str(tx: &str) -> HttpServerResult<()> {
     // 1. Remove all whitespace

crates/apollo_http_server/src/http_server_test.rs

@@ -88,6 +88,26 @@ async fn to_bytes(res: Response) -> Bytes {
     res.into_body().collect().await.unwrap().to_bytes()
 }
 
+/// Test that an HTTP server with a `allow_new_txs = false` config rejects new transactions.
+#[rstest]
+#[tokio::test]
+async fn allow_new_txs() {
+    let index = 10;
+    let tx = rpc_invoke_tx();
+
+    let mock_gateway_client = MockGatewayClient::new();
+    let mock_config_manager_client = get_mock_config_manager_client(false);
+
+    // TODO(Yael): avoid the hardcoded node offset index, consider dynamic allocation.
+    let http_client =
+        add_tx_http_client(mock_config_manager_client, mock_gateway_client, 1 + index).await;
+
+    // Send a transaction to the server.
+    let response = http_client.add_tx(tx.clone()).await;
+    let status = response.status();
+    assert!(!status.is_success(), "{status:?}");
+}
+
 #[tokio::test]
 async fn error_into_response() {
     let error = HttpServerError::DeserializationError(
@@ -127,7 +147,7 @@ async fn record_region_test(#[case] index: u16, #[case] tx: impl GatewayTransact
         .times(1)
         .return_const(Ok(GatewayOutput::Invoke(InvokeGatewayOutput::new(tx_hash_2))));
 
-    let mock_config_manager_client = get_mock_config_manager_client();
+    let mock_config_manager_client = get_mock_config_manager_client(true);
     // TODO(Yael): avoid the hardcoded node offset index, consider dynamic allocation.
     let http_client =
         add_tx_http_client(mock_config_manager_client, mock_gateway_client, 1 + index).await;
@@ -162,7 +182,7 @@ async fn record_region_gateway_failing_tx(#[case] index: u16, #[case] tx: impl G
         )),
     ));
 
-    let mock_config_manager_client = get_mock_config_manager_client();
+    let mock_config_manager_client = get_mock_config_manager_client(true);
     let http_client =
         add_tx_http_client(mock_config_manager_client, mock_gateway_client, 3 + index).await;
 
@@ -212,7 +232,7 @@ async fn test_response(#[case] index: u16, #[case] tx: impl GatewayTransaction)
         expected_internal_err,
     ));
 
-    let mock_config_manager_client = get_mock_config_manager_client();
+    let mock_config_manager_client = get_mock_config_manager_client(true);
     let http_client =
         add_tx_http_client(mock_config_manager_client, mock_gateway_client, 5 + index).await;
 
@@ -281,7 +301,7 @@ async fn test_unsupported_tx_version(
     }
 
     let mock_gateway_client = MockGatewayClient::new();
-    let mock_config_manager_client = get_mock_config_manager_client();
+    let mock_config_manager_client = get_mock_config_manager_client(true);
     let http_client =
         add_tx_http_client(mock_config_manager_client, mock_gateway_client, 9 + index).await;
 
@@ -302,7 +322,7 @@ async fn sanitizing_error_message() {
     tx_object.insert("version".to_string(), Value::String(malicious_version.to_string())).unwrap();
 
     let mock_gateway_client = MockGatewayClient::new();
-    let mock_config_manager_client = get_mock_config_manager_client();
+    let mock_config_manager_client = get_mock_config_manager_client(true);
     let http_client = add_tx_http_client(mock_config_manager_client, mock_gateway_client, 13).await;
 
     let serialized_err =

crates/apollo_http_server/src/metrics_test.rs

@@ -55,7 +55,7 @@ async fn add_tx_metrics_test(#[case] index: u16, #[case] tx: impl GatewayTransac
         )))
     });
 
-    let mock_config_manager_client = get_mock_config_manager_client();
+    let mock_config_manager_client = get_mock_config_manager_client(true);
 
     // Initialize the metrics directly instead of spawning a monitoring endpoint task.
     let recorder = PrometheusBuilder::new().build_recorder();
@@ -89,7 +89,7 @@ async fn add_tx_serde_failure_metrics_test() {
         .times(1)
         .return_once(move |_| Ok(success_gateway_client_output()));
 
-    let mock_config_manager_client = get_mock_config_manager_client();
+    let mock_config_manager_client = get_mock_config_manager_client(true);
 
     // Initialize the metrics directly instead of spawning a monitoring endpoint task.
     let recorder = PrometheusBuilder::new().build_recorder();

crates/apollo_http_server/src/test_utils.rs

@@ -183,12 +183,12 @@ pub fn deprecated_gateway_declare_tx() -> DeprecatedGatewayTransactionV3 {
     DeprecatedGatewayTransactionV3::from(declare_tx())
 }
 
-// A mock config manager client returning the default http server dynamic config for an unlimited
-// number of requests.
-pub fn get_mock_config_manager_client() -> MockConfigManagerClient {
+// A mock config manager client returning the an http server dynamic config that accepts/rejects
+// transactions for an unlimited number of requests.
+pub fn get_mock_config_manager_client(accept_new_txs: bool) -> MockConfigManagerClient {
     let mut mock_config_manager_client = MockConfigManagerClient::new();
     mock_config_manager_client
         .expect_get_http_server_dynamic_config()
-        .returning(move || Ok(HttpServerDynamicConfig::default()));
+        .returning(move || Ok(HttpServerDynamicConfig { accept_new_txs, ..Default::default() }));
     mock_config_manager_client
 }