diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index ed117ef867e..672f94618d4 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -418,6 +418,26 @@ fails when no restart fired, or when a fired restart has no fault id or its held `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. +### Restart at native start (`--native-trigger`) + +A time-based fault on a remote workspace often lands while the run is still preparing. `--native-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 `` (0-600) after the next matching event of the task's running run. + +| Target | Matching run event | +| --- | --- | +| `native-start` | `native.process_start_requested` (the executor is about to start the runner) | +| `provider-event` | the first `session.started`, `turn.submitted`, `turn.started` or `turn.accepted` | + +For example, `--native-trigger native-start+5` restarts the server five seconds after native start. `native-trigger.json` records, +per armed fault, the run and event that fired it, the delay after the event, and its fault id. A fault whose task rested first, or +that saw no matching event, is recorded with `missedReason` (`task_rested` or `no_matching_event`). The wave summary adds a "Native +trigger" section and fails when no restart fired. It takes `--triggers time` only, not `--storage-trigger`, and not G1, H1 or Q1. + +`--native-trigger-outage ` (10-900, with `--storage s3` and no `--storage-faults`) also starts an ST1 outage of that +length the moment each trigger fires, just before the restart. The recovered run then prepares while the store is down. The +windows are written to `storage-faults.json` and removed from the speed grade like planned storage faults. + 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 aab847bef0c..11358e0d13d 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -7,6 +7,7 @@ import { STORAGE_FAULT_KINDS, type StorageFaultKind } from "./storage-chaos.js"; import { validateMultiServerOptions } from "./multi-server.js"; import { parseTimeFaultOptions } from "./time-faults-config.js"; import { parseStorageTrigger } from "./storage-trigger.js"; +import { parseNativeTrigger, parseNativeTriggerOutage } from "./native-trigger.js"; import { parseRepeatedRestartOptions } from "./repeated-restarts.js"; import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, isProviderFault, parseProviderDelays, parseProviderMessages, type ProviderFaultKind } from "./provider-faults.js"; import { PROVIDER_FAULT_ADAPTERS, parseProviderFaultRestart } from "./acpx-provider-faults.js"; @@ -112,6 +113,8 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC "storage-faults": { type: "string" }, "storage-fault-seconds": { type: "string" }, "storage-trigger": { type: "string" }, + "native-trigger": { type: "string" }, + "native-trigger-outage": { type: "string" }, "provider-fault-message": { type: "string" }, "provider-fault-delays": { type: "string" }, "db-pressure": { type: "string" }, @@ -128,7 +131,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", "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", "rr-stop-on"] 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", "native-trigger", "native-trigger-outage", "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", "rr-stop-on"] 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, @@ -291,6 +294,8 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC options.storageFaultMs = [min * 1000, max * 1000]; } if (values["storage-trigger"] !== undefined) options.storageTrigger = parseStorageTrigger(values["storage-trigger"], options); + if (values["native-trigger"] !== undefined) options.nativeTrigger = parseNativeTrigger(values["native-trigger"], options); + if (values["native-trigger-outage"] !== undefined) options.nativeTrigger = { ...options.nativeTrigger!, outageMs: parseNativeTriggerOutage(values["native-trigger-outage"], options) }; return { mode: "wave", options, keep }; } diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index d68d6a60e9e..78130a1ed3b 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -15,10 +15,11 @@ import { setTimeout as delay } from "node:timers/promises"; import { ChaosApi, isSocketTypeOfServiceRace } from "./api.js"; 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 { StorageChaos, planStorageFaults, replayStorageWindows, type StorageFaultReceipt, type StorageFaultWindow } from "./storage-chaos.js"; import { startStorageFaultProxy } from "./storage-proxy.js"; import { StorageAuditor, auditsStorage, storageAuditMarkdown } from "./storage-audit.js"; import { exposedAfterLaunch, StorageTrigger, storageTriggerMarkdown, storageTriggerShortfall } from "./storage-trigger.js"; +import { NativeTrigger, nativeTriggerMarkdown, nativeTriggerShortfall, triggeredOutageWindow } from "./native-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"; @@ -254,6 +255,7 @@ async function main(): Promise { let objectStore: ObjectStore | null = null; let storageChaos: StorageChaos | null = null; let storageTrigger: StorageTrigger | null = null; + let nativeTrigger: NativeTrigger | null = null; let storageProxy: Awaited> | null = null; let credentialRotation: CredentialRotation | null = null; let storageAuditor: StorageAuditor | null = null; @@ -414,6 +416,39 @@ async function main(): Promise { }); storageProxy.observe(storageTrigger.listener); } + const outageMs = plan.options.nativeTrigger?.outageMs; + const outageWindows: StorageFaultWindow[] = []; + const outageClock = Date.now(); + if (outageMs && storageProxy && !storageChaos && !plan.replayOf) { + storageChaos = new StorageChaos(outageWindows, outageClock, credentialRotation?.wrap(storageProxy) ?? storageProxy, (kind, data) => timeline.add(kind, data)); + storageChaos.start(1_000, outageClock); + } + if (plan.options.nativeTrigger && !plan.replayOf) { + nativeTrigger = new NativeTrigger(plan.options.nativeTrigger, { + canFire: () => !interrupted && !faultsQueued && !supervisor.faultInProgress && supervisor.serverReady, + isOpen: taskId => observers.some(observer => observer.plan.taskId === taskId && !observer.done && observer.record.restAt === null), + issueId: taskId => observers.find(observer => observer.plan.taskId === taskId)?.record.issueId ?? null, + runs: issueId => api.get(`/api/issues/${issueId}/runs`, 5_000), + events: (runId, afterSeq, limit) => api.get(`/api/heartbeat-runs/${runId}/events?afterSeq=${afterSeq}&limit=${limit}`, 5_000), + 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; + if (outageMs) { + outageWindows.push(triggeredOutageWindow(Date.now() - outageClock, outageMs)); + storageChaos?.tick(); + } + fireFault(armed.kind, armed.holdMs, armed.taskId, exposed, undefined, undefined, armed.restartSteps); + await faultWork[queued]; + })().finally(() => { faultsQueued -= 1; }); + faultWork.push(work); + return work; + }, + }); + } supervisor.onCrashPoint = receipt => { const owner = pointResults.find(result => result.token === receipt.crashPoint?.token); @@ -467,8 +502,9 @@ async function main(): Promise { errors: check.errors.map(error => redactKnownSecrets(error, knownSecretValues())) }); }; - while (!interrupted && (pending.length || active.length || storageTrigger?.busy)) { + while (!interrupted && (pending.length || active.length || storageTrigger?.busy || nativeTrigger?.busy)) { const now = Date.now(); + await nativeTrigger?.tick(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 }); @@ -485,7 +521,7 @@ 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 && !storageTrigger?.busy + while (!crashPoint && active.length < plan.options.concurrency && pending.length && supervisor.serverReady && !supervisor.faultInProgress && !storageTrigger?.busy && !nativeTrigger?.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)))) { @@ -519,6 +555,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 (nativeTrigger) { nativeTrigger.arm({ taskId: observer.plan.taskId, kind: due.kind, holdMs: due.holdMs, armedAt: Date.now(), ...(due.restartSteps ? { restartSteps: due.restartSteps } : {}) }); 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())); @@ -557,6 +594,12 @@ async function main(): Promise { timeline.add("storage_trigger_settled", { shortfall, receipts: storageTrigger.receipts.length }); if (shortfall) harnessError ??= new Error(shortfall); } + if (nativeTrigger) { + const shortfall = nativeTriggerShortfall(nativeTrigger.finish(faults)); + storageSection = [storageSection, nativeTriggerMarkdown(nativeTrigger.receipts)].filter(Boolean).join("\n"); + timeline.add("native_trigger_settled", { shortfall, receipts: nativeTrigger.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"); @@ -739,6 +782,7 @@ async function main(): Promise { storageChaos?.stop(); await credentialRotation?.settled(); if (credentialRotation?.failures.length) harnessError ??= new Error(`ST5R credential rotation failed: ${credentialRotation.failures.join("; ")}`); + if (nativeTrigger) await writeFile(path.join(waveDir, "native-trigger.json"), `${JSON.stringify(nativeTrigger.finish(faults), null, 2)}\n`, { mode: 0o600 }).catch(() => undefined); 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); diff --git a/tests/restart-chaos/native-trigger.test.ts b/tests/restart-chaos/native-trigger.test.ts new file mode 100644 index 00000000000..5eb1d278b7b --- /dev/null +++ b/tests/restart-chaos/native-trigger.test.ts @@ -0,0 +1,163 @@ +import { describe, expect, it } from "vitest"; +import { parseCommand } from "./config.js"; +import { NATIVE_START_EVENT, NativeTrigger, runningRunIds, triggeredOutageWindow, nativeTriggerMarkdown, nativeTriggerMatches, nativeTriggerShortfall, type ArmedNativeFault, type NativeTriggerEvent, type NativeTriggerHooks } from "./native-trigger.js"; +import type { FaultReceipt } from "./types.js"; + +function harness(options: { events?: Record; runs?: unknown[]; open?: boolean; canFire?: () => boolean } = {}) { + const fired: ArmedNativeFault[] = []; + const recorded: string[] = []; + const events = options.events ?? {}; + const hooks: NativeTriggerHooks = { + canFire: options.canFire ?? (() => true), + isOpen: () => options.open ?? true, + issueId: taskId => `issue-${taskId}`, + runs: async () => options.runs ?? [{ runId: "run-1", status: "running" }], + events: async (runId, afterSeq, limit) => (events[runId] ?? []).filter(event => event.seq > afterSeq).slice(0, limit), + fire: async armed => { fired.push(armed); }, + record: kind => { recorded.push(kind); }, + }; + return { hooks, fired, recorded, events }; +} + +const iso = (ms: number) => new Date(ms).toISOString(); +const armed = (armedAt: number): ArmedNativeFault => ({ taskId: "t01", kind: "K2", holdMs: 0, armedAt }); + +describe("--native-trigger parsing", () => { + const base = ["--env", "local", "--adapters", "codex", "--workloads", "A1", "--faults", "K2"]; + + it("takes a target with an optional delay", () => { + const parsed = parseCommand([...base, "--native-trigger", "native-start+5"]); + expect(parsed.mode === "wave" && parsed.options.nativeTrigger).toEqual({ target: "native-start", delayMs: 5_000 }); + const provider = parseCommand([...base, "--native-trigger", "provider-event"]); + expect(provider.mode === "wave" && provider.options.nativeTrigger).toEqual({ target: "provider-event", delayMs: 0 }); + }); + + it("rejects unknown targets, long delays and conflicting trigger options", () => { + expect(() => parseCommand([...base, "--native-trigger", "boot"])).toThrow(/takes native-start or provider-event/); + expect(() => parseCommand([...base, "--native-trigger", "native-start+601"])).toThrow(/0-600/); + expect(() => parseCommand([...base, "--storage", "s3", "--storage-trigger", "run-log", "--native-trigger", "native-start"])).toThrow(/pick one/); + expect(() => parseCommand([...base, "--workloads", "C1", "--triggers", "comment", "--native-trigger", "native-start"])).toThrow(/--triggers time/); + expect(() => parseCommand([...base, "--workloads", "A1,G1", "--native-trigger", "native-start"])).toThrow(/board-acting workloads \(G1\)/); + }); + + it("starts an ST1 outage when the trigger fires only with s3 storage and no planned storage faults", () => { + const parsed = parseCommand([...base, "--storage", "s3", "--native-trigger", "native-start+5", "--native-trigger-outage", "120"]); + expect(parsed.mode === "wave" && parsed.options.nativeTrigger).toEqual({ target: "native-start", delayMs: 5_000, outageMs: 120_000 }); + expect(() => parseCommand([...base, "--storage", "s3", "--native-trigger-outage", "120"])).toThrow(/needs --native-trigger/); + expect(() => parseCommand([...base, "--native-trigger", "native-start", "--native-trigger-outage", "120"])).toThrow(/needs --storage s3/); + expect(() => parseCommand([...base, "--storage", "s3", "--storage-faults", "ST1", "--native-trigger", "native-start", "--native-trigger-outage", "120"])).toThrow(/drop --storage-faults/); + expect(() => parseCommand([...base, "--storage", "s3", "--native-trigger", "native-start", "--native-trigger-outage", "5"])).toThrow(/10 to 900/); + expect(triggeredOutageWindow(4_000, 120_000)).toEqual({ kind: "ST1", startMs: 4_000, durationMs: 120_000, + rules: [{ fault: { kind: "outage", status: 503 } }] }); + }); + + it("reads running run ids from the issue runs API shape", () => { + expect(runningRunIds([{ runId: "a", status: "running" }, { runId: "b", status: "failed" }, { id: "c", status: "running" }, null])).toEqual(["a", "c"]); + expect(runningRunIds(null)).toEqual([]); + }); + + it("matches native start and the first provider events only", () => { + expect(nativeTriggerMatches("native-start", NATIVE_START_EVENT)).toBe(true); + expect(nativeTriggerMatches("native-start", "native.process_identity_recorded")).toBe(false); + expect(nativeTriggerMatches("provider-event", "session.started")).toBe(true); + expect(nativeTriggerMatches("provider-event", "lifecycle")).toBe(false); + }); +}); + +describe("native-triggered restart", () => { + it("fires the armed fault the delay after a native start that followed arming", async () => { + const armedAt = 1_000_000; + const h = harness({ events: { "run-1": [ + { seq: 1, eventType: NATIVE_START_EVENT, createdAt: iso(armedAt - 5_000) }, + { seq: 2, eventType: "lifecycle", createdAt: iso(armedAt + 1_000) }, + { seq: 3, eventType: NATIVE_START_EVENT, createdAt: iso(armedAt + 2_000) }, + ] } }); + const trigger = new NativeTrigger({ target: "native-start", delayMs: 4_000 }, h.hooks); + trigger.arm(armed(armedAt)); + await trigger.tick(armedAt + 3_000); + expect(h.fired).toEqual([]); + expect(trigger.receipts[0]).toMatchObject({ seenAt: armedAt + 2_000, runId: "run-1", eventType: NATIVE_START_EVENT }); + await trigger.tick(armedAt + 6_000); + expect(h.fired).toEqual([armed(armedAt)]); + expect(trigger.receipts[0]!.firedAt).toBe(armedAt + 6_000); + await trigger.tick(armedAt + 7_000); + expect(h.fired).toHaveLength(1); + }); + + it("waits while another fault is in progress and only reads running runs", async () => { + let free = false; + const armedAt = 2_000_000; + const h = harness({ canFire: () => free, + runs: [{ runId: "old", status: "failed" }, { runId: "run-1", status: "running" }], + events: { old: [{ seq: 1, eventType: NATIVE_START_EVENT, createdAt: iso(armedAt + 500) }], + "run-1": [{ seq: 7, eventType: "session.started", createdAt: iso(armedAt + 1_000) }] } }); + const trigger = new NativeTrigger({ target: "provider-event", delayMs: 0 }, h.hooks); + trigger.arm(armed(armedAt)); + await trigger.tick(armedAt + 2_000); + expect(h.fired).toEqual([]); + free = true; + await trigger.tick(armedAt + 3_000); + expect(h.fired).toHaveLength(1); + expect(trigger.receipts[0]).toMatchObject({ runId: "run-1", eventType: "session.started" }); + }); + + it("stays busy while a seen event waits out its delay and still fails a wave in which no restart fired", async () => { + const armedAt = 4_000_000; + let open = true; + const h = harness({ events: { "run-1": [{ seq: 1, eventType: NATIVE_START_EVENT, createdAt: iso(armedAt + 10) }] } }); + h.hooks.isOpen = () => open; + const trigger = new NativeTrigger({ target: "native-start", delayMs: 600_000 }, h.hooks); + trigger.arm(armed(armedAt)); + expect(trigger.busy).toBe(false); + await trigger.tick(armedAt + 1_000); + expect(trigger.busy).toBe(true); + open = false; + await trigger.tick(armedAt + 2_000); + expect(trigger.busy).toBe(false); + const receipts = trigger.finish([]); + expect(receipts[0]).toMatchObject({ missedReason: "task_rested_after_event", firedAt: null }); + expect(nativeTriggerShortfall(receipts)).toMatch(/1 saw the event, none fired/); + }); + + it("reads at most three event pages per run in one pass and resumes from its cursor", async () => { + const armedAt = 5_000_000; + const deltas = Array.from({ length: 700 }, (_, index) => ({ seq: index + 1, eventType: "item.delta", createdAt: iso(armedAt + 1) })); + const h = harness({ events: { "run-1": [...deltas, { seq: 701, eventType: NATIVE_START_EVENT, createdAt: iso(armedAt + 2) }] } }); + let reads = 0; + const events = h.hooks.events; + h.hooks.events = async (...args) => { reads += 1; return events(...args); }; + const trigger = new NativeTrigger({ target: "native-start", delayMs: 0 }, h.hooks); + trigger.arm(armed(armedAt)); + await trigger.tick(armedAt + 100); + expect(reads).toBe(3); + expect(h.fired).toEqual([]); + await trigger.tick(armedAt + 200); + expect(h.fired).toHaveLength(1); + expect(trigger.receipts[0]).toMatchObject({ eventType: NATIVE_START_EVENT }); + }); + + it("records a missed trigger when the task rests or no event arrives, and links fired faults", async () => { + const resting = harness({ open: false }); + const rested = new NativeTrigger({ target: "native-start", delayMs: 0 }, resting.hooks); + rested.arm(armed(1)); + await rested.tick(10); + expect(rested.receipts[0]).toMatchObject({ missedReason: "task_rested", firedAt: null }); + expect(nativeTriggerShortfall(rested.finish([]))).toMatch(/none fired/); + + const quiet = new NativeTrigger({ target: "native-start", delayMs: 0 }, harness().hooks); + quiet.arm(armed(1)); + await quiet.tick(10); + expect(quiet.finish([])[0]).toMatchObject({ missedReason: "no_matching_event" }); + + const armedAt = 3_000_000; + const h = harness({ events: { "run-1": [{ seq: 1, eventType: NATIVE_START_EVENT, createdAt: iso(armedAt + 10) }] } }); + const trigger = new NativeTrigger({ target: "native-start", delayMs: 0 }, h.hooks); + trigger.arm(armed(armedAt)); + await trigger.tick(armedAt + 100); + const fault = { faultId: "f01", requestedBy: "t01", requestedAt: armedAt + 200 } as FaultReceipt; + const receipts = trigger.finish([fault]); + expect(receipts[0]!.faultId).toBe("f01"); + expect(nativeTriggerShortfall(receipts)).toBeNull(); + expect(nativeTriggerMarkdown(receipts)).toContain("| t01 | K2 | f01 | run-1 | native.process_start_requested | 0.1 s |"); + }); +}); diff --git a/tests/restart-chaos/native-trigger.ts b/tests/restart-chaos/native-trigger.ts new file mode 100644 index 00000000000..68a68b0de82 --- /dev/null +++ b/tests/restart-chaos/native-trigger.ts @@ -0,0 +1,182 @@ +import type { FaultKind, FaultReceipt, WaveOptions, WorkloadId } from "./types.js"; +import type { RestartStep } from "./repeated-restarts.js"; +import { storageFaultRules, type StorageFaultWindow } from "./storage-chaos.js"; + +export const NATIVE_TRIGGER_TARGETS = ["native-start", "provider-event"] as const; +export type NativeTriggerTarget = typeof NATIVE_TRIGGER_TARGETS[number]; +export interface NativeTriggerSpec { target: NativeTriggerTarget; delayMs: number; outageMs?: number } + +export const NATIVE_TRIGGER_BOARD_WORKLOADS: readonly WorkloadId[] = ["G1", "H1", "Q1"]; +export const NATIVE_START_EVENT = "native.process_start_requested"; +const PROVIDER_EVENT_TYPES = new Set(["session.started", "turn.submitted", "turn.started", "turn.accepted"]); +const EVENT_PAGE = 200; +const EVENT_PAGES_PER_LOOK = 3; + +export function nativeTriggerMatches(target: NativeTriggerTarget, eventType: string): boolean { + return target === "native-start" ? eventType === NATIVE_START_EVENT : PROVIDER_EVENT_TYPES.has(eventType); +} + +export function parseNativeTrigger(value: string, options: Pick): NativeTriggerSpec { + if (options.storageTrigger) throw new Error("--native-trigger and --storage-trigger both arm the planned fault; pick one"); + if (options.stormMinutes || options.crashPoint) throw new Error("--native-trigger arms each task's planned fault; a storm or crash-point wave has none"); + if (options.matrix) throw new Error("--native-trigger arms faults by time; --matrix plans trigger cells whose windows it would skip"); + if ((options.triggers ?? []).some(trigger => trigger !== "time")) throw new Error("--native-trigger replaces the fault's trigger; use it with --triggers time"); + const board = options.workloads.filter(workload => NATIVE_TRIGGER_BOARD_WORKLOADS.includes(workload)); + if (board.length) throw new Error(`--native-trigger keeps a fault armed after its planned time; board-acting workloads (${board.join(", ")}) would act before the restart`); + const match = /^([a-z-]+)(?:\+(\d+))?$/.exec(value); + const target = match?.[1]; + const seconds = match?.[2] === undefined ? 0 : Number(match[2]); + if (!target || !(NATIVE_TRIGGER_TARGETS as readonly string[]).includes(target) || seconds > 600) { + throw new Error(`--native-trigger takes ${NATIVE_TRIGGER_TARGETS.join(" or ")}, optionally + (0-600), for example native-start+5`); + } + return { target: target as NativeTriggerTarget, delayMs: seconds * 1000 }; +} + +export function parseNativeTriggerOutage(value: string, options: Pick): number { + if (!options.nativeTrigger) throw new Error("--native-trigger-outage needs --native-trigger"); + if (options.storage !== "s3") throw new Error("--native-trigger-outage needs --storage s3"); + if (options.storageFaults?.length) throw new Error("--native-trigger-outage starts its own ST1 window; drop --storage-faults"); + const seconds = /^\d+$/.test(value) ? Number(value) : NaN; + if (!(seconds >= 10 && seconds <= 900)) throw new Error("--native-trigger-outage takes 10 to 900 seconds"); + return seconds * 1000; +} + +export function triggeredOutageWindow(startMs: number, durationMs: number): StorageFaultWindow { + return { kind: "ST1", startMs, durationMs, rules: storageFaultRules("ST1", () => 0) }; +} + +export interface ArmedNativeFault { taskId: string; kind: FaultKind; holdMs: number; armedAt: number; restartSteps?: RestartStep[] } + +export interface NativeTriggerReceipt { + target: NativeTriggerTarget; delayMs: number; taskId: string; kind: FaultKind; armedAt: number; + seenAt: number | null; runId: string | null; eventType: string | null; + firedAt: number | null; missedAt: number | null; missedReason?: string; faultId?: string; +} + +export function runningRunIds(rows: readonly unknown[] | null): string[] { + return (rows ?? []).flatMap(row => { + const run = row as { runId?: unknown; id?: unknown; status?: unknown } | null; + const id = run?.runId ?? run?.id; + return typeof id === "string" && run?.status === "running" ? [id] : []; + }); +} + +export interface NativeTriggerEvent { seq: number; eventType: string; createdAt: string } + +export interface NativeTriggerHooks { + canFire(): boolean; + isOpen(taskId: string): boolean; + issueId(taskId: string): string | null; + runs(issueId: string): Promise; + events(runId: string, afterSeq: number, limit: number): Promise; + fire(armed: ArmedNativeFault): Promise; + record(kind: string, data: Record): void; +} + +export class NativeTrigger { + readonly receipts: NativeTriggerReceipt[] = []; + private readonly armed: Array<{ fault: ArmedNativeFault; receipt: NativeTriggerReceipt }> = []; + private readonly cursors = new Map(); + private firing = false; + + constructor(readonly spec: NativeTriggerSpec, private readonly hooks: NativeTriggerHooks) {} + + get busy(): boolean { return this.firing || this.armed.some(entry => entry.receipt.seenAt !== null); } + + arm(fault: ArmedNativeFault): void { + const receipt: NativeTriggerReceipt = { target: this.spec.target, delayMs: this.spec.delayMs, taskId: fault.taskId, + kind: fault.kind, armedAt: fault.armedAt, seenAt: null, runId: null, eventType: null, firedAt: null, missedAt: null }; + this.receipts.push(receipt); + this.armed.push({ fault, receipt }); + this.hooks.record("native_trigger_armed", { taskId: fault.taskId, fault: fault.kind, target: this.spec.target, delayMs: this.spec.delayMs }); + } + + async tick(now = Date.now()): Promise { + if (this.firing) return; + for (const entry of [...this.armed]) { + if (!this.hooks.isOpen(entry.fault.taskId)) { + this.armed.splice(this.armed.indexOf(entry), 1); + this.miss(entry.receipt, now, entry.receipt.seenAt === null ? "task_rested" : "task_rested_after_event"); + continue; + } + if (entry.receipt.seenAt === null) await this.look(entry.receipt); + const { seenAt } = entry.receipt; + if (seenAt === null || now < seenAt + this.spec.delayMs || !this.hooks.canFire()) continue; + this.armed.splice(this.armed.indexOf(entry), 1); + entry.receipt.firedAt = now; + this.firing = true; + this.hooks.record("native_trigger_fired", { taskId: entry.fault.taskId, fault: entry.fault.kind, target: this.spec.target, + runId: entry.receipt.runId, eventType: entry.receipt.eventType, afterEventMs: now - seenAt }); + void this.hooks.fire(entry.fault).catch(error => { + this.hooks.record("native_trigger_fire_failed", { taskId: entry.fault.taskId, error: String(error) }); + }).finally(() => { this.firing = false; }); + return; + } + } + + private async look(receipt: NativeTriggerReceipt): Promise { + const issueId = this.hooks.issueId(receipt.taskId); + if (!issueId) return; + const runs = await this.hooks.runs(issueId).catch(() => null); + for (const runId of runningRunIds(runs)) { + let afterSeq = this.cursors.get(runId) ?? 0; + for (let page = 0; page < EVENT_PAGES_PER_LOOK; page += 1) { + const events = await this.hooks.events(runId, afterSeq, EVENT_PAGE).catch(() => null); + if (!events?.length) break; + for (const event of events) { + afterSeq = Math.max(afterSeq, event.seq); + const at = Date.parse(event.createdAt); + if (receipt.seenAt === null && Number.isFinite(at) && at >= receipt.armedAt && nativeTriggerMatches(this.spec.target, event.eventType)) { + receipt.seenAt = at; + receipt.runId = runId; + receipt.eventType = event.eventType; + } + } + if (events.length < EVENT_PAGE) break; + } + this.cursors.set(runId, afterSeq); + if (receipt.seenAt !== null) return; + } + } + + private miss(receipt: NativeTriggerReceipt, at: number, reason: string) { + receipt.missedAt = at; + receipt.missedReason = reason; + this.hooks.record("native_trigger_missed", { taskId: receipt.taskId, fault: receipt.kind, target: this.spec.target, reason }); + } + + finish(faults: readonly FaultReceipt[], now = Date.now()): NativeTriggerReceipt[] { + for (const { receipt } of this.armed.splice(0)) { + this.miss(receipt, now, receipt.seenAt !== null ? "task_rested_after_event" + : this.hooks.isOpen(receipt.taskId) ? "no_matching_event" : "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 nativeTriggerShortfall(receipts: readonly NativeTriggerReceipt[]): string | null { + const fired = receipts.filter(receipt => receipt.firedAt !== null); + if (!fired.length) { + const seen = receipts.filter(receipt => receipt.seenAt !== null).length; + return `--native-trigger: no restart followed a matching run event (${receipts.length} armed, ${seen} saw the event, none fired)`; + } + const unproven = fired.filter(receipt => !receipt.faultId); + return unproven.length ? `--native-trigger: ${unproven.length} fired trigger(s) produced no fault: ${unproven.map(receipt => receipt.taskId).join(", ")}` : null; +} + +export function nativeTriggerMarkdown(receipts: readonly NativeTriggerReceipt[]): string { + const first = receipts[0]; + const lines = [`## Native trigger (${first ? `${first.target}+${first.delayMs / 1000}s` : "none"})`, "", + `${receipts.filter(receipt => receipt.faultId).length} of ${receipts.length} armed faults fired after a matching run event.`, "", + "| Task | Fault | Fault id | Run | Event | Fired after event |", "| --- | --- | --- | --- | --- | --- |", + ...receipts.map(receipt => `| ${receipt.taskId} | ${receipt.kind} | ${receipt.faultId ?? "-"} | ${receipt.runId?.slice(0, 8) ?? "-"} | ${ + receipt.eventType ?? `not reached (${receipt.missedReason ?? "unknown"})`} | ${receipt.firedAt !== null && receipt.seenAt !== null + ? `${((receipt.firedAt - receipt.seenAt) / 1000).toFixed(1)} s` : "-"} |`)]; + return `${lines.join("\n")}\n`; +} diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index aeea6588252..4d63f6c09b6 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -100,6 +100,7 @@ export interface WaveOptions { storageFaultMs?: [number, number]; storageWindows?: import("./storage-chaos.js").StorageFaultWindow[]; storageTrigger?: import("./storage-trigger.js").StorageTriggerTarget; + nativeTrigger?: import("./native-trigger.js").NativeTriggerSpec; workspaceCommit?: string; workspaceGitUrl?: string; /** Provider message faults from `--faults` (PV1 delay, PV2 drop, PV3 duplicate); see provider-faults.ts. */