fix(gateway): track cancelled requests

CHANGELOG.md

@@ -17,6 +17,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - `--rpc.gateway-trace-timeout` CLI option (default 30s) bounding how long `starknet_traceTransaction` and `starknet_traceBlockTransactions` may spend on the feeder gateway fallback path.
+- Gateway-client metrics now expose the `gateway_requests_in_flight` gauge.
+- Added `--rpc.gateway-add-transaction-timeout` (default 60s) for gateway submissions made by `starknet_addInvokeTransaction`, `starknet_addDeclareTransaction`, and `starknet_addDeployAccountTransaction`.
 - **BREAKING**: The RPC server now pings websocket peers that have been quiet for `--rpc.websocket.ping-interval` and closes connections that leave `--rpc.websocket.max-missed-pings` pings unanswered. Clients are required to answer pings, which RFC 6455 mandates and which browsers and the mainstream websocket libraries handle for you, but only while the client is reading from the connection. A client that stops reading for longer than the ping interval times the missed ping limit is disconnected. The keepalive cannot be turned off, so `--rpc.websocket.ping-interval`, `--rpc.websocket.initial-frame-timeout` and `--rpc.websocket.max-missed-pings` all reject `0`. Raise the ping interval rather than trying to disable it.
 - Concurrently open RPC websocket connections are now limited to 1024 by default, configurable with the `--rpc.websocket.max-connections` CLI option. Upgrade requests over the limit are rejected with HTTP 503.
 - RPC websocket connections that don't send anything after being established now time out, configurable with the `--rpc.websocket.initial-frame-timeout` CLI option.

crates/common/src/test_utils.rs

@@ -27,6 +27,7 @@ pub mod metrics {
         Counter,
         CounterFn,
         Gauge,
+        GaugeFn,
         Histogram,
         Key,
         KeyName,
@@ -43,17 +44,21 @@ pub mod metrics {
     #[derive(Debug, Default)]
     pub struct FakeRecorder(FakeRecorderHandle);
 
-    /// Handle to the [`FakeRecorder`], which allows to get the current value of
-    /// counters.
+    /// Handle to the [`FakeRecorder`], which exposes current counter and gauge
+    /// values.
     #[derive(Clone, Debug, Default)]
     pub struct FakeRecorderHandle {
         counters: Arc<RwLock<HashMap<Key, Arc<FakeCounterFn>>>>,
+        gauges: Arc<RwLock<HashMap<Key, Arc<FakeGaugeFn>>>>,
         methods: Option<&'static [&'static str]>,
     }
 
     #[derive(Debug, Default)]
     struct FakeCounterFn(AtomicU64);
 
+    #[derive(Debug, Default)]
+    struct FakeGaugeFn(AtomicU64);
+
     impl Recorder for FakeRecorder {
         fn describe_counter(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
         fn describe_gauge(&self, _: KeyName, _: Option<Unit>, _: SharedString) {}
@@ -88,8 +93,13 @@ pub mod metrics {
             }
         }
 
-        fn register_gauge(&self, _: &Key, _metadata: &Metadata<'_>) -> Gauge {
-            unimplemented!()
+        fn register_gauge(&self, key: &Key, _metadata: &Metadata<'_>) -> Gauge {
+            if self.is_key_used(key) {
+                let mut gauges = self.0.gauges.write().unwrap();
+                Gauge::from_arc(gauges.entry(key.clone()).or_default().clone())
+            } else {
+                Gauge::noop()
+            }
         }
         fn register_histogram(&self, _: &Key, _metadata: &Metadata<'_>) -> Histogram {
             // Ignored in tests for now
@@ -98,13 +108,14 @@ pub mod metrics {
     }
 
     impl FakeRecorder {
-        /// Creates a [`FakeRecorder`] which only holds counter values for
+        /// Creates a [`FakeRecorder`] which only holds metric values for
         /// `methods`.
         ///
-        /// All other methods use the [no-op counters](`https://docs.rs/metrics/latest/metrics/struct.Counter.html#method.noop`)
+        /// Metrics for all other methods use no-op handles.
         pub fn new_for(methods: &'static [&'static str]) -> Self {
             Self(FakeRecorderHandle {
                 counters: Arc::default(),
+                gauges: Arc::default(),
                 methods: Some(methods),
             })
         }
@@ -169,6 +180,25 @@ pub mod metrics {
                 .0
                 .load(Ordering::Relaxed)
         }
+
+        /// Gets the current value for a gauge registered with a `method` label.
+        pub fn get_gauge_value(
+            &self,
+            gauge_name: &'static str,
+            method_name: impl Into<Cow<'static, str>>,
+        ) -> f64 {
+            let gauges = self.gauges.read().unwrap();
+            f64::from_bits(
+                gauges
+                    .get(&Key::from_parts(
+                        gauge_name,
+                        vec![Label::new("method", method_name.into())],
+                    ))
+                    .expect("Unregistered gauge name")
+                    .0
+                    .load(Ordering::Relaxed),
+            )
+        }
     }
 
     impl CounterFn for FakeCounterFn {
@@ -179,4 +209,26 @@ pub mod metrics {
             unimplemented!()
         }
     }
+
+    impl GaugeFn for FakeGaugeFn {
+        fn increment(&self, val: f64) {
+            self.0
+                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
+                    Some((f64::from_bits(current) + val).to_bits())
+                })
+                .unwrap();
+        }
+
+        fn decrement(&self, val: f64) {
+            self.0
+                .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
+                    Some((f64::from_bits(current) - val).to_bits())
+                })
+                .unwrap();
+        }
+
+        fn set(&self, val: f64) {
+            self.0.store(val.to_bits(), Ordering::Relaxed);
+        }
+    }
 }

