fix(schedules): rotate deferred carriers and stop the redacting sweep failing live runs

Addresses the review findings on the reconciliation hardening.

A carrier whose accounting was deferred got no write at all, so it kept its
`updatedAt` and stayed at the head of the `updatedAt`-ordered recovery batch on
every tick, starving every other claimed carrier and never becoming eligible for
retention. This was a regression from the previous commit: settling used to bump
the timestamp for every examined row, and skipping the settle for already-terminal
carriers removed the bump with it. A payload with no `scheduledFor` can never
reconcile, so such a row pinned a batch slot permanently. Bump `updatedAt`
unconditionally and keep only the reconciled marker conditional.

The stale-execution sweep terminalized `redacting` logs on the execution deadline.
That deadline bounds execution, while `redacting` covers payload masking after the
run already finished -- so a run that used most of its budget entered redaction
with the deadline due, and the sweep failed it five minutes later while the worker
was still masking. Schedule recovery then read the log as a failed occurrence and
counted a failure that never happened, even though the worker's terminal write
later restored `completed`. Sweep `redacting` on the generic stale window only.

`getScheduleNextRunAt` falls back to a daily cadence when a schedule has no cron
expression. Deployment cannot persist such a schedule, so the branch is
unreachable, but this change widened its use from failure recovery to every
outcome -- log a warning when it fires rather than silently guessing a cadence.

`executionDeadlineAt` was missing from the shared `workflowExecutionLogs` schema
mock, so it read as `undefined` and assertions comparing against that column were
trivially true. Add it.

Author: waleedlatif1

apps/sim/app/api/cron/cleanup-stale-executions/route.test.ts

