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 "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..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,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 { 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); }