Skip to content

Commit 15dd973

Browse files
d-csclaude
andauthored
feat(core,webapp,run-engine): stamp a shard key onto run, batch and waitpoint ids (#4788)
## Summary Adds the id-minting half of sharding run data across several databases. Every entity that co-locates with a run now carries the run's shard key inside its own id, so its row is routable on its own instead of needing a directory table or a scatter across shards. Nothing changes for users yet. With no shard descriptors configured, every mint path produces exactly the ids it produces today, and the trigger path issues no extra query. ## Design A run's mint target travels as a single object carrying the kind and, when sharded, the shard character. The shard and the caller's region both occupy index 24 of a run-ops id, so passing them together makes it impossible for a caller to set two competing sources for one slot. A child run, a batch and a batch item read the shard from their parent's id rather than resolving a fresh one, so a run tree never splits across databases. Three services carried that branch separately, and one had already drifted, so it now lives in one function. Waitpoints mint through one shared pure function used by both the webapp and the run engine. They have to agree byte for byte, because the routing store refuses a waitpoint whose id is not stamped for the shard it is being written to: ```ts mintWaitpointIdForShard(key) // standalone token: the environment's shard mintWaitpointIdFor(anchorId) // co-located: the anchor's shard, or a cuid ``` The core is always freshly minted rather than derived from the anchor, since a derived body would be byte-identical to the run's own id. One latent bug fixed on the way: the failed-run path duplicated the mint branch inline and had drifted, so a child of a sharded parent would have been written to a different database from its parent. ## Guarding the create sites The expensive failure here is a waitpoint minted without its anchor's shard: one of the five create sites writes through a path that has no stamp check, so a miss there strands a blocked run with nothing logged. An enumerated census plus a source scan fails when a new create site appears, when an existing one stops passing its anchor, or when a site is added to a file the scan does not yet cover. The census was written before any site was converted, so it went red on the first commit and green as the last site landed. Both holes an earlier draft had, a file-granular count and a scan that missed the directory these mints used to live in, were confirmed closed by reintroducing them and watching the guard fail. ## Before enabling a shard Merging this is inert: with the mint list empty the resolver returns before it reads anything, and ids are identical to a measured `main` baseline. Verified against a live shard locally, including that the resolver issues no query across thirty triggers with no shard configured. Enabling is gated on two other pull requests, both open, both by the same author, each of which owns the file involved: - **#4781** adds the gen-2 shard arm to read-through. Without it a gen-2 run cannot wait on a token at all: the wait route resolves the waitpoint through read-through, which is shard-blind, so the wait fails. Do not set the mint list before it merges. - **#4780** generalises the distinct-database sentinel. Without it a shard pointed at the same physical database as the gen-1 store boots without complaint, which voids the disjointness the fan-out sums rely on. Testing also turned up a silent read-path gap that neither pull request covers: the paths that hydrate runs from ClickHouse through a fixed pair of Postgres clients drop gen-2 rows on the floor, so the runs list would show fewer rows than its own count with nothing logged. That needs its own change before a shard carries real traffic, and it is filed as such. ## Notes for reviewers Four commits in the middle of the stack do not typecheck in isolation: a signature change and its call-site repairs are separate commits, so bisecting inside the stack needs care. Commit `845ab06` also understates itself, since it rewrites the primary trigger path's mint alongside the failed-run path it names. No changeset and no server-changes entry: every path is inert while the feature is off, so there is nothing to tell users yet. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
1 parent 4e00651 commit 15dd973

42 files changed

Lines changed: 1885 additions & 142 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

apps/webapp/app/models/waitpointTag.server.ts

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,17 @@ export async function createWaitpointTag({
99
environmentId,
1010
projectId,
1111
residency,
12+
shardKey,
1213
}: {
1314
tag: string;
1415
environmentId: string;
1516
projectId: string;
1617
// Residency from the env mint kind: a tag has no owning run, so a minted-new env pins it to NEW
1718
// instead of defaulting to the draining legacy DB.
1819
residency?: "NEW" | "LEGACY";
20+
// The environment's gen-2 mint shard, when it has one. A tag has no id the router can read, so
21+
// without this the row lands on a gen-1 store while the token it describes lands on the shard.
22+
shardKey?: string;
1923
}) {
2024
if (tag.trim().length === 0) return;
2125

@@ -30,7 +34,8 @@ export async function createWaitpointTag({
3034
projectId,
3135
},
3236
undefined,
33-
residency
37+
residency,
38+
shardKey
3439
);
3540
} catch (error) {
3641
if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") {

apps/webapp/app/routes/api.v1.waitpoints.tokens.ts

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import {
1616
type PrismaClientOrTransaction,
1717
} from "~/db.server";
1818
import { resolveRunIdMintKind } from "~/v3/engineVersion.server";
19+
import { resolveMintShard } from "~/v3/runOpsMigration/runOpsMintShard.server";
1920
import { logger } from "~/services/logger.server";
2021
import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
2122
import { publicAccessTokenResponseHeaders } from "~/services/publicAccessTokenResponse.server";
@@ -69,6 +70,15 @@ const { action } = createActionApiRoute(
6970
});
7071
const residency = mintKind === "runOpsId" ? "NEW" : "LEGACY";
7172

73+
// No extra query: the org flags are already loaded on the authenticated env.
74+
const standaloneShardKey =
75+
mintKind === "runOpsId"
76+
? await resolveMintShard({
77+
id: authentication.environment.id,
78+
orgFeatureFlags: authentication.environment.organization.featureFlags,
79+
})
80+
: undefined;
81+
7282
//upsert tags
7383
let tags: { id: string; name: string }[] = [];
7484
const bodyTags = typeof body.tags === "string" ? [body.tags] : body.tags;
@@ -86,6 +96,7 @@ const { action } = createActionApiRoute(
8696
environmentId: authentication.environment.id,
8797
projectId: authentication.environment.projectId,
8898
residency,
99+
shardKey: standaloneShardKey,
89100
});
90101
if (tagRecord) {
91102
tags.push(tagRecord);
@@ -101,6 +112,7 @@ const { action } = createActionApiRoute(
101112
timeout,
102113
tags: bodyTags,
103114
standaloneResidency: residency,
115+
standaloneShardKey,
104116
});
105117

