Skip to main content

max / makenotwork

14.6 KB · 430 lines History Blame Raw
1 //! DB-layer contract tests for `db::webhook_events`, the webhook dedup +
2 //! retry-queue module.
3 //!
4 //! Audit Run 16 flagged `webhook_events` as a Testing/Concurrency cold spot
5 //! (type B+): the dedup marker and the retry-claim query are exercised only
6 //! indirectly through the Stripe webhook flow, with "no two-concurrent-
7 //! deliveries test." These call the `db::webhook_events` functions directly so
8 //! the two race-sensitive invariants live where they belong:
9 //!
10 //! 1. `mark_event_processed` is idempotent under concurrent duplicate delivery
11 //! (Stripe at-least-once), N racing marks of one event leave exactly one row.
12 //! 2. `get_retryable_events` claims-and-defers so an overlapping scheduler tick
13 //! (or a second replica) can't re-claim the same due events within the window.
14 //!
15 //! Plus the backoff/dead-letter progression and the retention prune.
16
17 use crate::harness::db::TestDb;
18 use makenotwork::db::webhook_events;
19
20 /// Count rows in `processed_webhook_events` (dedup markers).
21 async fn processed_count(pool: &sqlx::PgPool) -> i64 {
22 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM processed_webhook_events")
23 .fetch_one(pool)
24 .await
25 .expect("count processed")
26 }
27
28 /// Insert a failed event and return its id, backdating `next_retry_at` into the
29 /// past so it is immediately due (the column defaults to NOW()+60s).
30 async fn seed_due_event(pool: &sqlx::PgPool, event_type: &str) -> uuid::Uuid {
31 webhook_events::insert_failed_event(pool, "stripe", event_type, "{}", None, "boom")
32 .await
33 .expect("insert failed event");
34 sqlx::query_scalar::<_, uuid::Uuid>(
35 "UPDATE webhook_events SET next_retry_at = NOW() - INTERVAL '1 minute' \
36 WHERE event_type = $1 RETURNING id",
37 )
38 .bind(event_type)
39 .fetch_one(pool)
40 .await
41 .expect("backdate next_retry_at")
42 }
43
44 // ── dedup marker (`processed_webhook_events`) ────────────────────────────────
45
46 #[tokio::test]
47 async fn is_event_processed_false_before_mark_true_after() {
48 let db = TestDb::new().await;
49 let id = "evt_dedup_basic";
50
51 assert!(
52 !webhook_events::is_event_processed(&db.pool, id)
53 .await
54 .unwrap(),
55 "an unseen event must not read as processed"
56 );
57
58 webhook_events::mark_event_processed(&db.pool, id)
59 .await
60 .unwrap();
61
62 assert!(
63 webhook_events::is_event_processed(&db.pool, id)
64 .await
65 .unwrap(),
66 "a marked event must read as processed"
67 );
68 }
69
70 #[tokio::test]
71 async fn mark_event_processed_is_idempotent() {
72 let db = TestDb::new().await;
73 let id = "evt_dedup_idem";
74
75 // Serial re-marks (redelivery) must not error and must not duplicate the row.
76 webhook_events::mark_event_processed(&db.pool, id)
77 .await
78 .unwrap();
79 webhook_events::mark_event_processed(&db.pool, id)
80 .await
81 .unwrap();
82 webhook_events::mark_event_processed(&db.pool, id)
83 .await
84 .unwrap();
85
86 assert_eq!(
87 processed_count(&db.pool).await,
88 1,
89 "re-marking must be a no-op"
90 );
91 }
92
93 /// The core concurrency invariant: Stripe delivers at-least-once, so two
94 /// deliveries of the same event can race the post-handler mark. `ON CONFLICT DO
95 /// NOTHING` must collapse them to exactly one dedup row, never a duplicate-key
96 /// error, never two rows.
97 #[tokio::test]
98 async fn mark_event_processed_concurrent_duplicates_leave_one_row() {
99 let db = TestDb::new().await;
100 let id = "evt_dedup_race";
101
102 let mut handles = Vec::new();
103 for _ in 0..24 {
104 let pool = db.pool.clone();
105 let id = id.to_string();
106 handles.push(tokio::spawn(async move {
107 webhook_events::mark_event_processed(&pool, &id).await
108 }));
109 }
110 for h in handles {
111 // No task may observe a unique-violation, the ON CONFLICT swallows it.
112 h.await
113 .expect("task panicked")
114 .expect("mark must not error under contention");
115 }
116
117 assert_eq!(
118 processed_count(&db.pool).await,
119 1,
120 "concurrent duplicate deliveries must yield exactly one dedup row"
121 );
122 }
123
124 /// `try_lock_event` must serialize same-event deliveries without blocking: while
125 /// one delivery holds the per-event lock, a second delivery of the *same* event
126 /// gets `None` (→ the handler returns 503 for Stripe to redeliver) rather than
127 /// parking a connection, while a *different* event acquires its lock freely. When
128 /// the first guard drops, the same event becomes lockable again (Run 23 Conc/Perf:
129 /// blocking `pg_advisory_xact_lock` → non-blocking `pg_try_advisory_xact_lock`).
130 #[tokio::test]
131 async fn try_lock_event_is_non_blocking_and_per_event() {
132 let db = TestDb::new().await;
133
134 // First delivery of event A wins the lock and holds it.
135 let held = webhook_events::try_lock_event(&db.pool, "evt_lock_A")
136 .await
137 .expect("first try_lock must not error")
138 .expect("first delivery acquires the lock");
139
140 // A second, concurrent delivery of the SAME event must not acquire it, and
141 // must return immediately (None), not block.
142 assert!(
143 webhook_events::try_lock_event(&db.pool, "evt_lock_A")
144 .await
145 .expect("contended try_lock must not error")
146 .is_none(),
147 "a same-event delivery must get None while the lock is held"
148 );
149
150 // A DIFFERENT event hashes to a different key and is unaffected.
151 assert!(
152 webhook_events::try_lock_event(&db.pool, "evt_lock_B")
153 .await
154 .expect("distinct-event try_lock must not error")
155 .is_some(),
156 "a distinct event must acquire its own lock"
157 );
158
159 // Releasing the first guard frees event A for the next delivery.
160 //
161 // Rolled back explicitly rather than dropped. Dropping a sqlx `Transaction`
162 // does not run the ROLLBACK inline, it queues it onto the connection, to be
163 // flushed when that connection is next used or returned to the pool. The
164 // lock is therefore released *eventually*, not by the time `drop` returns,
165 // and asserting re-acquisition straight after a bare `drop(held)` raced that
166 // flush: this test failed roughly one run in three under load, and was the
167 // only flake in the suite once Sando started gating on it. Production is
168 // unaffected (the loser sheds its connection and Stripe redelivers later),
169 // but the deterministic release is what a caller can actually rely on, so it
170 // is what the test asserts.
171 held.rollback()
172 .await
173 .expect("releasing the held lock must not error");
174 assert!(
175 webhook_events::try_lock_event(&db.pool, "evt_lock_A")
176 .await
177 .expect("post-release try_lock must not error")
178 .is_some(),
179 "the lock must be re-acquirable once the holder releases it"
180 );
181 }
182
183 #[tokio::test]
184 async fn prune_processed_events_respects_retention_boundary() {
185 let db = TestDb::new().await;
186
187 webhook_events::mark_event_processed(&db.pool, "evt_old")
188 .await
189 .unwrap();
190 webhook_events::mark_event_processed(&db.pool, "evt_fresh")
191 .await
192 .unwrap();
193 // Age the old marker past a 30-day retention; leave the fresh one at NOW().
194 sqlx::query("UPDATE processed_webhook_events SET processed_at = NOW() - INTERVAL '40 days' WHERE event_id = 'evt_old'")
195 .execute(&db.pool)
196 .await
197 .unwrap();
198
199 let deleted = webhook_events::prune_processed_events(&db.pool, 30)
200 .await
201 .unwrap();
202 assert_eq!(deleted, 1, "exactly the aged marker is pruned");
203
204 assert!(
205 !webhook_events::is_event_processed(&db.pool, "evt_old")
206 .await
207 .unwrap()
208 );
209 assert!(
210 webhook_events::is_event_processed(&db.pool, "evt_fresh")
211 .await
212 .unwrap(),
213 "a within-retention marker must survive the prune"
214 );
215 }
216
217 // ── retry queue (`webhook_events`) claim-and-defer ───────────────────────────
218
219 /// A due event is claimed once, and the same immediate tick can't re-claim it:
220 /// `get_retryable_events` pushes `next_retry_at` out by 2 minutes as it selects,
221 /// so an overlapping scheduler tick (or a second replica) gets nothing. This is
222 /// the "two-concurrent-deliveries" guard the audit noted was untested.
223 #[tokio::test]
224 async fn get_retryable_events_claims_and_defers() {
225 let db = TestDb::new().await;
226 seed_due_event(&db.pool, "claim.a").await;
227 seed_due_event(&db.pool, "claim.b").await;
228
229 let first = webhook_events::get_retryable_events(&db.pool)
230 .await
231 .unwrap();
232 assert_eq!(
233 first.len(),
234 2,
235 "both due events are claimed on the first tick"
236 );
237
238 let second = webhook_events::get_retryable_events(&db.pool)
239 .await
240 .unwrap();
241 assert!(
242 second.is_empty(),
243 "a claimed event's next_retry_at is deferred, so an immediate re-tick claims nothing"
244 );
245 }
246
247 /// Two overlapping ticks racing the same due events must partition them, never
248 /// hand the same event to both callers (would double-run the handler).
249 #[tokio::test]
250 async fn get_retryable_events_concurrent_ticks_do_not_double_claim() {
251 let db = TestDb::new().await;
252 for i in 0..6 {
253 seed_due_event(&db.pool, &format!("race.claim.{i}")).await;
254 }
255
256 let p1 = db.pool.clone();
257 let p2 = db.pool.clone();
258 let (a, b) = tokio::join!(
259 tokio::spawn(async move { webhook_events::get_retryable_events(&p1).await }),
260 tokio::spawn(async move { webhook_events::get_retryable_events(&p2).await }),
261 );
262 let a = a.unwrap().unwrap();
263 let b = b.unwrap().unwrap();
264
265 let mut ids: Vec<_> = a.iter().chain(b.iter()).map(|e| e.id).collect();
266 let total = ids.len();
267 ids.sort();
268 ids.dedup();
269 assert_eq!(
270 ids.len(),
271 total,
272 "no event id may be claimed by both concurrent ticks"
273 );
274 assert_eq!(
275 ids.len(),
276 6,
277 "every due event is claimed exactly once across the two ticks"
278 );
279 }
280
281 #[tokio::test]
282 async fn get_retryable_events_excludes_future_and_exhausted() {
283 let db = TestDb::new().await;
284
285 // Due now, eligible.
286 seed_due_event(&db.pool, "elig.due").await;
287 // Scheduled in the future, not yet eligible.
288 webhook_events::insert_failed_event(&db.pool, "stripe", "elig.future", "{}", None, "e")
289 .await
290 .unwrap();
291 // Due but retry-exhausted (attempts >= 5), must be skipped so it can't storm.
292 let exhausted = seed_due_event(&db.pool, "elig.exhausted").await;
293 sqlx::query("UPDATE webhook_events SET attempts = 5 WHERE id = $1")
294 .bind(exhausted)
295 .execute(&db.pool)
296 .await
297 .unwrap();
298
299 let due = webhook_events::get_retryable_events(&db.pool)
300 .await
301 .unwrap();
302 let types: Vec<_> = due.iter().map(|e| e.event_type.as_str()).collect();
303 assert_eq!(
304 types,
305 ["elig.due"],
306 "only the due, non-exhausted event is returned"
307 );
308 }
309
310 // ── backoff + dead-letter progression ────────────────────────────────────────
311
312 #[tokio::test]
313 async fn schedule_retry_sets_retrying_and_backs_off() {
314 let db = TestDb::new().await;
315 let id = seed_due_event(&db.pool, "backoff.evt").await;
316
317 // First failed retry: attempt count 1, status -> retrying, deferred forward.
318 webhook_events::schedule_retry(&db.pool, id, 1, "still failing")
319 .await
320 .unwrap();
321
322 let (status, attempts, deferred): (String, i32, bool) = sqlx::query_as(
323 "SELECT status, attempts, next_retry_at > NOW() FROM webhook_events WHERE id = $1",
324 )
325 .bind(id)
326 .fetch_one(&db.pool)
327 .await
328 .unwrap();
329 assert_eq!(status, "retrying");
330 assert_eq!(attempts, 1);
331 assert!(
332 deferred,
333 "next_retry_at is pushed into the future by the backoff"
334 );
335
336 // Immediately after scheduling, it is not due, so a tick skips it.
337 assert!(
338 webhook_events::get_retryable_events(&db.pool)
339 .await
340 .unwrap()
341 .is_empty(),
342 "a just-rescheduled event is not yet due"
343 );
344 }
345
346 #[tokio::test]
347 async fn schedule_retry_dead_letters_at_max_attempts() {
348 let db = TestDb::new().await;
349 let id = seed_due_event(&db.pool, "dead.evt").await;
350
351 // Reaching the max attempts marks the event dead, off the retry queue,
352 // onto the operator dead-letter list.
353 webhook_events::schedule_retry(&db.pool, id, 5, "exhausted")
354 .await
355 .unwrap();
356
357 let status: String = sqlx::query_scalar("SELECT status FROM webhook_events WHERE id = $1")
358 .bind(id)
359 .fetch_one(&db.pool)
360 .await
361 .unwrap();
362 assert_eq!(status, "dead");
363
364 assert!(
365 webhook_events::get_retryable_events(&db.pool)
366 .await
367 .unwrap()
368 .is_empty(),
369 "a dead event must never be re-claimed for retry"
370 );
371 let dead = webhook_events::get_dead_events(&db.pool).await.unwrap();
372 assert!(
373 dead.iter().any(|e| e.id == id),
374 "a dead event surfaces on the operator list"
375 );
376 }
377
378 #[tokio::test]
379 async fn retry_dead_event_resurrects_only_dead_rows() {
380 let db = TestDb::new().await;
381
382 // A live (failed) event is not a dead-letter, so resetting it is a no-op.
383 let live = seed_due_event(&db.pool, "resurrect.live").await;
384 assert!(
385 !webhook_events::retry_dead_event(&db.pool, live)
386 .await
387 .unwrap(),
388 "retry_dead_event must only act on status = 'dead'"
389 );
390
391 // A genuinely dead event flips back to 'failed' and becomes claimable again.
392 let dead = seed_due_event(&db.pool, "resurrect.dead").await;
393 webhook_events::schedule_retry(&db.pool, dead, 5, "exhausted")
394 .await
395 .unwrap();
396 assert!(
397 webhook_events::retry_dead_event(&db.pool, dead)
398 .await
399 .unwrap()
400 );
401
402 let status: String = sqlx::query_scalar("SELECT status FROM webhook_events WHERE id = $1")
403 .bind(dead)
404 .fetch_one(&db.pool)
405 .await
406 .unwrap();
407 assert_eq!(
408 status, "failed",
409 "a resurrected event returns to the retry queue"
410 );
411 }
412
413 #[tokio::test]
414 async fn mark_processed_removes_event_from_queue() {
415 let db = TestDb::new().await;
416 let id = seed_due_event(&db.pool, "done.evt").await;
417
418 webhook_events::mark_processed(&db.pool, id).await.unwrap();
419
420 let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events WHERE id = $1")
421 .bind(id)
422 .fetch_one(&db.pool)
423 .await
424 .unwrap();
425 assert_eq!(
426 remaining, 0,
427 "a successfully processed event is deleted from the queue"
428 );
429 }
430