From 009f5fea7f4333dd1f18f487e8819d6217072fbc Mon Sep 17 00:00:00 2001 From: Waleed Date: Thu, 13 Aug 2026 17:55:37 -0700 Subject: [PATCH] fix(logs): record how long a cancelled run had been going (#6686) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(logs): record how long a cancelled run had been going A cancelled run got an end timestamp and no duration. Every other terminal transition writes both — the completion path sets them together, and a paused run already records its elapsed time — but cancellation writes the log row directly rather than through completion, so it had no in-memory duration to store and simply omitted the column. That is not cosmetic. `GET /api/v2/logs` filters on `minDurationMs` and `maxDurationMs`, and a null column drops the row out of every such query, so cancellations are invisible to exactly the searches someone runs when investigating cancellations. The published contract also says the end timestamp is null only while a run is active, which a cancelled run is not. Both cancellation writes now derive the duration in the same statement from the row's own `started_at`, through one shared expression so the two cannot drift apart the way they did from the completion path. The end instant is computed once and reused, so the stamped end and the derived duration describe the same moment rather than two clock reads. The instant is bound as an explicit `timestamp` rather than a `Date`: `started_at` is `timestamp without time zone` holding a UTC wall clock, and a driver-bound date would infer `timestamptz` and make the interval depend on the session zone. The floor of one millisecond matches the completion path, so a run cancelled inside its first millisecond still records that it ran. * fix(logs): saturate the cancelled-run duration at the column ceiling The column is `integer`, so an untimed run cancelled after roughly twenty-five days overflowed the cast. That cost more than the duration it was recording: the direct write is caught and logged, so the row would have stayed `running` with no end timestamp at all, and the workflow-group write would have failed its transaction and taken the whole cancellation with it. Saturating keeps the terminal write. A duration wrong in its last digits is a smaller lie than a run that never ended. * fix(logs): record the duration on the other two cancellation writes Review found the first pass had only covered two of four terminal cancellation writes. The two it missed spell the timestamp `endedAt: now` rather than `endedAt: new Date()`, so the search that found the first pair could never have found them — and one of them is the common case: a workflow-group run with a live cell sidecar takes that branch, and the direct cancel skips its own log update whenever group cancellation handled the run, so it was the only writer for those cancellations. The other is the paused-cancellation write, which the first pass reported as already correct on the strength of that same search. A paused run records its duration when it pauses; cancelling it did not. All four now derive the duration the same way, and the sweep for the remaining ones went over every `status: 'cancelled'` write rather than one spelling of the timestamp beside it. * fix(logs): let a recorded duration outlive a later cancellation A paused run measures its own active duration at the pause checkpoint. The previous commit then had cancellation overwrite that with wall clock from the start, which quietly redefines the column for those runs to include the time the run spent waiting rather than working — filling a gap by discarding an answer someone else had already computed. The duration now coalesces onto whatever the row already carries, so a cancellation only supplies the value when nothing else did. Every other cancellation path leaves the column null, so the change is inert there. * fix(logs): only a paused run keeps the duration it recorded Preserving any duration already on the row was too broad. Resuming flips the log back to running and leaves the pause checkpoint value behind, so a resumed run carries a stale reading while it is accruing time again; cancelling it would have frozen that pre-resume figure and disagreed with the resume completion path, which measures wall clock. What separates the two is the row's status rather than whether the column is populated. A paused run is not accruing, so its recorded active duration stands. A running one recomputes. --- .../cancel-workflow-execution.test.ts | 13 +++ .../execution/cancel-workflow-execution.ts | 8 +- apps/sim/lib/logs/execution/duration.test.ts | 91 +++++++++++++++++++ apps/sim/lib/logs/execution/duration.ts | 49 ++++++++++ .../table/workflow-group-cancellation.test.ts | 7 ++ .../lib/table/workflow-group-cancellation.ts | 16 +++- .../human-in-the-loop-manager.test.ts | 1 + .../executor/human-in-the-loop-manager.ts | 8 +- 8 files changed, 189 insertions(+), 4 deletions(-) create mode 100644 apps/sim/lib/logs/execution/duration.test.ts create mode 100644 apps/sim/lib/logs/execution/duration.ts diff --git a/apps/sim/lib/execution/cancel-workflow-execution.test.ts b/apps/sim/lib/execution/cancel-workflow-execution.test.ts index 77d7da49e7..53621352a0 100644 --- a/apps/sim/lib/execution/cancel-workflow-execution.test.ts +++ b/apps/sim/lib/execution/cancel-workflow-execution.test.ts @@ -281,4 +281,17 @@ describe('cancelWorkflowExecution', () => { expect(mockPublishWorkflowGroupCancellationEvent).not.toHaveBeenCalled() expect(mockUpdateSet).toHaveBeenCalledWith(expect.objectContaining({ status: 'cancelled' })) }) + + /** + * A cancelled run is terminal, so it owes the same two fields every other + * terminal write records. Without the duration it is invisible to the + * `minDurationMs`/`maxDurationMs` filters on `GET /api/v2/logs`. + */ + it('records how long the cancelled run had been going, not just when it stopped', async () => { + await cancelWorkflowExecution(INPUT) + + const [values] = mockUpdateSet.mock.calls.at(-1) as [Record] + expect(values.endedAt).toBeInstanceOf(Date) + expect(values.totalDurationMs).toBeDefined() + }) }) diff --git a/apps/sim/lib/execution/cancel-workflow-execution.ts b/apps/sim/lib/execution/cancel-workflow-execution.ts index 1ca5562076..1f391c864a 100644 --- a/apps/sim/lib/execution/cancel-workflow-execution.ts +++ b/apps/sim/lib/execution/cancel-workflow-execution.ts @@ -12,6 +12,7 @@ import { } from '@/lib/execution/cancellation' import { createExecutionEventWriter, readExecutionMetaState } from '@/lib/execution/event-buffer' import { abortManualExecution } from '@/lib/execution/manual-cancellation' +import { elapsedDurationMsSql } from '@/lib/logs/execution/duration' import { captureServerEvent } from '@/lib/posthog/server' import { cancelWorkflowGroupExecution, @@ -377,9 +378,14 @@ export async function cancelWorkflowExecution( !pausedCancelled ) { try { + const cancelledAt = new Date() await db .update(workflowExecutionLogs) - .set({ status: 'cancelled', endedAt: new Date() }) + .set({ + status: 'cancelled', + endedAt: cancelledAt, + totalDurationMs: elapsedDurationMsSql(cancelledAt), + }) .where( and( eq(workflowExecutionLogs.executionId, executionId), diff --git a/apps/sim/lib/logs/execution/duration.test.ts b/apps/sim/lib/logs/execution/duration.test.ts new file mode 100644 index 0000000000..7729e6e0f6 --- /dev/null +++ b/apps/sim/lib/logs/execution/duration.test.ts @@ -0,0 +1,91 @@ +/** + * @vitest-environment node + */ + +// Renders the real expression against the real drizzle dialect and schema. It +// is a raw `sql` template, so a rendering or type-cast bug only surfaces when +// Postgres executes it — the global drizzle/schema mocks would hide it. +import { describe, expect, it, vi } from 'vitest' + +vi.unmock('drizzle-orm') +vi.unmock('@sim/db') +vi.unmock('@sim/db/schema') + +process.env.DATABASE_URL ??= 'postgresql://user:pass@localhost:5432/test' + +const { PgDialect } = await import('drizzle-orm/pg-core') +const { elapsedDurationMsSql } = await import('@/lib/logs/execution/duration') + +function render(endedAt: Date) { + return new PgDialect().sqlToQuery(elapsedDurationMsSql(endedAt)) +} + +describe('elapsedDurationMsSql', () => { + it('measures against the row started_at rather than a second clock read', () => { + const { sql } = render(new Date('2026-08-13T12:00:05.000Z')) + + expect(sql).toContain('"started_at"') + expect(sql).not.toContain('now()') + }) + + /** + * `started_at` is `timestamp without time zone` holding a UTC wall clock. A + * driver-bound `Date` infers `timestamptz`, which would make the interval + * depend on the session zone; the explicit cast is what keeps it stable. + */ + it('binds the end instant as a zone-free timestamp', () => { + const { sql, params } = render(new Date('2026-08-13T12:00:05.000Z')) + + expect(sql).toContain('::timestamp') + expect(params).toContain('2026-08-13T12:00:05.000Z') + expect(params.some((param) => Array.isArray(param))).toBe(false) + }) + + /** The column is `integer`, and a sub-millisecond run still ran. */ + it('yields a whole number of milliseconds, floored at one', () => { + const { sql } = render(new Date('2026-08-13T12:00:05.000Z')) + + expect(sql).toContain('GREATEST(1,') + expect(sql).toContain('ROUND(') + expect(sql).toContain('::integer') + }) + + /** + * An untimed run cancelled after ~24.8 days exceeds `integer`. Without the + * ceiling the cast raises, and the terminal write is lost entirely — the row + * stays `running` with no end timestamp, which is worse than a saturated + * duration. + */ + it('saturates at the column ceiling instead of overflowing the cast', () => { + const { sql, params } = render(new Date('2026-08-13T12:00:05.000Z')) + + expect(sql).toContain('LEAST(') + expect(params).toContain(2_147_483_647) + }) + + /** + * A paused run records its *active* duration at the pause checkpoint. Elapsed + * wall clock through a later cancel includes the time it sat waiting, so + * overwriting would silently redefine what the column means for that run. + */ + it('keeps the duration a paused run already recorded', () => { + const { sql } = render(new Date('2026-08-13T12:00:05.000Z')) + + expect(sql).toContain(`"status" = 'pending' THEN COALESCE(`) + expect(sql).toContain('"total_duration_ms"') + }) + + /** + * Resuming flips the row back to `running` and leaves the checkpoint value + * behind, so a resumed run carries a stale duration while it is accruing time + * again. Preserving it would freeze a cancelled run at its pre-resume reading. + */ + it('recomputes for a running row rather than trusting a stale checkpoint', () => { + const { sql } = render(new Date('2026-08-13T12:00:05.000Z')) + + const elseBranch = sql.slice(sql.indexOf('ELSE')) + expect(elseBranch).toContain('LEAST(') + expect(elseBranch).not.toContain('COALESCE(') + expect(elseBranch).not.toContain('"total_duration_ms"') + }) +}) diff --git a/apps/sim/lib/logs/execution/duration.ts b/apps/sim/lib/logs/execution/duration.ts new file mode 100644 index 0000000000..59fc4a2ebf --- /dev/null +++ b/apps/sim/lib/logs/execution/duration.ts @@ -0,0 +1,49 @@ +import { workflowExecutionLogs } from '@sim/db/schema' +import { type SQL, sql } from 'drizzle-orm' + +/** + * Elapsed run time for a terminal write that does not go through + * `completeWorkflowExecution`, expressed against the row's own `started_at`. + * + * Cancellation writes the log row directly rather than through the completion + * path, so it has no in-memory duration to store. Deriving it in the same + * statement keeps `ended_at` and `total_duration_ms` describing one instant, + * and keeps a cancelled run visible to the duration filters on + * `GET /api/v2/logs` — a null there reads as "no duration recorded" and drops + * the run out of every `minDurationMs`/`maxDurationMs` query. + * + * `ended_at` is bound as an explicit `timestamp` rather than a `Date`, because + * `started_at` is `timestamp without time zone` holding a UTC wall clock: an + * ISO string casts to the same naive reading, while a driver-bound `Date` + * would infer `timestamptz` and make the interval depend on the session zone. + * + * Floored at 1ms to match `completeWorkflowExecution`, so a cancellation that + * lands inside the same millisecond as the start still records that it ran. + * + * Saturated at the column's own ceiling rather than left to overflow. The + * column is `integer`, so an untimed run cancelled after ~24.8 days would + * otherwise raise `numeric_value_out_of_range` — which costs more than the + * duration it was recording: the direct write is caught and logged, leaving + * the row `running` with no end timestamp at all, and the workflow-group write + * fails its transaction and takes the whole cancellation with it. A saturated + * duration is wrong in the last digit; a failed terminal write is wrong about + * whether the run ended. + * + * A duration a *paused* run already recorded wins, and only that one. Pausing + * writes the run's active duration at the checkpoint, which elapsed wall clock + * through a later cancel would redefine to include the time it sat waiting. + * + * The status is what distinguishes it, not merely the column being populated: + * resuming flips the row back to `running` and leaves that checkpoint value + * behind, so a resumed run carries a stale duration while it is once again + * accruing time. Keeping it there would freeze a cancelled run at its + * pre-resume reading and disagree with the resume completion path, which + * measures wall clock. A `running` row therefore always recomputes; only a + * `pending` one — paused, and not accruing — keeps what it has. + */ +const INT4_MAX_MS = 2_147_483_647 + +export function elapsedDurationMsSql(endedAt: Date): SQL { + const elapsed = sql`LEAST(${INT4_MAX_MS}, GREATEST(1, ROUND(EXTRACT(EPOCH FROM (${endedAt.toISOString()}::timestamp - ${workflowExecutionLogs.startedAt})) * 1000)))::integer` + return sql`CASE WHEN ${workflowExecutionLogs.status} = 'pending' THEN COALESCE(${workflowExecutionLogs.totalDurationMs}, ${elapsed}) ELSE ${elapsed} END` +} diff --git a/apps/sim/lib/table/workflow-group-cancellation.test.ts b/apps/sim/lib/table/workflow-group-cancellation.test.ts index 70e33d088a..6514827e2e 100644 --- a/apps/sim/lib/table/workflow-group-cancellation.test.ts +++ b/apps/sim/lib/table/workflow-group-cancellation.test.ts @@ -80,6 +80,7 @@ describe('cancelWorkflowGroupExecution', () => { expect(dbChainMockFns.set).toHaveBeenNthCalledWith(1, { status: 'cancelled', endedAt: expect.any(Date), + totalDurationMs: expect.anything(), executionDeadlineAt: null, }) expect(dbChainMockFns.set).toHaveBeenNthCalledWith(2, { @@ -218,9 +219,15 @@ describe('cancelWorkflowGroupExecution', () => { }) expect(dbChainMockFns.update).toHaveBeenCalledOnce() + /** + * `totalDurationMs` is derived in-statement from the row's `started_at`, so + * a cancelled run carries the duration every other terminal write records + * and stays visible to the `/api/v2/logs` duration filters. + */ expect(dbChainMockFns.set).toHaveBeenCalledWith({ status: 'cancelled', endedAt: expect.any(Date), + totalDurationMs: expect.anything(), executionDeadlineAt: null, }) const logUpdateValues = collectConditionValues(dbChainMockFns.where.mock.calls[2]?.[0]) diff --git a/apps/sim/lib/table/workflow-group-cancellation.ts b/apps/sim/lib/table/workflow-group-cancellation.ts index 3f097af058..4b67d3c2b2 100644 --- a/apps/sim/lib/table/workflow-group-cancellation.ts +++ b/apps/sim/lib/table/workflow-group-cancellation.ts @@ -3,6 +3,7 @@ import { tableRowExecutions, userTableDefinitions, workflowExecutionLogs } from import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { and, eq, inArray } from 'drizzle-orm' +import { elapsedDurationMsSql } from '@/lib/logs/execution/duration' import { appendTableEvent } from '@/lib/table/events' const logger = createLogger('WorkflowGroupCancellation') @@ -149,9 +150,15 @@ export async function cancelWorkflowGroupExecution( return { result: { kind: 'already_cancelled_without_sidecar' } as const } } + const cancelledAt = new Date() const [cancelledLog] = await tx .update(workflowExecutionLogs) - .set({ status: 'cancelled', endedAt: new Date(), executionDeadlineAt: null }) + .set({ + status: 'cancelled', + endedAt: cancelledAt, + totalDurationMs: elapsedDurationMsSql(cancelledAt), + executionDeadlineAt: null, + }) .where( and( eq(workflowExecutionLogs.workspaceId, options.workspaceId), @@ -187,7 +194,12 @@ export async function cancelWorkflowGroupExecution( if (workflowLogActive) { const [cancelledLog] = await tx .update(workflowExecutionLogs) - .set({ status: 'cancelled', endedAt: now, executionDeadlineAt: null }) + .set({ + status: 'cancelled', + endedAt: now, + totalDurationMs: elapsedDurationMsSql(now), + executionDeadlineAt: null, + }) .where( and( eq(workflowExecutionLogs.workspaceId, options.workspaceId), diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts index 30a1706697..5754349fc6 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.test.ts @@ -1019,6 +1019,7 @@ describe('PauseResumeManager paused cancellation after pause release', () => { expect(dbChainMockFns.set).toHaveBeenCalledWith({ status: 'cancelled', endedAt: expect.any(Date), + totalDurationMs: expect.anything(), executionDeadlineAt: null, }) const casConditions = flattenMockConditions(dbChainMockFns.where.mock.calls.at(-1)?.[0]) 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 75951205a4..a10c75122a 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 @@ -33,6 +33,7 @@ import { } from '@/lib/execution/payloads/large-value-metadata' import { compactBlockLogs, compactExecutionPayload } from '@/lib/execution/payloads/serializer' import { preprocessExecution } from '@/lib/execution/preprocessing' +import { elapsedDurationMsSql } from '@/lib/logs/execution/duration' import { LoggingSession } from '@/lib/logs/execution/logging-session' import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server' import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core' @@ -2615,7 +2616,12 @@ export class PauseResumeManager { if (!cancellationAlreadyTerminal) { const [cancelledExecution] = await tx .update(workflowExecutionLogs) - .set({ status: 'cancelled', endedAt: now, executionDeadlineAt: null }) + .set({ + status: 'cancelled', + endedAt: now, + totalDurationMs: elapsedDurationMsSql(now), + executionDeadlineAt: null, + }) .where( and( eq(workflowExecutionLogs.executionId, executionId),