From 7e411ac162e6bcc6c05c78e541e76557010aba94 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Thu, 20 Mar 2025 15:07:19 +0000 Subject: [PATCH] New PENDING_VERSION system which now requires queues to exist at dequeue time --- .../app/components/runs/v3/TaskRunStatus.tsx | 14 +- apps/webapp/app/database-types.ts | 1 + apps/webapp/app/hooks/useFilterTasks.ts | 5 - apps/webapp/app/models/taskQueue.server.ts | 4 +- apps/webapp/app/models/taskRun.server.ts | 1 + .../v3/ApiRetrieveRunPresenter.server.ts | 9 +- .../v3/ApiRunListPresenter.server.ts | 3 + .../presenters/v3/TestTaskPresenter.server.ts | 3 +- .../route.tsx | 2 +- .../route.tsx | 5 - .../v3/marqs/sharedQueueConsumer.server.ts | 2 +- .../v3/services/alerts/deliverAlert.server.ts | 2 +- .../services/createBackgroundWorker.server.ts | 153 ++++++--- .../createDeployedBackgroundWorker.server.ts | 7 +- ...createDeploymentBackgroundWorker.server.ts | 105 ++++-- .../services/createTaskRunAttempt.server.ts | 2 +- .../v3/services/pauseEnvironment.server.ts | 2 +- .../app/v3/taskRunHeartbeatFailed.server.ts | 1 + apps/webapp/app/v3/taskStatus.ts | 1 + docs/v3-openapi.yaml | 6 +- .../migration.sql | 4 + .../migration.sql | 20 ++ .../database/prisma/schema.prisma | 7 +- .../run-engine/src/engine/db/worker.ts | 29 +- .../run-engine/src/engine/index.ts | 24 +- .../src/engine/systems/dequeueSystem.ts | 55 +-- ...orkerSystem.ts => pendingVersionSystem.ts} | 57 ++-- .../src/engine/systems/runAttemptSystem.ts | 2 +- .../src/engine/tests/notDeployed.test.ts | 150 -------- .../src/engine/tests/pendingVersion.test.ts | 319 ++++++++++++++++++ .../run-engine/src/engine/tests/setup.ts | 126 ++++++- .../run-engine/src/engine/workerCatalog.ts | 2 +- .../run-engine/tsconfig.src.json | 3 +- .../run-engine/tsconfig.test.json | 2 +- packages/core/src/v3/apiClient/runStream.ts | 3 + packages/core/src/v3/index.ts | 1 + packages/core/src/v3/schemas/api.ts | 2 + packages/core/src/v3/schemas/common.ts | 2 +- packages/core/src/v3/schemas/runEngine.ts | 1 + packages/core/src/v3/tryCatch.ts | 15 + 40 files changed, 793 insertions(+), 359 deletions(-) create mode 100644 internal-packages/database/prisma/migrations/20250320111737_add_pending_version_task_run_status/migration.sql create mode 100644 internal-packages/database/prisma/migrations/20250320130354_add_many_to_many_relationship_task_queue_background_worker/migration.sql rename internal-packages/run-engine/src/engine/systems/{waitingForWorkerSystem.ts => pendingVersionSystem.ts} (54%) delete mode 100644 internal-packages/run-engine/src/engine/tests/notDeployed.test.ts create mode 100644 internal-packages/run-engine/src/engine/tests/pendingVersion.test.ts create mode 100644 packages/core/src/v3/tryCatch.ts diff --git a/apps/webapp/app/components/runs/v3/TaskRunStatus.tsx b/apps/webapp/app/components/runs/v3/TaskRunStatus.tsx index 325d20d9e..dbb44fc99 100644 --- a/apps/webapp/app/components/runs/v3/TaskRunStatus.tsx +++ b/apps/webapp/app/components/runs/v3/TaskRunStatus.tsx @@ -21,6 +21,7 @@ import { cn } from "~/utils/cn"; export const allTaskRunStatuses = [ "DELAYED", "WAITING_FOR_DEPLOY", + "PENDING_VERSION", "PENDING", "EXECUTING", "RETRYING_AFTER_FAILURE", @@ -37,7 +38,7 @@ export const allTaskRunStatuses = [ ] as const satisfies Readonly>; export const filterableTaskRunStatuses = [ - "WAITING_FOR_DEPLOY", + "PENDING_VERSION", "DELAYED", "PENDING", "WAITING_TO_RESUME", @@ -56,7 +57,10 @@ export const filterableTaskRunStatuses = [ const taskRunStatusDescriptions: Record = { DELAYED: "Task has been delayed and is waiting to be executed.", PENDING: "Task is waiting to be executed.", - WAITING_FOR_DEPLOY: "Task needs to be deployed first to start executing.", + PENDING_VERSION: + "Task is waiting for a version update because it cannot execute without additional information (task, queue, etc.).", + WAITING_FOR_DEPLOY: + "Task is waiting for a version update because it cannot execute without additional information (task, queue, etc.).", EXECUTING: "Task is currently being executed.", RETRYING_AFTER_FAILURE: "Task is being reattempted after a failure.", WAITING_TO_RESUME: `You have used a "wait" function. When the wait is complete, the task will resume execution.`, @@ -73,6 +77,7 @@ const taskRunStatusDescriptions: Record = { export const QUEUED_STATUSES = [ "PENDING", + "PENDING_VERSION", "WAITING_FOR_DEPLOY", "DELAYED", ] satisfies TaskRunStatus[]; @@ -120,6 +125,7 @@ export function TaskRunStatusIcon({ return ; case "PENDING": return ; + case "PENDING_VERSION": case "WAITING_FOR_DEPLOY": return ; case "EXECUTING": @@ -158,6 +164,7 @@ export function runStatusClassNameColor(status: TaskRunStatus): string { case "PENDING": case "DELAYED": return "text-charcoal-500"; + case "PENDING_VERSION": case "WAITING_FOR_DEPLOY": return "text-amber-500"; case "EXECUTING": @@ -194,8 +201,9 @@ export function runStatusTitle(status: TaskRunStatus): string { return "Delayed"; case "PENDING": return "Queued"; + case "PENDING_VERSION": case "WAITING_FOR_DEPLOY": - return "Waiting for deploy"; + return "Pending version"; case "EXECUTING": return "Executing"; case "WAITING_TO_RESUME": diff --git a/apps/webapp/app/database-types.ts b/apps/webapp/app/database-types.ts index 6214843f6..1fcd822c0 100644 --- a/apps/webapp/app/database-types.ts +++ b/apps/webapp/app/database-types.ts @@ -29,6 +29,7 @@ export const TaskRunAttemptStatus = { export const TaskRunStatus = { PENDING: "PENDING", + PENDING_VERSION: "PENDING_VERSION", WAITING_FOR_DEPLOY: "WAITING_FOR_DEPLOY", EXECUTING: "EXECUTING", WAITING_TO_RESUME: "WAITING_TO_RESUME", diff --git a/apps/webapp/app/hooks/useFilterTasks.ts b/apps/webapp/app/hooks/useFilterTasks.ts index 6bb64a9d8..7b95cf812 100644 --- a/apps/webapp/app/hooks/useFilterTasks.ts +++ b/apps/webapp/app/hooks/useFilterTasks.ts @@ -4,7 +4,6 @@ type Task = { id: string; friendlyId: string; taskIdentifier: string; - exportName: string; filePath: string; triggerSource: string; }; @@ -17,10 +16,6 @@ export function useFilterTasks({ tasks }: { tasks: T[] }) { return true; } - if (task.exportName.toLowerCase().includes(text.toLowerCase())) { - return true; - } - if (task.filePath.toLowerCase().includes(text.toLowerCase())) { return true; } diff --git a/apps/webapp/app/models/taskQueue.server.ts b/apps/webapp/app/models/taskQueue.server.ts index a9c52caf9..ed3435556 100644 --- a/apps/webapp/app/models/taskQueue.server.ts +++ b/apps/webapp/app/models/taskQueue.server.ts @@ -1,4 +1,4 @@ -import { QueueOptions } from "@trigger.dev/core/v3/schemas"; +import { QueueManifest } from "@trigger.dev/core/v3/schemas"; import { TaskQueue } from "@trigger.dev/database"; import { prisma } from "~/db.server"; @@ -35,7 +35,7 @@ export async function findQueueInEnvironment( return; } - const queueConfig = QueueOptions.safeParse(task.queueConfig); + const queueConfig = QueueManifest.safeParse(task.queueConfig); if (queueConfig.success) { const taskQueueName = queueConfig.data.name diff --git a/apps/webapp/app/models/taskRun.server.ts b/apps/webapp/app/models/taskRun.server.ts index c0166515e..cfd13a424 100644 --- a/apps/webapp/app/models/taskRun.server.ts +++ b/apps/webapp/app/models/taskRun.server.ts @@ -125,6 +125,7 @@ export function batchTaskRunItemStatusForRunStatus( case TaskRunStatus.TIMED_OUT: return BatchTaskRunItemStatus.FAILED; case TaskRunStatus.PENDING: + case TaskRunStatus.PENDING_VERSION: case TaskRunStatus.WAITING_FOR_DEPLOY: case TaskRunStatus.WAITING_TO_RESUME: case TaskRunStatus.RETRYING_AFTER_FAILURE: diff --git a/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts index 35baef75c..ac6c226e8 100644 --- a/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts @@ -204,6 +204,9 @@ export class ApiRetrieveRunPresenter extends BasePresenter { case "DELAYED": { return "DELAYED"; } + case "PENDING_VERSION": { + return "PENDING_VERSION"; + } case "WAITING_FOR_DEPLOY": { return "WAITING_FOR_DEPLOY"; } @@ -257,7 +260,11 @@ export class ApiRetrieveRunPresenter extends BasePresenter { } static apiBooleanHelpersFromRunStatus(status: RunStatus) { - const isQueued = status === "QUEUED" || status === "WAITING_FOR_DEPLOY" || status === "DELAYED"; + const isQueued = + status === "QUEUED" || + status === "WAITING_FOR_DEPLOY" || + status === "DELAYED" || + status === "PENDING_VERSION"; const isExecuting = status === "EXECUTING" || status === "REATTEMPTING" || status === "FROZEN"; const isCompleted = status === "COMPLETED" || diff --git a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts index 78f95e324..15966e372 100644 --- a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts @@ -280,6 +280,9 @@ export class ApiRunListPresenter extends BasePresenter { switch (status) { case "DELAYED": return "DELAYED"; + case "PENDING_VERSION": { + return "PENDING_VERSION"; + } case "WAITING_FOR_DEPLOY": { return "WAITING_FOR_DEPLOY"; } diff --git a/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts b/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts index c38478176..7c768f407 100644 --- a/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts @@ -21,7 +21,7 @@ type Task = { id: string; taskIdentifier: string; filePath: string; - exportName: string; + exportName?: string; friendlyId: string; }; @@ -151,7 +151,6 @@ export class TestTaskPresenter { id: task.id, taskIdentifier: task.slug, filePath: task.filePath, - exportName: task.exportName, friendlyId: task.friendlyId, }; diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam._index/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam._index/route.tsx index 430d04e4d..25441b93e 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam._index/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam._index/route.tsx @@ -507,7 +507,7 @@ function TaskActivityGraph({ activity }: { activity: TaskActivity }) { isAnimationActive={false} /> - +
- {t.slug} diff --git a/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts b/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts index a3adcefdd..c13e5062a 100644 --- a/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts +++ b/apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts @@ -1688,7 +1688,7 @@ class SharedQueueTasks { task: { id: backgroundWorkerTask.slug, filePath: backgroundWorkerTask.filePath, - exportName: backgroundWorkerTask.exportName, + exportName: backgroundWorkerTask.exportName ?? backgroundWorkerTask.slug, }, attempt: { id: attempt.friendlyId, diff --git a/apps/webapp/app/v3/services/alerts/deliverAlert.server.ts b/apps/webapp/app/v3/services/alerts/deliverAlert.server.ts index 5eff1a6c5..4297f6a72 100644 --- a/apps/webapp/app/v3/services/alerts/deliverAlert.server.ts +++ b/apps/webapp/app/v3/services/alerts/deliverAlert.server.ts @@ -573,7 +573,7 @@ export class DeliverAlertService extends BaseService { alert.workerDeployment.worker?.tasks.map((task) => ({ id: task.slug, filePath: task.filePath, - exportName: task.exportName, + exportName: task.exportName ?? "@deprecated", triggerSource: task.triggerSource, })) ?? [], environment: { diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index f7eed4f41..7bde7d124 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -20,10 +20,12 @@ import { } from "../runQueue.server"; import { calculateNextBuildVersion } from "../utils/calculateNextBuildVersion"; import { clampMaxDuration } from "../utils/maxDuration"; -import { BaseService } from "./baseService.server"; +import { BaseService, ServiceValidationError } from "./baseService.server"; import { CheckScheduleService } from "./checkSchedule.server"; import { projectPubSub } from "./projectPubSub.server"; import { RegisterNextTaskScheduleInstanceService } from "./registerNextTaskScheduleInstance.server"; +import { tryCatch } from "@trigger.dev/core/v3"; +import { engine } from "../runEngine.server"; export class CreateBackgroundWorkerService extends BaseService { public async call( @@ -96,58 +98,97 @@ export class CreateBackgroundWorkerService extends BaseService { }); } - const tasksToBackgroundFiles = await createBackgroundFiles( - body.metadata.sourceFiles, - backgroundWorker, - environment, - this._prisma - ); - await createWorkerResources( - body.metadata, - backgroundWorker, - environment, - this._prisma, - tasksToBackgroundFiles - ); - await syncDeclarativeSchedules( - body.metadata.tasks, - backgroundWorker, - environment, - this._prisma + const [filesError, tasksToBackgroundFiles] = await tryCatch( + createBackgroundFiles( + body.metadata.sourceFiles, + backgroundWorker, + environment, + this._prisma + ) ); - try { - //send a notification that a new worker has been created - await projectPubSub.publish( - `project:${project.id}:env:${environment.id}`, - "WORKER_CREATED", - { - environmentId: environment.id, - environmentType: environment.type, - createdAt: backgroundWorker.createdAt, - taskCount: body.metadata.tasks.length, - type: "local", - } + if (filesError) { + logger.error("Error creating background worker files", { + error: filesError, + backgroundWorker, + environment, + }); + + throw new ServiceValidationError("Error creating background worker files"); + } + + const [resourcesError] = await tryCatch( + createWorkerResources( + body.metadata, + backgroundWorker, + environment, + this._prisma, + tasksToBackgroundFiles + ) + ); + + if (resourcesError) { + logger.error("Error creating worker resources", { + error: resourcesError, + backgroundWorker, + environment, + }); + throw new ServiceValidationError("Error creating worker resources"); + } + + const [schedulesError] = await tryCatch( + syncDeclarativeSchedules(body.metadata.tasks, backgroundWorker, environment, this._prisma) + ); + + if (schedulesError) { + logger.error("Error syncing declarative schedules", { + error: schedulesError, + backgroundWorker, + environment, + }); + throw new ServiceValidationError("Error syncing declarative schedules"); + } + + const [updateConcurrencyLimitsError] = await tryCatch( + updateEnvConcurrencyLimits(environment) + ); + + if (updateConcurrencyLimitsError) { + logger.error("Error updating environment concurrency limits", { + error: updateConcurrencyLimitsError, + backgroundWorker, + environment, + }); + } + + const [publishError] = await tryCatch( + projectPubSub.publish(`project:${project.id}:env:${environment.id}`, "WORKER_CREATED", { + environmentId: environment.id, + environmentType: environment.type, + createdAt: backgroundWorker.createdAt, + taskCount: body.metadata.tasks.length, + type: "local", + }) + ); + + if (publishError) { + logger.error("Error publishing WORKER_CREATED event", { + error: publishError, + backgroundWorker, + environment, + }); + } + + if (backgroundWorker.engine === "V2") { + const [schedulePendingVersionsError] = await tryCatch( + engine.scheduleEnqueueRunsForBackgroundWorker(backgroundWorker.id) ); - await updateEnvConcurrencyLimits(environment); - } catch (err) { - logger.error( - "Error publishing WORKER_CREATED event or updating global concurrency limits", - { - error: - err instanceof Error - ? { - name: err.name, - message: err.message, - stack: err.stack, - } - : err, - project, - environment, - backgroundWorker, - } - ); + if (schedulePendingVersionsError) { + logger.error("Error scheduling pending versions", { + error: schedulePendingVersionsError, + }); + } } return backgroundWorker; @@ -338,6 +379,20 @@ async function createWorkerQueue( runtimeEnvironmentId: worker.runtimeEnvironmentId, projectId: worker.projectId, type: queueType, + workers: { + connect: { + id: worker.id, + }, + }, + }, + }); + } else { + await prisma.taskQueue.update({ + where: { + id: taskQueue.id, + }, + data: { + workers: { connect: { id: worker.id } }, }, }); } diff --git a/apps/webapp/app/v3/services/createDeployedBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createDeployedBackgroundWorker.server.ts index a82bc1b85..4ec6b7f56 100644 --- a/apps/webapp/app/v3/services/createDeployedBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createDeployedBackgroundWorker.server.ts @@ -64,12 +64,7 @@ export class CreateDeployedBackgroundWorkerService extends BaseService { } try { - await createWorkerResources( - body.metadata.tasks, - backgroundWorker, - environment, - this._prisma - ); + await createWorkerResources(body.metadata, backgroundWorker, environment, this._prisma); await syncDeclarativeSchedules( body.metadata.tasks, backgroundWorker, diff --git a/apps/webapp/app/v3/services/createDeploymentBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createDeploymentBackgroundWorker.server.ts index 743103293..e85a59804 100644 --- a/apps/webapp/app/v3/services/createDeploymentBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createDeploymentBackgroundWorker.server.ts @@ -1,15 +1,14 @@ -import { CreateBackgroundWorkerRequestBody } from "@trigger.dev/core/v3"; -import type { BackgroundWorker } from "@trigger.dev/database"; +import { CreateBackgroundWorkerRequestBody, logger, tryCatch } from "@trigger.dev/core/v3"; +import { BackgroundWorkerId } from "@trigger.dev/core/v3/isomorphic"; +import type { BackgroundWorker, WorkerDeployment } from "@trigger.dev/database"; import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; -import { generateFriendlyId } from "../friendlyIdentifiers"; -import { BaseService } from "./baseService.server"; +import { BaseService, ServiceValidationError } from "./baseService.server"; import { createBackgroundFiles, createWorkerResources, syncDeclarativeSchedules, } from "./createBackgroundWorker.server"; import { TimeoutDeploymentService } from "./timeoutDeployment.server"; -import { BackgroundWorkerId } from "@trigger.dev/core/v3/isomorphic"; export class CreateDeploymentBackgroundWorkerService extends BaseService { public async call( @@ -61,45 +60,65 @@ export class CreateDeploymentBackgroundWorkerService extends BaseService { }); } - try { - const tasksToBackgroundFiles = await createBackgroundFiles( + const [filesError, tasksToBackgroundFiles] = await tryCatch( + createBackgroundFiles( body.metadata.sourceFiles, backgroundWorker, environment, this._prisma - ); - await createWorkerResources( - body.metadata.tasks, + ) + ); + + if (filesError) { + logger.error("Error creating background worker files", { + error: filesError, + }); + + const serviceError = new ServiceValidationError("Error creating background worker files"); + + await this.#failBackgroundWorkerDeployment(deployment, serviceError); + + throw serviceError; + } + + const [resourcesError] = await tryCatch( + createWorkerResources( + body.metadata, backgroundWorker, environment, this._prisma, tasksToBackgroundFiles - ); - await syncDeclarativeSchedules( - body.metadata.tasks, - backgroundWorker, - environment, - this._prisma - ); - } catch (error) { - const name = error instanceof Error ? error.name : "UnknownError"; - const message = error instanceof Error ? error.message : JSON.stringify(error); + ) + ); - await this._prisma.workerDeployment.update({ - where: { - id: deployment.id, - }, - data: { - status: "FAILED", - failedAt: new Date(), - errorData: { - name, - message, - }, - }, + if (resourcesError) { + logger.error("Error creating background worker resources", { + error: resourcesError, }); - throw error; + const serviceError = new ServiceValidationError( + "Error creating background worker resources" + ); + + await this.#failBackgroundWorkerDeployment(deployment, serviceError); + + throw serviceError; + } + + const [schedulesError] = await tryCatch( + syncDeclarativeSchedules(body.metadata.tasks, backgroundWorker, environment, this._prisma) + ); + + if (schedulesError) { + logger.error("Error syncing declarative schedules", { + error: schedulesError, + }); + + const serviceError = new ServiceValidationError("Error syncing declarative schedules"); + + await this.#failBackgroundWorkerDeployment(deployment, serviceError); + + throw serviceError; } // Link the deployment with the background worker @@ -119,4 +138,24 @@ export class CreateDeploymentBackgroundWorkerService extends BaseService { return backgroundWorker; }); } + + async #failBackgroundWorkerDeployment(deployment: WorkerDeployment, error: Error) { + await this._prisma.workerDeployment.update({ + where: { + id: deployment.id, + }, + data: { + status: "FAILED", + failedAt: new Date(), + errorData: { + name: error.name, + message: error.message, + }, + }, + }); + + await TimeoutDeploymentService.dequeue(deployment.id, this._prisma); + + throw error; + } } diff --git a/apps/webapp/app/v3/services/createTaskRunAttempt.server.ts b/apps/webapp/app/v3/services/createTaskRunAttempt.server.ts index e25937826..f8cad4fbf 100644 --- a/apps/webapp/app/v3/services/createTaskRunAttempt.server.ts +++ b/apps/webapp/app/v3/services/createTaskRunAttempt.server.ts @@ -192,7 +192,7 @@ export class CreateTaskRunAttemptService extends BaseService { task: { id: lockedBy.slug, filePath: lockedBy.filePath, - exportName: lockedBy.exportName, + exportName: lockedBy.exportName ?? "@deprecated", }, attempt: { id: taskRunAttempt.friendlyId, diff --git a/apps/webapp/app/v3/services/pauseEnvironment.server.ts b/apps/webapp/app/v3/services/pauseEnvironment.server.ts index a3e029e56..de0216989 100644 --- a/apps/webapp/app/v3/services/pauseEnvironment.server.ts +++ b/apps/webapp/app/v3/services/pauseEnvironment.server.ts @@ -1,9 +1,9 @@ -import { type AuthenticatedEnvironment } from "@internal/testcontainers"; import { type PrismaClientOrTransaction } from "@trigger.dev/database"; import { prisma } from "~/db.server"; import { logger } from "~/services/logger.server"; import { updateEnvConcurrencyLimits } from "../runQueue.server"; import { WithRunEngine } from "./baseService.server"; +import { AuthenticatedEnvironment } from "~/services/apiAuth.server"; export type PauseStatus = "paused" | "resumed"; diff --git a/apps/webapp/app/v3/taskRunHeartbeatFailed.server.ts b/apps/webapp/app/v3/taskRunHeartbeatFailed.server.ts index fa564c06b..8c1c4a7e3 100644 --- a/apps/webapp/app/v3/taskRunHeartbeatFailed.server.ts +++ b/apps/webapp/app/v3/taskRunHeartbeatFailed.server.ts @@ -98,6 +98,7 @@ export class TaskRunHeartbeatFailedService extends BaseService { break; } case "DELAYED": + case "PENDING_VERSION": case "WAITING_FOR_DEPLOY": { logger.debug( `[TaskRunHeartbeatFailedService] ${taskRun.status} Removing task run from queue`, diff --git a/apps/webapp/app/v3/taskStatus.ts b/apps/webapp/app/v3/taskStatus.ts index a360bde09..909cb64b8 100644 --- a/apps/webapp/app/v3/taskStatus.ts +++ b/apps/webapp/app/v3/taskStatus.ts @@ -16,6 +16,7 @@ export type FINAL_RUN_STATUSES = (typeof FINAL_RUN_STATUSES)[number]; export const NON_FINAL_RUN_STATUSES = [ "DELAYED", "PENDING", + "PENDING_VERSION", "WAITING_FOR_DEPLOY", "EXECUTING", "WAITING_TO_RESUME", diff --git a/docs/v3-openapi.yaml b/docs/v3-openapi.yaml index 1cc92a174..b94280419 100644 --- a/docs/v3-openapi.yaml +++ b/docs/v3-openapi.yaml @@ -1677,7 +1677,7 @@ components: type: string description: The status of the run enum: - - WAITING_FOR_DEPLOY + - PENDING_VERSION - QUEUED - EXECUTING - REATTEMPTING @@ -1752,7 +1752,7 @@ components: type: string description: The status of the run enum: - - WAITING_FOR_DEPLOY + - PENDING_VERSION - QUEUED - EXECUTING - REATTEMPTING @@ -1947,8 +1947,8 @@ components: type: string description: The status of the run enum: + - PENDING_VERSION - DELAYED - - WAITING_FOR_DEPLOY - QUEUED - EXECUTING - REATTEMPTING diff --git a/internal-packages/database/prisma/migrations/20250320111737_add_pending_version_task_run_status/migration.sql b/internal-packages/database/prisma/migrations/20250320111737_add_pending_version_task_run_status/migration.sql new file mode 100644 index 000000000..82db74877 --- /dev/null +++ b/internal-packages/database/prisma/migrations/20250320111737_add_pending_version_task_run_status/migration.sql @@ -0,0 +1,4 @@ +-- AlterEnum +ALTER TYPE "TaskRunStatus" +ADD + VALUE 'PENDING_VERSION'; \ No newline at end of file diff --git a/internal-packages/database/prisma/migrations/20250320130354_add_many_to_many_relationship_task_queue_background_worker/migration.sql b/internal-packages/database/prisma/migrations/20250320130354_add_many_to_many_relationship_task_queue_background_worker/migration.sql new file mode 100644 index 000000000..19360f89b --- /dev/null +++ b/internal-packages/database/prisma/migrations/20250320130354_add_many_to_many_relationship_task_queue_background_worker/migration.sql @@ -0,0 +1,20 @@ +-- CreateTable +CREATE TABLE "_BackgroundWorkerToTaskQueue" ("A" TEXT NOT NULL, "B" TEXT NOT NULL); + +-- CreateIndex +CREATE UNIQUE INDEX "_BackgroundWorkerToTaskQueue_AB_unique" ON "_BackgroundWorkerToTaskQueue"("A", "B"); + +-- CreateIndex +CREATE INDEX "_BackgroundWorkerToTaskQueue_B_index" ON "_BackgroundWorkerToTaskQueue"("B"); + +-- AddForeignKey +ALTER TABLE + "_BackgroundWorkerToTaskQueue" +ADD + CONSTRAINT "_BackgroundWorkerToTaskQueue_A_fkey" FOREIGN KEY ("A") REFERENCES "BackgroundWorker"("id") ON DELETE CASCADE ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE + "_BackgroundWorkerToTaskQueue" +ADD + CONSTRAINT "_BackgroundWorkerToTaskQueue_B_fkey" FOREIGN KEY ("B") REFERENCES "TaskQueue"("id") ON DELETE CASCADE ON UPDATE CASCADE; \ No newline at end of file diff --git a/internal-packages/database/prisma/schema.prisma b/internal-packages/database/prisma/schema.prisma index c450a68b3..3aca5e060 100644 --- a/internal-packages/database/prisma/schema.prisma +++ b/internal-packages/database/prisma/schema.prisma @@ -1614,6 +1614,7 @@ model BackgroundWorker { attempts TaskRunAttempt[] lockedRuns TaskRun[] files BackgroundWorkerFile[] + queues TaskQueue[] deployment WorkerDeployment? @@ -1902,7 +1903,10 @@ enum TaskRunStatus { /// Task is waiting to be executed by a worker PENDING - /// Task hasn't been deployed yet but is waiting to be executed + /// The run is pending a version update because it cannot execute without additional information (task, queue, etc.). Replaces WAITING_FOR_DEPLOY + PENDING_VERSION + + /// Task hasn't been deployed yet but is waiting to be executed. Deprecated in favor of PENDING_VERSION WAITING_FOR_DEPLOY /// Task is currently being executed by a worker @@ -2558,6 +2562,7 @@ model TaskQueue { attempts TaskRunAttempt[] tasks BackgroundWorkerTask[] + workers BackgroundWorker[] @@unique([runtimeEnvironmentId, name]) } diff --git a/internal-packages/run-engine/src/engine/db/worker.ts b/internal-packages/run-engine/src/engine/db/worker.ts index 2c3264615..34abf2cd3 100644 --- a/internal-packages/run-engine/src/engine/db/worker.ts +++ b/internal-packages/run-engine/src/engine/db/worker.ts @@ -3,6 +3,7 @@ import { BackgroundWorkerTask, Prisma, PrismaClientOrTransaction, + TaskQueue, WorkerDeployment, } from "@trigger.dev/database"; import { CURRENT_DEPLOYMENT_LABEL } from "@trigger.dev/core/v3/isomorphic"; @@ -30,7 +31,8 @@ type RunWithBackgroundWorkerTasksResult = | "NO_WORKER" | "TASK_NOT_IN_LATEST" | "TASK_NEVER_REGISTERED" - | "BACKGROUND_WORKER_MISMATCH"; + | "BACKGROUND_WORKER_MISMATCH" + | "QUEUE_NOT_FOUND"; message: string; run: RunWithMininimalEnvironment; } @@ -49,6 +51,7 @@ type RunWithBackgroundWorkerTasksResult = run: RunWithMininimalEnvironment; worker: BackgroundWorker; task: BackgroundWorkerTask; + queue: TaskQueue; deployment: WorkerDeployment | null; }; @@ -158,11 +161,23 @@ export async function getRunWithBackgroundWorkerTasks( } } + const queue = workerWithTasks.queues.find((queue) => queue.name === run.queue); + + if (!queue) { + return { + success: false as const, + code: "QUEUE_NOT_FOUND", + message: `Queue not found for run: ${run.id}`, + run, + }; + } + return { success: true as const, run, worker: workerWithTasks.worker, task: backgroundTask, + queue, deployment: workerWithTasks.deployment, }; } @@ -170,6 +185,7 @@ export async function getRunWithBackgroundWorkerTasks( type WorkerDeploymentWithWorkerTasks = { worker: BackgroundWorker; tasks: BackgroundWorkerTask[]; + queues: TaskQueue[]; deployment: WorkerDeployment | null; }; @@ -184,6 +200,7 @@ export async function getWorkerDeploymentFromWorker( include: { deployment: true, tasks: true, + queues: true, }, }); @@ -191,7 +208,7 @@ export async function getWorkerDeploymentFromWorker( return null; } - return { worker, tasks: worker.tasks, deployment: worker.deployment }; + return { worker, tasks: worker.tasks, queues: worker.queues, deployment: worker.deployment }; } export async function getMostRecentWorker( @@ -204,6 +221,7 @@ export async function getMostRecentWorker( }, include: { tasks: true, + queues: true, }, orderBy: { id: "desc", @@ -214,7 +232,7 @@ export async function getMostRecentWorker( return null; } - return { worker, tasks: worker.tasks, deployment: null }; + return { worker, tasks: worker.tasks, queues: worker.queues, deployment: null }; } export async function getWorkerById( @@ -228,6 +246,7 @@ export async function getWorkerById( include: { deployment: true, tasks: true, + queues: true, }, orderBy: { id: "desc", @@ -238,7 +257,7 @@ export async function getWorkerById( return null; } - return { worker, tasks: worker.tasks, deployment: worker.deployment }; + return { worker, tasks: worker.tasks, queues: worker.queues, deployment: worker.deployment }; } export async function getWorkerFromCurrentlyPromotedDeployment( @@ -258,6 +277,7 @@ export async function getWorkerFromCurrentlyPromotedDeployment( worker: { include: { tasks: true, + queues: true, }, }, }, @@ -272,6 +292,7 @@ export async function getWorkerFromCurrentlyPromotedDeployment( return { worker: promotion.deployment.worker, tasks: promotion.deployment.worker.tasks, + queues: promotion.deployment.worker.queues, deployment: promotion.deployment, }; } diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index d1e181514..80fa8f460 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -48,7 +48,7 @@ import { ReleaseConcurrencySystem } from "./systems/releaseConcurrencySystem.js" import { RunAttemptSystem } from "./systems/runAttemptSystem.js"; import { SystemResources } from "./systems/systems.js"; import { TtlSystem } from "./systems/ttlSystem.js"; -import { WaitingForWorkerSystem } from "./systems/waitingForWorkerSystem.js"; +import { PendingVersionSystem } from "./systems/pendingVersionSystem.js"; import { WaitpointSystem } from "./systems/waitpointSystem.js"; import { EngineWorker, HeartbeatTimeouts, RunEngineOptions, TriggerParams } from "./types.js"; import { workerCatalog } from "./workerCatalog.js"; @@ -73,7 +73,7 @@ export class RunEngine { checkpointSystem: CheckpointSystem; delayedRunSystem: DelayedRunSystem; ttlSystem: TtlSystem; - waitingForWorkerSystem: WaitingForWorkerSystem; + pendingVersionSystem: PendingVersionSystem; releaseConcurrencySystem: ReleaseConcurrencySystem; constructor(private readonly options: RunEngineOptions) { @@ -148,10 +148,10 @@ export class RunEngine { reason: payload.reason, }); }, - queueRunsWaitingForWorker: async ({ payload }) => { - await this.waitingForWorkerSystem.enqueueRunsWaitingForWorker({ - backgroundWorkerId: payload.backgroundWorkerId, - }); + queueRunsPendingVersion: async ({ payload }) => { + await this.pendingVersionSystem.enqueueRunsForBackgroundWorker( + payload.backgroundWorkerId + ); }, tryCompleteBatch: async ({ payload }) => { await this.batchSystem.performCompleteBatch({ batchId: payload.batchId }); @@ -266,7 +266,7 @@ export class RunEngine { enqueueSystem: this.enqueueSystem, }); - this.waitingForWorkerSystem = new WaitingForWorkerSystem({ + this.pendingVersionSystem = new PendingVersionSystem({ resources, enqueueSystem: this.enqueueSystem, }); @@ -724,14 +724,8 @@ export class RunEngine { }); } - async queueRunsWaitingForWorker({ - backgroundWorkerId, - }: { - backgroundWorkerId: string; - }): Promise { - return this.waitingForWorkerSystem.enqueueRunsWaitingForWorker({ - backgroundWorkerId, - }); + async scheduleEnqueueRunsForBackgroundWorker(backgroundWorkerId: string): Promise { + return this.pendingVersionSystem.scheduleResolvePendingVersionRuns(backgroundWorkerId); } /** diff --git a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts index 33bdb5656..86b307cee 100644 --- a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts @@ -186,6 +186,7 @@ export class DequeueSystem { } case "NO_WORKER": case "TASK_NEVER_REGISTERED": + case "QUEUE_NOT_FOUND": case "TASK_NOT_IN_LATEST": { this.$.logger.warn(`RunEngine.dequeueFromMasterQueue(): ${result.code}`, { runId, @@ -194,7 +195,7 @@ export class DequeueSystem { }); //not deployed yet, so we'll wait for the deploy - await this.#waitingForDeploy({ + await this.#pendingVersion({ orgId, runId, reason: result.message, @@ -232,7 +233,7 @@ export class DequeueSystem { result, }); //not deployed yet, so we'll wait for the deploy - await this.#waitingForDeploy({ + await this.#pendingVersion({ orgId, runId, reason: "No deployment or deployment image reference found for deployed run", @@ -340,43 +341,6 @@ export class DequeueSystem { maxAttempts = parsedConfig.data.maxAttempts; } - - const queue = await prisma.taskQueue.findUnique({ - where: { - runtimeEnvironmentId_name: { - runtimeEnvironmentId: result.run.runtimeEnvironmentId, - name: sanitizeQueueName(result.run.queue), - }, - }, - }); - - if (!queue) { - this.$.logger.debug( - "RunEngine.dequeueFromMasterQueue(): queue not found, so nacking message", - { - queueMessage: message, - taskRunQueue: result.run.queue, - runtimeEnvironmentId: result.run.runtimeEnvironmentId, - } - ); - - //will auto-retry - const gotRequeued = await this.$.runQueue.nackMessage({ orgId, messageId: runId }); - if (!gotRequeued) { - await this.runAttemptSystem.systemFailure({ - runId, - error: { - type: "INTERNAL_ERROR", - code: "TASK_DEQUEUED_QUEUE_NOT_FOUND", - message: `Tried to dequeue the run but the queue doesn't exist: ${result.run.queue}`, - }, - tx: prisma, - }); - } - - return null; - } - //update the run const lockedTaskRun = await prisma.taskRun.update({ where: { @@ -386,7 +350,7 @@ export class DequeueSystem { lockedAt: new Date(), lockedById: result.task.id, lockedToVersionId: result.worker.id, - lockedQueueId: queue.id, + lockedQueueId: result.queue.id, startedAt: result.run.startedAt ?? new Date(), baseCostInCents: this.options.machines.baseCostInCents, machinePreset: machinePreset.name, @@ -547,7 +511,7 @@ export class DequeueSystem { ); } - async #waitingForDeploy({ + async #pendingVersion({ orgId, runId, workerId, @@ -566,14 +530,14 @@ export class DequeueSystem { return startSpan( this.$.tracer, - "#waitingForDeploy", + "#pendingVersion", async (span) => { return this.$.runLock.lock([runId], 5_000, async (signal) => { //mark run as waiting for deploy const run = await prisma.taskRun.update({ where: { id: runId }, data: { - status: "WAITING_FOR_DEPLOY", + status: "PENDING_VERSION", }, select: { id: true, @@ -590,6 +554,11 @@ export class DequeueSystem { }, }); + this.$.logger.debug("RunEngine.dequeueFromMasterQueue(): Pending version", { + runId, + run, + }); + await this.executionSnapshotSystem.createExecutionSnapshot(prisma, { run, snapshot: { diff --git a/internal-packages/run-engine/src/engine/systems/waitingForWorkerSystem.ts b/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts similarity index 54% rename from internal-packages/run-engine/src/engine/systems/waitingForWorkerSystem.ts rename to internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts index 517e1c303..991d8b9aa 100644 --- a/internal-packages/run-engine/src/engine/systems/waitingForWorkerSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts @@ -1,25 +1,25 @@ import { EnqueueSystem } from "./enqueueSystem.js"; import { SystemResources } from "./systems.js"; -export type WaitingForWorkerSystemOptions = { +export type PendingVersionSystemOptions = { resources: SystemResources; enqueueSystem: EnqueueSystem; - queueRunsWaitingForWorkerBatchSize?: number; + queueRunsPendingVersionBatchSize?: number; }; -export class WaitingForWorkerSystem { +export class PendingVersionSystem { private readonly $: SystemResources; private readonly enqueueSystem: EnqueueSystem; - constructor(private readonly options: WaitingForWorkerSystemOptions) { + constructor(private readonly options: PendingVersionSystemOptions) { this.$ = options.resources; this.enqueueSystem = options.enqueueSystem; } - async enqueueRunsWaitingForWorker({ backgroundWorkerId }: { backgroundWorkerId: string }) { + async enqueueRunsForBackgroundWorker(backgroundWorkerId: string) { //It could be a lot of runs, so we will process them in a batch //if there are still more to process we will enqueue this function again - const maxCount = this.options.queueRunsWaitingForWorkerBatchSize ?? 200; + const maxCount = this.options.queueRunsPendingVersionBatchSize ?? 200; const backgroundWorker = await this.$.prisma.backgroundWorker.findFirst({ where: { @@ -33,24 +33,34 @@ export class WaitingForWorkerSystem { }, }, tasks: true, + queues: true, }, }); if (!backgroundWorker) { - this.$.logger.error("#queueRunsWaitingForWorker: background worker not found", { + this.$.logger.error("#enqueueRunsForBackgroundWorker: background worker not found", { id: backgroundWorkerId, }); return; } - const runsWaitingForDeploy = await this.$.prisma.taskRun.findMany({ + this.$.logger.debug("Finding PENDING_VERSION runs for background worker", { + workerId: backgroundWorker.id, + taskIdentifiers: backgroundWorker.tasks.map((task) => task.slug), + queues: backgroundWorker.queues.map((queue) => queue.name), + }); + + const pendingRuns = await this.$.prisma.taskRun.findMany({ where: { runtimeEnvironmentId: backgroundWorker.runtimeEnvironmentId, projectId: backgroundWorker.projectId, - status: "WAITING_FOR_DEPLOY", + status: "PENDING_VERSION", taskIdentifier: { in: backgroundWorker.tasks.map((task) => task.slug), }, + queue: { + in: backgroundWorker.queues.map((queue) => queue.name), + }, }, orderBy: { createdAt: "asc", @@ -59,9 +69,22 @@ export class WaitingForWorkerSystem { }); //none to process - if (!runsWaitingForDeploy.length) return; + if (!pendingRuns.length) return; - for (const run of runsWaitingForDeploy) { + this.$.logger.debug("Enqueueing PENDING_VERSION runs for background worker", { + workerId: backgroundWorker.id, + taskIdentifiers: pendingRuns.map((run) => run.taskIdentifier), + queues: pendingRuns.map((run) => run.queue), + runs: pendingRuns.map((run) => ({ + id: run.id, + taskIdentifier: run.taskIdentifier, + queue: run.queue, + createdAt: run.createdAt, + priorityMs: run.priorityMs, + })), + }); + + for (const run of pendingRuns) { await this.$.prisma.$transaction(async (tx) => { const updatedRun = await tx.taskRun.update({ where: { @@ -83,19 +106,15 @@ export class WaitingForWorkerSystem { } //enqueue more if needed - if (runsWaitingForDeploy.length > maxCount) { - await this.scheduleEnqueueRunsWaitingForWorker({ backgroundWorkerId }); + if (pendingRuns.length > maxCount) { + await this.scheduleResolvePendingVersionRuns(backgroundWorkerId); } } - async scheduleEnqueueRunsWaitingForWorker({ - backgroundWorkerId, - }: { - backgroundWorkerId: string; - }): Promise { + async scheduleResolvePendingVersionRuns(backgroundWorkerId: string): Promise { //we want this to happen in the background await this.$.worker.enqueue({ - job: "queueRunsWaitingForWorker", + job: "queueRunsPendingVersion", payload: { backgroundWorkerId }, }); } diff --git a/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts b/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts index 0ba498d77..6d5c9028f 100644 --- a/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts @@ -270,7 +270,7 @@ export class RunAttemptSystem { task: { id: run.lockedBy!.slug, filePath: run.lockedBy!.filePath, - exportName: run.lockedBy!.exportName, + exportName: run.lockedBy!.exportName ?? undefined, }, attempt: { number: nextAttemptNumber, diff --git a/internal-packages/run-engine/src/engine/tests/notDeployed.test.ts b/internal-packages/run-engine/src/engine/tests/notDeployed.test.ts deleted file mode 100644 index 73244e756..000000000 --- a/internal-packages/run-engine/src/engine/tests/notDeployed.test.ts +++ /dev/null @@ -1,150 +0,0 @@ -import { containerTest, assertNonNullable } from "@internal/testcontainers"; -import { trace } from "@internal/tracing"; -import { RunEngine } from "../index.js"; -import { setTimeout } from "timers/promises"; -import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js"; - -vi.setConfig({ testTimeout: 60_000 }); - -describe("RunEngine not deployed", () => { - containerTest("Not yet deployed", async ({ prisma, redisOptions }) => { - //create environment - const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); - - const engine = new RunEngine({ - prisma, - worker: { - redis: redisOptions, - workers: 1, - tasksPerWorker: 10, - pollIntervalMs: 100, - }, - queue: { - redis: redisOptions, - }, - runLock: { - redis: redisOptions, - }, - machines: { - defaultMachine: "small-1x", - machines: { - "small-1x": { - name: "small-1x" as const, - cpu: 0.5, - memory: 0.5, - centsPerMs: 0.0001, - }, - }, - baseCostInCents: 0.0001, - }, - //set this so we have to requeue the runs in two batches - queueRunsWaitingForWorkerBatchSize: 1, - tracer: trace.getTracer("test", "0.0.0"), - }); - - try { - const taskIdentifier = "test-task"; - - //trigger the run - const run = await engine.trigger( - { - number: 1, - friendlyId: "run_1234", - environment: authenticatedEnvironment, - taskIdentifier, - payload: "{}", - payloadType: "application/json", - context: {}, - traceContext: {}, - traceId: "t12345", - spanId: "s12345", - masterQueue: "main", - queue: "task/test-task", - isTest: false, - tags: [], - }, - prisma - ); - - //trigger another run - const run2 = await engine.trigger( - { - number: 2, - friendlyId: "run_1235", - environment: authenticatedEnvironment, - taskIdentifier, - payload: "{}", - payloadType: "application/json", - context: {}, - traceContext: {}, - traceId: "t12346", - spanId: "s12346", - masterQueue: "main", - queue: "task/test-task", - isTest: false, - tags: [], - }, - prisma - ); - - //should be queued - const executionDataR1 = await engine.getRunExecutionData({ runId: run.id }); - const executionDataR2 = await engine.getRunExecutionData({ runId: run2.id }); - assertNonNullable(executionDataR1); - assertNonNullable(executionDataR2); - expect(executionDataR1.snapshot.executionStatus).toBe("QUEUED"); - expect(executionDataR2.snapshot.executionStatus).toBe("QUEUED"); - - //dequeuing should fail - const dequeued = await engine.dequeueFromMasterQueue({ - consumerId: "test_12345", - masterQueue: run.masterQueue, - maxRunCount: 10, - }); - expect(dequeued.length).toBe(0); - - //queue should be empty - const queueLength = await engine.runQueue.lengthOfQueue(authenticatedEnvironment, run.queue); - expect(queueLength).toBe(0); - - //check the execution data now - const executionData2R1 = await engine.getRunExecutionData({ runId: run.id }); - const executionData2R2 = await engine.getRunExecutionData({ runId: run2.id }); - assertNonNullable(executionData2R1); - assertNonNullable(executionData2R2); - expect(executionData2R1.snapshot.executionStatus).toBe("RUN_CREATED"); - expect(executionData2R2.snapshot.executionStatus).toBe("RUN_CREATED"); - expect(executionData2R1.run.status).toBe("WAITING_FOR_DEPLOY"); - expect(executionData2R2.run.status).toBe("WAITING_FOR_DEPLOY"); - - //create background worker - const backgroundWorker = await setupBackgroundWorker( - engine, - authenticatedEnvironment, - taskIdentifier - ); - - //now we deploy the background worker - await engine.queueRunsWaitingForWorker({ backgroundWorkerId: backgroundWorker.worker.id }); - - //it's async so we wait - await setTimeout(500); - - //should now be queued - const executionData3R1 = await engine.getRunExecutionData({ runId: run.id }); - const executionData3R2 = await engine.getRunExecutionData({ runId: run2.id }); - assertNonNullable(executionData3R1); - assertNonNullable(executionData3R2); - expect(executionData3R1.snapshot.executionStatus).toBe("QUEUED"); - expect(executionData3R2.snapshot.executionStatus).toBe("QUEUED"); - expect(executionData3R1.run.status).toBe("PENDING"); - expect(executionData3R2.run.status).toBe("PENDING"); - - //queue should be empty - const queueLength2 = await engine.runQueue.lengthOfQueue(authenticatedEnvironment, run.queue); - expect(queueLength2).toBe(2); - } finally { - engine.quit(); - } - }); -}); diff --git a/internal-packages/run-engine/src/engine/tests/pendingVersion.test.ts b/internal-packages/run-engine/src/engine/tests/pendingVersion.test.ts new file mode 100644 index 000000000..c2c1ae338 --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/pendingVersion.test.ts @@ -0,0 +1,319 @@ +import { containerTest, assertNonNullable } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { RunEngine } from "../index.js"; +import { setTimeout } from "timers/promises"; +import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "./setup.js"; + +vi.setConfig({ testTimeout: 60_000 }); + +describe("RunEngine pending version", () => { + containerTest( + "When a run is triggered but the background task hasn't been created yet", + async ({ prisma, redisOptions }) => { + //create environment + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + //set this so we have to requeue the runs in two batches + queueRunsWaitingForWorkerBatchSize: 1, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + //trigger the run + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_1234", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345", + spanId: "s12345", + masterQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + }, + prisma + ); + + //trigger another run + const run2 = await engine.trigger( + { + number: 2, + friendlyId: "run_1235", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12346", + spanId: "s12346", + masterQueue: "main", + queue: "task/test-task", + isTest: false, + tags: [], + }, + prisma + ); + + //should be queued + const executionDataR1 = await engine.getRunExecutionData({ runId: run.id }); + const executionDataR2 = await engine.getRunExecutionData({ runId: run2.id }); + assertNonNullable(executionDataR1); + assertNonNullable(executionDataR2); + expect(executionDataR1.snapshot.executionStatus).toBe("QUEUED"); + expect(executionDataR2.snapshot.executionStatus).toBe("QUEUED"); + + await setupBackgroundWorker(engine, authenticatedEnvironment, ["test-task-other"]); + + //dequeuing should fail + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: run.masterQueue, + maxRunCount: 10, + }); + expect(dequeued.length).toBe(0); + + //queue should be empty + const queueLength = await engine.runQueue.lengthOfQueue( + authenticatedEnvironment, + run.queue + ); + expect(queueLength).toBe(0); + + //check the execution data now + const executionData2R1 = await engine.getRunExecutionData({ runId: run.id }); + const executionData2R2 = await engine.getRunExecutionData({ runId: run2.id }); + assertNonNullable(executionData2R1); + assertNonNullable(executionData2R2); + expect(executionData2R1.snapshot.executionStatus).toBe("RUN_CREATED"); + expect(executionData2R2.snapshot.executionStatus).toBe("RUN_CREATED"); + expect(executionData2R1.run.status).toBe("PENDING_VERSION"); + expect(executionData2R2.run.status).toBe("PENDING_VERSION"); + + //create background worker + const backgroundWorker = await setupBackgroundWorker( + engine, + authenticatedEnvironment, + taskIdentifier + ); + + //it's async so we wait + await setTimeout(500); + + //should now be queued + const executionData3R1 = await engine.getRunExecutionData({ runId: run.id }); + const executionData3R2 = await engine.getRunExecutionData({ runId: run2.id }); + assertNonNullable(executionData3R1); + assertNonNullable(executionData3R2); + expect(executionData3R1.snapshot.executionStatus).toBe("QUEUED"); + expect(executionData3R2.snapshot.executionStatus).toBe("QUEUED"); + expect(executionData3R1.run.status).toBe("PENDING"); + expect(executionData3R2.run.status).toBe("PENDING"); + + //queue should be empty + const queueLength2 = await engine.runQueue.lengthOfQueue( + authenticatedEnvironment, + run.queue + ); + expect(queueLength2).toBe(2); + } finally { + engine.quit(); + } + } + ); + + containerTest( + "When a run is triggered but the queue hasn't been created yet", + async ({ prisma, redisOptions }) => { + //create environment + const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + + const engine = new RunEngine({ + prisma, + worker: { + redis: redisOptions, + workers: 1, + tasksPerWorker: 10, + pollIntervalMs: 100, + }, + queue: { + redis: redisOptions, + }, + runLock: { + redis: redisOptions, + }, + machines: { + defaultMachine: "small-1x", + machines: { + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + }, + baseCostInCents: 0.0001, + }, + //set this so we have to requeue the runs in two batches + queueRunsWaitingForWorkerBatchSize: 1, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + await setupBackgroundWorker(engine, authenticatedEnvironment, taskIdentifier); + + //trigger the run + const run = await engine.trigger( + { + number: 1, + friendlyId: "run_1234", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345", + spanId: "s12345", + masterQueue: "main", + queue: "custom-queue", + isTest: false, + tags: [], + }, + prisma + ); + + //trigger another run + const run2 = await engine.trigger( + { + number: 2, + friendlyId: "run_1235", + environment: authenticatedEnvironment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12346", + spanId: "s12346", + masterQueue: "main", + queue: "custom-queue-2", + isTest: false, + tags: [], + }, + prisma + ); + + //should be queued + const executionDataR1 = await engine.getRunExecutionData({ runId: run.id }); + const executionDataR2 = await engine.getRunExecutionData({ runId: run2.id }); + assertNonNullable(executionDataR1); + assertNonNullable(executionDataR2); + expect(executionDataR1.snapshot.executionStatus).toBe("QUEUED"); + expect(executionDataR2.snapshot.executionStatus).toBe("QUEUED"); + + //dequeuing should fail + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: run.masterQueue, + maxRunCount: 10, + }); + expect(dequeued.length).toBe(0); + + //queue should be empty + const queueLength = await engine.runQueue.lengthOfQueue( + authenticatedEnvironment, + run.queue + ); + expect(queueLength).toBe(0); + + //check the execution data now + const executionData2R1 = await engine.getRunExecutionData({ runId: run.id }); + const executionData2R2 = await engine.getRunExecutionData({ runId: run2.id }); + assertNonNullable(executionData2R1); + assertNonNullable(executionData2R2); + expect(executionData2R1.snapshot.executionStatus).toBe("RUN_CREATED"); + expect(executionData2R2.snapshot.executionStatus).toBe("RUN_CREATED"); + expect(executionData2R1.run.status).toBe("PENDING_VERSION"); + expect(executionData2R2.run.status).toBe("PENDING_VERSION"); + + //create background worker + const backgroundWorker = await setupBackgroundWorker( + engine, + authenticatedEnvironment, + taskIdentifier, + undefined, + undefined, + { + customQueues: ["custom-queue", "custom-queue-2"], + } + ); + + //it's async so we wait + await setTimeout(500); + + //should now be queued + const executionData3R1 = await engine.getRunExecutionData({ runId: run.id }); + const executionData3R2 = await engine.getRunExecutionData({ runId: run2.id }); + assertNonNullable(executionData3R1); + assertNonNullable(executionData3R2); + expect(executionData3R1.snapshot.executionStatus).toBe("QUEUED"); + expect(executionData3R2.snapshot.executionStatus).toBe("QUEUED"); + expect(executionData3R1.run.status).toBe("PENDING"); + expect(executionData3R2.run.status).toBe("PENDING"); + + // custom-queue should have 1 run + const queueLength2 = await engine.runQueue.lengthOfQueue( + authenticatedEnvironment, + "custom-queue" + ); + expect(queueLength2).toBe(1); + + // custom-queue-2 should have 1 run + const queueLength3 = await engine.runQueue.lengthOfQueue( + authenticatedEnvironment, + "custom-queue-2" + ); + expect(queueLength3).toBe(1); + } finally { + engine.quit(); + } + } + ); +}); diff --git a/internal-packages/run-engine/src/engine/tests/setup.ts b/internal-packages/run-engine/src/engine/tests/setup.ts index 95bb9f29e..be8b331db 100644 --- a/internal-packages/run-engine/src/engine/tests/setup.ts +++ b/internal-packages/run-engine/src/engine/tests/setup.ts @@ -72,17 +72,30 @@ export async function setupBackgroundWorker( machineConfig?: MachineConfig, retryOptions?: RetryOptions, queueOptions?: { + customQueues?: string[]; releaseConcurrencyOnWaitpoint?: boolean; concurrencyLimit?: number | null; } ) { + const latestWorkers = await engine.prisma.backgroundWorker.findMany({ + where: { + runtimeEnvironmentId: environment.id, + }, + orderBy: { + createdAt: "desc", + }, + take: 1, + }); + + const nextVersion = calculateNextBuildVersion(latestWorkers[0]?.version); + const worker = await engine.prisma.backgroundWorker.create({ data: { friendlyId: generateFriendlyId("worker"), contentHash: "hash", projectId: environment.project.id, runtimeEnvironmentId: environment.id, - version: "20241015.1", + version: nextVersion, metadata: {}, }, }); @@ -116,8 +129,14 @@ export async function setupBackgroundWorker( tasks.push(task); const queueName = sanitizeQueueName(`task/${identifier}`); - const taskQueue = await engine.prisma.taskQueue.create({ - data: { + const taskQueue = await engine.prisma.taskQueue.upsert({ + where: { + runtimeEnvironmentId_name: { + name: queueName, + runtimeEnvironmentId: worker.runtimeEnvironmentId, + }, + }, + create: { friendlyId: generateFriendlyId("queue"), name: queueName, concurrencyLimit: @@ -127,11 +146,27 @@ export async function setupBackgroundWorker( runtimeEnvironmentId: worker.runtimeEnvironmentId, projectId: worker.projectId, type: "VIRTUAL", + workers: { + connect: { + id: worker.id, + }, + }, releaseConcurrencyOnWaitpoint: typeof queueOptions?.releaseConcurrencyOnWaitpoint === "boolean" ? queueOptions.releaseConcurrencyOnWaitpoint : undefined, }, + update: { + concurrencyLimit: + typeof queueOptions?.concurrencyLimit === "undefined" + ? 10 + : queueOptions.concurrencyLimit, + workers: { + connect: { + id: worker.id, + }, + }, + }, }); if (typeof taskQueue.concurrencyLimit === "number") { @@ -145,13 +180,55 @@ export async function setupBackgroundWorker( } } + for (const queueName of queueOptions?.customQueues ?? []) { + const taskQueue = await engine.prisma.taskQueue.upsert({ + where: { + runtimeEnvironmentId_name: { + name: queueName, + runtimeEnvironmentId: worker.runtimeEnvironmentId, + }, + }, + create: { + friendlyId: generateFriendlyId("queue"), + name: queueName, + concurrencyLimit: + typeof queueOptions?.concurrencyLimit === "undefined" + ? 10 + : queueOptions.concurrencyLimit, + runtimeEnvironmentId: worker.runtimeEnvironmentId, + projectId: worker.projectId, + type: "VIRTUAL", + workers: { + connect: { + id: worker.id, + }, + }, + releaseConcurrencyOnWaitpoint: + typeof queueOptions?.releaseConcurrencyOnWaitpoint === "boolean" + ? queueOptions.releaseConcurrencyOnWaitpoint + : undefined, + }, + update: { + concurrencyLimit: + typeof queueOptions?.concurrencyLimit === "undefined" + ? 10 + : queueOptions.concurrencyLimit, + workers: { + connect: { + id: worker.id, + }, + }, + }, + }); + } + if (environment.type !== "DEVELOPMENT") { const deployment = await engine.prisma.workerDeployment.create({ data: { friendlyId: generateFriendlyId("deployment"), contentHash: worker.contentHash, version: worker.version, - shortCode: "short_code", + shortCode: `short_code_${worker.version}`, imageReference: `trigger/${environment.project.externalRef}:${worker.version}.${environment.slug}`, status: "DEPLOYED", projectId: environment.project.id, @@ -160,14 +237,26 @@ export async function setupBackgroundWorker( }, }); - const promotion = await engine.prisma.workerDeploymentPromotion.create({ - data: { - label: CURRENT_DEPLOYMENT_LABEL, + const promotion = await engine.prisma.workerDeploymentPromotion.upsert({ + where: { + environmentId_label: { + environmentId: deployment.environmentId, + label: CURRENT_DEPLOYMENT_LABEL, + }, + }, + create: { + deploymentId: deployment.id, + environmentId: deployment.environmentId, + label: CURRENT_DEPLOYMENT_LABEL, + }, + update: { deploymentId: deployment.id, - environmentId: environment.id, }, }); + //now we deploy the background worker + await engine.scheduleEnqueueRunsForBackgroundWorker(worker.id); + return { worker, tasks, @@ -181,3 +270,24 @@ export async function setupBackgroundWorker( tasks, }; } + +function calculateNextBuildVersion(latestVersion?: string | null): string { + const today = new Date(); + const year = today.getFullYear(); + const month = today.getMonth() + 1; + const day = today.getDate(); + const todayFormatted = `${year}${month < 10 ? "0" : ""}${month}${day < 10 ? "0" : ""}${day}`; + + if (!latestVersion) { + return `${todayFormatted}.1`; + } + + const [date, buildNumber] = latestVersion.split("."); + + if (date === todayFormatted) { + const nextBuildNumber = parseInt(buildNumber, 10) + 1; + return `${date}.${nextBuildNumber}`; + } + + return `${todayFormatted}.1`; +} diff --git a/internal-packages/run-engine/src/engine/workerCatalog.ts b/internal-packages/run-engine/src/engine/workerCatalog.ts index e4d945d65..92eddce19 100644 --- a/internal-packages/run-engine/src/engine/workerCatalog.ts +++ b/internal-packages/run-engine/src/engine/workerCatalog.ts @@ -29,7 +29,7 @@ export const workerCatalog = { }), visibilityTimeoutMs: 5000, }, - queueRunsWaitingForWorker: { + queueRunsPendingVersion: { schema: z.object({ backgroundWorkerId: z.string(), }), diff --git a/internal-packages/run-engine/tsconfig.src.json b/internal-packages/run-engine/tsconfig.src.json index 5617aa970..6043e02ad 100644 --- a/internal-packages/run-engine/tsconfig.src.json +++ b/internal-packages/run-engine/tsconfig.src.json @@ -14,6 +14,7 @@ "isolatedModules": true, "preserveWatchOutput": true, "skipLibCheck": true, - "strict": true + "strict": true, + "customConditions": ["@triggerdotdev/source"] } } diff --git a/internal-packages/run-engine/tsconfig.test.json b/internal-packages/run-engine/tsconfig.test.json index d8c7d1c63..b68d234bd 100644 --- a/internal-packages/run-engine/tsconfig.test.json +++ b/internal-packages/run-engine/tsconfig.test.json @@ -1,5 +1,5 @@ { - "include": ["src/**/*.test.ts", "src/run-queue/tests/dequeueMessageFromMasterQueue.ts"], + "include": ["src/**/*.test.ts"], "references": [{ "path": "./tsconfig.src.json" }], "compilerOptions": { "composite": true, diff --git a/packages/core/src/v3/apiClient/runStream.ts b/packages/core/src/v3/apiClient/runStream.ts index 4fed06d15..788299451 100644 --- a/packages/core/src/v3/apiClient/runStream.ts +++ b/packages/core/src/v3/apiClient/runStream.ts @@ -424,6 +424,9 @@ function apiStatusFromRunStatus(status: string): RunStatus { case "DELAYED": { return "DELAYED"; } + case "PENDING_VERSION": { + return "PENDING_VERSION"; + } case "WAITING_FOR_DEPLOY": { return "WAITING_FOR_DEPLOY"; } diff --git a/packages/core/src/v3/index.ts b/packages/core/src/v3/index.ts index f1ba7ac18..098362a7d 100644 --- a/packages/core/src/v3/index.ts +++ b/packages/core/src/v3/index.ts @@ -24,6 +24,7 @@ export * from "./jwt.js"; export * from "./idempotencyKeys.js"; export * from "./streams/asyncIterableStream.js"; export * from "./utils/getEnv.js"; +export * from "./tryCatch.js"; export { formatDuration, formatDurationInDays, diff --git a/packages/core/src/v3/schemas/api.ts b/packages/core/src/v3/schemas/api.ts index 10a46d5a6..ad5a4d6b2 100644 --- a/packages/core/src/v3/schemas/api.ts +++ b/packages/core/src/v3/schemas/api.ts @@ -609,6 +609,8 @@ export const TimezonesResult = z.object({ export type TimezonesResult = z.infer; export const RunStatus = z.enum([ + /// Task is waiting for a version update because it cannot execute without additional information (task, queue, etc.). Replaces WAITING_FOR_DEPLOY + "PENDING_VERSION", /// Task hasn't been deployed yet but is waiting to be executed "WAITING_FOR_DEPLOY", /// Task is waiting to be executed by a worker diff --git a/packages/core/src/v3/schemas/common.ts b/packages/core/src/v3/schemas/common.ts index fd2384dfa..43030847d 100644 --- a/packages/core/src/v3/schemas/common.ts +++ b/packages/core/src/v3/schemas/common.ts @@ -242,7 +242,7 @@ export type TaskRun = z.infer; export const TaskRunExecutionTask = z.object({ id: z.string(), filePath: z.string(), - exportName: z.string(), + exportName: z.string().optional(), }); export type TaskRunExecutionTask = z.infer; diff --git a/packages/core/src/v3/schemas/runEngine.ts b/packages/core/src/v3/schemas/runEngine.ts index abd7f2b5f..c45a651c1 100644 --- a/packages/core/src/v3/schemas/runEngine.ts +++ b/packages/core/src/v3/schemas/runEngine.ts @@ -21,6 +21,7 @@ export type TaskRunExecutionStatus = export const TaskRunStatus = { DELAYED: "DELAYED", PENDING: "PENDING", + PENDING_VERSION: "PENDING_VERSION", WAITING_FOR_DEPLOY: "WAITING_FOR_DEPLOY", EXECUTING: "EXECUTING", WAITING_TO_RESUME: "WAITING_TO_RESUME", diff --git a/packages/core/src/v3/tryCatch.ts b/packages/core/src/v3/tryCatch.ts new file mode 100644 index 000000000..664c6251f --- /dev/null +++ b/packages/core/src/v3/tryCatch.ts @@ -0,0 +1,15 @@ +// Types for the result object with discriminated union +type Success = [null, T]; +type Failure = [E, null]; + +type Result = Success | Failure; + +// Main wrapper function +export async function tryCatch(promise: Promise): Promise> { + try { + const data = await promise; + return [null, data]; + } catch (error) { + return [error as E, null]; + } +}