perf(run-store): route id-set reads to the owning store, not both DBs (#4342)

ericallam · web-flow · commit e9ac98b7a11c · 2026-07-22T23:03:24.000+01:00
## Summary

The split run-store's id-set read path (`#findRunsByIdSet`, used by the
runs-list hydrate, the realtime hydrator, and engine sweeps) queried the
new store for the entire id set and then probed the legacy store for the
misses. A run's residency is a total function of its id (run-ops ids
live in the new store, every other id in legacy), so each id belongs to
exactly one store. Route each id to its owner and query each store only
for its own ids, in parallel. Same result set, and while a split is
active with most runs still on legacy it removes a wasted new-store
query from every id-set read.

## Change

`#findRunsByIdSet` now partitions the ids by `classifyResidency` and
runs one bounded query per store (skipping an empty side), in parallel,
mirroring `expireRunsBatch` and the single-run `#route`. `finalizeRows`
still applies orderBy/take/skip globally over the merged set.

This drops the id-set path's cross-store fallback, which existed to
prefer the new-store copy when the same id was present in both stores.
That collision cannot arise when each id maps to exactly one store
(nothing writes a legacy-shaped id into the new store), so the fallback
is dead code. The two id-set tests that asserted "new copy wins on
collision" now assert the routing invariant: a legacy-shaped id resolves
to the legacy store and the path never consults the new store.

