Skip to content

Fix issue with heavy contention on TaskQueue updating concurrency limit #1653

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jan 30, 2025
Merged

Fix issue with heavy contention on TaskQueue updating concurrency limit #1653

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9d0bf58
Fix issue with heavy contention on TaskQueue updating concurrency limit
ericallam Jan 30, 2025
96bd406
Remove concurrency limit when creating background worker tasks
ericallam Jan 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/funny-emus-pay.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@trigger.dev/core": patch
---

Allow setting concurrencyLimit to null to signal removing the concurrency limit on the queue
41 changes: 19 additions & 22 deletions apps/webapp/app/v3/services/createBackgroundWorker.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,29 +180,29 @@ export async function createBackgroundTasks(
),
0
)
: null;
: task.queue?.concurrencyLimit;

const taskQueue = await prisma.taskQueue.upsert({
let taskQueue = await prisma.taskQueue.findFirst({
where: {
runtimeEnvironmentId_name: {
runtimeEnvironmentId: worker.runtimeEnvironmentId,
name: queueName,
},
},
update: {
concurrencyLimit,
},
create: {
friendlyId: generateFriendlyId("queue"),
name: queueName,
concurrencyLimit,
runtimeEnvironmentId: worker.runtimeEnvironmentId,
projectId: worker.projectId,
type: task.queue?.name ? "NAMED" : "VIRTUAL",
name: queueName,
},
});

if (typeof taskQueue.concurrencyLimit === "number") {
if (!taskQueue) {
taskQueue = await prisma.taskQueue.create({
data: {
friendlyId: generateFriendlyId("queue"),
name: queueName,
concurrencyLimit,
runtimeEnvironmentId: worker.runtimeEnvironmentId,
projectId: worker.projectId,
type: task.queue?.name ? "NAMED" : "VIRTUAL",
},
});
}

if (typeof concurrencyLimit === "number") {
logger.debug("CreateBackgroundWorkerService: updating concurrency limit", {
workerId: worker.id,
taskQueue,
Expand All @@ -212,11 +212,7 @@ export async function createBackgroundTasks(
concurrencyLimit,
taskidentifier: task.id,
});
await marqs?.updateQueueConcurrencyLimits(
environment,
taskQueue.name,
taskQueue.concurrencyLimit
);
await marqs?.updateQueueConcurrencyLimits(environment, taskQueue.name, concurrencyLimit);
} else {
logger.debug("CreateBackgroundWorkerService: removing concurrency limit", {
workerId: worker.id,
Expand All @@ -227,6 +223,7 @@ export async function createBackgroundTasks(
concurrencyLimit,
taskidentifier: task.id,
});

await marqs?.removeQueueConcurrencyLimits(environment, taskQueue.name);
}
} catch (error) {
Expand Down
1 change: 0 additions & 1 deletion apps/webapp/app/v3/services/replayTaskRun.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,6 @@ export class ReplayTaskRunService extends BaseService {
queue: taskQueue
? {
name: taskQueue.name,
concurrencyLimit: taskQueue.concurrencyLimit ?? undefined,
}
: undefined,
concurrencyKey: existingTaskRun.concurrencyKey ?? undefined,
Expand Down
103 changes: 44 additions & 59 deletions apps/webapp/app/v3/services/triggerTask.server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -450,7 +450,7 @@ export class TriggerTaskService extends BaseService {
),
0
)
: null;
: body.options.queue?.concurrencyLimit;

let taskQueue = await tx.taskQueue.findFirst({
where: {
Expand All @@ -459,74 +459,47 @@ export class TriggerTaskService extends BaseService {
},
});

const existingConcurrencyLimit =
typeof taskQueue?.concurrencyLimit === "number"
? taskQueue.concurrencyLimit
: undefined;

if (taskQueue) {
if (existingConcurrencyLimit !== concurrencyLimit) {
taskQueue = await tx.taskQueue.update({
where: {
id: taskQueue.id,
},
data: {
concurrencyLimit:
typeof concurrencyLimit === "number" ? concurrencyLimit : null,
},
});

if (typeof taskQueue.concurrencyLimit === "number") {
logger.debug("TriggerTaskService: updating concurrency limit", {
runId: taskRun.id,
friendlyId: taskRun.friendlyId,
taskQueue,
orgId: environment.organizationId,
projectId: environment.projectId,
existingConcurrencyLimit,
concurrencyLimit,
queueOptions: body.options?.queue,
});
await marqs?.updateQueueConcurrencyLimits(
environment,
taskQueue.name,
taskQueue.concurrencyLimit
);
} else {
logger.debug("TriggerTaskService: removing concurrency limit", {
runId: taskRun.id,
friendlyId: taskRun.friendlyId,
taskQueue,
orgId: environment.organizationId,
projectId: environment.projectId,
existingConcurrencyLimit,
concurrencyLimit,
queueOptions: body.options?.queue,
});
await marqs?.removeQueueConcurrencyLimits(environment, taskQueue.name);
}
}
} else {
const queueId = generateFriendlyId("queue");

if (!taskQueue) {
// handle conflicts with existing queues
taskQueue = await tx.taskQueue.create({
data: {
friendlyId: queueId,
friendlyId: generateFriendlyId("queue"),
name: queueName,
concurrencyLimit,
runtimeEnvironmentId: environment.id,
projectId: environment.projectId,
type: "NAMED",
},
});
}

if (typeof taskQueue.concurrencyLimit === "number") {
await marqs?.updateQueueConcurrencyLimits(
environment,
taskQueue.name,
taskQueue.concurrencyLimit
);
}
if (typeof concurrencyLimit === "number") {
logger.debug("TriggerTaskService: updating concurrency limit", {
runId: taskRun.id,
friendlyId: taskRun.friendlyId,
taskQueue,
orgId: environment.organizationId,
projectId: environment.projectId,
concurrencyLimit,
queueOptions: body.options?.queue,
});

await marqs?.updateQueueConcurrencyLimits(
environment,
taskQueue.name,
concurrencyLimit
);
} else if (concurrencyLimit === null) {
logger.debug("TriggerTaskService: removing concurrency limit", {
runId: taskRun.id,
friendlyId: taskRun.friendlyId,
taskQueue,
orgId: environment.organizationId,
projectId: environment.projectId,
queueOptions: body.options?.queue,
});

await marqs?.removeQueueConcurrencyLimits(environment, taskQueue.name);
}
}

Expand Down Expand Up @@ -623,6 +596,18 @@ export class TriggerTaskService extends BaseService {
throw new ServiceValidationError(
`Cannot trigger ${taskId} with a one-time use token as it has already been used.`
);
} else if (
Array.isArray(target) &&
target.length == 2 &&
typeof target[0] === "string" &&
typeof target[1] === "string" &&
target[0] == "runtimeEnvironmentId" &&
target[1] == "name" &&
error.message.includes("prisma.taskQueue.create")
) {
throw new Error(
`Failed to trigger ${taskId} as the queue could not be created do to a unique constraint error, please try again.`
);
} else {
throw new ServiceValidationError(
`Cannot trigger ${taskId} as it has already been triggered with the same idempotency key.`
Expand Down
4 changes: 2 additions & 2 deletions packages/core/src/v3/schemas/schemas.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,8 +134,8 @@ export const QueueOptions = z.object({
name: z.string().optional(),
/** An optional property that specifies the maximum number of concurrent run executions.
*
* If this property is omitted, the task can potentially use up the full concurrency of an environment. */
concurrencyLimit: z.number().int().min(0).max(1000).optional(),
* If this property is omitted, the task can potentially use up the full concurrency of an environment */
concurrencyLimit: z.number().int().min(0).max(1000).optional().nullable(),
});

export type QueueOptions = z.infer<typeof QueueOptions>;
Expand Down
58 changes: 48 additions & 10 deletions references/v3-catalog/src/trigger/queues.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,16 +11,53 @@ export const queuesController = task({
length?: number;
waitSeconds?: number;
}) => {
await queuesTest.batchTriggerAndWait(
Array.from({ length }, (_, i) => ({
payload: { waitSeconds },
options: {
await Promise.all([
queuesTest.trigger(
{ waitSeconds },
{
queue: {
name: `queue-${i % numberOfQueues}`,
name: "controller-3",
concurrencyLimit: 9,
},
},
}))
);
}
),
queuesTest.trigger(
{ waitSeconds },
{
queue: {
name: "controller-3",
concurrencyLimit: 9,
},
}
),
queuesTest.trigger(
{ waitSeconds },
{
queue: {
name: "controller-3",
concurrencyLimit: 9,
},
}
),
queuesTest.trigger(
{ waitSeconds },
{
queue: {
name: "controller-3",
concurrencyLimit: 9,
},
}
),
queuesTest.trigger(
{ waitSeconds },
{
queue: {
name: "controller-3",
concurrencyLimit: 9,
},
}
),
]);
},
});

Expand All @@ -34,10 +71,11 @@ export const queuesTest = task({
export const namedQueueTask = task({
id: "queues/named-queue",
queue: {
name: "named-queue",
name: "controller",
concurrencyLimit: 9,
},
run: async () => {
logger.info("named-queue");
logger.info("named-queue 2");
},
});

Expand Down
Loading