Skip to content

Commit 256430f

Browse files
committed
fix(executions): skip archive materialization on status polls that read no output
readWorkflowExecutionStatus now builds the column-only response first and returns it with error/finalOutput/blockOutputs null unless the poll is for a failed run, a completed run with includeOutput, or has selectedOutputs - the only three readers of the materialized execution data. Other polls skip the archive fetch, parse, and display projection. Response key order is unchanged. Skipped polls no longer emit the archive-unavailable, provenance-fault, or write-time-projected-span warnings, and a projection exception on such a poll no longer turns it into a 500.
1 parent 7992b56 commit 256430f

2 files changed

Lines changed: 129 additions & 16 deletions

File tree

‎apps/sim/lib/workflows/executor/execution-status.test.ts‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
22
import { asyncJobsMock, asyncJobsMockFns } from '@sim/testing/mocks/async-jobs.mock'
33
import { traceStoreMock, traceStoreMockFns } from '@sim/testing/mocks/trace-store.mock'
4+
import { sleep } from '@sim/utils/helpers'
45
import { and } from 'drizzle-orm'
56
import { beforeEach, describe, expect, it, vi } from 'vitest'
67

@@ -12,6 +13,7 @@ vi.mock('@/lib/workflows/executor/paused-execution-metadata', () => ({
1213
getAutomaticResumeWaitingMetadata: vi.fn().mockReturnValue(null),
1314
}))
1415

16+
import { FunctionalOutputsUnavailableError } from '@/lib/logs/execution/functional-outputs'
1517
import { getWorkflowExecutionStatus } from '@/lib/workflows/executor/execution-status'
1618

1719
const mockGetJob = asyncJobsMockFns.mockJobQueue.getJob
@@ -552,3 +554,108 @@ describe('getWorkflowExecutionStatus settled resume attempts', () => {
552554
expect(status).toMatchObject({ status: 'cancelled', level: 'info', error: null })
553555
})
554556
})
557+
558+
describe('getWorkflowExecutionStatus polls that read no execution data', () => {
559+
const STALLED = Symbol('stalled')
560+
561+
function queueLog(status: string, pausedRows: unknown[] = []) {
562+
queueTableRows(schemaMock.workflowExecutionLogs, [
563+
{
564+
executionId: 'execution-1',
565+
workflowId: 'workflow-1',
566+
workspaceId: 'workspace-1',
567+
status,
568+
level: 'info',
569+
trigger: 'api',
570+
startedAt: new Date('2026-08-05T12:00:00.000Z'),
571+
endedAt: new Date('2026-08-05T12:00:01.000Z'),
572+
totalDurationMs: 1000,
573+
executionData: { finalOutput: { answer: 42 } },
574+
costTotal: '0.5',
575+
},
576+
])
577+
queueTableRows(schemaMock.resumeQueue, [])
578+
queueTableRows(schemaMock.pausedExecutions, pausedRows)
579+
}
580+
581+
/** Stands in for an archive read that has not come back from object storage yet. */
582+
function stallArchiveRead() {
583+
mockMaterializeForDisplayWithBlockOutputs.mockReturnValue(new Promise(() => {}))
584+
}
585+
586+
/** Settles with {@link STALLED} instead of hanging when the read waits on the archive. */
587+
function readOrStall(read: Parameters<typeof getWorkflowExecutionStatus>[0]) {
588+
return Promise.race([getWorkflowExecutionStatus(read), sleep(0).then(() => STALLED)])
589+
}
590+
591+
beforeEach(() => {
592+
resetDbChainMock()
593+
mockGetJob.mockResolvedValue(null)
594+
})
595+
596+
it.each([
597+
['a running run', 'running', input],
598+
['a cancelled run', 'cancelled', input],
599+
['a completed run polled without includeOutput', 'completed', input],
600+
['a running run asked for its output early', 'running', { ...input, includeOutput: true }],
601+
])('answers %s without waiting on the run archive', async (_, status, read) => {
602+
queueLog(status)
603+
stallArchiveRead()
604+
605+
await expect(readOrStall(read)).resolves.toMatchObject({
606+
status,
607+
cost: { total: 0.5 },
608+
error: null,
609+
finalOutput: null,
610+
blockOutputs: null,
611+
})
612+
})
613+
614+
it('answers a paused run without waiting on the run archive', async () => {
615+
queueLog('running', [
616+
{
617+
id: 'paused-1',
618+
status: 'paused',
619+
pausePoints: [{ contextId: 'context-1', blockId: 'block-1', resumeStatus: 'paused' }],
620+
metadata: {},
621+
resumedCount: 0,
622+
pausedAt: new Date('2026-08-05T12:00:00.500Z'),
623+
nextResumeAt: null,
624+
},
625+
])
626+
stallArchiveRead()
627+
628+
await expect(readOrStall({ ...input, includeOutput: true })).resolves.toMatchObject({
629+
status: 'paused',
630+
paused: { contextId: 'context-1' },
631+
finalOutput: null,
632+
})
633+
})
634+
635+
it('still reads selected block outputs from a run that is still running', async () => {
636+
queueLog('running')
637+
mockMaterializeForDisplayWithBlockOutputs.mockResolvedValue({
638+
executionData: {},
639+
blockOutputs: new Map([['block-1', { content: 'partial' }]]),
640+
})
641+
642+
await expect(
643+
getWorkflowExecutionStatus({ ...input, selectedOutputs: ['block-1'] })
644+
).resolves.toMatchObject({
645+
status: 'running',
646+
finalOutput: null,
647+
blockOutputs: { 'block-1': { content: 'partial' } },
648+
})
649+
})
650+
651+
it('still rejects selected outputs of a running run whose execution data was truncated', async () => {
652+
queueLog('running')
653+
mockMaterializeForDisplayWithBlockOutputs.mockRejectedValue(
654+
new FunctionalOutputsUnavailableError()
655+
)
656+
657+
await expect(
658+
getWorkflowExecutionStatus({ ...input, selectedOutputs: ['block-1'] })
659+
).rejects.toBeInstanceOf(FunctionalOutputsUnavailableError)
660+
})
661+
})

