diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index c7fd0370cb7..4165a84c898 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -385,6 +385,30 @@ GitHub `origin` over https at its `origin/master` commit, so the sandbox clones snapshots are Paperclip-sized. Its oracle reads the task's files through the board API, retrying for up to 4 minutes while the closing save makes the workspace readable, and runs its commands on a local copy. +### Restart during a transfer (`--storage-trigger`) + +A time-based fault rarely lands while the server is moving bytes. `--storage-trigger ` turns +each task's planned K fault into an armed fault: at its planned time (`--fault-after-seconds` pins it) +the fault is armed, and it fires the moment the proxy sees the next matching server request. The +proxy holds that request (up to 60 s) while the harness takes its pre-fault snapshot and restarts the +server, so the restart provably interrupts the transfer. + +| Target | Matching request | +| --- | --- | +| `snapshot-upload` | `PUT` with a body under `workspace-snapshots/` (single PUT or multipart part; not create or complete) | +| `snapshot-part` | multipart `UploadPart` (`PUT …?partNumber=`) under `workspace-snapshots/` | +| `snapshot-download` | `GET` under `workspace-snapshots/` (restore) | +| `run-log` | `PUT`/`POST` under `run-logs/` (run-log journal segments) | + +`storage-trigger.json` records, per armed fault, the request that fired it (method, path with upload +ids hidden, declared bytes, outcome `aborted`/`completed`), every request in flight at that +moment (the held one included), and the fault id it produced. A fault whose task rested first, or that saw no matching request +before the wave ended, is recorded with `missedReason` (`task_rested` or `no_matching_request`). The wave +summary adds a "Storage trigger" section with each armed fault's result and the count not reached. It +fails when no restart fired, or when a fired restart has no fault id or its held request was not +`aborted` (the hold expired first). It needs `--storage s3` and `--triggers time`, and does not take +G1, H1 or Q1: their board actions would run while the fault waits for a request. + Every Retry the grader needs is clicked in the real board UI: the harness opens the issue in a headless Chromium, waits for the visible Retry button and clicks it. A task whose Retry button is missing fails R3 as `ui_retry_button_missing (user saw: …)`; a click that sends no retry request fails as diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index 1463d90c1cc..b5d61a24a6d 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -5,6 +5,7 @@ import { applyJourneyOptions } from "./journey-matrix.js"; import { takesJourneyTriggers } from "./journeys.js"; import { STORAGE_FAULT_KINDS, type StorageFaultKind } from "./storage-chaos.js"; import { parseTimeFaultOptions } from "./time-faults-config.js"; +import { parseStorageTrigger } from "./storage-trigger.js"; import { parseRepeatedRestartOptions } from "./repeated-restarts.js"; import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, isProviderFault, parseProviderDelays, parseProviderMessage, type ProviderFaultKind } from "./provider-faults.js"; import { PROVIDER_FAULT_ADAPTERS, parseProviderFaultRestart } from "./acpx-provider-faults.js"; @@ -109,6 +110,7 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC storage: { type: "string" }, "storage-faults": { type: "string" }, "storage-fault-seconds": { type: "string" }, + "storage-trigger": { type: "string" }, "provider-fault-message": { type: "string" }, "provider-fault-delays": { type: "string" }, "db-pressure": { type: "string" }, @@ -124,7 +126,7 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC 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", "daytona-reuse-lease", "storm-minutes", "lifecycle", "k4-hold-max-seconds", "crash-point", "crash-point-hold-ms", "upgrade-from", "workspace-repo", "storage", "storage-faults", "storage-fault-seconds", "fault-after-seconds", "provider-fault-message", "provider-fault-delays", "db-pressure", "matrix", "server-from", "time-age-hours", "provider-fault-restart", "provider-fault-restart-after-seconds", "rr-gap-seconds"] as const) { + for (const key of ["env", "adapters", "workloads", "faults", "triggers", "concurrency", "tasks", "seed", "daytona-image", "daytona-cpu", "daytona-reuse-lease", "storm-minutes", "lifecycle", "k4-hold-max-seconds", "crash-point", "crash-point-hold-ms", "upgrade-from", "workspace-repo", "storage", "storage-faults", "storage-fault-seconds", "storage-trigger", "fault-after-seconds", "provider-fault-message", "provider-fault-delays", "db-pressure", "matrix", "server-from", "time-age-hours", "provider-fault-restart", "provider-fault-restart-after-seconds", "rr-gap-seconds"] 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, @@ -286,6 +288,7 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC if (!match || min < 5 || max < min || max > 900) throw new Error("--storage-fault-seconds takes - seconds, 5 to 900, min not above max"); options.storageFaultMs = [min * 1000, max * 1000]; } + if (values["storage-trigger"] !== undefined) options.storageTrigger = parseStorageTrigger(values["storage-trigger"], options); return { mode: "wave", options, keep }; } diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index 2b8b04e545b..09e566b9fd8 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -17,6 +17,7 @@ import { BoardUi } from "./board-ui.js"; import { DEFAULT_MINIO_BINARY, MINIO_RELEASE, minioBinary, objectStoreServerEnvironment, startMinio, type ObjectStore } from "./object-store.js"; import { StorageChaos, planStorageFaults, replayStorageWindows, type StorageFaultReceipt } from "./storage-chaos.js"; import { startStorageFaultProxy } from "./storage-proxy.js"; +import { exposedAfterLaunch, StorageTrigger, storageTriggerMarkdown, storageTriggerShortfall } from "./storage-trigger.js"; import { needsCredentialRotation, startCredentialRotation, type CredentialRotation } from "./storage-credentials.js"; import { describeProviderFault, installProviderFaultSeam, resolveRealCodex, withProviderFaultCheck, type ProviderFaultSeam } from "./provider-faults.js"; import { installAcpxSidecarSeam, withAcpxSidecarHook, type AcpxSidecarSeam } from "./acpx-provider-faults.js"; @@ -250,11 +251,13 @@ async function main(): Promise { let board: BoardUi | null = null; let objectStore: ObjectStore | null = null; let storageChaos: StorageChaos | null = null; + let storageTrigger: StorageTrigger | null = null; let storageProxy: Awaited> | null = null; let credentialRotation: CredentialRotation | null = null; let providerFaults: ProviderFaultSeam | null = null; let acpxSidecarSeam: AcpxSidecarSeam | null = null; let speedSection: string | null = null; + let storageSection: string | null = null; let harnessError: unknown; process.on("uncaughtException", error => { if (isSocketTypeOfServiceRace(error)) { timeline.add("harness_socket_race_ignored", { error: String(error) }); return; } @@ -385,6 +388,28 @@ async function main(): Promise { })); }; + let launching: Promise | null = null; + if (storageProxy && plan.options.storageTrigger) { + storageTrigger = new StorageTrigger(plan.options.storageTrigger, { + canFire: () => !interrupted && !faultsQueued && !supervisor.faultInProgress && supervisor.serverReady, + isOpen: taskId => observers.some(observer => observer.plan.taskId === taskId && !observer.done && observer.record.restAt === null), + record: (kind, data) => timeline.add(kind, data), + fire: armed => { + faultsQueued += 1; + const work = (async () => { + const exposed = await exposedAfterLaunch(launching, launched); + await Promise.all(exposed.map(candidate => candidate.snapshotBeforeFault())); + const queued = faultWork.length; + fireFault(armed.kind, armed.holdMs, armed.taskId, exposed, undefined, undefined, armed.restartSteps); + await faultWork[queued]; + })().finally(() => { faultsQueued -= 1; }); + faultWork.push(work); + return work; + }, + }); + storageProxy.observe(storageTrigger.listener); + } + supervisor.onCrashPoint = receipt => { const owner = pointResults.find(result => result.token === receipt.crashPoint?.token); if (owner) owner.fault = receipt; @@ -437,7 +462,7 @@ async function main(): Promise { errors: check.errors.map(error => redactKnownSecrets(error, knownSecretValues())) }); }; - while (!interrupted && (pending.length || active.length)) { + while (!interrupted && (pending.length || active.length || storageTrigger?.busy)) { 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. @@ -455,13 +480,14 @@ async function main(): Promise { } } // Recheck the storm clock per launch: ensureServing or a slow launch can cross the storm's end. - while (!crashPoint && active.length < plan.options.concurrency && pending.length && supervisor.serverReady && !supervisor.faultInProgress + while (!crashPoint && active.length < plan.options.concurrency && pending.length && supervisor.serverReady && !supervisor.faultInProgress && !storageTrigger?.busy && (!storm || storm.mayLaunch(Date.now())) // `--matrix`: no-fault baselines never share the server with a fault cell. && (!plan.options.matrix || matrixMayLaunch(pending[0]!.plan, active.filter(observer => !observer.done).map(observer => observer.plan)))) { const next = pending.shift()!; launched.push(next); - await next.launch(); + launching = next.launch().finally(() => { launching = null; }); + await launching; if (plan.options.storageWindows && storageChaos && launched.length === 1) storageChaos.start(); if (!next.done) active.push(next); } @@ -488,6 +514,7 @@ async function main(): Promise { const due = observer.faultDue(Date.now()); if (!due) continue; if (observer.record.restAt !== null) { timeline.add("fault_skipped_task_resting", { taskId: observer.plan.taskId, fault: due.kind }); continue; } + if (storageTrigger) { storageTrigger.arm({ taskId: observer.plan.taskId, kind: due.kind, holdMs: due.holdMs, armedAt: Date.now(), ...(due.restartSteps ? { restartSteps: due.restartSteps } : {}) }); continue; } const exposed = active.filter(candidate => !candidate.done); await Promise.all(exposed.map(candidate => candidate.snapshotBeforeFault())); if (!(await observer.confirmTrigger(due, Date.now()))) continue; @@ -519,6 +546,12 @@ async function main(): Promise { await delay(1_000); } await Promise.all(faultWork); + if (storageTrigger) { + const shortfall = storageTriggerShortfall(storageTrigger.finish(faults)); + storageSection = storageTriggerMarkdown(storageTrigger.receipts); + timeline.add("storage_trigger_settled", { shortfall, receipts: storageTrigger.receipts.length }); + if (shortfall) harnessError ??= new Error(shortfall); + } await stopPressure(); checkWorktree("wave_end"); if (worktreeChanged) harnessError ??= new Error("The worktree changed during the wave (build, install or branch switch); later restarts may have run different code"); @@ -684,6 +717,8 @@ async function main(): Promise { await credentialRotation?.settled(); if (credentialRotation?.failures.length) harnessError ??= new Error(`ST5R credential rotation failed: ${credentialRotation.failures.join("; ")}`); if (storageProxy) { + storageProxy.observe(null); + if (storageTrigger) await writeFile(path.join(waveDir, "storage-trigger.json"), `${JSON.stringify(storageTrigger.finish(faults), null, 2)}\n`, { mode: 0o600 }).catch(() => undefined); if (storageChaos) await writeFile(path.join(waveDir, "storage-faults.json"), `${JSON.stringify(storageChaos.receipts, null, 2)}\n`, { mode: 0o600 }).catch(() => undefined); await writeFile(path.join(waveDir, "storage-proxy.json"), `${JSON.stringify(storageProxy.counts(), null, 2)}\n`, { mode: 0o600 }).catch(() => undefined); await storageProxy.close().catch(() => undefined); @@ -696,7 +731,7 @@ async function main(): Promise { if (lines.length) { await appendLedger(resultsRoot, lines); const summary = await writeSummary(resultsRoot, plan.waveId); - await writeFile(path.join(waveDir, "SUMMARY.md"), speedSection ? `${summary}\n${speedSection}` : summary, { mode: 0o600 }); + await writeFile(path.join(waveDir, "SUMMARY.md"), [summary, speedSection, storageSection].filter(Boolean).join("\n"), { mode: 0o600 }); console.info(`\n${summary}`); } } diff --git a/tests/restart-chaos/storage-proxy.ts b/tests/restart-chaos/storage-proxy.ts index df90bfef56f..42d95da93dd 100644 --- a/tests/restart-chaos/storage-proxy.ts +++ b/tests/restart-chaos/storage-proxy.ts @@ -22,6 +22,8 @@ const CREDENTIAL_ERROR_MESSAGES: Record = { export interface StorageFaultRule { fault: StorageFault; methods?: readonly string[]; pathIncludes?: string; remaining?: number } +export type StorageRequestListener = (incoming: IncomingMessage, outgoing: ServerResponse) => number | void; + export interface StorageProxyCounts { requests: number; injected: number; byMethod: Record; byStatus: Record } const XML = "application/xml"; @@ -53,6 +55,7 @@ export async function startStorageFaultProxy(endpoint: string, random: () => num throw new Error("Storage fault injection requires a loopback store"); } let rules: StorageFaultRule[] = []; + let listener: StorageRequestListener | null = null; const counts: StorageProxyCounts = { requests: 0, injected: 0, byMethod: {}, byStatus: {} }; const sockets = new Set(); const delayed = new Set>(); @@ -83,9 +86,10 @@ export async function startStorageFaultProxy(endpoint: string, random: () => num }; const server = createServer((incoming, outgoing) => { const method = incoming.method ?? "GET"; - const rule = rules.find(candidate => matches(candidate, method, incoming.url ?? "/")); + const heldMs = listener?.(incoming, outgoing) || 0; + const rule = heldMs ? undefined : rules.find(candidate => matches(candidate, method, incoming.url ?? "/")); if (rule?.remaining !== undefined) rule.remaining -= 1; - const fault = rule?.fault ?? { kind: "pass" as const }; + const fault: StorageFault = heldMs ? { kind: "latency", ms: heldMs } : rule?.fault ?? { kind: "pass" }; if (fault.kind === "latency") { const delay = fault.ms + Math.round((fault.jitterMs ?? 0) * random()); if (fault.bytesPerSecond && !links.has(rule!)) links.set(rule!, linkRates(fault.bytesPerSecond)); @@ -119,6 +123,7 @@ export async function startStorageFaultProxy(endpoint: string, random: () => num counts: () => structuredClone(counts), setRules(next: readonly StorageFaultRule[]) { endLinks(); rules = next.map(rule => ({ ...rule })); }, clear() { endLinks(); rules = []; }, + observe(next: StorageRequestListener | null) { listener = next; }, async close() { for (const timer of delayed) clearTimeout(timer); delayed.clear(); diff --git a/tests/restart-chaos/storage-trigger.test.ts b/tests/restart-chaos/storage-trigger.test.ts new file mode 100644 index 00000000000..7e8fe0fb2ac --- /dev/null +++ b/tests/restart-chaos/storage-trigger.test.ts @@ -0,0 +1,200 @@ +import { createServer, request, type Server } from "node:http"; +import { once } from "node:events"; +import { afterEach, describe, expect, it } from "vitest"; +import { parseCommand } from "./config.js"; +import { startStorageFaultProxy } from "./storage-proxy.js"; +import { exposedAfterLaunch, redactStorageUrl, storageTriggerMarkdown, storageTriggerMatches, storageTriggerShortfall, StorageTrigger, type StorageTriggerReceipt, type ArmedStorageFault, type StorageTriggerHooks } from "./storage-trigger.js"; +import type { FaultReceipt } from "./types.js"; + +const closers: Array<() => Promise> = []; +afterEach(async () => { for (const close of closers.splice(0).reverse()) await close(); }); + +async function upstream() { + const seen: string[] = []; + const server: Server = createServer((incoming, outgoing) => { + incoming.resume(); + incoming.on("end", () => { seen.push(`${incoming.method} ${incoming.url}`); outgoing.writeHead(200).end("ok"); }); + }); + server.listen(0, "127.0.0.1"); await once(server, "listening"); + closers.push(async () => { server.closeAllConnections(); server.close(); }); + return { endpoint: `http://127.0.0.1:${(server.address() as { port: number }).port}`, seen }; +} + +function hooks(overrides: Partial = {}) { + const fired: Array<{ armed: ArmedStorageFault; path: string }> = []; + const events: string[] = []; + const value: StorageTriggerHooks = { + canFire: () => true, isOpen: () => true, + fire: async (armed, held) => { fired.push({ armed, path: held.path }); }, + record: kind => { events.push(kind); }, + ...overrides, + }; + return { value, fired, events }; +} + +function send(endpoint: string, method: string, route: string, body = "bytes") { + const outgoing = request(`${endpoint}${route}`, { method, headers: { "content-length": Buffer.byteLength(body) } }); + outgoing.on("error", () => undefined); + const response = new Promise(resolve => outgoing.on("response", incoming => { incoming.resume(); resolve(incoming.statusCode ?? 0); })); + outgoing.end(body); + return { outgoing, response }; +} + +const until = async (check: () => boolean) => { for (let index = 0; index < 200 && !check(); index += 1) await new Promise(resolve => setTimeout(resolve, 10)); }; +const snapshotPut = "/bucket/chaos/w/c1/workspace-snapshots/ws1/s1/u1.pcws"; + +describe("storage trigger targets", () => { + it("matches snapshot uploads, parts, downloads and run-log writes only", () => { + expect(storageTriggerMatches("snapshot-upload", "PUT", snapshotPut, 4096)).toBe(true); + expect(storageTriggerMatches("snapshot-upload", "PUT", `${snapshotPut}?partNumber=1&uploadId=u`, 4096)).toBe(true); + expect(storageTriggerMatches("snapshot-upload", "PUT", snapshotPut, 0)).toBe(false); + expect(storageTriggerMatches("snapshot-upload", "POST", `${snapshotPut}?uploads`, 0)).toBe(false); + expect(storageTriggerMatches("snapshot-upload", "POST", `${snapshotPut}?uploadId=u`, 120)).toBe(false); + expect(storageTriggerMatches("snapshot-upload", "GET", snapshotPut)).toBe(false); + expect(storageTriggerMatches("snapshot-part", "PUT", `${snapshotPut}?partNumber=2&uploadId=abc`)).toBe(true); + expect(storageTriggerMatches("snapshot-part", "PUT", snapshotPut)).toBe(false); + expect(storageTriggerMatches("snapshot-download", "GET", snapshotPut)).toBe(true); + expect(storageTriggerMatches("snapshot-download", "HEAD", snapshotPut)).toBe(false); + expect(storageTriggerMatches("run-log", "PUT", "/bucket/chaos/w/run-logs/c1/r1/000001.ndjson")).toBe(true); + expect(storageTriggerMatches("run-log", "GET", "/bucket/chaos/w/run-logs/c1/r1/000001.ndjson")).toBe(false); + expect(storageTriggerMatches("run-log", "PUT", snapshotPut)).toBe(false); + }); + + it("keeps part numbers but hides upload ids and signatures in receipts", () => { + expect(redactStorageUrl(`${snapshotPut}?partNumber=2&uploadId=abc&X-Amz-Signature=s`)).toBe(`${snapshotPut}?partNumber=2&uploadId=…`); + expect(redactStorageUrl(`${snapshotPut}?uploads`)).toBe(`${snapshotPut}?uploads`); + expect(redactStorageUrl(`${snapshotPut}?x-id=PutObject`)).toBe(`${snapshotPut}?x-id=PutObject`); + }); + + it("parses --storage-trigger only with s3 storage and time triggers", () => { + const base = ["--env", "local", "--adapters", "codex", "--workloads", "A1", "--faults", "K1"]; + const parsed = parseCommand([...base, "--storage", "s3", "--storage-trigger", "run-log"]); + expect(parsed.mode === "wave" && parsed.options.storageTrigger).toBe("run-log"); + expect(() => parseCommand([...base, "--storage-trigger", "run-log"])).toThrow(/needs --storage s3/); + expect(() => parseCommand([...base, "--storage", "s3", "--storage-trigger", "everything"])).toThrow(/takes one of/); + expect(() => parseCommand([...base, "--workloads", "C1", "--triggers", "comment", "--storage", "s3", "--storage-trigger", "run-log"])).toThrow(/--triggers time/); + expect(() => parseCommand([...base, "--workloads", "A1,G1", "--storage", "s3", "--storage-trigger", "run-log"])).toThrow(/board-acting workloads \(G1\)/); + expect(() => parseCommand([...base, "--workloads", "C1", "--matrix", "--storage", "s3", "--storage-trigger", "run-log"])).toThrow(/--matrix/); + }); +}); + +describe("storage-triggered restart", () => { + it("fires an armed fault on the matching request, holds it open and records what was in flight", async () => { + const store = await upstream(); + const proxy = await startStorageFaultProxy(store.endpoint); + closers.push(() => proxy.close()); + const { value, fired, events } = hooks(); + const trigger = new StorageTrigger("snapshot-upload", value, 5_000); + proxy.observe(trigger.listener); + + const before = send(proxy.endpoint, "PUT", snapshotPut); + expect(await before.response).toBe(200); + expect(fired).toEqual([]); + + trigger.arm({ taskId: "t01", kind: "K1", holdMs: 0, armedAt: Date.now(), restartSteps: [{ gapMs: 5_000, signal: "SIGKILL" }] }); + const other = send(proxy.endpoint, "PUT", "/bucket/chaos/w/run-logs/r1/1.ndjson"); + expect(await other.response).toBe(200); + const held = send(proxy.endpoint, "PUT", `${snapshotPut}?partNumber=1&uploadId=u`, "x".repeat(1024)); + await until(() => fired.length > 0); + expect(fired).toEqual([{ armed: expect.objectContaining({ taskId: "t01", kind: "K1", restartSteps: [{ gapMs: 5_000, signal: "SIGKILL" }] }), path: `${snapshotPut}?partNumber=1&uploadId=…` }]); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(store.seen.filter(line => line.includes("partNumber"))).toEqual([]); + + held.outgoing.destroy(); + await until(() => trigger.receipts[0]?.request?.outcome !== "in_flight"); + const faults = [{ faultId: "f01", requestedBy: "t01", requestedAt: Date.now() } as FaultReceipt]; + const [receipt] = trigger.finish(faults); + expect(receipt).toMatchObject({ target: "snapshot-upload", taskId: "t01", faultId: "f01", missedAt: null, + request: { method: "PUT", held: true, declaredBytes: 1024, outcome: "aborted", status: null } }); + expect(receipt!.inFlight.map(entry => entry.method)).toEqual(["PUT"]); + expect(events).toEqual(["storage_trigger_armed", "storage_trigger_fired"]); + expect(proxy.counts().byMethod.PUT).toBe(2); + }); + + it("skips a match while a fault cannot fire and records armed faults that never matched", async () => { + const store = await upstream(); + const proxy = await startStorageFaultProxy(store.endpoint); + closers.push(() => proxy.close()); + let ready = false; + const { value, fired } = hooks({ canFire: () => ready, isOpen: taskId => taskId !== "t02" }); + const trigger = new StorageTrigger("snapshot-download", value, 5_000); + proxy.observe(trigger.listener); + trigger.arm({ taskId: "t02", kind: "K2", holdMs: 0, armedAt: 1 }); + trigger.arm({ taskId: "t03", kind: "K4", holdMs: 30_000, armedAt: 2 }); + trigger.arm({ taskId: "t04", kind: "K1", holdMs: 0, armedAt: 3 }); + + expect(await send(proxy.endpoint, "GET", snapshotPut, "").response).toBe(200); + expect(fired).toEqual([]); + ready = true; + const held = send(proxy.endpoint, "GET", snapshotPut, ""); + await until(() => fired.length > 0); + expect(fired.map(entry => entry.armed.taskId)).toEqual(["t03"]); + held.outgoing.destroy(); + + trigger.arm({ taskId: "t02", kind: "K1", holdMs: 0, armedAt: 4 }); + const receipts = trigger.finish([]); + expect(receipts.map(receipt => [receipt.taskId, receipt.missedReason ?? null, receipt.firedAt !== null])).toEqual([ + ["t02", "task_rested", false], ["t03", null, true], ["t04", "no_matching_request", false], ["t02", "task_rested", false]]); + }); + + it("forwards a held request after the hold when the restart never came", async () => { + const store = await upstream(); + const proxy = await startStorageFaultProxy(store.endpoint); + closers.push(() => proxy.close()); + const { value } = hooks(); + const trigger = new StorageTrigger("run-log", value, 50); + proxy.observe(trigger.listener); + trigger.arm({ taskId: "t01", kind: "K1", holdMs: 0, armedAt: 1 }); + expect(await send(proxy.endpoint, "PUT", "/bucket/chaos/w/run-logs/r1/1.ndjson").response).toBe(200); + await until(() => trigger.receipts[0]?.request?.outcome === "completed"); + expect(trigger.receipts[0]!.request).toMatchObject({ held: true, outcome: "completed", status: 200 }); + }); +}); + +describe("storage trigger receipts", () => { + const receipt = (taskId: string, firedAt: number | null, outcome: "aborted" | "completed" = "aborted", faultId?: string): StorageTriggerReceipt => ({ + target: "snapshot-upload", taskId, kind: "K1", armedAt: 0, firedAt, missedAt: firedAt === null ? 1 : null, inFlight: [], + request: firedAt === null ? null : { id: 1, method: "PUT", path: snapshotPut, startedAt: firedAt, declaredBytes: 1, held: true, outcome, status: null, settledAt: firedAt + 1 }, + ...(faultId ? { faultId } : {}) }); + + it("links each fired receipt to the first fault its task requested at or after the fire", () => { + const trigger = new StorageTrigger("snapshot-upload", hooks().value); + trigger.receipts.push(receipt("t01", 100), receipt("t01", 500), receipt("t02", 300)); + const faults = [{ faultId: "f03", requestedBy: "t01", requestedAt: 520 }, { faultId: "f01", requestedBy: "t01", requestedAt: 110 }, + { faultId: "f00", requestedBy: "t01", requestedAt: 50 }, { faultId: "f02", requestedBy: "t02", requestedAt: 305 }] as FaultReceipt[]; + expect(trigger.finish(faults).map(entry => entry.faultId)).toEqual(["f01", "f03", "f02"]); + }); + + it("fails a wave whose restarts did not provably interrupt a held transfer", () => { + expect(storageTriggerShortfall([receipt("t01", 10, "aborted", "f01"), receipt("t02", null)])).toBeNull(); + expect(storageTriggerShortfall([receipt("t01", null), receipt("t02", null)])).toMatch(/no restart interrupted a matching transfer \(2 armed, none fired\)/); + expect(storageTriggerShortfall([])).toMatch(/0 armed/); + expect(storageTriggerShortfall([receipt("t01", 10, "completed", "f01")])).toMatch(/t01 fault=f01 outcome=completed/); + expect(storageTriggerShortfall([receipt("t01", 10, "aborted")])).toMatch(/t01 fault=none outcome=aborted/); + }); + + it("summarises interrupted and unreached faults for the wave summary", () => { + const text = storageTriggerMarkdown([receipt("t01", 10, "aborted", "f01"), { ...receipt("t02", null), missedReason: "task_rested" }]); + expect(text).toContain("1 of 2 armed faults interrupted a transfer; 1 not reached."); + expect(text).toContain("| t01 | K1 | f01 | aborted | `PUT ws1/s1/u1.pcws` |"); + expect(text).toContain("| t02 | K1 | - | not reached (task_rested) | - |"); + }); +}); + +describe("storage-triggered restart exposure", () => { + it("waits for a launch in flight and exposes the task it launched", async () => { + const launched = [{ id: "t01", done: false }, { id: "t02", done: true }]; + let finish!: () => void; + const launching = new Promise(resolve => { finish = resolve; }); + launched.push({ id: "t03", done: false }); + const exposed = exposedAfterLaunch(launching, launched); + let settled = false; + void exposed.then(() => { settled = true; }); + await new Promise(resolve => setTimeout(resolve, 10)); + expect(settled).toBe(false); + launched.push({ id: "t04", done: false }); + finish(); + expect((await exposed).map(task => task.id)).toEqual(["t01", "t03", "t04"]); + expect((await exposedAfterLaunch(null, launched)).map(task => task.id)).toEqual(["t01", "t03", "t04"]); + }); +}); diff --git a/tests/restart-chaos/storage-trigger.ts b/tests/restart-chaos/storage-trigger.ts new file mode 100644 index 00000000000..c8cc2e017e5 --- /dev/null +++ b/tests/restart-chaos/storage-trigger.ts @@ -0,0 +1,167 @@ +import type { IncomingMessage, ServerResponse } from "node:http"; +import type { FaultKind, FaultReceipt, WaveOptions, WorkloadId } from "./types.js"; +import type { StorageRequestListener } from "./storage-proxy.js"; +import type { RestartStep } from "./repeated-restarts.js"; + +export const STORAGE_TRIGGER_TARGETS = ["snapshot-upload", "snapshot-part", "snapshot-download", "run-log"] as const; +export type StorageTriggerTarget = typeof STORAGE_TRIGGER_TARGETS[number]; +export const STORAGE_TRIGGER_HOLD_MS = 60_000; + +export const STORAGE_TRIGGER_BOARD_WORKLOADS: readonly WorkloadId[] = ["G1", "H1", "Q1"]; + +export function storageTriggerMatches(target: StorageTriggerTarget, method: string, url: string, declaredBytes: number | null = null): boolean { + const [route = "", query = ""] = url.split("?", 2); + switch (target) { + case "snapshot-upload": return method === "PUT" && route.includes("/workspace-snapshots/") && (declaredBytes ?? 0) > 0; + case "snapshot-part": return method === "PUT" && route.includes("/workspace-snapshots/") && /(^|&)partNumber=/.test(query); + case "snapshot-download": return method === "GET" && route.includes("/workspace-snapshots/"); + case "run-log": return (method === "PUT" || method === "POST") && route.includes("/run-logs/"); + } +} + +export function parseStorageTrigger(value: string, options: Pick): StorageTriggerTarget { + if (options.storage !== "s3") throw new Error("--storage-trigger needs --storage s3"); + if (options.stormMinutes || options.crashPoint) throw new Error("--storage-trigger arms each task's planned fault; a storm or crash-point wave has none"); + if (options.matrix) throw new Error("--storage-trigger arms faults by time; --matrix plans trigger cells whose windows it would skip"); + if ((options.triggers ?? []).some(trigger => trigger !== "time")) throw new Error("--storage-trigger replaces the fault's trigger; use it with --triggers time"); + const board = options.workloads.filter(workload => STORAGE_TRIGGER_BOARD_WORKLOADS.includes(workload)); + if (board.length) throw new Error(`--storage-trigger keeps a fault armed after its planned time; board-acting workloads (${board.join(", ")}) would act before the restart`); + if (!(STORAGE_TRIGGER_TARGETS as readonly string[]).includes(value)) { + throw new Error(`--storage-trigger takes one of ${STORAGE_TRIGGER_TARGETS.join(", ")}`); + } + return value as StorageTriggerTarget; +} + +export function redactStorageUrl(url: string): string { + const [route = "", query] = url.split("?", 2); + if (query === undefined) return route; + const kept = query.split("&").filter(part => part && !/^x-amz-/i.test(part)).map(part => { + const [name = "", raw] = part.split("=", 2); + return name === "partNumber" || name === "x-id" || raw === undefined ? part : `${name}=…`; + }); + return kept.length ? `${route}?${kept.join("&")}` : route; +} + +export type StorageRequestOutcome = "in_flight" | "completed" | "aborted"; + +export interface StorageRequestRecord { + id: number; method: string; path: string; startedAt: number; declaredBytes: number | null; + held: boolean; outcome: StorageRequestOutcome; status: number | null; settledAt: number | null; +} + +export interface ArmedStorageFault { taskId: string; kind: FaultKind; holdMs: number; armedAt: number; restartSteps?: RestartStep[] } + +export interface StorageTriggerReceipt { + target: StorageTriggerTarget; taskId: string; kind: FaultKind; armedAt: number; + firedAt: number | null; missedAt: number | null; missedReason?: string; + request: StorageRequestRecord | null; inFlight: StorageRequestRecord[]; faultId?: string; +} + +export interface StorageTriggerHooks { + canFire(): boolean; + isOpen(taskId: string): boolean; + fire(armed: ArmedStorageFault, request: StorageRequestRecord): Promise; + record(kind: string, data: Record): void; +} + +export class StorageTrigger { + readonly receipts: StorageTriggerReceipt[] = []; + private readonly armed: Array<{ fault: ArmedStorageFault; receipt: StorageTriggerReceipt }> = []; + private readonly open = new Map(); + private nextId = 1; + private firing = false; + + constructor(readonly target: StorageTriggerTarget, private readonly hooks: StorageTriggerHooks, private readonly holdMs = STORAGE_TRIGGER_HOLD_MS) {} + + get busy(): boolean { return this.firing; } + + arm(fault: ArmedStorageFault): void { + const receipt: StorageTriggerReceipt = { target: this.target, taskId: fault.taskId, kind: fault.kind, armedAt: fault.armedAt, + firedAt: null, missedAt: null, request: null, inFlight: [] }; + this.receipts.push(receipt); + this.armed.push({ fault, receipt }); + this.hooks.record("storage_trigger_armed", { taskId: fault.taskId, fault: fault.kind, target: this.target }); + } + + readonly listener: StorageRequestListener = (incoming: IncomingMessage, outgoing: ServerResponse) => { + const method = incoming.method ?? "GET"; + const url = incoming.url ?? "/"; + const length = Number(incoming.headers["content-length"]); + const request: StorageRequestRecord = { id: this.nextId++, method, path: redactStorageUrl(url), startedAt: Date.now(), + declaredBytes: Number.isFinite(length) ? length : null, held: false, outcome: "in_flight", status: null, settledAt: null }; + this.open.set(request.id, request); + outgoing.on("close", () => { + this.open.delete(request.id); + request.outcome = outgoing.writableFinished ? "completed" : "aborted"; + request.status = outgoing.headersSent ? outgoing.statusCode : null; + request.settledAt = Date.now(); + }); + if (!storageTriggerMatches(this.target, method, url, request.declaredBytes)) return; + const next = this.take(request.startedAt); + if (!next) return; + request.held = true; + this.firing = true; + next.receipt.firedAt = request.startedAt; + next.receipt.request = request; + next.receipt.inFlight = [...this.open.values()]; + this.hooks.record("storage_trigger_fired", { taskId: next.fault.taskId, fault: next.fault.kind, target: this.target, + request: `${method} ${request.path}`, inFlight: next.receipt.inFlight.length }); + void this.hooks.fire(next.fault, request).catch(error => { + this.hooks.record("storage_trigger_fire_failed", { taskId: next.fault.taskId, error: String(error) }); + }).finally(() => { this.firing = false; }); + return this.holdMs; + }; + + private take(now: number) { + if (this.firing || !this.armed.length || !this.hooks.canFire()) return null; + while (this.armed.length) { + const next = this.armed.shift()!; + if (this.hooks.isOpen(next.fault.taskId)) return next; + this.miss(next.receipt, now, "task_rested"); + } + return null; + } + + private miss(receipt: StorageTriggerReceipt, at: number, reason: string) { + receipt.missedAt = at; + receipt.missedReason = reason; + this.hooks.record("storage_trigger_missed", { taskId: receipt.taskId, fault: receipt.kind, target: this.target, reason }); + } + + finish(faults: readonly FaultReceipt[], now = Date.now()): StorageTriggerReceipt[] { + for (const { receipt } of this.armed.splice(0)) this.miss(receipt, now, this.hooks.isOpen(receipt.taskId) ? "no_matching_request" : "task_rested"); + for (const receipt of this.receipts) { + if (receipt.firedAt === null) continue; + const fault = faults.filter(candidate => candidate.requestedBy === receipt.taskId && candidate.requestedAt >= receipt.firedAt!) + .sort((left, right) => left.requestedAt - right.requestedAt)[0]; + if (fault) receipt.faultId = fault.faultId; + } + return this.receipts; + } +} + +export function storageTriggerShortfall(receipts: readonly StorageTriggerReceipt[]): string | null { + const fired = receipts.filter(receipt => receipt.firedAt !== null); + if (!fired.length) return `--storage-trigger: no restart interrupted a matching transfer (${receipts.length} armed, none fired)`; + const unproven = fired.filter(receipt => !receipt.faultId || receipt.request?.outcome !== "aborted"); + if (!unproven.length) return null; + return `--storage-trigger: ${unproven.length} fired restart(s) did not provably interrupt the held request: ${unproven + .map(receipt => `${receipt.taskId} fault=${receipt.faultId ?? "none"} outcome=${receipt.request?.outcome ?? "none"}`).join(", ")}`; +} + +export function storageTriggerMarkdown(receipts: readonly StorageTriggerReceipt[]): string { + const fired = receipts.filter(receipt => receipt.firedAt !== null); + const proven = fired.filter(receipt => receipt.faultId && receipt.request?.outcome === "aborted"); + const lines = [`## Storage trigger (${receipts[0]?.target ?? "none"})`, "", + `${proven.length} of ${receipts.length} armed faults interrupted a transfer; ${receipts.length - fired.length} not reached.`, "", + "| Task | Fault | Fault id | Result | Request |", "| --- | --- | --- | --- | --- |", + ...receipts.map(receipt => `| ${receipt.taskId} | ${receipt.kind} | ${receipt.faultId ?? "-"} | ${receipt.firedAt === null + ? `not reached (${receipt.missedReason ?? "unknown"})` : receipt.request?.outcome ?? "unknown"} | ${receipt.request + ? `\`${receipt.request.method} ${receipt.request.path.split("/").slice(-3).join("/")}\`` : "-"} |`)]; + return `${lines.join("\n")}\n`; +} + +export async function exposedAfterLaunch(launching: Promise | null, launched: readonly T[]): Promise { + await launching; + return launched.filter(candidate => !candidate.done); +} diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index dd985a3f678..7f680dd70d0 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -96,6 +96,7 @@ export interface WaveOptions { storageFaults?: Array; storageFaultMs?: [number, number]; storageWindows?: import("./storage-chaos.js").StorageFaultWindow[]; + storageTrigger?: import("./storage-trigger.js").StorageTriggerTarget; workspaceCommit?: string; workspaceGitUrl?: string; /** Provider message faults from `--faults` (PV1 delay, PV2 drop, PV3 duplicate); see provider-faults.ts. */