From 3f493ce68673e507b403f04c83786c6d6fb1e89c Mon Sep 17 00:00:00 2001 From: Miracle656 Date: Sun, 30 Aug 2026 14:32:27 +0100 Subject: [PATCH] =?UTF-8?q?feat(indexer):=20complete=20the=20persist=20dea?= =?UTF-8?q?d-letter=20queue=20=E2=80=94=20dedup,=20separated=20metrics,=20?= =?UTF-8?q?backlog=20observability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebuilt against dev after #208 landed the queue itself (failed_events, per-event isolation, cursor advance past captured events) while this PR was in flight — the remaining #508 criteria, delivered as a small delta on that foundation: - Pending rows are unique per event (migration 0030, keyed by the same contract/ledger/index triple 0025 made canonical): a poison event re-encountered across polls updates one row with attempts folded in and the latest error kept, so the queue's pending count means distinct poisoned events. Existing duplicates are collapsed with their attempt counts preserved; replayed rows are history and never block recording a fresh failure of the same event. - The persist path counts on its own series instead of the parse-DLQ counter it shared: trident_indexer_persist_dead_lettered_total, with both counters now described accurately. - A silent DLQ is the same as data loss, so the pending depth is published as trident_indexer_persist_dead_letter_backlog (refreshed each active poll cycle and on every dead-letter write), alerted on by TridentIndexerPersistDeadLetterBacklog for as long as any pending row exists, with a runbook covering diagnosis, backfill replay and its scope limits, and the replayed_at bookkeeping that resolves the alert. promtool-validated; metrics catalog updated. - An env-gated test proves redelivery collapses to one row with folded attempts, and that a replayed row does not block a fresh failure. Closes #508 --- crates/indexer/src/db/mod.rs | 104 ++++++++++++++++++ crates/indexer/src/metrics.rs | 33 ++++++ crates/indexer/src/streamer/mod.rs | 19 +++- .../migrations/0030_failed_events_dedup.sql | 48 ++++++++ database/schema.sql | 5 + docs/metrics-catalog.md | 3 + docs/runbooks/alerts.md | 61 ++++++++++ monitoring/alerts.yml | 21 ++++ 8 files changed, 293 insertions(+), 1 deletion(-) create mode 100644 database/migrations/0030_failed_events_dedup.sql 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). # ---------------------------------------------------------------------------