106118
const waitpointId = WaitpointId.toFriendlyId(result.waitpoint.id);

apps/webapp/app/runEngine/services/triggerFailedTask.server.ts

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,15 @@
11
import type { RunEngine } from "@internal/run-engine";
22
import { TaskRunErrorCodes, type TaskRunError } from "@trigger.dev/core/v3";
3-
import { RunId, generateRunOpsId } from "@trigger.dev/core/v3/isomorphic";
3+
import { RunId } from "@trigger.dev/core/v3/isomorphic";
44
import type {
55
PrismaClientOrTransaction,
66
RuntimeEnvironmentType,
77
TaskRun,
88
} from "@trigger.dev/database";
99
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
1010
import { logger } from "~/services/logger.server";
11-
import { resolveRunIdMintKind } from "~/v3/engineVersion.server";
12-
import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server";
11+
import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
12+
import { resolveRunMintTarget } from "~/v3/runOpsMigration/resolveRunMintTarget.server";
1313
import { getEventRepository } from "~/v3/eventRepository/index.server";
1414
import { runStore as defaultRunStore } from "~/v3/runStore.server";
1515
import type { RunStore } from "@internal/run-store";
@@ -103,17 +103,16 @@ export class TriggerFailedTaskService {
103103
return args.runFriendlyId;
104104
}
105105

106-
const mintKind = args.parentRunFriendlyId
107-
? resolveInheritedMintKind(args.parentRunFriendlyId)
108-
: await resolveRunIdMintKind({
106+
return mintFriendlyIdForKind(
107+
await resolveRunMintTarget({
108+
environment: {
109109
organizationId: args.organizationId,
110110
id: args.environmentId,
111111
orgFeatureFlags: args.orgFeatureFlags,
112-
});
113-
114-
return mintKind === "runOpsId"
115-
? RunId.toFriendlyId(generateRunOpsId())
116-
: RunId.generate().friendlyId;
112+
},
113+
parentRunFriendlyId: args.parentRunFriendlyId,
114+
})
115+
);
117116
}
118117

