From 446ad0d03303ec1de4cea66541b05d0d7672fa20 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 14 Aug 2026 13:19:09 +0100 Subject: [PATCH] perf(run-engine,webapp): narrow the control-plane worker-version read to the columns dequeue uses 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, pulling large JSON columns none of this path reads (task payloadSchema, config, queueConfig, description; deployment externalBuildData, buildServerMetadata, errorData, git; queue rateLimit). Replace the includes with explicit selects of only the columns dequeue uses, in both the passthrough resolver and the app resolver: task { id, slug, machineConfig, retryConfig, maxDurationInSeconds }, deployment { id, friendlyId, imageReference, imagePlatform }, queue { id, name }. The shared ResolvedWorkerVersion element types narrow to match, which also shrinks each cached worker-version entry. No behavior change. --- .../controlPlaneCache.server.ts | 70 +++++++++-- .../controlPlaneResolver.server.ts | 62 +++++++-- .../src/engine/controlPlaneResolver.ts | 119 ++++++++++++++---- .../src/engine/systems/dequeueSystem.ts | 20 +-- ...dequeueSystem.controlPlaneResolver.test.ts | 96 ++++++++++++++ internal-packages/run-engine/src/index.ts | 3 + 6 files changed, 315 insertions(+), 55 deletions(-) diff --git a/apps/webapp/app/v3/runOpsMigration/controlPlaneCache.server.ts b/apps/webapp/app/v3/runOpsMigration/controlPlaneCache.server.ts index 01fd205030..c62d6d9593 100644 --- a/apps/webapp/app/v3/runOpsMigration/controlPlaneCache.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/controlPlaneCache.server.ts @@ -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/…) diff --git a/apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.ts b/apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.ts index 97d7a93ac4..573160b9da 100644 --- a/apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.ts @@ -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 { 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 { 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" }], }); diff --git a/internal-packages/run-engine/src/engine/controlPlaneResolver.ts b/internal-packages/run-engine/src/engine/controlPlaneResolver.ts index da5f7b0060..256314b1d9 100644 --- a/internal-packages/run-engine/src/engine/controlPlaneResolver.ts +++ b/internal-packages/run-engine/src/engine/controlPlaneResolver.ts @@ -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, + }, }; } } diff --git a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts index 8887c850a5..f8f31d848a 100644 --- a/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/dequeueSystem.ts @@ -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 { diff --git a/internal-packages/run-engine/src/engine/tests/dequeueSystem.controlPlaneResolver.test.ts b/internal-packages/run-engine/src/engine/tests/dequeueSystem.controlPlaneResolver.test.ts index 3ea56bde8a..3fce8def73 100644 --- a/internal-packages/run-engine/src/engine/tests/dequeueSystem.controlPlaneResolver.test.ts +++ b/internal-packages/run-engine/src/engine/tests/dequeueSystem.controlPlaneResolver.test.ts @@ -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)", diff --git a/internal-packages/run-engine/src/index.ts b/internal-packages/run-engine/src/index.ts index b385f5e753..d8eaadf649 100644 --- a/internal-packages/run-engine/src/index.ts +++ b/internal-packages/run-engine/src/index.ts @@ -18,6 +18,9 @@ export type { ResolvedEngineEnv, ResolvedAuthenticatedEnv, ResolvedWorkerVersion, + ResolvedWorkerTask, + ResolvedTaskQueue, + ResolvedWorkerDeployment, } from "./engine/controlPlaneResolver.js"; // Batch Queue exports