diff --git a/.changeset/bootstrap-sources.md b/.changeset/bootstrap-sources.md index 7952067..2c52014 100644 --- a/.changeset/bootstrap-sources.md +++ b/.changeset/bootstrap-sources.md @@ -2,4 +2,4 @@ "@atmo-dev/contrail": minor --- -Add source-neutral snapshot and ordered-change contracts, capture-first bootstrap orchestration, and a database-backed target that commits projection progress atomically for fresh generations. Persist source continuity epochs and the capture mark before snapshot preparation. Add resumable, host-aware relay/PDS snapshots plus source-confirmed Jetstream marks, bounded ordered replay, retention-expiry detection, required source-semantics gates, durable bounded failure categories, and aggregate candidate verification before completion. +Add source-neutral snapshot and ordered-change contracts, capture-first bootstrap orchestration, and a database-backed target that commits projection progress atomically for fresh generations. Persist source continuity epochs and the capture mark before snapshot preparation. Add resumable, host-aware relay/PDS snapshots plus source-confirmed Jetstream marks, bounded ordered replay, retention-expiry detection, required source-semantics gates, durable bounded failure categories, aggregate candidate verification, and an immutable deployment-tuple registry with compare-and-swap activation and rollback retention. diff --git a/packages/contrail/README.md b/packages/contrail/README.md index f0eac5b..dd3d69d 100644 --- a/packages/contrail/README.md +++ b/packages/contrail/README.md @@ -72,6 +72,14 @@ After a write to a user's PDS, `contrail.notify(uri)` can fetch the authoritativ Contrail stores source event time, repository revision, source cursor, CID, and local index time separately from record/application time. Durable tombstones reject stale resurrection, and live Jetstream projection commits its exact yielded cursor in the same transaction. A successful PDS `listRecords` page is a current authoritative observation, so it supersedes older durable state without a redundant version read; its version writes and page cursor still commit atomically. Tombstones are retained indefinitely; authoritative rebuild/retention tooling is planned separately. +## Fresh generations (experimental) + +`PdsSnapshotSource`, `JetstreamChangeSource`, `DatabaseBootstrapTarget`, and `bootstrapFreshProjection()` build an unpublished database with capture-first replay. The capture mark is durable before relay discovery starts; PDS partition cursors and Jetstream checkpoints commit with their records. Completion rebuilds deferred projections, verifies aggregate record/version consistency, and stores only bounded failure categories. + +Jetstream generation replay requires an operator-owned continuity epoch and retention guarantee. It uses real stream events as marks, never wall clock or a quiet socket. Optional busy watermark collections can prove progress without being projected. + +A separate control database can use `DatabaseGenerationRegistry` to store immutable `(code, definition, database, generation)` tuples. `activate(candidate, expectedActive)` switches one singleton pointer with compare-and-swap, retaining the previous ready tuple for rollback. There is intentionally no percentage traffic-split API; platform routing must resolve the one active tuple. + ## Runtime record validation Pass the record Lexicons for every configured collection and their transitive references to enable shared strict validation and CID verification: diff --git a/packages/contrail/src/core/generations.ts b/packages/contrail/src/core/generations.ts new file mode 100644 index 0000000..0c8b553 --- /dev/null +++ b/packages/contrail/src/core/generations.ts @@ -0,0 +1,406 @@ +import type { Database } from "./types"; +import { getDialect } from "./dialect"; +import type { SourcePosition } from "./sources"; +import type { BootstrapVerificationReport } from "./verification"; + +const ACTIVE_POINTER_ID = 1; + +export interface GenerationTuple { + /** Stable immutable deployment generation ID. */ + id: string; + /** Digest or immutable version for the executable artifact. */ + codeDigest: string; + /** Digest of the projection/lexicon definition. */ + definitionDigest: string; + /** Platform-owned locator for this generation's dedicated database. */ + databaseLocator: string; + schemaVersion: number; +} + +export interface GenerationReadiness { + through: SourcePosition; + verification: BootstrapVerificationReport; +} + +export type GenerationLifecycleState = + | "candidate" + | "ready" + | "active" + | "retained" + | "retired"; + +export interface GenerationRecord { + tuple: GenerationTuple; + readiness: GenerationReadiness | null; + state: GenerationLifecycleState; + createdAt: number; + readyAt: number | null; + lastActivatedAt: number | null; + retiredAt: number | null; +} + +export interface GenerationActivation { + previous: GenerationRecord | null; + active: GenerationRecord; +} + +interface GenerationRow { + id: string; + code_digest: string; + definition_digest: string; + database_locator: string; + schema_version: number | string; + readiness_json: string | null; + created_at: number | string; + ready_at: number | string | null; + last_activated_at: number | string | null; + retired_at: number | string | null; + active_id: string | null; +} + +function boundedText(value: string, label: string, maximum: number): string { + if (typeof value !== "string" || value.length === 0 || value.length > maximum) { + throw new TypeError(`${label} must contain 1-${maximum} characters`); + } + return value; +} + +function timestamp(value: number | string | null): number | null { + if (value === null) return null; + const parsed = Number(value); + if (!Number.isSafeInteger(parsed) || parsed < 0) { + throw new Error("Invalid durable generation timestamp"); + } + return parsed; +} + +function validateTuple(tuple: GenerationTuple): GenerationTuple { + boundedText(tuple.id, "generation id", 128); + boundedText(tuple.codeDigest, "code digest", 256); + boundedText(tuple.definitionDigest, "definition digest", 256); + boundedText(tuple.databaseLocator, "database locator", 2_048); + if (!Number.isSafeInteger(tuple.schemaVersion) || tuple.schemaVersion < 1) { + throw new TypeError("schemaVersion must be a positive safe integer"); + } + return tuple; +} + +function validateReadiness(readiness: GenerationReadiness): GenerationReadiness { + const { through, verification } = readiness; + boundedText(through.source, "readiness source", 128); + boundedText(through.epoch, "readiness epoch", 256); + boundedText(through.cursor, "readiness cursor", 2_048); + if (!verification.ok) { + throw new Error("A failed bootstrap verification cannot become ready"); + } + if ( + !Number.isSafeInteger(verification.verifiedAt) || + verification.verifiedAt < 0 || + !Array.isArray(verification.checks) || + !verification.checks.every( + (item) => + item && + typeof item.name === "string" && + item.name.length > 0 && + item.name.length <= 256 && + item.ok === true && + Number.isSafeInteger(item.failures) && + item.failures === 0, + ) + ) { + throw new Error("Generation readiness contains malformed verification"); + } + return readiness; +} + +function parseReadiness(serialized: string): GenerationReadiness { + let value: unknown; + try { + value = JSON.parse(serialized); + } catch { + throw new Error("Durable generation readiness is not valid JSON"); + } + if (!value || typeof value !== "object") { + throw new Error("Durable generation readiness is malformed"); + } + return validateReadiness(value as GenerationReadiness); +} + +function record(row: GenerationRow): GenerationRecord { + const tuple = validateTuple({ + id: row.id, + codeDigest: row.code_digest, + definitionDigest: row.definition_digest, + databaseLocator: row.database_locator, + schemaVersion: Number(row.schema_version), + }); + const retiredAt = timestamp(row.retired_at); + const lastActivatedAt = timestamp(row.last_activated_at); + const readiness = row.readiness_json + ? parseReadiness(row.readiness_json) + : null; + const state: GenerationLifecycleState = + retiredAt !== null + ? "retired" + : row.active_id === row.id + ? "active" + : readiness === null + ? "candidate" + : lastActivatedAt === null + ? "ready" + : "retained"; + return { + tuple, + readiness, + state, + createdAt: timestamp(row.created_at)!, + readyAt: timestamp(row.ready_at), + lastActivatedAt, + retiredAt, + }; +} + +function sameTuple(left: GenerationTuple, right: GenerationTuple): boolean { + return ( + left.id === right.id && + left.codeDigest === right.codeDigest && + left.definitionDigest === right.definitionDigest && + left.databaseLocator === right.databaseLocator && + left.schemaVersion === right.schemaVersion + ); +} + +/** Initialize a small control-plane registry. This database is separate from + * candidate projection databases and stores no record bodies or source errors. */ +export async function initGenerationRegistry(db: Database): Promise { + const bigint = getDialect(db).bigintType; + await db + .prepare( + `CREATE TABLE IF NOT EXISTS contrail_generations ( + id TEXT PRIMARY KEY, + code_digest TEXT NOT NULL, + definition_digest TEXT NOT NULL, + database_locator TEXT NOT NULL, + schema_version INTEGER NOT NULL, + readiness_json TEXT, + created_at ${bigint} NOT NULL, + ready_at ${bigint}, + last_activated_at ${bigint}, + retired_at ${bigint} + )`, + ) + .run(); + await db + .prepare( + `CREATE TABLE IF NOT EXISTS contrail_generation_activation ( + id INTEGER PRIMARY KEY CHECK (id = 1), + generation_id TEXT, + activated_at ${bigint}, + FOREIGN KEY (generation_id) REFERENCES contrail_generations(id) + )`, + ) + .run(); + await db + .prepare( + `INSERT INTO contrail_generation_activation (id, generation_id) + VALUES (?, NULL) ON CONFLICT(id) DO NOTHING`, + ) + .bind(ACTIVE_POINTER_ID) + .run(); +} + +/** Durable compare-and-swap registry for complete deployment tuples. + * + * Request routing must resolve this one active pointer; this API deliberately + * exposes no percentage split between independent generation databases. */ +export class DatabaseGenerationRegistry { + constructor(private readonly db: Database) {} + + async registerCandidate(tuple: GenerationTuple): Promise { + validateTuple(tuple); + await this.db + .prepare( + `INSERT INTO contrail_generations + (id, code_digest, definition_digest, database_locator, schema_version, + created_at) + VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO NOTHING`, + ) + .bind( + tuple.id, + tuple.codeDigest, + tuple.definitionDigest, + tuple.databaseLocator, + tuple.schemaVersion, + Date.now(), + ) + .run(); + const stored = await this.get(tuple.id); + if (!stored || !sameTuple(stored.tuple, tuple)) { + throw new Error(`Generation ${tuple.id} already names another tuple`); + } + return stored; + } + + async markReady( + id: string, + readiness: GenerationReadiness, + ): Promise { + boundedText(id, "generation id", 128); + validateReadiness(readiness); + const serialized = JSON.stringify(readiness); + await this.db + .prepare( + `UPDATE contrail_generations + SET readiness_json = ?, ready_at = ? + WHERE id = ? AND readiness_json IS NULL AND retired_at IS NULL`, + ) + .bind(serialized, Date.now(), id) + .run(); + const stored = await this.get(id); + if (!stored) throw new Error(`Unknown generation ${id}`); + if (stored.state === "retired") { + throw new Error(`Retired generation ${id} cannot become ready`); + } + if (JSON.stringify(stored.readiness) !== serialized) { + throw new Error(`Generation ${id} already has different readiness proof`); + } + return stored; + } + + async get(id: string): Promise { + boundedText(id, "generation id", 128); + const row = await this.db + .prepare( + `SELECT generation.*, activation.generation_id AS active_id + FROM contrail_generations AS generation + LEFT JOIN contrail_generation_activation AS activation + ON activation.id = ? + WHERE generation.id = ?`, + ) + .bind(ACTIVE_POINTER_ID, id) + .first(); + return row ? record(row) : null; + } + + async active(): Promise { + const row = await this.db + .prepare( + `SELECT generation.*, activation.generation_id AS active_id + FROM contrail_generation_activation AS activation + JOIN contrail_generations AS generation + ON generation.id = activation.generation_id + WHERE activation.id = ?`, + ) + .bind(ACTIVE_POINTER_ID) + .first(); + return row ? record(row) : null; + } + + /** Atomically switch the complete tuple if the caller still sees the expected + * active generation. The old tuple remains ready for explicit rollback. */ + async activate( + candidateId: string, + expectedActiveId: string | null, + ): Promise { + boundedText(candidateId, "candidate generation id", 128); + if (expectedActiveId !== null) { + boundedText(expectedActiveId, "expected generation id", 128); + } + const previous = await this.active(); + if ((previous?.tuple.id ?? null) !== expectedActiveId) { + throw new Error("Active generation changed before activation"); + } + + const switched = await this.db + .prepare( + `UPDATE contrail_generation_activation + SET generation_id = ?, activated_at = ? + WHERE id = ? + AND ((generation_id = ?) OR + (generation_id IS NULL AND CAST(? AS TEXT) IS NULL)) + AND EXISTS ( + SELECT 1 FROM contrail_generations + WHERE id = ? AND readiness_json IS NOT NULL AND retired_at IS NULL + ) + RETURNING generation_id`, + ) + .bind( + candidateId, + Date.now(), + ACTIVE_POINTER_ID, + expectedActiveId, + expectedActiveId, + candidateId, + ) + .first<{ generation_id: string }>(); + if (switched?.generation_id !== candidateId) { + const candidate = await this.get(candidateId); + if (!candidate) throw new Error(`Unknown generation ${candidateId}`); + if (candidate.state === "candidate") { + throw new Error(`Generation ${candidateId} is not ready`); + } + if (candidate.state === "retired") { + throw new Error(`Generation ${candidateId} is retired`); + } + throw new Error("Active generation changed during activation"); + } + + // The pointer switch above is the authoritative atomic action. This field + // only distinguishes retained generations in operator listings, so a + // bookkeeping failure must not misreport a successful activation as failed. + try { + await this.db + .prepare( + `UPDATE contrail_generations SET last_activated_at = ? + WHERE id = ?`, + ) + .bind(Date.now(), candidateId) + .run(); + } catch { + // Best-effort metadata; active() still resolves the switched tuple. + } + const activated = await this.get(candidateId); + if (!activated) throw new Error("Activated generation could not be resolved"); + // A later concurrent activation may already have moved the pointer again; + // the successful CAS still activated this tuple at its linearization point. + return { previous, active: { ...activated, state: "active" } }; + } + + async retire(id: string): Promise { + boundedText(id, "generation id", 128); + await this.db + .prepare( + `UPDATE contrail_generations SET retired_at = ? + WHERE id = ? AND retired_at IS NULL + AND id <> COALESCE( + (SELECT generation_id FROM contrail_generation_activation WHERE id = ?), + '' + )`, + ) + .bind(Date.now(), id, ACTIVE_POINTER_ID) + .run(); + const stored = await this.get(id); + if (!stored) throw new Error(`Unknown generation ${id}`); + if (stored.state === "active") { + throw new Error(`Active generation ${id} cannot be retired`); + } + if (stored.state !== "retired") { + throw new Error(`Generation ${id} could not be retired`); + } + return stored; + } + + async list(): Promise { + const rows = await this.db + .prepare( + `SELECT generation.*, activation.generation_id AS active_id + FROM contrail_generations AS generation + LEFT JOIN contrail_generation_activation AS activation + ON activation.id = ? + ORDER BY generation.created_at DESC, generation.id`, + ) + .bind(ACTIVE_POINTER_ID) + .all(); + return (rows.results ?? []).map(record); + } +} diff --git a/packages/contrail/src/index.ts b/packages/contrail/src/index.ts index c1d828e..874ea50 100644 --- a/packages/contrail/src/index.ts +++ b/packages/contrail/src/index.ts @@ -20,6 +20,7 @@ export * from "./core/ingest"; export * from "./core/sources"; export * from "./core/bootstrap"; export * from "./core/verification"; +export * from "./core/generations"; export * from "./core/pds-snapshot"; export * from "./core/jetstream-source"; export * from "./core/jetstream"; diff --git a/packages/contrail/tests/generation-registry.test.ts b/packages/contrail/tests/generation-registry.test.ts new file mode 100644 index 0000000..250c4d2 --- /dev/null +++ b/packages/contrail/tests/generation-registry.test.ts @@ -0,0 +1,121 @@ +import { describe, expect, it } from "vitest"; +import { + DatabaseGenerationRegistry, + initGenerationRegistry, + type GenerationReadiness, + type GenerationTuple, +} from "../src/index"; +import { createSqliteDatabase } from "../src/adapters/sqlite"; + +function tuple(id: string): GenerationTuple { + return { + id, + codeDigest: `code-${id}`, + definitionDigest: `definition-${id}`, + databaseLocator: `database-${id}`, + schemaVersion: 9, + }; +} + +function readiness(cursor: string): GenerationReadiness { + return { + through: { source: "jetstream", epoch: "epoch-one", cursor }, + verification: { + ok: true, + verifiedAt: Date.now(), + checks: [ + { name: "snapshot-partitions", ok: true, failures: 0 }, + { name: "visible-version:event", ok: true, failures: 0 }, + ], + }, + }; +} + +describe("database generation registry", () => { + it("atomically activates complete tuples and keeps the previous one for rollback", async () => { + const db = createSqliteDatabase(":memory:"); + await initGenerationRegistry(db); + const registry = new DatabaseGenerationRegistry(db); + + expect((await registry.registerCandidate(tuple("one"))).state).toBe( + "candidate", + ); + expect((await registry.markReady("one", readiness("10"))).state).toBe( + "ready", + ); + const first = await registry.activate("one", null); + expect(first.previous).toBeNull(); + expect(first.active).toMatchObject({ + state: "active", + tuple: tuple("one"), + }); + + await registry.registerCandidate(tuple("two")); + await registry.markReady("two", readiness("20")); + const second = await registry.activate("two", "one"); + expect(second.previous?.tuple.id).toBe("one"); + expect(second.active.tuple).toEqual(tuple("two")); + expect((await registry.get("one"))?.state).toBe("retained"); + expect((await registry.active())?.tuple).toEqual(tuple("two")); + + const rollback = await registry.activate("one", "two"); + expect(rollback.previous?.tuple.id).toBe("two"); + expect(rollback.active.tuple.id).toBe("one"); + expect((await registry.get("two"))?.state).toBe("retained"); + + await registry.retire("two"); + expect((await registry.get("two"))?.state).toBe("retired"); + expect((await registry.active())?.tuple.id).toBe("one"); + }); + + it("rejects stale, incomplete, retired, and tuple-changing activations", async () => { + const db = createSqliteDatabase(":memory:"); + await initGenerationRegistry(db); + const registry = new DatabaseGenerationRegistry(db); + await registry.registerCandidate(tuple("one")); + await registry.markReady("one", readiness("10")); + await registry.activate("one", null); + + await expect(registry.activate("one", null)).rejects.toThrow( + "changed before activation", + ); + await expect(registry.retire("one")).rejects.toThrow("cannot be retired"); + + await registry.registerCandidate(tuple("candidate")); + await expect(registry.activate("candidate", "one")).rejects.toThrow( + "not ready", + ); + expect((await registry.active())?.tuple.id).toBe("one"); + + await expect( + registry.registerCandidate({ + ...tuple("one"), + databaseLocator: "another-database", + }), + ).rejects.toThrow("already names another tuple"); + + await registry.retire("candidate"); + await expect( + registry.markReady("candidate", readiness("30")), + ).rejects.toThrow("cannot become ready"); + }); + + it("requires a successful aggregate verification proof before readiness", async () => { + const db = createSqliteDatabase(":memory:"); + await initGenerationRegistry(db); + const registry = new DatabaseGenerationRegistry(db); + await registry.registerCandidate(tuple("failed")); + const proof = readiness("10"); + proof.verification.ok = false; + proof.verification.checks[0] = { + name: "snapshot-partitions", + ok: false, + failures: 1, + }; + + await expect(registry.markReady("failed", proof)).rejects.toThrow( + "failed bootstrap verification", + ); + expect((await registry.get("failed"))?.state).toBe("candidate"); + }); +});