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); }