diff --git a/apps/sim/lib/logs/application/read-execution-snapshot.test.ts b/apps/sim/lib/logs/application/read-execution-snapshot.test.ts index 5c173e4f59d..625896c5a40 100644 --- a/apps/sim/lib/logs/application/read-execution-snapshot.test.ts +++ b/apps/sim/lib/logs/application/read-execution-snapshot.test.ts @@ -66,6 +66,13 @@ const workspaceContext = { billedAccountUserId: 'billing-owner-1', } +const completedRun = { + executionData: null, + endedAt: new Date('2026-08-05T12:00:01.000Z'), + totalDurationMs: 1000, + costTotal: '0.75', +} + const workflowRecord = { id: 'log-1', workflowId: 'workflow-1', @@ -77,17 +84,19 @@ const workflowRecord = { endedAt: new Date('2026-08-05T12:00:01.000Z'), totalDurationMs: 1000, costTotal: '0.75', - executionData: null, } /** * Answers `db.select(...)` calls in order. The snapshot read walks the workflow - * log, then (only when that missed) the job log, then the state snapshot. + * log, then (only when that missed) the job log, then the state snapshot, then + * the run's execution data. */ function queueSelects(...results: unknown[][]): void { for (const rows of results) { mocks.select.mockReturnValueOnce({ - from: () => ({ where: () => ({ limit: () => Promise.resolve(rows) }) }), + from: () => ({ + where: () => Object.assign(Promise.resolve(rows), { limit: () => Promise.resolve(rows) }), + }), }) } } @@ -135,7 +144,11 @@ describe('readExecutionSnapshot spend projection', () => { ...DEFAULT_PERMISSION_GROUP_CONFIG, hideCostInfo: true, }) - queueSelects([workflowRecord], [{ id: 'snapshot-1', stateData: { blocks: {} } }]) + queueSelects( + [workflowRecord], + [{ id: 'snapshot-1', stateData: { blocks: {} } }], + [completedRun] + ) const result = await read(principal, 'run-1') @@ -153,11 +166,31 @@ describe('readExecutionSnapshot spend projection', () => { ...DEFAULT_PERMISSION_GROUP_CONFIG, hideCostInfo: true, }) - queueSelects([workflowRecord], [{ id: 'snapshot-1', stateData: { blocks: {} } }]) + queueSelects( + [workflowRecord], + [{ id: 'snapshot-1', stateData: { blocks: {} } }], + [completedRun] + ) const result = await read(executorPrincipal, 'run-1') expect(result.executionMetadata.cost).toEqual({ total: 0.75 }) expect(permissionGroupScopeMockFns.mockResolvePermissionGroupConfig).not.toHaveBeenCalled() }) + + it('reports the timing and cost read with the execution data, not the earlier context', async () => { + queueSelects( + [{ ...workflowRecord, endedAt: null, totalDurationMs: null, costTotal: null }], + [{ id: 'snapshot-1', stateData: { blocks: {} } }], + [completedRun] + ) + + const result = await read(executorPrincipal, 'run-1') + + expect(result.executionMetadata).toMatchObject({ + endedAt: '2026-08-05T12:00:01.000Z', + totalDurationMs: 1000, + cost: { total: 0.75 }, + }) + }) }) diff --git a/apps/sim/lib/logs/application/read-execution-snapshot.ts b/apps/sim/lib/logs/application/read-execution-snapshot.ts index cdeb115139d..e60eca12175 100644 --- a/apps/sim/lib/logs/application/read-execution-snapshot.ts +++ b/apps/sim/lib/logs/application/read-execution-snapshot.ts @@ -23,7 +23,7 @@ import { resolveActiveWorkspaceApplicationContext, } from '@/lib/workspaces/application/context' -const CHILD_SNAPSHOT_QUERY_CHUNK_SIZE = 1_000 +const SNAPSHOT_QUERY_CHUNK_SIZE = 1_000 interface WorkflowExecutionRecord { kind: 'workflow' @@ -37,7 +37,6 @@ interface WorkflowExecutionRecord { endedAt: Date | null totalDurationMs: number | null costTotal: string | null - executionData: unknown } interface JobExecutionRecord { @@ -79,7 +78,6 @@ async function resolveExecutionSnapshotContext( endedAt: workflowExecutionLogs.endedAt, totalDurationMs: workflowExecutionLogs.totalDurationMs, costTotal: workflowExecutionLogs.costTotal, - executionData: workflowExecutionLogs.executionData, }) .from(workflowExecutionLogs) .where(eq(workflowExecutionLogs.executionId, executionId)) @@ -118,6 +116,47 @@ async function resolveExecutionSnapshotContext( return { ...workspace, executionId, record: { kind: 'job', ...jobRecord } } } +async function readSnapshotStates( + ids: string[], + signal: AbortSignal | undefined +): Promise> { + const states = new Map() + for (let index = 0; index < ids.length; index += SNAPSHOT_QUERY_CHUNK_SIZE) { + signal?.throwIfAborted() + const rows = await db + .select({ + id: workflowExecutionSnapshots.id, + stateData: workflowExecutionSnapshots.stateData, + }) + .from(workflowExecutionSnapshots) + .where( + inArray(workflowExecutionSnapshots.id, ids.slice(index, index + SNAPSHOT_QUERY_CHUNK_SIZE)) + ) + for (const row of rows) states.set(row.id, row.stateData) + } + return states +} + +/** + * The run's execution data with the fields a completion write changes alongside it, + * read in one statement after authorization: a caller the use case refuses never + * fetches the payload, and a run finishing mid-request cannot pair new data with + * stale timing or cost. + */ +async function readRunCompletion(executionId: string) { + const [row] = await db + .select({ + executionData: workflowExecutionLogs.executionData, + endedAt: workflowExecutionLogs.endedAt, + totalDurationMs: workflowExecutionLogs.totalDurationMs, + costTotal: workflowExecutionLogs.costTotal, + }) + .from(workflowExecutionLogs) + .where(eq(workflowExecutionLogs.executionId, executionId)) + .limit(1) + return row +} + function collectChildSnapshotIds(traceSpans: TraceSpan[]): string[] { const ids = new Set() const pending = [...traceSpans] @@ -175,17 +214,18 @@ const authorizedReadExecutionSnapshotUseCase = defineAuthorizedWorkspaceUseCase( } } - const [snapshot] = await db - .select() - .from(workflowExecutionSnapshots) - .where(eq(workflowExecutionSnapshots.id, record.stateSnapshotId)) - .limit(1) - if (!snapshot) { + const workflowState = (await readSnapshotStates([record.stateSnapshotId], input.signal)).get( + record.stateSnapshotId + ) + if (workflowState === undefined) { throw new OrchestrationError('not_found', 'Workflow state snapshot not found') } + input.signal?.throwIfAborted() + const completion = await readRunCompletion(record.executionId) + if (!completion) throw new OrchestrationError('not_found', 'Workflow execution not found') const executionData = (await materializeExecutionData( - record.executionData as Record | null, + completion.executionData as Record | null, { workspaceId: context.workspaceId, workflowId: record.workflowId, @@ -201,40 +241,23 @@ const authorizedReadExecutionSnapshotUseCase = defineAuthorizedWorkspaceUseCase( }) } - const childSnapshotIds = collectChildSnapshotIds(traceSpans) - const childWorkflowSnapshots: Array<{ id: string; stateData: unknown }> = [] - for (let index = 0; index < childSnapshotIds.length; index += CHILD_SNAPSHOT_QUERY_CHUNK_SIZE) { - input.signal?.throwIfAborted() - childWorkflowSnapshots.push( - ...(await db - .select({ - id: workflowExecutionSnapshots.id, - stateData: workflowExecutionSnapshots.stateData, - }) - .from(workflowExecutionSnapshots) - .where( - inArray( - workflowExecutionSnapshots.id, - childSnapshotIds.slice(index, index + CHILD_SNAPSHOT_QUERY_CHUNK_SIZE) - ) - )) - ) - } + const childWorkflowSnapshots = await readSnapshotStates( + collectChildSnapshotIds(traceSpans), + input.signal + ) input.signal?.throwIfAborted() return { executionId: record.executionId, workflowId: record.workflowId, - workflowState: snapshot.stateData as Record, - childWorkflowSnapshots: Object.fromEntries( - childWorkflowSnapshots.map((child) => [child.id, child.stateData]) - ), + workflowState: workflowState as Record, + childWorkflowSnapshots: Object.fromEntries(childWorkflowSnapshots), executionMetadata: { trigger: record.trigger, startedAt: record.startedAt.toISOString(), - endedAt: record.endedAt?.toISOString(), - totalDurationMs: record.totalDurationMs, - cost: projectCostTotal(record.costTotal, projection), + endedAt: completion.endedAt?.toISOString(), + totalDurationMs: completion.totalDurationMs, + cost: projectCostTotal(completion.costTotal, projection), }, } }, diff --git a/apps/sim/lib/logs/types.ts b/apps/sim/lib/logs/types.ts index 6c0bba8bd3c..186a7140ced 100644 --- a/apps/sim/lib/logs/types.ts +++ b/apps/sim/lib/logs/types.ts @@ -225,6 +225,21 @@ export function isPersistedWorkflowExecutionStatus( ): value is PersistedWorkflowExecutionStatus { return (PERSISTED_WORKFLOW_EXECUTION_STATUSES as readonly string[]).includes(value) } + +/** + * Statuses of a finished run's log. A resume can never claim a log in one of them (the + * same set the resume attempt-failure `CASE` preserves). + */ +const TERMINAL_WORKFLOW_EXECUTION_STATUSES = [ + 'cancelled', + 'failed', + 'completed', +] as const satisfies readonly PersistedWorkflowExecutionStatus[] + +export function isTerminalWorkflowExecutionStatus(status: string): boolean { + return (TERMINAL_WORKFLOW_EXECUTION_STATUSES as readonly string[]).includes(status) +} + /** * In-flight statuses a crashed worker can strand, which the stale-execution * cron terminalizes. `pending` and `paused` are excluded: both are written as diff --git a/apps/sim/lib/workflows/executor/execution-read-payloads.integration.ts b/apps/sim/lib/workflows/executor/execution-read-payloads.integration.ts new file mode 100644 index 00000000000..231e7ff245f --- /dev/null +++ b/apps/sim/lib/workflows/executor/execution-read-payloads.integration.ts @@ -0,0 +1,231 @@ +/** + * The run status poll, the paused-execution list, and the execution snapshot + * view against real PostgreSQL, recording every value the client receives over + * the wire: a poll fetches a run's execution data only when it reads it, the + * paused list never fetches the serialized execution snapshot it does not + * return, and the snapshot view fetches no execution data for a refused caller. + */ +import { + pausedExecutions, + user, + workflow, + workflowExecutionLogs, + workflowExecutionSnapshots, + workspace, +} from '@sim/db/schema' +import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' +import { createExecutorPrincipal } from '@sim/testing/factories/principal.factory' +import { generateId } from '@sim/utils/id' +import { eq } from 'drizzle-orm' +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest' + +const harness = vi.hoisted(() => ({ + received: [] as unknown[], + closeClients: [] as Array<() => Promise>, +})) + +vi.mock('@sim/db', async (importOriginal) => { + const actual = await importOriginal() + const { default: postgres } = await import('postgres') + const { drizzle } = await import('drizzle-orm/postgres-js') + const schema = await import('@sim/db/schema') + const { withUtcTimestamps } = await import('@sim/db/timestamps') + const { readTestDatabaseUrl } = await import('@sim/db/testing/test-infrastructure') + const client = postgres( + readTestDatabaseUrl(), + withUtcTimestamps({ + max: 4, + prepare: false, + fetch_types: false, + connection: {}, + onnotice: () => {}, + transform: { + value: { + from: (value: unknown) => { + harness.received.push(value) + return value + }, + }, + }, + }) + ) + harness.closeClients.push(() => client.end()) + const recorded = drizzle(client, { schema }) + return { ...actual, db: recorded, dbReplica: recorded, dbFor: () => recorded } +}) + +import { db } from '@sim/db' +import { LOGS_DELEGATION_AUDIENCE } from '@/lib/logs/application/authorization' +import { readExecutionSnapshotUseCase } from '@/lib/logs/application/read-execution-snapshot' +import { getWorkflowExecutionStatus } from '@/lib/workflows/executor/execution-status' +import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager' + +const ids = { + owner: `execution-read-payloads-${generateId()}`, + workspace: generateId(), + workflow: generateId(), + snapshot: generateId(), +} + +/** Whether `marker` reached the client in any value since the last reset. */ +function fetched(marker: string): boolean { + return JSON.stringify(harness.received).includes(marker) +} + +beforeAll(async () => { + const now = new Date() + await db.insert(user).values({ + id: ids.owner, + name: 'Execution Read Payloads', + email: `${ids.owner}@execution-read-payloads.test`, + emailVerified: true, + createdAt: now, + updatedAt: now, + }) + await insertWorkspaceFixture(db, { + id: ids.workspace, + name: 'Execution Read Payloads', + ownerId: ids.owner, + billedAccountUserId: ids.owner, + }) + await db.insert(workflow).values({ + id: ids.workflow, + userId: ids.owner, + workspaceId: ids.workspace, + name: 'Execution Read Payloads', + lastSynced: now, + createdAt: now, + updatedAt: now, + }) + await db.insert(workflowExecutionSnapshots).values({ + id: ids.snapshot, + workflowId: ids.workflow, + stateHash: ids.snapshot, + stateData: {}, + }) +}, 30_000) + +afterAll(async () => { + await db.delete(pausedExecutions).where(eq(pausedExecutions.workflowId, ids.workflow)) + await db.delete(workflowExecutionLogs).where(eq(workflowExecutionLogs.workflowId, ids.workflow)) + await db.delete(workflowExecutionSnapshots).where(eq(workflowExecutionSnapshots.id, ids.snapshot)) + await db.delete(workflow).where(eq(workflow.id, ids.workflow)) + await deleteWorkspaceFixture(db, eq(workspace.id, ids.workspace)) + await db.delete(user).where(eq(user.id, ids.owner)) + await Promise.all(harness.closeClients.map((close) => close())) +}) + +describe('run status poll', () => { + it.each([ + { status: 'running', includeOutput: true, readsExecutionData: false }, + { status: 'completed', includeOutput: false, readsExecutionData: false }, + { status: 'completed', includeOutput: true, readsExecutionData: true }, + { status: 'failed', includeOutput: false, readsExecutionData: true }, + ])( + 'fetches execution data for a $status run polled with includeOutput=$includeOutput only when it reads it', + async ({ status, includeOutput, readsExecutionData }) => { + const executionId = generateId() + const marker = `execution-data-${executionId}` + await db.insert(workflowExecutionLogs).values({ + id: generateId(), + workflowId: ids.workflow, + workspaceId: ids.workspace, + executionId, + stateSnapshotId: ids.snapshot, + level: status === 'failed' ? 'error' : 'info', + status, + trigger: 'api', + startedAt: new Date(), + executionData: { error: marker, finalOutput: { marker } }, + }) + harness.received.length = 0 + + const result = await getWorkflowExecutionStatus({ + workflowId: ids.workflow, + executionId, + includeOutput, + selectedOutputs: [], + workspaceId: ids.workspace, + viewerUserId: null, + }) + + expect(result?.status).toBe(status) + expect(fetched(marker)).toBe(readsExecutionData) + } + ) +}) + +describe('paused execution list', () => { + it('returns trigger ids without fetching the serialized execution snapshot', async () => { + const executionId = generateId() + const marker = `serialized-snapshot-${executionId}` + await db.insert(pausedExecutions).values({ + id: generateId(), + workflowId: ids.workflow, + executionId, + executionSnapshot: { snapshot: marker, triggerIds: ['trigger-1'] }, + pausePoints: { + approval: { + contextId: 'approval', + pauseKind: 'human', + resumeStatus: 'paused', + response: {}, + registeredAt: new Date().toISOString(), + snapshotReady: true, + }, + }, + totalPauseCount: 1, + }) + harness.received.length = 0 + + const listed = await PauseResumeManager.listPausedExecutions({ workflowId: ids.workflow }) + + expect(listed.map((paused) => [paused.executionId, paused.triggerIds])).toEqual([ + [executionId, ['trigger-1']], + ]) + expect(fetched(marker)).toBe(false) + }) +}) + +describe('execution snapshot view', () => { + it('fetches no execution data for a caller it then refuses', async () => { + const executionId = generateId() + const marker = `execution-data-${executionId}` + await db.insert(workflowExecutionLogs).values({ + id: generateId(), + workflowId: ids.workflow, + workspaceId: ids.workspace, + executionId, + stateSnapshotId: ids.snapshot, + level: 'info', + status: 'completed', + trigger: 'api', + startedAt: new Date(), + executionData: { finalOutput: { marker } }, + }) + harness.received.length = 0 + const otherWorkspaceExecutor = createExecutorPrincipal({ + subjectUserId: undefined, + workspaceId: generateId(), + audience: LOGS_DELEGATION_AUDIENCE, + delegationContext: { + kind: 'workflow_execution', + workflowId: ids.workflow, + currentWorkflow: { + workflowId: ids.workflow, + mode: 'deployment', + deploymentVersionId: 'deployment-1', + }, + }, + }) + + await expect( + readExecutionSnapshotUseCase.execute({ + principal: otherWorkspaceExecutor, + input: { executionId }, + }) + ).rejects.toMatchObject({ code: 'not_found' }) + + expect(fetched(marker)).toBe(false) + }) +}) diff --git a/apps/sim/lib/workflows/executor/execution-run-files.ts b/apps/sim/lib/workflows/executor/execution-run-files.ts index 84f71e6c5b3..3959752ccdb 100644 --- a/apps/sim/lib/workflows/executor/execution-run-files.ts +++ b/apps/sim/lib/workflows/executor/execution-run-files.ts @@ -6,15 +6,13 @@ import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import { collectUserFilesById } from '@/lib/core/utils/user-file' import { MAX_INLINE_MATERIALIZATION_BYTES } from '@/lib/execution/payloads/limits' import { materializeExecutionData } from '@/lib/logs/execution/trace-store' +import { isTerminalWorkflowExecutionStatus } from '@/lib/logs/types' import { downloadFile } from '@/lib/uploads/core/storage-service' import { formatFileSize, inferContextFromKey } from '@/lib/uploads/utils/file' import { isRunOutputFileKey } from '@/lib/workflows/executor/run-file-scope' import { classifyRunFileStorageError } from '@/lib/workflows/executor/run-file-storage-error' import type { UserFile } from '@/executor/types' -/** Run states whose recorded output is final and therefore safe to address. */ -const TERMINAL_RUN_STATUSES = new Set(['completed', 'failed', 'cancelled']) - export interface WorkflowRunFilesInput { workflowId: string runId: string @@ -68,7 +66,7 @@ export async function getWorkflowRunFiles( if (!logRow) return null - const terminal = TERMINAL_RUN_STATUSES.has(logRow.status) + const terminal = isTerminalWorkflowExecutionStatus(logRow.status) if (!terminal) { return { terminal: false, workspaceId: logRow.workspaceId, filesById: new Map() } } diff --git a/apps/sim/lib/workflows/executor/execution-status.ts b/apps/sim/lib/workflows/executor/execution-status.ts index d55ff02742a..8137808fcdf 100644 --- a/apps/sim/lib/workflows/executor/execution-status.ts +++ b/apps/sim/lib/workflows/executor/execution-status.ts @@ -1,6 +1,6 @@ import { db } from '@sim/db' import { pausedExecutions, resumeQueue, workflowExecutionLogs } from '@sim/db/schema' -import { and, eq, sql } from 'drizzle-orm' +import { and, eq, inArray, sql } from 'drizzle-orm' import type { WorkflowExecutionStatusResponse } from '@/lib/api/contracts/workflows' import { getJobQueue } from '@/lib/core/async-jobs' import type { Job } from '@/lib/core/async-jobs/types' @@ -286,6 +286,24 @@ export async function getWorkflowExecutionStatus( return (await getProjectedWorkflowExecutionStatus(input))?.status ?? null } +/** Run statuses whose error or final output a poll returns, so it reads their execution data. */ +function statusesReadingExecutionData(includeOutput: boolean): string[] { + return includeOutput ? ['failed', 'completed'] : ['failed'] +} + +/** + * `execution_data` only for a row whose status the poll will read it for, so a + * poll on an in-flight or output-less run does not pull and parse the payload. + */ +function executionDataReadByPoll( + input: Pick +) { + if (input.selectedOutputs.length > 0) return workflowExecutionLogs.executionData + return sql`CASE WHEN ${inArray(workflowExecutionLogs.status, statusesReadingExecutionData(input.includeOutput))} THEN ${workflowExecutionLogs.executionData} END`.mapWith( + workflowExecutionLogs.executionData + ) +} + async function readWorkflowExecutionStatus( input: GetWorkflowExecutionStatusInput ): Promise { @@ -302,7 +320,7 @@ async function readWorkflowExecutionStatus( startedAt: workflowExecutionLogs.startedAt, endedAt: workflowExecutionLogs.endedAt, totalDurationMs: workflowExecutionLogs.totalDurationMs, - executionData: workflowExecutionLogs.executionData, + executionData: executionDataReadByPoll(input), costTotal: workflowExecutionLogs.costTotal, }) .from(workflowExecutionLogs) @@ -446,7 +464,7 @@ async function readWorkflowExecutionStatus( // Mirrors the three output conditions below; a poll needing none skips the archive read. const readsExecutionData = - status === 'failed' || (includeOutput && status === 'completed') || selectedOutputs.length > 0 + selectedOutputs.length > 0 || statusesReadingExecutionData(includeOutput).includes(status) if (!readsExecutionData) { return { ...resource, error: null, finalOutput: null, blockOutputs: null } } diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index d7b811e187c..7687162d637 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -40,7 +40,7 @@ import { terminalExecutionLogFields, } from '@/lib/logs/execution/cancellation' import { LoggingSession } from '@/lib/logs/execution/logging-session' -import type { PersistedWorkflowExecutionStatus } from '@/lib/logs/types' +import { isTerminalWorkflowExecutionStatus } from '@/lib/logs/types' import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server' import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core' import { @@ -92,19 +92,6 @@ const execDb = dbFor('exec') const logger = createLogger('HumanInTheLoopManager') const RUN_BUFFER_UNAVAILABLE_ERROR = 'Run buffer temporarily unavailable' const RESUMABLE_PAUSED_STATUSES = ['paused', 'partially_resumed'] as const -/** - * Statuses of a finished execution's log, the same set the attempt-failure - * `CASE` preserves; a resume can never claim a log in one of them. - */ -const TERMINAL_EXECUTION_LOG_STATUSES = [ - 'cancelled', - 'failed', - 'completed', -] as const satisfies readonly PersistedWorkflowExecutionStatus[] - -function isTerminalExecutionLogStatus(status: string): boolean { - return (TERMINAL_EXECUTION_LOG_STATUSES as readonly string[]).includes(status) -} const CANCELLABLE_PAUSED_STATUSES = ['paused', 'partially_resumed'] as const const AUTOMATIC_RESUME_INTERVENTION_PREFIX = 'Automatic resume requires manual intervention: ' const PAUSED_CANCELLATION_QUEUE_FAILURE_REASON = 'Paused execution cancellation requested' @@ -411,6 +398,19 @@ interface PausedExecutionSummary { pausePoints: PausePointWithQueue[] } +/** The `paused_executions` columns a summary is built from, with the snapshot's trigger ids. */ +type PausedExecutionSummaryRow = Pick< + typeof pausedExecutions.$inferSelect, + | 'id' + | 'workflowId' + | 'executionId' + | 'status' + | 'metadata' + | 'pausedAt' + | 'updatedAt' + | 'expiresAt' +> & { triggerIds: string[] | null | undefined } + interface PausedExecutionDetail extends PausedExecutionSummary { executionSnapshot: SerializedSnapshot queue: ResumeQueueEntrySummary[] @@ -2523,7 +2523,7 @@ export class PauseResumeManager { pausedExecution !== undefined && isResumablePausedStatus(pausedExecution.status) && executionLog !== undefined && - !isTerminalExecutionLogStatus(executionLog.status) + !isTerminalWorkflowExecutionStatus(executionLog.status) ) }) } @@ -3488,8 +3488,24 @@ export class PauseResumeManager { } } + /** + * The summary reads only `triggerIds` from the execution snapshot, whose + * serialized `snapshot` string can run to megabytes per row, so that one key + * is projected instead of shipping and parsing the whole document. + */ const rows = await execDb - .select() + .select({ + id: pausedExecutions.id, + workflowId: pausedExecutions.workflowId, + executionId: pausedExecutions.executionId, + status: pausedExecutions.status, + pausePoints: pausedExecutions.pausePoints, + metadata: pausedExecutions.metadata, + pausedAt: pausedExecutions.pausedAt, + updatedAt: pausedExecutions.updatedAt, + expiresAt: pausedExecutions.expiresAt, + triggerIds: sql`${pausedExecutions.executionSnapshot}->'triggerIds'`, + }) .from(pausedExecutions) .where(whereClause) .orderBy(desc(pausedExecutions.pausedAt)) @@ -3558,7 +3574,10 @@ export class PauseResumeManager { return null } - const executionSummary = PauseResumeManager.normalizePausedExecution(row, pausePoints) + const executionSummary = PauseResumeManager.normalizePausedExecution( + { ...row, triggerIds: (row.executionSnapshot as SerializedSnapshot)?.triggerIds }, + pausePoints + ) return { ...executionSummary, @@ -3753,7 +3772,7 @@ export class PauseResumeManager { } private static normalizePausedExecution( - row: typeof pausedExecutions.$inferSelect, + row: PausedExecutionSummaryRow, pausePoints: PausePointWithQueue[] ): PausedExecutionSummary { return { @@ -3767,7 +3786,7 @@ export class PauseResumeManager { updatedAt: row.updatedAt ? row.updatedAt.toISOString() : null, expiresAt: row.expiresAt ? row.expiresAt.toISOString() : null, metadata: row.metadata as Record, - triggerIds: (row.executionSnapshot as SerializedSnapshot)?.triggerIds || [], + triggerIds: row.triggerIds || [], pausePoints, } }