Skip to content

Commit 02ebae6

Browse files
committed
fix(logs): stop loading execution data and snapshots that reads do not use
- Status polls select execution_data through a CASE only for the statuses the poll reads it for (failed; completed when output is requested; always for selected outputs), from one shared status rule. - The execution snapshot view no longer loads execution_data while resolving its context; it is read only after authorization, so a refused caller never fetches it. - The paused execution list projects execution_snapshot->'triggerIds' instead of loading whole rows with the serialized executor snapshot. - One shared terminal-status predicate replaces the copies in the HITL manager and run-file reads.
1 parent 42f3409 commit 02ebae6

7 files changed

Lines changed: 368 additions & 64 deletions

File tree

‎apps/sim/lib/logs/application/read-execution-snapshot.test.ts‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -77,17 +77,19 @@ const workflowRecord = {
7777
endedAt: new Date('2026-08-05T12:00:01.000Z'),
7878
totalDurationMs: 1000,
7979
costTotal: '0.75',
80-
executionData: null,
8180
}
8281

8382
/**
8483
* Answers `db.select(...)` calls in order. The snapshot read walks the workflow
85-
* log, then (only when that missed) the job log, then the state snapshot.
84+
* log, then (only when that missed) the job log, then the state snapshot, then
85+
* the run's execution data.
8686
*/
8787
function queueSelects(...results: unknown[][]): void {
8888
for (const rows of results) {
8989
mocks.select.mockReturnValueOnce({
90-
from: () => ({ where: () => ({ limit: () => Promise.resolve(rows) }) }),
90+
from: () => ({
91+
where: () => Object.assign(Promise.resolve(rows), { limit: () => Promise.resolve(rows) }),
92+
}),
9193
})
9294
}
9395
}
@@ -135,7 +137,11 @@ describe('readExecutionSnapshot spend projection', () => {
135137
...DEFAULT_PERMISSION_GROUP_CONFIG,
136138
hideCostInfo: true,
137139
})
138-
queueSelects([workflowRecord], [{ id: 'snapshot-1', stateData: { blocks: {} } }])
140+
queueSelects(
141+
[workflowRecord],
142+
[{ id: 'snapshot-1', stateData: { blocks: {} } }],
143+
[{ executionData: null }]
144+
)
139145

140146
const result = await read(principal, 'run-1')
141147

@@ -153,7 +159,11 @@ describe('readExecutionSnapshot spend projection', () => {
153159
...DEFAULT_PERMISSION_GROUP_CONFIG,
154160
hideCostInfo: true,
155161
})
156-
queueSelects([workflowRecord], [{ id: 'snapshot-1', stateData: { blocks: {} } }])
162+
queueSelects(
163+
[workflowRecord],
164+
[{ id: 'snapshot-1', stateData: { blocks: {} } }],
165+
[{ executionData: null }]
166+
)
157167

158168
const result = await read(executorPrincipal, 'run-1')
159169

‎apps/sim/lib/logs/application/read-execution-snapshot.ts‎

Lines changed: 47 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ import {
2323
resolveActiveWorkspaceApplicationContext,
2424
} from '@/lib/workspaces/application/context'
2525

26-
const CHILD_SNAPSHOT_QUERY_CHUNK_SIZE = 1_000
26+
const SNAPSHOT_QUERY_CHUNK_SIZE = 1_000
2727

