feat(table): per-plan table dispatch concurrency with env overrides (#5720)

* feat(table): per-plan table dispatch concurrency with env overrides

* fix(table): enforce shared concurrencyKey cap on database batchEnqueueAndWait

* refactor(table): thread dispatch concurrency via invocation instead of persisting it

* improvement(table): collapse dispatch concurrency env vars to FREE/PAID

Author: TheodoreSpeaks

apps/sim/.env.example

@@ -133,4 +133,6 @@ API_ENCRYPTION_KEY=your_api_encryption_key # Use `openssl rand -hex 32` to gener
 # EXECUTION_TIMEOUT_ASYNC_FREE=5400     # Async execution timeout in seconds
 # FREE_TABLES_LIMIT=5                   # Max user tables per workspace
 # FREE_TABLE_ROWS_LIMIT=50000           # Max rows per user table
+# TABLE_DISPATCH_CONCURRENCY_FREE=20    # Rows one table run executes in parallel on free tier
+# TABLE_DISPATCH_CONCURRENCY_PAID=50    # Rows one table run executes in parallel on paid tiers (billing disabled uses this)
 # FREE_STORAGE_LIMIT_GB=5               # File storage quota in GB

apps/sim/background/table-run-dispatcher.ts

@@ -7,6 +7,9 @@ const logger = createLogger('TableRunDispatcherTask')
 
 export interface TableRunDispatcherPayload {
   dispatchId: string
+  /** Invoker's plan-resolved window size. Absent on payloads from before the
+   *  field existed → dispatcher falls back to the legacy cap. */
+  concurrency?: number
 }
 
 /**
@@ -26,9 +29,9 @@ export const tableRunDispatcherTask = task({
     concurrencyLimit: 8,
   },
   run: async (payload: TableRunDispatcherPayload) => {
-    const { dispatchId } = payload
+    const { dispatchId, concurrency } = payload
     try {
-      await runDispatcherToCompletion(dispatchId)
+      await runDispatcherToCompletion(dispatchId, concurrency)
     } catch (err) {
       logger.error(`[${dispatchId}] dispatcher loop failed`, { error: toError(err).message })
       throw err

apps/sim/background/workflow-column-execution.ts

@@ -22,6 +22,7 @@ import { retryTableAdmission } from '@/lib/table/admission-retry'
 import { withCascadeLock } from '@/lib/table/cascade-lock'
 import { getColumnId } from '@/lib/table/column-keys'
 import { isExecCancelled } from '@/lib/table/deps'
+import { getMaxTableDispatchConcurrency } from '@/lib/table/dispatch-concurrency'
 import { appendTableEvent } from '@/lib/table/events'
 import type {
   RowData,
@@ -852,11 +853,15 @@ export const workflowGroupCellTask = task({
   id: 'workflow-group-cell',
   machine: 'medium-1x',
   retry: { maxAttempts: 1 },
-  // Combined with `concurrencyKey: tableId`, caps each table's sub-queue to
-  // 20 in-flight cell jobs while letting different tables run in parallel.
+  // Combined with `concurrencyKey: tableId`, caps each table's sub-queue of
+  // in-flight cell jobs while letting different tables run in parallel. The
+  // cap is the highest per-plan dispatch window so the queue never throttles
+  // below a plan's window — the dispatcher window is the real per-run limiter.
+  // Read at trigger.dev deploy time: raising a TABLE_DISPATCH_CONCURRENCY_*
+  // env var above the current max needs a trigger.dev redeploy to take effect.
   queue: {
     name: 'workflow-group-cell',
-    concurrencyLimit: 20,
+    concurrencyLimit: getMaxTableDispatchConcurrency(),
   },
   run: (payload: QueuedWorkflowGroupCellPayload, { signal }) =>
     executeWorkflowGroupCellJob(payload, signal),

apps/sim/lib/core/async-jobs/backends/database.test.ts

@@ -2,6 +2,7 @@
  * @vitest-environment node
  */
 import { dbChainMock, dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import { sleep } from '@sim/utils/helpers'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 vi.mock('@sim/db', () => ({
@@ -78,3 +79,36 @@ describe('DatabaseJobQueue enqueue', () => {
     })
   })
 })
+
+describe('DatabaseJobQueue batchEnqueueAndWait', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('caps overlapping batches sharing a concurrencyKey at the shared limit', async () => {
+    const queue = new DatabaseJobQueue()
+    let inFlight = 0
+    let maxInFlight = 0
+    const makeItem = () => ({
+      payload: {},
+      options: {
+        concurrencyKey: 'table-1',
+        concurrencyLimit: 2,
+        runner: async () => {
+          inFlight += 1
+          maxInFlight = Math.max(maxInFlight, inFlight)
+          await sleep(1)
+          inFlight -= 1
+        },
+      },
+    })
+
+    await Promise.all([
+      queue.batchEnqueueAndWait('workflow-group-cell', [makeItem(), makeItem()]),
+      queue.batchEnqueueAndWait('workflow-group-cell', [makeItem(), makeItem()]),
+    ])
+
+    expect(maxInFlight).toBe(2)
+  })
+})

