test(webapp): guards for the split-idempotency + realtime-batch review fixes

Caller-driven / unit guards for the production hardening in the base PR: claim-TTL floor,
publishClaim CAS result, reacquire fail-closed 503, expired/failed global-scope recreate
re-serialisation, and the realtime batch primary re-read. Split out of the base PR so the
production change stays reviewable on its own.

Author: d-cs

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/idempotencyExpiredRecreateReserialize.test.ts

@@ -0,0 +1,93 @@
+import { describe, expect, it, vi } from "vitest";
+
+// Devin #2: a global-scope (or scope-absent) key whose existing run is EXPIRED/FAILED gets its key
+// cleared and recreated. Under the run-ops split that recreate must serialise through the claim (same
+// cross-DB dup risk as the claim-loser cleared path) rather than fall through to an unserialised create.
+vi.mock("~/db.server", () => ({
+  prisma: {},
+  $replica: {},
+  runOpsNewPrisma: {},
+  runOpsLegacyPrisma: {},
+  runOpsNewReplica: {},
+  runOpsLegacyReplica: {},
+}));
+
+const h = vi.hoisted(() => ({ existingRun: null as unknown, splitEnabled: true }));
+
+vi.mock("~/v3/runStore.server", () => ({
+  runStore: { findRun: vi.fn(async () => h.existingRun) },
+}));
+vi.mock("~/v3/mollifier/mollifierBuffer.server", () => ({ getMollifierBuffer: () => null }));
+vi.mock("~/v3/mollifier/mollifierGate.server", () => ({
+  makeResolveMollifierFlag: () => async () => false,
+}));
+vi.mock("~/v3/runOpsMigration/splitMode.server", () => ({
+  isSplitEnabled: async () => h.splitEnabled,
+}));
+vi.mock("~/runEngine/concerns/idempotencyResidency.server", () => ({
+  resolveIdempotencyDedupClient: async () => ({}),
+}));
+
+import { IdempotencyKeyConcern } from "~/runEngine/concerns/idempotencyKeys.server";
+import type { TriggerTaskRequest } from "~/runEngine/types";
+
+function makeRequest(): TriggerTaskRequest {
+  return {
+    taskId: "my-task",
+    environment: { id: "env_a", organizationId: "org_1", organization: { featureFlags: {} } },
+    options: {},
+    body: { options: { idempotencyKey: "k-1" } }, // scope absent → treated as global
+  } as unknown as TriggerTaskRequest;
+}
+
+// handleExistingRun's documented contract for an expired/failed run: clear the key, return isCached:false.
+const CLEARED = {
+  isCached: false as const,
+  idempotencyKey: "k-1",
+  idempotencyKeyExpiresAt: new Date(Date.now() + 60_000),
+};
+
+describe("IdempotencyKeyConcern · expired/failed recreate re-serialisation", () => {
+  it("routes the cleared recreate through the claim under global-scope split (no unserialised create)", async () => {
+    h.existingRun = { id: "run_internal", friendlyId: "run_friendly" };
+    h.splitEnabled = true;
+    const concern = new IdempotencyKeyConcern({} as never, {} as never, {} as never);
+    vi.spyOn(
+      concern as never as { handleExistingRun: () => unknown },
+      "handleExistingRun"
+    ).mockResolvedValue(CLEARED);
+    const SENTINEL = { ...CLEARED, claim: { token: "t" } };
+    const reacquire = vi
+      .spyOn(
+        concern as never as { reacquireClearedGlobalWinner: () => unknown },
+        "reacquireClearedGlobalWinner"
+      )
+      .mockResolvedValue(SENTINEL);
+
+    const result = await concern.handleTriggerRequest(makeRequest(), undefined);
+
+    expect(reacquire).toHaveBeenCalledOnce();
+    expect(result).toBe(SENTINEL);
+  });
+
+  it("does NOT re-serialise when the split is off — plain recreate", async () => {
+    h.existingRun = { id: "run_internal", friendlyId: "run_friendly" };
+    h.splitEnabled = false;
+    const concern = new IdempotencyKeyConcern({} as never, {} as never, {} as never);
+    vi.spyOn(
+      concern as never as { handleExistingRun: () => unknown },
+      "handleExistingRun"
+    ).mockResolvedValue(CLEARED);
+    const reacquire = vi
+      .spyOn(
+        concern as never as { reacquireClearedGlobalWinner: () => unknown },
+        "reacquireClearedGlobalWinner"
+      )
+      .mockResolvedValue({} as never);
+
+    const result = await concern.handleTriggerRequest(makeRequest(), undefined);
+
+    expect(reacquire).not.toHaveBeenCalled();
+    expect(result).toBe(CLEARED);
+  });
+});

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/publishClaimResult.test.ts

