feat: trigger mollifier phase 1 scaffolding

Redis-backed burst-smoothing layer behind MOLLIFIER_ENABLED=0 (default).
With the kill switch off, the gate short-circuits on its first env check
and production behaviour is identical to main.

@trigger.dev/redis-worker:
- MollifierBuffer: atomic Lua-backed FIFO with accept / pop / ack /
  requeue / fail + TTL. Per-env queues with HSET entry storage,
  atomic RPOP + status transition, FIFO retry ordering.
- MollifierDrainer: generic round-robin worker with concurrency cap,
  retry semantics, and a stop deadline to avoid livelock on a hung
  handler. Phase 3 will wire the handler to engine.trigger().
- Full testcontainers-backed test suite (21 tests).

apps/webapp:
- evaluateGate cascade-check (kill switch -> org feature flag ->
  shadow mode -> trip evaluator -> mollify / shadow_log / pass_through).
  Dependencies injected for testability; the trip evaluator stub
  returns { divert: false } in phase 1.
- Inserted into RunEngineTriggerTaskService.call() before
  traceEventConcern.traceRun. The mollify branch throws (unreachable
  in phase 1).
- Lazy MollifierBuffer + MollifierDrainer singletons; no Redis
  connection unless MOLLIFIER_ENABLED=1.
- 12 MOLLIFIER_* env vars (all safe defaults) and a mollifierEnabled
  feature flag in the global catalog.
- Drainer booted from worker.server.ts on first import.
- Read-fallback stub for phase 3.
- Gate cascade tests + .env loader so env.server validates in vitest
  workers.

Phase 2 will land the real trip evaluator; phase 3 will activate the
buffer-write + drain path.

Author: d-cs

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

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+Add MollifierBuffer and MollifierDrainer primitives for burst smoothing (scaffolding only — not active without webapp wiring).

.server-changes/mollifier-phase-1-scaffolding.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Add scaffolding for the trigger mollifier (phase 1). New env vars (all default off), `evaluateGate` (the mollifier gate) wired into the trigger hot path as a no-op, lazy singletons for the dedicated mollifier Redis client and drainer. No behavioural change while `MOLLIFIER_ENABLED=0`.

apps/webapp/app/env.server.ts

@@ -1030,6 +1030,34 @@ const EnvironmentSchema = z
     COMMON_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
     COMMON_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
+    MOLLIFIER_ENABLED: z.string().default("0"),
+    MOLLIFIER_SHADOW_MODE: z.string().default("0"),
+    MOLLIFIER_REDIS_HOST: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_HOST),
+    MOLLIFIER_REDIS_PORT: z.coerce
+      .number()
+      .optional()
+      .transform(
+        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined),
+      ),
+    MOLLIFIER_REDIS_USERNAME: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_USERNAME),
+    MOLLIFIER_REDIS_PASSWORD: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_PASSWORD),
+    MOLLIFIER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
+    MOLLIFIER_TRIP_WINDOW_MS: z.coerce.number().int().positive().default(200),
+    MOLLIFIER_TRIP_THRESHOLD: z.coerce.number().int().positive().default(100),
+    MOLLIFIER_HOLD_MS: z.coerce.number().int().positive().default(500),
+    MOLLIFIER_DRAIN_CONCURRENCY: z.coerce.number().int().positive().default(50),
+    MOLLIFIER_ENTRY_TTL_S: z.coerce.number().int().positive().default(600),
+    MOLLIFIER_DRAIN_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3),
+
     BATCH_TRIGGER_PROCESS_JOB_VISIBILITY_TIMEOUT_MS: z.coerce
       .number()
       .int()

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

@@ -40,6 +40,7 @@ import type {
   TriggerTaskRequest,
   TriggerTaskValidator,
 } from "../types";
