triggerdotdev · d-cs · May 12, 2026 · May 12, 2026 · May 14, 2026 · May 14, 2026
diff --git a/.changeset/mollifier-buffer-extensions.md b/.changeset/mollifier-buffer-extensions.md
@@ -0,0 +1,6 @@
+---
+"@trigger.dev/redis-worker": minor
+"@trigger.dev/core": patch
+---
+
+Mollifier buffer feature set built on top of the initial primitives: idempotency-lookup with SETNX dedup, atomic snapshot-mutation API (`mutateSnapshot` with tag/metadata/delay/cancel patches), metadata CAS for lossless concurrent updates, watermark-paginated listing, claim primitives for pre-gate idempotency, ZSET-backed per-env queue, 30s post-ack grace TTL, and drop the accept-time entry TTL (drainer is now the only removal mechanism). `@trigger.dev/core` gains an optional `notice` field on the trigger response so the SDK can surface mollifier-queued guidance to customers.
diff --git a/.gitignore b/.gitignore
@@ -72,4 +72,5 @@ apps/**/public/build
 .mcp.log
 .mcp.json
 .cursor/debug.log
-ailogger-output.log
+ailogger-output.log
+.playwright-mcp/
diff --git a/.server-changes/mollifier.md b/.server-changes/mollifier.md
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Mollifier — Redis-backed burst buffer in front of `engine.trigger` with a fair drainer, full read/write parity for buffered runs across the API + dashboard + realtime stream, alertable `mollifier.stale_entries.current` gauge for drainer health, and `runFailed` alerts on drainer-terminal `SYSTEM_FAILURE` rows.
diff --git a/apps/webapp/app/components/runs/v3/CancelRunDialog.tsx b/apps/webapp/app/components/runs/v3/CancelRunDialog.tsx
@@ -1,6 +1,7 @@
 import { NoSymbolIcon } from "@heroicons/react/24/solid";
 import { DialogClose } from "@radix-ui/react-dialog";
 import { Form, useNavigation } from "@remix-run/react";
+import { useEffect, useRef } from "react";
 import { Button } from "~/components/primitives/Buttons";
 import { DialogContent, DialogHeader } from "~/components/primitives/Dialog";
 import { FormButtons } from "~/components/primitives/FormButtons";
@@ -10,14 +11,35 @@ import { SpinnerWhite } from "~/components/primitives/Spinner";
 type CancelRunDialogProps = {
   runFriendlyId: string;
   redirectPath: string;
+  // Optional: when provided, close the dialog as soon as the cancel
+  // action transitions to "loading" (the redirect is in flight). Lets
+  // the caller control the open state without interfering with the
+  // form's submit name=value pair the way `<DialogClose asChild>`
+  // around the submit button does.
+  onCancelSubmitted?: () => void;
 };
 