119118
async call(request: TriggerFailedTaskRequest): Promise<string | null> {

apps/webapp/app/runEngine/services/triggerTask.server.ts

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -28,9 +28,8 @@ import { parseDelay } from "~/utils/delays";
2828
import { removeNullBytesFromKey } from "~/utils/nullBytes";
2929
import { handleMetadataPacket } from "~/utils/packets";
3030
import { startSpan } from "~/v3/tracing.server";
31-
import { resolveRunIdMintKind } from "~/v3/engineVersion.server";
32-
import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server";
3331
import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
32+
import { resolveRunMintTarget } from "~/v3/runOpsMigration/resolveRunMintTarget.server";
3433
import type {
3534
TriggerTaskServiceOptions,
3635
TriggerTaskServiceResult,
@@ -218,15 +217,17 @@ export class RunEngineTriggerTaskService {
218217
parentRunFriendlyId?: string,
219218
region?: string
220219
): Promise<string> {
221-
const mintKind = parentRunFriendlyId
222-
? resolveInheritedMintKind(parentRunFriendlyId)
223-
: await resolveRunIdMintKind({
220+
return mintFriendlyIdForKind(
221+
await resolveRunMintTarget({
222+
environment: {
224223
organizationId: environment.organizationId,
225224
id: environment.id,
226225
orgFeatureFlags: environment.organization.featureFlags,
227-
});
228-
229-
return mintFriendlyIdForKind(mintKind, region);
226+
},
227+
parentRunFriendlyId,
228+
region,
229+
})
230+
);
230231
}
231232

232233
public async call({

apps/webapp/app/v3/runEngineHandlers.server.ts

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import {
1111
runOpsNewPrismaClient,
1212
runOpsNewReplicaClient,
1313
runOpsLegacyPrismaClient,
14+
runOpsShardHandles,
1415
} from "~/db.server";
1516
import { env } from "~/env.server";
1617
import { findEnvironmentById, findEnvironmentFromRun } from "~/models/runtimeEnvironment.server";
@@ -1060,6 +1061,7 @@ export function setupBatchQueueCallbacks() {
10601061
newReplica: runOpsNewReplicaClient,
10611062
newWriter: runOpsNewPrismaClient,
10621063
legacyWriter: runOpsLegacyPrismaClient,
1064+
shards: runOpsShardHandles,
10631065
tryCompleteBatch: (batchId) => engine.tryCompleteBatch({ batchId }),
10641066
});
10651067
});

apps/webapp/app/v3/runEngineHandlersShared.server.ts

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
* whole webapp service graph). The handlers wire the production defaults; tests
55
* inject per-container stores/replicas, so these helpers never import db.server.
66
*/
7+
import { resolveShard } from "@trigger.dev/core/v3/isomorphic";
78
import type { CompleteBatchResult } from "@internal/run-engine";
89
import type { RunOpsPrismaClient } from "@internal/run-ops-database";
910
import type { RunStore } from "@internal/run-store";
@@ -83,8 +84,23 @@ export async function resolveBatchRunOpsWriter(
8384
newReplica: RunOpsPrismaClient;
8485
newWriter: RunOpsPrismaClient;
8586
legacyWriter: RunOpsPrismaClient;
87+
shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>;
8688
}
8789
): Promise<RunOpsPrismaClient> {
90+
// The probe below is binary, so without this a gen-2 batch resolves to a store holding no such
91+
// row, and the update throws before the batch waitpoint completes.
92+
const shardKey = resolveShard(batchId);
93+
if (shardKey !== "new" && shardKey !== "legacy") {
94+
const shard = deps.shards?.find((s) => s.key === shardKey);
95+
if (!shard) {
96+
// Writing to a guessed store is what strands a run. Fail loud instead.
97+
throw new Error(
98+
`resolveBatchRunOpsWriter: batch "${batchId}" names shard "${shardKey}", which is not configured`
99+
);
100+
}
101+
return shard.writer;
102+
}
103+
88104
const onNew = await deps.newReplica.batchTaskRun.findFirst({
89105
where: { id: batchId },
90106
select: { id: true },
@@ -106,6 +122,7 @@ export type BatchCompletionDeps = {
106122
newReplica: RunOpsPrismaClient;
107123
newWriter: RunOpsPrismaClient;
108124
legacyWriter: RunOpsPrismaClient;
125+
shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>;
109126
tryCompleteBatch: (batchId: string) => Promise<unknown>;
110127
};
111128

@@ -136,6 +153,7 @@ export async function handleBatchCompletion(
136153
newReplica: deps.newReplica,
137154
newWriter: deps.newWriter,
138155
legacyWriter: deps.legacyWriter,
156+
shards: deps.shards,
139157
});
140158

141159
try {
Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
import { describe, expect, it, vi } from "vitest";
2+
import { classifyKind, mintWaitpointIdFor, resolveShard } from "@trigger.dev/core/v3/isomorphic";
3+
import { resolveInheritedMintKind } from "./resolveInheritedMintKind.server";
4+
import {
5+
mintAnchoredRunFriendlyId,
6+
mintFriendlyIdForKind,
7+
} from "./mintAnchoredRunFriendlyId.server";
8+
import { batchIdForMintKind } from "./mintBatchFriendlyId.server";
9+
import { resolveRunMintTarget } from "./resolveRunMintTarget.server";
10+
11+
// Gate off means resolveMintShard answers "new". Every assertion is "the id is what it was".
12+
const offShard = vi.fn().mockResolvedValue("new" as const);
13+
const environment = { organizationId: "org_1", id: "env_1", orgFeatureFlags: {} };
14+
15+
describe("gate off — run mint paths", () => {
16+
it("a root run on the run-ops path mints a gen-1 v1 id", async () => {
17+
const target = await resolveRunMintTarget({
18+
environment,
19+
region: "us-east-1",
20+
deps: {
21+
resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"),
22+
resolveMintShard: offShard,
23+
},
24+
});
25+
const body = mintFriendlyIdForKind(target).slice(4);
26+
expect(body.length).toBe(26);
27+
expect(body[24]).toBe("e"); // the region char, as today
28+
expect(body[25]).toBe("1");
29+
});
30+
31+
it("a root run on a non-cut-over org mints a cuid", async () => {
32+
const target = await resolveRunMintTarget({
33+
environment,
34+
deps: {
35+
resolveRunIdMintKind: vi.fn().mockResolvedValue("cuid"),
36+
resolveMintShard: offShard,
37+
},
38+
});
39+
expect(mintFriendlyIdForKind(target).slice(4).length).toBe(25);
40+
});
41+
42+
it("a child of a gen-1 parent keeps the caller's region char", async () => {
43+
// The pre-split code passed the region on both arms; dropping it on the inherited arm would
44+
// silently stamp the default.
45+
const target = await resolveRunMintTarget({
46+
environment,
47+
parentRunFriendlyId: `run_${"a".repeat(24)}01`,
48+
region: "us-east-1",
49+
deps: {
50+
resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"),
51+
resolveMintShard: offShard,
52+
},
53+
});
54+
const body = mintFriendlyIdForKind(target).slice(4);
55+
expect(body[24]).toBe("e");
56+
expect(body[25]).toBe("1");
57+
});
58+
59+
it("a gen-2 parent's shard still outranks the caller's region", async () => {
60+
const target = await resolveRunMintTarget({
61+
environment,
62+
parentRunFriendlyId: `run_${"a".repeat(24)}a2`,
63+
region: "us-east-1",
64+
deps: {
65+
resolveRunIdMintKind: vi.fn().mockResolvedValue("runOpsId"),
66+
resolveMintShard: offShard,
67+
},
68+
});
69+
expect(mintFriendlyIdForKind(target).slice(4)[24]).toBe("a");
70+
});
71+
72+
it("a child of a gen-1 parent mints a gen-1 v1 id", () => {
73+
const body = mintFriendlyIdForKind(resolveInheritedMintKind(`run_${"a".repeat(24)}01`)).slice(
74+
4
75+
);
76+
expect(body[25]).toBe("1");
77+
});
78+
79+
it("a child of a cuid parent mints a cuid", () => {
80+
expect(
81+
mintFriendlyIdForKind(resolveInheritedMintKind(`run_${"b".repeat(25)}`)).slice(4).length
82+
).toBe(25);
83+
});
84+
});
85+
86+
describe("gate off — batch and item paths", () => {
87+
it("a batch with no shard char mints a gen-1 v1 id", () => {
88+
const r = batchIdForMintKind({ kind: "runOpsId" });
89+
expect(r.id.length).toBe(26);
90+
expect(r.id[25]).toBe("1");
91+
expect(classifyKind(r.id)).toBe("runOpsId");
92+
});
93+
94+
it("a batch on a non-cut-over org mints a cuid", () => {
95+
expect(batchIdForMintKind({ kind: "cuid" }).id.length).toBe(25);
96+
});
97+
98+
it("a batch item anchored on a gen-1 batch mints a gen-1 v1 id", () => {
99+
const body = mintAnchoredRunFriendlyId(`batch_${"a".repeat(24)}01`).slice(4);
100+
expect(body[25]).toBe("1");
101+
});
102+
});
103+
104+
describe("gate off — waitpoint paths", () => {
105+
it("every gen-1 or legacy anchor yields a cuid waitpoint id", () => {
106+
for (const anchor of [`${"a".repeat(24)}01`, "c".repeat(25), undefined]) {
107+
const r = mintWaitpointIdFor(anchor);
108+
expect(r.id.length).toBe(25);
109+
expect(resolveShard(r.id)).toBe("legacy");
110+
}
111+
});
112+
});

apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,4 +28,16 @@ describe("mintAnchoredRunFriendlyId", () => {
2828
expect(parsed.format).toBe("b32hex");
2929
expect(parsed.format === "b32hex" && parsed.region).toBe(REGION_CODES["us-east-1"]);
3030
});
31+
32+
it("a gen-2 batch anchor mints an item on the batch's shard", () => {
33+
const body = mintAnchoredRunFriendlyId(`batch_${"a".repeat(24)}a2`).slice("run_".length);
34+
expect(body).toHaveLength(26);
35+
expect(body[24]).toBe("a");
36+
expect(body[25]).toBe("2");
37+
});
38+
39+
it("a gen-2 batch anchor ignores a caller region: the shard owns index 24", () => {
40+
const body = mintAnchoredRunFriendlyId(`batch_${"a".repeat(24)}a2`, "us-east-1").slice(4);
41+
expect(body[24]).toBe("a");
42+
});
3143
});
Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,20 @@
1-
import { generateRunOpsId, RunId, type ResidencyKind } from "@trigger.dev/core/v3/isomorphic";
1+
import { generateRunOpsId, generateRunOpsIdV2, RunId } from "@trigger.dev/core/v3/isomorphic";
2+
import type { MintTarget } from "./mintTarget";
23
import { resolveInheritedMintKind } from "./resolveInheritedMintKind.server";
34

4-
// Shared id-generation branch for every run-mint path: "runOpsId" -> NEW store, "cuid" -> LEGACY.
5-
export function mintFriendlyIdForKind(mintKind: ResidencyKind, region?: string): string {
6-
return mintKind === "runOpsId"
7-
? RunId.toFriendlyId(generateRunOpsId(region))
8-
: RunId.generate().friendlyId;
5+
// A shardChar selects one gen-2 shard and takes index 24; without one the region takes that slot.
6+
export function mintFriendlyIdForKind(target: MintTarget): string {
7+
if (target.kind !== "runOpsId") {
8+
return RunId.generate().friendlyId;
9+
}
10+
11+
return RunId.toFriendlyId(
12+
target.shardChar ? generateRunOpsIdV2(target.shardChar) : generateRunOpsId(target.region)
13+
);
914
}
1015

1116
// Anchor a batch item's mint on the BATCH's friendlyId (id-shape, zero I/O), never the per-org
1217
// flag, so the item and its BatchTaskRun stay co-resident across a mid-batch flag flip.
1318
export function mintAnchoredRunFriendlyId(batchFriendlyId: string, region?: string): string {
14-
return mintFriendlyIdForKind(resolveInheritedMintKind(batchFriendlyId), region);
19+
return mintFriendlyIdForKind({ ...resolveInheritedMintKind(batchFriendlyId), region });
1520
}

0 commit comments

Comments
 (0)