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

Author: TheodoreSpeaks

apps/sim/.env.example

@@ -133,4 +133,5 @@ 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 (also _PRO/_TEAM/_ENTERPRISE, default 50; billing disabled uses the highest configured tier)
 # FREE_STORAGE_LIMIT_GB=5               # File storage quota in GB

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/config/env.ts

@@ -103,6 +103,10 @@ 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_PRO:        z.number().optional(),                  // Rows one table run executes in parallel on Pro tier (default: 50)
+    TABLE_DISPATCH_CONCURRENCY_TEAM:       z.number().optional(),                  // Rows one table run executes in parallel on Max/Team tier (default: 50)
+    TABLE_DISPATCH_CONCURRENCY_ENTERPRISE: z.number().optional(),                  // Rows one table run executes in parallel on Enterprise tier (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 plan-bucketed 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 per plan', () => {
+    mockEnv.TABLE_DISPATCH_CONCURRENCY_FREE = '5'
+    mockEnv.TABLE_DISPATCH_CONCURRENCY_ENTERPRISE = '200'
+
+    expect(getTableDispatchConcurrency('free')).toBe(5)
+    expect(getTableDispatchConcurrency('pro_6000')).toBe(50)
+    expect(getTableDispatchConcurrency('enterprise')).toBe(200)
+  })
+
+  it('uses the highest configured tier when billing is disabled', () => {
+    mockFlags.isBillingEnabled = false
+    expect(getTableDispatchConcurrency(null)).toBe(50)
+
+    mockEnv.TABLE_DISPATCH_CONCURRENCY_TEAM = '120'
+    expect(getTableDispatchConcurrency(null)).toBe(120)
+  })
+})
+
+describe('getMaxTableDispatchConcurrency', () => {
+  beforeEach(() => {
+    for (const key of Object.keys(mockEnv)) delete mockEnv[key]
+  })
+
+  it('returns the highest per-plan value', () => {
+    expect(getMaxTableDispatchConcurrency()).toBe(50)
+
+    mockEnv.TABLE_DISPATCH_CONCURRENCY_PRO = '80'
+    expect(getMaxTableDispatchConcurrency()).toBe(80)
+  })
+})

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

@@ -0,0 +1,81 @@
+import { getPlanTypeForLimits, type PlanCategory } from '@/lib/billing/plan-helpers'
+import { env, envNumber } from '@/lib/core/config/env'
+import { isBillingEnabled } from '@/lib/core/config/env-flags'
+
+/**
+ * Default per-plan table dispatch concurrency — how many rows one table run
+ * executes in parallel (the dispatcher window size). Overridable via the
+ * `TABLE_DISPATCH_CONCURRENCY_{FREE,PRO,TEAM,ENTERPRISE}` env vars.
+ */
+export const DEFAULT_TABLE_DISPATCH_CONCURRENCY = {
+  free: 20,
+  pro: 50,
+  team: 50,
+  enterprise: 50,
+} as const satisfies Record<PlanCategory, number>
+
+/**
+ * Resolves per-plan dispatch concurrency, applying env overrides on top of
+ * the defaults.
+ */
+export function getTableDispatchConcurrencyLimits(): Record<PlanCategory, number> {
+  return {
+    free: envNumber(env.TABLE_DISPATCH_CONCURRENCY_FREE, DEFAULT_TABLE_DISPATCH_CONCURRENCY.free, {
+      min: 1,
+      integer: true,
+    }),
+    pro: envNumber(env.TABLE_DISPATCH_CONCURRENCY_PRO, DEFAULT_TABLE_DISPATCH_CONCURRENCY.pro, {
+      min: 1,
+      integer: true,
+    }),
+    team: envNumber(env.TABLE_DISPATCH_CONCURRENCY_TEAM, DEFAULT_TABLE_DISPATCH_CONCURRENCY.team, {
+      min: 1,
+      integer: true,
+    }),
+    enterprise: envNumber(
+      env.TABLE_DISPATCH_CONCURRENCY_ENTERPRISE,
+      DEFAULT_TABLE_DISPATCH_CONCURRENCY.enterprise,
+      { min: 1, integer: true }
+    ),
+  }
+}
+
+/**
+ * Dispatch concurrency for one payer plan. Billing-disabled deployments get
+ * the highest configured tier.
+ */
+export function getTableDispatchConcurrency(plan: string | null | undefined): number {
+  if (!isBillingEnabled) return getMaxTableDispatchConcurrency()
+  return getTableDispatchConcurrencyLimits()[getPlanTypeForLimits(plan)]
+}
+
+/**
+ * Highest configured dispatch concurrency across plans. 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 {
+  return Math.max(...Object.values(getTableDispatchConcurrencyLimits()))
+}
+
+/**
+ * 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 getMaxTableDispatchConcurrency()
+  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'
@@ -85,6 +80,9 @@ export interface DispatchRow {
   limit: DispatchLimit | null
   /** Units of `limit.type` already consumed (eligible rows dispatched). */
   processedCount: number
+  /** Rows executed in parallel per window, resolved from the payer's plan at
+   *  creation. Null on pre-column rows → legacy cap of 20. */
+  concurrency: number | null
   isManualRun: boolean
   /** User who triggered the run (for usage attribution); null for auto-fire. */
   triggeredByUserId: string | null
