From f5a2ea6b6d9c32ba255db77820ae4365e93254b8 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 22 Jul 2026 17:24:28 +0100 Subject: [PATCH 1/3] perf(run-store): route id-set reads to the owning store, not both DBs Run residency is a total function of the id (run-ops ids resolve to the new store, every other id to legacy), and writes and single-run reads already route by it. The id-set read path was the outlier: it queried the new store for the entire id set, then probed legacy for the misses. While a split is active with most runs still on legacy, that meant a wasted new-store query on every list hydrate and engine sweep. Partition the id set by residency and query each store only for its own ids, in parallel (matching expireRunsBatch). Same result set. The old cross-store fallback existed to prefer a new-store copy of a same-id-in-both collision; that cannot arise when each id maps to exactly one store, so it is removed. The open-predicate path is unchanged (an open where cannot route by id, so it still unions both stores and dedupes). --- .../runOpsStore.idSetResidencyRouting.test.ts | 121 ++++++++++++++++++ .../src/runOpsStore.mixedResidency.test.ts | 9 +- .../run-store/src/runOpsStore.test.ts | 6 +- .../run-store/src/runOpsStore.ts | 24 +--- 4 files changed, 132 insertions(+), 28 deletions(-) create mode 100644 internal-packages/run-store/src/runOpsStore.idSetResidencyRouting.test.ts diff --git a/internal-packages/run-store/src/runOpsStore.idSetResidencyRouting.test.ts b/internal-packages/run-store/src/runOpsStore.idSetResidencyRouting.test.ts new file mode 100644 index 0000000000..5757ef7c43 --- /dev/null +++ b/internal-packages/run-store/src/runOpsStore.idSetResidencyRouting.test.ts @@ -0,0 +1,121 @@ +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 index ba87c75ff5..cf21212016 100644 --- 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 index 75f838deb3..a7c5adbbfb 100644 --- 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 index ba49698076..8d99384a7b 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -324,33 +324,21 @@ 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. async #findRunsByIdSet( args: FindRunsArgs, ids: string[], client?: ReadClient ): Promise { 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); } From a69429ab34a88d7962a82a9733f9c0c7b3fe4660 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 22 Jul 2026 17:52:48 +0100 Subject: [PATCH 2/3] test(webapp): seed run-ops ids for new-store runs in connected-run tests Under deterministic id routing, a run in the new store has a run-ops id and a cuid always resolves to legacy. These WaitpointPresenter connected-run tests seeded new-store runs with default cuid ids and relied on the old cross-store scan to still find them. Give the new-store-resident seeds run-ops ids so they route to the store that holds them. Also formats the new run-store routing test. --- .../waitpointPresenter.connectedRunsBounded.test.ts | 2 ++ ...waitpointPresenter.danglingConnectedRuns.test.ts | 2 ++ ...enter.dedicatedConnectedRuns.readthrough.test.ts | 7 +++++-- .../waitpointPresenter.splitConnectedRuns.test.ts | 7 +++++-- .../src/runOpsStore.idSetResidencyRouting.test.ts | 13 ++++++++++++- 5 files changed, 26 insertions(+), 5 deletions(-) diff --git a/apps/webapp/test/waitpointPresenter.connectedRunsBounded.test.ts b/apps/webapp/test/waitpointPresenter.connectedRunsBounded.test.ts index 9e2b0bcafb..ba50d312fd 100644 --- 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 index db10bdff62..26c6c6125e 100644 --- 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 index 7b1838d472..990ad8f979 100644 --- 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 index c6523ac3d4..351a7cb7fa 100644 --- 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 { 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 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 } }, From 4d07473c089846e9f2c351b227c30351b2cfa9dc Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Wed, 22 Jul 2026 22:44:48 +0100 Subject: [PATCH 3/3] docs(run-store): correct findRuns routing comments after id-set change --- internal-packages/run-store/src/runOpsStore.ts | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 8d99384a7b..467df81df2 100644 --- 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 { - // 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,6 +324,12 @@ export class RoutingRunStore implements RunStore { return idList ? this.#findRunsByIdSet(args, idList, client) : this.#findRunsOpen(args, client); } + // 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[],