From da5be31099c627ac0fc6a0e8b9b53a73e5e735a4 Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Thu, 6 Aug 2026 03:18:42 +0200 Subject: [PATCH] Persist fresh bootstrap progress --- .changeset/bootstrap-sources.md | 2 +- packages/contrail/src/core/bootstrap.ts | 388 ++++++++++++++++++ packages/contrail/src/core/db/records.ts | 15 +- packages/contrail/src/core/db/schema.ts | 39 +- packages/contrail/src/core/ingest.ts | 3 + packages/contrail/src/core/sources.ts | 30 +- packages/contrail/src/core/types.ts | 4 +- packages/contrail/src/index.ts | 1 + .../contrail/tests/bootstrap-sources.test.ts | 65 ++- .../tests/database-bootstrap-target.test.ts | 239 +++++++++++ .../contrail/tests/source-ordering.test.ts | 53 +++ 11 files changed, 809 insertions(+), 30 deletions(-) create mode 100644 packages/contrail/src/core/bootstrap.ts create mode 100644 packages/contrail/tests/database-bootstrap-target.test.ts diff --git a/.changeset/bootstrap-sources.md b/.changeset/bootstrap-sources.md index e7bc460..65c7198 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 plus capture-first bootstrap orchestration for fresh projection generations. Anchor the legacy PDS discovery path before its first relay request so newly created repositories remain recoverable through Jetstream replay. +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 anchor legacy PDS discovery before its first relay request so newly created repositories remain recoverable through Jetstream replay. diff --git a/packages/contrail/src/core/bootstrap.ts b/packages/contrail/src/core/bootstrap.ts new file mode 100644 index 0000000..5a8f388 --- /dev/null +++ b/packages/contrail/src/core/bootstrap.ts @@ -0,0 +1,388 @@ +import type { ContrailConfig, Database, IngestEvent, Statement } from "./types"; +import { recordTimeUs, createIngestEvent, ingestRecords } from "./ingest"; +import { getDependentNsids, recordsTableName } from "./types"; +import { rebuildDerivedProjections } from "./db/records"; +import type { + BootstrapRunState, + BootstrapTarget, + MutationBatch, + PreparedSnapshot, + SnapshotBatch, + SourceMutation, + SourcePosition, +} from "./sources"; + +const BOOTSTRAP_STATE_ID = 1; + +interface BootstrapStateRow { + phase: BootstrapRunState["phase"]; + snapshot_json: string; + capture_source: string; + capture_epoch: string; + capture_cursor: string; + snapshot_complete: number; + catchup_source: string | null; + catchup_epoch: string | null; + catchup_cursor: string | null; + change_source: string | null; + change_epoch: string | null; + change_cursor: string | null; +} + +interface BootstrapProgressRow { + partition: string; + cursor: string | null; + completed: number; +} + +export interface DatabaseBootstrapTargetOptions { + /** Skip FTS/count maintenance while loading and rebuild it before complete. */ + deferDerivedProjections?: boolean; + /** Additional actors already known to be in acquisition scope. */ + knownDids?: ReadonlySet; +} + +function position( + source: string | null, + epoch: string | null, + cursor: string | null, + label: string, +): SourcePosition | null { + if (source === null && epoch === null && cursor === null) return null; + if (source === null || epoch === null || cursor === null) { + throw new Error(`Incomplete durable ${label} position`); + } + return { source, epoch, cursor }; +} + +function isObject(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +function parsePreparedSnapshot(serialized: string): PreparedSnapshot { + let value: unknown; + try { + value = JSON.parse(serialized); + } catch { + throw new Error("Durable bootstrap snapshot is not valid JSON"); + } + if ( + !isObject(value) || + typeof value.id !== "string" || + typeof value.provider !== "string" || + (value.consistency !== "sampled-current-state" && + value.consistency !== "point-in-time") || + !isObject(value.collections) || + !isObject(value.semantics) + ) { + throw new Error("Durable bootstrap snapshot is malformed"); + } + return value as unknown as PreparedSnapshot; +} + +function sourceTimeUs(value: number, label: string): number { + if (!Number.isSafeInteger(value) || value < 0) { + throw new TypeError(`${label} must be a non-negative safe microsecond value`); + } + return value; +} + +function assertMutationPosition( + mutation: SourceMutation, + checkpoint: SourcePosition, +): SourcePosition { + const eventPosition = mutation.position ?? checkpoint; + if ( + eventPosition.source !== checkpoint.source || + eventPosition.epoch !== checkpoint.epoch + ) { + throw new Error( + `Mutation ${mutation.uri} belongs to ${eventPosition.source}/${eventPosition.epoch}, ` + + `but its batch belongs to ${checkpoint.source}/${checkpoint.epoch}`, + ); + } + return eventPosition; +} + +/** Database-backed projection target for one unpublished fresh generation. */ +export class DatabaseBootstrapTarget implements BootstrapTarget { + private knownDids: Set | undefined; + private knownDidsLoaded = false; + + constructor( + private readonly db: Database, + private readonly config: ContrailConfig, + private readonly options: DatabaseBootstrapTargetOptions = {}, + ) { + if (options.knownDids) this.knownDids = new Set(options.knownDids); + } + + async load(): Promise { + const row = await this.db + .prepare("SELECT * FROM bootstrap_state WHERE id = ?") + .bind(BOOTSTRAP_STATE_ID) + .first(); + if (!row) return null; + if (!(["snapshot", "catchup", "complete"] as string[]).includes(row.phase)) { + throw new Error(`Invalid durable bootstrap phase: ${row.phase}`); + } + const snapshot = parsePreparedSnapshot(row.snapshot_json); + const progressRows = await this.db + .prepare( + "SELECT partition, cursor, completed FROM bootstrap_snapshot_progress WHERE bootstrap_id = ? ORDER BY partition", + ) + .bind(BOOTSTRAP_STATE_ID) + .all(); + const captureFrom = position( + row.capture_source, + row.capture_epoch, + row.capture_cursor, + "capture", + ); + if (!captureFrom) throw new Error("Durable bootstrap capture is missing"); + return { + phase: row.phase, + snapshot, + captureFrom, + snapshotProgress: (progressRows.results ?? []).map((item) => ({ + partition: item.partition, + cursor: item.cursor, + complete: item.completed === 1, + })), + snapshotComplete: row.snapshot_complete === 1, + catchupThrough: position( + row.catchup_source, + row.catchup_epoch, + row.catchup_cursor, + "catch-up target", + ), + changeCheckpoint: position( + row.change_source, + row.change_epoch, + row.change_cursor, + "change checkpoint", + ), + }; + } + + async begin( + snapshot: PreparedSnapshot, + captureFrom: SourcePosition, + ): Promise { + const now = Date.now(); + await this.db + .prepare( + `INSERT INTO bootstrap_state + (id, phase, snapshot_json, capture_source, capture_epoch, + capture_cursor, snapshot_complete, started_at, updated_at) + VALUES (?, 'snapshot', ?, ?, ?, ?, 0, ?, ?)`, + ) + .bind( + BOOTSTRAP_STATE_ID, + JSON.stringify(snapshot), + captureFrom.source, + captureFrom.epoch, + captureFrom.cursor, + now, + now, + ) + .run(); + } + + async applySnapshotBatch( + snapshot: PreparedSnapshot, + batch: SnapshotBatch, + ): Promise { + const observedAtUs = sourceTimeUs( + batch.sourceTimeUs, + "Snapshot sourceTimeUs", + ); + const indexedAt = Date.now() * 1000; + const events = batch.records.map((item) => + createIngestEvent({ + uri: item.uri, + did: item.did, + collection: item.collection, + rkey: item.rkey, + operation: "update", + cid: item.cid, + value: item.value, + timeUs: recordTimeUs( + item.value, + item.collection, + this.config, + observedAtUs, + ), + indexedAt, + source: { + id: `snapshot:${snapshot.provider}`, + epoch: snapshot.through?.epoch ?? snapshot.id, + time_us: observedAtUs, + revision: null, + cursor: batch.progress.cursor ?? batch.progress.partition, + }, + }), + ); + const now = Date.now(); + const progress = this.db + .prepare( + `INSERT INTO bootstrap_snapshot_progress + (bootstrap_id, partition, cursor, completed, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(bootstrap_id, partition) DO UPDATE SET + cursor = excluded.cursor, + completed = excluded.completed, + updated_at = excluded.updated_at`, + ) + .bind( + BOOTSTRAP_STATE_ID, + batch.progress.partition, + batch.progress.cursor, + batch.progress.complete ? 1 : 0, + now, + ); + const checkpoint = this.db + .prepare( + `UPDATE bootstrap_state + SET snapshot_complete = ?, updated_at = ? + WHERE id = ? AND phase = 'snapshot'`, + ) + .bind(batch.done ? 1 : 0, now, BOOTSTRAP_STATE_ID); + await this.apply(events, [progress, checkpoint], true); + } + + async beginCatchup(through: SourcePosition): Promise { + await this.db + .prepare( + `UPDATE bootstrap_state + SET phase = 'catchup', catchup_source = ?, catchup_epoch = ?, + catchup_cursor = ?, change_source = capture_source, + change_epoch = capture_epoch, change_cursor = capture_cursor, + updated_at = ? + WHERE id = ? AND phase = 'snapshot' AND snapshot_complete = 1`, + ) + .bind( + through.source, + through.epoch, + through.cursor, + Date.now(), + BOOTSTRAP_STATE_ID, + ) + .run(); + const state = await this.load(); + if (state?.phase !== "catchup") { + throw new Error("Cannot begin catch-up before the snapshot is complete"); + } + } + + async applyMutationBatch(batch: MutationBatch): Promise { + const indexedAt = Date.now() * 1000; + const events = batch.mutations.map((mutation) => { + const eventPosition = assertMutationPosition(mutation, batch.checkpoint); + const observedAtUs = sourceTimeUs( + mutation.sourceTimeUs, + `Mutation ${mutation.uri} sourceTimeUs`, + ); + return createIngestEvent({ + uri: mutation.uri, + did: mutation.did, + collection: mutation.collection, + rkey: mutation.rkey, + operation: mutation.operation === "delete" ? "delete" : "update", + cid: mutation.operation === "delete" ? null : mutation.cid, + value: mutation.operation === "delete" ? undefined : mutation.value, + timeUs: + mutation.operation === "delete" + ? observedAtUs + : recordTimeUs( + mutation.value, + mutation.collection, + this.config, + observedAtUs, + ), + indexedAt, + source: { + id: eventPosition.source, + epoch: eventPosition.epoch, + time_us: observedAtUs, + revision: mutation.revision ?? null, + cursor: eventPosition.cursor, + }, + }); + }); + const checkpoint = this.db + .prepare( + `UPDATE bootstrap_state + SET change_source = ?, change_epoch = ?, change_cursor = ?, updated_at = ? + WHERE id = ? AND phase = 'catchup'`, + ) + .bind( + batch.checkpoint.source, + batch.checkpoint.epoch, + batch.checkpoint.cursor, + Date.now(), + BOOTSTRAP_STATE_ID, + ); + await this.apply(events, [checkpoint], false); + } + + async complete(): Promise { + if (this.options.deferDerivedProjections) { + await rebuildDerivedProjections(this.db, this.config); + } + await this.db + .prepare( + `UPDATE bootstrap_state + SET phase = 'complete', finished_at = ?, updated_at = ? + WHERE id = ? AND phase = 'catchup' + AND change_source = catchup_source + AND change_epoch = catchup_epoch + AND change_cursor = catchup_cursor`, + ) + .bind(Date.now(), Date.now(), BOOTSTRAP_STATE_ID) + .run(); + const state = await this.load(); + if (state?.phase !== "complete") { + throw new Error("Cannot complete bootstrap before catch-up reaches its target"); + } + } + + private async apply( + events: IngestEvent[], + checkpoints: Statement[], + authoritativeSourceObservation: boolean, + ): Promise { + const knownDids = await this.getKnownDids(); + const result = await ingestRecords(this.db, events, this.config, { + knownDids, + skipDerivedProjections: this.options.deferDerivedProjections === true, + authoritativeSourceObservation, + trailingStatements: checkpoints, + }); + if (knownDids) { + for (const did of result.discoveredDids) knownDids.add(did); + } + } + + private async getKnownDids(): Promise | undefined> { + if (getDependentNsids(this.config).length === 0) return undefined; + if (this.knownDidsLoaded) return this.knownDids ?? new Set(); + const known = this.knownDids ?? new Set(); + const identityRows = await this.db + .prepare("SELECT did FROM identities") + .all<{ did: string }>(); + for (const row of identityRows.results ?? []) known.add(row.did); + for (const [shortName, collection] of Object.entries( + this.config.collections, + )) { + if (collection.discover === false) continue; + const rows = await this.db + .prepare(`SELECT DISTINCT did FROM ${recordsTableName(shortName)}`) + .all<{ did: string }>(); + for (const row of rows.results ?? []) known.add(row.did); + } + this.knownDids = known; + this.knownDidsLoaded = true; + return known; + } +} diff --git a/packages/contrail/src/core/db/records.ts b/packages/contrail/src/core/db/records.ts index e60680f..0f60317 100644 --- a/packages/contrail/src/core/db/records.ts +++ b/packages/contrail/src/core/db/records.ts @@ -553,6 +553,7 @@ export interface RecordVersionInfo { operation: "create" | "update" | "delete"; cid: string | null; source_id: string; + source_epoch: string | null; source_revision: string | null; source_time_us: number; source_cursor: string | null; @@ -569,6 +570,7 @@ function versionForEvent(event: IngestEvent): RecordVersionInfo { operation: event.operation, cid: event.cid, source_id: source?.id ?? "legacy-caller", + source_epoch: source?.epoch ?? null, // Before 0.13.1 callers had no separate source clock, so time_us is the // least surprising compatibility fallback for hand-built IngestEvents. source_time_us: source?.time_us ?? event.time_us, @@ -616,6 +618,7 @@ export function compareRecordVersions( } if ( left.source_id === right.source_id && + left.source_epoch === right.source_epoch && left.source_cursor !== null && right.source_cursor !== null && left.source_cursor !== right.source_cursor @@ -631,6 +634,9 @@ export function compareRecordVersions( if (left.source_id !== right.source_id) { return left.source_id < right.source_id ? -1 : 1; } + const leftEpoch = left.source_epoch ?? ""; + const rightEpoch = right.source_epoch ?? ""; + if (leftEpoch !== rightEpoch) return leftEpoch < rightEpoch ? -1 : 1; return 0; } @@ -645,7 +651,7 @@ export async function lookupRecordVersions( const placeholders = chunk.map(() => "?").join(","); const rows = await db .prepare( - `SELECT uri, did, collection, rkey, operation, cid, source_id, source_revision, source_time_us, source_cursor, indexed_at FROM record_versions WHERE uri IN (${placeholders})`, + `SELECT uri, did, collection, rkey, operation, cid, source_id, source_epoch, source_revision, source_time_us, source_cursor, indexed_at FROM record_versions WHERE uri IN (${placeholders})`, ) .bind(...chunk) .all(); @@ -770,7 +776,7 @@ const RECORD_UPSERT_BINDINGS = 7; const RECORD_UPSERT_ROWS = Math.floor( MAX_STATEMENT_BINDINGS / RECORD_UPSERT_BINDINGS ); -const RECORD_VERSION_BINDINGS = 11; +const RECORD_VERSION_BINDINGS = 12; const RECORD_VERSION_ROWS = Math.floor( MAX_STATEMENT_BINDINGS / RECORD_VERSION_BINDINGS, ); @@ -789,12 +795,12 @@ function buildRecordVersionStatements( for (let index = 0; index < events.length; index += RECORD_VERSION_ROWS) { const chunk = events.slice(index, index + RECORD_VERSION_ROWS); const values = chunk - .map(() => "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") + .map(() => "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") .join(", "); statements.push( db .prepare( - `INSERT INTO record_versions (uri, did, collection, rkey, operation, cid, source_id, source_revision, source_time_us, source_cursor, indexed_at) VALUES ${values} ON CONFLICT(uri) DO UPDATE SET did = excluded.did, collection = excluded.collection, rkey = excluded.rkey, operation = excluded.operation, cid = excluded.cid, source_id = excluded.source_id, source_revision = excluded.source_revision, source_time_us = excluded.source_time_us, source_cursor = excluded.source_cursor, indexed_at = excluded.indexed_at`, + `INSERT INTO record_versions (uri, did, collection, rkey, operation, cid, source_id, source_epoch, source_revision, source_time_us, source_cursor, indexed_at) VALUES ${values} ON CONFLICT(uri) DO UPDATE SET did = excluded.did, collection = excluded.collection, rkey = excluded.rkey, operation = excluded.operation, cid = excluded.cid, source_id = excluded.source_id, source_epoch = excluded.source_epoch, source_revision = excluded.source_revision, source_time_us = excluded.source_time_us, source_cursor = excluded.source_cursor, indexed_at = excluded.indexed_at`, ) .bind( ...chunk.flatMap((event) => { @@ -811,6 +817,7 @@ function buildRecordVersionStatements( version.operation, retainedCid, version.source_id, + version.source_epoch, version.source_revision, version.source_time_us, version.source_cursor, diff --git a/packages/contrail/src/core/db/schema.ts b/packages/contrail/src/core/db/schema.ts index d673ff4..e7116d3 100644 --- a/packages/contrail/src/core/db/schema.ts +++ b/packages/contrail/src/core/db/schema.ts @@ -17,7 +17,7 @@ import { getSearchableFields } from "../search"; import { buildLabelsSchema } from "../labels/schema"; import { getMeta, setMeta } from "./meta"; -export const CONTRAIL_SCHEMA_VERSION = 8; +export const CONTRAIL_SCHEMA_VERSION = 9; const SCHEMA_FINGERPRINT_KEY = "schema_fingerprint"; function getResolved(config: ContrailConfig): ResolvedMaps { @@ -67,6 +67,33 @@ CREATE TABLE IF NOT EXISTS backfill_state ( heartbeat_at ${dialect.bigintType}, finished_at ${dialect.bigintType} ); +CREATE TABLE IF NOT EXISTS bootstrap_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + phase TEXT NOT NULL CHECK (phase IN ('snapshot', 'catchup', 'complete')), + snapshot_json TEXT NOT NULL, + capture_source TEXT NOT NULL, + capture_epoch TEXT NOT NULL, + capture_cursor TEXT NOT NULL, + snapshot_complete INTEGER NOT NULL DEFAULT 0, + catchup_source TEXT, + catchup_epoch TEXT, + catchup_cursor TEXT, + change_source TEXT, + change_epoch TEXT, + change_cursor TEXT, + started_at ${dialect.bigintType} NOT NULL, + updated_at ${dialect.bigintType} NOT NULL, + finished_at ${dialect.bigintType} +); +CREATE TABLE IF NOT EXISTS bootstrap_snapshot_progress ( + bootstrap_id INTEGER NOT NULL, + partition TEXT NOT NULL, + cursor TEXT, + completed INTEGER NOT NULL DEFAULT 0, + updated_at ${dialect.bigintType} NOT NULL, + PRIMARY KEY (bootstrap_id, partition), + FOREIGN KEY (bootstrap_id) REFERENCES bootstrap_state(id) +); CREATE TABLE IF NOT EXISTS identities ( did TEXT PRIMARY KEY, handle TEXT, @@ -82,6 +109,7 @@ CREATE TABLE IF NOT EXISTS record_versions ( operation TEXT NOT NULL CHECK (operation IN ('create', 'update', 'delete')), cid TEXT, source_id TEXT NOT NULL, + source_epoch TEXT, source_revision TEXT, source_time_us ${dialect.bigintType} NOT NULL, source_cursor TEXT, @@ -351,6 +379,11 @@ const MIGRATIONS: MigrationOp[] = [ column: "retry_exhausted", columnDef: "INTEGER NOT NULL DEFAULT 0", }, + { + table: "record_versions", + column: "source_epoch", + columnDef: "TEXT", + }, { table: "discovery", column: "retries", @@ -414,8 +447,8 @@ async function seedLegacyRecordVersions( const table = recordsTableName(shortName); await db .prepare( - `INSERT INTO record_versions (uri, did, collection, rkey, operation, cid, source_id, source_revision, source_time_us, source_cursor, indexed_at) - SELECT uri, did, ?, rkey, 'update', cid, 'legacy', NULL, indexed_at, NULL, indexed_at FROM ${table} + `INSERT INTO record_versions (uri, did, collection, rkey, operation, cid, source_id, source_epoch, source_revision, source_time_us, source_cursor, indexed_at) + SELECT uri, did, ?, rkey, 'update', cid, 'legacy', NULL, NULL, indexed_at, NULL, indexed_at FROM ${table} WHERE 1 = 1 ON CONFLICT(uri) DO NOTHING`, ) .bind(collection.collection) diff --git a/packages/contrail/src/core/ingest.ts b/packages/contrail/src/core/ingest.ts index d665358..48527cf 100644 --- a/packages/contrail/src/core/ingest.ts +++ b/packages/contrail/src/core/ingest.ts @@ -60,6 +60,9 @@ export function createIngestEvent(input: RecordEventInput): IngestEvent { indexed_at: indexedAt, source: { id: input.source?.id ?? "local", + ...(input.source?.epoch === undefined + ? {} + : { epoch: input.source.epoch }), // Preserve pre-0.13.1 ordering for callers that do not yet provide a // separate source clock; adapters always pass source.time_us explicitly. time_us: input.source?.time_us ?? input.timeUs, diff --git a/packages/contrail/src/core/sources.ts b/packages/contrail/src/core/sources.ts index 1ed787f..ca86817 100644 --- a/packages/contrail/src/core/sources.ts +++ b/packages/contrail/src/core/sources.ts @@ -43,11 +43,21 @@ export interface PreparedSnapshot { through?: SourcePosition; } +export interface SnapshotProgress { + /** Stable provider-owned partition, such as one collection/repository pair. */ + partition: string; + /** Opaque resume token within this partition, or null once complete. */ + cursor: string | null; + complete: boolean; +} + export interface SnapshotBatch { records: SnapshotRecord[]; - /** Opaque resume token within this exact prepared snapshot. */ - progress: string; - /** True only after every requested collection has been emitted. */ + /** Source observation time for ordering snapshot rows against later changes. */ + sourceTimeUs: number; + /** Progress for the partition represented by this batch. */ + progress: SnapshotProgress; + /** True only after every requested snapshot partition has completed. */ done: boolean; } @@ -61,7 +71,7 @@ export interface SnapshotSource { }): Promise; read(options: { snapshot: PreparedSnapshot; - progress?: string; + progress?: SnapshotProgress[]; signal?: AbortSignal; }): AsyncIterable; } @@ -119,7 +129,7 @@ export interface BootstrapRunState { phase: BootstrapPhase; snapshot: PreparedSnapshot; captureFrom: SourcePosition; - snapshotProgress: string | null; + snapshotProgress: SnapshotProgress[]; snapshotComplete: boolean; catchupThrough: SourcePosition | null; changeCheckpoint: SourcePosition | null; @@ -230,7 +240,7 @@ export async function bootstrapFreshProjection(options: { phase: "snapshot", snapshot, captureFrom, - snapshotProgress: null, + snapshotProgress: [], snapshotComplete: false, catchupThrough: null, changeCheckpoint: null, @@ -243,7 +253,7 @@ export async function bootstrapFreshProjection(options: { let sawDone = false; for await (const batch of snapshotSource.read({ snapshot: state.snapshot, - ...(state.snapshotProgress === null + ...(state.snapshotProgress.length === 0 ? {} : { progress: state.snapshotProgress }), signal, @@ -252,7 +262,11 @@ export async function bootstrapFreshProjection(options: { throw new Error(`Snapshot ${state.snapshot.id} emitted data after done`); } await target.applySnapshotBatch(state.snapshot, batch); - state.snapshotProgress = batch.progress; + const progressIndex = state.snapshotProgress.findIndex( + (item) => item.partition === batch.progress.partition, + ); + if (progressIndex < 0) state.snapshotProgress.push(batch.progress); + else state.snapshotProgress[progressIndex] = batch.progress; state.snapshotComplete = batch.done; sawDone = batch.done; } diff --git a/packages/contrail/src/core/types.ts b/packages/contrail/src/core/types.ts index 0be2255..f81c162 100644 --- a/packages/contrail/src/core/types.ts +++ b/packages/contrail/src/core/types.ts @@ -521,8 +521,10 @@ export interface RecordRow { } export interface MutationSource { - /** Stable adapter identifier, for example `jetstream` or `pds-backfill`. */ + /** Stable logical source identifier, for example `jetstream` or `pds-backfill`. */ id: string; + /** Continuity epoch for opaque cursors. Missing on legacy adapters. */ + epoch?: string | null; /** Source observation/event time, independent of record application time. */ time_us: number; /** Monotonic repository revision when the source provides one. */ diff --git a/packages/contrail/src/index.ts b/packages/contrail/src/index.ts index 96058d0..038290f 100644 --- a/packages/contrail/src/index.ts +++ b/packages/contrail/src/index.ts @@ -18,6 +18,7 @@ export type { ResolvedIdentity } from "./core/client"; // Ingestion and maintenance. export * from "./core/ingest"; export * from "./core/sources"; +export * from "./core/bootstrap"; export * from "./core/jetstream"; export * from "./core/persistent"; export * from "./core/backfill"; diff --git a/packages/contrail/tests/bootstrap-sources.test.ts b/packages/contrail/tests/bootstrap-sources.test.ts index f8f36e2..1f32ea8 100644 --- a/packages/contrail/tests/bootstrap-sources.test.ts +++ b/packages/contrail/tests/bootstrap-sources.test.ts @@ -7,6 +7,7 @@ import { type MutationBatch, type PreparedSnapshot, type SnapshotBatch, + type SnapshotProgress, type SnapshotRecord, type SnapshotSource, type SourceMutation, @@ -76,7 +77,7 @@ class MemoryTarget implements BootstrapTarget { phase: "snapshot", snapshot: structuredClone(snapshot), captureFrom: structuredClone(captureFrom), - snapshotProgress: null, + snapshotProgress: [], snapshotComplete: false, catchupThrough: null, changeCheckpoint: null, @@ -86,7 +87,11 @@ class MemoryTarget implements BootstrapTarget { async applySnapshotBatch(_snapshot: PreparedSnapshot, batch: SnapshotBatch) { for (const item of batch.records) this.records.set(item.uri, item); this.snapshotBatches++; - this.state!.snapshotProgress = batch.progress; + const index = this.state!.snapshotProgress.findIndex( + (item) => item.partition === batch.progress.partition, + ); + if (index < 0) this.state!.snapshotProgress.push(batch.progress); + else this.state!.snapshotProgress[index] = batch.progress; this.state!.snapshotComplete = batch.done; } @@ -121,7 +126,7 @@ class MemoryTarget implements BootstrapTarget { function snapshotSource(options: { snapshot: PreparedSnapshot; - batches(progress?: string): SnapshotBatch[]; + batches(progress?: SnapshotProgress[]): SnapshotBatch[]; calls?: string[]; }): SnapshotSource { return { @@ -131,7 +136,8 @@ function snapshotSource(options: { return structuredClone(options.snapshot); }, async *read({ progress }) { - options.calls?.push(`snapshot:${progress ?? "start"}`); + const cursor = progress?.find((item) => item.partition === "main")?.cursor; + options.calls?.push(`snapshot:${cursor ?? "start"}`); for (const batch of options.batches(progress)) yield structuredClone(batch); }, }; @@ -190,7 +196,8 @@ describe("bootstrap source orchestration", () => { // Alice was sampled at different moments: A already reflects cursor // 3, B is stale at cursor 2, and C did not exist when sampled. records: [record("a", 3), record("b", 2)], - progress: "done", + sourceTimeUs: 3, + progress: { partition: "main", cursor: null, complete: true }, done: true, }, ], @@ -238,7 +245,12 @@ describe("bootstrap source orchestration", () => { }), calls, batches: () => [ - { records: [record("a", 5)], progress: "done", done: true }, + { + records: [record("a", 5)], + sourceTimeUs: 5, + progress: { partition: "main", cursor: null, complete: true }, + done: true, + }, ], }); const changes = changeSource({ @@ -275,15 +287,35 @@ describe("bootstrap source orchestration", () => { const snapshot = snapshotSource({ snapshot: prepared(), batches(progress) { - reads.push(progress); - if (progress === "part-1") { + const cursor = progress?.find((item) => item.partition === "main")?.cursor; + reads.push(cursor); + if (cursor === "part-1") { return [ - { records: [record("b", 2)], progress: "done", done: true }, + { + records: [record("b", 2)], + sourceTimeUs: 2, + progress: { partition: "main", cursor: null, complete: true }, + done: true, + }, ]; } return [ - { records: [record("a", 1)], progress: "part-1", done: false }, - { records: [record("b", 2)], progress: "done", done: true }, + { + records: [record("a", 1)], + sourceTimeUs: 1, + progress: { + partition: "main", + cursor: "part-1", + complete: false, + }, + done: false, + }, + { + records: [record("b", 2)], + sourceTimeUs: 2, + progress: { partition: "main", cursor: null, complete: true }, + done: true, + }, ]; }, }); @@ -291,7 +323,7 @@ describe("bootstrap source orchestration", () => { const target = new MemoryTarget(); const apply = target.applySnapshotBatch.bind(target); target.applySnapshotBatch = async (preparedSnapshot, batch) => { - if (batch.progress === "done" && failSecondBatch) { + if (batch.progress.complete && failSecondBatch) { failSecondBatch = false; throw new Error("injected snapshot failure"); } @@ -325,7 +357,14 @@ describe("bootstrap source orchestration", () => { it("commits an empty change batch to prove progress through the target", async () => { const snapshot = snapshotSource({ snapshot: prepared(), - batches: () => [{ records: [], progress: "done", done: true }], + batches: () => [ + { + records: [], + sourceTimeUs: 1, + progress: { partition: "main", cursor: null, complete: true }, + done: true, + }, + ], }); const target = new MemoryTarget(); diff --git a/packages/contrail/tests/database-bootstrap-target.test.ts b/packages/contrail/tests/database-bootstrap-target.test.ts new file mode 100644 index 0000000..c269a44 --- /dev/null +++ b/packages/contrail/tests/database-bootstrap-target.test.ts @@ -0,0 +1,239 @@ +import { describe, expect, it } from "vitest"; +import { + DatabaseBootstrapTarget, + bootstrapFreshProjection, + initSchema, + queryRecords, + resolveConfig, + type ChangeSource, + type PreparedSnapshot, + type SnapshotSource, + type SourcePosition, +} from "../src/index"; +import { createSqliteDatabase } from "../src/adapters/sqlite"; + +const COLLECTION = "com.example.event"; +const DID = "did:plc:bootstrap"; + +function sourcePosition(cursor: number): SourcePosition { + return { source: "jetstream-test", epoch: "epoch-one", cursor: String(cursor) }; +} + +function snapshot(): PreparedSnapshot { + return { + id: "pds-snapshot-one", + provider: "pds", + consistency: "sampled-current-state", + collections: { [COLLECTION]: { state: "complete" } }, + semantics: { + ordinaryRecords: true, + ordinaryDeletes: true, + accountLifecycle: false, + repositoryReplacement: false, + verifiedCommits: false, + explicitHead: true, + }, + }; +} + +function value(name: string) { + return { $type: COLLECTION, name }; +} + +function record(rkey: string, name: string) { + return { + uri: `at://${DID}/${COLLECTION}/${rkey}`, + did: DID, + collection: COLLECTION, + rkey, + cid: `cid-${rkey}-${name}`, + value: value(name), + }; +} + +function config() { + return resolveConfig({ + namespace: "com.example", + profiles: [], + constellation: false, + collections: { event: { collection: COLLECTION } }, + }); +} + +describe("database bootstrap target", () => { + it("projects a sampled snapshot and ordered tail with durable epochs", async () => { + const resolved = config(); + const db = createSqliteDatabase(":memory:"); + await initSchema(db, resolved); + const prepared = snapshot(); + const snapshotSource: SnapshotSource = { + id: "pds", + async prepare() { + return prepared; + }, + async *read() { + yield { + records: [record("a", "old"), record("b", "remove")], + sourceTimeUs: 1, + progress: { partition: "pds", cursor: null, complete: true }, + done: true, + }; + }, + }; + let mark = 0; + const changeSource: ChangeSource = { + id: "jetstream", + async mark() { + mark++; + return sourcePosition(mark === 1 ? 1 : 3); + }, + async *read({ through }) { + yield { + mutations: [ + { + operation: "put", + ...record("a", "new"), + sourceTimeUs: 2, + position: sourcePosition(2), + }, + { + operation: "delete", + uri: `at://${DID}/${COLLECTION}/b`, + did: DID, + collection: COLLECTION, + rkey: "b", + sourceTimeUs: 3, + position: sourcePosition(3), + }, + ], + checkpoint: through, + caughtUp: true, + }; + }, + }; + const target = new DatabaseBootstrapTarget(db, resolved, { + deferDerivedProjections: true, + }); + + await bootstrapFreshProjection({ + collections: [COLLECTION], + snapshotSource, + changeSource, + target, + }); + + const records = await queryRecords(db, resolved, { collection: "event" }); + expect(records.records).toHaveLength(1); + expect(JSON.parse(records.records[0].record).name).toBe("new"); + expect(await target.load()).toMatchObject({ + phase: "complete", + snapshotProgress: [ + { partition: "pds", cursor: null, complete: true }, + ], + snapshotComplete: true, + captureFrom: sourcePosition(1), + catchupThrough: sourcePosition(3), + changeCheckpoint: sourcePosition(3), + }); + expect( + await db + .prepare( + "SELECT source_id, source_epoch, source_cursor FROM record_versions WHERE uri = ?", + ) + .bind(`at://${DID}/${COLLECTION}/a`) + .first(), + ).toEqual({ + source_id: "jetstream-test", + source_epoch: "epoch-one", + source_cursor: "2", + }); + }); + + it("rolls projection back when snapshot progress cannot commit", async () => { + const resolved = config(); + const db = createSqliteDatabase(":memory:"); + await initSchema(db, resolved); + const target = new DatabaseBootstrapTarget(db, resolved); + const prepared = snapshot(); + await target.begin(prepared, sourcePosition(1)); + await db + .prepare( + `CREATE TRIGGER fail_bootstrap_progress + BEFORE INSERT ON bootstrap_snapshot_progress + BEGIN SELECT RAISE(ABORT, 'injected bootstrap checkpoint failure'); END`, + ) + .run(); + + const batch = { + records: [record("a", "atomic")], + sourceTimeUs: 1, + progress: { partition: "pds", cursor: null, complete: true }, + done: true, + }; + await expect(target.applySnapshotBatch(prepared, batch)).rejects.toThrow( + "injected bootstrap checkpoint failure", + ); + + expect( + (await queryRecords(db, resolved, { collection: "event" })).records, + ).toHaveLength(0); + expect(await target.load()).toMatchObject({ + phase: "snapshot", + snapshotProgress: [], + snapshotComplete: false, + }); + + await db.prepare("DROP TRIGGER fail_bootstrap_progress").run(); + await target.applySnapshotBatch(prepared, batch); + await target.beginCatchup(sourcePosition(1)); + await target.complete(); + + expect( + (await queryRecords(db, resolved, { collection: "event" })).records, + ).toHaveLength(1); + expect((await target.load())?.phase).toBe("complete"); + }); + + it("rejects a mutation position from another epoch before advancing progress", async () => { + const resolved = config(); + const db = createSqliteDatabase(":memory:"); + await initSchema(db, resolved); + const target = new DatabaseBootstrapTarget(db, resolved); + const prepared = snapshot(); + await target.begin(prepared, sourcePosition(1)); + await target.applySnapshotBatch(prepared, { + records: [], + sourceTimeUs: 1, + progress: { partition: "pds", cursor: null, complete: true }, + done: true, + }); + await target.beginCatchup(sourcePosition(3)); + + await expect( + target.applyMutationBatch({ + mutations: [ + { + operation: "put", + ...record("a", "wrong-epoch"), + sourceTimeUs: 2, + position: { + source: "jetstream-test", + epoch: "epoch-two", + cursor: "2", + }, + }, + ], + checkpoint: sourcePosition(3), + caughtUp: true, + }), + ).rejects.toThrow("belongs to jetstream-test/epoch-two"); + + expect(await target.load()).toMatchObject({ + phase: "catchup", + changeCheckpoint: sourcePosition(1), + }); + expect( + (await queryRecords(db, resolved, { collection: "event" })).records, + ).toHaveLength(0); + }); +}); diff --git a/packages/contrail/tests/source-ordering.test.ts b/packages/contrail/tests/source-ordering.test.ts index 576cba1..08b363b 100644 --- a/packages/contrail/tests/source-ordering.test.ts +++ b/packages/contrail/tests/source-ordering.test.ts @@ -44,6 +44,7 @@ function mutation(options: { name?: string; recordTime?: number; sourceId?: string; + epoch?: string; }): IngestEvent { return createIngestEvent({ uri: URI, @@ -60,6 +61,7 @@ function mutation(options: { indexedAt: options.sourceTime + 10_000, source: { id: options.sourceId ?? "fake-source", + ...(options.epoch === undefined ? {} : { epoch: options.epoch }), time_us: options.sourceTime, revision: options.revision ?? null, cursor: String(options.sourceTime), @@ -119,6 +121,57 @@ describe("durable source ordering", () => { expect(stale.dropped.superseded).toBe(2); }); + it("does not compare opaque cursors across source epochs", async () => { + const resolved = config(); + const db = await setup(resolved); + await ingestRecords( + db, + [ + mutation({ + operation: "delete", + sourceTime: 100, + sourceId: "same-source", + epoch: "epoch-a", + revision: null, + }), + ], + resolved, + ); + + const nextEpoch = await ingestRecords( + db, + [ + { + ...mutation({ + operation: "delete", + sourceTime: 100, + sourceId: "same-source", + epoch: "epoch-b", + revision: null, + }), + source: { + id: "same-source", + epoch: "epoch-b", + time_us: 100, + revision: null, + cursor: "1", + }, + }, + ], + resolved, + ); + + expect(nextEpoch.accepted).toHaveLength(1); + expect( + await db + .prepare( + "SELECT source_epoch, source_cursor FROM record_versions WHERE uri = ?", + ) + .bind(URI) + .first(), + ).toEqual({ source_epoch: "epoch-b", source_cursor: "1" }); + }); + it("does not let a stale delete remove a newer record", async () => { const resolved = config(); const db = await setup(resolved); -- 2.51.2