feat(indexer): handle ledger reorgs and rollbacks without stale rows (#196)

crates/indexer/src/config.rs

@@ -74,6 +74,8 @@ pub struct Config {
     /// Operator-configured classic assets to resolve SAC events for (issue
     /// #262). Each is a `code:issuer` pair, or the literal `native` for XLM.
     pub tracked_sac_assets: Vec<crate::parser::sac::TrackedAsset>,
+    /// Maximum allowable reorg depth in ledgers before halting for operator intervention (issue #196).
+    pub max_reorg_depth: u64,
 }
 
 /// Default Postgres pool size for the indexer. It is a single writer with low
@@ -173,6 +175,7 @@ impl Config {
             60,
             2_592_000,
         );
+        let max_reorg_depth = parse_bounded_u64("MAX_REORG_DEPTH", 128, 1, 10_000);
         let db_pool_size = parse_pool_size("INDEXER_DB_POOL_SIZE", DEFAULT_DB_POOL_SIZE);
 
         // Collect all parse/range errors at once.
@@ -220,6 +223,7 @@ impl Config {
                 "TOKEN_METADATA_REFRESH_INTERVAL_SECS",
                 token_metadata_refresh_interval_secs.as_ref(),
             ),
+            ("MAX_REORG_DEPTH", max_reorg_depth.as_ref()),
         ] {
             if let Err(e) = result {
                 errors.push(format!("[indexer] {key}: {e}"));
@@ -347,6 +351,7 @@ impl Config {
             ),
             network_passphrase,
             tracked_sac_assets,
+            max_reorg_depth: max_reorg_depth.unwrap(),
         })
     }
 }

crates/indexer/src/db/mod.rs

@@ -719,6 +719,95 @@ pub async fn get_cursor(pool: &PgPool) -> Result<u64, TridentError> {
         .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("cursor parse")))
 }
 
+/// Fetch recent ledger sequences and hashes from `ledger_metadata` ordered by sequence descending.
+/// Used for reorg detection (issue #196).
+pub async fn get_recent_ledger_metadata(
+    pool: &PgPool,
+    limit: i64,
+) -> Result<Vec<(u64, String)>, TridentError> {
+    let rows: Vec<(i64, String)> = sqlx::query_as(
+        r#"
+        SELECT ledger_sequence, ledger_hash
+        FROM ledger_metadata
+        ORDER BY ledger_sequence DESC
+        LIMIT $1
+        "#,
+    )
+    .bind(limit)
+    .fetch_all(pool)
+    .await
+    .map_err(|e| {
+        TridentError::storage(anyhow::Error::new(e).context("get_recent_ledger_metadata"))
+    })?;
+
+    Ok(rows
+        .into_iter()
+        .map(|(seq, hash)| (seq as u64, hash))
+        .collect())
+}
+
+/// Atomically remove all indexed data from `from_sequence` onwards and rewind the cursor to `new_cursor` (issue #196).
+pub async fn handle_reorg_rollback(
+    pool: &PgPool,
+    from_sequence: u64,
+    new_cursor: u64,
+) -> Result<(), TridentError> {
+    let mut tx = pool
+        .begin()
+        .await
+        .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("reorg begin tx")))?;
+
+    let seq_i64 = from_sequence as i64;
+
+    // Delete token projections first (foreign key / dependent rows)
+    sqlx::query("DELETE FROM token_events WHERE ledger_sequence >= $1")
+        .bind(seq_i64)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("reorg delete token_events")))?;
+
+    // Delete invocation metrics
+    sqlx::query("DELETE FROM contract_invocation_metrics WHERE ledger_sequence >= $1")
+        .bind(seq_i64)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("reorg delete contract_invocation_metrics")))?;
+
+    // Delete storage snapshots
+    sqlx::query("DELETE FROM contract_storage_snapshots WHERE ledger_sequence >= $1")
+        .bind(seq_i64)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("reorg delete contract_storage_snapshots")))?;
+
+    // Delete primary soroban events
+    sqlx::query("DELETE FROM soroban_events WHERE ledger_sequence >= $1")
+        .bind(seq_i64)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("reorg delete soroban_events")))?;
+
+    // Delete ledger metadata
+    sqlx::query("DELETE FROM ledger_metadata WHERE ledger_sequence >= $1")
+        .bind(seq_i64)
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("reorg delete ledger_metadata")))?;
+
+    // Rewind cursor in system_state
+    sqlx::query("UPDATE system_state SET value = $1, updated_at = NOW() WHERE key = 'latest_ledger_cursor'")
+        .bind(new_cursor.to_string())
+        .execute(&mut *tx)
+        .await
+        .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("reorg rewind cursor")))?;
+
+    tx.commit()
+        .await
+        .map_err(|e| TridentError::storage(anyhow::Error::new(e).context("reorg commit tx")))?;
+
+    Ok(())
+}
+
 /// Write indexer health metrics into the `system_state` health columns after
 /// every successful poll cycle (issue #62).
 ///

crates/indexer/src/metrics.rs

