//! Webhook event retry queue; persist and retry failed webhook deliveries. use crate::error::Result; use chrono::{DateTime, Utc}; use sqlx::{PgPool, Postgres, Transaction}; /// A failed webhook event pending retry. #[derive(Debug, sqlx::FromRow)] #[allow(dead_code)] pub struct DbWebhookEvent { pub id: uuid::Uuid, pub source: String, pub event_type: String, pub payload: String, pub signature: Option, pub status: String, pub attempts: i32, pub last_error: Option, pub next_retry_at: DateTime, pub created_at: DateTime, } /// Whether a webhook event ID has already been processed. /// /// This is a read used to short-circuit a redelivered event. The matching write /// ([`mark_event_processed`]) happens only *after* the handler succeeds, so a /// crash mid-processing can never leave a "processed" marker with no work done, /// the event gets reprocessed on redelivery. (The handlers are /// idempotent, so a reprocess is safe; this read just avoids the redundant /// work.) The old `try_mark_event_processed` marked *before* processing and so /// could strand an event if the process died in the gap before the retry row /// was written, that ordering no longer exists. #[tracing::instrument(skip_all)] pub async fn is_event_processed(pool: &PgPool, event_id: &str) -> Result { let exists = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM processed_webhook_events WHERE event_id = $1)", ) .bind(event_id) .fetch_one(pool) .await?; Ok(exists) } /// Record a webhook event ID as processed. Call this only *after* the event's /// side effects are durably committed. `ON CONFLICT DO NOTHING` makes it /// idempotent, so a concurrent duplicate or a redelivery is harmless. #[tracing::instrument(skip_all)] pub async fn mark_event_processed(pool: &PgPool, event_id: &str) -> Result<()> { sqlx::query( "INSERT INTO processed_webhook_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING", ) .bind(event_id) .execute(pool) .await?; Ok(()) } /// Serialize concurrent redeliveries of one webhook event, without blocking. /// /// Returns `Some(tx)` holding a per-event `pg_advisory_xact_lock` when the lock /// is free, or `None` when another delivery of the *same* event already holds it. /// Hold the returned guard across the whole dedup-read -> process -> mark /// sequence: while it is held, a second concurrent delivery of the same event /// gets `None` here and the caller returns 503, so Stripe redelivers after the /// first delivery has committed its [`mark_event_processed`] row, and the /// redelivery's dedup read then short-circuits. /// /// This is the structural counterpart to the check-then-act dedup read. On its /// own that read has a TOCTOU window: two concurrent deliveries both observe /// "not processed" and both run the handler, so exactly-once rests entirely on /// every handler's own idempotency. With this lock held, the read is race-free /// and concurrent double-processing is impossible, a future non-idempotent /// handler cannot double-fire on a redelivery race. /// /// Why *try* rather than block (`pg_advisory_xact_lock`): a blocking acquire /// would park the pooled connection for the whole time the in-flight delivery /// runs, including its outbound Stripe calls, so a redelivery storm on one hot /// event could pin several connections just *waiting*. `pg_try_advisory_xact_lock` /// returns immediately; the loser sheds its connection and lets Stripe's own /// backoff redeliver, which is strictly cheaper than holding a conn to win a race /// the dedup marker will settle anyway. Only same-event contention is affected, /// distinct events hash to distinct keys and never contend. /// /// Robustness: the lock is transaction-scoped, so dropping the guard on any /// early return, `?`, or panic rolls the (write-free) transaction back and /// releases the lock. It cannot leak the way a pooled session lock would. When /// the lock is *not* acquired the returned `tx` is dropped here holding nothing, /// so there is no lock to leak. /// /// That release is prompt but not synchronous: dropping a sqlx `Transaction` /// queues the ROLLBACK onto the connection rather than awaiting it, so the lock /// clears when the connection is next used or returned to the pool. Nothing in /// the webhook path depends on the difference (the loser has already shed its /// connection and Stripe redelivers on its own backoff). Callers that do need /// the lock gone before their next acquire must `rollback().await` explicitly. The key is namespaced (`stripe_webhook:` prefix) /// so it shares no space with the other `hashtextextended` advisory locks in the /// codebase (reports, oauth). #[tracing::instrument(skip_all)] pub async fn try_lock_event<'a>( pool: &'a PgPool, event_id: &str, ) -> Result>> { let mut tx = pool.begin().await?; let acquired = sqlx::query_scalar::<_, bool>( "SELECT pg_try_advisory_xact_lock(hashtextextended('stripe_webhook:' || $1::text, 0))", ) .bind(event_id) .fetch_one(&mut *tx) .await?; Ok(acquired.then_some(tx)) } /// Delete processed-event dedup markers older than `days`. These markers only /// guard against Stripe *redelivering* an event, which it stops doing within a /// few days; 30 days is the retention the table was created with (migration /// 065) but never enforced, without this prune the table grows one row per /// webhook for the life of the deployment (Run #21 Performance SERIOUS). #[tracing::instrument(skip_all)] pub async fn prune_processed_events(pool: &PgPool, days: i64) -> Result { let result = sqlx::query( "DELETE FROM processed_webhook_events \ WHERE processed_at < NOW() - make_interval(days => $1::int)", ) .bind(days as i32) .execute(pool) .await?; Ok(result.rows_affected()) } /// Insert a failed webhook event for later retry. #[tracing::instrument(skip_all)] pub async fn insert_failed_event( pool: &PgPool, source: &str, event_type: &str, payload: &str, signature: Option<&str>, error: &str, ) -> Result<()> { sqlx::query( r"INSERT INTO webhook_events (source, event_type, payload, signature, last_error) VALUES ($1, $2, $3, $4, $5)", ) .bind(source) .bind(event_type) .bind(payload) .bind(signature) .bind(error) .execute(pool) .await?; Ok(()) } /// Fetch events that are due for retry (status = failed/retrying, next_retry_at <= now). /// Excludes events that have exhausted retries to prevent retry storms if the /// `schedule_retry` dead-letter update fails. /// Returns up to 10 at a time. #[tracing::instrument(skip_all)] pub async fn get_retryable_events(pool: &PgPool) -> Result> { // Atomically CLAIM the due events rather than plain-SELECT them: select // `FOR UPDATE SKIP LOCKED` and push `next_retry_at` out so a second scheduler // replica (or an overlapping tick) can't grab the same rows and double-run the // handler (audit Run 13 Conc). A claimed event whose process crashes before // resolution becomes eligible again after the claim window, handlers are // idempotent, so at-least-once is safe. Post-processing (`mark_processed` / // `schedule_retry`) rewrites `status`/`next_retry_at` for the terminal state. let events = sqlx::query_as::<_, DbWebhookEvent>( r" UPDATE webhook_events SET next_retry_at = NOW() + INTERVAL '2 minutes' WHERE id IN ( SELECT id FROM webhook_events WHERE status IN ('failed', 'retrying') AND attempts < 5 AND next_retry_at <= NOW() ORDER BY next_retry_at LIMIT 10 FOR UPDATE SKIP LOCKED ) RETURNING * ", ) .fetch_all(pool) .await?; Ok(events) } /// Mark an event as successfully processed and delete it. #[tracing::instrument(skip_all)] pub async fn mark_processed(pool: &PgPool, id: uuid::Uuid) -> Result<()> { sqlx::query("DELETE FROM webhook_events WHERE id = $1") .bind(id) .execute(pool) .await?; Ok(()) } /// Increment attempts and schedule next retry with exponential backoff. /// Backoff: 1m, 5m, 30m, 2h, 24h. After 5 attempts, mark as dead. #[tracing::instrument(skip_all)] pub async fn schedule_retry( pool: &PgPool, id: uuid::Uuid, attempt: i32, error: &str, ) -> Result<()> { let max_attempts = 5; if attempt >= max_attempts { sqlx::query( "UPDATE webhook_events SET status = 'dead', attempts = $2, last_error = $3 WHERE id = $1", ) .bind(id) .bind(attempt) .bind(error) .execute(pool) .await?; } else { // Exponential backoff: 60s, 300s, 1800s, 7200s, 86400s let delay_secs: i64 = match attempt { 0 => 60, 1 => 300, 2 => 1800, 3 => 7200, _ => 86400, }; sqlx::query( r"UPDATE webhook_events SET status = 'retrying', attempts = $2, last_error = $3, next_retry_at = NOW() + make_interval(secs => $4::double precision) WHERE id = $1", ) .bind(id) .bind(attempt) .bind(error) .bind(delay_secs as f64) .execute(pool) .await?; } Ok(()) } /// Get dead events for admin review. #[allow(dead_code)] #[tracing::instrument(skip_all)] pub async fn get_dead_events(pool: &PgPool) -> Result> { let events = sqlx::query_as::<_, DbWebhookEvent>( "SELECT * FROM webhook_events WHERE status = 'dead' ORDER BY created_at DESC LIMIT 50", ) .fetch_all(pool) .await?; Ok(events) } /// Reset a dead event for retry. #[allow(dead_code)] #[tracing::instrument(skip_all)] pub async fn retry_dead_event(pool: &PgPool, id: uuid::Uuid) -> Result { let result = sqlx::query( "UPDATE webhook_events SET status = 'failed', next_retry_at = NOW() WHERE id = $1 AND status = 'dead'", ) .bind(id) .execute(pool) .await?; Ok(result.rows_affected() > 0) }