+import { evaluateGate } from "~/v3/mollifier/mollifierGate.server";
 import { QueueSizeLimitExceededError, ServiceValidationError } from "~/v3/services/common.server";
 
 class NoopTriggerRacepointSystem implements TriggerRacepointSystem {
@@ -315,6 +316,16 @@ export class RunEngineTriggerTaskService {
         rootScheduleId: parentAnnotations?.rootScheduleId || options.scheduleId || undefined,
       };
 
+      const mollifierOutcome = await evaluateGate({
+        envId: environment.id,
+        orgId: environment.organizationId,
+      });
+      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(
           triggerRequest,

apps/webapp/app/services/worker.server.ts

@@ -26,6 +26,7 @@ import { ResumeBatchRunService } from "~/v3/services/resumeBatchRun.server";
 import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.server";
 import { RetryAttemptService } from "~/v3/services/retryAttempt.server";
 import { TimeoutDeploymentService } from "~/v3/services/timeoutDeployment.server";
+import { getMollifierDrainer } from "~/v3/mollifier/mollifierDrainer.server";
 import { GraphileMigrationHelperService } from "./db/graphileMigrationHelper.server";
 import { sendEmail } from "./email.server";
 import { logger } from "./logger.server";
@@ -128,6 +129,8 @@ export async function init() {
   if (env.WORKER_ENABLED === "true") {
     await workerQueue.initialize();
   }
+
+  getMollifierDrainer();
 }
 
 function getWorkerQueue() {

apps/webapp/app/v3/featureFlags.ts

@@ -8,6 +8,7 @@ export const FEATURE_FLAG = {
   hasAiAccess: "hasAiAccess",
   hasComputeAccess: "hasComputeAccess",
   hasPrivateConnections: "hasPrivateConnections",
+  mollifierEnabled: "mollifierEnabled",
 } as const;
 
 export const FeatureFlagCatalog = {
@@ -18,6 +19,7 @@ export const FeatureFlagCatalog = {
   [FEATURE_FLAG.hasAiAccess]: z.coerce.boolean(),
   [FEATURE_FLAG.hasComputeAccess]: z.coerce.boolean(),
   [FEATURE_FLAG.hasPrivateConnections]: z.coerce.boolean(),
+  [FEATURE_FLAG.mollifierEnabled]: z.coerce.boolean(),
 };
 
 export type FeatureFlagKey = keyof typeof FeatureFlagCatalog;

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

@@ -0,0 +1,28 @@
+import { MollifierBuffer } from "@trigger.dev/redis-worker";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { singleton } from "~/utils/singleton";
+
+function initializeMollifierBuffer(): MollifierBuffer {
+  logger.debug("Initializing mollifier buffer", {
+    host: env.MOLLIFIER_REDIS_HOST,
+  });
+
+  return new MollifierBuffer({
+    redisOptions: {
+      keyPrefix: "",
+      host: env.MOLLIFIER_REDIS_HOST,
+      port: env.MOLLIFIER_REDIS_PORT,
+      username: env.MOLLIFIER_REDIS_USERNAME,
+      password: env.MOLLIFIER_REDIS_PASSWORD,
+      enableAutoPipelining: true,
+      ...(env.MOLLIFIER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+    },
+    entryTtlSeconds: env.MOLLIFIER_ENTRY_TTL_S,
+  });
+}
+
+export function getMollifierBuffer(): MollifierBuffer | null {
+  if (env.MOLLIFIER_ENABLED !== "1") return null;
+  return singleton("mollifierBuffer", initializeMollifierBuffer);
+}

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

@@ -0,0 +1,36 @@
+import { MollifierDrainer } 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";
+
+function initializeMollifierDrainer(): MollifierDrainer {
+  const buffer = getMollifierBuffer();
+  if (!buffer) {
+    // Should be unreachable: getMollifierDrainer() guards on the same env flag as getMollifierBuffer().
+    throw new Error("MollifierDrainer initialised without a buffer — env vars inconsistent");
+  }
+
+  logger.debug("Initializing mollifier drainer", {
+    concurrency: env.MOLLIFIER_DRAIN_CONCURRENCY,
+    maxAttempts: env.MOLLIFIER_DRAIN_MAX_ATTEMPTS,
+  });
+
+  const drainer = new MollifierDrainer({
+    buffer,
+    handler: async () => {
+      throw new Error("MollifierDrainer phase 1: no handler wired");
+    },
+    concurrency: env.MOLLIFIER_DRAIN_CONCURRENCY,
+    maxAttempts: env.MOLLIFIER_DRAIN_MAX_ATTEMPTS,
+    isRetryable: () => false,
+  });
+
+  drainer.start();
+  return drainer;
+}
+
+export function getMollifierDrainer(): MollifierDrainer | null {
+  if (env.MOLLIFIER_ENABLED !== "1") return null;
+  return singleton("mollifierDrainer", initializeMollifierDrainer);
+}

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

@@ -0,0 +1,74 @@
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { flag } from "~/v3/featureFlags.server";
+import { FEATURE_FLAG } from "~/v3/featureFlags";
+
+export type TripDecision =
+  | { divert: false }
+  | { divert: true; reason: "per_env_rate" };
+
+export type GateOutcome =
+  | { action: "pass_through" }
+  | { action: "mollify"; decision: Extract<TripDecision, { divert: true }> }
+  | { action: "shadow_log"; decision: Extract<TripDecision, { divert: true }> };
+
+export type GateInputs = {
+  envId: string;
+  orgId: string;
+};
+
+export type TripEvaluator = (inputs: GateInputs) => Promise<TripDecision>;
+
+export type GateDependencies = {
+  isMollifierEnabled: () => boolean;
+  isShadowModeOn: () => boolean;
+  resolveOrgFlag: () => Promise<boolean>;
+  evaluator: TripEvaluator;
+  logShadow: (inputs: GateInputs, reason: "per_env_rate") => void;
+};
+
+const stubTripEvaluator: TripEvaluator = async () => ({ divert: false });
+
+export const defaultGateDependencies: GateDependencies = {
+  isMollifierEnabled: () => env.MOLLIFIER_ENABLED === "1",
+  isShadowModeOn: () => env.MOLLIFIER_SHADOW_MODE === "1",
+  resolveOrgFlag: () =>
+    flag({ key: FEATURE_FLAG.mollifierEnabled, defaultValue: false }),
+  evaluator: stubTripEvaluator,
+  logShadow: (inputs, reason) =>
+    logger.info("mollifier shadow decision", {
+      envId: inputs.envId,
+      orgId: inputs.orgId,
+      reason,
+    }),
+};
+
+export async function evaluateGate(
+  inputs: GateInputs,
+  deps: Partial<GateDependencies> = {},
+): Promise<GateOutcome> {
+  const d = { ...defaultGateDependencies, ...deps };
+
+  if (!d.isMollifierEnabled()) {
+    return { action: "pass_through" };
+  }
+
+  const orgFlagEnabled = await d.resolveOrgFlag();
+  const shadowOn = d.isShadowModeOn();
+
+  if (!orgFlagEnabled && !shadowOn) {
+    return { action: "pass_through" };
+  }
+
+  const decision = await d.evaluator(inputs);
+  if (!decision.divert) {
+    return { action: "pass_through" };
+  }
+
+  if (orgFlagEnabled) {
+    return { action: "mollify", decision };
+  }
+
+  d.logShadow(inputs, decision.reason);
+  return { action: "shadow_log", decision };
+}

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

@@ -0,0 +1,16 @@
+import { logger } from "~/services/logger.server";
+
+export type ReadFallbackInput = {
+  runId: string;
+  environmentId: string;
+  organizationId: string;
+};
+
+export async function findRunByIdWithMollifierFallback(
+  input: ReadFallbackInput,
+): Promise<null> {
+  logger.debug("mollifier read-fallback called (phase 1 stub)", {
+    runId: input.runId,
+  });
+  return null;
+}