diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 59dd2eef70c..33f9d27ef02 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -608,6 +608,87 @@ flag is `adapter_failed`, including three tasks that passed after a retry, which warning until `adapter_failed` names its cause. The audit saw only failure codes; the error-text part of U3 is checked only in waves run with this harness. +## Time faults (TM1–TM3, `--time-age-hours`) + +```sh +# Freeze, sleep and aged outage across running tasks. +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex --workloads A1,C1 --faults TM1,TM2,TM3 --concurrency 2 --tasks 3 --seed 301 +# The aged wait: an approval left pending for a day across a restart. +node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local \ + --adapters codex --workloads G1 --faults TM3 --triggers approval --concurrency 2 --tasks 2 --seed 302 +``` + +The fourth fault family. Every other wave is minutes long on a fresh instance, so nothing time-bound +expires. A time fault makes deadlines lapse during a run or across a restart. No product code changes. + +| Fault | What the harness does | Deadlines it crosses | +| --- | --- | --- | +| TM1 freeze | `SIGSTOP` the server process alone for 95–150 s, then `SIGCONT`. The runner, its provider and Postgres keep running. | Controller-instance lease (90 s), execution-control deadline (60 s), legacy run lease (60 s). The runner keeps calling a server that does not answer. | +| TM2 sleep | `SIGSTOP` the server and every process under it for 20–45 s. While they are stopped, shift every database timestamp back by the age. Then `SIGCONT` the same processes. | Everything stored in the database: native and remote execution leases, scheduled retries, liveness and silence watchdogs, accounting closure, and pending waits. The processes' own monotonic timers barely moved. This stands in for a laptop that slept. | +| TM3 aged outage | `SIGKILL` the server tree (as K1). While it is down, shift every database timestamp back by the age. Then start a replacement. | The same deadlines, reconciled by a fresh boot that finds runs, leases, wakes and waits that look hours old. | + +The age is 26 hours by default; `--time-age-hours` sets 1–72. The pause and the age are the cheapest +faithful stand-ins: +- A real outage of a day would hold the slot for a day. +- The relevant deadlines are fixed constants, not environment settings. + +**Aging.** One transaction moves every time in the `public` schema back by the same interval, so the +database is what it would be had everything happened the age earlier. Future deadlines move into the +past. The shift covers: +- every `timestamp` column; +- every time inside a `jsonb` column: ISO datetime strings keep their format, and whole numbers in the + wave's epoch-millisecond window move too. + +The JSON pass matters. Receipts and snapshots keep copies of times (a continuation receipt stores the +comment's `updatedAt` and compares it later). Shifting only the columns would fail those comparisons, a +harness artifact rather than a product failure. Triggers are turned off for the shift (replica role) so +they do not rewrite derived columns. For the time-bound tables (runs, native finalizations, controller +instances, wakes, environment and workspace leases, workspace operations, approvals, decisions, +interactions, resume claims, routine triggers and recovery actions), the identity, state and +timestamp columns of each row are written before the shift to `/time-faults/.json`. +The per-table counts are in the fault receipt (`timeFault.aging` in `faults.json`) and in the +timeline (`time_fault_aged`). + +If a frozen server's open transaction holds a lock the shift needs, the shift times out after 3 s. +The tree then runs for 2 s, is frozen again, and the shift retries; `lockRetries` counts these. An +age must exceed the wave's elapsed time, so shifted and unshifted values never overlap. A fault that +would break this fails as a harness error. + +**Grading.** The product sees only the aged values. The observer and grader read through +`TimeShiftedApi`, which adds the age back to the ISO times and epoch milliseconds that a shift moved. +R4 overlaps, R5 evidence, U1–U4 and the speed budgets are therefore judged in real time. Evidence +bundles show the corrected times too. + +The age is longer than the wave has run, so a moved value lands before the wave started, and nothing +the wave wrote is earlier than that. One exception stays aged: a deadline that was more than (age +minus elapsed wave time) in the future when the shift ran. + +Each later shift also widens the epoch window by the earlier ages, so a value shifted once is shifted +again. + +A freeze is a fault window like an outage, from `SIGSTOP` to ready. If a server exits by itself after +`SIGCONT` (for example a lost-lease self-fence), the harness records `serverExitedAfterThaw`, cleans +up the exited server's tree, starts a replacement and grades the tasks. If the cleanup cannot prove +the old tree stopped, the harness records `cleanupError`, starts no replacement, and the fault fails as a harness error. + +**Pass rule.** Each task is graded on R1–R6 and its T-oracle as usual: +- the task reaches its correct terminal state, or a diagnosable rest; +- no run executed twice (R4); +- nothing is stuck (R5, R6); +- no Retry for something the system should recover by itself (R3). + +For the aged wait (G1 with `--triggers approval`), the approval must still be pending and accepted +when the board decides after a day, and the approved subtask must exist exactly once. + +**Limits.** +- Local only. On Daytona the remote runner is not under the server, so it cannot be frozen. +- Agent JWT expiry is not covered. Native runs mint a token per tool call, so a short + `PAPERCLIP_AGENT_JWT_TTL_SECONDS` tests nothing there. +- The fixed 1-hour runtime-tools token is not covered either. +- Times kept outside Postgres are not aged: in-memory timers and the runner's own state files. + Neither are times inside `text` columns or HMACs computed over a time. + ## Fault timing (`--triggers`) Each task samples one trigger with its cell. `time` (default) fires at the @@ -650,6 +731,7 @@ re-arms the task's own triggered faults instead of using their offsets. | `daytona-image.ts` | Docker-free extraction of the image's runnerd and provider pack, cached per digest, and the controller/image content-ID check. | | `provider-faults.ts`, `provider-shim/` | PV1–PV3: the resolve hook, the Codex stdio shim, per-task rules and the `provider_fault_applied` check. | | `question-handoff.ts` | Q1 saved question: the board's answer step and its oracle. | +| `time-faults.ts`, `time-faults-config.ts` | TM1–TM3: process freezing, database aging with recorded rows, and the time-corrected API the grader reads through. | | `daytona-sweep.ts` | Daytona R6: read-only reconciliation of this wave's labelled sandboxes with leases; cleanup only through Paperclip. | ## Output (gitignored) diff --git a/tests/restart-chaos/config.ts b/tests/restart-chaos/config.ts index 5ede7a843a8..d2dee675226 100644 --- a/tests/restart-chaos/config.ts +++ b/tests/restart-chaos/config.ts @@ -4,6 +4,7 @@ import { parseArgs } from "node:util"; import { applyJourneyOptions } from "./journey-matrix.js"; import { takesJourneyTriggers } from "./journeys.js"; import { STORAGE_FAULT_KINDS } from "./storage-chaos.js"; +import { parseTimeFaultOptions } from "./time-faults-config.js"; import { PROVIDER_FAULT_KINDS, PROVIDER_FAULT_WORKLOADS, isProviderFault, parseProviderDelays, parseProviderMessage, type ProviderFaultKind } from "./provider-faults.js"; import { ADAPTER_IDS, ENVIRONMENT_KINDS, FAULT_KINDS, TRIGGER_IDS, WORKLOAD_IDS, @@ -111,13 +112,14 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC "db-pressure": { type: "string" }, matrix: { type: "boolean" }, "server-from": { type: "string" }, + "time-age-hours": { 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", "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"] 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", "fault-after-seconds", "provider-fault-message", "provider-fault-delays", "db-pressure", "matrix", "server-from", "time-age-hours"] 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, @@ -207,6 +209,7 @@ export function parseCommand(argv: readonly string[], now = Date.now()): ParsedC } else if (values["provider-fault-message"] !== undefined || values["provider-fault-delays"] !== undefined) { throw new Error("--provider-fault-message and --provider-fault-delays need a provider fault (PV1, PV2 or PV3) in --faults"); } + Object.assign(options, parseTimeFaultOptions(options, values["time-age-hours"])); const upgradeFrom = values["upgrade-from"]; if (options.faults.includes("K14") !== (upgradeFrom !== undefined)) { throw new Error("K14 (upgrade mid-run) and --upgrade-from go together: pass both or neither"); diff --git a/tests/restart-chaos/journey-matrix.ts b/tests/restart-chaos/journey-matrix.ts index 7ea15b1958a..88b26271794 100644 --- a/tests/restart-chaos/journey-matrix.ts +++ b/tests/restart-chaos/journey-matrix.ts @@ -13,6 +13,7 @@ import path from "node:path"; import { WORKLOAD_TRIGGERS } from "./journeys.js"; import { withProviderFaultRules } from "./provider-faults.js"; import { createRng, k4HoldRange, TRIGGER_DELAY_RANGE_MS } from "./scheduler.js"; +import { isTimeFault, timeFaultHoldMs } from "./time-faults-config.js"; import { workloadDefinitions } from "./workloads.js"; import type { AdapterId, TaskFaultKind, TaskPlan, TriggerId, WaveOptions, WavePlan, WorkloadId } from "./types.js"; @@ -55,7 +56,7 @@ export function planMatrixTasks(options: Pick { timeline.add("provider_faults_installed", { realCodex: realCodex.command, codexVersion: realCodex.version, rules: plan.tasks.flatMap(task => task.providerFault ? [`${task.taskId}:${describeProviderFault(task.providerFault)}`] : []) }); } + if (plan.options.timeAgeMs) supervisor.timeAgeMs = plan.options.timeAgeMs; await supervisor.start(serverEnvironment); - const api = new ChaosApi(supervisor.baseUrl); + // TM2/TM3 age the database; the observer and grader read run times back at their real times. + const api: ChaosApi = new TimeShiftedApi(supervisor.baseUrl, supervisor.agingShifts); const lab = await seedChaosLab({ api, env: plan.options.env, waitForReadyAgents: true, cancelled: () => interrupted, daytonaImage: plan.options.daytonaImage, daytonaCpu: plan.options.daytonaCpu, daytonaReuseLease: plan.options.daytonaReuseLease, adapters: [...new Set([...plan.tasks.map(task => task.adapter), diff --git a/tests/restart-chaos/scheduler.ts b/tests/restart-chaos/scheduler.ts index c3383562fb9..3f33112c573 100644 --- a/tests/restart-chaos/scheduler.ts +++ b/tests/restart-chaos/scheduler.ts @@ -1,6 +1,7 @@ import { isProviderFault, withProviderFaultRules } from "./provider-faults.js"; import { takesJourneyTriggers } from "./journeys.js"; import { workloadDefinitions } from "./workloads.js"; +import { isTimeFault, timeFaultHoldMs } from "./time-faults-config.js"; import type { AdapterId, FaultKind, ScheduledFault, StormFault, StormPlan, TaskFaultKind, TaskPlan, TriggerId, WavePlan, WaveOptions, WorkloadId } from "./types.js"; import type { ReplayRef } from "./config.js"; @@ -121,7 +122,7 @@ export function planTasks(options: Pick { const READY_TIMEOUT_MS = 300_000; const STOP_TIMEOUT_MS = 30_000; -export const FAULT_SIGNAL: Record = { K1: "SIGKILL", K2: "SIGTERM", K4: "SIGKILL", K6: "SIGKILL", K14: "SIGTERM" }; +export const FAULT_SIGNAL: Record = { K1: "SIGKILL", K2: "SIGTERM", K4: "SIGKILL", K6: "SIGKILL", K14: "SIGTERM", + TM1: "SIGSTOP", TM2: "SIGSTOP", TM3: "SIGKILL" }; /** The checkout's commit, for the K14 receipt. */ function checkoutSha(root: string): string | null { @@ -122,6 +124,11 @@ export class ChaosSupervisor { onCrashPoint: ((receipt: FaultReceipt) => void) | null = null; /** Crash-point suite: keep the control plane down this long after a self-kill (a K4-style hold). */ crashPointHoldMs = 0; + /** TM2/TM3: how far each aging fault shifts the wave's database back. */ + timeAgeMs = DEFAULT_TIME_AGE_MS; + /** Every aging shift so far, in order; the grader's API reads times through them (time-faults.ts). */ + readonly agingShifts: AgingShift[] = []; + private readonly createdAt = Date.now(); /** The checkout the server runs from: the older build (`--upgrade-from`) until the first K14, then this one. */ private serverRoot: string; /** The environment overrides the wave passed to `start`, kept for every restart and for the K14 switch. */ @@ -407,10 +414,11 @@ export class ChaosSupervisor { try { if (!previous?.pid || this.exited(previous)) throw new Error("No live Paperclip server to fault"); let outage: ControllerOutageReceipt | undefined; - outage = await replaceController({ + if (kind === "TM1" || kind === "TM2") await this.freeze(kind, previous, holdMs, receipt); + else outage = await replaceController({ requestId: `chaos-${receipt.faultId}-${requestedBy}`, previousPid: previous.pid, - outage: { signal: receipt.signal, holdMs }, + outage: { signal: receipt.signal === "SIGTERM" ? "SIGTERM" : "SIGKILL", holdMs }, // A K4 hold may outlast the Runner's 7-minute reconnect grace on // purpose (--k4-hold-max-seconds); the replacement still gets the // normal startup allowance after the hold. @@ -423,6 +431,12 @@ export class ChaosSupervisor { waitForUnhealthy: async () => { await this.waitForHealthToStop(); receipt.stoppedAt = Date.now(); this.timeline.add("fault_server_down", { faultId: receipt.faultId }); }, // K1/K2/K4 keep the controller runtime cache; K11 will discard it. discardCache: async () => {}, + // TM3: the server is down; every database timestamp moves back by the age before the replacement starts. + ...(kind === "TM3" ? { afterStopped: async () => { + const timeFault: TimeFaultRecord = { kind }; + receipt.timeFault = timeFault; + timeFault.aging = await this.age(receipt.faultId); + } } : {}), start: async () => { receipt.restartedAt = Date.now(); // K14: the replacement is the current build, on the same database and instance directory. @@ -452,6 +466,44 @@ export class ChaosSupervisor { return receipt; } + private async age(faultId: string, onLockTimeout?: () => Promise) { + const aging = await ageWaveDatabase({ connectionString: databaseUrlFromConfig(this.paths.configPath), faultId, ageMs: this.timeAgeMs, + evidenceDir: path.join(path.dirname(this.paths.serverLogPath), "time-faults"), waveStartedAt: this.createdAt, + priorAgeMs: this.agingShifts.reduce((sum, shift) => sum + shift.ageMs, 0), + ...(onLockTimeout ? { onLockTimeout } : {}) }); + this.agingShifts.push({ faultId, ageMs: aging.ageMs, appliedAt: aging.appliedAt, waveStartedAt: aging.waveStartedAt }); + this.timeline.add("time_fault_aged", { faultId, ageMs: aging.ageMs, totalRows: aging.totalRows, triggersDisabled: aging.triggersDisabled, + lockRetries: aging.lockRetries, tables: aging.tables.map(table => `${table.table}:${table.rows}`), evidence: aging.evidencePath }); + return aging; + } + + /** + * TM1 freezes the server process alone; TM2 freezes it with every process under it and ages the database + * while they are stopped. Both continue the same processes after the hold: no restart. A server that exits + * on its own after it is continued (a lost-lease self-fence) is recorded and replaced, and the tasks are graded. + */ + private async freeze(kind: "TM1" | "TM2", server: ChildProcess, holdMs: number, receipt: FaultReceipt): Promise { + const timeFault: TimeFaultRecord = { kind }; + receipt.timeFault = timeFault; + const times = await runFreezeFault({ kind, rootPid: server.pid!, holdMs, record: timeFault, freezer: new ProcessFreezer(), + ...(kind === "TM2" ? { age: (onLockTimeout: () => Promise) => this.age(receipt.faultId, onLockTimeout) } : {}), + cancelled: () => this.shuttingDown, + onFrozen: pids => this.timeline.add("fault_server_frozen", { faultId: receipt.faultId, pids, holdMs }), + onThawed: frozenMs => this.timeline.add("fault_server_thawed", { faultId: receipt.faultId, frozenMs }), + waitHealthy: () => this.waitForHealth(server, receipt.phases), + exitStatus: () => this.exited(server) ? { code: server.exitCode, signal: server.signalCode } : null, + replaceExited: async () => { + this.timeline.add("time_fault_server_exited_after_thaw", { faultId: receipt.faultId, ...timeFault.serverExitedAfterThaw }); + // Kills what is left of the exited server's tree; rejects when it cannot prove the tree stopped. + try { await this.stopper.crashForRestart(server); } finally { if (this.child === server) this.child = null; } + }, + startReplacement: async () => this.waitForHealth(await this.spawnServer(), receipt.phases), + }); + receipt.stoppedAt = times.stoppedAt; + receipt.restartedAt = times.restartedAt; + receipt.healthyAt = times.healthyAt; + } + /** * K14: move the server from the older build to this checkout. Only the first K14 switches; a later * one restarts the current build (recorded with `switched: false`). diff --git a/tests/restart-chaos/time-faults-config.ts b/tests/restart-chaos/time-faults-config.ts new file mode 100644 index 00000000000..71be84524af --- /dev/null +++ b/tests/restart-chaos/time-faults-config.ts @@ -0,0 +1,46 @@ +/** TM planning and `--time-age-hours` validation, kept apart from time-faults.ts so the parser and scheduler load no database code. */ +import type { WaveOptions } from "./types.js"; + +export const DEFAULT_TIME_AGE_HOURS = 26; +export const TIME_FAULT_KINDS = ["TM1", "TM2", "TM3"] as const; +export type TimeFaultKind = (typeof TIME_FAULT_KINDS)[number]; + +export function isTimeFault(kind: string): kind is TimeFaultKind { + return (TIME_FAULT_KINDS as readonly string[]).includes(kind); +} + +/** How long each freeze lasts. TM1 outlasts the 90 s controller lease; TM2's time passes through aging. */ +export const TIME_FREEZE_HOLD_RANGE_MS: Record<"TM1" | "TM2", readonly [number, number]> = { + TM1: [95_000, 150_000], + TM2: [20_000, 45_000], +}; + +/** Default age: past a day, so hourly, daily and 1 h deadlines all lapse. `--time-age-hours` sets it (1–72). */ +export const DEFAULT_TIME_AGE_MS = DEFAULT_TIME_AGE_HOURS * 3_600_000; + +/** Seeded freeze length for a TM task; 0 for TM3 (its time passes through aging). Draws only for TM1/TM2. */ +export function timeFaultHoldMs(kind: TimeFaultKind, int: (min: number, max: number) => number): number { + return kind === "TM3" ? 0 : int(...TIME_FREEZE_HOLD_RANGE_MS[kind]); +} + +const MIN_HOURS = 1; +const MAX_HOURS = 72; + +export function parseTimeFaultOptions(options: Pick, + rawAgeHours: string | undefined): Pick { + const time = options.faults.filter(isTimeFault); + const ages = time.some(kind => kind === "TM2" || kind === "TM3"); + if (rawAgeHours !== undefined && !ages) throw new Error("--time-age-hours sets the TM2/TM3 age; add TM2 or TM3 to --faults"); + if (!time.length) return {}; + // A freeze sends SIGSTOP to processes under the local server; a remote runner is not one of them. + if (options.env !== "local") throw new Error("Time faults (TM*) freeze and age a local instance; run them with --env local"); + if (options.stormMinutes) throw new Error("Time faults (TM*) are planned per task; do not combine them with --storm-minutes"); + if (options.crashPoint) throw new Error("Time faults (TM*) do not combine with --crash-point"); + if (options.faults.includes("K14")) throw new Error("Time faults (TM*) do not combine with K14; age the upgraded build in its own wave"); + if (!ages) return {}; + if (rawAgeHours === undefined) return { timeAgeMs: DEFAULT_TIME_AGE_HOURS * 3_600_000 }; + if (!/^[0-9]+$/.test(rawAgeHours)) throw new Error("--time-age-hours must be a whole number of hours"); + const hours = Number(rawAgeHours); + if (hours < MIN_HOURS || hours > MAX_HOURS) throw new Error(`--time-age-hours must be between ${MIN_HOURS} and ${MAX_HOURS}`); + return { timeAgeMs: hours * 3_600_000 }; +} diff --git a/tests/restart-chaos/time-faults-db.test.ts b/tests/restart-chaos/time-faults-db.test.ts new file mode 100644 index 00000000000..f64ade2ba29 --- /dev/null +++ b/tests/restart-chaos/time-faults-db.test.ts @@ -0,0 +1,99 @@ +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createDb, type Db } from "../../packages/db/src/index.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../packages/db/src/test-embedded-postgres.js"; +import { ageWaveDatabase, unshiftJson } from "./time-faults.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const HOUR = 3_600_000; +const NOON = Date.parse("2026-11-09T12:00:00.123Z"); +const ms = (value: unknown) => new Date(value as string).getTime(); + +describe.skipIf(!support.supported)("aging a wave database", () => { + let temporary: Awaited>; + let db: Db; + let evidenceDir: string; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("chaos-time-"); + db = createDb(temporary.connectionString, { maxConnections: 1 }); + evidenceDir = await mkdtemp(path.join(os.tmpdir(), "chaos-time-evidence-")); + await db.$client.unsafe(`create table chaos_time_comments (id text primary key, updated_at timestamptz not null, body text not null)`); + await db.$client.unsafe(`create table chaos_time_runs (id text primary key, status text, lease_expires_at timestamptz, finished_at timestamptz, context jsonb)`); + await db.$client.unsafe(`create table chaos_time_touch (id text primary key, updated_at timestamptz)`); + // A trigger that would rewrite updated_at on every update: the shift must not fire it. + await db.$client.unsafe(`create function chaos_time_touch_now() returns trigger language plpgsql as $$ + begin new.updated_at := now(); return new; end $$`); + await db.$client.unsafe(`create trigger chaos_time_touch_trigger before update on chaos_time_touch for each row execute function chaos_time_touch_now()`); + await db.$client.unsafe(`insert into chaos_time_comments values ('c1', $1, '2026-11-09T12:00:00Z stays text')`, [new Date(NOON).toISOString()]); + await db.$client.unsafe(`insert into chaos_time_touch values ('t1', $1)`, [new Date(NOON).toISOString()]); + await db.$client.unsafe(`insert into chaos_time_runs values ('r1', 'running', $1, null, $2::jsonb)`, [ + new Date(NOON + 20 * 60_000).toISOString(), + JSON.stringify({ commentUpdatedAt: new Date(NOON).toISOString(), pgStyle: "2026-11-09 12:00:00.5+00", plain: "2026-11-09T12:00:00Z", + wallMs: NOON, count: 42, version: 20_261_109, note: "not 2026-11-09T12:00:00Z", nested: [{ at: "2026-11-09T12:00:00.123456+05:30" }] }), + ]); + }, 120_000); + + afterAll(async () => { + await db?.$client.end(); + await temporary?.cleanup(); + if (evidenceDir) await rm(evidenceDir, { recursive: true, force: true }); + }); + + it("moves every time back by the age, including copies inside JSON, and records what it moved", async () => { + const age = 26 * HOUR; + const result = await ageWaveDatabase({ connectionString: temporary.connectionString, faultId: "f01", ageMs: age, evidenceDir, + waveStartedAt: NOON - HOUR, now: () => NOON + 30 * 60_000 }); + expect(result.triggersDisabled).toBe(true); + expect(result.tables.map(table => table.table)).toEqual(expect.arrayContaining(["chaos_time_comments", "chaos_time_runs", "chaos_time_touch"])); + + const [comment] = await db.$client.unsafe(`select updated_at, body from chaos_time_comments`) as unknown as Array<{ updated_at: unknown; body: string }>; + const [run] = await db.$client.unsafe(`select lease_expires_at, finished_at, context from chaos_time_runs`) as unknown as Array<{ + lease_expires_at: unknown; finished_at: unknown; context: Record }>; + const [touched] = await db.$client.unsafe(`select updated_at from chaos_time_touch`) as unknown as Array<{ updated_at: unknown }>; + expect(ms(comment!.updated_at)).toBe(NOON - age); + expect(comment!.body).toBe("2026-11-09T12:00:00Z stays text"); + expect(ms(touched!.updated_at)).toBe(NOON - age); + // A deadline 20 minutes ahead is now long past. + expect(ms(run!.lease_expires_at)).toBe(NOON + 20 * 60_000 - age); + expect(run!.finished_at).toBeNull(); + // A copy of a time still equals the column it was copied from (continuation receipts compare them). + expect(run!.context.commentUpdatedAt).toBe(new Date(ms(comment!.updated_at)).toISOString()); + expect(run!.context).toMatchObject({ + pgStyle: "2026-11-08 10:00:00.5+00", plain: "2026-11-08T10:00:00Z", wallMs: NOON - age, count: 42, version: 20_261_109, + note: "not 2026-11-09T12:00:00Z", nested: [{ at: "2026-11-08T10:00:00.123456+05:30" }], + }); + + const evidence = JSON.parse(await readFile(result.evidencePath, "utf8")) as { ageMs: number; tables: Array<{ table: string; rows: number }> }; + expect(evidence.ageMs).toBe(age); + expect(evidence.tables).toEqual(expect.arrayContaining([expect.objectContaining({ table: "chaos_time_runs", rows: 1 })])); + + // The grader reads the moved times back at their real time. + const shift = { faultId: "f01", ageMs: age, appliedAt: NOON + 30 * 60_000, waveStartedAt: NOON - HOUR }; + expect(unshiftJson({ at: new Date(ms(comment!.updated_at)).toISOString() }, [shift])).toEqual({ at: new Date(NOON).toISOString() }); + }); + + it("ages the same rows a second time and keeps every copy equal to its column", async () => { + const age = 26 * HOUR; + const second = await ageWaveDatabase({ connectionString: temporary.connectionString, faultId: "f02", ageMs: age, evidenceDir, + waveStartedAt: NOON - HOUR, priorAgeMs: age, now: () => NOON + 40 * 60_000 }); + expect(second.waveStartedAt).toBe(NOON - HOUR); + const [comment] = await db.$client.unsafe(`select updated_at from chaos_time_comments`) as unknown as Array<{ updated_at: unknown }>; + const [run] = await db.$client.unsafe(`select context from chaos_time_runs`) as unknown as Array<{ context: Record }>; + expect(ms(comment!.updated_at)).toBe(NOON - 2 * age); + expect(run!.context.commentUpdatedAt).toBe(new Date(NOON - 2 * age).toISOString()); + // The epoch copy was below the first window after one shift; the second window still reaches it. + expect(run!.context.wallMs).toBe(NOON - 2 * age); + expect(run!.context.count).toBe(42); + const shifts = [{ faultId: "f01", ageMs: age, appliedAt: NOON + 30 * 60_000, waveStartedAt: NOON - HOUR }, + { faultId: "f02", ageMs: age, appliedAt: NOON + 40 * 60_000, waveStartedAt: NOON - HOUR }]; + expect(unshiftJson(run!.context, shifts)).toMatchObject({ commentUpdatedAt: new Date(NOON).toISOString(), wallMs: NOON }); + }); + + it("refuses an age the wave has nearly outlived", async () => { + await expect(ageWaveDatabase({ connectionString: temporary.connectionString, faultId: "f02", ageMs: HOUR, evidenceDir, + waveStartedAt: NOON - HOUR, now: () => NOON })).rejects.toThrow(/too close to the age/); + }); +}); diff --git a/tests/restart-chaos/time-faults.test.ts b/tests/restart-chaos/time-faults.test.ts new file mode 100644 index 00000000000..cdc891a708d --- /dev/null +++ b/tests/restart-chaos/time-faults.test.ts @@ -0,0 +1,221 @@ +import { describe, expect, it } from "vitest"; +import { parseCommand } from "./config.js"; +import { planTasks } from "./scheduler.js"; +import { planMatrixWave } from "./journey-matrix.js"; +import { TIME_FREEZE_HOLD_RANGE_MS, isTimeFault } from "./time-faults-config.js"; +import { ProcessFreezer, agingStatements, descendantPids, parseProcessTable, runFreezeFault, unshiftJson, unshiftMs, type AgingResult, type TimeFaultRecord } from "./time-faults.js"; + +const HOUR = 3_600_000; + +describe("time fault options", () => { + it("accepts TM faults locally with a default 26 h age", () => { + const parsed = parseCommand(["--faults", "TM1,TM2,TM3", "--adapters", "codex", "--workloads", "A1", "--seed", "1"]); + if (parsed.mode !== "wave") throw new Error("expected a wave"); + expect(parsed.options.faults).toEqual(["TM1", "TM2", "TM3"]); + expect(parsed.options.timeAgeMs).toBe(26 * HOUR); + }); + + it("takes --time-age-hours only with an aging fault", () => { + const parsed = parseCommand(["--faults", "TM3", "--time-age-hours", "6", "--workloads", "A1", "--seed", "1"]); + expect(parsed.mode === "wave" && parsed.options.timeAgeMs).toBe(6 * HOUR); + expect(() => parseCommand(["--faults", "TM1", "--time-age-hours", "6", "--workloads", "A1"])).toThrow(/add TM2 or TM3/); + expect(() => parseCommand(["--faults", "K1", "--time-age-hours", "6", "--workloads", "A1"])).toThrow(/add TM2 or TM3/); + expect(() => parseCommand(["--faults", "TM3", "--time-age-hours", "0", "--workloads", "A1"])).toThrow(/between 1 and 72/); + expect(() => parseCommand(["--faults", "TM3", "--time-age-hours", "100", "--workloads", "A1"])).toThrow(/between 1 and 72/); + }); + + it("refuses combinations a freeze or an age cannot serve", () => { + expect(() => parseCommand(["--faults", "TM1", "--storm-minutes", "10"])).toThrow(/storm/); + expect(() => parseCommand(["--faults", "TM2", "--env", "daytona", "--daytona-image", `r/i@sha256:${"a".repeat(64)}`])).toThrow(/--env local/); + expect(() => parseCommand(["--faults", "TM3,K14", "--upgrade-from", "/tmp/x"])).toThrow(/K14/); + }); + + it("gives TM1 and TM2 a seeded freeze and TM3 none", () => { + const plan = planTasks({ seed: 7, tasks: 9, workloads: ["A1"], adapters: ["codex"], faults: ["TM1", "TM2", "TM3"] }); + for (const task of plan) { + if (task.fault === "TM3") expect(task.holdMs).toBe(0); + else { + const [min, max] = TIME_FREEZE_HOLD_RANGE_MS[task.fault as "TM1" | "TM2"]; + expect(task.holdMs).toBeGreaterThanOrEqual(min); + expect(task.holdMs).toBeLessThanOrEqual(max); + } + } + // TM1 outlasts the 90 s controller-instance lease. + expect(TIME_FREEZE_HOLD_RANGE_MS.TM1[0]).toBeGreaterThan(90_000); + }); + + it("gives matrix TM cells the same seeded freeze", () => { + const parsed = parseCommand(["--matrix", "--faults", "TM1,TM2,TM3", "--adapters", "codex", "--workloads", "A1", "--seed", "9"]); + if (parsed.mode !== "wave") throw new Error("expected a wave"); + const tasks = planMatrixWave("w-20261110T000000-s9", parsed.options).tasks.filter(task => isTimeFault(task.fault)); + expect(tasks.map(task => task.fault).sort()).toEqual(["TM1", "TM2", "TM3"]); + for (const task of tasks) { + if (task.fault === "TM3") expect(task.holdMs).toBe(0); + else expect(task.holdMs).toBeGreaterThanOrEqual(TIME_FREEZE_HOLD_RANGE_MS[task.fault as "TM1" | "TM2"][0]); + } + }); +}); + +describe("process freezing", () => { + const table = parseProcessTable([ + " 100 1 node server-entry.ts", + " 200 100 paperclip-runnerd --serve", + " 300 200 codex app-server", + " 400 100 /opt/pg/bin/postgres -D data", + " 500 400 postgres: checkpointer", + " 600 1 node other-session main.ts", + ].join("\n")); + + it("walks the server's tree, parents first, and never freezes Postgres", () => { + expect(descendantPids(100, table)).toEqual([200, 300]); + expect(descendantPids(600, table)).toEqual([]); + }); + + it("freezes the root first and continues everything it froze", () => { + const signals: string[] = []; + const freezer = new ProcessFreezer((pid, signal) => { signals.push(`${signal}:${pid}`); }, () => table); + expect(freezer.freeze(100, true)).toEqual([100, 200, 300]); + freezer.thaw(); + expect(signals).toEqual(["SIGSTOP:100", "SIGSTOP:200", "SIGSTOP:300", "SIGCONT:100", "SIGCONT:200", "SIGCONT:300"]); + expect(freezer.pids).toEqual([]); + }); + + it("TM1 freezes only the server", () => { + const signals: string[] = []; + const freezer = new ProcessFreezer((pid, signal) => { signals.push(`${signal}:${pid}`); }, () => table); + expect(freezer.freeze(100, false)).toEqual([100]); + freezer.thaw(); + expect(signals).toEqual(["SIGSTOP:100", "SIGCONT:100"]); + }); + + it("skips a process that exited before it could be frozen", () => { + const freezer = new ProcessFreezer(pid => { if (pid === 200) throw Object.assign(new Error("gone"), { code: "ESRCH" }); }, () => table); + expect(freezer.freeze(100, true)).toEqual([100, 300]); + }); +}); + +describe("aging", () => { + const window = { lo: 1_000, hi: 2_000 }; + + it("shifts every timestamp and JSON column of a table in one statement", () => { + const statements = agingStatements([ + { table: "heartbeat_runs", column: "created_at", kind: "timestamp" }, + { table: "heartbeat_runs", column: "context_snapshot", kind: "json" }, + { table: "agents", column: "updated_at", kind: "timestamp" }, + ], 2 * HOUR, window); + expect(statements.map(statement => statement.table)).toEqual(["agents", "heartbeat_runs"]); + expect(statements[1]!.sql).toBe("update public.\"heartbeat_runs\" set \"created_at\" = \"created_at\" - interval '7200000 milliseconds', " + + "\"context_snapshot\" = pg_temp.paperclip_shift_json(\"context_snapshot\", 7200000, 1000, 2000) " + + "where \"created_at\" is not null or \"context_snapshot\" is not null"); + }); + + it("takes whole seconds only, so fractional digits never change", () => { + expect(() => agingStatements([], 0, window)).toThrow(/positive/); + expect(() => agingStatements([], 1_500, window)).toThrow(/whole number of seconds/); + }); + + it("quotes identifiers", () => { + expect(agingStatements([{ table: "we\"ird", column: "at", kind: "timestamp" }], 1_000, window)[0]!.sql).toContain("public.\"we\"\"ird\""); + }); +}); + +describe("reading aged times at their real time", () => { + const applied = Date.parse("2026-11-09T12:00:00.000Z"); + const waveStartedAt = applied - 20 * 60_000; + const shift = { faultId: "f01", ageMs: 26 * HOUR, appliedAt: applied, waveStartedAt }; + + it("restores values the shift moved and leaves later writes alone", () => { + const writtenBefore = Date.parse("2026-11-09T11:55:00.000Z"); + expect(unshiftMs(writtenBefore - shift.ageMs, [shift])).toBe(writtenBefore); + const writtenAfter = applied + 30_000; + expect(unshiftMs(writtenAfter, [shift])).toBe(writtenAfter); + // A value written after the thaw from a time the frozen server kept in memory. + expect(unshiftMs(waveStartedAt + 60_000, [shift])).toBe(waveStartedAt + 60_000); + }); + + it("restores a moved deadline that was still in the future", () => { + const leaseExpiresAt = applied + 20 * 60_000; + expect(unshiftMs(leaseExpiresAt - shift.ageMs, [shift])).toBe(leaseExpiresAt); + }); + + it("undoes two shifts in order", () => { + const second = { faultId: "f02", ageMs: 26 * HOUR, appliedAt: applied + 10 * 60_000, waveStartedAt }; + const beforeBoth = applied - 60_000; + const between = applied + 5 * 60_000; + expect(unshiftMs(beforeBoth - shift.ageMs - second.ageMs, [shift, second])).toBe(beforeBoth); + expect(unshiftMs(between - second.ageMs, [shift, second])).toBe(between); + }); + + it("rewrites ISO strings and epoch milliseconds anywhere in a response and nothing else", () => { + const run = { id: "r1", status: "running", startedAt: new Date(applied - 60_000 - shift.ageMs).toISOString(), + wallStartedAtMs: applied - 60_000 - shift.ageMs, durationMs: 93_600_000, attempt: 2, + finishedAt: null, note: "2026-11-08", events: [{ createdAt: new Date(applied + 1_000).toISOString(), at: applied + 1_000 }] }; + expect(unshiftJson(run, [shift])).toEqual({ ...run, startedAt: new Date(applied - 60_000).toISOString(), wallStartedAtMs: applied - 60_000 }); + expect(unshiftJson(run, [])).toBe(run); + }); +}); + +describe("freeze sequence", () => { + const healthy = 5_000; + function harness(overrides: Partial[0]> = {}) { + const calls: string[] = []; + let clock = 1_000; + const record: TimeFaultRecord = { kind: "TM2" }; + const input: Parameters[0] = { + kind: "TM2", rootPid: 100, holdMs: 3_000, record, + freezer: { freeze: (pid, tree) => { calls.push(`freeze:${pid}:${tree}`); return [100, 200]; }, thaw: () => { calls.push("thaw"); } }, + cancelled: () => false, + waitHealthy: async () => { calls.push("wait"); return healthy; }, + exitStatus: () => null, + replaceExited: async () => { calls.push("cleanup"); }, + startReplacement: async () => { calls.push("replace"); return healthy + 1; }, + now: () => clock, sleep: async ms => { clock += ms; }, + ...overrides, + }; + return { input, calls, record }; + } + + it("freezes, holds, always thaws, then waits for the same server", async () => { + const { input, calls, record } = harness(); + await expect(runFreezeFault(input)).resolves.toEqual({ stoppedAt: 1_000, restartedAt: 4_000, healthyAt: healthy }); + expect(calls).toEqual(["freeze:100:true", "thaw", "wait"]); + expect(record).toMatchObject({ frozenPids: [100, 200], frozenAt: 1_000, thawedAt: 4_000 }); + }); + + it("ages while frozen and lets a lock holder run briefly before retrying", async () => { + const { input, calls, record } = harness({ age: async onLockTimeout => { calls.push("age"); await onLockTimeout(); calls.push("aged"); + return { faultId: "f01" } as AgingResult; } }); + await runFreezeFault(input); + expect(calls).toEqual(["freeze:100:true", "age", "thaw", "freeze:100:true", "aged", "thaw", "wait"]); + expect(record.aging).toEqual({ faultId: "f01" }); + }); + + it("thaws when aging fails", async () => { + const { input, calls } = harness({ age: async () => { throw new Error("shift failed"); } }); + await expect(runFreezeFault(input)).rejects.toThrow("shift failed"); + expect(calls).toEqual(["freeze:100:true", "thaw"]); + }); + + it("replaces a server that exited after the thaw and records it", async () => { + const { input, calls, record } = harness({ waitHealthy: async () => { throw new Error("exited"); }, + exitStatus: () => ({ code: 1, signal: null }) }); + await expect(runFreezeFault(input)).resolves.toMatchObject({ healthyAt: healthy + 1 }); + expect(calls).toEqual(["freeze:100:true", "thaw", "cleanup", "replace"]); + expect(record.serverExitedAfterThaw).toEqual({ code: 1, signal: null }); + expect(record.cleanupError).toBeUndefined(); + }); + + it("fails on a cleanup failure without starting a second server", async () => { + const { input, calls, record } = harness({ waitHealthy: async () => { throw new Error("exited"); }, + exitStatus: () => ({ code: null, signal: "SIGKILL" }), replaceExited: async () => { throw new Error("runner 300 still alive"); } }); + await expect(runFreezeFault(input)).rejects.toThrow(/could not be shown stopped: runner 300 still alive/); + expect(record.cleanupError).toBe("runner 300 still alive"); + expect(calls).not.toContain("replace"); + }); + + it("fails without a replacement when a live server never answers", async () => { + const { input, calls } = harness({ waitHealthy: async () => { throw new Error("not healthy"); } }); + await expect(runFreezeFault(input)).rejects.toThrow("not healthy"); + expect(calls).not.toContain("replace"); + }); +}); diff --git a/tests/restart-chaos/time-faults.ts b/tests/restart-chaos/time-faults.ts new file mode 100644 index 00000000000..e14e43aa096 --- /dev/null +++ b/tests/restart-chaos/time-faults.ts @@ -0,0 +1,439 @@ +/** + * Time faults (TM1–TM3): the fourth fault family, next to restarts (K*), storage + * (ST*) and provider messages (PV*). Every other wave is minutes long on a fresh + * instance, so nothing time-bound ever expires. These faults make deadlines lapse + * during a run or across a restart, without changing product code: + * + * - TM1 freeze: SIGSTOP the server process alone (not its runners, not Postgres) + * for 95–150 s, then SIGCONT. Longer than the controller-instance lease (90 s), + * the execution-control deadline (60 s) and the legacy run lease (60 s), so the + * server's own leases lapse while it believes it is healthy (a GC pause), while + * the runner keeps working and calling a server that does not answer. + * - TM2 sleep: SIGSTOP the server and every process under it (runners, providers), + * shift every timestamp in the wave's database back by the age (default 26 h), + * then SIGCONT. A laptop-sleep stand-in: the database says the age has passed, + * while the frozen processes' own monotonic timers barely moved. + * - TM3 aged outage: SIGKILL the server tree (as K1), shift every timestamp back + * by the age while it is down, then restart. The replacement must reconcile + * runs, leases, wakes and waits that look hours old. With `--workloads G1 + * --triggers approval` this is the aged wait: an approval pending for a day + * across a restart (the class of #20340). + * + * Aging shifts every `timestamp` column of every table in the `public` schema by + * the same interval, so the database is exactly what it would be had everything + * happened the age earlier. Triggers are disabled for the shift (replica role) + * so derived columns are not rewritten. The rows of the time-bound tables are + * recorded with their values before the shift in `time-faults/.json`. + * + * Grading reads run times through `TimeShiftedApi`, which adds the age back to + * any timestamp the shift moved, so R4 overlaps and speed budgets are judged on + * real time. The product itself sees only the aged values. + */ +import { execFileSync } from "node:child_process"; +import { readFileSync } from "node:fs"; +import { mkdir, writeFile } from "node:fs/promises"; +import path from "node:path"; +import { createDb } from "../../packages/db/src/index.js"; +import { ChaosApi } from "./api.js"; +import type { TimeFaultKind } from "./time-faults-config.js"; + +export { DEFAULT_TIME_AGE_MS, TIME_FAULT_KINDS, TIME_FREEZE_HOLD_RANGE_MS, isTimeFault, timeFaultHoldMs, type TimeFaultKind } from "./time-faults-config.js"; + +/** Tables whose rows carry the deadlines a run depends on; their rows are recorded in the evidence. */ +export const TIME_BOUND_TABLES = [ + "heartbeat_runs", "native_run_finalizations", "controller_instances", "agent_wakeup_requests", "environment_leases", + "execution_workspace_runtime_leases", "workspace_operations", "approvals", "decisions", "issue_thread_interactions", + "budget_resume_claims", "routine_triggers", "issue_recovery_actions", +] as const; +const RECORDED_ROWS_PER_TABLE = 200; + +export interface AgingShift { + faultId: string; + ageMs: number; + /** Harness clock when the shift committed. */ + appliedAt: number; + /** + * Wave start on the harness clock. Every time the wave wrote is at or after it, and the age is + * longer than the wave has run, so a value this shift moved lands before it, unless the value was + * a deadline more than `ageMs - (appliedAt - waveStartedAt)` in the future. + */ + waveStartedAt: number; +} + +export interface AgedTable { table: string; columns: string[]; rows: number } + +export interface AgingResult extends AgingShift { + startedAt: number; + tables: AgedTable[]; + totalRows: number; + /** False when the database role could not disable triggers (the shift then ran with them on). */ + triggersDisabled: boolean; + /** TM2: lock waits on a frozen server's connection that needed a brief thaw before retrying. */ + lockRetries: number; + evidencePath: string; +} + +export interface TimeFaultRecord { + kind: TimeFaultKind; + /** TM1/TM2: the processes frozen (server first). */ + frozenPids?: number[]; + frozenAt?: number; + thawedAt?: number; + aging?: AgingResult; + /** The frozen server exited on its own after SIGCONT (for example a lost-lease self-fence); a replacement was started. */ + serverExitedAfterThaw?: { code: number | null; signal: string | null }; + /** The exited server's process tree could not be shown stopped; the fault fails as a harness error. */ + cleanupError?: string; +} + +/** + * TM1/TM2 sequence, with every side effect injected (supervisor.ts supplies them): + * freeze, optionally age while frozen, hold, always thaw, then wait for the same + * server. A server that exited on its own after the thaw is cleaned up and replaced; + * a cleanup that cannot prove the old tree stopped is recorded, not swallowed. + */ +export async function runFreezeFault(input: { + kind: "TM1" | "TM2"; rootPid: number; holdMs: number; record: TimeFaultRecord; + freezer: Pick; + age?: (onLockTimeout: () => Promise) => Promise; + cancelled: () => boolean; + onFrozen?: (pids: number[]) => void; onThawed?: (frozenMs: number) => void; + waitHealthy: () => Promise; + exitStatus: () => { code: number | null; signal: string | null } | null; + replaceExited: () => Promise; + startReplacement: () => Promise; + now?: () => number; sleep?: (ms: number) => Promise; +}): Promise<{ stoppedAt: number; restartedAt: number; healthyAt: number }> { + const now = input.now ?? Date.now; + const sleep = input.sleep ?? ((ms: number) => new Promise(resolve => setTimeout(resolve, ms))); + const { record } = input; + const tree = input.kind === "TM2"; + let stoppedAt = now(); + try { + record.frozenPids = input.freezer.freeze(input.rootPid, tree); + stoppedAt = record.frozenAt = now(); + input.onFrozen?.(record.frozenPids); + if (input.age) { + record.aging = await input.age(async () => { + // A frozen connection holds a lock the shift needs: let the tree run briefly, then freeze it again. + input.freezer.thaw(); + await sleep(2_000); + record.frozenPids = [...new Set([...(record.frozenPids ?? []), ...input.freezer.freeze(input.rootPid, tree)])]; + }); + } + while (now() - stoppedAt < input.holdMs) { + if (input.cancelled()) throw new Error("Harness shutdown interrupted the freeze"); + await sleep(Math.min(1_000, input.holdMs - (now() - stoppedAt))); + } + } finally { + input.freezer.thaw(); + record.thawedAt = now(); + input.onThawed?.(record.thawedAt - (record.frozenAt ?? record.thawedAt)); + } + const restartedAt = record.thawedAt; + try { + return { stoppedAt, restartedAt, healthyAt: await input.waitHealthy() }; + } catch (error) { + const exit = input.exitStatus(); + if (!exit) throw error; + record.serverExitedAfterThaw = exit; + try { + await input.replaceExited(); + } catch (cleanup) { + // An old runner may still be alive and writing: fail the fault rather than start a second server beside it. + record.cleanupError = cleanup instanceof Error ? cleanup.message : String(cleanup); + throw new Error(`Exited server's process tree could not be shown stopped: ${record.cleanupError}`); + } + return { stoppedAt, restartedAt, healthyAt: await input.startReplacement() }; + } +} + +export interface ProcessRow { pid: number; ppid: number; command: string } + +export function parseProcessTable(output: string): ProcessRow[] { + return output.split("\n").flatMap(line => { + const match = /^\s*(\d+)\s+(\d+)\s+(.*)$/.exec(line); + return match ? [{ pid: Number(match[1]), ppid: Number(match[2]), command: match[3]!.trim() }] : []; + }); +} + +/** + * Every process under `rootPid`, parents before children. Postgres never runs + * under the server (it is wrapper-owned, beside it), and anything that looks like + * Postgres is excluded anyway: freezing the database is a different fault. + */ +export function descendantPids(rootPid: number, table: readonly ProcessRow[]): number[] { + const children = new Map(); + for (const row of table) children.set(row.ppid, [...(children.get(row.ppid) ?? []), row]); + const out: number[] = []; + const queue = [...(children.get(rootPid) ?? [])]; + while (queue.length) { + const row = queue.shift()!; + if (/(^|\/)postgres(\s|$)|postmaster/.test(row.command)) continue; + out.push(row.pid); + queue.push(...(children.get(row.pid) ?? [])); + } + return out; +} + +function readProcessTable(): ProcessRow[] { + return parseProcessTable(execFileSync("ps", ["-A", "-o", "pid=,ppid=,command="], { encoding: "utf8", maxBuffer: 16 * 1024 * 1024 })); +} + +const frozenEverywhere = new Set(); +let exitHookInstalled = false; + +/** + * SIGSTOP/SIGCONT for processes this wave started. Whatever happens to the + * harness, a frozen process is continued on exit, so it can still be stopped. + */ +export class ProcessFreezer { + private frozen: number[] = []; + + constructor(private readonly signal: (pid: number, signal: NodeJS.Signals) => void = (pid, sig) => process.kill(pid, sig), + private readonly table: () => ProcessRow[] = readProcessTable) { + if (!exitHookInstalled) { + exitHookInstalled = true; + process.on("exit", () => { for (const pid of frozenEverywhere) try { process.kill(pid, "SIGCONT"); } catch { /* gone */ } }); + } + } + + get pids(): readonly number[] { return this.frozen; } + + /** Freezes the root first (it then cannot spawn), then its tree, rescanning once for late children. */ + freeze(rootPid: number, tree: boolean): number[] { + this.stop(rootPid); + if (tree) { + for (let pass = 0; pass < 2; pass += 1) { + for (const pid of descendantPids(rootPid, this.table())) if (!this.frozen.includes(pid)) this.stop(pid); + } + } + return [...this.frozen]; + } + + thaw(): void { + for (const pid of this.frozen) { + try { this.signal(pid, "SIGCONT"); } catch { /* exited meanwhile */ } + frozenEverywhere.delete(pid); + } + this.frozen = []; + } + + private stop(pid: number) { + try { this.signal(pid, "SIGSTOP"); } catch { return; } + this.frozen.push(pid); + frozenEverywhere.add(pid); + } +} + +/** The wave's database URL from the instance config the supervisor wrote. */ +export function databaseUrlFromConfig(configPath: string): string { + const config = JSON.parse(readFileSync(configPath, "utf8")) as { database?: { connectionString?: string } }; + const url = config.database?.connectionString; + if (!url) throw new Error("Time fault: the instance config has no database connection string"); + return url; +} + +const quote = (identifier: string) => `"${identifier.replace(/"/g, "\"\"")}"`; + +export type AgingColumn = { table: string; column: string; kind: "timestamp" | "json" }; + +/** + * A session function (pg_temp) that shifts every time inside a JSON value, so a + * copy of a time (a snapshot, a receipt, an event payload) still equals the + * column it was copied from: + * - an ISO datetime string keeps its exact format, separator and offset; + * - a whole number in the epoch-millisecond window [lo, hi] (the wave's own times) moves by the age. + * The age is whole seconds, so fractional digits never change. + */ +export const SHIFT_JSON_FUNCTION = [ + "create or replace function pg_temp.paperclip_shift_json(v jsonb, age_ms bigint, lo bigint, hi bigint)", + "returns jsonb language plpgsql immutable as $fn$", + "declare t text; m text[]; r jsonb; shifted timestamp;", + "begin", + " case jsonb_typeof(v)", + " when 'object' then", + " select coalesce(jsonb_object_agg(e.key, pg_temp.paperclip_shift_json(e.value, age_ms, lo, hi)), '{}'::jsonb) into r from jsonb_each(v) e;", + " return r;", + " when 'array' then", + " select coalesce(jsonb_agg(pg_temp.paperclip_shift_json(e.value, age_ms, lo, hi) order by e.ord), '[]'::jsonb) into r", + " from jsonb_array_elements(v) with ordinality e(value, ord);", + " return r;", + " when 'string' then", + " t := v #>> '{}';", + " m := regexp_match(t, '^([0-9]{4}-[0-9]{2}-[0-9]{2})([T ])([0-9]{2}:[0-9]{2}:[0-9]{2})([.][0-9]+)?(Z|[+-][0-9]{2}(:?[0-9]{2})?)?$');", + " if m is null then return v; end if;", + " begin", + " shifted := (m[1] || ' ' || m[3])::timestamp - make_interval(secs => age_ms / 1000);", + " exception when others then return v;", + " end;", + " return to_jsonb(to_char(shifted, 'YYYY-MM-DD') || m[2] || to_char(shifted, 'HH24:MI:SS') || coalesce(m[4], '') || coalesce(m[5], ''));", + " when 'number' then", + " t := v #>> '{}';", + " if t ~ '^[0-9]{12,14}$' and t::numeric between lo and hi then return to_jsonb(t::bigint - age_ms); end if;", + " return v;", + " else", + " return v;", + " end case;", + "end $fn$", +].join("\n"); + +/** One UPDATE per table, shifting all of its timestamp and JSON columns at once. */ +export function agingStatements(columns: ReadonlyArray, ageMs: number, + epochWindow: { lo: number; hi: number }): Array<{ table: string; columns: string[]; sql: string }> { + if (!Number.isSafeInteger(ageMs) || ageMs <= 0 || ageMs % 1000 !== 0) throw new Error("Aging needs a positive whole number of seconds"); + const byTable = new Map(); + for (const column of columns) byTable.set(column.table, [...(byTable.get(column.table) ?? []), column]); + const shift = (column: AgingColumn) => column.kind === "timestamp" + ? `${quote(column.column)} - interval '${ageMs} milliseconds'` + : `pg_temp.paperclip_shift_json(${quote(column.column)}, ${ageMs}, ${Math.trunc(epochWindow.lo)}, ${Math.trunc(epochWindow.hi)})`; + return [...byTable.entries()].sort(([a], [b]) => a.localeCompare(b)).map(([table, cols]) => { + const present = cols.map(col => `${quote(col.column)} is not null`).join(" or "); + return { table, columns: cols.map(col => col.column), + sql: `update public.${quote(table)} set ${cols.map(col => `${quote(col.column)} = ${shift(col)}`).join(", ")} where ${present}` }; + }); +} + +const LOCK_NOT_AVAILABLE = "55P03"; +const MAX_LOCK_ATTEMPTS = 6; + +/** + * Shifts every timestamp in the wave's database back by `ageMs` in one + * transaction. `onLockTimeout` runs between attempts when a lock is held (TM2: + * a frozen server's open transaction); it may thaw and refreeze the server. + */ +export async function ageWaveDatabase(input: { connectionString: string; faultId: string; ageMs: number; evidenceDir: string; + /** Wave start on the harness clock: the shift must move every row the wave wrote below anything it did not move. */ + waveStartedAt: number; + /** Total age of earlier shifts in this wave: their epoch milliseconds already sit that much below the window. */ + priorAgeMs?: number; + onLockTimeout?: () => Promise; now?: () => number }): Promise { + const now = input.now ?? Date.now; + const startedAt = now(); + if (startedAt - input.waveStartedAt >= input.ageMs - 60_000) { + throw new Error(`Time fault ${input.faultId}: the wave has run ${Math.round((startedAt - input.waveStartedAt) / 60_000)} min, ` + + "too close to the age; raise --time-age-hours so shifted and unshifted times cannot overlap"); + } + const db = createDb(input.connectionString); + const sql = db.$client; + try { + const columns = await sql.unsafe(`select c.table_name as "table", c.column_name as "column", + case when c.data_type = 'jsonb' then 'json' else 'timestamp' end as kind + from information_schema.columns c join information_schema.tables t + on t.table_schema = c.table_schema and t.table_name = c.table_name and t.table_type = 'BASE TABLE' + where c.table_schema = 'public' and c.data_type in ('timestamp with time zone', 'timestamp without time zone', 'jsonb') + order by c.table_name, c.ordinal_position`) as unknown as AgingColumn[]; + const statements = agingStatements(columns, input.ageMs, epochWindow(input.waveStartedAt, input.priorAgeMs ?? 0, startedAt)); + const identityColumns = await sql.unsafe(`select table_name as "table", column_name as "column" from information_schema.columns + where table_schema = 'public' and table_name = any($1) and column_name in ('id', 'boot_id', 'run_id', 'issue_id', 'status', 'phase', 'state') + order by table_name, ordinal_position`, [[...TIME_BOUND_TABLES]]) as unknown as Array<{ table: string; column: string }>; + const [role] = await sql.unsafe("select rolsuper from pg_roles where rolname = current_user") as unknown as Array<{ rolsuper: boolean }>; + const triggersDisabled = role?.rolsuper === true; + const recorded = new Set(TIME_BOUND_TABLES); + let lockRetries = 0; + for (let attempt = 1; ; attempt += 1) { + try { + const outcome = await sql.begin(async tx => { + await tx.unsafe("set local lock_timeout = '3s'"); + if (triggersDisabled) await tx.unsafe("set local session_replication_role = replica"); + await tx.unsafe(SHIFT_JSON_FUNCTION); + const before: Record = {}; + for (const statement of statements) { + if (!recorded.has(statement.table)) continue; + // Identity, state and times only: payload columns can carry run output. + const keys = identityColumns.filter(key => key.table === statement.table).map(key => key.column); + const times = columns.filter(column => column.table === statement.table && column.kind === "timestamp").map(column => column.column); + const select = [...keys, ...times].map(quote).join(", "); + const order = times.includes("created_at") ? " order by created_at desc" : ""; + before[statement.table] = await tx.unsafe(`select ${select} from public.${quote(statement.table)}${order} limit ${RECORDED_ROWS_PER_TABLE}`) as unknown as unknown[]; + } + const tables: AgedTable[] = []; + for (const statement of statements) { + const result = await tx.unsafe(statement.sql); + tables.push({ table: statement.table, columns: statement.columns, rows: result.count ?? 0 }); + } + return { before, tables }; + }); + const appliedAt = now(); + const tables = outcome.tables.filter(table => table.rows > 0); + const evidencePath = path.join(input.evidenceDir, `${input.faultId}.json`); + await mkdir(input.evidenceDir, { recursive: true, mode: 0o700 }); + await writeFile(evidencePath, `${JSON.stringify({ faultId: input.faultId, ageMs: input.ageMs, startedAt, appliedAt, + note: "Rows of the time-bound tables as they were before the shift; after the shift every timestamp shown is ageMs earlier.", + tables, rowsBefore: outcome.before }, null, 2)}\n`, { mode: 0o600 }); + return { faultId: input.faultId, ageMs: input.ageMs, appliedAt, waveStartedAt: input.waveStartedAt, startedAt, tables, + totalRows: tables.reduce((sum, table) => sum + table.rows, 0), triggersDisabled, lockRetries, evidencePath }; + } catch (error) { + if ((error as { code?: string }).code !== LOCK_NOT_AVAILABLE || attempt >= MAX_LOCK_ATTEMPTS) throw error; + lockRetries += 1; + await input.onLockTimeout?.(); + } + } + } finally { + await sql.end({ timeout: 5 }).catch(() => undefined); + } +} + +const DAY_MS = 86_400_000; + +/** + * Epoch milliseconds the wave itself could have written: from a day before it started (less every + * earlier shift, so a value aged once is aged again) to a week ahead. + */ +export function epochWindow(waveStartedAt: number, priorAgeMs: number, now: number): { lo: number; hi: number } { + return { lo: waveStartedAt - DAY_MS - priorAgeMs, hi: now + 7 * DAY_MS }; +} + +const ISO_TIMESTAMP = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})$/; + +/** + * The real time of a database time the shifts moved; other values are returned unchanged. Latest + * shift first. A moved value lands before the wave started (the age outlasts the wave); nothing the + * wave wrote is earlier. A deadline more than the age minus the wave's run time ahead stays aged. + */ +export function unshiftMs(value: number, shifts: readonly AgingShift[]): number { + let real = value; + for (const shift of [...shifts].sort((a, b) => b.appliedAt - a.appliedAt)) { + if (real < shift.waveStartedAt) real += shift.ageMs; + } + return real; +} + +/** Whole numbers inside JSON that aging treats as epoch milliseconds, read back the same way. */ +function shiftedEpochMs(node: number, shifts: readonly AgingShift[]): boolean { + if (!Number.isSafeInteger(node) || node < 1e11 || node >= 1e14) return false; + const total = shifts.reduce((sum, shift) => sum + shift.ageMs, 0); + const earliest = Math.min(...shifts.map(shift => shift.waveStartedAt)); + return node >= earliest - DAY_MS - total; +} + +/** Copies `value`, with every ISO timestamp and epoch-millisecond number the shifts moved put back at its real time. */ +export function unshiftJson(value: T, shifts: readonly AgingShift[]): T { + if (!shifts.length) return value; + const visit = (node: unknown): unknown => { + if (typeof node === "string") { + if (!ISO_TIMESTAMP.test(node)) return node; + const ms = Date.parse(node); + if (!Number.isFinite(ms)) return node; + const real = unshiftMs(ms, shifts); + return real === ms ? node : new Date(real).toISOString(); + } + if (typeof node === "number") return shiftedEpochMs(node, shifts) ? unshiftMs(node, shifts) : node; + if (Array.isArray(node)) return node.map(visit); + if (node && typeof node === "object") return Object.fromEntries(Object.entries(node).map(([key, child]) => [key, visit(child)])); + return node; + }; + return visit(value) as T; +} + +/** + * The board API as the grader should read it in a wave with aged rows: run and + * event times come back at their real times. Writes are unchanged. + */ +export class TimeShiftedApi extends ChaosApi { + constructor(baseUrl: string, readonly shifts: AgingShift[]) { super(baseUrl); } + + override get(route: string, timeoutMs?: number): Promise { + return super.get(route, timeoutMs).then(value => unshiftJson(value, this.shifts)); + } +} diff --git a/tests/restart-chaos/types.ts b/tests/restart-chaos/types.ts index e6bec5e3acf..4a8afc07bd2 100644 --- a/tests/restart-chaos/types.ts +++ b/tests/restart-chaos/types.ts @@ -1,9 +1,13 @@ /** Shared types for the control-plane restart chaos harness (plan §5). */ import type { ProviderFaultEvent, ProviderFaultKind, ProviderFaultRule, ProviderMessage } from "./provider-faults.js"; import type { QuestionArtifacts, QuestionRecord } from "./question-handoff.js"; +import type { TimeFaultRecord } from "./time-faults.js"; -/** K14 (upgrade mid-run): a graceful stop, then the replacement starts from the current build (`--upgrade-from`). */ -export const FAULT_KINDS = ["K1", "K2", "K4", "K6", "K14"] as const; +/** + * K14 (upgrade mid-run): a graceful stop, then the replacement starts from the current build (`--upgrade-from`). + * TM1–TM3 (time faults, time-faults.ts): freeze the server, sleep with aged rows, or an aged outage. + */ +export const FAULT_KINDS = ["K1", "K2", "K4", "K6", "K14", "TM1", "TM2", "TM3"] 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. @@ -102,6 +106,8 @@ export interface WaveOptions { matrix?: boolean; /** `--server-from `: every server of the wave runs from this prepared checkout (for example a pre-fix commit). */ serverFrom?: string; + /** TM2/TM3: how far every database timestamp is shifted back (`--time-age-hours`, default 26 h). */ + timeAgeMs?: number; } export interface StormFault { @@ -225,7 +231,7 @@ export interface FaultReceipt { kind: FaultKind; /** Task that scheduled the fault; every other running task is also exposed. */ requestedBy: string; - signal: "SIGKILL" | "SIGTERM"; + signal: "SIGKILL" | "SIGTERM" | "SIGSTOP"; holdMs: number; requestedAt: number; startedAt: number; @@ -242,6 +248,8 @@ export interface FaultReceipt { crashPoint?: { name: string; hit: number; pid: number; token?: string }; /** K14: the build the stopped server ran and the build its replacement runs. */ upgrade?: { fromRoot: string; fromSha: string | null; toRoot: string; toSha: string | null; switched: boolean }; + /** TM1–TM3: what was frozen and which rows were aged. */ + timeFault?: TimeFaultRecord; serverLogOffsetAtStart: number; serverLogOffsetAtReady: number | null; error: string | null;