diff --git a/apps/sim/lib/workflows/executor/execution-queries.ts b/apps/sim/lib/workflows/executor/execution-queries.ts index 03bef36954a..69ed459a5ac 100644 --- a/apps/sim/lib/workflows/executor/execution-queries.ts +++ b/apps/sim/lib/workflows/executor/execution-queries.ts @@ -1,7 +1,7 @@ import { db } from '@sim/db' import { pausedExecutions, workflowExecutionLogs } from '@sim/db/schema' import { and, asc, desc, eq, gt, gte, lt, lte, or, sql } from 'drizzle-orm' -import { PAUSE_POINT_ROWS } from '@/lib/workflows/executor/pause-point-sql' +import { somePausePointMatches } from '@/lib/workflows/executor/pause-point-sql' export type WorkflowExecutionStatus = | 'pending' @@ -27,6 +27,10 @@ export interface ListWorkflowExecutionsInput { order: 'asc' | 'desc' } +/** A point a resume can still act on: any `resumeStatus` but `resumed` or `failed`, absent included. */ +const POINT_UNSETTLED = + '!(@.resumeStatus.type() == "string" && (@.resumeStatus == "resumed" || @.resumeStatus == "failed"))' + /** * An open pause record reads `paused` unless the log has finished and no point is left to resume, * the same rule `readWorkflowExecutionStatus` applies to a single run. @@ -35,10 +39,7 @@ const executionStatus = sql`CASE WHEN ${pausedExecutions.status} IN ('paused', 'partially_resumed') AND NOT ( ${workflowExecutionLogs.status} IN ('completed', 'failed', 'cancelled') - AND NOT EXISTS ( - select 1 from ${PAUSE_POINT_ROWS} - where coalesce(point.value ->> 'resumeStatus', 'paused') NOT IN ('resumed', 'failed') - ) + AND NOT ${somePausePointMatches(POINT_UNSETTLED)} ) THEN 'paused' ELSE ${workflowExecutionLogs.status} diff --git a/apps/sim/lib/workflows/executor/pause-point-sql.ts b/apps/sim/lib/workflows/executor/pause-point-sql.ts index 372b84b9b01..eb1e2cf87d8 100644 --- a/apps/sim/lib/workflows/executor/pause-point-sql.ts +++ b/apps/sim/lib/workflows/executor/pause-point-sql.ts @@ -1,21 +1,19 @@ import { pausedExecutions } from '@sim/db/schema' -import { sql } from 'drizzle-orm' +import { type SQL, sql } from 'drizzle-orm' /** - * Every pause point of a `paused_executions` row, as the derived table `point(value)`. + * Whether any pause point of a `paused_executions` row matches the jsonpath `filter`, which + * reads the point as `@`. * * `pause_points` is an object keyed by context id, or an array on legacy rows; both shapes are - * read, as `normalizePausePoints` reads them, and each set function only ever receives the shape - * it accepts. + * read, as `normalizePausePoints` reads them. A `jsonb_path_exists` call per row keeps this cheap + * enough to evaluate in a `WHERE` across every run of a workflow, which a correlated + * set-returning subquery is not. */ -export const PAUSE_POINT_ROWS = sql`( - select value from jsonb_each( - case when jsonb_typeof(${pausedExecutions.pausePoints}) = 'object' - then ${pausedExecutions.pausePoints} else '{}'::jsonb end - ) - union all - select value from jsonb_array_elements( - case when jsonb_typeof(${pausedExecutions.pausePoints}) = 'array' - then ${pausedExecutions.pausePoints} else '[]'::jsonb end - ) -) as point(value)` +export function somePausePointMatches(filter: string): SQL { + return sql`(case jsonb_typeof(${pausedExecutions.pausePoints}) + when 'object' then jsonb_path_exists(${pausedExecutions.pausePoints}, ${`$.* ? (${filter})`}::jsonpath) + when 'array' then jsonb_path_exists(${pausedExecutions.pausePoints}, ${`$[*] ? (${filter})`}::jsonpath) + else false + end)` +} diff --git a/apps/sim/lib/workflows/executor/paused-runs.ts b/apps/sim/lib/workflows/executor/paused-runs.ts index 07219f5c584..5ecd14f08fe 100644 --- a/apps/sim/lib/workflows/executor/paused-runs.ts +++ b/apps/sim/lib/workflows/executor/paused-runs.ts @@ -1,6 +1,6 @@ import { db } from '@sim/db' import { pausedExecutions, workflow } from '@sim/db/schema' -import { and, eq, inArray, isNull, sql } from 'drizzle-orm' +import { and, eq, inArray, isNull } from 'drizzle-orm' import { type CursorKey, encodeKeyset, @@ -14,7 +14,7 @@ import { resumeKeyset, textKey, } from '@/lib/api/list-query' -import { PAUSE_POINT_ROWS } from '@/lib/workflows/executor/pause-point-sql' +import { somePausePointMatches } from '@/lib/workflows/executor/pause-point-sql' /** A paused run of an active workflow, with the raw `pause_points` its detail is built from. */ interface WorkspacePausedRunRow { @@ -32,12 +32,15 @@ const PAUSED_RUN_KEYS: readonly KeysetKey[] = [ textKey(pausedExecutions.id, (row) => row.id), ] -/** Whether `point.value` is a pause point still waiting on a human, with `mapPausePoints`' defaults. */ -const POINT_WAITS_ON_HUMAN = sql`coalesce(point.value ->> 'pauseKind', 'human') <> 'time' - and coalesce(point.value ->> 'resumeStatus', 'paused') = 'paused'` +/** + * A pause point still waiting on a human, with `mapPausePoints`' defaults: a missing or null + * `pauseKind` reads `human` and a missing or null `resumeStatus` reads `paused`. + */ +const POINT_WAITS_ON_HUMAN = + '!(@.pauseKind.type() == "string" && @.pauseKind == "time") && (!exists(@.resumeStatus) || @.resumeStatus == null || (@.resumeStatus.type() == "string" && @.resumeStatus == "paused"))' /** A paused run with a pause point still waiting on a human. */ -const WAITS_ON_HUMAN = sql`exists (select 1 from ${PAUSE_POINT_ROWS} where ${POINT_WAITS_ON_HUMAN})` +const WAITS_ON_HUMAN = somePausePointMatches(POINT_WAITS_ON_HUMAN) /** The cursor that resumes the list after `row`. */ export function pausedRunCursorKeys(row: WorkspacePausedRunRow): CursorKey[] {