improvement(wait): tighten poll route and pause-manager helpers

- Parallelize per-row dispatch with Promise.all
- Add status='paused' guard on nextResumeAt rewrite to prevent clobbering concurrent resumes
- Extract computeEarliestResumeAt + PauseResumeManager.setNextResumeAt helpers
- Use canonical PausePoint type in poll route (drop StoredPausePoint)
- Narrow UNIT_TO_MS via as const + WaitUnit guard
- Bump LOCK_TTL_SECONDS above route maxDuration
- Clearer error when allowedPauseKinds rejects a resume

Author: TheodoreSpeaks

apps/sim/app/api/resume/poll/route.ts

@@ -8,22 +8,30 @@ import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { acquireLock, releaseLock } from '@/lib/core/config/redis'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
+import {
+  computeEarliestResumeAt,
+  PauseResumeManager,
+} from '@/lib/workflows/executor/human-in-the-loop-manager'
+import type { PausePoint } from '@/executor/types'
 
 const logger = createLogger('TimePauseResumePoll')
 
 export const dynamic = 'force-dynamic'
 export const maxDuration = 120
 
 const LOCK_KEY = 'time-pause-resume-poll-lock'
-const LOCK_TTL_SECONDS = 120
+const LOCK_TTL_SECONDS = 180
 const POLL_BATCH_LIMIT = 200
 
-interface StoredPausePoint {
-  contextId?: string
-  resumeStatus?: string
-  pauseKind?: string
-  resumeAt?: string
+interface DispatchFailure {
+  executionId: string
+  contextId: string
+  error: string
+}
+
+interface RowResult {
+  dispatched: number
+  failures: DispatchFailure[]
 }
 
 export const GET = withRouteHandler(async (request: NextRequest) => {
@@ -40,10 +48,6 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     )
   }
 
-  let claimedRows = 0
-  let dispatched = 0
-  const failures: { executionId: string; contextId: string; error: string }[] = []
-
   try {
     const now = new Date()
 
@@ -65,89 +69,21 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       )
       .limit(POLL_BATCH_LIMIT)
 
-    claimedRows = dueRows.length
-
-    for (const row of dueRows) {
-      const points = (row.pausePoints ?? {}) as Record<string, StoredPausePoint>
-      const metadata = (row.metadata ?? {}) as Record<string, unknown>
-      const userId = typeof metadata.executorUserId === 'string' ? metadata.executorUserId : ''
-
-      const duePoints: StoredPausePoint[] = []
-      let nextRemaining: Date | null = null
-
-      for (const point of Object.values(points)) {
-        if (point.pauseKind !== 'time' || !point.resumeAt) continue
-        if (point.resumeStatus && point.resumeStatus !== 'paused') continue
-
-        const resumeAt = new Date(point.resumeAt)
-        if (Number.isNaN(resumeAt.getTime())) continue
-
-        if (resumeAt <= now) {
-          duePoints.push(point)
-        } else if (!nextRemaining || resumeAt < nextRemaining) {
-          nextRemaining = resumeAt
-        }
-      }
-
-      for (const point of duePoints) {
-        const contextId = point.contextId
-        if (!contextId) continue
-        try {
-          const enqueueResult = await PauseResumeManager.enqueueOrStartResume({
-            executionId: row.executionId,
-            contextId,
-            resumeInput: {},
-            userId,
-          })
-
-          if (enqueueResult.status === 'starting') {
-            PauseResumeManager.startResumeExecution({
-              resumeEntryId: enqueueResult.resumeEntryId,
-              resumeExecutionId: enqueueResult.resumeExecutionId,
-              pausedExecution: enqueueResult.pausedExecution,
-              contextId: enqueueResult.contextId,
-              resumeInput: enqueueResult.resumeInput,
-              userId: enqueueResult.userId,
-            }).catch((error) => {
-              logger.error('Background time-pause resume failed', {
-                executionId: row.executionId,
-                contextId,
-                error: toError(error).message,
-              })
-            })
-          }
-          dispatched++
-        } catch (error) {
-          const message = toError(error).message
-          logger.warn('Failed to dispatch time-pause resume', {
-            executionId: row.executionId,
-            contextId,
-            error: message,
-          })
-          failures.push({ executionId: row.executionId, contextId, error: message })
-        }
-      }
-
-      // We never auto-retry a failed dispatch: workflow blocks aren't idempotent, and an
-      // operator must investigate stranded rows by hand. Setting nextResumeAt to the next
-      // future pause (or null) drops the row out of the poll, surfacing the failure.
-      await db
-        .update(pausedExecutions)
-        .set({ nextResumeAt: nextRemaining })
-        .where(eq(pausedExecutions.id, row.id))
-    }
+    const results = await Promise.all(dueRows.map((row) => dispatchRow(row, now)))
+    const dispatched = results.reduce((sum, r) => sum + r.dispatched, 0)
+    const failures = results.flatMap((r) => r.failures)
 
     logger.info('Time-pause resume poll completed', {
       requestId,
-      claimedRows,
+      claimedRows: dueRows.length,
       dispatched,
       failureCount: failures.length,
     })
 
     return NextResponse.json({
       success: true,
       requestId,
-      claimedRows,
+      claimedRows: dueRows.length,
       dispatched,
       failures,
     })
@@ -159,3 +95,79 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     await releaseLock(LOCK_KEY, requestId).catch(() => {})
   }
 })
