fix(mollifier): keep drainer loop alive across transient redis errors

processOneFromEnv now catches buffer.pop() failures so one env's hiccup
doesn't reject Promise.all and bubble up to the loop's outer catch. The
polling loop itself wraps each runOnce in try/catch and backs off with
capped exponential delay (up to 5s) instead of exiting permanently on the
first listEnvs/pop error. Stop semantics are unchanged: only the stopping
flag breaks the loop.

Adds two regression tests using a stub buffer (no Redis container) so
fault injection is deterministic.

Author: d-cs

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

@@ -3,3 +3,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.

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

@@ -1,5 +1,5 @@
 import { redisTest } from "@internal/testcontainers";
-import { describe, expect, vi } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import { Logger } from "@trigger.dev/core/logger";
 import { MollifierBuffer } from "./buffer.js";
 import { MollifierDrainer } from "./drainer.js";
@@ -217,6 +217,115 @@ describe("MollifierDrainer error handling", () => {
   });
 });
 
+// Transient Redis errors used to permanently kill the loop because
+// `processOneFromEnv` didn't catch `buffer.pop()` rejections — the error
+// bubbled through `Promise.all` → `runOnce` → `loop`'s outer catch and
+// left `isRunning = false`. These tests use a stubbed buffer (no Redis
+// container) so we can deterministically inject failures from `listEnvs`
+// and `pop` without racing against a real client.
+describe("MollifierDrainer resilience to transient buffer errors", () => {
+  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("survives a transient listEnvs failure and resumes draining", async () => {
+    let listCalls = 0;
+    const popped: string[] = [];
+    const buffer = makeStubBuffer({
+      listEnvs: async () => {
+        listCalls += 1;
+        if (listCalls === 1) {
+          throw new Error("simulated redis blip");
+        }
+        return ["env_a"];
+      },
+      pop: async () => {
+        const runId = `run_${popped.length + 1}`;
+        if (popped.length >= 2) return null;
+        popped.push(runId);
+        return {
+          runId,
+          envId: "env_a",
+          orgId: "org_1",
+          payload: "{}",
+          attempts: 0,
+          createdAt: new Date(),
+        } as any;
+      },
+    });
+
+    const handled: string[] = [];
+    const drainer = new MollifierDrainer({
+      buffer,
+      handler: async (input) => {
+        handled.push(input.runId);
+      },
+      concurrency: 1,
+      maxAttempts: 3,
+      isRetryable: () => false,
+      pollIntervalMs: 20,
+      logger: new Logger("test-drainer", "log"),
+    });
+
+    drainer.start();
+    const deadline = Date.now() + 3_000;
+    while (handled.length < 2 && Date.now() < deadline) {
+      await new Promise((r) => setTimeout(r, 20));
+    }
+    await drainer.stop({ timeoutMs: 1_000 });
+
+    expect(handled).toEqual(["run_1", "run_2"]);
+    expect(listCalls).toBeGreaterThan(1);
+  });
+
+  it("a pop failure for one env doesn't poison the rest of the batch", async () => {
+    const buffer = makeStubBuffer({
+      listEnvs: async () => ["bad", "good"],
+      pop: async (envId: string) => {
+        if (envId === "bad") {
+          throw new Error("simulated pop failure on bad env");
+        }
+        return {
+          runId: "run_good",
+          envId: "good",
+          orgId: "org_1",
+          payload: "{}",
+          attempts: 0,
+          createdAt: new Date(),
+        } as any;
+      },
+    });
+
+    const handled: string[] = [];
+    const drainer = new MollifierDrainer({
+      buffer,
+      handler: async (input) => {
+        handled.push(input.runId);
+      },
+      concurrency: 5,
+      maxAttempts: 3,
+      isRetryable: () => false,
+      logger: new Logger("test-drainer", "log"),
+    });
+
+    const result = await drainer.runOnce();
+    expect(result.drained).toBe(1);
+    expect(result.failed).toBe(1);
+    expect(handled).toEqual(["run_good"]);
+  });
+});
+
 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

@@ -90,21 +90,43 @@ export class MollifierDrainer<TPayload = unknown> {
     }
   }
 
+  // Transient Redis errors (e.g. a connection blip in `listEnvs` or `pop`)
+  // must not kill the polling loop permanently. We log each `runOnce`
+  // failure, back off so we don't spin tight on a sustained outage, and
+  // resume. The loop only exits when `stop()` flips `stopping`.
   private async loop(): Promise<void> {
     try {
+      let consecutiveErrors = 0;
       while (!this.stopping) {
-        const result = await this.runOnce();
-        if (result.drained === 0 && result.failed === 0) {
-          await this.delay(this.pollIntervalMs);
+        try {
+          const result = await this.runOnce();
+          consecutiveErrors = 0;
+          if (result.drained === 0 && result.failed === 0) {
+            await this.delay(this.pollIntervalMs);
+          }
+        } catch (err) {
+          consecutiveErrors += 1;
+          this.logger.error("MollifierDrainer.runOnce failed; backing off", {
+            err,
+            consecutiveErrors,
+          });
+          await this.delay(this.backoffMs(consecutiveErrors));
         }
       }
-    } catch (err) {
-      this.logger.error("MollifierDrainer loop crashed", { err });
     } finally {
       this.isRunning = false;
     }
   }
 
+  // Exponential backoff capped at 5s. Keeps the loop responsive after a
+  // brief blip while preventing a tight retry loop during a long Redis
+  // outage. 1 → 200ms, 2 → 400ms, 3 → 800ms, 4 → 1.6s, 5 → 3.2s, 6+ → 5s.
+  private backoffMs(consecutiveErrors: number): number {
+    const base = Math.max(this.pollIntervalMs, 100);
+    const capped = Math.min(base * 2 ** (consecutiveErrors - 1), 5_000);
+    return capped;
+  }
+
   private delay(ms: number): Promise<void> {
     return new Promise((resolve) => setTimeout(resolve, ms));
   }
@@ -115,8 +137,18 @@ export class MollifierDrainer<TPayload = unknown> {
     return [...envs.slice(start), ...envs.slice(0, start)];
   }
 
+  // A `pop()` failure for one env (e.g. a Redis hiccup mid-batch) must not
+  // poison the rest of the batch — `Promise.all` would otherwise reject and
+  // bubble all the way to `loop()`. Catch here so the failed env is just
+  // counted as "failed" for this tick and we move on.
   private async processOneFromEnv(envId: string): Promise<"drained" | "failed" | "empty"> {
-    const entry = await this.buffer.pop(envId);
+    let entry: BufferEntry | null;
+    try {
+      entry = await this.buffer.pop(envId);
+    } catch (err) {
+      this.logger.error("MollifierDrainer.pop failed", { envId, err });
+      return "failed";
+    }
     if (!entry) return "empty";
     return this.processEntry(entry);
   }