crates/gateway-client/src/builder.rs

@@ -932,7 +932,10 @@ mod tests {
             let server = MockServer::start().await;
             let client = expect_compressed(&server).await;
             client
-                .add_invoke_transaction(InvokeFunction::V3(v3_non_empty_proof()))
+                .add_invoke_transaction(
+                    InvokeFunction::V3(v3_non_empty_proof()),
+                    std::time::Duration::MAX,
+                )
                 .await
                 .unwrap();
         }
@@ -942,7 +945,10 @@ mod tests {
             let server = MockServer::start().await;
             let client = expect_uncompressed(&server).await;
             client
-                .add_invoke_transaction(InvokeFunction::V3(v3_empty_proof()))
+                .add_invoke_transaction(
+                    InvokeFunction::V3(v3_empty_proof()),
+                    std::time::Duration::MAX,
+                )
                 .await
                 .unwrap();
         }

crates/gateway-client/src/lib.rs

@@ -105,6 +105,7 @@ pub trait GatewayApi: Sync {
     async fn add_invoke_transaction<'tx>(
         &self,
         invoke: request::add_transaction::InvokeFunction<'tx>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::InvokeResponse, SequencerError> {
         unimplemented!();
     }
@@ -113,13 +114,15 @@ pub trait GatewayApi: Sync {
         &self,
         declare: request::add_transaction::Declare<'tx>,
         token: Option<String>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::DeclareResponse, SequencerError> {
         unimplemented!();
     }
 
     async fn add_deploy_account<'tx>(
         &self,
         deploy: request::add_transaction::DeployAccount<'tx>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::DeployAccountResponse, SequencerError> {
         unimplemented!();
     }
@@ -199,23 +202,28 @@ impl<T: GatewayApi + Sync + Send> GatewayApi for Arc<T> {
     async fn add_invoke_transaction<'tx>(
         &self,
         invoke: request::add_transaction::InvokeFunction<'tx>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::InvokeResponse, SequencerError> {
-        self.as_ref().add_invoke_transaction(invoke).await
+        self.as_ref().add_invoke_transaction(invoke, timeout).await
     }
 
     async fn add_declare_transaction<'tx>(
         &self,
         declare: request::add_transaction::Declare<'tx>,
         token: Option<String>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::DeclareResponse, SequencerError> {
-        self.as_ref().add_declare_transaction(declare, token).await
+        self.as_ref()
+            .add_declare_transaction(declare, token, timeout)
+            .await
     }
 
     async fn add_deploy_account<'tx>(
         &self,
         deploy: request::add_transaction::DeployAccount<'tx>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::DeployAccountResponse, SequencerError> {
