fix(webapp): honour cancel-wins conflict + requeue on transient PG outage in mollifier drainer

d-cs · claude · d-cs · commit df65a3b50b1e · 2026-05-27T16:40:39.000+01:00
The mollifier drainer's cancel bifurcation called engine.createCancelledRun
without handling its documented conflict contract: when the normal trigger
replay path races ahead and materialises a live (non-CANCELED) row, the engine
throws a conflict so the caller can "decide between engine.cancelRun() and
skipping". The handler did neither — the conflict propagated, isRetryablePgError
returned false, and the drainer buffer.fail()'d the entry, silently losing the
cancellation while the run kept executing. Now route conflicts to
engine.cancelRun() so the cancel actually wins.

Separately, when engine.trigger fails non-retryably and the SYSTEM_FAILURE
fallback write then fails because PG is transiently unreachable, rethrowing the
original non-retryable error made the drainer buffer.fail() the entry — losing
the run with no PG row ever landing, and dropping the write error entirely.
Rethrow the retryable write error instead so the drainer requeues; the failure
row lands once PG recovers. Non-retryable write failures still rethrow the
original error as before.

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts b/apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts
@@ -1,6 +1,7 @@
 import { context, trace, TraceFlags } from "@opentelemetry/api";
 import type { RunEngine } from "@internal/run-engine";
 import type { PrismaClientOrTransaction } from "@trigger.dev/database";
+import { RunId } from "@trigger.dev/core/v3/isomorphic";
 import type { MollifierDrainerHandler } from "@trigger.dev/redis-worker";
 import { startSpan } from "~/v3/tracing.server";
 import type { MollifierSnapshot } from "./mollifierSnapshot.server";
@@ -72,14 +73,41 @@ export function createDrainerHandler(deps: {
           span.setAttribute("mollifier.run_friendly_id", input.runId);
           span.setAttribute("mollifier.cancel_bifurcation", true);
           span.setAttribute("taskRunId", input.runId);
-          await deps.engine.createCancelledRun(
-            {
-              snapshot: input.payload as any,
-              cancelledAt: new Date(cancelledAtStr),
-              cancelReason,
-            },
-            deps.prisma,
-          );
+          try {
+            await deps.engine.createCancelledRun(
+              {
+                snapshot: input.payload as any,
+                cancelledAt: new Date(cancelledAtStr),
+                cancelReason,
+              },
+              deps.prisma,
+            );
+          } catch (err) {
+            // createCancelledRun throws a conflict when the normal trigger
+            // replay path won the race and already materialised a live
+            // (non-CANCELED) row for this friendlyId. Its contract leaves
+            // the resolution to us: honour the cancel by actually
+            // cancelling the now-live run. Letting the conflict propagate
+            // would instead reach the drainer's terminal-failure path
+            // (isRetryablePgError() is false for it), buffer.fail() the
+            // entry, and silently lose the cancellation while the run
+            // keeps executing.
+            const isConflict =
+              err instanceof Error && err.message.startsWith("createCancelledRun conflict");
+            if (!isConflict) {
+              throw err;
+            }
+            span.setAttribute("mollifier.cancel_conflict", true);
+            const friendlyId =
+              typeof input.payload.friendlyId === "string"
+                ? input.payload.friendlyId
+                : input.runId;
+            await deps.engine.cancelRun({
+              runId: RunId.fromFriendlyId(friendlyId),
+              completedAt: new Date(cancelledAtStr),
+              reason: cancelReason,
+            });
+          }
         });
       });
       return;
@@ -158,9 +186,23 @@ export function createDrainerHandler(deps: {
                 typeof snapshot.lockedQueueId === "string" ? snapshot.lockedQueueId : undefined,
             });
           } catch (writeErr) {
-            // Class A — PG itself is failing. Rethrow the original
-            // error so the drainer falls back to buffer.fail. Include
-            // the write error in the log line at the drainer layer.
+            // The terminal SYSTEM_FAILURE write itself failed. If it
+            // failed because PG is transiently unreachable, rethrow the
+            // *write* error so the drainer requeues — buffer.fail()ing on
+            // the original non-retryable error would lose the run with no
+            // PG row ever landing. Once PG recovers the requeued entry
+            // writes its failure row and the customer sees it.
+            if (isRetryablePgError(writeErr)) {
+              span.setAttribute("mollifier.terminal_write_retryable", true);
+              throw writeErr;
+            }
+            // PG reachable but the write was rejected for another reason
+            // (genuinely bad snapshot). Rethrow the original trigger error
+            // so the drainer falls back to buffer.fail.
+            span.setAttribute(
+              "mollifier.terminal_write_error",
+              writeErr instanceof Error ? writeErr.message : String(writeErr)
+            );
             throw err;
           }
         }
