diff --git a/.changeset/mollifier-redis-worker-primitives.md b/.changeset/mollifier-redis-worker-primitives.md index 8485a8b22..30bc1a808 100644 --- a/.changeset/mollifier-redis-worker-primitives.md +++ b/.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. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 2989ebe52..5ac1fbf62 100644 --- a/apps/webapp/app/env.server.ts +++ b/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() diff --git a/apps/webapp/app/v3/mollifier/mollifierDrainer.server.ts b/apps/webapp/app/v3/mollifier/mollifierDrainer.server.ts index db8fc5ccb..f3f683349 100644 --- a/apps/webapp/app/v3/mollifier/mollifierDrainer.server.ts +++ b/apps/webapp/app/v3/mollifier/mollifierDrainer.server.ts @@ -51,6 +51,7 @@ function initializeMollifierDrainer(): MollifierDrainer }, 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. diff --git a/packages/redis-worker/src/mollifier/drainer.test.ts b/packages/redis-worker/src/mollifier/drainer.test.ts index c67cf0327..a8b1d5852 100644 --- a/packages/redis-worker/src/mollifier/drainer.test.ts +++ b/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 & { [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({ diff --git a/packages/redis-worker/src/mollifier/drainer.ts b/packages/redis-worker/src/mollifier/drainer.ts index f0af56697..8905625c5 100644 --- a/packages/redis-worker/src/mollifier/drainer.ts +++ b/packages/redis-worker/src/mollifier/drainer.ts @@ -19,6 +19,15 @@ export type MollifierDrainerOptions = { 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 { 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; private envCursor = 0; @@ -45,6 +55,7 @@ export class MollifierDrainer { 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 { 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 { 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