feat(mollifier): trigger burst smoothing — Phase 1 (trip evaluator + dual-write monitoring + drainer ack loop)

Phase 1 of the trigger-burst smoothing initiative. Adds the A-side trip
evaluator (atomic Lua sliding-window per env) and wires it into the trigger
hot path. When the per-org mollifierEnabled feature flag is on AND the
evaluator says divert, the canonical replay payload is buffered to Redis
(via buffer.accept) AND the trigger continues through engine.trigger —
i.e. dual-write. The drainer pops + acks (no-op handler) to prove the
dequeue mechanism works end-to-end. Operators audit by joining
mollifier.buffered (write) and mollifier.drained (consume) logs by runId.

Buffer primitives hardened:
- accept is idempotent on duplicate runId (Lua EXISTS guard)
- pop skips orphan queue references (entry HASH TTL'd while runId queued)
- fail no-ops on missing entry (no partial FAILED hash leak)
- mollifier:envs set pruned on draining pop, restored on requeue
- 16-row truth-table test enumerates the gate cascade
- BufferedTriggerPayload defines the canonical replay shape Phase 2 will
  use to invoke engine.trigger
- payload hash for audit-equivalence computed off the hot path (in the
  drainer) to avoid CPU during a spike

Regression tests in apps/webapp/test/engine/triggerTask.test.ts pin the
mollifier integration:
- validation throws BEFORE the gate runs (no orphan buffer write on
  rejected triggers)
- mollify dual-write happy path (Postgres + Redis both reflect the run)
- pass_through path does NOT call buffer.accept
- engine.trigger throwing AFTER buffer.accept leaves an orphan
  (documented behaviour — drainer auto-cleans; audit-trail surfaces it)
- idempotency-key match short-circuits BEFORE the gate is consulted
- debounce match produces an orphan (documented behaviour — Phase 2
  must lift handleDebounce upfront before buffer.accept)

Behaviour with MOLLIFIER_ENABLED=0 (default) is byte-identical to main.
With MOLLIFIER_ENABLED=1 and the flag off, only mollifier.would_mollify
logs fire (no buffer state). With the flag on, dual-write activates.

Includes two opt-in *.fuzz.test.ts suites (gated on FUZZ=1) that
randomise operation sequences against evaluateTrip and the drainer to
find timing edges. They are clearly marked TEMPORARY in their headers.

Author: d-cs

.changeset/mollifier-redis-worker-primitives.md

@@ -2,4 +2,4 @@
 "@trigger.dev/redis-worker": patch
 ---
 
-Add MollifierBuffer and MollifierDrainer primitives for burst smoothing (scaffolding only — not active without webapp wiring).
+Add MollifierBuffer (with `accept`, `pop`, `ack`, `requeue`, `fail`, and `evaluateTrip`) and MollifierDrainer primitives for trigger burst smoothing. `evaluateTrip` is an atomic Lua sliding-window trip evaluator used by the webapp gate to detect per-env trigger bursts. Webapp shadow-mode logging is wired; buffer writes and drainer activation are deferred to a follow-up.

.server-changes/mollifier-phase-2-shadow-mode.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Wire the real A-side trip evaluator into the mollifier gate. With `MOLLIFIER_SHADOW_MODE=1`, each trigger evaluates the per-env sliding-window rate counter; bursts above threshold are logged as `mollifier.would_mollify` (no buffer write — phase 3 activates that). Emits the `mollifier.decisions` OTel counter. Behaviour with `MOLLIFIER_ENABLED=0` (default) is unchanged.

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -40,9 +40,21 @@ import type {
   TriggerTaskRequest,
   TriggerTaskValidator,
 } from "../types";
-import { evaluateGate } from "~/v3/mollifier/mollifierGate.server";
+import {
+  evaluateGate as defaultEvaluateGate,
+  type GateOutcome,
+} from "~/v3/mollifier/mollifierGate.server";
+import { getMollifierBuffer as defaultGetMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
+import { buildBufferedTriggerPayload } from "~/v3/mollifier/bufferedTriggerPayload.server";
+import { serialiseSnapshot, type MollifierBuffer } from "@trigger.dev/redis-worker";
 import { QueueSizeLimitExceededError, ServiceValidationError } from "~/v3/services/common.server";
 
+export type MollifierEvaluateGate = (
+  inputs: { envId: string; orgId: string; taskId: string },
+) => Promise<GateOutcome>;
+
+export type MollifierGetBuffer = () => MollifierBuffer | null;
+
 class NoopTriggerRacepointSystem implements TriggerRacepointSystem {
   async waitForRacepoint(options: { racepoint: TriggerRacepoints; id: string }): Promise<void> {
     return;
@@ -60,6 +72,11 @@ export class RunEngineTriggerTaskService {
   private readonly traceEventConcern: TraceEventConcern;
   private readonly triggerRacepointSystem: TriggerRacepointSystem;
   private readonly metadataMaximumSize: number;
+  // Mollifier hooks are DI'd so tests can drive the call-site's mollify branch
+  // deterministically (stub the gate to return mollify, inject a real or fake
+  // buffer). In production both default to the live module-level singletons.
+  private readonly evaluateGate: MollifierEvaluateGate;
+  private readonly getMollifierBuffer: MollifierGetBuffer;
 
   constructor(opts: {
     prisma: PrismaClientOrTransaction;
@@ -72,6 +89,8 @@ export class RunEngineTriggerTaskService {
     tracer: Tracer;
     metadataMaximumSize: number;
     triggerRacepointSystem?: TriggerRacepointSystem;
+    evaluateGate?: MollifierEvaluateGate;
+    getMollifierBuffer?: MollifierGetBuffer;
   }) {
     this.prisma = opts.prisma;
     this.engine = opts.engine;
@@ -83,6 +102,8 @@ export class RunEngineTriggerTaskService {
     this.traceEventConcern = opts.traceEventConcern;
     this.metadataMaximumSize = opts.metadataMaximumSize;
     this.triggerRacepointSystem = opts.triggerRacepointSystem ?? new NoopTriggerRacepointSystem();
+    this.evaluateGate = opts.evaluateGate ?? defaultEvaluateGate;
+    this.getMollifierBuffer = opts.getMollifierBuffer ?? defaultGetMollifierBuffer;
   }
 
   public async call({
@@ -316,15 +337,11 @@ export class RunEngineTriggerTaskService {
         rootScheduleId: parentAnnotations?.rootScheduleId || options.scheduleId || undefined,
       };
 
-      const mollifierOutcome = await evaluateGate({
+      const mollifierOutcome = await this.evaluateGate({
         envId: environment.id,
         orgId: environment.organizationId,
+        taskId,
       });
-      if (mollifierOutcome.action === "mollify") {
-        throw new Error(
-          "MollifierGate.mollify reached in phase 1 — should be unreachable until phase 3 wiring lands",
-        );
-      }
 
       try {
         return await this.traceEventConcern.traceRun(
@@ -338,6 +355,74 @@ export class RunEngineTriggerTaskService {
 
             const payloadPacket = await this.payloadProcessor.process(triggerRequest);
 
+            // Phase 1 dual-write: if the org has the mollifier feature flag
+            // enabled and the per-env trip evaluator says divert, write the
+            // canonical replay payload to the buffer AND continue through
+            // engine.trigger as normal. The buffer entry is an audit/preview
+            // copy; the drainer's no-op handler consumes it to prove the
+            // dequeue mechanism works. Phase 2 will replace engine.trigger
+            // (below) with a synthesised 200 response and rely on the
+            // drainer to perform the Postgres write via replay.
+            if (mollifierOutcome.action === "mollify") {
+              const buffer = this.getMollifierBuffer();
+              if (buffer) {
+                const canonicalPayload = buildBufferedTriggerPayload({
+                  runFriendlyId,
+                  taskId,
+                  envId: environment.id,
+                  envType: environment.type,
+                  envSlug: environment.slug,
+                  orgId: environment.organizationId,
+                  orgSlug: environment.organization.slug,
+                  projectId: environment.projectId,
+                  projectRef: environment.project.externalRef,
+                  body,
+                  idempotencyKey: idempotencyKey ?? null,
+                  idempotencyKeyExpiresAt: idempotencyKey
+                    ? idempotencyKeyExpiresAt ?? null
+                    : null,
+                  tags,
+                  parentRunFriendlyId: parentRun?.friendlyId ?? null,
+                  traceContext: event.traceContext,
+                  triggerSource,
+                  triggerAction,
+                  serviceOptions: options,
+                  createdAt: new Date(),
+                });
+
+                try {
+                  const serialisedPayload = serialiseSnapshot(canonicalPayload);
+                  await buffer.accept({
+                    runId: runFriendlyId,
+                    envId: environment.id,
+                    orgId: environment.organizationId,
+                    payload: serialisedPayload,
+                  });
+                  // Light log on the hot path — keep this synchronous work
+                  // O(1) per trigger. The drainer computes the payload hash
+                  // off-path; operators correlate `mollifier.buffered` →
+                  // `mollifier.drained` by runId.
+                  logger.info("mollifier.buffered", {
+                    runId: runFriendlyId,
+                    envId: environment.id,
+                    orgId: environment.organizationId,
+                    taskId,
+                    payloadBytes: serialisedPayload.length,
+                  });
+                } catch (err) {
+                  // Fail-open: buffer write must never block the customer's
+                  // trigger. engine.trigger below is the primary write path
+                  // in Phase 1 — the customer still gets a valid run.
+                  logger.error("mollifier.buffer_accept_failed", {
+                    runId: runFriendlyId,
+                    envId: environment.id,
+                    taskId,
+                    err: err instanceof Error ? err.message : String(err),
+                  });
+                }
+              }
+            }
+
             const taskRun = await this.engine.trigger(
               {
                 friendlyId: runFriendlyId,

apps/webapp/app/v3/mollifier/bufferedTriggerPayload.server.ts

@@ -0,0 +1,106 @@
+import type { TriggerTaskRequestBody } from "@trigger.dev/core/v3";
+import type { TriggerTaskServiceOptions } from "~/v3/services/triggerTask.server";
+
+// Canonical payload shape written to the mollifier buffer when the gate
+// decides to mollify a trigger. Phase 1 ALSO calls engine.trigger directly
+// (dual-write) so this is currently an audit/preview record. Phase 2 will
+// make the buffer the primary write path: the drainer's handler will read
+// this payload and replay it through engine.trigger to create the run in
+// Postgres, and read-fallback endpoints will synthesise a Run view from it
+// while it is still QUEUED.
+//
+// CONTRACT: this shape must contain everything needed for Phase 2's
+// drainer-replay to reconstruct an equivalent engine.trigger call. Phase 1
+// emits it to logs; Phase 2 will serialise it into Redis and rebuild it on
+// the drain side. Keep it serialisable — no functions, no class instances.
+export type BufferedTriggerPayload = {
+  runFriendlyId: string;
+
+  // Routing identifiers — let the drainer re-fetch full AuthenticatedEnvironment
+  // at replay time rather than embedding it in the payload.
+  envId: string;
+  envType: string;
+  envSlug: string;
+  orgId: string;
+  orgSlug: string;
+  projectId: string;
+  projectRef: string;
+
+  // Task identifier — looked up against the locked BackgroundWorkerTask
+  // at replay time to recover task-defaults.
+  taskId: string;
+
+  // Customer-supplied trigger body (payload, options, context).
+  body: TriggerTaskRequestBody;
+
+  // Resolved values from upstream concerns. The drainer should NOT re-resolve
+  // these — that would create a second idempotency-key check, etc.
+  idempotencyKey: string | null;
+  idempotencyKeyExpiresAt: string | null;
+  tags: string[];
+
+  // Parent/root linkage for nested triggers.
+  parentRunFriendlyId: string | null;
+
+  // Trace context — propagates the original triggering span across the
+  // buffer→drain boundary so the run's lifecycle stays under one trace.
+  traceContext: Record<string, unknown>;
+
+  // Annotations + service options that influence routing/replay.
+  triggerSource: string;
+  triggerAction: string;
+  serviceOptions: TriggerTaskServiceOptions;
+
+  // Wall-clock instants relevant to the run.
+  createdAt: string;
+};
+
+// Assemble the canonical payload from the inputs available at the point
+// `evaluateGate` returns "mollify" in `RunEngineTriggerTaskService.call`.
+// All fields must be derivable from data already in scope at that call site;
+// nothing should require an extra DB lookup.
+export function buildBufferedTriggerPayload(input: {
+  runFriendlyId: string;
+  taskId: string;
+  envId: string;
+  envType: string;
+  envSlug: string;
+  orgId: string;
+  orgSlug: string;
+  projectId: string;
+  projectRef: string;
+  body: TriggerTaskRequestBody;
+  idempotencyKey: string | null;
+  idempotencyKeyExpiresAt: Date | null;
+  tags: string[];
+  parentRunFriendlyId: string | null;
+  traceContext: Record<string, unknown>;
+  triggerSource: string;
+  triggerAction: string;
+  serviceOptions: TriggerTaskServiceOptions;
+  createdAt: Date;
+}): BufferedTriggerPayload {
+  return {
+    runFriendlyId: input.runFriendlyId,
+    envId: input.envId,
+    envType: input.envType,
+    envSlug: input.envSlug,
+    orgId: input.orgId,
+    orgSlug: input.orgSlug,
+    projectId: input.projectId,
+    projectRef: input.projectRef,
+    taskId: input.taskId,
+    body: input.body,
+    idempotencyKey: input.idempotencyKey,
+    idempotencyKeyExpiresAt: input.idempotencyKeyExpiresAt
+      ? input.idempotencyKeyExpiresAt.toISOString()
+      : null,
+    tags: input.tags,
+    parentRunFriendlyId: input.parentRunFriendlyId,
+    traceContext: input.traceContext,
+    triggerSource: input.triggerSource,
+    triggerAction: input.triggerAction,
+    serviceOptions: input.serviceOptions,
+    createdAt: input.createdAt.toISOString(),
+  };
+}

apps/webapp/app/v3/mollifier/mollifierDrainer.server.ts

@@ -1,10 +1,12 @@
-import { MollifierDrainer } from "@trigger.dev/redis-worker";
+import { createHash } from "node:crypto";
+import { MollifierDrainer, serialiseSnapshot } from "@trigger.dev/redis-worker";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { singleton } from "~/utils/singleton";
 import { getMollifierBuffer } from "./mollifierBuffer.server";
+import type { BufferedTriggerPayload } from "./bufferedTriggerPayload.server";
 
-function initializeMollifierDrainer(): MollifierDrainer {
+function initializeMollifierDrainer(): MollifierDrainer<BufferedTriggerPayload> {
   const buffer = getMollifierBuffer();
   if (!buffer) {
     // Should be unreachable: getMollifierDrainer() guards on the same env flag as getMollifierBuffer().
@@ -16,21 +18,46 @@ function initializeMollifierDrainer(): MollifierDrainer {
     maxAttempts: env.MOLLIFIER_DRAIN_MAX_ATTEMPTS,
   });
 
-  const drainer = new MollifierDrainer({
+  // Phase 1 handler: no-op ack. The trigger has ALREADY been written to
+  // Postgres via engine.trigger (dual-write at the call site). Popping +
+  // acking here proves the dequeue mechanism works end-to-end without
+  // duplicating the work. Phase 2 will replace this with an engine.trigger
+  // replay that performs the actual Postgres write.
+  const drainer = new MollifierDrainer<BufferedTriggerPayload>({
     buffer,
-    handler: async () => {
-      throw new Error("MollifierDrainer phase 1: no handler wired");
+    handler: async (input) => {
+      // Hash the (re-serialised, canonical) payload on the drain side rather
+      // than on the trigger hot path. Burst-time CPU stays with engine.trigger;
+      // the drainer is the natural place for the audit-equivalence checksum.
+      // Re-serialisation is identity for the BufferedTriggerPayload shape
+      // (only strings/numbers/plain objects), so this hash matches what the
+      // call site wrote into Redis.
+      const reserialised = serialiseSnapshot(input.payload);
+      const payloadHash = createHash("sha256").update(reserialised).digest("hex");
+      logger.info("mollifier.drained", {
+        runId: input.runId,
+        envId: input.envId,
+        orgId: input.orgId,
+        taskId: input.payload.taskId,
+        attempts: input.attempts,
+        ageMs: Date.now() - input.createdAt.getTime(),
+        payloadBytes: reserialised.length,
+        payloadHash,
+      });
     },
     concurrency: env.MOLLIFIER_DRAIN_CONCURRENCY,
     maxAttempts: env.MOLLIFIER_DRAIN_MAX_ATTEMPTS,
+    // A no-op handler shouldn't throw, but if something does (e.g. an
+    // unexpected deserialise failure), don't loop — let it FAIL terminally
+    // so the entry is observable in metrics.
     isRetryable: () => false,
   });
 
   drainer.start();
   return drainer;
 }
 
-export function getMollifierDrainer(): MollifierDrainer | null {
+export function getMollifierDrainer(): MollifierDrainer<BufferedTriggerPayload> | null {
   if (env.MOLLIFIER_ENABLED !== "1") return null;
   return singleton("mollifierDrainer", initializeMollifierDrainer);
 }