diff --git a/.vscode/launch.json b/.vscode/launch.json index 8242758d3..da6e7674a 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -138,8 +138,8 @@ "type": "node-terminal", "request": "launch", "name": "Debug RunEngine tests", - "command": "pnpm run test --filter @internal/run-engine", - "cwd": "${workspaceFolder}", + "command": "pnpm run test ./src/engine/tests/attemptFailures.test.ts -t 'OOM fails after retrying on larger machine'", + "cwd": "${workspaceFolder}/internal-packages/run-engine", "sourceMaps": true }, { diff --git a/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts b/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts index 3bb2ecf66..49483a967 100644 --- a/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts +++ b/apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts @@ -2,7 +2,7 @@ import { ActionFunctionArgs, json, LoaderFunctionArgs } from "@remix-run/server- import { z } from "zod"; import { prisma } from "~/db.server"; import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server"; -import { marqs } from "~/v3/marqs/index.server"; +import { engine } from "~/v3/runEngine.server"; import { updateEnvConcurrencyLimits } from "~/v3/runQueue.server"; const ParamsSchema = z.object({ @@ -113,20 +113,20 @@ export async function loader({ request, params }: LoaderFunctionArgs) { Object.fromEntries(requestUrl.searchParams.entries()) ); - const concurrencyLimit = await marqs.getEnvConcurrencyLimit(environment); - const currentConcurrency = await marqs.currentConcurrencyOfEnvironment(environment); - const reserveConcurrency = await marqs.reserveConcurrencyOfEnvironment(environment); + const concurrencyLimit = await engine.runQueue.getEnvConcurrencyLimit(environment); + const currentConcurrency = await engine.runQueue.currentConcurrencyOfEnvironment(environment); + const reserveConcurrency = await engine.runQueue.reserveConcurrencyOfEnvironment(environment); if (searchParams.queue) { - const queueConcurrencyLimit = await marqs.getQueueConcurrencyLimit( + const queueConcurrencyLimit = await engine.runQueue.getQueueConcurrencyLimit( environment, searchParams.queue ); - const queueCurrentConcurrency = await marqs.currentConcurrencyOfQueue( + const queueCurrentConcurrency = await engine.runQueue.currentConcurrencyOfQueue( environment, searchParams.queue ); - const queueReserveConcurrency = await marqs.reserveConcurrencyOfQueue( + const queueReserveConcurrency = await engine.runQueue.reserveConcurrencyOfQueue( environment, searchParams.queue ); diff --git a/apps/webapp/app/v3/services/triggerTaskV2.server.ts b/apps/webapp/app/v3/services/triggerTaskV2.server.ts index 1d1f6cb55..a1d431b65 100644 --- a/apps/webapp/app/v3/services/triggerTaskV2.server.ts +++ b/apps/webapp/app/v3/services/triggerTaskV2.server.ts @@ -4,6 +4,9 @@ import { packetRequiresOffloading, QueueOptions, SemanticInternalAttributes, + TaskRunError, + taskRunErrorEnhancer, + taskRunErrorToString, TriggerTaskRequestBody, } from "@trigger.dev/core/v3"; import { @@ -271,7 +274,7 @@ export class TriggerTaskServiceV2 extends WithRunEngine { immediate: true, }, async (event, traceContext, traceparent) => { - const run = await autoIncrementCounter.incrementInTransaction( + const result = await autoIncrementCounter.incrementInTransaction( `v3-run:${environment.id}:${taskId}`, async (num, tx) => { const lockedToBackgroundWorker = body.options?.lockToVersion @@ -374,7 +377,13 @@ export class TriggerTaskServiceV2 extends WithRunEngine { this._prisma ); - return { run: taskRun, isCached: false }; + const error = taskRun.error ? TaskRunError.parse(taskRun.error) : undefined; + + if (error) { + event.failWithError(error); + } + + return { run: taskRun, error, isCached: false }; }, async (_, tx) => { const counter = await tx.taskRunNumberCounter.findFirst({ @@ -390,7 +399,13 @@ export class TriggerTaskServiceV2 extends WithRunEngine { this._prisma ); - return run; + if (result?.error) { + throw new ServiceValidationError( + taskRunErrorToString(taskRunErrorEnhancer(result.error)) + ); + } + + return result; } ); } catch (error) { diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 8c849ec6a..95d78af40 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -13,11 +13,9 @@ import { parsePacket, RetryOptions, RunExecutionData, - sanitizeError, - shouldRetryError, StartRunAttemptResult, TaskRunError, - taskRunErrorEnhancer, + TaskRunErrorCodes, TaskRunExecution, TaskRunExecutionResult, TaskRunFailedExecutionResult, @@ -52,8 +50,9 @@ import { assertNever } from "assert-never"; import { nanoid } from "nanoid"; import { EventEmitter } from "node:events"; import { z } from "zod"; -import { RunQueue } from "../run-queue/index.js"; import { FairQueueSelectionStrategy } from "../run-queue/fairQueueSelectionStrategy.js"; +import { RunQueue, RunQueueReserveConcurrencyOptions } from "../run-queue/index.js"; +import { RunQueueFullKeyProducer } from "../run-queue/keyProducer.js"; import { MinimalAuthenticatedEnvironment } from "../shared/index.js"; import { MAX_TASK_RUN_ATTEMPTS } from "./consts.js"; import { getRunWithBackgroundWorkerTasks } from "./db/worker.js"; @@ -62,6 +61,7 @@ import { EventBusEvents } from "./eventBus.js"; import { executionResultFromSnapshot, getLatestExecutionSnapshot } from "./executionSnapshots.js"; import { RunLocker } from "./locking.js"; import { getMachinePreset } from "./machinePresets.js"; +import { retryOutcomeFromCompletion } from "./retrying.js"; import { isCheckpointable, isDequeueableExecutionStatus, @@ -70,8 +70,6 @@ import { isPendingExecuting, } from "./statuses.js"; import { HeartbeatTimeouts, RunEngineOptions, TriggerParams } from "./types.js"; -import { RunQueueFullKeyProducer } from "../run-queue/keyProducer.js"; -import { retryOutcomeFromCompletion } from "./retrying.js"; const workerCatalog = { finishWaitpoint: { @@ -423,6 +421,8 @@ export class RunEngine { completedByTaskRunId: taskRun.id, }); + let reserveConcurrencyOptions: RunQueueReserveConcurrencyOptions | undefined; + //triggerAndWait or batchTriggerAndWait if (resumeParentOnCompletion && parentTaskRunId) { //this will block the parent run from continuing until this waitpoint is completed (and removed) @@ -438,9 +438,7 @@ export class RunEngine { tx: prisma, }); - //release the concurrency - //if the queue is the same then it's recursive and we need to release that too otherwise we could have a deadlock - const parentRun = await prisma.taskRun.findUnique({ + const parentRun = await prisma.taskRun.findFirst({ select: { queue: true, }, @@ -448,12 +446,13 @@ export class RunEngine { id: parentTaskRunId, }, }); - const releaseRunConcurrency = parentRun?.queue === taskRun.queue; - await this.runQueue.releaseConcurrency( - environment.organization.id, - parentTaskRunId, - releaseRunConcurrency - ); + + if (parentRun) { + reserveConcurrencyOptions = { + messageId: parentTaskRunId, + recursiveQueue: parentRun?.queue === taskRun.queue, + }; + } } //Make sure lock extension succeeded @@ -551,14 +550,27 @@ export class RunEngine { //enqueue the run if it's not delayed if (!taskRun.delayUntil) { - await this.#enqueueRun({ + const { wasEnqueued, error } = await this.#enqueueRun({ run: taskRun, env: environment, timestamp: Date.now() - taskRun.priorityMs, workerId, runnerId, tx: prisma, + reserveConcurrency: reserveConcurrencyOptions, }); + + if (error) { + // Fail the run immediately + taskRun = await prisma.taskRun.update({ + where: { id: taskRun.id }, + data: { + status: "SYSTEM_FAILURE", + completedAt: new Date(), + error, + }, + }); + } } }); @@ -3212,6 +3224,7 @@ export class RunEngine { completedWaitpoints, workerId, runnerId, + reserveConcurrency, }: { run: TaskRun; env: MinimalAuthenticatedEnvironment; @@ -3228,10 +3241,11 @@ export class RunEngine { }[]; workerId?: string; runnerId?: string; - }) { + reserveConcurrency?: RunQueueReserveConcurrencyOptions; + }): Promise<{ wasEnqueued: boolean; error?: TaskRunError }> { const prisma = tx ?? this.prisma; - await this.runLock.lock([run.id], 5000, async (signal) => { + return await this.runLock.lock([run.id], 5000, async (signal) => { const newSnapshot = await this.#createExecutionSnapshot(prisma, { run: run, snapshot: { @@ -3252,7 +3266,7 @@ export class RunEngine { masterQueues.push(run.secondaryMasterQueue); } - await this.runQueue.enqueueMessage({ + const wasEnqueued = await this.runQueue.enqueueMessage({ env, masterQueues, message: { @@ -3267,7 +3281,21 @@ export class RunEngine { timestamp, attempt: 0, }, + reserveConcurrency, }); + + if (!wasEnqueued) { + return { + wasEnqueued: false, + error: { + type: "INTERNAL_ERROR", + code: TaskRunErrorCodes.RECURSIVE_WAIT_DEADLOCK, + message: `This run will never execute because it was triggered recursively and the task has no remaining concurrency available`, + } satisfies TaskRunError, + }; + } + + return { wasEnqueued }; }); } diff --git a/internal-packages/run-engine/src/engine/tests/attemptFailures.test.ts b/internal-packages/run-engine/src/engine/tests/attemptFailures.test.ts index e46f2ae2f..6ff3f4d7e 100644 --- a/internal-packages/run-engine/src/engine/tests/attemptFailures.test.ts +++ b/internal-packages/run-engine/src/engine/tests/attemptFailures.test.ts @@ -10,164 +10,160 @@ import { expect } from "vitest"; import { RunEngine } from "../index.js"; describe("RunEngine attempt failures", () => { - containerTest( - "Retry user error and succeed", - { timeout: 15_000 }, - async ({ prisma, redisOptions }) => { - //create environment - const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + containerTest("Retry user error and succeed", 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, - }, + 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: { - defaultMachine: "small-1x", - machines: { - "small-1x": { - name: "small-1x" as const, - cpu: 0.5, - memory: 0.5, - centsPerMs: 0.0001, - }, + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, }, - baseCostInCents: 0.0001, }, - tracer: trace.getTracer("test", "0.0.0"), + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + //create background worker + const backgroundWorker = await setupBackgroundWorker( + prisma, + 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", + queueName: "task/test-task", + isTest: false, + tags: [], + }, + prisma + ); + + //dequeue the run + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: run.masterQueue, + maxRunCount: 10, }); - try { - const taskIdentifier = "test-task"; + //create an attempt + const attemptResult = await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: dequeued[0].snapshot.id, + }); - //create background worker - const backgroundWorker = await setupBackgroundWorker( - prisma, - 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", - queueName: "task/test-task", - isTest: false, - tags: [], + //fail the attempt + const error = { + type: "BUILT_IN_ERROR" as const, + name: "UserError", + message: "This is a user error", + stackTrace: "Error: This is a user error\n at :1:1", + }; + const result = await engine.completeRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: attemptResult.snapshot.id, + completion: { + ok: false, + id: dequeued[0].run.id, + error, + retry: { + timestamp: Date.now(), + delay: 0, }, - prisma - ); + }, + }); + expect(result.attemptStatus).toBe("RETRY_IMMEDIATELY"); + expect(result.snapshot.executionStatus).toBe("PENDING_EXECUTING"); + expect(result.run.status).toBe("RETRYING_AFTER_FAILURE"); - //dequeue the run - const dequeued = await engine.dequeueFromMasterQueue({ - consumerId: "test_12345", - masterQueue: run.masterQueue, - maxRunCount: 10, - }); + //state should be pending + const executionData3 = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(executionData3); + expect(executionData3.snapshot.executionStatus).toBe("PENDING_EXECUTING"); + //only when the new attempt is created, should the attempt be increased + expect(executionData3.run.attemptNumber).toBe(1); + expect(executionData3.run.status).toBe("RETRYING_AFTER_FAILURE"); - //create an attempt - const attemptResult = await engine.startRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: dequeued[0].snapshot.id, - }); + //create a second attempt + const attemptResult2 = await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: executionData3.snapshot.id, + }); + expect(attemptResult2.run.attemptNumber).toBe(2); - //fail the attempt - const error = { - type: "BUILT_IN_ERROR" as const, - name: "UserError", - message: "This is a user error", - stackTrace: "Error: This is a user error\n at :1:1", - }; - const result = await engine.completeRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: attemptResult.snapshot.id, - completion: { - ok: false, - id: dequeued[0].run.id, - error, - retry: { - timestamp: Date.now(), - delay: 0, - }, - }, - }); - expect(result.attemptStatus).toBe("RETRY_IMMEDIATELY"); - expect(result.snapshot.executionStatus).toBe("PENDING_EXECUTING"); - expect(result.run.status).toBe("RETRYING_AFTER_FAILURE"); + //now complete it successfully + const result2 = await engine.completeRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: attemptResult2.snapshot.id, + completion: { + ok: true, + id: dequeued[0].run.id, + output: `{"foo":"bar"}`, + outputType: "application/json", + }, + }); + expect(result2.snapshot.executionStatus).toBe("FINISHED"); + expect(result2.run.attemptNumber).toBe(2); + expect(result2.run.status).toBe("COMPLETED_SUCCESSFULLY"); - //state should be pending - const executionData3 = await engine.getRunExecutionData({ runId: run.id }); - assertNonNullable(executionData3); - expect(executionData3.snapshot.executionStatus).toBe("PENDING_EXECUTING"); - //only when the new attempt is created, should the attempt be increased - expect(executionData3.run.attemptNumber).toBe(1); - expect(executionData3.run.status).toBe("RETRYING_AFTER_FAILURE"); + //waitpoint should have been completed, with the output + const runWaitpointAfter = await prisma.waitpoint.findMany({ + where: { + completedByTaskRunId: run.id, + }, + }); + expect(runWaitpointAfter.length).toBe(1); + expect(runWaitpointAfter[0].type).toBe("RUN"); + expect(runWaitpointAfter[0].output).toBe(`{"foo":"bar"}`); + expect(runWaitpointAfter[0].outputIsError).toBe(false); - //create a second attempt - const attemptResult2 = await engine.startRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: executionData3.snapshot.id, - }); - expect(attemptResult2.run.attemptNumber).toBe(2); - - //now complete it successfully - const result2 = await engine.completeRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: attemptResult2.snapshot.id, - completion: { - ok: true, - id: dequeued[0].run.id, - output: `{"foo":"bar"}`, - outputType: "application/json", - }, - }); - expect(result2.snapshot.executionStatus).toBe("FINISHED"); - expect(result2.run.attemptNumber).toBe(2); - expect(result2.run.status).toBe("COMPLETED_SUCCESSFULLY"); - - //waitpoint should have been completed, with the output - const runWaitpointAfter = await prisma.waitpoint.findMany({ - where: { - completedByTaskRunId: run.id, - }, - }); - expect(runWaitpointAfter.length).toBe(1); - expect(runWaitpointAfter[0].type).toBe("RUN"); - expect(runWaitpointAfter[0].output).toBe(`{"foo":"bar"}`); - expect(runWaitpointAfter[0].outputIsError).toBe(false); - - //state should be completed - const executionData4 = await engine.getRunExecutionData({ runId: run.id }); - assertNonNullable(executionData4); - expect(executionData4.snapshot.executionStatus).toBe("FINISHED"); - expect(executionData4.run.attemptNumber).toBe(2); - expect(executionData4.run.status).toBe("COMPLETED_SUCCESSFULLY"); - } finally { - engine.quit(); - } + //state should be completed + const executionData4 = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(executionData4); + expect(executionData4.snapshot.executionStatus).toBe("FINISHED"); + expect(executionData4.run.attemptNumber).toBe(2); + expect(executionData4.run.status).toBe("COMPLETED_SUCCESSFULLY"); + } finally { + engine.quit(); } - ); + }); - containerTest("Fail (no more retries)", { timeout: 15_000 }, async ({ prisma, redisOptions }) => { + containerTest("Fail (no more retries)", async ({ prisma, redisOptions }) => { //create environment const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); @@ -278,120 +274,116 @@ describe("RunEngine attempt failures", () => { } }); - containerTest( - "Fail (not a retriable error)", - { timeout: 15_000 }, - async ({ prisma, redisOptions }) => { - //create environment - const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + containerTest("Fail (not a retriable error)", 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, - }, + 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: { - defaultMachine: "small-1x", - machines: { - "small-1x": { - name: "small-1x" as const, - cpu: 0.5, - memory: 0.5, - centsPerMs: 0.0001, - }, + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, }, - baseCostInCents: 0.0001, }, - tracer: trace.getTracer("test", "0.0.0"), + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + //create background worker + await setupBackgroundWorker(prisma, authenticatedEnvironment, taskIdentifier, undefined, { + maxAttempts: 1, }); - 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", + queueName: "task/test-task", + isTest: false, + tags: [], + }, + prisma + ); - //create background worker - await setupBackgroundWorker(prisma, authenticatedEnvironment, taskIdentifier, undefined, { - maxAttempts: 1, - }); + //dequeue the run + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: run.masterQueue, + maxRunCount: 10, + }); - //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", - queueName: "task/test-task", - isTest: false, - tags: [], + //create an attempt + const attemptResult = await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: dequeued[0].snapshot.id, + }); + + //fail the attempt with an unretriable error + const error = { + type: "INTERNAL_ERROR" as const, + code: "DISK_SPACE_EXCEEDED" as const, + }; + const result = await engine.completeRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: attemptResult.snapshot.id, + completion: { + ok: false, + id: dequeued[0].run.id, + error, + retry: { + timestamp: Date.now(), + delay: 0, }, - prisma - ); + }, + }); + expect(result.attemptStatus).toBe("RUN_FINISHED"); + expect(result.snapshot.executionStatus).toBe("FINISHED"); + expect(result.run.status).toBe("CRASHED"); - //dequeue the run - const dequeued = await engine.dequeueFromMasterQueue({ - consumerId: "test_12345", - masterQueue: run.masterQueue, - maxRunCount: 10, - }); - - //create an attempt - const attemptResult = await engine.startRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: dequeued[0].snapshot.id, - }); - - //fail the attempt with an unretriable error - const error = { - type: "INTERNAL_ERROR" as const, - code: "DISK_SPACE_EXCEEDED" as const, - }; - const result = await engine.completeRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: attemptResult.snapshot.id, - completion: { - ok: false, - id: dequeued[0].run.id, - error, - retry: { - timestamp: Date.now(), - delay: 0, - }, - }, - }); - expect(result.attemptStatus).toBe("RUN_FINISHED"); - expect(result.snapshot.executionStatus).toBe("FINISHED"); - expect(result.run.status).toBe("CRASHED"); - - //state should be pending - const executionData3 = await engine.getRunExecutionData({ runId: run.id }); - assertNonNullable(executionData3); - expect(executionData3.snapshot.executionStatus).toBe("FINISHED"); - //only when the new attempt is created, should the attempt be increased - expect(executionData3.run.attemptNumber).toBe(1); - expect(executionData3.run.status).toBe("CRASHED"); - } finally { - engine.quit(); - } + //state should be pending + const executionData3 = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(executionData3); + expect(executionData3.snapshot.executionStatus).toBe("FINISHED"); + //only when the new attempt is created, should the attempt be increased + expect(executionData3.run.attemptNumber).toBe(1); + expect(executionData3.run.status).toBe("CRASHED"); + } finally { + engine.quit(); } - ); + }); - containerTest("OOM fail", { timeout: 15_000 }, async ({ prisma, redisOptions }) => { + containerTest("OOM fail", async ({ prisma, redisOptions }) => { //create environment const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); @@ -498,332 +490,324 @@ describe("RunEngine attempt failures", () => { } }); - containerTest( - "OOM retry on larger machine", - { timeout: 15_000 }, - async ({ prisma, redisOptions }) => { - //create environment - const authenticatedEnvironment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + containerTest("OOM retry on larger machine", 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, - }, + 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: { - defaultMachine: "small-1x", - machines: { - "small-1x": { - name: "small-1x" as const, - cpu: 0.5, - memory: 0.5, - centsPerMs: 0.0001, - }, - "small-2x": { - name: "small-2x" as const, - cpu: 1, - memory: 1, - centsPerMs: 0.0002, - }, + "small-1x": { + name: "small-1x" as const, + cpu: 0.5, + memory: 0.5, + centsPerMs: 0.0001, + }, + "small-2x": { + name: "small-2x" as const, + cpu: 1, + memory: 1, + centsPerMs: 0.0002, }, - baseCostInCents: 0.0001, }, - tracer: trace.getTracer("test", "0.0.0"), + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + //create background worker + await setupBackgroundWorker(prisma, authenticatedEnvironment, taskIdentifier, undefined, { + outOfMemory: { + machine: "small-2x", + }, }); - try { - const taskIdentifier = "test-task"; - - //create background worker - await setupBackgroundWorker(prisma, authenticatedEnvironment, taskIdentifier, undefined, { - outOfMemory: { - machine: "small-2x", - }, - }); - - //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", - queueName: "task/test-task", - isTest: false, - tags: [], - }, - prisma - ); - - //dequeue the run - const dequeued = await engine.dequeueFromMasterQueue({ - consumerId: "test_12345", - masterQueue: run.masterQueue, - maxRunCount: 10, - }); - - //create an attempt - const attemptResult = await engine.startRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: dequeued[0].snapshot.id, - }); - - //fail the attempt with an OOM error - const error = { - type: "INTERNAL_ERROR" as const, - code: "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE" as const, - message: "Process exited with code -1 after signal SIGKILL.", - stackTrace: "JavaScript heap out of memory", - }; - - const result = await engine.completeRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: attemptResult.snapshot.id, - completion: { - ok: false, - id: dequeued[0].run.id, - error, - }, - }); - - // The run should be retried with a larger machine - expect(result.attemptStatus).toBe("RETRY_QUEUED"); - expect(result.snapshot.executionStatus).toBe("QUEUED"); - expect(result.run.status).toBe("RETRYING_AFTER_FAILURE"); - - //state should be pending - const executionData = await engine.getRunExecutionData({ runId: run.id }); - assertNonNullable(executionData); - expect(executionData.snapshot.executionStatus).toBe("QUEUED"); - expect(executionData.run.attemptNumber).toBe(1); - expect(executionData.run.status).toBe("RETRYING_AFTER_FAILURE"); - - //create a second attempt - const attemptResult2 = await engine.startRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: executionData.snapshot.id, - }); - expect(attemptResult2.run.attemptNumber).toBe(2); - - //now complete it successfully - const result2 = await engine.completeRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: attemptResult2.snapshot.id, - completion: { - ok: true, - id: dequeued[0].run.id, - output: `{"foo":"bar"}`, - outputType: "application/json", - }, - }); - expect(result2.snapshot.executionStatus).toBe("FINISHED"); - expect(result2.run.attemptNumber).toBe(2); - expect(result2.run.status).toBe("COMPLETED_SUCCESSFULLY"); - - //waitpoint should have been completed, with the output - const runWaitpointAfter = await prisma.waitpoint.findMany({ - where: { - completedByTaskRunId: run.id, - }, - }); - expect(runWaitpointAfter.length).toBe(1); - expect(runWaitpointAfter[0].type).toBe("RUN"); - expect(runWaitpointAfter[0].output).toBe(`{"foo":"bar"}`); - expect(runWaitpointAfter[0].outputIsError).toBe(false); - - //state should be completed - const executionData4 = await engine.getRunExecutionData({ runId: run.id }); - assertNonNullable(executionData4); - expect(executionData4.snapshot.executionStatus).toBe("FINISHED"); - expect(executionData4.run.attemptNumber).toBe(2); - expect(executionData4.run.status).toBe("COMPLETED_SUCCESSFULLY"); - } finally { - engine.quit(); - } - } - ); - - containerTest( - "OOM fails after retrying on larger machine", - { timeout: 15_000 }, - 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, + //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", + queueName: "task/test-task", + isTest: false, + tags: [], }, - 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, - }, - "small-2x": { - name: "small-2x" as const, - cpu: 1, - memory: 1, - centsPerMs: 0.0002, - }, - }, - baseCostInCents: 0.0001, - }, - tracer: trace.getTracer("test", "0.0.0"), + prisma + ); + + //dequeue the run + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: run.masterQueue, + maxRunCount: 10, }); - try { - const taskIdentifier = "test-task"; + //create an attempt + const attemptResult = await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: dequeued[0].snapshot.id, + }); - //create background worker - await setupBackgroundWorker(prisma, authenticatedEnvironment, taskIdentifier, undefined, { - maxTimeoutInMs: 10, - maxAttempts: 10, - outOfMemory: { - machine: "small-2x", - }, - }); + //fail the attempt with an OOM error + const error = { + type: "INTERNAL_ERROR" as const, + code: "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE" as const, + message: "Process exited with code -1 after signal SIGKILL.", + stackTrace: "JavaScript heap out of memory", + }; - //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", - queueName: "task/test-task", - isTest: false, - tags: [], - }, - prisma - ); + const result = await engine.completeRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: attemptResult.snapshot.id, + completion: { + ok: false, + id: dequeued[0].run.id, + error, + }, + }); - //dequeue the run - const dequeued = await engine.dequeueFromMasterQueue({ - consumerId: "test_12345", - masterQueue: run.masterQueue, - maxRunCount: 10, - }); + // The run should be retried with a larger machine + expect(result.attemptStatus).toBe("RETRY_QUEUED"); + expect(result.snapshot.executionStatus).toBe("QUEUED"); + expect(result.run.status).toBe("RETRYING_AFTER_FAILURE"); - //create first attempt - const attemptResult = await engine.startRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: dequeued[0].snapshot.id, - }); + //state should be pending + const executionData = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(executionData); + expect(executionData.snapshot.executionStatus).toBe("QUEUED"); + expect(executionData.run.attemptNumber).toBe(1); + expect(executionData.run.status).toBe("RETRYING_AFTER_FAILURE"); - //fail the first attempt with an OOM error - const error = { - type: "INTERNAL_ERROR" as const, - code: "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE" as const, - message: "Process exited with code -1 after signal SIGKILL.", - stackTrace: "JavaScript heap out of memory", - }; + //create a second attempt + const attemptResult2 = await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: executionData.snapshot.id, + }); + expect(attemptResult2.run.attemptNumber).toBe(2); - const result = await engine.completeRunAttempt({ - runId: dequeued[0].run.id, - snapshotId: attemptResult.snapshot.id, - completion: { - ok: false, - id: dequeued[0].run.id, - error, - }, - }); + //now complete it successfully + const result2 = await engine.completeRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: attemptResult2.snapshot.id, + completion: { + ok: true, + id: dequeued[0].run.id, + output: `{"foo":"bar"}`, + outputType: "application/json", + }, + }); + expect(result2.snapshot.executionStatus).toBe("FINISHED"); + expect(result2.run.attemptNumber).toBe(2); + expect(result2.run.status).toBe("COMPLETED_SUCCESSFULLY"); - // The run should be retried with a larger machine - expect(result.attemptStatus).toBe("RETRY_QUEUED"); - expect(result.snapshot.executionStatus).toBe("QUEUED"); - expect(result.run.status).toBe("RETRYING_AFTER_FAILURE"); + //waitpoint should have been completed, with the output + const runWaitpointAfter = await prisma.waitpoint.findMany({ + where: { + completedByTaskRunId: run.id, + }, + }); + expect(runWaitpointAfter.length).toBe(1); + expect(runWaitpointAfter[0].type).toBe("RUN"); + expect(runWaitpointAfter[0].output).toBe(`{"foo":"bar"}`); + expect(runWaitpointAfter[0].outputIsError).toBe(false); - //state should be queued - const executionData = await engine.getRunExecutionData({ runId: run.id }); - assertNonNullable(executionData); - expect(executionData.snapshot.executionStatus).toBe("QUEUED"); - expect(executionData.run.attemptNumber).toBe(1); - expect(executionData.run.status).toBe("RETRYING_AFTER_FAILURE"); - - //wait for 1s - await setTimeout(1_000); - - //dequeue again - const dequeued2 = await engine.dequeueFromMasterQueue({ - consumerId: "test_12345", - masterQueue: run.masterQueue, - maxRunCount: 10, - }); - expect(dequeued2.length).toBe(1); - - //create second attempt - const attemptResult2 = await engine.startRunAttempt({ - runId: dequeued2[0].run.id, - snapshotId: dequeued2[0].snapshot.id, - }); - expect(attemptResult2.run.attemptNumber).toBe(2); - - //fail the second attempt with the same OOM error - const result2 = await engine.completeRunAttempt({ - runId: dequeued2[0].run.id, - snapshotId: attemptResult2.snapshot.id, - completion: { - ok: false, - id: dequeued2[0].run.id, - error, - retry: { - timestamp: Date.now(), - delay: 0, - }, - }, - }); - - // The run should fail after the second OOM - expect(result2.attemptStatus).toBe("RUN_FINISHED"); - expect(result2.snapshot.executionStatus).toBe("FINISHED"); - expect(result2.run.status).toBe("CRASHED"); - - //final state should be crashed - const finalExecutionData = await engine.getRunExecutionData({ runId: run.id }); - assertNonNullable(finalExecutionData); - expect(finalExecutionData.snapshot.executionStatus).toBe("FINISHED"); - expect(finalExecutionData.run.attemptNumber).toBe(2); - expect(finalExecutionData.run.status).toBe("CRASHED"); - } finally { - engine.quit(); - } + //state should be completed + const executionData4 = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(executionData4); + expect(executionData4.snapshot.executionStatus).toBe("FINISHED"); + expect(executionData4.run.attemptNumber).toBe(2); + expect(executionData4.run.status).toBe("COMPLETED_SUCCESSFULLY"); + } finally { + engine.quit(); } - ); + }); + + containerTest("OOM fails after retrying on larger machine", 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, + }, + "small-2x": { + name: "small-2x" as const, + cpu: 1, + memory: 1, + centsPerMs: 0.0002, + }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + const taskIdentifier = "test-task"; + + //create background worker + await setupBackgroundWorker(prisma, authenticatedEnvironment, taskIdentifier, undefined, { + maxTimeoutInMs: 10, + maxAttempts: 10, + outOfMemory: { + machine: "small-2x", + }, + }); + + //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", + queueName: "task/test-task", + isTest: false, + tags: [], + }, + prisma + ); + + //dequeue the run + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: run.masterQueue, + maxRunCount: 10, + }); + + //create first attempt + const attemptResult = await engine.startRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: dequeued[0].snapshot.id, + }); + + //fail the first attempt with an OOM error + const error = { + type: "INTERNAL_ERROR" as const, + code: "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE" as const, + message: "Process exited with code -1 after signal SIGKILL.", + stackTrace: "JavaScript heap out of memory", + }; + + const result = await engine.completeRunAttempt({ + runId: dequeued[0].run.id, + snapshotId: attemptResult.snapshot.id, + completion: { + ok: false, + id: dequeued[0].run.id, + error, + }, + }); + + // The run should be retried with a larger machine + expect(result.attemptStatus).toBe("RETRY_QUEUED"); + expect(result.snapshot.executionStatus).toBe("QUEUED"); + expect(result.run.status).toBe("RETRYING_AFTER_FAILURE"); + + //state should be queued + const executionData = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(executionData); + expect(executionData.snapshot.executionStatus).toBe("QUEUED"); + expect(executionData.run.attemptNumber).toBe(1); + expect(executionData.run.status).toBe("RETRYING_AFTER_FAILURE"); + + //wait for 1s + await setTimeout(5_000); + + //dequeue again + const dequeued2 = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: run.masterQueue, + maxRunCount: 10, + }); + expect(dequeued2.length).toBe(1); + + //create second attempt + const attemptResult2 = await engine.startRunAttempt({ + runId: dequeued2[0].run.id, + snapshotId: dequeued2[0].snapshot.id, + }); + expect(attemptResult2.run.attemptNumber).toBe(2); + + //fail the second attempt with the same OOM error + const result2 = await engine.completeRunAttempt({ + runId: dequeued2[0].run.id, + snapshotId: attemptResult2.snapshot.id, + completion: { + ok: false, + id: dequeued2[0].run.id, + error, + retry: { + timestamp: Date.now(), + delay: 0, + }, + }, + }); + + // The run should fail after the second OOM + expect(result2.attemptStatus).toBe("RUN_FINISHED"); + expect(result2.snapshot.executionStatus).toBe("FINISHED"); + expect(result2.run.status).toBe("CRASHED"); + + //final state should be crashed + const finalExecutionData = await engine.getRunExecutionData({ runId: run.id }); + assertNonNullable(finalExecutionData); + expect(finalExecutionData.snapshot.executionStatus).toBe("FINISHED"); + expect(finalExecutionData.run.attemptNumber).toBe(2); + expect(finalExecutionData.run.status).toBe("CRASHED"); + } finally { + engine.quit(); + } + }); }); diff --git a/internal-packages/run-engine/src/engine/tests/reserveConcurrency.test.ts b/internal-packages/run-engine/src/engine/tests/reserveConcurrency.test.ts new file mode 100644 index 000000000..d2ba5f1b3 --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/reserveConcurrency.test.ts @@ -0,0 +1,585 @@ +import { + assertNonNullable, + containerTest, + setupAuthenticatedEnvironment, + setupBackgroundWorker, +} from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { expect } from "vitest"; +import { RunEngine } from "../index.js"; +import { setTimeout } from "node:timers/promises"; +import { TaskRunErrorCodes } from "@trigger.dev/core/v3/schemas"; + +vi.setConfig({ testTimeout: 60_000 }); + +describe("Reserve concurrency", () => { + containerTest( + "triggerAndWait reserves concurrency on the environment when triggering a child task on a different queue", + 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, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + await engine.runQueue.updateEnvConcurrencyLimits({ + ...authenticatedEnvironment, + maximumConcurrencyLimit: 1, + }); + + const parentTask = "parent-task"; + const childTask = "child-task"; + + //create background worker + await setupBackgroundWorker(prisma, authenticatedEnvironment, [parentTask, childTask]); + + //trigger the run + const parentRun = await engine.trigger( + { + number: 1, + friendlyId: "run_p1234", + environment: authenticatedEnvironment, + taskIdentifier: parentTask, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345", + spanId: "s12345", + masterQueue: "main", + queueName: `task/${parentTask}`, + isTest: false, + tags: [], + }, + prisma + ); + + //dequeue parent + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: parentRun.masterQueue, + maxRunCount: 10, + }); + + //create an attempt + const initialExecutionData = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(initialExecutionData); + const attemptResult = await engine.startRunAttempt({ + runId: parentRun.id, + snapshotId: initialExecutionData.snapshot.id, + }); + + const childRun = await engine.trigger( + { + number: 1, + friendlyId: "run_c1234", + environment: authenticatedEnvironment, + taskIdentifier: childTask, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345", + spanId: "s12345", + masterQueue: "main", + queueName: `task/${childTask}`, + isTest: false, + tags: [], + resumeParentOnCompletion: true, + parentTaskRunId: parentRun.id, + }, + prisma + ); + + const childExecutionData = await engine.getRunExecutionData({ runId: childRun.id }); + assertNonNullable(childExecutionData); + expect(childExecutionData.snapshot.executionStatus).toBe("QUEUED"); + + const parentExecutionData = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(parentExecutionData); + expect(parentExecutionData.snapshot.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS"); + + //check the waitpoint blocking the parent run + const runWaitpoint = await prisma.taskRunWaitpoint.findFirst({ + where: { + taskRunId: parentRun.id, + }, + include: { + waitpoint: true, + }, + }); + assertNonNullable(runWaitpoint); + expect(runWaitpoint.waitpoint.type).toBe("RUN"); + expect(runWaitpoint.waitpoint.completedByTaskRunId).toBe(childRun.id); + + //dequeue the child run + const dequeuedChild = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: childRun.masterQueue, + maxRunCount: 10, + }); + + expect(dequeuedChild.length).toBe(1); + + //start the child run + const childAttempt = await engine.startRunAttempt({ + runId: childRun.id, + snapshotId: dequeuedChild[0].snapshot.id, + }); + + // complete the child run + await engine.completeRunAttempt({ + runId: childRun.id, + snapshotId: childAttempt.snapshot.id, + completion: { + id: childRun.id, + ok: true, + output: '{"foo":"bar"}', + outputType: "application/json", + }, + }); + + //child snapshot + const childExecutionDataAfter = await engine.getRunExecutionData({ runId: childRun.id }); + assertNonNullable(childExecutionDataAfter); + expect(childExecutionDataAfter.snapshot.executionStatus).toBe("FINISHED"); + + const waitpointAfter = await prisma.waitpoint.findFirst({ + where: { + id: runWaitpoint.waitpointId, + }, + }); + expect(waitpointAfter?.completedAt).not.toBeNull(); + expect(waitpointAfter?.status).toBe("COMPLETED"); + expect(waitpointAfter?.output).toBe('{"foo":"bar"}'); + + await setTimeout(500); + + const runWaitpointAfter = await prisma.taskRunWaitpoint.findFirst({ + where: { + taskRunId: parentRun.id, + }, + include: { + waitpoint: true, + }, + }); + expect(runWaitpointAfter).toBeNull(); + + //parent snapshot + const parentExecutionDataAfter = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(parentExecutionDataAfter); + expect(parentExecutionDataAfter.snapshot.executionStatus).toBe("EXECUTING"); + expect(parentExecutionDataAfter.completedWaitpoints?.length).toBe(1); + expect(parentExecutionDataAfter.completedWaitpoints![0].id).toBe(runWaitpoint.waitpointId); + expect(parentExecutionDataAfter.completedWaitpoints![0].completedByTaskRun?.id).toBe( + childRun.id + ); + expect(parentExecutionDataAfter.completedWaitpoints![0].output).toBe('{"foo":"bar"}'); + } finally { + engine.quit(); + } + } + ); + + containerTest( + "triggerAndWait reserves concurrency on the environment and the queue when triggering a child task on the same queue", + 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, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + await engine.runQueue.updateEnvConcurrencyLimits({ + ...authenticatedEnvironment, + maximumConcurrencyLimit: 1, + }); + + const parentTask = "parent-task"; + const childTask = "child-task"; + + //create background worker + await setupBackgroundWorker(prisma, authenticatedEnvironment, [parentTask, childTask]); + + //trigger the run + const parentRun = await engine.trigger( + { + number: 1, + friendlyId: "run_p1234", + environment: authenticatedEnvironment, + taskIdentifier: parentTask, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345", + spanId: "s12345", + masterQueue: "main", + queueName: "shared-queue", + queue: { + concurrencyLimit: 1, + }, + isTest: false, + tags: [], + }, + prisma + ); + + //dequeue parent + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: parentRun.masterQueue, + maxRunCount: 10, + }); + + expect(dequeued.length).toBe(1); + + //create an attempt + const initialExecutionData = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(initialExecutionData); + const attemptResult = await engine.startRunAttempt({ + runId: parentRun.id, + snapshotId: initialExecutionData.snapshot.id, + }); + + expect(attemptResult).toBeDefined(); + + const childRun = await engine.trigger( + { + number: 1, + friendlyId: "run_c1234", + environment: authenticatedEnvironment, + taskIdentifier: childTask, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345", + spanId: "s12345", + masterQueue: "main", + queueName: "shared-queue", + queue: { + concurrencyLimit: 1, + }, + isTest: false, + tags: [], + resumeParentOnCompletion: true, + parentTaskRunId: parentRun.id, + }, + prisma + ); + + const childExecutionData = await engine.getRunExecutionData({ runId: childRun.id }); + assertNonNullable(childExecutionData); + expect(childExecutionData.snapshot.executionStatus).toBe("QUEUED"); + + const parentExecutionData = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(parentExecutionData); + expect(parentExecutionData.snapshot.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS"); + + //check the waitpoint blocking the parent run + const runWaitpoint = await prisma.taskRunWaitpoint.findFirst({ + where: { + taskRunId: parentRun.id, + }, + include: { + waitpoint: true, + }, + }); + assertNonNullable(runWaitpoint); + expect(runWaitpoint.waitpoint.type).toBe("RUN"); + expect(runWaitpoint.waitpoint.completedByTaskRunId).toBe(childRun.id); + + //dequeue the child run + const dequeuedChild = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: childRun.masterQueue, + maxRunCount: 10, + }); + + expect(dequeuedChild.length).toBe(1); + + //start the child run + const childAttempt = await engine.startRunAttempt({ + runId: childRun.id, + snapshotId: dequeuedChild[0].snapshot.id, + }); + + // complete the child run + await engine.completeRunAttempt({ + runId: childRun.id, + snapshotId: childAttempt.snapshot.id, + completion: { + id: childRun.id, + ok: true, + output: '{"foo":"bar"}', + outputType: "application/json", + }, + }); + + //child snapshot + const childExecutionDataAfter = await engine.getRunExecutionData({ runId: childRun.id }); + assertNonNullable(childExecutionDataAfter); + expect(childExecutionDataAfter.snapshot.executionStatus).toBe("FINISHED"); + + const waitpointAfter = await prisma.waitpoint.findFirst({ + where: { + id: runWaitpoint.waitpointId, + }, + }); + expect(waitpointAfter?.completedAt).not.toBeNull(); + expect(waitpointAfter?.status).toBe("COMPLETED"); + expect(waitpointAfter?.output).toBe('{"foo":"bar"}'); + + await setTimeout(500); + + const runWaitpointAfter = await prisma.taskRunWaitpoint.findFirst({ + where: { + taskRunId: parentRun.id, + }, + include: { + waitpoint: true, + }, + }); + expect(runWaitpointAfter).toBeNull(); + + //parent snapshot + const parentExecutionDataAfter = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(parentExecutionDataAfter); + expect(parentExecutionDataAfter.snapshot.executionStatus).toBe("EXECUTING"); + expect(parentExecutionDataAfter.completedWaitpoints?.length).toBe(1); + expect(parentExecutionDataAfter.completedWaitpoints![0].id).toBe(runWaitpoint.waitpointId); + expect(parentExecutionDataAfter.completedWaitpoints![0].completedByTaskRun?.id).toBe( + childRun.id + ); + expect(parentExecutionDataAfter.completedWaitpoints![0].output).toBe('{"foo":"bar"}'); + } finally { + engine.quit(); + } + } + ); + + containerTest( + "triggerAndWait fails with recursive deadlock error when there is no more reserve concurrency left when triggering a child task on the same queue", + 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, + }, + tracer: trace.getTracer("test", "0.0.0"), + }); + + try { + await engine.runQueue.updateEnvConcurrencyLimits({ + ...authenticatedEnvironment, + maximumConcurrencyLimit: 1, + }); + + const parentTask = "parent-task"; + const childTask = "child-task"; + + //create background worker + await setupBackgroundWorker(prisma, authenticatedEnvironment, [parentTask, childTask]); + + //trigger the run + const parentRun = await engine.trigger( + { + number: 1, + friendlyId: "run_p1234", + environment: authenticatedEnvironment, + taskIdentifier: parentTask, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345", + spanId: "s12345", + masterQueue: "main", + queueName: "shared-queue", + queue: { + concurrencyLimit: 1, + }, + isTest: false, + tags: [], + }, + prisma + ); + + //dequeue parent + const dequeued = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: parentRun.masterQueue, + maxRunCount: 10, + }); + + expect(dequeued.length).toBe(1); + + //create an attempt + const initialExecutionData = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(initialExecutionData); + const attemptResult = await engine.startRunAttempt({ + runId: parentRun.id, + snapshotId: initialExecutionData.snapshot.id, + }); + + expect(attemptResult).toBeDefined(); + + const childRun = await engine.trigger( + { + number: 1, + friendlyId: "run_c1234", + environment: authenticatedEnvironment, + taskIdentifier: childTask, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345", + spanId: "s12345", + masterQueue: "main", + queueName: "shared-queue", + queue: { + concurrencyLimit: 1, + }, + isTest: false, + tags: [], + resumeParentOnCompletion: true, + parentTaskRunId: parentRun.id, + }, + prisma + ); + + const childExecutionData = await engine.getRunExecutionData({ runId: childRun.id }); + assertNonNullable(childExecutionData); + expect(childExecutionData.snapshot.executionStatus).toBe("QUEUED"); + + const parentExecutionData = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(parentExecutionData); + expect(parentExecutionData.snapshot.executionStatus).toBe("EXECUTING_WITH_WAITPOINTS"); + + //dequeue the child run + const dequeuedChild = await engine.dequeueFromMasterQueue({ + consumerId: "test_12345", + masterQueue: childRun.masterQueue, + maxRunCount: 10, + }); + + expect(dequeuedChild.length).toBe(1); + + // Now try and trigger another child run on the same queue + const childRun2 = await engine.trigger( + { + number: 1, + friendlyId: "run_c12345", + environment: authenticatedEnvironment, + taskIdentifier: childTask, + payload: "{}", + payloadType: "application/json", + context: {}, + traceContext: {}, + traceId: "t12345_2", + spanId: "s12345_2", + masterQueue: "main", + queueName: "shared-queue", + queue: { + concurrencyLimit: 1, + }, + isTest: false, + tags: [], + resumeParentOnCompletion: true, + parentTaskRunId: parentRun.id, + }, + prisma + ); + + expect(childRun2.status).toBe("SYSTEM_FAILURE"); + expect(childRun2.error).toEqual({ + type: "INTERNAL_ERROR", + code: TaskRunErrorCodes.RECURSIVE_WAIT_DEADLOCK, + message: expect.any(String), + }); + } finally { + engine.quit(); + } + } + ); +}); diff --git a/references/test-tasks/src/trigger/test-reserve-concurrency-system.ts b/references/test-tasks/src/trigger/test-reserve-concurrency-system.ts index 05b8eba9a..4a0b04479 100644 --- a/references/test-tasks/src/trigger/test-reserve-concurrency-system.ts +++ b/references/test-tasks/src/trigger/test-reserve-concurrency-system.ts @@ -1,4 +1,4 @@ -import { logger, task } from "@trigger.dev/sdk/v3"; +import { batch, logger, task } from "@trigger.dev/sdk/v3"; import assert from "assert"; import { getEnvironmentStats, @@ -293,8 +293,10 @@ export const testEnvReserveConcurrency = task({ })) ); + const retrievedHoldBatch = await batch.retrieve(holdBatch.batchId); + // Wait for the hold tasks to be executing - await Promise.all(holdBatch.runs.map((run) => waitForRunStatus(run.id, ["EXECUTING"]))); + await Promise.all(retrievedHoldBatch.runs.map((run) => waitForRunStatus(run, ["EXECUTING"]))); // Now we will trigger a parent task that will trigger a child task const parentRun = await genericParentTask.trigger( @@ -341,7 +343,7 @@ export const testEnvReserveConcurrency = task({ ); // Wait for the hold tasks to be completed - await Promise.all(holdBatch.runs.map((run) => waitForRunStatus(run.id, ["COMPLETED"]))); + await Promise.all(retrievedHoldBatch.runs.map((run) => waitForRunStatus(run, ["COMPLETED"]))); await updateEnvironmentConcurrencyLimit(ctx.environment.id, 100);