Skip to main content

max / makenotwork

payments: seal webhook dedup and tip-split idempotency (audit Run 20 Phase 1) Close the Run 20 chronic tip-split money-loss and make webhook dedup a structural guarantee rather than a per-handler convention. - tip splits: add revenue_splits_tip_recipient_key (migration 163) and ON CONFLICT DO NOTHING to create_tip_splits, mirroring the transaction split backstop from 151. A redelivered tip can no longer double-credit. - tip handler: on the already-completed (None) branch, re-fetch the tip via the new get_tip_by_session and re-run the idempotent split write, matching the purchase/cart crash-recovery path. A crash between complete_tip and the split write no longer strands collaborators' shares. - webhook dedup: add db::webhook_events::lock_event, a per-event pg_advisory_xact_lock held across dedup-read -> process -> mark in both the v1 and v2 handlers. Concurrent redeliveries now serialize, so the check-then-act read is race-free and double-processing is impossible without relying on every handler's own idempotency. The lock is transaction-scoped, so any early return or panic releases it. - add an end-to-end tip-split crash-recovery regression test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-03 06:55 UTC
Commit: 75be30d11cf9f88d8ecd41cf23b983e7d7c81200
Parent: 04d605b
9 files changed, +340 insertions, -15 deletions
A server/.sqlx/query-00d9075c944110786011058b3022eaaeef8bad250de214ecb631b451b34d78f0.json +88
@@ -0,0 +1,88 @@
1 + {
2 + "db_name": "PostgreSQL",
3 + "query": "\n SELECT\n id AS \"id: TipId\", tipper_id AS \"tipper_id: UserId\", recipient_id AS \"recipient_id: UserId\",\n project_id AS \"project_id: ProjectId\", amount_cents AS \"amount_cents: Cents\", message,\n status AS \"status: super::TransactionStatus\", stripe_payment_intent_id, stripe_checkout_session_id,\n stripe_transfer_group,\n created_at AS \"created_at: chrono::DateTime<chrono::Utc>\",\n completed_at AS \"completed_at: chrono::DateTime<chrono::Utc>\"\n FROM tips\n WHERE stripe_checkout_session_id = $1\n ",
4 + "describe": {
5 + "columns": [
6 + {
7 + "ordinal": 0,
8 + "name": "id: TipId",
9 + "type_info": "Uuid"
10 + },
11 + {
12 + "ordinal": 1,
13 + "name": "tipper_id: UserId",
14 + "type_info": "Uuid"
15 + },
16 + {
17 + "ordinal": 2,
18 + "name": "recipient_id: UserId",
19 + "type_info": "Uuid"
20 + },
21 + {
22 + "ordinal": 3,
23 + "name": "project_id: ProjectId",
24 + "type_info": "Uuid"
25 + },
26 + {
27 + "ordinal": 4,
28 + "name": "amount_cents: Cents",
29 + "type_info": "Int4"
30 + },
31 + {
32 + "ordinal": 5,
33 + "name": "message",
34 + "type_info": "Varchar"
35 + },
36 + {
37 + "ordinal": 6,
38 + "name": "status: super::TransactionStatus",
39 + "type_info": "Varchar"
40 + },
41 + {
42 + "ordinal": 7,
43 + "name": "stripe_payment_intent_id",
44 + "type_info": "Varchar"
45 + },
46 + {
47 + "ordinal": 8,
48 + "name": "stripe_checkout_session_id",
49 + "type_info": "Varchar"
50 + },
51 + {
52 + "ordinal": 9,
53 + "name": "stripe_transfer_group",
54 + "type_info": "Varchar"
55 + },
56 + {
57 + "ordinal": 10,
58 + "name": "created_at: chrono::DateTime<chrono::Utc>",
59 + "type_info": "Timestamptz"
60 + },
61 + {
62 + "ordinal": 11,
63 + "name": "completed_at: chrono::DateTime<chrono::Utc>",
64 + "type_info": "Timestamptz"
65 + }
66 + ],
67 + "parameters": {
68 + "Left": [
69 + "Text"
70 + ]
71 + },
72 + "nullable": [
73 + false,
74 + false,
75 + false,
76 + true,
77 + false,
78 + true,
79 + false,
80 + true,
81 + true,
82 + true,
83 + false,
84 + true
85 + ]
86 + },
87 + "hash": "00d9075c944110786011058b3022eaaeef8bad250de214ecb631b451b34d78f0"
88 + }
A server/migrations/163_tip_splits_idempotency.sql +21
@@ -0,0 +1,21 @@
1 + -- Tip revenue-split idempotency (Run 20 Payments, chronic money-loss finding).
2 + --
3 + -- Migration 151 gave transaction revenue splits a unique backstop
4 + -- (`revenue_splits_tx_recipient_key`) so `create_transaction_splits` could use
5 + -- ON CONFLICT DO NOTHING and a crash-recovery redelivery re-runs safely. Tip
6 + -- splits never got the equivalent: `create_tip_splits` was a plain INSERT, so a
7 + -- webhook redelivery after a crash between `complete_tip` and the split write
8 + -- left collaborators permanently short their share.
9 + --
10 + -- This index is the backstop that lets `create_tip_splits` use ON CONFLICT and
11 + -- makes a double-split structurally impossible. Tip-split rows carry
12 + -- transaction_id IS NULL and tip_id IS NOT NULL; transaction-split rows are the
13 + -- reverse, so a full unique index on (tip_id, recipient_id) constrains only tip
14 + -- rows (transaction rows share the NULL tip_id, which PostgreSQL treats as
15 + -- distinct, so they never collide here).
16 + --
17 + -- NON-CONCURRENT by necessity (sqlx runs each migration in its own transaction).
18 + -- Acceptable at current revenue_splits size; revisit if it grows large.
19 + CREATE UNIQUE INDEX IF NOT EXISTS revenue_splits_tip_recipient_key
20 + ON revenue_splits (tip_id, recipient_id)
21 + WHERE tip_id IS NOT NULL;
M server/src/db/project_members.rs +6
@@ -234,10 +234,16 @@ pub async fn create_tip_splits(
234 234 let recipient_ids: Vec<UserId> = splits.iter().map(|(id, _, _)| *id).collect();
235 235 let amounts: Vec<i32> = splits.iter().map(|(_, a, _)| *a as i32).collect();
236 236 let percents: Vec<i16> = splits.iter().map(|(_, _, p)| *p).collect();
237 + // ON CONFLICT DO NOTHING (against the partial revenue_splits_tip_recipient_key,
238 + // migration 163): one split row per recipient per tip, so a crash-recovery
239 + // redelivery that re-runs `record_tip_splits` is a no-op rather than a
240 + // double-credit (Run 20 Payments chronic). The WHERE predicate matches the
241 + // partial index so PostgreSQL infers it as the arbiter.
237 242 sqlx::query(
238 243 r#"
239 244 INSERT INTO revenue_splits (tip_id, recipient_id, amount_cents, split_percent, status)
240 245 SELECT $1, UNNEST($2::uuid[]), UNNEST($3::int[]), UNNEST($4::smallint[]), 'pending'
246 + ON CONFLICT (tip_id, recipient_id) WHERE tip_id IS NOT NULL DO NOTHING
241 247 "#,
242 248 )
243 249 .bind(tip_id)
M server/src/db/tips.rs +32
@@ -84,6 +84,38 @@ pub async fn complete_tip<'e>(
84 84 Ok(tip)
85 85 }
86 86
87 + /// Fetch a tip by its Stripe checkout session id, regardless of status.
88 + ///
89 + /// Used for webhook crash recovery: when `complete_tip` returns `None` (the tip
90 + /// was already flipped to completed by an earlier delivery), the handler re-reads
91 + /// the tip here to re-run the idempotent split write, in case the first delivery
92 + /// crashed after completing the tip but before recording splits.
93 + #[tracing::instrument(skip(pool))]
94 + pub async fn get_tip_by_session(
95 + pool: &PgPool,
96 + stripe_checkout_session_id: &str,
97 + ) -> Result<Option<DbTip>> {
98 + let tip = sqlx::query_as!(
99 + DbTip,
100 + r#"
101 + SELECT
102 + id AS "id: TipId", tipper_id AS "tipper_id: UserId", recipient_id AS "recipient_id: UserId",
103 + project_id AS "project_id: ProjectId", amount_cents AS "amount_cents: Cents", message,
104 + status AS "status: super::TransactionStatus", stripe_payment_intent_id, stripe_checkout_session_id,
105 + stripe_transfer_group,
106 + created_at AS "created_at: chrono::DateTime<chrono::Utc>",
107 + completed_at AS "completed_at: chrono::DateTime<chrono::Utc>"
108 + FROM tips
109 + WHERE stripe_checkout_session_id = $1
110 + "#,
111 + stripe_checkout_session_id,
112 + )
113 + .fetch_optional(pool)
114 + .await?;
115 +
116 + Ok(tip)
117 + }
118 +
87 119 /// Get tips received by a creator, most recent first.
88 120 #[tracing::instrument(skip(pool))]
89 121 pub async fn get_tips_received(
M server/src/db/webhook_events.rs +37 -1
@@ -1,6 +1,6 @@
1 1 //! Webhook event retry queue; persist and retry failed webhook deliveries.
2 2
3 - use sqlx::PgPool;
3 + use sqlx::{PgPool, Postgres, Transaction};
4 4 use chrono::{DateTime, Utc};
5 5 use crate::error::Result;
6 6
@@ -57,6 +57,42 @@ 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.
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.
68 + ///
69 + /// This is the structural counterpart to the check-then-act dedup read. On its
70 + /// own that read has a TOCTOU window: two concurrent deliveries both observe
71 + /// "not processed" and both run the handler, so exactly-once rests entirely on
72 + /// every handler's own idempotency. With this lock held, the read is race-free
73 + /// and concurrent double-processing is impossible — a future non-idempotent
74 + /// handler cannot double-fire on a redelivery race.
75 + ///
76 + /// Robustness: the lock is transaction-scoped, so dropping the guard on any
77 + /// 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).
81 + #[tracing::instrument(skip_all)]
82 + pub async fn lock_event<'a>(
83 + pool: &'a PgPool,
84 + event_id: &str,
85 + ) -> Result<Transaction<'a, Postgres>> {
86 + let mut tx = pool.begin().await?;
87 + sqlx::query(
88 + "SELECT pg_advisory_xact_lock(hashtextextended('stripe_webhook:' || $1::text, 0))",
89 + )
90 + .bind(event_id)
91 + .execute(&mut *tx)
92 + .await?;
93 + Ok(tx)
94 + }
95 +
60 96 /// Delete processed-event dedup markers older than `days`. These markers only
61 97 /// guard against Stripe *redelivering* an event, which it stops doing within a
62 98 /// few days; 30 days is the retention the table was created with (migration
M server/src/routes/stripe/webhook/checkout.rs +20 -1
@@ -639,7 +639,26 @@ pub(super) async fn handle_tip_checkout_completed(
639 639 send_tip_email(state, &tip, tipper_id, recipient_id);
640 640 }
641 641 None => {
642 - tracing::info!(session_id = %session_id, "tip already completed, ignoring duplicate webhook");
642 + // No pending row flipped to completed. Either a benign duplicate of
643 + // an already-finalized tip, OR the first delivery crashed AFTER
644 + // completing the tip but BEFORE recording splits — in which case
645 + // collaborators hold a completed tip with no split rows. Re-fetch the
646 + // tip and re-run the idempotent split write (ON CONFLICT DO NOTHING,
647 + // migration 163); a genuine duplicate is a no-op. Run 20 Payments.
648 + let recovered = db::tips::get_tip_by_session(&state.db, &session_id)
649 + .await
650 + .context("re-fetch tip for crash recovery")?;
651 + if let Some(tip) = recovered
652 + && let Some(project_id) = tip.project_id
653 + {
654 + tracing::info!(
655 + session_id = %session_id, tip_id = %tip.id,
656 + "crash-recovery: re-running tip splits for already-completed tip"
657 + );
658 + record_tip_splits(state, tip.id, project_id, tip.amount_cents).await;
659 + } else {
660 + tracing::info!(session_id = %session_id, "tip already completed, ignoring duplicate webhook");
661 + }
643 662 }
644 663 }
645 664
M server/src/routes/stripe/webhook/mod.rs +21 -13
@@ -45,21 +45,29 @@ pub(in crate::routes::stripe) async fn webhook(
45 45 let event = stripe.verify_webhook(payload, signature)?;
46 46 tracing::info!(event_type = %event.type_, event_id = %event.id, "received webhook event");
47 47
48 + // Serialize concurrent redeliveries of this event id. Held across the whole
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,
58 + Err(e) => {
59 + tracing::error!(event_id = %event.id, error = ?e, "failed to acquire webhook event lock, returning 503 for retry");
60 + return Ok(StatusCode::SERVICE_UNAVAILABLE);
61 + }
62 + };
63 +
48 64 // Deduplicate: skip if we already processed this event ID. This is a READ;
49 65 // the "processed" row is written only after the handler succeeds (below), so
50 - // a crash mid-processing leaves no marker and Stripe's redelivery reprocesses
51 - // (handlers are idempotent). The old "mark before processing" ordering could
52 - // strand an event whose process died after the mark committed but before its
53 - // work or retry row landed.
54 - //
55 - // INVARIANT (load-bearing): this dedup is check-then-act, so two concurrent
56 - // deliveries of the same event both reach `process_webhook_event`. Safety
57 - // therefore rests entirely on every handler's side-effects being atomic and
58 - // idempotent — a status-guarded UPDATE, an ON CONFLICT write, or a money
59 - // mint that sits behind its own claim witness (see `FanPlusCreditClaim` in
60 - // `db::promo_codes`). A *non-idempotent* side-effect (sending an email,
61 - // calling an external API) MUST be gated behind such a claim, never run
62 - // before its atomic write — otherwise a redelivery double-fires it.
66 + // a crash mid-processing leaves no marker and Stripe's redelivery reprocesses.
67 + // The `_event_lock` above closes the check-then-act race, so this read now
68 + // guarantees exactly-once dispatch regardless of handler idempotency; per-
69 + // handler atomic guards (status-guarded UPDATEs, ON CONFLICT writes, the
70 + // `FanPlusCreditClaim` witness) remain as defense in depth.
63 71 match db::webhook_events::is_event_processed(&state.db, &event.id).await {
64 72 Ok(true) => {
65 73 tracing::info!(event_id = %event.id, "duplicate webhook event, skipping");
M server/src/routes/stripe/webhook_v2.rs +13
@@ -48,6 +48,19 @@ pub(super) async fn webhook_v2(
48 48
49 49 tracing::info!(event_type = %thin.event_type, event_id = %thin.id, "received v2 thin event");
50 50
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,
58 + Err(e) => {
59 + tracing::error!(event_id = %thin.id, error = ?e, "failed to acquire v2 webhook event lock, returning 503 for retry");
60 + return Ok(StatusCode::SERVICE_UNAVAILABLE);
61 + }
62 + };
63 +
51 64 // Deduplicate: skip if this event was already processed. Read-only; the
52 65 // "processed" row is written only after the handler succeeds (below), so a
53 66 // crash mid-processing leaves no marker and Stripe redelivers (the handler
M server/tests/workflows/payment_crash_recovery.rs +102
@@ -182,6 +182,108 @@ async fn crash_recovery_reruns_finalize_idempotently() {
182 182 assert_eq!(split_count, 1, "idempotent: still exactly one split");
183 183 }
184 184
185 + /// Deliver a `checkout.session.completed` webhook for a tip on `session_id`.
186 + async fn deliver_tip_webhook(
187 + h: &mut TestHarness,
188 + event_id: &str,
189 + session_id: &str,
190 + tipper_id: db::UserId,
191 + recipient_id: db::UserId,
192 + ) -> u16 {
193 + let mut meta = HashMap::new();
194 + meta.insert("checkout_type".to_string(), "tip".to_string());
195 + meta.insert("tipper_id".to_string(), tipper_id.to_string());
196 + meta.insert("recipient_id".to_string(), recipient_id.to_string());
197 + let session = serde_json::json!({
198 + "id": session_id,
199 + "object": "checkout_session",
200 + "mode": "payment",
201 + "metadata": meta,
202 + "payment_intent": "pi_tip_crash_recovery",
203 + });
204 + let payload = serde_json::json!({
205 + "id": event_id,
206 + "type": "checkout.session.completed",
207 + "data": {"object": session},
208 + })
209 + .to_string();
210 + let signature = crate::harness::stripe::sign_webhook_payload(
211 + &payload,
212 + crate::harness::stripe::TEST_WEBHOOK_SECRET,
213 + );
214 + h.client.request_with_headers(
215 + "POST",
216 + "/stripe/webhook",
217 + Some(&payload),
218 + &[
219 + ("stripe-signature", &signature),
220 + ("content-type", "application/json"),
221 + ],
222 + ).await.status.as_u16()
223 + }
224 +
225 + /// Tip revenue splits must survive a crash between `complete_tip` and the split
226 + /// write, and a redelivery must not double-credit collaborators (Run 20 chronic
227 + /// money-loss). Mirrors the purchase recovery test for the tip path.
228 + #[tokio::test]
229 + async fn tip_split_crash_recovery_is_idempotent() {
230 + let mut h = TestHarness::with_mocks().await;
231 +
232 + // Recipient (creator) with a project and a 50% collaborator.
233 + let recipient_id = h.signup("trecip", "trecip@test.com", "pass1234").await;
234 + let project_id: db::ProjectId = sqlx::query_scalar(
235 + "INSERT INTO projects (user_id, slug, title) VALUES ($1, 'tshop', 'TShop') RETURNING id",
236 + ).bind(recipient_id).fetch_one(&h.db).await.unwrap();
237 + let collab_id = h.signup("tcollab", "tcollab@test.com", "pass1234").await;
238 + sqlx::query(
239 + "INSERT INTO project_members (project_id, user_id, role, split_percent, added_by) VALUES ($1, $2, 'member', 50, $3)",
240 + )
241 + .bind(project_id)
242 + .bind(collab_id)
243 + .bind(recipient_id)
244 + .execute(&h.db)
245 + .await
246 + .unwrap();
247 +
248 + // A tipper and a pending tip attributed to the project.
249 + let tipper_id = h.signup("ttipper", "ttipper@test.com", "pass1234").await;
250 + let session_id = "cs_tip_crash_001";
251 + db::tips::create_tip(&h.db, tipper_id, recipient_id, Some(project_id), 1000, None, session_id)
252 + .await
253 + .unwrap();
254 + let tip_id: db::TipId = sqlx::query_scalar(
255 + "SELECT id FROM tips WHERE stripe_checkout_session_id = $1",
256 + ).bind(session_id).fetch_one(&h.db).await.unwrap();
257 +
258 + // First delivery completes the tip and records the split.
259 + let status = deliver_tip_webhook(&mut h, "evt_tip_1", session_id, tipper_id, recipient_id).await;
260 + assert_eq!(status, 200, "first tip webhook should succeed");
261 + tokio::time::sleep(std::time::Duration::from_millis(200)).await;
262 + let split_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE tip_id = $1")
263 + .bind(tip_id).fetch_one(&h.db).await.unwrap();
264 + assert_eq!(split_count, 1, "first delivery records the collaborator split");
265 +
266 + // Simulate a crash before the split write: tip stays completed, split gone.
267 + sqlx::query("DELETE FROM revenue_splits WHERE tip_id = $1").bind(tip_id).execute(&h.db).await.unwrap();
268 +
269 + // Redelivery (new event id): complete_tip returns None; the recovery branch
270 + // re-fetches the tip and re-runs the idempotent split write.
271 + let status = deliver_tip_webhook(&mut h, "evt_tip_2", session_id, tipper_id, recipient_id).await;
272 + assert_eq!(status, 200, "recovery tip webhook should succeed");
273 + tokio::time::sleep(std::time::Duration::from_millis(200)).await;
274 + let split_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE tip_id = $1")
275 + .bind(tip_id).fetch_one(&h.db).await.unwrap();
276 + assert_eq!(split_count, 1, "recovery re-records the missing tip split");
277 +
278 + // A second recovery must NOT duplicate (ON CONFLICT DO NOTHING, migration 163).
279 + let status = deliver_tip_webhook(&mut h, "evt_tip_3", session_id, tipper_id, recipient_id).await;
280 + assert_eq!(status, 200, "second recovery tip webhook should succeed");
281 + tokio::time::sleep(std::time::Duration::from_millis(200)).await;
282 + let split_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM revenue_splits WHERE tip_id = $1")
283 + .bind(tip_id).fetch_one(&h.db).await.unwrap();
284 + assert_eq!(split_count, 1, "idempotent: still exactly one tip split");
285 + }
286 +
185 287 /// The partial unique index on license_keys (transaction_id) structurally rejects
186 288 /// a second auto-minted key for the same transaction even if the pre-check is
187 289 /// bypassed.