@@ -0,0 +1,33 @@
+import { describe, expect, it, vi } from "vitest";
+import { publishClaim } from "~/v3/mollifier/idempotencyClaim.server";
+import type { MollifierBuffer } from "@trigger.dev/redis-worker";
+
+// publishClaim compare-and-sets on the caller's token; a `false` result means a stale claimant already
+// moved in (the winner's claim expired mid-pipeline), so the publish no-op'd. The caller must be able to
+// SEE that rather than silently assuming its run is canonical.
+function fakeBuffer(publishResult: boolean): MollifierBuffer {
+  return { publishClaim: vi.fn(async () => publishResult) } as unknown as MollifierBuffer;
+}
+
+const base = {
+  envId: "env_1",
+  taskIdentifier: "task",
+  idempotencyKey: "k",
+  token: "tok",
+  runId: "run_1",
+  ttlSeconds: 30,
+};
+
+describe("publishClaim result", () => {
+  it("returns false when the buffer compare-and-set no-ops (a stale claimant already moved in)", async () => {
+    expect(await publishClaim({ ...base, buffer: fakeBuffer(false) })).toBe(false);
+  });
+
+  it("returns true when the publish sets the winner", async () => {
+    expect(await publishClaim({ ...base, buffer: fakeBuffer(true) })).toBe(true);
+  });
+
+  it("returns true when the mollifier buffer is unavailable (nothing to converge)", async () => {
+    expect(await publishClaim({ ...base, buffer: null })).toBe(true);
+  });
+});

apps/webapp/test/reacquireClearedGlobalWinner.test.ts

@@ -0,0 +1,44 @@
+import { describe, expect, it, vi } from "vitest";
+
+// Force every reacquire attempt to RESOLVE to a winner (never `claimed`/`timed_out`); the spied
+// handleExistingRun keeps reporting that winner as cleared, so the bounded loop advances to exhaustion.
+vi.mock("~/v3/mollifier/idempotencyClaim.server", () => ({
+  resetResolvedClaim: vi.fn(async () => {}),
+  claimOrAwait: vi.fn(async () => ({ kind: "resolved", runId: "run_cleared" })),
+}));
+
+import { IdempotencyKeyConcern } from "~/runEngine/concerns/idempotencyKeys.server";
+
+describe("reacquireClearedGlobalWinner — fail closed on exhaustion", () => {
+  it("throws a retryable 503 after exhausting reacquires (never falls through to an unserialised create)", async () => {
+    const concern = new IdempotencyKeyConcern({} as never, {} as never, {} as never);
+    // Winner is findable but perpetually cleared → each pass advances staleRunId, exhausting the bound.
+    vi.spyOn(
+      concern as unknown as { resolveWinnerAcrossDbs: () => unknown },
+      "resolveWinnerAcrossDbs"
+    ).mockResolvedValue({ friendlyId: "run_cleared" });
+    vi.spyOn(
+      concern as unknown as { handleExistingRun: () => unknown },
+      "handleExistingRun"
+    ).mockResolvedValue({ isCached: false });
+
+    const request = { environment: { id: "env_1" }, taskId: "task" } as never;
+    const ctx = {
+      idempotencyKey: "k",
+      idempotencyKeyExpiresAt: new Date(Date.now() + 60_000),
+      dedupClient: {} as never,
+      ttlSeconds: 30,
+      clearedRunId: "run_cleared",
+      safetyNetMs: 5_000,
+      pollStepMs: 25,
+    };
+
+    await expect(
+      (
+        concern as unknown as {
+          reacquireClearedGlobalWinner: (...a: unknown[]) => Promise<unknown>;
+        }
+      ).reacquireClearedGlobalWinner(request, undefined, ctx)
+    ).rejects.toMatchObject({ name: "ServiceValidationError", status: 503 });
+  });
+});

apps/webapp/test/resolveBatchForRealtime.test.ts

@@ -0,0 +1,47 @@
+import { describe, expect, it, vi } from "vitest";
+import { resolveBatchTaskRunForRealtime } from "~/v3/realtime/resolveBatchForRealtime.server";
+
+// The realtime batch route reads the batch client-less (replica). Under replica lag a just-created batch
+// misses; `shouldRetryNotFound` covers the zodfetch GET, but the Electric ShapeStream consumer
+// (self-hosters) ignores `x-should-retry`, so the route must re-read the owning PRIMARY on a miss.
+// Passing a (non-replica) writer client flips each store leg to its own primary.
+function laggingStore(batch: { id: string; friendlyId: string }) {
+  return {
+    findBatchTaskRunByFriendlyId: vi.fn(
+      async (_friendlyId: string, _envId: string, _args: unknown, client?: unknown) =>
+        client ? batch : null
+    ),
+  };
+}
+
+describe("resolveBatchTaskRunForRealtime", () => {
+  it("re-reads the primary when the replica misses a fresh batch", async () => {
+    const store = laggingStore({ id: "b_1", friendlyId: "batch_1" });
+    const found = await resolveBatchTaskRunForRealtime("batch_1", "env_1", {
+      store: store as never,
+      writer: {} as never,
+    });
+    expect(found).toEqual({ id: "b_1", friendlyId: "batch_1" });
+    expect(store.findBatchTaskRunByFriendlyId).toHaveBeenCalledTimes(2);
+  });
+
+  it("returns null when the batch is genuinely absent on both replica and primary", async () => {
+    const store = { findBatchTaskRunByFriendlyId: vi.fn(async () => null) };
+    const found = await resolveBatchTaskRunForRealtime("nope", "env_1", {
+      store: store as never,
+      writer: {} as never,
+    });
+    expect(found).toBeNull();
+  });
+
+  it("does not re-read the primary when the replica already has the batch", async () => {
+    const store = {
+      findBatchTaskRunByFriendlyId: vi.fn(async () => ({ id: "b_2", friendlyId: "batch_2" })),
+    };
+    await resolveBatchTaskRunForRealtime("batch_2", "env_1", {
+      store: store as never,
+      writer: {} as never,
+    });
+    expect(store.findBatchTaskRunByFriendlyId).toHaveBeenCalledTimes(1);
+  });
+});