fix(billing): order Stripe-accepted reconcile values by when Stripe was read

Author: waleedlatif1

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

@@ -618,6 +618,38 @@ describe('cancel_at_period_end sync', () => {
     expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(true)
   })
 
+  it('orders concurrent reconciles of Stripe-side changes by when each read Stripe', async () => {
+    const pro = await createProUserInPaidOrganization()
+    await pauseProSubscriptionForOrgCoverage(pro.userId)
+    const pauseSync = await latestOutboxEventId(
+      OUTBOX_EVENT_TYPES.STRIPE_SYNC_CANCEL_AT_PERIOD_END,
+      pro.subscriptionId
+    )
+    stripe.failNextUpdateAfterApplying('subscriptions')
+    await expect(processEvent(pauseSync)).resolves.toBe('pending')
+
+    stripe.updateOutsideSim(pro.stripeSubscriptionId, { cancel_at_period_end: false })
+    const renewal = stripe.events.at(-1) as Stripe.Event
+    const firstRead = stripe.holdNextRequest('subscriptions.retrieve')
+    const secondRead = stripe.holdNextRequest('subscriptions.retrieve')
+    const reconcilingRenewal = deliver(renewal)
+    await firstRead.reached
+
+    stripe.updateOutsideSim(pro.stripeSubscriptionId, { cancel_at_period_end: true })
+    const reconcilingCancel = deliver(stripe.events.at(-1) as Stripe.Event)
+    await secondRead.reached
+
+    firstRead.release()
+    await reconcilingRenewal
+    secondRead.release()
+    await reconcilingCancel
+
+    expect((await storedSubscription(pro.subscriptionId)).cancelAtPeriodEnd).toBe(true)
+    await makeDue(pauseSync)
+    await expect(processEvent(pauseSync)).resolves.toBe('completed')
+    expect(stripe.subscription(pro.stripeSubscriptionId).cancel_at_period_end).toBe(true)
+  })
+
   it('keeps a value Sim committed after the reconcile read Stripe', async () => {
     const pro = await createProUserInPaidOrganization()
     await pauseProSubscriptionForOrgCoverage(pro.userId)

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

@@ -99,14 +99,22 @@ async function readDatabaseClock(executor: DbOrTx): Promise<number> {
  * an older value for the webhook reconcile to restore. The caller must hold the subscription row
  * lock (`FOR UPDATE`, or the `UPDATE` itself), per the lock order on
  * {@link lockSubscriptionForSyncRetry}.
+ *
+ * A Sim commit is stamped with the clock under that lock. A value taken from Stripe passes
+ * `observedAt`, the clock read just before Stripe was read, so it orders by when it was observed:
+ * a slower reconcile of an earlier Stripe read cannot outrank a later one.
  */
 async function commitIntent<T extends SyncIntentFields>(
   tx: DbOrTx,
   eventType: SubscriptionSyncEventType,
   subscriptionId: string,
-  fields: T
+  fields: T,
+  observedAt?: number
 ): Promise<T & { committedAt: number }> {
-  const committed = await withCommittedAt(tx, fields)
+  const committed =
+    observedAt === undefined
+      ? await withCommittedAt(tx, fields)
+      : { ...fields, committedAt: observedAt }
   await patchRetryableOutboxEvents(tx, eventType, subscriptionSubject(subscriptionId), committed)
   return committed
 }
@@ -446,7 +454,7 @@ export async function reconcileSubscriptionSyncFromStripe(event: Stripe.Event):
         if (liveCancelAtPeriodEnd === undefined) return false
         cancelAtPeriodEnd = liveCancelAtPeriodEnd
         if (intents.cancelSyncCarriesOtherThan(cancelAtPeriodEnd)) {
-          await commitIntent(tx, CANCEL_SYNC, row.id, { cancelAtPeriodEnd })
+          await commitIntent(tx, CANCEL_SYNC, row.id, { cancelAtPeriodEnd }, liveReadAt)
         }
       }
       const seats = intents.seats.status === 'value' ? intents.seats.value : current.seats