diff --git a/crates/sui-indexer-alt-graphql/src/api/subscription/events.rs b/crates/sui-indexer-alt-graphql/src/api/subscription/events.rs index a990a9ddc995..d35745e5ea41 100644 --- a/crates/sui-indexer-alt-graphql/src/api/subscription/events.rs +++ b/crates/sui-indexer-alt-graphql/src/api/subscription/events.rs @@ -32,6 +32,8 @@ use crate::task::streaming::StreamedCaches; use super::scan_then_live::Subscribable; impl Subscribable for Event { + const SUBSCRIPTION_TYPE: &'static str = "events"; + type Item = Self; type Cursor = CEvent; type Filter = EventFilter; diff --git a/crates/sui-indexer-alt-graphql/src/api/subscription/scan_then_live.rs b/crates/sui-indexer-alt-graphql/src/api/subscription/scan_then_live.rs index 8450bbbf7c94..ef4857a1ed25 100644 --- a/crates/sui-indexer-alt-graphql/src/api/subscription/scan_then_live.rs +++ b/crates/sui-indexer-alt-graphql/src/api/subscription/scan_then_live.rs @@ -42,6 +42,8 @@ use crate::task::streaming::CheckpointBroadcaster; use crate::task::streaming::ProcessedCheckpoint; use crate::task::streaming::StreamedCaches; use crate::task::streaming::SubscriptionBroadcast; +use crate::task::streaming::SubscriptionLifecycleGuard; +use crate::task::streaming::SubscriptionTerminationReason; use crate::task::streaming::broadcast_error; use crate::task::streaming::reconnect_error; use crate::task::streaming::wait_for_pipelines_catching_up_at; @@ -55,6 +57,9 @@ const BACKFILL_POLL_INTERVAL: Duration = Duration::from_millis(100); /// GraphQL type (transactions, events) supplies its cursor, filter, and matching, and the free /// functions here drive the shared machinery over it. pub(super) trait Subscribable { + /// The subscription kind, used to label this feed's aggregate subscriber metrics. + const SUBSCRIPTION_TYPE: &'static str; + /// The GraphQL node delivered in each edge. type Item: OutputType + Send + 'static; /// The opaque cursor minted for resumption. @@ -137,6 +142,7 @@ pub(super) fn subscribe( let handoff_threshold = config.broadcast_buffer as u64 / 2; stream! { + let guard = SubscriptionLifecycleGuard::new(S::SUBSCRIPTION_TYPE, broadcast.metrics()); let mut pending_receiver = None; let mut handoff: Option = None; let mut last_checkpoint: Option = None; @@ -159,6 +165,7 @@ pub(super) fn subscribe( let scanned = match scanned { Ok(scanned) => scanned, Err(e) => { + guard.terminate(SubscriptionTerminationReason::BackfillError); yield Err(e); return; } @@ -179,6 +186,7 @@ pub(super) fn subscribe( break; } yield Ok(edge); + guard.record_backfill_delivered(); } // A coverage marker (its `checkpoint` is the fully-scanned frontier): stop once // it has covered the handoff. @@ -194,7 +202,9 @@ pub(super) fn subscribe( // Phase 2: follow live from `handoff + 1` (a fresh receiver if there was no backfill). let receiver = pending_receiver.unwrap_or_else(|| broadcast.broadcaster().resubscribe()); - for await edge in live::(receiver, last_checkpoint, caches, resolver_limits, filter) { + for await edge in + live::(receiver, last_checkpoint, caches, resolver_limits, filter, guard) + { yield edge; } } @@ -271,6 +281,7 @@ fn live( caches: Arc, resolver_limits: sui_package_resolver::Limits, filter: S::Filter, + guard: SubscriptionLifecycleGuard, ) -> impl Stream, RpcError>> { stream! { let mut delivered_live = false; @@ -289,15 +300,24 @@ fn live( received = seq, "Unexpected gap between scan and live; disconnecting" ); + guard.terminate(SubscriptionTerminationReason::UnexpectedGap); yield Err(reconnect_error()); return; } } // Deliver each matching item as its own payload, ordered within the checkpoint. // Empty checkpoints yield nothing. - let edges = S::matching_edges(&checkpoint, &caches, &resolver_limits, &filter)?; + let edges = match S::matching_edges(&checkpoint, &caches, &resolver_limits, &filter) { + Ok(edges) => edges, + Err(e) => { + guard.terminate(SubscriptionTerminationReason::Error); + yield Err(e); + return; + } + }; for edge in edges { yield Ok(edge); + guard.record_delivered(checkpoint.summary.timestamp_ms); } last_checkpoint = Some(seq); delivered_live = true; @@ -305,10 +325,12 @@ fn live( // A lag before the first live checkpoint is catch-up overflow (likely kv-rpc lag). Err(broadcast::error::RecvError::Lagged(missed)) if !delivered_live => { warn!(missed, "Subscriber fell behind during catch-up; disconnecting"); + guard.terminate(SubscriptionTerminationReason::Lagged); yield Err(reconnect_error()); return; } Err(e) => { + guard.terminate(SubscriptionTerminationReason::from_recv_error(&e)); yield Err(broadcast_error(e)); return; } diff --git a/crates/sui-indexer-alt-graphql/src/api/subscription/transactions.rs b/crates/sui-indexer-alt-graphql/src/api/subscription/transactions.rs index d95c2fd3b6d6..d09ad89502b6 100644 --- a/crates/sui-indexer-alt-graphql/src/api/subscription/transactions.rs +++ b/crates/sui-indexer-alt-graphql/src/api/subscription/transactions.rs @@ -30,6 +30,8 @@ use crate::task::streaming::StreamedCaches; use super::scan_then_live::Subscribable; impl Subscribable for Transaction { + const SUBSCRIPTION_TYPE: &'static str = "transactions"; + type Item = Self; type Cursor = CTransaction; type Filter = TransactionFilter; diff --git a/crates/sui-indexer-alt-graphql/src/lib.rs b/crates/sui-indexer-alt-graphql/src/lib.rs index 85166dc2f72f..6cf421268fba 100644 --- a/crates/sui-indexer-alt-graphql/src/lib.rs +++ b/crates/sui-indexer-alt-graphql/src/lib.rs @@ -303,10 +303,14 @@ struct IdeEnabled(bool); #[derive(Clone, Copy)] struct SubscriptionsEnabled(bool); -/// Per-subscriber delivery rate in output nodes per second, surfaced to the subscription handler so -/// it can pace each payload by its cost. `0` disables pacing. -#[derive(Clone, Copy)] -struct SubscriptionThrottleRate(u32); +/// Per-subscriber delivery throttle settings surfaced to the subscription handler: the rate in +/// output nodes per second (`0` disables pacing) and the metric each payload's pacing delay is +/// observed into. +#[derive(Clone)] +struct SubscriptionThrottle { + nodes_per_second: u32, + delay_metric: prometheus::Histogram, +} /// Set-up and run the RPC service, using the provided arguments (expected to be extracted from the /// command-line). @@ -394,6 +398,8 @@ pub async fn start_rpc( metrics.clone(), ); + let subscription_metrics = Arc::new(SubscriptionMetrics::new(registry)); + let streaming_setup = match subscription_args.checkpoint_stream_url { Some(uri) => { let ledger_grpc = ledger_grpc_reader @@ -417,7 +423,7 @@ pub async fn start_rpc( readiness.clone(), ledger_grpc.clone(), watermark_task.watermarks_rx(), - Arc::new(SubscriptionMetrics::new(registry)), + subscription_metrics.clone(), ); let caches = Arc::new(StreamedCaches::new( streaming_packages, @@ -435,6 +441,7 @@ pub async fn start_rpc( None => None, }; + let throttle_delay_metric = subscription_metrics.subscriber_throttle_delay.clone(); let mut rpc = rpc .route(GRAPHQL_PATH, post(graphql).get(graphiql)) .route(GRAPHQL_SUBSCRIPTIONS_PATH, post(graphql_subscriptions)) @@ -473,11 +480,12 @@ pub async fn start_rpc( let subscriptions_enabled = streaming_setup.is_some(); rpc = rpc.layer(SubscriptionsEnabled(subscriptions_enabled)); - rpc = rpc.layer(SubscriptionThrottleRate( - config + rpc = rpc.layer(SubscriptionThrottle { + nodes_per_second: config .subscription .per_subscriber_max_output_nodes_per_second, - )); + delay_metric: throttle_delay_metric, + }); // The transaction subscription backfill waits on pipeline watermarks to gate delivery, so it // needs a live view of them. Captured before the watermark task is consumed by `run()`. @@ -513,6 +521,7 @@ pub async fn start_rpc( let subscription_broadcast = Arc::new(SubscriptionBroadcast::new( _broadcaster, first_live_checkpoint, + subscription_metrics.clone(), )); rpc = rpc .data(subscription_broadcast) @@ -584,7 +593,7 @@ async fn graphql_subscriptions( ConnectInfo(addr): ConnectInfo, Extension(schema): Extension>, Extension(SubscriptionsEnabled(subscriptions_enabled)): Extension, - Extension(SubscriptionThrottleRate(nodes_per_second)): Extension, + Extension(throttle_cfg): Extension, Extension(watermark): Extension, request: GraphQLRequest, ) -> axum::response::Response { @@ -600,7 +609,7 @@ async fn graphql_subscriptions( // Query depth is computed once by the query-limits extension during validation and stashed here, // so the throttle can add its depth surcharge to each payload's cost. let query_depth = QueryDepth::default(); - let throttle = Throttle::new(nodes_per_second); + let throttle = Throttle::new(throttle_cfg.nodes_per_second, throttle_cfg.delay_metric); let req = request .into_inner() .data(Session::new(addr)) diff --git a/crates/sui-indexer-alt-graphql/src/metrics/subscription.rs b/crates/sui-indexer-alt-graphql/src/metrics/subscription.rs index 490fc9953b09..65ab8c4ae361 100644 --- a/crates/sui-indexer-alt-graphql/src/metrics/subscription.rs +++ b/crates/sui-indexer-alt-graphql/src/metrics/subscription.rs @@ -1,12 +1,14 @@ // Copyright (c) Mysten Labs, Inc. // SPDX-License-Identifier: Apache-2.0 +use prometheus::Histogram; use prometheus::HistogramVec; use prometheus::IntCounter; use prometheus::IntCounterVec; use prometheus::IntGaugeVec; use prometheus::Registry; use prometheus::register_histogram_vec_with_registry; +use prometheus::register_histogram_with_registry; use prometheus::register_int_counter_vec_with_registry; use prometheus::register_int_counter_with_registry; use prometheus::register_int_gauge_vec_with_registry; @@ -17,6 +19,22 @@ const LAG_SEC_BUCKETS: &[f64] = &[ 0.95, 1.0, 2.0, 3.0, 4.0, 5.0, 10.0, 20.0, 50.0, 100.0, 1000.0, ]; +/// Histogram buckets for subscription lifetime, in seconds: sub-second churn through multi-hour +/// holds, with day/week edges to size the long-lived tail (subscriptions may run indefinitely). +const DURATION_SEC_BUCKETS: &[f64] = &[ + 1.0, 5.0, 15.0, 30.0, // seconds + 60.0, 300.0, 900.0, 1800.0, // 1m, 5m, 15m, 30m + 3600.0, 7200.0, 21600.0, // 1h, 2h, 6h + 43200.0, 86400.0, // 12h, 24h + 172800.0, 604800.0, // 2d, 7d +]; + +/// Histogram buckets for the per-payload throttle pause, in seconds: sub-millisecond when the rate +/// budget is generous, up to a few seconds for a heavy payload against a tight budget. +const THROTTLE_DELAY_SEC_BUCKETS: &[f64] = &[ + 0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, +]; + /// Metrics specific to the streaming subscription feature. Subscription payloads also record into /// the shared field-resolution (`fields_*`) and query-complexity (`input_nodes`, `output_nodes`) /// metrics on `RpcMetrics`. @@ -31,6 +49,15 @@ pub struct SubscriptionMetrics { pub upstream_latest_processed_checkpoint_timestamp_ms: IntGaugeVec, pub upstream_gap_recoveries: IntCounter, pub upstream_malformed_checkpoints: IntCounter, + + // Metrics aggregated across all subscribers. + pub active_subscriptions: IntGaugeVec, + pub subscriptions_opened: IntCounterVec, + pub subscription_terminations: IntCounterVec, + pub subscription_duration: HistogramVec, + pub payloads_delivered: IntCounterVec, + pub live_payload_delivery_checkpoint_timestamp_lag: HistogramVec, + pub subscriber_throttle_delay: Histogram, } impl SubscriptionMetrics { @@ -97,6 +124,57 @@ impl SubscriptionMetrics { registry, ) .unwrap(), + active_subscriptions: register_int_gauge_vec_with_registry!( + "graphql_subscription_active_subscriptions", + "Number of subscriptions currently active, by type", + &["type"], + registry, + ) + .unwrap(), + subscriptions_opened: register_int_counter_vec_with_registry!( + "graphql_subscription_opened", + "Total subscriptions opened, by type", + &["type"], + registry, + ) + .unwrap(), + subscription_terminations: register_int_counter_vec_with_registry!( + "graphql_subscription_terminations", + "Total subscriptions that have ended, by type and reason", + &["type", "reason"], + registry, + ) + .unwrap(), + subscription_duration: register_histogram_vec_with_registry!( + "graphql_subscription_duration_seconds", + "Lifetime of each subscription from open to close, in seconds, by type", + &["type"], + DURATION_SEC_BUCKETS.to_vec(), + registry, + ) + .unwrap(), + payloads_delivered: register_int_counter_vec_with_registry!( + "graphql_subscription_payloads_delivered", + "Total payloads delivered to subscribers, by type and phase (live or backfill)", + &["type", "phase"], + registry, + ) + .unwrap(), + live_payload_delivery_checkpoint_timestamp_lag: register_histogram_vec_with_registry!( + "graphql_subscription_live_payload_delivery_checkpoint_timestamp_lag", + "Seconds from a live payload's checkpoint timestamp to when it was delivered to a subscriber, by type (server-side; backfill deliveries excluded, since their lag is catch-up distance not staleness)", + &["type"], + LAG_SEC_BUCKETS.to_vec(), + registry, + ) + .unwrap(), + subscriber_throttle_delay: register_histogram_with_registry!( + "graphql_subscription_throttle_delay_seconds", + "Seconds the delivery throttle paused before the next payload, observed once per payload across all subscribers and phases (zero while the rate budget is not binding)", + THROTTLE_DELAY_SEC_BUCKETS.to_vec(), + registry, + ) + .unwrap(), } } @@ -142,6 +220,11 @@ impl SubscriptionMetrics { .with_label_values(&[phase]) .observe(lag_ms as f64 / 1000.0); } + + #[cfg(test)] + pub(crate) fn new_for_test() -> std::sync::Arc { + std::sync::Arc::new(Self::new(&Registry::new())) + } } #[cfg(test)] diff --git a/crates/sui-indexer-alt-graphql/src/task/streaming/checkpoint_stream_task.rs b/crates/sui-indexer-alt-graphql/src/task/streaming/checkpoint_stream_task.rs index 0162084840e9..f91d7745ff16 100644 --- a/crates/sui-indexer-alt-graphql/src/task/streaming/checkpoint_stream_task.rs +++ b/crates/sui-indexer-alt-graphql/src/task/streaming/checkpoint_stream_task.rs @@ -110,6 +110,10 @@ use super::checkpoint_resume::scan_checkpoints; #[cfg(any(feature = "staging", test))] use super::gap_recovery::CheckpointFetcher; use super::gap_recovery::recover_gap; +#[cfg(any(feature = "staging", test))] +use super::lifecycle::SubscriptionLifecycleGuard; +#[cfg(any(feature = "staging", test))] +use super::lifecycle::SubscriptionTerminationReason; use super::processed_checkpoint::ProcessedCheckpoint; use super::processed_checkpoint::ProcessedTransaction; @@ -188,17 +192,30 @@ pub(crate) struct SubscriptionBroadcast { /// channel. Distinct from any `resume_from` arg passed by subscribers (those refer to the /// kv-rpc resume fallback, not the live broadcast). first_live_checkpoint: u64, + /// Per-subscriber metrics, shared with every stream this broadcast spawns. + metrics: Arc, } #[cfg(any(feature = "staging", test))] impl SubscriptionBroadcast { - pub(crate) fn new(broadcaster: CheckpointBroadcaster, first_live_checkpoint: u64) -> Self { + pub(crate) fn new( + broadcaster: CheckpointBroadcaster, + first_live_checkpoint: u64, + metrics: Arc, + ) -> Self { Self { broadcaster, first_live_checkpoint, + metrics, } } + /// Per-subscriber metrics shared by every stream this broadcast spawns. + #[cfg(feature = "staging")] + pub(crate) fn metrics(&self) -> &SubscriptionMetrics { + &self.metrics + } + /// Direct access to the broadcast receiver template. Subscribers should call /// `.resubscribe()` to get their own receiver. #[cfg(feature = "staging")] @@ -233,6 +250,7 @@ impl SubscriptionBroadcast { let config = config.clone(); stream! { + let guard = SubscriptionLifecycleGuard::new("checkpoints", &self.metrics); let mut last_yielded: Option = resume_from; let mut receiver = self.broadcaster.resubscribe(); // `resubscribe` is future-only (delivers `handoff + 1`), so Phase 1 stops exactly at @@ -242,7 +260,14 @@ impl SubscriptionBroadcast { // Phase 1: scan toward the tip; within `handoff_threshold`, resubscribe + pin, stop at it. if let Some(start_after) = last_yielded { for await item in scan_checkpoints(fetcher, self.clone(), start_after, &config) { - let processed = item?; + let processed = match item { + Ok(processed) => processed, + Err(e) => { + guard.terminate(SubscriptionTerminationReason::BackfillError); + yield Err(e); + return; + } + }; let seq = processed.summary.sequence_number; last_yielded = Some(seq); yield Ok(processed); @@ -276,6 +301,7 @@ impl SubscriptionBroadcast { received = processed.summary.sequence_number, "Unexpected gap between scan and live; disconnecting" ); + guard.terminate(SubscriptionTerminationReason::UnexpectedGap); yield Err(reconnect_error()); return; } @@ -287,11 +313,13 @@ impl SubscriptionBroadcast { }, Err(broadcast::error::RecvError::Lagged(missed)) if !delivered_live => { warn!(missed, "Subscriber fell behind during catch-up (likely kv-rpc lag)"); + guard.terminate(SubscriptionTerminationReason::Lagged); yield Err(reconnect_error()); return; } // Slow subscriber (Lagged after going live) or closed channel: disconnect. Err(e) => { + guard.terminate(SubscriptionTerminationReason::from_recv_error(&e)); yield Err(broadcast_error(e)); return; } diff --git a/crates/sui-indexer-alt-graphql/src/task/streaming/lifecycle.rs b/crates/sui-indexer-alt-graphql/src/task/streaming/lifecycle.rs new file mode 100644 index 000000000000..2d2ac1fff2c1 --- /dev/null +++ b/crates/sui-indexer-alt-graphql/src/task/streaming/lifecycle.rs @@ -0,0 +1,217 @@ +// Copyright (c) Mysten Labs, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use prometheus::Histogram; +use prometheus::IntCounter; +use prometheus::IntCounterVec; +use prometheus::IntGauge; +use tokio::sync::broadcast; +use tokio::time::Instant; + +use crate::metrics::SubscriptionMetrics; + +/// Why a subscription ended, used to label `subscription_terminations`. +#[derive(Clone, Copy)] +pub(crate) enum SubscriptionTerminationReason { + ClientClosed, + Lagged, + Error, + Shutdown, + BackfillError, + /// A defensive disconnect for an invariant that should never fire; a nonzero count is a bug. + UnexpectedGap, +} + +impl SubscriptionTerminationReason { + fn metric_label(self) -> &'static str { + match self { + Self::ClientClosed => "client_closed", + Self::Lagged => "lagged", + Self::Error => "error", + Self::Shutdown => "shutdown", + Self::BackfillError => "backfill_error", + Self::UnexpectedGap => "unexpected_gap", + } + } + + pub(crate) fn from_recv_error(e: &broadcast::error::RecvError) -> Self { + match e { + broadcast::error::RecvError::Lagged(_) => Self::Lagged, + broadcast::error::RecvError::Closed => Self::Shutdown, + } + } +} + +/// Tracks one subscription for the lifetime of its stream: on construction bumps the +/// active-subscription gauge and the opened counter, and on drop decrements the gauge, records the +/// lifetime, and records a termination. Call `terminate` to record why it ended and consume the +/// guard; a guard dropped without that call (for example when the client disconnects and the stream +/// future is cancelled) records `client_closed`. +pub(crate) struct SubscriptionLifecycleGuard { + active_subscriptions: IntGauge, + subscription_terminations: IntCounterVec, + subscription_duration: Histogram, + live_delivered: IntCounter, + backfill_delivered: IntCounter, + live_delivery_lag: Histogram, + subscription_type: &'static str, + termination_reason: Option, + started_at: Instant, +} + +impl SubscriptionLifecycleGuard { + pub(crate) fn new(subscription_type: &'static str, metrics: &SubscriptionMetrics) -> Self { + let label = subscription_type; + let active_subscriptions = metrics.active_subscriptions.with_label_values(&[label]); + active_subscriptions.inc(); + metrics + .subscriptions_opened + .with_label_values(&[label]) + .inc(); + Self { + active_subscriptions, + subscription_terminations: metrics.subscription_terminations.clone(), + subscription_duration: metrics.subscription_duration.with_label_values(&[label]), + live_delivered: metrics + .payloads_delivered + .with_label_values(&[label, "live"]), + backfill_delivered: metrics + .payloads_delivered + .with_label_values(&[label, "backfill"]), + live_delivery_lag: metrics + .live_payload_delivery_checkpoint_timestamp_lag + .with_label_values(&[label]), + subscription_type, + termination_reason: None, + started_at: Instant::now(), + } + } + + /// Record why the subscription ended, consuming the guard so it cannot be used afterward. + /// + /// The guard's `Drop` at the end of this call is what records the termination, with the reason + /// set here. A guard dropped without calling this means the client closed the connection. + pub(crate) fn terminate(mut self, reason: SubscriptionTerminationReason) { + self.termination_reason = Some(reason); + } + + /// Record delivery of one live payload: bump the live counter and observe its freshness (now + /// minus its checkpoint's timestamp). One call per payload, so a checkpoint with N matching + /// transactions/events records N samples that climb as the throttle paces them out. + pub(crate) fn record_delivered(&self, checkpoint_timestamp_ms: u64) { + self.live_delivered.inc(); + let lag_ms = chrono::Utc::now().timestamp_millis() - checkpoint_timestamp_ms as i64; + self.live_delivery_lag + .observe(lag_ms.max(0) as f64 / 1000.0); + } + + /// Record delivery of one backfilled payload. Counted under the `backfill` phase and without a + /// freshness sample: backfill replays historical checkpoints, so their timestamp lag is catch-up + /// distance, not delivery staleness. + pub(crate) fn record_backfill_delivered(&self) { + self.backfill_delivered.inc(); + } +} + +impl Drop for SubscriptionLifecycleGuard { + fn drop(&mut self) { + self.active_subscriptions.dec(); + self.subscription_duration + .observe(self.started_at.elapsed().as_secs_f64()); + // No recorded reason means the stream was dropped without a terminal event: the client left. + let reason = self + .termination_reason + .unwrap_or(SubscriptionTerminationReason::ClientClosed); + self.subscription_terminations + .with_label_values(&[self.subscription_type, reason.metric_label()]) + .inc(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A guard dropped without a set reason bumps then clears the active gauge, counts one opened, + /// records a lifetime sample, and attributes the termination to `client_closed`. + #[tokio::test] + async fn drop_without_reason_is_client_closed() { + let metrics = SubscriptionMetrics::new_for_test(); + let label = "transactions"; + + { + let _guard = SubscriptionLifecycleGuard::new("transactions", &metrics); + assert_eq!(active(&metrics, label), 1); + assert_eq!(opened(&metrics, label), 1); + } + + assert_eq!(active(&metrics, label), 0); + assert_eq!(duration_samples(&metrics, label), 1); + assert_eq!(terminations(&metrics, label, "client_closed"), 1); + } + + /// A set reason overrides the default and labels the termination accordingly. + #[tokio::test] + async fn set_reason_is_recorded() { + let metrics = SubscriptionMetrics::new_for_test(); + let label = "events"; + + { + let guard = SubscriptionLifecycleGuard::new("events", &metrics); + guard.terminate(SubscriptionTerminationReason::Error); + } + + assert_eq!(active(&metrics, label), 0); + assert_eq!(terminations(&metrics, label, "error"), 1); + assert_eq!(terminations(&metrics, label, "client_closed"), 0); + } + + /// Live deliveries bump the live counter and one lag sample each; backfill deliveries bump only + /// the backfill counter (no lag sample). + #[tokio::test] + async fn record_delivered_counts_by_phase() { + let metrics = SubscriptionMetrics::new_for_test(); + let label = "transactions"; + let guard = SubscriptionLifecycleGuard::new("transactions", &metrics); + + guard.record_delivered(0); + guard.record_delivered(0); + guard.record_backfill_delivered(); + + assert_eq!(delivered(&metrics, label, "live"), 2); + assert_eq!(delivered(&metrics, label, "backfill"), 1); + assert_eq!(delivery_lag_samples(&metrics, label), 2); + } + + fn active(m: &SubscriptionMetrics, label: &str) -> i64 { + m.active_subscriptions.with_label_values(&[label]).get() + } + + fn opened(m: &SubscriptionMetrics, label: &str) -> u64 { + m.subscriptions_opened.with_label_values(&[label]).get() + } + + fn terminations(m: &SubscriptionMetrics, label: &str, reason: &str) -> u64 { + m.subscription_terminations + .with_label_values(&[label, reason]) + .get() + } + + fn duration_samples(m: &SubscriptionMetrics, label: &str) -> u64 { + m.subscription_duration + .with_label_values(&[label]) + .get_sample_count() + } + + fn delivered(m: &SubscriptionMetrics, label: &str, phase: &str) -> u64 { + m.payloads_delivered + .with_label_values(&[label, phase]) + .get() + } + + fn delivery_lag_samples(m: &SubscriptionMetrics, label: &str) -> u64 { + m.live_payload_delivery_checkpoint_timestamp_lag + .with_label_values(&[label]) + .get_sample_count() + } +} diff --git a/crates/sui-indexer-alt-graphql/src/task/streaming/mod.rs b/crates/sui-indexer-alt-graphql/src/task/streaming/mod.rs index 4d6dd8998b62..545d4d0c0ba6 100644 --- a/crates/sui-indexer-alt-graphql/src/task/streaming/mod.rs +++ b/crates/sui-indexer-alt-graphql/src/task/streaming/mod.rs @@ -45,6 +45,8 @@ mod checkpoint_resume; mod checkpoint_stream_task; mod gap_recovery; +#[cfg(any(feature = "staging", test))] +mod lifecycle; mod processed_checkpoint; mod streamed_cache_eviction; mod streamed_caches; @@ -71,6 +73,10 @@ pub(crate) use checkpoint_stream_task::broadcast_error; pub(crate) use checkpoint_stream_task::reconnect_error; #[cfg(feature = "staging")] pub(crate) use gap_recovery::wait_for_pipelines_catching_up_at; +#[cfg(feature = "staging")] +pub(crate) use lifecycle::SubscriptionLifecycleGuard; +#[cfg(feature = "staging")] +pub(crate) use lifecycle::SubscriptionTerminationReason; pub(crate) use processed_checkpoint::ProcessedCheckpoint; pub(crate) use processed_checkpoint::ProcessedTransaction; pub(crate) use streamed_cache_eviction::EvictableCache; diff --git a/crates/sui-indexer-alt-graphql/src/task/streaming/test_utils.rs b/crates/sui-indexer-alt-graphql/src/task/streaming/test_utils.rs index 43e093bd88fa..38ee1e118cc8 100644 --- a/crates/sui-indexer-alt-graphql/src/task/streaming/test_utils.rs +++ b/crates/sui-indexer-alt-graphql/src/task/streaming/test_utils.rs @@ -23,6 +23,7 @@ use tokio::sync::broadcast; use super::checkpoint_stream_task::SubscriptionBroadcast; use super::gap_recovery::CheckpointFetcher; use super::processed_checkpoint::ProcessedCheckpoint; +use crate::metrics::SubscriptionMetrics; /// Per-key behavior of the mock fetcher. #[derive(Debug, Clone)] @@ -113,7 +114,11 @@ pub(super) fn test_broadcast( let (tx, rx) = broadcast::channel(256); ( tx, - Arc::new(SubscriptionBroadcast::new(rx, first_live_checkpoint)), + Arc::new(SubscriptionBroadcast::new( + rx, + first_live_checkpoint, + SubscriptionMetrics::new_for_test(), + )), ) } diff --git a/crates/sui-indexer-alt-graphql/src/throttle.rs b/crates/sui-indexer-alt-graphql/src/throttle.rs index 607b0be7126a..73a4f5e13113 100644 --- a/crates/sui-indexer-alt-graphql/src/throttle.rs +++ b/crates/sui-indexer-alt-graphql/src/throttle.rs @@ -9,6 +9,7 @@ use async_graphql::Response; use async_graphql::Value; use futures::Stream; use futures::StreamExt; +use prometheus::Histogram; use crate::extensions::query_limits::QueryDepth; @@ -33,14 +34,20 @@ const DEPTH_NODE_COST: u32 = 2; /// /// So a 10-node payload at query depth 5 costs `10 + 5*2 = 20`, and at 40 nodes/second is held /// `20 / 40 = 0.5s`. A rate of `0` disables pacing, and payloads are never dropped or reordered. -#[derive(Clone, Copy)] +#[derive(Clone)] pub(crate) struct Throttle { nodes_per_second: u32, + /// Observes each payload's pacing delay in seconds (including zero when the budget is not + /// binding), for the `throttle_delay` metric. + delay_metric: Histogram, } impl Throttle { - pub(crate) fn new(nodes_per_second: u32) -> Self { - Self { nodes_per_second } + pub(crate) fn new(nodes_per_second: u32, delay_metric: Histogram) -> Self { + Self { + nodes_per_second, + delay_metric, + } } /// Pace `stream`, delivering each payload immediately then pausing before the next for its delay. @@ -55,6 +62,7 @@ impl Throttle { // backpressures its resolution. Depth is constant but only known once validation has // run, so read the slot here. let delay = self.calculate_delay(&response.data, query_depth.get()); + self.delay_metric.observe(delay.as_secs_f64()); yield response; if !delay.is_zero() { tokio::time::sleep(delay).await; @@ -102,6 +110,14 @@ mod tests { use super::*; + fn throttle(nodes_per_second: u32) -> Throttle { + Throttle::new(nodes_per_second, test_histogram()) + } + + fn test_histogram() -> Histogram { + Histogram::with_opts(prometheus::HistogramOpts::new("test", "test")).unwrap() + } + #[test] fn counts_scalars_and_objects() { // A single scalar. @@ -133,10 +149,7 @@ mod tests { #[test] fn zero_rate_disables_pacing() { let payload = value!({ "a": 1, "b": 2 }); - assert_eq!( - Throttle::new(0).calculate_delay(&payload, 100), - Duration::ZERO - ); + assert_eq!(throttle(0).calculate_delay(&payload, 100), Duration::ZERO); } #[test] @@ -145,20 +158,19 @@ mod tests { // No depth surcharge: cost 5 at 10 nodes/sec = 0.5s. assert_eq!( - Throttle::new(10).calculate_delay(&payload, 0), + throttle(10).calculate_delay(&payload, 0), Duration::from_millis(500) ); // Depth surcharge included: cost 5 + 5 * 2 = 15 at 15 nodes/sec = 1s. assert_eq!( - Throttle::new(15).calculate_delay(&payload, 5), + throttle(15).calculate_delay(&payload, 5), Duration::from_secs(1) ); // A higher rate paces the same payload proportionally faster. assert!( - Throttle::new(20).calculate_delay(&payload, 0) - < Throttle::new(10).calculate_delay(&payload, 0) + throttle(20).calculate_delay(&payload, 0) < throttle(10).calculate_delay(&payload, 0) ); } @@ -172,9 +184,8 @@ mod tests { Response::new(payload.clone()), Response::new(payload), ]; - let mut paced = Box::pin( - Throttle::new(10).wrap(futures::stream::iter(responses), QueryDepth::default()), - ); + let mut paced = + Box::pin(throttle(10).wrap(futures::stream::iter(responses), QueryDepth::default())); let start = tokio::time::Instant::now(); paced.next().await.unwrap(); @@ -193,8 +204,7 @@ mod tests { let query_depth = QueryDepth::new_for_test(3); let payload = value!({ "a": 1, "b": 2, "c": 3 }); // 4 output nodes let responses = vec![Response::new(payload.clone()), Response::new(payload)]; - let mut paced = - Box::pin(Throttle::new(20).wrap(futures::stream::iter(responses), query_depth)); + let mut paced = Box::pin(throttle(20).wrap(futures::stream::iter(responses), query_depth)); let start = tokio::time::Instant::now(); paced.next().await.unwrap(); @@ -208,13 +218,27 @@ mod tests { // A rate of 0 disables pacing, so both payloads arrive immediately with no gap. let payload = value!({ "a": 1, "b": 2 }); let responses = vec![Response::new(payload.clone()), Response::new(payload)]; - let mut paced = Box::pin( - Throttle::new(0).wrap(futures::stream::iter(responses), QueryDepth::default()), - ); + let mut paced = + Box::pin(throttle(0).wrap(futures::stream::iter(responses), QueryDepth::default())); let start = tokio::time::Instant::now(); paced.next().await.unwrap(); paced.next().await.unwrap(); assert_eq!(start.elapsed(), Duration::ZERO); } + + #[tokio::test(start_paused = true)] + async fn wrap_observes_one_delay_sample_per_payload() { + let metric = test_histogram(); + let payload = value!({ "a": 1, "b": 2 }); + let responses = vec![Response::new(payload.clone()), Response::new(payload)]; + let mut paced = Box::pin( + Throttle::new(10, metric.clone()) + .wrap(futures::stream::iter(responses), QueryDepth::default()), + ); + + while paced.next().await.is_some() {} + + assert_eq!(metric.get_sample_count(), 2); + } }