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..d32dc048ea6 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,25 @@ export async function resolveBatchRunOpsWriter( newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; + shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; } ): Promise { + // A gen-2 batch names its own shard in its id, so route by that and never probe. The + // probe below is binary — NEW, else assume LEGACY — so a gen-2 batch would fall through + // to a store that holds no such row, and the completion update would throw before the + // batch waitpoint could complete, leaving the parent run 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 +124,7 @@ export type BatchCompletionDeps = { newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; + shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; tryCompleteBatch: (batchId: string) => Promise; }; @@ -136,6 +155,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..e51dae720ae --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/gen2MintInertness.test.ts @@ -0,0 +1,114 @@ +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", so no shard char reaches a MintTarget. Every assertion +// below is "the id is what it was before gen-2 existed". +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, so a child run stamped the + // requested region. 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..a3f9466788b --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/resolveRunMintTarget.server.ts @@ -0,0 +1,57 @@ +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. Two stages, and the second 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. + * + * Every run-mint path routes through here. The branch used to be duplicated per service, + * and one copy had already drifted into minting gen-1 for a gen-2 parent. + */ +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, exactly as + // it did before this branch. 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, which is the state of every deployment that has configured + // no shard. Only a single-char key names a gen-2 shard. + 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..751ef67c2a9 100644 --- a/apps/webapp/test/runEngineHandlers.test.ts +++ b/apps/webapp/test/runEngineHandlers.test.ts @@ -490,6 +490,107 @@ describe("runEngineHandlers batch completion", () => { }); describe("runEngineHandlers batch residency routing", () => { + // A gen-2 batch lives on its own shard. The binary probe below it looks only on the + // NEW store and then assumes LEGACY, so without a shard arm the completion update runs + // on a database that has no such row: Prisma throws "no record was found for an + // update", the callback dies before tryCompleteBatch, the BATCH waitpoint stays + // PENDING and the parent run waits forever with nothing logged as a hang. + // Real clients on two 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: + // every wrong resolution therefore lands on a database that holds no such batch, which is + // the production failure this arm exists to prevent. + 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; + }, + } + ); + + // Committed on the shard, and the callback survived to run — the hang was the callback + // dying on "no record was found for an update" before it could reach 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); + + // Nothing for this batch reached the gen-1 database. + expect(await prisma17.batchTaskRun.findMany({ where: { id: gen2BatchId } })).toHaveLength(0); + expect( + await prisma17.batchTaskRunError.findMany({ where: { batchTaskRunId: gen2BatchId } }) + ).toHaveLength(0); + } + ); + + // Kept on a throwing double deliberately: "the NEW store is never probed" is an assertion + // about a call that must not happen, and only a client that throws when touched can make + // that observable. A real client would simply return null and the test would still pass. + 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..cc53cbb7bd1 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,14 @@ export class RunEngine { tags?: string[]; /** Standalone-token residency (no owning run) from the env mint kind; ignored when `runId` is set. */ standaloneResidency?: "NEW" | "LEGACY"; + /** + * The environment's mint shard, for a STANDALONE token with no owning run. It selects the + * shard the token's id is stamped for. When it names a gen-2 shard the implementation must + * IGNORE `standaloneResidency`: a residency hint outranks the id shape in the router and + * can only name a gen-1 store, so honouring it would land the row there while its + * completion routes to the shard. Only a Postgres implementation reads this. + */ + standaloneShardKey?: ShardKey; }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> { return this.waitpointSystem.createManualWaitpoint({ runId, @@ -1828,6 +1840,7 @@ export class RunEngine { timeout, tags, standaloneResidency, + standaloneShardKey, }); } @@ -1853,7 +1866,13 @@ 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 the routing store resolves the owner from the batch and + // validates the stamp against the BATCH's shard. On the normal path the two are + // the same char anyway, because the batch inherited this run's shard. They differ + // only if a batch ever blocks a run from another shard -- and then this is the + // stamp that matches the owner the router actually checks. + ...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..52d29858b59 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,14 @@ 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"; + /** + * The environment's mint shard, for a STANDALONE token with no owning run. It selects the + * shard the token's id is stamped for. When it names a gen-2 shard the implementation must + * IGNORE `standaloneResidency`: a residency hint outranks the id shape in the router and + * can only name a gen-1 store, so honouring it would land the row there while its + * completion routes to the shard. Only a Postgres implementation reads this. + */ + standaloneShardKey?: ShardKey; }): Promise<{ waitpoint: Waitpoint; isCached: boolean }> { const result = await this.coordinator.createManualWaitpoint({ runId, @@ -206,6 +216,7 @@ export class WaitpointSystem { timeout, tags, standaloneResidency, + standaloneShardKey, }); if (result.kind === "cached") { @@ -721,11 +732,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 +824,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..da8397a247b 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,8 @@ 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. + // The id is stamped for the anchor run's shard, so the waitpoint's own row is routable + // and its completion write needs no probe. A gen-1 or legacy anchor keeps a cuid. const upsertArgs = { where: { environmentId_idempotencyKey: { @@ -247,7 +249,7 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator }, }, create: { - ...WaitpointId.generate(), + ...mintWaitpointIdFor(runId), type: "DATETIME" as const, idempotencyKey: idempotencyKey ?? nanoid(24), idempotencyKeyExpiresAt, @@ -272,6 +274,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 +282,18 @@ 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 hint and lets + // the id route: `residency` outranks the id shape and can only name a gen-1 store. + 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 +340,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 +352,9 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator }, }, create: { - ...WaitpointId.generate(), + ...(standaloneShard !== undefined + ? mintWaitpointIdForShard(standaloneShard) + : mintWaitpointIdFor(runId)), type: "MANUAL", idempotencyKey: idempotencyKey ?? nanoid(24), idempotencyKeyExpiresAt, @@ -379,12 +392,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..9ee7505f810 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,11 @@ export type WaitpointCoordinator = { mintAssociatedWaitpointData(params: { projectId: string; environmentId: string; + /** + * The run this waitpoint belongs to. Its id names the shard the row must land on, and + * this write bypasses the routing store's stamp check, so an unstamped id is silent here. + */ + anchorRunId: string; }): AssociatedWaitpointData; createAssociatedWaitpoint(params: { runId: string; @@ -130,6 +136,14 @@ 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. It selects the + * shard the token's id is stamped for. When it names a gen-2 shard the implementation must + * IGNORE `standaloneResidency`: a residency hint outranks the id shape in the router and + * can only name a gen-1 store, so honouring it would land the row there while its + * completion routes to the shard. Only a Postgres implementation reads this. + */ + 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..9e959e8dd30 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMint.proof.test.ts @@ -0,0 +1,136 @@ +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; +} + +// Every production `.ts` under a root, walked rather than listed: a mint added in a new +// file, or moved back into `systems/` where these all lived until the coordinator seam was +// extracted, has to be visible here or the census is decorative. +// +// Test-support trees are excluded deliberately. A helper that writes a row through raw +// Prisma never reaches the routing store, so it cannot misroute; requiring it to be +// catalogued would fill the census with sites that carry no risk. +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: this fails for a fifth mint added inside an + // already-catalogued file, AND for a swapped anchor — mintWaitpointIdFor(undefined) in + // place of the run id — which a bare call-count would wave through. + 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, + }); + } + + // No mint in the file beyond the ones the catalog accounts for. + 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) => { + // The regex matches tokens inside comments too — deliberate. Any textual addition + // forces the census to be reconciled, so a new site cannot land unnoticed. + 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: Prisma's @default(cuid()) then mints a cuid on + // a gen-2 shard after the write, which no stamp check can see. + 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..be8c76d72bf --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintCatalog.ts @@ -0,0 +1,86 @@ +// If you add a site that creates a Postgres `Waitpoint` row, add a matching entry here or +// `waitpointMint.proof.test.ts` fails. Entries are one per site, anchored by symbol name, +// never by line number. +// +// Why: a site that mints a cuid for a gen-2 run writes a row the completion path cannot +// find. Three of the sites below fail loudly, because the routing store refuses an +// unstamped id on a gen-2 shard. The RUN row written through `createRun` does NOT — that +// write happens inside the run store, which has no such check — so a missed site there +// strands a parent run with no error. +// +// PURE module — no engine import, no env, no Prisma. +export type WaitpointMintSite = { + id: string; + type: "DATETIME" | "MANUAL" | "RUN" | "BATCH"; + /** Repo-relative source path. */ + site: string; + /** Enclosing method or symbol name — NEVER a line number. */ + symbol: string; + /** + * The exact mint expressions this site contains, verbatim. The proof test counts each one + * per file, so both a new mint inside an already-catalogued file and a swapped anchor + * (`mintWaitpointIdFor(undefined)` in place of the run id) fail until reconciled here. + * Empty for a site that writes a row from 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..b4f44bb7623 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintSites.test.ts @@ -0,0 +1,149 @@ +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 that calls the helper with +// a hand-written literal passes even when a site stops passing its anchor, which is the one +// regression that matters here. +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", + }); + + // The run's shard, not the environment's: a co-located waitpoint follows its run. + expect(captured.id?.[24]).toBe("a"); + }); +}); + +describe("mintAssociatedWaitpointData stamps the anchor's shard", () => { + // The row this mints is 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", () => { + // What blockRunWithCreatedBatch relies on: the router validates a BATCH waitpoint + // against the batch's shard, because the create names only completedByBatchId. + 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/runOpsStore.shardMap.test.ts b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts index 8f2cc8c6485..aa1871b0756 100644 --- a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts +++ b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts @@ -115,6 +115,11 @@ function fakeStore(slot: Slot, log: Call[], config: FakeConfig = {}): FakeStore return Promise.resolve((config.batch ?? null) as never); }) as FakeStore["findBatchTaskRunById"], + 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 +929,39 @@ describe("RoutingRunStore batch probe tolerates legitimate dual-residency", () = expect(seen).toEqual([["legacy", "a"]]); }); }); + +describe("RoutingRunStore waitpoint tags follow their environment's shard", () => { + // A tag row carries no id the router can read, and `residency` only ever names a gen-1 store. + // Without the shard hint an environment's tags land on a different database from the tokens + // they describe. Reads fan out over every store, so the row is still found later: the symptom + // is a tag attributed to the wrong database, not an error, which is why this needs a test. + 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"]); + }); +}); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 7551e552b34..1a7b6d40719 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -2243,11 +2243,20 @@ 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. Callers never mint a tag id, so id-shape cannot route one, + // and `residency` collapses to a gen-1 store — which would leave an environment's tags on a + // different database from the tokens they describe. The read side already fans out over every + // store, so a misplaced row is found but attributed to the wrong environment's database. + const store = + shardKey !== undefined && shardKey !== NEW_SHARD && shardKey !== LEGACY_SHARD + ? this.#shardStore(shardKey) + : this.#waitpointWriteStore(undefined, residency, data.id); return store.upsertWaitpointTag(data, undefined); } diff --git a/internal-packages/run-store/src/types.ts b/internal-packages/run-store/src/types.ts index 9ea39473e5b..1b1ee1e1ee8 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,11 @@ 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, when it has one. A tag carries no id the router can + // read, so this is the only way its row can follow its environment's tokens onto a shard. + // Takes precedence over `residency`, which can only ever name a gen-1 store. + 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/core/src/v3/isomorphic/friendlyId.ts b/packages/core/src/v3/isomorphic/friendlyId.ts index 416660ce446..c471ce9aa31 100644 --- a/packages/core/src/v3/isomorphic/friendlyId.ts +++ b/packages/core/src/v3/isomorphic/friendlyId.ts @@ -229,6 +229,31 @@ export function isRunOpsIdBody(body: string): boolean { return parseRunOpsIdBody(body) !== undefined; } +// Shape-only core check: exactly the alphabet base32hexDecode accepts, without decoding. +// RUN_OPS_ID_ALPHABET is [0-9a-v], so this and "the decode would not throw" are the same +// predicate. Routing only needs the shape; decoding the core to recover a timestamp and +// then discarding it costs ~30x more, on a path the router takes 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..5f9bd76bd48 --- /dev/null +++ b/packages/core/src/v3/isomorphic/waitpointMint.test.ts @@ -0,0 +1,117 @@ +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", () => { + // resolveShard used to decode the 24-char core to recover a timestamp and then discard + // it, which cost ~10x a shape check on the router's hot path. The alphabet is [0-9a-v], + // so "the shape matches" and "the decode would not throw" are the same predicate. These + // pin that equivalence, because a drift here 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..d05d7b60f43 --- /dev/null +++ b/packages/core/src/v3/isomorphic/waitpointMint.ts @@ -0,0 +1,26 @@ +import { generateRunOpsIdV2, WaitpointId } from "./friendlyId.js"; +import { resolveShard, type ShardKey } from "./runOpsResidency.js"; + +// A waitpoint id for a Postgres shard — NOT the Redis store format (type char at index +// 24, version "w"), which has no Postgres row to route. The core is always fresh: reusing +// the anchor's would produce a body identical to the run'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, in the webapp and the engine alike: +// the router refuses a waitpoint whose id 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)); +}