From 4ad58ffe3ce8d450400eab8a23d360363678cce5 Mon Sep 17 00:00:00 2001 From: Zaksans Date: Wed, 2 Sep 2026 01:32:23 +0200 Subject: [PATCH 1/5] fix(gateway): track cancelled requests --- CHANGELOG.md | 1 + crates/common/src/test_utils.rs | 58 ++++++- crates/gateway-client/src/metrics.rs | 152 ++++++++++++------ crates/gateway-client/tests/metrics.rs | 41 ++++- .../rpc/src/method/add_invoke_transaction.rs | 75 +++++++++ 5 files changed, 267 insertions(+), 60 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a270c58791..29b81b416f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ 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 and record dropped request futures as failures with `reason="cancelled"`, including their elapsed duration. - **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. diff --git a/crates/common/src/test_utils.rs b/crates/common/src/test_utils.rs index 5f711618cb..2487364281 100644 --- a/crates/common/src/test_utils.rs +++ b/crates/common/src/test_utils.rs @@ -21,12 +21,13 @@ pub mod metrics { use std::borrow::Cow; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; - use std::sync::{Arc, RwLock}; + use std::sync::{Arc, Mutex, RwLock}; use 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>>>, + gauges: Arc>>>, methods: Option<&'static [&'static str]>, } #[derive(Debug, Default)] struct FakeCounterFn(AtomicU64); + #[derive(Debug, Default)] + struct FakeGaugeFn(Mutex); + impl Recorder for FakeRecorder { fn describe_counter(&self, _: KeyName, _: Option, _: SharedString) {} fn describe_gauge(&self, _: KeyName, _: Option, _: 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>, + ) -> f64 { + let gauges = self.gauges.read().unwrap(); + let value = *gauges + .get(&Key::from_parts( + gauge_name, + vec![Label::new("method", method_name.into())], + )) + .expect("Unregistered gauge name") + .0 + .lock() + .unwrap(); + value + } } impl CounterFn for FakeCounterFn { @@ -179,4 +209,18 @@ pub mod metrics { unimplemented!() } } + + impl GaugeFn for FakeGaugeFn { + fn increment(&self, val: f64) { + *self.0.lock().unwrap() += val; + } + + fn decrement(&self, val: f64) { + *self.0.lock().unwrap() -= val; + } + + fn set(&self, val: f64) { + *self.0.lock().unwrap() = val; + } + } } diff --git a/crates/gateway-client/src/metrics.rs b/crates/gateway-client/src/metrics.rs index a88804e4a1..eb18e182bc 100644 --- a/crates/gateway-client/src/metrics.rs +++ b/crates/gateway-client/src/metrics.rs @@ -7,6 +7,7 @@ use super::{BlockId, SequencerError}; const METRIC_REQUESTS: &str = "gateway_requests_total"; const METRIC_FAILED_REQUESTS: &str = "gateway_requests_failed_total"; +const METRIC_REQUESTS_IN_FLIGHT: &str = "gateway_requests_in_flight"; const METRIC_REQUESTS_LATENCY: &str = "gateway_request_duration_seconds"; const METRICS: [&str; 2] = [METRIC_REQUESTS, METRIC_FAILED_REQUESTS]; const TAG_LATEST: &str = "latest"; @@ -16,7 +17,9 @@ const REASON_DECODE: &str = "decode"; const REASON_STARKNET: &str = "starknet"; const REASON_RATE_LIMITING: &str = "rate_limiting"; const REASON_TIMEOUT: &str = "timeout"; -const REASONS: [&str; 4] = [ +const REASON_CANCELLED: &str = "cancelled"; +const REASONS: [&str; 5] = [ + REASON_CANCELLED, REASON_DECODE, REASON_RATE_LIMITING, REASON_STARKNET, @@ -42,8 +45,9 @@ pub fn register() { }) }); - // Request latency for all methods + // Request latency and in-flight requests for all methods Request::::METHODS.iter().for_each(|&method| { + let _ = metrics::gauge!(METRIC_REQUESTS_IN_FLIGHT, "method" => method); let _ = metrics::histogram!(METRIC_REQUESTS_LATENCY, "method" => method); }); @@ -112,15 +116,18 @@ impl RequestMetadata { /// # Usage /// -/// Awaits future `f` and increments the following counters for a particular +/// Awaits future `f` and records the following metrics for a particular /// method: /// - `gateway_requests_total`, -/// - `gateway_requests_failed_total` if the future returns the `Err()` variant. +/// - `gateway_requests_in_flight` while the future is alive, +/// - `gateway_requests_failed_total` if the future returns the `Err()` variant +/// or is cancelled before completion. /// /// # Additional counter labels /// -/// 1. All the above counters are also duplicated for the special cases of: -/// `("get_block" | "get_state_update") AND ("latest" | "pending")`. +/// 1. `gateway_requests_total` and `gateway_requests_failed_total` are also +/// duplicated for the special cases of: `("get_block" | "get_state_update") +/// AND ("latest" | "pending")`. /// /// 2. `gateway_requests_failed_total` is also duplicated for the specific /// failure reasons: @@ -130,71 +137,114 @@ impl RequestMetadata { /// error variant /// - `rate_limiting` if the future returns an `Err()` variant, which carries /// the [`reqwest::StatusCode::TOO_MANY_REQUESTS`] status code +/// - `cancelled` if the future is dropped before completion pub async fn with_metrics( meta: RequestMetadata, f: impl Future>, ) -> Result { - /// Increments a counter and its block tag specific variants if they exist - fn increment(counter_name: &'static str, meta: RequestMetadata) { - let method = meta.method; - let tag = meta.tag; - metrics::counter!(counter_name, "method" => method).increment(1); - - if let ("get_block" | "get_state_update", Some(tag)) = (method, tag.as_str()) { - metrics::counter!(counter_name, "method" => method, "tag" => tag).increment(1); - } + let mut metrics = InFlightRequest::new(meta); + let result = f.await; + metrics.finish(&result); + result +} + +/// Increments a counter and its block tag specific variants if they exist. +fn increment(counter_name: &'static str, meta: RequestMetadata) { + let method = meta.method; + let tag = meta.tag; + metrics::counter!(counter_name, "method" => method).increment(1); + + if let ("get_block" | "get_state_update", Some(tag)) = (method, tag.as_str()) { + metrics::counter!(counter_name, "method" => method, "tag" => tag).increment(1); } +} - /// Increments the `gateway_requests_failed_total` counter for a given - /// failure `reason`, includes block tag specific variants if they exist - fn increment_failed(meta: RequestMetadata, reason: &'static str) { - let method = meta.method; - let tag = meta.tag; - metrics::counter!(METRIC_FAILED_REQUESTS, "method" => method, "reason" => reason) - .increment(1); +/// Increments the `gateway_requests_failed_total` counter for a given failure +/// `reason`, including block tag specific variants if they exist. +fn increment_failed(meta: RequestMetadata, reason: &'static str) { + let method = meta.method; + let tag = meta.tag; + metrics::counter!(METRIC_FAILED_REQUESTS, "method" => method, "reason" => reason).increment(1); - if let ("get_block" | "get_state_update", Some(tag)) = (method, tag.as_str()) { - metrics::counter!(METRIC_FAILED_REQUESTS, "method" => method, "tag" => tag, "reason" => reason).increment(1); - } + if let ("get_block" | "get_state_update", Some(tag)) = (method, tag.as_str()) { + metrics::counter!(METRIC_FAILED_REQUESTS, "method" => method, "tag" => tag, "reason" => reason).increment(1); } +} - increment(METRIC_REQUESTS, meta); +struct InFlightRequest { + meta: RequestMetadata, + started: std::time::Instant, + in_flight: metrics::Gauge, + finished: bool, +} - let started = std::time::Instant::now(); - let result = f.await; - let elapsed = started.elapsed(); +impl InFlightRequest { + fn new(meta: RequestMetadata) -> Self { + increment(METRIC_REQUESTS, meta); - metrics::histogram!(METRIC_REQUESTS_LATENCY, "method" => meta.method) - .record(elapsed.as_secs_f64()); + let in_flight = metrics::gauge!(METRIC_REQUESTS_IN_FLIGHT, "method" => meta.method); + in_flight.increment(1.0); - result.inspect_err(|e| { - increment(METRIC_FAILED_REQUESTS, meta); + Self { + meta, + started: std::time::Instant::now(), + in_flight, + finished: false, + } + } + + fn finish(&mut self, result: &Result) { + self.finish_timing(); + + let Err(error) = result else { + return; + }; - match &e { + increment(METRIC_FAILED_REQUESTS, self.meta); + match error { SequencerError::StarknetError(_) => { - increment_failed(meta, REASON_STARKNET); + increment_failed(self.meta, REASON_STARKNET); } - SequencerError::InvalidStarknetErrorVariant => { - increment_failed(meta, REASON_DECODE); + SequencerError::InvalidStarknetErrorVariant | SequencerError::InvalidResponse(_) => { + increment_failed(self.meta, REASON_DECODE); } - SequencerError::InvalidResponse(_) => { - increment_failed(meta, REASON_DECODE); + SequencerError::ReqwestError(error) if error.is_decode() => { + increment_failed(self.meta, REASON_DECODE); } - SequencerError::ReqwestError(e) if e.is_decode() => { - increment_failed(meta, REASON_DECODE); - } - SequencerError::ReqwestError(e) - if e.is_status() - && e.status().expect("error kind should be status") + SequencerError::ReqwestError(error) + if error.is_status() + && error.status().expect("error kind should be status") == reqwest::StatusCode::TOO_MANY_REQUESTS => { - increment_failed(meta, REASON_RATE_LIMITING); + increment_failed(self.meta, REASON_RATE_LIMITING); } - SequencerError::ReqwestError(e) if e.is_timeout() => { - increment_failed(meta, REASON_TIMEOUT); + SequencerError::ReqwestError(error) if error.is_timeout() => { + increment_failed(self.meta, REASON_TIMEOUT); } - SequencerError::ReqwestError(_) => {} - SequencerError::GatewayRequestCreationError(_) => {} + SequencerError::ReqwestError(_) | SequencerError::GatewayRequestCreationError(_) => {} + } + } + + fn finish_timing(&mut self) { + self.finished = true; + self.in_flight.decrement(1.0); + metrics::histogram!(METRIC_REQUESTS_LATENCY, "method" => self.meta.method) + .record(self.started.elapsed().as_secs_f64()); + } +} + +impl Drop for InFlightRequest { + fn drop(&mut self) { + if self.finished { + return; } - }) + + self.finish_timing(); + increment(METRIC_FAILED_REQUESTS, self.meta); + increment_failed(self.meta, REASON_CANCELLED); + tracing::debug!( + method = self.meta.method, + "Gateway request cancelled before completion" + ); + } } diff --git a/crates/gateway-client/tests/metrics.rs b/crates/gateway-client/tests/metrics.rs index cfca6437fa..4a3762314d 100644 --- a/crates/gateway-client/tests/metrics.rs +++ b/crates/gateway-client/tests/metrics.rs @@ -101,16 +101,52 @@ async fn all_counter_types_including_tags() { .collect::>() .await; + let cancellation_server = MockServer::start().await; + Mock::given(matchers::path("/feeder_gateway/get_block")) + .and(matchers::query_param("blockNumber", "124")) + .and(matchers::query_param("headerOnly", "true")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(std::time::Duration::from_secs(60)) + .set_body_json(serde_json::json!({ + "block_hash": "0x0", + "block_number": 124 + })), + ) + .mount(&cancellation_server) + .await; + let cancellation_client = Client::for_test(cancellation_server.uri().parse().unwrap()) + .unwrap() + .disable_retry_for_tests(); + let mut cancelled_request = + Box::pin(cancellation_client.block_header(BlockId::Number(BlockNumber::new_or_panic(124)))); + + tokio::time::timeout( + std::time::Duration::from_millis(50), + cancelled_request.as_mut(), + ) + .await + .unwrap_err(); + assert_eq!( + handle.get_gauge_value("gateway_requests_in_flight", method_name), + 1.0 + ); + drop(cancelled_request); + assert_eq!( + handle.get_gauge_value("gateway_requests_in_flight", method_name), + 0.0 + ); + // IMPORTANT // // We're not using any crate::sequencer::metrics consts here, because this // is public API and we'd like to catch if/when it changed (apparently // due to a bug) [ - ("gateway_requests_total", None, None, 21), + ("gateway_requests_total", None, None, 22), ("gateway_requests_total", Some("latest"), None, 7), ("gateway_requests_total", Some("pending"), None, 7), - ("gateway_requests_failed_total", None, None, 18), + ("gateway_requests_failed_total", None, None, 19), ("gateway_requests_failed_total", Some("latest"), None, 6), ("gateway_requests_failed_total", Some("pending"), None, 6), ("gateway_requests_failed_total", None, Some("starknet"), 3), @@ -157,6 +193,7 @@ async fn all_counter_types_including_tags() { Some("rate_limiting"), 3, ), + ("gateway_requests_failed_total", None, Some("cancelled"), 1), ] .into_iter() .for_each( diff --git a/crates/rpc/src/method/add_invoke_transaction.rs b/crates/rpc/src/method/add_invoke_transaction.rs index a1911451e8..7c8eb9c0ae 100644 --- a/crates/rpc/src/method/add_invoke_transaction.rs +++ b/crates/rpc/src/method/add_invoke_transaction.rs @@ -458,4 +458,79 @@ mod tests { .unwrap_err(); assert_eq!(actual_error, expected_error); } + + #[tokio::test] + async fn dropping_rpc_future_cancels_gateway_submission() { + use tokio::io::AsyncReadExt; + use tokio::sync::oneshot; + use tokio::time::{timeout, Duration}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let gateway_address = listener.local_addr().unwrap(); + let (request_started_tx, request_started_rx) = oneshot::channel(); + let (connection_closed_tx, connection_closed_rx) = oneshot::channel(); + + let gateway = tokio::spawn(async move { + let (mut connection, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let mut buffer = [0; 4096]; + + loop { + let bytes_read = connection.read(&mut buffer).await.unwrap(); + assert_ne!( + bytes_read, 0, + "gateway connection closed before request arrived" + ); + request.extend_from_slice(&buffer[..bytes_read]); + + let Some(headers_end) = request.windows(4).position(|x| x == b"\r\n\r\n") else { + continue; + }; + let headers = std::str::from_utf8(&request[..headers_end]).unwrap(); + let content_length = headers + .lines() + .find_map(|header| { + let (name, value) = header.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().unwrap()) + }) + .expect("request should have a content-length header"); + + if request.len() >= headers_end + 4 + content_length { + break; + } + } + + request_started_tx.send(()).unwrap(); + + // The gateway intentionally never sends a response. Dropping the + // RPC future must therefore close this outbound connection. + match connection.read(&mut buffer[..1]).await { + Ok(0) | Err(_) => {} + Ok(_) => panic!("unexpected data after the complete gateway request"), + } + connection_closed_tx.send(()).unwrap(); + }); + + let mut context = RpcContext::for_tests(); + context.sequencer = starknet_gateway_client::Client::for_test( + format!("http://{gateway_address}").parse().unwrap(), + ) + .unwrap() + .disable_retry_for_tests(); + + let rpc_request = tokio::spawn(add_invoke_transaction(context, v3_input())); + timeout(Duration::from_secs(1), request_started_rx) + .await + .expect("gateway did not receive the request") + .unwrap(); + + rpc_request.abort(); + assert!(rpc_request.await.unwrap_err().is_cancelled()); + timeout(Duration::from_secs(1), connection_closed_rx) + .await + .expect("gateway connection survived RPC cancellation") + .unwrap(); + gateway.await.unwrap(); + } } From 2723d6fd0727fe59d9eaaed9dce2499c2e048ea3 Mon Sep 17 00:00:00 2001 From: Zaksans Date: Wed, 2 Sep 2026 22:01:02 +0200 Subject: [PATCH 2/5] fix(rpc): address gateway cancellation review --- CHANGELOG.md | 3 +- crates/gateway-client/src/builder.rs | 10 +- crates/gateway-client/src/lib.rs | 92 ++++++++++++++++--- crates/gateway-client/src/metrics.rs | 20 ++-- crates/gateway-client/tests/metrics.rs | 3 +- crates/pathfinder/src/bin/pathfinder/main.rs | 1 + crates/pathfinder/src/config.rs | 14 +++ crates/rpc/src/context.rs | 4 + .../rpc/src/method/add_declare_transaction.rs | 1 + .../method/add_deploy_account_transaction.rs | 9 +- .../rpc/src/method/add_invoke_transaction.rs | 9 +- 11 files changed, 126 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 29b81b416f..0501f122ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,7 +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 and record dropped request futures as failures with `reason="cancelled"`, including their elapsed duration. +- Gateway-client metrics now expose the `gateway_requests_in_flight` gauge. Dropped request futures only decrement the gauge because graceful shutdown and disconnected clients are not gateway failures. +- Added `--rpc.gateway-add-transaction-timeout` (default 60s) for sequencer 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. diff --git a/crates/gateway-client/src/builder.rs b/crates/gateway-client/src/builder.rs index 6910801f2d..990912b6f9 100644 --- a/crates/gateway-client/src/builder.rs +++ b/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(); } diff --git a/crates/gateway-client/src/lib.rs b/crates/gateway-client/src/lib.rs index 978f5cf0b6..2f78346781 100644 --- a/crates/gateway-client/src/lib.rs +++ b/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 { unimplemented!(); } @@ -113,6 +114,7 @@ pub trait GatewayApi: Sync { &self, declare: request::add_transaction::Declare<'tx>, token: Option, + timeout: Duration, ) -> Result { unimplemented!(); } @@ -120,6 +122,7 @@ pub trait GatewayApi: Sync { async fn add_deploy_account<'tx>( &self, deploy: request::add_transaction::DeployAccount<'tx>, + timeout: Duration, ) -> Result { unimplemented!(); } @@ -199,23 +202,28 @@ impl GatewayApi for Arc { async fn add_invoke_transaction<'tx>( &self, invoke: request::add_transaction::InvokeFunction<'tx>, + timeout: Duration, ) -> Result { - 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, + timeout: Duration, ) -> Result { - 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 { - 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 { @@ -622,6 +630,7 @@ impl GatewayApi for Client { async fn add_invoke_transaction<'tx>( &self, invoke: request::add_transaction::InvokeFunction<'tx>, + timeout: Duration, ) -> Result { // 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, + timeout: Duration, ) -> Result { // 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 { // 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(); diff --git a/crates/gateway-client/src/metrics.rs b/crates/gateway-client/src/metrics.rs index eb18e182bc..e69ae90e3d 100644 --- a/crates/gateway-client/src/metrics.rs +++ b/crates/gateway-client/src/metrics.rs @@ -17,9 +17,7 @@ const REASON_DECODE: &str = "decode"; const REASON_STARKNET: &str = "starknet"; const REASON_RATE_LIMITING: &str = "rate_limiting"; const REASON_TIMEOUT: &str = "timeout"; -const REASON_CANCELLED: &str = "cancelled"; -const REASONS: [&str; 5] = [ - REASON_CANCELLED, +const REASONS: [&str; 4] = [ REASON_DECODE, REASON_RATE_LIMITING, REASON_STARKNET, @@ -120,8 +118,7 @@ impl RequestMetadata { /// method: /// - `gateway_requests_total`, /// - `gateway_requests_in_flight` while the future is alive, -/// - `gateway_requests_failed_total` if the future returns the `Err()` variant -/// or is cancelled before completion. +/// - `gateway_requests_failed_total` if the future returns the `Err()` variant. /// /// # Additional counter labels /// @@ -137,7 +134,6 @@ impl RequestMetadata { /// error variant /// - `rate_limiting` if the future returns an `Err()` variant, which carries /// the [`reqwest::StatusCode::TOO_MANY_REQUESTS`] status code -/// - `cancelled` if the future is dropped before completion pub async fn with_metrics( meta: RequestMetadata, f: impl Future>, @@ -239,12 +235,10 @@ impl Drop for InFlightRequest { return; } - self.finish_timing(); - increment(METRIC_FAILED_REQUESTS, self.meta); - increment_failed(self.meta, REASON_CANCELLED); - tracing::debug!( - method = self.meta.method, - "Gateway request cancelled before completion" - ); + // A dropped request future is not necessarily a gateway failure: it is + // also how graceful shutdown and disconnected RPC clients cancel work. + // Keep the live gauge accurate without manufacturing latency or failure + // samples for a request that never completed. + self.in_flight.decrement(1.0); } } diff --git a/crates/gateway-client/tests/metrics.rs b/crates/gateway-client/tests/metrics.rs index 4a3762314d..2b22ff0ab3 100644 --- a/crates/gateway-client/tests/metrics.rs +++ b/crates/gateway-client/tests/metrics.rs @@ -146,7 +146,7 @@ async fn all_counter_types_including_tags() { ("gateway_requests_total", None, None, 22), ("gateway_requests_total", Some("latest"), None, 7), ("gateway_requests_total", Some("pending"), None, 7), - ("gateway_requests_failed_total", None, None, 19), + ("gateway_requests_failed_total", None, None, 18), ("gateway_requests_failed_total", Some("latest"), None, 6), ("gateway_requests_failed_total", Some("pending"), None, 6), ("gateway_requests_failed_total", None, Some("starknet"), 3), @@ -193,7 +193,6 @@ async fn all_counter_types_including_tags() { Some("rate_limiting"), 3, ), - ("gateway_requests_failed_total", None, Some("cancelled"), 1), ] .into_iter() .for_each( diff --git a/crates/pathfinder/src/bin/pathfinder/main.rs b/crates/pathfinder/src/bin/pathfinder/main.rs index 8184c8a284..2bf1d5fbe5 100644 --- a/crates/pathfinder/src/bin/pathfinder/main.rs +++ b/crates/pathfinder/src/bin/pathfinder/main.rs @@ -275,6 +275,7 @@ Hint: This is usually caused by exceeding the file descriptor limit of your syst submission_tracker_time_limit: config.submission_tracker_time_limit, submission_tracker_size_limit: config.submission_tracker_size_limit, block_trace_cache_size: config.rpc_block_trace_cache_size, + gateway_add_transaction_timeout: config.rpc_gateway_add_transaction_timeout, gateway_trace_timeout: config.rpc_gateway_trace_timeout, compiler_concurrency_limit, compiler_resource_limits: config.compiler_resource_limits, diff --git a/crates/pathfinder/src/config.rs b/crates/pathfinder/src/config.rs index 7bc4c20de5..59c1b58d01 100644 --- a/crates/pathfinder/src/config.rs +++ b/crates/pathfinder/src/config.rs @@ -610,6 +610,16 @@ Setting this value too low may cause compilation of large classes to fail.", )] rpc_block_trace_cache_size: std::num::NonZeroUsize, + #[arg( + long = "rpc.gateway-add-transaction-timeout", + value_name = "Seconds", + long_help = "Maximum duration an `addInvokeTransaction`, `addDeclareTransaction`, or \ + `addDeployAccountTransaction` request may wait for the sequencer gateway.", + default_value = "60", + env = "PATHFINDER_RPC_GATEWAY_ADD_TRANSACTION_TIMEOUT" + )] + rpc_gateway_add_transaction_timeout: std::num::NonZeroU64, + #[arg( long = "rpc.gateway-trace-timeout", value_name = "Seconds", @@ -1213,6 +1223,7 @@ pub struct Config { pub submission_tracker_time_limit: NonZeroU64, pub submission_tracker_size_limit: NonZeroUsize, pub rpc_block_trace_cache_size: NonZeroUsize, + pub rpc_gateway_add_transaction_timeout: Duration, pub rpc_gateway_trace_timeout: Duration, pub consensus: Option, /// Integration testing config, only available on debug builds with `p2p` @@ -1536,6 +1547,9 @@ impl Config { submission_tracker_time_limit: args.submission_tracker_time_limit, submission_tracker_size_limit: args.submission_tracker_size_limit, rpc_block_trace_cache_size: args.rpc_block_trace_cache_size, + rpc_gateway_add_transaction_timeout: Duration::from_secs( + args.rpc_gateway_add_transaction_timeout.get(), + ), rpc_gateway_trace_timeout: Duration::from_secs(args.rpc_gateway_trace_timeout.get()), consensus: ConsensusConfig::parse_or_exit(args.consensus), integration_testing: integration_testing::IntegrationTestingConfig::parse( diff --git a/crates/rpc/src/context.rs b/crates/rpc/src/context.rs index d1a3638c39..974ba7ad90 100644 --- a/crates/rpc/src/context.rs +++ b/crates/rpc/src/context.rs @@ -82,6 +82,9 @@ pub struct RpcConfig { pub submission_tracker_time_limit: NonZeroU64, pub submission_tracker_size_limit: NonZeroUsize, pub block_trace_cache_size: NonZeroUsize, + /// Upper bound for submitting invoke, declare, and deploy-account + /// transactions to the sequencer gateway. + pub gateway_add_transaction_timeout: Duration, /// Upper bound on how long a single trace request may wait on the feeder /// gateway fallback path. The gateway client retries transport errors with /// an unbounded exponential backoff, so without this ceiling a slow or @@ -272,6 +275,7 @@ impl RpcContext { submission_tracker_time_limit: NonZeroU64::new(300).unwrap(), submission_tracker_size_limit: NonZeroUsize::new(30000).unwrap(), block_trace_cache_size: NonZeroUsize::new(1).unwrap(), + gateway_add_transaction_timeout: Duration::from_secs(60), gateway_trace_timeout: Duration::from_secs(30), compiler_concurrency_limit: NonZeroUsize::new(1).unwrap(), compiler_resource_limits: pathfinder_compiler::ResourceLimits::for_test(), diff --git a/crates/rpc/src/method/add_declare_transaction.rs b/crates/rpc/src/method/add_declare_transaction.rs index 6c6a62280b..2bae3bcf95 100644 --- a/crates/rpc/src/method/add_declare_transaction.rs +++ b/crates/rpc/src/method/add_declare_transaction.rs @@ -243,6 +243,7 @@ pub async fn add_declare_transaction( account_deployment_data: &tx.account_deployment_data, }), input.token, + context.config.gateway_add_transaction_timeout, ) .await?; let new_tx = DeclareTransactionV3 { diff --git a/crates/rpc/src/method/add_deploy_account_transaction.rs b/crates/rpc/src/method/add_deploy_account_transaction.rs index 54baf93d04..4ccf2bbf00 100644 --- a/crates/rpc/src/method/add_deploy_account_transaction.rs +++ b/crates/rpc/src/method/add_deploy_account_transaction.rs @@ -188,8 +188,8 @@ pub(crate) async fn add_deploy_account_transaction_impl( BroadcastedDeployAccountTransaction::V3(tx) => { let response = context .sequencer - .add_deploy_account(add_transaction::DeployAccount::V3( - add_transaction::DeployAccountV3 { + .add_deploy_account( + add_transaction::DeployAccount::V3(add_transaction::DeployAccountV3 { signature: &tx.signature, nonce: tx.nonce, nonce_data_availability_mode: tx.nonce_data_availability_mode.into(), @@ -200,8 +200,9 @@ pub(crate) async fn add_deploy_account_transaction_impl( class_hash: tx.class_hash, contract_address_salt: tx.contract_address_salt, constructor_calldata: &tx.constructor_calldata, - }, - )) + }), + context.config.gateway_add_transaction_timeout, + ) .await?; let new_tx = DeployAccountTransactionV3 { contract_address: tx.deployed_contract_address(), diff --git a/crates/rpc/src/method/add_invoke_transaction.rs b/crates/rpc/src/method/add_invoke_transaction.rs index 7c8eb9c0ae..fc1d958a29 100644 --- a/crates/rpc/src/method/add_invoke_transaction.rs +++ b/crates/rpc/src/method/add_invoke_transaction.rs @@ -189,8 +189,8 @@ pub(crate) async fn add_invoke_transaction_impl( BroadcastedInvokeTransaction::V3(tx) => { let response = context .sequencer - .add_invoke_transaction(add_transaction::InvokeFunction::V3( - add_transaction::InvokeFunctionV3 { + .add_invoke_transaction( + add_transaction::InvokeFunction::V3(add_transaction::InvokeFunctionV3 { signature: &tx.signature, nonce: tx.nonce, nonce_data_availability_mode: tx.nonce_data_availability_mode.into(), @@ -203,8 +203,9 @@ pub(crate) async fn add_invoke_transaction_impl( account_deployment_data: &tx.account_deployment_data, proof_facts: &tx.proof_facts, proof: &tx.proof, - }, - )) + }), + context.config.gateway_add_transaction_timeout, + ) .await?; let new_tx = InvokeTransactionV3 { signature: tx.signature, From 1ff22475b0bad2013a694fc32f417af18689a355 Mon Sep 17 00:00:00 2001 From: Zaksans Date: Thu, 3 Sep 2026 21:33:06 +0200 Subject: [PATCH 3/5] fix(gateway): address follow-up review --- CHANGELOG.md | 4 +-- crates/common/src/test_utils.rs | 38 +++++++++++++++++----------- crates/gateway-client/src/metrics.rs | 12 --------- 3 files changed, 25 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0501f122ce..b345eba7bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,8 +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. Dropped request futures only decrement the gauge because graceful shutdown and disconnected clients are not gateway failures. -- Added `--rpc.gateway-add-transaction-timeout` (default 60s) for sequencer submissions made by `starknet_addInvokeTransaction`, `starknet_addDeclareTransaction`, and `starknet_addDeployAccountTransaction`. +- 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. diff --git a/crates/common/src/test_utils.rs b/crates/common/src/test_utils.rs index 2487364281..95608400c0 100644 --- a/crates/common/src/test_utils.rs +++ b/crates/common/src/test_utils.rs @@ -21,7 +21,7 @@ pub mod metrics { use std::borrow::Cow; use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; - use std::sync::{Arc, Mutex, RwLock}; + use std::sync::{Arc, RwLock}; use metrics::{ Counter, @@ -57,7 +57,7 @@ pub mod metrics { struct FakeCounterFn(AtomicU64); #[derive(Debug, Default)] - struct FakeGaugeFn(Mutex); + struct FakeGaugeFn(AtomicU64); impl Recorder for FakeRecorder { fn describe_counter(&self, _: KeyName, _: Option, _: SharedString) {} @@ -188,16 +188,16 @@ pub mod metrics { method_name: impl Into>, ) -> f64 { let gauges = self.gauges.read().unwrap(); - let value = *gauges - .get(&Key::from_parts( - gauge_name, - vec![Label::new("method", method_name.into())], - )) - .expect("Unregistered gauge name") - .0 - .lock() - .unwrap(); - value + 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), + ) } } @@ -212,15 +212,23 @@ pub mod metrics { impl GaugeFn for FakeGaugeFn { fn increment(&self, val: f64) { - *self.0.lock().unwrap() += val; + 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.lock().unwrap() -= val; + 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.lock().unwrap() = val; + self.0.store(val.to_bits(), Ordering::Relaxed); } } } diff --git a/crates/gateway-client/src/metrics.rs b/crates/gateway-client/src/metrics.rs index e69ae90e3d..55546ac3a6 100644 --- a/crates/gateway-client/src/metrics.rs +++ b/crates/gateway-client/src/metrics.rs @@ -171,7 +171,6 @@ struct InFlightRequest { meta: RequestMetadata, started: std::time::Instant, in_flight: metrics::Gauge, - finished: bool, } impl InFlightRequest { @@ -185,7 +184,6 @@ impl InFlightRequest { meta, started: std::time::Instant::now(), in_flight, - finished: false, } } @@ -222,8 +220,6 @@ impl InFlightRequest { } fn finish_timing(&mut self) { - self.finished = true; - self.in_flight.decrement(1.0); metrics::histogram!(METRIC_REQUESTS_LATENCY, "method" => self.meta.method) .record(self.started.elapsed().as_secs_f64()); } @@ -231,14 +227,6 @@ impl InFlightRequest { impl Drop for InFlightRequest { fn drop(&mut self) { - if self.finished { - return; - } - - // A dropped request future is not necessarily a gateway failure: it is - // also how graceful shutdown and disconnected RPC clients cancel work. - // Keep the live gauge accurate without manufacturing latency or failure - // samples for a request that never completed. self.in_flight.decrement(1.0); } } From 9822e0a35067aed44918db5893dd418099c85881 Mon Sep 17 00:00:00 2001 From: Zaksans Date: Mon, 7 Sep 2026 21:21:08 +0200 Subject: [PATCH 4/5] refactor(gateway): inline request timing recording --- crates/gateway-client/src/metrics.rs | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/crates/gateway-client/src/metrics.rs b/crates/gateway-client/src/metrics.rs index 55546ac3a6..1569c7bb63 100644 --- a/crates/gateway-client/src/metrics.rs +++ b/crates/gateway-client/src/metrics.rs @@ -1,8 +1,8 @@ //! Metrics related utilities use futures::Future; -use super::builder::stage::Method; use super::builder::Request; +use super::builder::stage::Method; use super::{BlockId, SequencerError}; const METRIC_REQUESTS: &str = "gateway_requests_total"; @@ -188,7 +188,8 @@ impl InFlightRequest { } fn finish(&mut self, result: &Result) { - self.finish_timing(); + metrics::histogram!(METRIC_REQUESTS_LATENCY, "method" => self.meta.method) + .record(self.started.elapsed().as_secs_f64()); let Err(error) = result else { return; @@ -218,11 +219,6 @@ impl InFlightRequest { SequencerError::ReqwestError(_) | SequencerError::GatewayRequestCreationError(_) => {} } } - - fn finish_timing(&mut self) { - metrics::histogram!(METRIC_REQUESTS_LATENCY, "method" => self.meta.method) - .record(self.started.elapsed().as_secs_f64()); - } } impl Drop for InFlightRequest { From 500ef87300e6c48e7a617fd351c9d4db78d2655a Mon Sep 17 00:00:00 2001 From: Zaksans Date: Mon, 7 Sep 2026 21:22:25 +0200 Subject: [PATCH 5/5] style(gateway): retain edition 2021 import order --- crates/gateway-client/src/metrics.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/gateway-client/src/metrics.rs b/crates/gateway-client/src/metrics.rs index 1569c7bb63..71fdb7f129 100644 --- a/crates/gateway-client/src/metrics.rs +++ b/crates/gateway-client/src/metrics.rs @@ -1,8 +1,8 @@ //! Metrics related utilities use futures::Future; -use super::builder::Request; use super::builder::stage::Method; +use super::builder::Request; use super::{BlockId, SequencerError}; const METRIC_REQUESTS: &str = "gateway_requests_total";