Function executeDatabaseScheduleJob

apps/sim/app/api/schedules/execute/route.ts:560–613

(
  jobQueue: JobQueue,
  jobId: string,
  payload: ScheduleExecutionPayload,
  schedule: DatabaseScheduleExecutionTarget,
  queuedAt: Date,
  requestId: string,
  delayMs: number
)

}

async function executeDatabaseScheduleJob(
  jobQueue: JobQueue,
  jobId: string,
  payload: ScheduleExecutionPayload,
  schedule: DatabaseScheduleExecutionTarget,
  queuedAt: Date,
  requestId: string,
  delayMs: number
): Promise<void> {
  if (delayMs > 0) await sleep(delayMs)

  const startResult = await runWithDatabaseScheduleStartTurn(() =>
    tryStartDatabaseScheduleJob(jobId)
  )
  if (startResult === 'not_pending') {
    logger.info(`[${requestId}] Database schedule execution job is no longer pending`, {
      scheduleId: schedule.id,
      workflowId: schedule.workflowId,
      jobId,
    })
    return
  }

  if (startResult === 'capacity_full') {
    logger.info(`[${requestId}] Deferred database schedule execution because capacity is full`, {
      scheduleId: schedule.id,
      workflowId: schedule.workflowId,
      jobId,
      concurrencyLimit: SCHEDULE_EXECUTION_CONCURRENCY_LIMIT,
    })
    return
  }

  try {
    const output = await executeScheduleJob(payload)
    await jobQueue.completeJob(jobId, output ?? null)
  } catch (error) {
    const errorMessage = toError(error).message
    logger.error(`[${requestId}] Schedule execution failed for workflow ${schedule.workflowId}`, {
      scheduleId: schedule.id,
      jobId,
      error: errorMessage,
    })
    await jobQueue.markJobFailed(jobId, errorMessage)
    await releaseScheduleLock(
      schedule.id,
      requestId,
      new Date(),
      `Failed to release lock for schedule ${schedule.id} after inline execution failure`,
      undefined,
      { expectedLastQueuedAt: queuedAt }
    )
  }
}

async function getPendingDatabaseScheduleJobs(limit: number) {
  if (limit <= 0) return []