@@ -40,6 +40,7 @@ pub const EFFECTIVE_POLL_INTERVAL_MS: &str = "trident_indexer_effective_poll_int
 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 REORGS_TOTAL: &str = "trident_indexer_reorgs_total";
 /// Count of ScVal values that hit the catch-all / debug-format fallback in
 /// `scval_to_string` or `scval_to_json` (issue #415). A high rate means the
 /// indexer is encountering Soroban types it cannot render as structured data.
@@ -135,6 +136,10 @@ pub fn install(port: u16) -> Result<(), TridentError> {
         RPC_FAILOVERS_TOTAL,
         "Times the indexer failed over to another RPC endpoint (issue #213)"
     );
+    describe_counter!(
+        REORGS_TOTAL,
+        "Total number of ledger reorganisations / rollbacks detected and reconciled (issue #196)"
+    );
     describe_gauge!(
         OUTBOX_BACKLOG,
         "Committed events not yet published to the Redis stream (issue #200)"
@@ -318,6 +323,10 @@ pub fn record_unhandled_scvariant() {
     counter!(UNHANDLED_SCVARIANT_TOTAL).increment(1);
 }
 
+pub fn record_reorg() {
+    counter!(REORGS_TOTAL).increment(1);
+}
+
 pub fn record_poll_duration(seconds: f64) {
     histogram!(POLL_DURATION_SECONDS).record(seconds);
 }

crates/indexer/src/streamer/mod.rs

@@ -498,10 +498,100 @@ impl Streamer {
         }
     }
 