2828
interface WorkflowExecutionRecord {
2929
kind: 'workflow'
@@ -37,7 +37,6 @@ interface WorkflowExecutionRecord {
3737
endedAt: Date | null
3838
totalDurationMs: number | null
3939
costTotal: string | null
40-
executionData: unknown
4140
}
4241

4342
interface JobExecutionRecord {
@@ -79,7 +78,6 @@ async function resolveExecutionSnapshotContext(
7978
endedAt: workflowExecutionLogs.endedAt,
8079
totalDurationMs: workflowExecutionLogs.totalDurationMs,
8180
costTotal: workflowExecutionLogs.costTotal,
82-
executionData: workflowExecutionLogs.executionData,
8381
})
8482
.from(workflowExecutionLogs)
8583
.where(eq(workflowExecutionLogs.executionId, executionId))
@@ -118,6 +116,40 @@ async function resolveExecutionSnapshotContext(
118116
return { ...workspace, executionId, record: { kind: 'job', ...jobRecord } }
119117
}
120118

119+
async function readSnapshotStates(
120+
ids: string[],
121+
signal: AbortSignal | undefined
122+
): Promise<Map<string, unknown>> {
123+
const states = new Map<string, unknown>()
124+
for (let index = 0; index < ids.length; index += SNAPSHOT_QUERY_CHUNK_SIZE) {
125+
signal?.throwIfAborted()
126+
const rows = await db
127+
.select({
128+
id: workflowExecutionSnapshots.id,
129+
stateData: workflowExecutionSnapshots.stateData,
130+
})
131+
.from(workflowExecutionSnapshots)
132+
.where(
133+
inArray(workflowExecutionSnapshots.id, ids.slice(index, index + SNAPSHOT_QUERY_CHUNK_SIZE))
134+
)
135+
for (const row of rows) states.set(row.id, row.stateData)
136+
}
137+
return states
138+
}
139+
140+
/**
141+
* Read here, after authorization, rather than with the context: a caller the
142+
* use case refuses never fetches the run's execution data.
143+
*/
144+
async function readExecutionData(executionId: string): Promise<unknown> {
145+
const [row] = await db
146+
.select({ executionData: workflowExecutionLogs.executionData })
147+
.from(workflowExecutionLogs)
148+
.where(eq(workflowExecutionLogs.executionId, executionId))
149+
.limit(1)
150+
return row?.executionData ?? null
151+
}
152+
121153
function collectChildSnapshotIds(traceSpans: TraceSpan[]): string[] {
122154
const ids = new Set<string>()
123155
const pending = [...traceSpans]
@@ -175,17 +207,16 @@ const authorizedReadExecutionSnapshotUseCase = defineAuthorizedWorkspaceUseCase(
175207
}
176208
}
177209

178-
const [snapshot] = await db
179-
.select()
180-
.from(workflowExecutionSnapshots)
181-
.where(eq(workflowExecutionSnapshots.id, record.stateSnapshotId))
182-
.limit(1)
183-
if (!snapshot) {
210+
const workflowState = (await readSnapshotStates([record.stateSnapshotId], input.signal)).get(
211+
record.stateSnapshotId
212+
)
213+
if (workflowState === undefined) {
184214
throw new OrchestrationError('not_found', 'Workflow state snapshot not found')
185215
}
186216

217+
input.signal?.throwIfAborted()
187218
const executionData = (await materializeExecutionData(
188-
record.executionData as Record<string, unknown> | null,
219+
(await readExecutionData(record.executionId)) as Record<string, unknown> | null,
189220
{
190221
workspaceId: context.workspaceId,
191222
workflowId: record.workflowId,
@@ -201,34 +232,17 @@ const authorizedReadExecutionSnapshotUseCase = defineAuthorizedWorkspaceUseCase(
201232
})
202233
}
203234

204-
const childSnapshotIds = collectChildSnapshotIds(traceSpans)
205-
const childWorkflowSnapshots: Array<{ id: string; stateData: unknown }> = []
206-
for (let index = 0; index < childSnapshotIds.length; index += CHILD_SNAPSHOT_QUERY_CHUNK_SIZE) {
207-
input.signal?.throwIfAborted()
208-
childWorkflowSnapshots.push(
209-
...(await db
210-
.select({
211-
id: workflowExecutionSnapshots.id,
212-
stateData: workflowExecutionSnapshots.stateData,
213-
})
214-
.from(workflowExecutionSnapshots)
215-
.where(
216-
inArray(
217-
workflowExecutionSnapshots.id,
218-
childSnapshotIds.slice(index, index + CHILD_SNAPSHOT_QUERY_CHUNK_SIZE)
219-
)
220-
))
221-
)
222-
}
235+
const childWorkflowSnapshots = await readSnapshotStates(
236+
collectChildSnapshotIds(traceSpans),
237+
input.signal
238+
)
223239

224240
input.signal?.throwIfAborted()
225241
return {
226242
executionId: record.executionId,
227243
workflowId: record.workflowId,
228-
workflowState: snapshot.stateData as Record<string, unknown>,
229-
childWorkflowSnapshots: Object.fromEntries(
230-
childWorkflowSnapshots.map((child) => [child.id, child.stateData])
231-
),
244+
workflowState: workflowState as Record<string, unknown>,
245+
childWorkflowSnapshots: Object.fromEntries(childWorkflowSnapshots),
232246
executionMetadata: {
233247
trigger: record.trigger,
234248
startedAt: record.startedAt.toISOString(),

‎apps/sim/lib/logs/types.ts‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -225,6 +225,21 @@ export function isPersistedWorkflowExecutionStatus(
225225
): value is PersistedWorkflowExecutionStatus {
226226
return (PERSISTED_WORKFLOW_EXECUTION_STATUSES as readonly string[]).includes(value)
227227
}
228+
229+
/**
230+
* Statuses of a finished run's log. A resume can never claim a log in one of them (the
231+
* same set the resume attempt-failure `CASE` preserves).
232+
*/
233+
const TERMINAL_WORKFLOW_EXECUTION_STATUSES = [
234+
'cancelled',
235+
'failed',
236+
'completed',
237+
] as const satisfies readonly PersistedWorkflowExecutionStatus[]
238+
239+
export function isTerminalWorkflowExecutionStatus(status: string): boolean {
240+
return (TERMINAL_WORKFLOW_EXECUTION_STATUSES as readonly string[]).includes(status)
241+
}
242+
228243
/**
229244
* In-flight statuses a crashed worker can strand, which the stale-execution
230245
* cron terminalizes. `pending` and `paused` are excluded: both are written as

0 commit comments

Comments
 (0)