improvement(cron): drop single-flight guard, rely on DB row claiming

Author: TheodoreSpeaks

apps/sim/app/api/schedules/execute/route.test.ts

@@ -877,21 +877,5 @@ describe('Scheduled Workflow Execution API Route', () => {
       const data = await response.json()
       expect(data).toMatchObject({ status: 'started' })
     })
-
-    it('returns already_running while a tick is in flight in the same process', async () => {
-      // Both calls run synchronously up to their `return` (no awaits before the
-      // single-flight guard), so the detached tick from the first call cannot
-      // progress past its first `await` before the second call checks the guard.
-      const first = GET(createMockRequest())
-      const second = GET(createMockRequest())
-
-      const [firstData, secondData] = await Promise.all([
-        first.then((r) => r.json()),
-        second.then((r) => r.json()),
-      ])
-
-      expect(firstData).toMatchObject({ status: 'started' })
-      expect(secondData).toMatchObject({ status: 'already_running' })
-    })
   })
 })

apps/sim/app/api/schedules/execute/route.ts

@@ -14,7 +14,7 @@ import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
 import { JOB_STATUS, type Job } from '@/lib/core/async-jobs/types'
 import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
-import { createSingleFlight } from '@/lib/core/utils/background'
+import { runDetached } from '@/lib/core/utils/background'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import {
@@ -1072,12 +1072,6 @@ async function processJobItem(job: ClaimedJob, queuedAt: Date, requestId: string
   }
 }
 
-/**
- * A tick self-bounds at `MAX_TICK_DURATION_MS`; allow a grace window beyond that
- * before a hung tick is considered stale and a new one is allowed to take over.
- */
-const scheduleTickGuard = createSingleFlight({ staleAfterMs: MAX_TICK_DURATION_MS + 60_000 })
-
 interface ScheduleTickResult {
   processedCount: number
   totalSchedules: number
@@ -1191,11 +1185,11 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     return authError
   }
 
-  const started = scheduleTickGuard.run('schedule-execution-tick', () => runScheduleTick(requestId))
+  runDetached('schedule-execution-tick', () => runScheduleTick(requestId))
 
   const response = {
-    message: started ? 'Scheduled execution started' : 'Scheduled execution already in progress',
-    status: started ? 'started' : 'already_running',
+    message: 'Scheduled execution started',
+    status: 'started',
   } satisfies ExecuteSchedulesResponse
 
   return NextResponse.json(response, { status: 202 })

apps/sim/lib/api/contracts/schedules.ts

@@ -134,7 +134,7 @@ const messageResponseSchema = z.object({
 
 export const executeSchedulesResponseSchema = z.object({
   message: z.string(),
-  status: z.enum(['started', 'already_running']),
+  status: z.literal('started'),
 })
 
 export type ExecuteSchedulesResponse = z.output<typeof executeSchedulesResponseSchema>

apps/sim/lib/core/utils/background.test.ts

@@ -2,7 +2,7 @@
  * @vitest-environment node
  */
 import { describe, expect, it, vi } from 'vitest'
-import { createSingleFlight, runDetached } from '@/lib/core/utils/background'
+import { runDetached } from '@/lib/core/utils/background'
 
 const flushMicrotasks = () => new Promise((resolve) => setTimeout(resolve, 0))
 
@@ -33,88 +33,3 @@ describe('runDetached', () => {
     await flushMicrotasks()
   })
 })
