fix(mollifier): bound drainer per-tick env fan-out via maxEnvsPerTick

mollifier:envs is a Redis SET that grows with the count of envs that
currently have buffered entries. Under normal operation that's small,
but an extended drainer outage can leave entries piled up across
thousands of envs — at which point runOnce would queue one
processOneFromEnv per env through pLimit, ballooning per-tick latency
and event-loop queue depth.

Cap per-tick fan-out at MOLLIFIER_DRAIN_MAX_ENVS_PER_TICK (default 500).
When the set fits within the cap, behaviour is unchanged (take all,
rotate cursor by 1 for fairness). When the set exceeds the cap, take a
rotating slice and advance the cursor by the slice size so successive
ticks sweep through the full set.

Tests use a stub buffer to drive listEnvs() deterministically with
thousands of envs without provisioning a real Redis.

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

@@ -5,3 +5,5 @@
 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. Phase 1 wires MollifierBuffer dual-write monitoring alongside the real trigger path and runs MollifierDrainer's pop/ack loop end-to-end with a no-op handler; full buffering and replayed drainer-side triggers land in later phases.
 
 MollifierDrainer's polling loop now survives transient Redis errors. `processOneFromEnv` catches `buffer.pop()` failures so one env's hiccup doesn't poison the rest of the batch, and the loop wraps each `runOnce` in a try/catch with capped exponential backoff (up to 5s) instead of dying permanently on the first `listEnvs`/`pop` error.
+
+MollifierDrainer accepts a new `maxEnvsPerTick` option (default 500) that bounds per-tick fan-out across the `mollifier:envs` SET. When the set grows beyond the cap (e.g. after an extended drainer outage left entries piled up across many envs), `runOnce` processes a rotating slice rather than queuing one `processOneFromEnv` job per env, and the cursor advances by the slice size so successive ticks sweep through the full set.

apps/webapp/app/env.server.ts

@@ -1049,6 +1049,7 @@ const EnvironmentSchema = z
     MOLLIFIER_ENTRY_TTL_S: z.coerce.number().int().positive().default(600),
     MOLLIFIER_DRAIN_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3),
     MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(30_000),
+    MOLLIFIER_DRAIN_MAX_ENVS_PER_TICK: z.coerce.number().int().positive().default(500),
 
     BATCH_TRIGGER_PROCESS_JOB_VISIBILITY_TIMEOUT_MS: z.coerce
       .number()

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

@@ -51,6 +51,7 @@ function initializeMollifierDrainer(): MollifierDrainer<BufferedTriggerPayload>
     },
     concurrency: env.MOLLIFIER_DRAIN_CONCURRENCY,
     maxAttempts: env.MOLLIFIER_DRAIN_MAX_ATTEMPTS,
+    maxEnvsPerTick: env.MOLLIFIER_DRAIN_MAX_ENVS_PER_TICK,
     // 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.

packages/redis-worker/src/mollifier/drainer.test.ts

@@ -326,6 +326,119 @@ describe("MollifierDrainer resilience to transient buffer errors", () => {
   });
 });
 
