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()
 }
 
+#[rstest]
+#[tokio::test]
+/// Test that an HTTP server with a `allow_new_txs = false` config rejects new transactions.
+async fn allow_new_txs() {
+    let index = 10;
+    let tx = rpc_invoke_tx();
+
+    let mock_gateway_client = MockGatewayClient::new();
+    let mock_config_manager_client = MockConfigManagerClient::new();
+
+    // 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(