diff --git a/apps/webapp/test/mollifierDrainerHandler.test.ts b/apps/webapp/test/mollifierDrainerHandler.test.ts
@@ -1,5 +1,6 @@
 import { describe, expect, it, vi } from "vitest";
 import { trace } from "@opentelemetry/api";
+import { RunId } from "@trigger.dev/core/v3/isomorphic";
 
 vi.mock("~/db.server", () => ({
   prisma: {},
@@ -180,6 +181,81 @@ describe("createDrainerHandler", () => {
     expect(createFailedTaskRun).toHaveBeenCalledOnce();
   });
 
+  it("honours the cancel when a buffered cancel races a materialised non-CANCELED row", async () => {
+    // Cancel-wins-over-trigger (Q4 bifurcation). If the normal trigger
+    // replay path materialised a live PENDING row before the cancel
+    // bifurcation drained, engine.createCancelledRun throws a conflict —
+    // its documented contract is that "the caller must decide between
+    // engine.cancelRun() and skipping". The drainer handler must honour
+    // the cancel intent by actually cancelling the now-live run; otherwise
+    // the conflict propagates, isRetryablePgError() returns false, and the
+    // drainer buffer.fail()s the entry — silently losing the cancellation
+    // while the run keeps executing.
+    const friendlyId = RunId.generate().friendlyId;
+    const createCancelledRun = vi.fn(async () => {
+      throw new Error(
+        `createCancelledRun conflict: existing run ${friendlyId} has status PENDING`
+      );
+    });
+    const cancelRun = vi.fn(async () => ({ alreadyFinished: false }));
+    const handler = createDrainerHandler({
+      engine: { createCancelledRun, cancelRun } as any,
+      prisma: {} as any,
+    });
+
+    await expect(
+      handler({
+        runId: friendlyId,
+        envId: "env_a",
+        orgId: "org_1",
+        payload: {
+          friendlyId,
+          taskIdentifier: "t",
+          environment: envFixture,
+          cancelledAt: new Date().toISOString(),
+          cancelReason: "Canceled by user",
+        },
+        attempts: 0,
+        createdAt: new Date(),
+      } as any)
+    ).resolves.toBeUndefined();
+
+    // The live run is actually cancelled, by its internal id.
+    expect(cancelRun).toHaveBeenCalledOnce();
+    expect(cancelRun.mock.calls[0][0].runId).toBe(RunId.fromFriendlyId(friendlyId));
+  });
+
+  it("requeues on a transient PG outage during the SYSTEM_FAILURE fallback write", async () => {
+    // engine.trigger failed non-retryably, so we try to write a terminal
+    // SYSTEM_FAILURE row. If THAT write fails because PG is transiently
+    // unreachable, rethrowing the *original* non-retryable error makes the
+    // drainer buffer.fail() the entry — losing the run with no PG row ever
+    // landing. Rethrow the retryable write error instead so the drainer
+    // requeues; once PG recovers the failure row lands and the customer
+    // sees it.
+    const trigger = vi.fn(async () => {
+      throw new Error("validation failed: payload too large");
+    });
+    const createFailedTaskRun = vi.fn(async () => {
+      throw new Error("Can't reach database server");
+    });
+    const handler = createDrainerHandler({
+      engine: { trigger, createFailedTaskRun } as any,
+      prisma: {} as any,
+    });
+
+    await expect(
+      handler({
+        runId: "run_x",
+        envId: "env_a",
+        orgId: "org_1",
+        payload: { taskIdentifier: "t", environment: envFixture },
+        attempts: 0,
+        createdAt: new Date(),
+      } as any)
+    ).rejects.toThrow("Can't reach database server");
+  });
+
   it("rethrows the original error when the snapshot lacks an environment block", async () => {
     const triggerErr = new Error("engine rejected the snapshot");
     const trigger = vi.fn(async () => {
