feat(webapp,run-engine): cancel API supports buffered runs (Phase C1)

d-cs · d-cs · commit d4f73421303f · 2026-05-20T17:14:40.000+01:00
Per the Q4 mollifier-cancel design — first mutation endpoint.

engine.createCancelledRun: new run-engine method that writes a CANCELED
TaskRun row directly from a buffer snapshot. Skips queue insertion,
waitpoint creation, and concurrency reservation (run never executes).
Emits runCancelled so the existing handler writes the TaskEvent
cancellation row. P2002 from double-pop is caught and returns the
existing row without re-emitting.

Drainer bifurcation: mollifierDrainerHandler routes to
createCancelledRun when snapshot.cancelledAt is set. Cancel-wins-
over-trigger — customer intent is terminal.

Cancel route: wraps the call in mutateWithFallback. PG-row hits go
through the existing CancelTaskRunService. Buffered-QUEUED hits land
a mark_cancelled patch on the snapshot via mutateSnapshot. busy
snapshots wait for drainer resolution then call the PG service
against the resulting row. 404 / 503 surface for genuine missing
or drainer-hung cases.

Known follow-up: the Q3 wait-and-bounce for cancel-of-buffered-FAILED
relies on the drainer eventually writing a SYSTEM_FAILURE PG row on
terminal materialisation failure. That drainer-side write isn't
implemented yet (the failed-drain path today only marks the buffer
entry hash FAILED). Cancel-of-state-3 will currently 503 after 2s
instead of returning the SYSTEM_FAILURE row. Acceptable rare-race
behaviour; flagged for a follow-up alongside the drainer sweeper work.
diff --git a/.server-changes/mollifier-cancel-buffered-runs.md b/.server-changes/mollifier-cancel-buffered-runs.md
@@ -0,0 +1,12 @@
+---
+area: webapp
+type: feature
+---
+
+Cancel API (`POST /api/v2/runs/{id}/cancel`) now works on buffered runs. Per the Q4 mollifier-cancel design:
+
+- `engine.createCancelledRun` (new method in `@internal/run-engine`): writes a `CANCELED` TaskRun row directly from a buffer snapshot, bypassing the trigger/queue pipeline. Skips run-queue insertion (no execution needed), waitpoint creation (single-`triggerAndWait` can't enter the buffer), and concurrency reservation. Emits `runCancelled` so the existing handler writes the TaskEvent cancellation row. Idempotent: P2002 unique-constraint violations from double-pop after a drainer requeue return the existing row without re-emitting.
+
+- Drainer bifurcation (`mollifierDrainerHandler.server.ts`): when the snapshot carries `cancelledAt`, route to `createCancelledRun` instead of `engine.trigger`. Cancel-wins-over-trigger ordering — customer intent is terminal.
+
+- Cancel route (`api.v2.runs.$runParam.cancel.ts`): wraps the call in `mutateWithFallback`. PG-row hits go through the existing `CancelTaskRunService`. Buffered-run hits land a `mark_cancelled` patch on the snapshot via `mutateSnapshot`. `busy` snapshots wait for drainer resolution then call the PG service against the resulting row. Genuine 404s and timeouts surface as 404/503 respectively.
diff --git a/apps/webapp/app/routes/api.v2.runs.$runParam.cancel.ts b/apps/webapp/app/routes/api.v2.runs.$runParam.cancel.ts
@@ -1,8 +1,9 @@
 import { json } from "@remix-run/server-runtime";
 import { z } from "zod";
-import { $replica } from "~/db.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
 import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server";
+import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -17,29 +18,48 @@ const { action } = createActionApiRoute(
       action: "write",
       resource: (params) => ({ type: "runs", id: params.runParam }),
     },
-    findResource: async (params, auth) => {
-      return $replica.taskRun.findFirst({
-        where: {
-          friendlyId: params.runParam,
-          runtimeEnvironmentId: auth.environment.id,
-        },
-      });
-    },
+    // PG-side authorisation is performed inside mutateWithFallback. Routing
+    // the resource through findResource (which would require a PG-or-buffer
+    // resolved discriminated union here) would duplicate the resolution
+    // mutateWithFallback already does, so we pass `null` to signal "open"
+    // and let the helper do the lookup atomically with the mutation.
+    findResource: async () => null,
   },
