fix(webapp): harden run-ops split idempotency + realtime batch reads

Follow-ups from review of the read-your-writes / idempotency work. Tests are
included here for now and will be split into the stacked tests PR.

- Claim TTL floor: the mollifier claim TTL was min(max, keyTTL), so a short customer
  idempotencyKeyTTL (1-2s) could expire the claim mid-pipeline and let a polling
  loser re-claim -> cross-DB duplicate under the split. Floor it at a pipeline bound
  (TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS, default 5), independent of the key TTL.
- Publish compare-and-set: publishClaim returns the buffer CAS result; the trigger
  success path detects a no-op'd publish (claim expired, another claimant moved in)
  and logs it rather than silently treating its own run as canonical.
- Reacquire exhaustion: reacquireClearedGlobalWinner failed OPEN on exhausting the
  bounded cleared-winner reacquires (or an unfindable winner), creating an
  unserialised run that can dual-create across DBs. Fail CLOSED with a retryable 503
  so the SDK retry re-serialises (mirrors the timed_out branch).
- Realtime batch route: findResource re-reads the owning primary on a replica miss,
  so the Electric ShapeStream consumer (self-hosters), which ignores the
  retryable-404 header, doesn't strand a live just-created batch on a permanent 404.

Author: d-cs

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),
 

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"}`

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";
@@ -268,13 +269,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 +519,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 +584,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

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -752,14 +752,25 @@ 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,
           token: idempotencyClaim.token,
           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) {

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));
+}

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<void> {
+}): Promise<boolean> {
   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;
   }
 }
 

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<typeof runStore, "findBatchTaskRunByFriendlyId">;
+
+// 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)
+    );
+}

apps/webapp/test/claimTtl.test.ts

@@ -0,0 +1,55 @@
+import { describe, expect, it } from "vitest";
+import { computeClaimTtlSeconds } from "~/v3/mollifier/claimTtl";
+
+// The mollifier idempotency claim is a serialization lock that must outlive the winner's
+// create-and-publish pipeline. If its TTL is derived from the customer's key TTL alone, a short key
+// TTL expires the claim mid-pipeline and a polling loser re-claims → a second creator → cross-DB dup.
+describe("computeClaimTtlSeconds", () => {
+  const now = Date.parse("2026-01-01T00:00:00.000Z");
+  const minTtlSeconds = 5;
+  const maxTtlSeconds = 30;
+
+  it("floors a short key TTL at the pipeline minimum (claim can't expire mid-pipeline)", () => {
+    expect(
+      computeClaimTtlSeconds({
+        keyExpiresAt: new Date(now + 2_000),
+        now,
+        minTtlSeconds,
+        maxTtlSeconds,
+      })
+    ).toBe(5);
+  });
+
+  it("caps a long key TTL at the maximum", () => {
+    expect(
+      computeClaimTtlSeconds({
+        keyExpiresAt: new Date(now + 3_600_000),
+        now,
+        minTtlSeconds,
+        maxTtlSeconds,
+      })
+    ).toBe(30);
+  });
+
+  it("uses the key TTL when it sits between the floor and the cap", () => {
+    expect(
+      computeClaimTtlSeconds({
+        keyExpiresAt: new Date(now + 12_000),
+        now,
+        minTtlSeconds,
+        maxTtlSeconds,
+      })
+    ).toBe(12);
+  });
+
+  it("floors an already-expired key at the minimum", () => {
+    expect(
+      computeClaimTtlSeconds({
+        keyExpiresAt: new Date(now - 1_000),
+        now,
+        minTtlSeconds,
+        maxTtlSeconds,
+      })
+    ).toBe(5);
+  });
+});

apps/webapp/test/mollifierClaimResolution.test.ts

@@ -156,4 +156,30 @@ describe("IdempotencyKeyConcern · claim resolution", () => {
       h.orgFlag = true; // restore for any later tests in this file
     }
   });
+
+  it("floors the claim TTL for a short idempotency-key TTL so the claim can't expire mid-pipeline", async () => {
+    // A 2s customer key TTL must NOT shrink the claim below the pipeline floor — else the claim expires
+    // while the winner is still creating and a polling loser re-claims (cross-DB dup under the split).
+    // Capture the ttlSeconds the concern hands the buffer's claim.
+    let capturedTtl: number | undefined;
+    h.buffer = {
+      claimIdempotency: vi.fn(async (input: { ttlSeconds: number }) => {
+        capturedTtl = input.ttlSeconds;
+        return { kind: "claimed" as const };
+      }),
+      lookupIdempotency: vi.fn(async () => null),
+    } as unknown as MollifierBuffer;
+
+    const findFirst = vi.fn(async () => null);
+    const concern = makeConcern({ findFirst });
+
+    const request = makeRequest();
+    (request.body.options as { idempotencyKeyTTL?: string }).idempotencyKeyTTL = "2s";
+
+    const result = await concern.handleTriggerRequest(request, undefined);
+
+    expect(result.isCached).toBe(false);
+    // Floored at TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS (default 5), NOT the ~2s key TTL.
+    expect(capturedTtl).toBeGreaterThanOrEqual(5);
+  });
 });

apps/webapp/test/mollifierIdempotencyClaim.test.ts

@@ -181,21 +181,21 @@ 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");
       }),
     } as unknown as MollifierBuffer;
     await expect(
       publishClaim({ ...baseInput, token: "owner-token", runId: "run_X", buffer })
-    ).resolves.toBeUndefined();
+    ).resolves.toBe(true);
   });
 });