+    /// Check for chain reorganisation / rollback before polling events (issue #196).
+    ///
+    /// Reorg conditions:
+    /// 1. `latest_ledger < *cursor`: The chain tip reported by RPC regressed behind our current cursor.
+    /// 2. Ledger hash mismatch: Compare recent stored ledger hashes from `ledger_metadata` against RPC `getLedgers`.
+    ///
+    /// If a reorg is detected:
+    /// - Verify that the rewind depth does not exceed `config.max_reorg_depth`.
+    /// - If it exceeds max depth, emit metric, log error, and return a fatal error.
+    /// - If within bounds, atomically delete affected rows, rewind cursor in `system_state`, update `*cursor`,
+    ///   record metric `metrics::record_reorg()`, and log structured warning.
+    async fn check_and_handle_reorg(&mut self, cursor: &mut u64) -> Result<(), TridentError> {
+        if *cursor == 0 {
+            return Ok(());
+        }
+
+        let latest_ledger = match self.rpc.get_latest_ledger().await {
+            Ok(seq) => seq,
+            Err(e) => {
+                tracing::debug!(error = %e, "Failed to fetch latest ledger for reorg check; will check on next poll");
+                return Ok(());
+            }
+        };
+
+        let mut reorg_start: Option<u64> = None;
+
+        // Condition 1: Chain tip regressed behind current cursor
+        if latest_ledger < *cursor {
+            reorg_start = Some(latest_ledger + 1);
+        } else {
+            // Condition 2: Check stored ledger hashes against RPC for recent ledgers
+            let check_count = (self.config.max_reorg_depth.min(10)) as i64;
+            let recent_ledgers = db::get_recent_ledger_metadata(&self.db, check_count).await?;
+
+            for (seq, stored_hash) in recent_ledgers {
+                if let Ok(Some(rpc_hash)) = self.rpc.get_ledger(seq).await {
+                    if rpc_hash != stored_hash {
+                        tracing::warn!(
+                            sequence = seq,
+                            stored_hash = %stored_hash,
+                            rpc_hash = %rpc_hash,
+                            "Ledger hash mismatch detected indicating reorg"
+                        );
+                        reorg_start = Some(match reorg_start {
+                            Some(existing) => existing.min(seq),
+                            None => seq,
+                        });
+                    }
+                }
+            }
+        }
+
+        if let Some(reorg_seq) = reorg_start {
+            let reorg_depth = cursor.saturating_sub(reorg_seq) + 1;
+
+            if reorg_depth > self.config.max_reorg_depth {
+                metrics::record_reorg();
+                tracing::error!(
+                    reorg_start = reorg_seq,
+                    cursor = *cursor,
+                    reorg_depth = reorg_depth,
+                    max_reorg_depth = self.config.max_reorg_depth,
+                    "Deep ledger reorganisation detected exceeding maximum allowed depth; halting indexer"
+                );
+                return Err(TridentError::indexer(anyhow::anyhow!(
+                    "Deep reorg of depth {} exceeds max allowed depth {}",
+                    reorg_depth,
+                    self.config.max_reorg_depth
+                )));
+            }
+
+            let new_cursor = reorg_seq.saturating_sub(1);
+            tracing::warn!(
+                reorg_start = reorg_seq,
+                cursor_before = *cursor,
+                new_cursor = new_cursor,
+                reorg_depth = reorg_depth,
+                "Ledger reorganisation detected; rolling back affected ledger data and rewinding cursor"
+            );
+
+            metrics::record_reorg();
+            db::handle_reorg_rollback(&self.db, reorg_seq, new_cursor).await?;
+            *cursor = new_cursor;
+        }
+
+        Ok(())
+    }
+
     /// Execute a single poll cycle. Fetches all available pages from the RPC
     /// starting at `cursor`, persists each event, and advances the cursor.
     /// Returns the total number of events processed in this cycle.
     async fn poll_once(&mut self, cursor: &mut u64) -> Result<usize, TridentError> {
+        self.check_and_handle_reorg(cursor).await?;
+
         let poll_start = Instant::now();
         // Cursor and lag as this cycle begins, so catch-up throughput can be
         // measured over the cycle (issue #420). `last_chain_tip` is the tip
@@ -2325,4 +2415,126 @@ mod tests {
 
         pool.close().await;
     }
+
+    #[tokio::test]
+    async fn reorg_detection_and_rollback_removes_stale_data() {
+        let (db_url, redis_url) = require_services!();
+        let server = MockServer::start().await;
+
+        let pool = sqlx::PgPool::connect(&db_url).await.unwrap();
+        reset_db(&pool).await;
+
+        // Seed old events and ledger metadata at sequence 101 and 102 with cursor at 102
+        sqlx::query(
+            r#"
+            INSERT INTO ledger_metadata (ledger_sequence, ledger_hash, ledger_timestamp, event_count)
+            VALUES (100, 'hash100', '2024-01-01T00:00:00Z', 1),
+                   (101, 'old_hash101', '2024-01-01T00:00:05Z', 1),
+                   (102, 'old_hash102', '2024-01-01T00:00:10Z', 1)
+            "#,
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+
+        sqlx::query(
+            r#"
+            INSERT INTO soroban_events (id, contract_id, ledger_sequence, ledger_timestamp, transaction_hash, event_index, topic_json, value_json, raw_payload)
+            VALUES ('00000000-0000-0000-0000-000000000101', 'CDEMO', 101, '2024-01-01T00:00:05Z', 'tx101', 0, '[]', '{}', 'raw'),
+                   ('00000000-0000-0000-0000-000000000102', 'CDEMO', 102, '2024-01-01T00:00:10Z', 'tx102', 0, '[]', '{}', 'raw')
+            "#,
+        )
+        .execute(&pool)
+        .await
+        .unwrap();
+
+        sqlx::query("UPDATE system_state SET value = '102' WHERE key = 'latest_ledger_cursor'")
+            .execute(&pool)
+            .await
+            .unwrap();
+
+        // Mock RPC latest ledger as 100 (indicating rollback of 101 & 102)
+        Mock::given(method("POST"))
+            .and(path("/"))
+            .and(body_partial_json(serde_json::json!({ "method": "getLatestLedger" })))
+            .respond_with(rpc_ok(serde_json::json!({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "result": { "sequence": 100 }
+            })))
+            .mount(&server)
+            .await;
+
+        // Mock getLedgers
+        Mock::given(method("POST"))
+            .and(path("/"))
+            .and(body_partial_json(serde_json::json!({ "method": "getLedgers" })))
+            .respond_with(rpc_ok(serde_json::json!({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "result": { "ledgers": [] }
+            })))
+            .mount(&server)
+            .await;
+
+        // Mock getEvents returning empty to complete poll
+        Mock::given(method("POST"))
+            .and(path("/"))
+            .and(body_partial_json(serde_json::json!({ "method": "getEvents" })))
+            .respond_with(rpc_ok(events_page(100, 0)))
+            .mount(&server)
+            .await;
+
+        let mut s = make_streamer(&db_url, &redis_url, server.uri()).await;
+        let mut cursor = 102u64;
+
+        s.poll_once(&mut cursor).await.unwrap();
+
+        // Verify cursor was rewound to 100
+        assert_eq!(cursor, 100, "cursor should be rewound to reorg tip 100");
+
+        // Verify old stale events and metadata from 101 & 102 are deleted
+        let count_events: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM soroban_events WHERE ledger_sequence > 100")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert_eq!(count_events.0, 0, "stale soroban_events from reorged ledgers must be deleted");
+
+        let count_meta: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM ledger_metadata WHERE ledger_sequence > 100")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert_eq!(count_meta.0, 0, "stale ledger_metadata from reorged ledgers must be deleted");
+
+        pool.close().await;
+    }
+
+    #[tokio::test]
+    async fn deep_reorg_exceeding_max_depth_halts_with_error() {
+        let (db_url, redis_url) = require_services!();
+        let server = MockServer::start().await;
+
+        let pool = sqlx::PgPool::connect(&db_url).await.unwrap();
+        reset_db(&pool).await;
+
+        // Mock RPC latest ledger as 100 when cursor is 500 (depth 400 > max_reorg_depth 128)
+        Mock::given(method("POST"))
+            .and(path("/"))
+            .and(body_partial_json(serde_json::json!({ "method": "getLatestLedger" })))
+            .respond_with(rpc_ok(serde_json::json!({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "result": { "sequence": 100 }
+            })))
+            .mount(&server)
+            .await;
+
+        let mut s = make_streamer(&db_url, &redis_url, server.uri()).await;
+        let mut cursor = 500u64;
+
+        let result = s.poll_once(&mut cursor).await;
+        assert!(result.is_err(), "deep reorg exceeding max_reorg_depth must return Err");
+
+        pool.close().await;
+    }
 }