-        self.as_ref().add_deploy_account(deploy).await
+        self.as_ref().add_deploy_account(deploy, timeout).await
     }
 
     async fn block_traces(&self, block: BlockId) -> Result<BlockTrace, SequencerError> {
@@ -622,6 +630,7 @@ impl GatewayApi for Client {
     async fn add_invoke_transaction<'tx>(
         &self,
         invoke: request::add_transaction::InvokeFunction<'tx>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::InvokeResponse, SequencerError> {
         // Note that we don't do retries here.
         // This method is used to proxy an add transaction operation from the
@@ -635,7 +644,7 @@ impl GatewayApi for Client {
             .compress(self.compress_gateway_requests && !invoke.is_proof_empty())
             .post_with_json(
                 &request::add_transaction::AddTransaction::Invoke(invoke),
-                Some(Duration::MAX),
+                Some(timeout),
             )
             .await
     }
@@ -646,6 +655,7 @@ impl GatewayApi for Client {
         &self,
         declare: request::add_transaction::Declare<'tx>,
         token: Option<String>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::DeclareResponse, SequencerError> {
         // Note that we don't do retries here.
         // This method is used to proxy an add transaction operation from the
@@ -658,7 +668,7 @@ impl GatewayApi for Client {
             .retry(false)
             .post_with_json(
                 &request::add_transaction::AddTransaction::Declare(declare),
-                Some(Duration::MAX),
+                Some(timeout),
             )
             .await
     }
@@ -667,6 +677,7 @@ impl GatewayApi for Client {
     async fn add_deploy_account<'tx>(
         &self,
         deploy: request::add_transaction::DeployAccount<'tx>,
+        timeout: Duration,
     ) -> Result<reply::add_transaction::DeployAccountResponse, SequencerError> {
         // Note that we don't do retries here.
         // This method is used to proxy an add transaction operation from the
@@ -677,7 +688,7 @@ impl GatewayApi for Client {
             .retry(false)
             .post_with_json(
                 &request::add_transaction::AddTransaction::DeployAccount(deploy),
-                Some(Duration::MAX),
+                Some(timeout),
             )
             .await
     }
@@ -721,7 +732,6 @@ mod tests {
 
     use assert_matches::assert_matches;
     use pathfinder_common::macro_prelude::*;
-    use pathfinder_common::prelude::*;
     use pathfinder_crypto::Felt;
     use starknet_gateway_test_fixtures::testnet::*;
     use starknet_gateway_types::error::{test_response_from, KnownStarknetErrorCode};
@@ -909,7 +919,10 @@ mod tests {
                     calldata: &call,
                 });
 
-                let error = client.add_invoke_transaction(invoke).await.unwrap_err();
+                let error = client
+                    .add_invoke_transaction(invoke, Duration::MAX)
+                    .await
+                    .unwrap_err();
                 assert_matches!(
                     error,
                     SequencerError::StarknetError(e) => assert_eq!(e.code, KnownStarknetErrorCode::DeprecatedTransaction.into())
@@ -941,7 +954,48 @@ mod tests {
                     entry_point_selector: None,
                     calldata: &call,
                 });
-                client.add_invoke_transaction(invoke).await.unwrap();
+                client
+                    .add_invoke_transaction(invoke, Duration::MAX)
+                    .await
+                    .unwrap();
+            }
+
+            #[tokio::test]
+            async fn uses_per_request_timeout() {
+                use request::add_transaction::{InvokeFunction, InvokeFunctionV0V1};
+
+                let server = MockServer::start().await;
+                Mock::given(matchers::method("POST"))
+                    .and(matchers::path("/gateway/add_transaction"))
+                    .respond_with(
+                        ResponseTemplate::new(200)
+                            .set_delay(Duration::from_secs(60))
+                            .set_body_json(serde_json::json!({
+                                "code": "TRANSACTION_RECEIVED",
+                                "transaction_hash": "0x1"
+                            })),
+                    )
+                    .mount(&server)
+                    .await;
+                let client = Client::for_test(server.uri().parse().unwrap()).unwrap();
+                let (_, fee, sig, nonce, addr, call) = inputs();
+                let invoke = InvokeFunction::V1(InvokeFunctionV0V1 {
+                    max_fee: fee,
+                    signature: &sig,
+                    nonce: Some(nonce),
+                    sender_address: addr,
+                    entry_point_selector: None,
+                    calldata: &call,
+                });
+
+                let error = client
+                    .add_invoke_transaction(invoke, Duration::from_millis(20))
+                    .await
+                    .unwrap_err();
+
+                assert_matches!(error, SequencerError::ReqwestError(error) => {
+                    assert!(error.is_timeout());
+                });
             }
         }
 
@@ -975,7 +1029,7 @@ mod tests {
                     compiled_class_hash: None,
                 });
                 let error = client
-                    .add_declare_transaction(declare, None)
+                    .add_declare_transaction(declare, None, Duration::MAX)
                     .await
                     .unwrap_err();
                 assert_matches!(
@@ -1010,7 +1064,10 @@ mod tests {
                     compiled_class_hash: None,
                 });
 
-                client.add_declare_transaction(declare, None).await.unwrap();
+                client
+                    .add_declare_transaction(declare, None, Duration::MAX)
+                    .await
+                    .unwrap();
             }
 
             fn sierra_contract_class_from_fixture() -> SierraContractDefinition {
@@ -1090,7 +1147,10 @@ mod tests {
                     )),
                 });
 
-                client.add_declare_transaction(declare, None).await.unwrap();
+                client
+                    .add_declare_transaction(declare, None, Duration::MAX)
+                    .await
+                    .unwrap();
             }
         }
 
@@ -1193,7 +1253,11 @@ mod tests {
                 });
 
                 client
-                    .add_declare_transaction(declare, Some(EXPECTED_TOKEN.to_owned()))
+                    .add_declare_transaction(
+                        declare,
+                        Some(EXPECTED_TOKEN.to_owned()),
+                        Duration::MAX,
+                    )
                     .await
                     .unwrap();
             }
@@ -1222,7 +1286,7 @@ mod tests {
                 });
 
                 let err = client
-                    .add_declare_transaction(declare, None)
+                    .add_declare_transaction(declare, None, Duration::MAX)
                     .await
                     .unwrap_err();