diff --git a/apps/webapp/app/entry.server.tsx b/apps/webapp/app/entry.server.tsx index 074bc39d760..501f478fdf6 100644 --- a/apps/webapp/app/entry.server.tsx +++ b/apps/webapp/app/entry.server.tsx @@ -17,6 +17,8 @@ import { LocaleContextProvider } from "./components/primitives/LocaleProvider"; import type { OperatingSystemPlatform } from "./components/primitives/OperatingSystemProvider"; import { OperatingSystemContextProvider } from "./components/primitives/OperatingSystemProvider"; import { assertRunOpsSplitSentinel, Prisma } from "./db.server"; +import { assertSnapshotStoreBootFromEnv } from "./v3/snapshotStoreBoot.server"; +import { registerSnapshotStoreWiring } from "./v3/snapshotStoreWiring.server"; import { env } from "./env.server"; import { eventLoopMonitor, eventLoopUtilizationMonitor } from "./eventLoopMonitor.server"; import { logger } from "./services/logger.server"; @@ -325,6 +327,18 @@ singleton("AssertRunOpsSplitSentinel", () => { return true; }); +singleton("SnapshotStoreWiring", registerSnapshotStoreWiring); + +// Ordered after the wiring above: the boot check asserts the repair binding is set, and the +// binding is what the wiring installs. +singleton("AssertSnapshotStoreBoot", () => { + assertSnapshotStoreBootFromEnv().catch((error) => { + logger.error("Snapshot store boot check failed; refusing to start", { error }); + process.exit(1); + }); + return true; +}); + singleton("RunEngineEventBusHandlers", registerRunEngineEventBusHandlers); singleton("SetupBatchQueueCallbacks", setupBatchQueueCallbacks); // Attach the realtime run-changed publish delegations to the engine event bus. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 2b1fba86980..c425d6bec0e 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -1307,6 +1307,40 @@ const EnvironmentSchema = z .string() .default(process.env.REDIS_TLS_DISABLED ?? "false"), + // Execution-snapshot store. MODE here is only the FLOOR: the operational dial is the + // snapshotStoreMode feature flag, so it can move without a deploy. + RUN_ENGINE_SNAPSHOT_STORE_MODE: z + .enum(["off", "dual-write", "redis-read", "redis-only"]) + .default("off"), + RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS: z.coerce + .number() + .int() + .default(72 * 60 * 60 * 1000), + RUN_ENGINE_SNAPSHOT_STORE_ORPHAN_AGE_MS: z.coerce + .number() + .int() + .default(24 * 60 * 60 * 1000), + RUN_ENGINE_SNAPSHOT_STORE_CONFIRM_ORPHAN_AFTER_MS: z.coerce + .number() + .int() + .default(2 * 60 * 60 * 1000), + RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_SCHEDULE: z.string().default("0 */6 * * *"), + RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_JITTER_IN_MS: z.coerce.number().int().default(60_000), + // An existing run costs ~4 serial round trips and the orphan-marker clear cannot be batched + // (cross-slot pipelines are rejected), so a full pass is hours, not minutes. A budget that + // truncates every pass stops rule 2 converging, because it needs consecutive sightings. + RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_BUDGET_MS: z.coerce.number().int().default(10_800_000), + RUN_ENGINE_SNAPSHOT_STORE_ORG_MODE_CACHE_TTL_MS: z.coerce.number().int().default(30_000), + RUN_ENGINE_SNAPSHOT_STORE_ORG_MODE_CACHE_MAX: z.coerce.number().int().default(10_000), + // No fallback to REDIS_*: this is a distinct durable endpoint and must be set explicitly, or + // execution state silently lands on the general-purpose cache. + RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST: z.string().optional(), + RUN_ENGINE_SNAPSHOT_STORE_REDIS_PORT: z.coerce.number().optional(), + RUN_ENGINE_SNAPSHOT_STORE_REDIS_USERNAME: z.string().optional(), + RUN_ENGINE_SNAPSHOT_STORE_REDIS_PASSWORD: z.string().optional(), + RUN_ENGINE_SNAPSHOT_STORE_REDIS_TLS_DISABLED: z.string().default("false"), + RUN_ENGINE_SNAPSHOT_STORE_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"), + RUN_ENGINE_DEV_PRESENCE_REDIS_HOST: z .string() .optional() diff --git a/apps/webapp/app/routes/admin.api.v1.feature-flags.ts b/apps/webapp/app/routes/admin.api.v1.feature-flags.ts index e9da02effd9..6caac2d78d2 100644 --- a/apps/webapp/app/routes/admin.api.v1.feature-flags.ts +++ b/apps/webapp/app/routes/admin.api.v1.feature-flags.ts @@ -10,6 +10,10 @@ import { withoutDerivedKeys, } from "~/v3/featureFlags.server"; import { validatePartialFeatureFlags } from "~/v3/featureFlags"; +import { + globalOnlySnapshotStoreFlagError, + snapshotStoreFlagSaveError, +} from "~/v3/snapshotStoreFlagGuard.server"; export async function action({ request }: ActionFunctionArgs) { await requireAdminApiRequest(request); @@ -30,6 +34,18 @@ export async function action({ request }: ActionFunctionArgs) { ); } + const globalOnlyError = globalOnlySnapshotStoreFlagError(body as Record); + if (globalOnlyError) { + return json({ error: globalOnlyError }, { status: 400 }); + } + + const snapshotStoreError = snapshotStoreFlagSaveError(body as Record, { + redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST, + }); + if (snapshotStoreError) { + return json({ error: snapshotStoreError }, { status: 400 }); + } + // Both the strip and the branch derive from the graced-group table, so adding a group needs // no edit here. Naming the keys inline is how a new group ends up writing its stamp straight // from the request body, with no lock. diff --git a/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.feature-flags.ts b/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.feature-flags.ts index db1524f8551..c66f608d5d8 100644 --- a/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.feature-flags.ts +++ b/apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.feature-flags.ts @@ -6,8 +6,10 @@ import { env } from "~/env.server"; import { prisma } from "~/db.server"; import { requireAdminApiRequest } from "~/services/personalAccessToken.server"; import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server"; +import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server"; +import { invalidateSnapshotStoreOrgMode } from "~/v3/snapshotStoreMode.server"; import { selectMintBaselineSource, stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace"; -import { validatePartialFeatureFlags } from "~/v3/featureFlags"; +import { validatePartialFeatureFlags, withoutOrgForbiddenSnapshotKeys } from "~/v3/featureFlags"; import { flags as getGlobalFlags } from "~/v3/featureFlags.server"; const ParamsSchema = z.object({ @@ -71,9 +73,18 @@ export async function action({ request, params }: ActionFunctionArgs) { const { runOpsMintKindPrev: _ignoredPrev, runOpsMintKindFlippedAt: _ignoredFlippedAt, - ...requestedFlags + ...rawRequestedFlags } = validationResult.data; + const requestedFlags = withoutOrgForbiddenSnapshotKeys(rawRequestedFlags); + + const snapshotStoreError = snapshotStoreFlagSaveError(requestedFlags, { + redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST, + }); + if (snapshotStoreError) { + return json({ error: snapshotStoreError }, { status: 400 }); + } + // Seed the flip baseline from the current GLOBAL mint flags so an org's FIRST per-org override // is graced from the currently-effective global kind, not the hardcoded default "cuid". const globalFlags = (await getGlobalFlags()) as Record; @@ -129,6 +140,7 @@ export async function action({ request, params }: ActionFunctionArgs) { // Org feature flags are embedded in every env of the org; drop all its cached env rows. controlPlaneResolver.invalidateOrganization(organizationId); + invalidateSnapshotStoreOrgMode(organizationId); const updatedFlagsResult = updatedOrganization.featureFlags ? validatePartialFeatureFlags(updatedOrganization.featureFlags as Record) diff --git a/apps/webapp/app/routes/admin.api.v2.orgs.$organizationId.feature-flags.ts b/apps/webapp/app/routes/admin.api.v2.orgs.$organizationId.feature-flags.ts index 0071054bd3e..773a2b6376c 100644 --- a/apps/webapp/app/routes/admin.api.v2.orgs.$organizationId.feature-flags.ts +++ b/apps/webapp/app/routes/admin.api.v2.orgs.$organizationId.feature-flags.ts @@ -6,11 +6,14 @@ import { env } from "~/env.server"; import { prisma } from "~/db.server"; import { requireUser } from "~/services/session.server"; import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server"; +import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server"; +import { invalidateSnapshotStoreOrgMode } from "~/v3/snapshotStoreMode.server"; import { selectMintBaselineSource, stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace"; import { flags as getGlobalFlags } from "~/v3/featureFlags.server"; import { FEATURE_FLAG, validatePartialFeatureFlags, + withoutOrgForbiddenSnapshotKeys, getAllFlagControlTypes, } from "~/v3/featureFlags"; import { featuresForRequest } from "~/features.server"; @@ -123,6 +126,7 @@ export async function action({ request, params }: ActionFunctionArgs) { } controlPlaneResolver.invalidateOrganization(organizationId); + invalidateSnapshotStoreOrgMode(organizationId); return json({ success: true }); } @@ -138,9 +142,18 @@ export async function action({ request, params }: ActionFunctionArgs) { const { runOpsMintKindPrev: _ignoredPrev, runOpsMintKindFlippedAt: _ignoredFlippedAt, - ...requestedFlags + ...rawRequestedFlags } = validationResult.data; + const requestedFlags = withoutOrgForbiddenSnapshotKeys(rawRequestedFlags); + + const snapshotStoreError = snapshotStoreFlagSaveError(requestedFlags, { + redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST, + }); + if (snapshotStoreError) { + return json({ error: snapshotStoreError }, { status: 400 }); + } + // Seed the flip baseline from the current GLOBAL mint flags so an org's FIRST per-org override // is graced from the currently-effective global kind, not the hardcoded default "cuid". const globalFlags = (await getGlobalFlags()) as Record; @@ -181,6 +194,7 @@ export async function action({ request, params }: ActionFunctionArgs) { // Org feature flags are embedded in every env of the org; drop all its cached env rows. controlPlaneResolver.invalidateOrganization(organizationId); + invalidateSnapshotStoreOrgMode(organizationId); return json({ success: true }); } diff --git a/apps/webapp/app/routes/admin.feature-flags.tsx b/apps/webapp/app/routes/admin.feature-flags.tsx index e987812f520..bd3fec0c3de 100644 --- a/apps/webapp/app/routes/admin.feature-flags.tsx +++ b/apps/webapp/app/routes/admin.feature-flags.tsx @@ -17,6 +17,10 @@ import { lockedFlagsInPayload, validatePartialFeatureFlags, } from "~/v3/featureFlags"; +import { + globalOnlySnapshotStoreFlagError, + snapshotStoreFlagSaveError, +} from "~/v3/snapshotStoreFlagGuard.server"; import { flags as getGlobalFlags, replaceGlobalFeatureFlags } from "~/v3/featureFlags.server"; import { featuresForRequest } from "~/features.server"; import { Button } from "~/components/primitives/Buttons"; @@ -129,6 +133,18 @@ export const action = dashboardAction( ); } + const globalOnlyError = globalOnlySnapshotStoreFlagError(parsed.data.flags); + if (globalOnlyError) { + return json({ error: globalOnlyError }, { status: 400 }); + } + + const snapshotStoreError = snapshotStoreFlagSaveError(parsed.data.flags, { + redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST, + }); + if (snapshotStoreError) { + return json({ error: snapshotStoreError }, { status: 400 }); + } + await replaceGlobalFeatureFlags(prisma, { requestedFlags: validationResult.data as Record, catalogKeys: Object.keys(getAllFlagControlTypes()) as FeatureFlagKey[], diff --git a/apps/webapp/app/v3/featureFlags.ts b/apps/webapp/app/v3/featureFlags.ts index 3a88beb54bc..8caecf428c5 100644 --- a/apps/webapp/app/v3/featureFlags.ts +++ b/apps/webapp/app/v3/featureFlags.ts @@ -44,6 +44,14 @@ export const FEATURE_FLAG = { // System-wide kill switch for additional (scoped) environment API-key lookup. // Defaults off; enable during rollout once the new lookup path is trusted. additionalApiKeyLookupEnabled: "additionalApiKeyLookupEnabled", + // The execution-snapshot store rollout dial. A flag rather than an environment variable because + // a sustained append failure burns a task attempt per transition, so dial-down is a correctness + // control and cannot wait for a deploy. + snapshotStoreMode: "snapshotStoreMode", + // Per-org override, read from the org blob only. Deliberately narrower than the global key: + // snapshot reads are global, so an org at a read position would read state its own writes never + // created. Stripped from org payloads by withoutOrgForbiddenSnapshotKeys. + snapshotStoreOrgMode: "snapshotStoreOrgMode", } as const; export const FeatureFlagCatalog = { @@ -153,6 +161,8 @@ export const FeatureFlagCatalog = { [FEATURE_FLAG.additionalApiKeysEnabled]: z.boolean(), [FEATURE_FLAG.additionalApiKeyIssuanceEnabled]: z.boolean(), [FEATURE_FLAG.additionalApiKeyLookupEnabled]: z.boolean(), + [FEATURE_FLAG.snapshotStoreMode]: z.enum(["off", "dual-write", "redis-read", "redis-only"]), + [FEATURE_FLAG.snapshotStoreOrgMode]: z.enum(["off", "dual-write"]), }; export type FeatureFlagKey = keyof typeof FeatureFlagCatalog; @@ -188,8 +198,20 @@ export const ORG_LOCKED_FLAGS: FeatureFlagKey[] = [ FEATURE_FLAG.runOpsMintShardSetPrev, FEATURE_FLAG.runOpsMintShardSetFlippedAt, FEATURE_FLAG.runOpsMintShardOverride, + // The dial is deployment-wide; only snapshotStoreOrgMode is per-org. + FEATURE_FLAG.snapshotStoreMode, ]; +/** + * Drops keys an organisation must never supply. ORG_LOCKED_FLAGS is a UI predicate and no save path + * consults it, so the line is held here — the same way the mint grace stamps are stripped. + */ +export function withoutOrgForbiddenSnapshotKeys>(values: T): T { + if (!(FEATURE_FLAG.snapshotStoreMode in values)) return values; + const { [FEATURE_FLAG.snapshotStoreMode]: _dropped, ...rest } = values; + return rest as T; +} + /** * Flag groups where the operator sets a `primary` and the server computes the rest. The topology * lives here, not in the server module, because the admin page needs it too: unsetting a primary diff --git a/apps/webapp/app/v3/runEngine.server.ts b/apps/webapp/app/v3/runEngine.server.ts index 76ac35f349b..30a5c549433 100644 --- a/apps/webapp/app/v3/runEngine.server.ts +++ b/apps/webapp/app/v3/runEngine.server.ts @@ -12,6 +12,8 @@ import { runEnginePendingVersionLookup } from "./runEnginePendingVersionLookup.s import { pickRunOpsStoreForCompletion } from "./runOpsMigration/crossSeamGuard.server"; import { runEngineControlPlaneResolver } from "./runOpsMigration/runEngineControlPlaneResolver.server"; import { runStore } from "./runStore.server"; +import { getSnapshotSweepRunner } from "./snapshotStoreBindings.server"; +import { getSnapshotStoreConfig } from "./snapshotStoreInstance.server"; import { meter, tracer } from "./tracer.server"; export const engine = singleton("RunEngine", createRunEngine); @@ -241,6 +243,22 @@ function createRunEngine() { randomize: true, }, }, + // Omitted entirely when the snapshot store is unconfigured: passing a runner would register the + // cron job and log an unbound pass every interval on every install that does not use the store. + snapshotStore: getSnapshotStoreConfig().configured + ? { + runSweep: async (opts) => { + const run = getSnapshotSweepRunner(); + if (!run) { + return { outcome: "unbound" }; + } + return run(opts); + }, + sweepSchedule: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_SCHEDULE, + sweepJitterInMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_JITTER_IN_MS, + sweepBudgetMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_BUDGET_MS, + } + : undefined, // Debounce configuration debounce: { maxDebounceDurationMs: env.RUN_ENGINE_MAXIMUM_DEBOUNCE_DURATION_MS, diff --git a/apps/webapp/app/v3/runStore.server.ts b/apps/webapp/app/v3/runStore.server.ts index 7f51a940cec..e14bd42477f 100644 --- a/apps/webapp/app/v3/runStore.server.ts +++ b/apps/webapp/app/v3/runStore.server.ts @@ -20,6 +20,7 @@ import { } from "~/db.server"; import { env } from "~/env.server"; import { singleton } from "~/utils/singleton"; +import { decorateWithSnapshotStore } from "./snapshotStoreInstance.server"; import { resilienceForClient, type TransactionResilienceConfig, @@ -173,7 +174,12 @@ function tryResolveRunOpsHandles() { } } -export const runStore: RunStore = singleton("RunStore", () => { +/** + * The router with no snapshot decorator. One intended consumer: the orphan sweeper's rule-2 + * lookup, which must ask Postgres whether a run row exists and must never be able to ask Redis + * whether Redis is an orphan. Every other caller wants `runStore`. + */ +export const runStoreWithoutSnapshotDecorator: RunStore = singleton("RunStore.undecorated", () => { const handles = ROUTING_ENABLED ? tryResolveRunOpsHandles() : null; // Single-store passthrough: self-host (one DB), or a context without run-ops handles. if (!handles) { @@ -202,3 +208,7 @@ export const runStore: RunStore = singleton("RunStore", () => { legacyResilience: resilienceForClient(handles.legacyWriter), }); }); + +export const runStore: RunStore = singleton("RunStore", () => + decorateWithSnapshotStore(runStoreWithoutSnapshotDecorator) +); diff --git a/apps/webapp/app/v3/snapshotStoreBindings.server.ts b/apps/webapp/app/v3/snapshotStoreBindings.server.ts new file mode 100644 index 00000000000..e10bbe796b6 --- /dev/null +++ b/apps/webapp/app/v3/snapshotStoreBindings.server.ts @@ -0,0 +1,31 @@ +import type { SnapshotRepairEnqueuer } from "@internal/run-store"; + +export type SweepPassOutcome = { + outcome: "completed" | "partial" | "skipped_locked" | "failed" | "unbound" | "aborted"; + counts?: Record; +}; + +export type SweepRunner = (opts: { + deadline: number; + signal: AbortSignal; +}) => Promise; + +/** Late-bound so the run store never has to import the engine. A third module wires both at boot. */ +let repairEnqueuer: SnapshotRepairEnqueuer | undefined; +let sweepRunner: SweepRunner | undefined; + +export function setSnapshotRepairEnqueuer(fn: SnapshotRepairEnqueuer): void { + repairEnqueuer = fn; +} + +export function getSnapshotRepairEnqueuer(): SnapshotRepairEnqueuer | undefined { + return repairEnqueuer; +} + +export function setSnapshotSweepRunner(fn: SweepRunner): void { + sweepRunner = fn; +} + +export function getSnapshotSweepRunner(): SweepRunner | undefined { + return sweepRunner; +} diff --git a/apps/webapp/app/v3/snapshotStoreBoot.server.ts b/apps/webapp/app/v3/snapshotStoreBoot.server.ts new file mode 100644 index 00000000000..4614a759a93 --- /dev/null +++ b/apps/webapp/app/v3/snapshotStoreBoot.server.ts @@ -0,0 +1,123 @@ +import type { SnapshotStoreMode } from "@internal/run-store"; +import { logger } from "~/services/logger.server"; +import { globalFlagsRegistry } from "~/v3/globalFlagsRegistry.server"; +import { getSnapshotRepairEnqueuer } from "./snapshotStoreBindings.server"; +import { getSnapshotStoreConfig, getSnapshotSweepClient } from "./snapshotStoreInstance.server"; + +export type SnapshotStoreBootDeps = { + mode: SnapshotStoreMode; + hostConfigured: boolean; + completedTtlMs: number; + orphanAgeMs: number; + ping: () => Promise; + repairBound: () => boolean; + log: (message: string, fields?: Record) => void; + warn: (message: string, fields?: Record) => void; +}; + +const PING_TIMEOUT_MS = 5_000; +const FLAG_READY_TIMEOUT_MS = 10_000; + +export async function assertSnapshotStoreBoot(deps: SnapshotStoreBootDeps): Promise { + const pastOff = deps.mode !== "off"; + + if (pastOff && !deps.hostConfigured) { + throw new Error( + `Snapshot store dial is "${deps.mode}" but RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST is unset, so nothing is constructed; refusing to start.` + ); + } + + if (pastOff && !(deps.completedTtlMs > 0)) { + throw new Error("RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS must be a positive integer."); + } + + if (pastOff && !(deps.orphanAgeMs > 0)) { + throw new Error("RUN_ENGINE_SNAPSHOT_STORE_ORPHAN_AGE_MS must be a positive integer."); + } + + // Nothing enforces that a process which appends has imported the engine module, and an unbound + // enqueuer loses every repair job silently — which burns a task attempt per lost repair. + if (pastOff && !deps.repairBound()) { + throw new Error( + "Snapshot store dial is past off but the repair enqueuer is unbound; refusing to start." + ); + } + + if (pastOff) { + const reachable = await deps.ping(); + if (!reachable) { + if (deps.mode === "redis-only") { + throw new Error( + "Snapshot store dial is redis-only and the endpoint is unreachable; refusing to start." + ); + } + // Postgres is authoritative below redis-only, so a lost append costs nothing. Refusing here + // would bleed fleet capacity during a Redis fault to protect a write path that is free. + deps.warn("Snapshot store Redis is unreachable; booting because Postgres is authoritative", { + mode: deps.mode, + }); + } + } + + if (deps.mode === "redis-only") { + deps.warn( + "Snapshot store dial is redis-only but this build always writes Postgres snapshots. Double-writing is safe, but a Redis fault at this position fails run creation.", + { mode: deps.mode } + ); + } + + deps.log("snapshot store resolved", { + mode: deps.mode, + hostConfigured: deps.hostConfigured, + completedTtlMs: deps.completedTtlMs, + orphanAgeMs: deps.orphanAgeMs, + }); +} + +async function pingSweepClient(): Promise { + const client = getSnapshotSweepClient(); + if (!client) { + return false; + } + try { + const result = await Promise.race([ + client.ping(), + new Promise((_, reject) => + setTimeout(() => reject(new Error("ping timed out")), PING_TIMEOUT_MS) + ), + ]); + return result === "PONG"; + } catch { + return false; + } +} + +/** The env-reading adapter. The boot log line is not authoritative after boot: the dial can move. */ +export async function assertSnapshotStoreBootFromEnv(): Promise { + // Wait for the flag snapshot's first load. Without this the resolved dial is always the env + // floor, because a cold registry returns undefined, and the configuration check would only ever + // see a value no operator sets. A registry that never loads leaves the check on the floor, which + // fails toward inert. + await Promise.race([ + globalFlagsRegistry.isReady, + new Promise((resolve) => setTimeout(resolve, FLAG_READY_TIMEOUT_MS)), + ]); + + const config = getSnapshotStoreConfig(); + + await assertSnapshotStoreBoot({ + mode: config.mode, + hostConfigured: config.configured, + completedTtlMs: config.completedTtlMs, + orphanAgeMs: config.orphanAgeMs, + ping: pingSweepClient, + repairBound: () => !!getSnapshotRepairEnqueuer(), + log: (message, fields) => + logger.info(message, { + ...fields, + keyPrefix: config.keyPrefix, + clusterMode: config.clusterMode, + }), + warn: (message, fields) => logger.warn(message, fields), + }); +} diff --git a/apps/webapp/app/v3/snapshotStoreFlagGuard.server.ts b/apps/webapp/app/v3/snapshotStoreFlagGuard.server.ts new file mode 100644 index 00000000000..8e69e82e367 --- /dev/null +++ b/apps/webapp/app/v3/snapshotStoreFlagGuard.server.ts @@ -0,0 +1,38 @@ +import { FEATURE_FLAG } from "~/v3/featureFlags"; + +/** + * Refuses a dial flip that would be silent: with no host the store is never constructed, so a flag + * past `off` has no effect. Cannot live in validatePartialFeatureFlags, which client components + * import and which therefore can never read env. + */ +export function snapshotStoreFlagSaveError( + requested: Record, + opts: { redisHostConfigured: boolean } +): string | undefined { + if (opts.redisHostConfigured) { + return undefined; + } + + // Both keys, because either one past `off` is equally silent without a connection. + for (const key of [FEATURE_FLAG.snapshotStoreMode, FEATURE_FLAG.snapshotStoreOrgMode] as const) { + const value = requested[key]; + if (typeof value === "string" && value !== "off") { + return `Cannot set ${key} to "${value}": RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST is not configured in this deployment, so the snapshot store is never constructed and the flag would have no effect.`; + } + } + + return undefined; +} + +/** + * Refuses an organisation-only key on a global save. Nothing reads the global row for it, so a + * value saved there is inert, and an inert control an operator can set is worse than no control. + */ +export function globalOnlySnapshotStoreFlagError( + requested: Record +): string | undefined { + if (FEATURE_FLAG.snapshotStoreOrgMode in requested) { + return `${FEATURE_FLAG.snapshotStoreOrgMode} is per-organisation only; nothing reads it from the global flags, so setting it here would have no effect.`; + } + return undefined; +} diff --git a/apps/webapp/app/v3/snapshotStoreInstance.server.ts b/apps/webapp/app/v3/snapshotStoreInstance.server.ts new file mode 100644 index 00000000000..7ffd5ef2d06 --- /dev/null +++ b/apps/webapp/app/v3/snapshotStoreInstance.server.ts @@ -0,0 +1,154 @@ +import { + createRedisClient, + createRedisClusterClient, + type RedisClient, + type RedisOptions, +} from "@internal/redis"; +import { + RedisSnapshotStore, + TaskRunExecutionSnapshotStore, + type RunStore, +} from "@internal/run-store"; +import { env } from "~/env.server"; +import { logger } from "~/services/logger.server"; +import { singleton } from "~/utils/singleton"; +import { getSnapshotRepairEnqueuer } from "./snapshotStoreBindings.server"; +import { snapshotStoreModeResolver } from "./snapshotStoreMode.server"; +import { createSnapshotStoreMetrics } from "./snapshotStoreMetrics.server"; +import { meter } from "./tracer.server"; + +const KEY_PREFIX = "engine:"; + +function isConfigured(): boolean { + return !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST; +} + +function redisOptions(): RedisOptions { + return { + keyPrefix: KEY_PREFIX, + host: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST ?? undefined, + port: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_PORT ?? undefined, + username: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_USERNAME ?? undefined, + password: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_PASSWORD ?? undefined, + enableAutoPipelining: true, + ...(env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }), + }; +} + +function isClusterMode(): boolean { + return env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_CLUSTER_MODE_ENABLED === "1"; +} + +function buildClient(name: string): RedisClient { + const options = redisOptions(); + const onError = (error: Error) => + logger.error(`snapshot store redis client error (${name})`, { error }); + + if (!isClusterMode()) { + return createRedisClient(options, { onError }); + } + + return createRedisClusterClient( + { + nodes: [ + { + host: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST, + port: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_PORT, + }, + ], + redisOptions: options, + }, + { onError } + ); +} + +type Instance = { + sweepClient: RedisClient; + hotPathClient: RedisClient; + redisSnapshotStore: RedisSnapshotStore; + decorate: (store: RunStore) => RunStore; +}; + +const instance = singleton("snapshotStoreInstance", () => { + if (!isConfigured()) { + return undefined; + } + + const metrics = createSnapshotStoreMetrics(meter); + + // The sweep gets a connection of its own so a full scan of every master can never stall a + // transition append. It also backs the sweep's exclusion lock. + const sweepClient = buildClient("sweep"); + + const hotPathClient = buildClient("store"); + + const redisSnapshotStore = new RedisSnapshotStore({ + client: hotPathClient, + completedTtlMs: env.RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS, + metrics: metrics.store, + }); + + return { + sweepClient, + hotPathClient, + redisSnapshotStore, + decorate: (store: RunStore) => + new TaskRunExecutionSnapshotStore(store, { + store: redisSnapshotStore, + modeResolver: snapshotStoreModeResolver, + // Pinned: the field defaults to 0, which would mean no read ever reaches Redis. The + // organisation is the ramp unit, so there is no percentage to ramp. + readPercent: 100, + metrics: metrics.decorator, + onAppendFailure: async (args) => { + const enqueue = getSnapshotRepairEnqueuer(); + if (!enqueue) { + logger.error("snapshot repair enqueuer is unbound; repair job dropped", args); + return; + } + await enqueue(args); + }, + }), + }; +}); + +/** Returns the store verbatim when no snapshot-store Redis is configured. */ +export function decorateWithSnapshotStore(store: RunStore): RunStore { + return instance ? instance.decorate(store) : store; +} + +export function getSnapshotSweepClient(): RedisClient | undefined { + return instance?.sweepClient; +} + +export function getSnapshotStoreConfig() { + return { + configured: isConfigured(), + mode: snapshotStoreModeResolver.resolve(), + completedTtlMs: env.RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS, + orphanAgeMs: env.RUN_ENGINE_SNAPSHOT_STORE_ORPHAN_AGE_MS, + keyPrefix: KEY_PREFIX, + clusterMode: isClusterMode(), + }; +} + +const extraQuits: (() => Promise)[] = []; + +/** Lets the wiring module hand back a teardown for what it built, so this owns closing everything. */ +export function registerSnapshotStoreQuit(quit: () => Promise): void { + extraQuits.push(quit); +} + +export async function quitSnapshotStoreClients(): Promise { + if (!instance) { + return; + } + for (const quit of extraQuits) { + await quit().catch(() => undefined); + } + await instance.sweepClient.quit().catch(() => undefined); + // Last: an append in flight must still land. The store's own quit() returns early on a + // caller-supplied client, so closing the socket is ours. + await instance.redisSnapshotStore.quit().catch(() => undefined); + await instance.hotPathClient.quit().catch(() => undefined); +} diff --git a/apps/webapp/app/v3/snapshotStoreMetrics.server.ts b/apps/webapp/app/v3/snapshotStoreMetrics.server.ts new file mode 100644 index 00000000000..c6990f7bac7 --- /dev/null +++ b/apps/webapp/app/v3/snapshotStoreMetrics.server.ts @@ -0,0 +1,133 @@ +import type { Meter } from "@internal/tracing"; +import type { DecoratorMetrics, SnapshotStoreMetrics } from "@internal/run-store"; + +export type SnapshotSweepCounts = Record; + +// Metric attributes must be bounded: every one of these is a time series. The store and the +// decorator type their outcome strings loosely, so anything unrecognised collapses to "other" +// rather than minting a series. +const APPEND_OUTCOMES = ["written", "duplicate", "forked", "skippedNoKeyspace"] as const; +const APPEND_TTLS = ["none", "completion", "reapplied"] as const; +const WRITE_OUTCOMES = ["written", "staged", "post_expiry", "skipped", "failed"] as const; +const READ_SOURCES = ["redis", "postgres"] as const; +const SWEEP_OUTCOMES = [ + "completed", + "partial", + "skipped_locked", + "failed", + "unbound", + "aborted", +] as const; +const SWEEP_FIELDS = [ + "scanned", + "expired", + "deleted", + "skipped", + "pendingDeletion", + "nodes", +] as const; + +const WRITE_SITES = [ + "createRun", + "createCancelledRun", + "completeAttemptSuccess", + "expireRun", + "expireParkedRun", + "rescheduleRun", + "lockRunToWorker", + "createExecutionSnapshot", + "runInTransaction", +] as const; +const READ_METHODS = [ + "findLatestExecutionSnapshot", + "findExecutionSnapshot", + "findManyExecutionSnapshots", + "findSnapshotCompletedWaitpointIds", + "findSnapshotCompletedWaitpointIdsWithPresence", +] as const; +const SNAPSHOT_OPS = [ + "append", + "getById", + "getLatest", + "getSince", + "getSinceCreatedAt", + "getSnapshotWaitpointIds", +] as const; + +function bounded(value: string, allowed: readonly string[]): string { + return allowed.includes(value) ? value : "other"; +} + +/** + * Every instrument is created inside this function. At module scope they would register on every + * boot, including deployments with no snapshot-store Redis configured. + */ +export function createSnapshotStoreMetrics(meter: Meter) { + // Two layers, two counters. Sharing one would count a single logical write twice and mix + // {outcome, ttl} points with {site, outcome} points under one name, so no sum or grouping over it + // would mean anything. + const appendTotal = meter.createCounter("run_engine.snapshot_store.append_total"); + const writeTotal = meter.createCounter("run_engine.snapshot_store.write_total"); + const appendFailed = meter.createCounter("run_engine.snapshot_store.append_failed"); + const flushStaged = meter.createCounter("run_engine.snapshot_store.flush_staged"); + const readSource = meter.createCounter("run_engine.snapshot_store.read_source"); + const postExpiryWrite = meter.createCounter("run_engine.snapshot_store.post_expiry_write"); + const skippedNoKeyspace = meter.createCounter("run_engine.snapshot_store.skipped_no_keyspace"); + const cycleMismatch = meter.createCounter("run_engine.snapshot_store.cycle_mismatch"); + const entryBytes = meter.createHistogram("run_engine.snapshot_store.entry_bytes"); + const cycleKeyBytes = meter.createHistogram("run_engine.snapshot_store.cycle_key_bytes"); + const cycleCount = meter.createHistogram("run_engine.snapshot_store.cycle_count"); + const opLatency = meter.createHistogram("run_engine.snapshot_store.op_latency_ms"); + const sweepPass = meter.createCounter("run_engine.snapshot_store.sweep_pass_total"); + const sweepCounts = meter.createHistogram("run_engine.snapshot_store.sweep_counts"); + + const store: SnapshotStoreMetrics = { + recordAppend: (outcome, ttl) => + appendTotal.add(1, { + outcome: bounded(outcome, APPEND_OUTCOMES), + ttl: bounded(ttl, APPEND_TTLS), + }), + recordEntryBytes: (bytes) => entryBytes.record(bytes), + recordCycleKeyBytes: (bytes) => cycleKeyBytes.record(bytes), + recordCycleCount: (count) => cycleCount.record(count), + recordSkippedNoKeyspace: () => skippedNoKeyspace.add(1), + recordCycleMismatch: () => cycleMismatch.add(1), + recordLatency: (op, ms) => opLatency.record(ms, { op: bounded(op, SNAPSHOT_OPS) }), + }; + + const decorator: DecoratorMetrics = { + recordWrite: (site, outcome) => { + writeTotal.add(1, { + site: bounded(site, WRITE_SITES), + outcome: bounded(outcome, WRITE_OUTCOMES), + }); + if (outcome === "post_expiry") { + postExpiryWrite.add(1); + } + if (outcome === "staged") { + flushStaged.add(1); + } + }, + recordAppendFailed: (site) => appendFailed.add(1, { site: bounded(site, WRITE_SITES) }), + recordRead: (method, source) => + readSource.add(1, { + method: bounded(method, READ_METHODS), + source: bounded(source, READ_SOURCES), + }), + }; + + /** One emitter per pass, so a pass that throws is distinguishable from one that succeeded. */ + function recordSweepPass(outcome: string, counts?: SnapshotSweepCounts): void { + sweepPass.add(1, { outcome: bounded(outcome, SWEEP_OUTCOMES) }); + if (!counts) { + return; + } + for (const [field, value] of Object.entries(counts)) { + if (typeof value === "number" && SWEEP_FIELDS.includes(field as never)) { + sweepCounts.record(value, { field }); + } + } + } + + return { store, decorator, recordSweepPass }; +} diff --git a/apps/webapp/app/v3/snapshotStoreMode.server.ts b/apps/webapp/app/v3/snapshotStoreMode.server.ts new file mode 100644 index 00000000000..e5b4b199c4e --- /dev/null +++ b/apps/webapp/app/v3/snapshotStoreMode.server.ts @@ -0,0 +1,166 @@ +import { LRUCache } from "lru-cache"; +import type { SnapshotStoreMode, SnapshotStoreModeResolver } from "@internal/run-store"; +import { $replica, prisma } from "~/db.server"; +import { env } from "~/env.server"; +import { logger } from "~/services/logger.server"; +import { singleton } from "~/utils/singleton"; +import { FEATURE_FLAG, FeatureFlagCatalog } from "~/v3/featureFlags"; +import { globalFlagsRegistry } from "~/v3/globalFlagsRegistry.server"; + +/** A cached "this organisation has no override", distinct from "not cached". */ +export const NO_OVERRIDE = "__none__" as const; + +/** + * The dial positions, declared here rather than imported, so this module does not depend on the + * run-store package's build output to typecheck. The assertion below fails if the two ever diverge. + */ +type DialMode = "off" | "dual-write" | "redis-read" | "redis-only"; + +/** Only the write positions are settable per organisation: reads are global. */ +type OrgDialMode = "off" | "dual-write"; + +type AssertSame = [A] extends [B] ? ([B] extends [A] ? true : never) : never; +const _dialMatchesRunStore: AssertSame = true; +void _dialMatchesRunStore; + +type CachedOrgMode = OrgDialMode | typeof NO_OVERRIDE; + +/** + * What to cache for one organisation's blob value. An unparseable or absent value caches + * NO_OVERRIDE rather than nothing: caching nothing means every organisation without an override + * re-queries on every write, which is all of them until a ramp starts. + */ +export function cachedOrgModeFor(raw: unknown): CachedOrgMode { + const parsed = FeatureFlagCatalog[FEATURE_FLAG.snapshotStoreOrgMode].safeParse(raw); + return parsed.success ? parsed.data : NO_OVERRIDE; +} + +type OrgModeSource = { + /** Cache-only: returns undefined on a true miss rather than querying. */ + get(organizationId: string): CachedOrgMode | undefined; + /** Fire-and-forget, de-duplicated per organisation, never throws. */ + refresh(organizationId: string): void; + /** Re-reads from the primary after a write, so replica lag cannot re-cache the old value. */ + invalidate(organizationId: string): void; +}; + +/** What resolution needs. Invalidation is a save-path concern, not a read-path one. */ +type ResolverOrgSource = Pick; + +export function buildSnapshotStoreModeResolver(deps: { + globalMode: () => DialMode | undefined; + orgMode: ResolverOrgSource; + envFloor: DialMode; +}): SnapshotStoreModeResolver { + return { + resolve(organizationId?: string): DialMode { + const global = deps.globalMode() ?? deps.envFloor; + if (!organizationId) { + return global; + } + + const cached = deps.orgMode.get(organizationId); + if (cached === NO_OVERRIDE) { + return global; + } + if (cached !== undefined) { + return cached; + } + + // Deliberately no read here. Seven decorator methods accept a caller-supplied `tx`, so a + // query on this path can land inside another caller's open interactive transaction, on the + // same pool for single-DB and self-host. Serve the global answer, warm the cache off-path. + try { + deps.orgMode.refresh(organizationId); + } catch { + // a warm-up must never fail a state transition + } + return global; + }, + }; +} + +const DEFAULT_CACHE_MAX = 10_000; +const DEFAULT_CACHE_TTL_MS = 30_000; + +function createOrgModeSource(): OrgModeSource { + // Defaults inline as well as in the schema: this must not throw when a caller supplies a partial + // env, and an LRU with neither bound set is a constructor error. + const cache = new LRUCache({ + max: env.RUN_ENGINE_SNAPSHOT_STORE_ORG_MODE_CACHE_MAX ?? DEFAULT_CACHE_MAX, + ttl: env.RUN_ENGINE_SNAPSHOT_STORE_ORG_MODE_CACHE_TTL_MS ?? DEFAULT_CACHE_TTL_MS, + }); + const inFlight = new Set(); + // A replica read that started before an invalidation can land after the primary read and put the + // superseded value back. A per-organisation generation lets a stale load discard its own result. + const generations = new Map(); + const generationOf = (organizationId: string) => generations.get(organizationId) ?? 0; + + return { + get: (organizationId) => cache.get(organizationId), + invalidate: (organizationId) => { + // Drop first, so a resolve between now and the re-read falls back rather than serving a + // value the write just replaced. + const generation = generationOf(organizationId) + 1; + generations.set(organizationId, generation); + cache.delete(organizationId); + void load(organizationId, prisma, generation); + }, + refresh: (organizationId) => { + if (inFlight.has(organizationId)) { + return; + } + inFlight.add(organizationId); + + void load(organizationId, $replica, generationOf(organizationId)).finally(() => { + inFlight.delete(organizationId); + }); + }, + }; + + function load( + organizationId: string, + client: typeof prisma | typeof $replica, + generation: number + ) { + return client.organization + .findFirst({ where: { id: organizationId }, select: { featureFlags: true } }) + .then((row) => { + // Only the narrow per-org key. The blob is never passed as `overrides` for the global + // key, where a parsing override would win outright. + const raw = (row?.featureFlags as Record | null | undefined)?.[ + FEATURE_FLAG.snapshotStoreOrgMode + ]; + // A newer invalidation happened while this read was in flight, so its answer is stale. + if (generation < generationOf(organizationId)) { + return; + } + cache.set(organizationId, cachedOrgModeFor(raw)); + }) + .catch((error) => { + logger.warn("snapshotStoreMode: organisation override read failed", { + organizationId, + error, + }); + }); + } +} + +/** Built on first use, never at import: importing this module must have no side effect. */ +function orgModeSource(): OrgModeSource { + return singleton("snapshotStoreOrgModeSource", createOrgModeSource); +} + +export const snapshotStoreModeResolver: SnapshotStoreModeResolver = buildSnapshotStoreModeResolver({ + globalMode: () => globalFlagsRegistry.current()?.[FEATURE_FLAG.snapshotStoreMode], + orgMode: { + get: (organizationId) => orgModeSource().get(organizationId), + refresh: (organizationId) => orgModeSource().refresh(organizationId), + }, + envFloor: env.RUN_ENGINE_SNAPSHOT_STORE_MODE ?? "off", +}); + +/** Called by the organisation flag save path so the writing process sees a dial change at once. */ +export function invalidateSnapshotStoreOrgMode(organizationId: string): void { + orgModeSource().invalidate(organizationId); +} diff --git a/apps/webapp/app/v3/snapshotStoreWiring.server.ts b/apps/webapp/app/v3/snapshotStoreWiring.server.ts new file mode 100644 index 00000000000..9cf96b0e35c --- /dev/null +++ b/apps/webapp/app/v3/snapshotStoreWiring.server.ts @@ -0,0 +1,63 @@ +import { SnapshotOrphanSweeper } from "@internal/run-store"; +import { env } from "~/env.server"; +import { logger } from "~/services/logger.server"; +import { signalsEmitter } from "~/services/signals.server"; +import { engine } from "./runEngine.server"; +import { runStoreWithoutSnapshotDecorator } from "./runStore.server"; +import { buildSnapshotSweepRunner } from "./snapshotSweepRunner.server"; +import { setSnapshotRepairEnqueuer, setSnapshotSweepRunner } from "./snapshotStoreBindings.server"; +import { + getSnapshotSweepClient, + quitSnapshotStoreClients, + registerSnapshotStoreQuit, +} from "./snapshotStoreInstance.server"; + +/** + * The third module: it imports both sides, so neither the run store nor the engine has to import + * the other. Invoked from entry.server.tsx, beside the other boot registrations. + */ +export function registerSnapshotStoreWiring(): boolean { + const sweepClient = getSnapshotSweepClient(); + + if (!sweepClient) { + return false; + } + + setSnapshotRepairEnqueuer(async (args) => { + await engine.enqueueSnapshotRepair(args); + }); + + const sweeper = new SnapshotOrphanSweeper({ + // Its own connection, so a scan of every master can never stall a transition append. + client: sweepClient, + // The undecorated router: rule 2 asks Postgres whether a run row exists, and must never be + // able to ask Redis whether Redis is an orphan. + runStore: runStoreWithoutSnapshotDecorator, + completedTtlMs: env.RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS, + orphanAgeMs: env.RUN_ENGINE_SNAPSHOT_STORE_ORPHAN_AGE_MS, + confirmOrphanAfterMs: env.RUN_ENGINE_SNAPSHOT_STORE_CONFIRM_ORPHAN_AFTER_MS, + }); + + setSnapshotSweepRunner( + buildSnapshotSweepRunner({ + client: sweepClient, + sweep: async ({ deadline, signal }) => ({ ...(await sweeper.sweep({ deadline, signal })) }), + lockTtlMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_BUDGET_MS + 3_600_000, + }) + ); + + registerSnapshotStoreQuit(() => sweeper.quit()); + + // Close the sweeper and all three connections on the way out, the same way the other Redis-backed + // singletons do. `quitSnapshotStoreClients` is async and the signals emitter swallows listener + // rejections, so discard the promise explicitly rather than handing it a floating one. The caller + // wraps this function in `singleton`, so the listeners are registered once per process. + const onShutdown = (): void => { + void quitSnapshotStoreClients(); + }; + signalsEmitter.on("SIGTERM", onShutdown); + signalsEmitter.on("SIGINT", onShutdown); + + logger.info("snapshot store wiring registered"); + return true; +} diff --git a/apps/webapp/app/v3/snapshotSweepRunner.server.ts b/apps/webapp/app/v3/snapshotSweepRunner.server.ts new file mode 100644 index 00000000000..c39c86dca8d --- /dev/null +++ b/apps/webapp/app/v3/snapshotSweepRunner.server.ts @@ -0,0 +1,58 @@ +import type { RedisClient } from "@internal/redis"; +import { logger } from "~/services/logger.server"; +import type { SweepPassOutcome, SweepRunner } from "./snapshotStoreBindings.server"; + +const LOCK_KEY = "snapshot-sweep:lock"; + +// Compare-and-delete. A bare DEL would let a pass that overran its own lock delete its SUCCESSOR's +// lock on release, and two passes would then run together — the failure the lock exists to prevent. +const RELEASE_LUA = ` +if redis.call("get", KEYS[1]) == ARGV[1] then + return redis.call("del", KEYS[1]) +end +return 0 +`; + +type SweepCounts = Record; + +export function buildSnapshotSweepRunner(deps: { + client: RedisClient; + sweep: (opts: { deadline: number; signal: AbortSignal }) => Promise; + lockTtlMs: number; + fence?: () => string; +}): SweepRunner { + return async ({ deadline, signal }): Promise => { + // enqueueOnce gives no overlap protection: its dedup record IS the queue item and the ack + // deletes it, so it elects a winner only at start-up. Nothing extends the visibility timeout + // either, so a long pass is redelivered and would otherwise run beside itself. + const fence = + deps.fence?.() ?? `${process.pid}-${Date.now()}-${Math.random().toString(36).slice(2)}`; + const acquired = await deps.client.set(LOCK_KEY, fence, "PX", deps.lockTtlMs, "NX"); + + if (acquired !== "OK") { + return { outcome: "skipped_locked" }; + } + + try { + const counts = await deps.sweep({ deadline, signal }); + + if (signal.aborted) { + return { outcome: "aborted", counts }; + } + if (counts.partial === true) { + return { outcome: "partial", counts }; + } + return { outcome: "completed", counts }; + } catch (error) { + if (signal.aborted) { + return { outcome: "aborted" }; + } + logger.error("snapshot orphan sweep pass failed", { error }); + return { outcome: "failed" }; + } finally { + await deps.client + .eval(RELEASE_LUA, 1, LOCK_KEY, fence) + .catch((error) => logger.warn("snapshot sweep lock release failed", { error })); + } + }; +} diff --git a/apps/webapp/test/snapshotStoreBoot.test.ts b/apps/webapp/test/snapshotStoreBoot.test.ts new file mode 100644 index 00000000000..f56b6c639da --- /dev/null +++ b/apps/webapp/test/snapshotStoreBoot.test.ts @@ -0,0 +1,79 @@ +import { describe, expect, it } from "vitest"; +import { assertSnapshotStoreBoot } from "~/v3/snapshotStoreBoot.server"; + +type Recorded = { logs: string[]; warnings: string[]; pings: number }; + +function deps(overrides: Partial[0]> = {}) { + const recorded: Recorded = { logs: [], warnings: [], pings: 0 }; + const base = { + mode: "off" as const, + hostConfigured: false, + completedTtlMs: 1, + orphanAgeMs: 1, + ping: async () => { + recorded.pings += 1; + return true; + }, + repairBound: () => true, + log: (message: string) => recorded.logs.push(message), + warn: (message: string) => recorded.warnings.push(message), + }; + return { ...base, ...overrides, recorded }; +} + +describe("assertSnapshotStoreBoot", () => { + it("passes and logs when the dial is off and nothing is configured", async () => { + const d = deps(); + await expect(assertSnapshotStoreBoot(d)).resolves.toBeUndefined(); + expect(d.recorded.logs).toHaveLength(1); + }); + + it("does not probe reachability at off", async () => { + const d = deps(); + await expect(assertSnapshotStoreBoot(d)).resolves.toBeUndefined(); + expect(d.recorded.pings).toBe(0); + }); + + it("refuses a dial past off with no host", async () => { + await expect( + assertSnapshotStoreBoot(deps({ mode: "dual-write", hostConfigured: false })) + ).rejects.toThrow(/RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST/); + }); + + it("refuses a non-positive TTL once the dial is past off", async () => { + await expect( + assertSnapshotStoreBoot(deps({ mode: "dual-write", hostConfigured: true, completedTtlMs: 0 })) + ).rejects.toThrow(/COMPLETED_TTL_MS/); + await expect( + assertSnapshotStoreBoot(deps({ mode: "dual-write", hostConfigured: true, orphanAgeMs: -1 })) + ).rejects.toThrow(/ORPHAN_AGE_MS/); + }); + + it("refuses when the repair binding is unset and the dial is past off", async () => { + await expect( + assertSnapshotStoreBoot( + deps({ mode: "dual-write", hostConfigured: true, repairBound: () => false }) + ) + ).rejects.toThrow(/repair/i); + }); + + it("boots on an unreachable endpoint below redis-only, loudly", async () => { + const d = deps({ mode: "dual-write", hostConfigured: true, ping: async () => false }); + await expect(assertSnapshotStoreBoot(d)).resolves.toBeUndefined(); + expect(d.recorded.warnings.length).toBeGreaterThan(0); + }); + + it("refuses an unreachable endpoint at redis-only", async () => { + await expect( + assertSnapshotStoreBoot( + deps({ mode: "redis-only", hostConfigured: true, ping: async () => false }) + ) + ).rejects.toThrow(/unreachable/i); + }); + + it("warns at redis-only because this build still writes Postgres snapshots", async () => { + const d = deps({ mode: "redis-only", hostConfigured: true }); + await expect(assertSnapshotStoreBoot(d)).resolves.toBeUndefined(); + expect(d.recorded.warnings.join(" ")).toMatch(/Postgres/i); + }); +}); diff --git a/apps/webapp/test/snapshotStoreConstruction.test.ts b/apps/webapp/test/snapshotStoreConstruction.test.ts new file mode 100644 index 00000000000..8a76d96e586 --- /dev/null +++ b/apps/webapp/test/snapshotStoreConstruction.test.ts @@ -0,0 +1,69 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +// A socket-level assertion is impossible in this suite: test/setup.ts mocks ioredis with a +// LazyRedis subclass that forces lazyConnect, so no client ever dials and a "no connection was +// opened" test would pass even if the gate were broken. The property that does hold, and the one +// that matters, is that no client OBJECT is constructed — an unconstructed client cannot dial. +async function importInstanceModule() { + vi.resetModules(); + delete (globalThis as Record).__trigger_singletons; + return import("~/v3/snapshotStoreInstance.server"); +} + +afterEach(async () => { + vi.unstubAllEnvs(); + vi.resetModules(); + delete (globalThis as Record).__trigger_singletons; +}); + +describe("snapshot store construction gate", () => { + it("constructs nothing when the snapshot-store host is unset", async () => { + // The generic pair is set by test/setup.ts, so this also covers the no-fallback rule: if any + // variable in the block fell back to REDIS_HOST, the store would be constructed here. + vi.stubEnv("RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST", ""); + + const mod = await importInstanceModule(); + const sentinel = {} as never; + + expect(mod.decorateWithSnapshotStore(sentinel)).toBe(sentinel); + expect(mod.getSnapshotSweepClient()).toBeUndefined(); + expect(mod.getSnapshotStoreConfig().configured).toBe(false); + }); + + it("constructs the decorator and the sweep client once the host is set", async () => { + vi.stubEnv("RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST", "127.0.0.1"); + vi.stubEnv("RUN_ENGINE_SNAPSHOT_STORE_REDIS_PORT", "6379"); + vi.stubEnv("RUN_ENGINE_SNAPSHOT_STORE_REDIS_TLS_DISABLED", "true"); + + const mod = await importInstanceModule(); + const sentinel = {} as never; + + // Without this the negative above is satisfiable by an import that throws. + expect(mod.decorateWithSnapshotStore(sentinel)).not.toBe(sentinel); + expect(mod.getSnapshotSweepClient()).toBeDefined(); + expect(mod.getSnapshotStoreConfig().configured).toBe(true); + + await mod.quitSnapshotStoreClients(); + }); + + it("reports the resolved configuration for the boot log line", async () => { + vi.stubEnv("RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST", "127.0.0.1"); + vi.stubEnv("RUN_ENGINE_SNAPSHOT_STORE_REDIS_TLS_DISABLED", "true"); + vi.stubEnv("RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS", "1000"); + vi.stubEnv("RUN_ENGINE_SNAPSHOT_STORE_ORPHAN_AGE_MS", "2000"); + + const mod = await importInstanceModule(); + const config = mod.getSnapshotStoreConfig(); + + expect(config).toMatchObject({ + configured: true, + mode: "off", + completedTtlMs: 1000, + orphanAgeMs: 2000, + keyPrefix: "engine:", + clusterMode: false, + }); + + await mod.quitSnapshotStoreClients(); + }); +}); diff --git a/apps/webapp/test/snapshotStoreFlagGuard.test.ts b/apps/webapp/test/snapshotStoreFlagGuard.test.ts new file mode 100644 index 00000000000..f13e2c76f0c --- /dev/null +++ b/apps/webapp/test/snapshotStoreFlagGuard.test.ts @@ -0,0 +1,82 @@ +import { describe, expect, it } from "vitest"; +import { + globalOnlySnapshotStoreFlagError, + snapshotStoreFlagSaveError, +} from "~/v3/snapshotStoreFlagGuard.server"; + +describe("snapshotStoreFlagSaveError", () => { + it("refuses a flip past off when no host is configured", () => { + expect( + snapshotStoreFlagSaveError( + { snapshotStoreMode: "dual-write" }, + { redisHostConfigured: false } + ) + ).toMatch(/RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST/); + }); + + it("names the position it refused", () => { + expect( + snapshotStoreFlagSaveError( + { snapshotStoreMode: "redis-read" }, + { redisHostConfigured: false } + ) + ).toMatch(/redis-read/); + }); + + it("allows a flip past off once the host is configured", () => { + expect( + snapshotStoreFlagSaveError({ snapshotStoreMode: "dual-write" }, { redisHostConfigured: true }) + ).toBeUndefined(); + }); + + it("allows off with no host, because that is the default state", () => { + expect( + snapshotStoreFlagSaveError({ snapshotStoreMode: "off" }, { redisHostConfigured: false }) + ).toBeUndefined(); + }); + + it("ignores a payload that does not mention the dial", () => { + expect( + snapshotStoreFlagSaveError({ runOpsMintKind: "cuid" }, { redisHostConfigured: false }) + ).toBeUndefined(); + }); + + it("ignores a non-string dial value and leaves it to schema validation", () => { + expect( + snapshotStoreFlagSaveError({ snapshotStoreMode: 3 }, { redisHostConfigured: false }) + ).toBeUndefined(); + }); + + it("refuses a per-organisation flip past off when no host is configured", () => { + // Same silence as the global key: without a connection the store is never constructed. + expect( + snapshotStoreFlagSaveError( + { snapshotStoreOrgMode: "dual-write" }, + { redisHostConfigured: false } + ) + ).toMatch(/RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST/); + }); + + it("allows a per-organisation off with no host", () => { + expect( + snapshotStoreFlagSaveError({ snapshotStoreOrgMode: "off" }, { redisHostConfigured: false }) + ).toBeUndefined(); + }); +}); + +describe("globalOnlySnapshotStoreFlagError", () => { + it("refuses the per-organisation key on a global save", () => { + // Nothing reads it from the global row, so a value saved there is an inert control. + expect(globalOnlySnapshotStoreFlagError({ snapshotStoreOrgMode: "dual-write" })).toMatch( + /per-organisation only/ + ); + }); + + it("allows the global key", () => { + expect(globalOnlySnapshotStoreFlagError({ snapshotStoreMode: "dual-write" })).toBeUndefined(); + }); + + it("ignores unrelated payloads", () => { + expect(globalOnlySnapshotStoreFlagError({ runOpsMintKind: "cuid" })).toBeUndefined(); + }); +}); diff --git a/apps/webapp/test/snapshotStoreFlags.test.ts b/apps/webapp/test/snapshotStoreFlags.test.ts new file mode 100644 index 00000000000..db43bc111c8 --- /dev/null +++ b/apps/webapp/test/snapshotStoreFlags.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from "vitest"; +import { + FeatureFlagCatalog, + ORG_LOCKED_FLAGS, + withoutOrgForbiddenSnapshotKeys, +} from "~/v3/featureFlags"; + +describe("snapshot store dial catalog", () => { + it("accepts all four positions globally", () => { + for (const value of ["off", "dual-write", "redis-read", "redis-only"]) { + expect(FeatureFlagCatalog.snapshotStoreMode.safeParse(value).success).toBe(true); + } + }); + + it("rejects an unknown position", () => { + expect(FeatureFlagCatalog.snapshotStoreMode.safeParse("redis-write").success).toBe(false); + }); + + it("accepts only write positions per organisation", () => { + for (const value of ["off", "dual-write"]) { + expect(FeatureFlagCatalog.snapshotStoreOrgMode.safeParse(value).success).toBe(true); + } + // Reads are global, so an org at a read position would read state its own writes never created. + expect(FeatureFlagCatalog.snapshotStoreOrgMode.safeParse("redis-read").success).toBe(false); + expect(FeatureFlagCatalog.snapshotStoreOrgMode.safeParse("redis-only").success).toBe(false); + }); + + it("lists the global dial as org-locked", () => { + expect(ORG_LOCKED_FLAGS).toContain("snapshotStoreMode"); + }); +}); + +describe("withoutOrgForbiddenSnapshotKeys", () => { + it("removes the global dial and keeps everything else", () => { + expect( + withoutOrgForbiddenSnapshotKeys({ + snapshotStoreMode: "redis-only", + snapshotStoreOrgMode: "dual-write", + runOpsMintKind: "cuid", + }) + ).toEqual({ snapshotStoreOrgMode: "dual-write", runOpsMintKind: "cuid" }); + }); + + it("returns the same object when the dial is absent", () => { + const input = { runOpsMintKind: "cuid" }; + expect(withoutOrgForbiddenSnapshotKeys(input)).toBe(input); + }); + + it("removes the dial even when it is the only key", () => { + expect(withoutOrgForbiddenSnapshotKeys({ snapshotStoreMode: "dual-write" })).toEqual({}); + }); +}); diff --git a/apps/webapp/test/snapshotStoreMetrics.test.ts b/apps/webapp/test/snapshotStoreMetrics.test.ts new file mode 100644 index 00000000000..e57bdc711b7 --- /dev/null +++ b/apps/webapp/test/snapshotStoreMetrics.test.ts @@ -0,0 +1,50 @@ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; + +const SOURCE_PATH = join(process.cwd(), "app/v3/snapshotStoreMetrics.server.ts"); + +describe("snapshotStoreMetrics module shape", () => { + // An instrument created but never incremented emits no data point, so a MeterProvider cannot see + // one that drifted back to module scope, where it would register on every boot. + const source = readFileSync(SOURCE_PATH, "utf8"); + const factoryAt = source.indexOf("export function createSnapshotStoreMetrics"); + + it("exports the factory", () => { + expect(factoryAt).toBeGreaterThan(-1); + }); + + it("creates every instrument inside the factory", () => { + const creations = [ + ...source.matchAll(/create(Counter|Histogram|UpDownCounter|Observable\w*)\(/g), + ]; + expect(creations.length).toBeGreaterThan(0); + for (const match of creations) { + expect(match.index).toBeGreaterThan(factoryAt); + } + }); + + it("calls getMeter nowhere at module scope", () => { + expect(source).not.toMatch(/^\s*(const|let|var)\s+\w+\s*=\s*getMeter\(/m); + }); + + it("declares no counter that has no producer in this ticket", () => { + // A counter pinned at zero looks the same as a working one that found nothing. + expect(source).not.toMatch(/compare_divergence/); + expect(source).not.toMatch(/\btrimmed\b/); + }); + + it("gives the two layers separate counters", () => { + // Sharing one would count a single logical write twice and mix {outcome, ttl} points with + // {site, outcome} points under one name. + const appendBlock = source.slice( + source.indexOf("recordAppend:"), + source.indexOf("recordWrite:") + ); + const writeBlock = source.slice(source.indexOf("recordWrite:")); + expect(appendBlock).toMatch(/appendTotal\.add/); + expect(appendBlock).not.toMatch(/writeTotal\.add/); + expect(writeBlock).toMatch(/writeTotal\.add/); + expect(writeBlock).not.toMatch(/appendTotal\.add/); + }); +}); diff --git a/apps/webapp/test/snapshotStoreMode.test.ts b/apps/webapp/test/snapshotStoreMode.test.ts new file mode 100644 index 00000000000..e01fa3c8dbd --- /dev/null +++ b/apps/webapp/test/snapshotStoreMode.test.ts @@ -0,0 +1,112 @@ +import { describe, expect, it, vi } from "vitest"; +import type { SnapshotStoreMode } from "@internal/run-store"; +import { + buildSnapshotStoreModeResolver, + cachedOrgModeFor, + NO_OVERRIDE, +} from "~/v3/snapshotStoreMode.server"; + +function build(opts: { + globalMode?: SnapshotStoreMode; + perOrg?: Record; + envFloor?: SnapshotStoreMode; + refresh?: (organizationId: string) => void; +}) { + return buildSnapshotStoreModeResolver({ + globalMode: () => opts.globalMode, + orgMode: { + get: (id: string) => opts.perOrg?.[id], + refresh: opts.refresh ?? (() => {}), + }, + envFloor: opts.envFloor ?? "off", + }); +} + +describe("snapshot store mode resolver", () => { + it("falls back to the env floor when the global snapshot is cold", () => { + expect(build({ envFloor: "off" }).resolve()).toBe("off"); + expect(build({ envFloor: "dual-write" }).resolve()).toBe("dual-write"); + }); + + it("prefers the global flag over the floor", () => { + expect(build({ globalMode: "redis-read", envFloor: "off" }).resolve()).toBe("redis-read"); + }); + + it("prefers an organisation override over the global flag", () => { + const r = build({ globalMode: "off", perOrg: { org_a: "dual-write" } }); + expect(r.resolve("org_a")).toBe("dual-write"); + expect(r.resolve("org_b")).toBe("off"); + }); + + it("lets an organisation be off while the global flag is on", () => { + const r = build({ globalMode: "dual-write", perOrg: { org_a: "off" } }); + expect(r.resolve("org_a")).toBe("off"); + expect(r.resolve("org_b")).toBe("dual-write"); + }); + + it("serves the global answer on a cold organisation and schedules a refresh", () => { + const refresh = vi.fn(); + const r = build({ globalMode: "dual-write", refresh }); + expect(r.resolve("org_cold")).toBe("dual-write"); + expect(refresh).toHaveBeenCalledWith("org_cold"); + }); + + it("never lets a refresh failure reach the caller", () => { + const refresh = vi.fn(() => { + throw new Error("control plane unreachable"); + }); + const r = build({ globalMode: "off", refresh }); + expect(() => r.resolve("org_x")).not.toThrow(); + expect(r.resolve("org_x")).toBe("off"); + }); + + it("resolves an unknown organisation to the global answer, never a throw", () => { + const r = build({ globalMode: "off", perOrg: {} }); + expect(r.resolve("org_deleted")).toBe("off"); + }); + + it("caches an absent override rather than nothing", () => { + // Caching nothing means every organisation without an override re-queries on every write. + expect(cachedOrgModeFor(undefined)).toBe(NO_OVERRIDE); + expect(cachedOrgModeFor(null)).toBe(NO_OVERRIDE); + expect(cachedOrgModeFor("not-a-mode")).toBe(NO_OVERRIDE); + expect(cachedOrgModeFor("redis-read")).toBe(NO_OVERRIDE); + expect(cachedOrgModeFor("dual-write")).toBe("dual-write"); + }); + + it("stops querying once an absent override is cached", () => { + // Without a cached negative, every organisation with no override re-queries on every write, + // which is every organisation until a ramp starts. + const refresh = vi.fn(); + let cached: string | undefined; + const r = buildSnapshotStoreModeResolver({ + globalMode: () => "off", + orgMode: { + get: () => cached as never, + refresh: (id: string) => { + refresh(id); + cached = "__none__"; + }, + }, + envFloor: "off", + }); + + expect(r.resolve("org_a")).toBe("off"); + expect(refresh).toHaveBeenCalledTimes(1); + + expect(r.resolve("org_a")).toBe("off"); + expect(r.resolve("org_a")).toBe("off"); + expect(refresh).toHaveBeenCalledTimes(1); + }); + + it("does not consult the organisation source when no organisation is supplied", () => { + const get = vi.fn(() => undefined); + const r = buildSnapshotStoreModeResolver({ + globalMode: () => "redis-read", + orgMode: { get, refresh: () => {} }, + envFloor: "off", + }); + expect(r.resolve()).toBe("redis-read"); + expect(get).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/webapp/test/snapshotStoreModuleLoad.test.ts b/apps/webapp/test/snapshotStoreModuleLoad.test.ts new file mode 100644 index 00000000000..50ed8da596b --- /dev/null +++ b/apps/webapp/test/snapshotStoreModuleLoad.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from "vitest"; + +// Reproduces the CI break. Many webapp suites mock ~/db.server and ~/env.server with minimal +// objects, so a module-load side effect that reads a new env variable takes the whole file down on +// import. This is the exact mock shape of the suite that caught it. +vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} })); +vi.mock("~/env.server", () => ({ env: { SESSION_SECRET: "test-session-secret" } })); +vi.mock("~/services/logger.server", () => ({ + logger: { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} }, +})); + +describe("snapshot store modules under a minimal env mock", () => { + it("imports the mode resolver without constructing anything", async () => { + await expect(import("~/v3/snapshotStoreMode.server")).resolves.toBeDefined(); + }); + + it("resolves off rather than throwing", async () => { + const { snapshotStoreModeResolver } = await import("~/v3/snapshotStoreMode.server"); + expect(snapshotStoreModeResolver.resolve()).toBe("off"); + expect(snapshotStoreModeResolver.resolve("org_anything")).toBe("off"); + }); + + it("imports the instance module and stays undecorated", async () => { + const mod = await import("~/v3/snapshotStoreInstance.server"); + const sentinel = {} as never; + expect(mod.decorateWithSnapshotStore(sentinel)).toBe(sentinel); + expect(mod.getSnapshotSweepClient()).toBeUndefined(); + }); +}); diff --git a/apps/webapp/test/snapshotSweepRunner.test.ts b/apps/webapp/test/snapshotSweepRunner.test.ts new file mode 100644 index 00000000000..1be447c55fc --- /dev/null +++ b/apps/webapp/test/snapshotSweepRunner.test.ts @@ -0,0 +1,130 @@ +import { redisTest } from "@internal/testcontainers"; +import { expect } from "vitest"; +import { createRedisClient } from "@internal/redis"; +import { buildSnapshotSweepRunner } from "~/v3/snapshotSweepRunner.server"; + +const LOCK_KEY = "snapshot-sweep:lock"; +const CLEAN = { scanned: 1, expired: 0, deleted: 0, skipped: 1, partial: false }; + +function opts() { + return { deadline: Date.now() + 10_000, signal: new AbortController().signal }; +} + +redisTest("reports completed and releases the lock", async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + try { + const runner = buildSnapshotSweepRunner({ + client, + sweep: async () => CLEAN, + lockTtlMs: 60_000, + }); + + expect((await runner(opts())).outcome).toBe("completed"); + expect(await client.get(LOCK_KEY)).toBeNull(); + } finally { + await client.quit(); + } +}); + +redisTest("reports partial when the pass truncates", async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + try { + const runner = buildSnapshotSweepRunner({ + client, + sweep: async () => ({ ...CLEAN, partial: true }), + lockTtlMs: 60_000, + }); + + const result = await runner(opts()); + expect(result.outcome).toBe("partial"); + expect(result.counts).toMatchObject({ partial: true }); + } finally { + await client.quit(); + } +}); + +redisTest("skips without running when the lock is held", async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + try { + await client.set(LOCK_KEY, "someone-else", "PX", 60_000); + let ran = false; + + const runner = buildSnapshotSweepRunner({ + client, + sweep: async () => { + ran = true; + return CLEAN; + }, + lockTtlMs: 60_000, + }); + + expect((await runner(opts())).outcome).toBe("skipped_locked"); + expect(ran).toBe(false); + expect(await client.get(LOCK_KEY)).toBe("someone-else"); + } finally { + await client.quit(); + } +}); + +redisTest("reports failed and still releases its own lock", async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + try { + const runner = buildSnapshotSweepRunner({ + client, + sweep: async () => { + throw new Error("redis down mid-pass"); + }, + lockTtlMs: 60_000, + }); + + // Resolving is deliberate: the worker reschedules a cron job on acknowledge as well as on the + // dead-letter path, so a failure needs no throw to keep the chain alive. + expect((await runner(opts())).outcome).toBe("failed"); + expect(await client.get(LOCK_KEY)).toBeNull(); + } finally { + await client.quit(); + } +}); + +redisTest("an overrun pass cannot delete a successor's lock", async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + try { + const runner = buildSnapshotSweepRunner({ + client, + fence: () => "first-pass", + // Stand in for the lock expiring mid-pass and a successor claiming it. + sweep: async () => { + await client.set(LOCK_KEY, "successor", "PX", 60_000); + return CLEAN; + }, + lockTtlMs: 60_000, + }); + + await runner(opts()); + + expect(await client.get(LOCK_KEY)).toBe("successor"); + } finally { + await client.quit(); + } +}); + +redisTest("reports aborted when shutdown cancels the pass", async ({ redisOptions }) => { + const client = createRedisClient(redisOptions); + try { + const controller = new AbortController(); + const runner = buildSnapshotSweepRunner({ + client, + sweep: async () => { + controller.abort(); + return CLEAN; + }, + lockTtlMs: 60_000, + }); + + const result = await runner({ deadline: Date.now() + 10_000, signal: controller.signal }); + expect(result.outcome).toBe("aborted"); + expect(await client.get(LOCK_KEY)).toBeNull(); + } finally { + await client.quit(); + } +}); diff --git a/internal-packages/redis/package.json b/internal-packages/redis/package.json index 14cf7f33d61..eff0a779dd7 100644 --- a/internal-packages/redis/package.json +++ b/internal-packages/redis/package.json @@ -10,6 +10,8 @@ "@trigger.dev/core": "workspace:*" }, "scripts": { - "typecheck": "tsc --noEmit" + "typecheck": "tsc --noEmit", + "test": "vitest", + "test:watch": "vitest" } } diff --git a/internal-packages/redis/src/cluster.test.ts b/internal-packages/redis/src/cluster.test.ts new file mode 100644 index 00000000000..71d2989b22f --- /dev/null +++ b/internal-packages/redis/src/cluster.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from "vitest"; +import { Cluster, createRedisClusterClient, defaultReconnectOnError } from "./index.js"; + +// Port 6399 is deliberately closed: these assertions are about the client the factory builds, not +// about a connection. +const NODES = [{ host: "127.0.0.1", port: 6399 }]; + +type InnerOptions = { + options: { + redisOptions?: { + reconnectOnError?: unknown; + maxRetriesPerRequest?: number; + retryStrategy?: unknown; + keyPrefix?: string; + }; + }; +}; + +function innerOptionsOf(client: Cluster) { + return (client as unknown as InnerOptions).options.redisOptions; +} + +describe("createRedisClusterClient", () => { + it("returns a Cluster instance", async () => { + const client = createRedisClusterClient({ nodes: NODES }); + try { + expect(client).toBeInstanceOf(Cluster); + } finally { + await client.quit().catch(() => undefined); + } + }); + + it("installs defaultReconnectOnError on the inner per-node options", async () => { + const client = createRedisClusterClient({ nodes: NODES }); + try { + expect(innerOptionsOf(client)?.reconnectOnError).toBe(defaultReconnectOnError); + } finally { + await client.quit().catch(() => undefined); + } + }); + + it("carries the retry defaults onto the inner options", async () => { + const client = createRedisClusterClient({ nodes: NODES }); + try { + const inner = innerOptionsOf(client); + expect(inner?.maxRetriesPerRequest).toBeTypeOf("number"); + expect(inner?.retryStrategy).toBeTypeOf("function"); + } finally { + await client.quit().catch(() => undefined); + } + }); + + it("lets caller redisOptions override the defaults", async () => { + const client = createRedisClusterClient({ + nodes: NODES, + redisOptions: { keyPrefix: "engine:", maxRetriesPerRequest: 3 }, + }); + try { + const inner = innerOptionsOf(client); + expect(inner?.keyPrefix).toBe("engine:"); + expect(inner?.maxRetriesPerRequest).toBe(3); + } finally { + await client.quit().catch(() => undefined); + } + }); + + it("keeps mapping READONLY, LOADING and UNBLOCKED to a reconnect-and-retry", () => { + expect(defaultReconnectOnError(new Error("READONLY against a read only replica"))).toBe(2); + expect(defaultReconnectOnError(new Error("LOADING Redis is loading the dataset"))).toBe(2); + expect(defaultReconnectOnError(new Error("UNBLOCKED force unblock"))).toBe(2); + expect(defaultReconnectOnError(new Error("ERR unknown command"))).toBe(false); + }); +}); diff --git a/internal-packages/redis/src/index.ts b/internal-packages/redis/src/index.ts index 622efe613fc..3ea9adecb28 100644 --- a/internal-packages/redis/src/index.ts +++ b/internal-packages/redis/src/index.ts @@ -1,4 +1,10 @@ -import { type Cluster, Redis, type RedisOptions } from "ioredis"; +import { + Redis, + type Cluster, + type ClusterNode, + type ClusterOptions, + type RedisOptions, +} from "ioredis"; import { Logger } from "@trigger.dev/core/logger"; export { @@ -85,3 +91,44 @@ export function createRedisClient( return client; } + +export type RedisClusterClientOptions = { + nodes: ClusterNode[]; + clusterOptions?: Omit; + redisOptions?: RedisOptions; +}; + +/** + * Cluster-mode client. `defaultOptions` go on the INNER per-node options, so a role swap gets the + * same reconnect-and-retry treatment a single-node client already gets. + */ +export function createRedisClusterClient( + options: RedisClusterClientOptions, + handlers?: { onError?: (err: Error) => void } +): Cluster { + const client = new Redis.Cluster(options.nodes, { + ...options.clusterOptions, + redisOptions: { + ...defaultOptions, + ...options.redisOptions, + }, + }); + + if (process.env.VITEST) { + client.on("error", () => {}); + return client; + } + + client.on("error", (error) => { + if (handlers?.onError) { + handlers.onError(error); + } else { + logger.error(`Redis cluster client error:`, { + error, + keyPrefix: options.redisOptions?.keyPrefix, + }); + } + }); + + return client; +} diff --git a/internal-packages/redis/vitest.config.ts b/internal-packages/redis/vitest.config.ts new file mode 100644 index 00000000000..e07f05e842b --- /dev/null +++ b/internal-packages/redis/vitest.config.ts @@ -0,0 +1,10 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + include: ["**/*.test.ts"], + globals: true, + isolate: true, + testTimeout: 10_000, + }, +}); diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index ccfb60ca4d6..39ca85949cc 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -3,6 +3,7 @@ import { type Meter, type Tracer, type Counter, + type Histogram, getMeter, startSpan, trace, @@ -92,14 +93,21 @@ import { } from "./controlPlaneResolver.js"; import { TtlSystem } from "./systems/ttlSystem.js"; import { WaitpointSystem } from "./systems/waitpointSystem.js"; +import { SNAPSHOT_SWEEP_COUNT_FIELDS } from "./types.js"; import type { EngineWorker, HeartbeatTimeouts, ReportableQueue, RunEngineOptions, + SnapshotSweepOutcome, TriggerParams, } from "./types.js"; import { createTtlWorkerCatalog } from "./ttlWorkerCatalog.js"; +import { + DEFAULT_SNAPSHOT_SWEEP_BUDGET_MS, + resolveSnapshotSweepCron, + snapshotSweepVisibilityTimeoutMs, +} from "./snapshotSweepSchedule.js"; import { workerCatalog } from "./workerCatalog.js"; import pMap from "p-map"; @@ -111,6 +119,8 @@ export class RunEngine { private logger: Logger; private tracer: Tracer; private meter: Meter; + private snapshotSweepPassCounter?: Counter; + private snapshotSweepCountsHistogram?: Histogram; private snapshotsSinceReplicaMissCounter: Counter; private snapshotsSinceReplicaRetryDelay: { minMs: number; maxMs: number }; private heartbeatTimeouts: HeartbeatTimeouts; @@ -251,7 +261,25 @@ export class RunEngine { ...options.worker.redis, keyPrefix: `${options.worker.redis.keyPrefix}worker:`, }, - catalog: workerCatalog, + catalog: { + ...workerCatalog, + sweepSnapshotOrphans: { + ...workerCatalog.sweepSnapshotOrphans, + // Derived, not fixed. The runner's lock TTL is the budget plus an hour, so a hardcoded + // timeout would sit below the lock once the budget is raised, inverting the ordering the + // fence depends on. + visibilityTimeoutMs: snapshotSweepVisibilityTimeoutMs( + options.snapshotStore?.sweepBudgetMs + ), + cron: resolveSnapshotSweepCron({ + hasRunner: !!options.snapshotStore?.runSweep, + schedule: options.snapshotStore?.sweepSchedule, + fallback: workerCatalog.sweepSnapshotOrphans.cron, + }), + jitterInMs: + options.snapshotStore?.sweepJitterInMs ?? workerCatalog.sweepSnapshotOrphans.jitterInMs, + }, + }, concurrency: options.worker, pollIntervalMs: options.worker.pollIntervalMs, immediatePollIntervalMs: options.worker.immediatePollIntervalMs, @@ -275,6 +303,9 @@ export class RunEngine { repairSnapshot: async ({ payload }) => { await this.#handleRepairSnapshot(payload); }, + sweepSnapshotOrphans: async () => { + await this.#handleSweepSnapshotOrphans(); + }, expireRun: async ({ payload }) => { await this.ttlSystem.expireRun({ runId: payload.runId }); }, @@ -323,6 +354,23 @@ export class RunEngine { this.tracer = options.tracer; this.meter = options.meter ?? getMeter("run-engine"); + // Only when the sweep is actually wired: a deployment that does not use the snapshot store + // should register no series for it at all. + if (options.snapshotStore?.runSweep) { + this.snapshotSweepPassCounter = this.meter.createCounter( + "run_engine.snapshot_store.sweep_pass_total", + { + description: + "Orphan-sweep passes by outcome. A pass that throws emits outcome=failed, so silence is distinguishable from success", + } + ); + + this.snapshotSweepCountsHistogram = this.meter.createHistogram( + "run_engine.snapshot_store.sweep_counts", + { description: "Per-field counts from one orphan-sweep pass" } + ); + } + this.snapshotsSinceReplicaMissCounter = this.meter.createCounter( "run_engine.snapshots_since.replica_miss", { @@ -2465,6 +2513,23 @@ export class RunEngine { }; } + /** + * The append-failure compensator. Shares the stall watchdog's job id AND its availableAt, so the + * two cannot enqueue two repairs for one run and neither can win a race that changes the delay. + */ + async enqueueSnapshotRepair(payload: { + runId: string; + snapshotId: string; + executionStatus: string; + }): Promise { + return this.worker.enqueueOnce({ + id: `repair-in-progress-run:${payload.runId}`, + job: "repairSnapshot", + payload, + availableAt: new Date(Date.now() + this.repairSnapshotTimeoutMs), + }); + } + async #repairRun(runId: string, dryRun: boolean) { const snapshot = await getLatestExecutionSnapshot(this.prisma, runId, this.runStore); @@ -2891,6 +2956,53 @@ export class RunEngine { }); } + /** + * One emitter per pass, in a finally, so a throw is reported rather than silent. The deploy step + * gates dual-write on an observed pass, so an absent metric would read as a clean sweep. + */ + async #handleSweepSnapshotOrphans() { + const runSweep = this.options.snapshotStore?.runSweep; + + if (!runSweep) { + // The cron entry is registered when the engine is constructed, which happens before the + // webapp sets the binding, so an occurrence already queued at boot can arrive unbound. + this.snapshotSweepPassCounter?.add(1, { outcome: "unbound" }); + this.logger.error("sweepSnapshotOrphans ran with no sweep runner bound"); + return; + } + + const budgetMs = this.options.snapshotStore?.sweepBudgetMs ?? DEFAULT_SNAPSHOT_SWEEP_BUDGET_MS; + const controller = new AbortController(); + // The deadline is the sweep's own stopping rule; this is the backstop for a pass that has + // stopped reaching a batch boundary, so the signal is not merely decorative. + const abortAt = globalThis.setTimeout(() => controller.abort(), budgetMs + 60_000); + let outcome: SnapshotSweepOutcome = "failed"; + let counts: Partial> | undefined; + + try { + const result = await runSweep({ + deadline: Date.now() + budgetMs, + signal: controller.signal, + }); + outcome = result.outcome; + counts = result.counts; + } catch (error) { + // Deliberately not rethrown. Both the acknowledge path and the dead-letter path reschedule a + // cron job, so returning here continues the chain; throwing would only add a dead-letter + // entry for every transient blip. The outcome metric is the signal. + this.logger.error("sweepSnapshotOrphans threw", { error }); + } finally { + globalThis.clearTimeout(abortAt); + this.snapshotSweepPassCounter?.add(1, { outcome }); + for (const [field, value] of Object.entries(counts ?? {})) { + // Each field is a metric attribute, so an unrecognised key would mint a time series. + if (typeof value === "number" && SNAPSHOT_SWEEP_COUNT_FIELDS.includes(field as never)) { + this.snapshotSweepCountsHistogram?.record(value, { field }); + } + } + } + } + async #handleRepairSnapshot({ runId, snapshotId, diff --git a/internal-packages/run-engine/src/engine/snapshotSweepSchedule.test.ts b/internal-packages/run-engine/src/engine/snapshotSweepSchedule.test.ts new file mode 100644 index 00000000000..c20f6ea5492 --- /dev/null +++ b/internal-packages/run-engine/src/engine/snapshotSweepSchedule.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from "vitest"; +import { + DEFAULT_SNAPSHOT_SWEEP_BUDGET_MS, + resolveSnapshotSweepCron, + snapshotSweepVisibilityTimeoutMs, +} from "./snapshotSweepSchedule.js"; + +const FALLBACK = "0 */6 * * *"; + +describe("resolveSnapshotSweepCron", () => { + it("never schedules without a runner", () => { + expect(resolveSnapshotSweepCron({ hasRunner: false, fallback: FALLBACK })).toBeUndefined(); + expect( + resolveSnapshotSweepCron({ hasRunner: false, schedule: "* * * * *", fallback: FALLBACK }) + ).toBeUndefined(); + }); + + it("uses the fallback when no schedule is supplied", () => { + expect(resolveSnapshotSweepCron({ hasRunner: true, fallback: FALLBACK })).toBe(FALLBACK); + }); + + it("uses the supplied schedule", () => { + expect( + resolveSnapshotSweepCron({ hasRunner: true, schedule: "0 */12 * * *", fallback: FALLBACK }) + ).toBe("0 */12 * * *"); + }); + + it("does not let an empty schedule silently disable the job", () => { + expect(resolveSnapshotSweepCron({ hasRunner: true, schedule: "", fallback: FALLBACK })).toBe( + FALLBACK + ); + expect(resolveSnapshotSweepCron({ hasRunner: true, schedule: " ", fallback: FALLBACK })).toBe( + FALLBACK + ); + }); +}); + +describe("the unconfigured deployment", () => { + // The webapp must omit the whole options block, not pass a runner that reports unbound: a + // registered cron would log an unbound pass every interval on every install not using the store. + it("schedules nothing when the options block is absent", () => { + expect( + resolveSnapshotSweepCron({ hasRunner: false, schedule: "0 */6 * * *", fallback: FALLBACK }) + ).toBeUndefined(); + }); +}); + +describe("snapshotSweepVisibilityTimeoutMs", () => { + // The runner's lock TTL is the budget plus an hour. The delivery window has to stay above it, or + // a redelivery arrives while the previous pass still holds the fence. + const lockTtl = (budget: number) => budget + 60 * 60 * 1000; + + it("stays above the lock TTL at the default budget", () => { + expect(snapshotSweepVisibilityTimeoutMs()).toBeGreaterThan( + lockTtl(DEFAULT_SNAPSHOT_SWEEP_BUDGET_MS) + ); + }); + + it("stays above the lock TTL for a raised budget", () => { + for (const budget of [60_000, 10_800_000, 43_200_000, 86_400_000]) { + expect(snapshotSweepVisibilityTimeoutMs(budget)).toBeGreaterThan(lockTtl(budget)); + } + }); +}); diff --git a/internal-packages/run-engine/src/engine/snapshotSweepSchedule.ts b/internal-packages/run-engine/src/engine/snapshotSweepSchedule.ts new file mode 100644 index 00000000000..20d967650c7 --- /dev/null +++ b/internal-packages/run-engine/src/engine/snapshotSweepSchedule.ts @@ -0,0 +1,26 @@ +/** + * `undefined` never schedules the job. An empty schedule falls back to the default rather than + * through: an empty string is falsy, so `setupCron` would filter the job out with nothing logged. + */ +export function resolveSnapshotSweepCron(opts: { + hasRunner: boolean; + schedule?: string; + fallback: string; +}): string | undefined { + if (!opts.hasRunner) { + return undefined; + } + return opts.schedule?.trim() ? opts.schedule : opts.fallback; +} + +/** Default budget for one sweep pass, when the caller supplies none. */ +export const DEFAULT_SNAPSHOT_SWEEP_BUDGET_MS = 10_800_000; + +/** + * Keeps the delivery window strictly above the runner's lock TTL, which is the budget plus an hour. + * Two hours of headroom, so a pass that overruns its budget still holds a lock that outlives the + * delivery it belongs to. + */ +export function snapshotSweepVisibilityTimeoutMs(budgetMs?: number): number { + return (budgetMs ?? DEFAULT_SNAPSHOT_SWEEP_BUDGET_MS) + 2 * 60 * 60 * 1000; +} diff --git a/internal-packages/run-engine/src/engine/tests/helpers/snapshotRepairEngine.ts b/internal-packages/run-engine/src/engine/tests/helpers/snapshotRepairEngine.ts new file mode 100644 index 00000000000..3924a5ef4ae --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/helpers/snapshotRepairEngine.ts @@ -0,0 +1,24 @@ +// A minimal engine for asserting the append-failure repair binding. No decorator and no Redis +// snapshot store: the property under test is the job id's dedupe, which lives on the engine. +import { trace } from "@internal/tracing"; + +export function engineOptionsForSnapshotRepair(prisma: unknown, redisOptions: unknown) { + return { + prisma, + worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 }, + queue: { + redis: redisOptions, + masterQueueConsumersDisabled: true, + processWorkerQueueDebounceMs: 50, + }, + runLock: { redis: redisOptions }, + machines: { + defaultMachine: "small-1x" as const, + machines: { + "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 }, + }, + baseCostInCents: 0.0001, + }, + tracer: trace.getTracer("test", "0.0.0"), + }; +} diff --git a/internal-packages/run-engine/src/engine/tests/snapshotRepairBinding.test.ts b/internal-packages/run-engine/src/engine/tests/snapshotRepairBinding.test.ts new file mode 100644 index 00000000000..ba14c58034e --- /dev/null +++ b/internal-packages/run-engine/src/engine/tests/snapshotRepairBinding.test.ts @@ -0,0 +1,48 @@ +import { containerTest } from "@internal/testcontainers"; +import { expect } from "vitest"; +import { RunEngine } from "../index.js"; +import { engineOptionsForSnapshotRepair } from "./helpers/snapshotRepairEngine.js"; + +containerTest( + "a second repair for the same run enqueues no second job", + async ({ prisma, redisOptions }) => { + const engine = new RunEngine(engineOptionsForSnapshotRepair(prisma, redisOptions) as never); + + try { + const payload = { + runId: "run_repair_dedupe", + snapshotId: "snap_1", + executionStatus: "EXECUTING", + }; + + // The stall watchdog uses this same job id, so the two compensators must collapse to one. + expect(await engine.enqueueSnapshotRepair(payload)).toBe(true); + expect(await engine.enqueueSnapshotRepair({ ...payload, snapshotId: "snap_2" })).toBe(false); + } finally { + await engine.quit(); + } + } +); + +containerTest("a different run gets its own repair job", async ({ prisma, redisOptions }) => { + const engine = new RunEngine(engineOptionsForSnapshotRepair(prisma, redisOptions) as never); + + try { + expect( + await engine.enqueueSnapshotRepair({ + runId: "run_a", + snapshotId: "snap_a", + executionStatus: "EXECUTING", + }) + ).toBe(true); + expect( + await engine.enqueueSnapshotRepair({ + runId: "run_b", + snapshotId: "snap_b", + executionStatus: "EXECUTING", + }) + ).toBe(true); + } finally { + await engine.quit(); + } +}); diff --git a/internal-packages/run-engine/src/engine/types.ts b/internal-packages/run-engine/src/engine/types.ts index 2516776373d..cb47f0f9f11 100644 --- a/internal-packages/run-engine/src/engine/types.ts +++ b/internal-packages/run-engine/src/engine/types.ts @@ -50,6 +50,27 @@ export type CrossSeamGuardHook = (input: { routeKind: "MANUAL" | "DATETIME" | "RESUME_TOKEN" | "IDEMPOTENCY_REUSE" | "RUN"; }) => Promise; +export type SnapshotSweepOutcome = + | "completed" + | "partial" + | "skipped_locked" + | "failed" + | "unbound" + | "aborted"; + +export const SNAPSHOT_SWEEP_COUNT_FIELDS = [ + "scanned", + "expired", + "deleted", + "skipped", + "pendingDeletion", + "nodes", + "partial", +] as const; + +/** Derived from the list above, so the runtime filter and the type cannot drift apart. */ +type SnapshotSweepCountField = (typeof SNAPSHOT_SWEEP_COUNT_FIELDS)[number]; + export type RunEngineOptions = { prisma: PrismaClient; readOnlyPrisma?: PrismaReplicaClient; @@ -166,6 +187,22 @@ export type RunEngineOptions = { randomize?: boolean; }; }; + /** + * The execution-snapshot orphan sweep. The engine owns scheduling only: the webapp owns what + * runs, because a pass needs a run store and its own Redis client and the engine opens neither. + */ + snapshotStore?: { + /** Bounded on purpose: both fields become metric attributes, so each value is a time series. */ + runSweep?: (opts: { deadline: number; signal: AbortSignal }) => Promise<{ + outcome: SnapshotSweepOutcome; + counts?: Partial>; + }>; + /** Cron. Absent or empty falls back to the catalog default. */ + sweepSchedule?: string; + sweepJitterInMs?: number; + /** Ceiling on one pass. Must stay below the job's visibility timeout. */ + sweepBudgetMs?: number; + }; debounce?: { redis?: RedisOptions; /** diff --git a/internal-packages/run-engine/src/engine/workerCatalog.ts b/internal-packages/run-engine/src/engine/workerCatalog.ts index 5def3071b9e..d7742255636 100644 --- a/internal-packages/run-engine/src/engine/workerCatalog.ts +++ b/internal-packages/run-engine/src/engine/workerCatalog.ts @@ -83,4 +83,19 @@ export const workerCatalog = { }), visibilityTimeoutMs: 30_000, }, + sweepSnapshotOrphans: { + schema: z.object({ + timestamp: z.number(), + lastTimestamp: z.number().optional(), + cron: z.string(), + }), + // The default budget plus two hours, so it stays strictly above the runner's lock TTL + // (budget plus one hour). Ordered that way, a lock outlives the delivery it belongs to. + visibilityTimeoutMs: 18_000_000, + cron: "0 */6 * * *", + jitterInMs: 60_000, + // Load-bearing. A throw takes the dead-letter path, which also reschedules, so the cron chain + // survives a failed pass. With retries it would not behave that way. + retry: { maxAttempts: 1 }, + }, }; diff --git a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.modeResolver.test.ts b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.modeResolver.test.ts new file mode 100644 index 00000000000..b1f425e929e --- /dev/null +++ b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.modeResolver.test.ts @@ -0,0 +1,72 @@ +import { describe, expect, it } from "vitest"; +import { + TaskRunExecutionSnapshotStore, + type SnapshotStoreMode, + type SnapshotStoreModeResolver, +} from "./taskRunExecutionSnapshotStore.js"; + +function storeWith(options: { + mode?: SnapshotStoreMode; + modeResolver?: SnapshotStoreModeResolver; +}) { + return new TaskRunExecutionSnapshotStore({} as never, { + store: {} as never, + ...options, + }); +} + +function resolverOf(perOrg: Record, global: SnapshotStoreMode) { + return { resolve: (orgId?: string) => (orgId ? (perOrg[orgId] ?? global) : global) }; +} + +describe("TaskRunExecutionSnapshotStore mode resolution", () => { + it("prefers the resolver over the static mode", () => { + const store = storeWith({ mode: "off", modeResolver: resolverOf({}, "dual-write") }); + expect(store.mode).toBe("dual-write"); + }); + + it("falls back to the static mode when no resolver is supplied", () => { + expect(storeWith({ mode: "redis-read" }).mode).toBe("redis-read"); + }); + + it("defaults to off with neither", () => { + expect(storeWith({}).mode).toBe("off"); + }); + + it("resolves per organisation", () => { + const store = storeWith({ modeResolver: resolverOf({ org_a: "dual-write" }, "off") }); + expect(store.writesRedisForTest("org_a")).toBe(true); + expect(store.writesRedisForTest("org_b")).toBe(false); + }); + + it("lets an organisation be off while the global answer is on", () => { + const store = storeWith({ modeResolver: resolverOf({ org_a: "off" }, "dual-write") }); + expect(store.writesRedisForTest("org_a")).toBe(false); + expect(store.writesRedisForTest("org_b")).toBe(true); + }); + + it("sees a resolver answer that changes after construction", () => { + let current: SnapshotStoreMode = "off"; + const store = storeWith({ modeResolver: { resolve: () => current } }); + expect(store.mode).toBe("off"); + current = "dual-write"; + expect(store.mode).toBe("dual-write"); + }); + + it("resolves the global answer when no organisation is supplied", () => { + const store = storeWith({ modeResolver: resolverOf({ org_a: "off" }, "redis-read") }); + expect(store.writesRedisForTest()).toBe(true); + }); + + it("resolves the fatal-birth decision per organisation, not globally", () => { + // A lost birth append is fatal only where Postgres holds nothing. An organisation still on a + // dual-write position must not have its run creation failed by the global position. + const store = storeWith({ + modeResolver: resolverOf({ org_dual: "dual-write" }, "redis-only"), + }); + + expect(store.modeForTest("org_dual")).toBe("dual-write"); + expect(store.modeForTest("org_other")).toBe("redis-only"); + expect(store.modeForTest()).toBe("redis-only"); + }); +}); diff --git a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.off.test.ts b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.off.test.ts index 437872106e7..aa6e95beeeb 100644 --- a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.off.test.ts +++ b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.off.test.ts @@ -53,15 +53,31 @@ describe("TaskRunExecutionSnapshotStore at mode off", () => { mode: "off", }) as unknown as Record unknown>; + // Both exceptions return a wrapped handle rather than the delegate's value verbatim, because + // the dial can move at runtime and an unwrapped handle would let a later write bypass the + // decorator with no signal. Every other method is still a pass-through, and the exploding + // Redis store is what proves none of them reaches Redis. + const WRAPPED = ["runInTransaction", "forWaitpointCompletion"]; + for (const name of RUN_STORE_METHOD_NAMES) { - if (name === "runInTransaction") continue; + if (WRAPPED.includes(name)) continue; expect(await decorated[name]("arg-one", "arg-two")).toBe(`result:${name}`); } - expect(calls).toEqual(RUN_STORE_METHOD_NAMES.filter((n) => n !== "runInTransaction")); + const handle = await decorated.forWaitpointCompletion("waitpoint", {}); + expect(handle).toBeInstanceOf(TaskRunExecutionSnapshotStore); + expect((handle as unknown as { delegate: unknown }).delegate).toBe( + "result:forWaitpointCompletion" + ); + + // forWaitpointCompletion is called after the loop, so it lands last rather than in place. + expect(calls).toEqual([ + ...RUN_STORE_METHOD_NAMES.filter((n) => !WRAPPED.includes(n)), + "forWaitpointCompletion", + ]); }); - it("hands the delegate's own store to a transaction callback", async () => { + it("wraps the transaction callback's store but never reaches Redis", async () => { const inner = forwardingProbe().store; let seen: unknown; const delegate = { @@ -82,7 +98,11 @@ describe("TaskRunExecutionSnapshotStore at mode off", () => { seen = store; }); - expect(seen).toBe(inner); + // The facade is always installed: this method holds only a runId, so it cannot know whether a + // per-organisation dial would put any write inside on Redis. The exploding Redis store is what + // proves the position still costs nothing at `off`. + expect(seen).toBeInstanceOf(TaskRunExecutionSnapshotStore); + expect(seen).not.toBe(inner); }); it("reports every other dial position as one that writes Redis", () => { diff --git a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.staging.test.ts b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.staging.test.ts index 9952f569e2b..0c86aad7d14 100644 --- a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.staging.test.ts +++ b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.staging.test.ts @@ -216,7 +216,7 @@ describe("the staging facade", () => { ); containerTest( - "hands the transaction callback the plain delegate at mode off", + "stages nothing and touches no Redis inside a transaction at mode off", async ({ prisma, redisOptions }) => { const { decorated, redis } = build(prisma as never, redisOptions as never, "off"); try { @@ -242,7 +242,10 @@ describe("the staging facade", () => { seen = store; }); - expect(seen).not.toBeInstanceOf(TaskRunExecutionSnapshotStore); + // The callback now always receives the staging facade: this method holds only a runId, and + // a per-organisation dial lives in that organisation's blob, so it cannot know whether any + // write inside will reach Redis. What must hold at `off` is that nothing is appended. + expect(seen).toBeInstanceOf(TaskRunExecutionSnapshotStore); expect(await redis.getLatest(runId)).toBeNull(); } finally { await redis.quit(); @@ -269,7 +272,7 @@ describe("the staging facade", () => { ); containerTest( - "returns the plain handle from forWaitpointCompletion at mode off", + "wraps the forWaitpointCompletion handle at mode off too", async ({ prisma, redisOptions }) => { const { decorated, redis } = build(prisma as never, redisOptions as never, "off"); try { @@ -277,7 +280,9 @@ describe("the staging facade", () => { routeKind: "MANUAL", } as never); - expect(handle).not.toBeInstanceOf(TaskRunExecutionSnapshotStore); + // Leaving it unwrapped was the one hole a future snapshot write could slip through with no + // signal, and the dial can now move at runtime, so the handle is wrapped at every position. + expect(handle).toBeInstanceOf(TaskRunExecutionSnapshotStore); } finally { await redis.quit(); } diff --git a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts index c6bc145a91d..302401449da 100644 --- a/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts +++ b/internal-packages/run-store/src/taskRunExecutionSnapshotStore.ts @@ -68,6 +68,15 @@ const WAITPOINT_CHUNK_SIZE = 100; */ export type SnapshotStoreMode = "off" | "dual-write" | "redis-read" | "redis-only"; +/** + * Resolves the dial for one write. MUST be synchronous and MUST NOT query: seven methods here take + * a caller-supplied `tx`, so this class cannot see a caller's transaction boundary, and a read + * issued here could land inside someone else's open interactive transaction. + */ +export type SnapshotStoreModeResolver = { + resolve(organizationId?: string): SnapshotStoreMode; +}; + /** * Enqueues the existing `repairSnapshot` job for a run whose append was lost. The decorator lives in * run-store and cannot reach the engine's worker, so the binding is injected. That binding must @@ -90,6 +99,8 @@ export type TaskRunExecutionSnapshotStoreOptions = { store: RedisSnapshotStore; /** Defaults to `off`, which is a pure pass-through that never touches Redis. */ mode?: SnapshotStoreMode; + /** Takes precedence over `mode`, and is re-read on every write so the dial can move at runtime. */ + modeResolver?: SnapshotStoreModeResolver; /** Percentage of runs whose reads come from Redis at `redis-read` and `redis-only`. Defaults to 0. */ readPercent?: number; onAppendFailure?: SnapshotRepairEnqueuer; @@ -117,7 +128,8 @@ export type StagedAppend = { }; export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { - readonly mode: SnapshotStoreMode; + readonly #staticMode: SnapshotStoreMode; + protected readonly modeResolver?: SnapshotStoreModeResolver; protected readonly redis: RedisSnapshotStore; protected readonly readPercent: number; protected readonly onAppendFailure?: SnapshotRepairEnqueuer; @@ -129,7 +141,8 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { constructor(delegate: RunStore, options: TaskRunExecutionSnapshotStoreOptions) { super(delegate); this.redis = options.store; - this.mode = options.mode ?? "off"; + this.#staticMode = options.mode ?? "off"; + this.modeResolver = options.modeResolver; this.readPercent = options.readPercent ?? 0; this.onAppendFailure = options.onAppendFailure; this.faults = options.faults; @@ -138,9 +151,28 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { this.staging = options.staging; } - /** True in every position that appends to Redis. */ - protected get writesRedis(): boolean { - return this.mode !== "off"; + get mode(): SnapshotStoreMode { + return this.modeResolver?.resolve() ?? this.#staticMode; + } + + /** The resolved position for one organisation. Falls back to the global answer when unknown. */ + protected modeFor(organizationId?: string): SnapshotStoreMode { + return this.modeResolver?.resolve(organizationId) ?? this.#staticMode; + } + + /** True in every position that appends to Redis, for this entry's organisation. */ + protected writesRedisFor(organizationId?: string): boolean { + return this.modeFor(organizationId) !== "off"; + } + + /** Test seam for the per-organisation predicate. Not for production callers. */ + writesRedisForTest(organizationId?: string): boolean { + return this.writesRedisFor(organizationId); + } + + /** Test seam for the resolved position. Not for production callers. */ + modeForTest(organizationId?: string): SnapshotStoreMode { + return this.modeFor(organizationId); } /** @@ -159,12 +191,9 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { runId: string | undefined, fn: (store: RunStore, tx: PrismaClientOrTransaction) => Promise ): Promise { - if (!this.writesRedis) { - // At `off` the callback must receive the delegate's own store, untouched, so a transaction - // behaves exactly as it does without the decorator in the chain. - return this.delegate.runInTransaction(runId, fn); - } - + // Always stage. This method holds only a runId, and a per-organisation dial lives in that + // organisation's blob, so "can any write in here reach Redis" cannot be answered here. Each + // entry resolves its own mode as it is staged, and an organisation at `off` stages nothing. const staged: StagedAppend[] = []; const result = await this.delegate.runInTransaction(runId, (store, tx) => @@ -195,10 +224,6 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { ): Promise { const store = await this.delegate.forWaitpointCompletion(waitpointId, context); - if (!this.writesRedis) { - return store; - } - // Carry the staging buffer through. Without it, a handle taken inside a transaction appends // immediately, which is the exact ordering the facade exists to prevent. return this.#wrap(store, this.staging); @@ -212,8 +237,9 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { #wrap(store: RunStore, staging?: StagedAppend[]): TaskRunExecutionSnapshotStore { return new TaskRunExecutionSnapshotStore(store, { store: this.redis, - mode: this.mode, + mode: this.#staticMode, readPercent: this.readPercent, + ...(this.modeResolver && { modeResolver: this.modeResolver }), logger: this.logger, ...(this.onAppendFailure && { onAppendFailure: this.onAppendFailure }), ...(this.faults && { faults: this.faults }), @@ -230,7 +256,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { params: CreateRunInput, tx?: PrismaClientOrTransaction ): Promise { - if (!this.writesRedis) { + if (!this.writesRedisFor(params.snapshot?.organizationId)) { return this.delegate.createRun(params, tx); } @@ -246,7 +272,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { params: CreateCancelledRunInput, tx?: PrismaClientOrTransaction ): Promise { - if (!this.writesRedis) { + if (!this.writesRedisFor(params.snapshot?.organizationId)) { return this.delegate.createCancelledRun(params, tx); } @@ -275,7 +301,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { args: { select: S }, tx?: PrismaClientOrTransaction ): Promise> { - if (!this.writesRedis) { + if (!this.writesRedisFor(data.snapshot?.organizationId)) { return this.delegate.completeAttemptSuccess(runId, data, args, tx); } @@ -300,7 +326,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { args: { select: S }, tx?: PrismaClientOrTransaction ): Promise> { - if (!this.writesRedis) { + if (!this.writesRedisFor(data.snapshot?.organizationId)) { return this.delegate.expireRun(runId, data as never, args, tx); } @@ -327,7 +353,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { }, tx?: PrismaClientOrTransaction ): Promise<{ count: number }> { - if (!this.writesRedis) { + if (!this.writesRedisFor(data.snapshot?.organizationId)) { return this.delegate.expireParkedRun(runId, data as never, tx); } @@ -353,7 +379,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { ): Promise { // The delegate writes a snapshot only when one is supplied, so an absent snapshot is a plain run // update with nothing for Redis to mirror. - if (!this.writesRedis || !data.snapshot) { + if (!data.snapshot || !this.writesRedisFor(data.snapshot?.organizationId)) { return this.delegate.rescheduleRun(runId, data, tx); } @@ -374,7 +400,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { data: LockRunData, tx?: PrismaClientOrTransaction ): Promise>> { - if (!this.writesRedis) { + if (!this.writesRedisFor(data.snapshot?.organizationId)) { return this.delegate.lockRunToWorker(runId, data, tx); } @@ -404,7 +430,7 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { input: CreateExecutionSnapshotInput, tx?: PrismaClientOrTransaction ): Promise> { - if (!this.writesRedis) { + if (!this.writesRedisFor(input?.organizationId)) { return this.delegate.createExecutionSnapshot(input, tx); } @@ -483,7 +509,10 @@ export class TaskRunExecutionSnapshotStore extends DelegatingRunStore { error, }); - if (this.mode === "redis-only") { + // The same organisation dial that decided to append decides whether a lost birth is + // fatal. Using the global position here would fail run creation for an organisation whose + // own position still has Postgres authoritative. + if (this.modeFor(entry.organizationId) === "redis-only") { throw error; } return;