Skip to content

Commit d3da633

Browse files
authored
fix(executions): skip archive materialization on status polls that read no output (#9308)
readWorkflowExecutionStatus returns the column-only response with error/finalOutput/blockOutputs null unless the run failed, a completed run was polled with includeOutput, or selectedOutputs were requested - 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. Adds regression tests for skipped polls (fail fast via a macrotask sentinel) and for selected outputs on a running run, including the truncated-row rejection.
1 parent 93c71ad commit d3da633

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,4 +1,5 @@
11
import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
2+
import { flushMacrotask } from '@sim/testing/helpers/async'
23
import { asyncJobsMock, asyncJobsMockFns } from '@sim/testing/mocks/async-jobs.mock'
34
import { traceStoreMock, traceStoreMockFns } from '@sim/testing/mocks/trace-store.mock'
45
import { and } from 'drizzle-orm'
@@ -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), flushMacrotask().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)