fix(webapp): persist concurrency overrides on deploy (#2636)
* fix(webapp): display correct concurrency override base value * fix(webapp): persist concurrency overrides on deploy * fix(webapp): use correct override base value type * fix(webapp): override input is bounded by env concurrency
This commit is contained in:
+6
-9
@@ -12,6 +12,7 @@ import { DialogClose } from "@radix-ui/react-dialog";
|
||||
import { Form, useNavigation, useSearchParams, type MetaFunction } from "@remix-run/react";
|
||||
import { type ActionFunctionArgs, type LoaderFunctionArgs } from "@remix-run/server-runtime";
|
||||
import type { RuntimeEnvironmentType } from "@trigger.dev/database";
|
||||
import type { QueueItem } from "@trigger.dev/core/v3/schemas";
|
||||
import { useEffect, useState } from "react";
|
||||
import { typedjson, useTypedLoaderData } from "remix-typedjson";
|
||||
import { z } from "zod";
|
||||
@@ -926,12 +927,7 @@ function QueueOverrideConcurrencyButton({
|
||||
queue,
|
||||
environmentConcurrencyLimit,
|
||||
}: {
|
||||
queue: {
|
||||
id: string;
|
||||
name: string;
|
||||
concurrencyLimit: number | null;
|
||||
concurrency?: { overriddenAt: Date | null };
|
||||
};
|
||||
queue: QueueItem;
|
||||
environmentConcurrencyLimit: number;
|
||||
}) {
|
||||
const navigation = useNavigation();
|
||||
@@ -970,10 +966,10 @@ function QueueOverrideConcurrencyButton({
|
||||
{isOverridden ? (
|
||||
<Paragraph>
|
||||
This queue's concurrency limit is currently overridden to {currentLimit}.
|
||||
{queue.concurrencyLimit !== null &&
|
||||
` The original limit set in code was ${queue.concurrencyLimit}.`}{" "}
|
||||
{typeof queue.concurrency?.base === "number" &&
|
||||
` The original limit set in code was ${queue.concurrency.base}.`}{" "}
|
||||
You can update the override or remove it to restore the{" "}
|
||||
{queue.concurrencyLimit !== null
|
||||
{typeof queue.concurrency?.base === "number"
|
||||
? "limit set in code"
|
||||
: "environment concurrency limit"}
|
||||
.
|
||||
@@ -995,6 +991,7 @@ function QueueOverrideConcurrencyButton({
|
||||
name="concurrencyLimit"
|
||||
id="concurrencyLimit"
|
||||
min="0"
|
||||
max={environmentConcurrencyLimit}
|
||||
value={concurrencyLimit}
|
||||
onChange={(e) => setConcurrencyLimit(e.target.value)}
|
||||
placeholder={currentLimit.toString()}
|
||||
|
||||
@@ -359,7 +359,7 @@ async function createWorkerQueue(
|
||||
) {
|
||||
let queueName = sanitizeQueueName(queue.name);
|
||||
|
||||
const concurrencyLimit =
|
||||
const baseConcurrencyLimit =
|
||||
typeof queue.concurrencyLimit === "number"
|
||||
? Math.max(
|
||||
Math.min(
|
||||
@@ -373,24 +373,26 @@ async function createWorkerQueue(
|
||||
|
||||
const taskQueue = await upsertWorkerQueueRecord(
|
||||
queueName,
|
||||
concurrencyLimit ?? null,
|
||||
baseConcurrencyLimit ?? null,
|
||||
orderableName,
|
||||
queueType,
|
||||
worker,
|
||||
prisma
|
||||
);
|
||||
|
||||
const newConcurrencyLimit = taskQueue.concurrencyLimit;
|
||||
|
||||
if (!taskQueue.paused) {
|
||||
if (typeof concurrencyLimit === "number") {
|
||||
if (typeof newConcurrencyLimit === "number") {
|
||||
logger.debug("createWorkerQueue: updating concurrency limit", {
|
||||
workerId: worker.id,
|
||||
taskQueue,
|
||||
orgId: environment.organizationId,
|
||||
projectId: environment.projectId,
|
||||
environmentId: environment.id,
|
||||
concurrencyLimit,
|
||||
concurrencyLimit: newConcurrencyLimit,
|
||||
});
|
||||
await updateQueueConcurrencyLimits(environment, taskQueue.name, concurrencyLimit);
|
||||
await updateQueueConcurrencyLimits(environment, taskQueue.name, newConcurrencyLimit);
|
||||
} else {
|
||||
logger.debug("createWorkerQueue: removing concurrency limit", {
|
||||
workerId: worker.id,
|
||||
@@ -398,7 +400,7 @@ async function createWorkerQueue(
|
||||
orgId: environment.organizationId,
|
||||
projectId: environment.projectId,
|
||||
environmentId: environment.id,
|
||||
concurrencyLimit,
|
||||
concurrencyLimit: newConcurrencyLimit,
|
||||
});
|
||||
await removeQueueConcurrencyLimits(environment, taskQueue.name);
|
||||
}
|
||||
@@ -455,6 +457,8 @@ async function upsertWorkerQueueRecord(
|
||||
},
|
||||
});
|
||||
} else {
|
||||
const hasOverride = taskQueue.concurrencyLimitOverriddenAt !== null;
|
||||
|
||||
taskQueue = await prisma.taskQueue.update({
|
||||
where: {
|
||||
id: taskQueue.id,
|
||||
@@ -463,7 +467,9 @@ async function upsertWorkerQueueRecord(
|
||||
workers: { connect: { id: worker.id } },
|
||||
version: "V2",
|
||||
orderableName,
|
||||
concurrencyLimit,
|
||||
// If overridden, keep current limit and update base; otherwise update limit normally
|
||||
concurrencyLimit: hasOverride ? undefined : concurrencyLimit,
|
||||
concurrencyLimitBase: hasOverride ? concurrencyLimit : undefined,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user