fix(db-part-1): eliminate pool self-deadlock from nested checkouts inside transactions

Author: icecrasher321

apps/realtime/src/database/operations.ts

@@ -1,6 +1,12 @@
 import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
 import * as schema from '@sim/db'
-import { workflow, workflowBlocks, workflowEdges, workflowSubflows } from '@sim/db'
+import {
+  instrumentPoolClient,
+  workflow,
+  workflowBlocks,
+  workflowEdges,
+  workflowSubflows,
+} from '@sim/db'
 import { createLogger } from '@sim/logger'
 import {
   BLOCK_OPERATIONS,
@@ -27,13 +33,16 @@ const logger = createLogger('SocketDatabase')
 
 const connectionString = env.DATABASE_URL
 const socketDb = drizzle(
-  postgres(connectionString, {
-    prepare: false,
-    idle_timeout: 10,
-    connect_timeout: 20,
-    max: 15,
-    onnotice: () => {},
-  }),
+  instrumentPoolClient(
+    postgres(connectionString, {
+      prepare: false,
+      idle_timeout: 10,
+      connect_timeout: 20,
+      max: 15,
+      onnotice: () => {},
+    }),
+    'socketDb'
+  ),
   { schema }
 )
 

apps/sim/app/api/credential-sets/[id]/members/route.ts

@@ -185,16 +185,24 @@ export const DELETE = withRouteHandler(
 
       const requestId = generateId().slice(0, 8)
 
-      // Use transaction to ensure member deletion + webhook sync are atomic
-      await db.transaction(async (tx) => {
-        await tx.delete(credentialSetMember).where(eq(credentialSetMember.id, memberId))
-
-        const syncResult = await syncAllWebhooksForCredentialSet(id, requestId, tx)
+      await db.delete(credentialSetMember).where(eq(credentialSetMember.id, memberId))
+
+      // Runs after the deletion commits: the sync performs external HTTP
+      // (OAuth refresh, provider unsubscribe) and must not hold a pooled
+      // connection. A sync failure must not fail the committed mutation —
+      // it self-heals on the next membership change/deploy.
+      try {
+        const syncResult = await syncAllWebhooksForCredentialSet(id, requestId)
         logger.info('Synced webhooks after member removed', {
           credentialSetId: id,
           ...syncResult,
         })
-      })
+      } catch (syncError) {
+        logger.error('Webhook sync failed after member removal', {
+          credentialSetId: id,
+          error: syncError,
+        })
+      }
 
       logger.info('Removed member from credential set', {
         credentialSetId: id,

apps/sim/app/api/credential-sets/invite/[token]/route.ts

@@ -194,17 +194,27 @@ export const POST = withRouteHandler(
               )
             )
         }
+      })
 
