feat: add versioned queues and authorized multi-queue dequeue

Adds authenticated, weighted multi-queue consumption with separate
dispatch classes and execution phases. Existing consumers retain their
legacy queue selection; requested subscriptions are validated and
authorized before dequeue.

Mono-RevId: 1b2b009af1ec70bc9bc23eb7d1a9a78cb60862e7

Author: carderne

apps/webapp/app/env.server.ts

@@ -6,6 +6,7 @@ import { isValidDatabaseUrl } from "./utils/db";
 import { parseRunOpsShards, validateShardListAgainstNewUrl } from "~/v3/runOpsShards.server";
 import { isValidRegex } from "./utils/regex";
 import { isValidDuration } from "./services/realtime/duration.server";
+import { WorkerQueueSubscriptionPolicyEnv } from "./runEngine/concerns/workerQueueSubscriptions.server";
 
 // `z.string()` constrained to a `parseDuration`-parseable string (e.g.
 // `7d`, `1h`). Validated at boot so a typo'd duration fails fast.
@@ -1226,6 +1227,8 @@ const EnvironmentSchema = z
     RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS: z.coerce.number().int().default(200),
     RUN_ENGINE_DEQUEUE_BLOCKING_TIMEOUT_SECONDS: z.coerce.number().int().default(10),
     RUN_ENGINE_DEQUEUE_DISABLED_WORKER_QUEUES: z.string().optional(),
+    // JSON map of worker-group IDs to allowed v2 subscriptions. Unlisted groups are denied.
+    RUN_ENGINE_WORKER_QUEUE_SUBSCRIPTIONS: WorkerQueueSubscriptionPolicyEnv,
     RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS: z.coerce.number().int().default(1000),
     // Off by default. Enable on a single service (e.g. the engine worker) so only one
     // instance reports worker queue length, rather than every replica.

apps/webapp/app/routes/engine.v1.worker-actions.dequeue.e2e.test.ts

@@ -0,0 +1,100 @@
+import { startTestServer, type TestServer } from "@internal/testcontainers/webapp";
+import type { WorkerQueueSubscription } from "@trigger.dev/core/v3/workers";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import {
+  seedEngineFixtures,
+  workerHeaders,
+  type EngineFixtures,
+} from "../../test/bench/lib/engineFixtures";
+
+const WORKER_GROUP_ID = "worker-group-legacy-auth";
+const LEGACY_MASTER_QUEUE = "my workers";
+const subscription: WorkerQueueSubscription = {
+  class: "ondemand",
+  phase: "fresh",
+  compat: "container",
+  channel: "stable",
+};
+
+let server: TestServer;
+let fixtures: EngineFixtures;
+
+beforeAll(async () => {
+  server = await startTestServer({
+    extraEnv: {
+      RUN_ENGINE_WORKER_QUEUE_SUBSCRIPTIONS: JSON.stringify({
+        [WORKER_GROUP_ID]: [subscription],
+      }),
+    },
+    overrideEnv: { RUN_ENGINE_WORKER_ENABLED: "1" },
+  });
+  fixtures = await seedEngineFixtures(server.prisma, {
+    taskCount: 1,
+    workerGroupId: WORKER_GROUP_ID,
+    masterQueue: LEGACY_MASTER_QUEUE,
+    enableFastPath: true,
+  });
+}, 180_000);
+
+afterAll(async () => {
+  await server?.stop();
+}, 120_000);
+
+function workerAction(path: string, body: unknown) {
+  return server.webapp.fetch(path, {
+    method: "POST",
+    headers: workerHeaders(fixtures, "legacy-worker-instance"),
+    body: JSON.stringify(body),
+  });
+}
+
+describe("dequeue request validation", () => {
+  it("authenticates legacy workers and rejects invalid requests without popping their queue", async () => {
+    const heartbeat = await workerAction("/engine/v1/worker-actions/heartbeat", {
+      cpu: { used: 0, available: 1 },
+      memory: { used: 0, available: 1 },
+      tasks: [],
+    });
+    expect(heartbeat.status).toBe(200);
+
+    const trigger = await server.webapp.fetch(
+      `/api/v1/tasks/${fixtures.taskIdentifiers[0]!}/trigger`,
+      {
+        method: "POST",
+        headers: {
+          Authorization: `Bearer ${fixtures.environmentApiKey}`,
+          "content-type": "application/json",
+        },
+        body: JSON.stringify({ payload: { message: "dequeue route validation" } }),
+      }
+    );
+    expect(trigger.ok).toBe(true);
+
+    const invalidRequests: Array<[body: unknown, status: number]> = [
+      [{ queueClass: "restore" }, 400],
+      [{ maxRunCount: "invalid" }, 400],
+      [{ maxResources: { cpu: "invalid", memory: 1 } }, 400],
+      [{ subscriptions: [] }, 422],
+      [{ subscriptions: [{ ...subscription, weight: 2 }] }, 422],
+      [{ queueClass: "default", subscriptions: [subscription] }, 422],
+      [{ maxRunCount: "invalid", subscriptions: [] }, 400],
+      [{ queueClass: "invalid", subscriptions: [subscription] }, 400],
+    ];
+    for (const [body, status] of invalidRequests) {
+      expect((await workerAction("/engine/v1/worker-actions/dequeue", body)).status).toBe(status);
+    }
+
+    const invalidV2Region = await workerAction("/engine/v1/worker-actions/dequeue", {
+      subscriptions: [subscription],
+    });
+    expect(invalidV2Region.status).toBe(422);
+
+    const legacyDequeue = await workerAction("/engine/v1/worker-actions/dequeue", {});
+    expect(legacyDequeue.status).toBe(200);
+    expect(await legacyDequeue.json()).toHaveLength(1);
+
+    const emptyDequeue = await workerAction("/engine/v1/worker-actions/dequeue", {});
+    expect(emptyDequeue.status).toBe(200);
+    expect(await emptyDequeue.json()).toEqual([]);
+  }, 180_000);
+});

