From bef36e17a732d5d50a9ede9d0700a6fa45d57ae1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 13 Aug 2026 19:40:23 -0700 Subject: [PATCH 1/2] fix(execution): give a cancelled async run its terminal metadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The staging integration suite has been failing integ-cancel-async-api-key/async-execution-becomes-cancelled: the run reports status cancelled with a null endedAt and a null duration, so the assertion fails and the dependent worker-stop check never runs. The run resource falls back to the queue job whenever no execution-log row exists yet, and a cancel that lands before the worker has written that row leaves exactly that state. Trigger.dev marks a run canceled the moment it accepts the cancellation but only stamps its finish time when the worker drains, so for the seconds in between the job is terminal with no timestamp, and the projection faithfully reports a terminal status with nothing to date it. The run writes a correct log row when it finally drains, which is why the endpoint heals itself and the alarm fires intermittently rather than always. The backend was discarding the one timestamp that is always present: the retrieve response carries a required updatedAt beside the optional finishedAt. A finish time still wins wherever it exists, so nothing that already reports correctly changes, and the fallback is taken only once the mapped status is terminal — an active run's updatedAt marks progress, and reading it as an end would retire a run that is still going. It records the server's last transition for the run rather than the reader's clock, so it stays put across polls instead of growing. Also carries the duration on a fifth cancellation write, in the internal cancel route, that the earlier pass missed because its sweep covered lib and not app. --- .../[executionId]/cancel/route.test.ts | 2 + .../executions/[executionId]/cancel/route.ts | 8 +- .../async-jobs/backends/trigger-dev.test.ts | 74 +++++++++++++++++++ .../core/async-jobs/backends/trigger-dev.ts | 38 +++++++++- .../executor/execution-status.test.ts | 1 + 5 files changed, 120 insertions(+), 3 deletions(-) diff --git a/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.test.ts b/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.test.ts index a8591a5f2ca..84ba7ec5d3a 100644 --- a/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.test.ts +++ b/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.test.ts @@ -1320,6 +1320,7 @@ describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => { expect(mockSet).toHaveBeenCalledWith({ status: 'cancelled', endedAt: expect.any(Date), + totalDurationMs: expect.anything(), executionDeadlineAt: null, }) }) @@ -1340,6 +1341,7 @@ describe('POST /api/workflows/[id]/executions/[executionId]/cancel', () => { expect(mockSet).toHaveBeenCalledWith({ status: 'cancelled', endedAt: expect.any(Date), + totalDurationMs: expect.anything(), executionDeadlineAt: null, }) }) diff --git a/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts b/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts index 9848732d8d3..eb2c3352bff 100644 --- a/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts +++ b/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts @@ -24,6 +24,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 { workflowExecutionOriginSql } from '@/lib/logs/execution-origin' import { captureServerEvent } from '@/lib/posthog/server' import { @@ -215,7 +216,12 @@ async function claimExecutionLogCancellation(args: { const now = new Date() const [cancelledExecution] = await db .update(workflowExecutionLogs) - .set({ status: 'cancelled', endedAt: now, executionDeadlineAt: null }) + .set({ + status: 'cancelled', + endedAt: now, + totalDurationMs: elapsedDurationMsSql(now), + executionDeadlineAt: null, + }) .where( and( eq(workflowExecutionLogs.executionId, args.executionId), diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts index 853503f5901..51337d9373f 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts @@ -331,6 +331,80 @@ describe('TriggerDevJobQueue status mapping', () => { await expect(queue.getJob('run-1')).resolves.toMatchObject({ status: jobStatus }) }) + + it('dates a run cancelled before it was dequeued by its last transition', async () => { + mockRetrieve.mockResolvedValueOnce({ + id: 'run-1', + payload: {}, + status: 'CANCELED', + taskIdentifier: 'workflow-execution', + createdAt: new Date('2026-08-05T12:00:00.000Z'), + updatedAt: new Date('2026-08-05T12:00:02.000Z'), + }) + const queue = new TriggerDevJobQueue() + + await expect(queue.getJob('run-1')).resolves.toMatchObject({ + status: 'cancelled', + startedAt: undefined, + completedAt: new Date('2026-08-05T12:00:02.000Z'), + }) + }) + + it('dates a run cancelled mid-flight by its last transition until it drains', async () => { + mockRetrieve.mockResolvedValueOnce({ + id: 'run-1', + payload: {}, + status: 'CANCELED', + taskIdentifier: 'workflow-execution', + createdAt: new Date('2026-08-05T12:00:00.000Z'), + startedAt: new Date('2026-08-05T12:00:01.000Z'), + updatedAt: new Date('2026-08-05T12:00:03.000Z'), + }) + const queue = new TriggerDevJobQueue() + + await expect(queue.getJob('run-1')).resolves.toMatchObject({ + status: 'cancelled', + startedAt: new Date('2026-08-05T12:00:01.000Z'), + completedAt: new Date('2026-08-05T12:00:03.000Z'), + }) + }) + + it('prefers the reported finish over the last transition once the run has drained', async () => { + mockRetrieve.mockResolvedValueOnce({ + id: 'run-1', + payload: {}, + status: 'COMPLETED', + taskIdentifier: 'workflow-execution', + createdAt: new Date('2026-08-05T12:00:00.000Z'), + startedAt: new Date('2026-08-05T12:00:01.000Z'), + finishedAt: new Date('2026-08-05T12:00:04.000Z'), + updatedAt: new Date('2026-08-05T12:00:09.000Z'), + }) + const queue = new TriggerDevJobQueue() + + await expect(queue.getJob('run-1')).resolves.toMatchObject({ + status: 'completed', + completedAt: new Date('2026-08-05T12:00:04.000Z'), + }) + }) + + it('leaves a still-running job with no completion instant', async () => { + mockRetrieve.mockResolvedValueOnce({ + id: 'run-1', + payload: {}, + status: 'EXECUTING', + taskIdentifier: 'workflow-execution', + createdAt: new Date('2026-08-05T12:00:00.000Z'), + startedAt: new Date('2026-08-05T12:00:01.000Z'), + updatedAt: new Date('2026-08-05T12:00:03.000Z'), + }) + const queue = new TriggerDevJobQueue() + + await expect(queue.getJob('run-1')).resolves.toMatchObject({ + status: 'processing', + completedAt: undefined, + }) + }) }) describe('TriggerDevJobQueue cancellation', () => { diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts index f031226a2a7..3c941d69b05 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts @@ -213,6 +213,38 @@ function mapTriggerDevStatus(status: string): JobStatus { } } +const TERMINAL_JOB_STATUSES: readonly JobStatus[] = [ + JOB_STATUS.COMPLETED, + JOB_STATUS.FAILED, + JOB_STATUS.CANCELLED, +] + +/** + * Dates the end of a run that trigger.dev already reports as terminal. + * + * A cancellation flips the run to `CANCELED` the moment it is accepted, but + * `finishedAt` is only stamped once the worker actually drains — seconds later, + * or never for a run that was cancelled before it was ever dequeued. A reader + * polling inside that window sees a terminal job carrying no completion + * instant, and every consumer that derives an end timestamp or an elapsed + * duration from it reports null. + * + * `updatedAt` is trigger.dev's own record of when the run last changed, so for + * a terminal run it dates that final transition rather than the read. It is + * consulted only once the status is terminal: an active run's `updatedAt` + * describes progress, not an ending, and reporting it would end a run that is + * still going. + */ +function resolveRunCompletedAt( + finishedAt: Date | string | undefined, + updatedAt: Date | string | undefined, + status: JobStatus +): Date | undefined { + if (finishedAt) return new Date(finishedAt) + if (!TERMINAL_JOB_STATUSES.includes(status)) return undefined + return updatedAt ? new Date(updatedAt) : undefined +} + /** * Adapter that wraps the trigger.dev SDK to conform to JobQueueBackend interface. */ @@ -379,14 +411,16 @@ export class TriggerDevJobQueue implements JobQueueBackend { : undefined, } + const status = mapTriggerDevStatus(run.status) + return { id: run.id, type: run.taskIdentifier as JobType, payload: run.payload, - status: mapTriggerDevStatus(run.status), + status, createdAt: run.createdAt ? new Date(run.createdAt) : new Date(), startedAt: run.startedAt ? new Date(run.startedAt) : undefined, - completedAt: run.finishedAt ? new Date(run.finishedAt) : undefined, + completedAt: resolveRunCompletedAt(run.finishedAt, run.updatedAt, status), attempts: run.attemptCount ?? 1, maxAttempts: 3, error: run.error?.message, diff --git a/apps/sim/lib/workflows/executor/execution-status.test.ts b/apps/sim/lib/workflows/executor/execution-status.test.ts index 5996631349e..94f83e9453b 100644 --- a/apps/sim/lib/workflows/executor/execution-status.test.ts +++ b/apps/sim/lib/workflows/executor/execution-status.test.ts @@ -128,6 +128,7 @@ describe('getWorkflowExecutionStatus queue projection', () => { status: 'cancelled', level: 'info', endedAt: '2026-08-05T12:00:01.000Z', + totalDurationMs: 1000, error: null, }) }) From 34ed113d6c403d7e104bb34dd19bb8e79b723664 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 13 Aug 2026 20:01:16 -0700 Subject: [PATCH 2/2] refactor(execution): derive a cancelled run's terminal fields in one place MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The five cancellation paths each hand-assembled the same four-key payload for the workflow-execution log, and one of them had already drifted: the direct cancel never cleared `execution_deadline_at`, leaving a cancelled row carrying the deadline of an attempt that had stopped running. Extract the payload so the key set cannot vary between them, and leave the paths themselves alone — they differ in handle, claim predicate, whether they read the row back, and what they do when the claim is lost, so they stay separate statements. Bind the end instant through the `started_at` column encoder rather than a pre-stringified ISO literal. `check:sql-date-binding` exists to enforce exactly that binding; the literal passed only because it was already a string. Collapse the duration expression to one `COALESCE` over a valueless-`ELSE` `CASE`, which builds the elapsed fragment once instead of in both branches, and reuse it for the stale-execution sweeper, which carried its own copy along with a second int4 ceiling constant. Document the invariants the fix depends on where a reader meets them: that `total_duration_ms` means wall clock for a terminal row and active time for a paused one, and that a terminal job must carry its transition instant. --- .../cleanup-stale-executions/route.test.ts | 29 +++++++---- .../cron/cleanup-stale-executions/route.ts | 7 +-- .../executions/[executionId]/cancel/route.ts | 9 +--- .../core/async-jobs/backends/trigger-dev.ts | 7 +-- apps/sim/lib/core/async-jobs/types.ts | 15 ++++++ .../execution/cancel-workflow-execution.ts | 8 +-- .../lib/logs/execution/cancellation.test.ts | 49 +++++++++++++++++++ apps/sim/lib/logs/execution/cancellation.ts | 23 +++++++++ apps/sim/lib/logs/execution/duration.test.ts | 35 +++++++++---- apps/sim/lib/logs/execution/duration.ts | 27 ++++++---- .../lib/table/workflow-group-cancellation.ts | 16 ++---- .../executor/human-in-the-loop-manager.ts | 9 +--- packages/db/schema.ts | 7 +++ 13 files changed, 167 insertions(+), 74 deletions(-) create mode 100644 apps/sim/lib/logs/execution/cancellation.test.ts create mode 100644 apps/sim/lib/logs/execution/cancellation.ts diff --git a/apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts b/apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts index a1422424ef7..a58e0bfa8ec 100644 --- a/apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts +++ b/apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts @@ -37,6 +37,21 @@ function flattenConditions(condition: unknown): MockCondition[] { return [node, ...(node.conditions?.flatMap((child) => flattenConditions(child)) ?? [])] } +function hasToSQL(value: unknown): value is { toSQL: () => { sql: string; params: unknown[] } } { + return typeof value === 'object' && value !== null && 'toSQL' in value +} + +/** + * Collects the leaves of a nested `sql` expression. The duration expression is + * built by a shared helper, so the values it binds sit one level below the + * fragment this route assembles rather than directly in its own params. + */ +function flattenSqlParams(expression: { sql: string; params: unknown[] }): unknown[] { + return expression.params.flatMap((param) => + hasToSQL(param) ? flattenSqlParams(param.toSQL()) : [param] + ) +} + function createRequest() { return createMockRequest( 'GET', @@ -96,11 +111,7 @@ describe('stale execution cleanup deadline grace', () => { 'toSQL' in value && value.toSQL().sql.includes('EXTRACT(EPOCH') ) - const totalDurationExpression = update.totalDurationMs.toSQL() - const cleanupTimestamp = totalDurationExpression.params.find( - (value): value is { toSQL: () => { sql: string; params: unknown[] } } => - typeof value === 'object' && value !== null && 'toSQL' in value - ) + const totalDurationLeaves = flattenSqlParams(update.totalDurationMs.toSQL()) expect(errorExpression.sql).toContain('CASE') expect(errorExpression.sql).toContain('IS NOT NULL') @@ -111,11 +122,9 @@ describe('stale execution cleanup deadline grace', () => { ) expect(staleDurationExpression?.toSQL().sql).toContain('ROUND') expect(staleDurationExpression?.toSQL().params).toContain(workflowExecutionLogs.startedAt) - expect(totalDurationExpression.sql).toContain('LEAST') - expect(totalDurationExpression.sql).toContain('ROUND') - expect(totalDurationExpression.params).toContain(2_147_483_647) - expect(totalDurationExpression.params).toContain(workflowExecutionLogs.startedAt) - expect(cleanupTimestamp?.toSQL().params).toEqual([new Date('2026-08-03T12:10:00.000Z')]) + expect(totalDurationLeaves).toContain(2_147_483_647) + expect(totalDurationLeaves).toContain(workflowExecutionLogs.startedAt) + expect(totalDurationLeaves).toContainEqual(new Date('2026-08-03T12:10:00.000Z')) expect(update.endedAt).toEqual(new Date('2026-08-03T12:10:00.000Z')) } finally { vi.useRealTimers() diff --git a/apps/sim/app/api/cron/cleanup-stale-executions/route.ts b/apps/sim/app/api/cron/cleanup-stale-executions/route.ts index fa3dabf0a03..f77e35a0741 100644 --- a/apps/sim/app/api/cron/cleanup-stale-executions/route.ts +++ b/apps/sim/app/api/cron/cleanup-stale-executions/route.ts @@ -25,6 +25,7 @@ import { } from '@/lib/core/execution-limits' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import type { DbTransaction } from '@/lib/db/types' +import { elapsedDurationMsSql } from '@/lib/logs/execution/duration' import { deleteFile } from '@/lib/uploads/core/storage-service' const logger = createLogger('CleanupStaleExecutions') @@ -33,7 +34,6 @@ const STALE_THRESHOLD_MS = getExecutionReservationTtlMs() const STALE_THRESHOLD_MINUTES = Math.ceil(STALE_THRESHOLD_MS / 60000) const GENERIC_STALE_PROCESSING_ERROR = `Job terminated: stuck in processing for more than ${STALE_THRESHOLD_MINUTES} minutes` const EXECUTION_DEADLINE_ERROR = getTimeoutErrorMessage(undefined) -const MAX_INT32 = 2_147_483_647 /** * Table jobs run as detached workers with progress heartbeats, independently of workflow timeout * policy. Preserve their historical 90-minute task window plus five-minute cleanup grace. @@ -154,10 +154,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { const staleDurationMinutes = sql`ROUND( EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) / 60 )::integer` - const totalDurationMs = sql`LEAST( - ${MAX_INT32}, - ROUND(EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) * 1000) - )::integer` + const totalDurationMs = elapsedDurationMsSql(now) let workflowRowsConsidered = 0 while (workflowRowsConsidered < WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN) { const limit = Math.min( diff --git a/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts b/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts index eb2c3352bff..d696a5e6f90 100644 --- a/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts +++ b/apps/sim/app/api/workflows/[id]/executions/[executionId]/cancel/route.ts @@ -24,7 +24,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 { cancelledExecutionLogFields } from '@/lib/logs/execution/cancellation' import { workflowExecutionOriginSql } from '@/lib/logs/execution-origin' import { captureServerEvent } from '@/lib/posthog/server' import { @@ -216,12 +216,7 @@ async function claimExecutionLogCancellation(args: { const now = new Date() const [cancelledExecution] = await db .update(workflowExecutionLogs) - .set({ - status: 'cancelled', - endedAt: now, - totalDurationMs: elapsedDurationMsSql(now), - executionDeadlineAt: null, - }) + .set(cancelledExecutionLogFields(now)) .where( and( eq(workflowExecutionLogs.executionId, args.executionId), diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts index 3c941d69b05..4b3960740de 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts @@ -17,6 +17,7 @@ import { type JobQueueBackend, type JobStatus, type JobType, + TERMINAL_JOB_STATUSES, validateMaxDurationSeconds, } from '@/lib/core/async-jobs/types' import { recordExecutionCancellationBackendResult } from '@/lib/core/execution-limits/metrics' @@ -213,12 +214,6 @@ function mapTriggerDevStatus(status: string): JobStatus { } } -const TERMINAL_JOB_STATUSES: readonly JobStatus[] = [ - JOB_STATUS.COMPLETED, - JOB_STATUS.FAILED, - JOB_STATUS.CANCELLED, -] - /** * Dates the end of a run that trigger.dev already reports as terminal. * diff --git a/apps/sim/lib/core/async-jobs/types.ts b/apps/sim/lib/core/async-jobs/types.ts index d19abb513bc..fb5facc4b13 100644 --- a/apps/sim/lib/core/async-jobs/types.ts +++ b/apps/sim/lib/core/async-jobs/types.ts @@ -29,6 +29,13 @@ export const JOB_STATUS = { export type JobStatus = (typeof JOB_STATUS)[keyof typeof JOB_STATUS] +/** The statuses a job cannot leave; every one of them requires a `completedAt`. */ +export const TERMINAL_JOB_STATUSES: readonly JobStatus[] = [ + JOB_STATUS.COMPLETED, + JOB_STATUS.FAILED, + JOB_STATUS.CANCELLED, +] + export type JobType = | 'workflow-execution' | 'schedule-execution' @@ -84,6 +91,14 @@ export interface Job { status: JobStatus createdAt: Date startedAt?: Date + /** + * When the job reached its current status, required whenever that status is + * one of `TERMINAL_JOB_STATUSES`. Consumers derive both an end timestamp and + * an elapsed duration from it, so a terminal job that omits it reports null + * for each. A backend reading an eventually-consistent source must supply its + * best-known transition instant rather than leaving this unset — never the + * time of the read, which grows on every poll. + */ completedAt?: Date attempts: number maxAttempts: number diff --git a/apps/sim/lib/execution/cancel-workflow-execution.ts b/apps/sim/lib/execution/cancel-workflow-execution.ts index 1f391c864a0..8d542d3bb27 100644 --- a/apps/sim/lib/execution/cancel-workflow-execution.ts +++ b/apps/sim/lib/execution/cancel-workflow-execution.ts @@ -12,7 +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 { cancelledExecutionLogFields } from '@/lib/logs/execution/cancellation' import { captureServerEvent } from '@/lib/posthog/server' import { cancelWorkflowGroupExecution, @@ -381,11 +381,7 @@ export async function cancelWorkflowExecution( const cancelledAt = new Date() await db .update(workflowExecutionLogs) - .set({ - status: 'cancelled', - endedAt: cancelledAt, - totalDurationMs: elapsedDurationMsSql(cancelledAt), - }) + .set(cancelledExecutionLogFields(cancelledAt)) .where( and( eq(workflowExecutionLogs.executionId, executionId), diff --git a/apps/sim/lib/logs/execution/cancellation.test.ts b/apps/sim/lib/logs/execution/cancellation.test.ts new file mode 100644 index 00000000000..4c2485cb08e --- /dev/null +++ b/apps/sim/lib/logs/execution/cancellation.test.ts @@ -0,0 +1,49 @@ +/** + * @vitest-environment node + */ + +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 { cancelledExecutionLogFields } = await import('@/lib/logs/execution/cancellation') + +describe('cancelledExecutionLogFields', () => { + /** + * The five cancellation paths spread this payload into their own `.set()`. + * Hand-assembling it at each one had already dropped `executionDeadlineAt` at + * a single site, so the point of the factory is that the key set cannot vary + * between them — a field added here without a reason reaches all five. + */ + it('writes exactly the terminal fields, deadline cleared', () => { + const endedAt = new Date('2026-08-13T12:00:05.000Z') + + const fields = cancelledExecutionLogFields(endedAt) + + expect(Object.keys(fields).sort()).toEqual([ + 'endedAt', + 'executionDeadlineAt', + 'status', + 'totalDurationMs', + ]) + expect(fields.status).toBe('cancelled') + expect(fields.endedAt).toBe(endedAt) + expect(fields.executionDeadlineAt).toBeNull() + }) + + /** `ended_at` and `total_duration_ms` must describe the same instant. */ + it('derives the duration from the same instant it ends the run at', () => { + const endedAt = new Date('2026-08-13T12:00:05.000Z') + + const { params } = new PgDialect().sqlToQuery( + cancelledExecutionLogFields(endedAt).totalDurationMs + ) + + expect(params).toContain(endedAt.toISOString()) + }) +}) diff --git a/apps/sim/lib/logs/execution/cancellation.ts b/apps/sim/lib/logs/execution/cancellation.ts new file mode 100644 index 00000000000..09e68df1d12 --- /dev/null +++ b/apps/sim/lib/logs/execution/cancellation.ts @@ -0,0 +1,23 @@ +import { elapsedDurationMsSql } from '@/lib/logs/execution/duration' + +/** + * The fields every terminal cancellation sets on a `workflow_execution_logs` + * row, ready to spread into `.set()`. + * + * The five cancellation paths — direct, workflow-group with and without a + * sidecar, paused, and the async cancel route — differ in their database + * handle, their claim predicate, whether they read the row back, and what they + * do when the claim is lost, so they remain separate statements. What they must + * not differ in is the row they leave behind, and hand-assembling this payload + * at each one had already dropped `executionDeadlineAt` at a single site, + * leaving a cancelled run still carrying the deadline of an attempt that had + * stopped running. + */ +export function cancelledExecutionLogFields(endedAt: Date) { + return { + status: 'cancelled' as const, + endedAt, + totalDurationMs: elapsedDurationMsSql(endedAt), + executionDeadlineAt: null, + } +} diff --git a/apps/sim/lib/logs/execution/duration.test.ts b/apps/sim/lib/logs/execution/duration.test.ts index 7729e6e0f60..65a55086d6c 100644 --- a/apps/sim/lib/logs/execution/duration.test.ts +++ b/apps/sim/lib/logs/execution/duration.test.ts @@ -29,14 +29,16 @@ describe('elapsedDurationMsSql', () => { }) /** - * `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. + * `started_at` is `timestamp without time zone` holding a UTC wall clock. The + * end instant is bound through that column's mapper, which emits a UTC ISO + * string; the explicit cast drops the offset to the same naive reading rather + * than leaving the interval to depend on how Postgres resolves the operator. */ 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(sql).toContain('::timestamp -') + expect(sql).not.toContain('::timestamptz') expect(params).toContain('2026-08-13T12:00:05.000Z') expect(params.some((param) => Array.isArray(param))).toBe(false) }) @@ -71,8 +73,7 @@ describe('elapsedDurationMsSql', () => { 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"') + expect(sql).toContain(`"status" = 'pending' THEN "workflow_execution_logs"."total_duration_ms"`) }) /** @@ -83,9 +84,23 @@ describe('elapsedDurationMsSql', () => { 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"') + const fallback = sql.slice(sql.indexOf('END,')) + expect(fallback).toContain('LEAST(') + expect(fallback).not.toContain('"total_duration_ms"') + }) + + /** + * The preserved-checkpoint rule and the elapsed fallback are one `COALESCE` + * over a valueless-`ELSE` `CASE`, not a `CASE` repeating the elapsed + * expression in both branches. `status` is `NOT NULL` and `started_at` is + * `NOT NULL`, so the `CASE` yields NULL exactly for a non-`pending` row or a + * `pending` row that never recorded one — the two cases that must fall + * through — and the fallback can never itself be NULL. + */ + it('builds the elapsed expression once', () => { + const { sql, params } = render(new Date('2026-08-13T12:00:05.000Z')) + + expect(sql.match(/LEAST\(/g)).toHaveLength(1) + expect(params).toHaveLength(2) }) }) diff --git a/apps/sim/lib/logs/execution/duration.ts b/apps/sim/lib/logs/execution/duration.ts index 59fc4a2ebf0..ca9cce7fb83 100644 --- a/apps/sim/lib/logs/execution/duration.ts +++ b/apps/sim/lib/logs/execution/duration.ts @@ -1,6 +1,8 @@ import { workflowExecutionLogs } from '@sim/db/schema' import { type SQL, sql } from 'drizzle-orm' +const INT4_MAX_MS = 2_147_483_647 + /** * Elapsed run time for a terminal write that does not go through * `completeWorkflowExecution`, expressed against the row's own `started_at`. @@ -12,13 +14,19 @@ import { type SQL, sql } from 'drizzle-orm' * `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. + * `ended_at` is bound through `started_at`'s own column mapper and cast to + * `timestamp`, because that column is `timestamp without time zone` holding a + * UTC wall clock. The cast is what pins the reading: the mapper emits a UTC ISO + * instant, and casting it drops the offset to the same naive wall clock the + * column stores. Left uncast the interval would depend on Postgres resolving + * the subtraction to the `timestamp` overload rather than `timestamptz`, and + * getting that wrong yields a duration in the wrong zone rather than an error. * - * Floored at 1ms to match `completeWorkflowExecution`, so a cancellation that - * lands inside the same millisecond as the start still records that it ran. + * Floored at 1ms to match the `Math.max(1, durationMs)` the completion path + * applies, so a cancellation landing inside the same millisecond as the start + * still records that it ran. The floor doubles as the guard against a negative + * interval, which is otherwise reachable whenever the clock that stamped + * `started_at` runs ahead of the one cancelling. * * 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 @@ -41,9 +49,8 @@ import { type SQL, sql } from 'drizzle-orm' * 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` + const endedAtParam = sql.param(endedAt, workflowExecutionLogs.startedAt) + const elapsed = sql`LEAST(${INT4_MAX_MS}, GREATEST(1, ROUND(EXTRACT(EPOCH FROM (${endedAtParam}::timestamp - ${workflowExecutionLogs.startedAt})) * 1000)))::integer` + return sql`COALESCE(CASE WHEN ${workflowExecutionLogs.status} = 'pending' THEN ${workflowExecutionLogs.totalDurationMs} END, ${elapsed})` } diff --git a/apps/sim/lib/table/workflow-group-cancellation.ts b/apps/sim/lib/table/workflow-group-cancellation.ts index 4b67d3c2b2e..1231cc2b43a 100644 --- a/apps/sim/lib/table/workflow-group-cancellation.ts +++ b/apps/sim/lib/table/workflow-group-cancellation.ts @@ -3,7 +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 { cancelledExecutionLogFields } from '@/lib/logs/execution/cancellation' import { appendTableEvent } from '@/lib/table/events' const logger = createLogger('WorkflowGroupCancellation') @@ -153,12 +153,7 @@ export async function cancelWorkflowGroupExecution( const cancelledAt = new Date() const [cancelledLog] = await tx .update(workflowExecutionLogs) - .set({ - status: 'cancelled', - endedAt: cancelledAt, - totalDurationMs: elapsedDurationMsSql(cancelledAt), - executionDeadlineAt: null, - }) + .set(cancelledExecutionLogFields(cancelledAt)) .where( and( eq(workflowExecutionLogs.workspaceId, options.workspaceId), @@ -194,12 +189,7 @@ export async function cancelWorkflowGroupExecution( if (workflowLogActive) { const [cancelledLog] = await tx .update(workflowExecutionLogs) - .set({ - status: 'cancelled', - endedAt: now, - totalDurationMs: elapsedDurationMsSql(now), - executionDeadlineAt: null, - }) + .set(cancelledExecutionLogFields(now)) .where( and( eq(workflowExecutionLogs.workspaceId, options.workspaceId), 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 a10c75122a5..48029178fdd 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,7 +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 { cancelledExecutionLogFields } from '@/lib/logs/execution/cancellation' 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' @@ -2616,12 +2616,7 @@ export class PauseResumeManager { if (!cancellationAlreadyTerminal) { const [cancelledExecution] = await tx .update(workflowExecutionLogs) - .set({ - status: 'cancelled', - endedAt: now, - totalDurationMs: elapsedDurationMsSql(now), - executionDeadlineAt: null, - }) + .set(cancelledExecutionLogFields(now)) .where( and( eq(workflowExecutionLogs.executionId, executionId), diff --git a/packages/db/schema.ts b/packages/db/schema.ts index e6f89597652..09f51b084a5 100644 --- a/packages/db/schema.ts +++ b/packages/db/schema.ts @@ -440,6 +440,13 @@ export const workflowExecutionLogs = pgTable( /** Absolute deadline for the current active attempt; cleared while paused or terminal. */ executionDeadlineAt: timestamp('execution_deadline_at'), endedAt: timestamp('ended_at'), + /** + * Wall clock from `started_at` for a terminal row; for a `pending` (paused) + * row, the active duration recorded at the checkpoint, which excludes the + * time the run sits waiting. Resuming leaves that checkpoint value in place + * while the row accrues time again, so a `running` row's value is stale + * until the next terminal write recomputes it. + */ totalDurationMs: integer('total_duration_ms'), /**