feat(redis-worker,webapp): mollifier buffer extensions + snapshot type (#3752)

## Summary

Buffer-side data layer used by the rest of the mollifier phase-3 stack.

- `buffer.ts` gains entry inspection (`getEntry`), idempotency lookup
(`lookupIdempotency`), in-place snapshot mutation (`mutateSnapshot`),
and dwell tracking. All atomic via Lua.
- `mollifierSnapshot.server.ts`: shared `MollifierSnapshot` type plus
(de)serialise helpers.
- Drops the entry-TTL config and its env var. The drainer is the
recovery mechanism; an entry that survives the drainer should surface as
a stale-sweep alert, not silently TTL away.

Adds methods to the buffer interface; nothing consumes them yet.
Subsequent PRs in the stack wire trigger-time mollify, read-fallback,
and mutation paths against this surface.

## Test plan

- [x] \`pnpm run typecheck --filter webapp\` passes
- [x] \`pnpm run test --filter @trigger.dev/redis-worker
packages/redis-worker/src/mollifier/buffer.test.ts\` passes

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

.changeset/mollifier-buffer-extensions.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": minor
+---
+
+Mollifier buffer extensions: idempotency dedup, an atomic `mutateSnapshot` API, metadata CAS, claim primitives, and a `MollifierSnapshot` type. The buffer's Redis client now reconnects with jittered backoff so a fleet of clients doesn't stampede Redis in lockstep after a blip.

.changeset/mollifier-buffer-pipeline-list-entries.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+Pipeline the per-entry `HGETALL` fetches in `MollifierBuffer.listEntriesForEnv`. The previous serial implementation issued one Redis round-trip per runId returned by `LRANGE`, which dominated stale-sweep wall-time at any meaningful backlog (at the sweep's default maxCount=1000, this is ~1000 RTTs per env per pass). Behaviour is unchanged — entries are still skipped when the entry hash has been torn down by a concurrent drainer ack/fail between the LRANGE and the HGETALL.

apps/webapp/app/env.server.ts

@@ -1095,7 +1095,6 @@ const EnvironmentSchema = z
     TRIGGER_MOLLIFIER_TRIP_THRESHOLD: z.coerce.number().int().positive().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),

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

@@ -374,13 +374,13 @@ export class RunEngineTriggerTaskService {
 
             const payloadPacket = await this.payloadProcessor.process(triggerRequest);
 
-            // Phase 1 dual-write: if the org has the mollifier feature flag
+            // 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
+            // dequeue mechanism works. A later change replaces engine.trigger
+            // (below) with a synthesised 200 response and relies on the
             // drainer to perform the Postgres write via replay.
             if (mollifierOutcome?.action === "mollify") {
               const buffer = this.getMollifierBuffer();
@@ -430,8 +430,8 @@ export class RunEngineTriggerTaskService {
                   });
                 } 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.
+                  // trigger. engine.trigger below is still the primary write
+                  // path here — the customer still gets a valid run.
                   logger.error("mollifier.buffer_accept_failed", {
                     runId: runFriendlyId,
                     envId: environment.id,

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

@@ -2,17 +2,17 @@ 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.
+// decides to mollify a trigger. At this stage the call site ALSO calls
+// engine.trigger directly (dual-write), so this is currently an
+// audit/preview record. A later change makes the buffer the primary write
+// path: the drainer's handler reads this payload and replays it through
+// engine.trigger to create the run in Postgres, and read-fallback
+// endpoints 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.
+// CONTRACT: this shape must contain everything the drainer-replay needs to
+// reconstruct an equivalent engine.trigger call. Today it is emitted to
+// logs; later it is serialised into Redis and rebuilt on the drain side.
+// Keep it serialisable — no functions, no class instances.
 export type BufferedTriggerPayload = {
   runFriendlyId: string;
 

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

@@ -22,7 +22,6 @@ function initializeMollifierBuffer(): MollifierBuffer {
       enableAutoPipelining: true,
       ...(env.TRIGGER_MOLLIFIER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
     },
-    entryTtlSeconds: env.TRIGGER_MOLLIFIER_ENTRY_TTL_S,
   });
 }
 

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

@@ -68,11 +68,11 @@ function initializeMollifierDrainer(): MollifierDrainer<BufferedTriggerPayload>
     maxAttempts: env.TRIGGER_MOLLIFIER_DRAIN_MAX_ATTEMPTS,
   });
 
-  // 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.
+  // No-op ack handler: 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. A later change replaces this with an engine.trigger replay
+  // that performs the actual Postgres write.
   const drainer = new MollifierDrainer<BufferedTriggerPayload>({
     buffer,
     handler: async (input) => {

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

@@ -73,7 +73,7 @@ export type GateDependencies = {
 };
 
 // `options` is a thunk so env reads happen per-evaluation, not at module load.
-// Don't "simplify" to a plain object — Phase 2 dynamic config relies on the
+// Don't "simplify" to a plain object — dynamic config relies on the
 // gate observing whichever env values are live at trigger time.
 const defaultEvaluator = createRealTripEvaluator({
   getBuffer: () => getMollifierBuffer(),

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

@@ -0,0 +1,16 @@
+import { serialiseSnapshot, deserialiseSnapshot } from "@trigger.dev/redis-worker";
+
+// MollifierSnapshot is the JSON-serialisable shape of the input that would be
+// passed to engine.trigger(). The drainer deserialises and replays it.
+// Kept as Record<string, unknown> at this layer — the engine.trigger call site
+// casts it to the engine's typed input. This keeps the mollifier subdirectory
+// from depending on @internal/run-engine internals.
+export type MollifierSnapshot = Record<string, unknown>;
+
+export function serialiseMollifierSnapshot(input: MollifierSnapshot): string {
+  return serialiseSnapshot(input);
+}
+
+export function deserialiseMollifierSnapshot(serialised: string): MollifierSnapshot {
+  return deserialiseSnapshot<MollifierSnapshot>(serialised);
+}

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

@@ -35,8 +35,8 @@ export function createRealTripEvaluator(deps: CreateRealTripEvaluatorDeps): Trip
     } catch (err) {
       // Deliberate: no error counter here. Shadow mode means a silent miss is
       // harmless — fail-open is the safe direction. The error log + Sentry
-      // capture is sufficient operability for Phase 1. Revisit in Phase 2
-      // when buffer writes are the primary path and a missed evaluation has cost.
+      // capture is sufficient operability while this runs in shadow mode. Revisit
+      // once buffer writes are the primary path and a missed evaluation has cost.
       logger.error("mollifier trip evaluator: fail-open on error", {
         envId: inputs.envId,
         err: err instanceof Error ? err.message : String(err),