fix(billing): lock the subscription before the outbox row in every sync retry path

Author: waleedlatif1

apps/sim/app/api/v1/admin/outbox/[id]/requeue/route.ts

@@ -14,6 +14,7 @@ import {
 } from '@/lib/billing/enterprise-outbox-events'
 import {
   isSubscriptionSyncEventType,
+  lockSubscriptionForSyncRetry,
   recommitSubscriptionSync,
 } from '@/lib/billing/webhooks/subscription-sync'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
@@ -68,7 +69,17 @@ export const POST = withRouteHandler(
       const deliveryRevision = metadataIntent?.success
         ? metadataIntent.data.deliveryRevision + 1
         : null
+      const subscriptionId = toRecord(existing?.payload).subscriptionId
+      const subscriptionSync =
+        existing &&
+        isSubscriptionSyncEventType(existing.eventType) &&
+        typeof subscriptionId === 'string'
+          ? { eventType: existing.eventType, subscriptionId }
+          : null
       const result = await db.transaction(async (tx) => {
+        if (subscriptionSync) {
+          await lockSubscriptionForSyncRetry(tx, subscriptionSync.subscriptionId)
+        }
         const requeued = await tx
           .update(outboxEvent)
           .set({
@@ -86,13 +97,12 @@ export const POST = withRouteHandler(
           })
           .where(and(eq(outboxEvent.id, id), eq(outboxEvent.status, 'dead_letter')))
           .returning({ id: outboxEvent.id, eventType: outboxEvent.eventType })
-        const subscriptionId = toRecord(existing?.payload).subscriptionId
-        if (
-          requeued.length > 0 &&
-          isSubscriptionSyncEventType(requeued[0].eventType) &&
-          typeof subscriptionId === 'string'
-        ) {
-          await recommitSubscriptionSync(tx, requeued[0].eventType, subscriptionId)
+        if (subscriptionSync && requeued.length > 0) {
+          await recommitSubscriptionSync(
+            tx,
+            subscriptionSync.eventType,
+            subscriptionSync.subscriptionId
+          )
         }
         return requeued
       })

apps/sim/lib/admin/subscription-lifecycle.test.ts

@@ -103,6 +103,7 @@ describe('admin subscription cancellation', () => {
 
   it('requeues the same dead-lettered period-end cancellation operation', async () => {
     dbChainMockFns.returning.mockResolvedValueOnce([{ id: activeSubscription.id }])
+    queueTableRows(outboxEvent, [{ subscriptionId: 'sub-row-1' }])
     queueTableRows(outboxEvent, [
       {
         id: 'outbox-1',
@@ -124,6 +125,10 @@ describe('admin subscription cancellation', () => {
       expect.objectContaining({ status: 'pending', attempts: 0, lastError: null })
     )
     expect(dbChainMockFns.set).toHaveBeenCalledWith({ cancelAtPeriodEnd: true })
+    expect(billingSubscriptionSyncMockFns.mockLockSubscriptionForSyncRetry).toHaveBeenCalledWith(
+      expect.anything(),
+      'sub-row-1'
+    )
     expect(billingSubscriptionSyncMockFns.mockRecommitSubscriptionSync).toHaveBeenCalledWith(
       expect.anything(),
       'stripe.sync-cancel-at-period-end',
@@ -136,6 +141,7 @@ describe('admin subscription cancellation', () => {
   })
 
   it('replays an immediate cancellation after the webhook removed active entitlement', async () => {
+    queueTableRows(outboxEvent, [])
     queueTableRows(outboxEvent, [
       {
         id: 'outbox-1',
@@ -159,6 +165,7 @@ describe('admin subscription cancellation', () => {
   })
 
   it('rejects reuse of a cancellation operation id with different timing', async () => {
+    queueTableRows(outboxEvent, [])
     queueTableRows(outboxEvent, [
       {
         id: 'outbox-1',

apps/sim/lib/admin/subscription-lifecycle.ts

@@ -10,6 +10,7 @@ import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/util
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import {
   enqueueCancelAtPeriodEndSync,
+  lockSubscriptionForSyncRetry,
   recommitSubscriptionSync,
 } from '@/lib/billing/webhooks/subscription-sync'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
@@ -261,6 +262,24 @@ export async function requestDashboardSubscriptionCancellation({
       : 'admin-dashboard-cancel-at-period-end')
   const cancellation = await db.transaction(async (tx) => {
     await acquireOrganizationMutationLock(tx, organizationId)
+    const isThisOperation = and(
+      sql`${outboxEvent.payload} ->> 'operationId' = ${operationId}`,
+      sql`${outboxEvent.payload} ->> 'organizationId' = ${organizationId}`
+    )
+
+    const [retriedSync] = await tx
+      .select({ subscriptionId: sql<string | null>`${outboxEvent.payload} ->> 'subscriptionId'` })
+      .from(outboxEvent)
+      .where(
+        and(
+          eq(outboxEvent.eventType, OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END),
+          isThisOperation
+        )
+      )
+      .limit(1)
+    if (retriedSync?.subscriptionId) {
+      await lockSubscriptionForSyncRetry(tx, retriedSync.subscriptionId)
+    }
 
     const [existingOperation] = await tx
       .select({
@@ -277,8 +296,7 @@ export async function requestDashboardSubscriptionCancellation({
             OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
             OUTBOX_EVENT_TYPES.STRIPE_CANCEL_SUBSCRIPTION_IMMEDIATELY,
           ]),
-          sql`${outboxEvent.payload} ->> 'operationId' = ${operationId}`,
-          sql`${outboxEvent.payload} ->> 'organizationId' = ${organizationId}`
+          isThisOperation
         )
       )
       .for('update')

apps/sim/lib/billing/enterprise-provisioning.ts

@@ -77,7 +77,10 @@ import { TERMINAL_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/util
 import { countPendingSeatInvitations } from '@/lib/billing/validation/seat-management'
 import { withEnterpriseReconciliationLease } from '@/lib/billing/webhooks/enterprise-reconciliation-lease'
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
-import { recommitSubscriptionSync } from '@/lib/billing/webhooks/subscription-sync'
+import {
+  lockSubscriptionForSyncRetry,
+  recommitSubscriptionSync,
+} from '@/lib/billing/webhooks/subscription-sync'
 import { env } from '@/lib/core/config/env'
 import {
   continueOutboxHandler,
@@ -2104,6 +2107,9 @@ export async function retryEnterpriseFollowUpJob(
 
   const retried = await db.transaction(async (tx) => {
     await acquireOrganizationMutationLock(tx, operationPayload.request.organizationId)
+    if (snapshotDetail.kind === 'personal_subscription_cancellation') {
+      await lockSubscriptionForSyncRetry(tx, snapshotDetail.subjectId)
+    }
     const [row] = await tx
       .select({
         status: outboxEvent.status,
@@ -2120,7 +2126,9 @@ export async function retryEnterpriseFollowUpJob(
       !detail ||
       !getEnterpriseFollowUpOperationIds(row.eventType, row.payload).includes(operationId) ||
       (detail.kind === 'member_reconciliation' &&
-        detail.subjectId !== operationPayload.request.organizationId)
+        detail.subjectId !== operationPayload.request.organizationId) ||
+      detail.kind !== snapshotDetail.kind ||
+      detail.subjectId !== snapshotDetail.subjectId
     ) {
       throw new EnterpriseProvisioningError('Enterprise follow-up job not found')
     }

apps/sim/lib/billing/webhooks/stripe-sync-convergence.integration.ts

@@ -43,6 +43,7 @@ vi.mock('@sim/db', () => ({
 vi.mock('@/lib/billing/stripe-client', () => stripeClientMock)
 vi.mock('@/lib/core/config/env-flags', () => envFlagsMock)
 
+import { requestDashboardSubscriptionCancellation } from '@/lib/admin/subscription-lifecycle'
 import { createSimAuthAdapter } from '@/lib/auth/sim-auth-adapter'
 import {
   pauseProSubscriptionForOrgCoverage,
@@ -132,6 +133,7 @@ beforeAll(async () => {
     'organization',
     'workspace',
     'permissions',
+    'audit_log',
   ]) {
     await connection.unsafe(`CREATE TABLE "${table}" (LIKE public."${table}" INCLUDING ALL)`)
   }
@@ -283,6 +285,18 @@ async function storedSubscription(subscriptionId: string) {
   return row
 }
 
+/** Resolves once another backend is blocked on a lock, i.e. the racing transaction is parked. */
+async function untilAnotherTransactionWaitsOnALock() {
+  for (let attempt = 0; attempt < 200; attempt++) {
+    const [row] = await connection<{ waiting: number }[]>`
+      select count(*)::int as waiting from pg_stat_activity
+      where datname = current_database() and wait_event_type = 'Lock'`
+    if (row.waiting > 0) return
+    await new Promise<void>((resolve) => setImmediate(resolve))
+  }
+  throw new Error('The racing transaction never waited on a lock')
+}
+
 describe('cancel_at_period_end sync', () => {
   it('pushes the latest value when an earlier sync lands in Stripe after a newer one', async () => {
     const pro = await createProUserInPaidOrganization()
@@ -618,18 +632,6 @@ describe('cancel_at_period_end sync', () => {
 })
 
 describe('Team activation', () => {
-  /** Resolves once another backend is blocked on a lock, i.e. the racing transaction is parked. */
-  async function untilAnotherTransactionWaitsOnALock() {
-    for (let attempt = 0; attempt < 200; attempt++) {
-      const [row] = await connection<{ waiting: number }[]>`
-        select count(*)::int as waiting from pg_stat_activity
-        where datname = current_database() and wait_event_type = 'Lock'`
-      if (row.waiting > 0) return
-      await new Promise<void>((resolve) => setImmediate(resolve))
-    }
-    throw new Error('The racing transaction never waited on a lock')
-  }
-
   it('records the cleared cancellation when a cancel is committed while it activates Team', async () => {
     const owner = await createUser('owner')
     const subscriptionId = generateId()
@@ -688,6 +690,89 @@ describe('Team activation', () => {
   })
 })
 
+describe('operator retry', () => {
+  it('requeues a dead-lettered sync while a writer commits a new value for the subscription', async () => {
+    const pro = await createProUserInPaidOrganization()
+    await pauseProSubscriptionForOrgCoverage(pro.userId)
+    const pauseSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      pro.subscriptionId
+    )
+    await deadLetter(pauseSync)
+
+    let releaseWriter: () => void = () => {}
+    const writerHeld = new Promise<void>((resolve) => {
+      releaseWriter = resolve
+    })
+    const writing = testDatabase.transaction(async (tx) => {
+      await tx
+        .update(subscription)
+        .set({ cancelAtPeriodEnd: false })
+        .where(eq(subscription.id, pro.subscriptionId))
+      await writerHeld
+      await enqueueCancelAtPeriodEndSync(tx, {
+        stripeSubscriptionId: pro.stripeSubscriptionId,
+        subscriptionId: pro.subscriptionId,
+        cancelAtPeriodEnd: false,
+        reason: 'member-left-paid-org',
+      })
+    })
+    const requeuing = requeueFromAdminApi(pauseSync)
+    await untilAnotherTransactionWaitsOnALock()
+    releaseWriter()
+
+    await expect(Promise.all([writing, requeuing])).resolves.toBeDefined()
+    await deliverUnrelatedUpdate(pro.stripeSubscriptionId)
+    expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(false)
+  })
+
+  it('retries a dead-lettered dashboard cancellation while a writer commits a new value', async () => {
+    const org = await createOrganizationWithPlan('team')
+    const operationId = generateId()
+    const actor = { id: null, name: 'Admin', email: null }
+    await requestDashboardSubscriptionCancellation({
+      organizationId: org.organizationId,
+      operationId,
+      timing: 'period_end',
+      actor,
+    })
+    const cancelSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      org.subscriptionId
+    )
+    await deadLetter(cancelSync)
+
+    let releaseWriter: () => void = () => {}
+    const writerHeld = new Promise<void>((resolve) => {
+      releaseWriter = resolve
+    })
+    const writing = testDatabase.transaction(async (tx) => {
+      await tx
+        .update(subscription)
+        .set({ cancelAtPeriodEnd: false })
+        .where(eq(subscription.id, org.subscriptionId))
+      await writerHeld
+      await enqueueCancelAtPeriodEndSync(tx, {
+        stripeSubscriptionId: org.stripeSubscriptionId,
+        subscriptionId: org.subscriptionId,
+        cancelAtPeriodEnd: false,
+        reason: 'pro-to-team-conversion',
+      })
+    })
+    const retrying = requestDashboardSubscriptionCancellation({
+      organizationId: org.organizationId,
+      operationId,
+      timing: 'period_end',
+      actor,
+    })
+    await untilAnotherTransactionWaitsOnALock()
+    releaseWriter()
+
+    await expect(Promise.all([writing, retrying])).resolves.toBeDefined()
+    expect((await storedSubscription(org.subscriptionId)).cancelAtPeriodEnd).toBe(true)
+  })
+})
+
 describe('customer contact sync', () => {
   it('pushes the current owner when an earlier sync lands in Stripe after a newer one', async () => {
     const [first, second, third] = await Promise.all([

apps/sim/lib/billing/webhooks/subscription-sync.ts

@@ -9,6 +9,7 @@ import { requireStripeClient } from '@/lib/billing/stripe-client'
 import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import {
   enqueueOutboxEvent,
+  INFLIGHT_OUTBOX_STATUSES,
   listRetryableOutboxEvents,
   patchRetryableOutboxEvents,
 } from '@/lib/core/outbox/service'
@@ -90,7 +91,8 @@ async function withCommittedAt<T extends SyncIntentFields>(
  * onto every event of that type that can still run: pending, processing, or dead-lettered (each
  * operator retry path resets dead letters to pending). No event that can run again ever carries
  * an older value for the webhook reconcile to restore. The caller must hold the subscription row
- * lock (`FOR UPDATE`, or the `UPDATE` itself).
+ * lock (`FOR UPDATE`, or the `UPDATE` itself), per the lock order on
+ * {@link lockSubscriptionForSyncRetry}.
  */
 async function commitIntent<T extends SyncIntentFields>(
   tx: DbOrTx,
@@ -150,9 +152,32 @@ export async function recordCancelAtPeriodEnd(
 }
 
 /**
- * Re-commits the subscription's current DB value onto its in-flight sync events, for a
+ * Takes the subscription row lock for an operator retry of one of its sync events; call it
+ * before touching the event, then {@link recommitSubscriptionSync} after resetting it.
+ *
+ * Lock order for every writer of a subscription's synced fields and their outbox events:
+ * organization mutation lock (where taken) → subscription row → outbox rows. Committing a value
+ * rewrites the subscription's retryable sync events, dead letters included, so a retry that
+ * locked a dead-lettered event before the subscription would deadlock against any concurrent
+ * writer.
+ */
+export async function lockSubscriptionForSyncRetry(
+  tx: DbOrTx,
+  subscriptionId: string
+): Promise<void> {
+  await tx
+    .select({ id: subscription.id })
+    .from(subscription)
+    .where(eq(subscription.id, subscriptionId))
+    .for('update')
+    .limit(1)
+}
+
+/**
+ * Re-commits the subscription's current DB value onto its sync events that can still run, for a
  * dead-lettered event that was just reset to `pending`: the retry then carries the latest value
- * rather than the one it failed with. Takes the subscription row lock itself.
+ * rather than the one it failed with. The caller holds the lock from
+ * {@link lockSubscriptionForSyncRetry}, taken before the reset.
  */
 export async function recommitSubscriptionSync(
   tx: DbOrTx,
@@ -163,7 +188,6 @@ export async function recommitSubscriptionSync(
     .select({ cancelAtPeriodEnd: subscription.cancelAtPeriodEnd, seats: subscription.seats })
     .from(subscription)
     .where(eq(subscription.id, subscriptionId))
-    .for('update')
     .limit(1)
   if (!current) return
 
@@ -229,7 +253,7 @@ export function cancelAtPeriodEndSyncIdempotencyKey(eventId: string): string {
  */
 type InflightIntent<T> = { status: 'none' } | { status: 'legacy' } | { status: 'value'; value: T }
 
-const INFLIGHT_STATUSES = new Set(['pending', 'processing'])
+const INFLIGHT_STATUSES: ReadonlySet<string> = new Set(INFLIGHT_OUTBOX_STATUSES)
 
 function latestIntent<T>(
   events: { eventType: string; status: string; payload: unknown }[],

apps/sim/lib/core/outbox/service.ts

@@ -407,7 +407,7 @@ export async function findDeadLetteredEvents(
 }
 
 /** Statuses of an event whose side effect may still run. */
-const INFLIGHT_OUTBOX_STATUSES = ['pending', 'processing'] as const
+export const INFLIGHT_OUTBOX_STATUSES = ['pending', 'processing'] as const
 /**
  * Statuses an event can still run from: in flight, or dead-lettered, which every operator retry
  * path resets to `pending`. A `completed` event never runs again.

packages/testing/src/mocks/billing-subscription-sync.mock.ts

@@ -17,6 +17,7 @@ import { vi } from 'vitest'
  */
 export const billingSubscriptionSyncMockFns = {
   mockEnqueueCancelAtPeriodEndSync: vi.fn(async () => 'cancel-at-period-end-sync-event'),
+  mockLockSubscriptionForSyncRetry: vi.fn(async () => undefined),
   mockRecommitSubscriptionSync: vi.fn(async () => undefined),
   mockRecordCancelAtPeriodEnd: vi.fn(async () => undefined),
   mockIsSubscriptionSyncEventType: vi.fn(
@@ -42,6 +43,7 @@ export const billingSubscriptionSyncMockFns = {
  */
 export const billingSubscriptionSyncMock = {
   enqueueCancelAtPeriodEndSync: billingSubscriptionSyncMockFns.mockEnqueueCancelAtPeriodEndSync,
+  lockSubscriptionForSyncRetry: billingSubscriptionSyncMockFns.mockLockSubscriptionForSyncRetry,
   recommitSubscriptionSync: billingSubscriptionSyncMockFns.mockRecommitSubscriptionSync,
   recordCancelAtPeriodEnd: billingSubscriptionSyncMockFns.mockRecordCancelAtPeriodEnd,
   isSubscriptionSyncEventType: billingSubscriptionSyncMockFns.mockIsSubscriptionSyncEventType,