From d76bb9e4ea1ddac314eaa8ba549dd9651cf6db28 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Thu, 14 May 2026 13:18:18 +0100 Subject: [PATCH] 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. --- .../mollifier-redis-worker-primitives.md | 2 + .../src/mollifier/drainer.test.ts | 111 +++++++++++++++++- .../redis-worker/src/mollifier/drainer.ts | 44 ++++++- 3 files changed, 150 insertions(+), 7 deletions(-) diff --git a/.changeset/mollifier-redis-worker-primitives.md b/.changeset/mollifier-redis-worker-primitives.md index 3378750a7..8485a8b22 100644 --- a/.changeset/mollifier-redis-worker-primitives.md +++ b/.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. diff --git a/packages/redis-worker/src/mollifier/drainer.test.ts b/packages/redis-worker/src/mollifier/drainer.test.ts index 73fbf5285..c67cf0327 100644 --- a/packages/redis-worker/src/mollifier/drainer.test.ts +++ b/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 & { [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({ diff --git a/packages/redis-worker/src/mollifier/drainer.ts b/packages/redis-worker/src/mollifier/drainer.ts index e42c1b125..f0af56697 100644 --- a/packages/redis-worker/src/mollifier/drainer.ts +++ b/packages/redis-worker/src/mollifier/drainer.ts @@ -90,21 +90,43 @@ export class MollifierDrainer { } } + // 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 { 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 { return new Promise((resolve) => setTimeout(resolve, ms)); } @@ -115,8 +137,18 @@ export class MollifierDrainer { 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); }