perf(run-engine,webapp): narrow the control-plane worker-version read to the columns dequeue uses (#4619)

## Summary The worker-version resolve path fetched every column of every `BackgroundWorkerTask` for a worker (`include: { tasks: true }`), plus full `WorkerDeployment` and `TaskQueue` rows, just to match one task at dequeue. That pulls large JSON columns none of this path reads (task `payloadSchema`/`config`/`queueConfig`/`description`, deployment `externalBuildData`/`buildServerMetadata`/`errorData`/`git`, queue `rateLimit`), so each resolve transfers and deserializes far more than it uses. ## Fix Replace the includes with explicit `select`s of only the columns dequeue reads, in both the passthrough resolver and the app resolver: - task: `id`, `slug`, `machineConfig`, `retryConfig`, `maxDurationInSeconds` - deployment: `id`, `friendlyId`, `imageReference`, `imagePlatform` - queue: `id`, `name` (the queue matcher keys on both) The shared `ResolvedWorkerVersion` element types narrow to match (mirrored in the cache), which also shrinks each cached worker-version entry. ## Impact The `tasks` read fetches every task of a worker to match one, so its cost scales with task count and payload-schema size. For a worker with ~70 registered tasks, dropping the unread columns cuts the per-query transfer roughly: | Task shape | Before | After | Reduction | |---|---|---|---| | Light (no payload schema, small config) | ~28 KB | ~14 KB | ~54% | | Typical (mixed schemas / config) | ~62 KB | ~14 KB | ~77% | | Schema-heavy (large `payloadSchema`) | ~200 KB | ~14 KB | ~93% | The `after` size is roughly fixed because the kept columns are small; the win grows with how heavy the dropped JSON is. Narrowing `deployment` (four JSON columns off a single row) and `queues` saves further on top. No behavior change: pure read-shape narrowing, no flag and no schema change, so rollback is a plain revert. Verified with a red/green run-engine test that asserts the resolved task, deployment, and queue carry only the used columns, plus the queue feature-matrix runs (batch, retry-policy, machine-preset, plain trigger) that exercise the kept columns.
2026-08-14 15:11:30 +01:00
parent 1240d91e43
commit 8dc8e1b58b
6 changed files with 315 additions and 55 deletions
@@ -1,11 +1,4 @@
-import type {
-  BackgroundWorker,
-  BackgroundWorkerTask,
-  Prisma,
-  RuntimeEnvironmentType,
-  TaskQueue,
-  WorkerDeployment,
-} from "@trigger.dev/database";
+import type { BackgroundWorker, Prisma, RuntimeEnvironmentType } from "@trigger.dev/database";
 import { BoundedTtlCache } from "~/services/realtime/boundedTtlCache";
 import type { AuthenticatedEnvironment } from "@trigger.dev/core/v3/auth/environment";

@@ -49,12 +42,67 @@ export type ResolvedEnv = {
  concurrencyLimitBurstFactor: Prisma.Decimal;
 };

+/**
+ * The BackgroundWorkerTask columns the dequeue resolve path reads. Mirrors run-engine's
+ * `ResolvedWorkerTask` exactly. The unread heavy JSON columns (`payloadSchema`, `config`,
+ * `queueConfig`, `description`) are dropped so this hot control-plane read stops shipping
+ * ~62KB/query (and each cached entry stays small); `machineConfig`/`retryConfig` are read
+ * at dequeue and stay.
+ */
+export type ResolvedWorkerTask = {
+  id: string;
+  slug: string;
+  machineConfig: Prisma.JsonValue | null;
+  retryConfig: Prisma.JsonValue | null;
+  maxDurationInSeconds: number | null;
+};
+
+/** The `select` that yields a `ResolvedWorkerTask`. */
+export const resolvedWorkerTaskSelect = {
+  id: true,
+  slug: true,
+  machineConfig: true,
+  retryConfig: true,
+  maxDurationInSeconds: true,
+} satisfies Prisma.BackgroundWorkerTaskSelect;
+
+/** Mirrors run-engine's `ResolvedTaskQueue` exactly. `id` + `name` (the matcher keys on both). */
+export type ResolvedTaskQueue = {
+  id: string;
+  name: string;
+};
+
+/** The `select` that yields a `ResolvedTaskQueue`. */
+export const resolvedTaskQueueSelect = {
+  id: true,
+  name: true,
+} satisfies Prisma.TaskQueueSelect;
+
+/**
+ * Mirrors run-engine's `ResolvedWorkerDeployment` exactly. Drops the unread heavy JSON columns
+ * (`externalBuildData`, `buildServerMetadata`, `errorData`, `git`) from this single-row read.
+ */
+export type ResolvedWorkerDeployment = {
+  id: string;
+  friendlyId: string;
+  imageReference: string | null;
+  imagePlatform: string;
+};
+
+/** The `select` that yields a `ResolvedWorkerDeployment`. */
+export const resolvedWorkerDeploymentSelect = {
+  id: true,
+  friendlyId: true,
+  imageReference: true,
+  imagePlatform: true,
+} satisfies Prisma.WorkerDeploymentSelect;
+
 /** Mirrors `WorkerDeploymentWithWorkerTasks` in `dequeueSystem.ts` exactly. */
 export type ResolvedWorkerVersion = {
  worker: BackgroundWorker;
-  tasks: BackgroundWorkerTask[];
-  queues: TaskQueue[];
-  deployment: WorkerDeployment | null;
+  tasks: ResolvedWorkerTask[];
+  queues: ResolvedTaskQueue[];
+  deployment: ResolvedWorkerDeployment | null;
 };

 // The canonical authenticated-environment shape (slug/type/project/organization/orgMember/…)
@@ -10,6 +10,9 @@ import {
  ControlPlaneCache,
  DEFAULT_CP_CACHE_MAX_ENTRIES,
  DEFAULT_CP_CACHE_TTL_MS,
+  resolvedTaskQueueSelect,
+  resolvedWorkerDeploymentSelect,
+  resolvedWorkerTaskSelect,
  type ResolvedAuthenticatedEnv,
  type ResolvedEnv,
  type ResolvedWorkerVersion,
@@ -389,7 +392,11 @@ export class ControlPlaneResolver {
    if (backgroundWorkerId) {
      const worker = await client.backgroundWorker.findFirst({
        where: { id: backgroundWorkerId },
-        include: { deployment: true, tasks: true, queues: true },
+        include: {
+          deployment: { select: resolvedWorkerDeploymentSelect },
+          tasks: { select: resolvedWorkerTaskSelect },
+          queues: { select: resolvedTaskQueueSelect },
+        },
      });

      if (!worker) {
@@ -411,7 +418,16 @@ export class ControlPlaneResolver {
      where: { environmentId, label: CURRENT_DEPLOYMENT_LABEL },
      include: {
        deployment: {
-          include: { worker: { include: { tasks: true, queues: true } } },
+          select: {
+            ...resolvedWorkerDeploymentSelect,
+            type: true,
+            worker: {
+              include: {
+                tasks: { select: resolvedWorkerTaskSelect },
+                queues: { select: resolvedTaskQueueSelect },
+              },
+            },
+          },
        },
      },
    });
@@ -421,11 +437,17 @@ export class ControlPlaneResolver {
    }

    if (type === undefined || promotion.deployment.type === "MANAGED") {
+      const { worker } = promotion.deployment;
      return {
-        worker: promotion.deployment.worker,
-        tasks: promotion.deployment.worker.tasks,
-        queues: promotion.deployment.worker.queues,
-        deployment: promotion.deployment,
+        worker,
+        tasks: worker.tasks,
+        queues: worker.queues,
+        deployment: {
+          id: promotion.deployment.id,
+          friendlyId: promotion.deployment.friendlyId,
+          imageReference: promotion.deployment.imageReference,
+          imagePlatform: promotion.deployment.imagePlatform,
+        },
      };
    }

@@ -434,7 +456,15 @@ export class ControlPlaneResolver {
    const latestV2Deployment = await client.workerDeployment.findFirst({
      where: { environmentId, type: "MANAGED" },
      orderBy: [{ createdAt: "desc" }, { id: "desc" }],
-      include: { worker: { include: { tasks: true, queues: true } } },
+      select: {
+        ...resolvedWorkerDeploymentSelect,
+        worker: {
+          include: {
+            tasks: { select: resolvedWorkerTaskSelect },
+            queues: { select: resolvedTaskQueueSelect },
+          },
+        },
+      },
    });

    if (!latestV2Deployment?.worker) {
@@ -445,7 +475,12 @@ export class ControlPlaneResolver {
      worker: latestV2Deployment.worker,
      tasks: latestV2Deployment.worker.tasks,
      queues: latestV2Deployment.worker.queues,
-      deployment: latestV2Deployment,
+      deployment: {
+        id: latestV2Deployment.id,
+        friendlyId: latestV2Deployment.friendlyId,
+        imageReference: latestV2Deployment.imageReference,
+        imagePlatform: latestV2Deployment.imagePlatform,
+      },
    };
  }

@@ -455,7 +490,11 @@ export class ControlPlaneResolver {
  ): Promise<ResolvedWorkerVersion | null> {
    const worker = await client.backgroundWorker.findFirst({
      where: { id: workerId },
-      include: { deployment: true, tasks: true, queues: true },
+      include: {
+        deployment: { select: resolvedWorkerDeploymentSelect },
+        tasks: { select: resolvedWorkerTaskSelect },
+        queues: { select: resolvedTaskQueueSelect },
+      },
    });

    if (!worker) {
@@ -471,7 +510,10 @@ export class ControlPlaneResolver {
  ): Promise<ResolvedWorkerVersion | null> {
    const worker = await client.backgroundWorker.findFirst({
      where: { runtimeEnvironmentId: environmentId },
-      include: { tasks: true, queues: true },
+      include: {
+        tasks: { select: resolvedWorkerTaskSelect },
+        queues: { select: resolvedTaskQueueSelect },
+      },
      orderBy: [{ createdAt: "desc" }, { id: "desc" }],
    });

@@ -1,11 +1,8 @@
 import type {
  BackgroundWorker,
-  BackgroundWorkerTask,
  Prisma,
  PrismaClient,
  RuntimeEnvironmentType,
-  TaskQueue,
-  WorkerDeployment,
 } from "@trigger.dev/database";
 import { CURRENT_DEPLOYMENT_LABEL } from "@trigger.dev/core/v3/isomorphic";
 import type { AuthenticatedEnvironment } from "@trigger.dev/core/v3/auth/environment";
@@ -51,12 +48,70 @@ export type ResolvedEngineEnv = {
 */
 export type ResolvedAuthenticatedEnv = AuthenticatedEnvironment & { git: Prisma.JsonValue | null };

+/**
+ * The BackgroundWorkerTask columns the dequeue resolve path actually reads. The worker's
+ * whole task set (~73 rows) is fetched for one matched task, so pulling the unread heavy
+ * JSON columns (`payloadSchema`, `config`, `queueConfig`, `description`) shipped ~62KB/query
+ * on the hottest control-plane read. `machineConfig`/`retryConfig` are read at dequeue and stay.
+ */
+export type ResolvedWorkerTask = {
+  id: string;
+  slug: string;
+  machineConfig: Prisma.JsonValue | null;
+  retryConfig: Prisma.JsonValue | null;
+  maxDurationInSeconds: number | null;
+};
+
+/** The `select` that yields a `ResolvedWorkerTask`. */
+export const resolvedWorkerTaskSelect = {
+  id: true,
+  slug: true,
+  machineConfig: true,
+  retryConfig: true,
+  maxDurationInSeconds: true,
+} satisfies Prisma.BackgroundWorkerTaskSelect;
+
+/**
+ * The TaskQueue columns the dequeue resolve path uses: `id` and `name` (the matcher keys on
+ * `lockedQueueId`/`name`). Drops the unread `rateLimit` JSON and the concurrency/type scalars.
+ */
+export type ResolvedTaskQueue = {
+  id: string;
+  name: string;
+};
+
+/** The `select` that yields a `ResolvedTaskQueue`. */
+export const resolvedTaskQueueSelect = {
+  id: true,
+  name: true,
+} satisfies Prisma.TaskQueueSelect;
+
+/**
+ * The WorkerDeployment columns the dequeue resolve path reads: `id`, `friendlyId`,
+ * `imageReference`, `imagePlatform`. Drops the unread heavy JSON columns (`externalBuildData`,
+ * `buildServerMetadata`, `errorData`, `git`) that this single-row read otherwise ships.
+ */
+export type ResolvedWorkerDeployment = {
+  id: string;
+  friendlyId: string;
+  imageReference: string | null;
+  imagePlatform: string;
+};
+
+/** The `select` that yields a `ResolvedWorkerDeployment`. */
+export const resolvedWorkerDeploymentSelect = {
+  id: true,
+  friendlyId: true,
+  imageReference: true,
+  imagePlatform: true,
+} satisfies Prisma.WorkerDeploymentSelect;
+
 /** Identical to dequeue's `WorkerDeploymentWithWorkerTasks`. */
 export type ResolvedWorkerVersion = {
  worker: BackgroundWorker;
-  tasks: BackgroundWorkerTask[];
-  queues: TaskQueue[];
-  deployment: WorkerDeployment | null;
+  tasks: ResolvedWorkerTask[];
+  queues: ResolvedTaskQueue[];
+  deployment: ResolvedWorkerDeployment | null;
 };

 export interface ControlPlaneResolver {
@@ -207,9 +262,9 @@ export class PassthroughControlPlaneResolver implements ControlPlaneResolver {
        id: workerId,
      },
      include: {
-        deployment: true,
-        tasks: true,
-        queues: true,
+        deployment: { select: resolvedWorkerDeploymentSelect },
+        tasks: { select: resolvedWorkerTaskSelect },
+        queues: { select: resolvedTaskQueueSelect },
      },
    });

@@ -231,8 +286,8 @@ export class PassthroughControlPlaneResolver implements ControlPlaneResolver {
        runtimeEnvironmentId: environmentId,
      },
      include: {
-        tasks: true,
-        queues: true,
+        tasks: { select: resolvedWorkerTaskSelect },
+        queues: { select: resolvedTaskQueueSelect },
      },
      orderBy: [{ createdAt: "desc" }, { id: "desc" }],
    });
@@ -250,9 +305,9 @@ export class PassthroughControlPlaneResolver implements ControlPlaneResolver {
        id: workerId,
      },
      include: {
-        deployment: true,
-        tasks: true,
-        queues: true,
+        deployment: { select: resolvedWorkerDeploymentSelect },
+        tasks: { select: resolvedWorkerTaskSelect },
+        queues: { select: resolvedTaskQueueSelect },
      },
    });

@@ -278,11 +333,13 @@ export class PassthroughControlPlaneResolver implements ControlPlaneResolver {
      },
      include: {
        deployment: {
-          include: {
+          select: {
+            ...resolvedWorkerDeploymentSelect,
+            type: true,
            worker: {
              include: {
-                tasks: true,
-                queues: true,
+                tasks: { select: resolvedWorkerTaskSelect },
+                queues: { select: resolvedTaskQueueSelect },
              },
            },
          },
@@ -296,11 +353,17 @@ export class PassthroughControlPlaneResolver implements ControlPlaneResolver {

    if (promotion.deployment.type === "MANAGED") {
      // This is a run engine v2 deployment, so return it
+      const { worker } = promotion.deployment;
      return {
-        worker: promotion.deployment.worker,
-        tasks: promotion.deployment.worker.tasks,
-        queues: promotion.deployment.worker.queues,
-        deployment: promotion.deployment,
+        worker,
+        tasks: worker.tasks,
+        queues: worker.queues,
+        deployment: {
+          id: promotion.deployment.id,
+          friendlyId: promotion.deployment.friendlyId,
+          imageReference: promotion.deployment.imageReference,
+          imagePlatform: promotion.deployment.imagePlatform,
+        },
      };
    }

@@ -311,11 +374,12 @@ export class PassthroughControlPlaneResolver implements ControlPlaneResolver {
        type: "MANAGED",
      },
      orderBy: [{ createdAt: "desc" }, { id: "desc" }],
-      include: {
+      select: {
+        ...resolvedWorkerDeploymentSelect,
        worker: {
          include: {
-            tasks: true,
-            queues: true,
+            tasks: { select: resolvedWorkerTaskSelect },
+            queues: { select: resolvedTaskQueueSelect },
          },
        },
      },
@@ -329,7 +393,12 @@ export class PassthroughControlPlaneResolver implements ControlPlaneResolver {
      worker: latestV2Deployment.worker,
      tasks: latestV2Deployment.worker.tasks,
      queues: latestV2Deployment.worker.queues,
-      deployment: latestV2Deployment,
+      deployment: {
+        id: latestV2Deployment.id,
+        friendlyId: latestV2Deployment.friendlyId,
+        imageReference: latestV2Deployment.imageReference,
+        imagePlatform: latestV2Deployment.imagePlatform,
+      },
    };
  }
 }
@@ -6,14 +6,16 @@ import { generateInternalId, getMaxDuration, SnapshotId } from "@trigger.dev/cor
 import { placementTag } from "@trigger.dev/core/v3/serverOnly";
 import type {
  BackgroundWorker,
-  BackgroundWorkerTask,
  Prisma,
  PrismaClientOrTransaction,
  RuntimeEnvironmentType,
-  TaskQueue,
-  WorkerDeployment,
 } from "@trigger.dev/database";
 import type { BillingCache } from "../billingCache.js";
+import type {
+  ResolvedTaskQueue,
+  ResolvedWorkerDeployment,
+  ResolvedWorkerTask,
+} from "../controlPlaneResolver.js";

 import { sendNotificationToWorker } from "../eventBus.js";
 import { getMachinePreset } from "../machinePresets.js";
@@ -88,16 +90,16 @@ type RunWithBackgroundWorkerTasksResult =
      run: RunWithDequeueScalars;
      environmentType: RuntimeEnvironmentType;
      worker: BackgroundWorker;
-      task: BackgroundWorkerTask;
-      queue: TaskQueue;
-      deployment: WorkerDeployment | null;
+      task: ResolvedWorkerTask;
+      queue: ResolvedTaskQueue;
+      deployment: ResolvedWorkerDeployment | null;
    };

 type WorkerDeploymentWithWorkerTasks = {
  worker: BackgroundWorker;
-  tasks: BackgroundWorkerTask[];
-  queues: TaskQueue[];
-  deployment: WorkerDeployment | null;
+  tasks: ResolvedWorkerTask[];
+  queues: ResolvedTaskQueue[];
+  deployment: ResolvedWorkerDeployment | null;
 };

 export class DequeueSystem {
@@ -487,6 +487,102 @@ describe("DequeueSystem controlPlaneResolver (latest-v2 fallback + workerId bran
  );
 });

+describe("DequeueSystem controlPlaneResolver (worker-task read shape)", () => {
+  /**
+   * The resolver must ship only the columns the dequeue path reads: for each task id/slug +
+   * machineConfig/retryConfig/maxDurationInSeconds; for the deployment id/friendlyId/
+   * imageReference/imagePlatform; for each queue id/name. The heavy JSON columns it never reads
+   * (task payloadSchema/config/queueConfig/description, deployment externalBuildData/
+   * buildServerMetadata/errorData/git, queue rateLimit) must NOT come back, so the hot
+   * control-plane read stops shipping the large per-query payload.
+   */
+  containerTest(
+    "resolveWorkerVersion returns only the dequeue-read columns and drops the heavy JSON columns",
+    async ({ prisma }) => {
+      const taskSlug = "test-task";
+      const cp = await seedControlPlane(prisma as unknown as PrismaClient, "slim", taskSlug);
+
+      await prisma.backgroundWorkerTask.update({
+        where: { id: cp.task.id },
+        data: {
+          machineConfig: { preset: "small-2x" },
+          retryConfig: { maxAttempts: 5, factor: 2, minTimeoutInMs: 100, maxTimeoutInMs: 1000 },
+          maxDurationInSeconds: 120,
+          payloadSchema: { type: "object", properties: { message: { type: "string" } } },
+          config: { type: "ai-sdk-chat" },
+          queueConfig: { concurrencyLimit: 7 },
+          description: "must not be shipped on the dequeue read",
+        },
+      });
+
+      await prisma.workerDeployment.update({
+        where: { id: cp.deployment.id },
+        data: {
+          externalBuildData: { imageTag: "must-not-ship", buildId: "b1" },
+          buildServerMetadata: { logs: "must-not-ship" },
+          errorData: { message: "must-not-ship" },
+          git: { commitSha: "deadbeef", must: "not-ship" },
+        },
+      });
+
+      await prisma.taskQueue.update({
+        where: { id: cp.queue.id },
+        data: { rateLimit: { limit: 10, must: "not-ship" } },
+      });
+
+      const resolver = new PassthroughControlPlaneResolver({
+        prisma: prisma as unknown as PrismaClient,
+      });
+
+      const version = await resolver.resolveWorkerVersion({
+        environmentId: cp.environment.id,
+        type: "PRODUCTION",
+      });
+
+      assertNonNullable(version);
+      const task = version.tasks.find((t) => t.slug === taskSlug);
+      assertNonNullable(task);
+
+      expect(task.id).toBe(cp.task.id);
+      expect(task.slug).toBe(taskSlug);
+      expect(task.machineConfig).toEqual({ preset: "small-2x" });
+      expect(task.retryConfig).toMatchObject({ maxAttempts: 5 });
+      expect(task.maxDurationInSeconds).toBe(120);
+
+      expect("payloadSchema" in task).toBe(false);
+      expect("config" in task).toBe(false);
+      expect("queueConfig" in task).toBe(false);
+      expect("description" in task).toBe(false);
+      expect(Object.keys(task).sort()).toEqual([
+        "id",
+        "machineConfig",
+        "maxDurationInSeconds",
+        "retryConfig",
+        "slug",
+      ]);
+
+      assertNonNullable(version.deployment);
+      expect(version.deployment.id).toBe(cp.deployment.id);
+      expect("externalBuildData" in version.deployment).toBe(false);
+      expect("buildServerMetadata" in version.deployment).toBe(false);
+      expect("errorData" in version.deployment).toBe(false);
+      expect("git" in version.deployment).toBe(false);
+      expect(Object.keys(version.deployment).sort()).toEqual([
+        "friendlyId",
+        "id",
+        "imagePlatform",
+        "imageReference",
+      ]);
+
+      const queue = version.queues.find((q) => q.id === cp.queue.id);
+      assertNonNullable(queue);
+      expect(queue.name).toBe(cp.queueName);
+      expect("rateLimit" in queue).toBe(false);
+      expect(Object.keys(queue).sort()).toEqual(["id", "name"]);
+    }
+  );
+});
+
 describe("DequeueSystem controlPlaneResolver (single-DB passthrough)", () => {
  containerTest(
    "default passthrough dequeue is byte-identical (resolves env + worker version end-to-end)",
@@ -18,6 +18,9 @@ export type {
  ResolvedEngineEnv,
  ResolvedAuthenticatedEnv,
  ResolvedWorkerVersion,
+  ResolvedWorkerTask,
+  ResolvedTaskQueue,
+  ResolvedWorkerDeployment,
 } from "./engine/controlPlaneResolver.js";

 // Batch Queue exports