fix(webapp): mollifier mutate-fallback watches Redis, not the writer

d-cs · claude · d-cs · commit 1b7c73f130e3 · 2026-05-27T17:42:42.000+01:00
The wait-and-bounce loop for mutations racing a mid-drain run polled the
PG primary on a fixed 20ms cadence with no jitter — up to ~100 reads per
request, synchronized across concurrent waiters, piling load onto the
writer exactly when mollifier is engaged to shed it.

The drainer writes the canonical PG row BEFORE it acks (sets
`materialised`) or fails (deletes the entry), so the buffer entry's own
state is an authoritative, already-in-Redis signal for "is the row in PG
yet?". Watch that (cheap Redis getEntry) instead, and touch the primary
exactly once — for the actual mutation — only after it resolves. Poll
gaps now use jittered exponential backoff (20ms → 250ms cap).

Drops the per-poll PG timeout race (DEFAULT_PG_TIMEOUT_MS / pgTimeoutMs /
findRunInPgWithTimeout), unneeded now that PG is read once rather than in
a tight loop.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/.server-changes/mollifier-mutations.md b/.server-changes/mollifier-mutations.md
@@ -3,4 +3,4 @@ area: webapp
 type: feature
 ---
 
-Mollifier API mutations on buffered runs: tag, metadata, replay, reschedule, cancel, and idempotency-key reset via a buffer-snapshot fallback.
+Mollifier API mutations on buffered runs: tag, metadata, replay, reschedule, cancel, and idempotency-key reset via a buffer-snapshot fallback. When a mutation races a mid-drain run, the wait-and-bounce loop watches the buffer entry in Redis (cheap) and reads the primary exactly once for the actual mutation, instead of polling the writer on a fixed cadence; polls use jittered exponential backoff.
diff --git a/apps/webapp/app/v3/mollifier/mutateWithFallback.server.ts b/apps/webapp/app/v3/mollifier/mutateWithFallback.server.ts
@@ -10,8 +10,12 @@ import { getMollifierBuffer } from "./mollifierBuffer.server";
 
 // Wait/retry knobs per Q3 design. Exported for tests.
 export const DEFAULT_SAFETY_NET_MS = 2_000;
+// Initial gap between buffer polls; grows by BACKOFF_FACTOR up to
+// DEFAULT_MAX_POLL_STEP_MS so a slow drain doesn't poll at a tight fixed
+// cadence for the whole safety-net budget.
 export const DEFAULT_POLL_STEP_MS = 20;
