diff --git a/apps/sim/.env.example b/apps/sim/.env.example index f46957ed851..1bc29faf0cd 100644 --- a/apps/sim/.env.example +++ b/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 diff --git a/apps/sim/background/table-run-dispatcher.ts b/apps/sim/background/table-run-dispatcher.ts index 441348b8acb..8f8cf16e0ca 100644 --- a/apps/sim/background/table-run-dispatcher.ts +++ b/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 diff --git a/apps/sim/background/workflow-column-execution.ts b/apps/sim/background/workflow-column-execution.ts index 032ffd0643a..20c66c55d37 100644 --- a/apps/sim/background/workflow-column-execution.ts +++ b/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), diff --git a/apps/sim/lib/core/async-jobs/backends/database.test.ts b/apps/sim/lib/core/async-jobs/backends/database.test.ts index c30ee727ffc..1d7207031bc 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.test.ts +++ b/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) + }) +}) diff --git a/apps/sim/lib/core/async-jobs/backends/database.ts b/apps/sim/lib/core/async-jobs/backends/database.ts index 8685a4fe7e8..45feb6ed412 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.ts +++ b/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, diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index 460973b1baa..c82870975c5 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/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) diff --git a/apps/sim/lib/table/dispatch-concurrency.test.ts b/apps/sim/lib/table/dispatch-concurrency.test.ts new file mode 100644 index 00000000000..2581b6e987b --- /dev/null +++ b/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, + 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) + }) +}) diff --git a/apps/sim/lib/table/dispatch-concurrency.ts b/apps/sim/lib/table/dispatch-concurrency.ts new file mode 100644 index 00000000000..dd73b08c4a7 --- /dev/null +++ b/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 { + 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) +} diff --git a/apps/sim/lib/table/dispatcher.ts b/apps/sim/lib/table/dispatcher.ts index 44de520a68d..2206b03c34a 100644 --- a/apps/sim/lib/table/dispatcher.ts +++ b/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 { - 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 { + 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 { +export async function dispatcherStep( + dispatchId: string, + concurrency?: number +): Promise { 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> { 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,13 @@ export async function runWorkflowColumn(opts: { runDispatcherToCompletion, } = await import('./dispatcher') + // Per-window parallelism follows the invoker's plan, resolved once here and + // threaded through the dispatcher invocation (task payload / loop arg). + 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 @@ -871,14 +882,14 @@ export async function runWorkflowColumn(opts: { ]) await tasks.trigger( 'table-run-dispatcher', - { dispatchId }, + { dispatchId, concurrency }, { concurrencyKey: dispatchId, region: await resolveTriggerRegion() } ) } else { // Local / no-trigger.dev: drive the same loop in-process, fire-and-forget // so the HTTP request returns instantly (mirrors the trigger.dev path's // async fan-out). - void runDispatcherToCompletion(dispatchId).catch((err) => + void runDispatcherToCompletion(dispatchId, concurrency).catch((err) => logger.error(`[${requestId}] dispatcher loop failed`, { dispatchId, error: toError(err).message,