Skip to main content

max / makenotwork

10.1 KB · 279 lines History Blame Raw
1 //! Webhook event retry queue; persist and retry failed webhook deliveries.
2
3 use crate::error::Result;
4 use chrono::{DateTime, Utc};
5 use sqlx::{PgPool, Postgres, Transaction};
6
7 /// A failed webhook event pending retry.
8 #[derive(Debug, sqlx::FromRow)]
9 #[allow(dead_code)]
10 pub struct DbWebhookEvent {
11 pub id: uuid::Uuid,
12 pub source: String,
13 pub event_type: String,
14 pub payload: String,
15 pub signature: Option<String>,
16 pub status: String,
17 pub attempts: i32,
18 pub last_error: Option<String>,
19 pub next_retry_at: DateTime<Utc>,
20 pub created_at: DateTime<Utc>,
21 }
22
23 /// Whether a webhook event ID has already been processed.
24 ///
25 /// This is a read used to short-circuit a redelivered event. The matching write
26 /// ([`mark_event_processed`]) happens only *after* the handler succeeds, so a
27 /// crash mid-processing can never leave a "processed" marker with no work done,
28 /// the event gets reprocessed on redelivery. (The handlers are
29 /// idempotent, so a reprocess is safe; this read just avoids the redundant
30 /// work.) The old `try_mark_event_processed` marked *before* processing and so
31 /// could strand an event if the process died in the gap before the retry row
32 /// was written, that ordering no longer exists.
33 #[tracing::instrument(skip_all)]
34 pub async fn is_event_processed(pool: &PgPool, event_id: &str) -> Result<bool> {
35 let exists = sqlx::query_scalar::<_, bool>(
36 "SELECT EXISTS(SELECT 1 FROM processed_webhook_events WHERE event_id = $1)",
37 )
38 .bind(event_id)
39 .fetch_one(pool)
40 .await?;
41
42 Ok(exists)
43 }
44
45 /// Record a webhook event ID as processed. Call this only *after* the event's
46 /// side effects are durably committed. `ON CONFLICT DO NOTHING` makes it
47 /// idempotent, so a concurrent duplicate or a redelivery is harmless.
48 #[tracing::instrument(skip_all)]
49 pub async fn mark_event_processed(pool: &PgPool, event_id: &str) -> Result<()> {
50 sqlx::query(
51 "INSERT INTO processed_webhook_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING",
52 )
53 .bind(event_id)
54 .execute(pool)
55 .await?;
56
57 Ok(())
58 }
59
60 /// Serialize concurrent redeliveries of one webhook event, without blocking.
61 ///
62 /// Returns `Some(tx)` holding a per-event `pg_advisory_xact_lock` when the lock
63 /// is free, or `None` when another delivery of the *same* event already holds it.
64 /// Hold the returned guard across the whole dedup-read -> process -> mark
65 /// sequence: while it is held, a second concurrent delivery of the same event
66 /// gets `None` here and the caller returns 503, so Stripe redelivers after the
67 /// first delivery has committed its [`mark_event_processed`] row, and the
68 /// redelivery's dedup read then short-circuits.
69 ///
70 /// This is the structural counterpart to the check-then-act dedup read. On its
71 /// own that read has a TOCTOU window: two concurrent deliveries both observe
72 /// "not processed" and both run the handler, so exactly-once rests entirely on
73 /// every handler's own idempotency. With this lock held, the read is race-free
74 /// and concurrent double-processing is impossible, a future non-idempotent
75 /// handler cannot double-fire on a redelivery race.
76 ///
77 /// Why *try* rather than block (`pg_advisory_xact_lock`): a blocking acquire
78 /// would park the pooled connection for the whole time the in-flight delivery
79 /// runs, including its outbound Stripe calls, so a redelivery storm on one hot
80 /// event could pin several connections just *waiting*. `pg_try_advisory_xact_lock`
81 /// returns immediately; the loser sheds its connection and lets Stripe's own
82 /// backoff redeliver, which is strictly cheaper than holding a conn to win a race
83 /// the dedup marker will settle anyway. Only same-event contention is affected,
84 /// distinct events hash to distinct keys and never contend.
85 ///
86 /// Robustness: the lock is transaction-scoped, so dropping the guard on any
87 /// early return, `?`, or panic rolls the (write-free) transaction back and
88 /// releases the lock. It cannot leak the way a pooled session lock would. When
89 /// the lock is *not* acquired the returned `tx` is dropped here holding nothing,
90 /// so there is no lock to leak.
91 ///
92 /// That release is prompt but not synchronous: dropping a sqlx `Transaction`
93 /// queues the ROLLBACK onto the connection rather than awaiting it, so the lock
94 /// clears when the connection is next used or returned to the pool. Nothing in
95 /// the webhook path depends on the difference (the loser has already shed its
96 /// connection and Stripe redelivers on its own backoff). Callers that do need
97 /// the lock gone before their next acquire must `rollback().await` explicitly. The key is namespaced (`stripe_webhook:` prefix)
98 /// so it shares no space with the other `hashtextextended` advisory locks in the
99 /// codebase (reports, oauth).
100 #[tracing::instrument(skip_all)]
101 pub async fn try_lock_event<'a>(
102 pool: &'a PgPool,
103 event_id: &str,
104 ) -> Result<Option<Transaction<'a, Postgres>>> {
105 let mut tx = pool.begin().await?;
106 let acquired = sqlx::query_scalar::<_, bool>(
107 "SELECT pg_try_advisory_xact_lock(hashtextextended('stripe_webhook:' || $1::text, 0))",
108 )
109 .bind(event_id)
110 .fetch_one(&mut *tx)
111 .await?;
112 Ok(acquired.then_some(tx))
113 }
114
115 /// Delete processed-event dedup markers older than `days`. These markers only
116 /// guard against Stripe *redelivering* an event, which it stops doing within a
117 /// few days; 30 days is the retention the table was created with (migration
118 /// 065) but never enforced, without this prune the table grows one row per
119 /// webhook for the life of the deployment (Run #21 Performance SERIOUS).
120 #[tracing::instrument(skip_all)]
121 pub async fn prune_processed_events(pool: &PgPool, days: i64) -> Result<u64> {
122 let result = sqlx::query(
123 "DELETE FROM processed_webhook_events \
124 WHERE processed_at < NOW() - make_interval(days => $1::int)",
125 )
126 .bind(days as i32)
127 .execute(pool)
128 .await?;
129
130 Ok(result.rows_affected())
131 }
132
133 /// Insert a failed webhook event for later retry.
134 #[tracing::instrument(skip_all)]
135 pub async fn insert_failed_event(
136 pool: &PgPool,
137 source: &str,
138 event_type: &str,
139 payload: &str,
140 signature: Option<&str>,
141 error: &str,
142 ) -> Result<()> {
143 sqlx::query(
144 r"INSERT INTO webhook_events (source, event_type, payload, signature, last_error)
145 VALUES ($1, $2, $3, $4, $5)",
146 )
147 .bind(source)
148 .bind(event_type)
149 .bind(payload)
150 .bind(signature)
151 .bind(error)
152 .execute(pool)
153 .await?;
154
155 Ok(())
156 }
157
158 /// Fetch events that are due for retry (status = failed/retrying, next_retry_at <= now).
159 /// Excludes events that have exhausted retries to prevent retry storms if the
160 /// `schedule_retry` dead-letter update fails.
161 /// Returns up to 10 at a time.
162 #[tracing::instrument(skip_all)]
163 pub async fn get_retryable_events(pool: &PgPool) -> Result<Vec<DbWebhookEvent>> {
164 // Atomically CLAIM the due events rather than plain-SELECT them: select
165 // `FOR UPDATE SKIP LOCKED` and push `next_retry_at` out so a second scheduler
166 // replica (or an overlapping tick) can't grab the same rows and double-run the
167 // handler (audit Run 13 Conc). A claimed event whose process crashes before
168 // resolution becomes eligible again after the claim window, handlers are
169 // idempotent, so at-least-once is safe. Post-processing (`mark_processed` /
170 // `schedule_retry`) rewrites `status`/`next_retry_at` for the terminal state.
171 let events = sqlx::query_as::<_, DbWebhookEvent>(
172 r"
173 UPDATE webhook_events
174 SET next_retry_at = NOW() + INTERVAL '2 minutes'
175 WHERE id IN (
176 SELECT id FROM webhook_events
177 WHERE status IN ('failed', 'retrying')
178 AND attempts < 5
179 AND next_retry_at <= NOW()
180 ORDER BY next_retry_at
181 LIMIT 10
182 FOR UPDATE SKIP LOCKED
183 )
184 RETURNING *
185 ",
186 )
187 .fetch_all(pool)
188 .await?;
189
190 Ok(events)
191 }
192
193 /// Mark an event as successfully processed and delete it.
194 #[tracing::instrument(skip_all)]
195 pub async fn mark_processed(pool: &PgPool, id: uuid::Uuid) -> Result<()> {
196 sqlx::query("DELETE FROM webhook_events WHERE id = $1")
197 .bind(id)
198 .execute(pool)
199 .await?;
200
201 Ok(())
202 }
203
204 /// Increment attempts and schedule next retry with exponential backoff.
205 /// Backoff: 1m, 5m, 30m, 2h, 24h. After 5 attempts, mark as dead.
206 #[tracing::instrument(skip_all)]
207 pub async fn schedule_retry(
208 pool: &PgPool,
209 id: uuid::Uuid,
210 attempt: i32,
211 error: &str,
212 ) -> Result<()> {
213 let max_attempts = 5;
214
215 if attempt >= max_attempts {
216 sqlx::query(
217 "UPDATE webhook_events SET status = 'dead', attempts = $2, last_error = $3 WHERE id = $1",
218 )
219 .bind(id)
220 .bind(attempt)
221 .bind(error)
222 .execute(pool)
223 .await?;
224 } else {
225 // Exponential backoff: 60s, 300s, 1800s, 7200s, 86400s
226 let delay_secs: i64 = match attempt {
227 0 => 60,
228 1 => 300,
229 2 => 1800,
230 3 => 7200,
231 _ => 86400,
232 };
233
234 sqlx::query(
235 r"UPDATE webhook_events
236 SET status = 'retrying',
237 attempts = $2,
238 last_error = $3,
239 next_retry_at = NOW() + make_interval(secs => $4::double precision)
240 WHERE id = $1",
241 )
242 .bind(id)
243 .bind(attempt)
244 .bind(error)
245 .bind(delay_secs as f64)
246 .execute(pool)
247 .await?;
248 }
249
250 Ok(())
251 }
252
253 /// Get dead events for admin review.
254 #[allow(dead_code)]
255 #[tracing::instrument(skip_all)]
256 pub async fn get_dead_events(pool: &PgPool) -> Result<Vec<DbWebhookEvent>> {
257 let events = sqlx::query_as::<_, DbWebhookEvent>(
258 "SELECT * FROM webhook_events WHERE status = 'dead' ORDER BY created_at DESC LIMIT 50",
259 )
260 .fetch_all(pool)
261 .await?;
262
263 Ok(events)
264 }
265
266 /// Reset a dead event for retry.
267 #[allow(dead_code)]
268 #[tracing::instrument(skip_all)]
269 pub async fn retry_dead_event(pool: &PgPool, id: uuid::Uuid) -> Result<bool> {
270 let result = sqlx::query(
271 "UPDATE webhook_events SET status = 'failed', next_retry_at = NOW() WHERE id = $1 AND status = 'dead'",
272 )
273 .bind(id)
274 .execute(pool)
275 .await?;
276
277 Ok(result.rows_affected() > 0)
278 }
279