+
+interface DueRow {
+  id: string
+  executionId: string
+  workflowId: string
+  pausePoints: unknown
+  metadata: unknown
+}
+
+async function dispatchRow(row: DueRow, now: Date): Promise<RowResult> {
+  const points = (row.pausePoints ?? {}) as Record<string, PausePoint>
+  const metadata = (row.metadata ?? {}) as Record<string, unknown>
+  const userId = typeof metadata.executorUserId === 'string' ? metadata.executorUserId : ''
+
+  const eligiblePoints = Object.values(points).filter(
+    (point) =>
+      point.pauseKind === 'time' && (!point.resumeStatus || point.resumeStatus === 'paused')
+  )
+  const duePoints = eligiblePoints.filter((point) => {
+    if (!point.resumeAt) return false
+    const at = new Date(point.resumeAt)
+    return !Number.isNaN(at.getTime()) && at <= now
+  })
+
+  const failures: DispatchFailure[] = []
+  let dispatched = 0
+
+  for (const point of duePoints) {
+    if (!point.contextId) continue
+    try {
+      const enqueueResult = await PauseResumeManager.enqueueOrStartResume({
+        executionId: row.executionId,
+        contextId: point.contextId,
+        resumeInput: {},
+        userId,
+      })
+
+      if (enqueueResult.status === 'starting') {
+        PauseResumeManager.startResumeExecution({
+          resumeEntryId: enqueueResult.resumeEntryId,
+          resumeExecutionId: enqueueResult.resumeExecutionId,
+          pausedExecution: enqueueResult.pausedExecution,
+          contextId: enqueueResult.contextId,
+          resumeInput: enqueueResult.resumeInput,
+          userId: enqueueResult.userId,
+        }).catch((error) => {
+          logger.error('Background time-pause resume failed', {
+            executionId: row.executionId,
+            contextId: point.contextId,
+            error: toError(error).message,
+          })
+        })
+      }
+      dispatched++
+    } catch (error) {
+      const message = toError(error).message
+      logger.warn('Failed to dispatch time-pause resume', {
+        executionId: row.executionId,
+        contextId: point.contextId,
+        error: message,
+      })
+      failures.push({ executionId: row.executionId, contextId: point.contextId, error: message })
+    }
+  }
+
+  // We never auto-retry a failed dispatch: workflow blocks aren't idempotent, and
+  // an operator must investigate stranded rows by hand. The status='paused' guard
+  // also prevents clobbering when a concurrent manual resume has already advanced
+  // the row's state since we read it.
+  await PauseResumeManager.setNextResumeAt({
+    pausedExecutionId: row.id,
+    nextResumeAt: computeEarliestResumeAt(eligiblePoints, { after: now }),
+  })
+
+  return { dispatched, failures }
+}

apps/sim/executor/handlers/wait/wait-handler.ts

@@ -75,11 +75,17 @@ const sleep = async (ms: number, options: SleepOptions = {}): Promise<boolean> =
   })
 }
 
