fix(webapp,redis-worker): reacquire the idempotency claim when a cleared global winner is recreated

The global-scope claim serialized the initial create but not the clear-and-recreate
path: a claim loser resolving an expired/failed winner cleared the key and then created
a new run unguarded, so two concurrent cross-residency losers could each recreate — a
duplicate the per-DB unique index can't catch. Now, when a loser's resolved winner is
cleared, the recreate is re-serialized through the claim: compare-and-delete the stale
resolved slot (keyed on the observed runId — never an unconditional DEL, so a concurrent
reacquirer's freshly published winner is never wiped) then re-enter claimOrAwait; capped
at 5 attempts, then fall open to the create with the PG unique index as backstop. Adds
MollifierBuffer.resetResolvedClaim reusing the existing compare-and-delete Lua. Guarded
by two caller-driven tests (expired-key + failed-winner, two concurrent losers, real
Redis) that create two runs when reverted.

Author: d-cs

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -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
@@ -287,10 +294,31 @@ export class IdempotencyKeyConcern {
         if (globalUnderSplit) {
           const winner = await this.resolveWinnerAcrossDbs(outcome.runId, request.environment.id);
           if (winner) {
-            return await this.handleExistingRun(request, parentStore, 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,
             });
           }
         }
@@ -479,6 +507,86 @@ export class IdempotencyKeyConcern {
     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<IdempotencyKeyConcernResult> {
+    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

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<boolean> {
+  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<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }