address comments

Author: icecrasher321

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

@@ -20,6 +20,7 @@ const {
   mockExecuteJobInline,
   mockFeatureFlags,
   mockEnqueue,
+  mockGetJob,
   mockStartJob,
   mockCompleteJob,
   mockMarkJobFailed,
@@ -34,6 +35,7 @@ const {
     isDev: true,
   },
   mockEnqueue: vi.fn().mockResolvedValue('job-id-1'),
+  mockGetJob: vi.fn().mockResolvedValue(null),
   mockStartJob: vi.fn().mockResolvedValue(undefined),
   mockCompleteJob: vi.fn().mockResolvedValue(undefined),
   mockMarkJobFailed: vi.fn().mockResolvedValue(undefined),
@@ -54,6 +56,7 @@ vi.mock('@/lib/core/config/feature-flags', () => mockFeatureFlags)
 vi.mock('@/lib/core/async-jobs', () => ({
   getJobQueue: vi.fn().mockResolvedValue({
     enqueue: mockEnqueue,
+    getJob: mockGetJob,
     startJob: mockStartJob,
     completeJob: mockCompleteJob,
     markJobFailed: mockMarkJobFailed,
@@ -265,6 +268,8 @@ describe('Scheduled Workflow Execution API Route', () => {
         requestId: 'test-request-id',
       }),
       expect.objectContaining({
+        jobId: expect.stringMatching(/^schedule_[0-9a-f]{32}$/),
+        concurrencyKey: expect.stringMatching(/^schedule_[0-9a-f]{32}$/),
         metadata: expect.objectContaining({
           workflowId: 'workflow-1',
           workspaceId: 'workspace-1',

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

@@ -1,11 +1,14 @@
 import { db, workflowDeploymentVersion, workflowSchedule } from '@sim/db'
 import { createLogger } from '@sim/logger'
+import { sha256Hex } from '@sim/security/hash'
 import { toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
+import { Cron } from 'croner'
 import { and, eq, inArray, isNull, lt, lte, ne, not, or, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
+import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import {
@@ -15,11 +18,13 @@ import {
 } from '@/background/schedule-execution'
 
 export const dynamic = 'force-dynamic'
+export const maxDuration = 3600
 
 const logger = createLogger('ScheduledExecuteAPI')
 const MAX_CRON_CLAIMS = 20
 const RESERVED_WORKFLOW_CLAIMS = 10
 const RESERVED_JOB_CLAIMS = MAX_CRON_CLAIMS - RESERVED_WORKFLOW_CLAIMS
+const STALE_SCHEDULE_CLAIM_MS = getMaxExecutionTimeout()
 
 const dueFilter = (queuedAt: Date) =>
   and(
@@ -29,7 +34,8 @@ const dueFilter = (queuedAt: Date) =>
     ne(workflowSchedule.status, 'completed'),
     or(
       isNull(workflowSchedule.lastQueuedAt),
-      lt(workflowSchedule.lastQueuedAt, workflowSchedule.nextRunAt)
+      lt(workflowSchedule.lastQueuedAt, workflowSchedule.nextRunAt),
+      lt(workflowSchedule.lastQueuedAt, new Date(queuedAt.getTime() - STALE_SCHEDULE_CLAIM_MS))
     )
   )
 
@@ -46,6 +52,22 @@ const workflowScheduleFilter = (queuedAt: Date) =>
 const jobScheduleFilter = (queuedAt: Date) =>
   and(dueFilter(queuedAt), eq(workflowSchedule.sourceType, 'job'))
 
+function buildScheduleExecutionJobId(schedule: {
+  id: string
+  nextRunAt?: Date | null
+  lastQueuedAt?: Date | null
+}): string {
+  const occurrence =
+    schedule.nextRunAt?.toISOString() ?? schedule.lastQueuedAt?.toISOString() ?? 'due'
+  return `schedule_${sha256Hex(`${schedule.id}:${occurrence}`).slice(0, 32)}`
+}
+
+function getNextRunFromCronExpression(cronExpression?: string | null): Date | null {
+  if (!cronExpression) return null
+  const cron = new Cron(cronExpression)
+  return cron.nextRun()
+}
+
 async function claimWorkflowSchedules(queuedAt: Date, limit: number) {
   if (limit <= 0) return []
 
@@ -182,12 +204,40 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       }
 
       try {
+        const scheduleJobId = buildScheduleExecutionJobId(schedule)
+        const existingJob = await jobQueue.getJob(scheduleJobId)
+        if (existingJob && ['pending', 'processing'].includes(existingJob.status)) {
+          logger.info(`[${requestId}] Schedule execution job already exists`, {
+            scheduleId: schedule.id,
+            jobId: scheduleJobId,
+            status: existingJob.status,
+          })
+          return
+        }
+        if (existingJob) {
+          logger.info(`[${requestId}] Releasing stale schedule claim for finished job`, {
+            scheduleId: schedule.id,
+            jobId: scheduleJobId,
+            status: existingJob.status,
+          })
+          await releaseScheduleLock(
+            schedule.id,
+            requestId,
+            queuedAt,
+            `Released stale schedule ${schedule.id} for finished job ${scheduleJobId}`,
+            getNextRunFromCronExpression(schedule.cronExpression)
+          )
+          return
+        }
+
         const resolvedWorkflow = schedule.workflowId
           ? await workflowUtils?.getWorkflowById(schedule.workflowId)
           : null
         const resolvedWorkspaceId = resolvedWorkflow?.workspaceId
 
         const jobId = await jobQueue.enqueue('schedule-execution', payload, {
+          jobId: scheduleJobId,
+          concurrencyKey: scheduleJobId,
           metadata: {
             workflowId: schedule.workflowId ?? undefined,
             workspaceId: resolvedWorkspaceId ?? undefined,
@@ -198,6 +248,23 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
           `[${requestId}] Queued schedule execution task ${jobId} for workflow ${schedule.workflowId}`
         )
 
+        const queuedJob = await jobQueue.getJob(jobId)
+        if (queuedJob && !['pending', 'processing'].includes(queuedJob.status)) {
+          logger.info(`[${requestId}] Schedule execution job already finished`, {
+            scheduleId: schedule.id,
+            jobId,
+            status: queuedJob.status,
+          })
+          await releaseScheduleLock(
+            schedule.id,
+            requestId,
+            queuedAt,
+            `Released stale schedule ${schedule.id} for finished job ${jobId}`,
+            getNextRunFromCronExpression(schedule.cronExpression)
+          )
+          return
+        }
+
         if (shouldExecuteInline()) {
           try {
             await jobQueue.startJob(jobId)

apps/sim/background/schedule-execution.ts

@@ -11,8 +11,13 @@ import { generateId } from '@sim/utils/id'
 import { task } from '@trigger.dev/sdk'
 import { Cron } from 'croner'
 import { and, eq, isNull } from 'drizzle-orm'
+import { getHighestPrioritySubscription } from '@/lib/billing/core/subscription'
 import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types'
-import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core/execution-limits'
+import {
+  createTimeoutAbortController,
+  getExecutionTimeout,
+  getTimeoutErrorMessage,
+} from '@/lib/core/execution-limits'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
@@ -1007,6 +1012,8 @@ export async function executeJobInline(payload: JobExecutionPayload) {
   const promptText = buildJobPrompt(jobRecord)
 
   try {
+    const userSubscription = await getHighestPrioritySubscription(jobRecord.sourceUserId)
+    const mothershipJobTimeoutMs = getExecutionTimeout(userSubscription?.plan, 'sync')
     const url = buildAPIUrl('/api/mothership/execute')
     const headers = await buildAuthHeaders(jobRecord.sourceUserId)
 
@@ -1018,16 +1025,52 @@ export async function executeJobInline(payload: JobExecutionPayload) {
     }
 
     const startTime = new Date()
-    const response = await fetch(url.toString(), {
-      method: 'POST',
-      headers,
-      body: JSON.stringify(body),
-    })
-    const endTime = new Date()
-    const durationMs = endTime.getTime() - startTime.getTime()
+    const timeoutController = createTimeoutAbortController(mothershipJobTimeoutMs)
+    try {
+      const response = await fetch(url.toString(), {
+        method: 'POST',
+        headers,
+        body: JSON.stringify(body),
+        signal: timeoutController.signal,
+      })
+
+      if (!response.ok) {
+        const errorText = await response.text().catch(() => {
+          if (timeoutController.isTimedOut()) {
+            throw new Error(getTimeoutErrorMessage(null, timeoutController.timeoutMs))
+          }
+          return 'Unknown error'
+        })
+        const endTime = new Date()
+        const durationMs = endTime.getTime() - startTime.getTime()
 
-    if (!response.ok) {
-      const errorText = await response.text().catch(() => 'Unknown error')
+        await createJobLogEntry({
+          scheduleId: payload.scheduleId,
+          workspaceId: jobRecord.sourceWorkspaceId,
+          jobTitle: jobRecord.jobTitle,
+          startTime,
+          endTime,
+          durationMs,
+          success: false,
+          errorMessage: errorText,
+        })
+
+        throw new Error(`Mothership execution failed (${response.status}): ${errorText}`)
+      }
+
+      let responseBody: Record<string, any> = {}
+      let wasCompletedByTool = false
+      try {
+        responseBody = await response.json()
+        const toolCalls = responseBody?.toolCalls as Array<{ name?: string }> | undefined
+        wasCompletedByTool = toolCalls?.some((tc) => tc.name === 'complete_job') ?? false
+      } catch {
+        if (timeoutController.isTimedOut()) {
+          throw new Error(getTimeoutErrorMessage(null, timeoutController.timeoutMs))
+        }
+      }
+      const endTime = new Date()
+      const durationMs = endTime.getTime() - startTime.getTime()
 
       await createJobLogEntry({
         scheduleId: payload.scheduleId,
@@ -1036,92 +1079,71 @@ export async function executeJobInline(payload: JobExecutionPayload) {
         startTime,
         endTime,
         durationMs,
-        success: false,
-        errorMessage: errorText,
+        success: true,
+        responseBody,
       })
 
-      throw new Error(`Mothership execution failed (${response.status}): ${errorText}`)
-    }
+      const newRunCount = (jobRecord.runCount || 0) + 1
 
-    let responseBody: Record<string, any> = {}
-    let wasCompletedByTool = false
-    try {
-      responseBody = await response.json()
-      const toolCalls = responseBody?.toolCalls as Array<{ name?: string }> | undefined
-      wasCompletedByTool = toolCalls?.some((tc) => tc.name === 'complete_job') ?? false
-    } catch {
-      // Response may not be JSON; proceed with normal flow
-    }
+      logger.info(`[${requestId}] Job executed successfully`, {
+        scheduleId: payload.scheduleId,
+        runCount: newRunCount,
+        wasCompletedByTool,
+      })
 
-    await createJobLogEntry({
-      scheduleId: payload.scheduleId,
-      workspaceId: jobRecord.sourceWorkspaceId,
-      jobTitle: jobRecord.jobTitle,
-      startTime,
-      endTime,
-      durationMs,
-      success: true,
-      responseBody,
-    })
+      if (wasCompletedByTool) {
+        await applyScheduleUpdate(
+          payload.scheduleId,
+          {
+            lastRanAt: now,
+            updatedAt: now,
+            runCount: newRunCount,
+            failedCount: 0,
+            lastQueuedAt: null,
+          },
+          requestId,
+          `Error updating job ${payload.scheduleId} after completion`
+        )
+        return
+      }
 
-    const newRunCount = (jobRecord.runCount || 0) + 1
+      const isOneTime = !jobRecord.cronExpression
+      let nextRunAt: Date | null = null
 
-    logger.info(`[${requestId}] Job executed successfully`, {
-      scheduleId: payload.scheduleId,
-      runCount: newRunCount,
-      wasCompletedByTool,
-    })
+      if (!isOneTime && jobRecord.cronExpression) {
+        const validation = validateCronExpression(
+          jobRecord.cronExpression,
+          jobRecord.timezone || 'UTC'
+        )
+        nextRunAt = validation.nextRun || null
+      }
+
+      const maxRunsReached = jobRecord.maxRuns && newRunCount >= jobRecord.maxRuns
+      if (maxRunsReached) {
+        logger.info(`[${requestId}] Job hit maxRuns limit`, {
+          scheduleId: payload.scheduleId,
+          maxRuns: jobRecord.maxRuns,
+          runCount: newRunCount,
+        })
+      }
 
-    if (wasCompletedByTool) {
       await applyScheduleUpdate(
         payload.scheduleId,
         {
           lastRanAt: now,
           updatedAt: now,
-          runCount: newRunCount,
+          nextRunAt: isOneTime || maxRunsReached ? null : nextRunAt,
           failedCount: 0,
           lastQueuedAt: null,
+          runCount: newRunCount,
+          status: isOneTime || maxRunsReached ? 'completed' : 'active',
         },
         requestId,
-        `Error updating job ${payload.scheduleId} after completion`
-      )
-      return
-    }
-
-    const isOneTime = !jobRecord.cronExpression
-    let nextRunAt: Date | null = null
-
-    if (!isOneTime && jobRecord.cronExpression) {
-      const validation = validateCronExpression(
-        jobRecord.cronExpression,
-        jobRecord.timezone || 'UTC'
+        `Error updating job ${payload.scheduleId} after success`
       )
-      nextRunAt = validation.nextRun || null
-    }
-
-    const maxRunsReached = jobRecord.maxRuns && newRunCount >= jobRecord.maxRuns
-    if (maxRunsReached) {
-      logger.info(`[${requestId}] Job hit maxRuns limit`, {
-        scheduleId: payload.scheduleId,
-        maxRuns: jobRecord.maxRuns,
-        runCount: newRunCount,
-      })
+    } finally {
+      timeoutController.cleanup()
     }
-
-    await applyScheduleUpdate(
-      payload.scheduleId,
-      {
-        lastRanAt: now,
-        updatedAt: now,
-        nextRunAt: isOneTime || maxRunsReached ? null : nextRunAt,
-        failedCount: 0,
-        lastQueuedAt: null,
-        runCount: newRunCount,
-        status: isOneTime || maxRunsReached ? 'completed' : 'active',
-      },
-      requestId,
-      `Error updating job ${payload.scheduleId} after success`
-    )
   } catch (error) {
     const errorMessage = toError(error).message
     logger.error(`[${requestId}] Job execution failed`, {