From add067ae764f95d605335a1b6698e6b2b3d53d85 Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Fri, 21 Aug 2026 10:42:06 +0200 Subject: [PATCH] Stabilize change log operations and upgrades --- apps/atmo-rsvp/tests/contract.test.ts | 23 +++++++++- packages/contrail/README.md | 4 +- packages/contrail/src/cli/commands/changes.ts | 9 +++- packages/contrail/src/core/change-log.ts | 39 ++++++++++++++++ packages/contrail/src/index.ts | 7 ++- packages/contrail/tests/change-log.test.ts | 45 +++++++++++++++++++ 6 files changed, 123 insertions(+), 4 deletions(-) diff --git a/apps/atmo-rsvp/tests/contract.test.ts b/apps/atmo-rsvp/tests/contract.test.ts index bfea7e3..9f0030f 100644 --- a/apps/atmo-rsvp/tests/contract.test.ts +++ b/apps/atmo-rsvp/tests/contract.test.ts @@ -3,7 +3,11 @@ import { describePublicService } from "@atmo-dev/contrail"; import { createSqliteDatabase } from "@atmo-dev/contrail/sqlite"; import { createWorker } from "@atmo-dev/contrail/worker"; import { lexicons } from "../lexicons/generated"; -import { config } from "../src/contrail.config"; +import { config, searchGenerationConfig } from "../src/contrail.config"; +import { + createAtmoMeilisearchRuntime, + type AtmoMeilisearchEnv, +} from "../src/meilisearch"; const EXPECTED_METHODS = [ "rsvp.atmo.event.getRecord", @@ -71,6 +75,23 @@ describe("api.atmo.rsvp public contract", () => { publicService: { endpoint: "https://api.atmo.rsvp" }, }), ).not.toThrow(); + expect(searchGenerationConfig.changes?.consumers.search).toEqual({ + collections: ["community.lexicon.calendar.event"], + phases: ["historical", "live"], + initial: "current", + requiredForActivation: true, + }); + const search = createAtmoMeilisearchRuntime(); + expect(() => + createWorker>( + searchGenerationConfig, + { + lexicons, + publicService: { endpoint: "https://api.atmo.rsvp" }, + ...search, + }, + ), + ).not.toThrow(); }); it("publishes its service DID and permits browser auth headers", async () => { diff --git a/packages/contrail/README.md b/packages/contrail/README.md index 6eac71f..22ac887 100644 --- a/packages/contrail/README.md +++ b/packages/contrail/README.md @@ -151,7 +151,9 @@ Claims coalesce repeated URIs, hydrate in set-oriented collection queries, and r `initial: "current"` uses a durable snapshot-plus-tail coordinator. Repeatedly claim and idempotently acknowledge `contrail.changes.claimSnapshotPage()`, then drain `claimBootstrapChanges()` through its fixed target using ordinary hydrate/ack. Finally claim the stable generation-scoped activation token with `claimActivation()`, perform an idempotent destination swap, and call `completeActivation()`. A crash replays the same URI page, tail range, or activation token. Records updated or deleted while the keyset scan races are corrected by the anchored tail. -A consumer can be added to a populated log when all of its collection/phase pairs were already covered; its current/future anchor is the atomic current head, while history starts at the retained floor. Expanding coverage still fails closed without a fresh generation or explicit old-writer quiet boundary. `contrail changes status`, `retry`, `prune`, and explicitly confirmed `skip` expose private operations for SQLite or Wrangler D1 deployments. Pruning is bounded by the slowest durable consumer/bootstrap anchor. Skip records a bounded private audit reason and never occurs implicitly. Disabling or removing an existing log remains fail-closed. With no configured consumers, no change-log tables or append writes exist. +A consumer can be added to a populated log when all of its collection/phase pairs were already covered; its current/future anchor is the atomic current head, while history starts at the retained floor. Expanding coverage still fails closed without a fresh generation or explicit old-writer quiet boundary. `contrail changes status`, `retry`, `prune`, and explicitly confirmed `skip` expose private operations for SQLite or Wrangler D1 deployments. Status includes a conservative projection/head/batch/ack write plan. Pruning is bounded by the slowest durable consumer/bootstrap anchor. Skip records a bounded private audit reason and never occurs implicitly. Disabling or removing an existing log remains fail-closed. With no configured consumers, no change-log tables or append writes exist. + +Backups retain the database's change-log generation ID, positions, leases, bootstrap token, and consumer progress. Restoring that backup as the same generation is resumable and may redeliver an in-flight lease after expiry. Do not clone it as a new deployment generation while reusing destination cursors: build a fresh projection database and run current-state bootstrap. Restoring projection tables without the matching change tables is an unrecoverable delivery gap and must enter reset/bootstrap rather than silently continuing numeric positions. ## Local development diff --git a/packages/contrail/src/cli/commands/changes.ts b/packages/contrail/src/cli/commands/changes.ts index 1dfd205..0d608d0 100644 --- a/packages/contrail/src/cli/commands/changes.ts +++ b/packages/contrail/src/cli/commands/changes.ts @@ -1,5 +1,6 @@ import type { CAC } from "cac"; import { Contrail } from "../../contrail.js"; +import { getChangeLogCostPlan } from "../../core/change-log.js"; import type { Database } from "../../core/types.js"; import { resolveAndLoadConfig, @@ -143,8 +144,9 @@ export function registerChanges(cli: CAC): void { } const status = await contrail.changes.status(db); + const cost = getChangeLogCostPlan(contrail.config); if (commandOptions.json) { - console.log(JSON.stringify(status, null, 2)); + console.log(JSON.stringify({ ...status, cost }, null, 2)); return; } if (!status.enabled || !status.state) { @@ -156,6 +158,11 @@ export function registerChanges(cli: CAC): void { `floor=${status.state.retainedFloor} rows=${status.rows} ` + `changes=${status.changes} bytes=${status.bytes}`, ); + console.log( + ` write plan: projection=${cost.projectionStateWrites} ` + + `head=${cost.changeHeadWrites} batch=${cost.changeBatchWrites} ` + + `ack=${cost.acknowledgementWrites}`, + ); for (const item of status.consumers) { const retry = item.nextAttemptAt === null diff --git a/packages/contrail/src/core/change-log.ts b/packages/contrail/src/core/change-log.ts index b50a551..7db6713 100644 --- a/packages/contrail/src/core/change-log.ts +++ b/packages/contrail/src/core/change-log.ts @@ -16,6 +16,45 @@ import { export const MAX_CHANGE_BATCH_CHANGES = 500; export const MAX_CHANGE_BATCH_BYTES = 512_000; +export interface ChangeLogCostPlan { + enabled: boolean; + consumers: number; + coveragePairs: number; + projectionStateWrites: number; + changeHeadWrites: number; + changeBatchWrites: number; + acknowledgementWrites: number; + /** Total expected rows written by one relevant projection transaction. */ + relevantProjectionWrites: number; +} + +/** Conservative write-amplification report for one bounded projection batch. */ +export function getChangeLogCostPlan( + config: ContrailConfig, + mutationUris = 50, +): ChangeLogCostPlan { + if (!Number.isSafeInteger(mutationUris) || mutationUris < 1 || mutationUris > 500) { + throw new TypeError("mutationUris must be an integer between 1 and 500"); + } + const enabled = changesEnabled(config); + // One serialized revision update plus one predecessor-check update per forty + // unique URIs. The idempotent singleton INSERT normally writes no row. + const projectionStateWrites = 1 + Math.ceil(mutationUris / 40); + const changeHeadWrites = enabled ? 1 : 0; + const changeBatchWrites = enabled ? 1 : 0; + return { + enabled, + consumers: Object.keys(config.changes?.consumers ?? {}).length, + coveragePairs: changeLogCoverage(config).length, + projectionStateWrites, + changeHeadWrites, + changeBatchWrites, + acknowledgementWrites: enabled ? 1 : 0, + relevantProjectionWrites: + projectionStateWrites + changeHeadWrites + changeBatchWrites, + }; +} + export interface RecordChange { id: string; kind: "record"; diff --git a/packages/contrail/src/index.ts b/packages/contrail/src/index.ts index dfd17b6..cd07755 100644 --- a/packages/contrail/src/index.ts +++ b/packages/contrail/src/index.ts @@ -30,11 +30,16 @@ export * from "./core/backfill"; export * from "./core/status"; export * from "./core/diagnostics"; export { + getChangeLogCostPlan, getChangeLogState, MAX_CHANGE_BATCH_BYTES, MAX_CHANGE_BATCH_CHANGES, } from "./core/change-log"; -export type { ChangeLogState, RecordChange } from "./core/change-log"; +export type { + ChangeLogCostPlan, + ChangeLogState, + RecordChange, +} from "./core/change-log"; export * from "./core/changes"; export * from "./core/change-bootstrap"; export * from "./core/delivery"; diff --git a/packages/contrail/tests/change-log.test.ts b/packages/contrail/tests/change-log.test.ts index 00ae632..bb5a3c3 100644 --- a/packages/contrail/tests/change-log.test.ts +++ b/packages/contrail/tests/change-log.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { Contrail, createIngestEvent, + getChangeLogCostPlan, getChangeLogState, ingestRecords, initSchema, @@ -146,6 +147,16 @@ describe("transactional projection change log", () => { it("has no change-log schema or writes when disabled", async () => { const db = createSqliteDatabase(":memory:"); const resolved = config(); + expect(getChangeLogCostPlan(resolved, 50)).toEqual({ + enabled: false, + consumers: 0, + coveragePairs: 0, + projectionStateWrites: 3, + changeHeadWrites: 0, + changeBatchWrites: 0, + acknowledgementWrites: 0, + relevantProjectionWrites: 3, + }); await initSchema(db, resolved); const tables = await db @@ -169,6 +180,16 @@ describe("transactional projection change log", () => { it("initializes a fresh generation, registrations, and coverage ledger", async () => { const db = createSqliteDatabase(":memory:"); const resolved = loggedConfig(); + expect(getChangeLogCostPlan(resolved, 50)).toMatchObject({ + enabled: true, + consumers: 2, + coveragePairs: 2, + projectionStateWrites: 3, + changeHeadWrites: 1, + changeBatchWrites: 1, + acknowledgementWrites: 1, + relevantProjectionWrites: 5, + }); await initSchema(db, resolved); const state = await getChangeLogState(db); @@ -424,6 +445,30 @@ describe("transactional projection change log", () => { ); }); + it("upgrades retained pre-byte-count change batches without resetting consumers", async () => { + const db = createSqliteDatabase(":memory:"); + const resolved = loggedConfig(); + await initSchema(db, resolved); + await ingestRecords(db, [mutation({ sourceTime: 1 })], resolved); + await db + .prepare("ALTER TABLE change_batches DROP COLUMN encoded_bytes") + .run(); + await db + .prepare( + "UPDATE _contrail_meta SET value = 'old-change-schema' WHERE key = 'schema_fingerprint'", + ) + .run(); + + await initSchema(db, resolved); + const row = await db + .prepare("SELECT encoded_bytes, changes_json FROM change_batches") + .first<{ encoded_bytes: number; changes_json: string }>(); + expect(row?.encoded_bytes).toBe( + new TextEncoder().encode(row!.changes_json).byteLength, + ); + expect((await getChangeLogState(db))?.head).toBe("1"); + }); + it("retries a losing overlapping writer from fresh durable state", async () => { const real = createSqliteDatabase(":memory:"); const resolved = loggedConfig(); -- 2.51.2