-export function CancelRunDialog({ runFriendlyId, redirectPath }: CancelRunDialogProps) {
+export function CancelRunDialog({
+  runFriendlyId,
+  redirectPath,
+  onCancelSubmitted,
+}: CancelRunDialogProps) {
   const navigation = useNavigation();
 
   const formAction = `/resources/taskruns/${runFriendlyId}/cancel`;
   const isLoading = navigation.formAction === formAction;
 
+  const wasSubmitting = useRef(false);
+  useEffect(() => {
+    if (!onCancelSubmitted) return;
+    if (navigation.state === "submitting" && navigation.formAction === formAction) {
+      wasSubmitting.current = true;
+    } else if (wasSubmitting.current && navigation.state !== "submitting") {
+      wasSubmitting.current = false;
+      onCancelSubmitted();
+    }
+  }, [navigation.state, navigation.formAction, formAction, onCancelSubmitted]);
+
   return (
     <DialogContent key="cancel">
       <DialogHeader>Cancel this run?</DialogHeader>

diff --git a/apps/webapp/app/entry.server.tsx b/apps/webapp/app/entry.server.tsx
@@ -9,6 +9,7 @@ import { renderToPipeableStream } from "react-dom/server";
 import { PassThrough } from "stream";
 import * as Worker from "~/services/worker.server";
 import { initMollifierDrainerWorker } from "~/v3/mollifierDrainerWorker.server";
+import { initMollifierStaleSweepWorker } from "~/v3/mollifierStaleSweepWorker.server";
 import { bootstrap } from "./bootstrap";
 import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import {
@@ -219,6 +220,7 @@ Worker.init().catch((error) => {
 });
 
 initMollifierDrainerWorker();
+initMollifierStaleSweepWorker();
 
 bootstrap().catch((error) => {
   logError(error);

diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts
@@ -1062,13 +1062,16 @@ const EnvironmentSchema = z
     // Separate switch for the drainer (consumer side) so it can be split
     // off onto a dedicated worker service. Unset → inherits
     // TRIGGER_MOLLIFIER_ENABLED, so single-container self-hosters don't have to
-    // flip two switches. In multi-replica deployments, set this to "0"
-    // explicitly on every replica except the one dedicated drainer
-    // service — otherwise every replica's polling loop races for the
-    // same buffer entries. `TRIGGER_MOLLIFIER_ENABLED` is still the master kill
-    // switch; setting this to "1" while `TRIGGER_MOLLIFIER_ENABLED` is "0" is a
-    // no-op because the gate-side singleton refuses to construct a
-    // buffer when the system is off.
+    // flip two switches. Multi-replica drainers are correct — `popAndMarkDraining`
+    // is an atomic ZPOPMIN + status flip in one Lua call, so only one replica
+    // can win any given entry — but inefficient: polling load (SMEMBERS +
+    // per-env scans) multiplies by N, and `TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY`
+    // is per-process so engine load also multiplies. Splitting the drainer
+    // onto a dedicated worker keeps that traffic off the request-serving
+    // replicas. `TRIGGER_MOLLIFIER_ENABLED` is still the master kill switch;
+    // setting this to "1" while `TRIGGER_MOLLIFIER_ENABLED` is "0" is a
+    // no-op because the gate-side singleton refuses to construct a buffer
+    // when the system is off.
     TRIGGER_MOLLIFIER_DRAINER_ENABLED: z.string().default(process.env.TRIGGER_MOLLIFIER_ENABLED ?? "0"),
     TRIGGER_MOLLIFIER_SHADOW_MODE: z.string().default("0"),
     TRIGGER_MOLLIFIER_REDIS_HOST: z
@@ -1091,14 +1094,34 @@ const EnvironmentSchema = z
       .transform((v) => v ?? process.env.REDIS_PASSWORD),
     TRIGGER_MOLLIFIER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
     TRIGGER_MOLLIFIER_TRIP_WINDOW_MS: z.coerce.number().int().positive().default(200),
-    TRIGGER_MOLLIFIER_TRIP_THRESHOLD: z.coerce.number().int().positive().default(100),
+    TRIGGER_MOLLIFIER_TRIP_THRESHOLD: z.coerce.number().int().nonnegative().default(100),
     TRIGGER_MOLLIFIER_HOLD_MS: z.coerce.number().int().positive().default(500),
     TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY: z.coerce.number().int().positive().default(50),
-    TRIGGER_MOLLIFIER_ENTRY_TTL_S: z.coerce.number().int().positive().default(600),
     TRIGGER_MOLLIFIER_DRAIN_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3),
     TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(30_000),
     TRIGGER_MOLLIFIER_DRAIN_MAX_ORGS_PER_TICK: z.coerce.number().int().positive().default(500),
 
+    // Periodic sweep that scans buffer queue ZSETs for entries whose
+    // dwell exceeds the stale threshold. Independent of the drainer —
+    // its job is exactly to make a stuck/offline drainer visible to
+    // ops. Defaults: enabled when the mollifier is enabled, run every
+    // 5 minutes, alert on anything that's been dwelling for 5+ minutes
+    // (matches the sweep interval — "anything still here when we
+    // check" is the simplest threshold that converges).
+    TRIGGER_MOLLIFIER_STALE_SWEEP_ENABLED: z
+      .string()
+      .default(process.env.TRIGGER_MOLLIFIER_ENABLED ?? "0"),
+    TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(5 * 60_000),
+    TRIGGER_MOLLIFIER_STALE_SWEEP_THRESHOLD_MS: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(5 * 60_000),
+
     BATCH_TRIGGER_PROCESS_JOB_VISIBILITY_TIMEOUT_MS: z.coerce
       .number()
       .int()

diff --git a/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts
@@ -15,6 +15,10 @@ import assertNever from "assert-never";
 import { API_VERSIONS, CURRENT_API_VERSION, RunStatusUnspecifiedApiVersion } from "~/api/versions";
 import { $replica, prisma } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import {
+  findRunByIdWithMollifierFallback,
+  type SyntheticRun,
+} from "~/v3/mollifier/readFallback.server";
 import { generatePresignedUrl } from "~/v3/objectStore.server";
 import { tracer } from "~/v3/tracer.server";
 import { startSpanWithEnv } from "~/v3/tracing.server";
@@ -64,13 +68,34 @@ type CommonRelatedRun = Prisma.Result<
   "findFirstOrThrow"
 >;
 
-type FoundRun = NonNullable<Awaited<ReturnType<typeof ApiRetrieveRunPresenter.findRun>>>;
+// Full shape returned by findRun() — the commonRunSelect fields plus the
+// extras the route handler reads. Declared explicitly (not inferred via
+// ReturnType<typeof findRun>) so findRun can return a synthesised buffered
+// run without the type becoming self-referential.
+type FoundRun = CommonRelatedRun & {
+  traceId: string;
+  payload: string;
+  payloadType: string;
+  output: string | null;
+  outputType: string;
+  error: Prisma.JsonValue;
+  attempts: { id: string }[];
+  attemptNumber: number | null;
+  engine: "V1" | "V2";
+  taskEventStore: string;
+  parentTaskRun: CommonRelatedRun | null;
+  rootTaskRun: CommonRelatedRun | null;
+  childRuns: CommonRelatedRun[];
+};
 
 export class ApiRetrieveRunPresenter {
   constructor(private readonly apiVersion: API_VERSIONS) {}
 
-  public static async findRun(friendlyId: string, env: AuthenticatedEnvironment) {
-    return $replica.taskRun.findFirst({
+  public static async findRun(
+    friendlyId: string,
+    env: AuthenticatedEnvironment,
+  ): Promise<FoundRun | null> {
+    const pgRow = await $replica.taskRun.findFirst({
       where: {
         friendlyId,
         runtimeEnvironmentId: env.id,
@@ -102,6 +127,23 @@ export class ApiRetrieveRunPresenter {
         },
       },
     });
+
+    if (pgRow) return pgRow;
+
+    // Postgres miss → fall back to the mollifier buffer. When the gate
+    // diverted a trigger, the run lives in Redis until the drainer replays
+    // it through engine.trigger. Synthesise the FoundRun shape so call()
+    // returns a `QUEUED` (or `FAILED`) response with empty output, no
+    // attempts, no relations.
+    const buffered = await findRunByIdWithMollifierFallback({
+      runId: friendlyId,
+      environmentId: env.id,
+      organizationId: env.organizationId,
+    });
+
+    if (!buffered) return null;
+
+    return synthesiseFoundRunFromBuffer(buffered);
   }
 
   public async call(taskRun: FoundRun, env: AuthenticatedEnvironment) {
@@ -475,3 +517,75 @@ function resolveTriggerFunction(run: CommonRelatedRun): TriggerFunction {
     return run.resumeParentOnCompletion ? "triggerAndWait" : "trigger";
   }
 }
+
+// Build a FoundRun-shaped object from a buffered (mollified) run. The run
+// is in the Redis buffer; engine.trigger hasn't created the Postgres row
+// yet, so every field that comes from execution state (output, attempts,
+// completedAt, cost, relations) takes a default. The presenter's call()
+// handles QUEUED-state runs without surprise.
+function bufferedStatusToTaskRunStatus(status: SyntheticRun["status"]): TaskRunStatus {
+  switch (status) {
+    case "FAILED":
+      return "SYSTEM_FAILURE";
+    case "CANCELED":
+      return "CANCELED";
+    default:
+      return "PENDING";
+  }
+}
+
+function synthesiseFoundRunFromBuffer(buffered: SyntheticRun): FoundRun {
+  const status: TaskRunStatus = bufferedStatusToTaskRunStatus(buffered.status);
+
+  const errorJson: Prisma.JsonValue = buffered.error
+    ? {
+        type: "STRING_ERROR",
+        raw: `${buffered.error.code}: ${buffered.error.message}`,
+      }
+    : null;
+
+  const metadata: Prisma.JsonValue =
+    typeof buffered.metadata === "string" ? buffered.metadata : null;
+
+  return {
+    id: buffered.friendlyId,
+    friendlyId: buffered.friendlyId,
+    status,
+    taskIdentifier: buffered.taskIdentifier ?? "",
+    createdAt: buffered.createdAt,
+    startedAt: null,
+    updatedAt: buffered.cancelledAt ?? buffered.createdAt,
+    completedAt: buffered.cancelledAt ?? null,
+    expiredAt: null,
+    delayUntil: buffered.delayUntil ?? null,
+    metadata,
+    metadataType: buffered.metadataType ?? "application/json",
+    ttl: buffered.ttl ?? null,
+    costInCents: 0,
+    baseCostInCents: 0,
+    usageDurationMs: 0,
+    idempotencyKey: buffered.idempotencyKey ?? null,
+    idempotencyKeyOptions: buffered.idempotencyKeyOptions ?? null,
+    isTest: buffered.isTest,
+    depth: buffered.depth,
+    scheduleId: null,
+    lockedToVersion: buffered.lockedToVersion ? { version: buffered.lockedToVersion } : null,
+    resumeParentOnCompletion: buffered.resumeParentOnCompletion,
+    batch: null,
+    runTags: buffered.tags,
+    traceId: buffered.traceId ?? "",
+    payload: typeof buffered.payload === "string" ? buffered.payload : "",
+    payloadType: buffered.payloadType ?? "application/json",
+    output: null,
+    outputType: "application/json",
+    error: errorJson,
+    attempts: [],
+    attemptNumber: null,
+    engine: "V2",
+    taskEventStore: "taskEvent",
+    workerQueue: buffered.workerQueue ?? "main",
+    parentTaskRun: null,
+    rootTaskRun: null,
+    childRuns: [],
+  };
+}
diff --git a/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts
@@ -151,7 +151,8 @@ export const ApiRunListSearchParams = z.object({
     }),
 });
 
-type ApiRunListSearchParams = z.infer<typeof ApiRunListSearchParams>;
+export type ApiRunListSearchParamsType = z.infer<typeof ApiRunListSearchParams>;
+type ApiRunListSearchParams = ApiRunListSearchParamsType;
 
 export class ApiRunListPresenter extends BasePresenter {
   public async call(

diff --git a/apps/webapp/app/presenters/v3/RunStreamPresenter.server.ts b/apps/webapp/app/presenters/v3/RunStreamPresenter.server.ts
@@ -3,6 +3,8 @@ import { logger } from "~/services/logger.server";
 import { singleton } from "~/utils/singleton";
 import { ABORT_REASON_SEND_ERROR, createSSELoader, SendFunction } from "~/utils/sse";
 import { throttle } from "~/utils/throttle";
+import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
+import { deserialiseSnapshot } from "@trigger.dev/redis-worker";
 import { tracePubSub } from "~/v3/services/tracePubSub.server";
 
 const PING_INTERVAL = 5_000;
@@ -37,17 +39,45 @@ export class RunStreamPresenter {
           },
         });
 
-        if (!run) {
+        // Fall back to the mollifier buffer when the run isn't in PG yet.
+        // The buffered run has no execution events to stream, but we still
+        // attach a trace-pubsub subscription using the snapshot's traceId
+        // so that the moment the drainer materialises the row and execution
+        // begins, those events flow to this open SSE connection. Closing
+        // with 404 would force the dashboard to keep retrying.
+        let traceId: string | null = run?.traceId ?? null;
+        if (!traceId) {
+          const buffer = getMollifierBuffer();
+          if (buffer) {
+            try {
+              const entry = await buffer.getEntry(runFriendlyId);
+              if (entry) {
+                const snapshot = deserialiseSnapshot<{ traceId?: string }>(entry.payload);
+                if (typeof snapshot.traceId === "string") {
+                  traceId = snapshot.traceId;
+                }
+              }
+            } catch (err) {
+              logger.warn("RunStreamPresenter buffer fallback failed", {
+                runFriendlyId,
+                err: err instanceof Error ? err.message : String(err),
+              });
+            }
+          }
+        }
+
+        if (!traceId) {
           throw new Response("Not found", { status: 404 });
         }
+        const resolvedRun = { traceId };
 
         logger.info("RunStreamPresenter.start", {
           runFriendlyId,
-          traceId: run.traceId,
+          traceId: resolvedRun.traceId,
         });
 
         // Subscribe to trace updates
-        const { unsubscribe, eventEmitter } = await tracePubSub.subscribeToTrace(run.traceId);
+        const { unsubscribe, eventEmitter } = await tracePubSub.subscribeToTrace(resolvedRun.traceId);
 
         // Only send max every 1 second
         const throttledSend = throttle(
@@ -105,7 +135,7 @@ export class RunStreamPresenter {
           cleanup: () => {
             logger.info("RunStreamPresenter.cleanup", {
               runFriendlyId,
-              traceId: run.traceId,
+              traceId: resolvedRun.traceId,
             });
 
             // Remove message listener
@@ -119,13 +149,13 @@ export class RunStreamPresenter {
               .then(() => {
                 logger.info("RunStreamPresenter.cleanup.unsubscribe succeeded", {
                   runFriendlyId,
-                  traceId: run.traceId,
+                  traceId: resolvedRun.traceId,
                 });
               })
               .catch((error) => {
                 logger.error("RunStreamPresenter.cleanup.unsubscribe failed", {
                   runFriendlyId,
-                  traceId: run.traceId,
+                  traceId: resolvedRun.traceId,
                   error: {
                     name: error.name,
                     message: error.message,