apps/sim/lib/core/async-jobs/backends/database.ts

@@ -207,7 +207,23 @@ export class DatabaseJobQueue implements JobQueueBackend {
         inlineCancelKeyControllers.set(cancelKey, controller)
         tracked.push({ key: cancelKey, controller })
       }
-      return runner(item.payload, controller.signal).catch((err) => {
+      // Same shared-key semaphore as `runInline`: without it, overlapping
+      // batches on one concurrencyKey (e.g. two dispatches on one table) would
+      // each run their full window concurrently instead of sharing the cap.
+      const { concurrencyKey, concurrencyLimit } = item.options ?? {}
+      const run = async () => {
+        if (concurrencyKey && concurrencyLimit && concurrencyLimit > 0) {
+          await acquireSlot(concurrencyKey, concurrencyLimit)
+          try {
+            await runner(item.payload, controller.signal)
+          } finally {
+            releaseSlot(concurrencyKey)
+          }
+          return
+        }
+        await runner(item.payload, controller.signal)
+      }
+      return run().catch((err) => {
         logger.error(`[${type}] Inline run failed`, {
           cancelKey,
           error: toError(err).message,

apps/sim/lib/core/config/env.ts

@@ -103,6 +103,8 @@ export const env = createEnv({
     ENTERPRISE_TABLE_ROWS_LIMIT:           z.number().optional(),                  // Max rows per table on enterprise tier (default: 1000000)
     TABLE_MAX_ROW_SIZE_BYTES:              z.number().optional(),                  // Max serialized size in bytes of a single user-table row (default: 409600)
     TABLE_MAX_PAGE_BYTES:                  z.number().optional(),                  // Dev-preview: byte budget per row-page read; pages cut early past it (unset = disabled)
+    TABLE_DISPATCH_CONCURRENCY_FREE:       z.number().optional(),                  // Rows one table run executes in parallel on free tier (default: 20)
+    TABLE_DISPATCH_CONCURRENCY_PAID:       z.number().optional(),                  // Rows one table run executes in parallel on paid tiers (default: 50)
 
     // Credit-tier Stripe prices (monthly)
     STRIPE_PRICE_TIER_25_MO:               z.string().min(1).optional(),           // Pro: $25/mo (6,000 credits)

apps/sim/lib/table/dispatch-concurrency.test.ts

@@ -0,0 +1,82 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockEnv, mockFlags } = vi.hoisted(() => ({
+  mockEnv: {} as Record<string, string | undefined>,
+  mockFlags: { isBillingEnabled: true },
+}))
+
+vi.mock('@/lib/core/config/env', () => ({
+  env: mockEnv,
+  envNumber: (
+    value: number | string | undefined | null,
+    fallback: number,
+    options: { min?: number; integer?: boolean } = {}
+  ) => {
+    const parsed = Number(value)
+    const min = options.min ?? 0
+    return Number.isFinite(parsed) &&
+      parsed >= min &&
+      (!options.integer || Number.isInteger(parsed))
+      ? parsed
+      : fallback
+  },
+}))
+
+vi.mock('@/lib/core/config/env-flags', () => ({
+  get isBillingEnabled() {
+    return mockFlags.isBillingEnabled
+  },
+}))
+
+import {
+  getMaxTableDispatchConcurrency,
+  getTableDispatchConcurrency,
+} from '@/lib/table/dispatch-concurrency'
+
+describe('getTableDispatchConcurrency', () => {
+  beforeEach(() => {
+    for (const key of Object.keys(mockEnv)) delete mockEnv[key]
+    mockFlags.isBillingEnabled = true
+  })
+
+  it('resolves free vs paid defaults', () => {
+    expect(getTableDispatchConcurrency(null)).toBe(20)
+    expect(getTableDispatchConcurrency('free')).toBe(20)
+    expect(getTableDispatchConcurrency('pro_6000')).toBe(50)
+    expect(getTableDispatchConcurrency('team_25000')).toBe(50)
+    expect(getTableDispatchConcurrency('enterprise')).toBe(50)
+  })
+
+  it('applies env overrides', () => {
+    mockEnv.TABLE_DISPATCH_CONCURRENCY_FREE = '5'
+    mockEnv.TABLE_DISPATCH_CONCURRENCY_PAID = '200'
+
+    expect(getTableDispatchConcurrency('free')).toBe(5)
+    expect(getTableDispatchConcurrency('pro_6000')).toBe(200)
+    expect(getTableDispatchConcurrency('enterprise')).toBe(200)
+  })
+
+  it('uses the paid value when billing is disabled', () => {
+    mockFlags.isBillingEnabled = false
+    expect(getTableDispatchConcurrency(null)).toBe(50)
+
+    mockEnv.TABLE_DISPATCH_CONCURRENCY_PAID = '120'
+    expect(getTableDispatchConcurrency(null)).toBe(120)
+  })
+})
+
+describe('getMaxTableDispatchConcurrency', () => {
+  beforeEach(() => {
+    for (const key of Object.keys(mockEnv)) delete mockEnv[key]
+  })
+
+  it('returns the highest configured value', () => {
+    expect(getMaxTableDispatchConcurrency()).toBe(50)
+
+    mockEnv.TABLE_DISPATCH_CONCURRENCY_FREE = '80'
+    expect(getMaxTableDispatchConcurrency()).toBe(80)
+  })
+})

apps/sim/lib/table/dispatch-concurrency.ts

@@ -0,0 +1,72 @@
+import { getPlanTypeForLimits } from '@/lib/billing/plan-helpers'
+import { env, envNumber } from '@/lib/core/config/env'
+import { isBillingEnabled } from '@/lib/core/config/env-flags'
+
+/**
+ * Default table dispatch concurrency — how many rows one table run executes
+ * in parallel (the dispatcher window size). Free vs paid (Pro, Max,
+ * Enterprise); overridable via `TABLE_DISPATCH_CONCURRENCY_{FREE,PAID}`.
+ */
+export const DEFAULT_TABLE_DISPATCH_CONCURRENCY = {
+  free: 20,
+  paid: 50,
+} as const
+
+/**
+ * Resolves dispatch concurrency limits, applying env overrides on top of the
+ * defaults.
+ */
+export function getTableDispatchConcurrencyLimits(): { free: number; paid: number } {
+  return {
+    free: envNumber(env.TABLE_DISPATCH_CONCURRENCY_FREE, DEFAULT_TABLE_DISPATCH_CONCURRENCY.free, {
+      min: 1,
+      integer: true,
+    }),
+    paid: envNumber(env.TABLE_DISPATCH_CONCURRENCY_PAID, DEFAULT_TABLE_DISPATCH_CONCURRENCY.paid, {
+      min: 1,
+      integer: true,
+    }),
+  }
+}
+
+/**
+ * Dispatch concurrency for one payer plan. Billing-disabled deployments get
+ * the paid value.
+ */
+export function getTableDispatchConcurrency(plan: string | null | undefined): number {
+  const limits = getTableDispatchConcurrencyLimits()
+  if (!isBillingEnabled) return limits.paid
+  return getPlanTypeForLimits(plan) === 'free' ? limits.free : limits.paid
+}
+
+/**
+ * Highest configured dispatch concurrency. The `workflow-group-cell`
+ * trigger.dev queue cap derives from this so the server-side per-table
+ * ceiling never throttles below a plan's window.
+ */
+export function getMaxTableDispatchConcurrency(): number {
+  const limits = getTableDispatchConcurrencyLimits()
+  return Math.max(limits.free, limits.paid)
+}
+
+/**
+ * Resolves the workspace payer's plan and returns its dispatch concurrency.
+ * Uses the same billing attribution the cells are billed under, so the window
+ * follows whoever pays for the run.
+ */
+export async function resolveTableDispatchConcurrency(input: {
+  workspaceId: string
+  actorUserId?: string | null
+}): Promise<number> {
+  if (!isBillingEnabled) return getTableDispatchConcurrencyLimits().paid
+  const { resolveBillingAttribution, resolveSystemBillingAttribution } = await import(
+    '@/lib/billing/core/billing-attribution'
+  )
+  const attribution = input.actorUserId
+    ? await resolveBillingAttribution({
+        actorUserId: input.actorUserId,
+        workspaceId: input.workspaceId,
+      })
+    : await resolveSystemBillingAttribution(input.workspaceId)
+  return getTableDispatchConcurrency(attribution.payerSubscription?.plan)
+}

apps/sim/lib/table/dispatcher.ts

@@ -40,11 +40,6 @@ import {
 
 const logger = createLogger('TableRunDispatcher')
 
-/** Window size matches the cell-execution concurrency cap so one window
- *  saturates the pool before the next is loaded — yields a row-major
- *  scan-line crawl (rows 1-20 finish before 21-40 start). */
-const WINDOW_SIZE = TABLE_CONCURRENCY_LIMIT
-
 const ACTIVE_DISPATCH_STATUSES = ['pending', 'dispatching'] as const
 
 export type DispatchStatus = 'pending' | 'dispatching' | 'complete' | 'cancelled'
@@ -323,14 +318,23 @@ export async function readDispatch(dispatchId: string): Promise<DispatchRow | nu
 
 /** Drive `dispatcherStep` to completion. Shared between the trigger.dev task
  *  wrapper (`tableRunDispatcherTask`) and the in-process inline path so both
- *  runtimes use identical loop semantics + error logging. */
-export async function runDispatcherToCompletion(dispatchId: string): Promise<void> {
-  while ((await dispatcherStep(dispatchId)) === 'continue') {}
+ *  runtimes use identical loop semantics + error logging. `concurrency` is the
+ *  invoker's plan-resolved window size (see `resolveTableDispatchConcurrency`),
+ *  threaded via the task payload; absent on payloads from before the field
+ *  existed → legacy cap. */
+export async function runDispatcherToCompletion(
+  dispatchId: string,
+  concurrency?: number
+): Promise<void> {
+  while ((await dispatcherStep(dispatchId, concurrency)) === 'continue') {}
 }
 
 /** Run one window of the dispatcher state machine. Caller re-invokes (via the
  *  trigger.dev task wrapper) until the returned status is `'done'`. */
-export async function dispatcherStep(dispatchId: string): Promise<DispatcherStepResult> {
+export async function dispatcherStep(
+  dispatchId: string,
+  concurrency?: number
+): Promise<DispatcherStepResult> {
   const dispatch = await readDispatch(dispatchId)
   if (!dispatch) {
     logger.warn(`[${dispatchId}] dispatch row missing — aborting`)
@@ -380,6 +384,11 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
     })
   }
 
+  // Window size = the invoker's plan-resolved parallelism, so one window
+  // saturates the cell pool before the next is loaded — yields a row-major
+  // scan-line crawl. Payloads without the field fall back to the legacy cap.
+  const windowSize = concurrency ?? TABLE_CONCURRENCY_LIMIT
+
   const filters = [
     eq(userTableRows.tableId, dispatch.tableId),
     gt(userTableRows.position, dispatch.cursor),
@@ -429,7 +438,7 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
       .from(userTableRows)
       .where(and(...filters))
       .orderBy(asc(userTableRows.position))
-      .limit(WINDOW_SIZE)
+      .limit(windowSize)
   // Filtered scopes carry a jsonb predicate the planner can't estimate — left alone it
   // seq-scans the whole shared relation per window; keep it on the tenant's position index.
   const chunk = hasJsonbFilter
@@ -533,8 +542,8 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
     // (CRIU-checkpointed wait); database backend calls the cell-task runner
     // directly via Promise.all (skips async_jobs since we're awaiting in-
     // process anyway). Either way the parent dispatcher blocks until every
-    // cell in the window terminates — bounds queue depth at WINDOW_SIZE.
-    const items = await buildEnqueueItems(windowRuns)
+    // cell in the window terminates — bounds queue depth at the window size.
+    const items = await buildEnqueueItems(windowRuns, windowSize)
     const queue = await getJobQueue()
     try {
       await queue.batchEnqueueAndWait('workflow-group-cell', items)