From e7344906aa48989724197bbefc94da221ff1a570 Mon Sep 17 00:00:00 2001 From: Dan Sutton Date: Thu, 14 May 2026 16:48:00 +0100 Subject: [PATCH] chore(mollifier): drop fuzz tests to keep phase-1 PR focused drainer.fuzz.test.ts and evaluateTrip.fuzz.test.ts are valuable as ongoing property checks but aren't load-bearing for the phase-1 review. Moving them to a follow-up keeps this PR smaller without losing coverage of the production paths (buffer.test.ts and drainer.test.ts together cover the contract surface). --- .../src/mollifier/drainer.fuzz.test.ts | 184 ------------------ .../src/mollifier/evaluateTrip.fuzz.test.ts | 167 ---------------- 2 files changed, 351 deletions(-) delete mode 100644 packages/redis-worker/src/mollifier/drainer.fuzz.test.ts delete mode 100644 packages/redis-worker/src/mollifier/evaluateTrip.fuzz.test.ts diff --git a/packages/redis-worker/src/mollifier/drainer.fuzz.test.ts b/packages/redis-worker/src/mollifier/drainer.fuzz.test.ts deleted file mode 100644 index 682c0466d..000000000 --- a/packages/redis-worker/src/mollifier/drainer.fuzz.test.ts +++ /dev/null @@ -1,184 +0,0 @@ -// TEMPORARY: fuzz tests for Phase 1 validation of `MollifierDrainer`. -// -// Gated behind `FUZZ=1` so they don't run in CI. Invoke locally with -// `FUZZ=1 pnpm --filter @trigger.dev/redis-worker test src/mollifier/drainer.fuzz` -// during the live-monitoring window before Phase 2. -// -// Targets: drainer must drive every accepted entry to a terminal state -// (acked, FAILED, or TTL-expired) under random handler outcomes and random -// arrival timing across multiple envs. Seeded via SEED. -// Remove once the drainer is stable across two release cycles. - -import { redisTest } from "@internal/testcontainers"; -import { describe, expect, vi } from "vitest"; -import { Logger } from "@trigger.dev/core/logger"; -import { MollifierBuffer } from "./buffer.js"; -import { MollifierDrainer } from "./drainer.js"; -import { serialiseSnapshot } from "./schemas.js"; - -const FUZZ_ENABLED = process.env.FUZZ === "1"; -const maybeDescribe = FUZZ_ENABLED ? describe : describe.skip; - -function makeRng(seed: number): () => number { - let state = seed | 0; - return () => { - state = (state + 0x6d2b79f5) | 0; - let t = state; - t = Math.imul(t ^ (t >>> 15), t | 1); - t ^= t + Math.imul(t ^ (t >>> 7), t | 61); - return ((t ^ (t >>> 14)) >>> 0) / 4294967296; - }; -} - -type Outcome = "success" | "retryable" | "non_retryable"; - -class FuzzHandlerError extends Error { - constructor(public retryable: boolean) { - super(retryable ? "retryable" : "non_retryable"); - } -} - -maybeDescribe("MollifierDrainer fuzz", () => { - const seed = process.env.SEED ? Number(process.env.SEED) : Date.now() & 0xffff; - // eslint-disable-next-line no-console - console.log(`[fuzz] drainer seed=${seed}`); - - redisTest( - `random handler outcomes across envs drive every entry to terminal (seed=${seed})`, - { timeout: 120_000 }, - async ({ redisContainer }) => { - const buffer = new MollifierBuffer({ - redisOptions: { - host: redisContainer.getHost(), - port: redisContainer.getPort(), - password: redisContainer.getPassword(), - }, - entryTtlSeconds: 600, - logger: new Logger("fuzz", "warn"), - }); - - const rng = makeRng(seed); - const envIds = ["e0", "e1", "e2"]; - const entryCount = 60; - const maxAttempts = 3; - - // Pre-decide each runId's outcome distribution: 70% success, 15% retry, 15% fail. - const targetOutcome = new Map(); - for (let i = 0; i < entryCount; i++) { - const r = rng(); - const outcome: Outcome = r < 0.7 ? "success" : r < 0.85 ? "retryable" : "non_retryable"; - targetOutcome.set(`r_${i}`, outcome); - } - - // Track per-runId handler invocations + peak in-flight (separate from - // entry attempts so we can cross-check). - const handlerCalls = new Map(); - let inflight = 0; - let peakInflight = 0; - const concurrency = 4; - - const handler = vi.fn(async (input: { runId: string; attempts: number }) => { - inflight++; - if (inflight > peakInflight) peakInflight = inflight; - try { - await new Promise((r) => setTimeout(r, 5 + Math.floor(rng() * 20))); - handlerCalls.set(input.runId, (handlerCalls.get(input.runId) ?? 0) + 1); - const outcome = targetOutcome.get(input.runId)!; - if (outcome === "success") return; - throw new FuzzHandlerError(outcome === "retryable"); - } finally { - inflight--; - } - }); - - const drainer = new MollifierDrainer({ - buffer, - handler, - concurrency, - maxAttempts, - isRetryable: (err) => err instanceof FuzzHandlerError && err.retryable, - logger: new Logger("fuzz-drainer", "warn"), - }); - - try { - // Accept entries in random order across envs. - const order = Array.from({ length: entryCount }, (_, i) => i); - for (let i = order.length - 1; i > 0; i--) { - const j = Math.floor(rng() * (i + 1)); - const tmp = order[i] as number; - order[i] = order[j] as number; - order[j] = tmp; - } - for (const i of order) { - await buffer.accept({ - runId: `r_${i}`, - envId: envIds[i % envIds.length] as string, - orgId: "org_1", - payload: serialiseSnapshot({ i }), - }); - } - - // Drive runOnce until queues + draining all settle. - let safety = 200; - while (safety-- > 0) { - const before = await buffer.listEnvs(); - if (before.length === 0) { - // Also confirm no DRAINING entries linger. - const entryKeys = await buffer["redis"].keys("mollifier:entries:*"); - const drainingStillPresent = ( - await Promise.all( - entryKeys.map(async (k) => (await buffer["redis"].hget(k, "status")) === "DRAINING"), - ) - ).some((v) => v); - if (!drainingStillPresent) break; - } - await drainer.runOnce(); - } - expect(safety).toBeGreaterThan(0); - - // Invariant 1: concurrency cap honoured. - expect(peakInflight).toBeGreaterThan(1); - expect(peakInflight).toBeLessThanOrEqual(concurrency); - - // Invariant 2: every entry is in a terminal state. - for (let i = 0; i < entryCount; i++) { - const runId = `r_${i}`; - const stored = await buffer.getEntry(runId); - const outcome = targetOutcome.get(runId)!; - - if (outcome === "success") { - // success → acked → deleted - expect(stored, `expected r_${i} acked`).toBeNull(); - expect(handlerCalls.get(runId)).toBe(1); - } else if (outcome === "non_retryable") { - // non-retryable → FAILED on first attempt - expect(stored, `expected r_${i} present`).not.toBeNull(); - expect(stored!.status, `r_${i} status`).toBe("FAILED"); - expect(handlerCalls.get(runId)).toBe(1); - } else { - // retryable → retries until maxAttempts, then FAILED - expect(stored, `expected r_${i} present`).not.toBeNull(); - expect(stored!.status, `r_${i} status`).toBe("FAILED"); - expect(handlerCalls.get(runId), `r_${i} handler calls`).toBe(maxAttempts); - } - } - - // Invariant 3: no entry has attempts > maxAttempts. - const allEntryKeys = await buffer["redis"].keys("mollifier:entries:*"); - for (const k of allEntryKeys) { - const attempts = Number(await buffer["redis"].hget(k, "attempts")); - expect(attempts, `entry ${k} attempts`).toBeLessThanOrEqual(maxAttempts); - } - - // Invariant 4: no orphan queue references at end. - for (const env of await buffer.listEnvs()) { - const queueLen = await buffer["redis"].llen(`mollifier:queue:${env}`); - expect(queueLen, `env ${env} queue should be empty`).toBe(0); - } - } finally { - await drainer.stop({ timeoutMs: 1000 }); - await buffer.close(); - } - }, - ); -}); diff --git a/packages/redis-worker/src/mollifier/evaluateTrip.fuzz.test.ts b/packages/redis-worker/src/mollifier/evaluateTrip.fuzz.test.ts deleted file mode 100644 index 6dbac4162..000000000 --- a/packages/redis-worker/src/mollifier/evaluateTrip.fuzz.test.ts +++ /dev/null @@ -1,167 +0,0 @@ -// TEMPORARY: fuzz tests for Phase 1 validation of `MollifierBuffer.evaluateTrip`. -// -// Gated behind `FUZZ=1` so they don't run in CI. Invoke locally with -// `FUZZ=1 pnpm --filter @trigger.dev/redis-worker test src/mollifier/evaluateTrip.fuzz` -// during the live-monitoring window before Phase 2. -// -// Targets: concurrent INCR atomicity, env isolation under high concurrency, -// trip/hold-down semantics under random arrival timing. Seeded via SEED. -// Remove once the trip-evaluator surface is stable across two release cycles. - -import { redisTest } from "@internal/testcontainers"; -import { describe, expect } from "vitest"; -import { Logger } from "@trigger.dev/core/logger"; -import { MollifierBuffer } from "./buffer.js"; - -const FUZZ_ENABLED = process.env.FUZZ === "1"; -const maybeDescribe = FUZZ_ENABLED ? describe : describe.skip; - -function makeRng(seed: number): () => number { - let state = seed | 0; - return () => { - state = (state + 0x6d2b79f5) | 0; - let t = state; - t = Math.imul(t ^ (t >>> 15), t | 1); - t ^= t + Math.imul(t ^ (t >>> 7), t | 61); - return ((t ^ (t >>> 14)) >>> 0) / 4294967296; - }; -} - -function pick(rng: () => number, items: T[]): T { - const item = items[Math.floor(rng() * items.length)]; - // items is non-empty by precondition; non-null assertion silences the - // noUncheckedIndexedAccess rule without runtime cost. - return item as T; -} - -maybeDescribe("MollifierBuffer.evaluateTrip fuzz", () => { - const seed = process.env.SEED ? Number(process.env.SEED) : Date.now() & 0xffff; - // eslint-disable-next-line no-console - console.log(`[fuzz] evaluateTrip seed=${seed}`); - - redisTest( - `concurrent INCR across N envs preserves atomicity + isolation (seed=${seed})`, - { timeout: 60_000 }, - async ({ redisContainer }) => { - const buffer = new MollifierBuffer({ - redisOptions: { - host: redisContainer.getHost(), - port: redisContainer.getPort(), - password: redisContainer.getPassword(), - }, - entryTtlSeconds: 600, - logger: new Logger("fuzz", "warn"), - }); - - try { - const rng = makeRng(seed); - const envIds = ["e0", "e1", "e2", "e3", "e4"]; - // High threshold so we test pure count integrity, not trip semantics. - const opts = { windowMs: 5000, threshold: 1_000_000, holdMs: 100 }; - - const callsPerEnv = new Map(); - for (const e of envIds) callsPerEnv.set(e, 0); - - // Build a random concurrent workload: 500 calls distributed across envs. - const work = Array.from({ length: 500 }, () => { - const env = pick(rng, envIds); - callsPerEnv.set(env, (callsPerEnv.get(env) ?? 0) + 1); - return env; - }); - - const results = await Promise.all( - work.map(async (env) => ({ env, result: await buffer.evaluateTrip(env, opts) })), - ); - - // Atomicity: per-env counts returned must form a contiguous 1..N sequence. - for (const env of envIds) { - const observed = results - .filter((r) => r.env === env) - .map((r) => r.result.count) - .sort((a, b) => a - b); - const expected = Array.from({ length: callsPerEnv.get(env) ?? 0 }, (_, i) => i + 1); - expect(observed, `env ${env}`).toEqual(expected); - } - - // Isolation: no env's final count touches another's. (Implicit from - // the above, but assert explicitly: counts per env match issue count.) - for (const env of envIds) { - const final = await buffer["redis"].get(`mollifier:rate:${env}`); - expect(Number(final)).toBe(callsPerEnv.get(env)); - } - } finally { - await buffer.close(); - } - }, - ); - - redisTest( - `random arrivals near window/hold boundaries (seed=${seed}) preserve trip semantics`, - { timeout: 60_000 }, - async ({ redisContainer }) => { - const buffer = new MollifierBuffer({ - redisOptions: { - host: redisContainer.getHost(), - port: redisContainer.getPort(), - password: redisContainer.getPassword(), - }, - entryTtlSeconds: 600, - logger: new Logger("fuzz", "warn"), - }); - - try { - const rng = makeRng(seed ^ 0x9e3779b1); - // Short window + threshold + holdMs to push timing edges fast. - const opts = { windowMs: 80, threshold: 3, holdMs: 150 }; - const envId = "fuzz_env"; - - // Generate 60 random delays in [0, windowMs*1.2). Track the last time - // the Lua placed/refreshed the PSETEX marker (every call where - // count > threshold). Slack accounts for Lua-to-JS round-trip plus - // PSETEX millisecond granularity. - const calls = 60; - // Slack absorbs (a) PSETEX millisecond granularity, (b) Lua-to-JS - // round-trip on a loaded testcontainer (~5-50ms under load), - // (c) Date.now() vs Redis internal clock skew. holdMs=150ms so 100ms - // slack is generous without making the invariant tautological. - const SLACK_MS = 100; - let lastOverThresholdAt = -Infinity; - - for (let i = 0; i < calls; i++) { - const delayMs = Math.floor(rng() * Math.floor(opts.windowMs * 1.2)); - await new Promise((r) => setTimeout(r, delayMs)); - const { tripped, count } = await buffer.evaluateTrip(envId, opts); - const now = Date.now(); - - const overThreshold = count > opts.threshold; - - // Invariant A: if count > threshold this call, the Lua just PSETEX'd - // the marker, so EXISTS must observe it — tripped MUST be true. - if (overThreshold) { - expect(tripped, `i=${i}: over-threshold call must see tripped:true`).toBe(true); - } - - // Invariant B: if tripped:true but count <= threshold, the marker - // is carryover from a prior over-threshold INCR. That INCR must - // have happened within holdMs (+ slack for measurement noise). - if (tripped && !overThreshold) { - expect( - now - lastOverThresholdAt, - `i=${i}: tripped without over-threshold means marker must be recent`, - ).toBeLessThanOrEqual(opts.holdMs + SLACK_MS); - } - - if (overThreshold) lastOverThresholdAt = now; - } - - // Invariant C: after generous idle (> windowMs + holdMs + slack), - // the env resets to a fresh count of 1, tripped:false. - await new Promise((r) => setTimeout(r, opts.windowMs + opts.holdMs + 100)); - const reset = await buffer.evaluateTrip(envId, opts); - expect(reset).toEqual({ tripped: false, count: 1 }); - } finally { - await buffer.close(); - } - }, - ); -});