Skip to main content

max / makenotwork

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