diff --git a/crates/starknet_transaction_prover/README.md b/crates/starknet_transaction_prover/README.md index bbba0572439..f60ae657450 100644 --- a/crates/starknet_transaction_prover/README.md +++ b/crates/starknet_transaction_prover/README.md @@ -303,10 +303,12 @@ scrapes bypass CORS and JSON-RPC parsing, and the endpoint is unauthenticated. | `prover_http_requests_total` | counter | `method`, `status` | HTTP request count. `method` is a bounded enum (`GET`/`POST`/`PUT`/`DELETE`/`HEAD`/`OPTIONS`/`PATCH`/`other`); `status` is the HTTP status class (`1xx` through `5xx`, plus `other`, and `error` when the tower stack produced no response). Excludes `/health` and `/metrics` probes. | | `prover_http_request_duration_seconds` | histogram | `method`, `status` | End-to-end HTTP request latency, using the same bounded `method` and `status` label values as the request counter. Excludes `/health` and `/metrics` probes. | | `prover_http_inflight_requests` | gauge | none | Current count of HTTP requests being handled. Decremented via RAII so panics and cancellations don't leak. | -| `prover_prove_transaction_outcome_total` | counter | `outcome` | Every proving request that reaches the prover, so its total is the shared denominator for all proving rates. `outcome` is a bounded enum: `success`, `failure_validation`, `failure_blocked`, `failure_runner`, `failure_output_parse`, `failure_proving`. | -| `prover_prove_transaction_duration_seconds` | histogram | `outcome` | Duration of the whole proving call, covering input validation, the optional blocking check, the virtual OS run and proving. Recorded for failures too, so filter on `outcome` for success-only percentiles. | +| `prover_prove_transaction_outcome_total` | counter | `outcome` | Every proving request, whether served or rejected, so its total is the shared denominator for all proving rates. `outcome` is a bounded enum: `success`, `failure_validation`, `failure_blocked`, `failure_runner`, `failure_output_parse`, `failure_proving`, `rejected_queue_full`, `rejected_wait_timeout`. The two `rejected_*` values are the busy-rejects, which return JSON-RPC error `-32005`. | +| `prover_prove_transaction_duration_seconds` | histogram | `outcome` | Duration of the whole proving call, covering input validation, the optional blocking check, the virtual OS run and proving. Recorded for failures too, so filter on `outcome` for success-only percentiles. Rejected requests never reach it. | | `prover_os_run_duration_seconds` | histogram | none | Virtual OS execution time, recorded for successful runs only. | | `prover_stwo_prove_duration_seconds` | histogram | none | STWO proving time, recorded for successful runs only. Emitted only by builds with the `stwo_proving` feature. | +| `prover_queue_waiting_requests` | gauge | none | Requests admitted to the queue but still waiting for a worker slot. Decremented via RAII on slot acquisition, timeout, or client disconnect. | +| `prover_queue_wait_duration_seconds` | histogram | none | Time a request waited in the queue before acquiring a worker slot (successful acquisitions only). | | `prover_panics_total` | counter | none | Process panics caught by the global panic hook. Lets an alert watch the panic rate without a log search. | No user-controlled value becomes a label, so label cardinality stays bounded. diff --git a/crates/starknet_transaction_prover/src/main.rs b/crates/starknet_transaction_prover/src/main.rs index 7bc1345f78f..3a514379069 100644 --- a/crates/starknet_transaction_prover/src/main.rs +++ b/crates/starknet_transaction_prover/src/main.rs @@ -26,6 +26,7 @@ async fn main() -> anyhow::Result<()> { 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, @@ -80,7 +81,8 @@ async fn main() -> anyhow::Result<()> { ); // Build and start the JSON-RPC server. - let rpc_impl = ProvingRpcServerImpl::from_config(&config); + let saturation_monitor = SaturationMonitor::default(); + 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)?; diff --git a/crates/starknet_transaction_prover/src/server.rs b/crates/starknet_transaction_prover/src/server.rs index 5f97ec12327..f53b9e77f98 100644 --- a/crates/starknet_transaction_prover/src/server.rs +++ b/crates/starknet_transaction_prover/src/server.rs @@ -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; diff --git a/crates/starknet_transaction_prover/src/server/metrics.rs b/crates/starknet_transaction_prover/src/server/metrics.rs index a4f872fbb4f..0fd7489276f 100644 --- a/crates/starknet_transaction_prover/src/server/metrics.rs +++ b/crates/starknet_transaction_prover/src/server/metrics.rs @@ -50,6 +50,7 @@ const DURATION_HISTOGRAM_BUCKETS: &[(&str, &[f64])] = &[ (names::OS_RUN_DURATION_SECONDS, PROVING_DURATION_BUCKETS), (names::STWO_PROVE_DURATION_SECONDS, PROVING_DURATION_BUCKETS), (http_metrics::names::REQUEST_DURATION_SECONDS, HTTP_DURATION_BUCKETS), + (names::QUEUE_WAIT_DURATION_SECONDS, QUEUE_WAIT_DURATION_BUCKETS), ]; /// Bucket bounds, in seconds, for the HTTP latency histogram. The layers above @@ -62,6 +63,13 @@ const DURATION_HISTOGRAM_BUCKETS: &[(&str, &[f64])] = &[ const HTTP_DURATION_BUCKETS: &[f64] = &[0.005, 0.025, 0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 45.0, 60.0]; +/// Bucket bounds, in seconds, for the queue-wait histogram. A request either +/// finds a free worker slot at once or waits behind proofs, so the buckets are +/// densest near zero. The last boundary is the default queue-wait timeout, so +/// the count of waits that ran to the timeout is exact. +const QUEUE_WAIT_DURATION_BUCKETS: &[f64] = + &[0.001, 0.01, 0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0]; + /// Metric name constants, so `metrics!` calls elsewhere point at one /// definition instead of repeating string literals. pub mod names { @@ -81,6 +89,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 @@ -92,6 +104,10 @@ pub mod outcomes { pub const FAILURE_RUNNER: &str = "failure_runner"; pub const FAILURE_OUTPUT_PARSE: &str = "failure_output_parse"; pub const FAILURE_PROVING: &str = "failure_proving"; + /// Rejected at admission because the queue (running + waiting) was full. + pub const REJECTED_QUEUE_FULL: &str = "rejected_queue_full"; + /// Rejected after waiting past `queue_wait_timeout` for a worker slot. + pub const REJECTED_WAIT_TIMEOUT: &str = "rejected_wait_timeout"; } /// Initializes the global Prometheus exporter and emits the `build_info` @@ -117,6 +133,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/rpc_impl.rs b/crates/starknet_transaction_prover/src/server/rpc_impl.rs index 71eb158b02f..07e7e2134fe 100644 --- a/crates/starknet_transaction_prover/src/server/rpc_impl.rs +++ b/crates/starknet_transaction_prover/src/server/rpc_impl.rs @@ -1,21 +1,30 @@ //! 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; use jsonrpsee::core::RpcResult; use jsonrpsee::types::ErrorObjectOwned; use starknet_api::rpc_transaction::RpcTransaction; -use tokio::sync::Semaphore; +use tokio::sync::{Semaphore, SemaphorePermit}; use tokio::time::timeout; 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, GaugeGuard}; 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 do not depend on that feature, so gate the +// module to the non-proving config and 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 continuously rejecting requests, so health + /// reporting can read it. + 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, ) } } @@ -78,29 +93,16 @@ impl ProvingRpcServer for ProvingRpcServerImpl { ) -> 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. - let _admission = self.admission_semaphore.try_acquire().map_err(|_| { - warn!( - max_concurrent_requests = self.max_concurrent_requests, - "Rejected proving request: queue is full" - ); - service_busy(self.max_concurrent_requests) - })?; - - // 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)); - } - }; + let _admission = self + .admission_semaphore + .try_acquire() + .map_err(|_| self.record_busy_reject(outcomes::REJECTED_QUEUE_FULL, "queue is full"))?; + + // Binding order matters. `_permit` is bound last, so it drops first and releases the + // worker slot before `_saturation_clear_guard` clears the window. The reverse order would + // let a rejection open a new window between the clear and the release, with nothing left + // to clear it. + let (_saturation_clear_guard, _permit) = self.acquire_worker_slot().await?; self.prover.prove_transaction(block_id, transaction).await.map_err(|err| { warn!("prove_transaction failed: {:?}", err); @@ -108,3 +110,63 @@ impl ProvingRpcServer for ProvingRpcServerImpl { }) } } + +impl ProvingRpcServerImpl { + /// Waits for a worker slot. Served in arrival order, or cancelled if the client disconnects. + /// Tracks the queue depth while the request waits, and records the wait time once a slot is + /// won. + /// + /// Returns the permit alongside a `SaturationClearGuard`. The caller must keep both alive for + /// the proving run. + async fn acquire_worker_slot( + &self, + ) -> Result<(SaturationClearGuard, SemaphorePermit<'_>), ErrorObjectOwned> { + let wait_start = Instant::now(); + let _waiting_guard = GaugeGuard::acquire(metric_names::QUEUE_WAITING_REQUESTS); + 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()); + // Clear on acquisition, not only on the guard's drop. Otherwise a long proving + // run keeps reporting saturation that has already ended. + self.saturation_monitor.mark_progress(); + Ok(( + SaturationClearGuard { saturation_monitor: self.saturation_monitor.clone() }, + permit, + )) + } + Ok(Err(_)) => Err(internal_server_error("proving service is shutting down")), + Err(_) => Err(self.record_busy_reject( + outcomes::REJECTED_WAIT_TIMEOUT, + "timed out waiting for a worker slot", + )), + } + } + + /// Records a busy reject: counts it under `outcome` so served and rejected requests share + /// one denominator, opens the saturation window, logs the reject, and returns the `-32005` + /// error. The four steps stay in one function because a reject that counts but never opens + /// the window would under-report sustained overload. + fn record_busy_reject(&self, outcome: &'static str, reason: &str) -> ErrorObjectOwned { + metrics::counter!(metric_names::PROVE_TRANSACTION_OUTCOME_TOTAL, "outcome" => outcome) + .increment(1); + self.saturation_monitor.mark_rejected(); + warn!( + max_concurrent_requests = self.max_concurrent_requests, + outcome, "Rejected proving request: {reason}" + ); + service_busy(self.max_concurrent_requests) + } +} + +/// Clears the saturation window on worker-slot release. The clear runs in `Drop`, so proving +/// success, proving error and client disconnect all reach it. +struct SaturationClearGuard { + saturation_monitor: SaturationMonitor, +} + +impl Drop for SaturationClearGuard { + fn drop(&mut self) { + self.saturation_monitor.mark_progress(); + } +} 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..949ef0f3f89 --- /dev/null +++ b/crates/starknet_transaction_prover/src/server/rpc_impl_test.rs @@ -0,0 +1,151 @@ +//! Tests for the admission/queue reject paths in [`ProvingRpcServerImpl`]. +//! +//! Sizing the semaphores so the reject fires before the prover runs covers both busy-reject +//! outcomes without a live node or a real proving run. Zero admission capacity forces a +//! queue-full reject. Zero worker slots with a tiny wait timeout force 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, SaturationClearGuard}; +use crate::server::saturation::SaturationMonitor; +use crate::server::test_recorder::{metric_value, outcome_total_line, 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", &[])) +} + +#[tokio::test] +async fn full_queue_rejects_with_service_busy_and_counts_queue_full() { + let handle = shared_handle(); + let line = outcome_total_line(outcomes::REJECTED_QUEUE_FULL); + let before = metric_value(&handle.render(), &line); + + // max_concurrent + max_queued = 0 gives an admission capacity of 0, so admission rejects + // every request. + let saturation_monitor = SaturationMonitor::default(); + let rpc_impl = ProvingRpcServerImpl::new( + dummy_prover(), + 0, + 0, + Duration::from_secs(30), + saturation_monitor.clone(), + ); + 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"); + assert!( + saturation_monitor.saturated_for_at_least(Duration::ZERO), + "a queue-full reject must open the saturation window" + ); +} + +#[tokio::test] +async fn wait_timeout_rejects_with_service_busy_and_counts_wait_timeout() { + let handle = shared_handle(); + let line = outcome_total_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. Admission lets the + // request in, it waits for a worker that never frees, and the timeout rejects it. + let saturation_monitor = SaturationMonitor::default(); + let rpc_impl = ProvingRpcServerImpl::new( + dummy_prover(), + 0, + 1, + Duration::from_millis(10), + saturation_monitor.clone(), + ); + 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", + ); + assert!( + saturation_monitor.saturated_for_at_least(Duration::ZERO), + "a wait-timeout reject must open the saturation window" + ); +} + +/// Saturation must clear when an in-flight job releases its worker slot, even if no further +/// request arrives. This test opens the window directly instead of going through a +/// `prove_transaction` reject, whose extra `rejected_*` outcome would race the exact-delta +/// assertions of the two reject tests above on the shared metrics recorder. +#[test] +fn saturation_clear_guard_drop_clears_saturation_without_new_traffic() { + let saturation_monitor = SaturationMonitor::default(); + let in_flight_release_guard = + SaturationClearGuard { saturation_monitor: saturation_monitor.clone() }; + + // Rejects open the window while the in-flight job holds the only worker slot. + saturation_monitor.mark_rejected(); + assert!(saturation_monitor.saturated_for_at_least(Duration::ZERO)); + + // The in-flight job finishes and its guard drops, with no new request arriving. + drop(in_flight_release_guard); + assert!( + !saturation_monitor.saturated_for_at_least(Duration::ZERO), + "releasing the worker slot must clear the saturation window" + ); +} + +/// A request that wins a worker slot must clear the saturation window at that moment, not only +/// when the slot is later released. This is the only test that fails if the `mark_progress` call +/// on acquisition goes away and clearing is left to `SaturationClearGuard`'s drop. The assertion +/// runs while both the guard and the permit are still held, so only the acquisition can have +/// cleared the window. +#[tokio::test] +async fn accepted_request_clears_saturation_window_while_in_flight() { + let saturation_monitor = SaturationMonitor::default(); + let rpc_impl = ProvingRpcServerImpl::new( + dummy_prover(), + 1, + 0, + Duration::from_secs(30), + saturation_monitor.clone(), + ); + + // Open the window, as a burst of busy-rejects would. + saturation_monitor.mark_rejected(); + assert!(saturation_monitor.saturated_for_at_least(Duration::ZERO)); + + let _worker_slot = + rpc_impl.acquire_worker_slot().await.expect("the only worker slot is free, so it is won"); + assert!( + !saturation_monitor.saturated_for_at_least(Duration::ZERO), + "a request holding a worker slot must have cleared the saturation window" + ); +} 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..280de0c2c48 --- /dev/null +++ b/crates/starknet_transaction_prover/src/server/saturation.rs @@ -0,0 +1,52 @@ +//! Saturation tracking for the prover's concurrency-limited request path. +//! +//! `ProvingRpcServerImpl` records rejects, worker-slot acquisitions and releases here. +//! `saturated_for_at_least` then answers how long the service has been continuously rejecting +//! requests, which separates sustained overload from an isolated reject. + +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. +#[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_progress` (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 forward progress. Either a request acquired a worker slot, or a slot + /// came free because proving finished, proving failed, or the client disconnected. + /// Either way the service is not stuck rejecting, so the saturation window clears. + /// + /// Slot release has to count. If only a fresh acquisition cleared the window, a + /// service whose traffic dries up after a burst of rejects would keep reporting + /// saturation with nothing left to reject. + pub fn mark_progress(&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..e086bbbaf35 --- /dev/null +++ b/crates/starknet_transaction_prover/src/server/saturation_test.rs @@ -0,0 +1,47 @@ +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(); + // The window has just opened, so the zero-elapsed comparison is still 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, because 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 extends the window instead of restarting it. Operators + // care about how long the service has been rejecting, which is the time since + // the first rejection. + monitor.mark_rejected(); + assert!(monitor.saturated_for_at_least(Duration::from_millis(25))); +} + +/// Backs both `mark_progress` call sites, worker-slot acquire and worker-slot release, which the +/// monitor cannot tell apart. +#[test] +fn mark_progress_clears_the_window() { + let monitor = SaturationMonitor::default(); + monitor.mark_rejected(); + sleep(Duration::from_millis(10)); + monitor.mark_progress(); + assert!(!monitor.saturated_for_at_least(Duration::from_millis(0))); +}