feat(sdk,core,webapp,run-engine): runtime override for the combined queue concurrency limit

apps/webapp/app/presenters/v3/QueueListPresenter.server.ts

@@ -70,6 +70,7 @@ function buildQueueListWhere(
 
   return {
     runtimeEnvironmentId: environmentId,
+    role: "QUEUE" as const,
     version: "V2",
     name: trimmedQuery
       ? {

apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts

@@ -31,6 +31,7 @@ export async function getQueue(
         where: {
           friendlyId: queue,
           runtimeEnvironmentId: environment.id,
+          role: "QUEUE",
         },
       })
     );
@@ -44,6 +45,7 @@ export async function getQueue(
       where: {
         name: queueName,
         runtimeEnvironmentId: environment.id,
+        role: "QUEUE",
       },
     })
   );

apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts

@@ -0,0 +1,98 @@
+import { json } from "@remix-run/server-runtime";
+import { type RetrieveQueueParam, RetrieveQueueType } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { toQueueItem } from "~/presenters/v3/QueueRetrievePresenter.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { concurrencySystem } from "~/v3/services/concurrencySystemInstance.server";
+
+const BodySchema = z.object({
+  type: RetrieveQueueType.default("id"),
+  concurrencyLimit: z.number().int().min(0).max(100000),
+});
+
+const route = createActionApiRoute(
+  {
+    body: BodySchema,
+    params: z.object({
+      queueParam: z.string().transform((val) => val.replace(/%2F/g, "/")),
+    }),
+    authorization: {
+      action: "write",
+      resource: () => ({ type: "queues" }),
+    },
+  },
+  async ({ params, body, authentication }) => {
+    const input: RetrieveQueueParam =
+      body.type === "id"
+        ? params.queueParam
+        : {
+            type: body.type,
+            name: decodeURIComponent(params.queueParam).replace(/%2F/g, "/"),
+          };
+
+    return concurrencySystem.queues
+      .overrideTotalConcurrencyLimit(authentication.environment, input, body.concurrencyLimit)
+      .match(
+        (queue) => {
+          return json(
+            toQueueItem({
+              friendlyId: queue.friendlyId,
+              name: queue.name,
+              type: queue.type,
+              running: queue.running,
+              queued: queue.queued,
+              concurrencyLimit: queue.concurrencyLimit,
+              concurrencyLimitBase: queue.concurrencyLimitBase,
+              concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
+              concurrencyLimitOverriddenBy: null,
+              paused: queue.paused,
+            }),
+            { status: 200 }
+          );
+        },
+        (error) => {
+          switch (error.type) {
+            case "queue_not_found": {
+              return json({ error: "Queue not found" }, { status: 404 });
+            }
+            case "invalid_override":
+            case "concurrency_limit_exceeds_maximum": {
+              return json({ error: error.message }, { status: 400 });
+            }
+            case "queue_update_failed": {
+              return json(
+                { error: "Failed to update queue total concurrency limit" },
+                { status: 500 }
+              );
+            }
+            case "sync_queue_concurrency_to_engine_failed": {
+              return json({ error: "Failed to sync the total concurrency limit" }, { status: 500 });
+            }
+            case "get_queue_stats_failed": {
+              return json({ error: "Failed to read queue stats" }, { status: 500 });
+            }
+            case "other": {
+              return json(
+                { error: "Failed to update queue total concurrency limit" },
+                {
+                  status: 500,
+                }
+              );
+            }
+            default: {
+              return json(
+                { error: "Failed to update queue total concurrency limit" },
+                {
+                  status: 500,
+                }
+              );
+            }
+          }
+        }
+      );
+  }
+);
+
+export const action = route.action;
+/** The builder's loader answers non-POST methods with a 405. */
+export const loader = route.loader;

apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts

@@ -0,0 +1,99 @@
+import { json } from "@remix-run/server-runtime";
+import { type RetrieveQueueParam, RetrieveQueueType } from "@trigger.dev/core/v3";
+import { z } from "zod";
+import { toQueueItem } from "~/presenters/v3/QueueRetrievePresenter.server";
+import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { concurrencySystem } from "~/v3/services/concurrencySystemInstance.server";
+
+const BodySchema = z.object({
+  type: RetrieveQueueType.default("id"),
+});
+
+const route = createActionApiRoute(
+  {
+    body: BodySchema,
+    params: z.object({
+      queueParam: z.string().transform((val) => val.replace(/%2F/g, "/")),
+    }),
+    authorization: {
+      action: "write",
+      resource: () => ({ type: "queues" }),
+    },
+  },
+  async ({ params, body, authentication }) => {
+    const input: RetrieveQueueParam =
+      body.type === "id"
+        ? params.queueParam
+        : {
+            type: body.type,
+            name: decodeURIComponent(params.queueParam).replace(/%2F/g, "/"),
+          };
+
+    return concurrencySystem.queues
+      .resetTotalConcurrencyLimit(authentication.environment, input)
+      .match(
+        (queue) => {
+          return json(
+            toQueueItem({
+              friendlyId: queue.friendlyId,
+              name: queue.name,
+              type: queue.type,
+              running: queue.running,
+              queued: queue.queued,
+              concurrencyLimit: queue.concurrencyLimit,
+              concurrencyLimitBase: queue.concurrencyLimitBase,
+              concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt,
+              concurrencyLimitOverriddenBy: null,
+              paused: queue.paused,
+            }),
+            { status: 200 }
+          );
+        },
+        (error) => {
+          switch (error.type) {
+            case "queue_not_found": {
+              return json({ error: "Queue not found" }, { status: 404 });
+            }
+            case "queue_not_overridden": {
+              return json(
+                { error: "The queue total concurrency limit is not overridden" },
+                { status: 400 }
+              );
+            }
+            case "queue_update_failed": {
+              return json(
+                { error: "Failed to reset the queue total concurrency limit" },
+                { status: 500 }
+              );
+            }
+            case "sync_queue_concurrency_to_engine_failed": {
+              return json({ error: "Failed to sync the total concurrency limit" }, { status: 500 });
+            }
+            case "get_queue_stats_failed": {
+              return json({ error: "Failed to read queue stats" }, { status: 500 });
+            }
+            case "other": {
+              return json(
+                { error: "Failed to reset the queue total concurrency limit" },
+                {
+                  status: 500,
+                }
+              );
+            }
+            default: {
+              return json(
+                { error: "Failed to reset the queue total concurrency limit" },
+                {
+                  status: 500,
+                }
+              );
+            }
+          }
+        }
+      );
+  }
+);
+
+export const action = route.action;
+/** The builder's loader answers non-POST methods with a 405. */
+export const loader = route.loader;

apps/webapp/app/runEngine/concerns/queues.server.ts

@@ -220,13 +220,71 @@ export class DefaultQueueManager implements QueueManager {
       queueName = sanitizedQueueName;
     }
 
-    const requestedGates = request.body.options?.gates ?? taskGates ?? undefined;
-    const gates = requestedGates
-      ?.flatMap((gate) => {
-        const sanitized = sanitizeQueueName(gate.queue);
-        return sanitized ? [{ queue: sanitized, concurrencyKey: gate.concurrencyKey }] : [];
-      })
-      .slice(0, 2);
+    const triggerLimits = request.body.options?.concurrency;
+
+    for (const name of triggerLimits ?? []) {
+      if (!/^[a-zA-Z0-9_-]{1,122}$/.test(name)) {
+        throw new ServiceValidationError(
+          `Invalid concurrency limit name "${name}": names are 1-122 characters using only letters, numbers, underscores and hyphens.`
+        );
+      }
+    }
+
+    /**
+     * Trigger-time names replace the task's declared NAMED limits only. The task's
+     * inline limit rides in its stored gates as an anonymous "limit/task/" gate and
+     * always applies, so it is carried over into the replacement (an empty array
+     * clears the named limits but keeps the inline one).
+     */
+    const inlineTaskGates = (taskGates ?? []).filter((gate) =>
+      gate.queue.startsWith("limit/task/")
+    );
+    const concurrencyGates = triggerLimits
+      ? [
+          ...inlineTaskGates,
+          ...triggerLimits.map((name): { queue: string; concurrencyKey?: string } => ({
+            queue: `limit/${name}`,
+          })),
+        ]
+      : undefined;
+
+    /**
+     * The raw gates option replaces stored gates the same way concurrency does, so
+     * it also carries the inline gate over; a replay resending the stored gates
+     * collapses back to the original set through the dedupe below.
+     */
+    const rawGates = request.body.options?.gates;
+    const requestedGates =
+      concurrencyGates ??
+      (rawGates ? [...inlineTaskGates, ...rawGates] : undefined) ??
+      taskGates ??
+      undefined;
+
+    const seenGates = new Set<string>();
+    const gates = requestedGates?.flatMap((gate) => {
+      const sanitized = sanitizeQueueName(gate.queue);
+      if (!sanitized) {
+        return [];
+      }
+      const dedupeKey = `${sanitized}${gate.concurrencyKey ?? ""}`;
+      if (seenGates.has(dedupeKey)) {
+        return [];
+      }
+      seenGates.add(dedupeKey);
+      return [{ queue: sanitized, concurrencyKey: gate.concurrencyKey }];
+    });
+
+    /**
+     * Unreachable through the public schemas (three requested gates plus one inline
+     * gate is the ceiling), kept as a backstop so an overflowing set can never be
+     * silently truncated downstream. Replays of three-gate runs against a task that
+     * later gained an inline limit resolve to four and stay valid.
+     */
+    if (gates && gates.length > 4) {
+      throw new ServiceValidationError(
+        `A run can hold at most four gates; this request resolves to ${gates.length}.`
+      );
+    }
 
     return {
       queueName,