From f2d9670d0ed9079501810771e21b3b7e601f614d Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Fri, 21 Aug 2026 17:54:43 +0100 Subject: [PATCH] feat(webapp): resolve which shard an environment mints run roots into Adds the third stage of the run-id mint gate chain. `resolveMintShard(env)` returns the shard key an environment mints new roots into: the active shard list, then a per-env or per-org pin, then a rendezvous hash of the environment id. With `RUN_OPS_MINT_SHARDS` unset or empty it returns "new", which is today's behaviour, so this merges inert. `computeRunIdMintKind` and `mintFlipGrace.ts` are untouched. The grace pattern is cloned into `mintShardGrace.ts` rather than widened, so the existing cuid/runOpsId flip grace keeps its behaviour. Design notes: - Pure core plus env-bound wrapper, mirroring `runOpsMintKind.server.ts`. Determinism is a property of `computeMintShard` for fixed deps; the wrapper supplies the clock, exactly as `effectiveMintKind` takes `nowMs`. - Zero new queries on the trigger hot path. Both pins live in the org override blob that `mintRunFriendlyId` already holds. - HRW scores `sha256(envId \0 key)` at 64 bits, over a sorted key list, with a lexicographic tie-break. A 32-bit score collides at our environment count, and without the sort two deployments listing the same keys in a different CSV order would place environments differently. - `parseShardCsv` rejects anything outside [a-z0-9] and rejects the reserved keys at boot. `generateRunOpsIdV2` throws on an out-of-alphabet char, so an unvalidated key would become a throw on the mint path. - A pin outside the active set falls through to the hash and reports once per environment per process. Honouring it would leak the drain the active list performs; throwing would fail customer triggers whenever a pinned shard drains. The loud-on-unknown-key rule governs reading an id, not writing one. - "new" is a legal pin value, holding one org or environment on gen-1 while the rest of the fleet mints gen-2. Without it, a non-empty active set moves every environment at once. - The active-set grace is stamped by `RUN_OPS_MINT_SHARDS_PREV` and `RUN_OPS_MINT_SHARDS_FLIPPED_AT`. A prev list with no timestamp is dropped; a timestamp with an empty prev list graces a first activation. No changeset and no `.server-changes` note: nothing user-visible, and no caller carries the returned key into an id yet. --- apps/webapp/app/env.server.ts | 27 ++ apps/webapp/app/v3/featureFlags.ts | 37 ++- .../v3/runOpsMigration/mintShardGrace.test.ts | 148 +++++++++++ .../app/v3/runOpsMigration/mintShardGrace.ts | 95 +++++++ .../runOpsMintShard.server.test.ts | 248 ++++++++++++++++++ .../runOpsMigration/runOpsMintShard.server.ts | 169 ++++++++++++ 6 files changed, 722 insertions(+), 2 deletions(-) create mode 100644 apps/webapp/app/v3/runOpsMigration/mintShardGrace.test.ts create mode 100644 apps/webapp/app/v3/runOpsMigration/mintShardGrace.ts create mode 100644 apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.test.ts create mode 100644 apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index c9179306124..a7c572cd34e 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -4,6 +4,7 @@ import { BoolEnv } from "./utils/boolEnv"; import { isValidDatabaseUrl } from "./utils/db"; import { isValidRegex } from "./utils/regex"; import { isValidDuration } from "./services/realtime/duration.server"; +import { parseShardCsv } from "./v3/runOpsMigration/mintShardGrace"; // `z.string()` constrained to a `parseDuration`-parseable string (e.g. // `7d`, `1h`). Validated at boot so a typo'd duration fails fast. @@ -41,6 +42,23 @@ const parseMachinePresetCsv = (raw: string, ctx: z.RefinementCtx): MachinePreset return out; }; +// A CSV of gen-2 mint shard keys, validated at boot by parseShardCsv. Kept as the raw string: +// the resolution is built once in runOpsMintShard.server.ts, and this only has to fail fast. +const shardCsvString = () => + z + .string() + .default("") + .superRefine((raw, ctx) => { + try { + parseShardCsv(raw); + } catch (error) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: error instanceof Error ? error.message : "invalid shard key CSV", + }); + } + }); + const GithubAppEnvSchema = z.preprocess( (val) => { const obj = val as any; @@ -1998,6 +2016,15 @@ const EnvironmentSchema = z // (stale or fresh) resolves to the same kind for the whole window. See mintFlipGrace.ts. RUN_OPS_MINT_FLIP_GRACE_MS: z.coerce.number().int().default(90_000), + // Gen-2 mint shards — CSV of single-char [a-z0-9] keys eligible for ROOT minting. Unset or + // empty means no gen-2 minting, which is today's behaviour. Validated at boot: an invalid + // key would mint an id that cannot be routed. _PREV + _FLIPPED_AT stamp a set change so + // every process crosses the cutover together; set both, or the grace never applies. + // Removing a key stops new roots on it and never stops routing it. See mintShardGrace.ts. + RUN_OPS_MINT_SHARDS: shardCsvString(), + RUN_OPS_MINT_SHARDS_PREV: shardCsvString(), + RUN_OPS_MINT_SHARDS_FLIPPED_AT: z.string().datetime().optional(), + // Session replication (Postgres → ClickHouse sessions_v1). Shares Redis // with the runs replicator for leader locking but has its own slot and // publication so the two consume independently. diff --git a/apps/webapp/app/v3/featureFlags.ts b/apps/webapp/app/v3/featureFlags.ts index 7c775799178..740fccebb87 100644 --- a/apps/webapp/app/v3/featureFlags.ts +++ b/apps/webapp/app/v3/featureFlags.ts @@ -26,6 +26,9 @@ export const FEATURE_FLAG = { // Grace-linger stamp carried alongside runOpsMintKind on flip. See mintFlipGrace.ts. runOpsMintKindPrev: "runOpsMintKindPrev", runOpsMintKindFlippedAt: "runOpsMintKindFlippedAt", + // Gen-2 mint shard pins, read from the org override blob only. See runOpsMintShard.server.ts. + runOpsMintShard: "runOpsMintShard", + runOpsMintShardEnvPins: "runOpsMintShardEnvPins", queueMetricsUiEnabled: "queueMetricsUiEnabled", // Per-organization rollout for creating additional environment API keys. additionalApiKeysEnabled: "additionalApiKeysEnabled", @@ -89,6 +92,32 @@ export const FeatureFlagCatalog = { // by stampMintKindFlip on a genuine flip. Display-only (see ORG_LOCKED_FLAGS). [FEATURE_FLAG.runOpsMintKindPrev]: z.enum(["cuid", "runOpsId"]), [FEATURE_FLAG.runOpsMintKindFlippedAt]: z.string().datetime(), + // Pins one org to a gen-2 mint shard. "new" holds the org on gen-1 run-ops ids, which is how + // a canary keeps the fleet's default while one org moves. Only honored while the key is in + // the active set (RUN_OPS_MINT_SHARDS); a drained key falls through to the hash. + [FEATURE_FLAG.runOpsMintShard]: z + .string() + .refine((v) => v === "new" || /^[a-z0-9]$/.test(v), 'must be a single [a-z0-9] char, or "new"'), + // Per-environment pins as JSON: {"": ""}. A JSON string because + // this catalog is scalar-only. Rejected at write, so a typo cannot silently un-pin an env. + [FEATURE_FLAG.runOpsMintShardEnvPins]: z.string().superRefine((raw, ctx) => { + const fail = (message: string) => ctx.addIssue({ code: z.ZodIssueCode.custom, message }); + + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return fail("must be valid JSON"); + } + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + return fail("must be a JSON object mapping environment id to shard key"); + } + for (const [environmentId, value] of Object.entries(parsed)) { + if (typeof value !== "string" || !(value === "new" || /^[a-z0-9]$/.test(value))) { + fail(`"${environmentId}" must map to a single [a-z0-9] char, or "new"`); + } + } + }), // Per-org access to the Queue Metrics dashboard UI (view only; emission is global and // separate). Off unless enabled for the org. [FEATURE_FLAG.queueMetricsUiEnabled]: z.coerce.boolean(), @@ -101,11 +130,15 @@ export const FeatureFlagCatalog = { export type FeatureFlagKey = keyof typeof FeatureFlagCatalog; -// Infrastructure flags that are read-only on the global flags page. -// Shown with current/resolved value but no controls. +// Infrastructure flags, plus org-scoped-only flags, that are read-only on the global flags +// page. Shown with current/resolved value but no controls. An org-scoped-only flag belongs +// here because its resolver never reads a global row, so an editable global control would +// offer a setting that does nothing. export const GLOBAL_LOCKED_FLAGS: FeatureFlagKey[] = [ FEATURE_FLAG.defaultWorkerInstanceGroupId, FEATURE_FLAG.taskEventRepository, + FEATURE_FLAG.runOpsMintShard, + FEATURE_FLAG.runOpsMintShardEnvPins, ]; // Flags that are read-only on the org-level dialog. diff --git a/apps/webapp/app/v3/runOpsMigration/mintShardGrace.test.ts b/apps/webapp/app/v3/runOpsMigration/mintShardGrace.test.ts new file mode 100644 index 00000000000..9358685e1e8 --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/mintShardGrace.test.ts @@ -0,0 +1,148 @@ +import { describe, expect, it } from "vitest"; +import { generateRunOpsIdV2 } from "@trigger.dev/core/v3/isomorphic"; +import { + buildMintShardResolution, + effectiveMintShardSet, + isValidPinValue, + parseShardCsv, + SHARD_KEY_PATTERN, + type MintShardSetResolution, +} from "./mintShardGrace"; + +const GRACE_MS = 90_000; +const T = 1_000_000; + +describe("parseShardCsv", () => { + it("returns an empty list for unset, empty and whitespace input", () => { + expect(parseShardCsv(undefined)).toEqual([]); + expect(parseShardCsv("")).toEqual([]); + expect(parseShardCsv(" ")).toEqual([]); + expect(parseShardCsv(",,")).toEqual([]); + }); + + it("trims, dedupes and SORTS, so operator typing order cannot change HRW", () => { + expect(parseShardCsv("b, a ,b")).toEqual(["a", "b"]); + expect(parseShardCsv("a,b,c")).toEqual(parseShardCsv("c,b,a")); + expect(parseShardCsv("b,c,a")).toEqual(parseShardCsv("a,c,b")); + }); + + it("accepts every one of the 36 legal shard keys", () => { + const all = "abcdefghijklmnopqrstuvwxyz0123456789".split(""); + expect(parseShardCsv(all.join(","))).toEqual([...all].sort()); + }); + + it("throws on a key outside [a-z0-9]", () => { + // generateRunOpsIdV2 throws on these; an unvalidated key MUST fail at boot, not at mint. + expect(() => parseShardCsv("A")).toThrow(/shard key/i); + expect(() => parseShardCsv("ab")).toThrow(/shard key/i); + expect(() => parseShardCsv("a,-")).toThrow(/shard key/i); + expect(() => parseShardCsv("a,_")).toThrow(/shard key/i); + }); + + it("rejects the reserved keys by name", () => { + expect(() => parseShardCsv("new")).toThrow(/reserved/i); + expect(() => parseShardCsv("a,legacy")).toThrow(/reserved/i); + }); +}); + +// Core does not export its shard-char pattern, so pin the local one to the real minter. +describe("shard alphabet agrees with the core minter", () => { + it("accepts exactly the characters generateRunOpsIdV2 accepts", () => { + const candidates = [ + ..."abcdefghijklmnopqrstuvwxyz0123456789".split(""), + ..."ABZ-_. +/é!".split(""), + "", + "ab", + ]; + + for (const candidate of candidates) { + let minterAccepts = true; + try { + generateRunOpsIdV2(candidate); + } catch { + minterAccepts = false; + } + + expect(SHARD_KEY_PATTERN.test(candidate)).toBe(minterAccepts); + } + }); +}); + +describe("isValidPinValue", () => { + it('accepts a shard key, and accepts "new" as the gen-1 hold value', () => { + expect(isValidPinValue("a")).toBe(true); + expect(isValidPinValue("7")).toBe(true); + expect(isValidPinValue("new")).toBe(true); + }); + + it("rejects legacy, and rejects anything outside the alphabet", () => { + expect(isValidPinValue("legacy")).toBe(false); + expect(isValidPinValue("A")).toBe(false); + expect(isValidPinValue("ab")).toBe(false); + expect(isValidPinValue("")).toBe(false); + }); +}); + +describe("effectiveMintShardSet", () => { + it("returns set when there is no stamp", () => { + const r: MintShardSetResolution = { set: ["a", "b"] }; + expect(effectiveMintShardSet(r, T, GRACE_MS)).toEqual(["a", "b"]); + }); + + it("returns set when flippedAtMs is absent even though prevSet is present", () => { + const r: MintShardSetResolution = { set: ["a", "b"], prevSet: ["a"] }; + expect(effectiveMintShardSet(r, T, GRACE_MS)).toEqual(["a", "b"]); + }); + + it("serves prevSet inside the window and set at/after the boundary", () => { + const r: MintShardSetResolution = { set: ["a", "b"], prevSet: ["a"], flippedAtMs: T }; + expect(effectiveMintShardSet(r, T, GRACE_MS)).toEqual(["a"]); + expect(effectiveMintShardSet(r, T + GRACE_MS - 1, GRACE_MS)).toEqual(["a"]); + // Boundary is exclusive on the prev side, so every process crosses it together. + expect(effectiveMintShardSet(r, T + GRACE_MS, GRACE_MS)).toEqual(["a", "b"]); + expect(effectiveMintShardSet(r, T + GRACE_MS + 1, GRACE_MS)).toEqual(["a", "b"]); + }); + + it("represents a graced first activation as an empty prevSet", () => { + const r: MintShardSetResolution = { set: ["a"], prevSet: [], flippedAtMs: T }; + expect(effectiveMintShardSet(r, T, GRACE_MS)).toEqual([]); + expect(effectiveMintShardSet(r, T + GRACE_MS, GRACE_MS)).toEqual(["a"]); + }); + + it("serves a drain through the window", () => { + const r: MintShardSetResolution = { set: ["a"], prevSet: ["a", "b"], flippedAtMs: T }; + expect(effectiveMintShardSet(r, T + 1, GRACE_MS)).toEqual(["a", "b"]); + expect(effectiveMintShardSet(r, T + GRACE_MS, GRACE_MS)).toEqual(["a"]); + }); +}); + +describe("buildMintShardResolution", () => { + it("omits prevSet entirely when no flip timestamp is configured", () => { + // A prevSet with no timestamp can never apply, so it MUST NOT linger. + const r = buildMintShardResolution({ shards: "a,b", prev: "a", flippedAt: undefined }); + expect(r).toEqual({ set: ["a", "b"], prevSet: undefined, flippedAtMs: undefined }); + }); + + it("keeps an empty prevSet when a flip timestamp IS configured", () => { + const r = buildMintShardResolution({ + shards: "a", + prev: "", + flippedAt: new Date(T).toISOString(), + }); + expect(r).toEqual({ set: ["a"], prevSet: [], flippedAtMs: T }); + }); + + it("parses the flip timestamp and sorts both lists", () => { + const r = buildMintShardResolution({ + shards: "b,a", + prev: "c,a", + flippedAt: new Date(T).toISOString(), + }); + expect(r).toEqual({ set: ["a", "b"], prevSet: ["a", "c"], flippedAtMs: T }); + }); + + it("treats an unparseable timestamp as no stamp at all", () => { + const r = buildMintShardResolution({ shards: "a", prev: "b", flippedAt: "not-a-date" }); + expect(r).toEqual({ set: ["a"], prevSet: undefined, flippedAtMs: undefined }); + }); +}); diff --git a/apps/webapp/app/v3/runOpsMigration/mintShardGrace.ts b/apps/webapp/app/v3/runOpsMigration/mintShardGrace.ts new file mode 100644 index 00000000000..bfa3139cc25 --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/mintShardGrace.ts @@ -0,0 +1,95 @@ +import type { ShardKey } from "@trigger.dev/core/v3/isomorphic"; + +// Index 24 of a gen-2 id sits inside the pod name `runner-`, and a DNS-1123 label accepts +// lowercase only, so the alphabet is 36 keys and no wider. Core keeps its copy private; +// mintShardGrace.test.ts pins this pattern to generateRunOpsIdV2 instead. +export const SHARD_KEY_PATTERN = /^[a-z0-9]$/; + +// Neither may enter the active set: "new" already means "mint a gen-1 run-ops id" and +// "legacy" means the cuid store, which minting never selects. +const RESERVED_SHARD_KEYS: readonly string[] = ["new", "legacy"]; + +// "new" IS legal as a PIN, holding one org or environment on gen-1 while the rest of the fleet +// mints gen-2. Without it a non-empty active set moves every environment at once. +export const GEN_1_PIN_VALUE = "new"; + +export type MintShardSetResolution = { + set: string[]; + prevSet?: string[]; + flippedAtMs?: number; +}; + +export function isValidPinValue(value: unknown): value is ShardKey { + if (typeof value !== "string") return false; + return value === GEN_1_PIN_VALUE || SHARD_KEY_PATTERN.test(value); +} + +// Throws rather than dropping a bad key: generateRunOpsIdV2 throws on an out-of-alphabet char, +// so an unvalidated key must fail at boot and never at mint. +export function parseShardCsv(raw: string | undefined | null): string[] { + const keys = (raw ?? "") + .split(",") + .map((s) => s.trim()) + .filter(Boolean); + + const unique = new Set(); + for (const key of keys) { + if (RESERVED_SHARD_KEYS.includes(key)) { + throw new Error(`"${key}" is a reserved key and cannot be an active mint shard`); + } + if (!SHARD_KEY_PATTERN.test(key)) { + throw new Error(`invalid shard key "${key}": must be a single char in [a-z0-9]`); + } + unique.add(key); + } + + // Sorted so no placement can depend on the order an operator typed the CSV in. + return [...unique].sort(); +} + +// Cutover boundary, mirroring effectiveMintKind. `nowMs` is the reader's wall clock while +// `flippedAtMs` is operator-supplied, so this assumes NTP-synced hosts with skew << graceMs, +// letting every process cross [flippedAtMs, flippedAtMs + graceMs) together (OLD then NEW). +export function effectiveMintShardSet( + r: MintShardSetResolution, + nowMs: number, + graceMs: number +): string[] { + if (r.prevSet === undefined || r.flippedAtMs === undefined) { + return r.set; + } + return nowMs < r.flippedAtMs + graceMs ? r.prevSet : r.set; +} + +// A prevSet with no timestamp can never apply, so it is dropped. A timestamp with an EMPTY +// prevSet is meaningful: it graces a first activation, serving no shards for the window. +export function buildMintShardResolution(source: { + shards: string | undefined; + prev: string | undefined; + flippedAt: string | undefined; +}): MintShardSetResolution { + const parsed = source.flippedAt !== undefined ? Date.parse(source.flippedAt) : NaN; + const flippedAtMs = Number.isNaN(parsed) ? undefined : parsed; + + return { + set: parseShardCsv(source.shards), + prevSet: flippedAtMs === undefined ? undefined : parseShardCsv(source.prev), + flippedAtMs, + }; +} + +// Returns a message to log when the stamp is half-configured, otherwise undefined. Stays quiet +// while the active set is empty, so an unconfigured deployment logs nothing at boot. +export function mintShardStampWarning(source: { + shards: string | undefined; + prev: string | undefined; + flippedAt: string | undefined; +}): string | undefined { + if (parseShardCsv(source.shards).length === 0) { + return undefined; + } + if (parseShardCsv(source.prev).length > 0 && source.flippedAt === undefined) { + return "RUN_OPS_MINT_SHARDS_PREV is set but RUN_OPS_MINT_SHARDS_FLIPPED_AT is not; the shard-set grace window will never apply"; + } + return undefined; +} diff --git a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.test.ts b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.test.ts new file mode 100644 index 00000000000..806a0c1aea8 --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.test.ts @@ -0,0 +1,248 @@ +import { describe, expect, it } from "vitest"; +import { computeMintShard, type MintShardDeps } from "./runOpsMintShard.server"; +import { type MintShardSetResolution } from "./mintShardGrace"; + +const GRACE_MS = 90_000; +const T = 1_000_000; + +// Cuid-shaped ids, not sequential integers: a sequential space does not model the real +// key distribution the hash has to spread. +function envIds(count: number): string[] { + const ids: string[] = []; + for (let i = 0; i < count; i++) { + ids.push(`cm${(i * 2654435761).toString(36).padStart(10, "0")}${i.toString(36)}zzq`); + } + return ids; +} + +function deps( + resolution: MintShardSetResolution, + overrides: Partial = {} +): MintShardDeps { + return { + resolution, + nowMs: T + GRACE_MS + 1, + graceMs: GRACE_MS, + orgFeatureFlags: undefined, + ...overrides, + }; +} + +function orgFlags(flags: Record) { + return { orgFeatureFlags: flags }; +} + +function place(ids: string[], resolution: MintShardSetResolution): Map { + const out = new Map(); + for (const id of ids) { + out.set(id, computeMintShard({ id }, deps(resolution))); + } + return out; +} + +describe("computeMintShard — the no-shards answer", () => { + it("returns new when the active set is unset", () => { + expect(computeMintShard({ id: "env_1" }, deps({ set: [] }))).toBe("new"); + }); + + it("returns new when the active set is empty even with a stale stamp present", () => { + const resolution: MintShardSetResolution = { set: [], prevSet: ["a"], flippedAtMs: T }; + // The empty check MUST run before the grace, so an unset set is an unconditional kill switch. + expect(computeMintShard({ id: "env_1" }, deps(resolution, { nowMs: T + 1 }))).toBe("new"); + }); + + it("returns new when the grace serves an empty prevSet", () => { + const resolution: MintShardSetResolution = { set: ["a"], prevSet: [], flippedAtMs: T }; + expect(computeMintShard({ id: "env_1" }, deps(resolution, { nowMs: T + 1 }))).toBe("new"); + }); +}); + +describe("computeMintShard — determinism", () => { + it("returns the same value for the same environment on every call", () => { + const resolution: MintShardSetResolution = { set: ["a", "b", "c"] }; + const first = computeMintShard({ id: "env_stable" }, deps(resolution)); + for (let i = 0; i < 1000; i++) { + expect(computeMintShard({ id: "env_stable" }, deps(resolution))).toBe(first); + } + }); + + it("ignores the order the operator listed the keys in", () => { + const ids = envIds(200); + const canonical = place(ids, { set: ["a", "b", "c"] }); + for (const permutation of [ + ["c", "b", "a"], + ["b", "a", "c"], + ["a", "c", "b"], + ]) { + expect(place(ids, { set: permutation })).toEqual(canonical); + } + }); +}); + +describe("computeMintShard — pins", () => { + const resolution: MintShardSetResolution = { set: ["a", "b"] }; + + it("lets a per-env pin override the hash", () => { + const ids = envIds(50); + for (const id of ids) { + const pinned = computeMintShard( + { id }, + deps(resolution, orgFlags({ runOpsMintShardEnvPins: JSON.stringify({ [id]: "b" }) })) + ); + expect(pinned).toBe("b"); + } + }); + + it("lets a per-org pin override the hash when no per-env pin is set", () => { + const ids = envIds(50); + for (const id of ids) { + expect(computeMintShard({ id }, deps(resolution, orgFlags({ runOpsMintShard: "a" })))).toBe( + "a" + ); + } + }); + + it("lets a per-env pin beat a per-org pin", () => { + const result = computeMintShard( + { id: "env_1" }, + deps( + resolution, + orgFlags({ + runOpsMintShard: "a", + runOpsMintShardEnvPins: JSON.stringify({ env_1: "b" }), + }) + ) + ); + expect(result).toBe("b"); + }); + + it("holds an environment on gen-1 when the pin is new", () => { + expect( + computeMintShard({ id: "env_1" }, deps(resolution, orgFlags({ runOpsMintShard: "new" }))) + ).toBe("new"); + expect( + computeMintShard( + { id: "env_1" }, + deps(resolution, orgFlags({ runOpsMintShardEnvPins: JSON.stringify({ env_1: "new" }) })) + ) + ).toBe("new"); + }); + + it("falls through to the hash and reports when the pin is outside the active set", () => { + // Honouring a drained pin would leak the drain; throwing would fail customer triggers. + const rejected: string[] = []; + const result = computeMintShard( + { id: "env_1" }, + deps(resolution, { + ...orgFlags({ runOpsMintShard: "z" }), + onPinRejected: (info) => rejected.push(info.pin), + }) + ); + expect(result).toBe(computeMintShard({ id: "env_1" }, deps(resolution))); + expect(rejected).toEqual(["z"]); + }); + + it("honours a pin to a drained key for the whole grace window, then falls through", () => { + const draining: MintShardSetResolution = { set: ["a"], prevSet: ["a", "b"], flippedAtMs: T }; + const pinnedToB = orgFlags({ runOpsMintShard: "b" }); + expect(computeMintShard({ id: "env_1" }, deps(draining, { ...pinnedToB, nowMs: T + 1 }))).toBe( + "b" + ); + expect( + computeMintShard({ id: "env_1" }, deps(draining, { ...pinnedToB, nowMs: T + GRACE_MS })) + ).not.toBe("b"); + }); + + it("ignores an unparseable pin blob rather than un-pinning silently", () => { + const result = computeMintShard( + { id: "env_1" }, + deps(resolution, orgFlags({ runOpsMintShard: "a", runOpsMintShardEnvPins: "{not json" })) + ); + expect(result).toBe("a"); + }); + + it("falls back to the org pin when the blob holds an invalid value for this env", () => { + const result = computeMintShard( + { id: "env_1" }, + deps( + resolution, + orgFlags({ + runOpsMintShard: "a", + runOpsMintShardEnvPins: JSON.stringify({ env_1: "LEGACY" }), + }) + ) + ); + expect(result).toBe("a"); + }); + + it("ignores an invalid org pin value", () => { + const result = computeMintShard( + { id: "env_1" }, + deps(resolution, orgFlags({ runOpsMintShard: "legacy" })) + ); + expect(result).toBe(computeMintShard({ id: "env_1" }, deps(resolution))); + }); +}); + +describe("computeMintShard — rendezvous properties", () => { + const ids = envIds(10_000); + + it("spreads roughly evenly across the active set", () => { + for (const set of [ + ["a", "b"], + ["a", "b", "c"], + ["a", "b", "c", "d"], + ]) { + const counts = new Map(); + for (const shard of place(ids, { set }).values()) { + counts.set(shard, (counts.get(shard) ?? 0) + 1); + } + expect(counts.size).toBe(set.length); + const expected = ids.length / set.length; + for (const count of counts.values()) { + expect(Math.abs(count - expected) / expected).toBeLessThan(0.1); + } + } + }); + + it("moves about 1/(N+1) of environments when a shard is added", () => { + const cases: Array<{ from: string[]; to: string[]; expected: number }> = [ + { from: ["a"], to: ["a", "b"], expected: 1 / 2 }, + { from: ["a", "b"], to: ["a", "b", "c"], expected: 1 / 3 }, + { from: ["a", "b", "c"], to: ["a", "b", "c", "d"], expected: 1 / 4 }, + ]; + + for (const { from, to, expected } of cases) { + const before = place(ids, { set: from }); + const after = place(ids, { set: to }); + const added = to.filter((k) => !from.includes(k)); + let moved = 0; + for (const id of ids) { + if (before.get(id) === after.get(id)) continue; + moved++; + // HRW's defining property: a mover lands on the ADDED shard, never on a survivor. + expect(added).toContain(after.get(id)); + } + expect(Math.abs(moved / ids.length - expected) / expected).toBeLessThan(0.1); + } + }); + + it("moves only the environments that hashed to a removed shard", () => { + const before = place(ids, { set: ["a", "b", "c"] }); + const after = place(ids, { set: ["a", "b"] }); + for (const id of ids) { + if (before.get(id) === "c") { + expect(after.get(id)).not.toBe("c"); + } else { + expect(after.get(id)).toBe(before.get(id)); + } + } + }); + + it("also moves pinned environments when their shard is removed", () => { + // Criterion 6 is a property of the hash only. A pin to a removed key moves too. + const pinnedToC = orgFlags({ runOpsMintShard: "c" }); + expect(computeMintShard({ id: "env_1" }, deps({ set: ["a", "b", "c"] }, pinnedToC))).toBe("c"); + expect(computeMintShard({ id: "env_1" }, deps({ set: ["a", "b"] }, pinnedToC))).not.toBe("c"); + }); +}); diff --git a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts new file mode 100644 index 00000000000..20f20fcf944 --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts @@ -0,0 +1,169 @@ +import { createHash } from "node:crypto"; +import type { ShardKey } from "@trigger.dev/core/v3/isomorphic"; +import { env } from "~/env.server"; +import { logger } from "~/services/logger.server"; +import { FEATURE_FLAG } from "~/v3/featureFlags"; +import { + buildMintShardResolution, + effectiveMintShardSet, + GEN_1_PIN_VALUE, + isValidPinValue, + mintShardStampWarning, + type MintShardSetResolution, +} from "./mintShardGrace"; + +export type MintShardDeps = { + resolution: MintShardSetResolution; + nowMs: number; + graceMs: number; + orgFeatureFlags: unknown; + onPinRejected?: (info: { environmentId: string; pin: string; activeSet: string[] }) => void; +}; + +function asRecord(value: unknown): Record | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + return value as Record; +} + +// Map keys are environment INTERNAL ids (cuids), not friendly ids. An unparseable blob, or a +// blob whose value for this environment is invalid, yields no per-env pin and lets the +// per-org scalar decide — never a silent un-pin straight to the hash. +function readEnvPin(raw: unknown, environmentId: string): ShardKey | undefined { + if (typeof raw !== "string") return undefined; + + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return undefined; + } + + const pins = asRecord(parsed); + const pin = pins?.[environmentId]; + return isValidPinValue(pin) ? pin : undefined; +} + +// Both pins live in the org override blob the trigger path already holds, so resolving a mint +// shard costs no query. +function readPin(orgFeatureFlags: unknown, environmentId: string): ShardKey | undefined { + const blob = asRecord(orgFeatureFlags); + if (!blob) return undefined; + + const envPin = readEnvPin(blob[FEATURE_FLAG.runOpsMintShardEnvPins], environmentId); + if (envPin !== undefined) return envPin; + + const scalar = blob[FEATURE_FLAG.runOpsMintShard]; + return isValidPinValue(scalar) ? scalar : undefined; +} + +// 64 bits: a 32-bit score collides at this system's environment count, and an undetected tie +// would resolve by iteration order. The NUL separates the fields so no two input pairs can +// concatenate alike. This hash input is FROZEN once gen-2 minting is live: changing it +// re-places every environment, silently. +function shardScore(environmentId: string, key: string): bigint { + return createHash("sha256").update(`${environmentId}\0${key}`).digest().readBigUInt64BE(0); +} + +function hrwSelect(environmentId: string, activeSet: string[]): string { + let bestKey = activeSet[0]; + let bestScore = shardScore(environmentId, bestKey); + + for (let i = 1; i < activeSet.length; i++) { + const key = activeSet[i]; + const score = shardScore(environmentId, key); + if (score > bestScore || (score === bestScore && key > bestKey)) { + bestKey = key; + bestScore = score; + } + } + + return bestKey; +} + +// PURE CORE — no env, no clock, no I/O; tests drive this directly. Deterministic for fixed +// deps, which is what lets run minting and token minting agree on one answer. +// +// The empty-set check runs BEFORE the grace, so an unset active list is an unconditional kill +// switch that a stale stamp cannot reopen. A pin outside the active set falls through to the +// hash rather than throwing: honouring it would leak the drain the active list performs, and +// throwing would fail customer triggers whenever a pinned shard drains. +export function computeMintShard(environment: { id: string }, deps: MintShardDeps): ShardKey { + if (deps.resolution.set.length === 0) { + return "new"; + } + + const activeSet = effectiveMintShardSet(deps.resolution, deps.nowMs, deps.graceMs); + if (activeSet.length === 0) { + return "new"; + } + + const pin = readPin(deps.orgFeatureFlags, environment.id); + if (pin !== undefined) { + if (pin === GEN_1_PIN_VALUE) { + return "new"; + } + if (activeSet.includes(pin)) { + return pin; + } + deps.onPinRejected?.({ environmentId: environment.id, pin, activeSet }); + } + + return hrwSelect(environment.id, activeSet); +} + +// ENV-BOUND wrapper — the only place env is read. The resolution is built once: these are +// deploy-time values, so re-parsing per mint would burn CPU on the hottest path in the system. +const shardResolution: MintShardSetResolution = buildMintShardResolution({ + shards: env.RUN_OPS_MINT_SHARDS, + prev: env.RUN_OPS_MINT_SHARDS_PREV, + flippedAt: env.RUN_OPS_MINT_SHARDS_FLIPPED_AT, +}); + +const stampWarning = mintShardStampWarning({ + shards: env.RUN_OPS_MINT_SHARDS, + prev: env.RUN_OPS_MINT_SHARDS_PREV, + flippedAt: env.RUN_OPS_MINT_SHARDS_FLIPPED_AT, +}); +if (stampWarning) { + logger.warn(`[runOpsMintShard] ${stampWarning}`, { + RUN_OPS_MINT_SHARDS: env.RUN_OPS_MINT_SHARDS, + RUN_OPS_MINT_SHARDS_PREV: env.RUN_OPS_MINT_SHARDS_PREV, + }); +} + +// Once per environment per process: a stale pin sits on the root-trigger path and would +// otherwise log on every trigger for that environment, indefinitely. +const reportedPins = new Set(); + +function reportPinRejected(info: { + environmentId: string; + pin: string; + activeSet: string[]; +}): void { + if (reportedPins.has(info.environmentId)) return; + reportedPins.add(info.environmentId); + logger.error("[runOpsMintShard] pinned shard is not in the active set; using the hash", info); +} + +/** + * 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. + * + * Async despite doing no I/O, so the deploy-free active-set layer can add a read later without + * changing every call site. + * + * @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 { + return computeMintShard(environment, { + resolution: shardResolution, + nowMs: Date.now(), + graceMs: env.RUN_OPS_MINT_FLIP_GRACE_MS, + orgFeatureFlags: environment.orgFeatureFlags, + onPinRejected: reportPinRejected, + }); +}