Compare commits

...

2 Commits

Author SHA1 Message Date
Dan Sutton 8d92ce9ae6 test(webapp): assert malformed mint-flip stamp is inert when resolved 2026-07-09 17:19:48 +01:00
Dan Sutton fbfc17e9d4 fix(webapp): schedule run-ops mint-kind flips to avoid duplicate runs
Flipping an organization's run-ops mint kind (which database new runs are
minted on) took effect independently per process as each cached value expired.
For a window after a flip, two concurrent triggers using the same idempotency
key could mint on different databases, where the per-database unique constraint
can't dedupe them, producing a duplicate run.

The flip is now a deterministic wall-clock cutover: the admin feature-flag
routes stamp the previously-effective kind and a flip timestamp onto the
organization, and the mint read resolves the old kind until flippedAt + grace
(default 90s, chosen to outlast the caches a flip drains through), after which
every process crosses to the new kind together. During the window all processes
agree on one database, so a concurrent same-key collision lands on a single
database and the existing unique-constraint retry returns the one idempotent
run. A same-target re-save carries the in-flight stamp forward, so it can't
slide the cutover.
2026-07-09 17:08:50 +01:00
8 changed files with 336 additions and 21 deletions
+4
View File
@@ -1753,6 +1753,10 @@ const EnvironmentSchema = z
RUN_OPS_MINT_ENABLED: BoolEnv.default(false),
RUN_OPS_MINT_FLAG_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
RUN_OPS_MINT_FLAG_CACHE_MAX_ENTRIES: z.coerce.number().int().default(10_000),
// Deterministic wall-clock cutover after a runOpsMintKind flip. Must exceed the sum
// of RUN_OPS_MINT_FLAG_CACHE_TTL_MS and the control-plane cache TTL so every process
// (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),
// Session replication (Postgres → ClickHouse sessions_v1). Shares Redis
// with the runs replicator for leader locking but has its own slot and
@@ -1,9 +1,12 @@
import type { ActionFunctionArgs, LoaderFunctionArgs } from "@remix-run/server-runtime";
import { json } from "@remix-run/server-runtime";
import type { Prisma } from "@trigger.dev/database";
import { z } from "zod";
import { env } from "~/env.server";
import { prisma } from "~/db.server";
import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
import { stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace";
import { validatePartialFeatureFlags } from "~/v3/featureFlags";
const ParamsSchema = z.object({
@@ -82,10 +85,15 @@ export async function action({ request, params }: ActionFunctionArgs) {
? validatePartialFeatureFlags(organization.featureFlags as Record<string, unknown>)
: { success: false as const };
const mergedFlags = {
...(existingFlags.success ? existingFlags.data : {}),
...validationResult.data,
};
const mergedFlags = stampMintKindFlip(
existingFlags.success ? existingFlags.data : {},
{
...(existingFlags.success ? existingFlags.data : {}),
...validationResult.data,
},
Date.now(),
env.RUN_OPS_MINT_FLIP_GRACE_MS
);
// Update the organization's feature flags
const updatedOrganization = await prisma.organization.update({
@@ -93,7 +101,7 @@ export async function action({ request, params }: ActionFunctionArgs) {
id: organizationId,
},
data: {
featureFlags: mergedFlags,
featureFlags: mergedFlags as Prisma.InputJsonValue,
},
select: {
id: true,
@@ -2,9 +2,11 @@ import type { ActionFunctionArgs, LoaderFunctionArgs } from "@remix-run/server-r
import { json } from "@remix-run/server-runtime";
import { Prisma } from "@trigger.dev/database";
import { z } from "zod";
import { env } from "~/env.server";
import { prisma } from "~/db.server";
import { requireUser } from "~/services/session.server";
import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
import { stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace";
import { flags as getGlobalFlags } from "~/v3/featureFlags.server";
import {
FEATURE_FLAG,
@@ -119,6 +121,17 @@ export async function action({ request, params }: ActionFunctionArgs) {
);
}
featureFlags = validationResult.data;
const existingOrg = await prisma.organization.findFirst({
where: { id: organizationId },
select: { featureFlags: true },
});
featureFlags = stampMintKindFlip(
existingOrg?.featureFlags as Record<string, unknown> | null | undefined,
featureFlags,
Date.now(),
env.RUN_OPS_MINT_FLIP_GRACE_MS
);
}
try {
+9
View File
@@ -19,6 +19,9 @@ export const FEATURE_FLAG = {
computeMigrationRequireTemplate: "computeMigrationRequireTemplate",
devBranchesEnabled: "devBranchesEnabled",
runOpsMintKind: "runOpsMintKind",
// Grace-linger stamp carried alongside runOpsMintKind on flip. See mintFlipGrace.ts.
runOpsMintKindPrev: "runOpsMintKindPrev",
runOpsMintKindFlippedAt: "runOpsMintKindFlippedAt",
} as const;
export const FeatureFlagCatalog = {
@@ -54,6 +57,10 @@ export const FeatureFlagCatalog = {
// Per-org run-ops-id mint cutover. Defaults to "cuid"; only honored when
// RUN_OPS_MINT_ENABLED is on AND isSplitEnabled() is true.
[FEATURE_FLAG.runOpsMintKind]: z.enum(["cuid", "runOpsId"]),
// Grace-linger stamp: the previously-effective kind and the flip timestamp, written
// 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(),
};
export type FeatureFlagKey = keyof typeof FeatureFlagCatalog;
@@ -70,6 +77,8 @@ export const GLOBAL_LOCKED_FLAGS: FeatureFlagKey[] = [
export const ORG_LOCKED_FLAGS: FeatureFlagKey[] = [
FEATURE_FLAG.defaultWorkerInstanceGroupId,
FEATURE_FLAG.taskEventRepository,
FEATURE_FLAG.runOpsMintKindPrev,
FEATURE_FLAG.runOpsMintKindFlippedAt,
];
// Create a Zod schema from the existing catalog
@@ -0,0 +1,159 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { effectiveMintKind, stampMintKindFlip, type MintFlagResolution } from "./mintFlipGrace";
// GRACE-LINGER: during [flippedAt, flippedAt + GRACE) every process — stale or fresh —
// must resolve to the SAME (old) kind; at/after the cutover every process resolves to
// the SAME (new) kind. This collapses the cross-process divergence window.
const GRACE_MS = 90_000;
const T = 1_000_000;
describe("effectiveMintKind", () => {
beforeEach(() => vi.useFakeTimers());
afterEach(() => vi.useRealTimers());
it("returns r.kind directly when prev is missing", () => {
const r: MintFlagResolution = { kind: "cuid" };
expect(effectiveMintKind(r, T, GRACE_MS)).toBe("cuid");
});
it("returns r.kind directly when flippedAtMs is missing", () => {
const r: MintFlagResolution = { kind: "runOpsId", prev: "cuid" };
expect(effectiveMintKind(r, T, GRACE_MS)).toBe("runOpsId");
});
it("CORE: stale and fresh resolutions agree at every instant during grace, then both flip to the new kind at cutover", () => {
const stale: MintFlagResolution = { kind: "cuid" };
const fresh: MintFlagResolution = { kind: "runOpsId", prev: "cuid", flippedAtMs: T };
for (const now of [T, T + 1, T + 1_000, T + 45_000, T + GRACE_MS - 1]) {
const staleResolved = effectiveMintKind(stale, now, GRACE_MS);
const freshResolved = effectiveMintKind(fresh, now, GRACE_MS);
expect(staleResolved).toBe("cuid");
expect(freshResolved).toBe("cuid");
}
for (const now of [T + GRACE_MS, T + GRACE_MS + 1, T + GRACE_MS + 60_000]) {
expect(effectiveMintKind(fresh, now, GRACE_MS)).toBe("runOpsId");
}
});
it("boundary: exactly at flippedAt + GRACE resolves to the NEW kind", () => {
const fresh: MintFlagResolution = { kind: "runOpsId", prev: "cuid", flippedAtMs: T };
expect(effectiveMintKind(fresh, T + GRACE_MS - 1, GRACE_MS)).toBe("cuid");
expect(effectiveMintKind(fresh, T + GRACE_MS, GRACE_MS)).toBe("runOpsId");
});
});
describe("stampMintKindFlip", () => {
beforeEach(() => vi.useFakeTimers());
afterEach(() => vi.useRealTimers());
it("a genuine flip (cuid -> runOpsId) stamps prev and flippedAt", () => {
const existing = { runOpsMintKind: "cuid" };
const outgoing = { runOpsMintKind: "runOpsId" };
const result = stampMintKindFlip(existing, outgoing, T, GRACE_MS);
expect(result.runOpsMintKind).toBe("runOpsId");
expect(result.runOpsMintKindPrev).toBe("cuid");
expect(result.runOpsMintKindFlippedAt).toBe(new Date(T).toISOString());
});
it("defaults the existing effective kind to cuid when existing flags are null", () => {
const outgoing = { runOpsMintKind: "runOpsId" };
const result = stampMintKindFlip(null, outgoing, T, GRACE_MS);
expect(result.runOpsMintKind).toBe("runOpsId");
expect(result.runOpsMintKindPrev).toBe("cuid");
expect(result.runOpsMintKindFlippedAt).toBe(new Date(T).toISOString());
});
it("resubmitting the same target kind mid-grace carries the stamp forward untouched (does not reset the cutover clock)", () => {
const existing = {
runOpsMintKind: "runOpsId",
runOpsMintKindPrev: "cuid",
runOpsMintKindFlippedAt: new Date(T).toISOString(),
};
const now = T + 10_000;
const outgoing = { runOpsMintKind: "runOpsId", someOtherFlag: true };
const result = stampMintKindFlip(existing, outgoing, now, GRACE_MS);
expect(result.runOpsMintKind).toBe("runOpsId");
expect(result.runOpsMintKindPrev).toBe("cuid");
// Unrelated re-save: the cutover time must NOT slide forward.
expect(result.runOpsMintKindFlippedAt).toBe(new Date(T).toISOString());
expect(result.someOtherFlag).toBe(true);
});
it("an unchanged save after grace has elapsed carries the settled stamp forward and preserves unrelated flags", () => {
const existing = {
runOpsMintKind: "runOpsId",
runOpsMintKindPrev: "cuid",
runOpsMintKindFlippedAt: new Date(T).toISOString(),
};
const outgoing = { runOpsMintKind: "runOpsId", someOtherFlag: true };
const result = stampMintKindFlip(existing, outgoing, T + GRACE_MS + 5_000, GRACE_MS);
expect(result.runOpsMintKind).toBe("runOpsId");
expect(result.someOtherFlag).toBe(true);
expect(result.runOpsMintKindPrev).toBe("cuid");
expect(result.runOpsMintKindFlippedAt).toBe(new Date(T).toISOString());
});
it("a flip-back requested after the original grace has elapsed stamps prev := the new settled (now-effective) kind, timestamped now", () => {
const existing = {
runOpsMintKind: "runOpsId",
runOpsMintKindPrev: "cuid",
runOpsMintKindFlippedAt: new Date(T).toISOString(),
};
const now = T + GRACE_MS + 1_000;
const outgoing = { runOpsMintKind: "cuid" };
const result = stampMintKindFlip(existing, outgoing, now, GRACE_MS);
expect(result.runOpsMintKind).toBe("cuid");
expect(result.runOpsMintKindPrev).toBe("runOpsId");
expect(result.runOpsMintKindFlippedAt).toBe(new Date(now).toISOString());
});
it("a flip-back mid-grace re-stamps prev to the still-effective old kind, so it keeps serving that kind (no divergence)", () => {
const existing = {
runOpsMintKind: "runOpsId",
runOpsMintKindPrev: "cuid",
runOpsMintKindFlippedAt: new Date(T).toISOString(),
};
const now = T + 20_000;
const outgoing = { runOpsMintKind: "cuid" };
const result = stampMintKindFlip(existing, outgoing, now, GRACE_MS);
expect(result.runOpsMintKind).toBe("cuid");
expect(result.runOpsMintKindPrev).toBe("cuid");
expect(result.runOpsMintKindFlippedAt).toBe(new Date(now).toISOString());
});
it("defaults outgoing kind to 'cuid' when runOpsMintKind is absent", () => {
const existing = { runOpsMintKind: "runOpsId" };
const outgoing: Record<string, unknown> = {};
const result = stampMintKindFlip(existing, outgoing, T, GRACE_MS);
expect(result.runOpsMintKind).toBe("cuid");
expect(result.runOpsMintKindPrev).toBe("runOpsId");
expect(result.runOpsMintKindFlippedAt).toBe(new Date(T).toISOString());
});
it("treats a malformed existing flippedAt as no stamp", () => {
const existing = {
runOpsMintKind: "runOpsId",
runOpsMintKindPrev: "cuid",
runOpsMintKindFlippedAt: "not-a-date",
};
const outgoing = { runOpsMintKind: "runOpsId" };
const result = stampMintKindFlip(existing, outgoing, T, GRACE_MS);
expect(result.runOpsMintKind).toBe("runOpsId");
// The malformed flippedAt is carried forward verbatim, but an unparseable timestamp is
// inert when resolved (Date.parse -> NaN -> effectiveMintKind returns the target kind).
expect(result.runOpsMintKindFlippedAt).toBe("not-a-date");
expect(
effectiveMintKind({ kind: "runOpsId", prev: "cuid", flippedAtMs: NaN }, T, GRACE_MS)
).toBe("runOpsId");
});
});
@@ -0,0 +1,77 @@
import type { RunIdMintKind } from "./runOpsMintKind.server";
export type { RunIdMintKind };
export type MintFlagResolution = {
kind: RunIdMintKind;
prev?: RunIdMintKind;
flippedAtMs?: number;
};
const DEFAULT_MINT_KIND: RunIdMintKind = "cuid";
// Deterministic wall-clock cutover: during [flippedAt, flippedAt + graceMs) every
// process — stale (hasn't re-read the flag) or fresh (has the new stamp) — resolves to
// the OLD kind. At/after the cutover, every process resolves to the NEW kind.
export function effectiveMintKind(
r: MintFlagResolution,
nowMs: number,
graceMs: number
): RunIdMintKind {
if (r.prev === undefined || r.flippedAtMs === undefined) {
return r.kind;
}
return nowMs < r.flippedAtMs + graceMs ? r.prev : r.kind;
}
function readMintKind(flags: Record<string, unknown>, key: string): RunIdMintKind | undefined {
const value = flags[key];
return value === "cuid" || value === "runOpsId" ? value : undefined;
}
function resolveEffectiveFromFlags(
flags: Record<string, unknown> | null | undefined,
nowMs: number,
graceMs: number
): RunIdMintKind {
const source = flags ?? {};
const kind = readMintKind(source, "runOpsMintKind") ?? DEFAULT_MINT_KIND;
const prev = readMintKind(source, "runOpsMintKindPrev");
const flippedAtRaw = source.runOpsMintKindFlippedAt;
const parsed = typeof flippedAtRaw === "string" ? Date.parse(flippedAtRaw) : NaN;
const flippedAtMs = Number.isNaN(parsed) ? undefined : parsed;
return effectiveMintKind({ kind, prev, flippedAtMs }, nowMs, graceMs);
}
// Stamps a grace window only when the outgoing TARGET kind differs from the stored one (a
// genuine flip); prev := the currently-effective kind. A save that leaves the target kind
// unchanged carries any in-flight stamp forward, so it can't reset the cutover clock.
export function stampMintKindFlip(
existingFlags: Record<string, unknown> | null | undefined,
outgoingFlags: Record<string, unknown>,
nowMs: number,
graceMs: number
): Record<string, unknown> {
const storedKind = readMintKind(existingFlags ?? {}, "runOpsMintKind") ?? DEFAULT_MINT_KIND;
const outgoingKind = readMintKind(outgoingFlags, "runOpsMintKind") ?? DEFAULT_MINT_KIND;
outgoingFlags.runOpsMintKind = outgoingKind;
if (outgoingKind !== storedKind) {
// Genuine target change: serve the currently-effective kind through the new grace window.
outgoingFlags.runOpsMintKindPrev = resolveEffectiveFromFlags(existingFlags, nowMs, graceMs);
outgoingFlags.runOpsMintKindFlippedAt = new Date(nowMs).toISOString();
return outgoingFlags;
}
const existing = existingFlags ?? {};
const existingPrev = existing.runOpsMintKindPrev;
const existingFlippedAt = existing.runOpsMintKindFlippedAt;
if (existingPrev !== undefined) {
outgoingFlags.runOpsMintKindPrev = existingPrev;
}
if (existingFlippedAt !== undefined) {
outgoingFlags.runOpsMintKindFlippedAt = existingFlippedAt;
}
return outgoingFlags;
}
@@ -2,15 +2,15 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { BoundedTtlCache } from "~/services/realtime/boundedTtlCache";
import { computeRunIdMintKind, type RunIdMintKind } from "./runOpsMintKind.server";
// LOCK of the CURRENT (intentional) flip-latency behavior, NOT a change request.
// resolveRunIdMintKind caches the per-org mint kind in a process-singleton
// BoundedTtlCache (TTL RUN_OPS_MINT_FLAG_CACHE_TTL_MS, 30000ms default) with get/set
// and NO invalidation hook (runOpsMintKind.server.ts:38-45,56-81). So after a flag
// flip a process keeps minting the stale kind until its cached entry expires; in
// multi-instance prod each process expires independently. This suite reconstructs the
// same flag fn over a real cache and pins both edges of that window.
// LOCK of the raw per-process cache's flip-latency behavior in isolation, NOT a change
// request. Production resolveRunIdMintKind now wraps this same staleness in a deterministic
// wall-clock grace window (mintFlipGrace.ts) so every process resolves to the SAME effective
// kind for the whole window, then all cross together. This raw staleness is now an
// intentional, safe input to that resolution. computeRunIdMintKind is unaffected, so this
// suite's assertions stand as-is.
// Mirror of resolveRunIdMintKind's flag fn (runOpsMintKind.server.ts:56-81).
// Bare cached-flag closure — deliberately NOT the production flag fn, which now layers
// grace resolution on top (see runOpsMintKind.server.ts).
function makeCachedFlag(
cache: BoundedTtlCache<RunIdMintKind>,
liveFlag: () => RunIdMintKind
@@ -3,8 +3,10 @@ import { env } from "~/env.server";
import { logger } from "~/services/logger.server";
import { BoundedTtlCache } from "~/services/realtime/boundedTtlCache";
import { singleton } from "~/utils/singleton";
import { FEATURE_FLAG } from "~/v3/featureFlags";
import { FEATURE_FLAG, FeatureFlagCatalog } from "~/v3/featureFlags";
import { makeFlag } from "~/v3/featureFlags.server";
import { DEFAULT_CP_CACHE_TTL_MS } from "./controlPlaneCache.server";
import { effectiveMintKind, type MintFlagResolution } from "./mintFlipGrace";
import { isSplitEnabled } from "./splitMode.server";
export type RunIdMintKind = "cuid" | "runOpsId";
@@ -38,12 +40,38 @@ const flagFn = singleton("runOpsMintFlag", () => makeFlag($replica));
const mintCache = singleton(
"runOpsMintCache",
() =>
new BoundedTtlCache<RunIdMintKind>(
new BoundedTtlCache<MintFlagResolution>(
env.RUN_OPS_MINT_FLAG_CACHE_TTL_MS,
env.RUN_OPS_MINT_FLAG_CACHE_MAX_ENTRIES
)
);
// BOOT-TIME SAFETY CHECK (warning only, never throws): the grace window only collapses
// the cross-process divergence window if it outlasts BOTH caches a flag flip has to drain
// through — this process's own mint-flag cache AND the org-flags control-plane cache a
// stale process might still be reading through. If it doesn't, warn loudly but keep booting.
const controlPlaneCacheTtlMs = env.CONTROL_PLANE_CACHE_TTL_MS ?? DEFAULT_CP_CACHE_TTL_MS;
if (env.RUN_OPS_MINT_FLIP_GRACE_MS <= env.RUN_OPS_MINT_FLAG_CACHE_TTL_MS + controlPlaneCacheTtlMs) {
logger.warn(
"[runOpsMintKind] RUN_OPS_MINT_FLIP_GRACE_MS does not exceed the sum of " +
"RUN_OPS_MINT_FLAG_CACHE_TTL_MS and the control-plane cache TTL; a flag flip can still " +
"cross-DB-duplicate a concurrent root trigger during the divergence window",
{
RUN_OPS_MINT_FLIP_GRACE_MS: env.RUN_OPS_MINT_FLIP_GRACE_MS,
RUN_OPS_MINT_FLAG_CACHE_TTL_MS: env.RUN_OPS_MINT_FLAG_CACHE_TTL_MS,
controlPlaneCacheTtlMs,
}
);
}
function readValidatedFlag<T extends "runOpsMintKindPrev" | "runOpsMintKindFlippedAt">(
overrides: Record<string, unknown>,
key: T
): string | undefined {
const parsed = FeatureFlagCatalog[FEATURE_FLAG[key]].safeParse(overrides[FEATURE_FLAG[key]]);
return parsed.success ? parsed.data : undefined;
}
export async function resolveRunIdMintKind(environment: {
organizationId: string;
id: string;
@@ -54,10 +82,14 @@ export async function resolveRunIdMintKind(environment: {
masterEnabled: env.RUN_OPS_MINT_ENABLED,
splitEnabled: isSplitEnabled,
flag: async (orgId, orgFeatureFlags) => {
// The cache stores only "cuid"|"runOpsId" (never undefined), so the cache's
// "stored-undefined == miss" caveat never applies here.
// The cache stores the full { kind, prev, flippedAtMs } trio (never undefined), so the
// cache's "stored-undefined == miss" caveat never applies here. A cache HIT still passes
// back through effectiveMintKind so a cached-but-stale entry crosses the grace boundary
// on schedule, without needing an invalidation hook.
const cached = mintCache.get(orgId);
if (cached !== undefined) return cached;
if (cached !== undefined) {
return effectiveMintKind(cached, Date.now(), env.RUN_OPS_MINT_FLIP_GRACE_MS);
}
// Hot-path pass-through: use the org flags the authenticated environment already
// carries; only fall back to a DB read when the caller did NOT pass them (non-trigger
@@ -72,13 +104,26 @@ export async function resolveRunIdMintKind(environment: {
})
)?.featureFlags;
const overridesRecord = (overrides as Record<string, unknown>) ?? {};
const kind = await flagFn({
key: FEATURE_FLAG.runOpsMintKind,
defaultValue: "cuid",
overrides: (overrides as Record<string, unknown>) ?? {},
overrides: overridesRecord,
});
mintCache.set(orgId, kind);
return kind;
// Read the grace-linger stamp DIRECTLY from the already-in-memory org overrides — no
// extra DB read, and no forcing these optional fields through flagFn's defaultValue path.
const prev = readValidatedFlag(overridesRecord, "runOpsMintKindPrev") as
| RunIdMintKind
| undefined;
const flippedAtRaw = readValidatedFlag(overridesRecord, "runOpsMintKindFlippedAt");
const parsedFlippedAt = flippedAtRaw !== undefined ? Date.parse(flippedAtRaw) : NaN;
const flippedAtMs = Number.isNaN(parsedFlippedAt) ? undefined : parsedFlippedAt;
const resolution: MintFlagResolution = { kind, prev, flippedAtMs };
mintCache.set(orgId, resolution);
return effectiveMintKind(resolution, Date.now(), env.RUN_OPS_MINT_FLIP_GRACE_MS);
},
});
}