diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index c638a17a2a0..e3dbfc135dd 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -34,6 +34,26 @@ A replay inherits the masks saved in the source wave's `plan.json`; Each fault receipt records the host 1-minute load average (`loadAvg1m`); treat timeout failures above about 16 as load-suspect. +## Restart storm (K13, `--storm-minutes`) + +```sh +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex,claude,opencode --concurrency 6 --storm-minutes 60 --seed 7 +``` + +A storm wave fires faults on its own seeded clock instead of one fault per +task. Each fault is drawn from `--faults` (default K1, K2, K4, K6) and fires +2–4 minutes after the previous one settled (server ready), so a K4 hold never +eats the next gap. Each fault hits every task in flight. While the storm lasts +the harness launches a new task whenever one rests, keeping `--concurrency` +in flight; `--tasks` (default 200) is only the pool. A pool that runs dry +before the storm ends fails the wave as a harness error (`storm_pool_exhausted`). When the storm ends no +new task or fault starts, the in-flight tasks drain to rest or their deadline, +and only launched tasks are graded. Storm tasks show fault `K13` in the ledger; +`storm_fault`, `storm_started` and `storm_ended` are in the timeline and the +receipts are in `faults.json`. A replay of a storm task fires the faults that +crossed it, without a new storm. Only `--triggers time` applies. + ## Daytona (`--env daytona`) The control plane stays local; agent runs execute in Daytona sandboxes from an diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index 29d1cd33ca2..04019859270 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -85,13 +85,14 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC "daytona-cpu": { type: "string" }, keep: { type: "boolean" }, "rc06-masked": { type: "boolean" }, + "storm-minutes": { type: "string" }, }, }); const keep = values.keep === true; const pollMs = integer(values["poll-ms"], "poll-ms", 3_000, 2_000, 5_000); const budgetCents = integer(values["budget-cents"], "budget-cents", 3_000, 1, 100_000); if (values.replay !== undefined) { - for (const key of ["env", "adapters", "workloads", "faults", "triggers", "concurrency", "tasks", "seed", "daytona-image", "daytona-cpu"] as const) { + for (const key of ["env", "adapters", "workloads", "faults", "triggers", "concurrency", "tasks", "seed", "daytona-image", "daytona-cpu", "storm-minutes"] as const) { if (values[key] !== undefined) throw new Error(`--replay reproduces the saved plan; do not combine it with --${key}`); } return { mode: "replay", ref: parseReplayRef(values.replay), keep, @@ -109,14 +110,18 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC // B5 grades bench.sh progress read from the live workspace while it runs. A // Daytona run's live workspace is inside the sandbox, so B5 is local-only. const workloadChoices = env === "daytona" ? WORKLOAD_IDS.filter(id => id !== "B5") : WORKLOAD_IDS; + // K13 restart storm: faults come from the storm clock, and `--tasks` is the + // pool that keeps `--concurrency` tasks in flight until the storm ends. + const stormMinutes = values["storm-minutes"] === undefined ? undefined : integer(values["storm-minutes"], "storm-minutes", 60, 1, 240); const options: WaveOptions = { env, adapters: list(values.adapters, ADAPTER_IDS, "adapters", ["codex", "claude"]), workloads: list(values.workloads, workloadChoices, "workloads", workloadChoices), - faults: list(values.faults, FAULT_KINDS, "faults", ["K1", "K2", "K4"]), + faults: list(values.faults, FAULT_KINDS, "faults", stormMinutes ? FAULT_KINDS : ["K1", "K2", "K4"]), triggers: list(values.triggers, TRIGGER_IDS, "triggers", ["time"]), concurrency: integer(values.concurrency, "concurrency", 4, 1, 20), - tasks: integer(values.tasks, "tasks", 8, 1, 200), + tasks: integer(values.tasks, "tasks", stormMinutes ? 200 : 8, 1, 200), + ...(stormMinutes ? { stormMinutes } : {}), seed: integer(values.seed, "seed", now % 2_147_483_647, 0, 2_147_483_647), budgetCents, pollMs, diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index e3a97c19d5e..8ade9bbfe8b 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -20,11 +20,11 @@ import { knownSecretValues, writeEvidenceBundle } from "./evidence.js"; import { gradeTask } from "./grader.js"; import { appendLedger, ledgerLine, writeSummary, type LedgerLine } from "./ledger.js"; import { TaskObserver } from "./observer.js"; -import { observedFaultSchedule, planWave, replayWave } from "./scheduler.js"; +import { StormRun, observedFaultSchedule, planWave, replayWave } from "./scheduler.js"; import { seedChaosLab } from "./seed.js"; import { ChaosSupervisor } from "./supervisor.js"; import { Timeline } from "./timeline.js"; -import type { FaultReceipt, OracleArtifacts, WavePlan } from "./types.js"; +import type { FaultKind, FaultReceipt, OracleArtifacts, TriggerId, WavePlan } from "./types.js"; const repositoryRoot = path.resolve(import.meta.dirname, "../.."); const resultsRoot = path.resolve(process.env.RESTART_CHAOS_RESULTS_DIR ?? path.join(import.meta.dirname, "results")); @@ -104,7 +104,8 @@ async function main(): Promise { const gitSha = git(["rev-parse", "HEAD"]); const gitDirty = git(["status", "--porcelain"]).length > 0; timeline.add("wave_planned", { waveId: plan.waveId, seed: plan.seed, gitSha, gitDirty, replayOf: plan.replayOf ?? null, - tasks: plan.tasks.map(task => `${task.taskId}:${task.adapter}/${task.workload}/${task.fault}${task.trigger ? `^${task.trigger}+${task.triggerDelayMs ?? 0}ms` : ""}@${task.fireAfterMs}ms${task.holdMs ? ` hold ${task.holdMs}ms` : ""}`) }); + tasks: plan.tasks.map(task => `${task.taskId}:${task.adapter}/${task.workload}/${task.fault}${task.trigger ? `^${task.trigger}+${task.triggerDelayMs ?? 0}ms` : ""}@${task.fireAfterMs}ms${task.holdMs ? ` hold ${task.holdMs}ms` : ""}`), + ...(plan.storm ? { storm: { durationMs: plan.storm.durationMs, faults: plan.storm.faults.map(fault => `${fault.kind}+${fault.gapMs}ms${fault.holdMs ? ` hold ${fault.holdMs}ms` : ""}`) } } : {}) }); preflight(plan); const serverEnvironment = plan.options.env === "daytona" ? await prepareDaytona(plan, timeline) : {}; @@ -154,17 +155,42 @@ async function main(): Promise { const protectedRoots = [os.homedir(), supervisor.paths.temporaryRoot, resultsRoot]; const observers = plan.tasks.map(task => new TaskObserver(task, api, lab, timeline, supervisor, protectedRoots)); const pending = [...observers]; + const launched: TaskObserver[] = []; const active: TaskObserver[] = []; const lastPoll = new Map(); let lastDoneAt = Date.now(); + // K13: storm faults fire on their own clock while tasks keep launching; then the storm drains. + const storm = plan.storm ? new StormRun(plan.storm, Date.now()) : null; + if (storm) timeline.add("storm_started", { durationMs: storm.plan.durationMs, plannedFaults: storm.plan.faults.length }); + // Faults are serialized; each one hits every task in flight when it starts. + const fireFault = (kind: FaultKind, holdMs: number, requestedBy: string, exposed: readonly TaskObserver[], + trigger?: { trigger: TriggerId; delayMs: number }, onSettled?: () => void) => { + faultsQueued += 1; + faultWork.push(supervisor.fault(kind, holdMs, requestedBy, trigger).finally(() => { faultsQueued -= 1; onSettled?.(); }).then(receipt => { + faults.push(receipt); + for (const candidate of exposed) { candidate.noteFault(receipt.faultId, receipt.startedAt); candidate.extendDeadline((receipt.readyAt ?? Date.now()) - receipt.startedAt); } + }, (error: Error & { receipt?: FaultReceipt }) => { + if (error.receipt) faults.push(error.receipt); + faultFailures += 1; + timeline.add("fault_error", { taskId: requestedBy, error: error.message }); + })); + }; while (!interrupted && (pending.length || active.length)) { const now = Date.now(); + if (storm?.endIfDue(now)) { + // Unlaunched pool tasks are not part of the wave; the in-flight ones drain without new faults. + timeline.add("storm_ended", { faultsFired: storm.state.fired, inFlight: active.length, unlaunched: pending.length }); + pending.length = 0; + } // A failed fault or an unexpected exit leaves the server not ready. // Recover within a bound, or fail the wave instead of waiting forever. if (!supervisor.serverReady && !supervisor.faultInProgress) await supervisor.ensureServing(); - while (active.length < plan.options.concurrency && pending.length && supervisor.serverReady && !supervisor.faultInProgress) { + // Recheck the storm clock per launch: ensureServing or a slow launch can cross the storm's end. + while (active.length < plan.options.concurrency && pending.length && supervisor.serverReady && !supervisor.faultInProgress + && (!storm || storm.mayLaunch(Date.now()))) { const next = pending.shift()!; + launched.push(next); await next.launch(); if (!next.done) active.push(next); } @@ -182,15 +208,19 @@ async function main(): Promise { if (!(await observer.confirmTrigger(due, Date.now()))) continue; const trigger = due.trigger && due.trigger !== "time" ? { trigger: due.trigger, delayMs: due.delayMs ?? 0 } : undefined; if (trigger) timeline.add("trigger_fired", { taskId: observer.plan.taskId, fault: due.kind, ...trigger }); - faultsQueued += 1; - faultWork.push(supervisor.fault(due.kind, due.holdMs, observer.plan.taskId, trigger).finally(() => { faultsQueued -= 1; }).then(receipt => { - faults.push(receipt); - for (const candidate of exposed) { candidate.noteFault(receipt.faultId, receipt.startedAt); candidate.extendDeadline((receipt.readyAt ?? Date.now()) - receipt.startedAt); } - }, (error: Error & { receipt?: FaultReceipt }) => { - if (error.receipt) faults.push(error.receipt); - faultFailures += 1; - timeline.add("fault_error", { taskId: observer.plan.taskId, error: error.message }); - })); + fireFault(due.kind, due.holdMs, observer.plan.taskId, exposed, trigger); + } + if (storm && !faultsQueued && !supervisor.faultInProgress && supervisor.serverReady) { + const due = storm.due(Date.now()); + if (due) { + const exposed = active.filter(candidate => !candidate.done); + await Promise.all(exposed.map(candidate => candidate.snapshotBeforeFault())); + // The snapshots take time; a storm that ended meanwhile is draining and gets no new fault. + if (storm.commit(due, Date.now())) { + timeline.add("storm_fault", { index: storm.state.fired, kind: due.kind, gapMs: due.gapMs, holdMs: due.holdMs, exposed: exposed.map(candidate => candidate.plan.taskId) }); + fireFault(due.kind, due.holdMs, "storm", exposed, undefined, () => storm.settled(Date.now())); + } + } } for (let index = active.length - 1; index >= 0; index -= 1) { if (active[index]!.done) { active.splice(index, 1); lastDoneAt = Date.now(); } @@ -198,6 +228,11 @@ async function main(): Promise { await delay(1_000); } await Promise.all(faultWork); + // A storm whose task pool ran dry early did not cover its storm; it is not a passing wave. + if (storm && !interrupted && storm.exhaustedEarly(Date.now())) { + timeline.add("storm_pool_exhausted", { faultsFired: storm.state.fired, launched: launched.length }); + harnessError ??= new Error(`Storm task pool (--tasks ${plan.options.tasks}) ran out before the storm ended; raise --tasks`); + } // An injected fault that did not complete invalidates the wave, even when // bounded recovery brought the server back and the tasks passed. if (faultFailures) harnessError ??= new Error(`${faultFailures} injected fault(s) failed; see fault_error in the timeline`); @@ -214,7 +249,7 @@ async function main(): Promise { } const attributed = new Set(); const secrets = knownSecretValues(); - for (const observer of observers) { + for (const observer of launched) { observer.record.sweep = await observer.sweep(); if (sandboxSweep) { const runIds = new Set(observer.record.samples.flatMap(sample => (sample.runs ?? []).map(run => run.id))); diff --git a/tests/restart-chaos/scheduler.test.ts b/tests/restart-chaos/scheduler.test.ts index 8f883a4c150..c926079c866 100644 --- a/tests/restart-chaos/scheduler.test.ts +++ b/tests/restart-chaos/scheduler.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import { parseCommand, parseReplayRef, waveIdFor } from "./config.js"; -import { K4_HOLD_RANGE_MS, createRng, observedFaultSchedule, planTasks, planWave, replayWave, taskFaultSchedule } from "./scheduler.js"; +import { K4_HOLD_RANGE_MS, STORM_GAP_RANGE_MS, createRng, observedFaultSchedule, planStorm, planTasks, planWave, replayWave, + StormRun, stormActive, stormFaultDue, taskFaultSchedule, type StormState } from "./scheduler.js"; import type { WaveOptions } from "./types.js"; import { workloadDefinitions } from "./workloads.js"; @@ -142,3 +143,113 @@ describe("Daytona options", () => { expect(() => parseCommand(["--env", "local", "--daytona-cpu", "2"])).toThrow(/only to --env daytona/); }); }); + +describe("K13 restart storm", () => { + const stormOptions: WaveOptions = { ...options, faults: ["K1", "K2", "K4", "K6"], tasks: 40, concurrency: 6, stormMinutes: 60 }; + + it("plans a seeded storm of 2–4 minute gaps drawn from the chosen faults that fills the duration", () => { + const storm = planStorm(42, 60, stormOptions.faults); + expect(planStorm(42, 60, stormOptions.faults)).toEqual(storm); + expect(storm.durationMs).toBe(3_600_000); + const gaps = storm.faults.map(fault => fault.gapMs); + expect(gaps.every(gap => gap >= STORM_GAP_RANGE_MS[0] && gap <= STORM_GAP_RANGE_MS[1])).toBe(true); + expect(gaps.reduce((sum, gap) => sum + gap, 0)).toBeGreaterThanOrEqual(storm.durationMs); + expect(gaps.slice(0, -1).reduce((sum, gap) => sum + gap, 0)).toBeLessThan(storm.durationMs); + for (const fault of storm.faults) { + expect(stormOptions.faults).toContain(fault.kind); + if (fault.kind === "K4") expect(fault.holdMs).toBeGreaterThanOrEqual(K4_HOLD_RANGE_MS[0]); + else expect(fault.holdMs).toBe(0); + } + expect(new Set(storm.faults.map(fault => fault.kind)).size).toBeGreaterThan(1); + }); + + it("gives storm tasks no fault of their own", () => { + const plan = planWave("w-20261106T101112-s42", stormOptions); + expect(plan.storm?.faults.length).toBeGreaterThan(10); + expect(plan.tasks).toHaveLength(40); + for (const task of plan.tasks) { + expect(task.fault).toBe("K13"); + expect(taskFaultSchedule(task)).toEqual([]); + } + expect(planWave("w-20261106T101112-s42", options).storm).toBeUndefined(); + expect(() => planTasks({ ...stormOptions, triggers: ["comment"], workloads: ["C1"] })).toThrow(/own clock/); + }); + + it("fires each fault after its gap from the previous fault settling, and stops when the storm ends", () => { + const storm = { durationMs: 600_000, faults: [{ kind: "K1" as const, holdMs: 0, gapMs: 120_000 }, + { kind: "K4" as const, holdMs: 200_000, gapMs: 180_000 }, { kind: "K2" as const, holdMs: 0, gapMs: 120_000 }] }; + const state: StormState = { startedAt: 0, fired: 0, lastSettledAt: null }; + expect(stormFaultDue(storm, state, 119_999)).toBeNull(); + expect(stormFaultDue(storm, state, 120_000)).toEqual(storm.faults[0]); + state.fired = 1; state.lastSettledAt = 150_000; + expect(stormFaultDue(storm, state, 329_999)).toBeNull(); + expect(stormFaultDue(storm, state, 330_000)).toEqual(storm.faults[1]); + // The K4 hold settles late; the next gap runs from then, not from when it fired. + state.fired = 2; state.lastSettledAt = 560_000; + expect(stormActive(storm, state, 599_999)).toBe(true); + expect(stormFaultDue(storm, state, 599_999)).toBeNull(); + expect(stormActive(storm, state, 600_000)).toBe(false); + expect(stormFaultDue(storm, state, 700_000)).toBeNull(); + state.fired = 3; + expect(stormFaultDue({ ...storm, durationMs: 10_000_000 }, state, 9_000_000)).toBeNull(); + }); + + it("rechecks the storm clock after slow work: no launch and no fault once the storm has ended", () => { + const storm = { durationMs: 300_000, faults: [{ kind: "K1" as const, holdMs: 0, gapMs: 120_000 }, { kind: "K2" as const, holdMs: 0, gapMs: 120_000 }] }; + const state: StormState = { startedAt: 0, fired: 1, lastSettledAt: 170_000 }; + expect(stormActive(storm, state, 299_999)).toBe(true); + expect(stormActive(storm, state, 300_000)).toBe(false); + // A fault selected before the end, with pre-fault snapshots finishing after it, no longer fires. + const due = stormFaultDue(storm, state, 295_000); + expect(due).toEqual(storm.faults[1]); + expect(stormFaultDue(storm, state, 295_500)).toBe(due); + expect(stormFaultDue(storm, state, 301_000)).toBeNull(); + }); + + it("drives the wave loop's storm decisions with a fresh clock across slow awaits", () => { + const plan = { durationMs: 300_000, faults: [{ kind: "K1" as const, holdMs: 0, gapMs: 120_000 }, { kind: "K2" as const, holdMs: 0, gapMs: 120_000 }] }; + const run = new StormRun(plan, 0); + // Iteration at 120 s: launch allowed, fault chosen, snapshots finish in time, fault committed. + expect(run.endIfDue(120_000)).toBe(false); + expect(run.mayLaunch(120_000)).toBe(true); + const first = run.due(120_000)!; + expect(run.commit(first, 121_000)).toBe(true); + expect(run.state.fired).toBe(1); + expect(run.due(150_000)).toBeNull(); // the next gap waits for the fault to settle + run.settled(170_000); + // Iteration at 295 s chooses the second fault, but the snapshots end after the storm: nothing fires. + const second = run.due(295_000)!; + expect(second).toEqual(plan.faults[1]); + expect(run.commit(second, 301_000)).toBe(false); + expect(run.state.fired).toBe(1); + // A launch decided at 299 s but taken after ensureServing crossed the end is refused. + expect(run.mayLaunch(299_000)).toBe(true); + expect(run.mayLaunch(300_500)).toBe(false); + expect(run.endIfDue(300_500)).toBe(true); + expect(run.endIfDue(301_000)).toBe(false); + expect(run.exhaustedEarly(301_000)).toBe(false); + }); + + it("flags a task pool that runs dry before the storm ends", () => { + const run = new StormRun({ durationMs: 3_600_000, faults: [{ kind: "K1", holdMs: 0, gapMs: 120_000 }] }, 0); + expect(run.exhaustedEarly(600_000)).toBe(true); + expect(run.exhaustedEarly(3_600_000)).toBe(false); + }); + + it("parses --storm-minutes with storm defaults and keeps replays storm-free", () => { + const parsed = parseCommand(["--storm-minutes", "60", "--concurrency", "6", "--adapters", "codex,claude,opencode", "--seed", "5"]); + if (parsed.mode !== "wave") throw new Error("expected wave"); + expect(parsed.options).toMatchObject({ stormMinutes: 60, faults: ["K1", "K2", "K4", "K6"], tasks: 200, concurrency: 6 }); + const plain = parseCommand(["--seed", "5"]); + if (plain.mode !== "wave") throw new Error("expected wave"); + expect(plain.options).toMatchObject({ faults: ["K1", "K2", "K4"], tasks: 8 }); + expect(plain.options.stormMinutes).toBeUndefined(); + expect(() => parseCommand(["--storm-minutes", "0"])).toThrow(/between 1 and 240/); + expect(() => parseCommand(["--replay", "w-20261106T101112-s42/t03", "--storm-minutes", "5"])).toThrow(/reproduces the saved plan/); + const plan = planWave("w-20261106T101112-s42", stormOptions); + const replay = replayWave(plan, { waveId: plan.waveId, taskId: "t02" }, 1, {}, [{ kind: "K6", holdMs: 0, atMs: 90_000 }]); + expect(replay.options.stormMinutes).toBeUndefined(); + expect(replay.storm).toBeUndefined(); + expect(taskFaultSchedule(replay.tasks[0]!)).toEqual([{ kind: "K6", holdMs: 0, atMs: 90_000 }]); + }); +}); diff --git a/tests/restart-chaos/scheduler.ts b/tests/restart-chaos/scheduler.ts index b9f307fa349..7a3f86a657f 100644 --- a/tests/restart-chaos/scheduler.ts +++ b/tests/restart-chaos/scheduler.ts @@ -1,5 +1,5 @@ import { workloadDefinitions } from "./workloads.js"; -import type { AdapterId, FaultKind, ScheduledFault, TaskPlan, TriggerId, WavePlan, WaveOptions, WorkloadId } from "./types.js"; +import type { AdapterId, FaultKind, ScheduledFault, StormFault, StormPlan, TaskFaultKind, TaskPlan, TriggerId, WavePlan, WaveOptions, WorkloadId } from "./types.js"; import type { ReplayRef } from "./config.js"; /** mulberry32: small, fast and fully determined by its 32-bit seed. */ @@ -46,16 +46,19 @@ export function triggerFitsWorkload(trigger: TriggerId, workload: WorkloadId): b * Pairwise-friendly sampling: cycle a seeded shuffle of the full * workload × adapter × fault product so small waves still spread across cells. */ -export function planTasks(options: Pick): TaskPlan[] { +export function planTasks(options: Pick): TaskPlan[] { const rng = createRng(options.seed); const triggers = options.triggers?.length ? options.triggers : ["time" as const]; + // A storm task has no fault of its own: the wave's storm crosses it. + const faults: readonly TaskFaultKind[] = options.stormMinutes ? ["K13"] : options.faults; + if (options.stormMinutes && triggers.some(trigger => trigger !== "time")) throw new Error("A K13 storm fires on its own clock; use --triggers time"); for (const trigger of triggers) { if (!options.workloads.some(workload => triggerFitsWorkload(trigger, workload))) { throw new Error(`Trigger "${trigger}" needs workload ${trigger === "comment" ? "C1" : "B5"} in --workloads`); } } - const cells: Array<{ workload: WorkloadId; adapter: AdapterId; fault: FaultKind; trigger: TriggerId }> = []; - for (const workload of options.workloads) for (const adapter of options.adapters) for (const fault of options.faults) { + const cells: Array<{ workload: WorkloadId; adapter: AdapterId; fault: TaskFaultKind; trigger: TriggerId }> = []; + for (const workload of options.workloads) for (const adapter of options.adapters) for (const fault of faults) { for (const trigger of triggers) if (triggerFitsWorkload(trigger, workload)) cells.push({ workload, adapter, fault, trigger }); } const tasks: TaskPlan[] = []; @@ -71,7 +74,7 @@ export function planTasks(options: Pick>> 0); + const durationMs = minutes * 60_000; + const sequence: StormFault[] = []; + for (let planned = 0; planned < durationMs;) { + const gapMs = rng.int(...STORM_GAP_RANGE_MS); + const kind = faults[rng.int(0, faults.length - 1)]!; + sequence.push({ kind, gapMs, holdMs: kind === "K4" ? rng.int(...K4_HOLD_RANGE_MS) : 0 }); + planned += gapMs; + } + return { durationMs, faults: sequence }; +} + +export interface StormState { + startedAt: number; + /** Storm faults fired so far. */ + fired: number; + /** When the previous storm fault finished (server ready, or the fault failed); null before the first. */ + lastSettledAt: number | null; +} + +/** New tasks launch and storm faults fire only while the storm lasts; then it drains. */ +export function stormActive(storm: StormPlan, state: StormState, now: number): boolean { + return now < state.startedAt + storm.durationMs; +} + +/** + * The storm clock the wave loop consults. Every decision takes a fresh `now`, + * because the loop awaits slow work (server recovery, launches, pre-fault + * snapshots) between choosing an action and taking it. + */ +export class StormRun { + readonly state: StormState; + ended = false; + + constructor(readonly plan: StormPlan, startedAt: number) { + this.state = { startedAt, fired: 0, lastSettledAt: null }; + } + + /** True once, on the transition to the drain. */ + endIfDue(now: number): boolean { + if (this.ended || stormActive(this.plan, this.state, now)) return false; + this.ended = true; + return true; + } + + /** Ask immediately before each launch. */ + mayLaunch(now: number): boolean { + return !this.ended && stormActive(this.plan, this.state, now); + } + + /** The next fault, never while the previous one is still unsettled: each gap runs from its settle. */ + due(now: number): StormFault | null { + return this.ended || this.unsettled ? null : stormFaultDue(this.plan, this.state, now); + } + + /** Claim a fault chosen before slow work. False when the storm ended meanwhile; nothing is consumed. */ + commit(fault: StormFault, now: number): boolean { + if (this.due(now) !== fault) return false; + this.state.fired += 1; + this.unsettled = true; + return true; + } + + settled(now: number) { this.state.lastSettledAt = now; this.unsettled = false; } + + private unsettled = false; + + /** The task pool ran dry with storm time left: the wave did not cover its storm. */ + exhaustedEarly(now: number): boolean { + return !this.ended && stormActive(this.plan, this.state, now); + } +} + +/** + * The next storm fault once its gap has passed. Gaps run from the previous + * fault settling, so a K4 hold never eats the next gap. The caller must not + * ask while a fault is still in progress. + */ +export function stormFaultDue(storm: StormPlan, state: StormState, now: number): StormFault | null { + if (!stormActive(storm, state, now)) return null; + const next = storm.faults[state.fired]; + if (!next) return null; + return now >= (state.lastSettledAt ?? state.startedAt) + next.gapMs ? next : null; } /** Faults that actually crossed a task, as offsets from its launch. */ @@ -106,6 +210,7 @@ export function observedFaultSchedule(createdAt: number | null, faultIds: readon /** The faults a task schedules: its replayed sequence, or its one planned fault. */ export function taskFaultSchedule(task: TaskPlan): ScheduledFault[] { if (task.replayFaults?.length) return task.replayFaults; + if (task.fault === "K13") return []; const planned: ScheduledFault = { kind: task.fault, holdMs: task.holdMs, atMs: task.fireAfterMs }; const trigger = task.trigger ?? "time"; if (trigger === "time") return [planned]; @@ -131,7 +236,8 @@ export function replayWave(source: WavePlan, ref: ReplayRef, now = Date.now(), waveId: `${base}-replay-${task.taskId}-${now}`, seed: source.seed, createdAt: new Date(now).toISOString(), - options: { ...source.options, ...overrides, tasks: 1, concurrency: 1, ...replayMasks(source.options.masks, overrides.masks) }, + // A replay fires only the faults that crossed the task, never a new storm. + options: { ...withoutStorm(source.options), ...overrides, tasks: 1, concurrency: 1, ...replayMasks(source.options.masks, overrides.masks) }, tasks: [{ ...task, ...(observed.length ? { replayFaults: observed.map(fault => ({ ...fault })) } : {}) }], replayOf: { ...ref }, }; diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index ed8cb7e9fcc..43432730be9 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -2,6 +2,8 @@ export const FAULT_KINDS = ["K1", "K2", "K4", "K6"] as const; export type FaultKind = (typeof FAULT_KINDS)[number]; +/** A task's planned fault. "K13" (restart storm) tasks have none of their own; the wave's storm crosses them. */ +export type TaskFaultKind = FaultKind | "K13"; export const WORKLOAD_IDS = ["A1", "B1", "B5", "C1"] as const; export type WorkloadId = (typeof WORKLOAD_IDS)[number]; @@ -39,6 +41,23 @@ export interface WaveOptions { * recorded per wave. Absent on plans written before masks existed. */ masks?: string[]; + /** + * K13 restart storm: fire faults drawn from `faults` every 2–4 minutes for + * this many minutes, keeping `concurrency` tasks in flight, then drain. + */ + stormMinutes?: number; +} + +export interface StormFault { + kind: FaultKind; + holdMs: number; + /** Wait this long after the previous storm fault settled (or the storm started). */ + gapMs: number; +} + +export interface StormPlan { + durationMs: number; + faults: StormFault[]; } export interface TaskPlan { @@ -46,7 +65,7 @@ export interface TaskPlan { index: number; workload: WorkloadId; adapter: AdapterId; - fault: FaultKind; + fault: TaskFaultKind; /** Fire the fault this long after the task's issue is created. */ fireAfterMs: number; /** Outage hold between stop and restart. 0 for K1/K2. */ @@ -78,6 +97,8 @@ export interface WavePlan { createdAt: string; options: WaveOptions; tasks: TaskPlan[]; + /** K13 only: the seeded storm fault sequence. */ + storm?: StormPlan; /** Present when this wave replays one task of an earlier wave. */ replayOf?: { waveId: string; taskId: string }; }