starknet_transaction_prover: track saturation and queue depth on the request path

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.

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)?;
 

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;

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<Promethe
     // Pre-register at zero so the series exists in scrapes before the first panic.
     metrics::counter!(names::PANICS_TOTAL).increment(0);
     super::http_metrics::preregister_http_metrics();
+    // Queue depth starts at zero. Busy-rejects are folded into the outcome counter, so
+    // pre-register both reject outcomes too. A rejection-rate query then has series from startup.
+    metrics::gauge!(names::QUEUE_WAITING_REQUESTS).set(0.0);
+    metrics::counter!(
+        names::PROVE_TRANSACTION_OUTCOME_TOTAL,
+        "outcome" => outcomes::REJECTED_QUEUE_FULL,
+    )
+    .increment(0);
+    metrics::counter!(
+        names::PROVE_TRANSACTION_OUTCOME_TOTAL,
+        "outcome" => outcomes::REJECTED_WAIT_TIMEOUT,
+    )
+    .increment(0);
     Ok(handle)
 }
 

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,33 +93,80 @@ impl ProvingRpcServer for ProvingRpcServerImpl {
     ) -> RpcResult<ProveTransactionResult> {
         // 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);
             ErrorObjectOwned::from(err)
         })
     }
 }
+
+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();
+    }
+}