feat(webapp): dashboard cancel/replay/idempotencyKey-reset handle buffered runs (Phase D)

d-cs · d-cs · commit 39e3bab39256 · 2026-05-20T18:28:05.000+01:00
Parallels Phase C's API-side work for the three dashboard mutation
routes.

D1 cancel — PG miss → buffer.mutateSnapshot('mark_cancelled'). Org-
membership verified against the buffered run's orgId (dashboard URL
doesn't carry an envId so the API-side env-scoped auth doesn't apply).
busy returns a "retry in a moment" message.

D2 replay — PG miss → findRunByIdWithMollifierFallback; B4-extended
SyntheticRun cast to TaskRun and fed to ReplayTaskRunService.
Project/env slugs for the redirect path looked up from the entry's
envId.

D3 idempotencyKey reset — PG miss → buffer.getEntry + readFallback to
read snapshot's idempotencyKey + taskIdentifier; org-membership
verified against entry orgId; existing ResetIdempotencyKeyService
(extended in B6b to clear both stores) handles the actual reset.
diff --git a/.server-changes/mollifier-dashboard-buffered-runs.md b/.server-changes/mollifier-dashboard-buffered-runs.md
@@ -0,0 +1,10 @@
+---
+area: webapp
+type: feature
+---
+
+Dashboard mutation routes handle buffered runs (Phase D — parallels Phase C's API-side work).
+
+- `POST /resources/taskruns/{runParam}/cancel`: PG miss falls through to `buffer.mutateSnapshot('mark_cancelled')`. Org-membership is verified against the buffered run's `orgId` (the dashboard URL doesn't carry an envId so the API-side env-scoped auth doesn't apply). `busy` returns a "retry in a moment" message.
+- `POST /resources/taskruns/{runParam}/replay`: PG miss falls through to `findRunByIdWithMollifierFallback`; the B4-extended `SyntheticRun` is cast to `TaskRun` and fed to `ReplayTaskRunService`. Project/env slugs needed for the success-redirect are looked up from the entry's `envId`.
+- `POST /resources/orgs/.../runs/{runParam}/idempotencyKey/reset`: PG miss falls through to buffer; reads `idempotencyKey` + `taskIdentifier` from the snapshot; org-membership verified against the entry's `orgId`. The existing `ResetIdempotencyKeyService` (extended in B6b to clear both stores) handles the actual reset.
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.idempotencyKey.reset.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.idempotencyKey.reset.tsx
@@ -5,6 +5,8 @@ import { logger } from "~/services/logger.server";
 import { requireUserId } from "~/services/session.server";
 import { ResetIdempotencyKeyService } from "~/v3/services/resetIdempotencyKey.server";
 import { v3RunParamsSchema } from "~/utils/pathBuilder";
+import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
 
 export const action: ActionFunction = async ({ request, params }) => {
   const userId = await requireUserId(request);
@@ -37,17 +39,53 @@ export const action: ActionFunction = async ({ request, params }) => {
       },
     });
 
