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

@@ -45,8 +45,10 @@ const DEPRECATED_GATEWAY_DEPLOY_ACCOUNT_TX_RESPONSE_JSON_PATH: &str =
 
 const EXPECTED_TX_HASH: TransactionHash = TransactionHash(Felt::ONE);
 
+// TODO(Tsabary): find a better way to allocate different ports to different tests.
+
 // The http_server is oblivious to the GateWayOutput type, so we always return invoke.
-pub fn default_gateway_output() -> GatewayOutput {
+fn default_gateway_output() -> GatewayOutput {
     GatewayOutput::Invoke(InvokeGatewayOutput::new(EXPECTED_TX_HASH))
 }
 
@@ -88,6 +90,28 @@ async fn to_bytes(res: Response) -> Bytes {
     res.into_body().collect().await.unwrap().to_bytes()
 }
 
+// TODO(Tsabary): there's a discrepancy with the used `StatusCode` crates: reqwest and hyper. Some
+// tests expect the former and some the other, based on the used test utils. Better sort this out
+// and make consistent across.
+
+// Uses add_tx_http_client with index 0.
+/// Test that an HTTP server with a `allow_new_txs = false` config rejects new transactions.
+#[rstest]
+#[tokio::test]
+async fn allow_new_txs() {
+    let tx = rpc_invoke_tx();
+
+    let mock_gateway_client = MockGatewayClient::new();
+    let mock_config_manager_client = get_mock_config_manager_client(false);
+
+    let http_client = add_tx_http_client(mock_config_manager_client, mock_gateway_client, 0).await;
+
+    // Send a transaction to the server.
+    let response = http_client.add_tx(tx.clone()).await;
+    let status = response.status();
+    assert_eq!(status, reqwest::StatusCode::SERVICE_UNAVAILABLE, "{status:?}");
+}
+
 #[tokio::test]
 async fn error_into_response() {
     let error = HttpServerError::DeserializationError(
@@ -106,6 +130,7 @@ async fn error_into_response() {
     );
 }
 
+// Uses add_tx_http_client with indices 1,2.
 #[traced_test]
 #[rstest]
 #[case::add_deprecated_gateway_tx(0, deprecated_gateway_invoke_tx())]
@@ -127,7 +152,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;
@@ -146,6 +171,7 @@ async fn record_region_test(#[case] index: u16, #[case] tx: impl GatewayTransact
     ));
 }
 
+// Uses add_tx_http_client with indices 3,4.
 #[traced_test]
 #[rstest]
 #[case::add_deprecated_gateway_tx(0, deprecated_gateway_invoke_tx())]
@@ -162,7 +188,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;
 
@@ -171,6 +197,7 @@ async fn record_region_gateway_failing_tx(#[case] index: u16, #[case] tx: impl G
     assert!(!logs_contain("Recorded transaction transaction_hash="));
 }
 
+// Uses add_tx_http_client with indices 5,6,7.
 #[rstest]
 #[case::add_deprecated_gateway_invoke(0, deprecated_gateway_invoke_tx())]
 #[case::add_deprecated_gateway_deploy_account(1, deprecated_gateway_deploy_account_tx())]
@@ -212,7 +239,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;
 
@@ -231,6 +258,7 @@ async fn test_response(#[case] index: u16, #[case] tx: impl GatewayTransaction)
     assert_eq!(error_str, expected_gateway_client_err_str);
 }
 
+// Uses add_tx_http_client with indices 9,10,11.
 #[rstest]
 #[case::missing_version(
     0,
@@ -281,7 +309,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;
 
@@ -291,6 +319,7 @@ async fn test_unsupported_tx_version(
     assert_eq!(starknet_error, expected_err);
 }
 
+// Uses add_tx_http_client with index 13.
 #[tokio::test]
 async fn sanitizing_error_message() {
     // Set the tx version to be a problematic text.
@@ -302,7 +331,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

@@ -33,6 +33,7 @@ fn success_gateway_client_output() -> GatewayOutput {
     GatewayOutput::Invoke(InvokeGatewayOutput::new(TransactionHash::default()))
 }
 
+// Uses add_tx_http_client with indices 14,15.
 #[rstest]
 #[case::add_deprecated_gateway_tx(0, deprecated_gateway_invoke_tx())]
 #[case::add_rpc_tx(1, rpc_invoke_tx())]
@@ -55,7 +56,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();
@@ -80,6 +81,7 @@ async fn add_tx_metrics_test(#[case] index: u16, #[case] tx: impl GatewayTransac
     ADDED_TRANSACTIONS_FAILURE.assert_eq::<usize>(&metrics, FAILURE_TXS_TO_SEND);
 }
 
+// Uses add_tx_http_client with index 16.
 #[tokio::test]
 async fn add_tx_serde_failure_metrics_test() {
     let mut mock_gateway_client = MockGatewayClient::new();
@@ -89,7 +91,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

@@ -153,6 +153,9 @@ impl GatewayTransaction for TransactionSerialization {
     }
 }
 
+// Creates a client for testing the http server functionality. Must be invoked with different
+// `port_index` values to support concurrent execution, otherwise leading to `Address already in use
+// (os error 98)` errors.
 pub async fn add_tx_http_client(
     mock_config_manager_client: MockConfigManagerClient,
     mock_gateway_client: MockGatewayClient,
@@ -183,12 +186,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
 }