diff --git a/crates/starknet_transaction_prover/src/main.rs b/crates/starknet_transaction_prover/src/main.rs index 4e7f9e03dc3..a445baf57b6 100644 --- a/crates/starknet_transaction_prover/src/main.rs +++ b/crates/starknet_transaction_prover/src/main.rs @@ -11,6 +11,7 @@ fn main() { async fn main() -> anyhow::Result<()> { use std::net::SocketAddr; use std::sync::Arc; + use std::time::Duration; use anyhow::Context; use clap::Parser; @@ -21,11 +22,13 @@ async fn main() -> anyhow::Result<()> { TransportMode, }; use starknet_transaction_prover::server::cors::{build_cors_layer, cors_mode}; + use starknet_transaction_prover::server::health::HealthLayer; use starknet_transaction_prover::server::log_redact::redact_url_host; use starknet_transaction_prover::server::metrics::install_exporter; use starknet_transaction_prover::server::panic::install_panic_hook; use starknet_transaction_prover::server::rpc_api::ProvingRpcServer; use starknet_transaction_prover::server::rpc_impl::ProvingRpcServerImpl; + use starknet_transaction_prover::server::saturation::SaturationMonitor; use starknet_transaction_prover::server::shutdown::spawn_signal_bridge; use starknet_transaction_prover::server::{ start_server, @@ -77,8 +80,16 @@ async fn main() -> anyhow::Result<()> { "Starting Starknet transaction prover." ); + // Shared saturation tracker — writer is `ProvingRpcServerImpl` on every + // permit acquire/reject; reader is `HealthLayer`. See `saturation.rs`. + let saturation_monitor = SaturationMonitor::default(); + let health_layer = HealthLayer::new( + saturation_monitor.clone(), + Duration::from_millis(config.health_max_saturated_ms), + ); + // Build and start the JSON-RPC server. - let rpc_impl = ProvingRpcServerImpl::from_config(&config); + let rpc_impl = ProvingRpcServerImpl::from_config(&config, saturation_monitor); let addr = SocketAddr::new(config.ip, config.port); let cors_layer = build_cors_layer(&config.cors_allow_origin)?; @@ -110,6 +121,7 @@ async fn main() -> anyhow::Result<()> { cors_layer, ohttp_layer, metrics_layer, + health_layer, ) .await?; diff --git a/crates/starknet_transaction_prover/src/server.rs b/crates/starknet_transaction_prover/src/server.rs index d46ddd51c37..71ad820f8d5 100644 --- a/crates/starknet_transaction_prover/src/server.rs +++ b/crates/starknet_transaction_prover/src/server.rs @@ -53,10 +53,10 @@ pub const OHTTP_JSONRPSEE_BODY_BUILDER: fn(Full) -> HttpBody = HttpBody:: /// - `RequestSpanLayer` sits BELOW `OhttpLayer` so it spans the decapsulated inner request with a /// fresh, envelope-unlinkable id (see `request_span`). macro_rules! prover_http_middleware { - ($metrics_layer:expr, $cors_layer:expr, $ohttp_layer:expr $(,)?) => { + ($health_layer:expr, $metrics_layer:expr, $cors_layer:expr, $ohttp_layer:expr $(,)?) => { ServiceBuilder::new() .layer(RequestLogLayer) - .layer(HealthLayer) + .layer($health_layer) .option_layer($metrics_layer) .layer(HttpMetricsLayer) .option_layer($cors_layer) @@ -84,6 +84,7 @@ pub mod request_log; pub mod request_span; pub mod rpc_api; pub mod rpc_impl; +pub mod saturation; pub mod shutdown; #[cfg(test)] pub mod test_recorder; @@ -94,6 +95,7 @@ pub use http_metrics::HttpMetricsLayer; pub use metrics::{MetricsLayer, METRICS_PATH}; pub use request_log::{RequestLogLayer, REQUEST_ID_HEADER}; pub use request_span::RequestSpanLayer; +pub use saturation::SaturationMonitor; #[cfg(test)] mod rpc_spec_test; @@ -116,6 +118,7 @@ pub async fn start_server( cors_layer: Option, ohttp_layer: Option, metrics_layer: Option, + health_layer: HealthLayer, ) -> anyhow::Result<(SocketAddr, ServerHandle)> { match transport { TransportMode::Http => { @@ -126,7 +129,12 @@ pub async fn start_server( let server = ServerBuilder::default() .set_config(server_config) // See `prover_http_middleware!` for the full layer-order rationale. - .set_http_middleware(prover_http_middleware!(metrics_layer, cors_layer, ohttp_layer)) + .set_http_middleware(prover_http_middleware!( + health_layer, + metrics_layer, + cors_layer, + ohttp_layer + )) .build(&addr) .await .context(format!("Failed to bind JSON-RPC server to {addr}"))?; @@ -145,6 +153,7 @@ pub async fn start_server( cors_layer, ohttp_layer, metrics_layer, + health_layer, ) .await } diff --git a/crates/starknet_transaction_prover/src/server/config.rs b/crates/starknet_transaction_prover/src/server/config.rs index 80a0a44daa7..50706752884 100644 --- a/crates/starknet_transaction_prover/src/server/config.rs +++ b/crates/starknet_transaction_prover/src/server/config.rs @@ -36,6 +36,10 @@ const DEFAULT_COMPILED_CLASS_CACHE_SIZE: usize = 600; /// 5 MiB — matches the convention used elsewhere in the sequencer. pub(crate) const DEFAULT_MAX_REQUEST_BODY_SIZE: u32 = 5 * 1024 * 1024; const DEFAULT_OHTTP_KEY_CACHE_MAX_AGE_SECS: u64 = 3600; +/// Default saturation window before `/health` returns 503. 10 seconds +/// matches "service is rejecting requests for a sustained period" without +/// flipping on a single in-flight burst. +const DEFAULT_HEALTH_MAX_SATURATED_MS: u64 = 10_000; /// Transport mode for the JSON-RPC server. #[derive(Clone, Debug)] @@ -103,6 +107,7 @@ struct RawServiceConfig { max_request_body_size: u32, ohttp_enabled: bool, ohttp_key_cache_max_age_secs: u64, + health_max_saturated_ms: u64, } impl Default for RawServiceConfig { @@ -130,6 +135,7 @@ impl Default for RawServiceConfig { max_request_body_size: DEFAULT_MAX_REQUEST_BODY_SIZE, ohttp_enabled: false, ohttp_key_cache_max_age_secs: DEFAULT_OHTTP_KEY_CACHE_MAX_AGE_SECS, + health_max_saturated_ms: DEFAULT_HEALTH_MAX_SATURATED_MS, } } } @@ -167,6 +173,11 @@ pub struct ServiceConfig { pub ohttp_enabled: bool, /// Cache-Control max-age for the `GET /ohttp-keys` response (seconds). pub ohttp_key_cache_max_age_secs: u64, + /// Saturation window (milliseconds) before `/health` flips to 503. The + /// service is "saturated" when it has been continuously rejecting + /// proving requests due to the concurrency limit; once that has held + /// for this many milliseconds, load balancers should drain the pod. + pub health_max_saturated_ms: u64, } /// Applies an optional CLI override to a config field, logging `old -> new` when it changes. @@ -370,6 +381,11 @@ impl ServiceConfig { &mut config.ohttp_key_cache_max_age_secs, args.ohttp_key_cache_max_age_secs, ); + override_field( + "health_max_saturated_ms", + &mut config.health_max_saturated_ms, + args.health_max_saturated_ms, + ); // Validate required fields. if config.rpc_node_url.is_empty() { @@ -466,6 +482,7 @@ impl ServiceConfig { max_request_body_size: config.max_request_body_size, ohttp_enabled: config.ohttp_enabled, ohttp_key_cache_max_age_secs: config.ohttp_key_cache_max_age_secs, + health_max_saturated_ms: config.health_max_saturated_ms, }) } } @@ -584,6 +601,14 @@ pub struct CliArgs { #[arg(long, value_enum, value_name = "FORMAT", env = "LOG_FORMAT", default_value_t = LogFormat::Text)] pub log_format: LogFormat, + /// Saturation window (milliseconds) before `/health` returns 503 + /// (default: 10000). The service is "saturated" when it has been + /// continuously rejecting proving requests due to the concurrency + /// limit; once that has held for this many milliseconds, load + /// balancers should drain the pod. + #[arg(long, value_name = "MILLIS", env = "HEALTH_MAX_SATURATED_MS")] + pub health_max_saturated_ms: Option, + /// Hidden escape hatch: override the embedded bouncer config (block capacity limits) with a /// custom JSON file. Not advertised because the embedded defaults are tuned for this prover /// (including high `l1_gas` / `message_segment_length`: virtual OS output is not L1-bound; it diff --git a/crates/starknet_transaction_prover/src/server/config_test.rs b/crates/starknet_transaction_prover/src/server/config_test.rs index fc800836e65..aceb06b2194 100644 --- a/crates/starknet_transaction_prover/src/server/config_test.rs +++ b/crates/starknet_transaction_prover/src/server/config_test.rs @@ -62,6 +62,7 @@ fn base_args() -> CliArgs { ohttp_enabled: false, ohttp_key_cache_max_age_secs: None, log_format: LogFormat::Text, + health_max_saturated_ms: None, } } @@ -153,6 +154,7 @@ fn cors_allow_origin_rejects_non_array_in_config_file() { ohttp_enabled: false, ohttp_key_cache_max_age_secs: None, log_format: LogFormat::Text, + health_max_saturated_ms: None, }; let error = ServiceConfig::from_args(args).unwrap_err(); diff --git a/crates/starknet_transaction_prover/src/server/health.rs b/crates/starknet_transaction_prover/src/server/health.rs index ec49cf8a59c..9f26d61e60d 100644 --- a/crates/starknet_transaction_prover/src/server/health.rs +++ b/crates/starknet_transaction_prover/src/server/health.rs @@ -1,9 +1,13 @@ //! HTTP `/health` endpoint as a tower middleware layer. //! //! Short-circuits `GET /health` before the jsonrpsee service sees the request -//! (which would 405 a GET). Any other request passes through unchanged. +//! (which would 405 a GET). Returns 503 once its `SaturationMonitor` reports the +//! service has been continuously rejecting requests for the configured threshold, +//! so load balancers can drain the pod. Body is opaque (no timestamps, counters, +//! or upstream URLs). use std::task::{Context, Poll}; +use std::time::Duration; use bytes::Bytes; use futures::future::{ready, Either, Ready}; @@ -12,6 +16,8 @@ use http_body_util::Full; use jsonrpsee::server::HttpBody; use tower::{Layer, Service}; +use crate::server::saturation::SaturationMonitor; + #[cfg(test)] #[path = "health_test.rs"] mod health_test; @@ -19,21 +25,57 @@ mod health_test; pub const HEALTH_PATH: &str = "/health"; const HEALTHY_BODY: &[u8] = br#"{"status":"ok"}"#; +/// Body returned by `GET /health` when saturated. Reason is an opaque code, +/// no internal state included. +const SATURATED_BODY: &[u8] = br#"{"status":"unhealthy","reason":"saturated"}"#; -#[derive(Clone, Copy, Default)] -pub struct HealthLayer; +/// Returns `503` once `saturation.saturated_for_at_least` crosses +/// `saturation_threshold`, and `200` otherwise. Tests that only need the healthy +/// path pass a fresh `SaturationMonitor::default()`, which never reports saturated. +#[derive(Clone)] +pub struct HealthLayer { + saturation: SaturationMonitor, + saturation_threshold: Duration, +} + +impl HealthLayer { + pub fn new(monitor: SaturationMonitor, threshold: Duration) -> Self { + Self { saturation: monitor, saturation_threshold: threshold } + } +} impl Layer for HealthLayer { type Service = HealthService; fn layer(&self, inner: S) -> Self::Service { - HealthService { inner } + HealthService { + inner, + saturation: self.saturation.clone(), + saturation_threshold: self.saturation_threshold, + } } } #[derive(Clone)] pub struct HealthService { inner: S, + saturation: SaturationMonitor, + saturation_threshold: Duration, +} + +impl HealthService { + fn health_response(&self) -> Response { + let (status, body) = if self.saturation.saturated_for_at_least(self.saturation_threshold) { + (StatusCode::SERVICE_UNAVAILABLE, SATURATED_BODY) + } else { + (StatusCode::OK, HEALTHY_BODY) + }; + Response::builder() + .status(status) + .header(header::CONTENT_TYPE, "application/json") + .body(HttpBody::new(Full::new(Bytes::from_static(body)))) + .expect("response build with a static body is infallible") + } } impl Service> for HealthService @@ -52,12 +94,7 @@ where fn call(&mut self, request: Request) -> Self::Future { if request.method() == Method::GET && request.uri().path() == HEALTH_PATH { - let response = Response::builder() - .status(StatusCode::OK) - .header(header::CONTENT_TYPE, "application/json") - .body(HttpBody::new(Full::new(Bytes::from_static(HEALTHY_BODY)))) - .expect("response build with a static body is infallible"); - return Either::Left(ready(Ok(response))); + return Either::Left(ready(Ok(self.health_response()))); } Either::Right(self.inner.call(request)) } diff --git a/crates/starknet_transaction_prover/src/server/health_test.rs b/crates/starknet_transaction_prover/src/server/health_test.rs index 288855aa25e..2d2bdcf93a4 100644 --- a/crates/starknet_transaction_prover/src/server/health_test.rs +++ b/crates/starknet_transaction_prover/src/server/health_test.rs @@ -1,3 +1,5 @@ +use std::time::Duration; + use bytes::Bytes; use http::{Method, Request, Response, StatusCode}; use http_body_util::{BodyExt, Full}; @@ -5,6 +7,7 @@ use jsonrpsee::server::HttpBody; use tower::{Layer, ServiceExt}; use crate::server::health::{HealthLayer, HEALTH_PATH}; +use crate::server::saturation::SaturationMonitor; /// Inner stub returning 418 so we can tell whether `HealthLayer` short-circuited. fn fallthrough_service() -> impl tower::Service< @@ -38,7 +41,8 @@ async fn read_body(response: Response) -> (StatusCode, Vec, http:: #[tokio::test] async fn get_health_returns_200_with_json_body() { - let svc = HealthLayer.layer(fallthrough_service()); + let svc = HealthLayer::new(SaturationMonitor::default(), Duration::from_millis(0)) + .layer(fallthrough_service()); let response = svc.oneshot(empty_request(Method::GET, HEALTH_PATH)).await.unwrap(); @@ -50,7 +54,8 @@ async fn get_health_returns_200_with_json_body() { #[tokio::test] async fn non_get_health_falls_through() { - let svc = HealthLayer.layer(fallthrough_service()); + let svc = HealthLayer::new(SaturationMonitor::default(), Duration::from_millis(0)) + .layer(fallthrough_service()); let response = svc.oneshot(empty_request(Method::POST, HEALTH_PATH)).await.unwrap(); @@ -60,10 +65,55 @@ async fn non_get_health_falls_through() { #[tokio::test] async fn get_other_path_falls_through() { - let svc = HealthLayer.layer(fallthrough_service()); + let svc = HealthLayer::new(SaturationMonitor::default(), Duration::from_millis(0)) + .layer(fallthrough_service()); let response = svc.oneshot(empty_request(Method::GET, "/")).await.unwrap(); let (status, _body, _) = read_body(response).await; assert_eq!(status, StatusCode::IM_A_TEAPOT); } + +#[tokio::test] +async fn unsaturated_health_returns_200_when_monitor_is_supplied() { + let monitor = SaturationMonitor::default(); + let svc = HealthLayer::new(monitor, Duration::from_millis(0)).layer(fallthrough_service()); + + let response = svc.oneshot(empty_request(Method::GET, HEALTH_PATH)).await.unwrap(); + + let (status, body, _) = read_body(response).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body, br#"{"status":"ok"}"#); +} + +#[tokio::test] +async fn saturated_for_at_least_threshold_returns_503_with_opaque_body() { + let monitor = SaturationMonitor::default(); + monitor.mark_rejected(); + // Zero threshold so the saturation is immediately past it. Avoids + // sleeping in the test. + let svc = HealthLayer::new(monitor, Duration::from_millis(0)).layer(fallthrough_service()); + + let response = svc.oneshot(empty_request(Method::GET, HEALTH_PATH)).await.unwrap(); + + let (status, body, _) = read_body(response).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + let body_text = std::str::from_utf8(&body).unwrap(); + assert!(body_text.contains("saturated")); + // No internal state — no timestamps, no permits, no upstream URLs. + assert!(!body_text.contains("Instant")); + assert!(!body_text.chars().any(|c| c.is_ascii_digit()), "body had digits: {body_text}"); +} + +#[tokio::test] +async fn recovery_clears_saturation_and_health_returns_to_200() { + let monitor = SaturationMonitor::default(); + monitor.mark_rejected(); + monitor.mark_accepted(); + let svc = HealthLayer::new(monitor, Duration::from_millis(0)).layer(fallthrough_service()); + + let response = svc.oneshot(empty_request(Method::GET, HEALTH_PATH)).await.unwrap(); + let (status, body, _) = read_body(response).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body, br#"{"status":"ok"}"#); +} diff --git a/crates/starknet_transaction_prover/src/server/metrics.rs b/crates/starknet_transaction_prover/src/server/metrics.rs index 84a40f328e4..f392cb59b20 100644 --- a/crates/starknet_transaction_prover/src/server/metrics.rs +++ b/crates/starknet_transaction_prover/src/server/metrics.rs @@ -39,6 +39,10 @@ pub mod names { pub const OS_RUN_DURATION_SECONDS: &str = "prover_os_run_duration_seconds"; /// Stwo proving sub-step duration. Bucketed. pub const STWO_PROVE_DURATION_SECONDS: &str = "prover_stwo_prove_duration_seconds"; + /// Requests admitted to the queue but not yet running (waiting for a worker slot). Gauge. + pub const QUEUE_WAITING_REQUESTS: &str = "prover_queue_waiting_requests"; + /// Time a request waited in the queue before acquiring a worker slot. Bucketed. + pub const QUEUE_WAIT_DURATION_SECONDS: &str = "prover_queue_wait_duration_seconds"; } /// Fixed, bounded set of values for the `outcome` label on @@ -76,6 +80,19 @@ pub fn install_exporter(version: &str, git_sha: &str) -> anyhow::Result outcomes::REJECTED_QUEUE_FULL, + ) + .increment(0); + metrics::counter!( + names::PROVE_TRANSACTION_OUTCOME_TOTAL, + "outcome" => outcomes::REJECTED_WAIT_TIMEOUT, + ) + .increment(0); Ok(handle) } diff --git a/crates/starknet_transaction_prover/src/server/ohttp_integration_test.rs b/crates/starknet_transaction_prover/src/server/ohttp_integration_test.rs index fe637c40b4e..3dfed12224b 100644 --- a/crates/starknet_transaction_prover/src/server/ohttp_integration_test.rs +++ b/crates/starknet_transaction_prover/src/server/ohttp_integration_test.rs @@ -12,6 +12,7 @@ //! (enabled via the crate's `testing` feature in this crate's dev-deps). use std::io::Read; +use std::time::Duration; use flate2::read::GzDecoder; use http::header; @@ -31,6 +32,7 @@ use tower_ohttp::OhttpLayer; use crate::server::request_log::{RequestLogLayer, REQUEST_ID_HEADER}; use crate::server::request_span::RequestSpanLayer; +use crate::server::saturation::SaturationMonitor; use crate::server::{HealthLayer, HttpMetricsLayer, MetricsLayer, OHTTP_JSONRPSEE_BODY_BUILDER}; const DEFAULT_BODY_LIMIT: usize = 102_400; @@ -100,9 +102,13 @@ async fn production_chain_compresses_inner_not_outer() { OHTTP_JSONRPSEE_BODY_BUILDER, ); - let mut svc = - prover_http_middleware!(None::, None::, Some(ohttp_layer)) - .service(tower::service_fn(jsonrpsee_echo_service)); + let mut svc = prover_http_middleware!( + HealthLayer::new(SaturationMonitor::default(), Duration::from_millis(0)), + None::, + None::, + Some(ohttp_layer) + ) + .service(tower::service_fn(jsonrpsee_echo_service)); // Body must be large enough for gzip to actually compress. let body_json = serde_json::json!({ @@ -207,9 +213,13 @@ async fn ohttp_inner_request_id_unlinkable_from_envelope() { ) }); - let mut svc = - prover_http_middleware!(None::, None::, Some(ohttp_layer)) - .service(echo_id); + let mut svc = prover_http_middleware!( + HealthLayer::new(SaturationMonitor::default(), Duration::from_millis(0)), + None::, + None::, + Some(ohttp_layer) + ) + .service(echo_id); // The envelope carries a client-chosen inner id that must be discarded. let (encapsulated, client_response) = encapsulate_bhttp_request( diff --git a/crates/starknet_transaction_prover/src/server/request_body_size_test.rs b/crates/starknet_transaction_prover/src/server/request_body_size_test.rs index 457505bea7d..98619fad0b0 100644 --- a/crates/starknet_transaction_prover/src/server/request_body_size_test.rs +++ b/crates/starknet_transaction_prover/src/server/request_body_size_test.rs @@ -18,8 +18,10 @@ use starknet_api::transaction::fields::Calldata; use starknet_types_core::felt::Felt; use crate::server::config::{TransportMode, DEFAULT_MAX_REQUEST_BODY_SIZE}; +use crate::server::health::HealthLayer; use crate::server::mock_rpc::MockProvingRpc; use crate::server::rpc_api::ProvingRpcServer; +use crate::server::saturation::SaturationMonitor; use crate::server::start_server; const NUM_CALLDATA_FELTS: usize = 5_000; @@ -36,6 +38,7 @@ async fn start_test_http_server(max_request_body_size: u32) -> (SocketAddr, Serv None, None, None, + HealthLayer::new(SaturationMonitor::default(), std::time::Duration::from_millis(0)), ) .await .expect("Failed to start HTTP server") diff --git a/crates/starknet_transaction_prover/src/server/rpc_impl.rs b/crates/starknet_transaction_prover/src/server/rpc_impl.rs index 71eb158b02f..d18f6d84b05 100644 --- a/crates/starknet_transaction_prover/src/server/rpc_impl.rs +++ b/crates/starknet_transaction_prover/src/server/rpc_impl.rs @@ -1,7 +1,7 @@ //! JSON-RPC trait implementation for the proving service. use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, Instant}; use async_trait::async_trait; use blockifier_reexecution::state_reader::rpc_objects::BlockId; @@ -15,7 +15,16 @@ use tracing::warn; use crate::proving::virtual_snos_prover::{ProveTransactionResult, RpcVirtualSnosProver}; use crate::server::config::ServiceConfig; use crate::server::errors::{internal_server_error, service_busy}; +use crate::server::metrics::{names as metric_names, outcomes}; use crate::server::rpc_api::ProvingRpcServer; +use crate::server::saturation::SaturationMonitor; + +// `dummy_prover()` builds an `RpcVirtualSnosProver`, which prepares recursive-prover precomputes +// under `stwo_proving`; the reject paths under test are feature-independent, so gate the module to +// the non-proving config to keep it fast. +#[cfg(all(test, not(feature = "stwo_proving")))] +#[path = "rpc_impl_test.rs"] +mod rpc_impl_test; /// Starknet RPC specification version (matches the pinned `starknet_specs_rev`). pub(crate) const SPEC_VERSION: &str = "0.10.3-rc.2"; @@ -32,6 +41,9 @@ pub struct ProvingRpcServerImpl { max_concurrent_requests: usize, /// Backstop on the FIFO wait so a stuck worker can't pin a waiter's connection indefinitely. queue_wait_timeout: Duration, + /// Tracks how long the service has been rejecting requests so + /// `/health` can flip to 503. Cloned cheaply (Arc internally). + saturation_monitor: SaturationMonitor, } impl ProvingRpcServerImpl { @@ -41,6 +53,7 @@ impl ProvingRpcServerImpl { max_concurrent_requests: usize, max_queued_requests: usize, queue_wait_timeout: Duration, + saturation_monitor: SaturationMonitor, ) -> Self { Self { prover, @@ -50,17 +63,19 @@ impl ProvingRpcServerImpl { )), max_concurrent_requests, queue_wait_timeout, + saturation_monitor, } } /// Creates a new ProvingRpcServerImpl from configuration. - pub fn from_config(config: &ServiceConfig) -> Self { + pub fn from_config(config: &ServiceConfig, saturation_monitor: SaturationMonitor) -> Self { let prover = RpcVirtualSnosProver::new(&config.prover_config); Self::new( prover, config.max_concurrent_requests, config.max_queued_requests, Duration::from_millis(config.queue_wait_timeout_millis), + saturation_monitor, ) } } @@ -77,8 +92,16 @@ impl ProvingRpcServer for ProvingRpcServerImpl { transaction: RpcTransaction, ) -> RpcResult { // Admission: cap queue length (running + waiting). Reject with -32005 only when the queue - // is full; held for the whole request, so a client disconnect frees the slot. + // is full; held for the whole request, so a client disconnect frees the slot. Busy-rejects + // are folded into the outcome counter (as a `rejected_*` outcome) so every request — + // served or shed — shares one denominator for rate calculations. let _admission = self.admission_semaphore.try_acquire().map_err(|_| { + metrics::counter!( + metric_names::PROVE_TRANSACTION_OUTCOME_TOTAL, + "outcome" => outcomes::REJECTED_QUEUE_FULL, + ) + .increment(1); + self.saturation_monitor.mark_rejected(); warn!( max_concurrent_requests = self.max_concurrent_requests, "Rejected proving request: queue is full" @@ -87,18 +110,37 @@ impl ProvingRpcServer for ProvingRpcServerImpl { })?; // Wait FIFO for a worker slot (tokio's Semaphore is fair), with queue_wait_timeout as a - // backstop. Served in arrival order, or cancelled if the client disconnects. - let _permit = match timeout(self.queue_wait_timeout, self.concurrency_semaphore.acquire()) - .await - { - Ok(Ok(permit)) => permit, - Ok(Err(_)) => return Err(internal_server_error("proving service is shutting down")), - Err(_) => { - warn!( - max_concurrent_requests = self.max_concurrent_requests, - "Rejected proving request: timed out waiting for a worker slot" - ); - return Err(service_busy(self.max_concurrent_requests)); + // backstop. `QueueWaitingGuard` decrements the queue-depth gauge on every exit from the + // wait — slot acquired, timeout, shutdown, or client disconnect — so the gauge can't leak. + let wait_start = Instant::now(); + let _permit = { + metrics::gauge!(metric_names::QUEUE_WAITING_REQUESTS).increment(1.0); + let _waiting_guard = QueueWaitingGuard; + match timeout(self.queue_wait_timeout, self.concurrency_semaphore.acquire()).await { + Ok(Ok(permit)) => { + metrics::histogram!(metric_names::QUEUE_WAIT_DURATION_SECONDS) + .record(wait_start.elapsed().as_secs_f64()); + // A worker slot opened up and this request is about to prove — clear any + // saturation window so /health can recover. + self.saturation_monitor.mark_accepted(); + permit + } + Ok(Err(_)) => { + return Err(internal_server_error("proving service is shutting down")); + } + Err(_) => { + metrics::counter!( + metric_names::PROVE_TRANSACTION_OUTCOME_TOTAL, + "outcome" => outcomes::REJECTED_WAIT_TIMEOUT, + ) + .increment(1); + self.saturation_monitor.mark_rejected(); + warn!( + max_concurrent_requests = self.max_concurrent_requests, + "Rejected proving request: timed out waiting for a worker slot" + ); + return Err(service_busy(self.max_concurrent_requests)); + } } }; @@ -108,3 +150,14 @@ impl ProvingRpcServer for ProvingRpcServerImpl { }) } } + +/// Decrements [`metric_names::QUEUE_WAITING_REQUESTS`] on drop. Using a guard rather than an +/// explicit decrement covers the timeout, shutdown, and cancellation (client-disconnect) paths so +/// the gauge always returns to its true depth. +struct QueueWaitingGuard; + +impl Drop for QueueWaitingGuard { + fn drop(&mut self) { + metrics::gauge!(metric_names::QUEUE_WAITING_REQUESTS).decrement(1.0); + } +} diff --git a/crates/starknet_transaction_prover/src/server/rpc_impl_test.rs b/crates/starknet_transaction_prover/src/server/rpc_impl_test.rs new file mode 100644 index 00000000000..d994dd0eb81 --- /dev/null +++ b/crates/starknet_transaction_prover/src/server/rpc_impl_test.rs @@ -0,0 +1,95 @@ +//! Tests for the admission/queue reject paths in [`ProvingRpcServerImpl`]. +//! +//! Both busy-reject outcomes are exercised deterministically without a live node or a real +//! proving run by sizing the semaphores so the reject fires before the prover is ever called: +//! zero admission capacity forces a queue-full reject; zero worker slots with a tiny wait timeout +//! forces a wait-timeout reject. + +use std::time::Duration; + +use blockifier_reexecution::state_reader::rpc_objects::BlockId; +use blockifier_test_utils::calldata::create_calldata; +use starknet_api::core::ContractAddress; +use starknet_api::rpc_transaction::RpcTransaction; + +use crate::config::ProverConfig; +use crate::proving::virtual_snos_prover::RpcVirtualSnosProver; +use crate::server::metrics::{names as metric_names, outcomes}; +use crate::server::rpc_api::ProvingRpcServer; +use crate::server::rpc_impl::ProvingRpcServerImpl; +use crate::server::saturation::SaturationMonitor; +use crate::server::test_recorder::{metric_value, shared_handle}; +use crate::test_utils::{build_client_side_rpc_invoke, DUMMY_ACCOUNT_ADDRESS}; + +/// JSON-RPC error code returned by `service_busy` (see `server::errors`). +const SERVICE_BUSY_CODE: i32 = -32005; + +fn dummy_prover() -> RpcVirtualSnosProver { + let config = + ProverConfig { rpc_node_url: "http://localhost:1".to_string(), ..Default::default() }; + RpcVirtualSnosProver::new(&config) +} + +/// The reject fires at admission/wait, before the transaction is inspected, so any request works. +fn dummy_request() -> RpcTransaction { + let account = ContractAddress::try_from(DUMMY_ACCOUNT_ADDRESS).unwrap(); + build_client_side_rpc_invoke(account, create_calldata(account, "noop", &[])) +} + +fn outcome_line(outcome: &str) -> String { + format!("{}{{outcome=\"{}\"}}", metric_names::PROVE_TRANSACTION_OUTCOME_TOTAL, outcome) +} + +#[tokio::test] +async fn full_queue_rejects_with_service_busy_and_counts_queue_full() { + let handle = shared_handle(); + let line = outcome_line(outcomes::REJECTED_QUEUE_FULL); + let before = metric_value(&handle.render(), &line); + + // max_concurrent + max_queued = 0 → admission capacity 0 → every request is shed at admission. + let rpc_impl = ProvingRpcServerImpl::new( + dummy_prover(), + 0, + 0, + Duration::from_secs(30), + SaturationMonitor::default(), + ); + let error = rpc_impl + .prove_transaction(BlockId::Latest, dummy_request()) + .await + .expect_err("a full queue must reject"); + + assert_eq!(error.code(), SERVICE_BUSY_CODE); + assert_eq!(metric_value(&handle.render(), &line) - before, 1.0, "rejected_queue_full delta"); +} + +#[tokio::test] +async fn wait_timeout_rejects_with_service_busy_and_counts_wait_timeout() { + let handle = shared_handle(); + let line = outcome_line(outcomes::REJECTED_WAIT_TIMEOUT); + let before = metric_value(&handle.render(), &line); + let gauge_before = metric_value(&handle.render(), metric_names::QUEUE_WAITING_REQUESTS); + + // One queue slot but zero worker slots, with a tiny backstop timeout: the request is admitted, + // waits for a worker that never frees, and is shed on the timeout. + let rpc_impl = ProvingRpcServerImpl::new( + dummy_prover(), + 0, + 1, + Duration::from_millis(10), + SaturationMonitor::default(), + ); + let error = rpc_impl + .prove_transaction(BlockId::Latest, dummy_request()) + .await + .expect_err("a wait-timeout must reject"); + + assert_eq!(error.code(), SERVICE_BUSY_CODE); + assert_eq!(metric_value(&handle.render(), &line) - before, 1.0, "rejected_wait_timeout delta"); + // The queue-depth guard ran on the timeout path, so the gauge returns to its prior value. + assert_eq!( + metric_value(&handle.render(), metric_names::QUEUE_WAITING_REQUESTS), + gauge_before, + "queue-depth gauge returned to baseline", + ); +} diff --git a/crates/starknet_transaction_prover/src/server/rpc_spec_test.rs b/crates/starknet_transaction_prover/src/server/rpc_spec_test.rs index e706bc503e1..c0e558cf5ae 100644 --- a/crates/starknet_transaction_prover/src/server/rpc_spec_test.rs +++ b/crates/starknet_transaction_prover/src/server/rpc_spec_test.rs @@ -127,6 +127,7 @@ fn rpc_module() -> RpcModule { TEST_MAX_CONCURRENT_REQUESTS, 0, std::time::Duration::from_secs(30), + crate::server::SaturationMonitor::default(), ); rpc_impl.into_rpc() } diff --git a/crates/starknet_transaction_prover/src/server/saturation.rs b/crates/starknet_transaction_prover/src/server/saturation.rs new file mode 100644 index 00000000000..2e823201e09 --- /dev/null +++ b/crates/starknet_transaction_prover/src/server/saturation.rs @@ -0,0 +1,64 @@ +//! Saturation tracking for the prover's concurrency-limited request path. +//! +//! Used by both `ProvingRpcServerImpl` (writer) and `HealthLayer` (reader) +//! so `/health` can flip to 503 once the service has been rejecting +//! requests for a sustained period. +//! +//! Mechanism: +//! - On every rejection from the concurrency semaphore: if the monitor is currently "clear", set a +//! timestamp marking when saturation started. +//! - On every successful acquire: clear the timestamp. +//! - `/health` consults [`SaturationMonitor::saturated_for_at_least`] with the configured threshold +//! and returns 503 if it passed. +//! +//! Saturation is therefore measured by *traffic that the service has tried +//! to serve*. With no traffic at all, the monitor reports healthy — there +//! is no saturation event to time. This mirrors the proof-interceptor's +//! upstream-reachability tracking (healthy until sustained failures cross a +//! threshold) so the two services behave the same way from a load-balancer's +//! perspective. + +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +#[cfg(test)] +#[path = "saturation_test.rs"] +mod saturation_test; + +/// Cheap-to-clone handle to the shared saturation state. The interior +/// `Arc>` makes both `mark_rejected`/`mark_accepted` and the +/// read path lock-free at the API surface. +#[derive(Clone, Default)] +pub struct SaturationMonitor { + state: Arc>>, +} + +impl SaturationMonitor { + /// Record a rejection. Starts the saturation window if this is the + /// first rejection since the last `mark_accepted` (or since startup). + pub fn mark_rejected(&self) { + let mut state = self.state.lock().expect("saturation lock poisoned"); + if state.is_none() { + *state = Some(Instant::now()); + } + } + + /// Record a successful acquire. Clears the saturation window so a + /// transient burst of rejections doesn't keep `/health` red forever + /// once the service recovers. + pub fn mark_accepted(&self) { + let mut state = self.state.lock().expect("saturation lock poisoned"); + *state = None; + } + + /// Returns true when the service has been continuously rejecting + /// requests for at least `threshold`. Returns false when the service + /// has handled at least one request successfully within the window or + /// has not seen any traffic at all. + pub fn saturated_for_at_least(&self, threshold: Duration) -> bool { + self.state + .lock() + .expect("saturation lock poisoned") + .is_some_and(|started_at| started_at.elapsed() >= threshold) + } +} diff --git a/crates/starknet_transaction_prover/src/server/saturation_test.rs b/crates/starknet_transaction_prover/src/server/saturation_test.rs new file mode 100644 index 00000000000..3f60b2876f1 --- /dev/null +++ b/crates/starknet_transaction_prover/src/server/saturation_test.rs @@ -0,0 +1,45 @@ +use std::thread::sleep; +use std::time::Duration; + +use crate::server::saturation::SaturationMonitor; + +#[test] +fn starts_healthy_before_any_traffic() { + let monitor = SaturationMonitor::default(); + assert!(!monitor.saturated_for_at_least(Duration::from_millis(0))); + assert!(!monitor.saturated_for_at_least(Duration::from_secs(10))); +} + +#[test] +fn rejection_starts_window_and_threshold_eventually_passes() { + let monitor = SaturationMonitor::default(); + monitor.mark_rejected(); + // Window has just opened — zero-elapsed comparison must still be true + // (we are at or past the 0ms threshold). + assert!(monitor.saturated_for_at_least(Duration::from_millis(0))); + // Not yet at the 50ms threshold — the rejection happened just now. + assert!(!monitor.saturated_for_at_least(Duration::from_millis(50))); + sleep(Duration::from_millis(60)); + assert!(monitor.saturated_for_at_least(Duration::from_millis(50))); +} + +#[test] +fn repeated_rejections_do_not_reset_the_window() { + let monitor = SaturationMonitor::default(); + monitor.mark_rejected(); + sleep(Duration::from_millis(30)); + // A second rejection should extend, not restart, the saturation + // window — operators care about "how long has this been bad", which + // is the time since the first rejection. + monitor.mark_rejected(); + assert!(monitor.saturated_for_at_least(Duration::from_millis(25))); +} + +#[test] +fn one_successful_acquire_clears_the_window() { + let monitor = SaturationMonitor::default(); + monitor.mark_rejected(); + sleep(Duration::from_millis(10)); + monitor.mark_accepted(); + assert!(!monitor.saturated_for_at_least(Duration::from_millis(0))); +} diff --git a/crates/starknet_transaction_prover/src/server/tls.rs b/crates/starknet_transaction_prover/src/server/tls.rs index 16534c829e7..05480cde08e 100644 --- a/crates/starknet_transaction_prover/src/server/tls.rs +++ b/crates/starknet_transaction_prover/src/server/tls.rs @@ -53,6 +53,7 @@ pub async fn start_tls_server( cors_layer: Option, ohttp_layer: Option, metrics_layer: Option, + health_layer: HealthLayer, ) -> anyhow::Result<(SocketAddr, ServerHandle)> { let tls_acceptor = load_tls_acceptor(cert_path, key_path)?; @@ -63,7 +64,12 @@ pub async fn start_tls_server( // See `prover_http_middleware!` for the full layer-order rationale. let svc_builder = ServerBuilder::default() .set_config(server_config) - .set_http_middleware(prover_http_middleware!(metrics_layer, cors_layer, ohttp_layer)) + .set_http_middleware(prover_http_middleware!( + health_layer, + metrics_layer, + cors_layer, + ohttp_layer + )) .to_service_builder(); let listener = TcpListener::bind(addr)