@@ -126,11 +126,10 @@ describe('stale execution cleanup deadline grace', () => {
             condition.right.getTime() === expectedThreshold.getTime()
         )
 
-      expect(deadlineComparisons).toHaveLength(3)
+      expect(deadlineComparisons).toHaveLength(2)
       expect(deadlineComparisons.map(({ right }) => right)).toEqual([
         expectedThreshold,
         expectedThreshold,
-        expectedThreshold,
       ])
 
       const executionUpdateIndex = dbChainMockFns.update.mock.calls.findIndex(
@@ -169,6 +168,35 @@ describe('stale execution cleanup deadline grace', () => {
     }
   })
 
+  it('sweeps redacting logs on the generic window, never the execution deadline', async () => {
+    const response = await GET(createRequest())
+
+    expect(response.status).toBe(200)
+    const redactingPredicates = dbChainMockFns.where.mock.calls
+      .map(([condition]) => flattenConditions(condition))
+      .filter((conditions) =>
+        conditions.some(
+          (condition) =>
+            condition.type === 'eq' &&
+            condition.left === workflowExecutionLogs.status &&
+            condition.right === 'redacting'
+        )
+      )
+
+    expect(redactingPredicates.length).toBeGreaterThan(0)
+    for (const conditions of redactingPredicates) {
+      expect(
+        conditions.some((condition) => condition.left === workflowExecutionLogs.executionDeadlineAt)
+      ).toBe(false)
+      expect(
+        conditions.some(
+          (condition) =>
+            condition.type === 'lt' && condition.left === workflowExecutionLogs.startedAt
+        )
+      ).toBe(true)
+    }
+  })
+
   it('terminalizes stale running and redacting execution logs', async () => {
     const response = await GET(createRequest())
 

apps/sim/app/api/cron/cleanup-stale-executions/route.ts

@@ -27,7 +27,10 @@ import {
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import type { DbTransaction } from '@/lib/db/types'
 import { elapsedDurationMsSql } from '@/lib/logs/execution/duration'
-import { STALE_SWEEPABLE_EXECUTION_STATUSES } from '@/lib/logs/types'
+import {
+  STALE_SWEEPABLE_EXECUTION_STATUSES,
+  type StaleSweepableExecutionStatus,
+} from '@/lib/logs/types'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
 import {
   carrierNotIrrecoverableSql,
@@ -148,13 +151,28 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
     let currentWorkflowBatchSize = 0
 
     try {
-      const staleExecutionTimePredicate = or(
-        lt(workflowExecutionLogs.executionDeadlineAt, staleDeadlineThreshold),
-        and(
-          isNull(workflowExecutionLogs.executionDeadlineAt),
-          lt(workflowExecutionLogs.startedAt, staleThreshold)
-        )
-      )
+      /**
+       * `running` is swept on its execution deadline plus the cleanup grace, or
+       * on the generic stale window when it has no deadline.
+       *
+       * `redacting` gets the generic window only. That status is set *after* the
+       * run finished, while a live worker masks the payload, so the execution
+       * deadline is already in the past by the time redaction starts — the
+       * deadline rule would fail a masking pass that is merely slow, and
+       * schedule recovery would read that as a failed occurrence even though
+       * the worker is about to persist `completed`. A redaction that genuinely
+       * crashed still clears within the generic window.
+       */
+      const staleExecutionTimePredicate = (status: StaleSweepableExecutionStatus) =>
+        status === 'redacting'
+          ? lt(workflowExecutionLogs.startedAt, staleThreshold)
+          : or(
+              lt(workflowExecutionLogs.executionDeadlineAt, staleDeadlineThreshold),
+              and(
+                isNull(workflowExecutionLogs.executionDeadlineAt),
+                lt(workflowExecutionLogs.startedAt, staleThreshold)
+              )
+            )
       const cleanupTimestamp = sql.param(now, workflowExecutionLogs.startedAt)
       const staleDurationMinutes = sql<number>`ROUND(
         EXTRACT(EPOCH FROM (${cleanupTimestamp} - ${workflowExecutionLogs.startedAt})) / 60
@@ -169,7 +187,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       for (const executionStatus of STALE_SWEEPABLE_EXECUTION_STATUSES) {
         const staleExecutionPredicate = and(
           eq(workflowExecutionLogs.status, executionStatus),
-          staleExecutionTimePredicate
+          staleExecutionTimePredicate(executionStatus)
         )
         while (workflowRowsConsidered < WORKFLOW_EXECUTION_MAX_ROWS_PER_RUN) {
           const limit = Math.min(
@@ -194,7 +212,18 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
                 endedAt: now,
                 executionDeadlineAt: null,
                 totalDurationMs,
-                executionData: sql`jsonb_set(
+                executionData:
+                  executionStatus === 'redacting'
+                    ? sql`jsonb_set(
+                  COALESCE(execution_data, '{}'::jsonb),
+                  ARRAY['error'],
+                  to_jsonb(
+                    ${'Execution terminated: worker timeout or crash after '}::text
+                      || ${staleDurationMinutes}::text
+                      || ' minutes'
+                  )
+                )`
+                    : sql`jsonb_set(
                   COALESCE(execution_data, '{}'::jsonb),
                   ARRAY['error'],
                   to_jsonb(

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

@@ -789,6 +789,49 @@ describe('Scheduled Workflow Execution API Route', () => {
     expect(mockExecuteScheduleJob).not.toHaveBeenCalled()
   })
 
+  it('rotates a deferred carrier to the back of the batch instead of starving it', async () => {
+    mockShouldExecuteInline.mockReturnValue(true)
+    const claimedAt = new Date('2025-01-01T00:00:00.000Z')
+    const newerClaim = new Date('2025-01-01T00:05:00.000Z')
+    mockProcessingCounts(0, 0)
+    orderByLimitMock.mockResolvedValueOnce([
+      {
+        id: 'completed-job-id',
+        status: 'completed',
+        payload: {
+          scheduleId: 'schedule-1',
+          workflowId: 'workflow-1',
+          executionId: 'execution-1',
+          now: claimedAt.toISOString(),
+          scheduledFor: claimedAt.toISOString(),
+        },
+      },
+    ])
+    queueTableRows(mockWorkflowExecutionLogs, [
+      { executionId: 'execution-1', workflowId: 'workflow-1', status: 'completed' },
+    ])
+    queueTableRows(mockWorkflowSchedule, [
+      {
+        archivedAt: null,
+        lastQueuedAt: newerClaim,
+        nextRunAt: claimedAt,
+        status: 'active',
+      },
+    ])
+    mockApplyScheduleSuccessUpdate.mockResolvedValueOnce(false)
+
+    await runScheduleTick('test-request-id')
+
+    const carrierWrites = dbChainMockFns.set.mock.calls
+      .map(([values]) => values as Record<string, unknown>)
+      .filter((values) => values.updatedAt instanceof Date)
+
+    expect(carrierWrites).toContainEqual({ updatedAt: expect.any(Date) })
+    expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
+      expect.objectContaining({ metadata: expect.anything() })
+    )
+  })
+
   it('marks malformed terminal carriers irrecoverable so they leave the recovery batch', async () => {
     mockShouldExecuteInline.mockReturnValue(true)
     mockProcessingCounts(0, 0)

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

@@ -474,14 +474,29 @@ function staleScheduleExecutionJobsFilter(now: Date) {
   )
 }
 
+/**
+ * Recovery cadence for an occurrence the worker never got to account for.
+ *
+ * Deployment refuses to persist a schedule without a valid cron expression
+ * (`deployScheduleBlocks` writes `validateScheduleBlock`'s `cronExpression`,
+ * and an invalid one fails the deploy), so every reachable schedule takes the
+ * cron branch and lands on its real cadence. The day fallback exists only so a
+ * row that somehow violates that invariant still advances instead of re-firing
+ * the same occurrence forever — log it, because the cadence it picks is a
+ * guess.
+ */
 function getScheduleNextRunAt(
-  schedule: { cronExpression?: string | null; timezone?: string },
+  schedule: { scheduleId?: string; cronExpression?: string | null; timezone?: string },
   now: Date
 ): Date {
-  return (
-    getNextRunFromCronExpression(schedule.cronExpression, schedule.timezone) ??
-    new Date(now.getTime() + 24 * 60 * 60 * 1000)
-  )
+  const nextRunAt = getNextRunFromCronExpression(schedule.cronExpression, schedule.timezone)
+  if (nextRunAt) return nextRunAt
+
+  logger.warn('Recovering a schedule with no cron expression; falling back to a daily cadence', {
+    scheduleId: schedule.scheduleId,
+    timezone: schedule.timezone,
+  })
+  return new Date(now.getTime() + 24 * 60 * 60 * 1000)
 }
 
 function isTerminalJobStatus(status: string): boolean {
@@ -983,15 +998,24 @@ async function recoverStaleDatabaseScheduleJobs(now: Date): Promise<void> {
       })
 
       if (disabled) disabledScheduleIds.add(payload.scheduleId)
-      if (reconciled) {
-        await tx
-          .update(asyncJobs)
-          .set({
-            metadata: buildCarrierReconciledMetadata(asyncJobs.metadata),
-            updatedAt: now,
-          })
-          .where(eq(asyncJobs.id, row.id))
-      }
+
+      /**
+       * `updatedAt` is bumped even when accounting was deferred. The batch is
+       * ordered by `updatedAt`, so a carrier that keeps failing to reconcile —
+       * a newer claim holds the schedule, or the payload carries no
+       * `scheduledFor` to match an occurrence against — would otherwise sit at
+       * the head of every tick forever and starve every other claimed carrier
+       * out of the batch. Only the reconciled marker gates retention; the bump
+       * just rotates the row to the back of the queue.
+       */
+      await tx
+        .update(asyncJobs)
+        .set(
+          reconciled
+            ? { metadata: buildCarrierReconciledMetadata(asyncJobs.metadata), updatedAt: now }
+            : { updatedAt: now }
+        )
+        .where(eq(asyncJobs.id, row.id))
     }
   })
 

apps/sim/lib/logs/types.ts

@@ -243,6 +243,8 @@ export const STALE_SWEEPABLE_EXECUTION_STATUSES = [
   'redacting',
 ] as const satisfies readonly PersistedWorkflowExecutionStatus[]
 
+export type StaleSweepableExecutionStatus = (typeof STALE_SWEEPABLE_EXECUTION_STATUSES)[number]
+
 export interface CompletedWorkflowExecutionLog extends WorkflowExecutionLog {
   persistedStatus: PersistedWorkflowExecutionStatus
 }

packages/testing/src/mocks/schema.mock.ts

@@ -187,6 +187,7 @@ export const schemaMock = {
     status: 'status',
     trigger: 'trigger',
     startedAt: 'startedAt',
+    executionDeadlineAt: 'executionDeadlineAt',
     endedAt: 'endedAt',
     totalDurationMs: 'totalDurationMs',
     executionData: 'executionData',