-
-describe('createSingleFlight', () => {
-  it('starts work and reports active while in flight', async () => {
-    const guard = createSingleFlight({ staleAfterMs: 60_000 })
-    let release: () => void = () => {}
-    const gate = new Promise<void>((resolve) => {
-      release = resolve
-    })
-
-    const started = guard.run('task', () => gate)
-    expect(started).toBe(true)
-    expect(guard.isActive()).toBe(true)
-
-    release()
-    await flushMicrotasks()
-    expect(guard.isActive()).toBe(false)
-  })
-
-  it('refuses a second run while one is already in flight', async () => {
-    const guard = createSingleFlight({ staleAfterMs: 60_000 })
-    let release: () => void = () => {}
-    const gate = new Promise<void>((resolve) => {
-      release = resolve
-    })
-
-    expect(guard.run('task', () => gate)).toBe(true)
-    expect(guard.run('task', () => Promise.resolve())).toBe(false)
-
-    release()
-    await flushMicrotasks()
-    expect(guard.run('task', () => Promise.resolve())).toBe(true)
-  })
-
-  it('clears the active flag even when work rejects', async () => {
-    const guard = createSingleFlight({ staleAfterMs: 60_000 })
-
-    expect(guard.run('task', () => Promise.reject(new Error('boom')))).toBe(true)
-    await flushMicrotasks()
-    expect(guard.isActive()).toBe(false)
-  })
-
-  it('takes over a stale run whose work never settles', async () => {
-    const guard = createSingleFlight({ staleAfterMs: 10 })
-
-    // A run whose promise never settles — its `finally` never fires.
-    expect(guard.run('task', () => new Promise<void>(() => {}))).toBe(true)
-    expect(guard.run('task', () => Promise.resolve())).toBe(false)
-
-    await new Promise((resolve) => setTimeout(resolve, 20))
-
-    const second = vi.fn().mockResolvedValue(undefined)
-    expect(guard.run('task', second)).toBe(true)
-    await flushMicrotasks()
-    expect(second).toHaveBeenCalledTimes(1)
-    expect(guard.isActive()).toBe(false)
-  })
-
-  it('does not let a late stale run clear a newer run slot', async () => {
-    const guard = createSingleFlight({ staleAfterMs: 10 })
-
-    let releaseStale: () => void = () => {}
-    const stale = new Promise<void>((resolve) => {
-      releaseStale = resolve
-    })
-    expect(guard.run('task', () => stale)).toBe(true)
-
-    await new Promise((resolve) => setTimeout(resolve, 20))
-
-    // New run takes over the stale slot.
-    let releaseFresh: () => void = () => {}
-    const fresh = new Promise<void>((resolve) => {
-      releaseFresh = resolve
-    })
-    expect(guard.run('task', () => fresh)).toBe(true)
-
-    // The original stale run settling late must not release the newer slot.
-    releaseStale()
-    await flushMicrotasks()
-    expect(guard.isActive()).toBe(true)
-
-    releaseFresh()
-    await flushMicrotasks()
-    expect(guard.isActive()).toBe(false)
-  })
-})

apps/sim/lib/core/utils/background.ts

@@ -24,65 +24,3 @@ export function runDetached(label: string, work: () => Promise<unknown>): void {
       logger.error(`Background task failed: ${label}`, toError(error))
     })
 }
-
-interface SingleFlightOptions {
-  /**
-   * How long a run may hold the slot before it is treated as stale. A later
-   * `run` call past this window takes over and starts a fresh run, so a hung
-   * task (one whose promise never settles) cannot wedge the slot permanently.
-   * This is the in-process equivalent of a distributed lock's TTL.
-   */
-  staleAfterMs: number
-}
-
-/**
- * A per-process single-flight guard. Prevents a long-running detached task from
- * piling up when it is invoked again before the previous run finishes.
- *
- * This guards a single Node process only — cross-replica deduplication must be
- * handled by the underlying work (e.g. database row claiming or a distributed
- * lock).
- *
- * A held slot is released when its run settles, or — if the run hangs and never
- * settles — taken over by the next `run` call after `staleAfterMs`. Ownership is
- * tracked by token so a stale run that settles late cannot clear a newer run's
- * slot.
- */
-export function createSingleFlight({ staleAfterMs }: SingleFlightOptions) {
-  let activeToken: symbol | null = null
-  let activeSince = 0
-
-  return {
-    /** Whether a run currently holds the slot in this process. */
-    isActive: (): boolean => activeToken !== null,
-
-    /**
-     * Starts `work` detached unless a non-stale run already holds the slot.
-     *
-     * @returns `true` if a new run started, `false` if a run was already in flight.
-     */
-    run(label: string, work: () => Promise<unknown>): boolean {
-      const now = Date.now()
-      if (activeToken !== null) {
-        if (now - activeSince < staleAfterMs) return false
-        logger.warn(
-          `Single-flight "${label}" held for ${now - activeSince}ms (> ${staleAfterMs}ms); starting a new run`
-        )
-      }
-
-      const token = Symbol(label)
-      activeToken = token
-      activeSince = now
-      runDetached(label, () =>
-        Promise.resolve()
-          .then(work)
-          .finally(() => {
-            if (activeToken === token) {
-              activeToken = null
-            }
-          })
-      )
-      return true
-    },
-  }
-}