@@ -190,6 +188,8 @@ export async function insertDispatch(input: {
   mode: DispatchMode
   scope: DispatchScope
   limit?: DispatchLimit | null
+  /** Per-window parallelism from the payer's plan (see `resolveTableDispatchConcurrency`). */
+  concurrency: number
   isManualRun: boolean
   triggeredByUserId?: string | null
 }): Promise<string> {
@@ -202,6 +202,7 @@ export async function insertDispatch(input: {
     mode: input.mode,
     scope: input.scope,
     limit: input.limit ?? null,
+    concurrency: input.concurrency,
     status: 'pending',
     // -1 = "haven't started." First window's filter `position > -1` matches
     // position 0; subsequent iterations advance to `lastPosition` which then
@@ -291,6 +292,7 @@ export async function listActiveDispatches(tableId: string): Promise<DispatchRow
     cursor: row.cursor,
     limit: (row.limit as DispatchLimit | null) ?? null,
     processedCount: row.processedCount,
+    concurrency: row.concurrency,
     isManualRun: row.isManualRun,
     triggeredByUserId: row.triggeredByUserId,
     requestedAt: row.requestedAt,
@@ -315,6 +317,7 @@ export async function readDispatch(dispatchId: string): Promise<DispatchRow | nu
     cursor: row.cursor,
     limit: (row.limit as DispatchLimit | null) ?? null,
     processedCount: row.processedCount,
+    concurrency: row.concurrency,
     isManualRun: row.isManualRun,
     triggeredByUserId: row.triggeredByUserId,
     requestedAt: row.requestedAt,
@@ -380,6 +383,11 @@ export async function dispatcherStep(dispatchId: string): Promise<DispatcherStep
     })
   }
 
+  // Window size = the dispatch's plan-resolved parallelism, so one window
+  // saturates the cell pool before the next is loaded — yields a row-major
+  // scan-line crawl. Pre-column dispatches fall back to the legacy cap.
+  const windowSize = dispatch.concurrency ?? TABLE_CONCURRENCY_LIMIT
+
   const filters = [
     eq(userTableRows.tableId, dispatch.tableId),
     gt(userTableRows.position, dispatch.cursor),
@@ -429,7 +437,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 +541,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)
@@ -792,6 +800,7 @@ export async function markActiveDispatchesCancelled(
     cursor: row.cursor,
     limit: (row.limit as DispatchLimit | null) ?? null,
     processedCount: row.processedCount,
+    concurrency: row.concurrency,
     isManualRun: row.isManualRun,
     triggeredByUserId: row.triggeredByUserId,
     requestedAt: row.requestedAt,

apps/sim/lib/table/workflow-columns.ts

@@ -35,6 +35,7 @@ const logger = createLogger('WorkflowGroupScheduler')
 import { getColumnId } from '@/lib/table/column-keys'
 import { USER_TABLE_ROWS_SQL_NAME } from '@/lib/table/constants'
 import { areGroupDepsSatisfied, areOutputsFilled, isExecInFlight } from '@/lib/table/deps'
+import { resolveTableDispatchConcurrency } from '@/lib/table/dispatch-concurrency'
 import type { DispatchLimit, DispatchMode } from '@/lib/table/dispatcher'
 import { buildFilterClause } from '@/lib/table/sql'
 
@@ -237,7 +238,8 @@ export function buildPendingRuns(
  *  id. The cell-job import pulls in the executor + blocks stack, so skip it on
  *  trigger.dev to avoid a multi-second dispatcher cold-start. */
 export async function buildEnqueueItems(
-  pendingRuns: WorkflowGroupCellPayload[]
+  pendingRuns: WorkflowGroupCellPayload[],
+  concurrencyLimit: number = TABLE_CONCURRENCY_LIMIT
 ): Promise<Array<{ payload: QueuedWorkflowGroupCellPayload; options: EnqueueOptions }>> {
   if (pendingRuns.length === 0) return []
 
@@ -292,7 +294,7 @@ export async function buildEnqueueItems(
         },
       },
       concurrencyKey: runOpts.tableId,
-      concurrencyLimit: TABLE_CONCURRENCY_LIMIT,
+      concurrencyLimit,
       tags: cellTagsFor(runOpts),
       ...(runner ? { runner } : {}),
       cancelKey: cellCancelKey(runOpts.tableId, runOpts.rowId, runOpts.groupId),
@@ -391,7 +393,9 @@ export type QueuedWorkflowGroupCellPayload = Omit<
   billingAttribution: BillingAttributionSnapshot
 }
 
-/** Per-table concurrency cap. Mirrors trigger.dev's `concurrencyLimit: 20`. */
+/** Legacy per-table concurrency cap. The live cap is per-plan (see
+ *  `getTableDispatchConcurrency`); this remains the fallback for dispatch rows
+ *  that predate the `concurrency` column and for non-dispatch cell enqueues. */
 export const TABLE_CONCURRENCY_LIMIT = 20
 
 /**
@@ -745,6 +749,14 @@ export async function runWorkflowColumn(opts: {
     runDispatcherToCompletion,
   } = await import('./dispatcher')
 
+  // Per-window parallelism follows the payer's plan, resolved once here and
+  // persisted on the dispatch row so the dispatcher loop reads a stable value
+  // across checkpointed waits and task retries.
+  const concurrency = await resolveTableDispatchConcurrency({
+    workspaceId,
+    actorUserId: triggeredByUserId,
+  })
+
   // Always insert a `table_run_dispatches` row, and insert it FIRST — before
   // the prior-run cancel and the bulk clear below, which can take seconds on
   // a large table. The client shows its Stop control optimistically from the
@@ -768,6 +780,7 @@ export async function runWorkflowColumn(opts: {
         : {}),
     },
     limit,
+    concurrency,
     isManualRun,
     triggeredByUserId,
   })

packages/db/migrations/0263_brief_warhawk.sql

@@ -0,0 +1 @@
+ALTER TABLE "table_run_dispatches" ADD COLUMN "concurrency" integer;