diff --git a/crates/indexer/src/db/mod.rs b/crates/indexer/src/db/mod.rs index 9443d3c6..e45af73b 100644 --- a/crates/indexer/src/db/mod.rs +++ b/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 { + 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. /// diff --git a/crates/indexer/src/metrics.rs b/crates/indexer/src/metrics.rs index 3c64052c..43b67201 100644 --- a/crates/indexer/src/metrics.rs +++ b/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); } diff --git a/crates/indexer/src/streamer/mod.rs b/crates/indexer/src/streamer/mod.rs index 469263cf..62d23757 100644 --- a/crates/indexer/src/streamer/mod.rs +++ b/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 = 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 diff --git a/database/migrations/0030_failed_events_dedup.sql b/database/migrations/0030_failed_events_dedup.sql new file mode 100644 index 00000000..f02ac729 --- /dev/null +++ b/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; diff --git a/database/schema.sql b/database/schema.sql index c3092af0..c093a588 100644 --- a/database/schema.sql +++ b/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; diff --git a/docs/metrics-catalog.md b/docs/metrics-catalog.md index ed97c6d5..2aaeeb05 100644 --- a/docs/metrics-catalog.md +++ b/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). | diff --git a/docs/runbooks/alerts.md b/docs/runbooks/alerts.md index 507f5946..abbb63aa 100644 --- a/docs/runbooks/alerts.md +++ b/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('');` 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 --to-ledger [--contract ] + ``` + + 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. diff --git a/monitoring/alerts.yml b/monitoring/alerts.yml index a5b38fab..5f40e302 100644 --- a/monitoring/alerts.yml +++ b/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). # ---------------------------------------------------------------------------