From e4c9b5c99a102cb4c9e8eb869fbc6e01893c3bcd Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sat, 18 Jul 2026 17:25:29 +0100 Subject: [PATCH 1/3] fix(webapp,run-engine,run-store,redis-worker): read-your-writes + global-scope idempotency correctness under the run-ops split MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Production code only — the guarding tests are in the stacked PR. 1) Read-your-writes: run-store reads that gate a mutation or feed a public GET / realtime response were routed to a lagging read replica, so a just-written run/waitpoint/batch could spuriously miss under replica lag. Route those reads to the owning primary (findRun/findWaitpoint/findBatchTaskRunByFriendlyId -> *OnPrimary, a primary re-read on a miss, or a retryable 404 where the SDK polls). Additive: the happy path is unchanged; a primary read happens only on a miss. 2) Global-scope idempotency across the split: a global-scope key carries no per-run salt, so the same (env, task, key) triggered concurrently from parents resident on different run-ops DBs could dedup-miss on each DB and create a duplicate run (the per-DB unique index can't enforce cross-DB uniqueness). Serialize such triggers (global scope, or scope-absent, while split is active) through the existing Redis idempotency claim, resolve the winner by id across both DBs, and reacquire the claim on the expired/failed clear-and-recreate path. run/attempt scope embed the run id in their hash and never contend. --- .../app/models/runtimeEnvironment.server.ts | 24 +- .../v3/ApiWaitpointPresenter.server.ts | 53 ++- .../presenters/v3/BatchPresenter.server.ts | 14 +- .../app/routes/api.v1.batches.$batchId.ts | 5 + ....runs.$runFriendlyId.input-streams.wait.ts | 30 +- ...uns.$runFriendlyId.session-streams.wait.ts | 28 +- .../app/routes/api.v1.runs.$runId.metadata.ts | 13 + ...i.v1.sessions.$session.end-and-continue.ts | 15 +- ...ens.$waitpointFriendlyId.callback.$hash.ts | 11 +- .../app/routes/api.v2.batches.$batchId.ts | 5 + .../routes/realtime.v1.batches.$batchId.ts | 5 + .../app/routes/realtime.v1.runs.$runId.ts | 28 +- .../realtime.v1.streams.$runId.$streamId.ts | 40 +- ...streams.$runId.$target.$streamId.append.ts | 40 +- ...ime.v1.streams.$runId.$target.$streamId.ts | 91 ++-- ...ltime.v1.streams.$runId.input.$streamId.ts | 59 +-- ...am.runs.$runParam.idempotencyKey.reset.tsx | 27 +- ...ram.realtime.v1.sessions.$sessionId.$io.ts | 41 +- ...am.realtime.v1.streams.$runId.$streamId.ts | 30 +- ...ltime.v1.streams.$runId.input.$streamId.ts | 30 +- .../route.tsx | 28 +- .../route.tsx | 10 +- .../resources.taskruns.$runParam.cancel.ts | 35 +- .../resources.taskruns.$runParam.replay.ts | 7 +- .../app/routes/sync.traces.runs.$traceId.ts | 11 +- .../concerns/idempotencyKeys.server.ts | 423 +++++++++++++----- .../app/services/dashboardAgent.server.ts | 4 +- .../realtime/sessionRunManager.server.ts | 6 +- .../v3/mollifier/idempotencyClaim.server.ts | 33 ++ .../app/v3/services/batchTriggerV3.server.ts | 23 +- .../src/engine/systems/runAttemptSystem.ts | 21 +- .../src/engine/systems/ttlSystem.ts | 34 +- .../run-store/src/runOpsStore.ts | 7 +- packages/redis-worker/src/mollifier/buffer.ts | 17 + 34 files changed, 818 insertions(+), 430 deletions(-) diff --git a/apps/webapp/app/models/runtimeEnvironment.server.ts b/apps/webapp/app/models/runtimeEnvironment.server.ts index 987394ea40c..939cd55e94d 100644 --- a/apps/webapp/app/models/runtimeEnvironment.server.ts +++ b/apps/webapp/app/models/runtimeEnvironment.server.ts @@ -274,19 +274,17 @@ export async function findEnvironmentFromRun( ): Promise { // Run-ops scalars (runTags/batchId/runtimeEnvironmentId) from the run store; the env half is // resolved via the control-plane resolver so the run-ops DB can split without a cross-DB join. - const taskRun = await runStore.findRun( - { - id: runId, - }, - { - select: { - runTags: true, - batchId: true, - runtimeEnvironmentId: true, - }, - }, - tx ?? $replica - ); + const select = { + runTags: true, + batchId: true, + runtimeEnvironmentId: true, + } as const; + let taskRun = await runStore.findRun({ id: runId }, { select }, tx ?? $replica); + if (!taskRun) { + // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary before + // treating it as absent, so runMetadataUpdated doesn't drop a live run's final metadata + publish. + taskRun = await runStore.findRun({ id: runId }, { select }, prisma); + } if (!taskRun) { return null; } diff --git a/apps/webapp/app/presenters/v3/ApiWaitpointPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiWaitpointPresenter.server.ts index c4477efefac..eb1d0dd0cdf 100644 --- a/apps/webapp/app/presenters/v3/ApiWaitpointPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/ApiWaitpointPresenter.server.ts @@ -42,29 +42,36 @@ export class ApiWaitpointPresenter extends BasePresenter { return this.trace("call", async (span) => { // The store routes by the waitpointId's residency (id shape) and reads the owning // store's replica. waitpointId is pre-decoded from the friendlyId via WaitpointId.toId. - const waitpoint = await this.runStore.findWaitpoint({ - where: { - id: waitpointId, - environmentId: environment.id, - }, - select: { - id: true, - friendlyId: true, - type: true, - status: true, - idempotencyKey: true, - userProvidedIdempotencyKey: true, - idempotencyKeyExpiresAt: true, - inactiveIdempotencyKey: true, - output: true, - outputType: true, - outputIsError: true, - completedAfter: true, - completedAt: true, - createdAt: true, - tags: true, - }, - }); + const where = { + id: waitpointId, + environmentId: environment.id, + }; + const select = { + id: true, + friendlyId: true, + type: true, + status: true, + idempotencyKey: true, + userProvidedIdempotencyKey: true, + idempotencyKeyExpiresAt: true, + inactiveIdempotencyKey: true, + output: true, + outputType: true, + outputIsError: true, + completedAfter: true, + completedAt: true, + createdAt: true, + tags: true, + } as const; + + let waitpoint = await this.runStore.findWaitpoint({ where, select }); + + // Read-your-writes on a public GET: a just-minted token may not be on the owning store's + // replica yet, so a replica miss would 404 a live token. Re-read the owning primary before + // concluding it doesn't exist (mirrors the metadata GET loader + the complete/callback paths). + if (!waitpoint) { + waitpoint = await this.runStore.findWaitpointOnPrimary({ where, select }); + } if (!waitpoint) { logger.error(`WaitpointPresenter: Waitpoint not found`, { diff --git a/apps/webapp/app/presenters/v3/BatchPresenter.server.ts b/apps/webapp/app/presenters/v3/BatchPresenter.server.ts index b9190a6f686..3e9ef0be858 100644 --- a/apps/webapp/app/presenters/v3/BatchPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/BatchPresenter.server.ts @@ -47,13 +47,25 @@ export class BatchPresenter extends BasePresenter { // The BatchTaskRun (run-ops) is read through the run store, which routes by residency. The // runtimeEnvironment (control-plane) is resolved separately because the cross-seam FK is // dropped, so the batch row cannot single-SQL join to control-plane RuntimeEnvironment. - const batch = await this.runStore.findBatchTaskRunByFriendlyId( + let batch = await this.runStore.findBatchTaskRunByFriendlyId( batchId, environmentId, { include: BATCH_INCLUDE }, this._replica ); + // Read-your-writes: findBatchTaskRunByFriendlyId defaults to (and here reads) the replica, so a + // batch created within the replica's apply window returns null under lag. Re-read from the owning + // primary on a miss so a live batch's detail page never spuriously 404s ("Batch not found"). + if (!batch) { + batch = await this.runStore.findBatchTaskRunByFriendlyId( + batchId, + environmentId, + { include: BATCH_INCLUDE }, + this._prisma + ); + } + if (!batch) { throw new Error("Batch not found"); } diff --git a/apps/webapp/app/routes/api.v1.batches.$batchId.ts b/apps/webapp/app/routes/api.v1.batches.$batchId.ts index 01acd3d140f..713716f20ee 100644 --- a/apps/webapp/app/routes/api.v1.batches.$batchId.ts +++ b/apps/webapp/app/routes/api.v1.batches.$batchId.ts @@ -12,6 +12,11 @@ export const loader = createLoaderApiRoute( params: ParamsSchema, allowJWT: true, corsStrategy: "all", + // A just-created batch may not yet have replicated to the read replica this client-less + // findBatchTaskRunByFriendlyId lookup routes to; return a retryable 404 so the SDK retries through + // replica lag rather than stranding a live batch on a permanent 404 (mirrors the run-get routes, + // e.g. api.v3.runs.$runId). + shouldRetryNotFound: true, findResource: (params, auth) => { return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id, { include: { errors: true }, diff --git a/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts b/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts index 3c081b2280b..024779ac666 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts @@ -33,21 +33,23 @@ const { action, loader } = createActionApiRoute( }, async ({ authentication, body, params }) => { try { - const run = await runStore.findRun( - { - friendlyId: params.runFriendlyId, - runtimeEnvironmentId: authentication.environment.id, + const where = { + friendlyId: params.runFriendlyId, + runtimeEnvironmentId: authentication.environment.id, + }; + const args = { + select: { + id: true, + friendlyId: true, + realtimeStreamsVersion: true, + streamBasinName: true, }, - { - select: { - id: true, - friendlyId: true, - realtimeStreamsVersion: true, - streamBasinName: true, - }, - }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 fails the .wait() registration on a run + // that exists. Re-read the owning primary on a replica miss. + const run = + (await runStore.findRun(where, args, $replica)) ?? + (await runStore.findRunOnPrimary(where, args)); if (!run) { return json({ error: "Run not found" }, { status: 404 }); diff --git a/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts b/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts index f6c58ab748b..c00ff51b3be 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.session-streams.wait.ts @@ -39,20 +39,22 @@ const { action, loader } = createActionApiRoute( }, async ({ authentication, body, params }) => { try { - const run = await runStore.findRun( - { - friendlyId: params.runFriendlyId, - runtimeEnvironmentId: authentication.environment.id, + const where = { + friendlyId: params.runFriendlyId, + runtimeEnvironmentId: authentication.environment.id, + }; + const args = { + select: { + id: true, + friendlyId: true, + realtimeStreamsVersion: true, }, - { - select: { - id: true, - friendlyId: true, - realtimeStreamsVersion: true, - }, - }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 fails the .wait() registration on a run + // that exists. Re-read the owning primary on a replica miss. + const run = + (await runStore.findRun(where, args, $replica)) ?? + (await runStore.findRunOnPrimary(where, args)); if (!run) { return json({ error: "Run not found" }, { status: 404 }); diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts b/apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts index 58cf572f44d..ac26302f8a3 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts +++ b/apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts @@ -64,6 +64,19 @@ export async function loader({ request, params }: LoaderFunctionArgs) { ); } + // Read-your-writes: a run drained from the buffer to the primary but not yet replicated misses + // both the replica read and the buffer. Re-read the owning primary before 404ing. + const primaryRun = await runStore.findRunOnPrimary( + { friendlyId: parsed.data.runId, runtimeEnvironmentId: env.id }, + { select: { metadata: true, metadataType: true } } + ); + if (primaryRun) { + return json( + { metadata: primaryRun.metadata, metadataType: primaryRun.metadataType }, + { status: 200 } + ); + } + return json({ error: "Run not found" }, { status: 404 }); } diff --git a/apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts b/apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts index 35770928c2d..c6e5a90f667 100644 --- a/apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts +++ b/apps/webapp/app/routes/api.v1.sessions.$session.end-and-continue.ts @@ -75,7 +75,7 @@ const { action, loader } = createActionApiRoute( // SDK exposes via `ctx.run.id`). Internally `Session.currentRunId` // stores the TaskRun.id cuid, so resolve before handing to the // optimistic-claim service. - const callingRun = await runStore.findRun( + let callingRun = await runStore.findRun( { friendlyId: body.callingRunId, runtimeEnvironmentId: authentication.environment.id, @@ -83,6 +83,19 @@ const { action, loader } = createActionApiRoute( { select: { id: true } }, $replica ); + if (!callingRun) { + // Replica lag: `callingRunId` is the agent's own live run (it is executing this request), so it + // exists on the owning primary even when the read replica has not caught up. Re-read the primary + // before 404ing — otherwise a lagging replica turns a legitimate handoff into a spurious + // "callingRunId not found in this environment". + callingRun = await runStore.findRunOnPrimary( + { + friendlyId: body.callingRunId, + runtimeEnvironmentId: authentication.environment.id, + }, + { select: { id: true } } + ); + } if (!callingRun) { return json({ error: "callingRunId not found in this environment" }, { status: 404 }); } diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts index 502db2aca79..431a7eb2582 100644 --- a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts +++ b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts @@ -36,13 +36,22 @@ export async function action({ request, params }: ActionFunctionArgs) { // Resolve wherever the waitpoint resides. The store routes by the waitpoint id's residency // (id-shape) and probes both run-ops DBs, so a token on either store resolves; the env is // resolved below from the row via the control-plane resolver. - const waitpoint = await runStore.findWaitpoint({ + let waitpoint = await runStore.findWaitpoint({ where: { id: waitpointId, }, select: { id: true, status: true, environmentId: true }, }); + if (!waitpoint) { + // Read-your-writes: a token whose callback fires right after mint may not have replicated + // yet. Re-read the owning primary before 404ing (mirrors complete.ts's primary fallback). + waitpoint = await runStore.findWaitpointOnPrimary({ + where: { id: waitpointId }, + select: { id: true, status: true, environmentId: true }, + }); + } + if (!waitpoint) { return json({ error: "Waitpoint not found" }, { status: 404 }); } diff --git a/apps/webapp/app/routes/api.v2.batches.$batchId.ts b/apps/webapp/app/routes/api.v2.batches.$batchId.ts index b6ef4136bd4..4f23f14bca6 100644 --- a/apps/webapp/app/routes/api.v2.batches.$batchId.ts +++ b/apps/webapp/app/routes/api.v2.batches.$batchId.ts @@ -12,6 +12,11 @@ export const loader = createLoaderApiRoute( params: ParamsSchema, allowJWT: true, corsStrategy: "all", + // A just-created batch may not yet have replicated to the read replica this client-less + // findBatchTaskRunByFriendlyId lookup routes to; return a retryable 404 so the SDK retries through + // replica lag rather than stranding a live batch on a permanent 404 (mirrors the run-get routes, + // e.g. api.v3.runs.$runId). + shouldRetryNotFound: true, findResource: (params, auth) => { return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id, { include: { errors: true }, diff --git a/apps/webapp/app/routes/realtime.v1.batches.$batchId.ts b/apps/webapp/app/routes/realtime.v1.batches.$batchId.ts index 29ad61d6646..b95cd4491db 100644 --- a/apps/webapp/app/routes/realtime.v1.batches.$batchId.ts +++ b/apps/webapp/app/routes/realtime.v1.batches.$batchId.ts @@ -13,6 +13,11 @@ export const loader = createLoaderApiRoute( params: ParamsSchema, allowJWT: true, corsStrategy: "all", + // A just-created batch may not yet have replicated to the read replica this client-less + // findBatchTaskRunByFriendlyId lookup routes to; return a retryable 404 so the SDK retries through + // replica lag rather than stranding a live batch on a permanent 404 (mirrors the run-get routes, + // e.g. api.v3.runs.$runId). + shouldRetryNotFound: true, findResource: (params, auth) => { return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id); }, diff --git a/apps/webapp/app/routes/realtime.v1.runs.$runId.ts b/apps/webapp/app/routes/realtime.v1.runs.$runId.ts index 51b47b22afd..c65bf0ecd43 100644 --- a/apps/webapp/app/routes/realtime.v1.runs.$runId.ts +++ b/apps/webapp/app/routes/realtime.v1.runs.$runId.ts @@ -15,22 +15,24 @@ export const loader = createLoaderApiRoute( allowJWT: true, corsStrategy: "all", findResource: async (params, authentication) => { - return runStore.findRun( - { - friendlyId: params.runId, - runtimeEnvironmentId: authentication.environment.id, - }, - { - include: { - batch: { - select: { - friendlyId: true, - }, + const where = { + friendlyId: params.runId, + runtimeEnvironmentId: authentication.environment.id, + }; + const args = { + include: { + batch: { + select: { + friendlyId: true, }, }, }, - $replica - ); + }; + // Replica lag can null out a run that already exists on the owning primary. A spurious 404 + // here permanently fails the client's realtime subscription (the SSE client treats 404 as + // "stream gone" — nonRetryableStatuses). Re-read the primary on a replica miss. + const run = await runStore.findRun(where, args, $replica); + return run ?? runStore.findRunOnPrimary(where, args); }, authorization: { action: "read", diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts index 895e336fd8b..ab941fd05ad 100644 --- a/apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts +++ b/apps/webapp/app/routes/realtime.v1.streams.$runId.$streamId.ts @@ -16,29 +16,29 @@ export const loader = createLoaderApiRoute( allowJWT: true, corsStrategy: "all", findResource: async (params, auth) => { - const run = await runStore.findRun( - { - friendlyId: params.runId, - runtimeEnvironmentId: auth.environment.id, - }, - { - select: { - id: true, - friendlyId: true, - taskIdentifier: true, - runTags: true, - realtimeStreamsVersion: true, - streamBasinName: true, - batch: { - select: { - friendlyId: true, - }, + const where = { + friendlyId: params.runId, + runtimeEnvironmentId: auth.environment.id, + }; + const args = { + select: { + id: true, + friendlyId: true, + taskIdentifier: true, + runTags: true, + realtimeStreamsVersion: true, + streamBasinName: true, + batch: { + select: { + friendlyId: true, }, }, }, - $replica - ); - return run; + }; + // Replica lag can null out a live run; a spurious 404 permanently fails the SSE subscription + // (the client treats 404 as "stream gone"). Re-read the owning primary on a replica miss. + const run = await runStore.findRun(where, args, $replica); + return run ?? runStore.findRunOnPrimary(where, args); }, authorization: { action: "read", diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.append.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.append.ts index 12400e8d0b0..7195a4ae9a6 100644 --- a/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.append.ts +++ b/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.append.ts @@ -27,29 +27,31 @@ const { action } = createActionApiRoute( maxContentLength: MAX_APPEND_BODY_BYTES, }, async ({ request, params, authentication }) => { - const run = await runStore.findRun( - { - friendlyId: params.runId, - runtimeEnvironmentId: authentication.environment.id, - }, - { - select: { - id: true, - friendlyId: true, - parentTaskRun: { - select: { - friendlyId: true, - }, + const where = { + friendlyId: params.runId, + runtimeEnvironmentId: authentication.environment.id, + }; + const args = { + select: { + id: true, + friendlyId: true, + parentTaskRun: { + select: { + friendlyId: true, }, - rootTaskRun: { - select: { - friendlyId: true, - }, + }, + rootTaskRun: { + select: { + friendlyId: true, }, }, }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 permanently fails the append. Re-read the + // owning primary on a replica miss. + const run = + (await runStore.findRun(where, args, $replica)) ?? + (await runStore.findRunOnPrimary(where, args)); if (!run) { return new Response("Run not found", { status: 404 }); diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts index 01a2b550d45..31c55645743 100644 --- a/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts +++ b/apps/webapp/app/routes/realtime.v1.streams.$runId.$target.$streamId.ts @@ -19,32 +19,34 @@ const { action } = createActionApiRoute( params: ParamsSchema, }, async ({ request, params, authentication }) => { - const run = await runStore.findRun( - { - friendlyId: params.runId, - runtimeEnvironmentId: authentication.environment.id, - }, - { - select: { - id: true, - friendlyId: true, - streamBasinName: true, - parentTaskRun: { - select: { - friendlyId: true, - streamBasinName: true, - }, + const where = { + friendlyId: params.runId, + runtimeEnvironmentId: authentication.environment.id, + }; + const args = { + select: { + id: true, + friendlyId: true, + streamBasinName: true, + parentTaskRun: { + select: { + friendlyId: true, + streamBasinName: true, }, - rootTaskRun: { - select: { - friendlyId: true, - streamBasinName: true, - }, + }, + rootTaskRun: { + select: { + friendlyId: true, + streamBasinName: true, }, }, }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 permanently fails the ingest client. + // Re-read the owning primary on a replica miss. + const run = + (await runStore.findRun(where, args, $replica)) ?? + (await runStore.findRunOnPrimary(where, args)); if (!run) { return new Response("Run not found", { status: 404 }); @@ -154,32 +156,33 @@ const loader = createLoaderApiRoute( allowJWT: false, corsStrategy: "none", findResource: async (params, authentication) => { - return runStore.findRun( - { - friendlyId: params.runId, - runtimeEnvironmentId: authentication.environment.id, - }, - { - select: { - id: true, - friendlyId: true, - streamBasinName: true, - parentTaskRun: { - select: { - friendlyId: true, - streamBasinName: true, - }, + const where = { + friendlyId: params.runId, + runtimeEnvironmentId: authentication.environment.id, + }; + const args = { + select: { + id: true, + friendlyId: true, + streamBasinName: true, + parentTaskRun: { + select: { + friendlyId: true, + streamBasinName: true, }, - rootTaskRun: { - select: { - friendlyId: true, - streamBasinName: true, - }, + }, + rootTaskRun: { + select: { + friendlyId: true, + streamBasinName: true, }, }, }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 permanently fails the HEAD probe. + // Re-read the owning primary on a replica miss. + const run = await runStore.findRun(where, args, $replica); + return run ?? runStore.findRunOnPrimary(where, args); }, }, async ({ request, params, resource: run, authentication }) => { diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts index 78fe332b8af..3c67fdb94d4 100644 --- a/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts +++ b/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts @@ -39,22 +39,24 @@ const { action } = createActionApiRoute( }, }, async ({ request, params, authentication }) => { - const run = await runStore.findRun( - { - friendlyId: params.runId, - runtimeEnvironmentId: authentication.environment.id, - }, - { - select: { - id: true, - friendlyId: true, - completedAt: true, - realtimeStreamsVersion: true, - streamBasinName: true, - }, + const where = { + friendlyId: params.runId, + runtimeEnvironmentId: authentication.environment.id, + }; + const args = { + select: { + id: true, + friendlyId: true, + completedAt: true, + realtimeStreamsVersion: true, + streamBasinName: true, }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 permanently fails the input send. Re-read + // the owning primary on a replica miss. + const run = + (await runStore.findRun(where, args, $replica)) ?? + (await runStore.findRunOnPrimary(where, args)); if (!run) { return json({ ok: false, error: "Run not found" }, { status: 404 }); @@ -133,22 +135,23 @@ const loader = createLoaderApiRoute( allowJWT: true, corsStrategy: "all", findResource: async (params, auth) => { - return runStore.findRun( - { - friendlyId: params.runId, - runtimeEnvironmentId: auth.environment.id, - }, - { - include: { - batch: { - select: { - friendlyId: true, - }, + const where = { + friendlyId: params.runId, + runtimeEnvironmentId: auth.environment.id, + }; + const args = { + include: { + batch: { + select: { + friendlyId: true, }, }, }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 permanently fails the SSE tail (the + // client treats 404 as "stream gone"). Re-read the owning primary on a replica miss. + const run = await runStore.findRun(where, args, $replica); + return run ?? runStore.findRunOnPrimary(where, args); }, authorization: { action: "read", diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.idempotencyKey.reset.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.idempotencyKey.reset.tsx index 9567ce084e7..dea9ed18e60 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.idempotencyKey.reset.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.idempotencyKey.reset.tsx @@ -12,20 +12,19 @@ export const action: ActionFunction = async ({ request, params }) => { const { projectParam, organizationSlug, envParam, runParam } = v3RunParamsSchema.parse(params); try { - const taskRun = await runStore.findRun( - { - friendlyId: runParam, - }, - { - select: { - id: true, - idempotencyKey: true, - taskIdentifier: true, - projectId: true, - runtimeEnvironmentId: true, - }, - } - ); + const resetSelect = { + id: true, + idempotencyKey: true, + taskIdentifier: true, + projectId: true, + runtimeEnvironmentId: true, + }; + let taskRun = await runStore.findRun({ friendlyId: runParam }, { select: resetSelect }); + if (!taskRun) { + // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary + // before 404ing — this null gates the reset mutation below (mirrors cancel/replay). + taskRun = await runStore.findRunOnPrimary({ friendlyId: runParam }, { select: resetSelect }); + } if (!taskRun) { return jsonWithErrorMessage({}, request, "Run not found"); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts index c001d66d509..062b75503a6 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts @@ -1,6 +1,6 @@ import { type LoaderFunctionArgs } from "@remix-run/server-runtime"; import { z } from "zod"; -import { $replica } from "~/db.server"; +import { $replica, prisma } from "~/db.server"; import { runStore } from "~/v3/runStore.server"; import { findProjectBySlug } from "~/models/project.server"; import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; @@ -8,7 +8,7 @@ import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server"; import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server"; import { canonicalSessionAddressingKey, - resolveSessionByIdOrExternalId, + resolveSessionWithWriterFallback, } from "~/services/realtime/sessions.server"; import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server"; import { requireUserId } from "~/services/session.server"; @@ -51,22 +51,27 @@ export async function loader({ request, params }: LoaderFunctionArgs) { // Verify the run lives in this environment — keeps callers from // subscribing to arbitrary sessions via `/runs/$runParam/...`. - const run = await runStore.findRun( - { - friendlyId: runParam, - runtimeEnvironmentId: environment.id, - }, - { - select: { id: true, friendlyId: true }, - }, - $replica - ); + const runWhere = { + friendlyId: runParam, + runtimeEnvironmentId: environment.id, + }; + const runArgs = { + select: { id: true, friendlyId: true }, + }; + // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription + // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss. + const run = + (await runStore.findRun(runWhere, runArgs, $replica)) ?? + (await runStore.findRunOnPrimary(runWhere, runArgs)); if (!run) { return new Response("Run not found", { status: 404 }); } - const session = await resolveSessionByIdOrExternalId($replica, environment.id, sessionId); + // Replica lag can null out a just-created session; a spurious 404 breaks the dashboard Agent tab + // subscription (useRealtimeStream surfaces the error and does not auto-retry). Resolve replica-first + // with a writer fallback — the same helper the sibling `.in/append` route uses. + const session = await resolveSessionWithWriterFallback(environment.id, sessionId); if (!session) { return new Response("Session not found", { status: 404 }); @@ -76,10 +81,12 @@ export async function loader({ request, params }: LoaderFunctionArgs) { // this environment is enough to subscribe to any session in the same // environment — defeats the point of scoping subscriptions through the // run route. SessionRun.runId is indexed (@unique), so this is cheap. - const linkedSessionRun = await $replica.sessionRun.findFirst({ - where: { runId: run.id, sessionId: session.id }, - select: { id: true }, - }); + // Replica lag can null out the just-created run↔session linkage row; a spurious 404 breaks the + // dashboard Agent tab subscription (client does not auto-retry). Re-read the primary on a miss. + const linkWhere = { runId: run.id, sessionId: session.id }; + const linkedSessionRun = + (await $replica.sessionRun.findFirst({ where: linkWhere, select: { id: true } })) ?? + (await prisma.sessionRun.findFirst({ where: linkWhere, select: { id: true } })); if (!linkedSessionRun) { return new Response("Session not found for run", { status: 404 }); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts index 6c50b0f1463..4fbb309e2a8 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts @@ -45,21 +45,23 @@ export async function loader({ request, params }: LoaderFunctionArgs) { return new Response("Environment not found", { status: 404 }); } - const run = await runStore.findRun( - { - friendlyId: runId, - runtimeEnvironmentId: environment.id, + const runWhere = { + friendlyId: runId, + runtimeEnvironmentId: environment.id, + }; + const runArgs = { + select: { + id: true, + friendlyId: true, + realtimeStreamsVersion: true, + streamBasinName: true, }, - { - select: { - id: true, - friendlyId: true, - realtimeStreamsVersion: true, - streamBasinName: true, - }, - }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription + // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss. + const run = + (await runStore.findRun(runWhere, runArgs, $replica)) ?? + (await runStore.findRunOnPrimary(runWhere, runArgs)); if (!run) { return new Response("Run not found", { status: 404 }); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts index 1ecc7819c23..7eae8f6fdcf 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts @@ -47,21 +47,23 @@ export async function loader({ request, params }: LoaderFunctionArgs) { return new Response("Environment not found", { status: 404 }); } - const run = await runStore.findRun( - { - friendlyId: runId, - runtimeEnvironmentId: environment.id, - }, - { - select: { - id: true, - friendlyId: true, - realtimeStreamsVersion: true, - streamBasinName: true, - }, + const runWhere = { + friendlyId: runId, + runtimeEnvironmentId: environment.id, + }; + const runArgs = { + select: { + id: true, + friendlyId: true, + realtimeStreamsVersion: true, + streamBasinName: true, }, - $replica - ); + }; + // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab input-stream + // subscription (useRealtimeStream surfaces the error, no auto-retry). Re-read the primary on a miss. + const run = + (await runStore.findRun(runWhere, runArgs, $replica)) ?? + (await runStore.findRunOnPrimary(runWhere, runArgs)); if (!run) { return new Response("Run not found", { status: 404 }); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx index 3d8cec515c5..43381613433 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx @@ -60,18 +60,22 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { throw new Response("Not Found", { status: 404 }); } - const run = await runStore.findRun( - { friendlyId: runParam, projectId: project.id }, - { - select: { - id: true, - friendlyId: true, - realtimeStreamsVersion: true, - streamBasinName: true, - runtimeEnvironmentId: true, - }, - } - ); + const runWhere = { friendlyId: runParam, projectId: project.id }; + const runArgs = { + select: { + id: true, + friendlyId: true, + realtimeStreamsVersion: true, + streamBasinName: true, + runtimeEnvironmentId: true, + }, + }; + // Client-less findRun defaults to the read replica; replica lag can null out a live run and 404 a + // valid stream-viewer request (useRealtimeStream surfaces the error, no auto-retry). Re-read the + // owning primary on a replica miss. + const run = + (await runStore.findRun(runWhere, runArgs)) ?? + (await runStore.findRunOnPrimary(runWhere, runArgs)); if (!run) { throw new Response("Not Found", { status: 404 }); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route.tsx index 16592ba33f5..e1ed1b2cc70 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.waitpoints.$waitpointFriendlyId.complete/route.tsx @@ -81,7 +81,7 @@ export const action = async ({ request, params }: ActionFunctionArgs) => { const waitpointId = WaitpointId.toId(waitpointFriendlyId); - const waitpoint = await runStore.findWaitpoint({ + let waitpoint = await runStore.findWaitpoint({ select: { projectId: true, environmentId: true, @@ -90,6 +90,14 @@ export const action = async ({ request, params }: ActionFunctionArgs) => { id: waitpointId, }, }); + if (!waitpoint) { + // Read-your-writes: a just-minted token may not have replicated. Re-read the owning primary + // before the auth guard / "No waitpoint found" (mirrors the token complete/callback routes). + waitpoint = await runStore.findWaitpointOnPrimary({ + select: { projectId: true, environmentId: true }, + where: { id: waitpointId }, + }); + } if (waitpoint?.projectId !== project.id) { return redirectWithErrorMessage( diff --git a/apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts b/apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts index ce74becc7be..cf6b0317780 100644 --- a/apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts +++ b/apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts @@ -80,21 +80,26 @@ export const action = dashboardAction( // The project-scope + membership auth is a control-plane concern resolved // separately below; joining project/organization here is a cross-DB join // that returns nothing once the run lives in the run-ops DB. - const taskRun = await runStore.findRun( - { friendlyId: runParam }, - { - select: { - id: true, - engine: true, - status: true, - friendlyId: true, - taskEventStore: true, - createdAt: true, - completedAt: true, - projectId: true, - }, - } - ); + const cancelRunSelect = { + id: true, + engine: true, + status: true, + friendlyId: true, + taskEventStore: true, + createdAt: true, + completedAt: true, + projectId: true, + }; + let taskRun = await runStore.findRun({ friendlyId: runParam }, { select: cancelRunSelect }); + if (!taskRun) { + // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary + // before treating it as absent (mirrors resolveRunOrganizationId's primary fallback above). + taskRun = await runStore.findRun( + { friendlyId: runParam }, + { select: cancelRunSelect }, + prisma + ); + } // Project-scope + membership auth is control-plane only — keyed by the // run's projectId. A miss is treated as not-found (mirrors the old where). diff --git a/apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts b/apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts index 8074a3d3b16..d6fe04be99f 100644 --- a/apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts +++ b/apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts @@ -323,7 +323,12 @@ export const action = dashboardAction( try { // Run-ops read keyed by friendlyId only; membership auth is re-checked on the // control plane below, keyed off the resolved run's projectId. - const pgRun = await runStore.findRun({ friendlyId: runParam }); + let pgRun = await runStore.findRun({ friendlyId: runParam }); + if (!pgRun) { + // Read-your-writes: a just-created run may not have replicated. Re-read the owning primary + // before falling back to the mollifier buffer (mirrors resolveRunOrganizationId above). + pgRun = await runStore.findRun({ friendlyId: runParam }, prisma); + } // Mollifier read-fallback: if the original isn't in PG yet, synthesise a // TaskRun from the buffered snapshot. Needs project/org/env slugs for the diff --git a/apps/webapp/app/routes/sync.traces.runs.$traceId.ts b/apps/webapp/app/routes/sync.traces.runs.$traceId.ts index e89772a103d..5fd1f7c65c0 100644 --- a/apps/webapp/app/routes/sync.traces.runs.$traceId.ts +++ b/apps/webapp/app/routes/sync.traces.runs.$traceId.ts @@ -23,7 +23,7 @@ export async function loader({ params, request }: LoaderFunctionArgs) { return new Response("No user found in cookie", { status: 401 }); } - const run = await runStore.findRun( + let run = await runStore.findRun( { traceId, }, @@ -35,6 +35,15 @@ export async function loader({ params, request }: LoaderFunctionArgs) { $replica ); + if (!run) { + // Read-your-writes: a just-created run may not have replicated yet. Re-read the owning + // primary before 404ing so a live run's realtime trace feed isn't spuriously not-found. + run = await runStore.findRunOnPrimary( + { traceId }, + { select: { runtimeEnvironmentId: true } } + ); + } + if (!run) { return new Response("No run found", { status: 404 }); } diff --git a/apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts b/apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts index ea50261a5a1..561e6a5d476 100644 --- a/apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts +++ b/apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts @@ -1,5 +1,5 @@ -import { RunId } from "@trigger.dev/core/v3/isomorphic"; -import type { PrismaClientOrTransaction, TaskRun } from "@trigger.dev/database"; +import { ownerEngine, RunId } from "@trigger.dev/core/v3/isomorphic"; +import type { PrismaClientOrTransaction, TaskRun, Waitpoint } from "@trigger.dev/database"; import { env } from "~/env.server"; import { logger } from "~/services/logger.server"; import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server"; @@ -8,7 +8,7 @@ import type { RunEngine } from "~/v3/runEngine.server"; import { shouldIdempotencyKeyBeCleared } from "~/v3/taskStatus"; import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server"; import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server"; -import { claimOrAwait } from "~/v3/mollifier/idempotencyClaim.server"; +import { claimOrAwait, resetResolvedClaim } from "~/v3/mollifier/idempotencyClaim.server"; import { makeResolveMollifierFlag } from "~/v3/mollifier/mollifierGate.server"; import { runStore } from "~/v3/runStore.server"; import { runOpsLegacyPrisma, runOpsNewPrisma } from "~/db.server"; @@ -24,6 +24,13 @@ import type { TraceEventConcern, TriggerTaskRequest } from "../types"; // handleTriggerRequest. const resolveOrgMollifierFlag = makeResolveMollifierFlag(); +// Cap on the claim-loser recreate re-acquisition loop (see +// reacquireClearedGlobalWinner). Each pass reopens a stale resolved slot and +// re-enters the claim; bounded so a pathological stream of expired/failed +// winners can't spin forever. On exhaustion we fall open to the create with +// PG's unique index as the backstop. +const MAX_CLEARED_WINNER_REACQUIRES = 5; + // Claim ownership context returned to the caller when the // IdempotencyKeyConcern won a pre-gate claim. Caller MUST publish the // winning runId on pipeline success (`publishClaim`) or release the @@ -209,107 +216,11 @@ export class IdempotencyKeyConcern { } if (existingRun) { - // The idempotency key has expired - if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) { - logger.debug("[TriggerTaskService][call] Idempotency key has expired", { - idempotencyKey: request.options?.idempotencyKey, - run: existingRun, - }); - - // Update the existing run to remove the idempotency key - await runStore.clearIdempotencyKey( - { byId: { runId: existingRun.id, idempotencyKey } }, - dedupClient - ); - - return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt }; - } - - // If the existing run failed or was expired, we clear the key and do a new run - if (shouldIdempotencyKeyBeCleared(existingRun.status)) { - logger.debug("[TriggerTaskService][call] Idempotency key should be cleared", { - idempotencyKey: request.options?.idempotencyKey, - runStatus: existingRun.status, - runId: existingRun.id, - }); - - // Update the existing run to remove the idempotency key - await runStore.clearIdempotencyKey( - { byId: { runId: existingRun.id, idempotencyKey } }, - dedupClient - ); - - return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt }; - } - - // We have an idempotent run, so we return it - const parentRunId = request.body.options?.parentRunId; - const resumeParentOnCompletion = request.body.options?.resumeParentOnCompletion; - - //We're using `andWait` so we need to block the parent run with a waitpoint - if (resumeParentOnCompletion && parentRunId) { - // `parentRunId` comes from the request body and isn't re-validated - // here, so confirm the parent run is in the caller's environment - // before wiring a waitpoint against it. - const parentRunInternalId = RunId.fromFriendlyId(parentRunId); - const parentRunInCallerEnv = await runStore.findRun( - { - id: parentRunInternalId, - runtimeEnvironmentId: request.environment.id, - }, - { select: { id: true } }, - this.prisma - ); - if (!parentRunInCallerEnv) { - throw new ServiceValidationError("Parent run not found in the calling environment", 404); - } - - // Get or create waitpoint lazily (existing run may not have one if it was standalone) - let associatedWaitpoint = existingRun.associatedWaitpoint; - if (!associatedWaitpoint) { - associatedWaitpoint = await this.engine.getOrCreateRunWaitpoint({ - runId: existingRun.id, - projectId: request.environment.projectId, - environmentId: request.environment.id, - }); - } - - await this.traceEventConcern.traceIdempotentRun( - request, - parentStore, - { - existingRun, - idempotencyKey, - incomplete: associatedWaitpoint.status === "PENDING", - isError: associatedWaitpoint.outputIsError, - }, - async (event) => { - const spanId = - request.options?.parentAsLinkType === "replay" - ? event.spanId - : event.traceparent?.spanId - ? `${event.traceparent.spanId}:${event.spanId}` - : event.spanId; - - await this.engine.blockRunWithWaitpoint({ - runId: parentRunInternalId, - waitpoints: associatedWaitpoint!.id, - spanIdToComplete: spanId, - batch: request.options?.batchId - ? { - id: request.options.batchId, - index: request.options.batchIndex ?? 0, - } - : undefined, - projectId: request.environment.projectId, - organizationId: request.environment.organizationId, - tx: dedupClient, - }); - } - ); - } - - return { isCached: true, run: existingRun }; + return await this.handleExistingRun(request, parentStore, existingRun, { + idempotencyKey, + idempotencyKeyExpiresAt, + dedupClient, + }); } // Pre-gate claim — closes the PG+buffer race during gate transition. @@ -325,20 +236,37 @@ export class IdempotencyKeyConcern { // claim's Redis SETNX keeps its RTT off the hot path for those requests // during staged rollout. The org-flag check is a pure in-memory read of // `Organization.featureFlags`, no DB query. + // + // Under the run-ops split the claim ALSO acts as the only cross-DB mutex a + // `global`-scope key has: that key is per (environment, task), so it can be + // triggered concurrently from two parents on DIFFERENT physical DBs where + // each probe misses and the per-DB unique index can't enforce uniqueness. + // For that case the claim is eligible regardless of the per-org flag AND of + // resumeParentOnCompletion (the loser wires its parent waitpoint against the + // winner in the resolved branch below). An absent scope (pre-hashed key / + // older SDK) is treated conservatively as possibly-global — harmless for a + // real run/attempt key, whose hash already embeds the parent id so two + // parents mint DISTINCT keys that never share a claim slot. + const idempotencyKeyScope = request.body.options?.idempotencyKeyOptions?.scope; + const globalUnderSplit = + (idempotencyKeyScope === "global" || idempotencyKeyScope === undefined) && + (await isSplitEnabled()); + const claimEligible = - !request.body.options?.resumeParentOnCompletion && !request.body.options?.debounce && !request.options?.oneTimeUseToken && - (await resolveOrgMollifierFlag({ - envId: request.environment.id, - orgId: request.environment.organizationId, - taskId: request.taskId, - orgFeatureFlags: - (request.environment.organization?.featureFlags as - | Record - | null - | undefined) ?? null, - })); + (globalUnderSplit || + (!request.body.options?.resumeParentOnCompletion && + (await resolveOrgMollifierFlag({ + envId: request.environment.id, + orgId: request.environment.organizationId, + taskId: request.taskId, + orgFeatureFlags: + (request.environment.organization?.featureFlags as + | Record + | null + | undefined) ?? null, + })))); if (claimEligible) { const ttlSeconds = Math.max( 1, @@ -356,6 +284,44 @@ export class IdempotencyKeyConcern { pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS, }); if (outcome.kind === "resolved") { + // Global-under-split loser: the winner lives on ITS parent's DB, which + // may differ from this loser's parent DB. Resolve the winner by id + // across both DBs (classify the winner friendlyId → NEW/LEGACY client, + // then let the router route+fall-back by id-shape) and feed it through + // the same existing-run handling the PG-hit path uses — so expiry-clear, + // status-clear, and the resumeParentOnCompletion waitpoint wiring all + // apply to the loser exactly as they would to a plain cached hit. + if (globalUnderSplit) { + const winner = await this.resolveWinnerAcrossDbs(outcome.runId, request.environment.id); + if (winner) { + const resolved = await this.handleExistingRun(request, parentStore, winner, { + idempotencyKey, + idempotencyKeyExpiresAt, + dedupClient, + }); + // A LIVE winner (cached hit, or andWait waitpoint wired) is terminal. + if (resolved.isCached) { + return resolved; + } + // CRITICAL (CodeRabbit): the resolved winner was EXPIRED or FAILED, so + // handleExistingRun cleared its key and would have us CREATE a new run. + // The initial create is serialised by the claim, but this clear-and- + // recreate is not — and under the split the per-DB unique index can't + // dedup a cross-residency recreate, so two concurrent losers clearing + // the SAME winner would each create → duplicate run. Re-serialise the + // recreate through the claim so exactly one caller recreates (and + // publishes) while the rest resolve to the fresh run. + return await this.reacquireClearedGlobalWinner(request, parentStore, { + idempotencyKey, + idempotencyKeyExpiresAt, + dedupClient, + ttlSeconds, + clearedRunId: outcome.runId, + safetyNetMs: env.TRIGGER_MOLLIFIER_CLAIM_WAIT_MS, + pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS, + }); + } + } // Another concurrent trigger committed first. Re-resolve via the // existing checks: writer-side PG findFirst first (defeats // replica lag), then buffer fallback for the buffered case. @@ -421,4 +387,231 @@ export class IdempotencyKeyConcern { return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt }; } + + // Resolve an already-existing idempotent run: honour key expiry / status + // clearing, and for `andWait` (resumeParentOnCompletion) block the calling + // parent on the run's waitpoint. Extracted so both the PG-hit path and the + // cross-DB claim-loser path resolve an existing run identically. + private async handleExistingRun( + request: TriggerTaskRequest, + parentStore: string | undefined, + existingRun: TaskRun & { associatedWaitpoint?: Waitpoint | null }, + ctx: { + idempotencyKey: string; + idempotencyKeyExpiresAt: Date; + dedupClient: PrismaClientOrTransaction; + } + ): Promise { + const { idempotencyKey, idempotencyKeyExpiresAt, dedupClient } = ctx; + + // The idempotency key has expired + if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) { + logger.debug("[TriggerTaskService][call] Idempotency key has expired", { + idempotencyKey: request.options?.idempotencyKey, + run: existingRun, + }); + + // Update the existing run to remove the idempotency key + await runStore.clearIdempotencyKey( + { byId: { runId: existingRun.id, idempotencyKey } }, + dedupClient + ); + + return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt }; + } + + // If the existing run failed or was expired, we clear the key and do a new run + if (shouldIdempotencyKeyBeCleared(existingRun.status)) { + logger.debug("[TriggerTaskService][call] Idempotency key should be cleared", { + idempotencyKey: request.options?.idempotencyKey, + runStatus: existingRun.status, + runId: existingRun.id, + }); + + // Update the existing run to remove the idempotency key + await runStore.clearIdempotencyKey( + { byId: { runId: existingRun.id, idempotencyKey } }, + dedupClient + ); + + return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt }; + } + + // We have an idempotent run, so we return it + const parentRunId = request.body.options?.parentRunId; + const resumeParentOnCompletion = request.body.options?.resumeParentOnCompletion; + + //We're using `andWait` so we need to block the parent run with a waitpoint + if (resumeParentOnCompletion && parentRunId) { + // `parentRunId` comes from the request body and isn't re-validated + // here, so confirm the parent run is in the caller's environment + // before wiring a waitpoint against it. + const parentRunInternalId = RunId.fromFriendlyId(parentRunId); + const parentRunInCallerEnv = await runStore.findRun( + { + id: parentRunInternalId, + runtimeEnvironmentId: request.environment.id, + }, + { select: { id: true } }, + this.prisma + ); + if (!parentRunInCallerEnv) { + throw new ServiceValidationError("Parent run not found in the calling environment", 404); + } + + // Get or create waitpoint lazily (existing run may not have one if it was standalone) + let associatedWaitpoint = existingRun.associatedWaitpoint; + if (!associatedWaitpoint) { + associatedWaitpoint = await this.engine.getOrCreateRunWaitpoint({ + runId: existingRun.id, + projectId: request.environment.projectId, + environmentId: request.environment.id, + }); + } + + await this.traceEventConcern.traceIdempotentRun( + request, + parentStore, + { + existingRun, + idempotencyKey, + incomplete: associatedWaitpoint.status === "PENDING", + isError: associatedWaitpoint.outputIsError, + }, + async (event) => { + const spanId = + request.options?.parentAsLinkType === "replay" + ? event.spanId + : event.traceparent?.spanId + ? `${event.traceparent.spanId}:${event.spanId}` + : event.spanId; + + await this.engine.blockRunWithWaitpoint({ + runId: parentRunInternalId, + waitpoints: associatedWaitpoint!.id, + spanIdToComplete: spanId, + batch: request.options?.batchId + ? { + id: request.options.batchId, + index: request.options.batchIndex ?? 0, + } + : undefined, + projectId: request.environment.projectId, + organizationId: request.environment.organizationId, + tx: dedupClient, + }); + } + ); + } + + return { isCached: true, run: existingRun }; + } + + // Re-serialise a cross-DB recreate through the claim after a claim-loser's + // resolved winner turned out to be EXPIRED / FAILED (its key was cleared by + // handleExistingRun). Without this, concurrent losers clearing the same + // winner each create a new run on their own DB — the per-DB unique index + // can't dedup a cross-residency pair, so the initial-create's serialisation + // is lost on the recreate. Each pass: compare-and-delete the stale resolved + // slot (keyed on the cleared runId — never an unconditional DEL, so a + // reacquirer that already re-published a NEW winner is not wiped), then + // re-enter claimOrAwait. Exactly one caller wins the re-claim and recreates + // (returning its claim to publish); the rest resolve to the fresh run. If a + // re-claim resolves to ANOTHER cleared winner we advance and loop, bounded + // by MAX_CLEARED_WINNER_REACQUIRES; on exhaustion (or an unfindable + // resolution) we fall open to the create, matching the initial claim's + // fail-open posture (PG unique index as backstop). + private async reacquireClearedGlobalWinner( + request: TriggerTaskRequest, + parentStore: string | undefined, + ctx: { + idempotencyKey: string; + idempotencyKeyExpiresAt: Date; + dedupClient: PrismaClientOrTransaction; + ttlSeconds: number; + clearedRunId: string; + safetyNetMs: number; + pollStepMs: number; + } + ): Promise { + const { idempotencyKey, idempotencyKeyExpiresAt, dedupClient, ttlSeconds } = ctx; + const claimInput = { + envId: request.environment.id, + taskIdentifier: request.taskId, + idempotencyKey, + }; + let staleRunId = ctx.clearedRunId; + for (let attempt = 0; attempt < MAX_CLEARED_WINNER_REACQUIRES; attempt++) { + await resetResolvedClaim({ ...claimInput, runId: staleRunId }); + + const outcome = await claimOrAwait({ + ...claimInput, + ttlSeconds, + safetyNetMs: ctx.safetyNetMs, + pollStepMs: ctx.pollStepMs, + }); + + if (outcome.kind === "timed_out") { + throw new ServiceValidationError("Idempotency claim resolution timed out", 503); + } + if (outcome.kind === "claimed") { + // We own the recreate. Caller MUST publish the new runId / release on error. + return { + isCached: false, + idempotencyKey, + idempotencyKeyExpiresAt, + claim: { ...claimInput, token: outcome.token }, + }; + } + // resolved: another caller won the recreate. Honour it like any cached + // hit (incl. andWait wiring). If it too was cleared, advance and loop. + const winner = await this.resolveWinnerAcrossDbs(outcome.runId, request.environment.id); + if (!winner) { + logger.warn("idempotency reacquire resolved but runId not findable", { + envId: request.environment.id, + taskIdentifier: request.taskId, + claimedRunId: outcome.runId, + }); + break; + } + const resolved = await this.handleExistingRun(request, parentStore, winner, { + idempotencyKey, + idempotencyKeyExpiresAt, + dedupClient, + }); + if (resolved.isCached) { + return resolved; + } + staleRunId = outcome.runId; + } + return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt }; + } + + // Resolve a claim winner (a run friendlyId) across both split DBs. Classify + // the id-shape to pick the writer client for read-your-writes, then read by + // id — the routing store routes to the owning store and falls back to the + // other, so a winner on either DB is found. Returns null when the id can't be + // classified or the row genuinely isn't there (caller falls through). + private async resolveWinnerAcrossDbs( + winnerFriendlyId: string, + environmentId: string + ): Promise<(TaskRun & { associatedWaitpoint?: Waitpoint | null }) | null> { + let internalId: string; + try { + internalId = RunId.fromFriendlyId(winnerFriendlyId); + } catch { + return null; + } + let client: PrismaClientOrTransaction; + try { + client = ownerEngine(internalId) === "NEW" ? runOpsNewPrisma : runOpsLegacyPrisma; + } catch { + client = this.prisma; + } + return runStore.findRun( + { id: internalId, runtimeEnvironmentId: environmentId }, + { include: { associatedWaitpoint: true } }, + client + ); + } } diff --git a/apps/webapp/app/services/dashboardAgent.server.ts b/apps/webapp/app/services/dashboardAgent.server.ts index 14eb51fcbef..039ee3c3ada 100644 --- a/apps/webapp/app/services/dashboardAgent.server.ts +++ b/apps/webapp/app/services/dashboardAgent.server.ts @@ -212,7 +212,9 @@ export async function resolveRunCommit( environmentId: string, runFriendlyId: string ): Promise<{ sha: string; version: string; dirty: boolean } | null> { - const run = await runStore.findRun( + // Read-your-writes: a just-locked run's lockedToVersionId may not have replicated. Read the owning + // primary so a live, pinned run resolves its commit instead of silently falling back to branch head. + const run = await runStore.findRunOnPrimary( { friendlyId: runFriendlyId, runtimeEnvironmentId: environmentId }, { select: { lockedToVersionId: true } } ); diff --git a/apps/webapp/app/services/realtime/sessionRunManager.server.ts b/apps/webapp/app/services/realtime/sessionRunManager.server.ts index 5ff462154bf..10cd966d95c 100644 --- a/apps/webapp/app/services/realtime/sessionRunManager.server.ts +++ b/apps/webapp/app/services/realtime/sessionRunManager.server.ts @@ -478,8 +478,10 @@ export async function swapSessionRun(params: SwapSessionRunParams): Promise { - // Use the read replica — this is a hot-path probe and stale-by-ms is - // fine. The append handler re-checks if it ends up reusing the runId. + // Use the read replica — hot-path probe, stale-by-ms is fine. The dangerous + // stale shape (looks-vanished → double-trigger) is closed by the writer re-probe + // in ensureRunForSession; a stale-non-final reuse self-heals via the durable S2 + // stream + next-append re-probe (there is no append-handler re-check). // `friendlyId` is fetched alongside `status` so the dead-run-detection // branch in `ensureRunForSession` can forward the public-form id as // `payload.previousRunId` without a second read. `Session.currentRunId` diff --git a/apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts b/apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts index 47c9733c927..196017b438d 100644 --- a/apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts +++ b/apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts @@ -213,6 +213,39 @@ export async function releaseClaim(input: { } } +// Reopen a stale RESOLVED claim slot whose winner was cleared (expired / +// failed), so the claim-loser reacquire path can re-serialise a cross-DB +// recreate through a fresh claim. Compare-and-delete on the observed winner +// runId (buffer-side) so a concurrent reacquirer's freshly published winner +// is never wiped. Best-effort: on buffer-null / error we no-op and let the +// reacquire's claimOrAwait proceed (fail-open — the caller caps attempts and +// ultimately falls through to the create, PG unique index as backstop). +export async function resetResolvedClaim(input: { + envId: string; + taskIdentifier: string; + idempotencyKey: string; + runId: string; + buffer?: MollifierBuffer | null; +}): Promise { + const buffer = input.buffer === undefined ? getMollifierBuffer() : input.buffer; + if (!buffer) return false; + try { + return await buffer.resetResolvedClaim({ + envId: input.envId, + taskIdentifier: input.taskIdentifier, + idempotencyKey: input.idempotencyKey, + expectedRunId: input.runId, + }); + } catch (err) { + logger.warn("idempotency resolved-claim reset failed", { + envId: input.envId, + taskIdentifier: input.taskIdentifier, + err: err instanceof Error ? err.message : String(err), + }); + return false; + } +} + function defaultSleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } diff --git a/apps/webapp/app/v3/services/batchTriggerV3.server.ts b/apps/webapp/app/v3/services/batchTriggerV3.server.ts index a50ba637474..fdb87d2d2ca 100644 --- a/apps/webapp/app/v3/services/batchTriggerV3.server.ts +++ b/apps/webapp/app/v3/services/batchTriggerV3.server.ts @@ -195,18 +195,23 @@ export class BatchTriggerV3Service extends BaseService { span.setAttribute("batchId", batchId); const dependentAttempt = body?.dependentAttempt - ? await this.runStore.findTaskRunAttempt({ - // Scope to the caller's environment (see dependentAttemptWhere). - where: dependentAttemptWhere(body.dependentAttempt, environment.id), - include: { - taskRun: { - select: { - id: true, - status: true, + ? await this.runStore.findTaskRunAttempt( + { + // Scope to the caller's environment (see dependentAttemptWhere). + where: dependentAttemptWhere(body.dependentAttempt, environment.id), + include: { + taskRun: { + select: { + id: true, + status: true, + }, }, }, }, - }) + // Read-your-writes: the dependent parent attempt's terminal state must be seen even + // when the read replica lags, so route this read to the owning primary. + this._prisma + ) : undefined; if ( diff --git a/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts b/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts index 7f53c7648de..922b41cd225 100644 --- a/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/runAttemptSystem.ts @@ -170,7 +170,9 @@ export class RunAttemptSystem { } public async resolveTaskRunContext(runId: string): Promise { - const run = await this.$.runStore.findRun( + // read-your-writes: a just-created/locked run may not be on a replica yet; read the owning primary + // so a live run's execution context resolves instead of throwing a spurious 404. + const run = await this.$.runStore.findRunOnPrimary( { id: runId, }, @@ -711,8 +713,8 @@ export class RunAttemptSystem { const completedAt = new Date(); - // Read current usage values to calculate new totals (safe under runLock) - const currentRun = await this.$.runStore.findRun( + // Read current usage totals on the owning primary (read-your-writes; a replica lags) + const currentRun = await this.$.runStore.findRunOnPrimary( { id: runId }, { select: { @@ -904,7 +906,8 @@ export class RunAttemptSystem { const failedAt = new Date(); const retryResult = await retryOutcomeFromCompletion( - this.$.readOnlyPrisma, + // read-your-writes: lock-time maxAttempts/lockedRetryConfig may not be on a replica yet + this.$.prisma, this.$.runStore, { runId, @@ -917,7 +920,9 @@ export class RunAttemptSystem { // Force requeue means it was crashed so the attempt span needs to be closed if (forceRequeue) { - const minimalRun = await this.$.runStore.findRun( + // read-your-writes: the run was just written in this flow; read the owning primary so the + // requeue event re-read cannot false-miss on a lagging replica (mirrors the :906 read). + const minimalRun = await this.$.runStore.findRunOnPrimary( { id: runId, }, @@ -1375,7 +1380,7 @@ export class RunAttemptSystem { // Calculate updated usage if we have attempt duration data let usageUpdate: { usageDurationMs: number; costInCents: number } | undefined; if (attemptDurationMs !== undefined) { - const currentRun = await this.$.runStore.findRun( + const currentRun = await this.$.runStore.findRunOnPrimary( { id: runId }, { select: { @@ -1589,8 +1594,8 @@ export class RunAttemptSystem { const truncatedError = this.#truncateTaskRunError(error); - // Read current usage values to calculate new totals - const currentRun = await this.$.runStore.findRun( + // Read current usage totals on the owning primary (read-your-writes; a replica lags) + const currentRun = await this.$.runStore.findRunOnPrimary( { id: runId }, { select: { diff --git a/internal-packages/run-engine/src/engine/systems/ttlSystem.ts b/internal-packages/run-engine/src/engine/systems/ttlSystem.ts index af6f41eac2e..0fb3b8387cb 100644 --- a/internal-packages/run-engine/src/engine/systems/ttlSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/ttlSystem.ts @@ -158,22 +158,26 @@ export class TtlSystem { const skipped: { runId: string; reason: string }[] = []; // Fetch all runs in a single query (no snapshot data needed) - const runs = await this.$.runStore.findRuns({ - where: { id: { in: runIds } }, - select: { - id: true, - spanId: true, - status: true, - lockedAt: true, - ttl: true, - taskEventStore: true, - createdAt: true, - associatedWaitpoint: { select: { id: true } }, - organizationId: true, - projectId: true, - runtimeEnvironmentId: true, + const runs = await this.$.runStore.findRuns( + { + where: { id: { in: runIds } }, + select: { + id: true, + spanId: true, + status: true, + lockedAt: true, + ttl: true, + taskEventStore: true, + createdAt: true, + associatedWaitpoint: { select: { id: true } }, + organizationId: true, + projectId: true, + runtimeEnvironmentId: true, + }, + // read-your-writes: the queue slot is already claimed; a lagging replica would orphan the run }, - }); + this.$.prisma + ); // Filter runs that can be expired const runsToExpire: typeof runs = []; diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index f6b613a393f..97c0759e5e1 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -1244,6 +1244,11 @@ export class RoutingRunStore implements RunStore { args as Record ); const id = RoutingRunStore.#waitpointId((args as { where?: unknown }).where); + // A colocated lookup (no id, resolved via `coLocateWithRunId`) is the (env,idempotencyKey) dedup + // probe of createDateTimeWaitpoint/createManualWaitpoint: read-your-writes within the owning + // run/tree. On a retry it must observe attempt 1's just-written waitpoint to short-circuit, so it + // reads the owning store's PRIMARY — the replica can lag and miss it, re-arming/re-blocking the run. + const coLocatedDedup = id === undefined && opts?.coLocateWithRunId !== undefined; const store = id !== undefined ? await this.#resolveWaitpointStore(id, client !== undefined) @@ -1254,7 +1259,7 @@ export class RoutingRunStore implements RunStore { store !== undefined ? ((await store.findWaitpoint( scalarArgs as typeof args, - RoutingRunStore.#ownPrimary(store, client) + coLocatedDedup ? store.primaryReadClient : RoutingRunStore.#ownPrimary(store, client) )) as Record | null) : (((await this.#new.findWaitpoint( scalarArgs as typeof args, diff --git a/packages/redis-worker/src/mollifier/buffer.ts b/packages/redis-worker/src/mollifier/buffer.ts index d2281cd4332..179de94eca3 100644 --- a/packages/redis-worker/src/mollifier/buffer.ts +++ b/packages/redis-worker/src/mollifier/buffer.ts @@ -462,6 +462,23 @@ export class MollifierBuffer { await this.redis.releaseMollifierClaim(claimKey, `${PENDING_PREFIX}${input.token}`); } + // Reopen a RESOLVED claim slot whose winner was cleared, so the claim-loser + // reacquire path can re-serialise a cross-DB recreate through a fresh claim. + // Compare-and-delete on the observed `expectedRunId` (never an unconditional + // DEL) so a concurrent reacquirer that already re-published a NEW winner is + // never wiped. Reuses the lookup self-heal's compare-and-delete Lua. Returns + // true if the stale slot was cleared. + async resetResolvedClaim( + input: IdempotencyLookupInput & { expectedRunId: string } + ): Promise { + const claimKey = makeIdempotencyClaimKey(input); + const deleted = (await this.redis.delMollifierKeyIfEquals( + claimKey, + input.expectedRunId + )) as number; + return deleted === 1; + } + // Read the current claim value, used by the wait/poll loop on losers // to detect "pending" → "resolved" transitions and timeouts. async readClaim(input: IdempotencyLookupInput): Promise { From fb8fcfe619d8d17c79d61c56d17ee3ac814c29b1 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sat, 18 Jul 2026 17:35:20 +0100 Subject: [PATCH 2/3] chore(run-store): remove the unused #routeWaitpointWrite helper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dead since #4272 inlined per-write residency routing at each call site — the private helper had zero callers (only doc-comments referenced it by name), tripping eslint no-unused-private-class-members. Remove it and reword the comments that referenced it. --- .../run-store/src/runOpsStore.ts | 22 ++++--------------- 1 file changed, 4 insertions(+), 18 deletions(-) diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 97c0759e5e1..56f6addebdf 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -138,20 +138,6 @@ export class RoutingRunStore implements RunStore { return this.#routeOrNew(runId).runInTransaction(runId, fn); } - // A waitpoint WRITE co-locates with its run by id-shape (cuid → LEGACY, run-ops id → NEW, - // unclassifiable → LEGACY), mirroring how `blockRunWithWaitpointEdges` routes the edge by - // run id. The caller's `tx` is never forwarded: a routed write must run on the OWNING store's - // OWN client so the row lands in that store's database (same-store atomicity comes from the - // owning store opening its own transaction, never from a caller-supplied one). - #routeWaitpointWrite( - id: string | undefined, - _tx?: PrismaClientOrTransaction - ): { store: RunStore; tx?: PrismaClientOrTransaction } { - const store = - typeof id === "string" && this.#classifySafe(id) === "NEW" ? this.#new : this.#legacy; - return { store, tx: undefined }; - } - // Resolve which store ACTUALLY holds a waitpoint id: drain-on-read can relocate a cuid // waitpoint onto NEW while keeping its id, so probe the id-shape's home then the other. // `onPrimary` probes each store's own primary (read-your-writes callers; a fresh row may not @@ -1187,7 +1173,7 @@ export class RoutingRunStore implements RunStore { opts?.residency, RoutingRunStore.#waitpointId(data) ); - // Never forward the caller's tx into a routed write (matches #routeWaitpointWrite). + // Never forward the caller's tx into a routed write (it runs on the owning store's own client). return store.createWaitpoint(args, undefined); } @@ -1706,7 +1692,7 @@ export class RoutingRunStore implements RunStore { ): Promise { // Route by the batch's classifiable internal id: run-ops id→NEW, cuid→LEGACY. The caller's // `tx` is never forwarded — the create runs on the owning store's own client so the batch and - // its co-resident child runs/items land on the same DB. Mirrors #routeWaitpointWrite / + // its co-resident child runs/items land on the same DB. Mirrors the by-id waitpoint-write routing / // updateBatchTaskRun. const store = await this.#routeOrNewForWrite(data.id); return store.createBatchTaskRun(data, undefined); @@ -1723,7 +1709,7 @@ export class RoutingRunStore implements RunStore { const id = typeof args.where.id === "string" ? args.where.id : (args.where.friendlyId ?? undefined); // The caller's `tx` is never forwarded — the update runs on the owning store's own client so - // it targets the DB the batch actually lives on. Mirrors #routeWaitpointWrite. + // it targets the DB the batch actually lives on. Mirrors the by-id waitpoint-write routing. const store = this.#routeOrNew(id); return store.updateBatchTaskRun(args, undefined); } @@ -1885,7 +1871,7 @@ export class RoutingRunStore implements RunStore { // WaitpointTag — a standalone entity (no run/waitpoint FK) keyed by (environmentId, name). // --------------------------------------------------------------------------- - // Callers never mint a tag id (defaults to cuid), so #routeWaitpointWrite always resolves LEGACY + // Callers never mint a tag id (defaults to cuid), so a tag write always resolves LEGACY // today — deliberately single-homed, like standalone waitpoint tokens. If tag-id minting is ever made // residency-aware, findManyWaitpointTags must de-dupe by (environmentId, name) or names will duplicate. upsertWaitpointTag( From 1c625e17e9515316e639601d5beeca30736babe5 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Sun, 19 Jul 2026 16:08:32 +0100 Subject: [PATCH 3/3] =?UTF-8?q?fix(webapp,run-store):=20review=20follow-up?= =?UTF-8?q?s=20=E2=80=94=20split=20idempotency=20+=20realtime=20batch=20ha?= =?UTF-8?q?rdening?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Production follow-ups from the CodeRabbit/Claude/Devin review. Caller-driven guard tests are in the stacked tests PR; the two test edits here are coupled to the production change (the publishClaim signature change invalidates a main assertion; the fan-out removal obsoletes a main test). - Claim TTL floor (TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS, default 5) independent of the customer key TTL, so a short key TTL can't expire the claim mid-pipeline and let a loser re-claim. - publishClaim returns the buffer CAS result; the trigger success path detects + logs a no-op'd publish. - reacquireClearedGlobalWinner fails closed with a retryable 503 (exhaustion / unfindable winner); the expired/failed clear-and-recreate path routes through it too (Devin), so that recreate is serialised. - Realtime batch route re-reads the owning primary on a replica miss (backend-agnostic; closes the Electric ShapeStream permanent-404 for self-hosters). - Remove the classifiable-id cross-store fan-out from findRun: a run's id-shape fixes its residency for life, so the single-store read is correct and the fan-out was dead code. --- apps/webapp/app/env.server.ts | 3 + .../routes/realtime.v1.batches.$batchId.ts | 15 +- .../concerns/idempotencyKeys.server.ts | 70 ++++- .../runEngine/services/triggerTask.server.ts | 13 +- apps/webapp/app/v3/mollifier/claimTtl.ts | 12 + .../v3/mollifier/idempotencyClaim.server.ts | 10 +- .../resolveBatchForRealtime.server.ts | 24 ++ .../test/mollifierIdempotencyClaim.test.ts | 8 +- ...OpsStore.residencyMismatchFallback.test.ts | 296 ------------------ .../run-store/src/runOpsStore.ts | 33 +- 10 files changed, 128 insertions(+), 356 deletions(-) create mode 100644 apps/webapp/app/v3/mollifier/claimTtl.ts create mode 100644 apps/webapp/app/v3/realtime/resolveBatchForRealtime.server.ts delete mode 100644 internal-packages/run-store/src/runOpsStore.residencyMismatchFallback.test.ts diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 4e2d4c5845f..973560ca07f 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -1311,6 +1311,9 @@ const EnvironmentSchema = z // claim TTL), how long a waiter blocks before timing out, and the // waiter poll interval. TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS: z.coerce.number().int().positive().default(30), + // Pipeline floor: the claim never shrinks below this even for a short customer key TTL, so it + // can't expire mid-pipeline and let a loser re-claim (cross-DB duplicate under the split). + TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS: z.coerce.number().int().positive().default(5), TRIGGER_MOLLIFIER_CLAIM_WAIT_MS: z.coerce.number().int().positive().default(5_000), TRIGGER_MOLLIFIER_CLAIM_POLL_MS: z.coerce.number().int().positive().default(25), diff --git a/apps/webapp/app/routes/realtime.v1.batches.$batchId.ts b/apps/webapp/app/routes/realtime.v1.batches.$batchId.ts index b95cd4491db..8a761b32f2e 100644 --- a/apps/webapp/app/routes/realtime.v1.batches.$batchId.ts +++ b/apps/webapp/app/routes/realtime.v1.batches.$batchId.ts @@ -2,7 +2,7 @@ import { z } from "zod"; import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server"; import { resolveRealtimeStreamClient } from "~/services/realtime/resolveRealtimeStreamClient.server"; import { anyResource, createLoaderApiRoute } from "~/services/routeBuilders/apiBuilder.server"; -import { runStore } from "~/v3/runStore.server"; +import { resolveBatchTaskRunForRealtime } from "~/v3/realtime/resolveBatchForRealtime.server"; const ParamsSchema = z.object({ batchId: z.string(), @@ -13,14 +13,13 @@ export const loader = createLoaderApiRoute( params: ParamsSchema, allowJWT: true, corsStrategy: "all", - // A just-created batch may not yet have replicated to the read replica this client-less - // findBatchTaskRunByFriendlyId lookup routes to; return a retryable 404 so the SDK retries through - // replica lag rather than stranding a live batch on a permanent 404 (mirrors the run-get routes, - // e.g. api.v3.runs.$runId). + // A just-created batch may not yet have replicated to the read replica the client-less lookup uses. + // shouldRetryNotFound stamps a retryable 404 for the zodfetch GET; the realtime resolver ALSO + // re-reads the owning primary on a replica miss, so the Electric ShapeStream consumer (which ignores + // x-should-retry) doesn't strand a live batch on a permanent 404. Mirrors the run-get routes. shouldRetryNotFound: true, - findResource: (params, auth) => { - return runStore.findBatchTaskRunByFriendlyId(params.batchId, auth.environment.id); - }, + findResource: (params, auth) => + resolveBatchTaskRunForRealtime(params.batchId, auth.environment.id), authorization: { action: "read", // See sibling note in api.v1.batches.$batchId.ts — `{type: "runs"}` diff --git a/apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts b/apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts index 561e6a5d476..f6696865e94 100644 --- a/apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts +++ b/apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts @@ -9,6 +9,7 @@ import { shouldIdempotencyKeyBeCleared } from "~/v3/taskStatus"; import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server"; import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server"; import { claimOrAwait, resetResolvedClaim } from "~/v3/mollifier/idempotencyClaim.server"; +import { computeClaimTtlSeconds } from "~/v3/mollifier/claimTtl"; import { makeResolveMollifierFlag } from "~/v3/mollifier/mollifierGate.server"; import { runStore } from "~/v3/runStore.server"; import { runOpsLegacyPrisma, runOpsNewPrisma } from "~/db.server"; @@ -180,6 +181,14 @@ export class IdempotencyKeyConcern { } ); + // `global`-scope (or scope-absent) keys under the split have no per-run salt, so the Redis claim is + // their only cross-DB dedup mutex. Computed here (not just in the claim block below) because the + // expired/failed clear-and-recreate path must serialise through it too. + const idempotencyKeyScope = request.body.options?.idempotencyKeyOptions?.scope; + const globalUnderSplit = + (idempotencyKeyScope === "global" || idempotencyKeyScope === undefined) && + (await isSplitEnabled()); + const existingRun = idempotencyKey ? await runStore.findRun( { @@ -216,11 +225,37 @@ export class IdempotencyKeyConcern { } if (existingRun) { - return await this.handleExistingRun(request, parentStore, existingRun, { + const handled = await this.handleExistingRun(request, parentStore, existingRun, { idempotencyKey, idempotencyKeyExpiresAt, dedupClient, }); + // A LIVE cached hit (or andWait waitpoint wiring) is terminal. + if (handled.isCached) { + return handled; + } + // isCached === false → the existing run was EXPIRED/FAILED, so handleExistingRun cleared its key + // and we must recreate. For a global-scope key under split that recreate has to be claim- + // serialised too — otherwise two concurrent cross-residency recreates each create a run the + // per-DB unique index can't dedup (the same hole reacquireClearedGlobalWinner closes on the + // claim-loser path). Non-split / non-global: the plain unserialised recreate is safe. + if (globalUnderSplit) { + return await this.reacquireClearedGlobalWinner(request, parentStore, { + idempotencyKey, + idempotencyKeyExpiresAt, + dedupClient, + ttlSeconds: computeClaimTtlSeconds({ + keyExpiresAt: idempotencyKeyExpiresAt, + now: Date.now(), + minTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS, + maxTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS, + }), + clearedRunId: existingRun.friendlyId, + safetyNetMs: env.TRIGGER_MOLLIFIER_CLAIM_WAIT_MS, + pollStepMs: env.TRIGGER_MOLLIFIER_CLAIM_POLL_MS, + }); + } + return handled; } // Pre-gate claim — closes the PG+buffer race during gate transition. @@ -247,11 +282,8 @@ export class IdempotencyKeyConcern { // older SDK) is treated conservatively as possibly-global — harmless for a // real run/attempt key, whose hash already embeds the parent id so two // parents mint DISTINCT keys that never share a claim slot. - const idempotencyKeyScope = request.body.options?.idempotencyKeyOptions?.scope; - const globalUnderSplit = - (idempotencyKeyScope === "global" || idempotencyKeyScope === undefined) && - (await isSplitEnabled()); - + // (idempotencyKeyScope / globalUnderSplit are computed above — they also gate the expired/failed + // recreate serialisation.) const claimEligible = !request.body.options?.debounce && !request.options?.oneTimeUseToken && @@ -268,13 +300,12 @@ export class IdempotencyKeyConcern { | undefined) ?? null, })))); if (claimEligible) { - const ttlSeconds = Math.max( - 1, - Math.min( - env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS, - Math.ceil((idempotencyKeyExpiresAt.getTime() - Date.now()) / 1000) - ) - ); + const ttlSeconds = computeClaimTtlSeconds({ + keyExpiresAt: idempotencyKeyExpiresAt, + now: Date.now(), + minTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS, + maxTtlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS, + }); const outcome = await claimOrAwait({ envId: request.environment.id, taskIdentifier: request.taskId, @@ -519,8 +550,8 @@ export class IdempotencyKeyConcern { // (returning its claim to publish); the rest resolve to the fresh run. If a // re-claim resolves to ANOTHER cleared winner we advance and loop, bounded // by MAX_CLEARED_WINNER_REACQUIRES; on exhaustion (or an unfindable - // resolution) we fall open to the create, matching the initial claim's - // fail-open posture (PG unique index as backstop). + // resolution) we fail CLOSED with a retryable 503 so the SDK retry re-serialises, + // rather than fall open to an unserialised cross-DB create. private async reacquireClearedGlobalWinner( request: TriggerTaskRequest, parentStore: string | undefined, @@ -584,7 +615,14 @@ export class IdempotencyKeyConcern { } staleRunId = outcome.runId; } - return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt }; + // Exhausted the bounded reacquires (or the winner was unfindable). Rather than fall through to an + // UNSERIALISED create — which under global-scope-split can dual-create across DBs (the per-DB unique + // index can't dedup cross-residency) — fail closed with a retryable 503 so the SDK retry re-serialises + // through a fresh claim (mirrors the timed_out branch above). + throw new ServiceValidationError( + "Idempotency claim could not be re-serialised after repeated cleared winners", + 503 + ); } // Resolve a claim winner (a run friendlyId) across both split DBs. Classify diff --git a/apps/webapp/app/runEngine/services/triggerTask.server.ts b/apps/webapp/app/runEngine/services/triggerTask.server.ts index f4981fece16..ff22a479a1d 100644 --- a/apps/webapp/app/runEngine/services/triggerTask.server.ts +++ b/apps/webapp/app/runEngine/services/triggerTask.server.ts @@ -752,7 +752,7 @@ export class RunEngineTriggerTaskService { // Pipeline returned successfully — publish the claim if we held // one. Waiters polling for our key resolve to this runId. if (idempotencyClaim && result?.run?.friendlyId) { - await publishMollifierClaim({ + const published = await publishMollifierClaim({ envId: idempotencyClaim.envId, taskIdentifier: idempotencyClaim.taskIdentifier, idempotencyKey: idempotencyClaim.idempotencyKey, @@ -760,6 +760,17 @@ export class RunEngineTriggerTaskService { runId: result.run.friendlyId, ttlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS, }); + if (!published) { + // Our claim expired mid-pipeline and another claimant took it, so this publish no-op'd: a + // different run is now canonical for this key while we return ours (a cross-DB dup under the + // split). Rare now the claim TTL is floored (C1); surfaced for monitoring pending auto- + // convergence (re-resolve the current winner + cancel this orphan). + logger.warn("mollifier claim publish no-op'd; winner lost the claim mid-pipeline", { + envId: idempotencyClaim.envId, + taskIdentifier: idempotencyClaim.taskIdentifier, + runId: result.run.friendlyId, + }); + } } return result; } catch (err) { diff --git a/apps/webapp/app/v3/mollifier/claimTtl.ts b/apps/webapp/app/v3/mollifier/claimTtl.ts new file mode 100644 index 00000000000..7fc8687cec7 --- /dev/null +++ b/apps/webapp/app/v3/mollifier/claimTtl.ts @@ -0,0 +1,12 @@ +// The claim is a serialization lock that must outlive the winner's create-and-publish pipeline. A short +// customer key TTL must NOT shrink it below a pipeline floor (else the claim expires mid-pipeline and a +// polling loser re-claims → cross-DB duplicate). Floor at `minTtlSeconds` independent of key TTL; cap at max. +export function computeClaimTtlSeconds(input: { + keyExpiresAt: Date; + now: number; + minTtlSeconds: number; + maxTtlSeconds: number; +}): number { + const keyTtlSeconds = Math.ceil((input.keyExpiresAt.getTime() - input.now) / 1000); + return Math.min(input.maxTtlSeconds, Math.max(input.minTtlSeconds, keyTtlSeconds)); +} diff --git a/apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts b/apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts index 196017b438d..191ff62058b 100644 --- a/apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts +++ b/apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts @@ -163,12 +163,14 @@ export async function publishClaim(input: { runId: string; ttlSeconds?: number; buffer?: MollifierBuffer | null; -}): Promise { +}): Promise { const buffer = input.buffer === undefined ? getMollifierBuffer() : input.buffer; - if (!buffer) return; + if (!buffer) return true; const ttlSeconds = input.ttlSeconds ?? DEFAULT_CLAIM_TTL_SECONDS; try { - await buffer.publishClaim({ + // false = compare-and-set no-op: a stale claimant (our TTL expired) already moved in, so our + // publish did NOT set the winner. The caller decides how to converge. + return await buffer.publishClaim({ envId: input.envId, taskIdentifier: input.taskIdentifier, idempotencyKey: input.idempotencyKey, @@ -182,6 +184,8 @@ export async function publishClaim(input: { taskIdentifier: input.taskIdentifier, err: err instanceof Error ? err.message : String(err), }); + // Unknown publish state (transient error) — the claim TTL is the safety net; don't signal a no-op. + return true; } } diff --git a/apps/webapp/app/v3/realtime/resolveBatchForRealtime.server.ts b/apps/webapp/app/v3/realtime/resolveBatchForRealtime.server.ts new file mode 100644 index 00000000000..edd318f9230 --- /dev/null +++ b/apps/webapp/app/v3/realtime/resolveBatchForRealtime.server.ts @@ -0,0 +1,24 @@ +import { prisma } from "~/db.server"; +import { runStore } from "~/v3/runStore.server"; + +type BatchStore = Pick; + +// The realtime batch route reads the batch client-less (replica), which can miss a just-created batch +// under replica lag. `shouldRetryNotFound` covers the zodfetch GET, but the Electric ShapeStream +// consumer (self-hosters) ignores `x-should-retry`, so re-read the owning primary on a miss — passing a +// non-replica writer flips each store leg to its own primary — to avoid a permanent 404. +export function resolveBatchTaskRunForRealtime( + friendlyId: string, + environmentId: string, + deps?: { store?: BatchStore; writer?: unknown } +) { + const store = deps?.store ?? runStore; + const writer = deps?.writer ?? prisma; + return store + .findBatchTaskRunByFriendlyId(friendlyId, environmentId) + .then( + (onReplica) => + onReplica ?? + store.findBatchTaskRunByFriendlyId(friendlyId, environmentId, undefined, writer as never) + ); +} diff --git a/apps/webapp/test/mollifierIdempotencyClaim.test.ts b/apps/webapp/test/mollifierIdempotencyClaim.test.ts index 1b093250f1e..84083087a4b 100644 --- a/apps/webapp/test/mollifierIdempotencyClaim.test.ts +++ b/apps/webapp/test/mollifierIdempotencyClaim.test.ts @@ -181,13 +181,13 @@ describe("publishClaim", () => { expect(buffer.publishClaim).toHaveBeenCalledOnce(); }); - it("no-op when buffer is null", async () => { + it("no buffer → true (nothing to converge, not a no-op'd publish)", async () => { await expect( publishClaim({ ...baseInput, token: "owner-token", runId: "run_X", buffer: null }) - ).resolves.toBeUndefined(); + ).resolves.toBe(true); }); - it("swallows errors so trigger pipeline isn't broken by Redis hiccups", async () => { + it("swallows errors so trigger pipeline isn't broken by Redis hiccups (returns true, unknown state)", async () => { const buffer = { publishClaim: vi.fn(async () => { throw new Error("ECONNREFUSED"); @@ -195,7 +195,7 @@ describe("publishClaim", () => { } as unknown as MollifierBuffer; await expect( publishClaim({ ...baseInput, token: "owner-token", runId: "run_X", buffer }) - ).resolves.toBeUndefined(); + ).resolves.toBe(true); }); }); diff --git a/internal-packages/run-store/src/runOpsStore.residencyMismatchFallback.test.ts b/internal-packages/run-store/src/runOpsStore.residencyMismatchFallback.test.ts deleted file mode 100644 index 8897c295662..00000000000 --- a/internal-packages/run-store/src/runOpsStore.residencyMismatchFallback.test.ts +++ /dev/null @@ -1,296 +0,0 @@ -// RED→GREEN lock for RoutingRunStore.findRun's ON-MISS FAN-OUT for a CLASSIFIABLE id. -// -// THE BUG: findRun for a classifiable id routed to the SINGLE owning store (by id-shape -// classification) and returned whatever it gave — no on-miss fallback. When a run's PHYSICAL -// residency does not match its id-shape classification (e.g. a pre-#4154 27-char base62 run that -// lives on the NEW store but now classifies LEGACY), findRun routed to the wrong store, missed, -// and returned null → a spurious 404 — even though the run is physically present on the OTHER DB -// (and runs.list surfaces it). The unclassifiable path already fanned out NEW→LEGACY; this makes -// the classifiable path equally robust. -// -// Uses the REAL two-physical-DB split (heteroRunOpsPostgresTest: prisma14 = full/legacy on PG14, -// prisma17 = dedicated run-ops subset on PG17). NEVER mocked. The residency/classification MISMATCH -// is simulated deterministically by injecting a custom `classify` fn into the RoutingRunStore -// constructor — the physical row is written to the NEW store while `classify` reports its id LEGACY. - -import { heteroRunOpsPostgresTest } from "@internal/testcontainers"; -import { ownerEngine, type Residency } from "@trigger.dev/core/v3/isomorphic"; -import type { PrismaClient } from "@trigger.dev/database"; -import type { RunOpsPrismaClient } from "@internal/run-ops-database"; -import { describe, expect } from "vitest"; -import { PostgresRunStore } from "./PostgresRunStore.js"; -import { RoutingRunStore } from "./runOpsStore.js"; -import type { CreateRunInput, RunStore } from "./types.js"; - -// ownerEngine classifies by internal-id LENGTH/version char: 25 chars → cuid → LEGACY, -// a v1 body (version "1" at index 25) → run-ops id → NEW. -function cuidLegacy(seed: string): string { - return (seed + "c".repeat(25)).slice(0, 25); -} -function runOpsNew(seed: string): string { - return (seed.replace(/[^0-9a-v]/g, "0") + "k".repeat(24)).slice(0, 24) + "01"; -} - -function makeDedicatedStore(prisma17: RunOpsPrismaClient) { - return new PostgresRunStore({ - prisma: prisma17 as never, - readOnlyPrisma: prisma17 as never, - schemaVariant: "dedicated", - }); -} - -function makeLegacyStore(prisma14: PrismaClient) { - return new PostgresRunStore({ - prisma: prisma14, - readOnlyPrisma: prisma14, - schemaVariant: "legacy", - }); -} - -// Wrap a real store so findRun/findRunOnPrimary calls are COUNTED while every method still delegates -// to the REAL PostgresRunStore (this is instrumentation, not a behavior mock — the underlying reads, -// writes, getters all run for real). Lets us assert the FAST PATH does not touch the other store. -function countingReads( - inner: RunStore, - counts: { findRun: number; findRunOnPrimary: number } -): RunStore { - return new Proxy(inner, { - get(target, prop) { - // Read via target[prop] so getters (e.g. primaryReadClient) run with `this` = the real store. - const value = (target as unknown as Record)[prop]; - if (typeof value !== "function") return value; - if (prop === "findRun" || prop === "findRunOnPrimary") { - return (...args: unknown[]) => { - counts[prop as "findRun" | "findRunOnPrimary"] += 1; - return (value as (...a: unknown[]) => unknown).apply(target, args); - }; - } - return (value as (...a: unknown[]) => unknown).bind(target); - }, - }) as unknown as RunStore; -} - -async function seedLegacyEnvironment(prisma14: PrismaClient, suffix: string) { - const organization = await prisma14.organization.create({ - data: { title: `Org ${suffix}`, slug: `org-${suffix}` }, - }); - const project = await prisma14.project.create({ - data: { - name: `Project ${suffix}`, - slug: `project-${suffix}`, - externalRef: `proj_${suffix}`, - organizationId: organization.id, - }, - }); - const environment = await prisma14.runtimeEnvironment.create({ - data: { - type: "DEVELOPMENT", - slug: "dev", - projectId: project.id, - organizationId: organization.id, - apiKey: `tr_dev_${suffix}`, - pkApiKey: `pk_dev_${suffix}`, - shortcode: `short_${suffix}`, - }, - }); - return { - organizationId: organization.id, - projectId: project.id, - runtimeEnvironmentId: environment.id, - environmentId: environment.id, - }; -} - -function buildCreateRunInput(params: { - runId: string; - friendlyId: string; - organizationId: string; - projectId: string; - runtimeEnvironmentId: string; -}): CreateRunInput { - return { - data: { - id: params.runId, - engine: "V2", - status: "PENDING", - friendlyId: params.friendlyId, - runtimeEnvironmentId: params.runtimeEnvironmentId, - environmentType: "DEVELOPMENT", - organizationId: params.organizationId, - projectId: params.projectId, - taskIdentifier: "my-task", - payload: "{}", - payloadType: "application/json", - traceContext: {}, - traceId: `trace_${params.runId}`, - spanId: `span_${params.runId}`, - queue: "task/my-task", - isTest: false, - taskEventStore: "taskEvent", - depth: 0, - }, - snapshot: { - engine: "V2", - executionStatus: "RUN_CREATED", - description: "Run was created", - runStatus: "PENDING", - environmentId: params.runtimeEnvironmentId, - environmentType: "DEVELOPMENT", - projectId: params.projectId, - organizationId: params.organizationId, - }, - }; -} - -// Insert a TaskRun row DIRECTLY onto the NEW (dedicated) store, bypassing routing, so we can force a -// residency/classification MISMATCH: the row is physically on #new while `classify` calls its id LEGACY. -async function insertRunOnNewStore( - prisma17: RunOpsPrismaClient, - params: { - runId: string; - friendlyId: string; - environmentId: string; - organizationId: string; - projectId: string; - } -) { - await prisma17.taskRun.create({ - data: { - id: params.runId, - engine: "V2", - status: "PENDING", - friendlyId: params.friendlyId, - runtimeEnvironmentId: params.environmentId, - environmentType: "DEVELOPMENT", - organizationId: params.organizationId, - projectId: params.projectId, - taskIdentifier: "my-task", - payload: "{}", - payloadType: "application/json", - traceContext: {}, - traceId: `trace_${params.runId}`, - spanId: `span_${params.runId}`, - queue: "task/my-task", - isTest: false, - taskEventStore: "taskEvent", - depth: 0, - }, - }); -} - -describe("RoutingRunStore.findRun — on-miss fan-out for a classifiable id (residency ≠ classification)", () => { - // ── THE BUG: a run physically on #new whose id classifies LEGACY must still be found ── - // Without the on-miss fallback, findRun routes to #legacy (per classification), misses, returns null. - heteroRunOpsPostgresTest( - "returns a #new-resident run whose id classifies LEGACY (owning-store miss → other-store fallback)", - async ({ prisma14, prisma17 }) => { - const env = await seedLegacyEnvironment(prisma14, "mm1"); - const newStore = makeDedicatedStore(prisma17); - const legacyStore = makeLegacyStore(prisma14); - - // A run-ops-shaped id (so ownerEngine would say NEW), but we FORCE classify → LEGACY to model - // a residency/classification mismatch: physically on #new, classified LEGACY. - const mismatchId = runOpsNew("mm1"); - const classify = (id: string): Residency => (id === mismatchId ? "LEGACY" : ownerEngine(id)); - const router = new RoutingRunStore({ new: newStore, legacy: legacyStore, classify }); - - await insertRunOnNewStore(prisma17, { - runId: mismatchId, - friendlyId: "run_mm1", - environmentId: env.environmentId, - organizationId: env.organizationId, - projectId: env.projectId, - }); - - // Physical residency sanity: on #new only. - expect(await prisma17.taskRun.findUnique({ where: { id: mismatchId } })).not.toBeNull(); - expect(await prisma14.taskRun.findUnique({ where: { id: mismatchId } })).toBeNull(); - - // classify → LEGACY routes to #legacy (miss); the fix falls back to #new and finds the run. - const byId = (await router.findRun({ id: mismatchId }, { select: { id: true } })) as Record< - string, - unknown - > | null; - expect(byId?.id).toBe(mismatchId); - - // Same on the read-your-writes primary variant (a caller-passed writer → findRunOnPrimary). - const byIdPrimary = (await router.findRun( - { id: mismatchId }, - { select: { id: true } }, - prisma14 - )) as Record | null; - expect(byIdPrimary?.id).toBe(mismatchId); - } - ); - - // ── FAST PATH: a run found in its CLASSIFIED store is a SINGLE read (no second-store probe) ── - heteroRunOpsPostgresTest( - "does NOT read the other store when the classified (owning) store hits", - async ({ prisma14, prisma17 }) => { - const env = await seedLegacyEnvironment(prisma14, "mm2"); - - // NEW-resident run-ops-id run: owning store = #new. Wrap #legacy to catch any stray probe. - const newCounts = { findRun: 0, findRunOnPrimary: 0 }; - const legacyCounts = { findRun: 0, findRunOnPrimary: 0 }; - const newStore = countingReads(makeDedicatedStore(prisma17), newCounts); - const legacyStore = countingReads(makeLegacyStore(prisma14), legacyCounts); - const router = new RoutingRunStore({ new: newStore, legacy: legacyStore }); - - const newId = runOpsNew("mm2n"); // classifies NEW - await router.createRun( - buildCreateRunInput({ - runId: newId, - friendlyId: "run_mm2_new", - organizationId: env.organizationId, - projectId: env.projectId, - runtimeEnvironmentId: env.runtimeEnvironmentId, - }) - ); - - const hit = (await router.findRun({ id: newId }, { select: { id: true } })) as Record< - string, - unknown - > | null; - expect(hit?.id).toBe(newId); - // Owning store read exactly once; the other store NOT touched (fast path preserved). - expect(newCounts.findRun).toBe(1); - expect(legacyCounts.findRun).toBe(0); - expect(legacyCounts.findRunOnPrimary).toBe(0); - - // Symmetric: a cuid run whose owning store is #legacy must not probe #new on a hit. - const legacyId = cuidLegacy("mm2l"); // classifies LEGACY - await router.createRun( - buildCreateRunInput({ - runId: legacyId, - friendlyId: "run_mm2_legacy", - organizationId: env.organizationId, - projectId: env.projectId, - runtimeEnvironmentId: env.runtimeEnvironmentId, - }) - ); - newCounts.findRun = 0; - legacyCounts.findRun = 0; - - const hitLegacy = (await router.findRun( - { id: legacyId }, - { select: { id: true } } - )) as Record | null; - expect(hitLegacy?.id).toBe(legacyId); - expect(legacyCounts.findRun).toBe(1); - expect(newCounts.findRun).toBe(0); - } - ); - - // ── A genuine miss on BOTH stores still returns null (fan-out exhausted) ── - heteroRunOpsPostgresTest( - "returns null when the run is on neither store", - async ({ prisma14, prisma17 }) => { - const newStore = makeDedicatedStore(prisma17); - const legacyStore = makeLegacyStore(prisma14); - const router = new RoutingRunStore({ new: newStore, legacy: legacyStore }); - expect( - await router.findRun({ id: cuidLegacy("ghost") }, { select: { id: true } }) - ).toBeNull(); - } - ); -}); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 56f6addebdf..d14ae525224 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -229,9 +229,11 @@ export class RoutingRunStore implements RunStore { const onPrimary = readYourWrites(argsOrClient, _client); const id = idFromWhere(where); if (id !== undefined) { - // Residency-classifiable (id/friendlyId): route to the owning store, then fall back to the - // OTHER store on a miss. - return this.#findRunRouted(id, where, args, onPrimary); + // Residency-classifiable (id/friendlyId): id-shape is destiny for a run's whole life — a run-ops + // id lives on NEW, a cuid on LEGACY — so read the one owning store (no cross-store fan-out). + const store = this.#routeOrNew(id); + const method = onPrimary ? "findRunOnPrimary" : "findRun"; + return (store[method] as (...rest: unknown[]) => Promise)(where, args); } // Unclassifiable where (e.g. spanId, idempotencyKey): the run may live on either DB, // so fan out NEW-first then LEGACY rather than defaulting to NEW — defaulting silently @@ -239,31 +241,6 @@ export class RoutingRunStore implements RunStore { return this.#findRunUnrouted(where, args, onPrimary); } - // A classifiable id names its OWNING store by id-shape, but physical residency can diverge from - // classification (a pre-#4154 base62 run lives on #new yet classifies LEGACY). Read the owning - // store first — a hit is a SINGLE read (the fast path) — then, ONLY on a miss, probe the OTHER - // store before returning null, so a run whose residency ≠ its id-shape is found rather than 404'd. - // Both legs run the SAME index-covered TaskRun lookup; the fan-out cost is paid only on the (rare) - // miss. Mirrors #findRunUnrouted's shape but keyed on the classified owner. - async #findRunRouted( - id: string, - where: Prisma.TaskRunWhereInput, - args: unknown, - onPrimary: boolean - ): Promise { - const method = onPrimary ? "findRunOnPrimary" : "findRun"; - const owning = this.#routeOrNew(id); - const fromOwning = await (owning[method] as (...rest: unknown[]) => Promise)( - where, - args - ); - if (fromOwning != null) { - return fromOwning; - } - const other = owning === this.#new ? this.#legacy : this.#new; - return (other[method] as (...rest: unknown[]) => Promise)(where, args); - } - async #findRunUnrouted( where: Prisma.TaskRunWhereInput, args: unknown,