-export const DEFAULT_PG_TIMEOUT_MS = 50;
+export const DEFAULT_MAX_POLL_STEP_MS = 250;
+const BACKOFF_FACTOR = 1.7;
 
 export type MutateWithFallbackInput<TResponse> = {
   runId: string;
@@ -28,13 +32,16 @@ export type MutateWithFallbackInput<TResponse> = {
   // Override defaults for tests.
   safetyNetMs?: number;
   pollStepMs?: number;
-  pgTimeoutMs?: number;
+  maxPollStepMs?: number;
   // Test injection.
   getBuffer?: () => MollifierBuffer | null;
   prismaWriter?: TaskRunReader;
   prismaReplica?: TaskRunReader;
   sleep?: (ms: number) => Promise<void>;
   now?: () => number;
+  // Jitter source; defaults to Math.random. Inject `() => 0` for
+  // deterministic poll timing in tests.
+  random?: () => number;
 };
 
 export type MutateWithFallbackOutcome<TResponse> =
@@ -92,32 +99,49 @@ export async function mutateWithFallback<TResponse>(
     return { kind: "not_found" };
   }
 
-  // result === "busy" — entry is DRAINING / FAILED / materialised. Wait
-  // for the drainer to terminate the entry into PG (success or
-  // SYSTEM_FAILURE) and route through pgMutation.
+  // result === "busy" — the entry is mid-handoff (DRAINING) or already
+  // materialised. We do NOT poll the primary for the row to appear: that
+  // piles read load onto the writer at exactly the moment mollifier exists
+  // to shed it. Instead we watch the buffer entry itself (cheap Redis
+  // reads). The drainer writes the PG row BEFORE it acks (sets
+  // `materialised`) or fails (deletes the entry), so the entry's own state
+  // is an authoritative, already-in-Redis signal for "is the row in PG
+  // yet?". Only once it resolves do we touch the primary — exactly once,
+  // for the real mutation.
   const safetyNetMs = input.safetyNetMs ?? DEFAULT_SAFETY_NET_MS;
-  const pollStepMs = input.pollStepMs ?? DEFAULT_POLL_STEP_MS;
-  const pgTimeoutMs = input.pgTimeoutMs ?? DEFAULT_PG_TIMEOUT_MS;
+  const maxPollStepMs = input.maxPollStepMs ?? DEFAULT_MAX_POLL_STEP_MS;
+  const random = input.random ?? Math.random;
   const deadline = now() + safetyNetMs;
+  let step = input.pollStepMs ?? DEFAULT_POLL_STEP_MS;
 
   while (now() < deadline) {
     if (input.abortSignal?.aborted) {
       return { kind: "timed_out" };
     }
 
-    const row = await findRunInPgWithTimeout(
-      writer,
-      input.runId,
-      input.environmentId,
-      pgTimeoutMs,
-    );
-    if (row) {
-      const response = await input.pgMutation(row);
-      return { kind: "pg", response };
+    const entry = await buffer.getEntry(input.runId);
+    // Resolved when the entry is gone (`fail` deleted it after writing a
+    // terminal SYSTEM_FAILURE row) or materialised (`ack` after a
+    // successful trigger / cancel write). In both cases the PG row is now
+    // committed on the primary, so read it once and route through the
+    // canonical PG mutation path.
+    if (entry === null || entry.materialised === true) {
+      const row = await findRunInPg(writer, input.runId, input.environmentId);
+      if (row) {
+        const response = await input.pgMutation(row);
+        return { kind: "pg", response };
+      }
+      // Entry gone with no PG row: the drainer's terminal write itself
+      // failed (PG unreachable). Nothing to mutate.
+      return { kind: "not_found" };
     }
-
+    // Still QUEUED (requeued after a retryable drain error) or DRAINING —
+    // the run hasn't reached PG. Back off with jitter so concurrent
+    // waiters on the same draining run don't requery in lockstep.
     if (now() >= deadline) break;
-    await sleep(pollStepMs);
+    const jittered = step + Math.floor(random() * step);
+    await sleep(jittered);
+    step = Math.min(Math.ceil(step * BACKOFF_FACTOR), maxPollStepMs);
   }
 
   logger.warn("mollifier mutate-with-fallback: drainer resolution timed out", {
@@ -148,32 +172,6 @@ async function findRunInPg(
   });
 }
 
-async function findRunInPgWithTimeout(
-  client: TaskRunReader,
-  friendlyId: string,
-  environmentId: string,
-  timeoutMs: number,
-): Promise<TaskRun | null> {
-  // One slow PG query shouldn't burn the whole safety-net budget.
-  // Promise.race against a timer; on timeout we treat the poll as a miss
-  // and the outer loop tries again on the next tick.
-  const timeoutToken = Symbol("pg-timeout");
-  let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
-  const timeoutPromise = new Promise<typeof timeoutToken>((resolve) => {
-    timeoutHandle = setTimeout(() => resolve(timeoutToken), timeoutMs);
-  });
-  try {
-    const winner = await Promise.race([
-      findRunInPg(client, friendlyId, environmentId),
-      timeoutPromise,
-    ]);
-    if (winner === timeoutToken) return null;
-    return winner;
-  } finally {
-    if (timeoutHandle) clearTimeout(timeoutHandle);
-  }
-}
-
 function defaultSleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
diff --git a/apps/webapp/test/mollifierMutateWithFallback.test.ts b/apps/webapp/test/mollifierMutateWithFallback.test.ts
@@ -6,7 +6,11 @@ vi.mock("~/db.server", () => ({
 }));
 
 import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
-import type { MollifierBuffer, MutateSnapshotResult } from "@trigger.dev/redis-worker";
+import type {
+  BufferEntry,
+  MollifierBuffer,
+  MutateSnapshotResult,
+} from "@trigger.dev/redis-worker";
 import type { TaskRun } from "@trigger.dev/database";
 
 type FindFirst = ReturnType<typeof vi.fn>;
@@ -22,9 +26,30 @@ function fakePrisma(rows: Array<TaskRun | null>): PrismaStub {
 function bufferReturning(result: MutateSnapshotResult): MollifierBuffer {
   return {
     mutateSnapshot: vi.fn(async () => result),
+    getEntry: vi.fn(async () => null),
   } as unknown as MollifierBuffer;
 }
 
+// Buffer whose mutateSnapshot returns "busy" and whose getEntry walks a
+// scripted sequence of entry states (the drainer's progress). The last
+// element repeats once the sequence is exhausted.
+function bufferBusy(entries: Array<BufferEntry | null>): MollifierBuffer {
+  const getEntry = vi.fn();
+  for (const e of entries) getEntry.mockResolvedValueOnce(e);
+  getEntry.mockResolvedValue(entries.length ? entries[entries.length - 1] : null);
+  return {
+    mutateSnapshot: vi.fn(async () => "busy" as const),
+    getEntry,
+  } as unknown as MollifierBuffer;
+}
+
+const entryDraining = (): BufferEntry =>
+  ({ status: "DRAINING", materialised: false }) as unknown as BufferEntry;
+const entryQueued = (): BufferEntry =>
+  ({ status: "QUEUED", materialised: false }) as unknown as BufferEntry;
+const entryMaterialised = (): BufferEntry =>
+  ({ status: "DRAINING", materialised: true }) as unknown as BufferEntry;
+
 const fakeRun = (overrides: Partial<TaskRun> = {}): TaskRun =>
   ({
     id: "pg_id",
@@ -101,55 +126,138 @@ describe("mutateWithFallback", () => {
     expect(pgMutation).toHaveBeenCalledWith(row);
   });
 
-  it("replica miss + buffer busy + writer resolves mid-wait → pgMutation", async () => {
+  it("busy → watches buffer through DRAINING, materialises, hits primary exactly once", async () => {
     const row = fakeRun();
     const pgMutation = vi.fn(async () => "pg-after-wait");
-    // Replica misses; writer misses twice, then hits.
-    const writer = fakePrisma([null, null, row]);
+    // Writer is read ONCE, only after the buffer reports materialised.
+    const writer = fakePrisma([row]);
+    const buffer = bufferBusy([entryDraining(), entryDraining(), entryMaterialised()]);
     let nowValue = 0;
     const result = await mutateWithFallback({
       ...baseInput,
       pgMutation,
       synthesisedResponse: () => "snap",
       prismaReplica: fakePrisma([null]) as unknown as typeof import("~/db.server").$replica,
       prismaWriter: writer as unknown as typeof import("~/db.server").prisma,
-      getBuffer: () => bufferReturning("busy"),
-      sleep: async () => {
-        nowValue += 20;
+      getBuffer: () => buffer,
+      sleep: async (ms) => {
+        nowValue += ms;
       },
       now: () => nowValue,
       safetyNetMs: 2000,
       pollStepMs: 20,
-      pgTimeoutMs: 50,
+      random: () => 0,
     });
     expect(result).toEqual({ kind: "pg", response: "pg-after-wait" });
     expect(pgMutation).toHaveBeenCalledWith(row);
-    // Writer should have been polled 3 times before the hit.
-    expect(writer.taskRun.findFirst).toHaveBeenCalledTimes(3);
+    // Detection happened against Redis (3 polls), the primary exactly once.
+    expect(buffer.getEntry).toHaveBeenCalledTimes(3);
+    expect(writer.taskRun.findFirst).toHaveBeenCalledTimes(1);
+  });
+
+  it("busy → entry deleted by terminal fail, writer finds SYSTEM_FAILURE row → pgMutation", async () => {
+    const row = fakeRun();
+    const pgMutation = vi.fn(async () => "pg-failed-row");
+    const writer = fakePrisma([row]);
+    const buffer = bufferBusy([entryDraining(), null]);
+    let nowValue = 0;
+    const result = await mutateWithFallback({
+      ...baseInput,
+      pgMutation,
+      synthesisedResponse: () => "snap",
+      prismaReplica: fakePrisma([null]) as unknown as typeof import("~/db.server").$replica,
+      prismaWriter: writer as unknown as typeof import("~/db.server").prisma,
+      getBuffer: () => buffer,
+      sleep: async (ms) => {
+        nowValue += ms;
+      },
+      now: () => nowValue,
+      safetyNetMs: 2000,
+      pollStepMs: 20,
+      random: () => 0,
+    });
+    expect(result).toEqual({ kind: "pg", response: "pg-failed-row" });
+    expect(writer.taskRun.findFirst).toHaveBeenCalledTimes(1);
+  });
+
+  it("busy → entry deleted but no PG row (terminal write failed) → not_found", async () => {
+    const buffer = bufferBusy([null]);
+    const writer = fakePrisma([null]);
+    let nowValue = 0;
+    const result = await mutateWithFallback({
+      ...baseInput,
+      pgMutation: async () => "pg",
+      synthesisedResponse: () => "snap",
+      prismaReplica: fakePrisma([null]) as unknown as typeof import("~/db.server").$replica,
+      prismaWriter: writer as unknown as typeof import("~/db.server").prisma,
+      getBuffer: () => buffer,
+      sleep: async (ms) => {
+        nowValue += ms;
+      },
+      now: () => nowValue,
+      safetyNetMs: 2000,
+      pollStepMs: 20,
+      random: () => 0,
+    });
+    expect(result).toEqual({ kind: "not_found" });
+    expect(writer.taskRun.findFirst).toHaveBeenCalledTimes(1);
+  });
+
+  it("busy → requeued (back to QUEUED) then materialises; doesn't resolve early", async () => {
+    const row = fakeRun();
+    const pgMutation = vi.fn(async () => "pg-after-requeue");
+    const writer = fakePrisma([row]);
+    // QUEUED (requeued after a retryable drain error) must NOT be treated
+    // as "done" — the run hasn't reached PG. Only the later materialise does.
+    const buffer = bufferBusy([entryQueued(), entryDraining(), entryMaterialised()]);
+    let nowValue = 0;
+    const result = await mutateWithFallback({
+      ...baseInput,
+      pgMutation,
+      synthesisedResponse: () => "snap",
+      prismaReplica: fakePrisma([null]) as unknown as typeof import("~/db.server").$replica,
+      prismaWriter: writer as unknown as typeof import("~/db.server").prisma,
+      getBuffer: () => buffer,
+      sleep: async (ms) => {
+        nowValue += ms;
+      },
+      now: () => nowValue,
+      safetyNetMs: 2000,
+      pollStepMs: 20,
+      random: () => 0,
+    });
+    expect(result).toEqual({ kind: "pg", response: "pg-after-requeue" });
+    expect(buffer.getEntry).toHaveBeenCalledTimes(3);
+    expect(writer.taskRun.findFirst).toHaveBeenCalledTimes(1);
   });
 
-  it("replica miss + buffer busy + drainer never resolves → timed_out", async () => {
+  it("busy → drainer never resolves (stays DRAINING) → timed_out, primary never touched", async () => {
+    const writer = fakePrisma([]);
+    const buffer = bufferBusy([entryDraining()]);
     let nowValue = 0;
     const result = await mutateWithFallback({
       ...baseInput,
       pgMutation: async () => "pg",
       synthesisedResponse: () => "snap",
       prismaReplica: fakePrisma([null]) as unknown as typeof import("~/db.server").$replica,
-      prismaWriter: fakePrisma([null, null, null, null, null]) as unknown as typeof import("~/db.server").prisma,
-      getBuffer: () => bufferReturning("busy"),
-      sleep: async () => {
-        nowValue += 20;
+      prismaWriter: writer as unknown as typeof import("~/db.server").prisma,
+      getBuffer: () => buffer,
+      sleep: async (ms) => {
+        nowValue += ms;
       },
       now: () => nowValue,
-      safetyNetMs: 60,
+      safetyNetMs: 100,
       pollStepMs: 20,
-      pgTimeoutMs: 5,
+      random: () => 0,
     });
     expect(result).toEqual({ kind: "timed_out" });
+    // The whole point: while the run is still draining we never read the primary.
+    expect(writer.taskRun.findFirst).toHaveBeenCalledTimes(0);
   });
 
   it("abort signal during wait → timed_out without further polls", async () => {
-    const writer = fakePrisma([null, null, null]);
+    const writer = fakePrisma([]);
+    const buffer = bufferBusy([entryDraining(), entryDraining()]);
     const controller = new AbortController();
     let nowValue = 0;
     const result = await mutateWithFallback({
@@ -158,20 +266,21 @@ describe("mutateWithFallback", () => {
       synthesisedResponse: () => "snap",
       prismaReplica: fakePrisma([null]) as unknown as typeof import("~/db.server").$replica,
       prismaWriter: writer as unknown as typeof import("~/db.server").prisma,
-      getBuffer: () => bufferReturning("busy"),
-      sleep: async () => {
-        nowValue += 20;
+      getBuffer: () => buffer,
+      sleep: async (ms) => {
+        nowValue += ms;
         controller.abort();
       },
       now: () => nowValue,
       safetyNetMs: 2000,
       pollStepMs: 20,
-      pgTimeoutMs: 5,
+      random: () => 0,
       abortSignal: controller.signal,
     });
     expect(result).toEqual({ kind: "timed_out" });
-    // One poll happened before the sleep+abort.
-    expect(writer.taskRun.findFirst).toHaveBeenCalledTimes(1);
+    // One buffer poll happened before the sleep+abort; primary untouched.
+    expect(buffer.getEntry).toHaveBeenCalledTimes(1);
+    expect(writer.taskRun.findFirst).toHaveBeenCalledTimes(0);
   });
 
   it("buffer is null (mollifier disabled) → not_found after replica miss", async () => {
