From 402109603f16d7d0013152564dbfe4862ec3cba2 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 7 Jul 2026 12:12:57 +0100 Subject: [PATCH 1/5] fix(webapp): anchor batch item run-ops residency to the batch friendlyId On the v4 (RunEngine V2) batch paths, RunEngineBatchTriggerService (api.v2) and the BatchQueue item callback (api.v3), a parentless batch's item runs minted their run-ops residency from a fresh per-org flag read at processing time. A mint-flag flip between batch creation and item processing could then mint an item into a different physical store than its BatchTaskRun row: an FK violation against the run-ops DB's live TaskRun_batchId_fkey (batch on legacy, item on new), or a silent orphan (batch on new, item on legacy). Anchor each item's mint on the batch's own friendlyId (pure id-shape, zero new queries), mirroring the already-safe BatchTriggerV3Service, and anchor the pre-failed-run fallback the same way since it also sets batchId. Consolidate the shared id-generation branch into mintFriendlyIdForKind so every mint path stays in lockstep. Single-DB mode is unchanged: a cuid-shaped batch friendlyId yields a cuid item. --- .server-changes/fix-batch-item-mint-anchor.md | 6 + .../runEngine/services/batchTrigger.server.ts | 11 ++ .../services/triggerFailedTask.server.test.ts | 42 ++++ .../services/triggerFailedTask.server.ts | 14 +- .../runEngine/services/triggerTask.server.ts | 6 +- .../webapp/app/v3/runEngineHandlers.server.ts | 24 +++ .../mintAnchoredRunFriendlyId.server.test.ts | 31 +++ .../mintAnchoredRunFriendlyId.server.ts | 15 ++ .../app/v3/services/batchTriggerV3.server.ts | 6 +- .../batchQueueItemResidencyAnchoring.test.ts | 179 +++++++++++++++++ ...gineBatchTriggerResidencyAnchoring.test.ts | 185 ++++++++++++++++++ 11 files changed, 509 insertions(+), 10 deletions(-) create mode 100644 .server-changes/fix-batch-item-mint-anchor.md create mode 100644 apps/webapp/app/runEngine/services/triggerFailedTask.server.test.ts create mode 100644 apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts create mode 100644 apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts create mode 100644 apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts create mode 100644 apps/webapp/test/runEngineBatchTriggerResidencyAnchoring.test.ts diff --git a/.server-changes/fix-batch-item-mint-anchor.md b/.server-changes/fix-batch-item-mint-anchor.md new file mode 100644 index 0000000000..0ee7526123 --- /dev/null +++ b/.server-changes/fix-batch-item-mint-anchor.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Anchor batch item run-ops residency on the batch's own friendlyId (not a fresh per-org flag read) in the run-engine batch trigger service and the BatchQueue item callback, on both the success path and the pre-failed-run failure path, so a mid-batch mint-flag flip can no longer mint an item (or a pre-failed item) into a different physical store than its BatchTaskRun row. diff --git a/apps/webapp/app/runEngine/services/batchTrigger.server.ts b/apps/webapp/app/runEngine/services/batchTrigger.server.ts index 772770becc..5e29d15892 100644 --- a/apps/webapp/app/runEngine/services/batchTrigger.server.ts +++ b/apps/webapp/app/runEngine/services/batchTrigger.server.ts @@ -18,6 +18,7 @@ import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; import { logger } from "~/services/logger.server"; import { batchTriggerWorker } from "~/v3/batchTriggerWorker.server"; import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server"; +import { mintAnchoredRunFriendlyId } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server"; import { mintBatchFriendlyId } from "~/v3/runOpsMigration/mintBatchFriendlyId.server"; import { downloadPacketFromObjectStore, @@ -588,6 +589,9 @@ export class RunEngineBatchTriggerService extends WithRunEngine { parentRunId, resumeParentOnCompletion, batch: { id: batch.id, index: workingIndex }, + // Anchor the pre-failed run on the BATCH's residency (same as the happy path) so it + // co-resides with its BatchTaskRun row regardless of a mid-batch mint-flag flip. + runFriendlyId: mintAnchoredRunFriendlyId(batch.friendlyId, item.options?.region), options: item.options as Record, traceContext: options?.traceContext as Record | undefined, spanParentAsLink: options?.spanParentAsLink, @@ -677,6 +681,12 @@ export class RunEngineBatchTriggerService extends WithRunEngine { const triggerTaskService = new TriggerTaskService(); + // Anchor the item's mint on the BATCH's own friendlyId (not a fresh per-org flag read) so + // an org's mint flag flipping between batch creation and this (possibly much later, + // worker-processed) item never splits the item from its BatchTaskRun row. See + // mintAnchoredRunFriendlyId.server.ts. + const runFriendlyId = mintAnchoredRunFriendlyId(batch.friendlyId, item.options?.region); + const result = await triggerTaskService.call( item.task, environment, @@ -695,6 +705,7 @@ export class RunEngineBatchTriggerService extends WithRunEngine { spanParentAsLink: options?.spanParentAsLink, batchId: batch.id, batchIndex: currentIndex, + runFriendlyId, realtimeStreamsVersion: options?.realtimeStreamsVersion, triggerSource: options?.triggerSource ?? "api", triggerAction: options?.triggerAction ?? "trigger", diff --git a/apps/webapp/app/runEngine/services/triggerFailedTask.server.test.ts b/apps/webapp/app/runEngine/services/triggerFailedTask.server.test.ts new file mode 100644 index 0000000000..a6d48d0e6b --- /dev/null +++ b/apps/webapp/app/runEngine/services/triggerFailedTask.server.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it, vi } from "vitest"; + +// Empty singletons satisfy the module-level wiring imports; the mint method under test is driven +// directly via (service as any) and never touches the DB (same boundary as triggerTask.server.test.ts). +vi.mock("~/db.server", () => ({ + prisma: {}, + $replica: {}, + runOpsNewPrisma: {}, + runOpsLegacyPrisma: {}, + runOpsNewReplica: {}, + runOpsLegacyReplica: {}, +})); +vi.mock("~/v3/runOpsMigration/splitMode.server", () => ({ isSplitEnabled: async () => false })); + +import { classifyKind, generateRunOpsId, RunId } from "@trigger.dev/core/v3/isomorphic"; +import { TriggerFailedTaskService } from "./triggerFailedTask.server"; + +function buildService() { + return new TriggerFailedTaskService({ prisma: {} as any, engine: {} as any }); +} + +describe("TriggerFailedTaskService.mintFailedRunFriendlyId", () => { + it("returns the caller-supplied runFriendlyId verbatim (override wins over any mint)", async () => { + const override = RunId.toFriendlyId(generateRunOpsId()); + const minted = await (buildService() as any).mintFailedRunFriendlyId({ + organizationId: "org_1", + environmentId: "env_1", + runFriendlyId: override, + }); + expect(minted).toBe(override); + }); + + it("without an override, still inherits a run-ops (NEW) parent by id-shape", async () => { + const parentRunFriendlyId = RunId.toFriendlyId(generateRunOpsId()); + const minted = await (buildService() as any).mintFailedRunFriendlyId({ + organizationId: "org_1", + environmentId: "env_1", + parentRunFriendlyId, + }); + expect(classifyKind(minted)).toBe("runOpsId"); + }); +}); diff --git a/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts b/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts index ee420c835d..d319bc2722 100644 --- a/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts +++ b/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts @@ -43,6 +43,9 @@ export type TriggerFailedTaskRequest = { spanParentAsLink?: boolean; errorCode?: TaskRunErrorCodes; + + /** Pre-minted friendlyId; when set it wins over the mint. Batch callers pass a batch-anchored id. */ + runFriendlyId?: string; }; /** @@ -86,14 +89,20 @@ export class TriggerFailedTaskService { // Mint a failed run's friendlyId. The id-kind decides which store the run is // born in (cuid → legacy store, run-ops id → new store); the whole subgraph of a - // run must agree. Root failed runs mint by the environment's setting; child - // failed runs inherit the parent's current store so they never split. + // run must agree. A caller-supplied runFriendlyId (batch-anchored id) wins verbatim; + // otherwise root failed runs mint by the environment's setting and child failed runs + // inherit the parent's current store so they never split. private async mintFailedRunFriendlyId(args: { organizationId: string; environmentId: string; orgFeatureFlags?: unknown; parentRunFriendlyId?: string; + runFriendlyId?: string; }): Promise { + if (args.runFriendlyId) { + return args.runFriendlyId; + } + const mintKind = args.parentRunFriendlyId ? resolveInheritedMintKind(args.parentRunFriendlyId) : await resolveRunIdMintKind({ @@ -125,6 +134,7 @@ export class TriggerFailedTaskService { environmentId: request.environment.id, orgFeatureFlags: request.environment.organization.featureFlags, parentRunFriendlyId: request.parentRunId, + runFriendlyId: request.runFriendlyId, }); mintedFriendlyId = failedRunFriendlyId; diff --git a/apps/webapp/app/runEngine/services/triggerTask.server.ts b/apps/webapp/app/runEngine/services/triggerTask.server.ts index 6a72ad34a1..f4981fece1 100644 --- a/apps/webapp/app/runEngine/services/triggerTask.server.ts +++ b/apps/webapp/app/runEngine/services/triggerTask.server.ts @@ -14,7 +14,6 @@ import { TriggerTraceContext, } from "@trigger.dev/core/v3"; import { - generateRunOpsId, parseTraceparent, RunId, serializeTraceparent, @@ -28,6 +27,7 @@ import { handleMetadataPacket } from "~/utils/packets"; import { startSpan } from "~/v3/tracing.server"; import { resolveRunIdMintKind } from "~/v3/engineVersion.server"; import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server"; +import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server"; import type { TriggerTaskServiceOptions, TriggerTaskServiceResult, @@ -153,9 +153,7 @@ export class RunEngineTriggerTaskService { orgFeatureFlags: environment.organization.featureFlags, }); - return mintKind === "runOpsId" - ? RunId.toFriendlyId(generateRunOpsId(region)) - : RunId.generate().friendlyId; + return mintFriendlyIdForKind(mintKind, region); } public async call({ diff --git a/apps/webapp/app/v3/runEngineHandlers.server.ts b/apps/webapp/app/v3/runEngineHandlers.server.ts index d6c4dd5fe3..fcefb782c1 100644 --- a/apps/webapp/app/v3/runEngineHandlers.server.ts +++ b/apps/webapp/app/v3/runEngineHandlers.server.ts @@ -28,6 +28,7 @@ import { getEventRepositoryForStore, recordRunDebugLog } from "./eventRepository import { roomFromFriendlyRunId, socketIo } from "./handleSocketIo.server"; import { engine } from "./runEngine.server"; import { runStore } from "./runStore.server"; +import { mintAnchoredRunFriendlyId } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server"; import { isSplitEnabled } from "~/v3/runOpsMigration/splitMode.server"; import { PerformTaskRunAlertsService } from "./services/alerts/performTaskRunAlerts.server"; import { @@ -837,6 +838,12 @@ export function setupBatchQueueCallbacks() { parentRunId: meta.parentRunId, resumeParentOnCompletion: meta.resumeParentOnCompletion, batch: { id: batchId, index: itemIndex }, + // Anchor the pre-failed run on the BATCH's residency so it co-resides with its + // BatchTaskRun row regardless of a mid-batch mint-flag flip. + runFriendlyId: mintAnchoredRunFriendlyId( + friendlyId, + (item.options as { region?: string } | undefined)?.region + ), traceContext: meta.traceContext as Record | undefined, spanParentAsLink: meta.spanParentAsLink, }); @@ -874,6 +881,14 @@ export function setupBatchQueueCallbacks() { // Normalize payload - for application/store (R2 paths), this passes through as-is const payload = normalizePayload(item.payload, item.payloadType); + // Anchor the item's mint on the BATCH's own friendlyId (not a fresh per-org flag + // read) so an org's mint flag flipping between batch creation and this queue-driven + // (possibly much later) callback never splits the item from its BatchTaskRun row. + const runFriendlyId = mintAnchoredRunFriendlyId( + friendlyId, + (item.options as { region?: string } | undefined)?.region + ); + const result = await triggerTaskService.call( item.task, environment, @@ -893,6 +908,7 @@ export function setupBatchQueueCallbacks() { spanParentAsLink: meta.spanParentAsLink, batchId, batchIndex: itemIndex, + runFriendlyId, realtimeStreamsVersion: meta.realtimeStreamsVersion, planType: meta.planType, triggerSource: meta.parentRunId ? "sdk" : (meta.triggerSource ?? "api"), @@ -929,6 +945,10 @@ export function setupBatchQueueCallbacks() { parentRunId: meta.parentRunId, resumeParentOnCompletion: meta.resumeParentOnCompletion, batch: { id: batchId, index: itemIndex }, + runFriendlyId: mintAnchoredRunFriendlyId( + friendlyId, + (item.options as { region?: string } | undefined)?.region + ), options: item.options as Record, traceContext: meta.traceContext as Record | undefined, spanParentAsLink: meta.spanParentAsLink, @@ -1009,6 +1029,10 @@ export function setupBatchQueueCallbacks() { parentRunId: meta.parentRunId, resumeParentOnCompletion: meta.resumeParentOnCompletion, batch: { id: batchId, index: itemIndex }, + runFriendlyId: mintAnchoredRunFriendlyId( + friendlyId, + (item.options as { region?: string } | undefined)?.region + ), options: item.options as Record, traceContext: meta.traceContext as Record | undefined, spanParentAsLink: meta.spanParentAsLink, diff --git a/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts new file mode 100644 index 0000000000..558731447a --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.test.ts @@ -0,0 +1,31 @@ +import { + BatchId, + classifyKind, + generateRunOpsId, + parseRunId, + REGION_CODES, +} from "@trigger.dev/core/v3/isomorphic"; +import { describe, expect, it } from "vitest"; +import { mintAnchoredRunFriendlyId } from "./mintAnchoredRunFriendlyId.server"; + +describe("mintAnchoredRunFriendlyId", () => { + it("a run-ops (NEW) batch anchor yields a run-ops (NEW) item friendlyId", () => { + const batchFriendlyId = BatchId.toFriendlyId(generateRunOpsId()); + const itemFriendlyId = mintAnchoredRunFriendlyId(batchFriendlyId); + expect(classifyKind(itemFriendlyId)).toBe("runOpsId"); + }); + + it("a cuid (LEGACY) batch anchor yields a cuid (LEGACY) item friendlyId", () => { + const batchFriendlyId = BatchId.generate().friendlyId; + const itemFriendlyId = mintAnchoredRunFriendlyId(batchFriendlyId); + expect(classifyKind(itemFriendlyId)).toBe("cuid"); + }); + + it("stamps the requested region char into a run-ops id", () => { + const batchFriendlyId = BatchId.toFriendlyId(generateRunOpsId()); + const itemFriendlyId = mintAnchoredRunFriendlyId(batchFriendlyId, "us-east-1"); + const parsed = parseRunId(itemFriendlyId); + expect(parsed.format).toBe("b32hex"); + expect(parsed.format === "b32hex" && parsed.region).toBe(REGION_CODES["us-east-1"]); + }); +}); diff --git a/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts new file mode 100644 index 0000000000..0f5da2e56f --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/mintAnchoredRunFriendlyId.server.ts @@ -0,0 +1,15 @@ +import { generateRunOpsId, RunId, type ResidencyKind } from "@trigger.dev/core/v3/isomorphic"; +import { resolveInheritedMintKind } from "./resolveInheritedMintKind.server"; + +// Shared id-generation branch for every run-mint path: "runOpsId" -> NEW store, "cuid" -> LEGACY. +export function mintFriendlyIdForKind(mintKind: ResidencyKind, region?: string): string { + return mintKind === "runOpsId" + ? RunId.toFriendlyId(generateRunOpsId(region)) + : RunId.generate().friendlyId; +} + +// Anchor a batch item's mint on the BATCH's friendlyId (id-shape, zero I/O), never the per-org +// flag, so the item and its BatchTaskRun stay co-resident across a mid-batch flag flip. +export function mintAnchoredRunFriendlyId(batchFriendlyId: string, region?: string): string { + return mintFriendlyIdForKind(resolveInheritedMintKind(batchFriendlyId), region); +} diff --git a/apps/webapp/app/v3/services/batchTriggerV3.server.ts b/apps/webapp/app/v3/services/batchTriggerV3.server.ts index bee0504a3d..299508e32c 100644 --- a/apps/webapp/app/v3/services/batchTriggerV3.server.ts +++ b/apps/webapp/app/v3/services/batchTriggerV3.server.ts @@ -12,7 +12,6 @@ import { Prisma, } from "@trigger.dev/database"; import type { RunStore } from "@internal/run-store"; -import { generateRunOpsId, RunId } from "@trigger.dev/core/v3/isomorphic"; import { z } from "zod"; import type { PrismaClientOrTransaction } from "~/db.server"; import { prisma } from "~/db.server"; @@ -26,6 +25,7 @@ import { getEntitlement } from "~/services/platform.v3.server"; import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server"; import { resolveRunIdMintKind, type RunIdMintKind } from "~/v3/engineVersion.server"; import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server"; +import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server"; import { mintBatchFriendlyId } from "~/v3/runOpsMigration/mintBatchFriendlyId.server"; import { batchTriggerWorker } from "../batchTriggerWorker.server"; import { legacyRunEngineWorker } from "../legacyRunEngineWorker.server"; @@ -361,9 +361,7 @@ export class BatchTriggerV3Service extends BaseService { orgFeatureFlags: environment.organization.featureFlags, }); - return mintKind === "runOpsId" - ? RunId.toFriendlyId(generateRunOpsId(region)) - : RunId.generate().friendlyId; + return mintFriendlyIdForKind(mintKind, region); } async #prepareRunData( diff --git a/apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts b/apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts new file mode 100644 index 0000000000..6ff733dcd5 --- /dev/null +++ b/apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts @@ -0,0 +1,179 @@ +// REGRESSION: the 2-phase batch API (createBatch.server.ts + streamBatchItems.server.ts, wired +// through BatchQueue to setupBatchQueueCallbacks) must anchor each item's mint on the BATCH's own +// friendlyId, like batchTriggerV3.server.ts's mintChildFriendlyId does — not re-resolve the +// per-org mint flag, which can flip between batch creation and this async callback. Covers the +// happy path AND the pre-failed-run failure branch, in BOTH residency directions. + +import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + findEnvironmentById: vi.fn(), + triggerTaskServiceCall: vi.fn(), + triggerFailedTaskCall: vi.fn(), + setBatchProcessItemCallback: vi.fn(), +})); + +vi.mock("~/db.server", () => ({ + prisma: {}, + $replica: {}, + runOpsNewPrisma: {}, + runOpsLegacyPrisma: {}, + runOpsNewReplica: {}, + runOpsLegacyReplica: {}, +})); + +vi.mock("~/models/runtimeEnvironment.server", () => ({ + findEnvironmentById: mocks.findEnvironmentById, + findEnvironmentFromRun: vi.fn(), +})); + +vi.mock("~/v3/services/triggerTask.server", () => ({ + TriggerTaskService: class { + call(...args: unknown[]) { + return mocks.triggerTaskServiceCall(...args); + } + }, +})); + +vi.mock("~/runEngine/services/triggerFailedTask.server", () => ({ + TriggerFailedTaskService: class { + call(...args: unknown[]) { + return mocks.triggerFailedTaskCall(...args); + } + }, +})); + +vi.mock("~/v3/runEngine.server", () => ({ + engine: { + setBatchProcessItemCallback: mocks.setBatchProcessItemCallback, + setBatchCompletionCallback: vi.fn(), + tryCompleteBatch: vi.fn(), + }, +})); + +vi.mock("~/v3/runOpsMigration/splitMode.server", () => ({ isSplitEnabled: async () => false })); + +import { + BatchId, + classifyKind, + generateRunOpsId, + ownerEngine, +} from "@trigger.dev/core/v3/isomorphic"; +import { setupBatchQueueCallbacks } from "~/v3/runEngineHandlers.server"; + +vi.setConfig({ testTimeout: 30_000 }); + +type ProcessItemCallback = (args: { + batchId: string; + friendlyId: string; + itemIndex: number; + item: Record; + meta: Record; + attempt: number; + isFinalAttempt: boolean; +}) => Promise; + +let processItemCallback: ProcessItemCallback; + +beforeAll(() => { + setupBatchQueueCallbacks(); + processItemCallback = mocks.setBatchProcessItemCallback.mock.calls[0][0] as ProcessItemCallback; +}); + +beforeEach(() => { + mocks.findEnvironmentById.mockReset().mockResolvedValue({ + id: "env_1", + organizationId: "org_1", + organization: { featureFlags: {} }, + }); + mocks.triggerTaskServiceCall.mockReset(); + mocks.triggerFailedTaskCall.mockReset(); +}); + +async function runItem(friendlyId: string, isFinalAttempt = false) { + await processItemCallback({ + batchId: "batch_internal_1", + friendlyId, + itemIndex: 0, + item: { task: "some-task", payload: "{}", payloadType: "application/json", options: {} }, + meta: { environmentId: "env_1" }, + attempt: 1, + isFinalAttempt, + }); +} + +describe("setupBatchQueueCallbacks — batch item residency anchoring", () => { + // Split is off in this test, so re-resolving the per-org flag would yield cuid — a NEW anchor + // yielding runOpsId proves the anchor wins over what the flag would say. + it("happy path: a run-ops (NEW) batch anchor mints a run-ops item", async () => { + const friendlyId = BatchId.toFriendlyId(generateRunOpsId()); + expect(ownerEngine(friendlyId)).toBe("NEW"); + mocks.triggerTaskServiceCall.mockResolvedValue({ + run: { id: "run_internal_1", friendlyId: "run_fake", status: "PENDING" }, + isCached: false, + }); + + await runItem(friendlyId); + + expect(mocks.triggerTaskServiceCall).toHaveBeenCalledTimes(1); + const [, , , options] = mocks.triggerTaskServiceCall.mock.calls[0] as [ + unknown, + unknown, + unknown, + { runFriendlyId?: string } | undefined, + ]; + expect(options?.runFriendlyId).toBeDefined(); + expect(classifyKind(options!.runFriendlyId!)).toBe("runOpsId"); + }); + + // The cuid direction catches a "always mint run-ops id regardless of anchor" bug. + it("happy path: a cuid (LEGACY) batch anchor mints a cuid item", async () => { + const friendlyId = BatchId.generate().friendlyId; + expect(ownerEngine(friendlyId)).toBe("LEGACY"); + mocks.triggerTaskServiceCall.mockResolvedValue({ + run: { id: "run_internal_1", friendlyId: "run_fake", status: "PENDING" }, + isCached: false, + }); + + await runItem(friendlyId); + + expect(mocks.triggerTaskServiceCall).toHaveBeenCalledTimes(1); + const [, , , options] = mocks.triggerTaskServiceCall.mock.calls[0] as [ + unknown, + unknown, + unknown, + { runFriendlyId?: string } | undefined, + ]; + expect(options?.runFriendlyId).toBeDefined(); + expect(classifyKind(options!.runFriendlyId!)).toBe("cuid"); + }); + + // The pre-failed run (TriggerTaskService returned undefined on the final attempt) carries + // batchId (→ live TaskRun.batchId FK), so it too must be anchored to the batch's residency. + it("failure branch: a run-ops (NEW) batch anchors the pre-failed run to the batch, not the flag", async () => { + const friendlyId = BatchId.toFriendlyId(generateRunOpsId()); + expect(ownerEngine(friendlyId)).toBe("NEW"); + mocks.triggerTaskServiceCall.mockResolvedValue(undefined); // item trigger fails + mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake"); + + await runItem(friendlyId, true); + + expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1); + const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }]; + expect(request.runFriendlyId).toBeDefined(); + expect(classifyKind(request.runFriendlyId!)).toBe("runOpsId"); + }); + + it("failure branch: a cuid (LEGACY) batch anchors the pre-failed run to a cuid id", async () => { + const friendlyId = BatchId.generate().friendlyId; + mocks.triggerTaskServiceCall.mockResolvedValue(undefined); // item trigger fails + mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake"); + + await runItem(friendlyId, true); + + expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1); + const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }]; + expect(request.runFriendlyId).toBeDefined(); + expect(classifyKind(request.runFriendlyId!)).toBe("cuid"); + }); +}); diff --git a/apps/webapp/test/runEngineBatchTriggerResidencyAnchoring.test.ts b/apps/webapp/test/runEngineBatchTriggerResidencyAnchoring.test.ts new file mode 100644 index 0000000000..b98c7d0d2e --- /dev/null +++ b/apps/webapp/test/runEngineBatchTriggerResidencyAnchoring.test.ts @@ -0,0 +1,185 @@ +// REGRESSION: batch + items must be co-resident (TaskRun.batchId FKs into BatchTaskRun on the +// run-ops NEW DB). RunEngineBatchTriggerService (api.v2.tasks.batch.ts) must anchor each item's +// mint on the BATCH's own friendlyId, like batchTriggerV3.server.ts's mintChildFriendlyId does, +// not re-resolve the per-org flag (which can flip between batch creation and async processing). +// Covers BOTH the happy path (item minted directly) AND the failure branch (pre-failed run via +// TriggerFailedTaskService), in BOTH residency directions. Drives the real `processBatchTaskRun` +// entrypoint with a fake `_engine` (no DB/Redis needed) and captures the trigger-pipeline inputs. + +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + findEnvironmentById: vi.fn(), + triggerTaskServiceCall: vi.fn(), + triggerFailedTaskCall: vi.fn(), +})); + +vi.mock("~/db.server", () => ({ + prisma: {}, + $replica: {}, + runOpsNewPrisma: {}, + runOpsLegacyPrisma: {}, + runOpsNewReplica: {}, + runOpsLegacyReplica: {}, +})); + +vi.mock("~/models/runtimeEnvironment.server", () => ({ + findEnvironmentById: mocks.findEnvironmentById, +})); + +vi.mock("~/v3/services/triggerTask.server", () => ({ + TriggerTaskService: class { + call(...args: unknown[]) { + return mocks.triggerTaskServiceCall(...args); + } + }, +})); + +vi.mock("~/runEngine/services/triggerFailedTask.server", () => ({ + TriggerFailedTaskService: class { + call(...args: unknown[]) { + return mocks.triggerFailedTaskCall(...args); + } + }, +})); + +import { + BatchId, + classifyKind, + generateRunOpsId, + ownerEngine, +} from "@trigger.dev/core/v3/isomorphic"; +import type { AuthenticatedEnvironment } from "~/services/apiAuth.server"; +import { RunEngineBatchTriggerService } from "~/runEngine/services/batchTrigger.server"; + +vi.setConfig({ testTimeout: 30_000 }); + +const environment = { + id: "env_1", + organizationId: "org_1", + organization: { featureFlags: {} }, +} as unknown as AuthenticatedEnvironment; + +function makeBatch(batchFriendlyId: string) { + return { + id: "batch_internal_1", + friendlyId: batchFriendlyId, + runtimeEnvironmentId: "env_1", + runCount: 1, + payload: JSON.stringify([{ task: "some-task", payload: "{}" }]), + payloadType: "application/json", + options: {}, + }; +} + +function makeFakeEngine(batch: ReturnType) { + return { + runStore: { + findBatchTaskRunById: vi.fn().mockResolvedValue(batch), + updateBatchTaskRun: vi.fn().mockResolvedValue({ processingJobsCount: 1, runCount: 1 }), + }, + tryCompleteBatch: vi.fn().mockResolvedValue(undefined), + }; +} + +async function processBatch(batchFriendlyId: string) { + const batch = makeBatch(batchFriendlyId); + const service = new RunEngineBatchTriggerService( + "sequential", + {} as any, + makeFakeEngine(batch) as any + ); + await service.processBatchTaskRun({ + batchId: batch.id, + processingId: "0", + range: { start: 0, count: 50 }, + attemptCount: 0, + strategy: "sequential", + }); +} + +function anchoredRunFriendlyIdArg(mock: ReturnType): string | undefined { + const [, , , options] = mock.mock.calls[0] as [ + unknown, + unknown, + unknown, + { runFriendlyId?: string } | undefined, + ]; + return options?.runFriendlyId; +} + +beforeEach(() => { + mocks.findEnvironmentById.mockReset().mockResolvedValue(environment); + mocks.triggerTaskServiceCall.mockReset(); + mocks.triggerFailedTaskCall.mockReset(); +}); + +describe("RunEngineBatchTriggerService — batch item residency anchoring", () => { + // Split is off in this test, so re-resolving the per-org flag would yield cuid — a NEW anchor + // yielding runOpsId proves the anchor wins over what the flag would say. + it("happy path: a run-ops (NEW) batch anchor mints a run-ops item, overriding what the flag would resolve", async () => { + const batchFriendlyId = BatchId.toFriendlyId(generateRunOpsId()); + expect(ownerEngine(batchFriendlyId)).toBe("NEW"); + + mocks.triggerTaskServiceCall.mockResolvedValue({ + run: { id: "run_internal_1", friendlyId: "run_fake", status: "PENDING" }, + isCached: false, + }); + + await processBatch(batchFriendlyId); + + expect(mocks.triggerTaskServiceCall).toHaveBeenCalledTimes(1); + const runFriendlyId = anchoredRunFriendlyIdArg(mocks.triggerTaskServiceCall); + expect(runFriendlyId).toBeDefined(); + expect(classifyKind(runFriendlyId!)).toBe("runOpsId"); + }); + + // The cuid direction catches a "always mint run-ops id regardless of anchor" bug. + it("happy path: a cuid (LEGACY) batch anchor mints a cuid item", async () => { + const batchFriendlyId = BatchId.generate().friendlyId; // cuid → LEGACY + expect(ownerEngine(batchFriendlyId)).toBe("LEGACY"); + + mocks.triggerTaskServiceCall.mockResolvedValue({ + run: { id: "run_internal_1", friendlyId: "run_fake", status: "PENDING" }, + isCached: false, + }); + + await processBatch(batchFriendlyId); + + expect(mocks.triggerTaskServiceCall).toHaveBeenCalledTimes(1); + const runFriendlyId = anchoredRunFriendlyIdArg(mocks.triggerTaskServiceCall); + expect(runFriendlyId).toBeDefined(); + expect(classifyKind(runFriendlyId!)).toBe("cuid"); + }); + + // The pre-failed run created by TriggerFailedTaskService still carries batchId (→ live + // TaskRun.batchId FK), so it too must be anchored to the batch's residency, not the per-org flag. + it("failure branch: a run-ops (NEW) batch anchors the pre-failed run to the batch, not the flag", async () => { + const batchFriendlyId = BatchId.toFriendlyId(generateRunOpsId()); + expect(ownerEngine(batchFriendlyId)).toBe("NEW"); + + mocks.triggerTaskServiceCall.mockResolvedValue(undefined); // item trigger fails + mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake"); + + await processBatch(batchFriendlyId); + + expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1); + const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }]; + expect(request.runFriendlyId).toBeDefined(); + expect(classifyKind(request.runFriendlyId!)).toBe("runOpsId"); + }); + + it("failure branch: a cuid (LEGACY) batch anchors the pre-failed run to a cuid id", async () => { + const batchFriendlyId = BatchId.generate().friendlyId; // cuid → LEGACY + + mocks.triggerTaskServiceCall.mockResolvedValue(undefined); // item trigger fails + mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake"); + + await processBatch(batchFriendlyId); + + expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1); + const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }]; + expect(request.runFriendlyId).toBeDefined(); + expect(classifyKind(request.runFriendlyId!)).toBe("cuid"); + }); +}); From 6e1b607f9e127614cf42480c088bf608bcdd14ac Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 7 Jul 2026 12:13:08 +0100 Subject: [PATCH 2/5] test(run-ops-database): verify real bidirectional schema parity The parity test only read the dedicated run-ops schema and matched model headers with regexes, so it never compared columns against the control-plane schema and could not detect a run-subgraph scalar column that diverged between the two physical schemas. Parse both schemas and assert bidirectional scalar-column parity (type, nullability, array-ness, default) across the run-subgraph models, scoped to those models so an unrelated control-plane edit can't break it, and fail on any unparsed field line so a format change can't silently bypass the check. --- .../prisma/schema.parity.test.ts | 262 +++++++++++++++--- 1 file changed, 217 insertions(+), 45 deletions(-) diff --git a/internal-packages/run-ops-database/prisma/schema.parity.test.ts b/internal-packages/run-ops-database/prisma/schema.parity.test.ts index 3598b9c811..6bb0a1bd90 100644 --- a/internal-packages/run-ops-database/prisma/schema.parity.test.ts +++ b/internal-packages/run-ops-database/prisma/schema.parity.test.ts @@ -2,11 +2,24 @@ import { describe, expect, it } from "vitest"; import { readFileSync } from "node:fs"; import { resolve } from "node:path"; -// Asserts every scalar column of the run-subgraph models in the control-plane -// schema also exists in the dedicated schema (so the dedicated DB can hold the -// same row shape), and that the dedicated schema contains NO reference to a -// control-plane model name. -const CONTROL_PLANE_MODELS = [ +// Parses both the control-plane schema (`@trigger.dev/database`) and this +// dedicated run-ops schema, then diffs every SCALAR field of each run-subgraph +// model BIDIRECTIONALLY: a field present in either schema must exist in the +// other with matching type/nullability/array-ness/`@default`, since a run row +// must be writable to either physical database. +// +// Relation fields are excluded by design: the dedicated schema drops +// control-plane `@relation`s to control-plane-owned models and replaces the +// implicit many-to-many waitpoint relations with explicit FK-free join models +// (`TaskRunWaitpoint`, `WaitpointRunConnection`, `CompletedWaitpoint`). Only +// the scalar FK columns backing those relations (e.g. `projectId`) are +// compared; the relation object fields (e.g. `project`) are skipped. + +const CONTROL_PLANE_SCHEMA_PATH = "../../database/prisma/schema.prisma"; +const DEDICATED_SCHEMA_PATH = "./schema.prisma"; + +// Must never appear as a relation target in the dedicated schema. +const CONTROL_PLANE_ONLY_MODELS = [ "Organization", "OrgMember", "Project", @@ -19,73 +32,232 @@ const CONTROL_PLANE_MODELS = [ "TaskQueue", ]; -function readSchema(rel: string) { +// Must have every scalar column reproduced in the dedicated schema. +const RUN_SUBGRAPH_MODELS = [ + "TaskRun", + "TaskRunAttempt", + "TaskRunExecutionSnapshot", + "TaskRunWaitpoint", + "TaskRunCheckpoint", + "CheckpointRestoreEvent", + "TaskRunTag", + "Waitpoint", + "WaitpointTag", + "BatchTaskRun", + "TaskRunDependency", + "BatchTaskRunItem", + "BatchTaskRunError", + "Checkpoint", +]; + +function readSchema(rel: string): string { return readFileSync(resolve(__dirname, rel), "utf8"); } -// Prisma comments (`///` docs and `//` lines) may legitimately mention -// control-plane model names in prose, which would false-match the drift -// regexes below. Strip them so parity assertions only see real schema syntax. -function stripComments(schema: string) { - return schema.replace(/\/\/.*$/gm, ""); +// Strips `/* */` block comments and `//`/`///` line comments so prose mentioning +// model names can't false-match below. (Neither schema has `/*` inside a string.) +function stripComments(schema: string): string { + return schema.replace(/\/\*[\s\S]*?\*\//g, "").replace(/\/\/.*$/gm, ""); +} + +interface FieldInfo { + type: string; + optional: boolean; + array: boolean; + default: string | null; +} + +type ModelFields = Map; + +// Prisma model blocks don't nest, so a lazy match to the next `}`-only line suffices. +function extractModelBlocks(schema: string): Map { + const blocks = new Map(); + const re = /^model\s+(\w+)\s*\{([\s\S]*?)\n\}/gm; + let match: RegExpExecArray | null; + while ((match = re.exec(schema))) { + blocks.set(match[1], match[2]); + } + return blocks; +} + +// Raw argument of `@default(...)`, honoring nested parens (`@default(cuid())`). Null if absent. +function extractDefault(attrs: string): string | null { + const marker = "@default("; + const idx = attrs.indexOf(marker); + if (idx === -1) return null; + const start = idx + marker.length; + let depth = 0; + for (let i = start; i < attrs.length; i++) { + if (attrs[i] === "(") { + depth++; + } else if (attrs[i] === ")") { + if (depth === 0) return attrs.slice(start, i); + depth--; + } + } + return attrs.slice(start); +} + +// Excludes relation fields: anything carrying `@relation`, plus back-relation fields +// (e.g. `checkpoints Checkpoint[]`) identified by their type being another model name. +// `unparsed` collects any content line the field regex fails to match, so a future +// multi-line field can never silently vanish from the diff. +function parseScalarFields( + body: string, + modelNames: Set +): { fields: ModelFields; unparsed: string[] } { + const fields: ModelFields = new Map(); + const unparsed: string[] = []; + for (const rawLine of body.split("\n")) { + const line = rawLine.trim(); + if (!line || line.startsWith("@@")) continue; + const match = line.match(/^(\w+)\s+([A-Za-z_]\w*)(\[\])?(\?)?\s*(.*)$/); + if (!match) { + unparsed.push(line); + continue; + } + const [, name, type, arrayMark, optionalMark, rest] = match; + if (rest.includes("@relation")) continue; + if (modelNames.has(type)) continue; + fields.set(name, { + type, + array: arrayMark === "[]", + optional: optionalMark === "?", + default: extractDefault(rest), + }); + } + return { fields, unparsed }; +} + +function parseSchema(schema: string) { + const stripped = stripComments(schema); + const modelBlocks = extractModelBlocks(stripped); + const modelNames = new Set(modelBlocks.keys()); + const models = new Map(); + const unparsed: string[] = []; + for (const [name, body] of modelBlocks) { + const parsed = parseScalarFields(body, modelNames); + models.set(name, parsed.fields); + for (const line of parsed.unparsed) { + unparsed.push(`${name}: ${line}`); + } + } + return { models, modelNames, unparsed }; +} + +function describeField(field: FieldInfo | undefined): string { + if (!field) return ""; + const array = field.array ? "[]" : ""; + const optional = field.optional ? "?" : ""; + const withDefault = field.default !== null ? ` @default(${field.default})` : ""; + return `${field.type}${array}${optional}${withDefault}`; } +const controlPlane = parseSchema(readSchema(CONTROL_PLANE_SCHEMA_PATH)); +const dedicated = parseSchema(readSchema(DEDICATED_SCHEMA_PATH)); + describe("dedicated run-ops schema parity", () => { + it("includes all 14 run-subgraph models in both schemas", () => { + for (const model of RUN_SUBGRAPH_MODELS) { + expect(Array.from(dedicated.modelNames)).toContain(model); + expect(Array.from(controlPlane.modelNames)).toContain(model); + } + }); + + it("fails on any content line the field parser doesn't recognise (no silent skips)", () => { + // Scope the control-plane check to run-subgraph models so an unrelated control-plane schema + // edit can't break this run-ops test; the dedicated schema contains only run-ops models. + const inRunSubgraph = (entry: string) => + RUN_SUBGRAPH_MODELS.some((model) => entry.startsWith(`${model}: `)); + expect(controlPlane.unparsed.filter(inRunSubgraph)).toEqual([]); + expect(dedicated.unparsed).toEqual([]); + }); + + it("reproduces every scalar column of each run-subgraph model bidirectionally with matching type/nullability/array/default", () => { + const mismatches: string[] = []; + + for (const modelName of RUN_SUBGRAPH_MODELS) { + const controlFields = controlPlane.models.get(modelName); + const dedicatedFields = dedicated.models.get(modelName); + if (!controlFields || !dedicatedFields) { + mismatches.push(`${modelName}: model not found in one of the two schemas`); + continue; + } + + const fieldNames = new Set([...controlFields.keys(), ...dedicatedFields.keys()]); + for (const fieldName of fieldNames) { + const controlField = controlFields.get(fieldName); + const dedicatedField = dedicatedFields.get(fieldName); + + if (!controlField) { + mismatches.push( + `${modelName}.${fieldName}: dedicated has ${describeField( + dedicatedField + )} but the control-plane schema has no such scalar field` + ); + continue; + } + if (!dedicatedField) { + mismatches.push( + `${modelName}.${fieldName}: control-plane has ${describeField( + controlField + )} but the dedicated schema has no such scalar field` + ); + continue; + } + + const matches = + dedicatedField.type === controlField.type && + dedicatedField.array === controlField.array && + dedicatedField.optional === controlField.optional && + dedicatedField.default === controlField.default; + + if (!matches) { + mismatches.push( + `${modelName}.${fieldName}: control-plane=${describeField( + controlField + )} dedicated=${describeField(dedicatedField)}` + ); + } + } + } + + expect(mismatches).toEqual([]); + }); + it("references no control-plane model as a relation target", () => { - const dedicated = stripComments(readSchema("./schema.prisma")); - for (const model of CONTROL_PLANE_MODELS) { + const dedicatedText = stripComments(readSchema(DEDICATED_SCHEMA_PATH)); + for (const model of CONTROL_PLANE_ONLY_MODELS) { // A relation target appears as ` fieldName Model @relation(...)`. A bare // scalar column like `projectId String` is fine; the model TYPE must be absent. const relationTarget = new RegExp( `@relation[^\\n]*\\b${model}\\b|\\b${model}\\b[^\\n]*@relation` ); - expect(dedicated).not.toMatch(relationTarget); - expect(dedicated).not.toMatch(new RegExp(`\\s${model}(\\?|\\[\\])?\\s`)); - } - }); - - it("includes all 14 run-subgraph models", () => { - const dedicated = readSchema("./schema.prisma"); - for (const m of [ - "TaskRun", - "TaskRunAttempt", - "TaskRunExecutionSnapshot", - "TaskRunWaitpoint", - "TaskRunCheckpoint", - "CheckpointRestoreEvent", - "TaskRunTag", - "Waitpoint", - "WaitpointTag", - "BatchTaskRun", - "TaskRunDependency", - "BatchTaskRunItem", - "BatchTaskRunError", - "Checkpoint", - ]) { - expect(dedicated).toMatch(new RegExp(`model ${m} \\{`)); + expect(dedicatedText).not.toMatch(relationTarget); + expect(dedicatedText).not.toMatch(new RegExp(`\\s${model}(\\?|\\[\\])?\\s`)); } }); it("keeps the group-(A) waitpoint-block references FK-FREE (scalar columns / explicit FK-free join models)", () => { - const dedicated = stripComments(readSchema("./schema.prisma")); + const dedicatedText = stripComments(readSchema(DEDICATED_SCHEMA_PATH)); // TaskRunWaitpoint must NOT carry a `@relation` to Waitpoint/TaskRun/BatchTaskRun. - const trw = dedicated.match(/model TaskRunWaitpoint \{[\s\S]*?\n\}/)![0]; + const trw = dedicatedText.match(/model TaskRunWaitpoint \{[\s\S]*?\n\}/)![0]; expect(trw).not.toMatch(/@relation/); expect(trw).toMatch(/waitpointId\s+String/); expect(trw).toMatch(/taskRunId\s+String/); // The two implicit M2M sets are replaced by explicit FK-free join models. - expect(dedicated).toMatch(/model WaitpointRunConnection \{/); - expect(dedicated).toMatch(/model CompletedWaitpoint \{/); - const wrc = dedicated.match(/model WaitpointRunConnection \{[\s\S]*?\n\}/)![0]; + expect(dedicatedText).toMatch(/model WaitpointRunConnection \{/); + expect(dedicatedText).toMatch(/model CompletedWaitpoint \{/); + const wrc = dedicatedText.match(/model WaitpointRunConnection \{[\s\S]*?\n\}/)![0]; expect(wrc).not.toMatch(/@relation/); // Waitpoint completion back-refs are scalar, not relations. - const wp = dedicated.match(/model Waitpoint \{[\s\S]*?\n\}/)![0]; + const wp = dedicatedText.match(/model Waitpoint \{[\s\S]*?\n\}/)![0]; expect(wp).not.toMatch(/completedByTaskRun\s+TaskRun\s*\?\s*@relation/); }); it("keeps the group-(B) co-resident references as real FKs (e.g. TaskRunAttempt.taskRun)", () => { - const dedicated = stripComments(readSchema("./schema.prisma")); - const attempt = dedicated.match(/model TaskRunAttempt \{[\s\S]*?\n\}/)![0]; + const dedicatedText = stripComments(readSchema(DEDICATED_SCHEMA_PATH)); + const attempt = dedicatedText.match(/model TaskRunAttempt \{[\s\S]*?\n\}/)![0]; // The attempt->run relation stays a real FK (always co-resident). expect(attempt).toMatch(/taskRun\s+TaskRun\s+@relation/); }); From e9ee0727711edf97a0e175c7d00d730a3ab6044d Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 7 Jul 2026 12:13:08 +0100 Subject: [PATCH 3/5] fix(webapp): signal when run-ops split read fan-out is silently disabled The read fan-out gate decides split-mode fan-out purely by the JS object identity of the NEW vs control-plane clients. If a future change made the NEW client alias the control-plane client while both split URLs are set, fan-out would silently disable with no signal, and no test exercised the real wiring from topology selection into the gate. Warn when both run-ops URLs are set but the NEW client is not a distinct instance, and add a glue test feeding the real topology output into the real gate. The gate's enabled value is unchanged. --- .../run-ops-split-read-gate-warn.md | 6 ++ apps/webapp/app/db.server.ts | 1 + .../v3/runOpsMigration/runOpsSplitReadGate.ts | 14 ++- .../test/runOpsSplitReadGate.glue.test.ts | 87 +++++++++++++++++ apps/webapp/test/runOpsSplitReadGate.test.ts | 94 ++++++++++++++++++- 5 files changed, 200 insertions(+), 2 deletions(-) create mode 100644 .server-changes/run-ops-split-read-gate-warn.md create mode 100644 apps/webapp/test/runOpsSplitReadGate.glue.test.ts diff --git a/.server-changes/run-ops-split-read-gate-warn.md b/.server-changes/run-ops-split-read-gate-warn.md new file mode 100644 index 0000000000..6bbc7a03a9 --- /dev/null +++ b/.server-changes/run-ops-split-read-gate-warn.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Warn when the run-ops split read configuration is present but read fan-out is silently disabled, instead of failing silently diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index 7805628f72..67cc366103 100644 --- a/apps/webapp/app/db.server.ts +++ b/apps/webapp/app/db.server.ts @@ -290,6 +290,7 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({ controlPlaneReplica: $replica, hasNewUrl: !!env.RUN_OPS_DATABASE_URL, hasLegacyUrl: !!env.RUN_OPS_LEGACY_DATABASE_URL, + logger, }); // Boot-time interlock: if the flag is on but the distinct-DB sentinel does not diff --git a/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts b/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts index 59eef29fbc..ce8c2d3d5d 100644 --- a/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts +++ b/apps/webapp/app/v3/runOpsMigration/runOpsSplitReadGate.ts @@ -6,9 +6,21 @@ export function computeRunOpsSplitReadEnabled(args: { controlPlaneReplica: unknown; hasNewUrl: boolean; hasLegacyUrl: boolean; + logger?: { warn: (msg: string, meta?: Record) => void }; }): boolean { const newIsDistinctDedicatedClient = args.newReplica !== args.controlPlaneWriter && args.newReplica !== args.controlPlaneReplica; - return newIsDistinctDedicatedClient && args.hasNewUrl && args.hasLegacyUrl; + const enabled = newIsDistinctDedicatedClient && args.hasNewUrl && args.hasLegacyUrl; + + // Configured for split but the identity check failed: fan-out is being silently disabled. + if (!newIsDistinctDedicatedClient && args.hasNewUrl && args.hasLegacyUrl) { + args.logger?.warn( + "run-ops split read fan-out is configured (RUN_OPS_DATABASE_URL and " + + "RUN_OPS_LEGACY_DATABASE_URL are both set) but the NEW client is not a distinct " + + "instance from the control-plane client; read fan-out is silently disabled." + ); + } + + return enabled; } diff --git a/apps/webapp/test/runOpsSplitReadGate.glue.test.ts b/apps/webapp/test/runOpsSplitReadGate.glue.test.ts new file mode 100644 index 0000000000..f4fc5b6890 --- /dev/null +++ b/apps/webapp/test/runOpsSplitReadGate.glue.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it, vi } from "vitest"; +import { selectRunOpsTopology } from "~/db.server"; +import { computeRunOpsSplitReadEnabled } from "~/v3/runOpsMigration/runOpsSplitReadGate"; + +// Glue test: nothing previously asserted that selectRunOpsTopology's ACTUAL output, fed into +// computeRunOpsSplitReadEnabled, yields the right gate. Each unit was only tested against +// hand-rolled objects, so a refactor that made the NEW client alias a control-plane client +// (even with correct URLs) would silently disable read fan-out with zero test failure. +const cp = { writer: { __tag: "cp-writer" } as any, replica: { __tag: "cp-replica" } as any }; + +describe("selectRunOpsTopology -> computeRunOpsSplitReadEnabled (seam)", () => { + it("split-configured with a genuinely distinct NEW client: gate TRUE, no warn", () => { + const dedicatedNew = { __tag: "dedicated-new" } as any; + const topo = selectRunOpsTopology( + { splitEnabled: true, legacyUrl: "postgres://legacy", newUrl: "postgres://new" }, + { + controlPlane: cp, + buildNewWriter: vi.fn().mockReturnValue(dedicatedNew), + buildNewReplica: vi.fn(), + } + ); + const warn = vi.fn(); + + const enabled = computeRunOpsSplitReadEnabled({ + newReplica: topo.newRunOps.replica, + controlPlaneWriter: topo.controlPlane.writer, + controlPlaneReplica: topo.controlPlane.replica, + hasNewUrl: true, + hasLegacyUrl: true, + logger: { warn }, + }); + + expect(enabled).toBe(true); + expect(warn).not.toHaveBeenCalled(); + }); + + it("regression: both URLs set but the client factory aliases the control-plane instance -> gate FALSE, warn fires", () => { + // Stand-in for the bug this test guards against: a builder refactor that accidentally + // returns the shared control-plane client instead of opening a dedicated connection. + const topo = selectRunOpsTopology( + { splitEnabled: true, legacyUrl: "postgres://legacy", newUrl: "postgres://new" }, + { + controlPlane: cp, + buildNewWriter: vi.fn().mockReturnValue(cp.replica), + buildNewReplica: vi.fn(), + } + ); + const warn = vi.fn(); + + const enabled = computeRunOpsSplitReadEnabled({ + newReplica: topo.newRunOps.replica, + controlPlaneWriter: topo.controlPlane.writer, + controlPlaneReplica: topo.controlPlane.replica, + hasNewUrl: true, + hasLegacyUrl: true, + logger: { warn }, + }); + + expect(enabled).toBe(false); + expect(warn).toHaveBeenCalledTimes(1); + }); + + it("single mode (URLs unset): topology naturally collapses to control-plane refs -> gate FALSE, no warn", () => { + const topo = selectRunOpsTopology( + { splitEnabled: false }, + { + controlPlane: cp, + buildNewWriter: vi.fn(), + buildNewReplica: vi.fn(), + } + ); + const warn = vi.fn(); + + const enabled = computeRunOpsSplitReadEnabled({ + newReplica: topo.newRunOps.replica, + controlPlaneWriter: topo.controlPlane.writer, + controlPlaneReplica: topo.controlPlane.replica, + hasNewUrl: false, + hasLegacyUrl: false, + logger: { warn }, + }); + + expect(enabled).toBe(false); + expect(warn).not.toHaveBeenCalled(); + expect(topo.newRunOps.replica).toBe(cp.replica); // sanity: genuinely the shared instance + }); +}); diff --git a/apps/webapp/test/runOpsSplitReadGate.test.ts b/apps/webapp/test/runOpsSplitReadGate.test.ts index c9238bcff3..4deb0bb532 100644 --- a/apps/webapp/test/runOpsSplitReadGate.test.ts +++ b/apps/webapp/test/runOpsSplitReadGate.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { computeRunOpsSplitReadEnabled } from "~/v3/runOpsMigration/runOpsSplitReadGate"; // Distinct sentinel objects standing in for the prisma client singletons. @@ -72,4 +72,96 @@ describe("computeRunOpsSplitReadEnabled", () => { false ); }); + + // Observability regression guard: split-configured (both URLs set) but the NEW client is not a + // distinct instance must WARN loudly. Without this signal, an accidental refactor that makes the + // NEW client alias a control-plane client silently disables read fan-out with zero error/warning. + describe("warn signal when configured-but-aliased", () => { + it("warns when both URLs are set but NEW aliases the control-plane replica", () => { + const warn = vi.fn(); + const enabled = computeRunOpsSplitReadEnabled({ + newReplica: cpReplica, // aliasing regression: NEW === control-plane replica + controlPlaneWriter: cpWriter, + controlPlaneReplica: cpReplica, + hasNewUrl: true, + hasLegacyUrl: true, + logger: { warn }, + }); + + expect(enabled).toBe(false); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0][0]).toMatch(/split.*configured/i); + }); + + it("warns when both URLs are set but NEW aliases the control-plane writer", () => { + const warn = vi.fn(); + const enabled = computeRunOpsSplitReadEnabled({ + newReplica: cpWriter, // aliasing regression: NEW === control-plane writer + controlPlaneWriter: cpWriter, + controlPlaneReplica: cpReplica, + hasNewUrl: true, + hasLegacyUrl: true, + logger: { warn }, + }); + + expect(enabled).toBe(false); + expect(warn).toHaveBeenCalledTimes(1); + }); + + it("does NOT warn in ordinary single mode (both URLs unset, clients naturally aliased)", () => { + const warn = vi.fn(); + const enabled = computeRunOpsSplitReadEnabled({ + newReplica: cpReplica, + controlPlaneWriter: cpWriter, + controlPlaneReplica: cpReplica, + hasNewUrl: false, + hasLegacyUrl: false, + logger: { warn }, + }); + + expect(enabled).toBe(false); + expect(warn).not.toHaveBeenCalled(); + }); + + it("does NOT warn when only one URL is set (not truly configured for split)", () => { + const warn = vi.fn(); + computeRunOpsSplitReadEnabled({ + newReplica: cpReplica, + controlPlaneWriter: cpWriter, + controlPlaneReplica: cpReplica, + hasNewUrl: true, + hasLegacyUrl: false, + logger: { warn }, + }); + + expect(warn).not.toHaveBeenCalled(); + }); + + it("does NOT warn when the NEW client is genuinely distinct (healthy split)", () => { + const warn = vi.fn(); + const enabled = computeRunOpsSplitReadEnabled({ + newReplica: dedicatedNew, + controlPlaneWriter: cpWriter, + controlPlaneReplica: cpReplica, + hasNewUrl: true, + hasLegacyUrl: true, + logger: { warn }, + }); + + expect(enabled).toBe(true); + expect(warn).not.toHaveBeenCalled(); + }); + + it("does not throw when no logger is supplied (logger stays optional)", () => { + expect(() => + computeRunOpsSplitReadEnabled({ + newReplica: cpReplica, + controlPlaneWriter: cpWriter, + controlPlaneReplica: cpReplica, + hasNewUrl: true, + hasLegacyUrl: true, + }) + ).not.toThrow(); + }); + }); }); From efd56f726192865c4fff9b59d1a97936774903db Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 7 Jul 2026 12:46:29 +0100 Subject: [PATCH 4/5] chore(webapp): drop redundant server-changes entry for the read fan-out warn --- .server-changes/run-ops-split-read-gate-warn.md | 6 ------ 1 file changed, 6 deletions(-) delete mode 100644 .server-changes/run-ops-split-read-gate-warn.md diff --git a/.server-changes/run-ops-split-read-gate-warn.md b/.server-changes/run-ops-split-read-gate-warn.md deleted file mode 100644 index 6bbc7a03a9..0000000000 --- a/.server-changes/run-ops-split-read-gate-warn.md +++ /dev/null @@ -1,6 +0,0 @@ ---- -area: webapp -type: fix ---- - -Warn when the run-ops split read configuration is present but read fan-out is silently disabled, instead of failing silently From 149909d2ad1031221c40123f3d658565256af96e Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 7 Jul 2026 13:50:51 +0100 Subject: [PATCH 5/5] refactor(webapp,run-ops-database): address bot review feedback Give TriggerFailedTaskService.callWithoutTraceEvents the same batch-anchored runFriendlyId override as call(), so a future batch caller of it can't mint an unanchored pre-failed run. Hoist the repeated anchored-mint expression in the BatchQueue item callback into a single local helper. Use a type alias instead of an interface in the schema-parity test, and cover the two remaining pre-failed-run branches (pre-marked error item, thrown trigger error). --- .../services/triggerFailedTask.server.ts | 3 ++ .../webapp/app/v3/runEngineHandlers.server.ts | 33 +++++------- .../batchQueueItemResidencyAnchoring.test.ts | 51 +++++++++++++++++-- .../prisma/schema.parity.test.ts | 4 +- 4 files changed, 65 insertions(+), 26 deletions(-) diff --git a/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts b/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts index d319bc2722..f8ba67f344 100644 --- a/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts +++ b/apps/webapp/app/runEngine/services/triggerFailedTask.server.ts @@ -331,6 +331,8 @@ export class TriggerFailedTaskService { resumeParentOnCompletion?: boolean; batch?: { id: string; index: number }; errorCode?: TaskRunErrorCodes; + /** Pre-minted friendlyId; when set it wins over the mint. Batch callers pass a batch-anchored id. */ + runFriendlyId?: string; }): Promise { // Held for the catch's log line; the in-try `const` is what consumers use. let mintedFriendlyId: string | undefined; @@ -345,6 +347,7 @@ export class TriggerFailedTaskService { // single replica lookup by organizationId only when there is no parent. orgFeatureFlags: undefined, parentRunFriendlyId: opts.parentRunId, + runFriendlyId: opts.runFriendlyId, }); mintedFriendlyId = failedRunFriendlyId; diff --git a/apps/webapp/app/v3/runEngineHandlers.server.ts b/apps/webapp/app/v3/runEngineHandlers.server.ts index fcefb782c1..ccda9baa45 100644 --- a/apps/webapp/app/v3/runEngineHandlers.server.ts +++ b/apps/webapp/app/v3/runEngineHandlers.server.ts @@ -809,6 +809,14 @@ export function setupBatchQueueCallbacks() { }, }, async (span) => { + // Anchor every item mint on the BATCH's friendlyId so a mid-batch mint-flag flip + // can't split an item (or pre-failed item) from its BatchTaskRun row. + const mintItemRunFriendlyId = () => + mintAnchoredRunFriendlyId( + friendlyId, + (item.options as { region?: string } | undefined)?.region + ); + const triggerFailedTaskService = new TriggerFailedTaskService({ prisma, engine, @@ -838,12 +846,7 @@ export function setupBatchQueueCallbacks() { parentRunId: meta.parentRunId, resumeParentOnCompletion: meta.resumeParentOnCompletion, batch: { id: batchId, index: itemIndex }, - // Anchor the pre-failed run on the BATCH's residency so it co-resides with its - // BatchTaskRun row regardless of a mid-batch mint-flag flip. - runFriendlyId: mintAnchoredRunFriendlyId( - friendlyId, - (item.options as { region?: string } | undefined)?.region - ), + runFriendlyId: mintItemRunFriendlyId(), traceContext: meta.traceContext as Record | undefined, spanParentAsLink: meta.spanParentAsLink, }); @@ -881,13 +884,7 @@ export function setupBatchQueueCallbacks() { // Normalize payload - for application/store (R2 paths), this passes through as-is const payload = normalizePayload(item.payload, item.payloadType); - // Anchor the item's mint on the BATCH's own friendlyId (not a fresh per-org flag - // read) so an org's mint flag flipping between batch creation and this queue-driven - // (possibly much later) callback never splits the item from its BatchTaskRun row. - const runFriendlyId = mintAnchoredRunFriendlyId( - friendlyId, - (item.options as { region?: string } | undefined)?.region - ); + const runFriendlyId = mintItemRunFriendlyId(); const result = await triggerTaskService.call( item.task, @@ -945,10 +942,7 @@ export function setupBatchQueueCallbacks() { parentRunId: meta.parentRunId, resumeParentOnCompletion: meta.resumeParentOnCompletion, batch: { id: batchId, index: itemIndex }, - runFriendlyId: mintAnchoredRunFriendlyId( - friendlyId, - (item.options as { region?: string } | undefined)?.region - ), + runFriendlyId: mintItemRunFriendlyId(), options: item.options as Record, traceContext: meta.traceContext as Record | undefined, spanParentAsLink: meta.spanParentAsLink, @@ -1029,10 +1023,7 @@ export function setupBatchQueueCallbacks() { parentRunId: meta.parentRunId, resumeParentOnCompletion: meta.resumeParentOnCompletion, batch: { id: batchId, index: itemIndex }, - runFriendlyId: mintAnchoredRunFriendlyId( - friendlyId, - (item.options as { region?: string } | undefined)?.region - ), + runFriendlyId: mintItemRunFriendlyId(), options: item.options as Record, traceContext: meta.traceContext as Record | undefined, spanParentAsLink: meta.spanParentAsLink, diff --git a/apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts b/apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts index 6ff733dcd5..8fdd2b159f 100644 --- a/apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts +++ b/apps/webapp/test/batchQueueItemResidencyAnchoring.test.ts @@ -2,7 +2,8 @@ // through BatchQueue to setupBatchQueueCallbacks) must anchor each item's mint on the BATCH's own // friendlyId, like batchTriggerV3.server.ts's mintChildFriendlyId does — not re-resolve the // per-org mint flag, which can flip between batch creation and this async callback. Covers the -// happy path AND the pre-failed-run failure branch, in BOTH residency directions. +// happy path (both residency directions) and all three pre-failed-run branches: trigger returned +// undefined, pre-marked error item, and a thrown trigger error. import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -90,12 +91,21 @@ beforeEach(() => { mocks.triggerFailedTaskCall.mockReset(); }); -async function runItem(friendlyId: string, isFinalAttempt = false) { +async function runItem( + friendlyId: string, + isFinalAttempt = false, + itemOptions: Record = {} +) { await processItemCallback({ batchId: "batch_internal_1", friendlyId, itemIndex: 0, - item: { task: "some-task", payload: "{}", payloadType: "application/json", options: {} }, + item: { + task: "some-task", + payload: "{}", + payloadType: "application/json", + options: itemOptions, + }, meta: { environmentId: "env_1" }, attempt: 1, isFinalAttempt, @@ -176,4 +186,39 @@ describe("setupBatchQueueCallbacks — batch item residency anchoring", () => { expect(request.runFriendlyId).toBeDefined(); expect(classifyKind(request.runFriendlyId!)).toBe("cuid"); }); + + // Pre-marked error items (e.g. oversized payloads) are pre-failed before the trigger runs; that + // pre-failed run also carries batchId, so it must anchor to the batch too. + it("pre-marked error branch: a run-ops (NEW) batch anchors the pre-failed run to the batch", async () => { + const friendlyId = BatchId.toFriendlyId(generateRunOpsId()); + expect(ownerEngine(friendlyId)).toBe("NEW"); + mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake"); + + await runItem(friendlyId, false, { + __error: "payload too large", + __errorCode: "PAYLOAD_TOO_LARGE", + }); + + expect(mocks.triggerTaskServiceCall).not.toHaveBeenCalled(); + expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1); + const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }]; + expect(request.runFriendlyId).toBeDefined(); + expect(classifyKind(request.runFriendlyId!)).toBe("runOpsId"); + }); + + // The catch branch (the item trigger throws) creates a pre-failed run on the final attempt; it + // carries batchId, so it must anchor to the batch too. + it("catch branch: a run-ops (NEW) batch anchors the pre-failed run when the trigger throws", async () => { + const friendlyId = BatchId.toFriendlyId(generateRunOpsId()); + expect(ownerEngine(friendlyId)).toBe("NEW"); + mocks.triggerTaskServiceCall.mockRejectedValue(new Error("trigger boom")); + mocks.triggerFailedTaskCall.mockResolvedValue("run_prefailed_fake"); + + await runItem(friendlyId, true); + + expect(mocks.triggerFailedTaskCall).toHaveBeenCalledTimes(1); + const [request] = mocks.triggerFailedTaskCall.mock.calls[0] as [{ runFriendlyId?: string }]; + expect(request.runFriendlyId).toBeDefined(); + expect(classifyKind(request.runFriendlyId!)).toBe("runOpsId"); + }); }); diff --git a/internal-packages/run-ops-database/prisma/schema.parity.test.ts b/internal-packages/run-ops-database/prisma/schema.parity.test.ts index 6bb0a1bd90..5e29f00ebc 100644 --- a/internal-packages/run-ops-database/prisma/schema.parity.test.ts +++ b/internal-packages/run-ops-database/prisma/schema.parity.test.ts @@ -60,12 +60,12 @@ function stripComments(schema: string): string { return schema.replace(/\/\*[\s\S]*?\*\//g, "").replace(/\/\/.*$/gm, ""); } -interface FieldInfo { +type FieldInfo = { type: string; optional: boolean; array: boolean; default: string | null; -} +}; type ModelFields = Map;