github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

New PENDING_VERSION system which now requires queues to exist at dequeue time

Browse Source
This commit is contained in:
Eric Allam
2025-03-20 15:07:19 +00:00
parent 7a58439728
commit 7e411ac162
40 changed files with 793 additions and 359 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/webapp/app/components/runs/v3/TaskRunStatus.tsx
+11 -3
View File
@@ -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<Array<TaskRunStatus>>;
export const filterableTaskRunStatuses = [
"WAITING_FOR_DEPLOY",
"PENDING_VERSION",
"DELAYED",
"PENDING",
"WAITING_TO_RESUME",
@@ -56,7 +57,10 @@ export const filterableTaskRunStatuses = [
const taskRunStatusDescriptions: Record<TaskRunStatus, string> = {
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<TaskRunStatus, string> = {
export const QUEUED_STATUSES = [
"PENDING",
"PENDING_VERSION",
"WAITING_FOR_DEPLOY",
"DELAYED",
] satisfies TaskRunStatus[];
@@ -120,6 +125,7 @@ export function TaskRunStatusIcon({
return <ClockIcon className={cn(runStatusClassNameColor(status), className)} />;
case "PENDING":
return <RectangleStackIcon className={cn(runStatusClassNameColor(status), className)} />;
case "PENDING_VERSION":
case "WAITING_FOR_DEPLOY":
return <RectangleStackIcon className={cn(runStatusClassNameColor(status), className)} />;
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":
apps/webapp/app/database-types.ts
+1
View File
@@ -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",
apps/webapp/app/hooks/useFilterTasks.ts
-5
View File
@@ -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<T extends Task>({ tasks }: { tasks: T[] }) {
return true;
}
if (task.exportName.toLowerCase().includes(text.toLowerCase())) {
return true;
}
if (task.filePath.toLowerCase().includes(text.toLowerCase())) {
return true;
}
apps/webapp/app/models/taskQueue.server.ts
+2 -2
View File
@@ -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
apps/webapp/app/models/taskRun.server.ts
+1
View File
@@ -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:
apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts
+8 -1
View File
@@ -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" ||
apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts
+3
View File
@@ -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";
}
apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts
+1 -2
View File
@@ -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,
};
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam._index/route.tsx
+1 -1
View File
@@ -507,7 +507,7 @@ function TaskActivityGraph({ activity }: { activity: TaskActivity }) {
isAnimationActive={false}
/>
<Bar dataKey="PENDING" fill="#5F6570" stackId="a" strokeWidth={0} barSize={10} />
<Bar dataKey="WAITING_FOR_DEPLOY" fill="#F59E0B" stackId="a" strokeWidth={0} barSize={10} />
<Bar dataKey="PENDING_VERSION" fill="#F59E0B" stackId="a" strokeWidth={0} barSize={10} />
<Bar dataKey="EXECUTING" fill="#3B82F6" stackId="a" strokeWidth={0} barSize={10} />
<Bar
dataKey="RETRYING_AFTER_FAILURE"
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.deployments.$deploymentParam/route.tsx
-5
View File
@@ -236,11 +236,6 @@ export default function Page() {
<TableRow key={t.slug}>
<TableCell>
<div className="inline-flex flex-col gap-0.5">
<TaskFunctionName
variant="extra-small"
functionName={t.exportName}
className="-ml-1 inline-flex"
/>
<Paragraph variant="extra-small" className="text-text-dimmed">
{t.slug}
</Paragraph>
apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts
+1 -1
View File
@@ -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,
apps/webapp/app/v3/services/alerts/deliverAlert.server.ts
+1 -1
View File
@@ -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: {
apps/webapp/app/v3/services/createBackgroundWorker.server.ts
+104 -49
View File
@@ -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 } },
},
});
}
apps/webapp/app/v3/services/createDeployedBackgroundWorker.server.ts
+1 -6
View File
@@ -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,
apps/webapp/app/v3/services/createDeploymentBackgroundWorker.server.ts
+72 -33
View File
@@ -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;
}
}
apps/webapp/app/v3/services/createTaskRunAttempt.server.ts
+1 -1
View File
@@ -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,
apps/webapp/app/v3/services/pauseEnvironment.server.ts
+1 -1
View File
@@ -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";
apps/webapp/app/v3/taskRunHeartbeatFailed.server.ts
+1
View File
@@ -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`,
apps/webapp/app/v3/taskStatus.ts
+1
View File
@@ -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",
docs/v3-openapi.yaml
+3 -3
View File
@@ -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
internal-packages/database/prisma/migrations/20250320111737_add_pending_version_task_run_status/migration.sql
+4
View File
@@ -0,0 +1,4 @@
-- AlterEnum
ALTER TYPE "TaskRunStatus"
ADD
VALUE 'PENDING_VERSION';
internal-packages/database/prisma/migrations/20250320130354_add_many_to_many_relationship_task_queue_background_worker/migration.sql
+20
View File
@@ -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;
internal-packages/database/prisma/schema.prisma
+6 -1
View File
@@ -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])
}
internal-packages/run-engine/src/engine/db/worker.ts
+25 -4
View File
@@ -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,
};
}
internal-packages/run-engine/src/engine/index.ts
+9 -15
View File
@@ -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<void> {
return this.waitingForWorkerSystem.enqueueRunsWaitingForWorker({
backgroundWorkerId,
});
async scheduleEnqueueRunsForBackgroundWorker(backgroundWorkerId: string): Promise<void> {
return this.pendingVersionSystem.scheduleResolvePendingVersionRuns(backgroundWorkerId);
}
/**
internal-packages/run-engine/src/engine/systems/dequeueSystem.ts
+12 -43
View File
@@ -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: {
internal-packages/run-engine/src/engine/systems/waitingForWorkerSystem.ts → internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts
+38 -19
View File
@@ -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<void> {
async scheduleResolvePendingVersionRuns(backgroundWorkerId: string): Promise<void> {
//we want this to happen in the background
await this.$.worker.enqueue({
job: "queueRunsWaitingForWorker",
job: "queueRunsPendingVersion",
payload: { backgroundWorkerId },
});
}
internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts
+1 -1
View File
@@ -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,
internal-packages/run-engine/src/engine/tests/notDeployed.test.ts
-150
View File
@@ -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();
}
});
});
internal-packages/run-engine/src/engine/tests/pendingVersion.test.ts
+319
View File
@@ -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();
}
}
);
});
internal-packages/run-engine/src/engine/tests/setup.ts
+118 -8
View File
@@ -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`;
}
internal-packages/run-engine/src/engine/workerCatalog.ts
+1 -1
View File
@@ -29,7 +29,7 @@ export const workerCatalog = {
}),
visibilityTimeoutMs: 5000,
},
queueRunsWaitingForWorker: {
queueRunsPendingVersion: {
schema: z.object({
backgroundWorkerId: z.string(),
}),
internal-packages/run-engine/tsconfig.src.json
+2 -1
View File
@@ -14,6 +14,7 @@
"isolatedModules": true,
"preserveWatchOutput": true,
"skipLibCheck": true,
"strict": true
"strict": true,
"customConditions": ["@triggerdotdev/source"]
}
}
internal-packages/run-engine/tsconfig.test.json
+1 -1
View File
@@ -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,
packages/core/src/v3/apiClient/runStream.ts
+3
View File
@@ -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";
}
packages/core/src/v3/index.ts
+1
View File
@@ -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,
packages/core/src/v3/schemas/api.ts
+2
View File
@@ -609,6 +609,8 @@ export const TimezonesResult = z.object({
export type TimezonesResult = z.infer<typeof TimezonesResult>;
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
packages/core/src/v3/schemas/common.ts
+1 -1
View File
@@ -242,7 +242,7 @@ export type TaskRun = z.infer<typeof TaskRun>;
export const TaskRunExecutionTask = z.object({
id: z.string(),
filePath: z.string(),
exportName: z.string(),
exportName: z.string().optional(),
});
export type TaskRunExecutionTask = z.infer<typeof TaskRunExecutionTask>;
packages/core/src/v3/schemas/runEngine.ts
+1
View File
@@ -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",
packages/core/src/v3/tryCatch.ts
+15
View File
@@ -0,0 +1,15 @@
// Types for the result object with discriminated union
type Success<T> = [null, T];
type Failure<E> = [E, null];
type Result<T, E = Error> = Success<T> | Failure<E>;
// Main wrapper function
export async function tryCatch<T, E = Error>(promise: Promise<T>): Promise<Result<T, E>> {
try {
const data = await promise;
return [null, data];
} catch (error) {
return [error as E, null];
}
}