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..f67f0860f2b 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,15 @@ const { action } = createActionApiRoute( }); const residency = mintKind === "runOpsId" ? "NEW" : "LEGACY"; + // No extra query: the org flags 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 +96,7 @@ const { action } = createActionApiRoute( environmentId: authentication.environment.id, projectId: authentication.environment.projectId, residency, + shardKey: standaloneShardKey, }); if (tagRecord) { tags.push(tagRecord); @@ -101,6 +112,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..a20d531ba7e 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,23 @@ export async function resolveBatchRunOpsWriter( newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; + shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; } ): Promise { + // 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. + 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 +122,7 @@ export type BatchCompletionDeps = { newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; + shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; tryCompleteBatch: (batchId: string) => Promise; }; @@ -136,6 +153,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..3485115ea97 --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/gen2MintInertness.test.ts @@ -0,0 +1,112 @@ +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"; + +// Gate off means resolveMintShard answers "new". Every assertion is "the id is what it was". +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..3adbc6e7321 100644 --- a/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts @@ -1,15 +1,20 @@ -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; +// A shardChar selects one gen-2 shard and takes index 24; without one the region takes that slot. +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..088eaed5c09 100644 --- a/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/mintBatchFriendlyId.server.ts @@ -1,45 +1,36 @@ -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. 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..334e22d5ccd --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/resolveRunMintTarget.server.ts @@ -0,0 +1,48 @@ +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, +}; + +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 unless a gen-2 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 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..360bc9fd863 100644 --- a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts @@ -69,14 +69,18 @@ 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 { + // Answer before reading anything, so an unconfigured deployment adds no control-plane query to + // 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..b86e5a40a64 100644 --- a/apps/webapp/app/v3/services/batchTriggerV3.server.ts +++ b/apps/webapp/app/v3/services/batchTriggerV3.server.ts @@ -362,15 +362,19 @@ export class BatchTriggerV3Service extends BaseService { anchorFriendlyId?: string, region?: string ): Promise { - const mintKind = anchorFriendlyId + // Not routed through resolveRunMintTarget: the root arm is unreachable in production and + // resolveMintKind is injected so a test can drive it 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..63f7e88505f 100644 --- a/apps/webapp/test/runEngineHandlers.test.ts +++ b/apps/webapp/test/runEngineHandlers.test.ts @@ -490,6 +490,97 @@ 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. + // The shard is prisma14 and both gen-1 slots are prisma17, so any 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; + }, + } + ); + + 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: 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..7917ba68303 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(), + // From the batch, not the blocked run: the create passes only completedByBatchId, + // which is the owner the router validates 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..def58f7bc37 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"; @@ -247,7 +247,7 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator }, }, create: { - ...WaitpointId.generate(), + ...mintWaitpointIdFor(runId), type: "DATETIME" as const, idempotencyKey: idempotencyKey ?? nanoid(24), idempotencyKeyExpiresAt, @@ -272,6 +272,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 +280,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 +337,8 @@ 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 are re-evaluated per attempt, so a retry after a conflict tries 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 +348,9 @@ export class LegacyPostgresWaitpointCoordinator implements WaitpointCoordinator }, }, create: { - ...WaitpointId.generate(), + ...(standaloneShard !== undefined + ? mintWaitpointIdForShard(standaloneShard) + : mintWaitpointIdFor(runId)), type: "MANUAL", idempotencyKey: idempotencyKey ?? nanoid(24), idempotencyKeyExpiresAt, @@ -379,12 +388,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..412e1ed97ec 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; + /** This write skips the router's stamp check. */ + anchorRunId: string; }): AssociatedWaitpointData; createAssociatedWaitpoint(params: { runId: string; @@ -108,7 +111,11 @@ 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. */ + /** + * Co-locates the waitpoint with this run's DB. There is deliberately no standalone arm: omitting + * it on a gen-2 environment 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 +137,8 @@ export type CreateManualWaitpointParams = { * full rationale. Only a Postgres implementation reads this. */ standaloneResidency?: "NEW" | "LEGACY"; + /** For a standalone token. When it names a gen-2 shard, ignore `standaloneResidency`. */ + 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..4a7a863bd4d --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMint.proof.test.ts @@ -0,0 +1,124 @@ +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, not listed, so a mint in a new file is visible. Test trees excluded: a raw-Prisma +// helper 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] : []; + }); +} + +const MINT_CALL = /mintWaitpointIdFor(?:Shard)?\(/g; +const UNSTAMPED_MINT = /WaitpointId\.generate\(/g; +const WAITPOINT_WRITE = /waitpoint\.create\(|upsertWaitpoint\(|createWaitpoint\(/g; + +// Scanning the catalog would count its own string data. +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"]; + +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, so a swapped anchor fails too. + 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: 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) => { + // Worst case is a create with no id: @default(cuid()) 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..7f5df203b65 --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintCatalog.ts @@ -0,0 +1,70 @@ +// A site creating a Postgres `Waitpoint` row needs an entry here or `waitpointMint.proof.test.ts` +// fails. Most unstamped mints fail loudly at the router, but `createRun` writes inside the run +// store, which has no stamp check. +export type WaitpointMintSite = { + id: string; + type: "DATETIME" | "MANUAL" | "RUN" | "BATCH"; + site: string; + /** Never a line number. */ + symbol: string; + /** Verbatim, counted per file, so a swapped anchor fails too. Empty if 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", + }, + // These 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..53edce67d1f --- /dev/null +++ b/internal-packages/run-engine/src/engine/waitpointCoordinator/waitpointMintSites.test.ts @@ -0,0 +1,143 @@ +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: 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", () => { + 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", () => { + 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..2dc6717e2bf --- /dev/null +++ b/internal-packages/run-store/src/placement.proof.test.ts @@ -0,0 +1,138 @@ +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"; + +/** Parsed from source, not imported as a type, so a failure can name the unclassified method. */ +function interfaceMethods(): string[] { + const source = read(TYPES); + const start = source.indexOf("export interface RunStore {"); + expect(start).toBeGreaterThan(-1); + + 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]; +} + +/** Not brace-matching: an inline object type in a signature makes that fiddly to get right. */ +function methodBody(source: string, method: string): string | undefined { + // Last, not first: 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"); + }); + + 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: [] }); + }); + + 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: [] }); + }); + + 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: [] }); + }); + + // Per body, not per file: three creates share one route expression. + 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..4768c389075 --- /dev/null +++ b/internal-packages/run-store/src/placementCatalog.ts @@ -0,0 +1,222 @@ +// Every `RunStore` method appears below once, as a read or a write, and `placement.proof.test.ts` +// fails until a new one is classified. The combination that must never exist is residency-only +// routing with a silent miss: `WaitpointTag` sat there, misplacing rows with tests passing. + +/** `residency` cannot name a gen-2 shard. */ +type PlacementBasis = "own-id" | "owner-id" | "shard-hint" | "fan-out" | "residency"; + +/** `silent`: the write succeeds on the wrong database, or an `updateMany` affects zero rows. */ +type MissMode = "loud" | "silent"; + +export type PlacementSite = { + method: string; + basis: PlacementBasis; + missMode: MissMode; + /** Verbatim, and every arm: the first arm that matches is what routes. */ + routes: readonly string[]; + /** Required for `residency` and `fan-out`, where safety is a claim not a mechanism. */ + why?: string; +}; + +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.", + }, +]; + +/** Listed so the union covers the interface: a `getOrCreateThing` must not pass as a read. */ +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..14b5fb15bf1 100644 --- a/internal-packages/run-store/src/runOpsStore.nShardMatrix.test.ts +++ b/internal-packages/run-store/src/runOpsStore.nShardMatrix.test.ts @@ -303,3 +303,155 @@ describe("RoutingRunStore four-store matrix — pagination merge", () => { } ); }); + +// A misplaced tag row still reads back, because the read fans out, so only a per-database count +// can see it. Hence containers 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, + "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 constraint is per-database, so a collapse still inserts two rows: this catches + // 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" + ); + + const found = await router.findManyWaitpointTags({ + where: { environmentId: env.environmentId }, + }); + expect(found.map((r) => r.name)).toEqual(["tag-readback"]); + } + ); + + // What a rollout produces: tags exist, then the environment is pinned. + 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" + ); + + 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, or each result set holds one row and a name-only key would pass. + 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..aff0d50764e 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,99 @@ describe("RoutingRunStore batch probe tolerates legitimate dual-residency", () = expect(seen).toEqual([["legacy", "a"]]); }); }); + +describe("RoutingRunStore waitpoint tags follow their environment's shard", () => { + 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", () => { + // A caller passing both used to write to a gen-1 database silently: 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..fa923244d06 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -1565,14 +1565,15 @@ export class RoutingRunStore implements RunStore { } return this.#shardStore(key); } - if (residency !== undefined) { + // `residency` names only NEW or LEGACY, so a stamped gen-2 id wins. + 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 +2244,51 @@ 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); + // + const store = + shardKey !== undefined && shardKey !== NEW_SHARD && shardKey !== LEGACY_SHARD + ? this.#shardStore(shardKey) + : this.#waitpointWriteStore(undefined, residency, data.id); return store.upsertWaitpointTag(data, undefined); } + // Both keys, in order. Drain can mirror a tag onto NEW keeping its id, and NEW wins. Then by + // name, because the per-database unique index lets a store mint its own cuid for a tag it has + // not seen. Dropping a row is safe: nothing reads a tag's id. + #mergeTags>(legs: Array<{ key: ShardKey; rows: R[] }>): R[] { + const survivors = this.#mergeById(legs); + const survivorSet = new Set(survivors as R[]); + + // Survivors only, 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, not rebuild: position matters 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 +2319,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..41fecf90bb5 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, + // 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..b38aa2899e8 100644 --- a/packages/core/src/v3/isomorphic/friendlyId.ts +++ b/packages/core/src/v3/isomorphic/friendlyId.ts @@ -229,6 +229,27 @@ export function isRunOpsIdBody(body: string): boolean { return parseRunOpsIdBody(body) !== undefined; } +// Same alphabet base32hexDecode accepts, so this and "the decode would not throw" are one +// predicate. Decoding a timestamp to discard it costs ~30x more on the router's hot path. +const RUN_OPS_ID_CORE_PATTERN = /^[0-9a-v]{24}$/; + +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)) + ); +} + +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..a320cae852d --- /dev/null +++ b/packages/core/src/v3/isomorphic/waitpointMint.test.ts @@ -0,0 +1,114 @@ +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", () => { + // Pins the shape/decode 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..f4327d40f0e --- /dev/null +++ b/packages/core/src/v3/isomorphic/waitpointMint.ts @@ -0,0 +1,23 @@ +import { generateRunOpsIdV2, WaitpointId } from "./friendlyId.js"; +import { resolveShard, type ShardKey } from "./runOpsResidency.js"; + +// A Postgres waitpoint id, not the Redis store format (version "w"), which has no 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 unstamped id on a shard. +export function mintWaitpointIdFor(anchorId: string | undefined): { + id: string; + friendlyId: string; +} { + return anchorId === undefined + ? WaitpointId.generate() + : mintWaitpointIdForShard(resolveShard(anchorId)); +}