apps/webapp/app/routes/engine.v1.worker-actions.dequeue.ts

@@ -8,12 +8,20 @@ import { createActionWorkerApiRoute } from "~/services/routeBuilders/apiBuilder.
 export const action = createActionWorkerApiRoute(
   {
     body: z.compile(WorkerApiDequeueRequestBody),
+    bodyValidationErrorStatus: (error) =>
+      error.issues.every((issue) => issue.path[0] === "subscriptions") ? 422 : 400,
   },
   async ({
     authenticatedWorker,
     runnerId,
     body,
   }): Promise<TypedResponse<WorkerApiDequeueResponseBody>> => {
-    return json(await authenticatedWorker.dequeue({ runnerId, queueClass: body.queueClass }));
+    return json(
+      await authenticatedWorker.dequeue({
+        runnerId,
+        queueClass: body.queueClass,
+        subscriptions: body.subscriptions,
+      })
+    );
   }
 );

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

@@ -17,8 +17,11 @@ const disabledWorkerQueues = parseDisabledWorkerQueues(
   env.RUN_ENGINE_DEQUEUE_DISABLED_WORKER_QUEUES
 );
 
-export function isWorkerQueueDequeueDisabled(workerQueue: string): boolean {
-  return matchesDisabledWorkerQueue(workerQueue, disabledWorkerQueues);
+export function isWorkerQueueDequeueDisabled(
+  workerQueue: string,
+  version: "legacy" | "v2"
+): boolean {
+  return matchesDisabledWorkerQueue(workerQueue, disabledWorkerQueues, version);
 }
 
 export function recordBlockedDequeue(workerQueue: string): void {

apps/webapp/app/runEngine/concerns/workerQueueDequeue.server.test.ts

@@ -0,0 +1,111 @@
+import { RunEngine } from "@internal/run-engine";
+import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "@internal/run-engine/tests";
+import { assertNonNullable, containerTest } from "@internal/testcontainers";
+import { trace } from "@internal/tracing";
+import { generateFriendlyId } from "@trigger.dev/core/v3/isomorphic";
+import type { WorkerQueueSubscription } from "@trigger.dev/core/v3/workers";
+import { expect } from "vitest";
+import { dequeueWorkerQueues } from "~/runEngine/concerns/workerQueueDequeue.server";
+import { createWorkerQueueConsumer } from "~/runEngine/concerns/workerQueueSubscriptions.server";
+
+containerTest(
+  "authorizes the entire subscription set before pop and retains legacy dequeue",
+  async ({ prisma, redisOptions }) => {
+    const engine = new RunEngine({
+      prisma,
+      worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
+      queue: { redis: redisOptions, masterQueueConsumersDisabled: true },
+      runLock: { redis: redisOptions },
+      machines: {
+        defaultMachine: "small-1x",
+        machines: {
+          "small-1x": { name: "small-1x", cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
+        },
+        baseCostInCents: 0.0005,
+      },
+      tracer: trace.getTracer("multi-queue-test"),
+    });
+
+    try {
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      const taskIdentifier = "test-task";
+      await setupBackgroundWorker(engine, environment, taskIdentifier);
+      const stable: WorkerQueueSubscription = {
+        class: "ondemand",
+        phase: "fresh",
+        compat: "container",
+        channel: "stable",
+      };
+      const canary: WorkerQueueSubscription = { ...stable, channel: "canary" };
+      const scheduled: WorkerQueueSubscription = { ...stable, class: "scheduled" };
+      const worker = createWorkerQueueConsumer({
+        workerGroupId: environment.project.defaultWorkerGroupId!,
+        workerInstanceId: "test-instance",
+        masterQueue: "legacy-region",
+        region: "us-east-1",
+        workloadType: "CONTAINER" as const,
+        allowedSubscriptions: [stable, scheduled],
+      });
+
+      const lanes = [
+        "us-east-1:v2:ondemand:fresh:container:stable",
+        "us-east-1:v2:scheduled:fresh:container:stable",
+        "legacy-region",
+        "legacy-region:scheduled",
+      ];
+      const runs = [];
+      for (const [index, workerQueue] of lanes.entries()) {
+        runs.push(
+          await engine.trigger(
+            {
+              number: index + 1,
+              friendlyId: generateFriendlyId("run"),
+              environment,
+              taskIdentifier,
+              payload: "{}",
+              payloadType: "application/json",
+              context: {},
+              traceContext: {},
+              traceId: "test-trace",
+              spanId: "test-span",
+              workerQueue,
+              enableFastPath: true,
+              queue: `task/${taskIdentifier}`,
+              isTest: false,
+              tags: [],
+            },
+            prisma
+          )
+        );
+      }
+
+      expect(() =>
+        dequeueWorkerQueues({ engine, worker, subscriptions: [stable, canary] })
+      ).toThrow("not authorized");
+      const unchanged = await engine.getRunExecutionData({ runId: runs[0]!.id });
+      expect(unchanged?.snapshot.executionStatus).toBe("QUEUED");
+
+      const deliveredRunIds = new Set<string>();
+      for (const remaining of [1, 0]) {
+        const delivery = await dequeueWorkerQueues({
+          engine,
+          worker,
+          subscriptions: [stable, scheduled],
+        });
+        expect(delivery).toHaveLength(1);
+        assertNonNullable(delivery[0]);
+        deliveredRunIds.add(delivery[0].run.id);
+        expect(delivery[0].snapshot.executionStatus).toBe("PENDING_EXECUTING");
+        expect(delivery[0].workerQueueLength).toBe(remaining);
+      }
+      expect(deliveredRunIds).toEqual(new Set([runs[0]!.id, runs[1]!.id]));
+
+      expect((await dequeueWorkerQueues({ engine, worker }))[0]?.run.id).toBe(runs[2]!.id);
+      expect(
+        (await dequeueWorkerQueues({ engine, worker, queueClass: "scheduled" }))[0]?.run.id
+      ).toBe(runs[3]!.id);
+    } finally {
+      await engine.quit();
+    }
+  }
+);

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

@@ -0,0 +1,47 @@
+import type { RunEngine } from "@internal/run-engine";
+import type {
+  WeightedWorkerQueueSubscription,
+  WorkerQueueClass,
+} from "@trigger.dev/core/v3/workers";
+import { isWorkerQueueDequeueDisabled, recordBlockedDequeue } from "./dequeueGate.server";
+import { workerQueueForClass } from "./workerQueueSplit.server";
+import {
+  resolveWorkerQueueSubscriptions,
+  type WorkerQueueConsumer,
+} from "./workerQueueSubscriptions.server";
+
+export function dequeueWorkerQueues({
+  engine,
+  worker,
+  runnerId,
+  queueClass,
+  subscriptions,
+}: {
+  engine: RunEngine;
+  worker: WorkerQueueConsumer;
+  runnerId?: string;
+  queueClass?: WorkerQueueClass;
+  subscriptions?: WeightedWorkerQueueSubscription[];
+}) {
+  const workerQueues = subscriptions
+    ? resolveWorkerQueueSubscriptions(worker, subscriptions)
+    : [{ queue: workerQueueForClass(worker.masterQueue, queueClass), weight: 1 }];
+
+  const enabledQueues = workerQueues.filter(({ queue }) => {
+    if (isWorkerQueueDequeueDisabled(queue, subscriptions ? "v2" : "legacy")) {
+      recordBlockedDequeue(queue);
+      return false;
+    }
+    return true;
+  });
+  if (enabledQueues.length === 0) {
+    return Promise.resolve([]);
+  }
+
+  return engine.dequeueFromWorkerQueues({
+    consumerId: worker.workerInstanceId,
+    workerQueues: enabledQueues,
+    workerId: worker.workerInstanceId,
+    runnerId,
+  });
+}