feat(indexer): reconcile indexed event counts against the RPC source

.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)

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

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<dyn std::error::Error>> {
     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).

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