Skip to main content

max / makenotwork

Payments resilience: fan-out dead-letter, billing guards, inbound idempotency /audit Run 13 payments-resilience cold spots: - fan_ops: a failed fan pause/suspend/cancel op was only logged. It now opens a WAM ticket listing the failed subscription IDs, so a dropped op (a fan still charged after a creator pause, a suspended creator's fans retaining access) is surfaced for manual reconciliation. Threaded state.wam through all 6 call sites. - synckit activate_billing: guard the UPDATE with `AND billing_status = 'draft'` and treat 0 rows as a Conflict, so a replay/TOCTOU can't silently resurrect a canceled (terminal) or already-active app at the DB layer (the handler already checked, but the write was unguarded). - shared subscription writer macro: add `start <= end` to the period guard (was `end > 0` only), so an inverted Stripe period writes nothing instead of stamping a backwards window — parity with synckit's writer. - Postmark inbound issues: pre-check the MessageID before create_issue / create_comment so a Postmark redelivery (invited by our 5xx-transient responses) can't create a duplicate issue or reply. New regression inbound_duplicate_delivery_creates_one_issue. Deferred (LOW, flagged): full scan-worker-pool / background drain on shutdown — current behavior is idempotent-safe (jobs re-claimed on restart) and the stateful scheduler is already drained. Integration: git_issues + synckit_billing + admin + subscriptions green (59). clippy --lib clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-01 21:57 UTC
Commit: 4462fabaac9c6128ca927e8888d0afae0af58c95
Parent: b06bc60
9 files changed, +113 insertions, -8 deletions
M server/src/db/subscription_writer.rs +6 -1
@@ -58,7 +58,12 @@ macro_rules! define_stripe_subscription_writer {
58 58 period: Option<(i64, i64)>,
59 59 ) -> $crate::error::Result<Option<$row>> {
60 60 let (period_start, period_end) = match period {
61 - Some((start, end)) if end > 0 => (
61 + // `end > 0` rejects the thin/zero webhook (no 1970 period); the
62 + // additional `start <= end` rejects an inverted range, so a
63 + // malformed Stripe period writes nothing (the `COALESCE` keeps the
64 + // existing values) rather than stamping a backwards window — parity
65 + // with `synckit_billing::apply_billing_update` (audit Run 13).
66 + Some((start, end)) if end > 0 && start <= end => (
62 67 ::chrono::DateTime::from_timestamp(start, 0),
63 68 ::chrono::DateTime::from_timestamp(end, 0),
64 69 ),
M server/src/db/synckit_billing.rs +12 -2
@@ -90,7 +90,12 @@ pub async fn activate_billing(
90 90 period_start: DateTime<Utc>,
91 91 period_end: DateTime<Utc>,
92 92 ) -> Result<()> {
93 - sqlx::query(
93 + // Guard `billing_status = 'draft'` at the DB layer, not just in the handler:
94 + // activation is only ever valid from draft, so a replay or a TOCTOU race that
95 + // reaches here against a canceled (terminal) or already-active app must NOT
96 + // silently resurrect it / orphan a live subscription (audit Run 13). 0 rows
97 + // affected → the app left draft; surface a conflict rather than succeeding.
98 + let result = sqlx::query(
94 99 r#"
95 100 UPDATE sync_apps SET
96 101 billing_status = 'active',
@@ -101,7 +106,7 @@ pub async fn activate_billing(
101 106 gb_per_key = $6,
102 107 current_period_start = $7,
103 108 current_period_end = $8
104 - WHERE id = $1
109 + WHERE id = $1 AND billing_status = 'draft'
105 110 "#,
106 111 )
107 112 .bind(app_id)
@@ -114,6 +119,11 @@ pub async fn activate_billing(
114 119 .bind(period_end)
115 120 .execute(pool)
116 121 .await?;
122 + if result.rows_affected() == 0 {
123 + return Err(crate::error::AppError::Conflict(
124 + "Billing activation is only valid for a draft app".to_string(),
125 + ));
126 + }
117 127 Ok(())
118 128 }
119 129
M server/src/payments/fan_ops.rs +30 -5
@@ -60,29 +60,54 @@ impl FanSubOp {
60 60 /// Apply `op` to every subscription in `sub_ids` on the background queue. Returns
61 61 /// immediately; the loop runs off the request/webhook hot path. Per-subscription
62 62 /// failures are logged and do not abort the rest. No-op for an empty list.
63 + ///
64 + /// A failed op is not merely logged: any failures open a WAM ticket (when `wam`
65 + /// is configured) so a dropped pause/suspend/cancel — a fan still charged after a
66 + /// creator pause, or a suspended creator's fans retaining access — is actively
67 + /// surfaced for manual reconciliation rather than lost in the logs (audit Run 13
68 + /// Resilience: fan-out had no dead-letter). The failed subscription IDs are
69 + /// listed in the ticket body.
63 70 pub fn spawn_fan_sub_fanout(
64 71 bg: &BackgroundTx,
65 72 stripe: Arc<dyn PaymentProvider>,
66 73 account_id: StripeAccountId,
67 74 sub_ids: Vec<String>,
68 75 op: FanSubOp,
76 + wam: Option<crate::wam_client::WamClient>,
69 77 ) {
70 78 if sub_ids.is_empty() {
71 79 return;
72 80 }
73 81 bg.spawn("fan-sub stripe fan-out", async move {
74 82 let total = sub_ids.len();
75 - let mut failed = 0u32;
83 + let mut failed_ids: Vec<&str> = Vec::new();
76 84 for sub_id in &sub_ids {
77 85 if let Err(e) = op.apply(&stripe, sub_id, account_id.as_str()).await {
78 - failed += 1;
86 + failed_ids.push(sub_id);
79 87 tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed");
80 88 }
81 89 }
82 - if failed > 0 {
83 - tracing::warn!(total, failed, op = op.label(), "fan subscription fan-out completed with failures");
84 - } else {
90 + if failed_ids.is_empty() {
85 91 tracing::debug!(total, op = op.label(), "fan subscription fan-out completed");
92 + return;
93 + }
94 + let failed = failed_ids.len();
95 + tracing::warn!(total, failed, op = op.label(), "fan subscription fan-out completed with failures");
96 + if let Some(wam) = wam {
97 + let title = format!(
98 + "Fan subscription fan-out incomplete: {} of {total} '{}' ops failed",
99 + failed,
100 + op.label()
101 + );
102 + let body = format!(
103 + "Applying '{}' to a creator's fan subscriptions (account {}) left {failed} of {total} \
104 + unreconciled. These subscriptions may still be charging (or retaining access) against \
105 + the creator's current state and need manual reconciliation in Stripe.\n\nFailed subscription IDs:\n{}",
106 + op.label(),
107 + account_id.as_str(),
108 + failed_ids.join("\n"),
109 + );
110 + wam.create_ticket(&title, Some(&body), "high", "fan-fanout-incomplete", Some(account_id.as_str())).await;
86 111 }
87 112 });
88 113 }
M server/src/routes/admin/moderation.rs +1
@@ -84,6 +84,7 @@ pub(super) async fn admin_decide_appeal(
84 84 account_id.clone(),
85 85 ids,
86 86 crate::payments::fan_ops::FanSubOp::Resume,
87 + state.wam.clone(),
87 88 );
88 89 }
89 90 }
M server/src/routes/admin/users.rs +3
@@ -188,6 +188,7 @@ pub(super) async fn admin_suspend_user(
188 188 account_id.clone(),
189 189 ids,
190 190 crate::payments::fan_ops::FanSubOp::Pause,
191 + state.wam.clone(),
191 192 );
192 193 let paused = db::subscriptions::pause_subscriptions_for_creator(&state.db, id).await?;
193 194 tracing::info!(user_id = %id, stripe_queued = queued, db_paused = paused, "paused fan subscriptions for suspended creator");
@@ -236,6 +237,7 @@ pub(super) async fn admin_unsuspend_user(
236 237 account_id.clone(),
237 238 ids,
238 239 crate::payments::fan_ops::FanSubOp::Resume,
240 + state.wam.clone(),
239 241 );
240 242 tracing::info!(user_id = %id, resumed = queued, "resumed fan subscriptions for unsuspended creator");
241 243 }
@@ -298,6 +300,7 @@ pub(super) async fn admin_terminate_user(
298 300 account_id.clone(),
299 301 ids,
300 302 crate::payments::fan_ops::FanSubOp::Cancel,
303 + state.wam.clone(),
301 304 );
302 305 }
303 306
M server/src/routes/api/users/profile.rs +1
@@ -343,6 +343,7 @@ pub(in crate::routes::api) async fn pause_creator(
343 343 stripe_account_id.clone(),
344 344 ids,
345 345 crate::payments::fan_ops::FanSubOp::CancelAtPeriodEnd(true),
346 + state.wam.clone(),
346 347 );
347 348 }
348 349 }
M server/src/routes/postmark/issues.rs +33
@@ -140,6 +140,23 @@ async fn handle_new_issue(
140 140 }
141 141 }
142 142
143 + // Idempotency: our 5xx-on-transient design invites Postmark to redeliver, so
144 + // a retry after a prior success must not create a DUPLICATE issue. If this
145 + // MessageID already maps to an issue, treat it as already-processed (audit
146 + // Run 13 Payments idempotency).
147 + if !payload.message_id.is_empty() {
148 + match db::issues::get_issue_id_by_any_message_id(&state.db, &[&payload.message_id]).await {
149 + Ok(Some(_)) => {
150 + tracing::info!(message_id = %payload.message_id, "inbound-issues: duplicate delivery; issue already created");
151 + return HandlerOutcome::Terminal(StatusCode::OK);
152 + }
153 + Ok(None) => {}
154 + Err(e) => {
155 + return HandlerOutcome::Transient(anyhow::Error::new(e).context("inbound-issues: idempotency check"));
156 + }
157 + }
158 + }
159 +
143 160 // Create the issue
144 161 let title = payload.subject.trim();
145 162 if title.is_empty() {
@@ -289,6 +306,22 @@ async fn handle_issue_reply(
289 306 return HandlerOutcome::Terminal(StatusCode::OK);
290 307 }
291 308
309 + // Idempotency: a redelivered reply (our 5xx invites Postmark to retry) must
310 + // not create a DUPLICATE comment. If this MessageID is already mapped, skip
311 + // (audit Run 13 Payments idempotency).
312 + if !payload.message_id.is_empty() {
313 + match db::issues::get_issue_id_by_any_message_id(&state.db, &[&payload.message_id]).await {
314 + Ok(Some(_)) => {
315 + tracing::info!(message_id = %payload.message_id, "inbound-issues: duplicate reply delivery; comment already recorded");
316 + return HandlerOutcome::Terminal(StatusCode::OK);
317 + }
318 + Ok(None) => {}
319 + Err(e) => {
320 + return HandlerOutcome::Transient(anyhow::Error::new(e).context("inbound-issues: reply idempotency check"));
321 + }
322 + }
323 + }
324 +
292 325 // Look up the issue
293 326 let issue = match db::issues::get_issue_by_id(&state.db, issue_id).await {
294 327 Ok(Some(i)) => i,
M server/src/routes/stripe/webhook/checkout.rs +1
@@ -554,6 +554,7 @@ pub(super) async fn handle_creator_tier_checkout_completed(
554 554 stripe_account_id.clone(),
555 555 ids,
556 556 crate::payments::fan_ops::FanSubOp::CancelAtPeriodEnd(false),
557 + state.wam.clone(),
557 558 );
558 559 }
559 560
M server/tests/workflows/git_issues.rs +26
@@ -351,6 +351,32 @@ async fn inbound_spoofed_sender_is_rejected() {
351 351 }
352 352
353 353 #[tokio::test]
354 + async fn inbound_duplicate_delivery_creates_one_issue() {
355 + // Postmark redelivers on our 5xx-transient responses; a retry carrying the
356 + // same MessageID must not create a duplicate issue (Run 13 idempotency).
357 + let tmp = tempfile::TempDir::new().unwrap();
358 + make_test_repo(tmp.path());
359 + let mut h = setup_with_inbound(&tmp).await;
360 +
361 + h.client.set_bearer_token("test-inbound-secret");
362 + // Reusing the same payload string reuses the same MessageID.
363 + let payload = inbound_payload(
364 + "testowner+testrepo@issues.makenot.work",
365 + "testowner@example.com",
366 + "Duplicate delivery",
367 + "This should only create one issue.",
368 + );
369 + let r1 = h.client.post_json("/postmark/inbound-issues", &payload).await;
370 + assert_eq!(r1.status, 200, "first delivery: {}", r1.text);
371 + let r2 = h.client.post_json("/postmark/inbound-issues", &payload).await;
372 + assert_eq!(r2.status, 200, "redelivery ack: {}", r2.text);
373 +
374 + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM issues")
375 + .fetch_one(&h.db).await.unwrap();
376 + assert_eq!(count, 1, "a redelivered MessageID must not create a duplicate issue");
377 + }
378 +
379 + #[tokio::test]
354 380 async fn inbound_missing_auth_is_rejected() {
355 381 // No SPF/DKIM verdict at all (enforce mode) — the sender is unauthenticated,
356 382 // so no issue is attributed to the claimed From user.