The open-predicate path (`#findRunsOpen`) is unchanged: an open `where`
has no id to route on, so it still unions both stores and dedupes.
diff --git a/apps/webapp/test/waitpointPresenter.connectedRunsBounded.test.ts b/apps/webapp/test/waitpointPresenter.connectedRunsBounded.test.ts
@@ -64,6 +64,7 @@ vi.mock("~/presenters/v3/NextRunListPresenter.server", () => ({
 
 import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
 import { PostgresRunStore, RoutingRunStore } from "@internal/run-store";
+import { generateRunOpsId } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClient } from "@trigger.dev/database";
 import type { RunOpsPrismaClient } from "@internal/run-ops-database";
 import {
@@ -156,6 +157,7 @@ async function seedRun(
 ) {
   return (prisma as PrismaClient).taskRun.create({
     data: {
+      id: `run_${generateRunOpsId()}`,
       friendlyId,
       taskIdentifier: "my-task",
       status: "PENDING",
diff --git a/apps/webapp/test/waitpointPresenter.danglingConnectedRuns.test.ts b/apps/webapp/test/waitpointPresenter.danglingConnectedRuns.test.ts
@@ -58,6 +58,7 @@ vi.mock("~/presenters/v3/NextRunListPresenter.server", () => ({
 
 import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
 import { PostgresRunStore, RoutingRunStore } from "@internal/run-store";
+import { generateRunOpsId } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClient } from "@trigger.dev/database";
 import type { RunOpsPrismaClient } from "@internal/run-ops-database";
 import {
@@ -141,6 +142,7 @@ async function seedWaitpoint(prisma: RunOpsPrismaClient, ctx: SeedContext, frien
 async function seedRun(prisma: RunOpsPrismaClient, ctx: SeedContext, friendlyId: string) {
   return (prisma as unknown as PrismaClient).taskRun.create({
     data: {
+      id: `run_${generateRunOpsId()}`,
       friendlyId,
       taskIdentifier: "my-task",
       status: "PENDING",
diff --git a/apps/webapp/test/waitpointPresenter.dedicatedConnectedRuns.readthrough.test.ts b/apps/webapp/test/waitpointPresenter.dedicatedConnectedRuns.readthrough.test.ts
@@ -59,6 +59,7 @@ vi.mock("~/presenters/v3/NextRunListPresenter.server", () => ({
 
 import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
 import { PostgresRunStore, RoutingRunStore } from "@internal/run-store";
+import { generateRunOpsId } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClient } from "@trigger.dev/database";
 import type { RunOpsPrismaClient } from "@internal/run-ops-database";
 import { WaitpointPresenter } from "~/presenters/v3/WaitpointPresenter.server";
@@ -147,10 +148,12 @@ async function seedWaitpoint(
 async function seedRun(
   prisma: PrismaClient | RunOpsPrismaClient,
   ctx: SeedContext,
-  friendlyId: string
+  friendlyId: string,
+  id?: string
 ) {
   return (prisma as PrismaClient).taskRun.create({
     data: {
+      ...(id ? { id } : {}),
       friendlyId,
       taskIdentifier: "my-task",
       status: "PENDING",
@@ -216,7 +219,7 @@ describe("WaitpointPresenter against the REAL dedicated run-ops client", () => {
       const waitpoint = await seedWaitpoint(prisma14, ctx, "waitpoint_crossdb");
 
       // The connected run + join live only on the NEW dedicated DB (co-resident with the run).
-      const run = await seedRun(prisma17, ctx, "run_crossnew");
+      const run = await seedRun(prisma17, ctx, "run_crossnew", `run_${generateRunOpsId()}`);
       await prisma17.waitpointRunConnection.create({
         data: { taskRunId: run.id, waitpointId: waitpoint.id },
       });
diff --git a/apps/webapp/test/waitpointPresenter.splitConnectedRuns.test.ts b/apps/webapp/test/waitpointPresenter.splitConnectedRuns.test.ts
@@ -62,6 +62,7 @@ vi.mock("~/presenters/v3/NextRunListPresenter.server", () => ({
 
 import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
 import { PostgresRunStore, RoutingRunStore } from "@internal/run-store";
+import { generateRunOpsId } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClient } from "@trigger.dev/database";
 import type { RunOpsPrismaClient } from "@internal/run-ops-database";
 import {
@@ -128,10 +129,12 @@ async function seedParents(prisma: PrismaClient, slug: string): Promise<SeedCont
 async function seedRun(
   prisma: PrismaClient | RunOpsPrismaClient,
   ctx: SeedContext,
-  friendlyId: string
+  friendlyId: string,
+  id?: string
 ) {
   return (prisma as PrismaClient).taskRun.create({
     data: {
+      ...(id ? { id } : {}),
       friendlyId,
       taskIdentifier: "my-task",
       status: "PENDING",
@@ -175,7 +178,7 @@ describe("WaitpointPresenter — connected runs SPLIT across both physical DBs",
       // 2 connected runs resident + joined on NEW (below the limit on its own).
       const NEW_RUN_FRIENDLY_IDS = ["run_split_new0", "run_split_new1"];
       for (const friendlyId of NEW_RUN_FRIENDLY_IDS) {
-        const run = await seedRun(prisma17, ctx, friendlyId);
+        const run = await seedRun(prisma17, ctx, friendlyId, `run_${generateRunOpsId()}`);
         await prisma17.waitpointRunConnection.create({
           data: { taskRunId: run.id, waitpointId: waitpoint.id },
         });
diff --git a/internal-packages/run-store/src/runOpsStore.idSetResidencyRouting.test.ts b/internal-packages/run-store/src/runOpsStore.idSetResidencyRouting.test.ts
@@ -0,0 +1,132 @@
+import { heteroPostgresTest } from "@internal/testcontainers";
+import type { PrismaClient } from "@trigger.dev/database";
+import { classifyResidency } from "@trigger.dev/core/v3/isomorphic";
+import { describe, expect } from "vitest";
+import { PostgresRunStore } from "./PostgresRunStore.js";
+import { RoutingRunStore } from "./runOpsStore.js";
+
+const ORG_ID = "orgroute0000000000000001";
+const PROJ_ID = "projroute00000000000001";
+const ENV_ID = "envroute0000000000000001";
+
+const newId = (i: number) => "k".repeat(20) + String(i).padStart(4, "0") + "01";
+const cuidId = (i: number) => "c".repeat(21) + String(i).padStart(4, "0");
+
+async function seedShared(prisma: PrismaClient, suffix: string) {
+  await prisma.organization.create({
+    data: { id: ORG_ID, title: `Route ${suffix}`, slug: `route-${suffix}` },
+  });
+  await prisma.project.create({
+    data: {
+      id: PROJ_ID,
+      name: `Route ${suffix}`,
+      slug: `route-${suffix}`,
+      externalRef: `proj_route_${suffix}`,
+      organizationId: ORG_ID,
+    },
+  });
+  await prisma.runtimeEnvironment.create({
+    data: {
+      id: ENV_ID,
+      type: "PRODUCTION",
+      slug: "prod",
+      projectId: PROJ_ID,
+      organizationId: ORG_ID,
+      apiKey: `tr_prod_${suffix}`,
+      pkApiKey: `pk_prod_${suffix}`,
+      shortcode: `short_${suffix}`,
+    },
+  });
+}
+
+const BASE = new Date("2026-01-01T00:00:00.000Z").getTime();
+
+async function seedRun(prisma: PrismaClient, id: string, offsetSec: number) {
+  await prisma.taskRun.create({
+    data: {
+      id,
+      engine: "V2",
+      status: "COMPLETED_SUCCESSFULLY",
+      friendlyId: `run_${id}`,
+      runtimeEnvironmentId: ENV_ID,
+      environmentType: "PRODUCTION",
+      organizationId: ORG_ID,
+      projectId: PROJ_ID,
+      taskIdentifier: "route-task",
+      payload: "{}",
+      payloadType: "application/json",
+      traceId: `trace_${id}`,
+      spanId: `span_${id}`,
+      queue: "task/route",
+      isTest: false,
+      taskEventStore: "taskEvent",
+      depth: 0,
+      createdAt: new Date(BASE + offsetSec * 1000),
+    },
+  });
+}
+
+describe("RoutingRunStore id-set residency routing", () => {
+  heteroPostgresTest(
+    "routes each id to its owning store and merges in orderBy order",
+    { timeout: 120000 },
+    async ({ prisma14, prisma17 }) => {
+      for (let i = 0; i < 5; i++) {
+        expect(classifyResidency(newId(i))).toBe("NEW");
+        expect(classifyResidency(cuidId(i))).toBe("LEGACY");
+      }
+
+      await seedShared(prisma14, "legacy");
+      await seedShared(prisma17, "new");
+
+      for (let i = 0; i < 5; i++) {
+        await seedRun(prisma17, newId(i), i * 2 + 1);
+        await seedRun(prisma14, cuidId(i), i * 2);
+      }
+
+      const legacyStore = new PostgresRunStore({ prisma: prisma14, readOnlyPrisma: prisma14 });
+      const newStore = new PostgresRunStore({ prisma: prisma17, readOnlyPrisma: prisma17 });
+      const router = new RoutingRunStore({ new: newStore, legacy: legacyStore });
+
+      const mixedIds = [0, 1, 2, 3, 4].flatMap((i) => [newId(i), cuidId(i)]);
+      const globalDesc = [
+        newId(4),
+        cuidId(4),
+        newId(3),
+        cuidId(3),
+        newId(2),
+        cuidId(2),
+        newId(1),
+        cuidId(1),
+        newId(0),
+        cuidId(0),
+      ];
+
+      const all = (await router.findRuns({
+        where: { id: { in: mixedIds } },
+        orderBy: { createdAt: "desc" },
+        take: 100,
+      })) as Array<{ id: string }>;
+      expect(all.map((r) => r.id)).toEqual(globalDesc);
+
+      const top4 = (await router.findRuns({
+        where: { id: { in: mixedIds } },
+        orderBy: { createdAt: "desc" },
+        take: 4,
+      })) as Array<{ id: string }>;
+      expect(top4.map((r) => r.id)).toEqual(globalDesc.slice(0, 4));
+
+      const newOnly = (await router.findRuns({
+        where: { id: { in: [newId(0), newId(2), newId(4)] } },
+        orderBy: { createdAt: "asc" },
+      })) as Array<{ id: string }>;
+      expect(newOnly.map((r) => r.id)).toEqual([newId(0), newId(2), newId(4)]);
+
+      const legacyOnly = (await router.findRuns({
+        where: { id: { in: [cuidId(1), cuidId(3)] } },
+        orderBy: { createdAt: "asc" },
+      })) as Array<{ id: string }>;
+      expect(legacyOnly.map((r) => r.id)).toEqual([cuidId(1), cuidId(3)]);
+    }
+  );
+});
diff --git a/internal-packages/run-store/src/runOpsStore.mixedResidency.test.ts b/internal-packages/run-store/src/runOpsStore.mixedResidency.test.ts
@@ -262,11 +262,8 @@ describe("RoutingRunStore — mixed-residency matrix (cuid #legacy + run-ops id
     }
   );
 
-  // ── Case 1b: NEW-wins on id collision in #findRunsByIdSet ──
-  // The copy→fence window can leave the same id on both DBs. The id-set path queries NEW first; an id
-  // already found on NEW must NOT be re-fetched from LEGACY, so the NEW copy wins.
   heteroRunOpsPostgresTest(
-    "case 1b: findRuns by id-set with a colliding id resolves to the NEW copy",
+    "case 1b: findRuns by id-set routes a cuid id to LEGACY only, ignoring any NEW copy",
     async ({ prisma14, prisma17 }) => {
       const { router } = makeSplitRouter(prisma14, prisma17);
       const env = await seedSharedEnv(prisma14, "m1b");
@@ -304,8 +301,8 @@ describe("RoutingRunStore — mixed-residency matrix (cuid #legacy + run-ops id
         where: { id: { in: [collidingId] } },
         select: { id: true, taskIdentifier: true },
       });
-      expect(rows).toHaveLength(1); // deduped, not double-reported
-      expect((rows[0] as any).taskIdentifier).toBe("new-copy-wins"); // NEW wins
+      expect(rows).toHaveLength(1);
+      expect((rows[0] as any).taskIdentifier).toBe("my-task");
     }
   );
 
diff --git a/internal-packages/run-store/src/runOpsStore.test.ts b/internal-packages/run-store/src/runOpsStore.test.ts
@@ -1004,10 +1004,8 @@ describe("RoutingRunStore.findRuns split-mode fan-out + drain", () => {
     }
   );
 
-  // A run present on BOTH DBs (the copy->fence migration window) must be returned ONCE,
-  // and the NEW copy wins.
   heteroPostgresTest(
-    "id-set dedupes a run present on both DBs, preferring NEW",
+    "id-set routes a cuid id to its LEGACY owner and does not consult NEW",
     async ({ prisma14, prisma17 }) => {
       const legacyStore = new PostgresRunStore({ prisma: prisma14, readOnlyPrisma: prisma14 });
       const newStore = new PostgresRunStore({ prisma: prisma17, readOnlyPrisma: prisma17 });
@@ -1031,7 +1029,7 @@ describe("RoutingRunStore.findRuns split-mode fan-out + drain", () => {
         select: { id: true, taskIdentifier: true },
       })) as Array<{ id: string; taskIdentifier: string }>;
       expect(rows).toHaveLength(1);
-      expect(rows[0]!.taskIdentifier).toBe("from-new");
+      expect(rows[0]!.taskIdentifier).toBe("from-legacy");
     }
   );
 
diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts
@@ -301,12 +301,12 @@ export class RoutingRunStore implements RunStore {
     },
     client?: ReadClient
   ): Promise<unknown> {
-    // SPLIT-mode fan-out across NEW + LEGACY. A `findRuns` `where` can span ids of mixed
-    // residency, so we resolve each owning store and merge, preserving orderBy/take/skip.
-    // The caller's client is never forwarded verbatim (it is the control-plane client); its
-    // presence routes each leg to that store's OWN primary (read-your-writes), else each store
-    // reads its own replica as before. NEW wins on id collisions (the copy->fence migration
-    // window) so a half-migrated run is never double-reported.
+    // SPLIT-mode routing across NEW + LEGACY. A bounded id set is routed per id to its owning
+    // store by residency (#findRunsByIdSet); an open predicate with no id to route on unions both
+    // stores and dedupes NEW-wins (#findRunsOpen). Either way orderBy/take/skip are re-imposed
+    // globally over the merged rows. The caller's client is never forwarded verbatim (it is the
+    // control-plane client); its presence routes each leg to that store's OWN primary
+    // (read-your-writes), else each store reads its own replica as before.
     return this.#findRunsRouted(args, client);
   }
 
@@ -324,33 +324,27 @@ export class RoutingRunStore implements RunStore {
     return idList ? this.#findRunsByIdSet(args, idList, client) : this.#findRunsOpen(args, client);
   }
 
-  // Bounded id-set (the list hydrate + engine sweeps). Query NEW for the whole set first
-  // (it holds run-ops runs); probe LEGACY only for the ids NEW missed that could still live
-  // there (cuid). The two id sets are disjoint by construction, so the merge needs no dedupe.
+  // Bounded id-set (the list hydrate + engine sweeps). Residency is a total function of the id
+  // (classifyResidency), so route each id to its owning store and query each store only for its
+  // own ids, in parallel; never query NEW for a cuid or LEGACY for a run-ops id. The partitions
+  // are disjoint by construction, so the merge needs no dedupe. take/skip are never pushed per
+  // store (that would truncate a store's page before the merge knows membership); finalizeRows
+  // re-imposes orderBy/take/skip once, globally, over the merged rows.
   async #findRunsByIdSet(
     args: FindRunsArgs,
     ids: string[],
     client?: ReadClient
   ): Promise<unknown[]> {
     const { args: selArgs, addedFields } = ensureProjected(args);
-    // The id set already bounds the per-store result, so never push take/skip down — doing
-    // so would truncate a store's page before the merge knows membership and mis-attribute
-    // rows. take/skip are applied once, globally, in finalizeRows.
     const fan = { ...selArgs, take: undefined, skip: undefined };
+    const newIds = ids.filter((id) => this.#classifySafe(id) === "NEW");
+    const legacyIds = ids.filter((id) => this.#classifySafe(id) !== "NEW");
     const findNew = this.#findManyOn(this.#new, client);
     const findLegacy = this.#findManyOn(this.#legacy, client);
-
-    const newRows = await findNew(fan);
-    const foundIds = new Set(newRows.map((r) => r.id as string));
-
-    const toLegacy: string[] = [];
-    for (const id of ids) {
-      if (foundIds.has(id)) continue;
-      if (this.#classifySafe(id) === "NEW") continue; // run-ops id: cannot live on LEGACY
-      toLegacy.push(id);
-    }
-
-    const legacyRows = toLegacy.length > 0 ? await findLegacy(narrowToIds(fan, toLegacy)) : [];
+    const [newRows, legacyRows] = await Promise.all([
+      newIds.length > 0 ? findNew(narrowToIds(fan, newIds)) : [],
+      legacyIds.length > 0 ? findLegacy(narrowToIds(fan, legacyIds)) : [],
+    ]);
     return finalizeRows([...newRows, ...legacyRows], args, addedFields);
   }
 
