Skip to main content

max / makenotwork

server: non-blocking per-event webhook lock (try_lock_event) Blocking pg_advisory_xact_lock became a non-blocking pg_try_advisory_xact_lock at all three call sites (v1 handler, v2 thin handler, retry worker); the loser returns 503 / re-queues instead of pinning a pooled connection while blocked during a same-event redelivery storm. New test pins the per-event non-blocking semantics. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-05 15:49 UTC
Commit: 6c2657ed21b30a6113d74d9fab7ac6e106fb593b
Parent: 049dea0
5 files changed, +110 insertions, -35 deletions
M server/src/db/webhook_events.rs +28 -16
@@ -57,14 +57,15 @@ pub async fn mark_event_processed(pool: &PgPool, event_id: &str) -> Result<()> {
57 57 Ok(())
58 58 }
59 59
60 - /// Serialize concurrent redeliveries of one webhook event.
60 + /// Serialize concurrent redeliveries of one webhook event, without blocking.
61 61 ///
62 - /// Returns a transaction holding a per-event `pg_advisory_xact_lock`. Hold the
63 - /// returned guard across the whole dedup-read -> process -> mark sequence: a
64 - /// second concurrent delivery of the same event blocks in this call until the
65 - /// first delivery returns (releasing the lock) and, by then, has durably
66 - /// committed its [`mark_event_processed`] row — so the second delivery's dedup
67 - /// read sees it and short-circuits.
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.
68 69 ///
69 70 /// This is the structural counterpart to the check-then-act dedup read. On its
70 71 /// own that read has a TOCTOU window: two concurrent deliveries both observe
@@ -73,24 +74,35 @@ pub async fn mark_event_processed(pool: &PgPool, event_id: &str) -> Result<()> {
73 74 /// and concurrent double-processing is impossible — a future non-idempotent
74 75 /// handler cannot double-fire on a redelivery race.
75 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 + ///
76 86 /// Robustness: the lock is transaction-scoped, so dropping the guard on any
77 87 /// early return, `?`, or panic rolls the (write-free) transaction back and
78 - /// releases the lock. It cannot leak the way a pooled session lock would. The
79 - /// key is namespaced (`stripe_webhook:` prefix) so it shares no space with the
80 - /// other `hashtextextended` advisory locks in the codebase (reports, oauth).
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. The key is namespaced (`stripe_webhook:` prefix)
91 + /// so it shares no space with the other `hashtextextended` advisory locks in the
92 + /// codebase (reports, oauth).
81 93 #[tracing::instrument(skip_all)]
82 - pub async fn lock_event<'a>(
94 + pub async fn try_lock_event<'a>(
83 95 pool: &'a PgPool,
84 96 event_id: &str,
85 - ) -> Result<Transaction<'a, Postgres>> {
97 + ) -> Result<Option<Transaction<'a, Postgres>>> {
86 98 let mut tx = pool.begin().await?;
87 - sqlx::query(
88 - "SELECT pg_advisory_xact_lock(hashtextextended('stripe_webhook:' || $1::text, 0))",
99 + let acquired = sqlx::query_scalar::<_, bool>(
100 + "SELECT pg_try_advisory_xact_lock(hashtextextended('stripe_webhook:' || $1::text, 0))",
89 101 )
90 102 .bind(event_id)
91 - .execute(&mut *tx)
103 + .fetch_one(&mut *tx)
92 104 .await?;
93 - Ok(tx)
105 + Ok(acquired.then_some(tx))
94 106 }
95 107
96 108 /// Delete processed-event dedup markers older than `days`. These markers only
M server/src/routes/stripe/webhook/mod.rs +14 -8
@@ -47,14 +47,20 @@ pub(in crate::routes::stripe) async fn webhook(
47 47
48 48 // Serialize concurrent redeliveries of this event id. Held across the whole
49 49 // dedup-read -> process -> mark sequence below, the lock makes the dedup read
50 - // race-free: a second delivery of the same event blocks until this one
51 - // returns and has durably committed its processed-event mark, then reads it
52 - // and skips. Dropping `_event_lock` on any return path rolls the (write-free)
53 - // lock transaction back and releases it — it cannot leak. See
54 - // `db::webhook_events::lock_event`. On acquisition failure we return 503 so
55 - // Stripe retries, matching the dedup-read failure path.
56 - let _event_lock = match db::webhook_events::lock_event(&state.db, &event.id).await {
57 - Ok(tx) => tx,
50 + // race-free. We *try* the lock rather than block on it: if a second delivery
51 + // of the same event arrives while this one is mid-flight, it gets `None` and
52 + // returns 503 immediately instead of parking a pooled connection for the
53 + // duration of the in-flight handler (Run 23 Conc/Perf). Stripe redelivers
54 + // after this one commits its processed-event mark, and the redelivery's dedup
55 + // read then short-circuits. Dropping `_event_lock` on any return path rolls
56 + // the (write-free) lock transaction back and releases it — it cannot leak.
57 + // See `db::webhook_events::try_lock_event`.
58 + let _event_lock = match db::webhook_events::try_lock_event(&state.db, &event.id).await {
59 + Ok(Some(tx)) => tx,
60 + Ok(None) => {
61 + tracing::info!(event_id = %event.id, "concurrent delivery of this webhook event is in flight; returning 503 for redelivery");
62 + return Ok(StatusCode::SERVICE_UNAVAILABLE);
63 + }
58 64 Err(e) => {
59 65 tracing::error!(event_id = %event.id, error = ?e, "failed to acquire webhook event lock, returning 503 for retry");
60 66 return Ok(StatusCode::SERVICE_UNAVAILABLE);
M server/src/routes/stripe/webhook_v2.rs +11 -6
@@ -49,12 +49,17 @@ pub(super) async fn webhook_v2(
49 49 tracing::info!(event_type = %thin.event_type, event_id = %thin.id, "received v2 thin event");
50 50
51 51 // Serialize concurrent redeliveries of this event id (see
52 - // `db::webhook_events::lock_event`). Held across dedup-read -> process ->
53 - // mark, it makes the check-then-act read below race-free. Dropping
54 - // `_event_lock` on any return releases the lock; on acquisition failure we
55 - // return 503 so Stripe retries.
56 - let _event_lock = match db::webhook_events::lock_event(&state.db, &thin.id).await {
57 - Ok(tx) => tx,
52 + // `db::webhook_events::try_lock_event`). Held across dedup-read -> process ->
53 + // mark, it makes the check-then-act read below race-free. We *try* the lock
54 + // rather than block: a same-event delivery arriving mid-flight gets `None` and
55 + // returns 503 immediately instead of parking a pooled connection (Run 23
56 + // Conc/Perf). Dropping `_event_lock` on any return releases the lock.
57 + let _event_lock = match db::webhook_events::try_lock_event(&state.db, &thin.id).await {
58 + Ok(Some(tx)) => tx,
59 + Ok(None) => {
60 + tracing::info!(event_id = %thin.id, "concurrent delivery of this v2 event is in flight; returning 503 for redelivery");
61 + return Ok(StatusCode::SERVICE_UNAVAILABLE);
62 + }
58 63 Err(e) => {
59 64 tracing::error!(event_id = %thin.id, error = ?e, "failed to acquire v2 webhook event lock, returning 503 for retry");
60 65 return Ok(StatusCode::SERVICE_UNAVAILABLE);
M server/src/scheduler/webhooks.rs +11 -5
@@ -47,12 +47,18 @@ pub(super) async fn retry_failed_webhooks(state: &AppState) {
47 47 };
48 48
49 49 // Hold the per-event lock across dedup-read -> process -> mark, exactly
50 - // like the live handler. On lock failure, leave the row queued for the
51 - // next tick. Named binding (not bare `_`) so the guard lives — and holds
52 - // the lock — for the whole iteration.
50 + // like the live handler. We *try* the lock: if a live redelivery (or
51 + // another tick) already holds it, leave this row queued and move on rather
52 + // than blocking the retry loop on a pooled connection. Named binding (not
53 + // bare `_`) so the guard lives — and holds the lock — for the whole
54 + // iteration.
53 55 let _event_lock = match &stripe_event_id {
54 - Some(eid) => match db::webhook_events::lock_event(&state.db, eid).await {
55 - Ok(tx) => Some(tx),
56 + Some(eid) => match db::webhook_events::try_lock_event(&state.db, eid).await {
57 + Ok(Some(tx)) => Some(tx),
58 + Ok(None) => {
59 + tracing::info!(event_id = %event.id, "webhook event locked by another worker; leaving queued for next tick");
60 + continue;
61 + }
56 62 Err(e) => {
57 63 tracing::error!(event_id = %event.id, error = ?e, "failed to lock webhook event for retry; will retry next tick");
58 64 continue;
M server/tests/workflows/db_webhook_events.rs +46
@@ -103,6 +103,52 @@ async fn mark_event_processed_concurrent_duplicates_leave_one_row() {
103 103 );
104 104 }
105 105
106 + /// `try_lock_event` must serialize same-event deliveries without blocking: while
107 + /// one delivery holds the per-event lock, a second delivery of the *same* event
108 + /// gets `None` (→ the handler returns 503 for Stripe to redeliver) rather than
109 + /// parking a connection, while a *different* event acquires its lock freely. When
110 + /// the first guard drops, the same event becomes lockable again (Run 23 Conc/Perf:
111 + /// blocking `pg_advisory_xact_lock` → non-blocking `pg_try_advisory_xact_lock`).
112 + #[tokio::test]
113 + async fn try_lock_event_is_non_blocking_and_per_event() {
114 + let db = TestDb::new().await;
115 +
116 + // First delivery of event A wins the lock and holds it.
117 + let held = webhook_events::try_lock_event(&db.pool, "evt_lock_A")
118 + .await
119 + .expect("first try_lock must not error")
120 + .expect("first delivery acquires the lock");
121 +
122 + // A second, concurrent delivery of the SAME event must not acquire it — and
123 + // must return immediately (None), not block.
124 + assert!(
125 + webhook_events::try_lock_event(&db.pool, "evt_lock_A")
126 + .await
127 + .expect("contended try_lock must not error")
128 + .is_none(),
129 + "a same-event delivery must get None while the lock is held"
130 + );
131 +
132 + // A DIFFERENT event hashes to a different key and is unaffected.
133 + assert!(
134 + webhook_events::try_lock_event(&db.pool, "evt_lock_B")
135 + .await
136 + .expect("distinct-event try_lock must not error")
137 + .is_some(),
138 + "a distinct event must acquire its own lock"
139 + );
140 +
141 + // Releasing the first guard frees event A for the next delivery.
142 + drop(held);
143 + assert!(
144 + webhook_events::try_lock_event(&db.pool, "evt_lock_A")
145 + .await
146 + .expect("post-release try_lock must not error")
147 + .is_some(),
148 + "the lock must be re-acquirable once the holder drops"
149 + );
150 + }
151 +
106 152 #[tokio::test]
107 153 async fn prune_processed_events_respects_retention_boundary() {
108 154 let db = TestDb::new().await;