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

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)