diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index c63e0aa2d4f..3b697eb22df 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -28,6 +28,7 @@ import { singleton } from "./utils/singleton"; import { registerDatabaseMetricsSource } from "./utils/databaseMetrics.server"; import { isSplitEnabled, + assertShardsRequireSplit, assertSplitRealtimeInterlock, } from "./v3/runOpsMigration/splitMode.server"; import { computeRunOpsSplitReadEnabled } from "./v3/runOpsMigration/runOpsSplitReadGate"; @@ -617,6 +618,12 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({ // interlock). Async, so it cannot live in the synchronous singleton factory — called // fire-and-forget from the eager-boot path (routing is wired synchronously at module load). export async function assertRunOpsSplitSentinel(): Promise { + // Shard interlock first: shard clients are only built on the split-on arm, so this case has to be + // checked BEFORE the split-off early return below, which would otherwise skip it in silence. + assertShardsRequireSplit({ + splitFlagEnabled: env.RUN_OPS_SPLIT_ENABLED, + shards: env.RUN_OPS_SHARDS, + }); if (!env.RUN_OPS_SPLIT_ENABLED) return; // Realtime interlock (synchronous): Electric replicates only from the control-plane // DB, so split-on without the native realtime backend leaves NEW-resident runs diff --git a/apps/webapp/app/presenters/v3/BatchListPresenter.server.ts b/apps/webapp/app/presenters/v3/BatchListPresenter.server.ts index 5440b602877..0d186279175 100644 --- a/apps/webapp/app/presenters/v3/BatchListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/BatchListPresenter.server.ts @@ -75,6 +75,7 @@ export class BatchListPresenter extends BasePresenter { runOpsNew?: RunOpsPrismaClient; // new run-ops client (run-ops brand ⇒ guard classifies as runops) runOpsLegacyReplica?: RunOpsPrismaClient; // legacy run-ops READ REPLICA only — never the legacy primary controlPlaneReplica?: PrismaClientOrTransaction; // control-plane DB (for project) + shardReplicas?: ReadonlyArray<{ key: string; replica: RunOpsPrismaClient }>; splitEnabled?: boolean; // resolved boot constant } ) { @@ -110,9 +111,11 @@ export class BatchListPresenter extends BasePresenter { // unsound across the residency split: legacy cuid ids ("c…") sort ABOVE new run-ops ids ("0…") // under id order, so a new-only page can hide pre-flip legacy batches that belong ahead of it. // Ordering is by createdAt (id tiebreak), which is chronologically correct across both schemes. - const [newRows, legacyRows] = await Promise.all([ + const shardReplicas = this.readRoute.shardReplicas ?? []; + const [newRows, legacyRows, ...shardRows] = await Promise.all([ scan(this.readRoute.runOpsNew ?? passthrough), scan(this.readRoute.runOpsLegacyReplica ?? passthrough), + ...shardReplicas.map((shard) => scan(shard.replica)), ]); // De-dupe by id (new wins), re-sort under the page's keyset order, re-apply the over-fetch LIMIT. @@ -125,6 +128,11 @@ export class BatchListPresenter extends BasePresenter { byId.set(row.id, row); } } + for (const rows of shardRows) { + for (const row of rows) { + byId.set(row.id, row); + } + } // forward => newest-first (createdAt DESC), backward => oldest-first (ASC); id is the stable // tiebreak (ASCII codepoint, NEVER localeCompare). @@ -167,7 +175,23 @@ export class BatchListPresenter extends BasePresenter { ).batchTaskRun.findFirst({ where: { runtimeEnvironmentId: environmentId }, }); - return Boolean(onLegacy); + if (onLegacy) { + return true; + } + + const shardReplicas = this.readRoute.shardReplicas ?? []; + if (shardReplicas.length === 0) { + return false; + } + + const onShards = await Promise.all( + shardReplicas.map((shard) => + shard.replica.batchTaskRun.findFirst({ + where: { runtimeEnvironmentId: environmentId }, + }) + ) + ); + return onShards.some(Boolean); } public async call({ diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.batches/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.batches/route.tsx index ab139d8d312..5a8574c799a 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.batches/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.batches/route.tsx @@ -56,6 +56,7 @@ import { runOpsSplitReadEnabled, type PrismaClientOrTransaction, } from "~/db.server"; +import { runOpsNonAliasedShardReplicas } from "~/v3/runOpsMigration/shardHandles.server"; import { docsPath, EnvironmentParamSchema, @@ -104,6 +105,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { runOpsNew: runOpsNewReplicaClient, runOpsLegacyReplica: runOpsLegacyReplicaClient, controlPlaneReplica: $replica as unknown as PrismaClientOrTransaction, + shardReplicas: runOpsNonAliasedShardReplicas, splitEnabled: runOpsSplitReadEnabled, }); const list = await presenter.call({ diff --git a/apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts b/apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts index ed7fb0cb237..b8349eba269 100644 --- a/apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/distinctDbSentinel.server.ts @@ -64,6 +64,48 @@ export async function probeControlPlaneCoresidency( export type DistinctTarget = { id: string; url: string }; +/** Injection seam for the retry tests: no containers, no real waiting. */ +export type DistinctProbeOptions = { + logger?: { warn: (msg: string, meta?: Record) => void }; + readFingerprint?: (url: string) => Promise; + /** Total attempts per target, including the first. Bounded so boot latency stays bounded. */ + attempts?: number; + sleep?: (ms: number) => Promise; +}; + +const DEFAULT_PROBE_ATTEMPTS = 3; +const RETRY_BASE_DELAY_MS = 250; + +const defaultSleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * Read one fingerprint, retrying a bounded number of times. + * + * The probe fails CLOSED, and that must not change: "distinct" is a positive claim a failed probe + * cannot support. But failing closed on the first blip means one shard being briefly unreachable + * collapses the deployment to single-DB, and the boot interlock then refuses the boot for the whole + * fleet. A transient error deserves a retry; a persistent one still fails closed, just later. + */ +async function readFingerprintWithRetry( + url: string, + read: (url: string) => Promise, + attempts: number, + sleep: (ms: number) => Promise +): Promise { + let lastError: unknown; + for (let attempt = 1; attempt <= attempts; attempt++) { + try { + return await read(url); + } catch (error) { + lastError = error; + if (attempt < attempts) { + await sleep(RETRY_BASE_DELAY_MS * attempt); + } + } + } + throw lastError; +} + /** * Set uniqueness over every store that owns its own database. Fail-closed: a probe that cannot * answer returns NOT distinct, because "distinct" is a positive claim a failed probe cannot support. @@ -77,14 +119,22 @@ export type DistinctTarget = { id: string; url: string }; */ export async function probeDistinctStores( targets: DistinctTarget[], - opts?: { logger?: { warn: (msg: string, meta?: Record) => void } } + opts?: DistinctProbeOptions ): Promise<{ distinct: true } | { distinct: false; reason: string }> { if (targets.length < 2) { return { distinct: true }; } + const read = opts?.readFingerprint ?? readDatabaseFingerprint; + const attempts = opts?.attempts ?? DEFAULT_PROBE_ATTEMPTS; + const sleep = opts?.sleep ?? defaultSleep; + try { - const fingerprints = await Promise.all(targets.map((t) => readDatabaseFingerprint(t.url))); + // Retry per TARGET, not around the whole set: one slow shard must not re-probe the stores that + // already answered. A duplicate verdict below is final and is never retried. + const fingerprints = await Promise.all( + targets.map((t) => readFingerprintWithRetry(t.url, read, attempts, sleep)) + ); const seen = new Map(); for (const [index, target] of targets.entries()) { @@ -104,7 +154,9 @@ export async function probeDistinctStores( return { distinct: true }; } catch (error) { - const reason = `distinct-db sentinel probe failed; failing closed (single-DB). ${String(error)}`; + const reason = + `distinct-db sentinel probe failed after ${opts?.attempts ?? DEFAULT_PROBE_ATTEMPTS} ` + + `attempt(s); failing closed (single-DB). ${String(error)}`; opts?.logger?.warn(reason, { error }); return { distinct: false, reason }; } @@ -115,7 +167,7 @@ export async function probeDistinctStores( export async function probeDistinctDatabases( legacyUrl: string, newUrl: string, - opts?: { logger?: { warn: (msg: string, meta?: Record) => void } } + opts?: DistinctProbeOptions ): Promise<{ distinct: true } | { distinct: false; reason: string }> { return probeDistinctStores( [ diff --git a/apps/webapp/app/v3/runOpsMigration/shardHandles.server.test.ts b/apps/webapp/app/v3/runOpsMigration/shardHandles.server.test.ts index b90c1dfe7c4..170b08b98ec 100644 --- a/apps/webapp/app/v3/runOpsMigration/shardHandles.server.test.ts +++ b/apps/webapp/app/v3/runOpsMigration/shardHandles.server.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { buildShardHandleMaps } from "./shardHandles.server"; +import { buildShardHandleMaps, nonAliasedShardReplicas } from "./shardHandles.server"; // Two distinct sentinels per shard: the maps must not cross writer and replica. function handle(key: string) { @@ -35,3 +35,26 @@ describe("buildShardHandleMaps", () => { expect(replicas.get("a")).not.toEqual({ tag: "a-writer" }); }); }); + +describe("nonAliasedShardReplicas", () => { + it("yields an empty list when no shard is configured", () => { + expect(nonAliasedShardReplicas([])).toEqual([]); + }); + + it("keeps the configured order and carries each shard's replica", () => { + expect(nonAliasedShardReplicas([handle("b"), handle("a")])).toEqual([ + { key: "b", replica: { tag: "b-replica" } }, + { key: "a", replica: { tag: "a-replica" } }, + ]); + }); + + it("drops a shard that declares aliasOf", () => { + expect(nonAliasedShardReplicas([{ ...handle("a"), aliasOf: "new" }, handle("b")])).toEqual([ + { key: "b", replica: { tag: "b-replica" } }, + ]); + }); + + it("never carries a writer in place of a replica", () => { + expect(nonAliasedShardReplicas([handle("a")])[0]?.replica).not.toEqual({ tag: "a-writer" }); + }); +}); diff --git a/apps/webapp/app/v3/runOpsMigration/shardHandles.server.ts b/apps/webapp/app/v3/runOpsMigration/shardHandles.server.ts index cbe827be4dc..89c673663cc 100644 --- a/apps/webapp/app/v3/runOpsMigration/shardHandles.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/shardHandles.server.ts @@ -5,14 +5,16 @@ * is what keeps every gen-2 arm unreachable today. */ import type { PrismaClient } from "@trigger.dev/database"; +import type { RunOpsPrismaClient } from "@internal/run-ops-database"; import type { ShardKey } from "@trigger.dev/core/v3/isomorphic"; import type { PrismaReplicaClient } from "~/db.server"; import { runOpsShardHandles } from "~/db.server"; type ShardHandle = { key: string; - writer: unknown; - replica: unknown; + writer: RunOpsPrismaClient; + replica: RunOpsPrismaClient; + aliasOf?: string; }; export function buildShardHandleMaps(handles: ShardHandle[]): { @@ -22,8 +24,8 @@ export function buildShardHandleMaps(handles: ShardHandle[]): { const replicas = new Map(); const writers = new Map(); for (const handle of handles) { - replicas.set(handle.key, handle.replica as PrismaReplicaClient); - writers.set(handle.key, handle.writer as PrismaClient); + replicas.set(handle.key, handle.replica as unknown as PrismaReplicaClient); + writers.set(handle.key, handle.writer as unknown as PrismaClient); } return { replicas, writers }; } @@ -40,7 +42,17 @@ function resolveShardHandles(): ShardHandle[] { } } -const maps = buildShardHandleMaps(resolveShardHandles()); +export function nonAliasedShardReplicas( + handles: ReadonlyArray<{ key: string; replica: TClient; aliasOf?: string }> +): ReadonlyArray<{ key: string; replica: TClient }> { + return handles + .filter((handle) => handle.aliasOf === undefined) + .map((handle) => ({ key: handle.key, replica: handle.replica })); +} + +const handles = resolveShardHandles(); +const maps = buildShardHandleMaps(handles); export const runOpsShardReplicas = maps.replicas; export const runOpsShardWriters = maps.writers; +export const runOpsNonAliasedShardReplicas = nonAliasedShardReplicas(handles); diff --git a/apps/webapp/app/v3/runOpsMigration/splitMode.server.ts b/apps/webapp/app/v3/runOpsMigration/splitMode.server.ts index b9a4e3dfdf2..b62942cb87d 100644 --- a/apps/webapp/app/v3/runOpsMigration/splitMode.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/splitMode.server.ts @@ -7,7 +7,11 @@ import { env } from "~/env.server"; import { logger } from "~/services/logger.server"; import { probeDistinctStores as defaultProbe } from "./distinctDbSentinel.server"; -import { nonAliasedShards, type ShardTarget } from "~/v3/runOpsShards.server"; +import { + nonAliasedShards, + type RunOpsShardDescriptor, + type ShardTarget, +} from "~/v3/runOpsShards.server"; export type SplitModeConfig = { flagEnabled: boolean; @@ -73,6 +77,35 @@ export function assertSplitRealtimeInterlock(config: SplitRealtimeInterlockConfi } } +export type ShardsRequireSplitConfig = { + splitFlagEnabled: boolean; + /** Raw descriptors. The alias exemption is applied here so no call site can forget it. */ + shards: RunOpsShardDescriptor[]; +}; + +/** + * Boot-time shard interlock (pure predicate). Shard clients are only built on the split-on arm of + * `selectRunOpsTopology`, so a shard configured while the split flag is off is dropped in silence: + * no client, no fan-out leg, and any row already resident on that database vanishes from every + * list with no error. The other two ways split can end up disabled (URLs missing, sentinel not + * distinct) already refuse to boot; this closes the one that does not. + */ +export function assertShardsRequireSplit(config: ShardsRequireSplitConfig): void { + if (config.splitFlagEnabled) { + return; + } + // An aliased shard owns no database: it shares its target's client by reference, so its rows are + // still read with the split off and nothing is dropped. Exempt here exactly as it is exempt from + // the distinctness sentinel, the coresidency loop and replication. + const owning = nonAliasedShards(config.shards).map((shard) => shard.key); + if (owning.length === 0) { + return; + } + throw new Error( + `RUN_OPS_SHARDS configures shard(s) ${owning.join(", ")} but RUN_OPS_SPLIT_ENABLED is off, so no shard client is built and rows on those databases would be silently missing; refusing to start.` + ); +} + let cached: Promise | undefined; export function isSplitEnabled(): Promise { diff --git a/apps/webapp/test/batchListPresenter.readroute.test.ts b/apps/webapp/test/batchListPresenter.readroute.test.ts index 9b1d55c9591..e7026f2663d 100644 --- a/apps/webapp/test/batchListPresenter.readroute.test.ts +++ b/apps/webapp/test/batchListPresenter.readroute.test.ts @@ -16,14 +16,18 @@ vi.mock("~/db.server", () => ({ import { heteroPostgresTest, heteroRunOpsPostgresTest, + makeNShardRunOpsPostgresTest, postgresTest, } from "@internal/testcontainers"; +import { generateRunOpsId, generateRunOpsIdV2 } from "@trigger.dev/core/v3/isomorphic"; import type { PrismaClient } from "@trigger.dev/database"; import type { RunOpsPrismaClient } from "@internal/run-ops-database"; import { + type BatchList, type BatchListOptions, BatchListPresenter, } from "~/presenters/v3/BatchListPresenter.server"; +import { nonAliasedShardReplicas } from "~/v3/runOpsMigration/shardHandles.server"; vi.setConfig({ testTimeout: 120_000 }); @@ -163,7 +167,7 @@ async function mirrorEnvParents( } async function createBatch( - prisma: PrismaClient, + prisma: PrismaClient | RunOpsPrismaClient, ctx: SeedContext, batch: { id: string; @@ -174,7 +178,7 @@ async function createBatch( createdAt?: Date; } ) { - return prisma.batchTaskRun.create({ + return (prisma as PrismaClient).batchTaskRun.create({ data: { id: batch.id, friendlyId: batch.friendlyId, @@ -621,3 +625,289 @@ describe("BatchListPresenter run-ops read routing (PG14 control-plane/legacy + P } ); }); + +describe("BatchListPresenter gen-2 shard legs (legacy PG14 + new PG17 + 2 shard PG17 databases)", () => { + const twoShardTest = makeNShardRunOpsPostgresTest(2); + + const shardPresenter = ( + legacyPrisma: PrismaClient, + newPrisma: RunOpsPrismaClient, + shardReplicas: ReadonlyArray<{ key: string; replica: RunOpsPrismaClient }> + ) => + new BatchListPresenter(legacyPrisma, legacyPrisma, { + runOpsNew: newPrisma, + runOpsLegacyReplica: legacyPrisma as unknown as RunOpsPrismaClient, + controlPlaneReplica: legacyPrisma, + splitEnabled: true, + shardReplicas, + }); + + twoShardTest( + "a gen-2 batch on its shard appears in the list alongside gen-1 and legacy batches", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const shardA = shardPrismas[0]!; + const ctx = await seedParents(legacyPrisma, "gen2-visible"); + + const legacyId = "cmm00000000000000000legac"; + const newId = generateRunOpsId(); + const shardId = generateRunOpsIdV2("a"); + + await createBatch(legacyPrisma, ctx, { + id: legacyId, + friendlyId: "fr_legacy", + createdAt: new Date(Date.now() - 3 * 60_000), + }); + await createBatch(newPrisma, ctx, { + id: newId, + friendlyId: "fr_new", + createdAt: new Date(Date.now() - 2 * 60_000), + }); + await createBatch(shardA, ctx, { + id: shardId, + friendlyId: "fr_shard_a", + createdAt: new Date(Date.now() - 1 * 60_000), + }); + + const page = await shardPresenter(legacyPrisma, newPrisma, [ + { key: "a", replica: shardA }, + ]).call(baseCall(ctx, { pageSize: 10 })); + + expect(page.batches.map((b) => b.id)).toEqual([shardId, newId, legacyId]); + expect(page.batches.map((b) => b.friendlyId)).toContain("fr_shard_a"); + + const withoutShardLeg = await shardPresenter(legacyPrisma, newPrisma, []).call( + baseCall(ctx, { pageSize: 10 }) + ); + expect(withoutShardLeg.batches.map((b) => b.id)).toEqual([newId, legacyId]); + } + ); + + twoShardTest( + "a page spanning legacy, new and two shards is ordered by createdAt then id across all stores", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const [shardA, shardB] = [shardPrismas[0]!, shardPrismas[1]!]; + const ctx = await seedParents(legacyPrisma, "gen2-order"); + + const t0 = new Date(Date.now() - 10 * 60_000); + const at = (minutes: number) => new Date(t0.getTime() + minutes * 60_000); + + const legacyId = "cmm00000000000000000order"; + const newId = generateRunOpsId(); + const shardAId = generateRunOpsIdV2("a"); + const shardBId = generateRunOpsIdV2("b"); + + await createBatch(legacyPrisma, ctx, { + id: legacyId, + friendlyId: "fr_o_legacy", + createdAt: at(0), + }); + await createBatch(newPrisma, ctx, { id: newId, friendlyId: "fr_o_new", createdAt: at(1) }); + await createBatch(shardA, ctx, { id: shardAId, friendlyId: "fr_o_a", createdAt: at(2) }); + await createBatch(shardB, ctx, { id: shardBId, friendlyId: "fr_o_b", createdAt: at(3) }); + + const tieTime = at(4); + const tieOnNew = generateRunOpsId(); + const tieOnShardB = generateRunOpsIdV2("b"); + await createBatch(newPrisma, ctx, { + id: tieOnNew, + friendlyId: "fr_tie_new", + createdAt: tieTime, + }); + await createBatch(shardB, ctx, { + id: tieOnShardB, + friendlyId: "fr_tie_b", + createdAt: tieTime, + }); + + const page = await shardPresenter(legacyPrisma, newPrisma, [ + { key: "a", replica: shardA }, + { key: "b", replica: shardB }, + ]).call(baseCall(ctx, { pageSize: 10 })); + + const tieHead = tieOnNew > tieOnShardB ? tieOnNew : tieOnShardB; + const tieTail = tieOnNew > tieOnShardB ? tieOnShardB : tieOnNew; + expect(page.batches.map((b) => b.id)).toEqual([ + tieHead, + tieTail, + shardBId, + shardAId, + newId, + legacyId, + ]); + } + ); + + twoShardTest( + "paging forward then backward across boundaries that span stores loses and repeats no batch", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const [shardA, shardB] = [shardPrismas[0]!, shardPrismas[1]!]; + const ctx = await seedParents(legacyPrisma, "gen2-paging"); + + const t0 = Date.now() - 60 * 60_000; + const seeded: string[] = []; + for (let i = 0; i < 8; i++) { + const store = [legacyPrisma, newPrisma, shardA, shardB][i % 4]!; + const id = + i % 4 === 0 + ? `cmm0000000000000000pag${i}` + : i % 4 === 1 + ? generateRunOpsId() + : generateRunOpsIdV2(i % 4 === 2 ? "a" : "b"); + await createBatch(store, ctx, { + id, + friendlyId: `fr_pag_${i}`, + createdAt: new Date(t0 + i * 60_000), + }); + seeded.push(id); + } + const newestFirst = [...seeded].reverse(); + + const presenter = shardPresenter(legacyPrisma, newPrisma, [ + { key: "a", replica: shardA }, + { key: "b", replica: shardB }, + ]); + + const forward: string[][] = []; + let cursor: string | undefined; + for (let guard = 0; guard < 10; guard++) { + const page = await presenter.call( + baseCall(ctx, { pageSize: 3, direction: "forward", cursor }) + ); + forward.push(page.batches.map((b) => b.id)); + if (!page.pagination.next) break; + cursor = page.pagination.next; + } + expect(forward.flat()).toEqual(newestFirst); + + let backCursor: string | undefined; + cursor = undefined; + for (let guard = 0; guard < 10; guard++) { + const page = await presenter.call( + baseCall(ctx, { pageSize: 3, direction: "forward", cursor }) + ); + backCursor = page.pagination.previous; + if (!page.pagination.next) break; + cursor = page.pagination.next; + } + + const backward: string[][] = []; + for (let guard = 0; guard < 10 && backCursor; guard++) { + const page: BatchList = await presenter.call( + baseCall(ctx, { pageSize: 3, direction: "backward", cursor: backCursor }) + ); + backward.unshift(page.batches.map((b) => b.id)); + backCursor = page.pagination.previous; + } + + const backwardIds = backward.flat(); + expect(backwardIds).toHaveLength(6); + expect(new Set(backwardIds).size).toBe(backwardIds.length); + expect(backwardIds).toEqual(newestFirst.slice(0, 6)); + } + ); + + twoShardTest( + "the empty-state probe reports batches present when only a shard holds batches", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const shardB = shardPrismas[1]!; + const ctx = await seedParents(legacyPrisma, "gen2-probe"); + + await createBatch(shardB, ctx, { + id: generateRunOpsIdV2("b"), + friendlyId: "fr_probe_shard", + }); + + const presenter = shardPresenter(legacyPrisma, newPrisma, [ + { key: "a", replica: shardPrismas[0]! }, + { key: "b", replica: shardB }, + ]); + + const page = await presenter.call(baseCall(ctx, { friendlyId: "fr_does_not_exist" })); + expect(page.batches).toHaveLength(0); + expect(page.hasAnyBatches).toBe(true); + + const withoutShardLeg = await shardPresenter(legacyPrisma, newPrisma, []).call( + baseCall(ctx, { friendlyId: "fr_does_not_exist" }) + ); + expect(withoutShardLeg.hasAnyBatches).toBe(false); + } + ); + + twoShardTest( + "a duplicated id resolves by precedence: a shard copy outranks new, and new outranks legacy", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const shardA = shardPrismas[0]!; + const ctx = await seedParents(legacyPrisma, "gen2-precedence"); + + const onBothGenOne = "cmm000000000000000000prec"; + await createBatch(legacyPrisma, ctx, { + id: onBothGenOne, + friendlyId: "fr_prec_gen1", + status: "PENDING", + createdAt: new Date(Date.now() - 60_000), + }); + await createBatch(newPrisma, ctx, { + id: onBothGenOne, + friendlyId: "fr_prec_gen1", + status: "COMPLETED", + createdAt: new Date(Date.now() - 60_000), + }); + + const onNewAndShard = generateRunOpsIdV2("a"); + await createBatch(newPrisma, ctx, { + id: onNewAndShard, + friendlyId: "fr_prec_shard", + status: "PENDING", + createdAt: new Date(Date.now() - 30_000), + }); + await createBatch(shardA, ctx, { + id: onNewAndShard, + friendlyId: "fr_prec_shard", + status: "COMPLETED", + createdAt: new Date(Date.now() - 30_000), + }); + + const page = await shardPresenter(legacyPrisma, newPrisma, [ + { key: "a", replica: shardA }, + ]).call(baseCall(ctx, { pageSize: 10 })); + + expect(page.batches.map((b) => b.id).filter((id) => id === onBothGenOne)).toHaveLength(1); + expect(page.batches.map((b) => b.id).filter((id) => id === onNewAndShard)).toHaveLength(1); + expect(page.batches.find((b) => b.id === onBothGenOne)?.status).toBe("COMPLETED"); + expect(page.batches.find((b) => b.id === onNewAndShard)?.status).toBe("COMPLETED"); + } + ); + + twoShardTest( + "an aliased shard contributes no leg, and its rows still arrive once via the aliased store", + async ({ legacyPrisma, newPrisma, shardPrismas }) => { + const shardB = shardPrismas[1]!; + const ctx = await seedParents(legacyPrisma, "gen2-alias"); + + const soakId = generateRunOpsIdV2("a"); + const realShardId = generateRunOpsIdV2("b"); + await createBatch(newPrisma, ctx, { + id: soakId, + friendlyId: "fr_soak", + createdAt: new Date(Date.now() - 60_000), + }); + await createBatch(shardB, ctx, { id: realShardId, friendlyId: "fr_real_shard" }); + + const aliasedSpy = spyClient(newPrisma as unknown as PrismaClient); + + const legs = nonAliasedShardReplicas([ + { key: "a", replica: aliasedSpy.client as unknown as RunOpsPrismaClient, aliasOf: "new" }, + { key: "b", replica: shardB }, + ]); + expect(legs.map((leg) => leg.key)).toEqual(["b"]); + + const page = await shardPresenter( + legacyPrisma, + aliasedSpy.client as unknown as RunOpsPrismaClient, + legs + ).call(baseCall(ctx, { pageSize: 10 })); + expect(page.batches.map((b) => b.id)).toEqual([realShardId, soakId]); + expect(aliasedSpy.counts.findMany).toBe(1); + } + ); +}); diff --git a/apps/webapp/test/runOpsSplitMode.test.ts b/apps/webapp/test/runOpsSplitMode.test.ts index fd7da6f356c..efe2ddca16a 100644 --- a/apps/webapp/test/runOpsSplitMode.test.ts +++ b/apps/webapp/test/runOpsSplitMode.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from "vitest"; import { PostgreSqlContainer } from "@testcontainers/postgresql"; import { computeSplitEnabled, + assertShardsRequireSplit, assertSplitRealtimeInterlock, } from "~/v3/runOpsMigration/splitMode.server"; import { probeDistinctDatabases } from "~/v3/runOpsMigration/distinctDbSentinel.server"; @@ -155,6 +156,58 @@ describe("assertSplitRealtimeInterlock (pure)", () => { }); }); +describe("assertShardsRequireSplit (pure)", () => { + const owning = (key: string) => ({ + key, + region: "local", + url: `postgres://${key}`, + replication: { slotName: `s_${key}`, publicationName: `p_${key}`, originGeneration: 2 }, + }); + const aliased = (key: string) => ({ key, region: "local", aliasOf: "new" as const }); + + it("allows shards when the split flag is on", () => { + expect(() => + assertShardsRequireSplit({ splitFlagEnabled: true, shards: [owning("a")] }) + ).not.toThrow(); + }); + + it("allows the split flag off when no shard is configured", () => { + expect(() => assertShardsRequireSplit({ splitFlagEnabled: false, shards: [] })).not.toThrow(); + }); + + // Shards are only built on the split-on arm of selectRunOpsTopology, so configuring one while + // the split flag is off silently drops it: no client, no leg, and any row already resident on + // that database disappears from every list with no error. + it("refuses to boot when a shard that owns a database is configured but the split flag is off", () => { + expect(() => + assertShardsRequireSplit({ splitFlagEnabled: false, shards: [owning("a"), owning("b")] }) + ).toThrow(/RUN_OPS_SHARDS/); + }); + + it("names the dropped shards so the operator can see which ones they are", () => { + expect(() => + assertShardsRequireSplit({ splitFlagEnabled: false, shards: [owning("a"), owning("b")] }) + ).toThrow(/a, b/); + }); + + // An aliased shard owns no database: it shares its target's client by reference, so its rows are + // still read with the split off. Refusing to boot for one is a false positive. + it("allows an alias-only config with the split flag off", () => { + expect(() => + assertShardsRequireSplit({ splitFlagEnabled: false, shards: [aliased("a")] }) + ).not.toThrow(); + }); + + it("refuses only for the owning shards when the config mixes both", () => { + expect(() => + assertShardsRequireSplit({ + splitFlagEnabled: false, + shards: [aliased("a"), owning("b")], + }) + ).toThrow(/shard\(s\) b /); + }); +}); + describe("distinct-DB sentinel (real Postgres)", () => { it("reports NOT distinct when both URLs hit the same physical cluster", async () => { const pg = await new PostgreSqlContainer("docker.io/postgres:14").start(); diff --git a/apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts b/apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts index 562d50b63d5..5ada87c5f9d 100644 --- a/apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts +++ b/apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts @@ -1,6 +1,6 @@ import { heteroPostgresTest } from "@internal/testcontainers"; import { PrismaClient } from "@trigger.dev/database"; -import { describe, expect, vi } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { probeDistinctDatabases, probeDistinctStores, @@ -66,6 +66,78 @@ describe("probeDistinctDatabases", () => { ); }); +// A transient failure on ONE store must not refuse the boot fleet-wide. The probe fails closed, +// so an unretried blip on any shard collapses the whole deployment to single-DB and the boot +// interlock then throws. Retry a bounded number of times, then fail closed exactly as before. +describe("probeDistinctStores bounded retry", () => { + const fp = (sysId: string, db: string) => ({ systemIdentifier: sysId, databaseName: db }); + + it("recovers when a transient failure clears within the retry budget", async () => { + let calls = 0; + const readFingerprint = vi.fn(async (url: string) => { + calls++; + if (calls === 2) throw new Error("ECONNREFUSED"); + return fp("sys", url); + }); + const result = await probeDistinctStores( + [ + { id: "new", url: "a" }, + { id: "shard-a", url: "b" }, + ], + { readFingerprint, attempts: 3, sleep: async () => {} } + ); + expect(result).toEqual({ distinct: true }); + }); + + it("fails closed once the retry budget is exhausted", async () => { + const readFingerprint = vi.fn(async () => { + throw new Error("ECONNREFUSED"); + }); + const result = await probeDistinctStores( + [ + { id: "new", url: "a" }, + { id: "shard-a", url: "b" }, + ], + { readFingerprint, attempts: 3, sleep: async () => {} } + ); + expect(result).toMatchObject({ distinct: false }); + }); + + it("bounds the attempts it makes", async () => { + const readFingerprint = vi.fn(async () => { + throw new Error("ECONNREFUSED"); + }); + await probeDistinctStores( + [ + { id: "a", url: "a" }, + { id: "b", url: "b" }, + ], + { + readFingerprint, + attempts: 3, + sleep: async () => {}, + } + ); + // 2 targets x 3 attempts each, and no more. + expect(readFingerprint).toHaveBeenCalledTimes(6); + }); + + // A duplicate is a correct, final answer. Retrying it would delay every boot of a genuinely + // misconfigured deployment for no benefit. + it("does not retry a genuine duplicate", async () => { + const readFingerprint = vi.fn(async () => fp("sys", "same")); + const result = await probeDistinctStores( + [ + { id: "new", url: "a" }, + { id: "shard-a", url: "b" }, + ], + { readFingerprint, attempts: 3, sleep: async () => {} } + ); + expect(result).toMatchObject({ distinct: false }); + expect(readFingerprint).toHaveBeenCalledTimes(2); + }); +}); + describe("probeDistinctStores (set uniqueness at N)", () => { heteroPostgresTest( "reports distinct for two separate physical clusters",