From 807d5b91c1eeab1ec6be965306ab9f66b4f4372e Mon Sep 17 00:00:00 2001 From: dogskip Date: Sun, 19 Jul 2026 07:05:07 +0900 Subject: [PATCH] feat: add reproducible scenario catalog --- README.md | 6 + docs/data-model.md | 23 ++ .../plans/2026-07-19-reproducible-catalog.md | 60 +++++ docs/test-contract.md | 5 +- docs/threat-model.md | 10 +- package.json | 3 +- schema/sqlite/001_catalog.sql | 51 ++++ src/app.ts | 193 ++++----------- src/catalog.ts | 222 ++++++++++++++++++ src/index.ts | 2 + src/scenario.ts | 186 +++++++++++++++ src/server.ts | 6 +- test/app.test.ts | 58 +++++ test/catalog.test.ts | 64 +++++ test/scenario.test.ts | 36 +++ 15 files changed, 771 insertions(+), 154 deletions(-) create mode 100644 docs/data-model.md create mode 100644 docs/superpowers/plans/2026-07-19-reproducible-catalog.md create mode 100644 schema/sqlite/001_catalog.sql create mode 100644 src/catalog.ts create mode 100644 src/scenario.ts create mode 100644 test/catalog.test.ts create mode 100644 test/scenario.test.ts diff --git a/README.md b/README.md index f488ea1..bfb9f83 100644 --- a/README.md +++ b/README.md @@ -4,12 +4,16 @@ Causal Lab is a deterministic network simulator for studying an observed-remove Each replica records dotted operations and a version vector. Put and remove operations carry the exact dots they observed; tombstones make late and duplicated delivery harmless. Healing performs a reliable anti-entropy pass and reports whether all replicas converged to the same canonical state. +An optional content-addressed SQLite catalog persists validated scenarios and immutable deterministic run receipts. It is deliberately outside the simulation core: no storage call, clock, or generated identifier can affect convergence behavior. + ## Boundary - at most 12 replicas and 10,000 scheduled events - at most 2 MiB per JSON request - no wall-clock time, sockets, telemetry, hosted state, or dynamic code execution in the simulation core - scenario identifiers, keys, values, and replica names are bounded before allocation +- canonical scenario and run IDs are SHA-256 receipts over versioned, sorted-key JSON +- SQLite STRICT tables enforce digest, JSON, byte-count, boolean, and foreign-key contracts The precise invariants are in [`docs/test-contract.md`](docs/test-contract.md). Security assumptions are in [`docs/threat-model.md`](docs/threat-model.md). @@ -17,6 +21,8 @@ The precise invariants are in [`docs/test-contract.md`](docs/test-contract.md). Node 24 or newer and pnpm 10 are required. The core never reads wall-clock time. `pnpm start` serves `POST /v1/run` on `127.0.0.1:8787`; the host remains restricted to `127.0.0.1` or `::1`. +Set `CAUSAL_LAB_DB` to a trusted SQLite file path to add the scenario catalog routes. Without it the original stateless API and its failure surface remain unchanged. See `docs/data-model.md` for the exact tables, receipts, backup boundary, and recovery contract. + ```sh pnpm test pnpm typecheck diff --git a/docs/data-model.md b/docs/data-model.md new file mode 100644 index 0000000..fca2929 --- /dev/null +++ b/docs/data-model.md @@ -0,0 +1,23 @@ +# Scenario catalog data model + +The simulator remains a pure deterministic core. Storage is an optional adapter enabled by CAUSAL_LAB_DB, so a database failure cannot alter CRDT or virtual-time semantics and removing the setting restores the original stateless service. + +## Identity + +Validated JSON is encoded with causal-lab-json-v1: object keys are sorted, arrays retain order, and only finite JSON values are admitted. The scenario ID is SHA-256 over the version label, a separator, and those canonical bytes. Key order and whitespace therefore do not change identity, while any semantic field does. + +Run IDs bind a scenario ID to the complete canonical report. trace_sha256 separately makes trace verification and indexing cheap. Repeating one deterministic scenario produces the same immutable receipt rather than another timestamp-based row. + +## Tables + +- scenarios stores scenario_id, canonicalization, canonical_json, and definition_bytes. +- runs stores run_id, scenario_id, report_json, trace_sha256, processed_events, and converged. +- schema_migrations records the applied schema version and owner. + +All tables use STRICT typing. Digest shape, JSON validity, byte and event bounds, booleans, and the scenario foreign key are database constraints. Triggers reject updates and deletes from content-addressed rows. On every open the adapter runs integrity_check and foreign_key_check. + +## Operations and recovery + +POST /v1/scenarios stores a validated definition. GET /v1/scenarios/:id returns it. POST /v1/scenarios/:id/runs executes and stores the deterministic receipt, and GET /v1/runs/:id reads it. These routes exist only when the catalog is configured; POST /v1/run remains stateless and backward compatible. + +For a live backup, use the SQLite backup API or briefly stop the single process and copy the database together with its WAL and shared-memory files. A portable rollback is an export of canonical scenario and report JSON, verified again by their IDs before import. diff --git a/docs/superpowers/plans/2026-07-19-reproducible-catalog.md b/docs/superpowers/plans/2026-07-19-reproducible-catalog.md new file mode 100644 index 0000000..c4b9c6b --- /dev/null +++ b/docs/superpowers/plans/2026-07-19-reproducible-catalog.md @@ -0,0 +1,60 @@ +# Reproducible Scenario Catalog Implementation Plan + +**Goal:** Add a content-addressed SQLite catalog for scenarios and deterministic run receipts while keeping CRDT semantics storage-independent. +**Architecture:** Parsing and execution become pure reusable functions. A catalog adapter stores canonical scenario JSON and immutable reports; the simulator never imports storage. +**Tech Stack:** TypeScript 5.9, Node.js 24 node:sqlite, Hono, Vitest, SQLite STRICT tables +**Verification:** pnpm lint; pnpm typecheck; pnpm test; pnpm build; PRAGMA foreign_key_check and integrity_check. + +--- + +## Three-pass review + +1. Repository evidence: validation and execution live inside one HTTP handler, preventing reusable scenario identity. +2. External standard: CRDTs converge from the same update set; SQLite offers strict typing and atomic single-file storage. +3. Adversarial review: timestamps, random IDs, mutable reports, arbitrary invariant code, and user-controlled SQL were rejected. + +## Decision + +Canonicalize validated scenarios with sorted keys and SHA-256. The digest is the scenario ID. Store one immutable report and trace digest per scenario. Existing POST /v1/run stays compatible; catalog routes exist only when configured. + +## SQLite schema + +- scenarios(scenario_id, canonical_json, definition_bytes) +- runs(run_id, scenario_id, report_json, trace_sha256, processed_events, converged) +- schema_migrations(version, applied_by) + +IDs are 64 lowercase hex characters, JSON is checked, and reports reference scenarios. + +## Implementation tasks + +### Task 1: Extract pure scenario execution + +**Files:** Create src/scenario.ts and test/scenario.test.ts; modify src/app.ts. + +- [x] Add failing tests for identity across key order and byte-identical execution. + +### Task 2: Implement the catalog + +**Files:** Create schema/sqlite/001_catalog.sql, src/catalog.ts, and test/catalog.test.ts. + +- [x] Test migration idempotence, immutable replay, hash rejection, foreign keys, reopen, and integrity checks. +- [x] Use prepared statements and disable extensions. + +### Task 3: Add catalog routes + +**Files:** Modify src/app.ts, src/server.ts, and test/app.test.ts. + +- [x] Add create, fetch, and run receipt routes without changing stateless limits. + +### Task 4: Document recovery + +**Files:** Modify README.md, docs/test-contract.md, and docs/threat-model.md; create docs/data-model.md. + +- [x] Document WAL-aware backup and canonical JSON export rollback. + +## Risk and rollback + +- Medium: isolate the active-development Node SQLite API in one adapter. +- Medium: version canonicalization and pin it with fixtures. +- Low: fail startup on integrity or foreign-key errors. +- Rollback: omit CAUSAL_LAB_DB; stateless execution remains unchanged. diff --git a/docs/test-contract.md b/docs/test-contract.md index d9bd45e..9f6530e 100644 --- a/docs/test-contract.md +++ b/docs/test-contract.md @@ -9,6 +9,9 @@ The suite must demonstrate: 5. a reliable anti-entropy pass after healing converges every replica; 6. different delivery histories that contain the same operations converge to the same canonical map; 7. replica, event, field, probability, and HTTP body limits are enforced before state mutation. +8. object key order and whitespace do not change a scenario ID, while semantic changes do; +9. repeated deterministic execution produces one immutable run receipt across database reopen; +10. catalog migrations are idempotent and integrity, foreign-key, hash, byte-count, and trace checks pass; +11. catalog routes are absent when storage is not configured and preserve the stateless run contract. Limits: 12 replicas, 10,000 events, 2 MiB JSON, 128-byte identifiers and keys, and 4 KiB values. - diff --git a/docs/threat-model.md b/docs/threat-model.md index c54437e..e5d82a8 100644 --- a/docs/threat-model.md +++ b/docs/threat-model.md @@ -2,7 +2,7 @@ ## Assets and inputs -The simulator protects deterministic trace order, operation identity, version-vector monotonicity, observed-remove semantics, and bounded resource use. Scenario JSON, replica names, keys, values, topology changes, and network probabilities are untrusted. +The simulator protects deterministic trace order, operation identity, version-vector monotonicity, observed-remove semantics, bounded resource use, scenario identity, and immutable run receipts. Scenario JSON, replica names, keys, values, topology changes, network probabilities, catalog IDs, and database contents are untrusted. ## Defended cases @@ -12,8 +12,12 @@ The simulator protects deterministic trace order, operation identity, version-ve - operations are immutable data; duplicate delivery is idempotent and remove tombstones survive out-of-order delivery - virtual time is an integer controlled by the simulation, not by timers or the host clock - public failures report contract errors without stack traces or environment paths +- storage is injected at the HTTP boundary and is never imported by CRDT or simulation modules +- prepared statements bind all catalog values; extension loading and double-quoted string literals are disabled +- catalog creation executes the bounded pure scenario once, so semantically invalid replica references are never persisted +- database files reject symbolic-link targets and are restricted to process-owner permissions +- STRICT tables, immutable triggers, foreign keys, digest recomputation, and open-time health checks detect drift ## Non-goals -This is not a production database, Byzantine protocol, consensus system, authentication service, or performance benchmark. It does not defend against a process-account compromise or attempt to model every transport behavior. - +This is not a Byzantine protocol, consensus system, authentication service, or performance benchmark. The optional local catalog is not a multi-writer distributed database. It does not defend against a process-account compromise or attempt to model every transport behavior. diff --git a/package.json b/package.json index 486cfdb..f382b87 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,8 @@ "causal-lab": "./dist/server.js" }, "files": [ - "dist" + "dist", + "schema" ], "scripts": { "build": "tsc -p tsconfig.build.json", diff --git a/schema/sqlite/001_catalog.sql b/schema/sqlite/001_catalog.sql new file mode 100644 index 0000000..cea82d6 --- /dev/null +++ b/schema/sqlite/001_catalog.sql @@ -0,0 +1,51 @@ +CREATE TABLE IF NOT EXISTS schema_migrations ( + version INTEGER PRIMARY KEY CHECK (version > 0), + applied_by TEXT NOT NULL CHECK (applied_by = 'causal-lab') +) STRICT; + +CREATE TABLE IF NOT EXISTS scenarios ( + scenario_id TEXT PRIMARY KEY CHECK ( + length(scenario_id) = 64 AND scenario_id NOT GLOB '*[^0-9a-f]*' + ), + canonicalization TEXT NOT NULL CHECK (canonicalization = 'causal-lab-json-v1'), + canonical_json TEXT NOT NULL CHECK (json_valid(canonical_json)), + definition_bytes INTEGER NOT NULL CHECK (definition_bytes > 0 AND definition_bytes <= 2097152) +) STRICT, WITHOUT ROWID; + +CREATE TABLE IF NOT EXISTS runs ( + run_id TEXT PRIMARY KEY CHECK ( + length(run_id) = 64 AND run_id NOT GLOB '*[^0-9a-f]*' + ), + scenario_id TEXT NOT NULL, + report_json TEXT NOT NULL CHECK (json_valid(report_json)), + trace_sha256 TEXT NOT NULL CHECK ( + length(trace_sha256) = 64 AND trace_sha256 NOT GLOB '*[^0-9a-f]*' + ), + processed_events INTEGER NOT NULL CHECK (processed_events >= 0), + converged INTEGER NOT NULL CHECK (converged IN (0, 1)), + FOREIGN KEY (scenario_id) REFERENCES scenarios(scenario_id) ON DELETE RESTRICT +) STRICT, WITHOUT ROWID; + +CREATE INDEX IF NOT EXISTS runs_by_scenario ON runs(scenario_id, run_id); + +CREATE TRIGGER IF NOT EXISTS scenarios_immutable_update +BEFORE UPDATE ON scenarios BEGIN + SELECT RAISE(ABORT, 'scenarios are immutable'); +END; + +CREATE TRIGGER IF NOT EXISTS scenarios_immutable_delete +BEFORE DELETE ON scenarios BEGIN + SELECT RAISE(ABORT, 'scenarios are immutable'); +END; + +CREATE TRIGGER IF NOT EXISTS runs_immutable_update +BEFORE UPDATE ON runs BEGIN + SELECT RAISE(ABORT, 'runs are immutable'); +END; + +CREATE TRIGGER IF NOT EXISTS runs_immutable_delete +BEFORE DELETE ON runs BEGIN + SELECT RAISE(ABORT, 'runs are immutable'); +END; + +INSERT OR IGNORE INTO schema_migrations(version, applied_by) VALUES (1, 'causal-lab'); diff --git a/src/app.ts b/src/app.ts index b7708f5..9173bab 100644 --- a/src/app.ts +++ b/src/app.ts @@ -1,23 +1,12 @@ import { Hono } from "hono"; import { bodyLimit } from "hono/body-limit"; +import type { ScenarioCatalog } from "./catalog.js"; import { ContractError } from "./crdt.js"; -import { Simulation, type SimulationConfig } from "./simulation.js"; +import { executeScenario, parseScenario } from "./scenario.js"; export const MAX_HTTP_BODY_BYTES = 2 * 1024 * 1024; -const MAX_STEPS = 1_000; -type ScenarioStep = - | Readonly<{ at: number; action: "put"; replica: string; key: string; value: string }> - | Readonly<{ at: number; action: "remove"; replica: string; key: string }> - | Readonly<{ at: number; action: "partition"; left: string; right: string }> - | Readonly<{ at: number; action: "heal" }>; - -type Scenario = Readonly<{ - config: SimulationConfig; - steps: readonly ScenarioStep[]; -}>; - -export function createApp(): Hono { +export function createApp(catalog?: ScenarioCatalog): Hono { const app = new Hono(); app.post( "/v1/run", @@ -27,29 +16,7 @@ export function createApp(): Hono { }), async (context) => { try { - const scenario = parseScenario(await context.req.json()); - const simulation = new Simulation(scenario.config); - let previousTime = 0; - for (const step of scenario.steps) { - simulation.advance(step.at - previousTime); - previousTime = step.at; - switch (step.action) { - case "put": - simulation.put(step.replica, step.key, step.value); - break; - case "remove": - simulation.remove(step.replica, step.key); - break; - case "partition": - simulation.partition(step.left, step.right); - break; - case "heal": - simulation.healAll(); - break; - } - } - simulation.runUntilIdle(); - return context.json(simulation.report()); + return context.json(executeScenario(parseScenario(await context.req.json()))); } catch (error) { if (error instanceof ContractError || error instanceof SyntaxError) { return context.json({ error: "scenario violates the simulation contract" }, 422); @@ -58,118 +25,48 @@ export function createApp(): Hono { } }, ); - app.notFound((context) => context.json({ error: "route was not found" }, 404)); - app.onError((_error, context) => context.json({ error: "simulation failed" }, 500)); - return app; -} - -function parseScenario(input: unknown): Scenario { - const scenario = expectObject(input, ["config", "steps"], "scenario"); - const configInput = expectObject( - scenario.config, - ["replicas", "seed", "minLatency", "maxLatency", "dropRate", "duplicateRate"], - "config", - ); - if (!Array.isArray(configInput.replicas) || !configInput.replicas.every(isString)) { - throw new ContractError("replicas must be strings"); - } - if (!Array.isArray(scenario.steps) || scenario.steps.length > MAX_STEPS) { - throw new ContractError(`steps must contain at most ${MAX_STEPS} entries`); - } - const config: SimulationConfig = { - replicas: configInput.replicas, - seed: expectNumber(configInput.seed, "seed"), - minLatency: expectNumber(configInput.minLatency, "minLatency"), - maxLatency: expectNumber(configInput.maxLatency, "maxLatency"), - dropRate: expectNumber(configInput.dropRate, "dropRate"), - duplicateRate: expectNumber(configInput.duplicateRate, "duplicateRate"), - }; - - let previousTime = 0; - const steps = scenario.steps.map((inputStep): ScenarioStep => { - const base = expectObject(inputStep, undefined, "step"); - const action = base.action; - if (!isAction(action)) { - throw new ContractError("step action is invalid"); - } - const allowed = { - put: ["at", "action", "replica", "key", "value"], - remove: ["at", "action", "replica", "key"], - partition: ["at", "action", "left", "right"], - heal: ["at", "action"], - }[action]; - const step = expectObject(inputStep, allowed, "step"); - const at = expectNumber(step.at, "step time"); - if (!Number.isSafeInteger(at) || at < previousTime || at > 1_000_000_000) { - throw new ContractError("step times must be non-decreasing safe integers"); - } - previousTime = at; - switch (action) { - case "put": - return { - at, - action, - replica: expectString(step.replica, "replica"), - key: expectString(step.key, "key"), - value: expectString(step.value, "value"), - }; - case "remove": - return { - at, - action, - replica: expectString(step.replica, "replica"), - key: expectString(step.key, "key"), - }; - case "partition": - return { - at, - action, - left: expectString(step.left, "left replica"), - right: expectString(step.right, "right replica"), - }; - case "heal": - return { at, action }; - } - }); - return { config, steps }; -} - -function expectObject( - value: unknown, - allowed: readonly string[] | undefined, - field: string, -): Record { - if (typeof value !== "object" || value === null || Array.isArray(value)) { - throw new ContractError(`${field} must be an object`); - } - const object = value as Record; - if (allowed !== undefined) { - const allowedFields = new Set(allowed); - if (Object.keys(object).some((key) => !allowedFields.has(key))) { - throw new ContractError(`${field} contains an unknown field`); - } - } - return object; -} - -function expectString(value: unknown, field: string): string { - if (!isString(value)) { - throw new ContractError(`${field} must be a string`); - } - return value; -} -function expectNumber(value: unknown, field: string): number { - if (typeof value !== "number" || !Number.isFinite(value)) { - throw new ContractError(`${field} must be a finite number`); + if (catalog !== undefined) { + app.post( + "/v1/scenarios", + bodyLimit({ + maxSize: MAX_HTTP_BODY_BYTES, + onError: (context) => context.json({ error: "request body is too large" }, 413), + }), + async (context) => { + try { + const stored = catalog.putScenario(parseScenario(await context.req.json())); + return context.json(stored, 201); + } catch (error) { + if (error instanceof ContractError || error instanceof SyntaxError) { + return context.json({ error: "scenario violates the simulation contract" }, 422); + } + throw error; + } + }, + ); + app.get("/v1/scenarios/:id", (context) => { + const scenario = catalog.getScenario(context.req.param("id")); + return scenario === undefined + ? context.json({ error: "scenario was not found" }, 404) + : context.json(scenario); + }); + app.post("/v1/scenarios/:id/runs", (context) => { + const scenario = catalog.getScenario(context.req.param("id")); + if (scenario === undefined) { + return context.json({ error: "scenario was not found" }, 404); + } + return context.json(catalog.putRun(scenario.id, executeScenario(scenario.scenario)), 201); + }); + app.get("/v1/runs/:id", (context) => { + const run = catalog.getRun(context.req.param("id")); + return run === undefined + ? context.json({ error: "run was not found" }, 404) + : context.json(run); + }); } - return value; -} - -function isString(value: unknown): value is string { - return typeof value === "string"; -} -function isAction(value: unknown): value is ScenarioStep["action"] { - return isString(value) && ["put", "remove", "partition", "heal"].includes(value); + app.notFound((context) => context.json({ error: "route was not found" }, 404)); + app.onError((_error, context) => context.json({ error: "simulation failed" }, 500)); + return app; } diff --git a/src/catalog.ts b/src/catalog.ts new file mode 100644 index 0000000..49b6760 --- /dev/null +++ b/src/catalog.ts @@ -0,0 +1,222 @@ +import { createHash } from "node:crypto"; +import { chmodSync, existsSync, lstatSync, mkdirSync, readFileSync } from "node:fs"; +import { dirname } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { + CANONICALIZATION, + canonicalJson, + executeScenario, + parseScenario, + scenarioIdentity, + type Scenario, +} from "./scenario.js"; +import type { SimulationReport } from "./simulation.js"; + +const HASH_PATTERN = /^[0-9a-f]{64}$/; +const MIGRATION = readFileSync( + new URL("../schema/sqlite/001_catalog.sql", import.meta.url), + "utf8", +); + +export type ScenarioReference = Readonly<{ + id: string; + canonicalization: typeof CANONICALIZATION; +}>; + +export type StoredScenario = ScenarioReference & Readonly<{ scenario: Scenario }>; + +export type StoredRun = Readonly<{ + id: string; + scenarioId: string; + traceSha256: string; + processedEvents: number; + converged: boolean; + report: SimulationReport; +}>; + +export class ScenarioCatalog { + readonly #database: DatabaseSync; + + constructor(path: string) { + if (path.length === 0) { + throw new Error("catalog path cannot be empty"); + } + if (path !== ":memory:") { + if (existsSync(path) && lstatSync(path).isSymbolicLink()) { + throw new Error("catalog path cannot be a symbolic link"); + } + mkdirSync(dirname(path), { mode: 0o700, recursive: true }); + } + this.#database = new DatabaseSync(path, { + allowExtension: false, + enableDoubleQuotedStringLiterals: false, + enableForeignKeyConstraints: true, + timeout: 5_000, + }); + try { + this.#database.exec("PRAGMA journal_mode = WAL; PRAGMA synchronous = FULL;"); + this.#migrate(); + this.assertHealthy(); + if (path !== ":memory:") { + chmodSync(path, 0o600); + } + } catch (error) { + this.#database.close(); + throw error; + } + } + + putScenario(scenario: Scenario): ScenarioReference { + executeScenario(scenario); + const identity = scenarioIdentity(scenario); + this.#database + .prepare( + "INSERT OR IGNORE INTO scenarios " + + "(scenario_id, canonicalization, canonical_json, definition_bytes) VALUES (?, ?, ?, ?)", + ) + .run(identity.id, CANONICALIZATION, identity.canonical, Buffer.byteLength(identity.canonical)); + const row = this.#database + .prepare("SELECT canonicalization, canonical_json FROM scenarios WHERE scenario_id = ?") + .get(identity.id) as { canonicalization: string; canonical_json: string } | undefined; + if ( + row?.canonicalization !== CANONICALIZATION || + row.canonical_json !== identity.canonical + ) { + throw new Error("scenario digest collision or catalog corruption"); + } + return { id: identity.id, canonicalization: CANONICALIZATION }; + } + + getScenario(id: string): StoredScenario | undefined { + if (!HASH_PATTERN.test(id)) { + return undefined; + } + const row = this.#database + .prepare( + "SELECT canonicalization, canonical_json, definition_bytes " + + "FROM scenarios WHERE scenario_id = ?", + ) + .get(id) as + | { canonicalization: string; canonical_json: string; definition_bytes: number } + | undefined; + if (row === undefined) { + return undefined; + } + if (row.canonicalization !== CANONICALIZATION) { + throw new Error("catalog canonicalization is not supported"); + } + if (Buffer.byteLength(row.canonical_json) !== row.definition_bytes) { + throw new Error("stored scenario byte count does not match its content"); + } + const scenario = parseScenario(JSON.parse(row.canonical_json) as unknown); + if (scenarioIdentity(scenario).id !== id) { + throw new Error("stored scenario failed its content identity check"); + } + return { id, canonicalization: CANONICALIZATION, scenario }; + } + + putRun(scenarioId: string, report: SimulationReport): StoredRun { + if (this.getScenario(scenarioId) === undefined) { + throw new Error("scenario does not exist"); + } + const reportJson = canonicalJson(report); + const traceSha256 = digest(canonicalJson(report.trace)); + const id = digest(`${scenarioId}\0${reportJson}`); + this.#database + .prepare( + "INSERT OR IGNORE INTO runs " + + "(run_id, scenario_id, report_json, trace_sha256, processed_events, converged) " + + "VALUES (?, ?, ?, ?, ?, ?)", + ) + .run(id, scenarioId, reportJson, traceSha256, report.processedEvents, report.converged ? 1 : 0); + const stored = this.getRun(id); + if ( + stored === undefined || + stored.scenarioId !== scenarioId || + canonicalJson(stored.report) !== reportJson + ) { + throw new Error("run digest collision or catalog corruption"); + } + return stored; + } + + getRun(id: string): StoredRun | undefined { + if (!HASH_PATTERN.test(id)) { + return undefined; + } + const row = this.#database + .prepare( + "SELECT scenario_id, report_json, trace_sha256, processed_events, converged " + + "FROM runs WHERE run_id = ?", + ) + .get(id) as + | { + scenario_id: string; + report_json: string; + trace_sha256: string; + processed_events: number; + converged: number; + } + | undefined; + if (row === undefined) { + return undefined; + } + const report = JSON.parse(row.report_json) as SimulationReport; + if ( + digest(`${row.scenario_id}\0${row.report_json}`) !== id || + digest(canonicalJson(report.trace)) !== row.trace_sha256 || + report.processedEvents !== row.processed_events || + report.converged !== (row.converged === 1) + ) { + throw new Error("stored run failed its receipt checks"); + } + return { + id, + scenarioId: row.scenario_id, + traceSha256: row.trace_sha256, + processedEvents: row.processed_events, + converged: row.converged === 1, + report, + }; + } + + assertHealthy(): void { + const migration = this.#database + .prepare("SELECT COALESCE(MAX(version), 0) AS version FROM schema_migrations") + .get() as { version: number }; + if (migration.version !== 1) { + throw new Error("catalog schema version is not supported"); + } + const integrity = this.#database.prepare("PRAGMA integrity_check").all() as Array<{ + integrity_check: string; + }>; + if (integrity.length !== 1 || integrity[0]?.integrity_check !== "ok") { + throw new Error("catalog integrity check failed"); + } + const foreignKeys = this.#database.prepare("PRAGMA foreign_key_check").all(); + if (foreignKeys.length !== 0) { + throw new Error("catalog foreign-key check failed"); + } + } + + close(): void { + if (this.#database.isOpen) { + this.#database.close(); + } + } + + #migrate(): void { + this.#database.exec("BEGIN IMMEDIATE"); + try { + this.#database.exec(MIGRATION); + this.#database.exec("COMMIT"); + } catch (error) { + this.#database.exec("ROLLBACK"); + throw error; + } + } +} + +function digest(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} diff --git a/src/index.ts b/src/index.ts index 3aab0c8..cd08131 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,3 +1,5 @@ export * from "./app.js"; +export * from "./catalog.js"; export * from "./crdt.js"; +export * from "./scenario.js"; export * from "./simulation.js"; diff --git a/src/scenario.ts b/src/scenario.ts new file mode 100644 index 0000000..90e2500 --- /dev/null +++ b/src/scenario.ts @@ -0,0 +1,186 @@ +import { createHash } from "node:crypto"; +import { ContractError } from "./crdt.js"; +import { Simulation, type SimulationConfig, type SimulationReport } from "./simulation.js"; + +export const MAX_STEPS = 1_000; +export const CANONICALIZATION = "causal-lab-json-v1"; + +export type ScenarioStep = + | Readonly<{ at: number; action: "put"; replica: string; key: string; value: string }> + | Readonly<{ at: number; action: "remove"; replica: string; key: string }> + | Readonly<{ at: number; action: "partition"; left: string; right: string }> + | Readonly<{ at: number; action: "heal" }>; + +export type Scenario = Readonly<{ + config: SimulationConfig; + steps: readonly ScenarioStep[]; +}>; + +export function parseScenario(input: unknown): Scenario { + const scenario = expectObject(input, ["config", "steps"], "scenario"); + const configInput = expectObject( + scenario.config, + ["replicas", "seed", "minLatency", "maxLatency", "dropRate", "duplicateRate"], + "config", + ); + if (!Array.isArray(configInput.replicas) || !configInput.replicas.every(isString)) { + throw new ContractError("replicas must be strings"); + } + if (!Array.isArray(scenario.steps) || scenario.steps.length > MAX_STEPS) { + throw new ContractError(`steps must contain at most ${MAX_STEPS} entries`); + } + const config: SimulationConfig = { + replicas: [...configInput.replicas], + seed: expectNumber(configInput.seed, "seed"), + minLatency: expectNumber(configInput.minLatency, "minLatency"), + maxLatency: expectNumber(configInput.maxLatency, "maxLatency"), + dropRate: expectNumber(configInput.dropRate, "dropRate"), + duplicateRate: expectNumber(configInput.duplicateRate, "duplicateRate"), + }; + + let previousTime = 0; + const steps = scenario.steps.map((inputStep): ScenarioStep => { + const base = expectObject(inputStep, undefined, "step"); + const action = base.action; + if (!isAction(action)) { + throw new ContractError("step action is invalid"); + } + const allowed = { + put: ["at", "action", "replica", "key", "value"], + remove: ["at", "action", "replica", "key"], + partition: ["at", "action", "left", "right"], + heal: ["at", "action"], + }[action]; + const step = expectObject(inputStep, allowed, "step"); + const at = expectNumber(step.at, "step time"); + if (!Number.isSafeInteger(at) || at < previousTime || at > 1_000_000_000) { + throw new ContractError("step times must be non-decreasing safe integers"); + } + previousTime = at; + switch (action) { + case "put": + return { + at, + action, + replica: expectString(step.replica, "replica"), + key: expectString(step.key, "key"), + value: expectString(step.value, "value"), + }; + case "remove": + return { + at, + action, + replica: expectString(step.replica, "replica"), + key: expectString(step.key, "key"), + }; + case "partition": + return { + at, + action, + left: expectString(step.left, "left replica"), + right: expectString(step.right, "right replica"), + }; + case "heal": + return { at, action }; + } + }); + return { config, steps }; +} + +export function executeScenario(scenario: Scenario): SimulationReport { + const simulation = new Simulation(scenario.config); + let previousTime = 0; + for (const step of scenario.steps) { + simulation.advance(step.at - previousTime); + previousTime = step.at; + switch (step.action) { + case "put": + simulation.put(step.replica, step.key, step.value); + break; + case "remove": + simulation.remove(step.replica, step.key); + break; + case "partition": + simulation.partition(step.left, step.right); + break; + case "heal": + simulation.healAll(); + break; + } + } + simulation.runUntilIdle(); + return simulation.report(); +} + +export function canonicalJson(value: unknown): string { + if (value === null || typeof value === "boolean" || typeof value === "string") { + return JSON.stringify(value); + } + if (typeof value === "number") { + if (!Number.isFinite(value)) { + throw new ContractError("canonical JSON requires finite numbers"); + } + return JSON.stringify(value); + } + if (Array.isArray(value)) { + return `[${value.map(canonicalJson).join(",")}]`; + } + if (typeof value === "object") { + const record = value as Record; + return `{${Object.keys(record) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonicalJson(record[key])}`) + .join(",")}}`; + } + throw new ContractError("value is not representable as canonical JSON"); +} + +export function scenarioIdentity(scenario: Scenario): Readonly<{ id: string; canonical: string }> { + const canonical = canonicalJson(scenario); + const id = createHash("sha256") + .update(CANONICALIZATION) + .update("\0") + .update(canonical) + .digest("hex"); + return { id, canonical }; +} + +function expectObject( + value: unknown, + allowed: readonly string[] | undefined, + field: string, +): Record { + if (typeof value !== "object" || value === null || Array.isArray(value)) { + throw new ContractError(`${field} must be an object`); + } + const object = value as Record; + if (allowed !== undefined) { + const allowedFields = new Set(allowed); + if (Object.keys(object).some((key) => !allowedFields.has(key))) { + throw new ContractError(`${field} contains an unknown field`); + } + } + return object; +} + +function expectString(value: unknown, field: string): string { + if (!isString(value)) { + throw new ContractError(`${field} must be a string`); + } + return value; +} + +function expectNumber(value: unknown, field: string): number { + if (typeof value !== "number" || !Number.isFinite(value)) { + throw new ContractError(`${field} must be a finite number`); + } + return value; +} + +function isString(value: unknown): value is string { + return typeof value === "string"; +} + +function isAction(value: unknown): value is ScenarioStep["action"] { + return isString(value) && ["put", "remove", "partition", "heal"].includes(value); +} diff --git a/src/server.ts b/src/server.ts index 64ac389..c3c8952 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,6 +1,7 @@ #!/usr/bin/env node import { serve } from "@hono/node-server"; import { createApp } from "./app.js"; +import { ScenarioCatalog } from "./catalog.js"; const hostname = process.env.CAUSAL_LAB_HOST ?? "127.0.0.1"; const port = Number(process.env.CAUSAL_LAB_PORT ?? "8787"); @@ -14,11 +15,14 @@ if (!Number.isSafeInteger(port) || port < 1 || port > 65_535) { process.exit(2); } -const server = serve({ fetch: createApp().fetch, hostname, port }); +const databasePath = process.env.CAUSAL_LAB_DB; +const catalog = databasePath === undefined ? undefined : new ScenarioCatalog(databasePath); +const server = serve({ fetch: createApp(catalog).fetch, hostname, port }); console.log(`causal-lab listening on http://${hostname}:${port}`); const shutdown = (): void => { server.close((error) => { + catalog?.close(); if (error !== undefined) { console.error("causal-lab shutdown failed"); process.exitCode = 1; diff --git a/test/app.test.ts b/test/app.test.ts index bbdbb37..3106cbf 100644 --- a/test/app.test.ts +++ b/test/app.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "vitest"; import { createApp } from "../src/app.js"; +import { ScenarioCatalog } from "../src/catalog.js"; describe("scenario API", () => { it("runs a bounded scenario and returns a convergence report", async () => { @@ -39,4 +40,61 @@ describe("scenario API", () => { }); expect(response.status).toBe(422); }); + + it("stores, runs, and reads a content-addressed scenario when a catalog is configured", async () => { + const catalog = new ScenarioCatalog(":memory:"); + const app = createApp(catalog); + const created = await app.request("/v1/scenarios", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(sampleScenario()), + }); + expect(created.status).toBe(201); + const reference = (await created.json()) as { id: string }; + + const fetched = await app.request(`/v1/scenarios/${reference.id}`); + expect(fetched.status).toBe(200); + + const executed = await app.request(`/v1/scenarios/${reference.id}/runs`, { method: "POST" }); + expect(executed.status).toBe(201); + const run = (await executed.json()) as { id: string; traceSha256: string }; + expect(run.traceSha256).toMatch(/^[0-9a-f]{64}$/); + expect((await app.request(`/v1/runs/${run.id}`)).status).toBe(200); + catalog.close(); + }); + + it("does not catalog a scenario that fails semantic simulation validation", async () => { + const catalog = new ScenarioCatalog(":memory:"); + const app = createApp(catalog); + const invalid = sampleScenario() as { + config: { replicas: string[] }; + steps: Array>; + }; + invalid.steps = [{ at: 0, action: "put", replica: "missing", key: "mode", value: "safe" }]; + + const response = await app.request("/v1/scenarios", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(invalid), + }); + expect(response.status).toBe(422); + catalog.close(); + }); }); + +function sampleScenario(): unknown { + return { + config: { + replicas: ["a", "b"], + seed: 7, + minLatency: 1, + maxLatency: 2, + dropRate: 0, + duplicateRate: 0, + }, + steps: [ + { at: 0, action: "put", replica: "a", key: "mode", value: "safe" }, + { at: 3, action: "heal" }, + ], + }; +} diff --git a/test/catalog.test.ts b/test/catalog.test.ts new file mode 100644 index 0000000..aa381b9 --- /dev/null +++ b/test/catalog.test.ts @@ -0,0 +1,64 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { afterEach, describe, expect, it } from "vitest"; +import { ScenarioCatalog } from "../src/catalog.js"; +import { executeScenario, parseScenario } from "../src/scenario.js"; + +const directories: string[] = []; + +afterEach(() => { + for (const directory of directories.splice(0)) { + rmSync(directory, { force: true, recursive: true }); + } +}); + +describe("scenario catalog", () => { + it("persists content-addressed scenarios and immutable run receipts across reopen", () => { + const directory = mkdtempSync(join(tmpdir(), "causal-lab-")); + directories.push(directory); + const path = join(directory, "catalog.sqlite"); + const scenario = parseScenario(sampleScenario()); + + const first = new ScenarioCatalog(path); + const reference = first.putScenario(scenario); + const repeated = first.putScenario(scenario); + expect(repeated).toEqual(reference); + const run = first.putRun(reference.id, executeScenario(scenario)); + expect(first.putRun(reference.id, executeScenario(scenario))).toEqual(run); + first.assertHealthy(); + first.close(); + + const reopened = new ScenarioCatalog(path); + expect(reopened.getScenario(reference.id)?.scenario).toEqual(scenario); + expect(reopened.getRun(run.id)).toEqual(run); + expect(reopened.getRun("not-a-digest")).toBeUndefined(); + reopened.close(); + + const database = new DatabaseSync(path); + expect(() => database.prepare("UPDATE scenarios SET definition_bytes = 1").run()).toThrow( + /immutable/, + ); + expect(database.prepare("PRAGMA foreign_key_check").all()).toEqual([]); + expect(database.prepare("PRAGMA integrity_check").get()).toEqual({ integrity_check: "ok" }); + database.close(); + }); +}); + +function sampleScenario(): unknown { + return { + config: { + replicas: ["a", "b"], + seed: 7, + minLatency: 1, + maxLatency: 2, + dropRate: 0, + duplicateRate: 0, + }, + steps: [ + { at: 0, action: "put", replica: "a", key: "mode", value: "safe" }, + { at: 3, action: "heal" }, + ], + }; +} diff --git a/test/scenario.test.ts b/test/scenario.test.ts new file mode 100644 index 0000000..a60137f --- /dev/null +++ b/test/scenario.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from "vitest"; +import { executeScenario, parseScenario, scenarioIdentity } from "../src/scenario.js"; + +const scenarioInput = { + config: { + replicas: ["a", "b"], + seed: 19, + minLatency: 1, + maxLatency: 3, + dropRate: 0.25, + duplicateRate: 0.5, + }, + steps: [ + { at: 0, action: "partition", left: "a", right: "b" }, + { at: 1, action: "put", replica: "a", key: "mode", value: "safe" }, + { at: 5, action: "heal" }, + ], +}; + +describe("reproducible scenarios", () => { + it("assigns the same identity regardless of object key order", () => { + const reordered = { + steps: scenarioInput.steps.map((step) => Object.fromEntries(Object.entries(step).reverse())), + config: Object.fromEntries(Object.entries(scenarioInput.config).reverse()), + }; + + const first = parseScenario(scenarioInput); + const second = parseScenario(reordered); + expect(scenarioIdentity(first)).toEqual(scenarioIdentity(second)); + }); + + it("executes the same validated scenario byte-identically", () => { + const scenario = parseScenario(scenarioInput); + expect(executeScenario(scenario)).toEqual(executeScenario(scenario)); + }); +});