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).
This commit is contained in:
Dan Sutton
@@ -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<string, Outcome>();
|
||||
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<string, number>();
|
||||
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();
|
||||
}
|
||||
},
|
||||
);
|
||||
});
|
||||
@@ -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<T>(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<string, number>();
|
||||
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();
|
||||
}
|
||||
},
|
||||
);
|
||||
});
|
||||
Reference in New Issue
Block a user