-    if (!taskRun) {
-      return jsonWithErrorMessage({}, request, "Run not found");
-    }
-
-    if (!taskRun.idempotencyKey) {
-      return jsonWithErrorMessage({}, request, "This run does not have an idempotency key");
+    // Resolve run from PG or the mollifier buffer (Q5). For a buffered
+    // run the snapshot carries the idempotencyKey + taskIdentifier; we
+    // also need the runtimeEnvironmentId to feed ResetIdempotencyKeyService
+    // (which clears both PG and the buffer lookup — B6b).
+    let resolved:
+      | { idempotencyKey: string; taskIdentifier: string; runtimeEnvironmentId: string }
+      | null = null;
+    if (taskRun) {
+      if (!taskRun.idempotencyKey) {
+        return jsonWithErrorMessage({}, request, "This run does not have an idempotency key");
+      }
+      resolved = {
+        idempotencyKey: taskRun.idempotencyKey,
+        taskIdentifier: taskRun.taskIdentifier,
+        runtimeEnvironmentId: taskRun.runtimeEnvironmentId,
+      };
+    } else {
+      const buffer = getMollifierBuffer();
+      const entry = buffer ? await buffer.getEntry(runParam) : null;
+      if (!entry) {
+        return jsonWithErrorMessage({}, request, "Run not found");
+      }
+      const member = await prisma.orgMember.findFirst({
+        where: { userId, organizationId: entry.orgId },
+        select: { id: true },
+      });
+      if (!member) {
+        return jsonWithErrorMessage({}, request, "Run not found");
+      }
+      const synthetic = await findRunByIdWithMollifierFallback({
+        runId: runParam,
+        environmentId: entry.envId,
+        organizationId: entry.orgId,
+      });
+      if (!synthetic?.idempotencyKey || !synthetic.taskIdentifier) {
+        return jsonWithErrorMessage({}, request, "This run does not have an idempotency key");
+      }
+      resolved = {
+        idempotencyKey: synthetic.idempotencyKey,
+        taskIdentifier: synthetic.taskIdentifier,
+        runtimeEnvironmentId: entry.envId,
+      };
     }
 
     const environment = await prisma.runtimeEnvironment.findUnique({
       where: {
-        id: taskRun.runtimeEnvironmentId,
+        id: resolved.runtimeEnvironmentId,
       },
       include: {
         project: {
@@ -64,7 +102,7 @@ export const action: ActionFunction = async ({ request, params }) => {
 
     const service = new ResetIdempotencyKeyService();
 
-    await service.call(taskRun.idempotencyKey, taskRun.taskIdentifier, {
+    await service.call(resolved.idempotencyKey, resolved.taskIdentifier, {
       ...environment,
       organizationId: environment.project.organizationId,
       organization: environment.project.organization,
diff --git a/apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts b/apps/webapp/app/routes/resources.taskruns.$runParam.cancel.ts
@@ -6,6 +6,7 @@ import { redirectWithErrorMessage, redirectWithSuccessMessage } from "~/models/m
 import { logger } from "~/services/logger.server";
 import { requireUserId } from "~/services/session.server";
 import { CancelTaskRunService } from "~/v3/services/cancelTaskRun.server";
+import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
 
 export const cancelSchema = z.object({
   redirectUrl: z.string(),
@@ -42,15 +43,56 @@ export const action: ActionFunction = async ({ request, params }) => {
       },
     });
 
-    if (!taskRun) {
+    if (taskRun) {
+      const cancelRunService = new CancelTaskRunService();
+      await cancelRunService.call(taskRun);
+      return redirectWithSuccessMessage(submission.value.redirectUrl, request, `Canceled run`);
+    }
+
+    // PG miss — try the mollifier buffer. The customer can hit cancel
+    // on a buffered run from the dashboard during the burst window.
+    // Q4 design: snapshot a `mark_cancelled` patch; the drainer's
+    // bifurcation routes the run to `engine.createCancelledRun` on
+    // next pop.
+    const buffer = getMollifierBuffer();
+    const entry = buffer ? await buffer.getEntry(runParam) : null;
+    if (!entry) {
       submission.error = { runParam: ["Run not found"] };
       return json(submission);
     }
 
-    const cancelRunService = new CancelTaskRunService();
-    await cancelRunService.call(taskRun);
+    // Dashboard auth: verify the requesting user is a member of the
+    // buffered run's org. The API path scopes by env id from the
+    // authenticated request; the dashboard route uses org-membership
+    // because the URL doesn't carry an envId.
+    const member = await prisma.orgMember.findFirst({
+      where: { userId, organizationId: entry.orgId },
+      select: { id: true },
+    });
+    if (!member) {
+      submission.error = { runParam: ["Run not found"] };
+      return json(submission);
+    }
 
-    return redirectWithSuccessMessage(submission.value.redirectUrl, request, `Canceled run`);
+    const result = await buffer!.mutateSnapshot(runParam, {
+      type: "mark_cancelled",
+      cancelledAt: new Date().toISOString(),
+      cancelReason: "Canceled by user",
+    });
+    if (result === "applied_to_snapshot") {
+      return redirectWithSuccessMessage(submission.value.redirectUrl, request, `Canceled run`);
+    }
+    if (result === "not_found") {
+      submission.error = { runParam: ["Run not found"] };
+      return json(submission);
+    }
+    // "busy" — drainer is materialising. Customer can retry; by then the
+    // PG row exists and the regular cancel path takes over.
+    return redirectWithErrorMessage(
+      submission.value.redirectUrl,
+      request,
+      "Run is materialising — retry in a moment"
+    );
   } catch (error) {
     if (error instanceof Error) {
       logger.error("Failed to cancel run", {
diff --git a/apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts b/apps/webapp/app/routes/resources.taskruns.$runParam.replay.ts
@@ -11,6 +11,9 @@ import { requireUser } from "~/services/session.server";
 import { sortEnvironments } from "~/utils/environmentSort";
 import { v3RunSpanPath } from "~/utils/pathBuilder";
 import { ReplayTaskRunService } from "~/v3/services/replayTaskRun.server";
+import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
+import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
+import type { TaskRun } from "@trigger.dev/database";
 import parseDuration from "parse-duration";
 import { findCurrentWorkerDeployment } from "~/v3/models/workerDeployment.server";
 import { queueTypeFromType } from "~/presenters/v3/QueueRetrievePresenter.server";
@@ -174,7 +177,7 @@ export const action: ActionFunction = async ({ request, params }) => {
   }
 
   try {
-    const taskRun = await prisma.taskRun.findFirst({
+    const pgRun = await prisma.taskRun.findFirst({
       where: {
         friendlyId: runParam,
       },
@@ -192,6 +195,45 @@ export const action: ActionFunction = async ({ request, params }) => {
       },
     });
 
+    // Mollifier read-fallback (Q2): if the original isn't in PG yet,
+    // synthesise a TaskRun from the buffered snapshot. The B4-extended
+    // SyntheticRun carries every field ReplayTaskRunService reads. We
+    // also need projectSlug + orgSlug + envSlug for the redirect path,
+    // so look those up via the snapshot's runtimeEnvironmentId.
+    let taskRun:
+      | (TaskRun & {
+          project: { slug: string; organization: { slug: string } };
+          runtimeEnvironment: { slug: string };
+        })
+      | null = pgRun ?? null;
+    if (!taskRun) {
+      const buffer = getMollifierBuffer();
+      const entry = buffer ? await buffer.getEntry(runParam) : null;
+      if (entry) {
+        const synthetic = await findRunByIdWithMollifierFallback({
+          runId: runParam,
+          environmentId: entry.envId,
+          organizationId: entry.orgId,
+        });
+        if (synthetic) {
+          const envRow = await prisma.runtimeEnvironment.findFirst({
+            where: { id: entry.envId },
+            select: {
+              slug: true,
+              project: { select: { slug: true, organization: { select: { slug: true } } } },
+            },
+          });
+          if (envRow) {
+            taskRun = {
+              ...(synthetic as unknown as TaskRun),
+              project: { slug: envRow.project.slug, organization: { slug: envRow.project.organization.slug } },
+              runtimeEnvironment: { slug: envRow.slug },
+            };
+          }
+        }
+      }
+    }
+
     if (!taskRun) {
       return redirectWithErrorMessage(submission.value.failedRedirect, request, "Run not found");
     }
