perf(webapp,run-store): group per-item run reads into batched queries

Several run-ops read paths issued one database query per item where a single
batched query returns the same data: batch results hydrated each member run
separately, the run retrieve endpoint resolved each locked worker version
separately, the dedicated-schema relation hydrator ran per row, and a few
services read runs in per-id loops. These now group into batched reads.

Adds a grouped findRunsByIds to the run store (one residency-partitioned read
for a set of run ids) and routes the presenters and services through it. The
dedicated-schema relation hydrator batches across all rows per relation. The
waitpoint connected-runs read also regains its display limit at the fetch, so
it no longer loads every connected run into memory before slicing to five.

Author: d-cs

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

@@ -1,4 +1,5 @@
 import type { BatchTaskRunExecutionResult } from "@trigger.dev/core/v3";
+import { ownerEngine } from "@trigger.dev/core/v3/isomorphic";
 import {
   $replica,
   type PrismaClientOrTransaction,
@@ -8,7 +9,6 @@ import {
 import type { TaskRunWithAttempts } from "~/models/taskRun.server";
 import { executionResultForTaskRun } from "~/models/taskRun.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
 import { runStore as defaultRunStore } from "~/v3/runStore.server";
 import { BasePresenter } from "./basePresenter.server";
 
@@ -43,11 +43,14 @@ const memberRunSelect = {
 } as const;
 
 /**
- * Split on: the batch row + its item rows resolve new-run-ops first, then the LEGACY RUN-OPS
- * READ REPLICA ONLY (never the legacy primary — there is no such handle); each member run is
- * hydrated independently via readThroughRun keyed on the member runId, so a batch whose members
- * span migrated + abandoned runs returns the complete reachable set (the batch-spanning-the-line
- * read; the dangling-reference termination gate is a separate, adjacent unit).
+ * Split on: the batch row + its members resolve new-run-ops first, then the LEGACY RUN-OPS READ
+ * REPLICA ONLY (never the legacy primary — there is no such handle). Members hydrate via ONE
+ * grouped read against `newClient` for the whole id set, then ONE grouped read against
+ * `legacyReplica` for just the misses that could still be legacy-resident — the same
+ * residency-partitioned shape as the old per-member read-through, but batched instead of fanned
+ * out one query per member. A batch whose members span migrated + abandoned runs returns the
+ * complete reachable set (the batch-spanning-the-line read; the dangling-reference termination
+ * gate is a separate, adjacent unit).
  *
  * Split off (single-DB / self-host): one passthrough read for the batch row + a single store
  * id-set hydrate for the members — no legacy read, no known-migrated probe, no second connection.
@@ -133,7 +136,7 @@ export class ApiBatchResultsPresenter extends BasePresenter {
   // Split: resolve the batch row new-first then off the legacy READ REPLICA only (a batch id may
   // be cuid or run-ops id, and a cuid-shaped id can still have been backfilled onto NEW, so id-shape
   // residency is not authoritative for the row — the new-first-then-legacy probe is), then
-  // hydrate every member run independently via the per-run read-through primitive.
+  // hydrate every member run in ONE grouped new-then-legacy read.
   async #callSplit(
     friendlyId: string,
     env: AuthenticatedEnvironment
@@ -175,41 +178,35 @@ export class ApiBatchResultsPresenter extends BasePresenter {
       };
     }
 
-    const readMemberRun = (client: PrismaClientOrTransaction, taskRunId: string) =>
-      client.taskRun.findFirst({
-        where: { id: taskRunId },
-        select: memberRunSelect,
-      }) as Promise<TaskRunWithAttempts | null>;
-
-    // Per-member fan-out: each member may live on a different DB, so a single nested include cannot
-    // cross the seam. Promise.all preserves batchRun.items order, unchanged from today.
-    const memberResults = await Promise.all(
-      batchRun.items.map(async (item) => {
-        const result = await readThroughRun<TaskRunWithAttempts>({
-          runId: item.taskRunId,
-          environmentId: env.id,
-          readNew: (client) => readMemberRun(client, item.taskRunId),
-          readLegacy: (replica) => readMemberRun(replica, item.taskRunId),
-          deps: {
-            splitEnabled: true,
-            // Pass the SAME resolved handles the batch row used, so the batch row and its members
-            // never resolve against different DBs. (Letting these fall through to readThroughRun's
-            // own module-level defaults would diverge from the batch read's `?? this._replica`.)
-            newClient,
-            legacyReplica,
-            isPastRetention: this.readThrough?.isPastRetention,
-          },
-        });
+    const taskRunIds = batchRun.items.map((item) => item.taskRunId);
 
-        // not-found / past-retention members are omitted (matches today's drop-undefined behavior);
-        // the dangling-reference termination gate (separate unit) governs whether that's permitted.
-        if (result.source === "not-found" || result.source === "past-retention") {
-          return undefined;
-        }
+    const newRows = (await newClient.taskRun.findMany({
+      where: { id: { in: taskRunIds } },
+      select: memberRunSelect,
+    })) as TaskRunWithAttempts[];
+    const runsById = new Map(newRows.map((run) => [run.id, run]));
 
-        return executionResultForTaskRun(result.value);
-      })
+    // A run-ops id can only live on NEW, so only misses that AREN'T run-ops-shaped are candidates
+    // for the legacy probe — mirrors readThroughRun's per-id "NEW residency skips legacy" rule.
+    const legacyCandidateIds = taskRunIds.filter(
+      (id) => !runsById.has(id) && ownerEngine(id) !== "NEW"
     );
+    if (legacyCandidateIds.length > 0) {
+      const legacyRows = (await legacyReplica.taskRun.findMany({
+        where: { id: { in: legacyCandidateIds } },
+        select: memberRunSelect,
+      })) as TaskRunWithAttempts[];
+      for (const run of legacyRows) {
+        runsById.set(run.id, run);
+      }
+    }
+
+    // not-found members are omitted (matches today's drop-undefined behavior); the
+    // dangling-reference termination gate (separate unit) governs whether that's permitted.
+    const memberResults = batchRun.items.map((item) => {
+      const run = runsById.get(item.taskRunId);
+      return run ? executionResultForTaskRun(run) : undefined;
+    });
 
     return {
       id: batchRun.friendlyId,

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

@@ -164,17 +164,13 @@ export class ApiRetrieveRunPresenter {
         collect(child.lockedToVersionId);
       }
 
+      const lockedWorkersByVersionId =
+        await controlPlaneResolver.resolveRunLockedWorkersByVersionIds([...distinctVersionIds]);
       const versionById = new Map<string, string | null>(
-        await Promise.all(
-          [...distinctVersionIds].map(
-            async (id) =>
-              [
-                id,
-                (await controlPlaneResolver.resolveRunLockedWorker({ lockedToVersionId: id }))
-                  ?.lockedToVersion?.version ?? null,
-              ] as const
-          )
-        )
+        [...distinctVersionIds].map((id) => [
+          id,
+          lockedWorkersByVersionId.get(id)?.lockedToVersion?.version ?? null,
+        ])
       );
 
       const resolveVersion = (id: string | null): { version: string } | null => {

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

@@ -11,6 +11,10 @@ import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server";
 
 export type WaitpointDetail = NonNullable<Awaited<ReturnType<WaitpointPresenter["call"]>>>;
 
+// Single-sourced bound for connected run friendlyIds: applied at the FETCH in #connectedRunIdsOn,
+// not just at display time.
+export const CONNECTED_RUNS_DISPLAY_LIMIT = 5;
+
 export class WaitpointPresenter extends BasePresenter {
   constructor(
     prisma?: PrismaClientOrTransaction,
@@ -98,16 +102,16 @@ export class WaitpointPresenter extends BasePresenter {
       const runs = await client.taskRun.findMany({
         where: { id: { in: runIds } },
         select: { friendlyId: true },
-        take: 5,
+        take: CONNECTED_RUNS_DISPLAY_LIMIT,
       });
       for (const run of runs) {
         friendlyIds.add(run.friendlyId);
       }
-      if (friendlyIds.size >= 5) {
+      if (friendlyIds.size >= CONNECTED_RUNS_DISPLAY_LIMIT) {
         break;
       }
     }
-    return Array.from(friendlyIds).slice(0, 5);
+    return Array.from(friendlyIds).slice(0, CONNECTED_RUNS_DISPLAY_LIMIT);
   }
 
   // Schema-aware read of the run ids linked to a waitpoint: the dedicated subset uses the explicit
@@ -125,11 +129,14 @@ export class WaitpointPresenter extends BasePresenter {
       const links = await joinDelegate.findMany({
         where: { waitpointId },
         select: { taskRunId: true },
+        take: CONNECTED_RUNS_DISPLAY_LIMIT,
       });
       return links.map((link) => link.taskRunId);
     }
+    // CONNECTED_RUNS_DISPLAY_LIMIT is a compile-time constant int, not an interpolated value.
     const rows = await client.$queryRaw<{ A: string }[]>`
       SELECT "A" FROM "_WaitpointRunConnections" WHERE "B" = ${waitpointId}
+      LIMIT ${CONNECTED_RUNS_DISPLAY_LIMIT}
     `;
     return rows.map((row) => row.A);
   }

apps/webapp/app/v3/runOpsMigration/controlPlaneResolver.server.test.ts

@@ -0,0 +1,168 @@
+// Real control-plane (single Postgres) proof that resolving the locked-worker version for many
+// runs' distinct `lockedToVersionId`s is a GROUPED query, not one `backgroundWorker.findFirst`
+// per id. The DB is never mocked: the call-counting proxy below delegates every call to the
+// real Prisma client (the real query still runs against the real container) - it only tallies
+// how many times each model.method pair was invoked.
+import { postgresTest } from "@internal/testcontainers";
+import { describe, expect } from "vitest";
+import type { PrismaClient, PrismaReplicaClient } from "@trigger.dev/database";
+import { ControlPlaneCache } from "./controlPlaneCache.server";
+import { ControlPlaneResolver } from "./controlPlaneResolver.server";
+
+// Wraps a real Prisma client so every `client.<model>.<method>(...)` call is tallied by
+// `"<model>.<method>"` before being forwarded, unmodified, to the real delegate. No behavior is
+// faked or short-circuited - this is instrumentation, not a mock.
+function createCallCountingProxy<T extends object>(
+  client: T
+): { client: T; counts: Map<string, number> } {
+  const counts = new Map<string, number>();
+  const bump = (key: string) => counts.set(key, (counts.get(key) ?? 0) + 1);
+  const wrappedModels = new Map<string, unknown>();
+
+  const wrapModel = (model: string, delegate: object) =>
+    new Proxy(delegate, {
+      get(target, prop, receiver) {
+        const value = Reflect.get(target, prop, receiver);
+        if (typeof prop === "string" && typeof value === "function") {
+          return (...args: unknown[]) => {
+            bump(`${model}.${prop}`);
+            return (value as (...a: unknown[]) => unknown).apply(target, args);
+          };
+        }
+        return value;
+      },
+    });
+
+  const proxy = new Proxy(client, {
+    get(target, prop, receiver) {
+      const value = Reflect.get(target, prop, receiver);
+      if (
+        typeof prop === "string" &&
+        value &&
+        typeof value === "object" &&
+        typeof (value as { findMany?: unknown }).findMany === "function"
+      ) {
+        if (!wrappedModels.has(prop)) {
+          wrappedModels.set(prop, wrapModel(prop, value as object));
+        }
+        return wrappedModels.get(prop);
+      }
+      return value;
+    },
+  });
+
+  return { client: proxy as T, counts };
+}
+
+let n = 0;
+
+async function seedEnv(prisma: PrismaClient) {
+  const s = n++;
+  const organization = await prisma.organization.create({
+    data: { title: `Org ${s}`, slug: `org-${s}` },
+  });
+  const project = await prisma.project.create({
+    data: {
+      name: `P ${s}`,
+      slug: `p-${s}`,
+      externalRef: `proj_${s}`,
+      organizationId: organization.id,
+    },
+  });
+  const environment = await prisma.runtimeEnvironment.create({
+    data: {
+      type: "PRODUCTION",
+      slug: `env-${s}`,
+      projectId: project.id,
+      organizationId: organization.id,
+      apiKey: `tr_${s}`,
+      pkApiKey: `pk_${s}`,
+      shortcode: `sc_${s}`,
+    },
+  });
+  return { organization, project, environment };
+}
+
+async function seedBackgroundWorker(
+  prisma: PrismaClient,
+  ctx: { projectId: string; runtimeEnvironmentId: string },
+  version: string
+) {
+  const s = n++;
+  return prisma.backgroundWorker.create({
+    data: {
+      friendlyId: `worker_${s}`,
+      version,
+      contentHash: `hash_${s}`,
+      projectId: ctx.projectId,
+      runtimeEnvironmentId: ctx.runtimeEnvironmentId,
+      metadata: {},
+    },
+  });
+}
+
+describe("ControlPlaneResolver.resolveRunLockedWorkersByVersionIds", () => {
+  postgresTest(
+    "issues ONE grouped findMany and ZERO per-id findFirst for N distinct ids",
+    async ({ prisma }) => {
+      const { project, environment } = await seedEnv(prisma);
+      const workers = await Promise.all(
+        [0, 1, 2].map((i) =>
+          seedBackgroundWorker(
+            prisma,
+            { projectId: project.id, runtimeEnvironmentId: environment.id },
+            `2024010${i}.0`
+          )
+        )
+      );
+
+      const { client: countedReplica, counts } = createCallCountingProxy(prisma);
+
+      const resolver = new ControlPlaneResolver({
+        controlPlanePrimary: prisma,
+        controlPlaneReplica: countedReplica as unknown as PrismaReplicaClient,
+        cache: new ControlPlaneCache(),
+        splitEnabled: () => true,
+      });
+
+      const ids = workers.map((w) => w.id);
+      const result = await resolver.resolveRunLockedWorkersByVersionIds(ids);
+
+      expect(counts.get("backgroundWorker.findMany") ?? 0).toBe(1);
+      expect(counts.get("backgroundWorker.findFirst") ?? 0).toBe(0);
+
+      for (const worker of workers) {
+        expect(result.get(worker.id)?.lockedToVersion?.version).toBe(worker.version);
+      }
+    }
+  );
+
+  postgresTest("cache-hit ids issue no query at all", async ({ prisma }) => {
+    const { project, environment } = await seedEnv(prisma);
+    const worker = await seedBackgroundWorker(
+      prisma,
+      { projectId: project.id, runtimeEnvironmentId: environment.id },
+      "20240101.0"
+    );
+
+    const cache = new ControlPlaneCache();
+    const { client: countedReplica, counts } = createCallCountingProxy(prisma);
+
+    const resolver = new ControlPlaneResolver({
+      controlPlanePrimary: prisma,
+      controlPlaneReplica: countedReplica as unknown as PrismaReplicaClient,
+      cache,
+      splitEnabled: () => true,
+    });
+
+    // Warm the cache.
+    await resolver.resolveRunLockedWorkersByVersionIds([worker.id]);
+    expect(counts.get("backgroundWorker.findMany") ?? 0).toBe(1);
+
+    // Second call for the same id is served entirely from cache - no query at all.
+    const result = await resolver.resolveRunLockedWorkersByVersionIds([worker.id]);
+    expect(counts.get("backgroundWorker.findMany") ?? 0).toBe(1);
+    expect(counts.get("backgroundWorker.findFirst") ?? 0).toBe(0);
+    expect(result.get(worker.id)?.lockedToVersion?.version).toBe(worker.version);
+  });
+});