+describe("MollifierDrainer per-tick env cap", () => {
+  // Bounding fan-out prevents one runOnce from queuing thousands of
+  // processOneFromEnv jobs when `mollifier:envs` is unexpectedly large.
+  // These tests use a stub buffer so we can drive the env list count
+  // deterministically without provisioning a real Redis with thousands
+  // of envs.
+  type StubBuffer = Partial<MollifierBuffer> & { [K in keyof MollifierBuffer]?: any };
+  function makeStubBuffer(overrides: StubBuffer): MollifierBuffer {
+    const base: StubBuffer = {
+      listEnvs: async () => [],
+      pop: async () => null,
+      ack: async () => {},
+      requeue: async () => {},
+      fail: async () => true,
+      getEntry: async () => null,
+      close: async () => {},
+    };
+    return { ...base, ...overrides } as unknown as MollifierBuffer;
+  }
+
+  it("processes at most maxEnvsPerTick envs per runOnce", async () => {
+    const allEnvs = Array.from({ length: 20 }, (_, i) => `env_${i}`);
+    const popped: string[] = [];
+    const buffer = makeStubBuffer({
+      listEnvs: async () => allEnvs,
+      pop: async (envId: string) => {
+        popped.push(envId);
+        return null; // empty queue — runOnce records this as "empty"
+      },
+    });
+
+    const drainer = new MollifierDrainer({
+      buffer,
+      handler: async () => {},
+      concurrency: 5,
+      maxAttempts: 3,
+      isRetryable: () => false,
+      maxEnvsPerTick: 5,
+      logger: new Logger("test-drainer", "log"),
+    });
+
+    await drainer.runOnce();
+    expect(popped).toHaveLength(5);
+  });
+
+  it("rotates through the full set across successive ticks when sliced", async () => {
+    const allEnvs = Array.from({ length: 12 }, (_, i) => `env_${i}`);
+    const popped: string[] = [];
+    const buffer = makeStubBuffer({
+      listEnvs: async () => allEnvs,
+      pop: async (envId: string) => {
+        popped.push(envId);
+        return null;
+      },
+    });
+
+    const drainer = new MollifierDrainer({
+      buffer,
+      handler: async () => {},
+      concurrency: 4,
+      maxAttempts: 3,
+      isRetryable: () => false,
+      maxEnvsPerTick: 4,
+      logger: new Logger("test-drainer", "log"),
+    });
+
+    // Three ticks = 12 / 4 → exactly one full sweep.
+    await drainer.runOnce();
+    await drainer.runOnce();
+    await drainer.runOnce();
+
+    expect(new Set(popped)).toEqual(new Set(allEnvs));
+    expect(popped).toHaveLength(12);
+  });
+
+  it("takes all envs and rotates by 1 when the set fits within the cap", async () => {
+    const allEnvs = ["env_a", "env_b", "env_c"];
+    const popsPerTick: string[][] = [];
+    let tick: string[] = [];
+    const buffer = makeStubBuffer({
+      listEnvs: async () => allEnvs,
+      pop: async (envId: string) => {
+        tick.push(envId);
+        return null;
+      },
+    });
+
+    const drainer = new MollifierDrainer({
+      buffer,
+      handler: async () => {},
+      concurrency: 3,
+      maxAttempts: 3,
+      isRetryable: () => false,
+      maxEnvsPerTick: 100, // way above n
+      logger: new Logger("test-drainer", "log"),
+    });
+
+    for (let i = 0; i < 3; i++) {
+      tick = [];
+      await drainer.runOnce();
+      popsPerTick.push(tick);
+    }
+
+    // Every tick covers every env (because cap > n), but the head-of-line
+    // env rotates by 1 each tick — preserves the original fairness behaviour.
+    for (const popped of popsPerTick) {
+      expect(new Set(popped)).toEqual(new Set(allEnvs));
+    }
+    expect(popsPerTick[0][0]).not.toEqual(popsPerTick[1][0]);
+    expect(popsPerTick[1][0]).not.toEqual(popsPerTick[2][0]);
+  });
+});
+
 describe("MollifierDrainer.start/stop", () => {
   redisTest("start polls and processes, stop halts the loop", { timeout: 20_000 }, async ({ redisContainer }) => {
     const buffer = new MollifierBuffer({

packages/redis-worker/src/mollifier/drainer.ts

@@ -19,6 +19,15 @@ export type MollifierDrainerOptions<TPayload> = {
   maxAttempts: number;
   isRetryable: (err: unknown) => boolean;
   pollIntervalMs?: number;
+  // Cap on how many envs `runOnce` processes per tick. When the
+  // `mollifier:envs` SET grows large (e.g. an extended drainer outage left
+  // entries piled up across thousands of envs), an uncapped fan-out queues
+  // one `processOneFromEnv` job per env through `pLimit`, ballooning
+  // per-tick latency and event-loop queue depth. With this cap the
+  // drainer rotates through the full set across multiple ticks instead.
+  // Defaults to 500; size for "typical worst-case envs-with-pending-
+  // entries" rather than total system env count.
+  maxEnvsPerTick?: number;
   logger?: Logger;
 };
 
@@ -33,6 +42,7 @@ export class MollifierDrainer<TPayload = unknown> {
   private readonly maxAttempts: number;
   private readonly isRetryable: (err: unknown) => boolean;
   private readonly pollIntervalMs: number;
+  private readonly maxEnvsPerTick: number;
   private readonly logger: Logger;
   private readonly limit: ReturnType<typeof pLimit>;
   private envCursor = 0;
@@ -45,6 +55,7 @@ export class MollifierDrainer<TPayload = unknown> {
     this.maxAttempts = options.maxAttempts;
     this.isRetryable = options.isRetryable;
     this.pollIntervalMs = options.pollIntervalMs ?? 100;
+    this.maxEnvsPerTick = options.maxEnvsPerTick ?? 500;
     this.logger = options.logger ?? new Logger("MollifierDrainer", "debug");
     this.limit = pLimit(options.concurrency);
   }
@@ -53,7 +64,7 @@ export class MollifierDrainer<TPayload = unknown> {
     const envs = await this.buffer.listEnvs();
     if (envs.length === 0) return { drained: 0, failed: 0 };
 
-    const ordered = this.rotate(envs);
+    const ordered = this.takeRotatingSlice(envs);
 
     const inflight: Promise<"drained" | "failed" | "empty">[] = [];
     for (const envId of ordered) {
@@ -131,10 +142,21 @@ export class MollifierDrainer<TPayload = unknown> {
     return new Promise((resolve) => setTimeout(resolve, ms));
   }
 
-  private rotate(envs: string[]): string[] {
-    const start = this.envCursor % envs.length;
-    this.envCursor = (this.envCursor + 1) % Math.max(envs.length, 1);
-    return [...envs.slice(start), ...envs.slice(0, start)];
+  // Take up to `maxEnvsPerTick` envs starting at the current cursor, with
+  // wrap-around. When the full set fits within the cap we take everything
+  // and advance the cursor by 1 — preserves the original head-of-line
+  // fairness rotation. When we have to slice, we advance the cursor by the
+  // slice size so successive ticks sweep through the full set rather than
+  // re-processing the same prefix on each tick.
+  private takeRotatingSlice(envs: string[]): string[] {
+    const n = envs.length;
+    const sliceSize = Math.min(this.maxEnvsPerTick, n);
+    const start = this.envCursor % n;
+    const advance = sliceSize < n ? sliceSize : 1;
+    this.envCursor = (this.envCursor + advance) % Math.max(n, 1);
+    const end = start + sliceSize;
+    if (end <= n) return envs.slice(start, end);
+    return [...envs.slice(start), ...envs.slice(0, end - n)];
   }
 
   // A `pop()` failure for one env (e.g. a Redis hiccup mid-batch) must not