diff --git a/apps/webapp/app/models/waitpointTag.server.ts b/apps/webapp/app/models/waitpointTag.server.ts index 0d521a5c83a..d2ad6a49a42 100644 --- a/apps/webapp/app/models/waitpointTag.server.ts +++ b/apps/webapp/app/models/waitpointTag.server.ts @@ -9,6 +9,7 @@ export async function createWaitpointTag({ environmentId, projectId, residency, + shardKey, }: { tag: string; environmentId: string; @@ -16,6 +17,9 @@ export async function createWaitpointTag({ // Residency from the env mint kind: a tag has no owning run, so a minted-new env pins it to NEW // instead of defaulting to the draining legacy DB. residency?: "NEW" | "LEGACY"; + // The environment's gen-2 mint shard, when it has one. A tag has no id the router can read, so + // without this the row lands on a gen-1 store while the token it describes lands on the shard. + shardKey?: string; }) { if (tag.trim().length === 0) return; @@ -30,7 +34,8 @@ export async function createWaitpointTag({ projectId, }, undefined, - residency + residency, + shardKey ); } catch (error) { if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") { diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.ts index 62322c527c7..92e49a001b4 100644 --- a/apps/webapp/app/routes/api.v1.waitpoints.tokens.ts +++ b/apps/webapp/app/routes/api.v1.waitpoints.tokens.ts @@ -16,6 +16,7 @@ import { type PrismaClientOrTransaction, } from "~/db.server"; import { resolveRunIdMintKind } from "~/v3/engineVersion.server"; +import { resolveMintShard } from "~/v3/runOpsMigration/runOpsMintShard.server"; import { logger } from "~/services/logger.server"; import { generateHttpCallbackUrl } from "~/services/httpCallback.server"; import { publicAccessTokenResponseHeaders } from "~/services/publicAccessTokenResponse.server"; @@ -69,6 +70,16 @@ const { action } = createActionApiRoute( }); const residency = mintKind === "runOpsId" ? "NEW" : "LEGACY"; + // The token's id is minted inside the engine, so the shard travels with the call. No + // extra query: the org flags this reads are already loaded on the authenticated env. + const standaloneShardKey = + mintKind === "runOpsId" + ? await resolveMintShard({ + id: authentication.environment.id, + orgFeatureFlags: authentication.environment.organization.featureFlags, + }) + : undefined; + //upsert tags let tags: { id: string; name: string }[] = []; const bodyTags = typeof body.tags === "string" ? [body.tags] : body.tags; @@ -86,6 +97,7 @@ const { action } = createActionApiRoute( environmentId: authentication.environment.id, projectId: authentication.environment.projectId, residency, + shardKey: standaloneShardKey, }); if (tagRecord) { tags.push(tagRecord); @@ -101,6 +113,7 @@ const { action } = createActionApiRoute( timeout, tags: bodyTags, standaloneResidency: residency, + standaloneShardKey, }); const waitpointId = WaitpointId.toFriendlyId(result.waitpoint.id); diff --git a/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts b/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts index f8ba67f3448..e1d8a0841aa 100644 --- a/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts +++ b/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts @@ -1,6 +1,6 @@ import type { RunEngine } from "@internal/run-engine"; import { TaskRunErrorCodes, type TaskRunError } from "@trigger.dev/core/v3"; -import { RunId, generateRunOpsId } from "@trigger.dev/core/v3/isomorphic"; +import { RunId } from "@trigger.dev/core/v3/isomorphic"; import type { PrismaClientOrTransaction, RuntimeEnvironmentType, @@ -8,8 +8,8 @@ import type { } from "@trigger.dev/database"; import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; -import { resolveRunIdMintKind } from "~/v3/engineVersion.server"; -import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server"; +import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server"; +import { resolveRunMintTarget } from "~/v3/runOpsMigration/resolveRunMintTarget.server"; import { getEventRepository } from "~/v3/eventRepository/index.server"; import { runStore as defaultRunStore } from "~/v3/runStore.server"; import type { RunStore } from "@internal/run-store"; @@ -103,17 +103,16 @@ export class TriggerFailedTaskService { return args.runFriendlyId; } - const mintKind = args.parentRunFriendlyId - ? resolveInheritedMintKind(args.parentRunFriendlyId) - : await resolveRunIdMintKind({ + return mintFriendlyIdForKind( + await resolveRunMintTarget({ + environment: { organizationId: args.organizationId, id: args.environmentId, orgFeatureFlags: args.orgFeatureFlags, - }); - - return mintKind === "runOpsId" - ? RunId.toFriendlyId(generateRunOpsId()) - : RunId.generate().friendlyId; + }, + parentRunFriendlyId: args.parentRunFriendlyId, + }) + ); } async call(request: TriggerFailedTaskRequest): Promise { diff --git a/apps/webapp/app/runEngine/services/triggerTask.server.ts b/apps/webapp/app/runEngine/services/triggerTask.server.ts index 8e9e99d7f09..d3320dbc219 100644 --- a/apps/webapp/app/runEngine/services/triggerTask.server.ts +++ b/apps/webapp/app/runEngine/services/triggerTask.server.ts @@ -28,9 +28,8 @@ import { parseDelay } from "~/utils/delays"; import { removeNullBytesFromKey } from "~/utils/nullBytes"; import { handleMetadataPacket } from "~/utils/packets"; import { startSpan } from "~/v3/tracing.server"; -import { resolveRunIdMintKind } from "~/v3/engineVersion.server"; -import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server"; import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server"; +import { resolveRunMintTarget } from "~/v3/runOpsMigration/resolveRunMintTarget.server"; import type { TriggerTaskServiceOptions, TriggerTaskServiceResult, @@ -218,15 +217,17 @@ export class RunEngineTriggerTaskService { parentRunFriendlyId?: string, region?: string ): Promise { - const mintKind = parentRunFriendlyId - ? resolveInheritedMintKind(parentRunFriendlyId) - : await resolveRunIdMintKind({ + return mintFriendlyIdForKind( + await resolveRunMintTarget({ + environment: { organizationId: environment.organizationId, id: environment.id, orgFeatureFlags: environment.organization.featureFlags, - }); - - return mintFriendlyIdForKind(mintKind, region); + }, + parentRunFriendlyId, + region, + }) + ); } public async call({ diff --git a/apps/webapp/app/v3/runEngineHandlers.server.ts b/apps/webapp/app/v3/runEngineHandlers.server.ts index c44bcc54cec..da5a5d89802 100644 --- a/apps/webapp/app/v3/runEngineHandlers.server.ts +++ b/apps/webapp/app/v3/runEngineHandlers.server.ts @@ -11,6 +11,7 @@ import { runOpsNewPrismaClient, runOpsNewReplicaClient, runOpsLegacyPrismaClient, + runOpsShardHandles, } from "~/db.server"; import { env } from "~/env.server"; import { findEnvironmentById, findEnvironmentFromRun } from "~/models/runtimeEnvironment.server"; @@ -1060,6 +1061,7 @@ export function setupBatchQueueCallbacks() { newReplica: runOpsNewReplicaClient, newWriter: runOpsNewPrismaClient, legacyWriter: runOpsLegacyPrismaClient, + shards: runOpsShardHandles, tryCompleteBatch: (batchId) => engine.tryCompleteBatch({ batchId }), }); }); diff --git a/apps/webapp/app/v3/runEngineHandlersShared.server.ts b/apps/webapp/app/v3/runEngineHandlersShared.server.ts index d8999e2332a..6af0e394211 100644 --- a/apps/webapp/app/v3/runEngineHandlersShared.server.ts +++ b/apps/webapp/app/v3/runEngineHandlersShared.server.ts @@ -4,6 +4,7 @@ * whole webapp service graph). The handlers wire the production defaults; tests * inject per-container stores/replicas, so these helpers never import db.server. */ +import { resolveShard } from "@trigger.dev/core/v3/isomorphic"; import type { CompleteBatchResult } from "@internal/run-engine"; import type { RunOpsPrismaClient } from "@internal/run-ops-database"; import type { RunStore } from "@internal/run-store"; @@ -83,8 +84,24 @@ export async function resolveBatchRunOpsWriter( newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; + shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; } ): Promise { + // A gen-2 batch names its shard in its id. The probe below is binary, so without this a gen-2 + // batch resolves to a store holding no such row and the update throws before the batch waitpoint + // completes, leaving the parent blocked with nothing logged. + const shardKey = resolveShard(batchId); + if (shardKey !== "new" && shardKey !== "legacy") { + const shard = deps.shards?.find((s) => s.key === shardKey); + if (!shard) { + // Writing to a guessed store is what strands a run. Fail loud instead. + throw new Error( + `resolveBatchRunOpsWriter: batch "${batchId}" names shard "${shardKey}", which is not configured` + ); + } + return shard.writer; + } + const onNew = await deps.newReplica.batchTaskRun.findFirst({ where: { id: batchId }, select: { id: true }, @@ -106,6 +123,7 @@ export type BatchCompletionDeps = { newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; + shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; tryCompleteBatch: (batchId: string) => Promise; }; @@ -136,6 +154,7 @@ export async function handleBatchCompletion( newReplica: deps.newReplica, newWriter: deps.newWriter, legacyWriter: deps.legacyWriter, + shards: deps.shards, }); try { diff --git a/apps/webapp/app/v3/runOpsMigration/gen2MintInertness.test.ts b/apps/webapp/app/v3/runOpsMigration/gen2MintInertness.test.ts new file mode 100644 index 00000000000..dfb906bfe97 --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/gen2MintInertness.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, it, vi } from "vitest"; +import { classifyKind, mintWaitpointIdFor, resolveShard } from "@trigger.dev/core/v3/isomorphic"; +import { resolveInheritedMintKind } from "./resolveInheritedMintKind.server"; +import { + mintAnchoredRunFriendlyId, + mintFriendlyIdForKind, +} from "./mintAnchoredRunFriendlyId.server"; +import { batchIdForMintKind } from "./mintBatchFriendlyId.server"; +import { resolveRunMintTarget } from "./resolveRunMintTarget.server"; + +// The gate is off when RUN_OPS_SHARDS is unset or runOpsMintShardSet is empty; either way +// resolveMintShard answers "new". Every assertion is "the id is what it was before gen-2". +const offShard = vi.fn().mockResolvedValue("new" as const); +const environment = { organizationId: "org_1", id: "env_1", orgFeatureFlags: {} }; + +describe("gate off — run mint paths", () => { + it("a root run on the run-ops path mints a gen-1 v1 id", async () => { + const target = await resolveRunMintTarget({ + environment, + region: "us-east-1", + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"), + resolveMintShard: offShard, + }, + }); + const body = mintFriendlyIdForKind(target).slice(4); + expect(body.length).toBe(26); + expect(body[24]).toBe("e"); // the region char, as today + expect(body[25]).toBe("1"); + }); + + it("a root run on a non-cut-over org mints a cuid", async () => { + const target = await resolveRunMintTarget({ + environment, + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("cuid"), + resolveMintShard: offShard, + }, + }); + expect(mintFriendlyIdForKind(target).slice(4).length).toBe(25); + }); + + it("a child of a gen-1 parent keeps the caller's region char", async () => { + // The pre-split code passed the region on both arms; dropping it on the inherited arm would + // silently stamp the default. + const target = await resolveRunMintTarget({ + environment, + parentRunFriendlyId: `run_${"a".repeat(24)}01`, + region: "us-east-1", + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"), + resolveMintShard: offShard, + }, + }); + const body = mintFriendlyIdForKind(target).slice(4); + expect(body[24]).toBe("e"); + expect(body[25]).toBe("1"); + }); + + it("a gen-2 parent's shard still outranks the caller's region", async () => { + const target = await resolveRunMintTarget({ + environment, + parentRunFriendlyId: `run_${"a".repeat(24)}a2`, + region: "us-east-1", + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"), + resolveMintShard: offShard, + }, + }); + expect(mintFriendlyIdForKind(target).slice(4)[24]).toBe("a"); + }); + + it("a child of a gen-1 parent mints a gen-1 v1 id", () => { + const body = mintFriendlyIdForKind(resolveInheritedMintKind(`run_${"a".repeat(24)}01`)).slice( + 4 + ); + expect(body[25]).toBe("1"); + }); + + it("a child of a cuid parent mints a cuid", () => { + expect( + mintFriendlyIdForKind(resolveInheritedMintKind(`run_${"b".repeat(25)}`)).slice(4).length + ).toBe(25); + }); +}); + +describe("gate off — batch and item paths", () => { + it("a batch with no shard char mints a gen-1 v1 id", () => { + const r = batchIdForMintKind({ kind: "runOpsId" }); + expect(r.id.length).toBe(26); + expect(r.id[25]).toBe("1"); + expect(classifyKind(r.id)).toBe("runOpsId"); + }); + + it("a batch on a non-cut-over org mints a cuid", () => { + expect(batchIdForMintKind({ kind: "cuid" }).id.length).toBe(25); + }); + + it("a batch item anchored on a gen-1 batch mints a gen-1 v1 id", () => { + const body = mintAnchoredRunFriendlyId(`batch_${"a".repeat(24)}01`).slice(4); + expect(body[25]).toBe("1"); + }); +}); + +describe("gate off — waitpoint paths", () => { + it("every gen-1 or legacy anchor yields a cuid waitpoint id", () => { + for (const anchor of [`${"a".repeat(24)}01`, "c".repeat(25), undefined]) { + const r = mintWaitpointIdFor(anchor); + expect(r.id.length).toBe(25); + expect(resolveShard(r.id)).toBe("legacy"); + } + }); +}); diff --git a/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts index 558731447a2..3beb4d746c8 100644 --- a/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts +++ b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts @@ -28,4 +28,16 @@ describe("mintAnchoredRunFriendlyId", () => { expect(parsed.format).toBe("b32hex"); expect(parsed.format === "b32hex" && parsed.region).toBe(REGION_CODES["us-east-1"]); }); + + it("a gen-2 batch anchor mints an item on the batch's shard", () => { + const body = mintAnchoredRunFriendlyId(`batch_${"a".repeat(24)}a2`).slice("run_".length); + expect(body).toHaveLength(26); + expect(body[24]).toBe("a"); + expect(body[25]).toBe("2"); + }); + + it("a gen-2 batch anchor ignores a caller region: the shard owns index 24", () => { + const body = mintAnchoredRunFriendlyId(`batch_${"a".repeat(24)}a2`, "us-east-1").slice(4); + expect(body[24]).toBe("a"); + }); }); diff --git a/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts index 0f5da2e56f7..d3de7bf8cb4 100644 --- a/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts @@ -1,15 +1,22 @@ -import { generateRunOpsId, RunId, type ResidencyKind } from "@trigger.dev/core/v3/isomorphic"; +import { generateRunOpsId, generateRunOpsIdV2, RunId } from "@trigger.dev/core/v3/isomorphic"; +import type { MintTarget } from "./mintTarget"; import { resolveInheritedMintKind } from "./resolveInheritedMintKind.server"; -// Shared id-generation branch for every run-mint path: "runOpsId" -> NEW store, "cuid" -> LEGACY. -export function mintFriendlyIdForKind(mintKind: ResidencyKind, region?: string): string { - return mintKind === "runOpsId" - ? RunId.toFriendlyId(generateRunOpsId(region)) - : RunId.generate().friendlyId; +// Shared id-generation branch for every run-mint path: "runOpsId" -> a dedicated store, +// "cuid" -> LEGACY. A shardChar selects one gen-2 shard and takes index 24; without one, +// the region takes that slot exactly as it does today. +export function mintFriendlyIdForKind(target: MintTarget): string { + if (target.kind !== "runOpsId") { + return RunId.generate().friendlyId; + } + + return RunId.toFriendlyId( + target.shardChar ? generateRunOpsIdV2(target.shardChar) : generateRunOpsId(target.region) + ); } // Anchor a batch item's mint on the BATCH's friendlyId (id-shape, zero I/O), never the per-org // flag, so the item and its BatchTaskRun stay co-resident across a mid-batch flag flip. export function mintAnchoredRunFriendlyId(batchFriendlyId: string, region?: string): string { - return mintFriendlyIdForKind(resolveInheritedMintKind(batchFriendlyId), region); + return mintFriendlyIdForKind({ ...resolveInheritedMintKind(batchFriendlyId), region }); } diff --git a/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.test.ts b/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.test.ts index 9973be57d1d..0e07a59d382 100644 --- a/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.test.ts +++ b/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.test.ts @@ -4,15 +4,23 @@ import { classifyKind } from "@trigger.dev/core/v3/isomorphic"; describe("batchIdForMintKind (pure)", () => { it("'runOpsId' kind -> 26-char classifiable NEW batch id (no 21-char ids)", () => { - const r = batchIdForMintKind("runOpsId"); + const r = batchIdForMintKind({ kind: "runOpsId" }); expect(r.friendlyId.startsWith("batch_")).toBe(true); expect(r.id.length).toBe(26); expect(classifyKind(r.id)).toBe("runOpsId"); expect(classifyKind(r.friendlyId)).toBe("runOpsId"); }); + it("a shard char mints a gen-2 batch id carrying that char", () => { + const r = batchIdForMintKind({ kind: "runOpsId", shardChar: "a" }); + expect(r.id.length).toBe(26); + expect(r.id[24]).toBe("a"); + expect(r.id[25]).toBe("2"); + expect(classifyKind(r.id)).toBe("runOpsId"); + }); + it("cuid -> 25-char classifiable LEGACY batch id", () => { - const r = batchIdForMintKind("cuid"); + const r = batchIdForMintKind({ kind: "cuid" }); expect(r.id.length).toBe(25); expect(classifyKind(r.id)).toBe("cuid"); expect(classifyKind(r.friendlyId)).toBe("cuid"); @@ -20,21 +28,26 @@ describe("batchIdForMintKind (pure)", () => { it("never mints a 21-char id", () => { for (const kind of ["cuid", "runOpsId"] as const) { - expect([25, 26]).toContain(batchIdForMintKind(kind).id.length); + expect([25, 26]).toContain(batchIdForMintKind({ kind }).id.length); } }); }); describe("resolveBatchMintKind", () => { const environment = { organizationId: "org_1", id: "env_1", orgFeatureFlags: {} }; + const NEW_PARENT = `run_${"a".repeat(24)}01`; + const LEGACY_PARENT = `run_${"a".repeat(25)}`; + const GEN2_PARENT = `run_${"a".repeat(24)}a2`; it("ROOT batch (no parent) resolves per-org kind via resolveRunIdMintKind", async () => { const resolveRunIdMintKind = vi.fn().mockResolvedValue("runOpsId"); - const kind = await resolveBatchMintKind({ + const resolveMintShard = vi.fn().mockResolvedValue("new"); + const target = await resolveBatchMintKind({ environment, - deps: { resolveRunIdMintKind }, + deps: { resolveRunIdMintKind, resolveMintShard }, }); - expect(kind).toBe("runOpsId"); + expect(target.kind).toBe("runOpsId"); + expect(target.shardChar).toBeUndefined(); expect(resolveRunIdMintKind).toHaveBeenCalledWith({ organizationId: "org_1", id: "env_1", @@ -42,66 +55,96 @@ describe("resolveBatchMintKind", () => { }); }); + it("ROOT batch mints by the mint policy when a shard is active", async () => { + const target = await resolveBatchMintKind({ + environment, + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"), + resolveMintShard: vi.fn().mockResolvedValue("a"), + }, + }); + expect(target).toEqual({ kind: "runOpsId", shardChar: "a", region: undefined }); + }); + it("ROOT batch on a non-cut-over org -> cuid", async () => { - const resolveRunIdMintKind = vi.fn().mockResolvedValue("cuid"); - const kind = await resolveBatchMintKind({ + const target = await resolveBatchMintKind({ environment, - deps: { resolveRunIdMintKind }, + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("cuid"), + resolveMintShard: vi.fn(), + }, }); - expect(kind).toBe("cuid"); + expect(target.kind).toBe("cuid"); }); it("CHILD batch inherits a run-ops (NEW) parent by id-shape", async () => { - const parentRunFriendlyId = `run_${"a".repeat(24) + "01"}`; const resolveRunIdMintKind = vi.fn(); - - const kind = await resolveBatchMintKind({ + const target = await resolveBatchMintKind({ environment, - parentRunFriendlyId, - deps: { resolveRunIdMintKind }, + parentRunFriendlyId: NEW_PARENT, + deps: { resolveRunIdMintKind, resolveMintShard: vi.fn() }, }); + expect(target).toEqual({ kind: "runOpsId" }); + expect(resolveRunIdMintKind).not.toHaveBeenCalled(); + }); - expect(kind).toBe("runOpsId"); + it("CHILD batch takes a gen-2 parent's shard char", async () => { + const resolveRunIdMintKind = vi.fn(); + const resolveMintShard = vi.fn(); + const target = await resolveBatchMintKind({ + environment, + parentRunFriendlyId: GEN2_PARENT, + deps: { resolveRunIdMintKind, resolveMintShard }, + }); + expect(target).toEqual({ kind: "runOpsId", shardChar: "a" }); expect(resolveRunIdMintKind).not.toHaveBeenCalled(); + expect(resolveMintShard).not.toHaveBeenCalled(); }); it("CHILD batch inherits a cuid (LEGACY) parent by id-shape", async () => { - const parentRunFriendlyId = `run_${"a".repeat(25)}`; const resolveRunIdMintKind = vi.fn(); - - const kind = await resolveBatchMintKind({ + const target = await resolveBatchMintKind({ environment, - parentRunFriendlyId, - deps: { resolveRunIdMintKind }, + parentRunFriendlyId: LEGACY_PARENT, + deps: { resolveRunIdMintKind, resolveMintShard: vi.fn() }, }); - - expect(kind).toBe("cuid"); + expect(target).toEqual({ kind: "cuid" }); expect(resolveRunIdMintKind).not.toHaveBeenCalled(); }); // mint-on-FLIP invariant: a child follows its parent's store even after the org flag // flips the other way. The flag resolver must NEVER be consulted for a child. it("FLIP 'cuid'->'runOpsId': a cuid (LEGACY) parent still mints a cuid child though the flag now says 'runOpsId'", async () => { - const parentRunFriendlyId = `run_${"a".repeat(25)}`; const resolveRunIdMintKind = vi.fn().mockResolvedValue("runOpsId"); // flag flipped to runOpsId - const kind = await resolveBatchMintKind({ + const target = await resolveBatchMintKind({ environment, - parentRunFriendlyId, - deps: { resolveRunIdMintKind }, + parentRunFriendlyId: LEGACY_PARENT, + deps: { resolveRunIdMintKind, resolveMintShard: vi.fn() }, }); - expect(kind).toBe("cuid"); + expect(target).toEqual({ kind: "cuid" }); expect(resolveRunIdMintKind).not.toHaveBeenCalled(); }); it("FLIP 'runOpsId'->'cuid': a run-ops (NEW) parent still mints a run-ops child though the flag now says 'cuid'", async () => { - const parentRunFriendlyId = `run_${"a".repeat(24) + "01"}`; const resolveRunIdMintKind = vi.fn().mockResolvedValue("cuid"); // flag flipped back to cuid - const kind = await resolveBatchMintKind({ + const target = await resolveBatchMintKind({ environment, - parentRunFriendlyId, - deps: { resolveRunIdMintKind }, + parentRunFriendlyId: NEW_PARENT, + deps: { resolveRunIdMintKind, resolveMintShard: vi.fn() }, }); - expect(kind).toBe("runOpsId"); + expect(target).toEqual({ kind: "runOpsId" }); expect(resolveRunIdMintKind).not.toHaveBeenCalled(); }); + + it("FLIP does not move a gen-2 child off its parent's shard", async () => { + const target = await resolveBatchMintKind({ + environment, + parentRunFriendlyId: GEN2_PARENT, + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("cuid"), + resolveMintShard: vi.fn().mockResolvedValue("b"), + }, + }); + expect(target).toEqual({ kind: "runOpsId", shardChar: "a" }); + }); }); diff --git a/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.ts b/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.ts index e2d8511e3ff..b08d9b9b33f 100644 --- a/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.ts @@ -1,45 +1,37 @@ -import { BatchId, generateRunOpsId } from "@trigger.dev/core/v3/isomorphic"; -import { - resolveRunIdMintKind as defaultResolveRunIdMintKind, - type RunIdMintKind, -} from "~/v3/engineVersion.server"; -import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server"; +import { BatchId, generateRunOpsId, generateRunOpsIdV2 } from "@trigger.dev/core/v3/isomorphic"; +import type { MintTarget } from "./mintTarget"; +import { resolveRunMintTarget, type RunMintDeps } from "./resolveRunMintTarget.server"; -type ResolveDeps = { - resolveRunIdMintKind: typeof defaultResolveRunIdMintKind; -}; +export function batchIdForMintKind(target: MintTarget): { id: string; friendlyId: string } { + if (target.kind !== "runOpsId") { + return BatchId.generate(); + } -const defaultDeps: ResolveDeps = { - resolveRunIdMintKind: defaultResolveRunIdMintKind, -}; + const id = target.shardChar + ? generateRunOpsIdV2(target.shardChar) + : generateRunOpsId(target.region); -export function batchIdForMintKind(kind: RunIdMintKind): { id: string; friendlyId: string } { - if (kind === "runOpsId") { - const id = generateRunOpsId(); - return { id, friendlyId: BatchId.toFriendlyId(id) }; - } - return BatchId.generate(); + return { id, friendlyId: BatchId.toFriendlyId(id) }; } +// A batch anchors on the parent RUN's id, never on another batch, and every call site +// passes that id optionally — so one call serves a root batch and a child batch. export async function resolveBatchMintKind(args: { environment: { organizationId: string; id: string; orgFeatureFlags?: unknown }; parentRunFriendlyId?: string; - deps?: Partial; -}): Promise { - const deps = { ...defaultDeps, ...args.deps }; - return args.parentRunFriendlyId - ? resolveInheritedMintKind(args.parentRunFriendlyId) - : deps.resolveRunIdMintKind({ - organizationId: args.environment.organizationId, - id: args.environment.id, - orgFeatureFlags: args.environment.orgFeatureFlags, - }); + deps?: Partial; +}): Promise { + return resolveRunMintTarget({ + environment: args.environment, + parentRunFriendlyId: args.parentRunFriendlyId, + deps: args.deps, + }); } export async function mintBatchFriendlyId(args: { environment: { organizationId: string; id: string; orgFeatureFlags?: unknown }; parentRunFriendlyId?: string; - deps?: Partial; + deps?: Partial; }): Promise<{ id: string; friendlyId: string }> { return batchIdForMintKind(await resolveBatchMintKind(args)); } diff --git a/apps/webapp/app/v3/runOpsMigration/mintTarget.ts b/apps/webapp/app/v3/runOpsMigration/mintTarget.ts new file mode 100644 index 00000000000..94355d55462 --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/mintTarget.ts @@ -0,0 +1,11 @@ +import type { ResidencyKind } from "@trigger.dev/core/v3/isomorphic"; + +/** + * Where one mint lands. `shardChar` and `region` both occupy index 24 of a run-ops id, so + * they travel together and cannot disagree. `shardChar` set means gen-2, region ignored. + */ +export type MintTarget = { + kind: ResidencyKind; + shardChar?: string; + region?: string; +}; diff --git a/apps/webapp/app/v3/runOpsMigration/resolveInheritedMintKind.server.test.ts b/apps/webapp/app/v3/runOpsMigration/resolveInheritedMintKind.server.test.ts index 3f135793f84..570cc496182 100644 --- a/apps/webapp/app/v3/runOpsMigration/resolveInheritedMintKind.server.test.ts +++ b/apps/webapp/app/v3/runOpsMigration/resolveInheritedMintKind.server.test.ts @@ -1,15 +1,68 @@ import { describe, expect, it } from "vitest"; import { resolveInheritedMintKind } from "./resolveInheritedMintKind.server"; +import { mintFriendlyIdForKind } from "./mintAnchoredRunFriendlyId.server"; -const NEW_PARENT = `run_${"a".repeat(24) + "01"}`; // run-ops id-shape -> NEW +const NEW_PARENT = `run_${"a".repeat(24)}01`; // run-ops v1 id-shape -> NEW const LEGACY_PARENT = `run_${"b".repeat(25)}`; // cuid id-shape -> LEGACY +const GEN2_PARENT = `run_${"a".repeat(24)}a2`; // gen-2, shard "a" describe("resolveInheritedMintKind (pure id-shape, shared across all mint paths)", () => { - it("inherits a run-ops (NEW) parent by id-shape -> 'runOpsId' kind", () => { - expect(resolveInheritedMintKind(NEW_PARENT)).toBe("runOpsId"); + it("inherits a run-ops (NEW) parent by id-shape -> runOpsId with NO shard char", () => { + expect(resolveInheritedMintKind(NEW_PARENT)).toEqual({ kind: "runOpsId" }); }); it("inherits a cuid (LEGACY) parent by id-shape -> cuid", () => { - expect(resolveInheritedMintKind(LEGACY_PARENT)).toBe("cuid"); + expect(resolveInheritedMintKind(LEGACY_PARENT)).toEqual({ kind: "cuid" }); + }); + + it("inherits a gen-2 parent's shard char, never a freshly resolved one", () => { + expect(resolveInheritedMintKind(GEN2_PARENT)).toEqual({ kind: "runOpsId", shardChar: "a" }); + }); + + it("accepts the bare internal form", () => { + expect(resolveInheritedMintKind(GEN2_PARENT.slice(4))).toEqual({ + kind: "runOpsId", + shardChar: "a", + }); + }); +}); + +describe("mintFriendlyIdForKind", () => { + it("a shard char mints a gen-2 id with that char at index 24 and '2' at 25", () => { + const body = mintFriendlyIdForKind({ kind: "runOpsId", shardChar: "a" }).slice("run_".length); + expect(body.length).toBe(26); + expect(body[24]).toBe("a"); + expect(body[25]).toBe("2"); + }); + + it("a shard char wins over a region: index 24 has ONE source", () => { + const body = mintFriendlyIdForKind({ + kind: "runOpsId", + shardChar: "a", + region: "us-east-1", + }).slice("run_".length); + expect(body[24]).toBe("a"); // not "e", the us-east-1 region char + }); + + it("no shard char mints a gen-1 v1 id, stamping the region as today", () => { + const body = mintFriendlyIdForKind({ kind: "runOpsId", region: "us-east-1" }).slice(4); + expect(body[24]).toBe("e"); + expect(body[25]).toBe("1"); + }); + + it("no shard char and no region mints a gen-1 v1 id with the default region char", () => { + const body = mintFriendlyIdForKind({ kind: "runOpsId" }).slice(4); + expect(body[24]).toBe("0"); + expect(body[25]).toBe("1"); + }); + + it("cuid kind mints a 25-char cuid", () => { + expect(mintFriendlyIdForKind({ kind: "cuid" }).slice(4).length).toBe(25); + }); + + it("an end-to-end inherit-then-mint keeps a child on the parent's shard", () => { + const body = mintFriendlyIdForKind(resolveInheritedMintKind(GEN2_PARENT)).slice(4); + expect(body[24]).toBe("a"); + expect(body[25]).toBe("2"); }); }); diff --git a/apps/webapp/app/v3/runOpsMigration/resolveInheritedMintKind.server.ts b/apps/webapp/app/v3/runOpsMigration/resolveInheritedMintKind.server.ts index 6ec9583c94b..825d910d7d9 100644 --- a/apps/webapp/app/v3/runOpsMigration/resolveInheritedMintKind.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/resolveInheritedMintKind.server.ts @@ -1,10 +1,16 @@ -import { ownerEngine } from "@trigger.dev/core/v3/isomorphic"; -import type { RunIdMintKind } from "./runOpsMintKind.server"; +import { resolveShard } from "@trigger.dev/core/v3/isomorphic"; +import type { MintTarget } from "./mintTarget"; // Mint a child in the SAME physical store as its anchor (parent run / owning batch), // regardless of the org's current mint flag — keeps a subgraph co-resident across a // flip. With no migration/drain, residency is a pure id-shape check (zero hot-path // I/O): a run-ops (NEW) parent mints run-ops children, a cuid (LEGACY) parent mints cuid. -export function resolveInheritedMintKind(parentRunFriendlyId: string): RunIdMintKind { - return ownerEngine(parentRunFriendlyId) === "NEW" ? "runOpsId" : "cuid"; +// A gen-2 parent hands down its OWN shard char, never a freshly resolved one: two runs in +// one tree must never split across shards. +export function resolveInheritedMintKind(parentRunFriendlyId: string): MintTarget { + const shard = resolveShard(parentRunFriendlyId); + + if (shard === "legacy") return { kind: "cuid" }; + if (shard === "new") return { kind: "runOpsId" }; + return { kind: "runOpsId", shardChar: shard }; } diff --git a/apps/webapp/app/v3/runOpsMigration/resolveRunMintTarget.server.test.ts b/apps/webapp/app/v3/runOpsMigration/resolveRunMintTarget.server.test.ts new file mode 100644 index 00000000000..a71fdcc2b5f --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/resolveRunMintTarget.server.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it, vi } from "vitest"; +import { resolveRunMintTarget } from "./resolveRunMintTarget.server"; + +const environment = { organizationId: "org_1", id: "env_1", orgFeatureFlags: {} }; +const GEN2_PARENT = `run_${"a".repeat(24)}a2`; +const LEGACY_PARENT = `run_${"b".repeat(25)}`; + +describe("resolveRunMintTarget — root", () => { + it("resolves the kind, then the shard, and returns both", async () => { + const resolveRunIdMintKind = vi.fn().mockResolvedValue("runOpsId"); + const resolveMintShard = vi.fn().mockResolvedValue("a"); + + const target = await resolveRunMintTarget({ + environment, + deps: { resolveRunIdMintKind, resolveMintShard }, + }); + + expect(target).toEqual({ kind: "runOpsId", shardChar: "a", region: undefined }); + expect(resolveMintShard).toHaveBeenCalledWith({ id: "env_1", orgFeatureFlags: {} }); + }); + + it("a 'new' shard result carries NO shard char, so the mint stays gen-1", async () => { + const target = await resolveRunMintTarget({ + environment, + region: "us-east-1", + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"), + resolveMintShard: vi.fn().mockResolvedValue("new"), + }, + }); + expect(target).toEqual({ kind: "runOpsId", region: "us-east-1" }); + }); + + it("never resolves a shard when the kind is cuid", async () => { + const resolveMintShard = vi.fn(); + const target = await resolveRunMintTarget({ + environment, + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("cuid"), + resolveMintShard, + }, + }); + expect(target).toEqual({ kind: "cuid" }); + expect(resolveMintShard).not.toHaveBeenCalled(); + }); +}); + +describe("resolveRunMintTarget — child", () => { + it("inherits a gen-2 parent's shard and consults NEITHER resolver", async () => { + const resolveRunIdMintKind = vi.fn(); + const resolveMintShard = vi.fn(); + + const target = await resolveRunMintTarget({ + environment, + parentRunFriendlyId: GEN2_PARENT, + deps: { resolveRunIdMintKind, resolveMintShard }, + }); + + expect(target).toEqual({ kind: "runOpsId", shardChar: "a" }); + expect(resolveRunIdMintKind).not.toHaveBeenCalled(); + expect(resolveMintShard).not.toHaveBeenCalled(); + }); + + it("a cuid parent still yields cuid though the flag now says runOpsId", async () => { + const target = await resolveRunMintTarget({ + environment, + parentRunFriendlyId: LEGACY_PARENT, + deps: { + resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"), + resolveMintShard: vi.fn().mockResolvedValue("a"), + }, + }); + expect(target).toEqual({ kind: "cuid" }); + }); +}); diff --git a/apps/webapp/app/v3/runOpsMigration/resolveRunMintTarget.server.ts b/apps/webapp/app/v3/runOpsMigration/resolveRunMintTarget.server.ts new file mode 100644 index 00000000000..6b421ae0d4b --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/resolveRunMintTarget.server.ts @@ -0,0 +1,53 @@ +import type { MintTarget } from "./mintTarget"; +import { resolveInheritedMintKind } from "./resolveInheritedMintKind.server"; +import { resolveRunIdMintKind as defaultResolveRunIdMintKind } from "./runOpsMintKind.server"; +import { resolveMintShard as defaultResolveMintShard } from "./runOpsMintShard.server"; + +export type RunMintDeps = { + resolveRunIdMintKind: typeof defaultResolveRunIdMintKind; + resolveMintShard: typeof defaultResolveMintShard; +}; + +const defaultDeps: RunMintDeps = { + resolveRunIdMintKind: defaultResolveRunIdMintKind, + resolveMintShard: defaultResolveMintShard, +}; + +/** + * Where one run mints. The second stage runs only for a root run already on the run-ops path: a + * child inherits its parent's shard by id-shape, so a tree never splits across shards. + */ +export async function resolveRunMintTarget(args: { + environment: { organizationId: string; id: string; orgFeatureFlags?: unknown }; + parentRunFriendlyId?: string; + region?: string; + deps?: Partial; +}): Promise { + if (args.parentRunFriendlyId) { + // The region still travels: it takes index 24 for an inherited gen-1 parent, and a gen-2 + // parent's shardChar outranks it. + return { ...resolveInheritedMintKind(args.parentRunFriendlyId), region: args.region }; + } + + const deps = { ...defaultDeps, ...args.deps }; + + const kind = await deps.resolveRunIdMintKind({ + organizationId: args.environment.organizationId, + id: args.environment.id, + orgFeatureFlags: args.environment.orgFeatureFlags, + }); + + if (kind !== "runOpsId") { + return { kind }; + } + + const shard = await deps.resolveMintShard({ + id: args.environment.id, + orgFeatureFlags: args.environment.orgFeatureFlags, + }); + + // A reserved key means gen-1, the state of every deployment with no shard configured. + return shard === "new" || shard === "legacy" + ? { kind, region: args.region } + : { kind, shardChar: shard, region: args.region }; +} diff --git a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts index c1c2b9ddd48..422af3712e9 100644 --- a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts @@ -69,14 +69,19 @@ function reportOverrideRejected(info: { override: string; activeSet: string[] }) /** * Which shard an environment mints new roots into. Call only after resolveRunIdMintKind has * returned "runOpsId". Returns "new" to mean a gen-1 run-ops id, which is today's behaviour. - * - * @knipignore the gen-2 write-path change is the first production caller; drop this tag there. */ export async function resolveMintShard(environment: { id: string; // Pass environment.organization.featureFlags from the trigger call site. orgFeatureFlags?: unknown; }): Promise { + // No shard descriptor means no shard can ever be minted into, so answer before reading + // anything: an unconfigured deployment keeps exactly today's code path, with no + // control-plane query on the trigger path, no cache write and no log line. + if (env.RUN_OPS_SHARDS.length === 0) { + return "new"; + } + return resolveMintShardWith(environment, { readFlags: readSetFlags, cache: liveCache, diff --git a/apps/webapp/app/v3/services/batchTriggerV3.server.ts b/apps/webapp/app/v3/services/batchTriggerV3.server.ts index 563ef446bcc..17a3bbb60d3 100644 --- a/apps/webapp/app/v3/services/batchTriggerV3.server.ts +++ b/apps/webapp/app/v3/services/batchTriggerV3.server.ts @@ -362,15 +362,20 @@ export class BatchTriggerV3Service extends BaseService { anchorFriendlyId?: string, region?: string ): Promise { - const mintKind = anchorFriendlyId + // Deliberately not routed through resolveRunMintTarget: the root arm below is + // unreachable in production (every call site passes an anchor), and resolveMintKind is + // injected so a test can drive that arm without a database. + const target = anchorFriendlyId ? resolveInheritedMintKind(anchorFriendlyId) - : await this.resolveMintKind({ - organizationId: environment.organizationId, - id: environment.id, - orgFeatureFlags: environment.organization.featureFlags, - }); + : { + kind: await this.resolveMintKind({ + organizationId: environment.organizationId, + id: environment.id, + orgFeatureFlags: environment.organization.featureFlags, + }), + }; - return mintFriendlyIdForKind(mintKind, region); + return mintFriendlyIdForKind({ ...target, region }); } async #prepareRunData( diff --git a/apps/webapp/test/engine/triggerFailedTask.withoutTraceEvents.test.ts b/apps/webapp/test/engine/triggerFailedTask.withoutTraceEvents.test.ts index a0be900fb82..eac5d75ec8a 100644 --- a/apps/webapp/test/engine/triggerFailedTask.withoutTraceEvents.test.ts +++ b/apps/webapp/test/engine/triggerFailedTask.withoutTraceEvents.test.ts @@ -90,4 +90,53 @@ describe("TriggerFailedTaskService — failed run residency (callWithoutTraceEve await engine.quit(); } ); + + containerTest( + "a pre-minted runFriendlyId passes through untouched", + async ({ prisma, redisOptions }) => { + const engine = makeEngine(prisma, redisOptions); + const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); + const taskIdentifier = "failed-residency-passthrough"; + await setupBackgroundWorker(engine, environment, taskIdentifier); + + const parentFriendlyId = RunId.toFriendlyId(generateRunOpsId()); + await engine.trigger( + { + friendlyId: parentFriendlyId, + environment, + taskIdentifier, + payload: "{}", + payloadType: "application/json", + traceId: "00000000000000000000000000000000", + spanId: "0000000000000000", + workerQueue: "main", + queue: `task/${taskIdentifier}`, + isTest: false, + tags: [], + } as any, + prisma + ); + + // A batch item arrives with its id already minted from the BATCH. Re-resolving it + // here would move the item off its batch's shard, so the pass-through has to win + // over the mint-target resolver. + const preMinted = RunId.toFriendlyId(generateRunOpsId()); + + const friendlyId = await makeService(prisma, engine).callWithoutTraceEvents({ + environmentId: environment.id, + environmentType: environment.type, + projectId: environment.projectId, + organizationId: environment.organizationId, + taskId: taskIdentifier, + payload: { test: "passthrough" }, + errorMessage: "boom", + parentRunId: parentFriendlyId, + runFriendlyId: preMinted, + }); + + expect(friendlyId).toBe(preMinted); + + await engine.quit(); + } + ); }); diff --git a/apps/webapp/test/runEngineHandlers.test.ts b/apps/webapp/test/runEngineHandlers.test.ts index 2c57d87506e..fda077e026c 100644 --- a/apps/webapp/test/runEngineHandlers.test.ts +++ b/apps/webapp/test/runEngineHandlers.test.ts @@ -490,6 +490,100 @@ describe("runEngineHandlers batch completion", () => { }); describe("runEngineHandlers batch residency routing", () => { + // See resolveBatchRunOpsWriter: without a shard arm a gen-2 batch resolves to a store holding + // no such row, and the parent waits forever with nothing logged. + // Real databases, so the assertion is where the rows landed rather than which object came back. + // The shard is prisma14 and both gen-1 slots are prisma17, so every wrong resolution lands on a + // database holding no such batch. + heteroPostgresTest( + "a gen-2 batch commits on its shard, and the gen-1 store stays empty", + async ({ prisma14, prisma17 }) => { + const shardSeed = await seedEnvironment(prisma14, "g2shard"); + const gen2BatchId = `${"a".repeat(24)}a2`; + await seedBatch(prisma14, { + id: gen2BatchId, + friendlyId: `batch_${gen2BatchId}`, + runtimeEnvironmentId: shardSeed.environment.id, + }); + + const shards = [{ key: "a", writer: prisma14 }] as const; + + const writer = await resolveBatchRunOpsWriter(gen2BatchId, { + newReplica: prisma17, + newWriter: prisma17, + legacyWriter: prisma17, + shards: shards as never, + }); + expect(writer).toBe(prisma14); + + let completed: string | undefined; + await handleBatchCompletion( + { + batchId: gen2BatchId, + runIds: ["run_friendly_1"], + successfulRunCount: 1, + failedRunCount: 1, + failures: [failure(0, "TRIGGER_ERROR")], + }, + { + splitEnabled: true, + newReplica: prisma17, + newWriter: prisma17, + legacyWriter: prisma17, + shards: shards as never, + tryCompleteBatch: async (id) => { + completed = id; + }, + } + ); + + // The hang was the callback dying on "no record was found for an update" before this. + const onShard = await prisma14.batchTaskRun.findFirstOrThrow({ where: { id: gen2BatchId } }); + expect(onShard.status).toBe("PARTIAL_FAILED"); + expect( + await prisma14.batchTaskRunError.findMany({ where: { batchTaskRunId: gen2BatchId } }) + ).toHaveLength(1); + expect(completed).toBe(gen2BatchId); + + expect(await prisma17.batchTaskRun.findMany({ where: { id: gen2BatchId } })).toHaveLength(0); + expect( + await prisma17.batchTaskRunError.findMany({ where: { batchTaskRunId: gen2BatchId } }) + ).toHaveLength(0); + } + ); + + // A throwing double deliberately: this asserts a call that must NOT happen, and a real client + // would return null and pass either way. + it("a gen-2 batch id never probes the gen-1 store", async () => { + const shardWriter = {} as never; + + const writer = await resolveBatchRunOpsWriter(`${"a".repeat(24)}a2`, { + newReplica: { + batchTaskRun: { + findFirst: async () => { + throw new Error("a gen-2 batch id must never probe the NEW store"); + }, + }, + } as never, + newWriter: {} as never, + legacyWriter: {} as never, + shards: [{ key: "a", writer: shardWriter as never }], + }); + + expect(writer).toBe(shardWriter); + }); + + it("an unconfigured shard key fails loud rather than writing elsewhere", async () => { + await expect( + resolveBatchRunOpsWriter(`${"a".repeat(24)}z2`, { + newReplica: {} as never, + newWriter: {} as never, + legacyWriter: {} as never, + shards: [{ key: "a", writer: {} as never }], + }) + ).rejects.toThrow(/shard/i); + }); + // True single-DB invariant: the topology's cpFallback makes newReplica and // legacyWriter the SAME control-plane client, so the probe always resolves to // that one client regardless of where length-classification would guess. diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index ccfb60ca4d6..4a4cdeb99a7 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -26,7 +26,8 @@ import { generateInternalId, parseNaturalLanguageDurationInMs, RunId, - WaitpointId, + mintWaitpointIdFor, + type ShardKey, } from "@trigger.dev/core/v3/isomorphic"; import { type PrismaClient, @@ -1087,6 +1088,7 @@ export class RunEngine { ? this.waitpointSystem.buildRunAssociatedWaitpoint({ projectId: environment.project.id, environmentId: environment.id, + anchorRunId: taskRunId, }) : undefined, }, @@ -1373,6 +1375,7 @@ export class RunEngine { ? this.waitpointSystem.buildRunAssociatedWaitpoint({ projectId: environment.project.id, environmentId: environment.id, + anchorRunId: taskRunId, }) : undefined; @@ -1807,6 +1810,7 @@ export class RunEngine { timeout, tags, standaloneResidency, + standaloneShardKey, }: { /** The run that will block on this waitpoint. Co-locates the waitpoint with the run's DB. */ runId?: string; @@ -1818,6 +1822,7 @@ export class RunEngine { tags?: string[]; /** Standalone-token residency (no owning run) from the env mint kind; ignored when `runId` is set. */ standaloneResidency?: "NEW" | "LEGACY"; + standaloneShardKey?: ShardKey; }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> { return this.waitpointSystem.createManualWaitpoint({ runId, @@ -1828,6 +1833,7 @@ export class RunEngine { timeout, tags, standaloneResidency, + standaloneShardKey, }); } @@ -1853,7 +1859,9 @@ export class RunEngine { const waitpoint = await this.runStore.createWaitpoint( { data: { - ...WaitpointId.generate(), + // Stamped from the batch, not the blocked run: this create passes only + // completedByBatchId, so that is the owner the router validates the stamp against. + ...mintWaitpointIdFor(batchId), type: "BATCH", idempotencyKey: batchId, userProvidedIdempotencyKey: false, diff --git a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts index 3dbed999445..9715a89cb9a 100644 --- a/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/waitpointSystem.ts @@ -1,4 +1,5 @@ import { timeoutError } from "@trigger.dev/core/v3"; +import type { ShardKey } from "@trigger.dev/core/v3/isomorphic"; import type { PrismaClientOrTransaction, TaskRun, @@ -184,6 +185,7 @@ export class WaitpointSystem { timeout, tags, standaloneResidency, + standaloneShardKey, }: { runId?: string; environmentId: string; @@ -196,6 +198,7 @@ export class WaitpointSystem { // the token lands on the run-ops DB (NEW) in a fully-minted-new deployment instead of defaulting // to LEGACY by its cuid id-shape. Ignored when `runId` is set (co-location wins). standaloneResidency?: "NEW" | "LEGACY"; + standaloneShardKey?: ShardKey; }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> { const result = await this.coordinator.createManualWaitpoint({ runId, @@ -206,6 +209,7 @@ export class WaitpointSystem { timeout, tags, standaloneResidency, + standaloneShardKey, }); if (result.kind === "cached") { @@ -721,11 +725,17 @@ export class WaitpointSystem { public buildRunAssociatedWaitpoint({ projectId, environmentId, + anchorRunId, }: { projectId: string; environmentId: string; + anchorRunId: string; }) { - return this.coordinator.mintAssociatedWaitpointData({ projectId, environmentId }); + return this.coordinator.mintAssociatedWaitpointData({ + projectId, + environmentId, + anchorRunId, + }); } /** @@ -807,7 +817,11 @@ export class WaitpointSystem { const snapshot = await getLatestExecutionSnapshot(prisma, runId, this.$.runStore); // Create waitpoint and link to run atomically - const waitpointData = this.buildRunAssociatedWaitpoint({ projectId, environmentId }); + const waitpointData = this.buildRunAssociatedWaitpoint({ + projectId, + environmentId, + anchorRunId: runId, + }); const waitpoint = await this.coordinator.createAssociatedWaitpoint({ runId, diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts index d1e48fa4f8d..4877075100b 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts @@ -1,6 +1,6 @@ import type { RunStore } from "@internal/run-store"; import { tryCatch } from "@trigger.dev/core/v3"; -import { WaitpointId } from "@trigger.dev/core/v3/isomorphic"; +import { mintWaitpointIdFor, mintWaitpointIdForShard } from "@trigger.dev/core/v3/isomorphic"; import type { Logger } from "@trigger.dev/core/logger"; import type { PrismaClient, Waitpoint } from "@trigger.dev/database"; import { boundedIn, Prisma } from "@trigger.dev/database"; @@ -239,6 +239,7 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator // The two `nanoid(24)` calls below are deliberately separate and produce DIFFERENT values: // the upsert `where` key must not match the `create` key, or a guaranteed-miss upsert becomes // a possible update. Do not hoist either to a shared constant. + // Stamped for the anchor run's shard, so the row is routable and completion needs no probe. const upsertArgs = { where: { environmentId_idempotencyKey: { @@ -247,7 +248,7 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator }, }, create: { - ...WaitpointId.generate(), + ...mintWaitpointIdFor(runId), type: "DATETIME" as const, idempotencyKey: idempotencyKey ?? nanoid(24), idempotencyKeyExpiresAt, @@ -272,6 +273,7 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator timeout, tags, standaloneResidency, + standaloneShardKey, }: CreateManualWaitpointParams): Promise { // Co-location invariant (see createDateTimeWaitpoint): when a `runId` is supplied the waitpoint // co-locates with that run's DB and the (env,idempotencyKey) dedup is per-run (co-resident). A @@ -279,11 +281,17 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator // owner, blocked later by whichever run waits on it (possibly cross-DB, resolved by the // run-co-resident block edge + completion fan-out). With no owner it reads the env mint kind via // `standaloneResidency` so a minted-new env keeps its tokens on NEW; unset, it routes by id-shape. No tx here. + // A gen-2 standalone token carries its shard in its own id, so it passes no residency hint. + const standaloneShard = runId ? undefined : standaloneShardKey; + const isGen2Standalone = + standaloneShard !== undefined && standaloneShard !== "new" && standaloneShard !== "legacy"; const colocate = runId ? { coLocateWithRunId: runId } - : standaloneResidency - ? { residency: standaloneResidency } - : undefined; + : isGen2Standalone + ? undefined + : standaloneResidency + ? { residency: standaloneResidency } + : undefined; const existingWaitpoint = idempotencyKey ? await this.runStore.findWaitpoint( { @@ -330,8 +338,9 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator while (attempts < maxRetries) { try { // As in createDateTimeWaitpoint, the two `nanoid(24)` calls are deliberately separate and - // differ. Both, and `WaitpointId.generate()`, are re-evaluated on every attempt: that is - // what makes a retry after a unique-constraint conflict try a fresh key. + // differ. Both, and the id mint, are re-evaluated on every attempt: that is what makes a + // retry after a unique-constraint conflict try a fresh key. The anchor does not change, + // so every attempt stays on the same shard. const waitpoint = await this.runStore.upsertWaitpoint( { where: { @@ -341,7 +350,9 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator }, }, create: { - ...WaitpointId.generate(), + ...(standaloneShard !== undefined + ? mintWaitpointIdForShard(standaloneShard) + : mintWaitpointIdFor(runId)), type: "MANUAL", idempotencyKey: idempotencyKey ?? nanoid(24), idempotencyKeyExpiresAt, @@ -379,12 +390,14 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator mintAssociatedWaitpointData({ projectId, environmentId, + anchorRunId, }: { projectId: string; environmentId: string; + anchorRunId: string; }): AssociatedWaitpointData { return { - ...WaitpointId.generate(), + ...mintWaitpointIdFor(anchorRunId), type: "RUN" as const, status: "PENDING" as const, idempotencyKey: nanoid(24), diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts index 8a50abb7d1c..80463b1cd1f 100644 --- a/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/types.ts @@ -1,5 +1,6 @@ import type { ReadClient } from "@internal/run-store"; import type { PrismaClientOrTransaction, Waitpoint } from "@trigger.dev/database"; +import type { ShardKey } from "@trigger.dev/core/v3/isomorphic"; /** * The waitpoint and edge state operations that `WaitpointSystem` delegates. @@ -24,6 +25,8 @@ export type WaitpointCoordinator = { mintAssociatedWaitpointData(params: { projectId: string; environmentId: string; + /** Names the shard the row lands on. This write skips the router's stamp check. */ + anchorRunId: string; }): AssociatedWaitpointData; createAssociatedWaitpoint(params: { runId: string; @@ -108,7 +111,13 @@ export type CreateWaitpointResult = | { kind: "created"; waitpoint: Waitpoint }; export type CreateDateTimeWaitpointParams = { - /** When set, the waitpoint co-locates with this run's DB and the dedup probe targets it. */ + /** + * When set, the waitpoint co-locates with this run's DB and the dedup probe targets it. + * + * Every production caller supplies it, and there is deliberately no standalone arm. Omitting it + * on a gen-2 environment mints a cuid and lands the row on a gen-1 store, silently. A standalone + * caller needs a shard hint here first, as `createManualWaitpointParams` has. + */ runId?: string; projectId: string; environmentId: string; @@ -130,6 +139,11 @@ export type CreateManualWaitpointParams = { * full rationale. Only a Postgres implementation reads this. */ standaloneResidency?: "NEW" | "LEGACY"; + /** + * The environment's mint shard, for a standalone token with no owning run. When it names a gen-2 + * shard the implementation must ignore `standaloneResidency`, which can only name a gen-1 store. + */ + standaloneShardKey?: ShardKey; }; /** The RUN-waitpoint row data. Pure — no store touch — so the mint is coordinator-owned. */ diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMint.proof.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMint.proof.test.ts new file mode 100644 index 00000000000..ebd13aeaec0 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMint.proof.test.ts @@ -0,0 +1,126 @@ +import { existsSync, readdirSync, readFileSync, statSync } from "node:fs"; +import path from "node:path"; +import { describe, expect, it } from "vitest"; +import { WAITPOINT_MINT_SITES } from "./waitpointMintCatalog"; + +function repoRoot(): string { + let dir = process.cwd(); + while (!existsSync(path.join(dir, "pnpm-workspace.yaml"))) { + const parent = path.dirname(dir); + if (parent === dir) throw new Error("repo root (pnpm-workspace.yaml) not found"); + dir = parent; + } + return dir; +} + +function read(relative: string): string { + return readFileSync(path.join(repoRoot(), relative), "utf8"); +} + +function count(source: string, pattern: RegExp): number { + return (source.match(pattern) ?? []).length; +} + +// Walked rather than listed, so a mint added in a new file is still visible. Test-support trees +// are excluded: a helper writing through raw Prisma never reaches the routing store. +const TEST_SUPPORT_DIRS = new Set(["tests", "__tests__", "fixtures"]); + +function walk(relativeRoot: string): string[] { + const absolute = path.join(repoRoot(), relativeRoot); + return readdirSync(absolute).flatMap((name) => { + const child = `${relativeRoot}/${name}`; + if (statSync(path.join(absolute, name)).isDirectory()) { + return TEST_SUPPORT_DIRS.has(name) ? [] : walk(child); + } + return name.endsWith(".ts") && !name.includes(".test.") ? [child] : []; + }); +} + +// The mint helpers are the only sanctioned way to produce a Postgres waitpoint id. +const MINT_CALL = /mintWaitpointIdFor(?:Shard)?\(/g; +const UNSTAMPED_MINT = /WaitpointId\.generate\(/g; +const WAITPOINT_WRITE = /waitpoint\.create\(|upsertWaitpoint\(|createWaitpoint\(/g; + +// The catalog holds the mint expressions as string data, so scanning it would count them. +const CATALOG_ITSELF = + "internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintCatalog.ts"; + +const ENGINE_SOURCES = walk("internal-packages/run-engine/src/engine").filter( + (f) => f !== CATALOG_ITSELF +); +const SCANNED = [...ENGINE_SOURCES, "internal-packages/run-store/src/PostgresRunStore.ts"]; + +// expression -> how many times the catalog says it appears in this file +function expectedMints(file: string): Map { + const expected = new Map(); + for (const site of WAITPOINT_MINT_SITES.filter((s) => s.site === file)) { + for (const expr of site.mints) { + expected.set(expr, (expected.get(expr) ?? 0) + 1); + } + } + return expected; +} + +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, "\\$&"); +} + +describe("waitpoint mint census — the catalog matches the source", () => { + it("scans the engine tree and the run-store writer, and finds files to scan", () => { + expect(ENGINE_SOURCES.length).toBeGreaterThan(10); + expect(SCANNED).toContain("internal-packages/run-engine/src/engine/systems/waitpointSystem.ts"); + }); + + // Per expression, not per file, so a swapped anchor fails too and not just a new site. + it.each(SCANNED)("%s has exactly the mint expressions the catalog claims", (file) => { + const source = read(file); + const expected = expectedMints(file); + + for (const [expr, n] of expected) { + expect({ expr, found: count(source, new RegExp(escapeRegExp(expr), "g")) }).toEqual({ + expr, + found: n, + }); + } + + const accounted = [...expected.values()].reduce((a, b) => a + b, 0); + expect(count(source, MINT_CALL)).toBe(accounted); + }); + + it.each(SCANNED)("%s mints no waitpoint id with the un-stamped helper", (file) => { + // Matches inside comments too, deliberately: any textual addition forces a reconcile. + expect(count(read(file), UNSTAMPED_MINT)).toBe(0); + }); + + it.each(SCANNED)("%s writes a waitpoint row only if it is catalogued", (file) => { + // A create with no id is the worst case: @default(cuid()) then mints one after the write. + const writes = count(read(file), WAITPOINT_WRITE); + const catalogued = WAITPOINT_MINT_SITES.some((s) => s.site === file); + expect(writes === 0 || catalogued).toBe(true); + }); + + it("every catalogued site names a file that exists", () => { + for (const site of WAITPOINT_MINT_SITES) { + expect({ site: site.site, exists: existsSync(path.join(repoRoot(), site.site)) }).toEqual({ + site: site.site, + exists: true, + }); + } + }); + + it("every catalogued site names its enclosing symbol in that file", () => { + for (const site of WAITPOINT_MINT_SITES) { + const symbol = site.symbol.split(" ")[0]!.replace("#", ""); + expect({ site: site.id, present: read(site.site).includes(symbol) }).toEqual({ + site: site.id, + present: true, + }); + } + }); + + it("no catalogued symbol is a line number", () => { + for (const site of WAITPOINT_MINT_SITES) { + expect(site.symbol).not.toMatch(/:\d+/); + } + }); +}); diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintCatalog.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintCatalog.ts new file mode 100644 index 00000000000..b42d2d4de9f --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintCatalog.ts @@ -0,0 +1,80 @@ +// Add a site that creates a Postgres `Waitpoint` row and add an entry here, or +// `waitpointMint.proof.test.ts` fails. One entry per site, anchored by symbol, never by line. +// +// A site that mints a cuid for a gen-2 run writes a row the completion path cannot find. Most +// fail loudly, because the router refuses an unstamped id on a gen-2 shard. The RUN row written +// through `createRun` does not: that write is inside the run store, which has no such check. +// +// Pure module: no engine import, no env, no Prisma. +export type WaitpointMintSite = { + id: string; + type: "DATETIME" | "MANUAL" | "RUN" | "BATCH"; + site: string; + /** Enclosing method or symbol name — NEVER a line number. */ + symbol: string; + /** + * Mint expressions this site contains, verbatim, counted per file, so a new mint and a swapped + * anchor both fail until reconciled. Empty for a site writing an id minted elsewhere. + */ + mints: readonly string[]; +}; + +const COORDINATOR = + "internal-packages/run-engine/src/engine/waitpointCoordinator/legacyPostgresCoordinator.ts"; +const ENGINE = "internal-packages/run-engine/src/engine/index.ts"; +const RUN_STORE = "internal-packages/run-store/src/PostgresRunStore.ts"; + +export const WAITPOINT_MINT_SITES: readonly WaitpointMintSite[] = [ + { + id: "coordinator.datetime", + mints: ["mintWaitpointIdFor(runId)"], + type: "DATETIME", + site: COORDINATOR, + symbol: "createDateTimeWaitpoint", + }, + { + id: "coordinator.manual", + mints: ["mintWaitpointIdForShard(standaloneShard)", "mintWaitpointIdFor(runId)"], + type: "MANUAL", + site: COORDINATOR, + symbol: "createManualWaitpoint", + }, + { + id: "coordinator.associated.mint", + mints: ["mintWaitpointIdFor(anchorRunId)"], + type: "RUN", + site: COORDINATOR, + symbol: "mintAssociatedWaitpointData", + }, + { + id: "coordinator.associated.create", + mints: [], + type: "RUN", + site: COORDINATOR, + symbol: "createAssociatedWaitpoint", + }, + { + id: "engine.batch", + mints: ["mintWaitpointIdFor(batchId)"], + type: "BATCH", + site: ENGINE, + symbol: "blockRunWithCreatedBatch", + }, + // The physical writers of the RUN row. They take an already-minted id rather than + // minting one, but they are the writes that bypass the routing store's stamp check, so a + // new writer here must be seen. + { + id: "runStore.createRun.nested", + mints: [], + type: "RUN", + site: RUN_STORE, + symbol: "createRun (nested associatedWaitpoint create)", + }, + { + id: "runStore.createRun.dedicated", + mints: [], + type: "RUN", + site: RUN_STORE, + symbol: "#createAssociatedWaitpoint", + }, +]; diff --git a/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintSites.test.ts b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintSites.test.ts new file mode 100644 index 00000000000..d1f0834d823 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintSites.test.ts @@ -0,0 +1,146 @@ +import type { RunStore } from "@internal/run-store"; +import type { Logger } from "@trigger.dev/core/logger"; +import type { PrismaClient, Waitpoint } from "@trigger.dev/database"; +import { describe, expect, it } from "vitest"; +import { LegacyPostgresWaitpointCoordinator } from "./legacyPostgresCoordinator.js"; + +// These drive the real create sites, not the mint helper: a test calling the helper directly +// passes even when a site stops passing its anchor. +const GEN2_RUN = `${"a".repeat(24)}a2`; +const GEN1_RUN = `${"a".repeat(24)}01`; +const GEN2_BATCH = `${"d".repeat(24)}b2`; + +type Captured = { id?: string; friendlyId?: string }; + +function coordinatorCapturing(captured: Captured) { + const runStore = { + findWaitpoint: async () => null, + upsertWaitpoint: async (args: { create: Captured }) => { + captured.id = args.create.id; + captured.friendlyId = args.create.friendlyId; + return { id: args.create.id } as unknown as Waitpoint; + }, + } as unknown as RunStore; + + return new LegacyPostgresWaitpointCoordinator({ + runStore, + prisma: {} as unknown as PrismaClient, + logger: { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: () => {}, + } as unknown as Logger, + }); +} + +describe("createDateTimeWaitpoint stamps the anchor's shard", () => { + it("a gen-2 run anchor yields a gen-2 waitpoint id", async () => { + const captured: Captured = {}; + await coordinatorCapturing(captured).createDateTimeWaitpoint({ + runId: GEN2_RUN, + projectId: "proj", + environmentId: "env", + completedAfter: new Date(), + }); + + expect(captured.id).toHaveLength(26); + expect(captured.id?.[24]).toBe("a"); + expect(captured.id?.[25]).toBe("2"); + expect(captured.friendlyId).toBe(`waitpoint_${captured.id}`); + }); + + it("a gen-1 run anchor keeps a cuid", async () => { + const captured: Captured = {}; + await coordinatorCapturing(captured).createDateTimeWaitpoint({ + runId: GEN1_RUN, + projectId: "proj", + environmentId: "env", + completedAfter: new Date(), + }); + + expect(captured.id).toHaveLength(25); + }); +}); + +describe("createManualWaitpoint stamps the anchor's shard", () => { + it("a gen-2 run anchor yields a gen-2 waitpoint id", async () => { + const captured: Captured = {}; + await coordinatorCapturing(captured).createManualWaitpoint({ + runId: GEN2_RUN, + projectId: "proj", + environmentId: "env", + }); + + expect(captured.id?.[24]).toBe("a"); + expect(captured.id?.[25]).toBe("2"); + }); + + it("a standalone token mints by the environment's shard, not by an anchor", async () => { + const captured: Captured = {}; + await coordinatorCapturing(captured).createManualWaitpoint({ + projectId: "proj", + environmentId: "env", + standaloneShardKey: "c", + }); + + expect(captured.id?.[24]).toBe("c"); + expect(captured.id?.[25]).toBe("2"); + }); + + it("a standalone token on a gen-1 environment keeps a cuid", async () => { + const captured: Captured = {}; + await coordinatorCapturing(captured).createManualWaitpoint({ + projectId: "proj", + environmentId: "env", + standaloneShardKey: "new", + standaloneResidency: "NEW", + }); + + expect(captured.id).toHaveLength(25); + }); + + it("an owning run outranks the environment shard", async () => { + const captured: Captured = {}; + await coordinatorCapturing(captured).createManualWaitpoint({ + runId: GEN2_RUN, + projectId: "proj", + environmentId: "env", + standaloneShardKey: "c", + }); + + expect(captured.id?.[24]).toBe("a"); + }); +}); + +describe("mintAssociatedWaitpointData stamps the anchor's shard", () => { + // Written inside the run store, which has no stamp check, so an unstamped id here strands the + // parent run with nothing logged. + const mint = (anchorRunId: string) => + coordinatorCapturing({}).mintAssociatedWaitpointData({ + projectId: "proj", + environmentId: "env", + anchorRunId, + }); + + it("a gen-2 run anchor yields a gen-2 waitpoint id", () => { + const data = mint(GEN2_RUN); + expect(data.id).toHaveLength(26); + expect(data.id[24]).toBe("a"); + expect(data.id[25]).toBe("2"); + expect(data.friendlyId).toBe(`waitpoint_${data.id}`); + }); + + it("a gen-1 run anchor keeps a cuid", () => { + expect(mint(GEN1_RUN).id).toHaveLength(25); + }); + + it("mints a fresh core, so the waitpoint id never equals the run's own body", () => { + expect(mint(GEN2_RUN).id).not.toBe(GEN2_RUN); + }); + + it("a batch anchor stamps the batch's shard", () => { + // The create names only completedByBatchId, so that is what the router validates against. + expect(mint(GEN2_BATCH).id[24]).toBe("b"); + }); +}); diff --git a/internal-packages/run-store/src/PostgresRunStore.ts b/internal-packages/run-store/src/PostgresRunStore.ts index df718b4a1af..22dc2f90c44 100644 --- a/internal-packages/run-store/src/PostgresRunStore.ts +++ b/internal-packages/run-store/src/PostgresRunStore.ts @@ -2757,8 +2757,10 @@ export class PostgresRunStore implements RunStore { async upsertWaitpointTag( data: { environmentId: string; name: string; projectId: string; id?: string }, tx?: PrismaClientOrTransaction, - // `residency` selects the store at the router; a single store has one client and ignores it. - _residency?: ShardKey + // `residency` and `shardKey` select the store at the router; a single store has one client + // and ignores both. + _residency?: ShardKey, + _shardKey?: ShardKey ): Promise { const prisma = tx ?? this.prisma; diff --git a/internal-packages/run-store/src/delegatingRunStore.ts b/internal-packages/run-store/src/delegatingRunStore.ts index c7fe3225c76..6735a742822 100644 --- a/internal-packages/run-store/src/delegatingRunStore.ts +++ b/internal-packages/run-store/src/delegatingRunStore.ts @@ -25,7 +25,7 @@ import type { WaitpointTag, } from "@trigger.dev/database"; import type { TaskRunError } from "@trigger.dev/core/v3/schemas"; -import type { Residency } from "@trigger.dev/core/v3/isomorphic"; +import type { Residency, ShardKey } from "@trigger.dev/core/v3/isomorphic"; import type { ClearIdempotencyKeyInput, CompletionSnapshotInput, @@ -715,9 +715,10 @@ export class DelegatingRunStore implements RunStore { // A tag has no owning run to co-locate with; when no minted `id` pins it by id-shape, a // minted-new env's tags read this residency (NEW) so they land with the env's tokens/runs // instead of defaulting to LEGACY. Single-store impls ignore it. - residency?: Residency + residency?: Residency, + shardKey?: ShardKey ): Promise { - return this.delegate.upsertWaitpointTag(data, tx, residency); + return this.delegate.upsertWaitpointTag(data, tx, residency, shardKey); } findManyWaitpointTags( diff --git a/internal-packages/run-store/src/placement.proof.test.ts b/internal-packages/run-store/src/placement.proof.test.ts new file mode 100644 index 00000000000..e40e3888424 --- /dev/null +++ b/internal-packages/run-store/src/placement.proof.test.ts @@ -0,0 +1,151 @@ +import { existsSync, readFileSync } from "node:fs"; +import path from "node:path"; +import { describe, expect, it } from "vitest"; +import { + GIVEN_RUN_ID_ROUTE, + PLACEMENT_SITES, + READ_ONLY_METHODS, + ROUTES_BY_GIVEN_RUN_ID, +} from "./placementCatalog.js"; + +function repoRoot(): string { + let dir = process.cwd(); + while (!existsSync(path.join(dir, "pnpm-workspace.yaml"))) { + const parent = path.dirname(dir); + if (parent === dir) throw new Error("repo root (pnpm-workspace.yaml) not found"); + dir = parent; + } + return dir; +} + +const read = (relative: string) => readFileSync(path.join(repoRoot(), relative), "utf8"); + +const TYPES = "internal-packages/run-store/src/types.ts"; +const STORE = "internal-packages/run-store/src/runOpsStore.ts"; + +/** + * Method names on the `RunStore` interface, overloads collapsed. Parsed from source rather than + * imported as a type, so a failure can name the method somebody forgot to classify. + */ +function interfaceMethods(): string[] { + const source = read(TYPES); + const start = source.indexOf("export interface RunStore {"); + expect(start).toBeGreaterThan(-1); + + // Declarations sit at two-space indent. A stray name from later in the file shows up as + // uncatalogued rather than being dropped. + const body = source.slice(start); + const names = new Set(); + for (const match of body.matchAll(/^ {2}([a-zA-Z][A-Za-z0-9]*)(<[^\n]*?>)?\(/gm)) { + names.add(match[1]!); + } + return [...names]; +} + +/** + * One method implementation: its declaration to the next member at the same indent. Not + * brace-matching, which a signature carrying an inline object type makes fiddly to get right. + */ +function methodBody(source: string, method: string): string | undefined { + // The last declaration: an overloaded method leads with bodiless signatures. + const declaration = new RegExp(`^ {2}(?:async )?${method}(?:<[^\\n]*?>)?\\(`, "gm"); + const matches = [...source.matchAll(declaration)]; + const start = matches.at(-1)?.index; + if (start === undefined) return undefined; + + const rest = source.slice(start + 3); + const next = rest.search(/^ {2}(?:async )?[a-zA-Z#][A-Za-z0-9]*(?:<[^\n]*?>)?\(/m); + return next === -1 ? rest : rest.slice(0, next); +} + +function catalogued(): { writes: string[]; all: string[] } { + const writes = [...ROUTES_BY_GIVEN_RUN_ID, ...PLACEMENT_SITES.map((s) => s.method)]; + return { writes, all: [...writes, ...READ_ONLY_METHODS] }; +} + +describe("run-store placement census — every write states what it routes by", () => { + it("parses a plausible interface, so a silent parse failure cannot pass the suite", () => { + const methods = interfaceMethods(); + expect(methods.length).toBeGreaterThan(50); + expect(methods).toContain("upsertWaitpointTag"); + expect(methods).toContain("findRun"); + }); + + // The point of the census: nobody adds a write without saying how it is placed. + it("classifies every interface method as exactly one of read or write", () => { + const methods = interfaceMethods(); + const { all } = catalogued(); + + const uncatalogued = methods.filter((m) => !all.includes(m)).sort(); + expect({ uncatalogued }).toEqual({ uncatalogued: [] }); + + const stale = all.filter((m) => !methods.includes(m)).sort(); + expect({ staleCatalogEntries: stale }).toEqual({ staleCatalogEntries: [] }); + }); + + it("never classifies a method as both a read and a write", () => { + const { writes } = catalogued(); + const both = writes.filter((m) => READ_ONLY_METHODS.includes(m)).sort(); + expect({ classifiedAsBoth: both }).toEqual({ classifiedAsBoth: [] }); + }); + + it("lists no method twice", () => { + const { all } = catalogued(); + const seen = new Set(); + const duplicates = all.filter((m) => (seen.has(m) ? true : (seen.add(m), false))).sort(); + expect({ duplicates }).toEqual({ duplicates: [] }); + }); + + // The forbidden cell: a row on a database its owner does not live on, with nothing to detect + // it. `upsertWaitpointTag` sat here and every functional test passed. + it("has no write that routes on residency alone and misses silently", () => { + const forbidden = PLACEMENT_SITES.filter( + (s) => s.basis === "residency" && s.missMode === "silent" + ).map((s) => s.method); + + expect({ residencyOnlySilentWrites: forbidden }).toEqual({ residencyOnlySilentWrites: [] }); + }); + + // Both are claims about safety rather than mechanisms, so each has to be argued in the catalog. + it("requires a written justification wherever safety is a claim, not a mechanism", () => { + const unjustified = PLACEMENT_SITES.filter( + (s) => (s.basis === "residency" || s.basis === "fan-out") && (s.why ?? "").trim().length < 40 + ).map((s) => s.method); + + expect({ unjustified }).toEqual({ unjustified: [] }); + }); + + it("gives every catalogued write at least one routing expression", () => { + const empty = PLACEMENT_SITES.filter((s) => s.routes.length === 0).map((s) => s.method); + expect({ withoutRoutes: empty }).toEqual({ withoutRoutes: [] }); + }); + + // Scoped to the method's own body, not the whole file: three creates share + // `#routeOrNew(params.data.id)`, so a file-wide search passes when one loses its route. + it.each(PLACEMENT_SITES)("$method still contains the routes the catalog claims", (site) => { + const body = methodBody(read(STORE), site.method); + + expect({ method: site.method, found: body !== undefined }).toEqual({ + method: site.method, + found: true, + }); + + for (const route of site.routes) { + expect({ method: site.method, route, present: body!.includes(route) }).toEqual({ + method: site.method, + route, + present: true, + }); + } + }); + + it.each(ROUTES_BY_GIVEN_RUN_ID)("%s routes on the run id it is given", (method) => { + const body = methodBody(read(STORE), method); + + expect({ method, found: body !== undefined }).toEqual({ method, found: true }); + expect({ method, routedOnGivenRunId: body!.includes(GIVEN_RUN_ID_ROUTE) }).toEqual({ + method, + routedOnGivenRunId: true, + }); + }); +}); diff --git a/internal-packages/run-store/src/placementCatalog.ts b/internal-packages/run-store/src/placementCatalog.ts new file mode 100644 index 00000000000..0e88ef707c1 --- /dev/null +++ b/internal-packages/run-store/src/placementCatalog.ts @@ -0,0 +1,242 @@ +// Every method on the `RunStore` interface appears exactly once below, as a read or as a write. +// `placement.proof.test.ts` diffs this catalog against the interface, so a new method fails the +// build until somebody classifies it. +// +// The waitpoint mint census is exhaustive over id production and cannot see a row with no minted +// id, which is how `WaitpointTag` wrote to a gen-1 store for a gen-2 environment with every +// functional test passing. This is exhaustive over placement instead: what does each write route +// by? The combination that must never exist is residency-only routing with a silent miss. +// +// Pure module: no store import, no Prisma, no env. + +/** What the routing decision is made from. `residency` cannot name a gen-2 shard. */ +type PlacementBasis = "own-id" | "owner-id" | "shard-hint" | "fan-out" | "residency"; + +/** + * `loud` — Prisma raises "no record was found for an update" and the caller sees it. + * `silent` — the write succeeds on the wrong database. A create inserts a row there; an + * `updateMany` reports zero rows affected, which callers read as "nothing to do". + */ +type MissMode = "loud" | "silent"; + +export type PlacementSite = { + method: string; + basis: PlacementBasis; + missMode: MissMode; + /** + * Routing expressions the implementation contains, verbatim. The proof test requires each to + * still be present, so weakening a route fails here first. List every arm: the first arm that + * matches is what routes, so a set that looks safe on its last arm proves nothing. + */ + routes: readonly string[]; + /** Required for `residency` and `fan-out`, where safety is a claim rather than a mechanism. */ + why?: string; +}; + +/** Handed a run id, routing on it. Listed by name; 20 identical entries would be rubber-stamped. */ +export const ROUTES_BY_GIVEN_RUN_ID: readonly string[] = [ + "startAttempt", + "completeAttemptSuccess", + "recordRetryOutcome", + "requeueRun", + "recordBulkActionMembership", + "cancelRun", + "failRunPermanently", + "finalizeRun", + "expireRun", + "lockRunToWorker", + "parkPendingVersion", + "promotePendingVersionRuns", + "expireParkedRun", + "suspendForCheckpoint", + "resumeFromCheckpoint", + "rescheduleRun", + "enqueueDelayedRun", + "rewriteDebouncedRun", + "pushTags", + "pushRealtimeStream", +]; + +export const GIVEN_RUN_ID_ROUTE = "#routeForWrite(runId)"; + +export const PLACEMENT_SITES: readonly PlacementSite[] = [ + { + method: "runInTransaction", + basis: "own-id", + missMode: "loud", + routes: ["#routeOrNew(runId)"], + }, + { + method: "createRun", + basis: "own-id", + missMode: "silent", + routes: ["#routeOrNew(params.data.id)"], + }, + { + method: "createCancelledRun", + basis: "own-id", + missMode: "silent", + routes: ["#routeOrNew(params.data.id)"], + }, + { + method: "createFailedRun", + basis: "own-id", + missMode: "silent", + routes: ["#routeOrNew(params.data.id)"], + }, + { + method: "updateMetadata", + basis: "own-id", + missMode: "loud", + routes: ["#routeOrNewForWrite(runId)"], + }, + { + method: "clearIdempotencyKey", + basis: "fan-out", + missMode: "silent", + routes: ["#route(params.byId.runId)", "#shardStore(NEW_SHARD)", "#shardsExcept(NEW_SHARD)"], + why: "Routes by run id when the caller has one. The predicate arm has no id at all, so it checks NEW and then every remaining store, gen-2 shards included: a key minted before an org flipped still lives on a run in another store, and missing it leaves a stale key deduping forever.", + }, + { + method: "expireRunsBatch", + basis: "fan-out", + missMode: "silent", + routes: ["#fanOutPartitioned(this.#probeOrder, runIds"], + why: "Partitions the id list by shape and calls each store with only its own ids, over the full probe order rather than a gen-1 pair. Nothing is missed because every id is routed individually.", + }, + { + method: "createExecutionSnapshot", + basis: "owner-id", + missMode: "silent", + routes: ["#routeOrNewForWrite(input.run.id)"], + }, + { + method: "createBatchTaskRunItem", + basis: "owner-id", + missMode: "silent", + routes: ["#routeForWrite(data.batchTaskRunId)"], + }, + { + method: "createTaskRunCheckpoint", + basis: "owner-id", + missMode: "silent", + routes: ["#route(ownerRunId)"], + }, + { + method: "blockRunWithWaitpointEdges", + basis: "owner-id", + missMode: "silent", + routes: ["#routeOrNewForWrite(params.runId)"], + }, + { + method: "deleteManyTaskRunWaitpoints", + basis: "owner-id", + missMode: "silent", + routes: [ + "#routeOrNewForWrite(taskRunId)", + "#sumCounts((store) => store.deleteManyTaskRunWaitpoints(args))", + ], + why: "Routes by the owning run id when the filter names one; otherwise sums across every store, so a delete cannot quietly skip a shard.", + }, + { + method: "createBatchTaskRun", + basis: "own-id", + missMode: "silent", + routes: ["#routeForWrite(data.id)"], + }, + { + method: "updateBatchTaskRun", + basis: "own-id", + missMode: "loud", + routes: ["#routeOrNew(id)"], + }, + { + method: "updateManyBatchTaskRun", + basis: "fan-out", + missMode: "silent", + routes: ["#routeOrNew(id)", "#sumCounts((store) => store.updateManyBatchTaskRun(args))"], + why: "Routes by batch id when the filter names one, and otherwise sums across every store. An updateMany reports zero rows rather than failing, so the fan-out is what keeps a filtered update from silently skipping a shard.", + }, + { + method: "updateManyBatchTaskRunItems", + basis: "fan-out", + missMode: "silent", + routes: ["#routeOrNew(id)", "#sumCounts((store) => store.updateManyBatchTaskRunItems(args))"], + why: "Same shape as updateManyBatchTaskRun: id when available, every store otherwise.", + }, + { + method: "createWaitpoint", + basis: "own-id", + missMode: "silent", + routes: ["#waitpointWriteStore("], + why: "Prefers a co-location anchor (the owning run or batch), then the waitpoint's own stamped id, and only then the residency hint. The anchor arm refuses an unstamped id against a gen-2 shard, and the residency arm is skipped entirely when the id names a gen-2 shard, because the hint cannot express that answer.", + }, + { + method: "upsertWaitpoint", + basis: "own-id", + missMode: "silent", + routes: ["#waitpointWriteStore(opts?.coLocateWithRunId, opts?.residency, waitpointId)"], + why: "As createWaitpoint: anchor, then the waitpoint's own stamped id, then residency. A residency hint never wins over an id naming a gen-2 shard.", + }, + { + method: "updateWaitpoint", + basis: "own-id", + missMode: "loud", + routes: ["#resolveWaitpointStore(id)", "#routeOrNew(opts.coLocateWithRunId)"], + why: "The waitpoint's own id wins; the co-location hint is only the fallback for a filter that names no id. Ordering matters here and the arms must stay in this order.", + }, + { + method: "updateManyWaitpoints", + basis: "fan-out", + missMode: "silent", + routes: [ + "#resolveWaitpointStore(id)", + "#sumCounts((store) => store.updateManyWaitpoints(args))", + ], + why: "Routes by waitpoint id when the filter names one, and sums across every store otherwise, because an updateMany that lands on the wrong database reports zero rows instead of failing.", + }, + { + method: "upsertWaitpointTag", + basis: "shard-hint", + missMode: "silent", + routes: ["#shardStore(shardKey)", "#waitpointWriteStore(undefined, residency, data.id)"], + why: "A tag row has no id the router can read and no owning row to follow, so the caller passes the environment's mint shard explicitly. Without that hint this write routes on residency alone, which cannot name a gen-2 shard: the row lands on a gen-1 store while the tokens it describes live on the shard, and because reads fan out the row is still found. That is the defect this catalog was built after.", + }, +]; + +/** + * Reads, listed only so the union covers the interface exactly: a new method named + * `getOrCreateThing` would otherwise pass for a read on the strength of its name. Read routing is + * not audited here, because a read that probes the wrong store finds nothing and moves on. + */ +export const READ_ONLY_METHODS: readonly string[] = [ + "findRun", + "findRunOrThrow", + "findRunOnPrimary", + "findRunOrThrowOnPrimary", + "findRuns", + "findRunsByIds", + "findRunsByIdempotencyKeys", + "findLatestExecutionSnapshot", + "findExecutionSnapshot", + "findManyExecutionSnapshots", + "findSnapshotCompletedWaitpointIds", + "findSnapshotCompletedWaitpointIdsWithPresence", + "findWaitpointConnectedRunIds", + "findWaitpointCompletedSnapshotIds", + "countPendingWaitpoints", + "countPendingWaitpointsWithPresence", + "findWaitpoint", + "findWaitpointOnPrimary", + "findManyWaitpoints", + "forWaitpointCompletion", + "findManyTaskRunWaitpoints", + "findTaskRunAttempt", + "findBatchTaskRunById", + "findBatchTaskRunByFriendlyId", + "findBatchTaskRunByIdempotencyKey", + "countBatchTaskRunItems", + "findManyBatchTaskRunItems", + "findBatchTaskRunItem", + "findManyWaitpointTags", +]; diff --git a/internal-packages/run-store/src/runOpsStore.nShardMatrix.test.ts b/internal-packages/run-store/src/runOpsStore.nShardMatrix.test.ts index 3f42e36171a..ec8ecb20dde 100644 --- a/internal-packages/run-store/src/runOpsStore.nShardMatrix.test.ts +++ b/internal-packages/run-store/src/runOpsStore.nShardMatrix.test.ts @@ -303,3 +303,163 @@ describe("RoutingRunStore four-store matrix — pagination merge", () => { } ); }); + +// A tag has no id to route by and no owning row to follow, and `residency` names only a gen-1 +// store. Nothing fails when the row is misplaced, because reads fan out and find it anyway, so +// only a per-database count can see it. Hence container tests rather than the fake-store suite. +describe("four-store matrix — a waitpoint tag lands on its environment's shard", () => { + matrixTest( + "the shard key routes the tag to shard a, and no gen-1 store receives it", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const router = makeMatrixRouter(legacyPrisma, newPrisma, shardPrismas); + const env = await seedLegacyEnv(legacyPrisma, "tag_shard_a"); + + await router.upsertWaitpointTag( + { environmentId: env.environmentId, name: "tag-on-a", projectId: env.projectId }, + undefined, + // The residency an environment minting gen-2 ids reports. On its own this names the gen-1 + // NEW store, so it is exactly the value that used to misplace the row. + "NEW", + "a" + ); + + expect(await shardPrismas[0]!.waitpointTag.count({ where: { name: "tag-on-a" } })).toBe(1); + expect(await newPrisma.waitpointTag.count({ where: { name: "tag-on-a" } })).toBe(0); + expect(await legacyPrisma.waitpointTag.count({ where: { name: "tag-on-a" } })).toBe(0); + expect(await shardPrismas[1]!.waitpointTag.count({ where: { name: "tag-on-a" } })).toBe(0); + } + ); + + matrixTest( + "two environments on different shards do not share a database", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const router = makeMatrixRouter(legacyPrisma, newPrisma, shardPrismas); + const envA = await seedLegacyEnv(legacyPrisma, "tag_two_a"); + const envB = await seedLegacyEnv(legacyPrisma, "tag_two_b"); + + await router.upsertWaitpointTag( + { environmentId: envA.environmentId, name: "shared-name", projectId: envA.projectId }, + undefined, + "NEW", + "a" + ); + await router.upsertWaitpointTag( + { environmentId: envB.environmentId, name: "shared-name", projectId: envB.projectId }, + undefined, + "NEW", + "b" + ); + + // The unique constraint is per-database, so a collapse onto one database still inserts two + // rows. The failure to catch is placement, not a constraint violation. + const onA = await shardPrismas[0]!.waitpointTag.findMany({ where: { name: "shared-name" } }); + const onB = await shardPrismas[1]!.waitpointTag.findMany({ where: { name: "shared-name" } }); + expect(onA.map((r) => r.environmentId)).toEqual([envA.environmentId]); + expect(onB.map((r) => r.environmentId)).toEqual([envB.environmentId]); + expect(await newPrisma.waitpointTag.count({ where: { name: "shared-name" } })).toBe(0); + } + ); + + matrixTest( + "with no shard key the tag still routes by residency, exactly as before", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const router = makeMatrixRouter(legacyPrisma, newPrisma, shardPrismas); + const env = await seedLegacyEnv(legacyPrisma, "tag_gen1"); + + await router.upsertWaitpointTag( + { environmentId: env.environmentId, name: "tag-gen1", projectId: env.projectId }, + undefined, + "NEW" + ); + + expect(await newPrisma.waitpointTag.count({ where: { name: "tag-gen1" } })).toBe(1); + expect(await shardPrismas[0]!.waitpointTag.count({ where: { name: "tag-gen1" } })).toBe(0); + } + ); + + matrixTest( + "a tag written to a shard is found by the read fan-out", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const router = makeMatrixRouter(legacyPrisma, newPrisma, shardPrismas); + const env = await seedLegacyEnv(legacyPrisma, "tag_readback"); + + await router.upsertWaitpointTag( + { environmentId: env.environmentId, name: "tag-readback", projectId: env.projectId }, + undefined, + "NEW", + "a" + ); + + // The read takes no shard hint, so this proves the write is reachable the normal way. + const found = await router.findManyWaitpointTags({ + where: { environmentId: env.environmentId }, + }); + expect(found.map((r) => r.name)).toEqual(["tag-readback"]); + } + ); + + // What a real rollout produces: an environment has tags, then it is pinned. Its old rows stay on + // the gen-1 store and the same name goes to the shard with its own cuid, so an id-keyed dedupe + // would list the name twice. + matrixTest( + "the same tag name on a gen-1 store and a shard is listed once", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const router = makeMatrixRouter(legacyPrisma, newPrisma, shardPrismas); + const env = await seedLegacyEnv(legacyPrisma, "tag_dupe"); + + await router.upsertWaitpointTag( + { environmentId: env.environmentId, name: "prod", projectId: env.projectId }, + undefined, + "NEW" + ); + await router.upsertWaitpointTag( + { environmentId: env.environmentId, name: "prod", projectId: env.projectId }, + undefined, + "NEW", + "a" + ); + + // Two physical rows with different ids, which the per-database unique index permits. + const onNew = await newPrisma.waitpointTag.findMany({ where: { name: "prod" } }); + const onShard = await shardPrismas[0]!.waitpointTag.findMany({ where: { name: "prod" } }); + expect(onNew).toHaveLength(1); + expect(onShard).toHaveLength(1); + expect(onNew[0]!.id).not.toBe(onShard[0]!.id); + + const found = await router.findManyWaitpointTags({ + where: { environmentId: env.environmentId }, + }); + expect(found.map((r) => r.name)).toEqual(["prod"]); + } + ); + + matrixTest( + "two environments keep their own tag of the same name", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const router = makeMatrixRouter(legacyPrisma, newPrisma, shardPrismas); + const envA = await seedLegacyEnv(legacyPrisma, "tag_dupe_a"); + const envB = await seedLegacyEnv(legacyPrisma, "tag_dupe_b"); + + await router.upsertWaitpointTag( + { environmentId: envA.environmentId, name: "prod", projectId: envA.projectId }, + undefined, + "NEW", + "a" + ); + await router.upsertWaitpointTag( + { environmentId: envB.environmentId, name: "prod", projectId: envB.projectId }, + undefined, + "NEW", + "b" + ); + + // No environment filter, so both rows reach the merge together. Filtering per call would + // hide a name-only dedupe: each result set would hold one row and collapse to itself. + const both = await router.findManyWaitpointTags({ where: { name: "prod" } }); + + expect(both.map((r) => r.environmentId).sort()).toEqual( + [envA.environmentId, envB.environmentId].sort() + ); + } + ); +}); diff --git a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts index 8f2cc8c6485..3cb2253c06b 100644 --- a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts +++ b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts @@ -115,6 +115,16 @@ function fakeStore(slot: Slot, log: Call[], config: FakeConfig = {}): FakeStore return Promise.resolve((config.batch ?? null) as never); }) as FakeStore["findBatchTaskRunById"], + upsertWaitpoint: ((args: { create?: { id?: string } }) => { + record("upsertWaitpoint"); + return Promise.resolve((args.create ?? {}) as never); + }) as FakeStore["upsertWaitpoint"], + + upsertWaitpointTag: ((data: { name: string }) => { + record("upsertWaitpointTag"); + return Promise.resolve({ id: `tag_${slot}`, name: data.name } as never); + }) as FakeStore["upsertWaitpointTag"], + countPendingWaitpointsWithPresence: ((waitpointIds: string[], _client?: ReadClient) => { record("countPendingWaitpointsWithPresence"); const pending = new Set(config.pendingWaitpointIds ?? []); @@ -924,3 +934,104 @@ describe("RoutingRunStore batch probe tolerates legitimate dual-residency", () = expect(seen).toEqual([["legacy", "a"]]); }); }); + +describe("RoutingRunStore waitpoint tags follow their environment's shard", () => { + // A tag has no id to route by and `residency` names only a gen-1 store, so without the hint the + // row lands on a different database from the tokens it describes. Reads fan out and still find + // it, so the symptom is placement rather than an error. + const tag = { environmentId: "env_1", name: "tag", projectId: "proj_1" }; + + const shardedRouter = () => { + const log: Call[] = []; + const router = new RoutingRunStore({ + new: fakeStore("new", log), + legacy: fakeStore("legacy", log), + shards: [{ key: "a", store: fakeStore("a", log) }], + }); + return { router, log }; + }; + + it("routes a tag to the gen-2 shard the environment mints on", async () => { + const { router, log } = shardedRouter(); + await router.upsertWaitpointTag(tag as never, undefined, "NEW", "a"); + expect(trace(log)).toEqual(["a:upsertWaitpointTag"]); + }); + + it("a gen-1 shard key still routes by residency", async () => { + const { router, log } = shardedRouter(); + await router.upsertWaitpointTag(tag as never, undefined, "NEW", "new"); + expect(trace(log)).toEqual(["new:upsertWaitpointTag"]); + }); + + it("no shard hint keeps today's behaviour exactly", async () => { + const { router, log } = shardedRouter(); + await router.upsertWaitpointTag(tag as never, undefined, "LEGACY"); + expect(trace(log)).toEqual(["legacy:upsertWaitpointTag"]); + }); +}); + +describe("RoutingRunStore waitpoint writes: a stamped gen-2 id outranks a residency hint", () => { + // `residency` names only NEW or LEGACY, so a stamped gen-2 id has to win. Before this, safety + // rested on every caller withholding the hint for a gen-2 id; a caller that passed both would + // have written the row to a gen-1 database silently, because a create never misses. + const GEN2 = `${"a".repeat(24)}a2`; + const GEN1 = `${"a".repeat(24)}01`; + const CUID = "c".repeat(25); + + const shardedRouter = () => { + const log: Call[] = []; + const router = new RoutingRunStore({ + new: fakeStore("new", log), + legacy: fakeStore("legacy", log), + shards: [{ key: "a", store: fakeStore("a", log) }], + }); + return { router, log }; + }; + + const upsert = (router: RoutingRunStore, id: string, residency?: "NEW" | "LEGACY") => + router.upsertWaitpoint( + { create: { id }, update: {}, where: { id } } as never, + undefined, + residency === undefined ? undefined : ({ residency } as never) + ); + + it("routes to the shard the id names even when the hint says NEW", async () => { + const { router, log } = shardedRouter(); + await upsert(router, GEN2, "NEW"); + expect(trace(log)).toEqual(["a:upsertWaitpoint"]); + }); + + it("routes to the shard the id names even when the hint says LEGACY", async () => { + const { router, log } = shardedRouter(); + await upsert(router, GEN2, "LEGACY"); + expect(trace(log)).toEqual(["a:upsertWaitpoint"]); + }); + + it("still honours the hint for a gen-1 run-ops id, which the hint can express", async () => { + const { router, log } = shardedRouter(); + await upsert(router, GEN1, "NEW"); + expect(trace(log)).toEqual(["new:upsertWaitpoint"]); + }); + + it("still honours the hint for a cuid, and does not read it as a shard", async () => { + const { router, log } = shardedRouter(); + await upsert(router, CUID, "NEW"); + expect(trace(log)).toEqual(["new:upsertWaitpoint"]); + }); + + it("with no hint at all, a gen-1 id still routes by its own shape", async () => { + const { router, log } = shardedRouter(); + await upsert(router, GEN1); + expect(trace(log)).toEqual(["new:upsertWaitpoint"]); + }); + + it("an owning run still outranks both, so a co-located waitpoint follows its run", async () => { + const { router, log } = shardedRouter(); + await router.upsertWaitpoint( + { create: { id: GEN2 }, update: {}, where: { id: GEN2 } } as never, + undefined, + { coLocateWithRunId: GEN2, residency: "NEW" } as never + ); + expect(trace(log)).toEqual(["a:upsertWaitpoint"]); + }); +}); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 7551e552b34..31a58c74f8e 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -1565,14 +1565,16 @@ export class RoutingRunStore implements RunStore { } return this.#shardStore(key); } - if (residency !== undefined) { + // A gen-2-stamped id names the only database this row can live on, and `residency` can name + // only NEW or LEGACY, so the id wins rather than every caller having to withhold the hint. + const stamped = typeof waitpointId === "string" ? this.#shardKeyOfSafe(waitpointId) : undefined; + const isGen2Stamped = + stamped !== undefined && stamped !== NEW_SHARD && stamped !== LEGACY_SHARD; + + if (residency !== undefined && !isGen2Stamped) { return this.#shardStore(residency === "NEW" ? NEW_SHARD : LEGACY_SHARD); } - return this.#shardStore( - typeof waitpointId === "string" - ? this.#shardKeyOfSafe(waitpointId) - : this.#idlessWaitpointShard - ); + return this.#shardStore(stamped ?? this.#idlessWaitpointShard); } upsertWaitpoint( @@ -2243,14 +2245,59 @@ export class RoutingRunStore implements RunStore { upsertWaitpointTag( data: { environmentId: string; name: string; projectId: string; id?: string }, tx?: PrismaClientOrTransaction, - residency?: Residency + residency?: Residency, + shardKey?: ShardKey ): Promise { // No owning run; route by the env's residency hint when present, else a minted id-shape, else // fall back to LEGACY (same precedence as a standalone waitpoint). Caller tx is never forwarded. - const store = this.#waitpointWriteStore(undefined, residency, data.id); + // + // A gen-2 shard hint wins outright: a tag has no id to route by, and `residency` names only a + // gen-1 store, which would leave the row on a different database from the tokens it describes. + const store = + shardKey !== undefined && shardKey !== NEW_SHARD && shardKey !== LEGACY_SHARD + ? this.#shardStore(shardKey) + : this.#waitpointWriteStore(undefined, residency, data.id); return store.upsertWaitpointTag(data, undefined); } + // Two collisions exist, so both keys are needed in this order. By id first: drain can mirror a + // tag onto NEW while it keeps its id, and NEW is authoritative. By (environmentId, name) second: + // the unique index is per-database, so a store that never saw the tag minted its own cuid for it + // and the id pass cannot tell they are one tag. + // + // Dropping a row is safe because nothing reads a tag's id: a waitpoint holds its tags as a string + // array and this table is a name registry. + #mergeTags>(legs: Array<{ key: ShardKey; rows: R[] }>): R[] { + const survivors = this.#mergeById(legs); + const survivorSet = new Set(survivors as R[]); + + // Legs arrive in #precedence order, so the last write wins. Restricted to id-pass survivors so + // a stale mirror cannot win its name back. + const winnerByName = new Map(); + for (const { rows } of legs) { + for (const row of rows) { + if (!survivorSet.has(row)) continue; + const key = RoutingRunStore.#tagNameKey(row); + if (key !== undefined) winnerByName.set(key, row); + } + } + + // Filter rather than rebuild: a winner keeps the position #mergeById gave it, which callers + // observe when `orderBy` is absent. + return (survivors as R[]).filter((row) => { + const key = RoutingRunStore.#tagNameKey(row); + return key === undefined || winnerByName.get(key) === row; + }); + } + + static #tagNameKey(row: Record): string | undefined { + const environmentId = row.environmentId; + const name = row.name; + return typeof environmentId === "string" && typeof name === "string" + ? `${environmentId}\u0000${name}` + : undefined; + } + // A tag keyed by (environmentId, name) can exist on BOTH DBs for one env (dual-resident, no // id-shape signal), so fan out NEW→LEGACY and de-dupe by id (NEW wins, matching the router's // NEW-wins invariant). take/skip are widened per-leg then re-imposed globally after the merge, @@ -2281,7 +2328,7 @@ export class RoutingRunStore implements RunStore { RoutingRunStore.#ownPrimary(store, client) )) as unknown as Array>, })); - const deduped = this.#mergeById(legs) as unknown as WaitpointTag[]; + const deduped = this.#mergeTags(legs) as unknown as WaitpointTag[]; const merged = args.orderBy ? (sortByOrderBy( deduped as unknown as Array>, diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts index 9ea39473e5b..036b844bdf6 100644 --- a/internal-packages/run-store/src/types.ts +++ b/internal-packages/run-store/src/types.ts @@ -12,7 +12,7 @@ import type { WaitpointTag, } from "@trigger.dev/database"; import type { TaskRunError } from "@trigger.dev/core/v3/schemas"; -import type { Residency } from "@trigger.dev/core/v3/isomorphic"; +import type { Residency, ShardKey } from "@trigger.dev/core/v3/isomorphic"; /** * Client accepted by the read methods. Reads route through the replica by @@ -958,7 +958,10 @@ export interface RunStore { // A tag has no owning run to co-locate with; when no minted `id` pins it by id-shape, a // minted-new env's tags read this residency (NEW) so they land with the env's tokens/runs // instead of defaulting to LEGACY. Single-store impls ignore it. - residency?: Residency + residency?: Residency, + // The environment's gen-2 mint shard. A tag has no id to route by, so this is the only way its + // row follows its environment's tokens onto a shard. Outranks `residency`. + shardKey?: ShardKey ): Promise; findManyWaitpointTags( args: { diff --git a/knip.json b/knip.json index c6e8aee8977..84456756ca1 100644 --- a/knip.json +++ b/knip.json @@ -25,8 +25,7 @@ "vite/node-globals-shim.js", "app/v3/otlpTransformWorker.ts" ], - "ignoreDependencies": ["@sentry/cli", "assert", "util"], - "ignore": ["app/v3/runOpsMigration/runOpsMintShard.server.ts"] + "ignoreDependencies": ["@sentry/cli", "assert", "util"] }, "internal-packages/dashboard-agent": { "entry": ["trigger.config.ts", "src/investigation-sweep.ts", "src/maintenance.ts"], diff --git a/packages/build/src/package.json b/packages/build/src/package.json new file mode 100644 index 00000000000..3dbc1ca591c --- /dev/null +++ b/packages/build/src/package.json @@ -0,0 +1,3 @@ +{ + "type": "module" +} diff --git a/packages/core/src/v3/isomorphic/friendlyId.ts b/packages/core/src/v3/isomorphic/friendlyId.ts index 416660ce446..7c78ae46b93 100644 --- a/packages/core/src/v3/isomorphic/friendlyId.ts +++ b/packages/core/src/v3/isomorphic/friendlyId.ts @@ -229,6 +229,30 @@ export function isRunOpsIdBody(body: string): boolean { return parseRunOpsIdBody(body) !== undefined; } +// Shape-only check over the same alphabet base32hexDecode accepts, so it is the same predicate as +// "the decode would not throw". Routing needs the shape only, and decoding a timestamp to discard +// it costs ~30x more on a path taken for every routed call. +const RUN_OPS_ID_CORE_PATTERN = /^[0-9a-v]{24}$/; + +/** Shape-only v1 body check for routing: 26 chars, version "1", region and core in range. */ +export function isRunOpsIdBodyShape(body: string): boolean { + return ( + body.length === RUN_OPS_ID_LENGTH && + body[RUN_OPS_ID_VERSION_INDEX] === RUN_OPS_ID_VERSION && + REGION_CHAR_PATTERN.test(body[RUN_OPS_ID_REGION_INDEX] ?? "") && + RUN_OPS_ID_CORE_PATTERN.test(body.slice(0, RUN_OPS_ID_CORE_LENGTH)) + ); +} + +/** Shape-only gen-2 body check for routing. Returns the shard char, or undefined. */ +export function runOpsIdV2ShardShape(body: string): string | undefined { + if (body.length !== RUN_OPS_ID_LENGTH) return undefined; + if (body[RUN_OPS_ID_VERSION_INDEX] !== RUN_OPS_ID_VERSION_2) return undefined; + const shard = body[RUN_OPS_ID_SHARD_INDEX] ?? ""; + if (!SHARD_CHAR_PATTERN.test(shard)) return undefined; + return RUN_OPS_ID_CORE_PATTERN.test(body.slice(0, RUN_OPS_ID_CORE_LENGTH)) ? shard : undefined; +} + /** Parse a `run_`-prefixed friendly id; anything not a well-formed v1/gen-2 id is legacy. */ export function parseRunId(id: string): ParsedRunId { if (!id.startsWith("run_")) return LEGACY_RUN_ID; diff --git a/packages/core/src/v3/isomorphic/index.ts b/packages/core/src/v3/isomorphic/index.ts index 3f372854735..5207dbc2c65 100644 --- a/packages/core/src/v3/isomorphic/index.ts +++ b/packages/core/src/v3/isomorphic/index.ts @@ -1,5 +1,6 @@ export * from "./friendlyId.js"; export * from "./runOpsResidency.js"; +export * from "./waitpointMint.js"; export * from "./duration.js"; export * from "./maxDuration.js"; export * from "./queueName.js"; diff --git a/packages/core/src/v3/isomorphic/runOpsResidency.ts b/packages/core/src/v3/isomorphic/runOpsResidency.ts index c0f98ee5ed9..2e0501ee5a9 100644 --- a/packages/core/src/v3/isomorphic/runOpsResidency.ts +++ b/packages/core/src/v3/isomorphic/runOpsResidency.ts @@ -1,4 +1,4 @@ -import { isRunOpsIdBody, parseRunOpsIdV2Body } from "./friendlyId.js"; +import { isRunOpsIdBodyShape, runOpsIdV2ShardShape } from "./friendlyId.js"; /** * The two store FAMILIES a run/waitpoint can reside in. "NEW" is the dedicated @@ -61,10 +61,10 @@ function internalForm(id: string): string { export function resolveShard(id: string): ShardKey { const body = internalForm(id); - const genTwo = parseRunOpsIdV2Body(body); - if (genTwo) return genTwo.shard; + const shard = runOpsIdV2ShardShape(body); + if (shard !== undefined) return shard; - return isRunOpsIdBody(body) ? "new" : "legacy"; + return isRunOpsIdBodyShape(body) ? "new" : "legacy"; } /** diff --git a/packages/core/src/v3/isomorphic/waitpointMint.test.ts b/packages/core/src/v3/isomorphic/waitpointMint.test.ts new file mode 100644 index 00000000000..8d35a7248ad --- /dev/null +++ b/packages/core/src/v3/isomorphic/waitpointMint.test.ts @@ -0,0 +1,115 @@ +import { describe, expect, it } from "vitest"; +import { mintWaitpointIdFor, mintWaitpointIdForShard } from "./waitpointMint.js"; +import { + generateRunOpsId, + generateRunOpsIdV2, + isValidShardChar, + parseRunOpsIdBody, + parseRunOpsIdV2Body, +} from "./friendlyId.js"; +import { resolveShard } from "./runOpsResidency.js"; + +const GEN2_RUN = `run_${"a".repeat(24)}a2`; // shard "a", version "2" +const GEN1_RUN = `run_${"a".repeat(24)}01`; // region "0", version "1" +const CUID_RUN = `run_${"b".repeat(25)}`; + +describe("mintWaitpointIdForShard", () => { + it("a gen-2 shard key mints a gen-2 body with that char at index 24", () => { + const r = mintWaitpointIdForShard("a"); + expect(r.id.length).toBe(26); + expect(r.id[24]).toBe("a"); + expect(r.id[25]).toBe("2"); + expect(r.friendlyId).toBe(`waitpoint_${r.id}`); + expect(parseRunOpsIdV2Body(r.id)?.shard).toBe("a"); + }); + + it("the reserved key 'new' mints a cuid, unchanged from today", () => { + const r = mintWaitpointIdForShard("new"); + expect(r.id.length).toBe(25); + expect(resolveShard(r.id)).toBe("legacy"); + }); + + it("the reserved key 'legacy' mints a cuid", () => { + expect(mintWaitpointIdForShard("legacy").id.length).toBe(25); + }); + + it("two calls for one shard never collide", () => { + expect(mintWaitpointIdForShard("a").id).not.toBe(mintWaitpointIdForShard("a").id); + }); + + it("every gen-2 id it mints routes back to its own shard", () => { + for (const key of ["a", "b", "0", "z", "9"]) { + expect(isValidShardChar(key)).toBe(true); + expect(resolveShard(mintWaitpointIdForShard(key).id)).toBe(key); + } + }); +}); + +describe("mintWaitpointIdFor", () => { + it("a gen-2 anchor stamps the anchor's shard char", () => { + const r = mintWaitpointIdFor(GEN2_RUN); + expect(r.id[24]).toBe("a"); + expect(r.id[25]).toBe("2"); + }); + + it("a gen-2 anchor yields a FRESH core, never the anchor's own body", () => { + const r = mintWaitpointIdFor(GEN2_RUN); + expect(r.id).not.toBe(GEN2_RUN.slice(4)); + expect(r.id.slice(0, 24)).not.toBe("a".repeat(24)); + }); + + it("accepts the bare internal form as well as the prefixed form", () => { + expect(mintWaitpointIdFor(GEN2_RUN.slice(4)).id[24]).toBe("a"); + }); + + it("a gen-1 v1 anchor mints a cuid", () => { + expect(mintWaitpointIdFor(GEN1_RUN).id.length).toBe(25); + }); + + it("a cuid anchor mints a cuid", () => { + expect(mintWaitpointIdFor(CUID_RUN).id.length).toBe(25); + }); + + it("no anchor mints a cuid", () => { + expect(mintWaitpointIdFor(undefined).id.length).toBe(25); + }); +}); + +describe("resolveShard shape checks match the decoding parsers", () => { + // The alphabet is [0-9a-v], so "the shape matches" and "the decode would not throw" are the + // same predicate. These pin that equivalence: a drift misroutes rather than erroring. + const classifyByDecode = (body: string): string => { + const genTwo = parseRunOpsIdV2Body(body); + if (genTwo) return genTwo.shard; + return parseRunOpsIdBody(body) !== undefined ? "new" : "legacy"; + }; + + it("agrees on freshly minted gen-1 and gen-2 bodies", () => { + for (let i = 0; i < 500; i++) { + const one = generateRunOpsId(); + const two = generateRunOpsIdV2("abcdefghijklmnopqrstuvwxyz0123456789"[i % 36]!); + expect(resolveShard(one)).toBe(classifyByDecode(one)); + expect(resolveShard(two)).toBe(classifyByDecode(two)); + } + }); + + it("agrees on 26-char strings carrying out-of-alphabet characters", () => { + const alpha = "0123456789abcdefghijklmnopqrstuvwxyz-_.ZW!"; + for (let i = 0; i < 2000; i++) { + let s = ""; + for (let j = 0; j < 26; j++) s += alpha[(i * 7 + j * 13) % alpha.length]; + for (const body of [s, s.slice(0, 25) + "1", s.slice(0, 25) + "2"]) { + expect({ body, shape: resolveShard(body) }).toEqual({ + body, + shape: classifyByDecode(body), + }); + } + } + }); + + it("agrees on the shapes the plan pins as legacy", () => { + for (const body of ["", "a", "a".repeat(25), "a".repeat(27), `${"a".repeat(24)}e2`]) { + expect(resolveShard(body)).toBe(classifyByDecode(body)); + } + }); +}); diff --git a/packages/core/src/v3/isomorphic/waitpointMint.ts b/packages/core/src/v3/isomorphic/waitpointMint.ts new file mode 100644 index 00000000000..4ce08dfaaa6 --- /dev/null +++ b/packages/core/src/v3/isomorphic/waitpointMint.ts @@ -0,0 +1,24 @@ +import { generateRunOpsIdV2, WaitpointId } from "./friendlyId.js"; +import { resolveShard, type ShardKey } from "./runOpsResidency.js"; + +// A Postgres waitpoint id, NOT the Redis store format (version "w" at index 25), which has no +// Postgres row to route. The core is always fresh, or the body would equal the anchor's own id. +export function mintWaitpointIdForShard(key: ShardKey): { id: string; friendlyId: string } { + if (key === "new" || key === "legacy") { + return WaitpointId.generate(); + } + + const id = generateRunOpsIdV2(key); + return { id, friendlyId: WaitpointId.toFriendlyId(id) }; +} + +// Every Postgres waitpoint mint goes through here: the router refuses an id that is not stamped +// for the shard it lands on. A gen-1 or legacy anchor keeps a cuid. +export function mintWaitpointIdFor(anchorId: string | undefined): { + id: string; + friendlyId: string; +} { + return anchorId === undefined + ? WaitpointId.generate() + : mintWaitpointIdForShard(resolveShard(anchorId)); +}