+      // Runs after the membership commits: the sync performs external HTTP
+      // (OAuth refresh, provider unsubscribe) and must not hold a pooled
+      // connection. A sync failure must not fail the committed mutation —
+      // it self-heals on the next membership change/deploy.
+      try {
         const syncResult = await syncAllWebhooksForCredentialSet(
           invitation.credentialSetId,
-          requestId,
-          tx
+          requestId
         )
         logger.info('Synced webhooks after member joined', {
           credentialSetId: invitation.credentialSetId,
           ...syncResult,
         })
-      })
+      } catch (syncError) {
+        logger.error('Webhook sync failed after invitation accept', {
+          credentialSetId: invitation.credentialSetId,
+          error: syncError,
+        })
+      }
 
       logger.info('Accepted credential set invitation', {
         invitationId: invitation.id,

apps/sim/app/api/credential-sets/memberships/route.ts

@@ -74,7 +74,6 @@ export const DELETE = withRouteHandler(async (req: NextRequest) => {
   try {
     const requestId = generateId().slice(0, 8)
 
-    // Use transaction to ensure revocation + webhook sync are atomic
     await db.transaction(async (tx) => {
       // Find and verify membership
       const [membership] = await tx
@@ -104,15 +103,26 @@ export const DELETE = withRouteHandler(async (req: NextRequest) => {
           updatedAt: new Date(),
         })
         .where(eq(credentialSetMember.id, membership.id))
+    })
 
-      // Sync webhooks to remove this user's credential webhooks
-      const syncResult = await syncAllWebhooksForCredentialSet(credentialSetId, requestId, tx)
+    // Runs after the revocation commits: the sync performs external HTTP
+    // (OAuth refresh, provider unsubscribe) and must not hold a pooled
+    // connection. A sync failure must not fail the committed mutation —
+    // it self-heals on the next membership change/deploy.
+    try {
+      const syncResult = await syncAllWebhooksForCredentialSet(credentialSetId, requestId)
       logger.info('Synced webhooks after member left', {
         credentialSetId,
         userId: session.user.id,
         ...syncResult,
       })
-    })
+    } catch (syncError) {
+      logger.error('Webhook sync failed after member left', {
+        credentialSetId,
+        userId: session.user.id,
+        error: syncError,
+      })
+    }
 
     logger.info('User left credential set', {
       credentialSetId,

apps/sim/app/api/v1/admin/workspaces/[id]/members/route.ts

@@ -47,6 +47,7 @@ import {
   adminV1ListWorkspaceMembersContract,
 } from '@/lib/api/contracts/v1/admin'
 import { parseRequest } from '@/lib/api/server'
+import { isWorkspaceOnEnterprisePlan } from '@/lib/billing/core/subscription'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { revokeWorkspaceCredentialMembershipsTx } from '@/lib/credentials/access'
 import { syncWorkspaceEnvCredentials } from '@/lib/credentials/environment'
@@ -247,7 +248,12 @@ export const POST = withRouteHandler(
         updatedAt: now,
       })
 
-      await applyWorkspaceAutoAddGroup(db, workspaceId, userId)
+      await applyWorkspaceAutoAddGroup(
+        db,
+        workspaceId,
+        userId,
+        await isWorkspaceOnEnterprisePlan(workspaceId)
+      )
 
       logger.info(`Admin API: Added user ${userId} to workspace ${workspaceId}`, {
         permissions: permissionLevel,

apps/sim/app/api/workspaces/[id]/permissions/route.ts

@@ -8,6 +8,7 @@ import { type NextRequest, NextResponse } from 'next/server'
 import { updateWorkspacePermissionsContract } from '@/lib/api/contracts/workspaces'
 import { parseRequest } from '@/lib/api/server'
 import { getSession } from '@/lib/auth'
+import { isWorkspaceOnEnterprisePlan } from '@/lib/billing/core/subscription'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { syncWorkspaceEnvCredentials } from '@/lib/credentials/environment'
 import { applyWorkspaceAutoAddGroup } from '@/lib/permission-groups/auto-add'
@@ -159,6 +160,12 @@ export const PATCH = withRouteHandler(
         existingPerms.map((p) => [p.userId, { permission: p.permissionType, email: p.email }])
       )
 
+      // Resolved before the transaction: the entitlement check reads billing
+      // tables on the global pool and must not run while the tx holds a
+      // pooled connection.
+      const hasNewMembers = body.updates.some((update) => !permLookup.has(update.userId))
+      const autoAddEntitled = hasNewMembers ? await isWorkspaceOnEnterprisePlan(workspaceId) : false
+
       await db.transaction(async (tx) => {
         for (const update of body.updates) {
           const isNew = !permLookup.has(update.userId)
@@ -184,7 +191,7 @@ export const PATCH = withRouteHandler(
           })
 
           if (isNew) {
-            await applyWorkspaceAutoAddGroup(tx, workspaceId, update.userId)
+            await applyWorkspaceAutoAddGroup(tx, workspaceId, update.userId, autoAddEntitled)
           }
         }
       })

apps/sim/lib/billing/core/usage-log.ts

@@ -77,11 +77,7 @@ interface UsageEntry {
   metadata?: UsageLogMetadata
 }
 
-/**
- * Parameters for the central recordUsage function.
- * This is the single entry point for all billing mutations.
- */
-export interface RecordUsageParams {
+interface RecordUsageBaseParams {
   /** The user being charged */
   userId: string
   /** One or more usage_log entries to record. Total cost is derived from these. */
@@ -92,19 +88,37 @@ export interface RecordUsageParams {
   workflowId?: string
   /** Execution context */
   executionId?: string
-  /** Billing entity scope, resolved by caller when already known. */
-  billingEntity?: BillingEntity
-  /** Billing period bounds, resolved by caller when already known. */
-  billingPeriod?: { start: Date; end: Date }
-  /**
-   * Optional transaction to run the ledger INSERT in. Callers that reconcile a
-   * read-then-insert under a lock (e.g. the per-execution advisory lock in the
-   * workflow completion path) pass their tx so the insert participates in the
-   * same locked transaction. Defaults to the pooled db.
-   */
-  tx?: DbOrTx
 }
 
+/**
+ * Parameters for the central recordUsage function.
+ * This is the single entry point for all billing mutations.
+ *
+ * Callers that pass `tx` (e.g. the per-execution advisory-lock reconciliation
+ * in the workflow completion path) must pre-resolve the billing context before
+ * opening the transaction: resolving it inside would run the subscription
+ * lookups on the global pool while the tx already holds a pooled connection,
+ * starving the pool under load (see recordCumulativeUsage for the history).
+ */
+export type RecordUsageParams = RecordUsageBaseParams &
+  (
+    | {
+        /** Transaction the ledger INSERT participates in. */
+        tx: DbOrTx
+        /** Billing entity scope, resolved before the transaction opened. */
+        billingEntity: BillingEntity
+        /** Billing period bounds, resolved before the transaction opened. */
+        billingPeriod: { start: Date; end: Date }
+      }
+    | {
+        tx?: undefined
+        /** Billing entity scope, resolved by caller when already known. */
+        billingEntity?: BillingEntity
+        /** Billing period bounds, resolved by caller when already known. */
+        billingPeriod?: { start: Date; end: Date }
+      }
+  )
+
 export function stableEventKey(parts: Record<string, unknown>): string {
   const payload = Object.keys(parts)
     .sort()

apps/sim/lib/invitations/core.test.ts

@@ -15,6 +15,7 @@ const {
   mockSyncUsageLimitsFromSubscription,
   mockSyncWorkspaceEnvCredentials,
   mockApplyWorkspaceAutoAddGroup,
+  mockIsWorkspaceOnEnterprisePlan,
   mockFeatureFlags,
 } = vi.hoisted(() => ({
   mockEnsureUserInOrganization: vi.fn(),
@@ -27,6 +28,7 @@ const {
   mockSyncUsageLimitsFromSubscription: vi.fn(),
   mockSyncWorkspaceEnvCredentials: vi.fn(),
   mockApplyWorkspaceAutoAddGroup: vi.fn(),
+  mockIsWorkspaceOnEnterprisePlan: vi.fn(async () => true),
   mockFeatureFlags: { isBillingEnabled: true },
 }))
 
@@ -60,6 +62,10 @@ vi.mock('@/lib/auth/active-organization', () => ({
   setActiveOrganizationForCurrentSession: mockSetActiveOrganizationForCurrentSession,
 }))
 
+vi.mock('@/lib/billing/core/subscription', () => ({
+  isWorkspaceOnEnterprisePlan: mockIsWorkspaceOnEnterprisePlan,
+}))
+
 vi.mock('@/lib/billing/core/usage', () => ({
   syncUsageLimitsFromSubscription: mockSyncUsageLimitsFromSubscription,
 }))

apps/sim/lib/invitations/core.ts

@@ -17,6 +17,7 @@ import { createLogger } from '@sim/logger'
 import { generateId } from '@sim/utils/id'
 import { and, eq, inArray, lte } from 'drizzle-orm'
 import { setActiveOrganizationForCurrentSession } from '@/lib/auth/active-organization'
+import { isWorkspaceOnEnterprisePlan } from '@/lib/billing/core/subscription'
 import { syncUsageLimitsFromSubscription } from '@/lib/billing/core/usage'
 import {
   acquireOrgMembershipLock,
@@ -433,6 +434,14 @@ export async function acceptInvitation(
 
   const acceptedWorkspaceIds: string[] = []
 
+  // Resolved before the transaction: the entitlement check reads billing
+  // tables on the global pool, which must not run while the tx below holds a
+  // pooled connection and the org-membership advisory lock.
+  const autoAddEntitlementByWorkspace = new Map<string, boolean>()
+  for (const workspaceId of new Set(inv.grants.map((grant) => grant.workspaceId))) {
+    autoAddEntitlementByWorkspace.set(workspaceId, await isWorkspaceOnEnterprisePlan(workspaceId))
+  }
+
   try {
     await db.transaction(async (tx) => {
       /**
@@ -502,7 +511,12 @@ export async function acceptInvitation(
           })
         }
 
-        await applyWorkspaceAutoAddGroup(tx, grant.workspaceId, input.userId)
+        await applyWorkspaceAutoAddGroup(
+          tx,
+          grant.workspaceId,
+          input.userId,
+          autoAddEntitlementByWorkspace.get(grant.workspaceId) ?? false
+        )
 
         acceptedWorkspaceIds.push(grant.workspaceId)
       }