From 256430fb9fceea815763e10bac27e0680608dad4 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 22 Oct 2026 13:16:00 -0700 Subject: [PATCH 1/2] 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. --- .../executor/execution-status.test.ts | 107 ++++++++++++++++++ .../workflows/executor/execution-status.ts | 38 ++++--- 2 files changed, 129 insertions(+), 16 deletions(-) diff --git a/apps/sim/lib/workflows/executor/execution-status.test.ts b/apps/sim/lib/workflows/executor/execution-status.test.ts index 30eaee80adf..cb11f858cd5 100644 --- a/apps/sim/lib/workflows/executor/execution-status.test.ts +++ b/apps/sim/lib/workflows/executor/execution-status.test.ts @@ -1,6 +1,7 @@ import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing' import { asyncJobsMock, asyncJobsMockFns } from '@sim/testing/mocks/async-jobs.mock' import { traceStoreMock, traceStoreMockFns } from '@sim/testing/mocks/trace-store.mock' +import { sleep } from '@sim/utils/helpers' import { and } from 'drizzle-orm' import { beforeEach, describe, expect, it, vi } from 'vitest' @@ -12,6 +13,7 @@ vi.mock('@/lib/workflows/executor/paused-execution-metadata', () => ({ getAutomaticResumeWaitingMetadata: vi.fn().mockReturnValue(null), })) +import { FunctionalOutputsUnavailableError } from '@/lib/logs/execution/functional-outputs' import { getWorkflowExecutionStatus } from '@/lib/workflows/executor/execution-status' const mockGetJob = asyncJobsMockFns.mockJobQueue.getJob @@ -552,3 +554,108 @@ describe('getWorkflowExecutionStatus settled resume attempts', () => { expect(status).toMatchObject({ status: 'cancelled', level: 'info', error: null }) }) }) + +describe('getWorkflowExecutionStatus polls that read no execution data', () => { + const STALLED = Symbol('stalled') + + function queueLog(status: string, pausedRows: unknown[] = []) { + queueTableRows(schemaMock.workflowExecutionLogs, [ + { + executionId: 'execution-1', + workflowId: 'workflow-1', + workspaceId: 'workspace-1', + status, + level: 'info', + trigger: 'api', + startedAt: new Date('2026-08-05T12:00:00.000Z'), + endedAt: new Date('2026-08-05T12:00:01.000Z'), + totalDurationMs: 1000, + executionData: { finalOutput: { answer: 42 } }, + costTotal: '0.5', + }, + ]) + queueTableRows(schemaMock.resumeQueue, []) + queueTableRows(schemaMock.pausedExecutions, pausedRows) + } + + /** Stands in for an archive read that has not come back from object storage yet. */ + function stallArchiveRead() { + mockMaterializeForDisplayWithBlockOutputs.mockReturnValue(new Promise(() => {})) + } + + /** Settles with {@link STALLED} instead of hanging when the read waits on the archive. */ + function readOrStall(read: Parameters[0]) { + return Promise.race([getWorkflowExecutionStatus(read), sleep(0).then(() => STALLED)]) + } + + beforeEach(() => { + resetDbChainMock() + mockGetJob.mockResolvedValue(null) + }) + + it.each([ + ['a running run', 'running', input], + ['a cancelled run', 'cancelled', input], + ['a completed run polled without includeOutput', 'completed', input], + ['a running run asked for its output early', 'running', { ...input, includeOutput: true }], + ])('answers %s without waiting on the run archive', async (_, status, read) => { + queueLog(status) + stallArchiveRead() + + await expect(readOrStall(read)).resolves.toMatchObject({ + status, + cost: { total: 0.5 }, + error: null, + finalOutput: null, + blockOutputs: null, + }) + }) + + it('answers a paused run without waiting on the run archive', async () => { + queueLog('running', [ + { + id: 'paused-1', + status: 'paused', + pausePoints: [{ contextId: 'context-1', blockId: 'block-1', resumeStatus: 'paused' }], + metadata: {}, + resumedCount: 0, + pausedAt: new Date('2026-08-05T12:00:00.500Z'), + nextResumeAt: null, + }, + ]) + stallArchiveRead() + + await expect(readOrStall({ ...input, includeOutput: true })).resolves.toMatchObject({ + status: 'paused', + paused: { contextId: 'context-1' }, + finalOutput: null, + }) + }) + + it('still reads selected block outputs from a run that is still running', async () => { + queueLog('running') + mockMaterializeForDisplayWithBlockOutputs.mockResolvedValue({ + executionData: {}, + blockOutputs: new Map([['block-1', { content: 'partial' }]]), + }) + + await expect( + getWorkflowExecutionStatus({ ...input, selectedOutputs: ['block-1'] }) + ).resolves.toMatchObject({ + status: 'running', + finalOutput: null, + blockOutputs: { 'block-1': { content: 'partial' } }, + }) + }) + + it('still rejects selected outputs of a running run whose execution data was truncated', async () => { + queueLog('running') + mockMaterializeForDisplayWithBlockOutputs.mockRejectedValue( + new FunctionalOutputsUnavailableError() + ) + + await expect( + getWorkflowExecutionStatus({ ...input, selectedOutputs: ['block-1'] }) + ).rejects.toBeInstanceOf(FunctionalOutputsUnavailableError) + }) +}) diff --git a/apps/sim/lib/workflows/executor/execution-status.ts b/apps/sim/lib/workflows/executor/execution-status.ts index 22d7867cee5..d55ff02742a 100644 --- a/apps/sim/lib/workflows/executor/execution-status.ts +++ b/apps/sim/lib/workflows/executor/execution-status.ts @@ -430,6 +430,27 @@ async function readWorkflowExecutionStatus( const cost = logRow.costTotal != null ? { total: Number(logRow.costTotal) } : null + const resource = { + executionId: logRow.executionId, + // Column is `set null` on workflow delete; the caller's param is the fallback. + workflowId: logRow.workflowId ?? workflowId, + status, + trigger: logRow.trigger, + level: logRow.level, + startedAt: logRow.startedAt.toISOString(), + endedAt: logRow.endedAt?.toISOString() ?? null, + totalDurationMs: logRow.totalDurationMs ?? null, + paused, + cost, + } + + // Mirrors the three output conditions below; a poll needing none skips the archive read. + const readsExecutionData = + status === 'failed' || (includeOutput && status === 'completed') || selectedOutputs.length > 0 + if (!readsExecutionData) { + return { ...resource, error: null, finalOutput: null, blockOutputs: null } + } + const requestedBlockIds = [ ...new Set(selectedOutputs.map((selector) => selector.split('.')[0]).filter(Boolean)), ] @@ -456,20 +477,5 @@ async function readWorkflowExecutionStatus( ? pickSelectedOutputs(selectedOutputs, materialized.blockOutputs) : null - return { - executionId: logRow.executionId, - // Column is `set null` on workflow delete; the caller's param is the fallback. - workflowId: logRow.workflowId ?? workflowId, - status, - trigger: logRow.trigger, - level: logRow.level, - startedAt: logRow.startedAt.toISOString(), - endedAt: logRow.endedAt?.toISOString() ?? null, - totalDurationMs: logRow.totalDurationMs ?? null, - paused, - cost, - error, - finalOutput, - blockOutputs, - } + return { ...resource, error, finalOutput, blockOutputs } } From 4630b014052968f86df24f7d3b162677a7e3ef45 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 22 Oct 2026 13:27:28 -0700 Subject: [PATCH 2/2] fix(executions): use the shared flushMacrotask helper for the stall sentinel --- apps/sim/lib/workflows/executor/execution-status.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/workflows/executor/execution-status.test.ts b/apps/sim/lib/workflows/executor/execution-status.test.ts index cb11f858cd5..f69cd41431a 100644 --- a/apps/sim/lib/workflows/executor/execution-status.test.ts +++ b/apps/sim/lib/workflows/executor/execution-status.test.ts @@ -1,7 +1,7 @@ import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing' +import { flushMacrotask } from '@sim/testing/helpers/async' import { asyncJobsMock, asyncJobsMockFns } from '@sim/testing/mocks/async-jobs.mock' import { traceStoreMock, traceStoreMockFns } from '@sim/testing/mocks/trace-store.mock' -import { sleep } from '@sim/utils/helpers' import { and } from 'drizzle-orm' import { beforeEach, describe, expect, it, vi } from 'vitest' @@ -585,7 +585,7 @@ describe('getWorkflowExecutionStatus polls that read no execution data', () => { /** Settles with {@link STALLED} instead of hanging when the read waits on the archive. */ function readOrStall(read: Parameters[0]) { - return Promise.race([getWorkflowExecutionStatus(read), sleep(0).then(() => STALLED)]) + return Promise.race([getWorkflowExecutionStatus(read), flushMacrotask().then(() => STALLED)]) } beforeEach(() => {