-const UNIT_TO_MS: Record<string, number> = {
+const UNIT_TO_MS = {
   seconds: 1000,
   minutes: 60 * 1000,
   hours: 60 * 60 * 1000,
   days: 24 * 60 * 60 * 1000,
+} as const satisfies Record<string, number>
+
+type WaitUnit = keyof typeof UNIT_TO_MS
+
+function isWaitUnit(value: string): value is WaitUnit {
+  return value in UNIT_TO_MS
 }
 
 /**
@@ -122,12 +128,10 @@ export class WaitBlockHandler implements BlockHandler {
       throw new Error('Wait amount must be a positive number')
     }
 
-    const unitMs = UNIT_TO_MS[timeUnit]
-    if (!unitMs) {
+    if (!isWaitUnit(timeUnit)) {
       throw new Error(`Unknown wait unit: ${timeUnit}`)
     }
-
-    const waitMs = timeValue * unitMs
+    const waitMs = timeValue * UNIT_TO_MS[timeUnit]
 
     if (waitMs > MAX_WAIT_MS) {
       throw new Error('Wait time exceeds maximum of 30 days')

apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts

@@ -126,6 +126,27 @@ interface StartResumeExecutionArgs {
   abortSignal?: AbortSignal
 }
 
+/**
+ * Returns the earliest `resumeAt` across `pauseKind: 'time'` pause points whose
+ * `resumeAt` is a valid date and (when `after` is provided) strictly later than it.
+ * Returns `null` when no candidate exists.
+ */
+export function computeEarliestResumeAt(
+  points: Iterable<Pick<PausePoint, 'pauseKind' | 'resumeAt'>>,
+  options: { after?: Date } = {}
+): Date | null {
+  const { after } = options
+  let earliest: Date | null = null
+  for (const point of points) {
+    if (point.pauseKind !== 'time' || !point.resumeAt) continue
+    const candidate = new Date(point.resumeAt)
+    if (Number.isNaN(candidate.getTime())) continue
+    if (after && candidate <= after) continue
+    if (!earliest || candidate < earliest) earliest = candidate
+  }
+  return earliest
+}
+
 export class PauseResumeManager {
   static async persistPauseResult(args: PersistPauseResultArgs): Promise<void> {
     const { workflowId, executionId, pausePoints, snapshotSeed, executorUserId } = args
@@ -147,13 +168,7 @@ export class PauseResumeManager {
       return acc
     }, {})
 
-    const nextResumeAt = pausePoints.reduce<Date | null>((earliest, point) => {
-      if (point.pauseKind !== 'time' || !point.resumeAt) return earliest
-      const candidate = new Date(point.resumeAt)
-      if (Number.isNaN(candidate.getTime())) return earliest
-      if (!earliest || candidate < earliest) return candidate
-      return earliest
-    }, null)
+    const nextResumeAt = computeEarliestResumeAt(pausePoints)
 
     const now = new Date()
 
@@ -232,7 +247,9 @@ export class PauseResumeManager {
 
       const pauseKind: PauseKind = pausePoint.pauseKind ?? 'human'
       if (allowedPauseKinds && !allowedPauseKinds.includes(pauseKind)) {
-        throw new Error(`Pause point cannot be resumed manually (pauseKind=${pauseKind})`)
+        throw new Error(
+          `Pause kind '${pauseKind}' is not allowed for this resume endpoint (allowed: ${allowedPauseKinds.join(', ')})`
+        )
       }
 
       const activeResume = await tx
@@ -1329,6 +1346,23 @@ export class PauseResumeManager {
     })
   }
 
+  /**
+   * Updates `next_resume_at` only when the row is still `status='paused'`.
+   * Guard prevents the cron poller from clobbering a freshly-written value when a
+   * concurrent manual resume has already advanced the row's state.
+   */
+  static async setNextResumeAt(args: {
+    pausedExecutionId: string
+    nextResumeAt: Date | null
+  }): Promise<void> {
+    await db
+      .update(pausedExecutions)
+      .set({ nextResumeAt: args.nextResumeAt })
+      .where(
+        and(eq(pausedExecutions.id, args.pausedExecutionId), eq(pausedExecutions.status, 'paused'))
+      )
+  }
+
   static async listPausedExecutions(options: {
     workflowId: string
     status?: string | string[]