fix(cron): add staleness takeover to single-flight guard

Author: TheodoreSpeaks

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

@@ -1072,7 +1072,11 @@ async function processJobItem(job: ClaimedJob, queuedAt: Date, requestId: string
   }
 }
 
-const scheduleTickGuard = createSingleFlight()
+/**
+ * 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

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

@@ -36,7 +36,7 @@ describe('runDetached', () => {
 
 describe('createSingleFlight', () => {
   it('starts work and reports active while in flight', async () => {
-    const guard = createSingleFlight()
+    const guard = createSingleFlight({ staleAfterMs: 60_000 })
     let release: () => void = () => {}
     const gate = new Promise<void>((resolve) => {
       release = resolve
@@ -52,7 +52,7 @@ describe('createSingleFlight', () => {
   })
 
   it('refuses a second run while one is already in flight', async () => {
-    const guard = createSingleFlight()
+    const guard = createSingleFlight({ staleAfterMs: 60_000 })
     let release: () => void = () => {}
     const gate = new Promise<void>((resolve) => {
       release = resolve
@@ -67,10 +67,54 @@ describe('createSingleFlight', () => {
   })
 
   it('clears the active flag even when work rejects', async () => {
-    const guard = createSingleFlight()
+    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

@@ -25,34 +25,61 @@ export function runDetached(label: string, work: () => Promise<unknown>): void {
     })
 }
 
+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() {
-  let active = false
+export function createSingleFlight({ staleAfterMs }: SingleFlightOptions) {
+  let activeToken: symbol | null = null
+  let activeSince = 0
 
   return {
-    /** Whether a run is currently in flight in this process. */
-    isActive: (): boolean => active,
+    /** Whether a run currently holds the slot in this process. */
+    isActive: (): boolean => activeToken !== null,
 
     /**
-     * Starts `work` detached if no run is active.
+     * Starts `work` detached unless a non-stale run already holds the slot.
      *
-     * @returns `true` if a new run started, `false` if one was already in flight.
+     * @returns `true` if a new run started, `false` if a run was already in flight.
      */
     run(label: string, work: () => Promise<unknown>): boolean {
-      if (active) return false
-      active = true
+      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(() => {
-            active = false
+            if (activeToken === token) {
+              activeToken = null
+            }
           })
       )
       return true