From 06acd48f73dd450ad7cdfe199c5005f17fc1fae2 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Sun, 7 Jun 2026 18:06:12 +0300 Subject: [PATCH] [ingest] add perf stats tracking for firehose connections --- Cargo.toml | 1 + docs/api/firehose.md | 38 +++++ src/control/firehose.rs | 9 ++ src/ingest/firehose.rs | 67 +++++++- src/ingest/firehose_stats.rs | 303 +++++++++++++++++++++++++++++++++++ src/ingest/mod.rs | 2 + src/state.rs | 6 + 7 files changed, 423 insertions(+), 3 deletions(-) create mode 100644 src/ingest/firehose_stats.rs diff --git a/Cargo.toml b/Cargo.toml index 9485ba2..820e3e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ indexer = [] indexer_stream = ["indexer"] jetstream = ["dep:zstd"] user-keyspace = [] +firehose-diagnostics = [] [dependencies] tokio = { version = "1.0", features = ["full"] } diff --git a/docs/api/firehose.md b/docs/api/firehose.md index 3a34b79..85d3d06 100644 --- a/docs/api/firehose.md +++ b/docs/api/firehose.md @@ -21,6 +21,42 @@ list all known firehose sources, including offline ones waiting for the retry lo "kind": "tcp_refused", "detail": "connection refused" }, + "stats": { + "connection_attempts": 2, + "successful_connections": 1, + "connect_errors": 1, + "stream_errors": 0, + "frames_read": 1200, + "bytes_read": 22000000, + "messages_decoded": 1200, + "messages_forwarded": 1198, + "messages_skipped": 2, + "forward_errors": 0, + "throttle_waits": 0, + "throttle_wait_micros": 0, + "should_process_micros": 8800, + "send_waits": 1198, + "send_wait_micros": 64000, + "connect_elapsed_micros": 410000, + "max_send_wait_micros": 9000, + "max_should_process_micros": 1200, + "max_throttle_wait_micros": 0, + "last_connect_attempt_at": 1717239900, + "last_connected_at": 1717239901, + "last_frame_at": 1717239958, + "last_decoded_at": 1717239958, + "last_forwarded_at": 1717239958, + "last_start_cursor": 123, + "last_seq": 1322, + "max_seq": 1322, + "message_kinds": { + "commit": 1197, + "sync": 0, + "identity": 0, + "account": 3, + "info": 0 + } + }, "host_status": "offline", "pds": { "host": "127.0.0.1", @@ -35,6 +71,8 @@ list all known firehose sources, including offline ones waiting for the retry lo `last_failure` is present while hydrant has recorded failure/backoff state for a source. `kind` is a compact category such as `dns`, `tcp_refused`, `tcp_timeout`, `tls`, `http_upgrade`, `websocket`, `decode`, `relay_error`, or `config`; `detail` contains the underlying error text. +`stats` is present only in builds compiled with the `firehose-diagnostics` feature. these counters are in-memory process diagnostics intended for polling and diffing. compare `last_seq`/`max_seq`, `frames_read`, `messages_forwarded`, `send_wait_micros`, `max_send_wait_micros`, and `stream_errors` across samples to distinguish source lag, reconnect churn, filtering, and worker-channel backpressure. + ### query parameters all filters are exact-match and optional. multiple filters are combined with logical `AND`. diff --git a/src/control/firehose.rs b/src/control/firehose.rs index 255c131..b72eeea 100644 --- a/src/control/firehose.rs +++ b/src/control/firehose.rs @@ -10,6 +10,8 @@ use url::Url; use crate::config::FirehoseSource; use crate::db::{self, keys}; +#[cfg(feature = "firehose-diagnostics")] +use crate::ingest::firehose_stats::FirehoseStatsSnapshot; use crate::ingest::{BufferTx, firehose::FirehoseIngestor}; use crate::state::AppState; @@ -63,6 +65,9 @@ pub struct FirehoseSourceInfo { pub retry_in_secs: Option, #[serde(skip_serializing_if = "Option::is_none")] pub last_failure: Option, + #[cfg(feature = "firehose-diagnostics")] + #[serde(skip_serializing_if = "Option::is_none")] + pub stats: Option, #[serde(skip_serializing_if = "Option::is_none")] pub host_status: Option<&'static str>, #[serde(skip_serializing_if = "Option::is_none")] @@ -213,6 +218,8 @@ impl FirehoseHandle { let throttle = self.state.throttler.snapshot(url); let pds = is_pds.then(|| self.pds_info(url, &meta)).flatten(); let host_status = pds.as_ref().map(|pds| pds.status); + #[cfg(feature = "firehose-diagnostics")] + let stats = self.state.firehose_stats.snapshot(url); let last_failure = throttle .last_failure @@ -233,6 +240,8 @@ impl FirehoseHandle { .then_some(throttle.throttled_until), retry_in_secs: throttle.retry_in_secs(now), last_failure, + #[cfg(feature = "firehose-diagnostics")] + stats, host_status, pds, }); diff --git a/src/ingest/firehose.rs b/src/ingest/firehose.rs index 423ea0f..45e967b 100644 --- a/src/ingest/firehose.rs +++ b/src/ingest/firehose.rs @@ -119,6 +119,17 @@ fn classify_websocket_error(err: &tokio_websockets::Error) -> FirehoseFailure { } } +#[cfg(feature = "firehose-diagnostics")] +fn message_stats(msg: &SubscribeReposMessage<'_>) -> (&'static str, Option) { + match msg { + SubscribeReposMessage::Commit(commit) => ("commit", Some(commit.seq)), + SubscribeReposMessage::Sync(sync) => ("sync", Some(sync.seq)), + SubscribeReposMessage::Identity(identity) => ("identity", Some(identity.seq)), + SubscribeReposMessage::Account(account) => ("account", Some(account.seq)), + SubscribeReposMessage::Info(_) => ("info", None), + } +} + trait AddJitter: rand::Rng { fn add_jitter(&mut self, timeout: Duration) -> Duration { let timeout_secs = timeout.as_secs_f32(); @@ -138,6 +149,8 @@ pub struct FirehoseIngestor { _verify_signatures: bool, throttle: ThrottleHandle, max_failures: usize, + #[cfg(feature = "firehose-diagnostics")] + stats: Arc, } impl FirehoseIngestor { @@ -152,6 +165,8 @@ impl FirehoseIngestor { max_failures: usize, ) -> Self { let throttle = state.throttler.get_handle(&relay_host).await; + #[cfg(feature = "firehose-diagnostics")] + let stats = state.firehose_stats.handle(&relay_host).await; Self { state, buffer_tx, @@ -162,6 +177,8 @@ impl FirehoseIngestor { _verify_signatures: verify_signatures, throttle, max_failures, + #[cfg(feature = "firehose-diagnostics")] + stats, } } @@ -191,6 +208,8 @@ impl FirehoseIngestor { Some(c) => info!(cursor = %c, "resuming from cursor"), None => info!("no cursor found, live tailing"), } + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_connect_attempt(start_cursor); let host_status = self.is_pds.then(|| { let meta = self.state.pds_meta.load(); @@ -210,6 +229,8 @@ impl FirehoseIngestor { Ok(s) => s, Err(e) => { let failure = classify_firehose_error(&e); + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_connect_error(failure.kind); let secs = match self.on_failure(&failure).await { Some(secs) => secs, None => { @@ -244,6 +265,8 @@ impl FirehoseIngestor { elapsed_ms = connect_started.elapsed().as_millis(), "firehose connected" ); + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_connected(connect_started.elapsed()); let mut marked_active = false; let active_sleep_secs = if cfg!(debug_assertions) { 1 } else { 60 }; let mut active_sleep = @@ -256,8 +279,15 @@ impl FirehoseIngestor { Ok(b) => b, Err(e) => break Err(e), }; + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_frame(bytes.len()); match decode_frame(&bytes) { Ok(msg) => { + #[cfg(feature = "firehose-diagnostics")] + { + let (kind, seq) = message_stats(&msg); + self.stats.record_decoded(kind, seq); + } if self.is_pds { let tier = { let meta = self.state.pds_meta.load(); @@ -269,8 +299,13 @@ impl FirehoseIngestor { self.state.tier_policy.resolve(host, override_name) }; let accounts = self.state.db.get_count(&count_key).await; + #[cfg(feature = "firehose-diagnostics")] + let throttle_started = Instant::now(); tokio::select! { - _ = self.throttle.wait_for_allow(accounts, &tier) => {} + _ = self.throttle.wait_for_allow(accounts, &tier) => { + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_throttle_wait(throttle_started.elapsed()); + } _ = self.enabled.changed() => { if !*self.enabled.borrow() { info!("firehose disabled, disconnecting"); @@ -342,10 +377,14 @@ impl FirehoseIngestor { match res { Ok(()) => {} Err(FirehoseError::StreamClosed { code: 1001, reason }) => { + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_stream_error("stream_closed"); debug!(reason = %reason, "host gone away"); tokio::time::sleep(Duration::from_secs(1)).await; } Err(FirehoseError::FutureCursor) => { + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_stream_error("future_cursor"); if self.is_pds && let Err(e) = self.set_host_status(HostStatus::Idle) { @@ -363,6 +402,8 @@ impl FirehoseIngestor { .map_or(Cow::Borrowed(""), Cow::Borrowed); let failure = FirehoseFailure::new("relay_error", format!("{error}: {message}")); + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_stream_error(failure.kind); error!(err = %error, "relay sent error: {message}"); let secs = match self.on_failure(&failure).await { Some(secs) => secs, @@ -391,6 +432,8 @@ impl FirehoseIngestor { } Err(e) => { let failure = classify_firehose_error(&e); + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_stream_error(failure.kind); let secs = match self.on_failure(&failure).await { Some(secs) => secs, None => { @@ -487,26 +530,44 @@ impl FirehoseIngestor { _ => return, }; + #[cfg(feature = "firehose-diagnostics")] + let should_process_started = Instant::now(); let process = self .should_process(did) .await .inspect_err(|e| error!(did = %did, err = %e, "failed to check if we should process")) .unwrap_or(false); + #[cfg(feature = "firehose-diagnostics")] + self.stats + .record_should_process(should_process_started.elapsed()); if !process { + #[cfg(feature = "firehose-diagnostics")] + self.stats.record_skipped(); trace!(did = %did, "skipping: not in filter"); return; } trace!(did = %did, "forwarding message to ingest buffer"); - if let Err(e) = self + #[cfg(feature = "firehose-diagnostics")] + let send_started = Instant::now(); + let res = self .buffer_tx .send(IngestMessage::Firehose { url: self.relay_host.clone(), is_pds: self.is_pds, msg: msg.into_static(), }) - .await + .await; + #[cfg(feature = "firehose-diagnostics")] { + let elapsed = send_started.elapsed(); + if res.is_ok() { + self.stats.record_forwarded(elapsed); + } else { + self.stats.record_forward_error(elapsed); + } + } + if let Err(e) = res { error!(err = %e, "failed to send message to buffer processor"); } } diff --git a/src/ingest/firehose_stats.rs b/src/ingest/firehose_stats.rs new file mode 100644 index 0000000..0a224d5 --- /dev/null +++ b/src/ingest/firehose_stats.rs @@ -0,0 +1,303 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicI64, AtomicU64, Ordering}; +use std::time::Duration; + +use parking_lot::Mutex; +use serde::Serialize; +use url::Url; + +#[derive(Default)] +pub struct FirehoseStats { + sources: scc::HashMap>, +} + +impl FirehoseStats { + pub async fn handle(&self, url: &Url) -> Arc { + self.sources + .entry_async(url.clone()) + .await + .or_insert_with(|| Arc::new(FirehoseSourceStats::default())) + .get() + .clone() + } + + pub fn snapshot(&self, url: &Url) -> Option { + self.sources.read_sync(url, |_, stats| stats.snapshot()) + } +} + +#[derive(Default)] +pub struct FirehoseSourceStats { + connection_attempts: AtomicU64, + successful_connections: AtomicU64, + connect_errors: AtomicU64, + stream_errors: AtomicU64, + frames_read: AtomicU64, + bytes_read: AtomicU64, + messages_decoded: AtomicU64, + messages_forwarded: AtomicU64, + messages_skipped: AtomicU64, + commit_messages: AtomicU64, + sync_messages: AtomicU64, + identity_messages: AtomicU64, + account_messages: AtomicU64, + info_messages: AtomicU64, + forward_errors: AtomicU64, + throttle_waits: AtomicU64, + throttle_wait_micros: AtomicU64, + should_process_micros: AtomicU64, + send_waits: AtomicU64, + send_wait_micros: AtomicU64, + connect_elapsed_micros: AtomicU64, + max_send_wait_micros: AtomicU64, + max_should_process_micros: AtomicU64, + max_throttle_wait_micros: AtomicU64, + last_connect_attempt_at: AtomicI64, + last_connected_at: AtomicI64, + last_frame_at: AtomicI64, + last_decoded_at: AtomicI64, + last_forwarded_at: AtomicI64, + last_error_at: AtomicI64, + last_start_cursor: AtomicI64, + last_seq: AtomicI64, + max_seq: AtomicI64, + last_error_kind: Mutex>, +} + +impl FirehoseSourceStats { + pub fn record_connect_attempt(&self, cursor: Option) { + self.connection_attempts.fetch_add(1, Ordering::Relaxed); + self.last_connect_attempt_at + .store(now_ts(), Ordering::Relaxed); + self.last_start_cursor + .store(cursor.unwrap_or(0), Ordering::Relaxed); + } + + pub fn record_connected(&self, elapsed: Duration) { + self.successful_connections.fetch_add(1, Ordering::Relaxed); + self.last_connected_at.store(now_ts(), Ordering::Relaxed); + self.connect_elapsed_micros + .fetch_add(duration_micros(elapsed), Ordering::Relaxed); + } + + pub fn record_connect_error(&self, kind: &'static str) { + self.connect_errors.fetch_add(1, Ordering::Relaxed); + self.record_error_kind(kind); + } + + pub fn record_stream_error(&self, kind: &'static str) { + self.stream_errors.fetch_add(1, Ordering::Relaxed); + self.record_error_kind(kind); + } + + pub fn record_frame(&self, len: usize) { + self.frames_read.fetch_add(1, Ordering::Relaxed); + self.bytes_read + .fetch_add(len.try_into().unwrap_or(u64::MAX), Ordering::Relaxed); + self.last_frame_at.store(now_ts(), Ordering::Relaxed); + } + + pub fn record_decoded(&self, kind: &'static str, seq: Option) { + self.messages_decoded.fetch_add(1, Ordering::Relaxed); + self.last_decoded_at.store(now_ts(), Ordering::Relaxed); + match kind { + "commit" => self.commit_messages.fetch_add(1, Ordering::Relaxed), + "sync" => self.sync_messages.fetch_add(1, Ordering::Relaxed), + "identity" => self.identity_messages.fetch_add(1, Ordering::Relaxed), + "account" => self.account_messages.fetch_add(1, Ordering::Relaxed), + "info" => self.info_messages.fetch_add(1, Ordering::Relaxed), + _ => 0, + }; + if let Some(seq) = seq { + self.last_seq.store(seq, Ordering::Relaxed); + self.max_seq.fetch_max(seq, Ordering::Relaxed); + } + } + + pub fn record_throttle_wait(&self, elapsed: Duration) { + let micros = duration_micros(elapsed); + if micros == 0 { + return; + } + self.throttle_waits.fetch_add(1, Ordering::Relaxed); + self.throttle_wait_micros + .fetch_add(micros, Ordering::Relaxed); + self.max_throttle_wait_micros + .fetch_max(micros, Ordering::Relaxed); + } + + pub fn record_should_process(&self, elapsed: Duration) { + let micros = duration_micros(elapsed); + self.should_process_micros + .fetch_add(micros, Ordering::Relaxed); + self.max_should_process_micros + .fetch_max(micros, Ordering::Relaxed); + } + + pub fn record_skipped(&self) { + self.messages_skipped.fetch_add(1, Ordering::Relaxed); + } + + pub fn record_forwarded(&self, elapsed: Duration) { + self.messages_forwarded.fetch_add(1, Ordering::Relaxed); + self.last_forwarded_at.store(now_ts(), Ordering::Relaxed); + self.record_send_wait(elapsed); + } + + pub fn record_forward_error(&self, elapsed: Duration) { + self.forward_errors.fetch_add(1, Ordering::Relaxed); + self.record_send_wait(elapsed); + } + + fn record_send_wait(&self, elapsed: Duration) { + let micros = duration_micros(elapsed); + self.send_waits.fetch_add(1, Ordering::Relaxed); + self.send_wait_micros.fetch_add(micros, Ordering::Relaxed); + self.max_send_wait_micros + .fetch_max(micros, Ordering::Relaxed); + } + + fn record_error_kind(&self, kind: &'static str) { + self.last_error_at.store(now_ts(), Ordering::Relaxed); + *self.last_error_kind.lock() = Some(kind); + } + + fn snapshot(&self) -> FirehoseStatsSnapshot { + FirehoseStatsSnapshot { + connection_attempts: self.load_u64(&self.connection_attempts), + successful_connections: self.load_u64(&self.successful_connections), + connect_errors: self.load_u64(&self.connect_errors), + stream_errors: self.load_u64(&self.stream_errors), + frames_read: self.load_u64(&self.frames_read), + bytes_read: self.load_u64(&self.bytes_read), + messages_decoded: self.load_u64(&self.messages_decoded), + messages_forwarded: self.load_u64(&self.messages_forwarded), + messages_skipped: self.load_u64(&self.messages_skipped), + message_kinds: FirehoseMessageStats { + commit: self.load_u64(&self.commit_messages), + sync: self.load_u64(&self.sync_messages), + identity: self.load_u64(&self.identity_messages), + account: self.load_u64(&self.account_messages), + info: self.load_u64(&self.info_messages), + }, + forward_errors: self.load_u64(&self.forward_errors), + throttle_waits: self.load_u64(&self.throttle_waits), + throttle_wait_micros: self.load_u64(&self.throttle_wait_micros), + should_process_micros: self.load_u64(&self.should_process_micros), + send_waits: self.load_u64(&self.send_waits), + send_wait_micros: self.load_u64(&self.send_wait_micros), + connect_elapsed_micros: self.load_u64(&self.connect_elapsed_micros), + max_send_wait_micros: self.load_u64(&self.max_send_wait_micros), + max_should_process_micros: self.load_u64(&self.max_should_process_micros), + max_throttle_wait_micros: self.load_u64(&self.max_throttle_wait_micros), + last_connect_attempt_at: nonzero_i64(&self.last_connect_attempt_at), + last_connected_at: nonzero_i64(&self.last_connected_at), + last_frame_at: nonzero_i64(&self.last_frame_at), + last_decoded_at: nonzero_i64(&self.last_decoded_at), + last_forwarded_at: nonzero_i64(&self.last_forwarded_at), + last_error_at: nonzero_i64(&self.last_error_at), + last_start_cursor: nonzero_i64(&self.last_start_cursor), + last_seq: nonzero_i64(&self.last_seq), + max_seq: nonzero_i64(&self.max_seq), + last_error_kind: *self.last_error_kind.lock(), + } + } + + fn load_u64(&self, atomic: &AtomicU64) -> u64 { + atomic.load(Ordering::Relaxed) + } +} + +#[derive(Debug, Clone, Serialize)] +pub struct FirehoseStatsSnapshot { + pub connection_attempts: u64, + pub successful_connections: u64, + pub connect_errors: u64, + pub stream_errors: u64, + pub frames_read: u64, + pub bytes_read: u64, + pub messages_decoded: u64, + pub messages_forwarded: u64, + pub messages_skipped: u64, + pub message_kinds: FirehoseMessageStats, + pub forward_errors: u64, + pub throttle_waits: u64, + pub throttle_wait_micros: u64, + pub should_process_micros: u64, + pub send_waits: u64, + pub send_wait_micros: u64, + pub connect_elapsed_micros: u64, + pub max_send_wait_micros: u64, + pub max_should_process_micros: u64, + pub max_throttle_wait_micros: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_connect_attempt_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_connected_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_frame_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_decoded_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_forwarded_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_error_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_start_cursor: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_seq: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_seq: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_error_kind: Option<&'static str>, +} + +#[derive(Debug, Clone, Serialize)] +pub struct FirehoseMessageStats { + pub commit: u64, + pub sync: u64, + pub identity: u64, + pub account: u64, + pub info: u64, +} + +fn now_ts() -> i64 { + chrono::Utc::now().timestamp() +} + +fn duration_micros(duration: Duration) -> u64 { + duration.as_micros().try_into().unwrap_or(u64::MAX) +} + +fn nonzero_i64(atomic: &AtomicI64) -> Option { + let value = atomic.load(Ordering::Relaxed); + (value != 0).then_some(value) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn snapshot_records_firehose_progress() { + let stats = FirehoseSourceStats::default(); + + stats.record_connect_attempt(Some(100)); + stats.record_connected(Duration::from_millis(12)); + stats.record_frame(42); + stats.record_decoded("commit", Some(101)); + stats.record_forwarded(Duration::from_micros(7)); + + let snapshot = stats.snapshot(); + assert_eq!(snapshot.connection_attempts, 1); + assert_eq!(snapshot.successful_connections, 1); + assert_eq!(snapshot.frames_read, 1); + assert_eq!(snapshot.bytes_read, 42); + assert_eq!(snapshot.messages_decoded, 1); + assert_eq!(snapshot.messages_forwarded, 1); + assert_eq!(snapshot.last_start_cursor, Some(100)); + assert_eq!(snapshot.last_seq, Some(101)); + assert_eq!(snapshot.max_seq, Some(101)); + assert_eq!(snapshot.message_kinds.commit, 1); + } +} diff --git a/src/ingest/mod.rs b/src/ingest/mod.rs index f38b323..1d9ec7a 100644 --- a/src/ingest/mod.rs +++ b/src/ingest/mod.rs @@ -1,4 +1,6 @@ pub mod firehose; +#[cfg(feature = "firehose-diagnostics")] +pub mod firehose_stats; #[cfg(feature = "indexer")] pub mod indexer; mod mailbox; diff --git a/src/state.rs b/src/state.rs index 165b558..838da92 100644 --- a/src/state.rs +++ b/src/state.rs @@ -10,6 +10,8 @@ use tokio::sync::Notify; use tokio::sync::watch; use url::Url; +#[cfg(feature = "firehose-diagnostics")] +use crate::ingest::firehose_stats::FirehoseStats; #[cfg(feature = "relay")] use crate::pds_daily_limit::PdsDailyLimit; use crate::{ @@ -31,6 +33,8 @@ pub struct AppState { pub(crate) pds_daily_limit: PdsDailyLimit, pub(crate) tier_policy: TierPolicy, pub firehose_cursors: scc::HashIndex, + #[cfg(feature = "firehose-diagnostics")] + pub(crate) firehose_stats: FirehoseStats, pub firehose_enabled: watch::Sender, #[cfg(feature = "indexer")] pub backfill_notify: Notify, @@ -108,6 +112,8 @@ impl AppState { pds_meta, tier_policy: config.tier_policy.clone(), firehose_cursors: relay_cursors, + #[cfg(feature = "firehose-diagnostics")] + firehose_stats: FirehoseStats::default(), #[cfg(feature = "indexer")] backfill_notify: Notify::new(), #[cfg(feature = "indexer")] -- 2.51.2