fix(api): match pause points with jsonb_path_exists so a status-filtered run list stays fast (#9971)

Author: waleedlatif1

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<string>`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}

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)`
+}

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<WorkspacePausedRunRow>[] = [
   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[] {