Merge branch 'mollifier-phase-2' into mollifier-phase-3

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

@@ -363,6 +363,88 @@ describe("MollifierDrainer resilience to transient buffer errors", () => {
     expect(result.failed).toBe(1);
     expect(handled).toEqual(["run_good"]);
   });
+
+  it("a requeue failure during retry recovery doesn't poison the rest of the batch", async () => {
+    // Regression: handler throws a retryable error → processEntry calls
+    // buffer.requeue() inside its catch block. If requeue() itself throws
+    // (Redis blip during error recovery), the rejection used to escape
+    // processOneFromEnv unwrapped and reject the runOnce Promise.all,
+    // dropping handler results from sibling envs in the same tick.
+    const handled: string[] = [];
+    const buffer = makeStubBuffer({
+      ...eachEnvAsOwnOrg(["bad", "good"]),
+      pop: async (envId: string) =>
+        ({
+          runId: envId === "bad" ? "run_bad" : "run_good",
+          envId,
+          orgId: "org_1",
+          payload: "{}",
+          attempts: 0,
+          createdAt: new Date(),
+        }) as any,
+      requeue: async () => {
+        throw new Error("simulated requeue failure");
+      },
+    });
+
+    const drainer = new MollifierDrainer({
+      buffer,
+      handler: async (input) => {
+        handled.push(input.runId);
+        if (input.runId === "run_bad") throw new Error("transient");
+      },
+      concurrency: 5,
+      maxAttempts: 3,
+      isRetryable: () => true,
+      logger: new Logger("test-drainer", "log"),
+    });
+
+    const result = await drainer.runOnce();
+    // Two envs scheduled, one handler succeeded (drained), one handler threw
+    // and its recovery requeue threw too — counted as failed, batch not poisoned.
+    expect(result.drained).toBe(1);
+    expect(result.failed).toBe(1);
+    expect(new Set(handled)).toEqual(new Set(["run_bad", "run_good"]));
+  });
+
+  it("a fail() throw during terminal recovery doesn't poison the rest of the batch", async () => {
+    // Regression: handler throws a non-retryable error → processEntry calls
+    // buffer.fail() inside its catch block. If fail() itself throws, the
+    // rejection used to escape unwrapped and reject runOnce's Promise.all.
+    const handled: string[] = [];
+    const buffer = makeStubBuffer({
+      ...eachEnvAsOwnOrg(["bad", "good"]),
+      pop: async (envId: string) =>
+        ({
+          runId: envId === "bad" ? "run_bad" : "run_good",
+          envId,
+          orgId: "org_1",
+          payload: "{}",
+          attempts: 0,
+          createdAt: new Date(),
+        }) as any,
+      fail: async () => {
+        throw new Error("simulated fail() failure");
+      },
+    });
+
+    const drainer = new MollifierDrainer({
+      buffer,
+      handler: async (input) => {
+        handled.push(input.runId);
+        if (input.runId === "run_bad") throw new Error("terminal");
+      },
+      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(new Set(handled)).toEqual(new Set(["run_bad", "run_good"]));
+  });
 });
 
 describe("MollifierDrainer per-tick org cap", () => {

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

@@ -226,10 +226,11 @@ export class MollifierDrainer<TPayload = unknown> {
     return sorted[idx]!;
   }
 
-  // A `pop()` failure for one env (e.g. a Redis hiccup mid-batch) must not
+  // A failure for one env (e.g. a Redis hiccup mid-batch in `pop`, or in
+  // `requeue`/`fail` during error recovery inside `processEntry`) 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.
+  // bubble all the way to `loop()`. Catch both stages 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"> {
     let entry: BufferEntry | null;
     try {
@@ -239,7 +240,16 @@ export class MollifierDrainer<TPayload = unknown> {
       return "failed";
     }
     if (!entry) return "empty";
-    return this.processEntry(entry);
+    try {
+      return await this.processEntry(entry);
+    } catch (err) {
+      this.logger.error("MollifierDrainer.processEntry failed", {
+        envId,
+        runId: entry.runId,
+        err,
+      });
+      return "failed";
+    }
   }
 
   private async processEntry(entry: BufferEntry): Promise<"drained" | "failed"> {