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

the run engine now works with the new reserve concurrency system

Browse Source
This commit is contained in:
Eric Allam
2025-03-12 11:44:16 +00:00
parent 555d9912ec
commit 1a233efd42
7 changed files with 1203 additions and 589 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.vscode/launch.json
Vendored
+2 -2
View File
@@ -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
},
{
apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts
+7 -7
View File
@@ -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
);
apps/webapp/app/v3/services/triggerTaskV2.server.ts
+18 -3
View File
@@ -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) {
internal-packages/run-engine/src/engine/index.ts
+47 -19
View File
@@ -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 };
});
}
internal-packages/run-engine/src/engine/tests/attemptFailures.test.ts
+539 -555
View File
File diff suppressed because it is too large Load Diff
internal-packages/run-engine/src/engine/tests/reserveConcurrency.test.ts
+585
View File
@@ -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();
}
}
);
});
references/test-tasks/src/trigger/test-reserve-concurrency-system.ts
+5 -3
View File
@@ -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);