-  async ({ resource }) => {
-    if (!resource) {
-      return json({ error: "Run not found" }, { status: 404 });
-    }
+  async ({ params, authentication }) => {
+    const runId = params.runParam;
+    const env = authentication.environment;
+    const cancelledAt = new Date();
+    const cancelReason = "Canceled by user";
 
-    const service = new CancelTaskRunService();
+    const outcome = await mutateWithFallback({
+      runId,
+      environmentId: env.id,
+      organizationId: env.organizationId,
+      bufferPatch: {
+        type: "mark_cancelled",
+        cancelledAt: cancelledAt.toISOString(),
+        cancelReason,
+      },
+      pgMutation: async (taskRun) => {
+        const service = new CancelTaskRunService();
+        try {
+          await service.call(taskRun);
+        } catch {
+          return json({ error: "Internal Server Error" }, { status: 500 });
+        }
+        return json({ id: taskRun.friendlyId }, { status: 200 });
+      },
+      synthesisedResponse: () => json({ id: runId }, { status: 200 }),
+      abortSignal: getRequestAbortSignal(),
+    });
 
-    try {
-      await service.call(resource);
-    } catch (error) {
-      return json({ error: "Internal Server Error" }, { status: 500 });
+    if (outcome.kind === "not_found") {
+      return json({ error: "Run not found" }, { status: 404 });
     }
-
-    return json({ id: resource.friendlyId }, { status: 200 });
+    if (outcome.kind === "timed_out") {
+      return json({ error: "Run materialisation timed out" }, { status: 503 });
+    }
+    return outcome.response;
   }
 );
 
diff --git a/apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts b/apps/webapp/app/v3/mollifier/mollifierDrainerHandler.server.ts
@@ -45,6 +45,39 @@ export function createDrainerHandler(deps: {
           })
         : context.active();
 
+    // Cancel-wins-over-trigger (Q4 bifurcation). If a cancel API call
+    // landed on this entry while it was QUEUED, the snapshot carries
+    // `cancelledAt` + `cancelReason`. Skip the normal materialise path
+    // and write a CANCELED PG row directly. The existing runCancelled
+    // handler writes the TaskEvent.
+    const cancelledAtStr =
+      typeof input.payload.cancelledAt === "string" ? input.payload.cancelledAt : undefined;
+    if (cancelledAtStr) {
+      const cancelReason =
+        typeof input.payload.cancelReason === "string"
+          ? input.payload.cancelReason
+          : "Canceled by user";
+      await context.with(parentContext, async () => {
+        await startSpan(tracer, "mollifier.drained.cancelled", async (span) => {
+          span.setAttribute("mollifier.drained", true);
+          span.setAttribute("mollifier.dwell_ms", dwellMs);
+          span.setAttribute("mollifier.attempts", input.attempts);
+          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,
+          );
+        });
+      });
+      return;
+    }
+
     await context.with(parentContext, async () => {
       await startSpan(tracer, "mollifier.drained", async (span) => {
         span.setAttribute("mollifier.drained", true);
diff --git a/apps/webapp/test/mollifierMollify.test.ts b/apps/webapp/test/mollifierMollify.test.ts
@@ -8,8 +8,10 @@ vi.mock("~/db.server", () => ({
 import { mollifyTrigger } from "~/v3/mollifier/mollifierMollify.server";
 import type { MollifierBuffer } from "@trigger.dev/redis-worker";
 
-function fakeBuffer(): { buffer: MollifierBuffer; accept: ReturnType<typeof vi.fn> } {
-  const accept = vi.fn(async () => undefined);
+function fakeBuffer(
+  acceptResult: Awaited<ReturnType<MollifierBuffer["accept"]>> = { kind: "accepted" },
+): { buffer: MollifierBuffer; accept: ReturnType<typeof vi.fn> } {
+  const accept = vi.fn(async () => acceptResult);
   return {
     buffer: { accept } as unknown as MollifierBuffer,
     accept,
@@ -39,6 +41,8 @@ describe("mollifyTrigger", () => {
       envId: "env_a",
       orgId: "org_1",
       payload: expect.any(String),
+      idempotencyKey: undefined,
+      taskIdentifier: undefined,
     });
     expect(result.run.friendlyId).toBe("run_friendly_1");
     expect(result.error).toBeUndefined();
@@ -50,6 +54,26 @@ describe("mollifyTrigger", () => {
     });
   });
 
+  it("echoes the winner's runId with isCached=true on duplicate_idempotency", async () => {
+    const { buffer } = fakeBuffer({
+      kind: "duplicate_idempotency",
+      existingRunId: "run_winner",
+    });
+    const result = await mollifyTrigger({
+      runFriendlyId: "run_loser",
+      environmentId: "env_a",
+      organizationId: "org_1",
+      engineTriggerInput: { taskIdentifier: "t", payload: "{}" },
+      decision: { divert: true, reason: "per_env_rate", count: 1, threshold: 1 },
+      buffer,
+      idempotencyKey: "key",
+      taskIdentifier: "t",
+    });
+    expect(result.run.friendlyId).toBe("run_winner");
+    expect(result.isCached).toBe(true);
+    expect(result.notice).toBeUndefined();
+  });
+
   it("snapshot is round-trippable: payload field is parseable JSON of engineTriggerInput", async () => {
     const { buffer, accept } = fakeBuffer();
     const engineInput = { taskIdentifier: "t", payload: "{}", tags: ["a", "b"] };
diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts
@@ -443,6 +443,154 @@ export class RunEngine {
 
   //MARK: - Run functions
 
+  /**
+   * Writes a TaskRun row in CANCELED state directly, bypassing the trigger
+   * pipeline. Used by the mollifier drainer when a cancel API call lands on
+   * a buffered run before it materialises (Q4 mollifier-cancel design).
+   *
+   * Skips: queue insertion (no execution), waitpoint creation (single-
+   * triggerAndWait can't enter the buffer; F4 bypass), concurrency
+   * reservation. Emits `runCancelled` so the existing TaskEvent handler
+   * writes the cancellation event row — the only side effect PG-side cancel
+   * has today per audit.
+   *
+   * Idempotent: if a row with the same friendlyId already exists (double
+   * drainer pop after requeue), Prisma's P2002 unique-constraint violation
+   * is caught and the existing row is returned. The duplicate runCancelled
+   * emission is skipped — the original drain's emit already wrote the
+   * TaskEvent.
+   */
+  async createCancelledRun(
+    {
+      snapshot,
+      cancelledAt,
+      cancelReason,
+    }: {
+      snapshot: TriggerParams;
+      cancelledAt: Date;
+      cancelReason: string;
+    },
+    tx?: PrismaClientOrTransaction,
+  ): Promise<TaskRun> {
+    const prisma = tx ?? this.prisma;
+    return startSpan(this.tracer, "createCancelledRun", async (span) => {
+      span.setAttribute("friendlyId", snapshot.friendlyId);
+      span.setAttribute("taskIdentifier", snapshot.taskIdentifier);
+      const id = RunId.fromFriendlyId(snapshot.friendlyId);
+      const error: TaskRunError = { type: "STRING_ERROR", raw: cancelReason };
+
+      try {
+        const taskRun = await prisma.taskRun.create({
+          data: {
+            id,
+            engine: "V2",
+            status: "CANCELED",
+            friendlyId: snapshot.friendlyId,
+            runtimeEnvironmentId: snapshot.environment.id,
+            environmentType: snapshot.environment.type,
+            organizationId: snapshot.environment.organization.id,
+            projectId: snapshot.environment.project.id,
+            idempotencyKey: snapshot.idempotencyKey,
+            idempotencyKeyExpiresAt: snapshot.idempotencyKeyExpiresAt,
+            idempotencyKeyOptions: snapshot.idempotencyKeyOptions,
+            taskIdentifier: snapshot.taskIdentifier,
+            payload: snapshot.payload,
+            payloadType: snapshot.payloadType,
+            context: snapshot.context,
+            traceContext: snapshot.traceContext,
+            traceId: snapshot.traceId,
+            spanId: snapshot.spanId,
+            parentSpanId: snapshot.parentSpanId,
+            lockedToVersionId: snapshot.lockedToVersionId,
+            taskVersion: snapshot.taskVersion,
+            sdkVersion: snapshot.sdkVersion,
+            cliVersion: snapshot.cliVersion,
+            concurrencyKey: snapshot.concurrencyKey,
+            queue: snapshot.queue,
+            lockedQueueId: snapshot.lockedQueueId,
+            workerQueue: snapshot.workerQueue,
+            isTest: snapshot.isTest,
+            taskEventStore: snapshot.taskEventStore,
+            runTags: snapshot.tags.length === 0 ? undefined : snapshot.tags,
+            oneTimeUseToken: snapshot.oneTimeUseToken,
+            parentTaskRunId: snapshot.parentTaskRunId,
+            rootTaskRunId: snapshot.rootTaskRunId,
+            replayedFromTaskRunFriendlyId: snapshot.replayedFromTaskRunFriendlyId,
+            batchId: snapshot.batch?.id,
+            resumeParentOnCompletion: snapshot.resumeParentOnCompletion,
+            depth: snapshot.depth,
+            seedMetadata: snapshot.seedMetadata,
+            seedMetadataType: snapshot.seedMetadataType,
+            metadata: snapshot.metadata,
+            metadataType: snapshot.metadataType,
+            machinePreset: snapshot.machine,
+            scheduleId: snapshot.scheduleId,
+            scheduleInstanceId: snapshot.scheduleInstanceId,
+            createdAt: snapshot.createdAt,
+            bulkActionGroupIds: snapshot.bulkActionId ? [snapshot.bulkActionId] : undefined,
+            planType: snapshot.planType,
+            realtimeStreamsVersion: snapshot.realtimeStreamsVersion,
+            streamBasinName: snapshot.streamBasinName,
+            annotations: snapshot.annotations,
+            completedAt: cancelledAt,
+            updatedAt: cancelledAt,
+            error: error as unknown as Prisma.InputJsonValue,
+            attemptNumber: 0,
+            executionSnapshots: {
+              create: {
+                engine: "V2",
+                executionStatus: "FINISHED",
+                description: "Run cancelled before materialisation",
+                runStatus: "CANCELED",
+                environmentId: snapshot.environment.id,
+                environmentType: snapshot.environment.type,
+                projectId: snapshot.environment.project.id,
+                organizationId: snapshot.environment.organization.id,
+              },
+            },
+          },
+        });
+
+        this.eventBus.emit("runCancelled", {
+          time: cancelledAt,
+          run: {
+            id: taskRun.id,
+            status: taskRun.status,
+            friendlyId: taskRun.friendlyId,
+            spanId: taskRun.spanId,
+            taskEventStore: taskRun.taskEventStore,
+            createdAt: taskRun.createdAt,
+            completedAt: taskRun.completedAt,
+            error,
+            updatedAt: taskRun.updatedAt,
+            attemptNumber: taskRun.attemptNumber ?? 0,
+          },
+          organization: { id: snapshot.environment.organization.id },
+          project: { id: snapshot.environment.project.id },
+          environment: { id: snapshot.environment.id },
+        });
+
+        return taskRun;
+      } catch (err) {
+        // P2002 = unique constraint violation. Double-pop after a drainer
+        // requeue can reach this. Idempotent: return the existing row
+        // without re-emitting.
+        if (
+          err instanceof Prisma.PrismaClientKnownRequestError &&
+          err.code === "P2002"
+        ) {
+          this.logger.info(
+            "createCancelledRun: row already exists, returning existing (idempotent)",
+            { friendlyId: snapshot.friendlyId },
+          );
+          const existing = await prisma.taskRun.findFirst({ where: { id } });
+          if (existing) return existing;
+        }
+        throw err;
+      }
+    });
+  }
+
   /** "Triggers" one run. */
   async trigger(
     {
