fix(webapp,run-store): review follow-ups — split idempotency + realtime batch hardening

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.

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";
@@ -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

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