feat(indexer): dead-letter queue for events that repeatedly fail to persist

crates/indexer/src/db/mod.rs

@@ -950,6 +950,16 @@ pub async fn insert_parse_error(
 /// JSONB so it can be inspected and replayed once the underlying cause (a
 /// constraint violation, an outage that outlasted the retry budget, etc.) is
 /// understood, without needing to re-fetch it from Stellar RPC.
+///
+/// Keyed over PENDING rows by the event's natural key — the same
+/// (contract_id, ledger_sequence, event_index) triple `event_uuid` and
+/// migration 0025 make canonical (issue #508): a poison
+/// event re-encountered across polls — an RPC redelivery, a backfill overlap
+/// — updates its existing row (attempt count folded in, latest error kept)
+/// instead of accumulating duplicates, so the pending row count equals the
+/// number of distinct poisoned events, which is exactly what the backlog
+/// gauge and its alert report. A row an operator already replayed is history
+/// and does not block recording a fresh failure of the same event.
 pub async fn insert_failed_event(
     pool: &PgPool,
     event: &SorobanEvent,
@@ -966,6 +976,10 @@ pub async fn insert_failed_event(
             (ledger_sequence, contract_id, transaction_hash, event_index,
              event_payload, error_message, attempts)
         VALUES ($1, $2, $3, $4, $5, $6, $7)
+        ON CONFLICT (contract_id, ledger_sequence, event_index) WHERE replayed_at IS NULL
+        DO UPDATE SET
+            attempts      = failed_events.attempts + EXCLUDED.attempts,
+            error_message = EXCLUDED.error_message
         "#,
     )
     .bind(event.ledger_sequence as i64)
@@ -982,6 +996,23 @@ pub async fn insert_failed_event(
     Ok(())
 }
 
+/// Number of dead-lettered events still awaiting replay. Published as the
+/// `trident_indexer_persist_dead_letter_backlog` gauge each active poll
+/// cycle (and on every dead-letter write), so the non-empty-DLQ alert has a
+/// live series to fire on — a silent queue is indistinguishable from data
+/// loss (issue #508). Counts only pending rows: a replayed row is resolved
+/// history, not backlog.
+pub async fn count_pending_failed_events(pool: &PgPool) -> Result<i64, TridentError> {
+    let row: (i64,) =
+        sqlx::query_as("SELECT COUNT(*) FROM failed_events WHERE replayed_at IS NULL")
+            .fetch_one(pool)
+            .await
+            .map_err(|e| {
+                TridentError::storage(anyhow::Error::new(e).context("count_pending_failed_events"))
+            })?;
+    Ok(row.0)
+}
+
 /// Contracts among `contract_ids` whose `token_metadata` row is still fresh
 /// (resolved or refreshed since `cutoff`), for either a positive or a cached
 /// negative ("not a token") result (issue #263). Contracts absent from this
@@ -1091,6 +1122,79 @@ mod tests {
         assert_ne!(a, b);
     }
 
+    /// Dead-lettering the same event twice must collapse to ONE pending row
+    /// with the attempt counts folded together (issue #508) — the pending
+    /// count is what the backlog gauge reports, and it must mean "distinct
+    /// poisoned events", not "retry bursts".
+    ///
+    /// Uses the shared test database (TEST_DATABASE_URL) like the other
+    /// integration tests; skips when it is not configured.
+    #[tokio::test]
+    async fn failed_event_redelivery_updates_one_pending_row() {
+        let db_url = match std::env::var("TEST_DATABASE_URL") {
+            Ok(url) => url,
+            Err(_) if std::env::var("REQUIRE_TEST_SERVICES").is_ok() => {
+                panic!("TEST_DATABASE_URL must be set when REQUIRE_TEST_SERVICES is set");
+            }
+            Err(_) => {
+                eprintln!("SKIP: TEST_DATABASE_URL not set");
+                return;
+            }
+        };
+        let pool = PgPool::connect(&db_url).await.unwrap();
+
+        // A unique contract per run, like the other failed_events test: the
+        // shared fixture reuses one transaction hash everywhere, so scoping
+        // by contract keeps parallel tests out of each other's rows.
+        let contract_id = format!("CDLQ_{}", Uuid::new_v4());
+        let event = make_event(&contract_id, 43, 0);
+        sqlx::query("DELETE FROM failed_events WHERE contract_id = $1")
+            .bind(&contract_id)
+            .execute(&pool)
+            .await
+            .expect("cleanup failed");
+
+        insert_failed_event(&pool, &event, "first failure", 4)
+            .await
+            .expect("first dead-letter failed");
+        insert_failed_event(&pool, &event, "second failure", 4)
+            .await
+            .expect("redelivered dead-letter must not error");
+
+        let (rows, attempts, error): (i64, i32, String) = sqlx::query_as(
+            "SELECT COUNT(*) OVER (), attempts, error_message FROM failed_events              WHERE contract_id = $1 AND replayed_at IS NULL",
+        )
+        .bind(&contract_id)
+        .fetch_one(&pool)
+        .await
+        .expect("row query failed");
+        assert_eq!(rows, 1, "redelivery must update, not duplicate");
+        assert_eq!(attempts, 8, "attempt counts fold together");
+        assert_eq!(error, "second failure", "latest error wins");
+
+        // A REPLAYED row is history: the same event failing again gets a
+        // fresh pending row rather than resurrecting the resolved one.
+        sqlx::query("UPDATE failed_events SET replayed_at = NOW() WHERE transaction_hash = $1")
+            .bind(&event.transaction_hash)
+            .execute(&pool)
+            .await
+            .expect("mark replayed");
+        insert_failed_event(&pool, &event, "regression after replay", 1)
+            .await
+            .expect("post-replay dead-letter failed");
+        let pending: (i64,) = sqlx::query_as(
+            "SELECT COUNT(*) FROM failed_events              WHERE contract_id = $1 AND replayed_at IS NULL",
+        )
+        .bind(&contract_id)
+        .fetch_one(&pool)
+        .await
+        .expect("pending count failed");
+        assert_eq!(
+            pending.0, 1,
+            "a replayed row must not block a fresh failure"
+        );
+    }
+
     /// Committing the same event twice must not error and the row count in
     /// `soroban_events` must remain 1.
     ///

crates/indexer/src/metrics.rs

@@ -33,6 +33,16 @@ pub const PARSE_ERRORS_TOTAL: &str = "trident_indexer_parse_errors_total";
 /// including ones that later succeed on retry: this counter only moves when an
 /// event is actually abandoned, which is what an alert should fire on.
 pub const DEAD_LETTERED_TOTAL: &str = "trident_indexer_dead_lettered_total";
+/// Deliberately separate from DEAD_LETTERED_TOTAL above: that one counts
+/// undecodable events captured in `parse_errors` (a poison message — retry
+/// never helps), while this counts well-formed events whose database commit
+/// failed after the retry budget and landed in `failed_events` for replay
+/// (issue #508). Conflating them made one number answer two different
+/// operational questions.
+pub const PERSIST_DEAD_LETTERED_TOTAL: &str = "trident_indexer_persist_dead_lettered_total";
+/// Current number of `failed_events` rows awaiting replay. Non-empty pages
+/// via TridentIndexerPersistDeadLetterBacklog (monitoring/alerts.yml).
+pub const PERSIST_DEAD_LETTER_BACKLOG: &str = "trident_indexer_persist_dead_letter_backlog";
 pub const POLL_DURATION_SECONDS: &str = "trident_indexer_poll_duration_seconds";
 pub const POLL_ERRORS_TOTAL: &str = "trident_indexer_poll_errors_total";
 pub const RPC_RETRIES_TOTAL: &str = "trident_indexer_rpc_retries_total";
@@ -120,6 +130,18 @@ pub fn install(port: u16) -> Result<(), TridentError> {
         "Events skipped (diagnostic, failed call, or contract filter)"
     );
     describe_counter!(PARSE_ERRORS_TOTAL, "Total events that failed XDR decoding");
+    describe_counter!(
+        DEAD_LETTERED_TOTAL,
+        "Undecodable events durably captured in parse_errors (issue #414)"
+    );
+    describe_counter!(
+        PERSIST_DEAD_LETTERED_TOTAL,
+        "Well-formed events captured in failed_events after exhausting the persist retry budget (issue #508)"
+    );
+    describe_gauge!(
+        PERSIST_DEAD_LETTER_BACKLOG,
+        "failed_events rows awaiting replay; non-empty pages via TridentIndexerPersistDeadLetterBacklog (issue #508)"
+    );
     describe_histogram!(
         POLL_DURATION_SECONDS,
         "Time per poll_once cycle, in seconds"
@@ -205,6 +227,9 @@ 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!(PERSIST_DEAD_LETTERED_TOTAL).increment(0);
+    gauge!(PERSIST_DEAD_LETTER_BACKLOG).set(0.0);
+    counter!(DEAD_LETTERED_TOTAL).increment(0);
     gauge!(RPC_ACTIVE_ENDPOINT).set(0.0);
     gauge!(OUTBOX_BACKLOG).set(0.0);
     gauge!(LEDGER_LAG).set(0.0);
@@ -325,6 +350,14 @@ pub fn record_parse_error() {
     counter!(PARSE_ERRORS_TOTAL).increment(1);
 }
 
+pub fn record_persist_dead_lettered() {
+    counter!(PERSIST_DEAD_LETTERED_TOTAL).increment(1);
+}
+
+pub fn set_persist_dead_letter_backlog(depth: i64) {
+    gauge!(PERSIST_DEAD_LETTER_BACKLOG).set(depth as f64);
+}
+
 pub fn record_dead_lettered() {
     counter!(DEAD_LETTERED_TOTAL).increment(1);
 }

crates/indexer/src/streamer/mod.rs

@@ -745,6 +745,14 @@ impl Streamer {
             metrics::record_events_processed(events_in_page as u64);
             metrics::record_events_skipped(skipped_in_page);
 
+            // Refresh the dead-letter backlog gauge every active cycle so an
+            // operator replay (which shrinks the queue out-of-band) is
+            // reflected without restarting the indexer. Best-effort: the
+            // gauge is observability, not control flow.
+            if let Ok(depth) = db::count_pending_failed_events(&self.db).await {
+                metrics::set_persist_dead_letter_backlog(depth);
+            }
+
             // Decide whether this page advances the cursor, and gather the
             // ledger provenance that must land in the same transaction.
             let mut next_cursor: Option<u64> = None;
@@ -1252,7 +1260,16 @@ async fn commit_page_with_fallback(
                 "Event failed to persist after per-event retries; dead-lettering"
             );
             match db::insert_failed_event(db, event, &e.to_string(), attempts).await {
-                Ok(()) => metrics::record_dead_lettered(),
+                Ok(()) => {
+                    // The PERSIST counter, not the parse one: these answer
+                    // different operational questions (issue #508), and the
+                    // backlog gauge refreshes immediately so the alert sees
+                    // the new row without waiting for the next active cycle.
+                    metrics::record_persist_dead_lettered();
+                    if let Ok(depth) = db::count_pending_failed_events(db).await {
+                        metrics::set_persist_dead_letter_backlog(depth);
+                    }
+                }
                 Err(dl_err) => {
                     // The dead-letter write goes to the same database the
                     // event's own INSERT just failed against. Failing here is

database/migrations/0030_failed_events_dedup.sql

@@ -0,0 +1,48 @@
+-- failed_events dedup + pending-uniqueness (issue #508, completing #208).
+--
+-- 0028 introduced the persist dead-letter queue but every exhausted retry
+-- burst INSERTs a fresh row, so one poison event re-encountered across polls
+-- (an RPC redelivery, a backfill overlap) accumulates duplicates and the
+-- queue's row count stops meaning "number of distinct poisoned events" —
+-- which is exactly the number the backlog gauge and its alert report.
+--
+-- A row's natural key is (contract_id, ledger_sequence, event_index) — the
+-- same triple 0025 established as soroban_events' natural key and
+-- event_uuid derives the deterministic id from. Uniqueness is
+-- enforced only over PENDING rows (replayed_at IS NULL): a row an operator
+-- has already replayed is history and must not block recording a fresh
+-- failure of the same event if it ever fails again.
+
+-- Collapse existing pending duplicates before the index can exist: keep the
+-- newest row per key and fold the attempt counts into it, so no evidence of
+-- how often the event failed is lost.
+WITH ranked AS (
+    SELECT id,
+           SUM(attempts) OVER (PARTITION BY contract_id, ledger_sequence, event_index) AS total_attempts,
+           ROW_NUMBER() OVER (
+               PARTITION BY contract_id, ledger_sequence, event_index
+               ORDER BY occurred_at DESC, id
+           ) AS rn
+    FROM failed_events
+    WHERE replayed_at IS NULL
+)
+UPDATE failed_events f
+SET attempts = ranked.total_attempts
+FROM ranked
+WHERE f.id = ranked.id AND ranked.rn = 1;
+
+DELETE FROM failed_events f
+USING (
+    SELECT id,
+           ROW_NUMBER() OVER (
+               PARTITION BY contract_id, ledger_sequence, event_index
+               ORDER BY occurred_at DESC, id
+           ) AS rn
+    FROM failed_events
+    WHERE replayed_at IS NULL
+) ranked
+WHERE f.id = ranked.id AND ranked.rn > 1;
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_failed_events_pending
+    ON failed_events (contract_id, ledger_sequence, event_index)
+    WHERE replayed_at IS NULL;

database/schema.sql

@@ -558,3 +558,8 @@ CREATE TABLE IF NOT EXISTS failed_events (
 
 CREATE INDEX IF NOT EXISTS idx_failed_events_occurred_at ON failed_events (occurred_at DESC);
 CREATE INDEX IF NOT EXISTS idx_failed_events_pending ON failed_events (occurred_at) WHERE replayed_at IS NULL;
+-- Pending rows are unique per event (migration 0030): redelivered failures
+-- update in place, so COUNT(*) of pending rows = distinct poisoned events.
+CREATE UNIQUE INDEX IF NOT EXISTS uq_failed_events_pending
+    ON failed_events (contract_id, ledger_sequence, event_index)
+    WHERE replayed_at IS NULL;

docs/metrics-catalog.md

@@ -18,6 +18,9 @@ port `9090`, set via `METRICS_PORT`). Defined in
 | `trident_indexer_events_total` | counter | — | events | Cumulative events indexed since process start. |
 | `trident_indexer_events_skipped_total` | counter | — | events | Events skipped: diagnostic/failed-call events, or filtered by the contract allowlist. |
 | `trident_indexer_parse_errors_total` | counter | — | events | Events that failed XDR decoding and were written to `parse_errors` instead of `soroban_events`. |
+| `trident_indexer_dead_lettered_total` | counter | — | events | Undecodable events durably written to `parse_errors` after the dead-letter insert's own retries (#414). |
+| `trident_indexer_persist_dead_lettered_total` | counter | — | events | Events that decoded fine but exhausted the persist retry budget and were durably captured in `failed_events` (#508). |
+| `trident_indexer_persist_dead_letter_backlog` | gauge | — | rows | `failed_events` rows awaiting replay (`replayed_at IS NULL`), refreshed each active poll cycle and on every dead-letter write; non-empty pages via `TridentIndexerPersistDeadLetterBacklog` (#508). |
 | `trident_indexer_poll_duration_seconds` | histogram | — | seconds | Wall-clock time of one `poll_once` cycle (may span multiple RPC pages). |
 | `trident_indexer_poll_errors_total` | counter | — | cycles | Poll cycles that returned an error (logged, cursor unaffected, retried next interval). |
 | `trident_indexer_rpc_retries_total` | counter | — | retries | Retries triggered by transient `getEvents` failures (exponential backoff). |

docs/runbooks/alerts.md

@@ -686,3 +686,64 @@ those, so it fires on the level rather than the slope.
    gone: `SELECT pg_drop_replication_slot('<name>');`
 3. If it is ordinary growth, treat it as
    `TridentDiskFillingWithin48Hours` above.
+
+## TridentIndexerPersistDeadLetterBacklog
+
+**Means:** `failed_events` has pending rows — at least one event decoded
+fine but its database commit kept failing through the whole-page retry, the
+per-event isolation retry, and its backoff budget, so the streamer captured
+the failing event (full payload + error message), counted it on
+`trident_indexer_persist_dead_lettered_total`, and advanced the cursor past
+it (issues #208/#508). The data is safe but missing from `soroban_events`
+until replayed.
+
+**Why this threshold:** any pending row at all means indexed data is
+incomplete, and rows leave the pending state only through the replay
+procedure below — so the alert stays up until the gap is actually closed. A
+silent DLQ is the same as data loss. `for: 5m` only absorbs scrape jitter.
+Pending rows are unique per event (migration 0030): the gauge counts
+distinct poisoned events, not retry bursts.
+
+**First steps:**
+
+1. Inspect the queue:
+
+   ```sql
+   SELECT contract_id, ledger_sequence, error_message, attempts, occurred_at
+   FROM failed_events WHERE replayed_at IS NULL ORDER BY occurred_at;
+   ```
+
+   `error_message` names the exact commit error, updated to the most recent
+   failure; `attempts` accumulates across redeliveries.
+2. Fix the underlying cause (a malformed field failing column conversion, a
+   constraint interaction, …). The failure survived the whole retry budget,
+   so replaying before the fix will just fail again.
+3. **Replay** once the fix is deployed: re-ingest the affected range with
+   the backfill CLI (idempotent — `ON CONFLICT DO NOTHING` absorbs the
+   events that did commit). Note its scope: backfill restores rows in
+   `soroban_events` only — it does not write outbox rows or token
+   projections, so replayed events are not delivered to Redis/webhook
+   subscribers and do not appear in `token_events`. Acceptable for
+   historical repair, but know what you are and are not restoring:
+
+   ```
+   trident-backfill --from-ledger <min> --to-ledger <max> [--contract <id>]
+   ```
+
+   using `SELECT MIN(ledger_sequence), MAX(ledger_sequence) FROM
+   failed_events WHERE replayed_at IS NULL;` for the range.
+4. Verify the events landed, then mark the rows replayed — this is what
+   resolves the alert (rows are kept as history, never deleted):
+
+   ```sql
+   UPDATE failed_events d
+   SET replayed_at = NOW()
+   FROM soroban_events e
+   WHERE d.replayed_at IS NULL
+     AND e.ledger_sequence = d.ledger_sequence
+     AND e.transaction_hash = d.transaction_hash
+     AND e.event_index = d.event_index;
+   ```
+
+   The gauge refreshes on the next active poll cycle; the alert clears once
+   no pending rows remain.

monitoring/alerts.yml

@@ -126,6 +126,27 @@ groups:
             recognise yet.
           runbook_url: "docs/runbooks/alerts.md#tridentindexerparseerrorratehigh"
 
+      # A non-empty persist dead-letter queue is data missing from
+      # soroban_events until someone replays it — a silently growing DLQ is
+      # the same as data loss. Rows leave the pending state only through the
+      # documented replay procedure, so this fires for as long as any exist.
+      - alert: TridentIndexerPersistDeadLetterBacklog
+        expr: trident_indexer_persist_dead_letter_backlog > 0
+        for: 5m
+        labels:
+          severity: warning
+          service: indexer
+        annotations:
+          summary: "Trident dead-letter queue holds {{ $value }} event(s) that failed to persist"
+          description: >
+            failed_events has pending rows: at least one event exhausted the
+            persist retry budget and was captured with its payload and error
+            message instead of being indexed. Ingestion has advanced past it,
+            but the data is missing from soroban_events until it is replayed.
+            Diagnose error_message, fix the underlying cause, then run the
+            replay procedure in the runbook and mark the rows replayed_at.
+          runbook_url: "docs/runbooks/alerts.md#tridentindexerpersistdeadletterbacklog"
+
   # ---------------------------------------------------------------------------
   # Stellar RPC health (#297).
   # ---------------------------------------------------------------------------