‎apps/sim/lib/workflows/executor/execution-status.ts‎

Lines changed: 22 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -430,6 +430,27 @@ async function readWorkflowExecutionStatus(
430430

431431
const cost = logRow.costTotal != null ? { total: Number(logRow.costTotal) } : null
432432

433+
const resource = {
434+
executionId: logRow.executionId,
435+
// Column is `set null` on workflow delete; the caller's param is the fallback.
436+
workflowId: logRow.workflowId ?? workflowId,
437+
status,
438+
trigger: logRow.trigger,
439+
level: logRow.level,
440+
startedAt: logRow.startedAt.toISOString(),
441+
endedAt: logRow.endedAt?.toISOString() ?? null,
442+
totalDurationMs: logRow.totalDurationMs ?? null,
443+
paused,
444+
cost,
445+
}
446+
447+
// Mirrors the three output conditions below; a poll needing none skips the archive read.
448+
const readsExecutionData =
449+
status === 'failed' || (includeOutput && status === 'completed') || selectedOutputs.length > 0
450+
if (!readsExecutionData) {
451+
return { ...resource, error: null, finalOutput: null, blockOutputs: null }
452+
}
453+
433454
const requestedBlockIds = [
434455
...new Set(selectedOutputs.map((selector) => selector.split('.')[0]).filter(Boolean)),
435456
]
@@ -456,20 +477,5 @@ async function readWorkflowExecutionStatus(
456477
? pickSelectedOutputs(selectedOutputs, materialized.blockOutputs)
457478
: null
458479

459-
return {
460-
executionId: logRow.executionId,
461-
// Column is `set null` on workflow delete; the caller's param is the fallback.
462-
workflowId: logRow.workflowId ?? workflowId,
463-
status,
464-
trigger: logRow.trigger,
465-
level: logRow.level,
466-
startedAt: logRow.startedAt.toISOString(),
467-
endedAt: logRow.endedAt?.toISOString() ?? null,
468-
totalDurationMs: logRow.totalDurationMs ?? null,
469-
paused,
470-
cost,
471-
error,
472-
finalOutput,
473-
blockOutputs,
474-
}
480+
return { ...resource, error, finalOutput, blockOutputs }
475481
}

0 commit comments

Comments
 (0)