diff --git a/.env.example b/.env.example index d20bf0ed..5b6c98ad 100644 --- a/.env.example +++ b/.env.example @@ -218,6 +218,15 @@ OUTBOX_POLL_INTERVAL_MS=100 OUTBOX_BATCH_SIZE=500 OUTBOX_BACKLOG_ALERT_THRESHOLD=10000 +# Ledger-range reconciliation against the RPC source (issue #511): every +# RECONCILE_INTERVAL_MS, compare indexed event counts for the most recent +# RECONCILE_LEDGER_SPAN settled ledgers (RECONCILE_TIP_MARGIN behind the +# tip) against getEvents, and report discrepant ledger ranges. +RECONCILE_ENABLED=true +RECONCILE_INTERVAL_MS=600000 +RECONCILE_LEDGER_SPAN=400 +RECONCILE_TIP_MARGIN=100 + # --- Lag alerting --- # OPTIONAL indexer default: empty (disabled) diff --git a/crates/indexer/src/config.rs b/crates/indexer/src/config.rs index 3eeeaaee..38427390 100644 --- a/crates/indexer/src/config.rs +++ b/crates/indexer/src/config.rs @@ -44,6 +44,15 @@ pub struct Config { pub outbox_batch_size: i64, /// Backlog size at which the relay warns that delivery is falling behind. pub outbox_backlog_alert_threshold: i64, + /// Whether the ledger-range reconciliation loop runs (issue #511). + pub reconcile_enabled: bool, + /// Time between reconciliation passes. + pub reconcile_interval: std::time::Duration, + /// How many settled ledgers each pass compares against the RPC. + pub reconcile_ledger_span: u64, + /// How far behind the chain tip the window sits, so in-flight ledgers + /// the streamer has not committed yet never read as discrepancies. + pub reconcile_tip_margin: u64, pub index_diagnostic: bool, /// Topic patterns pushed into the `getEvents` RPC filter alongside the /// contract allowlist (issue #203). Empty means "no topic narrowing". @@ -176,6 +185,10 @@ impl Config { let outbox_batch_size = parse_bounded_u64("OUTBOX_BATCH_SIZE", 500, 1, 10_000); let outbox_backlog_alert_threshold = parse_bounded_u64("OUTBOX_BACKLOG_ALERT_THRESHOLD", 10_000, 1, 10_000_000); + let reconcile_interval_ms = + parse_bounded_u64("RECONCILE_INTERVAL_MS", 600_000, 10_000, 86_400_000); + let reconcile_ledger_span = parse_bounded_u64("RECONCILE_LEDGER_SPAN", 400, 10, 100_000); + let reconcile_tip_margin = parse_bounded_u64("RECONCILE_TIP_MARGIN", 100, 0, 10_000); let alert_lag_threshold = parse_bounded_u64("ALERT_LAG_THRESHOLD", 200, 1, 1_000_000); let alert_cooldown_minutes = parse_bounded_u64("ALERT_COOLDOWN_MINUTES", 30, 1, 10_080); let statement_timeout_ms = @@ -234,6 +247,9 @@ impl Config { "OUTBOX_BACKLOG_ALERT_THRESHOLD", outbox_backlog_alert_threshold.as_ref(), ), + ("RECONCILE_INTERVAL_MS", reconcile_interval_ms.as_ref()), + ("RECONCILE_LEDGER_SPAN", reconcile_ledger_span.as_ref()), + ("RECONCILE_TIP_MARGIN", reconcile_tip_margin.as_ref()), ("ALERT_LAG_THRESHOLD", alert_lag_threshold.as_ref()), ("ALERT_COOLDOWN_MINUTES", alert_cooldown_minutes.as_ref()), ("DB_STATEMENT_TIMEOUT_MS", statement_timeout_ms.as_ref()), @@ -279,6 +295,12 @@ impl Config { .map(|v| v.eq_ignore_ascii_case("true")) .unwrap_or(false); + // Enabled unless explicitly turned off: the reconciler is the proof + // that indexed data matches the chain, so it defaults on (issue #511). + let reconcile_enabled = std::env::var("RECONCILE_ENABLED") + .map(|v| !v.eq_ignore_ascii_case("false")) + .unwrap_or(true); + let topic_filters = match std::env::var("INDEX_TOPIC_FILTERS") { Ok(spec) => match crate::rpc::filters::parse_topic_filters(&spec) { Ok(f) => f, @@ -323,6 +345,9 @@ impl Config { let outbox_poll_interval_ms = outbox_poll_interval_ms.unwrap(); let outbox_batch_size = outbox_batch_size.unwrap() as i64; let outbox_backlog_alert_threshold = outbox_backlog_alert_threshold.unwrap() as i64; + let reconcile_interval = std::time::Duration::from_millis(reconcile_interval_ms.unwrap()); + let reconcile_ledger_span = reconcile_ledger_span.unwrap(); + let reconcile_tip_margin = reconcile_tip_margin.unwrap(); let alert_lag_threshold = alert_lag_threshold.unwrap(); let alert_cooldown_minutes = alert_cooldown_minutes.unwrap(); let statement_timeout_ms = statement_timeout_ms.unwrap(); @@ -352,6 +377,10 @@ impl Config { outbox_poll_interval: Duration::from_millis(outbox_poll_interval_ms), outbox_batch_size, outbox_backlog_alert_threshold, + reconcile_enabled, + reconcile_interval, + reconcile_ledger_span, + reconcile_tip_margin, index_diagnostic, topic_filters, max_events_per_poll: max_events_per_poll as u32, @@ -926,6 +955,13 @@ mod tests { assert_eq!(cfg.outbox_poll_interval.as_millis(), 100); assert_eq!(cfg.outbox_batch_size, 500); assert_eq!(cfg.outbox_backlog_alert_threshold, 10_000); + assert!(cfg.reconcile_enabled); + assert_eq!( + cfg.reconcile_interval, + std::time::Duration::from_millis(600_000) + ); + assert_eq!(cfg.reconcile_ledger_span, 400); + assert_eq!(cfg.reconcile_tip_margin, 100); }); } diff --git a/crates/indexer/src/main.rs b/crates/indexer/src/main.rs index 1905219d..a09c2045 100644 --- a/crates/indexer/src/main.rs +++ b/crates/indexer/src/main.rs @@ -15,6 +15,7 @@ mod health; mod metrics; mod parser; mod poll; +mod reconcile; mod redis_stream; mod rpc; mod spec; @@ -163,6 +164,32 @@ async fn main() -> Result<(), Box> { let relay_shutdown = shutdown.clone(); let relay_handle = tokio::spawn(async move { relay.run(relay_shutdown).await }); + // Ledger-range reconciliation (issue #511): periodically re-fetches a + // settled window from the RPC and compares per-ledger event counts + // against the database, so silent under-indexing surfaces in minutes + // instead of at the next incident. Read-only; stops on the same shutdown + // signal, and needs no drain — a pass in flight holds no state worth + // finishing. + if cfg.reconcile_enabled { + let reconcile_rpc = rpc::RpcClient::with_endpoints( + cfg.stellar_rpc_urls.clone(), + &rpc::RpcHttpSettings { + connect_timeout: cfg.rpc_connect_timeout, + request_timeout: cfg.rpc_request_timeout, + pool_idle_timeout: cfg.rpc_pool_idle_timeout, + pool_max_idle_per_host: cfg.rpc_pool_max_idle_per_host, + tcp_keepalive: cfg.rpc_tcp_keepalive, + }, + )?; + let reconciler = reconcile::Reconciler::new(&cfg, db_pool.clone(), reconcile_rpc); + let reconcile_shutdown = shutdown.clone(); + tokio::spawn(async move { reconciler.run(reconcile_shutdown).await }); + } else { + tracing::warn!( + "RECONCILE_ENABLED=false: nothing is verifying indexed counts against the RPC source" + ); + } + // Allow the shutdown drain to finish its in-flight work before the process // is killed. Kubernetes/Fly terminationGracePeriodSeconds should be ≥ this // value + a small buffer (recommended: SHUTDOWN_GRACE_SECS + 5). diff --git a/crates/indexer/src/metrics.rs b/crates/indexer/src/metrics.rs index 3c64052c..816fccc6 100644 --- a/crates/indexer/src/metrics.rs +++ b/crates/indexer/src/metrics.rs @@ -41,6 +41,25 @@ pub const RPC_TIMEOUTS_TOTAL: &str = "trident_indexer_rpc_timeouts_total"; pub const RPC_ACTIVE_ENDPOINT: &str = "trident_indexer_rpc_active_endpoint"; pub const RPC_FAILOVERS_TOTAL: &str = "trident_indexer_rpc_failovers_total"; pub const OUTBOX_BACKLOG: &str = "trident_indexer_outbox_backlog"; + +/// Reconciliation loop (issue #511): passes that completed a full compare of +/// a settled ledger window against the RPC source. +pub const RECONCILE_PASSES_TOTAL: &str = "trident_indexer_reconcile_passes_total"; +/// Passes that aborted before producing a report (RPC or DB failure). A +/// failing reconciler reports nothing — which must never read as clean. +pub const RECONCILE_PASS_FAILURES_TOTAL: &str = "trident_indexer_reconcile_pass_failures_total"; +/// Events the RPC reports for reconciled windows that the database does not +/// account for — the silent-under-indexing signal this loop exists to catch. +pub const RECONCILE_MISSING_EVENTS_TOTAL: &str = "trident_indexer_reconcile_missing_events_total"; +/// Events the database holds that the RPC does not report for the window — +/// over-indexing, as wrong as under-indexing. +pub const RECONCILE_EXTRA_EVENTS_TOTAL: &str = "trident_indexer_reconcile_extra_events_total"; +/// Ledgers in the most recent pass whose counts disagreed. Stays non-zero on +/// every pass until the discrepancy is resolved, which is what the alert +/// fires on. +pub const RECONCILE_DISCREPANT_LEDGERS: &str = "trident_indexer_reconcile_discrepant_ledgers"; +/// Highest ledger covered by the most recent completed pass. +pub const RECONCILE_WINDOW_END_LEDGER: &str = "trident_indexer_reconcile_window_end_ledger"; pub const OUTBOX_PUBLISHED_TOTAL: &str = "trident_indexer_outbox_published_total"; pub const OUTBOX_PUBLISH_FAILURES_TOTAL: &str = "trident_indexer_outbox_publish_failures_total"; /// RPC call latency in seconds, labelled by `method` (e.g. `getEvents`) and @@ -157,6 +176,30 @@ pub fn install(port: u16) -> Result<(), TridentError> { OUTBOX_PUBLISH_FAILURES_TOTAL, "Outbox publish attempts that failed (issue #200)" ); + describe_counter!( + RECONCILE_PASSES_TOTAL, + "Reconciliation passes that completed a full window compare (issue #511)" + ); + describe_counter!( + RECONCILE_PASS_FAILURES_TOTAL, + "Reconciliation passes that aborted before producing a report (issue #511)" + ); + describe_counter!( + RECONCILE_MISSING_EVENTS_TOTAL, + "Events on the RPC source that the database does not account for (issue #511)" + ); + describe_counter!( + RECONCILE_EXTRA_EVENTS_TOTAL, + "Events in the database that the RPC source does not report (issue #511)" + ); + describe_gauge!( + RECONCILE_DISCREPANT_LEDGERS, + "Ledgers in the most recent reconciliation pass with disagreeing counts (issue #511)" + ); + describe_gauge!( + RECONCILE_WINDOW_END_LEDGER, + "Highest ledger covered by the most recent completed reconciliation pass (issue #511)" + ); describe_gauge!( HEARTBEAT_TIMESTAMP, "Unix timestamp (seconds) of the most recent completed poll cycle (#218)" @@ -205,6 +248,12 @@ pub fn install(port: u16) -> Result<(), TridentError> { counter!(RPC_FAILOVERS_TOTAL).increment(0); counter!(OUTBOX_PUBLISHED_TOTAL).increment(0); counter!(OUTBOX_PUBLISH_FAILURES_TOTAL).increment(0); + counter!(RECONCILE_PASSES_TOTAL).increment(0); + counter!(RECONCILE_PASS_FAILURES_TOTAL).increment(0); + counter!(RECONCILE_MISSING_EVENTS_TOTAL).increment(0); + counter!(RECONCILE_EXTRA_EVENTS_TOTAL).increment(0); + gauge!(RECONCILE_DISCREPANT_LEDGERS).set(0.0); + gauge!(RECONCILE_WINDOW_END_LEDGER).set(0.0); gauge!(RPC_ACTIVE_ENDPOINT).set(0.0); gauge!(OUTBOX_BACKLOG).set(0.0); gauge!(LEDGER_LAG).set(0.0); @@ -329,6 +378,30 @@ pub fn record_dead_lettered() { counter!(DEAD_LETTERED_TOTAL).increment(1); } +pub fn record_reconcile_pass_completed() { + counter!(RECONCILE_PASSES_TOTAL).increment(1); +} + +pub fn record_reconcile_pass_failed() { + counter!(RECONCILE_PASS_FAILURES_TOTAL).increment(1); +} + +pub fn record_reconcile_missing_events(count: u64) { + counter!(RECONCILE_MISSING_EVENTS_TOTAL).increment(count); +} + +pub fn record_reconcile_extra_events(count: u64) { + counter!(RECONCILE_EXTRA_EVENTS_TOTAL).increment(count); +} + +pub fn set_reconcile_discrepant_ledgers(count: i64) { + gauge!(RECONCILE_DISCREPANT_LEDGERS).set(count as f64); +} + +pub fn set_reconcile_window_end(ledger: u64) { + gauge!(RECONCILE_WINDOW_END_LEDGER).set(ledger as f64); +} + pub fn record_poll_duration(seconds: f64) { histogram!(POLL_DURATION_SECONDS).record(seconds); } diff --git a/crates/indexer/src/reconcile.rs b/crates/indexer/src/reconcile.rs new file mode 100644 index 00000000..4023c2f3 --- /dev/null +++ b/crates/indexer/src/reconcile.rs @@ -0,0 +1,848 @@ +//! # Ledger-range reconciliation against the RPC source (issue #511) +//! +//! Nothing else proves that what Trident indexed matches what the chain +//! actually emitted: the streamer trusts its own poll loop, and without an +//! independent check, silent under-indexing is invisible — the API returns a +//! confident, incomplete answer. +//! +//! This job periodically re-fetches a settled ledger window from `getEvents` +//! and compares per-ledger event counts against the database. The RPC side +//! applies **exactly the ingest pipeline's own selection rules** — the same +//! server-side filter plan, the diagnostic gate, the failed-call skip, and +//! the allowlist/`index_from` boundaries — because indexed counts are not raw +//! RPC counts, and comparing unlike sets would make every report a false +//! positive. The database side counts `soroban_events` plus `parse_errors` +//! (an event we saw but could not decode is accounted for, not missing). +//! +//! Discrepancies are reported as **specific ledger ranges** (contiguous +//! discrepant ledgers coalesced), logged with both counts, and surfaced via +//! the `trident_indexer_reconcile_*` metrics that the +//! `TridentIndexerReconciliationMismatch` alert fires on. +//! +//! ## Continuous, not on-demand — and why +//! +//! The job runs as a slow in-process loop (default: every 10 minutes over +//! the most recent ~400 settled ledgers, a deliberate match for the nightly +//! testnet-correctness suite's window). Continuous operation is what makes +//! under-indexing visible in minutes rather than at the next incident, and +//! at this cadence the extra RPC load is a rounding error next to the poll +//! loop. For arbitrary historical ranges there is already an on-demand path: +//! `trident-backfill --dry-run` walks any window and reports counts without +//! writing. Both choices are documented in the alert runbook. + +use std::collections::HashMap; + +use tokio_util::sync::CancellationToken; + +use trident_common::TridentError; + +use crate::config::Config; +use crate::db; +use crate::metrics; +use crate::rpc::filters::build_event_filters; +use crate::rpc::{EventFilter, RawEvent, RpcClient}; + +/// Page size for the reconciliation walk — same as the RPC maximum the +/// streamer uses. +const PAGE_LIMIT: u32 = 200; + +/// Hard cap on pages per pass, so a pathological RPC response can never spin +/// the walk forever. 400 pages × 200 events comfortably covers the default +/// window. +const MAX_PAGES: u32 = 400; + +/// One contiguous run of ledgers whose indexed counts disagree with the RPC. +#[derive(Debug, PartialEq, Eq)] +pub struct DiscrepantRange { + pub from_ledger: u64, + pub to_ledger: u64, + /// Events the RPC reports for this range (after ingest selection rules). + pub rpc_events: u64, + /// Events accounted for in the database (indexed + parse-error rows). + pub db_events: u64, +} + +/// The outcome of one reconciliation pass. +#[derive(Debug, Default)] +pub struct ReconcileReport { + pub window_start: u64, + pub window_end: u64, + pub rpc_events: u64, + pub db_events: u64, + pub discrepant_ranges: Vec, + /// True when the walk hit MAX_PAGES before covering the requested + /// window. The compare window is then CLAMPED to the fully-walked + /// prefix (`window_end` reflects the clamp), so the ranges in this + /// report are still real — but ledgers past `window_end` were not + /// verified this pass. + pub truncated: bool, +} + +impl ReconcileReport { + pub fn missing_events(&self) -> u64 { + self.discrepant_ranges + .iter() + .map(|r| r.rpc_events.saturating_sub(r.db_events)) + .sum() + } + + pub fn extra_events(&self) -> u64 { + self.discrepant_ranges + .iter() + .map(|r| r.db_events.saturating_sub(r.rpc_events)) + .sum() + } +} + +/// The reconciliation loop. Construct with [`Reconciler::new`], then `run` +/// alongside the streamer; it stops on the shared shutdown token. +pub struct Reconciler { + db: sqlx::PgPool, + rpc: RpcClient, + network: String, + index_diagnostic: bool, + topic_filters: Vec>, + interval: std::time::Duration, + ledger_span: u64, + tip_margin: u64, +} + +impl Reconciler { + pub fn new(cfg: &Config, db: sqlx::PgPool, rpc: RpcClient) -> Self { + Self { + db, + rpc, + network: cfg.network.clone(), + index_diagnostic: cfg.index_diagnostic, + topic_filters: cfg.topic_filters.clone(), + interval: cfg.reconcile_interval, + ledger_span: cfg.reconcile_ledger_span, + tip_margin: cfg.reconcile_tip_margin, + } + } + + /// Loop until shutdown: one pass, log/meter the report, sleep. A pass + /// failure is logged and retried next interval — the reconciler is a + /// safety net and must never take the indexer down with it. + pub async fn run(&self, shutdown: CancellationToken) { + tracing::info!( + interval_secs = self.interval.as_secs(), + ledger_span = self.ledger_span, + tip_margin = self.tip_margin, + "Reconciliation loop starting" + ); + loop { + if shutdown.is_cancelled() { + break; + } + match self.run_pass().await { + Ok(report) => self.publish(&report), + Err(e) => { + metrics::record_reconcile_pass_failed(); + tracing::warn!(error = %e, "Reconciliation pass failed; will retry next interval"); + } + } + tokio::select! { + _ = tokio::time::sleep(self.interval) => {} + _ = shutdown.cancelled() => break, + } + } + tracing::info!("Reconciliation loop stopping"); + } + + fn publish(&self, report: &ReconcileReport) { + metrics::record_reconcile_pass_completed(); + metrics::set_reconcile_window_end(report.window_end); + metrics::set_reconcile_discrepant_ledgers( + report + .discrepant_ranges + .iter() + .map(|r| r.to_ledger - r.from_ledger + 1) + .sum::() as i64, + ); + let missing = report.missing_events(); + let extra = report.extra_events(); + if missing > 0 { + metrics::record_reconcile_missing_events(missing); + } + if extra > 0 { + metrics::record_reconcile_extra_events(extra); + } + + if report.truncated { + tracing::warn!( + window_start = report.window_start, + window_end = report.window_end, + "Reconciliation window was clamped at the page cap; ledgers past window_end were not verified this pass" + ); + } + if report.discrepant_ranges.is_empty() { + tracing::info!( + window_start = report.window_start, + window_end = report.window_end, + rpc_events = report.rpc_events, + db_events = report.db_events, + "Reconciliation clean: indexed counts match the RPC source" + ); + return; + } + for range in &report.discrepant_ranges { + tracing::warn!( + from_ledger = range.from_ledger, + to_ledger = range.to_ledger, + rpc_events = range.rpc_events, + db_events = range.db_events, + "Reconciliation discrepancy: indexed counts disagree with the RPC source for this ledger range" + ); + } + tracing::warn!( + window_start = report.window_start, + window_end = report.window_end, + ranges = report.discrepant_ranges.len(), + missing_events = missing, + extra_events = extra, + "Reconciliation pass found discrepancies" + ); + } + + /// One reconciliation pass over the most recent settled window. + pub async fn run_pass(&self) -> Result { + let tip = self.rpc.get_latest_ledger().await?; + let window_end = tip.saturating_sub(self.tip_margin); + let window_start = window_end + .saturating_sub(self.ledger_span.saturating_sub(1)) + .max(1); + if window_end == 0 || window_start > window_end { + return Err(TridentError::rpc(anyhow::anyhow!( + "chain tip {tip} leaves no settled window behind a margin of {}", + self.tip_margin + ))); + } + + // Only compare ledgers the indexer has actually passed: a window + // ahead of the cursor is not yet indexed and would read as one giant + // false "missing" range. + let cursor = db::get_cursor(&self.db).await?; + let window_end = window_end.min(cursor); + if window_end < window_start { + return Err(TridentError::rpc(anyhow::anyhow!( + "indexer cursor {cursor} has not reached the settled window starting at {window_start}; nothing to reconcile yet" + ))); + } + + // Mirror the streamer's server-side filter plan exactly (issue #203): + // same allowlist source, same topic patterns, same degraded-mode + // fallback to client-side filtering. + let allowlist = { + let map = db::load_indexed_contracts(&self.db, &self.network).await?; + if map.is_empty() { + None + } else { + Some(map) + } + }; + let contract_ids = allowlist.as_ref().map(|map| { + map.keys() + .cloned() + .collect::>() + }); + let plan = build_event_filters(contract_ids.as_ref(), &self.topic_filters); + + let rpc_counts = self + .count_rpc_events(window_start, window_end, &plan.filters, allowlist.as_ref()) + .await?; + + // A walk that hit the page cap covered only part of the window. The + // comparable range ends one ledger BEFORE the interruption point (the + // interrupted ledger may be partially counted); comparing the full + // window would turn every un-walked ledger into a fake "extra events" + // discrepancy and page the on-call with garbage ranges. + let mut window_end = window_end; + if rpc_counts.truncated { + let comparable_end = rpc_counts.last_seen_ledger.saturating_sub(1); + if comparable_end < window_start { + return Err(TridentError::rpc(anyhow::anyhow!( + "reconciliation walk hit the page cap before completing a single ledger; lower RECONCILE_LEDGER_SPAN (window [{window_start}, {window_end}])" + ))); + } + tracing::warn!( + window_start, + requested_end = window_end, + comparable_end, + "Reconciliation walk hit the page cap; comparing the covered prefix only — lower RECONCILE_LEDGER_SPAN if this persists" + ); + window_end = comparable_end; + } + + let db_counts = self.count_db_events(window_start, window_end).await?; + + Ok(build_report( + window_start, + window_end, + &rpc_counts.per_ledger, + &db_counts, + rpc_counts.truncated, + )) + } + + /// Walk `getEvents` across the window and count events per ledger, + /// applying the ingest pipeline's client-side selection rules. + async fn count_rpc_events( + &self, + window_start: u64, + window_end: u64, + filters: &[EventFilter], + allowlist: Option<&HashMap>, + ) -> Result { + let mut per_ledger: HashMap = HashMap::new(); + let mut cursor: Option = None; + let mut start: Option = Some(window_start); + let mut truncated = true; + let mut last_seen_ledger = 0u64; + + 'pages: for _ in 0..MAX_PAGES { + let page = self + .rpc + .get_events(start, cursor.clone(), PAGE_LIMIT, filters) + .await?; + // Only the first request anchors by ledger; later ones resume by + // cursor (the RPC rejects requests carrying both). + start = None; + + let count = page.events.len(); + if count == 0 { + truncated = false; + break; + } + + for event in page.events { + let ledger: u64 = event.ledger.parse().map_err(|_| { + TridentError::parse(anyhow::anyhow!( + "event {} reported an unparseable ledger {:?}", + event.id, + event.ledger + )) + })?; + if ledger > window_end { + truncated = false; + break 'pages; + } + cursor = Some(event.page_cursor()); + last_seen_ledger = last_seen_ledger.max(ledger); + if self.counts_toward_index(&event, ledger, allowlist) { + *per_ledger.entry(ledger).or_insert(0) += 1; + } + } + + if count < PAGE_LIMIT as usize { + truncated = false; + break; + } + } + + Ok(RpcCounts { + per_ledger, + truncated, + last_seen_ledger, + }) + } + + /// The ingest pipeline's client-side selection rules, mirrored from + /// `Parser::parse_event_with_projection` and the streamer's allowlist + /// check. Any rule added there must be added here, or reconciliation + /// reports false positives — the parity test below pins the behavior. + fn counts_toward_index( + &self, + event: &RawEvent, + ledger: u64, + allowlist: Option<&HashMap>, + ) -> bool { + if event.event_type == "diagnostic" && !self.index_diagnostic { + return false; + } + if !event.in_successful_contract_call { + return false; + } + if let Some(filter) = allowlist { + let contract_id = event.contract_id.as_deref().unwrap_or_default(); + match filter.get(contract_id) { + None => return false, + Some(&index_from) if (ledger as i64) < index_from => return false, + _ => {} + } + } + true + } + + /// Per-ledger accounted-for counts from the database: indexed events plus + /// parse-error rows (seen but undecodable — captured, not missing). + async fn count_db_events( + &self, + window_start: u64, + window_end: u64, + ) -> Result, TridentError> { + let mut per_ledger: HashMap = HashMap::new(); + + let indexed: Vec<(i64, i64)> = sqlx::query_as( + "SELECT ledger_sequence, COUNT(*) FROM soroban_events + WHERE ledger_sequence BETWEEN $1 AND $2 + GROUP BY ledger_sequence", + ) + .bind(window_start as i64) + .bind(window_end as i64) + .fetch_all(&self.db) + .await + .map_err(|e| { + TridentError::storage(anyhow::Error::new(e).context("reconcile count events")) + })?; + for (ledger, count) in indexed { + *per_ledger.entry(ledger as u64).or_insert(0) += count as u64; + } + + let parse_errors: Vec<(i64, i64)> = sqlx::query_as( + "SELECT ledger_sequence, COUNT(*) FROM parse_errors + WHERE ledger_sequence BETWEEN $1 AND $2 + GROUP BY ledger_sequence", + ) + .bind(window_start as i64) + .bind(window_end as i64) + .fetch_all(&self.db) + .await + .map_err(|e| { + TridentError::storage(anyhow::Error::new(e).context("reconcile count parse errors")) + })?; + for (ledger, count) in parse_errors { + *per_ledger.entry(ledger as u64).or_insert(0) += count as u64; + } + + Ok(per_ledger) + } +} + +struct RpcCounts { + per_ledger: HashMap, + truncated: bool, + /// Highest ledger whose events were fully consumed before the walk + /// stopped. Meaningful only when `truncated`: the ledger the page cap + /// interrupted may be partially counted, so the comparable window ends + /// one ledger before it. + last_seen_ledger: u64, +} + +/// Compare per-ledger counts and coalesce contiguous discrepant ledgers into +/// ranges — the issue asks for *specific ledger ranges*, and one warn line +/// per affected range beats four hundred per-ledger lines. +fn build_report( + window_start: u64, + window_end: u64, + rpc: &HashMap, + db: &HashMap, + truncated: bool, +) -> ReconcileReport { + let mut report = ReconcileReport { + window_start, + window_end, + truncated, + ..Default::default() + }; + + // Counts outside the window (a partially walked ledger past a clamped + // end) are deliberately ignored: the loop below only reads ledgers + // inside [window_start, window_end]. + let mut open: Option = None; + for ledger in window_start..=window_end { + let rpc_count = rpc.get(&ledger).copied().unwrap_or(0); + let db_count = db.get(&ledger).copied().unwrap_or(0); + report.rpc_events += rpc_count; + report.db_events += db_count; + + if rpc_count == db_count { + if let Some(range) = open.take() { + report.discrepant_ranges.push(range); + } + continue; + } + + match open.as_mut() { + Some(range) => { + range.to_ledger = ledger; + range.rpc_events += rpc_count; + range.db_events += db_count; + } + None => { + open = Some(DiscrepantRange { + from_ledger: ledger, + to_ledger: ledger, + rpc_events: rpc_count, + db_events: db_count, + }); + } + } + } + if let Some(range) = open.take() { + report.discrepant_ranges.push(range); + } + + report +} + +#[cfg(test)] +mod tests { + use super::*; + + fn counts(pairs: &[(u64, u64)]) -> HashMap { + pairs.iter().copied().collect() + } + + #[test] + fn matching_counts_produce_a_clean_report() { + let rpc = counts(&[(10, 3), (11, 1)]); + let db = counts(&[(10, 3), (11, 1)]); + let report = build_report(10, 12, &rpc, &db, false); + assert!(report.discrepant_ranges.is_empty()); + assert_eq!(report.rpc_events, 4); + assert_eq!(report.db_events, 4); + } + + #[test] + fn contiguous_discrepant_ledgers_coalesce_into_one_range() { + // Ledgers 11-13 all disagree; 10 and 14 agree. + let rpc = counts(&[(10, 2), (11, 3), (12, 1), (13, 2)]); + let db = counts(&[(10, 2), (11, 1), (13, 1), (14, 0)]); + let report = build_report(10, 14, &rpc, &db, false); + assert_eq!( + report.discrepant_ranges, + vec![DiscrepantRange { + from_ledger: 11, + to_ledger: 13, + rpc_events: 6, + db_events: 2, + }] + ); + assert_eq!(report.missing_events(), 4); + assert_eq!(report.extra_events(), 0); + } + + #[test] + fn separate_discrepancies_report_separate_ranges() { + // Discrepant ledgers 10, 12, 14 with clean ledgers 11 and 13 between + // them: three distinct ranges, not one. + let rpc = counts(&[(10, 1), (14, 1)]); + let db = counts(&[(12, 5)]); + let report = build_report(10, 14, &rpc, &db, false); + assert_eq!(report.discrepant_ranges.len(), 3); + // Extra events (indexed rows the chain never emitted) are reported + // too — over-indexing is as wrong as under-indexing. + assert_eq!(report.missing_events(), 2); + assert_eq!(report.extra_events(), 5); + } + + #[test] + fn counts_outside_the_window_are_ignored() { + // A clamped (truncated) walk can leave a partially counted ledger + // past the compare window in the RPC map; it must not surface as a + // discrepancy. + let rpc = counts(&[(10, 1), (13, 7)]); + let db = counts(&[(10, 1)]); + let report = build_report(10, 12, &rpc, &db, true); + assert!(report.discrepant_ranges.is_empty()); + assert!(report.truncated); + } + + #[test] + fn ledgers_with_zero_events_on_both_sides_are_clean() { + let report = build_report(100, 500, &HashMap::new(), &HashMap::new(), false); + assert!(report.discrepant_ranges.is_empty()); + } + + // ----------------------------------------------------------------------- + // Integration: a full pass against a mock RPC and a real database. + // ----------------------------------------------------------------------- + + use serde_json::json; + use wiremock::matchers::{body_partial_json, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + /// Same env-gated skip the other DB-touching test modules use: silently + /// skipped without TEST_DATABASE_URL, hard-failed when + /// REQUIRE_TEST_SERVICES is set so CI cannot skip by accident. + fn test_db_url() -> Option { + match std::env::var("TEST_DATABASE_URL") { + Ok(url) if !url.is_empty() => Some(url), + _ => { + if std::env::var("REQUIRE_TEST_SERVICES").is_ok() { + panic!("REQUIRE_TEST_SERVICES is set but TEST_DATABASE_URL is missing"); + } + eprintln!("SKIP: TEST_DATABASE_URL not set"); + None + } + } + } + + fn raw_event( + ledger: u64, + idx: u32, + contract: &str, + event_type: &str, + successful: bool, + ) -> serde_json::Value { + json!({ + "type": event_type, + "ledger": ledger.to_string(), + "ledgerClosedAt": "2024-01-01T00:00:00Z", + "contractId": contract, + "id": format!("{ledger:016}-{idx}"), + "pagingToken": format!("{ledger}-{idx}"), + "txHash": format!("rechash{ledger}{idx}"), + "topic": ["AAAADwAAAAh0cmFuc2Zlcg=="], + "value": "", + "inSuccessfulContractCall": successful + }) + } + + /// These two tests mutate genuinely global state — the `indexed_contracts` + /// allowlist and the `latest_ledger_cursor` row — so they hold this for + /// their whole run instead of racing each other (and use a ledger window, + /// [30201, 30600], that no other suite's fixtures touch). + static RECONCILE_DB_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + + async fn seed_event(pool: &sqlx::PgPool, ledger: u64, idx: u32, contract: &str) { + sqlx::query( + "INSERT INTO soroban_events + (id, contract_id, ledger_sequence, ledger_timestamp, transaction_hash, + event_index, event_type, topics, data) + VALUES (gen_random_uuid(), $1, $2, NOW(), $3, $4, 'contract', '[]', '{}')", + ) + .bind(contract) + .bind(ledger as i64) + .bind(format!("rechash{ledger}{idx}")) + .bind(idx as i32) + .execute(pool) + .await + .expect("seed event"); + } + + /// The acceptance scenario for issue #511: a deliberately incomplete and + /// a deliberately over-indexed ledger are both reported with their exact + /// ranges, the ingest pipeline's skip rules are mirrored (a failed-call + /// event and a diagnostic event on the RPC side do NOT read as missing), + /// and a parse-error row counts as accounted for. + #[tokio::test] + async fn pass_reports_missing_and_extra_ledger_ranges() { + let _guard = RECONCILE_DB_LOCK.lock().await; + let Some(db_url) = test_db_url() else { return }; + let pool = sqlx::PgPool::connect(&db_url).await.expect("db connect"); + + sqlx::query("DELETE FROM soroban_events WHERE ledger_sequence BETWEEN 30201 AND 30600") + .execute(&pool) + .await + .expect("clear events"); + sqlx::query("DELETE FROM parse_errors WHERE ledger_sequence BETWEEN 30201 AND 30600") + .execute(&pool) + .await + .expect("clear parse errors"); + sqlx::query("DELETE FROM indexed_contracts") + .execute(&pool) + .await + .expect("clear allowlist"); + sqlx::query("UPDATE system_state SET value = '30600' WHERE key = 'latest_ledger_cursor'") + .execute(&pool) + .await + .expect("set cursor"); + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/")) + .and(body_partial_json(json!({"method": "getLatestLedger"}))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", "id": 1, + "result": {"sequence": 30700} + }))) + .mount(&server) + .await; + // Chain truth (window [30201, 30600] at span 400 / margin 100 / tip 1000): + // 30550: two countable events -> DB has 1 indexed + 1 parse + // error, so it is fully accounted for (clean). + // 30551: one FAILED-call event -> must not count (clean). + // 30552: one diagnostic event -> must not count (clean). + // 30553: one countable event -> DB has nothing (missing). + // 30560: nothing -> DB has one row (extra). + Mock::given(method("POST")) + .and(path("/")) + .and(body_partial_json(json!({"method": "getEvents"}))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", "id": 1, + "result": { + "events": [ + raw_event(30550, 0, "CRECON", "contract", true), + raw_event(30550, 1, "CRECON", "contract", true), + raw_event(30551, 0, "CRECON", "contract", false), + raw_event(30552, 0, "CRECON", "diagnostic", true), + raw_event(30553, 0, "CRECON", "contract", true), + ], + "latestLedger": 30700 + } + }))) + .mount(&server) + .await; + + seed_event(&pool, 30550, 0, "CRECON").await; + sqlx::query( + "INSERT INTO parse_errors (ledger_sequence, event_index, raw_payload, error_message) + VALUES (30550, 1, '{}', 'test decode failure')", + ) + .execute(&pool) + .await + .expect("seed parse error"); + seed_event(&pool, 30560, 0, "CRECON").await; + + let rpc = RpcClient::with_endpoints( + vec![server.uri()], + &crate::rpc::RpcHttpSettings { + connect_timeout: std::time::Duration::from_secs(5), + request_timeout: std::time::Duration::from_secs(30), + pool_idle_timeout: std::time::Duration::from_secs(90), + pool_max_idle_per_host: 8, + tcp_keepalive: std::time::Duration::from_secs(60), + }, + ) + .expect("rpc client"); + + let reconciler = Reconciler { + db: pool.clone(), + rpc, + network: "testnet".to_string(), + index_diagnostic: false, + topic_filters: Vec::new(), + interval: std::time::Duration::from_secs(600), + ledger_span: 400, + tip_margin: 100, + }; + + let report = reconciler.run_pass().await.expect("pass"); + + assert_eq!(report.window_start, 30201); + assert_eq!(report.window_end, 30600); + assert!(!report.truncated); + assert_eq!( + report.discrepant_ranges, + vec![ + DiscrepantRange { + from_ledger: 30553, + to_ledger: 30553, + rpc_events: 1, + db_events: 0, + }, + DiscrepantRange { + from_ledger: 30560, + to_ledger: 30560, + rpc_events: 0, + db_events: 1, + }, + ], + "exactly the corrupted ledgers must be reported, as specific ranges" + ); + assert_eq!(report.missing_events(), 1); + assert_eq!(report.extra_events(), 1); + + pool.close().await; + } + + /// The allowlist and per-contract index_from boundaries are applied to + /// the RPC side, mirroring the streamer — otherwise every skipped event + /// would read as missing. + #[tokio::test] + async fn allowlist_rules_are_mirrored_on_the_rpc_side() { + let _guard = RECONCILE_DB_LOCK.lock().await; + let Some(db_url) = test_db_url() else { return }; + let pool = sqlx::PgPool::connect(&db_url).await.expect("db connect"); + + sqlx::query("DELETE FROM soroban_events WHERE ledger_sequence BETWEEN 30201 AND 30600") + .execute(&pool) + .await + .expect("clear events"); + sqlx::query("DELETE FROM parse_errors WHERE ledger_sequence BETWEEN 30201 AND 30600") + .execute(&pool) + .await + .expect("clear parse errors"); + sqlx::query("DELETE FROM indexed_contracts") + .execute(&pool) + .await + .expect("clear allowlist"); + sqlx::query( + "INSERT INTO indexed_contracts (contract_id, network, index_from) + VALUES ('CLISTED', 'testnet', 30555)", + ) + .execute(&pool) + .await + .expect("seed allowlist"); + sqlx::query("UPDATE system_state SET value = '30600' WHERE key = 'latest_ledger_cursor'") + .execute(&pool) + .await + .expect("set cursor"); + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/")) + .and(body_partial_json(json!({"method": "getLatestLedger"}))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", "id": 1, + "result": {"sequence": 30700} + }))) + .mount(&server) + .await; + // 30550: listed contract but BELOW its index_from -> not counted. + // 30560: unlisted contract -> not counted. + // 30570: listed, at/above index_from -> counted; DB has it (clean). + Mock::given(method("POST")) + .and(path("/")) + .and(body_partial_json(json!({"method": "getEvents"}))) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", "id": 1, + "result": { + "events": [ + raw_event(30550, 0, "CLISTED", "contract", true), + raw_event(30560, 0, "CUNLISTED", "contract", true), + raw_event(30570, 0, "CLISTED", "contract", true), + ], + "latestLedger": 30700 + } + }))) + .mount(&server) + .await; + + seed_event(&pool, 30570, 0, "CLISTED").await; + + let rpc = RpcClient::with_endpoints( + vec![server.uri()], + &crate::rpc::RpcHttpSettings { + connect_timeout: std::time::Duration::from_secs(5), + request_timeout: std::time::Duration::from_secs(30), + pool_idle_timeout: std::time::Duration::from_secs(90), + pool_max_idle_per_host: 8, + tcp_keepalive: std::time::Duration::from_secs(60), + }, + ) + .expect("rpc client"); + + let reconciler = Reconciler { + db: pool.clone(), + rpc, + network: "testnet".to_string(), + index_diagnostic: false, + topic_filters: Vec::new(), + interval: std::time::Duration::from_secs(600), + ledger_span: 400, + tip_margin: 100, + }; + + let report = reconciler.run_pass().await.expect("pass"); + assert!( + report.discrepant_ranges.is_empty(), + "skip-rule parity must hold, got {:?}", + report.discrepant_ranges + ); + + pool.close().await; + } +} diff --git a/crates/indexer/src/rpc/mod.rs b/crates/indexer/src/rpc/mod.rs index 9a0e9009..87a92bd4 100644 --- a/crates/indexer/src/rpc/mod.rs +++ b/crates/indexer/src/rpc/mod.rs @@ -144,12 +144,11 @@ struct GetLedgersResult { /// `getLatestLedger` takes no parameters, but the JSON-RPC envelope this client /// builds always serialises a `params` member. /// -/// Test-gated alongside [`RpcClient::get_latest_ledger`], its only caller. -#[cfg(test)] +/// Wire types for [`RpcClient::get_latest_ledger`] (un-gated with it for the +/// reconciliation loop, issue #511). #[derive(Serialize)] struct EmptyParams {} -#[cfg(test)] #[derive(Deserialize)] struct GetLatestLedgerResult { sequence: u64, @@ -516,12 +515,13 @@ impl RpcClient { /// a caller who does not yet know the tip cannot supply. This method has no /// such precondition. /// - /// Test-gated: the poll loop learns the tip from the `getEvents` responses - /// it already makes, so the running indexer has no reason to spend an extra - /// round trip on it. Only the testnet correctness suite (issue #419), which - /// must choose a ledger window before it can request anything, needs it. - /// Remove the gate if a production caller ever appears. - #[cfg(test)] + /// The poll loop learns the tip from the `getEvents` responses it already + /// makes, so it never calls this. The testnet correctness suite (issue + /// #419) and the reconciliation loop (issue #511) both need to choose a + /// settled ledger window before they can request anything, which is + /// exactly the case this method exists for. (Previously `#[cfg(test)]` + /// with a note to remove the gate when a production caller appeared — + /// the reconciler is that caller.) pub async fn get_latest_ledger(&self) -> Result { let result: GetLatestLedgerResult = self .call("getLatestLedger", 3, EmptyParams {}, "getLatestLedger") diff --git a/crates/indexer/src/streamer/mod.rs b/crates/indexer/src/streamer/mod.rs index 469263cf..1687b433 100644 --- a/crates/indexer/src/streamer/mod.rs +++ b/crates/indexer/src/streamer/mod.rs @@ -1493,6 +1493,10 @@ mod tests { outbox_poll_interval: Duration::from_millis(10), outbox_batch_size: 500, outbox_backlog_alert_threshold: 10_000, + reconcile_enabled: false, + reconcile_interval: Duration::from_secs(600), + reconcile_ledger_span: 400, + reconcile_tip_margin: 100, metrics_port: 0, alert_webhook_url: None, alert_lag_threshold: 200, diff --git a/docs/ENVIRONMENT.md b/docs/ENVIRONMENT.md index f41a4a74..6e61b503 100644 --- a/docs/ENVIRONMENT.md +++ b/docs/ENVIRONMENT.md @@ -84,6 +84,10 @@ description is accurate. Keep this file honest by hand. | `OUTBOX_POLL_INTERVAL_MS` | Optional | `100` (min `10`, max `60000`) | How often the relay scans for unpublished events. | | `OUTBOX_BATCH_SIZE` | Optional | `500` (min `1`, max `10000`) | Max events published per relay pass. | | `OUTBOX_BACKLOG_ALERT_THRESHOLD` | Optional | `10000` (min `1`, max `10000000`) | Backlog size that logs an alert-worthy warning. | +| `RECONCILE_ENABLED` | Optional | `true` | Runs the ledger-range reconciliation loop that compares indexed event counts against the RPC source (#511). Disabling it means nothing verifies indexed data against the chain. | +| `RECONCILE_INTERVAL_MS` | Optional | `600000` (min `10000`, max `86400000`) | Time between reconciliation passes. | +| `RECONCILE_LEDGER_SPAN` | Optional | `400` (min `10`, max `100000`) | Settled ledgers each pass compares. | +| `RECONCILE_TIP_MARGIN` | Optional | `100` (min `0`, max `10000`) | Distance behind the chain tip the window sits, so in-flight ledgers never read as discrepancies. | ### Lag alerting diff --git a/docs/metrics-catalog.md b/docs/metrics-catalog.md index ed97c6d5..7f9948d0 100644 --- a/docs/metrics-catalog.md +++ b/docs/metrics-catalog.md @@ -26,6 +26,12 @@ port `9090`, set via `METRICS_PORT`). Defined in | `trident_indexer_last_poll_timestamp_seconds` | gauge | — | unix seconds | Set once per poll-loop iteration regardless of outcome — the dead-man's-switch (#218). Stale means the loop is hung, not just slow. | | `trident_indexer_db_pool_size` | gauge | — | connections | Current size of the indexer's own Postgres pool. | | `trident_indexer_db_pool_idle_connections` | gauge | — | connections | Idle connections in the indexer's own Postgres pool. | +| `trident_indexer_reconcile_passes_total` | counter | - | passes | Reconciliation passes that completed a full settled-window compare against the RPC source (#511). | +| `trident_indexer_reconcile_pass_failures_total` | counter | - | passes | Reconciliation passes that aborted before producing a report. While these grow, mismatch silence is unknown, not clean. | +| `trident_indexer_reconcile_missing_events_total` | counter | - | events | Events the RPC reports (after ingest selection rules) that the database does not account for - silent under-indexing. | +| `trident_indexer_reconcile_extra_events_total` | counter | - | events | Events in the database that the RPC does not report for the window - over-indexing. | +| `trident_indexer_reconcile_discrepant_ledgers` | gauge | - | ledgers | Ledgers in the most recent pass whose counts disagreed; stays non-zero every pass until resolved. Alerted via `TridentIndexerReconciliationMismatch`. | +| `trident_indexer_reconcile_window_end_ledger` | gauge | - | ledger | Highest ledger covered by the most recent completed pass. | | `trident_indexer_catchup_ledgers_per_second` | gauge | — | ledgers/sec | Backfill rate while behind the chain tip (issue #420). **Only exported while catching up** — absent, not zero, once the lag drops below 10 ledgers. See [performance.md](performance.md#indexer-catch-up-throughput). | | `trident_indexer_catchup_events_per_second` | gauge | — | events/sec | Backfill rate in events, over the same window as the gauge above. Reported alongside it because ledgers/sec alone hides whether a sparse or dense range is being processed. | diff --git a/docs/runbooks/alerts.md b/docs/runbooks/alerts.md index 507f5946..d8f33bf4 100644 --- a/docs/runbooks/alerts.md +++ b/docs/runbooks/alerts.md @@ -117,6 +117,54 @@ malformed events. 3. If it's a new, valid event shape, this is a parser bug — file/fix rather than treating it as transient. +## TridentIndexerReconciliationMismatch + +**Means:** the reconciliation loop (issue #511) re-fetched a settled ledger +window from `getEvents` - applying the ingest pipeline's own filter and skip +rules - and the per-ledger event counts disagree with what the database +holds (`soroban_events` plus `parse_errors`). Some ledgers are +under-indexed (missing events) or over-indexed (extra events). + +**Why this threshold:** any disagreement at all means the indexed data is +wrong for those ledgers; the gauge is refreshed every pass (default 10 +minutes), so `for: 15m` means at least two consecutive passes agreed on the +disagreement. Warning rather than critical while the detector is new; +ratchet to critical once it has run clean on testnet for a while. + +**First steps:** +1. Find the `Reconciliation discrepancy` warn logs - they name each ledger + range with the RPC and database counts + (`trident_indexer_reconcile_missing_events_total` vs + `_extra_events_total` says which direction). +2. For missing events, re-ingest the reported ranges: + `trident-backfill --from-ledger --to-ledger ` (idempotent). + Use `--dry-run` first to preview counts for any range on demand. +3. If the discrepancy reappears on later passes for NEW ranges, the ingest + pipeline is dropping events right now - check parse-error rates, RPC + health, and recent deploys before backfilling further. +4. Extra events (indexed rows the chain does not report) usually mean a + backfill wrote rows outside the allowlist rules or a duplicate-index bug + - inspect the rows in the reported range before deleting anything. + +## TridentIndexerReconciliationFailing + +**Means:** the reconciliation loop keeps aborting before producing a report +- `getLatestLedger`/`getEvents` failures, or database errors during the +count queries. + +**Why this threshold:** a single failed pass self-heals next interval; more +than two failures inside 30 minutes, sustained for 30 minutes, means the +loop has effectively stopped verifying. While this fires, the mismatch +alert's silence is unknown, not clean. + +**First steps:** +1. Check the `Reconciliation pass failed` warn logs for the error. +2. If RPC-related, see `TridentIndexerRPCErrorRateHigh` - the reconciler + shares the endpoint pool and fails alongside it. +3. If the indexer cursor has not yet reached the settled window (fresh + deploy, deep backfill), passes fail with "nothing to reconcile yet" - + expected until the indexer catches up. + ## TridentIndexerRPCErrorRateHigh **Means:** over 5% of Stellar RPC calls (`getEvents`/`getLedgers`) errored in diff --git a/monitoring/alerts.yml b/monitoring/alerts.yml index a5b38fab..bee2b968 100644 --- a/monitoring/alerts.yml +++ b/monitoring/alerts.yml @@ -126,6 +126,42 @@ groups: recognise yet. runbook_url: "docs/runbooks/alerts.md#tridentindexerparseerrorratehigh" + # --------------------------------------------------------------------------- + # Ledger-range reconciliation against the RPC source (#511). + # --------------------------------------------------------------------------- + - name: trident.indexer.reconciliation + rules: + - alert: TridentIndexerReconciliationMismatch + expr: trident_indexer_reconcile_discrepant_ledgers > 0 + for: 15m + labels: + severity: warning + service: indexer + annotations: + summary: "Trident indexed counts disagree with the RPC source for {{ $value }} ledger(s)" + description: > + The reconciliation loop compared indexed event counts per ledger + against getEvents over a settled window and found disagreement. + The indexer logs name the exact ledger ranges with both counts. + This is the silent-under-indexing signal: the API may be + returning confident, incomplete answers for these ledgers. + runbook_url: "docs/runbooks/alerts.md#tridentindexerreconciliationmismatch" + + - alert: TridentIndexerReconciliationFailing + expr: increase(trident_indexer_reconcile_pass_failures_total[30m]) > 2 + for: 30m + labels: + severity: warning + service: indexer + annotations: + summary: "Trident reconciliation passes are failing" + description: > + The reconciliation loop keeps aborting before producing a report + (RPC or database failures). While this fires, nothing is + verifying that indexed counts match the chain - treat the + mismatch alert's silence as unknown, not clean. + runbook_url: "docs/runbooks/alerts.md#tridentindexerreconciliationfailing" + # --------------------------------------------------------------------------- # Stellar RPC health (#297). # ---------------------------------------------------------------------------