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.
This commit is contained in:
Daniel Sutton
@@ -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.
|
||||
|
||||
@@ -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: {"<environmentId>": "<shard key>"}. 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.
|
||||
|
||||
@@ -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 });
|
||||
});
|
||||
});
|
||||
@@ -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-<id>`, 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<string>();
|
||||
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;
|
||||
}
|
||||
@@ -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> = {}
|
||||
): MintShardDeps {
|
||||
return {
|
||||
resolution,
|
||||
nowMs: T + GRACE_MS + 1,
|
||||
graceMs: GRACE_MS,
|
||||
orgFeatureFlags: undefined,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function orgFlags(flags: Record<string, unknown>) {
|
||||
return { orgFeatureFlags: flags };
|
||||
}
|
||||
|
||||
function place(ids: string[], resolution: MintShardSetResolution): Map<string, string> {
|
||||
const out = new Map<string, string>();
|
||||
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<string, number>();
|
||||
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");
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown> | undefined {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) return undefined;
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
|
||||
// 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<string>();
|
||||
|
||||
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<ShardKey> {
|
||||
return computeMintShard(environment, {
|
||||
resolution: shardResolution,
|
||||
nowMs: Date.now(),
|
||||
graceMs: env.RUN_OPS_MINT_FLIP_GRACE_MS,
|
||||
orgFeatureFlags: environment.orgFeatureFlags,
|
||||
onPinRejected: reportPinRejected,
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user