From cd6bfd7a19f0711c65f43535c4f6480920faa69b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Mon, 9 Nov 2026 20:27:25 +0100 Subject: [PATCH 1/4] feat: find a restore by the newest event seen, before anything is delivered, and test restores end to end on every engine The restore watch also watches the newest event each instance read, or committed as a forgetting, at its position: a copy that lost it is found at the next check, even one taken in the five seconds before the generation moved, and in the web role. The dispatcher checks at the start of every pass, and a pass that finds a restore reconciles the forget ledger before it delivers or claims anything. restoreContract in @bounda-dev/core/adapter/testing restores a backup with the app stopped and, where the engine can, running, after forgetting, key rotation and more writes, also under a web instance and without a ledger, and checks every person forgotten after the copy with expectForgotten. It runs on node:sqlite, libSQL, PostgreSQL and the Durable Object. The new "Backups and restores" guide has the steps. --- .changeset/restore-harness.md | 5 + docs/src/content/docs/guides/backups.md | 127 ++++ docs/src/content/docs/guides/deployment.mdx | 3 +- docs/src/content/docs/guides/personal-data.md | 9 +- packages/cloudflare/test/restore.test.ts | 101 ++++ .../core/src/adapter/sqlite/restore.test.ts | 186 ++++++ .../testing/checkpoint-store.contract.ts | 2 +- packages/core/src/adapter/testing/index.ts | 6 + .../src/adapter/testing/restore.contract.ts | 552 ++++++++++++++++++ packages/core/src/kernel/app.test.ts | 2 + packages/core/src/kernel/app.ts | 52 +- .../src/kernel/dispatch/dispatcher.test.ts | 36 ++ .../core/src/kernel/dispatch/dispatcher.ts | 7 + .../personal-data/ledger-append.test.ts | 17 +- .../src/kernel/personal-data/ledger-append.ts | 10 +- packages/core/src/kernel/reactive-harness.ts | 8 + .../core/src/kernel/scheduler/worker.test.ts | 84 +++ packages/core/src/kernel/scheduler/worker.ts | 45 +- .../core/src/kernel/shared/restores.test.ts | 182 +++++- packages/core/src/kernel/shared/restores.ts | 78 ++- packages/core/src/kernel/status.test.ts | 1 + packages/postgresql/src/restore.test.ts | 65 +++ packages/sqlite/src/restore.test.ts | 132 +++++ 23 files changed, 1667 insertions(+), 43 deletions(-) create mode 100644 .changeset/restore-harness.md create mode 100644 docs/src/content/docs/guides/backups.md create mode 100644 packages/cloudflare/test/restore.test.ts create mode 100644 packages/core/src/adapter/sqlite/restore.test.ts create mode 100644 packages/core/src/adapter/testing/restore.contract.ts create mode 100644 packages/postgresql/src/restore.test.ts create mode 100644 packages/sqlite/src/restore.test.ts diff --git a/.changeset/restore-harness.md b/.changeset/restore-harness.md new file mode 100644 index 00000000..6e93f2a8 --- /dev/null +++ b/.changeset/restore-harness.md @@ -0,0 +1,5 @@ +--- +"@bounda-dev/core": minor +--- + +The restore watch also watches the newest event each instance read, or committed as a forgetting, at its position: a copy that lost it is found at the next pass, even one taken in the five seconds before the generation moved, and an instance in the `web` role finds it as well. The dispatcher checks at the start of every pass, and a pass that finds a restore reconciles the forget ledger before anything is delivered or claimed, so a scheduled command of someone the copy brought back never runs. An app without a forget ledger logs an error when it finds a restore. `restoreContract` in `@bounda-dev/core/adapter/testing` is the end-to-end restore an adapter has to pass: persons write, a backup is taken, some are forgotten, the master key rotates, and the copy is restored with the app stopped and, when the engine can, running, also under an instance in the `web` role and without a forget ledger through `app.restored()`; `expectForgotten` checks every person forgotten after the copy, and the app goes on. It runs on node:sqlite, libSQL, PostgreSQL and the Durable Object. The new guide "Backups and restores" has the steps for each engine and what is guaranteed. diff --git a/docs/src/content/docs/guides/backups.md b/docs/src/content/docs/guides/backups.md new file mode 100644 index 00000000..e8ab0ff7 --- /dev/null +++ b/docs/src/content/docs/guides/backups.md @@ -0,0 +1,127 @@ +--- +title: Backups and restores +description: Taking a backup, restoring it with the app stopped or running, and what Bounda guarantees about the persons forgotten in between. +sidebar: + order: 12 +--- + +A backup is a copy of the database at one moment. Restoring it takes every table back to that +moment, the keys of the persons forgotten since included, so their data would open again. With a +[forget ledger](/guides/personal-data/#the-forget-ledger), a record of every forgetting kept +outside the database, Bounda forgets them again. This page is what an operator does, and what the +runtime then guarantees. + +## Before you need one + +- Configure a [forget ledger](/guides/personal-data/#the-forget-ledger) in a store the restore + does not touch, and set its `keepFor` longer than your oldest backup. +- Keep every [master key](/guides/personal-data/#rotating-the-master-key) a backup was taken under + as long as the backup: a backup restored after a rotation opens only with the old key in + `previous`, and the worker rewraps its keys under the current one again. +- Keep the read models that live in [another database](/reference/configuration/) on the same + schedule as the storage, and restore both to the same moment. + +## Taking a backup + +Take it with the app running, with the engine's own tools: they copy a consistent snapshot while +the app writes. + +- **SQLite.** `VACUUM INTO '/backups/app.db'`, or the `.backup` command of the `sqlite3` shell. +- **PostgreSQL.** `pg_dump --format custom`, of the schema the adapter uses, or the provider's + point-in-time recovery. +- **Durable Object.** Nothing to do: Cloudflare keeps 30 days of point-in-time history of every + object's storage. + +## Restoring + +At start, stop the app, put the copy back and start it: + +- **SQLite.** Copy the backup to the database's path, or next to it and point `path` at it. +- **PostgreSQL.** `pg_restore`, or a point-in-time restore to another server and `url` pointed at + it. +- **Durable Object.** `ctx.storage.onNextSessionRestoreBookmark(bookmark)` with a bookmark from + `getBookmarkForTime`, then `ctx.abort()`: the object restarts on the restored storage. + +The app finds the restore before it serves anything, by the forget ledger's mark, reads the +ledger, forgets again whoever the copy brought back, and the read models hear of it before the +start resolves. A ledger that does not answer within ten seconds is left to finish in the +background, unless the ledger is [strict](/guides/personal-data/#what-the-runtime-does-with-it). +Without a forget ledger nothing outside the database says it went back: run `bounda restored`, +then [`bounda forget --from`](/reference/cli/#bounda-forgotten-and-bounda-forget) with the list of +whom you forgot since the copy, saved before the restore. + +Under a running app, restore in place, in one transaction, so the app sees the database as it was +or as the copy has it, never half restored: + +- **SQLite.** In one write transaction on the app's own engine. Stop the app first to restore a + libSQL database with the `sqlite3` shell's `.restore`: SQLite's backup API, run by another + library on a file libSQL has open, left it corrupted in our tests. +- **PostgreSQL.** `pg_restore --clean --if-exists --single-transaction --schema `. + +Every instance finds it on its next pass, before the pass delivers any event or claims any +scheduled command, and an instance in the `web` role within five seconds; +[restores](/guides/personal-data/#restores) says how. It logs + +``` +the database went back to an earlier copy; background jobs start over +``` + +and reads the ledger at once. A request served between the restore and that pass reads the copy as +it is. A Durable Object has no running app to restore under: restoring its storage restarts it. +An app without a forget ledger logs an error as well, since it has no record of whom to forget. + +## What is guaranteed + +Bounda's restore tests run this on SQLite (node:sqlite and libSQL), PostgreSQL and the Durable +Object, with the app stopped and, where the engine allows it, running: persons write their data, +a backup is taken, some are forgotten, the master key rotates, more is written, and the copy is +restored. + +- **Whoever was forgotten after the copy is forgotten again,** with `reason: "restored"`: no key, + snapshot, idempotency record of a rejection or scheduled command of theirs, no row a projection + of `PersonalDataForgotten` removes, their fields `null` in the export, in every event and dead + letter and in the state of their processes, and none of their values in any row of the + database. +- **Their scheduled commands never run**, even when they fall due as the copy comes back. +- **Whoever was forgotten before the copy stays forgotten,** once: the copy holds their forgetting. + Someone who registered after the copy is gone with it, and their entry in the ledger finds no + key to remove. +- **Everyone else reads as before,** their keys wrapped under the previous master key open and are + rewrapped under the current one. +- **The app goes on:** projections stay current, new persons register, `forget` and the rotation + work, and the next normal start forgets nobody again. A failed reconciliation the copy + remembers does not hold the restored start back, and `bounda status` counts the restore. +- **Without a forget ledger,** `bounda restored` counts the restore and starts the jobs over, and + `bounda forget --from` forgets again whom you list. + +What ran after the copy was taken runs again: scheduled commands of persons nobody forgot, and +reactions to events the copy holds but its subscribers had not reached yet. Effects outside the +database, such as an email, happen twice unless their receiver deduplicates them. + +## For adapter authors + +`restoreContract` from `@bounda-dev/core/adapter/testing` is the scenario above. Give it a fresh +database, a ledger store apart from it, and the engine's backup and restore: + +```ts +import { restoreContract } from "@bounda-dev/core/adapter/testing"; + +describe("my adapter", () => { + restoreContract({ + hot: true, + create: async () => ({ + storage: myAdapter({ url }), + forgetLedger: myAdapter({ url: ledgerUrl }), + backUp: () => takeBackup(), + restore: async () => { + await putTheBackupBack(); + return myAdapter({ url }); + }, + }), + }); +}); +``` + +`hot` says whether the engine restores in place under a running app. Each forgotten person is +checked with [`expectForgotten`](/guides/personal-data/#in-tests) over the adapter's `inspect`, every table +under its prefix, and over what the app reads back of them. diff --git a/docs/src/content/docs/guides/deployment.mdx b/docs/src/content/docs/guides/deployment.mdx index b2873615..fa97da70 100644 --- a/docs/src/content/docs/guides/deployment.mdx +++ b/docs/src/content/docs/guides/deployment.mdx @@ -363,7 +363,8 @@ them again when the app starts on the restored copy, or at once when the copy is running app: every running instance finds that the database went back, and its background jobs start over (see [restores](/guides/personal-data/#restores)). Restored while the app was stopped, the ledger tells it at the next start, and without a ledger run `bounda restored`. Set the -ledger's `keepFor` longer than the oldest backup you keep. +ledger's `keepFor` longer than the oldest backup you keep. [Backups and restores](/guides/backups/) +has the steps for each engine. ## Observability diff --git a/docs/src/content/docs/guides/personal-data.md b/docs/src/content/docs/guides/personal-data.md index 091d5972..6d5832b9 100644 --- a/docs/src/content/docs/guides/personal-data.md +++ b/docs/src/content/docs/guides/personal-data.md @@ -1020,14 +1020,19 @@ and `bounda status` counts the restores found. How it is found depends on when t | The copy was restored | Found by | What is guaranteed | | --- | --- | --- | -| under the running app | every instance, on its next pass, or within five seconds in the `web` role | Any copy taken before the last write the instances saw. A copy taken after it differs from the database in nothing but waits, which the jobs read again anyway. | +| under the running app | every instance, on its next pass, before it delivers an event or claims a scheduled command, or within five seconds in the `web` role | Any copy taken before the last write the instances saw, or before the newest event an instance read or committed. A copy taken after both differs from the database in nothing but waits, which the jobs read again anyway. | | while the app was stopped, with a forget ledger | the first instance to start, before it serves anything | Any copy taken before the last mark the ledger holds, which follows every move of the counter and a stop after any write, or before a forgetting the ledger holds and the copy does not. Only after a crash can a copy of the last five seconds of writes, with no forgetting among them, go unseen: it differs from the database in waits alone. | | while the app was stopped, without a ledger | `bounda restored`, which you run after restoring | What you declare: the database alone cannot tell a copy from itself. | Under the running app, every instance moves a counter in `bounda_checkpoints` one forward after it wrote, at most every five seconds, and remembers what it read: a copy holds less, and the first instance to see it counts the restore in another row, which every other instance then sees -changed. An idle instance writes nothing, so neither does its watch. With a forget ledger the +changed. An idle instance writes nothing, so neither does its watch. Every instance also watches +the newest event it read, or committed as a forgetting, at its position: events are only ever +appended, so a copy that lost it is found at once, even one taken in the five seconds before the +counter moved. A pass that finds a restore reconciles the ledger before the dispatcher delivers +anything or the worker claims anything, so a scheduled command of someone the copy brought back +never runs. [Backups and restores](/guides/backups/) has the steps for each engine. With a forget ledger the counter is also raised in the ledger, as a mark outside the database, which is how a start tells a copy restored while the app was stopped; and a forgetting the ledger holds that the database does not is a restore too, whatever the counter says, so the forgettings a copy undid are diff --git a/packages/cloudflare/test/restore.test.ts b/packages/cloudflare/test/restore.test.ts new file mode 100644 index 00000000..c45b23ee --- /dev/null +++ b/packages/cloudflare/test/restore.test.ts @@ -0,0 +1,101 @@ +import { runInDurableObject } from "cloudflare:test"; +import { env } from "cloudflare:workers"; +import { restoreContract } from "@bounda-dev/core/adapter/testing"; +import { defineForgetLedger } from "@bounda-dev/core/config"; +import { describe } from "vitest"; +import { durableObjectAdapter } from "../src/adapter.ts"; +import { createKvForgetLedger } from "../src/forget-ledger.ts"; +import { insideObject } from "./inside-object.ts"; + +interface SchemaObject { + readonly type: string; + readonly name: string; + readonly sql: string; +} + +interface Copy { + readonly schema: readonly SchemaObject[]; + readonly rows: Readonly[]>>; +} + +// Everything the app keeps in the object's SQLite database; workerd's own tables stay. +const schemaOf = (sql: SqlStorage): SchemaObject[] => + sql + .exec<{ type: string; name: string; sql: string | null }>( + "SELECT type, name, sql FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' AND name NOT LIKE '_cf_%'", + ) + .toArray() + .flatMap(({ type, name, sql: text }) => (text === null ? [] : [{ type, name, sql: text }])); + +const tablesOf = (schema: readonly SchemaObject[]): readonly string[] => + schema.filter(({ type }) => type === "table").map(({ name }) => name); + +const copyOf = (sql: SqlStorage): Copy => { + const schema = schemaOf(sql); + return { + schema, + rows: Object.fromEntries( + tablesOf(schema).map((table) => [table, sql.exec(`SELECT * FROM "${table}"`).toArray()]), + ), + }; +}; + +// Workerd keeps no point-in-time history of an object's storage, which Cloudflare restores to a +// bookmark and then restarts the object: the copy is every table, index and trigger with its rows, +// and restoring puts back exactly those, in one transaction. +const restoreFrom = (storage: DurableObjectStorage, copy: Copy): void => { + storage.transactionSync(() => { + for (const table of tablesOf(schemaOf(storage.sql))) storage.sql.exec(`DROP TABLE "${table}"`); + for (const { sql: text } of copy.schema.filter(({ type }) => type === "table")) { + storage.sql.exec(text); + } + for (const [table, rows] of Object.entries(copy.rows)) { + for (const row of rows) { + const columns = Object.keys(row); + storage.sql.exec( + `INSERT INTO "${table}" (${columns.map((column) => `"${column}"`).join(", ")}) VALUES (${columns.map(() => "?").join(", ")})`, + ...columns.map((column) => row[column] ?? null), + ); + } + } + // After the rows, so no trigger writes what the copy already holds. + for (const { sql: text } of copy.schema.filter(({ type }) => type !== "table")) { + storage.sql.exec(text); + } + }); +}; + +let objects = 0; + +describe("restoring a Durable Object's storage", () => { + restoreContract({ + hot: false, + create: async () => { + objects += 1; + const stub = env.BARE.get(env.BARE.newUniqueId()); + const adapter = insideObject( + stub, + await runInDurableObject(stub, (_instance, state) => + durableObjectAdapter({ storage: state.storage, options: {} }), + ), + ); + let copy: Copy | null = null; + return { + storage: adapter, + // KV, as on Cloudflare: a restore of the object leaves it alone. + forgetLedger: defineForgetLedger(() => + createKvForgetLedger({ namespace: env.FORGET_LEDGER, prefix: `restore-${objects}/` }), + ), + backUp: async () => { + copy = await runInDurableObject(stub, (_instance, state) => copyOf(state.storage.sql)); + }, + restore: async () => { + const taken = copy; + if (taken === null) throw new Error("no copy was taken"); + await runInDurableObject(stub, (_instance, state) => restoreFrom(state.storage, taken)); + return adapter; + }, + }; + }, + }); +}); diff --git a/packages/core/src/adapter/sqlite/restore.test.ts b/packages/core/src/adapter/sqlite/restore.test.ts new file mode 100644 index 00000000..14c20480 --- /dev/null +++ b/packages/core/src/adapter/sqlite/restore.test.ts @@ -0,0 +1,186 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { backup, DatabaseSync } from "node:sqlite"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { type BoundaApp, createApp } from "../../kernel/app.ts"; +import { createRecordingLogger } from "../../kernel/test-support.ts"; +import type { EventPayloadArgs, PayloadArgs } from "../../modules/payload.ts"; +import type { Registry } from "../../modules/registry.ts"; +import type { Adapter } from "../adapter.ts"; +import { restoreContract } from "../testing/restore.contract.ts"; +import { createNodeSqliteAdapter } from "./node-sqlite.ts"; + +let directory: string; +let files = 0; + +beforeAll(async () => { + directory = await mkdtemp(join(tmpdir(), "bounda-restore-")); +}); + +afterAll(async () => { + await rm(directory, { recursive: true, force: true }); +}); + +const fresh = (name: string): string => { + files += 1; + return join(directory, `${name}-${files}.db`); +}; + +describe("restoring a node:sqlite database", () => { + restoreContract({ + hot: true, + create: async () => { + const path = fresh("store"); + const copy = fresh("backup"); + const { adapter, db } = createNodeSqliteAdapter(path); + return { + storage: adapter, + forgetLedger: createNodeSqliteAdapter(fresh("ledger")).adapter, + // An online copy, as a backup tool takes it while the app runs. + backUp: async () => { + db.exec(`VACUUM INTO '${copy}'`); + }, + // In place, through SQLite's backup API, as the sqlite3 shell's .restore does: the app's + // connection sees the copy at its next statement. + restore: async () => { + const source = new DatabaseSync(copy, { readOnly: true }); + try { + await backup(source, path); + } finally { + source.close(); + } + return adapter; + }, + }; + }, + }); +}); + +const registry = { + aggregates: { + customer: { + state: { initialState: {} }, + events: { + customerRegistered: { + payload: ({ z, personal }: EventPayloadArgs) => z.object({ name: personal(z.string()) }), + evolve: () => ({}), + }, + }, + commands: { + registerCustomer: { + module: { + payload: ({ z }: PayloadArgs) => z.object({ customerId: z.string(), name: z.string() }), + handler: ({ + command, + events, + }: { + command: { payload: { name: string } }; + events: Record unknown>; + }) => [events.customerRegistered?.({ name: command.payload.name })], + }, + }, + }, + policies: {}, + processes: {}, + }, + }, + readModels: {}, +} as unknown as Registry; + +const register = (app: BoundaApp, customerId: string): Promise => + ( + app.commands as unknown as Record Promise> + ).registerCustomer?.({ customerId, name: `name of ${customerId}` }) ?? Promise.resolve(); + +const nameOf = async (app: BoundaApp, customerId: string): Promise => + (await app.exportPersonalData(customerId)).events[0]?.data.name; + +// A database, its copy taken after c-1 registered, and c-1 forgotten since. +const forgottenSinceTheCopy = async (forgetLedger?: Adapter) => { + const path = fresh("store"); + const copy = fresh("backup"); + const { adapter, db } = createNodeSqliteAdapter(path); + const config = { + storage: adapter, + ...(forgetLedger === undefined + ? {} + : { runtime: { personalData: { forgetLedger: { store: forgetLedger } } } }), + }; + const before = await createApp({ registry, config }); + await register(before, "c-1"); + db.exec(`VACUUM INTO '${copy}'`); + await before.forget("c-1"); + await before.stop(); + const restore = async (): Promise => { + const source = new DatabaseSync(copy, { readOnly: true }); + try { + await backup(source, path); + } finally { + source.close(); + } + }; + return { config, restore }; +}; + +describe("a restore under a running app", () => { + it("is noticed even before the app's first pass", async () => { + const { config, restore } = await forgottenSinceTheCopy( + createNodeSqliteAdapter(fresh("ledger")).adapter, + ); + const app = await createApp({ registry, config }); + await restore(); + await app.runUntilIdle(); + expect(await nameOf(app, "c-1")).toBeNull(); + await app.stop(); + }); + + it("is noticed right after a forgetting, before any pass has read it", async () => { + const path = fresh("store"); + const copy = fresh("backup"); + const { adapter, db } = createNodeSqliteAdapter(path); + const app = await createApp({ + registry, + config: { + storage: adapter, + runtime: { + personalData: { + forgetLedger: { store: createNodeSqliteAdapter(fresh("ledger")).adapter }, + }, + }, + }, + }); + await register(app, "c-1"); + await app.runUntilIdle(); + db.exec(`VACUUM INTO '${copy}'`); + await app.forget("c-1"); + const source = new DatabaseSync(copy, { readOnly: true }); + try { + await backup(source, path); + } finally { + source.close(); + } + expect(await nameOf(app, "c-1")).toBe("name of c-1"); + await app.runUntilIdle(); + expect(await nameOf(app, "c-1")).toBeNull(); + await app.stop(); + }); + + it("without a forget ledger, is logged as bringing persons back", async () => { + const { config, restore } = await forgottenSinceTheCopy(); + const { logger, entries } = createRecordingLogger(); + const app = await createApp({ registry, config, logger }); + await restore(); + await app.runUntilIdle(); + expect(await nameOf(app, "c-1")).toBe("name of c-1"); + expect(entries).toContainEqual({ + level: "error", + message: + "without runtime.personalData.forgetLedger, whoever was forgotten after the copy was taken is back until forgotten again, as bounda forget --from does", + }); + expect( + entries.filter(({ message }) => message.startsWith("without runtime.personalData")), + ).toHaveLength(1); + await app.stop(); + }); +}); diff --git a/packages/core/src/adapter/testing/checkpoint-store.contract.ts b/packages/core/src/adapter/testing/checkpoint-store.contract.ts index c2c159da..3ac2f3d5 100644 --- a/packages/core/src/adapter/testing/checkpoint-store.contract.ts +++ b/packages/core/src/adapter/testing/checkpoint-store.contract.ts @@ -136,7 +136,7 @@ export const checkpointStoreContract: CheckpointStoreContractFunction = ({ creat // A stream whose head the test moves, as appends do. const stream = { head: 0 }; - const events = { lastPosition: async () => stream.head }; + const events = { lastPosition: async () => stream.head, readAll: async () => [] }; const watchOver = () => createRestoreWatch({ checkpoints: store, events, marks: null, logger: silentLogger }); const at = (ms: number) => new Date(start + ms); diff --git a/packages/core/src/adapter/testing/index.ts b/packages/core/src/adapter/testing/index.ts index 0448a73b..a15306ef 100644 --- a/packages/core/src/adapter/testing/index.ts +++ b/packages/core/src/adapter/testing/index.ts @@ -60,6 +60,12 @@ export type { TransactionLocking, } from "./read-model-transaction.contract.ts"; export { readModelTransactionContract } from "./read-model-transaction.contract.ts"; +export type { + RestoreContractArgs, + RestoreContractFunction, + RestoreTarget, +} from "./restore.contract.ts"; +export { restoreContract } from "./restore.contract.ts"; export type { SchedulerContractArgs, SchedulerContractFunction } from "./scheduler.contract.ts"; export { schedulerContract } from "./scheduler.contract.ts"; export type { diff --git a/packages/core/src/adapter/testing/restore.contract.ts b/packages/core/src/adapter/testing/restore.contract.ts new file mode 100644 index 00000000..2d991d48 --- /dev/null +++ b/packages/core/src/adapter/testing/restore.contract.ts @@ -0,0 +1,552 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { RuntimeRole } from "../../config/types.ts"; +import { type Clock, createFixedClock } from "../../contracts/clock.ts"; +import { ValidationError } from "../../contracts/errors.ts"; +import type { StoredEvent } from "../../contracts/event.ts"; +import { type BoundaApp, createApp } from "../../kernel/app.ts"; +import { + defineForgetLedger, + type ForgetLedgerSource, +} from "../../kernel/personal-data/forget-ledger.ts"; +import type { KeyWrapperDefinition } from "../../kernel/personal-data/key-wrapper.ts"; +import { masterKey } from "../../kernel/personal-data/master-key.ts"; +import { GENERATION_INTERVAL_MS } from "../../kernel/shared/restores.ts"; +import type { + PersonalDataForgottenEvent, + PersonalDataForgottenPayload, +} from "../../modules/bounda-events.ts"; +import type { CommandPayloadArgs, EventPayloadArgs, PayloadArgs } from "../../modules/payload.ts"; +import type { Registry } from "../../modules/registry.ts"; +import type { FieldsArgs } from "../../modules/view.ts"; +import { expectForgotten } from "../../testing/expect-forgotten.ts"; +import type { Adapter } from "../adapter.ts"; +import type { ForgetLedger } from "../storage/forget-ledger.ts"; +import type { Table } from "../storage/table.ts"; + +/** + * One database an app runs on, and the operator's tools for it. + */ +export interface RestoreTarget { + /** + * The app's storage, over the database `backUp` copies. + */ + readonly storage: Adapter; + /** + * Where the forget ledger lives: anything a restore of the storage's database leaves alone. + */ + readonly forgetLedger: ForgetLedgerSource; + /** + * Copies the storage's database, with an app open on it, as a backup tool does. + */ + backUp(): Promise; + /** + * Puts the copy back, and resolves to the storage an app opens on it from then on: `storage` + * itself for a restore made in place. Called with an app open on `storage` only when the + * contract's `hot` is `true`, and after it stopped otherwise. + */ + restore(): Promise; +} + +export interface RestoreContractArgs { + /** + * A fresh, empty database, with a ledger store apart from it. + */ + readonly create: () => Promise; + /** + * Whether the database can be restored under a running app, in place, as the restore tools of + * PostgreSQL and SQLite do: then the contract also restores under one. + */ + readonly hot: boolean; +} + +export interface RestoreContractFunction { + (args: RestoreContractArgs): void; +} + +interface CustomerState { + readonly email: string | null; +} + +interface CustomerRow { + readonly customerId: string; + readonly email: string; + readonly name: string; +} + +type Builders = Readonly unknown>>; + +// Every handler that fails quotes the personal field it read, as a careless one does, so a dead +// letter holds it. +const registry = { + aggregates: { + customer: { + stateFingerprint: "restore-contract", + state: { initialState: { email: null } satisfies CustomerState }, + events: { + customerRegistered: { + payload: ({ z, personal }: EventPayloadArgs) => + z.object({ email: personal(z.string()), name: personal(z.string()) }), + evolve: ({ event }: { event: { payload: { email: string | null } } }) => ({ + email: event.payload.email, + }), + }, + customerNoted: { + payload: ({ z, personal }: EventPayloadArgs) => z.object({ note: personal(z.string()) }), + evolve: ({ state }: { state: CustomerState }) => state, + }, + }, + commands: { + registerCustomer: { + module: { + payload: ({ z }: PayloadArgs) => + z.object({ customerId: z.string(), email: z.string(), name: z.string() }), + handler: ({ + command, + events, + }: { + command: { payload: { email: string; name: string } }; + events: Builders; + }) => [ + events.customerRegistered?.({ + email: command.payload.email, + name: command.payload.name, + }), + ], + }, + }, + confirmCustomer: { + module: { + payload: ({ z }: PayloadArgs) => z.object({ customerId: z.string() }), + rejections: ({ state }: { state: CustomerState }) => ({ + NotYet: `${state.email ?? "This customer"} cannot be confirmed yet`, + }), + handler: ({ reject }: { reject: (name: "NotYet") => never }) => reject("NotYet"), + }, + }, + noteCustomer: { + module: { + payload: ({ z, personal }: CommandPayloadArgs) => + z.object({ customerId: z.string(), note: personal(z.string()) }), + handler: ({ + command, + events, + }: { + command: { payload: { note: string } }; + events: Builders; + }) => [events.customerNoted?.({ note: command.payload.note })], + }, + }, + }, + policies: { + welcomeOnCustomerRegistered: { + module: { + handler: async ({ event }: { event: { payload: { email: string | null } } }) => { + throw new ValidationError(`the mailer refuses ${event.payload.email}`, []); + }, + }, + }, + }, + processes: { + onboarding: { + module: { + config: ({ + events, + }: { + events: { customer: { CustomerRegistered: string; CustomerNoted: string } }; + }) => ({ + startedBy: [events.customer.CustomerRegistered], + completedBy: [events.customer.CustomerNoted], + }), + state: ({ z, personal }: CommandPayloadArgs) => + z.object({ email: personal(z.string()).optional() }), + }, + handlers: { + customer: { + customerRegistered: { + handler: ({ event }: { event: { payload: { email: string | null } } }) => + event.payload.email === null ? {} : { email: event.payload.email }, + }, + }, + }, + }, + }, + }, + }, + readModels: { + customers: { + view: { + fields: ({ f }: FieldsArgs) => ({ + customerId: f.string().primaryKey(), + email: f.string(), + name: f.string(), + }), + }, + projections: { + customer: { + customerRegistered: { + project: async ({ + event, + table, + }: { + event: StoredEvent< + "CustomerRegistered", + { email: string | null; name: string | null } + >; + table: Table; + }) => { + await table.upsert({ + customerId: event.aggregateId, + email: event.payload.email ?? "", + name: event.payload.name ?? "", + }); + }, + }, + }, + bounda: { + personalDataForgotten: { + project: async ({ + event, + table, + }: { + event: PersonalDataForgottenEvent; + table: Table; + }) => { + await table.delete({ customerId: event.payload.personId }); + }, + }, + }, + }, + queries: { + getCustomer: { + payload: ({ z }: PayloadArgs) => z.object({ customerId: z.string() }), + repository: ({ table, customerId }: { table: Table; customerId: string }) => + table.findOne({ customerId }), + handler: ({ repositoryData }: { repositoryData: CustomerRow | null }) => repositoryData, + }, + }, + }, + }, +} as unknown as Registry; + +const START = new Date("2026-11-09T09:00:00.000Z"); +// Longer than the scenario runs for, so no periodic reconciliation hides a restore the app missed. +const RECONCILE_EVERY_MS = 24 * 60 * 60_000; +// When the scheduled commands written before the copy fall due. +const NOTE_DELAY_MS = 60 * 60_000; + +// Two master keys, the second the one a rotation moves to. +const bytes = (seed: number): string => + btoa(String.fromCharCode(...Array.from({ length: 32 }, (_, index) => (seed + index * 7) % 256))); +const env = { RESTORE_KEY_1: bytes(1), RESTORE_KEY_2: bytes(2) }; +const BEFORE_ROTATION = masterKey({ from: "RESTORE_KEY_1" }); +const ROTATED = masterKey({ from: "RESTORE_KEY_2", previous: ["RESTORE_KEY_1"] }); +const AFTER_ROTATION = masterKey({ from: "RESTORE_KEY_2" }); + +/** + * What a person writes: every value is distinct enough that finding it anywhere means it is + * theirs. + */ +const valuesOf = (person: string) => ({ + email: `${person}.restore-contract@example.com`, + name: `${person} Restorecontractname`, + note: `${person} restore-contract note`, +}); + +type Commands = Readonly< + Record< + string, + ( + payload: unknown, + options?: { readonly idempotencyKey?: string; readonly delay?: string }, + ) => Promise + > +>; +type Queries = Readonly Promise>>; + +interface Started { + readonly app: BoundaApp; + readonly commands: Commands; + readonly queries: Queries; +} + +/** + * A ledger whose reads fail while `failing` is set. + */ +const faultyLedger = (source: ForgetLedgerSource) => { + const state = { failing: false }; + const definition = defineForgetLedger(async (args) => { + const inner: ForgetLedger = + source.kind === "bounda-forget-ledger" + ? await source.create(args) + : await source.createForgetLedger({ tenant: args.tenant, keepForMs: args.keepForMs }); + return { + ...inner, + read: async (readArgs) => { + if (state.failing) throw new Error("the ledger is down"); + return inner.read(readArgs); + }, + }; + }); + return { definition, state }; +}; + +/** + * What every adapter must guarantee when its database is restored from a backup, with the app + * running or at its next start: nobody forgotten after the copy was taken can be read again + * anywhere (events, read models, snapshots, idempotency records, scheduled commands, dead letters, + * process state, export, the database's own rows), and the app goes on working. Call it inside a + * `describe` of the adapter's test file. + */ +export const restoreContract: RestoreContractFunction = ({ create, hot }) => { + describe("restore contract", () => { + const stops: (() => Promise)[] = []; + + afterEach(async () => { + for (const stop of stops.splice(0).reverse()) await stop(); + }); + + const start = async ( + storage: Adapter, + forgetLedger: ForgetLedgerSource | null, + keyWrapper: KeyWrapperDefinition, + clock: Clock, + role: RuntimeRole = "all", + ): Promise => { + const app = await createApp({ + registry, + config: { + storage, + runtime: { + role, + personalData: { + keyWrapper, + ...(forgetLedger === null + ? {} + : { forgetLedger: { store: forgetLedger, reconcileEvery: RECONCILE_EVERY_MS } }), + }, + policies: { retry: { strategy: "none" } }, + overrides: { customer: { snapshot: { every: 1 } } }, + }, + }, + clock, + env, + }); + stops.push(() => app.stop()); + return { + app, + commands: app.commands as unknown as Commands, + queries: app.queries as unknown as Queries, + }; + }; + + // Everything a person leaves behind: events, a read model row, a snapshot, an idempotency + // record of a rejection that quotes them, a scheduled command, a dead letter and a process. + const write = async ({ app, commands }: Started, person: string): Promise => { + const { email, name, note } = valuesOf(person); + await commands.registerCustomer?.( + { customerId: person, email, name }, + { idempotencyKey: `register-${person}` }, + ); + await commands + .confirmCustomer?.({ customerId: person }, { idempotencyKey: `confirm-${person}` }) + .catch(() => undefined); + await commands.noteCustomer?.({ customerId: person, note }, { delay: "1h" }); + await app.runUntilIdle(); + }; + + const forgottenReasons = async (app: BoundaApp, person: string): Promise => + (await app.events.read({ limit: 1_000 })) + .filter((event) => event.type === "PersonalDataForgotten" && event.aggregateId === person) + .map((event) => (event.payload as PersonalDataForgottenPayload).reason); + + // Every table of the storage, and what the app reads back of the person. + const expectForgottenIn = async ( + storage: Adapter, + { app, queries }: Started, + person: string, + ): Promise => { + await expectForgotten(storage, person, { + values: Object.values(valuesOf(person)), + places: { + query: await queries.getCustomer?.({ customerId: person }), + export: await app.exportPersonalData(person), + "dead letters": await app.deadLetters.list(), + }, + }); + }; + + const expectReadable = async ({ app, queries }: Started, person: string): Promise => { + const { email, name } = valuesOf(person); + expect(await queries.getCustomer?.({ customerId: person })).toEqual({ + customerId: person, + email, + name, + }); + const exported = await app.exportPersonalData(person); + expect(exported.events[0]?.data).toEqual({ email, name }); + }; + + const scenario = async (restoring: "hot" | "at start"): Promise => { + const target = await create(); + const clock = createFixedClock(START); + // Before the copy: ana, bob and dan register, cy registers and is forgotten. + const first = await start(target.storage, target.forgetLedger, BEFORE_ROTATION, clock); + for (const person of ["ana", "bob", "cy", "dan"]) await write(first, person); + await first.app.forget("cy"); + await first.app.runUntilIdle(); + await target.backUp(); + + // After the copy: ana is forgotten, the master key rotates, then bob and eve, who registered + // after the copy, are forgotten, and more is written. + await first.app.forget("ana"); + await first.app.runUntilIdle(); + await first.app.stop(); + clock.advance(60_000); + const rotated = await start(target.storage, target.forgetLedger, ROTATED, clock); + await rotated.app.runUntilIdle(); + await write(rotated, "eve"); + await rotated.app.forget("bob"); + await rotated.app.forget("eve"); + await rotated.app.runUntilIdle(); + for (const person of ["ana", "bob", "cy", "eve"]) { + await expectForgottenIn(target.storage, rotated, person); + } + + let storage: Adapter; + let restored: Started; + if (restoring === "hot") { + storage = await target.restore(); + } else { + await rotated.app.stop(); + storage = await target.restore(); + } + // The scheduled commands in the copy fall due: ana's and bob's must not run. + clock.advance(NOTE_DELAY_MS); + if (restoring === "hot") { + restored = rotated; + await restored.app.runUntilIdle(); + } else { + restored = await start(storage, target.forgetLedger, ROTATED, clock); + } + + // Everyone forgotten after the copy was taken, and still holding a key in it, is forgotten + // again; cy, forgotten before it, and eve, who is not in it, are not. + for (const person of ["ana", "bob"]) { + expect(await forgottenReasons(restored.app, person)).toEqual(["restored"]); + await expectForgottenIn(storage, restored, person); + } + expect(await forgottenReasons(restored.app, "cy")).toEqual(["requested"]); + expect(await forgottenReasons(restored.app, "eve")).toEqual([]); + await expectForgottenIn(storage, restored, "cy"); + await expectForgottenIn(storage, restored, "eve"); + expect(await restored.app.getLag()).toMatchObject({ forgetLedger: { reconciled: true } }); + // Found by the newest event this instance saw under a running app, by the ledger's mark at + // start. + expect((await restored.app.status()).restoresFound).toBe(1); + + // dan, whom nobody forgot, reads as before, his key rewrapped again under the new master key. + await restored.app.runUntilIdle(); + await expectReadable(restored, "dan"); + const noted = (await restored.app.events.read({ limit: 1_000 })) + .filter((event) => event.type === "CustomerNoted") + .map((event) => event.aggregateId); + expect(noted).toContain("dan"); + expect(noted).not.toContain("ana"); + expect(noted).not.toContain("bob"); + + // The app goes on: a new person registers and is projected, dan is forgotten. + await write(restored, "fay"); + await expectReadable(restored, "fay"); + await restored.app.forget("dan"); + await restored.app.runUntilIdle(); + await expectForgottenIn(storage, restored, "dan"); + await restored.app.stop(); + + // A normal start forgets nobody again, and opens fay's key without the old master key. + clock.advance(60_000); + const next = await start(storage, target.forgetLedger, AFTER_ROTATION, clock); + await expectReadable(next, "fay"); + for (const person of ["ana", "bob", "dan"]) { + expect((await forgottenReasons(next.app, person)).length).toBe(1); + } + }; + + it("forgets again at the start after a restore whoever was forgotten since the copy, everywhere, and goes on", async () => { + await scenario("at start"); + }); + + it.skipIf(!hot)( + "forgets again, at the worker's next pass, whoever a restore under the running app brings back", + async () => { + await scenario("hot"); + }, + ); + + it("reconciles at the start after a restore whatever the copy says of a failed reconciliation", async () => { + const target = await create(); + const clock = createFixedClock(START); + const { definition, state } = faultyLedger(target.forgetLedger); + const first = await start(target.storage, definition, BEFORE_ROTATION, clock); + await write(first, "ana"); + // A reconciliation fails, and its hold is in the copy. + state.failing = true; + clock.advance(RECONCILE_EVERY_MS); + await first.app.runUntilIdle(); + expect(await first.app.getLag()).toMatchObject({ forgetLedger: { reconciled: false } }); + await target.backUp(); + state.failing = false; + await first.app.forget("ana"); + await first.app.runUntilIdle(); + await first.app.stop(); + + const storage = await target.restore(); + const restored = await start(storage, definition, BEFORE_ROTATION, clock); + expect(await forgottenReasons(restored.app, "ana")).toEqual(["restored"]); + expect(await restored.app.getLag()).toMatchObject({ forgetLedger: { reconciled: true } }); + await expectForgottenIn(storage, restored, "ana"); + }); + + it.skipIf(!hot)( + "forgets again whoever a restore under an instance that runs no worker brings back", + async () => { + const target = await create(); + const clock = createFixedClock(START); + const web = await start(target.storage, target.forgetLedger, BEFORE_ROTATION, clock, "web"); + web.app.start(); + await write(web, "ana"); + await target.backUp(); + await web.app.forget("ana"); + await web.app.catchUpReadModels(); + await expectForgottenIn(target.storage, web, "ana"); + await target.restore(); + clock.advance(GENERATION_INTERVAL_MS); + // Its watch reconciles and catches the read models up by itself. + await vi.waitFor(async () => { + expect(await forgottenReasons(web.app, "ana")).toEqual(["restored"]); + await expectForgottenIn(target.storage, web, "ana"); + }); + }, + ); + + it("starts over when an operator says a database without a forget ledger was restored, which bounda forget --from then forgets again", async () => { + const target = await create(); + const clock = createFixedClock(START); + const first = await start(target.storage, null, BEFORE_ROTATION, clock); + await write(first, "ana"); + await target.backUp(); + await first.app.forget("ana"); + await first.app.runUntilIdle(); + await first.app.stop(); + + const storage = await target.restore(); + const restored = await start(storage, null, BEFORE_ROTATION, clock); + // Nothing outside the database says it went back. + expect((await restored.app.status()).restoresFound).toBe(0); + await restored.app.restored(); + expect((await restored.app.status()).restoresFound).toBe(1); + expect(await restored.app.forgetAgain([{ personId: "ana" }])).toEqual(["ana"]); + await restored.app.runUntilIdle(); + await expectForgottenIn(storage, restored, "ana"); + }); + }); +}; diff --git a/packages/core/src/kernel/app.test.ts b/packages/core/src/kernel/app.test.ts index f7c5bf89..d0fe79e4 100644 --- a/packages/core/src/kernel/app.test.ts +++ b/packages/core/src/kernel/app.test.ts @@ -824,7 +824,9 @@ describe("policies and processes follow the stream from when they are deployed", ]); expect(lag.maxLag).toBe(0); const { checkpointStore } = await storage.createStorage(); + // The generation the restore watch moved after the pass that projected the order. expect((await checkpointStore.list()).map((checkpoint) => checkpoint.subscriber)).toEqual([ + "runtime:generation", "projection:orderSummary", ]); await app.stop(); diff --git a/packages/core/src/kernel/app.ts b/packages/core/src/kernel/app.ts index e8ec3fd6..ae44c0f3 100644 --- a/packages/core/src/kernel/app.ts +++ b/packages/core/src/kernel/app.ts @@ -73,7 +73,7 @@ import { import { createScheduledCommandWorker } from "./scheduler/worker.ts"; import { guardedLogger } from "./shared/guarded-logger.ts"; import { ignoredRetries, type PendingRetries } from "./shared/pending-retries.ts"; -import { createRestoreWatch, GENERATION_INTERVAL_MS } from "./shared/restores.ts"; +import { createRestoreWatch, GENERATION_INTERVAL_MS, type Restored } from "./shared/restores.ts"; import { createSnapshotSubscriber, takesSnapshots } from "./snapshot/subscriber.ts"; import { type AppStatus, readAppStatus } from "./status.ts"; import { ATTRIBUTES, METRICS, meter, tracedOperation } from "./telemetry.ts"; @@ -492,10 +492,25 @@ export const assembleApp: AssembleAppFunction = async ({ const close = ledger.close.bind(ledger); opened.unshift(close); } + const restores = createRestoreWatch({ + checkpoints: adapted.checkpointStore, + events: adapted.eventStore, + marks: ledger, + logger, + }); const written = ledger === null ? adapted - : withLedgerAppend({ storage: adapted, ledger, keepForMs, clock, logger }); + : withLedgerAppend({ + storage: adapted, + ledger, + keepForMs, + clock, + logger, + committed: (forgettings) => { + for (const event of forgettings) restores.saw(event); + }, + }); const aggregates = buildAggregates({ registry, ports: ports.byAggregate, config }); const processDefinitions = buildProcesses({ registry, aggregates, config }); const storage = appStorage({ @@ -668,12 +683,14 @@ export const assembleApp: AssembleAppFunction = async ({ logger, errorText, }); - const restores = createRestoreWatch({ - checkpoints: written.checkpointStore, - events: written.eventStore, - marks: ledger, - logger, - }); + // A restore found outside a worker's pass: the jobs start over and the ledger is reconciled + // at once, unless this instance's worker handled it already. Resolves how many it forgot. It + // reads the worker, built after the dispatcher, only when called: no pass runs before. + const restartFound = async (found: Restored): Promise => { + if (!(await worker.restored(found))) return 0; + const forgotten = await reconciliation?.reconcile(clock.now()); + return typeof forgotten === "number" ? forgotten : 0; + }; const dispatcher = createDispatcher({ eventStore: storage.eventStore, checkpointStore: storage.checkpointStore, @@ -716,6 +733,12 @@ export const assembleApp: AssembleAppFunction = async ({ keys: storage.personalKeys, logger, }), + // Nothing is delivered on a restored database before whom it brought back is forgotten again: + // the pass then delivers that forgetting to the read models. + beforePass: async () => { + const found = await restores.check(clock.now()); + if (found !== null) await restartFound(found); + }, clock, logger, errorText, @@ -817,7 +840,7 @@ export const assembleApp: AssembleAppFunction = async ({ // holds it brought back are cleared before anything runs, and the reconciliation below forgets // again whom it brought back. const cold = await restores.start(); - if (cold !== null) await worker.restored(cold.first); + if (cold !== null) await worker.restored(cold); // An instance that runs no worker still watches for a restore, and reconciles the ledger at // once when it finds one, since no pass of its own would. const watchForRestores = (): (() => void) => { @@ -825,9 +848,8 @@ export const assembleApp: AssembleAppFunction = async ({ const next = (): void => { cancel = clock.after(GENERATION_INTERVAL_MS, async () => { const found = await restores.check(clock.now()); - if (found !== null) { - await worker.restored(found.first); - await reconciliation?.reconcile(clock.now()); + if (found !== null && (await restartFound(found)) > 0) { + await dispatcher.catchUp("projection"); } next(); }); @@ -855,7 +877,7 @@ export const assembleApp: AssembleAppFunction = async ({ } else if (forgotten > 0) { // The ledger held a forgetting the database does not: a restore, whatever the mark said. if (cold === null) { - await worker.restored((await restores.declare()).first, { reconciled: true }); + await worker.restored(await restores.declare(), { reconciled: true }); } await dispatcher.catchUp("projection"); } @@ -1082,9 +1104,7 @@ export const assembleApp: AssembleAppFunction = async ({ return forgotten; }, restored: async () => { - const { first } = await restores.declare(); - await worker.restored(first); - await reconciliation?.reconcile(clock.now()); + await restartFound(await restores.declare()); }, pendingRebuilds: async () => { const paused = await Promise.all( diff --git a/packages/core/src/kernel/dispatch/dispatcher.test.ts b/packages/core/src/kernel/dispatch/dispatcher.test.ts index 7f85e867..32b2e40b 100644 --- a/packages/core/src/kernel/dispatch/dispatcher.test.ts +++ b/packages/core/src/kernel/dispatch/dispatcher.test.ts @@ -47,6 +47,42 @@ const recorder = ( }; describe("createDispatcher", () => { + it("runs beforePass at the start of every pass, before it delivers, and delivers nothing when it throws", async () => { + const { eventStore, checkpointStore } = await storage(); + await appendMany(eventStore, 2); + const steps: string[] = []; + let failing = false; + const subscriber: Subscriber = { + name: "a", + kind: "projection", + process: async (events) => { + steps.push(`deliver ${events.length}`); + return events.length; + }, + }; + const dispatcher = createDispatcher({ + clock: createFixedClock(), + eventStore, + checkpointStore, + subscribers: [subscriber], + batchSize: 1, + pollIntervalMs: 10, + beforePass: async () => { + steps.push("before"); + if (failing) throw new Error("the store cannot be read"); + }, + logger: silentLogger, + errorText: "include", + }); + await dispatcher.processOnce(); + expect(steps).toEqual(["before", "deliver 1"]); + await dispatcher.catchUp("projection"); + expect(steps).toEqual(["before", "deliver 1", "before", "deliver 1", "before"]); + failing = true; + await expect(dispatcher.runUntilIdle()).rejects.toThrow("the store cannot be read"); + expect(steps.at(-1)).toBe("before"); + }); + it("delivers batches per subscriber in order and checkpoints after each", async () => { const { eventStore, checkpointStore } = await storage(); await appendMany(eventStore, 5); diff --git a/packages/core/src/kernel/dispatch/dispatcher.ts b/packages/core/src/kernel/dispatch/dispatcher.ts index 6157c056..b4e060d9 100644 --- a/packages/core/src/kernel/dispatch/dispatcher.ts +++ b/packages/core/src/kernel/dispatch/dispatcher.ts @@ -138,6 +138,11 @@ export interface CreateDispatcherArgs { * Where each subscriber's last failure is kept, its text sealed; without it, nowhere. */ readonly lastFailures?: LastFailures; + /** + * Runs at the start of every pass, before anything is delivered, such as checking whether the + * store was restored under the app. + */ + readonly beforePass?: () => Promise; readonly clock: Clock; readonly logger: Logger; readonly errorText: ErrorText; @@ -177,6 +182,7 @@ export const createDispatcher: CreateDispatcherFunction = ({ idleIntervalMs = pollIntervalMs, notifier, lastFailures, + beforePass, clock, logger, errorText, @@ -274,6 +280,7 @@ export const createDispatcher: CreateDispatcherFunction = ({ }; const pass = async (wait: boolean, patient: boolean, only?: SubscriberKind): Promise => { + await beforePass?.(); let advanced = false; for (const subscriber of delivering) { if (only !== undefined && subscriber.kind !== only) continue; diff --git a/packages/core/src/kernel/personal-data/ledger-append.test.ts b/packages/core/src/kernel/personal-data/ledger-append.test.ts index 22ed466e..7781247a 100644 --- a/packages/core/src/kernel/personal-data/ledger-append.test.ts +++ b/packages/core/src/kernel/personal-data/ledger-append.test.ts @@ -29,6 +29,7 @@ const setUp = async ({ readonly logger?: Logger; } = {}) => { const appended: (readonly ForgetLedgerEntry[])[] = []; + const committed: string[][] = []; const inner = await memory().createStorage(); const storage: Storage = withLedgerAppend({ storage: inner, @@ -44,13 +45,16 @@ const setUp = async ({ keepForMs, clock: createFixedClock(NOW), logger, + committed: (forgettings) => { + committed.push(forgettings.map(({ id }) => id)); + }, }); - return { storage, appended }; + return { storage, appended, committed }; }; describe("the append to the forget ledger as a transaction commits", () => { it("appends the forgettings a transaction committed, in one call, and nothing else", async () => { - const { storage, appended } = await setUp(); + const { storage, appended, committed } = await setUp(); const result = await storage.transact(async (transaction) => { await transaction.eventStore.append({ aggregateType: "order", @@ -75,8 +79,10 @@ describe("the append to the forget ledger as a transaction commits", () => { { personId: "c-2", eventId: "f-c-2-1", forgottenAt: "2026-01-01T00:00:00.000Z" }, ], ]); + expect(committed).toEqual([["f-c-1-1", "f-c-2-1"]]); await storage.transact(async () => undefined); expect(appended).toHaveLength(1); + expect(committed).toHaveLength(1); await storage.transact((transaction) => transaction.eventStore.reseal([{ position: 1, payload: { total: 9 }, person: "c-9" }]), ); @@ -87,7 +93,7 @@ describe("the append to the forget ledger as a transaction commits", () => { }); it("appends nothing for a transaction that rolls back, nor for an earlier run of the work", async () => { - const { storage, appended } = await setUp(); + const { storage, appended, committed } = await setUp(); await expect( storage.transact(async (transaction) => { await transaction.eventStore.append({ @@ -100,6 +106,7 @@ describe("the append to the forget ledger as a transaction commits", () => { }), ).rejects.toThrow("rolled back"); expect(appended).toEqual([]); + expect(committed).toEqual([]); const inner = await memory().createStorage(); let runs = 0; const retrying = withLedgerAppend({ @@ -138,7 +145,7 @@ describe("the append to the forget ledger as a transaction commits", () => { }); it("leaves out a forgetting older than keepFor", async () => { - const { storage, appended } = await setUp({ keepForMs: 60_000 }); + const { storage, appended, committed } = await setUp({ keepForMs: 60_000 }); await storage.transact((transaction) => transaction.eventStore.append({ aggregateType: "bounda", @@ -148,6 +155,8 @@ describe("the append to the forget ledger as a transaction commits", () => { }), ); expect(appended).toEqual([]); + // Watched for a restore all the same. + expect(committed).toEqual([["f-c-1-1"]]); const { storage: fresh, appended: kept } = await setUp({ keepForMs: 60_001 }); await fresh.transact((transaction) => transaction.eventStore.append({ diff --git a/packages/core/src/kernel/personal-data/ledger-append.ts b/packages/core/src/kernel/personal-data/ledger-append.ts index acb182ca..ebe866dc 100644 --- a/packages/core/src/kernel/personal-data/ledger-append.ts +++ b/packages/core/src/kernel/personal-data/ledger-append.ts @@ -48,6 +48,11 @@ export interface WithLedgerAppendArgs { readonly keepForMs: number | null; readonly clock: Clock; readonly logger: Logger; + /** + * Called with the forgettings of each transaction once it committed, before they go to the + * ledger. + */ + readonly committed?: (forgettings: readonly StoredEvent[]) => void; } export interface WithLedgerAppendFunction { @@ -68,6 +73,7 @@ export const withLedgerAppend: WithLedgerAppendFunction = ({ keepForMs, clock, logger, + committed, }) => { let stalled = false; return { @@ -95,7 +101,9 @@ export const withLedgerAppend: WithLedgerAppendFunction = ({ }, }); }); - const entries = ledgerEntries({ events: appended, keepForMs, now: clock.now() }); + const forgettings = appended.filter(isForgetting); + if (forgettings.length > 0) committed?.(forgettings); + const entries = ledgerEntries({ events: forgettings, keepForMs, now: clock.now() }); if (entries.length > 0 && !stalled) { const appending = ledger.append(entries); const outcome = await waitForLedger(clock, appending).catch((error: unknown) => ({ diff --git a/packages/core/src/kernel/reactive-harness.ts b/packages/core/src/kernel/reactive-harness.ts index 940a987f..54e35e4f 100644 --- a/packages/core/src/kernel/reactive-harness.ts +++ b/packages/core/src/kernel/reactive-harness.ts @@ -14,6 +14,7 @@ import { createRestrictedRetries } from "./dead-letters/restricted.ts"; import { createDeadLetterRetry, type DeadLetterRetry } from "./dead-letters/retry.ts"; import { createDispatcher, type Dispatcher } from "./dispatch/dispatcher.ts"; import type { AppStorage } from "./events/app-event-store.ts"; +import type { ForgetLedgerReconciliation } from "./personal-data/reconciliation.ts"; import { buildPolicies, type PoliciesRuntime } from "./policy/build-policies.ts"; import { createDelayedPolicies } from "./policy/delayed.ts"; import { createPolicyExecutor, type PolicyExecutor } from "./policy/executor.ts"; @@ -25,6 +26,7 @@ import { createProjectionSubscriber } from "./projection/runner.ts"; import { buildReadModels, type ReadModelsRuntime } from "./read-model/build-read-models.ts"; import { createScheduledCommandWorker, type ScheduledCommandWorker } from "./scheduler/worker.ts"; import { ignoredRetries } from "./shared/pending-retries.ts"; +import type { RestoreWatch } from "./shared/restores.ts"; import { defaultPorts, withPersonalKeys } from "./test-support.ts"; export interface ReactiveHarness { @@ -56,6 +58,8 @@ export interface CreateReactiveHarnessArgs { * The storage to run on; the in-memory adapter unless given. */ readonly adapter?: Adapter; + readonly restores?: RestoreWatch; + readonly reconciliation?: ForgetLedgerReconciliation; } export interface CreateReactiveHarnessFunction { @@ -71,6 +75,8 @@ export const createReactiveHarness: CreateReactiveHarnessFunction = async ({ config: overrides = {}, logger = silentLogger, adapter = memory(), + restores, + reconciliation, }) => { const clock = createFixedClock(); const storage = withPersonalKeys(await adapter.createStorage(), clock); @@ -146,6 +152,8 @@ export const createReactiveHarness: CreateReactiveHarnessFunction = async ({ config, }), restrictedRetries: createRestrictedRetries({ storage, retries, clock, logger }), + restores, + reconciliation, reactive: [POLICIES_SUBSCRIBER, PROCESSES_SUBSCRIBER], config, ids, diff --git a/packages/core/src/kernel/scheduler/worker.test.ts b/packages/core/src/kernel/scheduler/worker.test.ts index 47cb8d30..3c169a41 100644 --- a/packages/core/src/kernel/scheduler/worker.test.ts +++ b/packages/core/src/kernel/scheduler/worker.test.ts @@ -53,6 +53,90 @@ const failingRegistry: Registry = { }; describe("scheduled command worker", () => { + it("says, on a restore its pass finds without a forget ledger, that the persons forgotten since are back", async () => { + const { logger, entries } = createRecordingLogger("omit"); + let finds: { first: boolean } | null = { first: true }; + const harness = await createReactiveHarness({ + registry: orderRegistry, + logger, + restores: { + start: async () => null, + check: async () => { + const result = finds; + finds = null; + return result; + }, + declare: async () => ({ first: true }), + flush: async () => undefined, + found: async () => 0, + saw: () => undefined, + }, + }); + expect(await harness.worker.runOnce()).toBe(0); + expect(entries.map(({ level, message }) => [level, message])).toEqual([ + ["warn", "the database went back to an earlier copy; background jobs start over"], + [ + "error", + "without runtime.personalData.forgetLedger, whoever was forgotten after the copy was taken is back until forgotten again, as bounda forget --from does", + ], + ]); + }); + + it("starts the jobs over on a restore its pass finds, and reconciles before it claims anything, once", async () => { + const steps: string[] = []; + const found = { first: true }; + let finds: { first: boolean } | null = found; + const { logger, entries } = createRecordingLogger("omit"); + const harness = await createReactiveHarness({ + registry: orderRegistry, + logger, + restores: { + start: async () => null, + check: async () => { + steps.push("check"); + const result = finds; + finds = null; + return result; + }, + declare: async () => ({ first: true }), + flush: async () => undefined, + found: async () => 0, + saw: () => undefined, + }, + reconciliation: { + reconcile: async () => 0, + sweep: async () => { + steps.push("sweep"); + return 1; + }, + nextDueAt: () => null, + reconciled: () => true, + status: async () => ({}) as never, + restored: async (first) => { + steps.push(`restored ${first}`); + }, + }, + }); + const { scheduler } = harness.storage; + const nextDueAt = scheduler.nextDueAt.bind(scheduler); + vi.spyOn(scheduler, "nextDueAt").mockImplementation((args) => { + steps.push("claim"); + return nextDueAt(args); + }); + expect(await harness.worker.runOnce()).toBe(2); + expect(steps.slice(0, 5)).toEqual(["check", "restored true", "sweep", "claim", "sweep"]); + expect(entries.map(({ message }) => message)).toEqual([ + "the database went back to an earlier copy; background jobs start over", + ]); + // Found by this pass already: a dispatcher, or another pass, that shared the check leaves it. + expect(await harness.worker.restored(found)).toBe(false); + finds = found; + await harness.worker.runOnce(); + expect(steps.filter((step) => step.startsWith("restored"))).toEqual(["restored true"]); + expect(await harness.worker.restored({ first: false })).toBe(true); + expect(steps.at(-1)).toBe("restored false"); + }); + it("purges the expired idempotency keys once per interval, and goes on when it cannot", async () => { const { logger, entries } = createRecordingLogger("omit"); const harness = await createReactiveHarness({ registry: orderRegistry, logger }); diff --git a/packages/core/src/kernel/scheduler/worker.ts b/packages/core/src/kernel/scheduler/worker.ts index 35d5d7d8..2410ced1 100644 --- a/packages/core/src/kernel/scheduler/worker.ts +++ b/packages/core/src/kernel/scheduler/worker.ts @@ -32,7 +32,7 @@ import { PROCESS_DEADLINE_COMMAND, type ProcessDeadlinePayload } from "../proces import type { ProcessDeadlines } from "../process/deliver-deadline.ts"; import { PROCESSES_SUBSCRIBER } from "../process/runner.ts"; import { createMutex } from "../shared/mutex.ts"; -import type { RestoreWatch } from "../shared/restores.ts"; +import type { Restored, RestoreWatch } from "../shared/restores.ts"; import { classifyFailure, errorIdentity, errorLabel, retryDelayMs } from "../shared/retry.ts"; import { appendSystemEvent, @@ -64,9 +64,10 @@ export interface ScheduledCommandWorker { /** * Starts every background job over from what the database holds, as after a restore, between * passes; `first` also clears the holds the copy brought back, and `reconciled` leaves alone - * the reconciliation that found it. Never rejects. + * the reconciliation that found it. Resolves whether it started them over: a restore this + * worker's own pass handled already is left. Never rejects. */ - restored(first: boolean, options?: { readonly reconciled?: boolean }): Promise; + restored(found: Restored, options?: { readonly reconciled?: boolean }): Promise; } export interface CreateScheduledCommandWorkerArgs { @@ -558,6 +559,10 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction return storage.scheduler.claimDue({ now, limit: CLAIM_LIMIT, leaseMs }); }; + // What `restores.check` found last and was handled: a check shared with the dispatcher gives + // both the same finding. + let handled: Restored | null = null; + // The jobs' holds the copy brought back count failures it undid: the instance that found the // restore first clears them, and every one forgets what it knew of the jobs. Each never rejects. // `reconciled` when the reconciliation that just ran is what found the restore. @@ -565,6 +570,11 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction logger.warn("the database went back to an earlier copy; background jobs start over", { first, }); + if (reconciliation === undefined) { + logger.error( + "without runtime.personalData.forgetLedger, whoever was forgotten after the copy was taken is back until forgotten again, as bounda forget --from does", + ); + } purgedAt = Number.NEGATIVE_INFINITY; waits.clear(); onRestored?.(); @@ -583,7 +593,14 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction const runOnce = (): Promise => mutex.run(async () => { const restored = await restores?.check(clock.now(), { active: worked }); - if (restored !== undefined && restored !== null) await restart(restored.first); + let reconciledAtOnce = 0; + if (restored !== undefined && restored !== null && restored !== handled) { + handled = restored; + await restart(restored.first); + // Before anything is claimed: the scheduled commands of whom the copy brought back go + // with their forgetting. + reconciledAtOnce = (await reconciliation?.sweep(clock.now())) ?? 0; + } const claimedAt = clock.now(); const due = await claimDue(claimedAt); // The batch is claimed already: an error escaping here would leave every entry to lapse, and @@ -624,12 +641,20 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction // A forgetting the ledger holds and the database does not: the database went back, though // the generation did not show it, as for a copy taken just before a forget. if (forgottenAgain > 0 && restores !== undefined && (restored ?? null) === null) { - await restart((await restores.declare()).first, true); + handled = await restores.declare(); + await restart(handled.first, true); } const purged = (await letterRetention?.sweep(claimedAt)) ?? 0; const backfilled = (await backfill?.sweep(claimedAt)) ?? 0; if (unreadable !== undefined) throw unreadable.error; - const done = due.length + forgotten + rewrapped + forgottenAgain + purged + backfilled; + const done = + reconciledAtOnce + + due.length + + forgotten + + rewrapped + + forgottenAgain + + purged + + backfilled; worked = done > 0; return done; }); @@ -664,6 +689,12 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction runOnce, leaseMs, waitingDeadlines: () => waits.size, - restored: (first, { reconciled } = {}) => mutex.run(() => restart(first, reconciled)), + restored: (found, { reconciled } = {}) => + mutex.run(async () => { + if (found === handled) return false; + handled = found; + await restart(found.first, reconciled); + return true; + }), }; }; diff --git a/packages/core/src/kernel/shared/restores.test.ts b/packages/core/src/kernel/shared/restores.test.ts index dab7bc5e..2b55ba44 100644 --- a/packages/core/src/kernel/shared/restores.test.ts +++ b/packages/core/src/kernel/shared/restores.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import type { CheckpointStore } from "../../adapter/storage/checkpoint-store.ts"; import type { ForgetLedger } from "../../adapter/storage/forget-ledger.ts"; +import type { StoredEvent } from "../../contracts/event.ts"; import { memory } from "../../memory/index.ts"; import { createRecordingLogger } from "../test-support.ts"; import { @@ -27,7 +28,7 @@ const setUp = async ( const watchOver = () => createRestoreWatch({ checkpoints: options.wrap?.(checkpointStore) ?? checkpointStore, - events: { lastPosition: async () => stream.head }, + events: { lastPosition: async () => stream.head, readAll: async () => [] }, marks: options.marks === false ? null : (options.ledger?.(ledger) ?? ledger), logger, }); @@ -208,6 +209,7 @@ describe("createRestoreWatch", () => { if (down) throw new Error("down"); return 0; }, + readAll: async () => [], }, marks: ledger, logger, @@ -223,3 +225,181 @@ describe("createRestoreWatch", () => { ]); }); }); + +// A stream whose history a test rewrites, as a restore does: positions come back with other ids. +const rewritable = () => { + let events: { id: string; position: number }[] = []; + let reads = 0; + return { + events: { + lastPosition: async () => events.at(-1)?.position ?? 0, + readAll: async ({ afterPosition, limit }: { afterPosition: number; limit: number }) => { + reads += 1; + return events + .filter(({ position }) => position > afterPosition) + .slice(0, limit) as unknown as StoredEvent[]; + }, + }, + append: (...ids: string[]) => { + for (const id of ids) events.push({ id, position: (events.at(-1)?.position ?? 0) + 1 }); + }, + // Positions may skip, as a sequence does after a rollback. + skip: (count: number) => { + const last = events.at(-1)?.position ?? 0; + events.push({ id: `skipped-${last}`, position: last + count + 1 }); + }, + restore: (count: number) => { + events = events.slice(0, count); + }, + reads: () => reads, + }; +}; + +const watchingEvents = async () => { + const { checkpointStore } = await memory().createStorage(); + const stream = rewritable(); + const { logger } = createRecordingLogger("omit"); + const watch = createRestoreWatch({ + checkpoints: checkpointStore, + events: stream.events, + marks: null, + logger, + }); + return { checkpointStore, stream, watch }; +}; + +describe("createRestoreWatch, watching the newest event", () => { + it("finds nothing while events are only appended, whatever positions they skip", async () => { + const { stream, watch } = await watchingEvents(); + await watch.start(); + stream.append("e-1", "e-2"); + expect(await watch.check(at(0))).toBeNull(); + stream.skip(4); + expect(await watch.check(at(1))).toBeNull(); + expect(await watch.check(at(2))).toBeNull(); + }); + + it("finds a restore that took the stream back past the newest event it saw, before the generation moved", async () => { + const { checkpointStore, stream, watch } = await watchingEvents(); + stream.append("e-1"); + await watch.start(); + stream.append("e-2"); + await watch.check(at(0)); + expect(await checkpointStore.get(RESTORE_GENERATION)).toBe(1); + stream.append("e-3"); + // Within the interval: the generation does not move, the event does. + await watch.check(at(1)); + stream.restore(2); + expect(await watch.check(at(2))).toEqual({ first: true }); + expect(await checkpointStore.get(RESTORES_FOUND)).toBe(1); + expect(await watch.check(at(3))).toBeNull(); + }); + + it("finds a restore however much was written after it, as positions come back under other ids", async () => { + const { stream, watch } = await watchingEvents(); + stream.append("e-1", "e-2"); + await watch.start(); + stream.restore(1); + stream.append("e-2b", "e-3b"); + expect(await watch.check(at(0))).toEqual({ first: true }); + expect(await watch.check(at(1))).toBeNull(); + stream.restore(1); + stream.append("e-2c"); + expect(await watch.check(at(2))).toEqual({ first: true }); + // Another event at the position it watched last: the restore is counted once. + expect(await watch.check(at(3))).toBeNull(); + }); + + it("watches the event at its position again after a restore that put another one there", async () => { + const { stream, watch } = await watchingEvents(); + stream.append("e-1", "e-2"); + await watch.start(); + stream.restore(1); + stream.append("e-2b"); + expect(await watch.check(at(0))).toEqual({ first: true }); + expect(await watch.check(at(1))).toBeNull(); + }); + + it("finds a restore to an empty stream, and watches the stream again once events come", async () => { + const { stream, watch } = await watchingEvents(); + stream.append("e-1"); + await watch.start(); + stream.restore(0); + expect(await watch.check(at(0))).toEqual({ first: true }); + expect(await watch.check(at(1))).toBeNull(); + stream.append("e-1b"); + expect(await watch.check(at(2))).toBeNull(); + stream.restore(0); + expect(await watch.check(at(3))).toEqual({ first: true }); + }); + + it("watches an event it committed once it is newer than the one it read, and nothing older", async () => { + const { stream, watch } = await watchingEvents(); + stream.append("e-1", "e-2"); + await watch.start(); + watch.saw({ id: "e-1", position: 1 }); + watch.saw({ id: "elsewhere", position: 2 }); + expect(await watch.check(at(0))).toBeNull(); + stream.append("forgotten"); + watch.saw({ id: "forgotten", position: 3 }); + stream.restore(2); + expect(await watch.check(at(1))).toEqual({ first: true }); + }); + + it("watches an event it committed before its first reading", async () => { + const { stream, watch } = await watchingEvents(); + stream.append("e-1"); + watch.saw({ id: "e-1", position: 1 }); + stream.restore(0); + await watch.check(at(0)); + expect(await watch.found()).toBe(0); + stream.append("e-1b"); + watch.saw({ id: "e-1b", position: 1 }); + stream.restore(0); + expect(await watch.check(at(1))).toEqual({ first: true }); + }); + + it("watches the stream as it is after a restore was declared, so it is counted once", async () => { + const { stream, watch } = await watchingEvents(); + stream.append("e-1", "e-2"); + await watch.start(); + stream.restore(1); + expect(await watch.declare()).toEqual({ first: true }); + expect(await watch.check(at(0))).toBeNull(); + expect(await watch.found()).toBe(1); + }); + + it("reads the stream once a check while nothing is appended", async () => { + const { stream, watch } = await watchingEvents(); + stream.append("e-1"); + await watch.start(); + const before = stream.reads(); + await watch.check(at(0)); + expect(stream.reads() - before).toBe(1); + stream.append("e-2"); + await watch.check(at(1)); + expect(stream.reads() - before).toBe(3); + }); + + it("moves the generation at the next check for an active caller that joined one under way", async () => { + const { checkpointStore, stream, watch } = await watchingEvents(); + stream.append("e-1"); + await watch.start(); + await Promise.all([watch.check(at(0)), watch.check(at(0), { active: true })]); + expect(await checkpointStore.get(RESTORE_GENERATION)).toBe(0); + await watch.check(at(1)); + expect(await checkpointStore.get(RESTORE_GENERATION)).toBe(1); + }); + + it("shares one check between callers that overlap, and checks again after", async () => { + const { checkpointStore, stream, watch } = await watchingEvents(); + stream.append("e-1", "e-2"); + await watch.start(); + stream.restore(1); + const found = await Promise.all([watch.check(at(0)), watch.check(at(0), { active: true })]); + expect(found[0]).toEqual({ first: true }); + expect(found[1]).toBe(found[0]); + expect(await checkpointStore.get(RESTORES_FOUND)).toBe(1); + expect(await watch.check(at(1))).toBeNull(); + }); +}); diff --git a/packages/core/src/kernel/shared/restores.ts b/packages/core/src/kernel/shared/restores.ts index 617209e5..2858110b 100644 --- a/packages/core/src/kernel/shared/restores.ts +++ b/packages/core/src/kernel/shared/restores.ts @@ -64,6 +64,17 @@ export interface RestoreWatch { * How many restores the instances found, as the database counts them. */ found(): Promise; + /** + * An event this instance committed, such as a forgetting: watched from now on when it is newer + * than the one watched, so a copy taken before it is found even when the generation had not + * moved yet. + */ + saw(event: WatchedEvent): void; +} + +export interface WatchedEvent { + readonly id: string; + readonly position: number; } export interface CheckRestoreOptions { @@ -76,7 +87,7 @@ export interface CheckRestoreOptions { export interface CreateRestoreWatchArgs { readonly checkpoints: CheckpointStore; - readonly events: Pick; + readonly events: Pick; /** * The forget ledger, outside the database; `null` without one, and a copy restored while the * app was stopped is then found only when declared. @@ -95,10 +106,14 @@ interface Seen { } /** - * A generation that went back, or a count of restores that changed, means a restore. The first - * instance to see the generation go back moves the count past every value an instance can have - * read, so each other instance finds it changed, whether its own reading of the generation shows - * the restore or not. A copy taken after the last move goes unseen: nothing was written since. + * A generation that went back, a count of restores that changed, or the newest event this + * instance saw gone from its position, means a restore. The first instance to see the generation + * go back, or its event gone, moves the count past every value an instance can have read, so each + * other instance finds it changed, whether its own reading shows the restore or not. Events are + * only ever appended, so the newest one seen stays at its position until a restore takes the + * stream back past it: that finds a copy taken after the generation's last move, as one taken a + * moment before a forgetting. Positions are reused after a restore, hence the id. Concurrent checks + * share one reading. */ export const createRestoreWatch: CreateRestoreWatchFunction = ({ checkpoints, @@ -113,6 +128,27 @@ export const createRestoreWatch: CreateRestoreWatchFunction = ({ // While the ledger's mark could not be read, the generation stays where it is: moving it could // carry a copy restored while the app was stopped past the mark before it is compared. let unmarked = marks !== null; + let witness: WatchedEvent | null = null; + let checking: Promise | null = null; + // Activity of a caller that joined a check under way, for the next one. + let joinedActive = false; + + const eventAt = async (position: number): Promise => { + if (position === 0) return null; + const [event] = await events.readAll({ afterPosition: position - 1, limit: 1 }); + return event === undefined ? null : { id: event.id, position: event.position }; + }; + + // The event this instance watches stays where it was, or another took its place. + const witnessGone = async (): Promise => { + if (witness === null) return false; + return (await eventAt(witness.position))?.id !== witness.id; + }; + + const watchHead = async (last: number): Promise => { + if (witness?.position === last) return; + witness = await eventAt(last); + }; const read = async (): Promise => { const [generation, restores] = await Promise.all([ @@ -174,8 +210,10 @@ export const createRestoreWatch: CreateRestoreWatchFunction = ({ const check = async (now: Date, active: boolean): Promise => { const at = now.getTime(); - const [stored, last] = await Promise.all([read(), events.lastPosition()]); - pending ||= active || (head !== null && last !== head); + const joined = joinedActive; + joinedActive = false; + const [stored, last, gone] = await Promise.all([read(), events.lastPosition(), witnessGone()]); + pending ||= active || joined || (head !== null && last !== head); head = last; if (unmarked) { const restored = await compareWithMark(stored); @@ -185,7 +223,7 @@ export const createRestoreWatch: CreateRestoreWatchFunction = ({ let current = stored; if ( seen !== null && - (stored.generation < seen.generation || stored.restores !== seen.restores) + (stored.generation < seen.generation || stored.restores !== seen.restores || gone) ) { // A count past what this instance read was moved by the instance that found it first. if (stored.restores > seen.restores) { @@ -197,6 +235,8 @@ export const createRestoreWatch: CreateRestoreWatchFunction = ({ pending = true; } seen = current; + if (restored !== null) witness = null; + await watchHead(last); if (pending && !unmarked && at - movedAt >= GENERATION_INTERVAL_MS) await move(current, at); return restored; }; @@ -213,6 +253,7 @@ export const createRestoreWatch: CreateRestoreWatchFunction = ({ try { const stored = await read(); head = await events.lastPosition(); + await watchHead(head); seen = stored; const restored = await compareWithMark(stored); // A mark only goes up: past a restore it is already higher. @@ -222,11 +263,23 @@ export const createRestoreWatch: CreateRestoreWatchFunction = ({ return logged("the database could not be compared with the forget ledger's mark")(error); } }, - check: (now, { active = false } = {}) => - check(now, active).catch(logged("the database could not be checked for a restore")), + check: (now, { active = false } = {}) => { + if (checking !== null) { + joinedActive ||= active; + return checking; + } + checking = check(now, active) + .catch(logged("the database could not be checked for a restore")) + .finally(() => { + checking = null; + }); + return checking; + }, declare: async () => { const stored = await read(); const restored = await count(stored, Math.max(stored.restores, seen?.restores ?? 0), 0); + witness = null; + await watchHead(await events.lastPosition()); if (marks !== null && seen !== null) await marks.raiseMark(RESTORE_MARK, seen.generation); return restored; }, @@ -242,5 +295,10 @@ export const createRestoreWatch: CreateRestoreWatchFunction = ({ } }, found: () => checkpoints.get(RESTORES_FOUND), + saw: (event) => { + if (witness === null || event.position > witness.position) { + witness = { id: event.id, position: event.position }; + } + }, }; }; diff --git a/packages/core/src/kernel/status.test.ts b/packages/core/src/kernel/status.test.ts index e2e22bcf..e1caedf4 100644 --- a/packages/core/src/kernel/status.test.ts +++ b/packages/core/src/kernel/status.test.ts @@ -85,6 +85,7 @@ describe("readAppStatus", () => { declare: async () => ({ first: true }), flush: async () => undefined, found: async () => 0, + saw: () => undefined, }, }); expect( diff --git a/packages/postgresql/src/restore.test.ts b/packages/postgresql/src/restore.test.ts new file mode 100644 index 00000000..4d8eee14 --- /dev/null +++ b/packages/postgresql/src/restore.test.ts @@ -0,0 +1,65 @@ +import { restoreContract } from "@bounda-dev/core/adapter/testing"; +import { PostgreSqlContainer, type StartedPostgreSqlContainer } from "@testcontainers/postgresql"; +import { afterAll, describe } from "vitest"; +import { postgresql } from "./index.ts"; + +const startContainer = async (): Promise => { + try { + return await new PostgreSqlContainer("postgres:17").start(); + } catch { + return null; + } +}; + +const container = await startContainer(); +const url = container?.getConnectionUri() ?? ""; +let schemas = 0; + +afterAll(async () => { + await container?.stop(); +}); + +// The server's own tools, run where the server is. +const tool = async (command: readonly string[]): Promise => { + if (container === null) return; + const { exitCode, output } = await container.exec([ + ...command, + "--username", + container.getUsername(), + "--dbname", + container.getDatabase(), + ]); + if (exitCode !== 0) throw new Error(`${command[0]} exited with ${exitCode}: ${output}`); +}; + +describe.skipIf(container === null)("restoring a PostgreSQL schema", () => { + restoreContract({ + hot: true, + create: async () => { + schemas += 1; + const schema = `restored_${schemas}`; + const file = `/tmp/${schema}.dump`; + const storage = postgresql({ url, schema, maxConnections: 3 }); + return { + storage, + forgetLedger: postgresql({ url, schema: `ledger_${schemas}`, maxConnections: 2 }), + // pg_dump takes a consistent snapshot while the app writes. + backUp: () => tool(["pg_dump", "--schema", schema, "--format", "custom", "--file", file]), + // In place, in one transaction: the app's connections see the schema before the restore or + // after it. + restore: async () => { + await tool([ + "pg_restore", + "--clean", + "--if-exists", + "--single-transaction", + "--schema", + schema, + file, + ]); + return storage; + }, + }; + }, + }); +}); diff --git a/packages/sqlite/src/restore.test.ts b/packages/sqlite/src/restore.test.ts new file mode 100644 index 00000000..90f70b14 --- /dev/null +++ b/packages/sqlite/src/restore.test.ts @@ -0,0 +1,132 @@ +import { copyFile, mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { restoreContract } from "@bounda-dev/core/adapter/testing"; +import { type Client, createClient } from "@libsql/client"; +import { afterAll, beforeAll, describe } from "vitest"; +import { sqlite } from "./index.ts"; + +let directory: string; +let files = 0; +const clients: Client[] = []; + +beforeAll(async () => { + directory = await mkdtemp(join(tmpdir(), "bounda-restore-")); +}); + +afterAll(async () => { + for (const client of clients) client.close(); + // libSQL lets go of a file that saw a write transaction only once its statements are garbage + // collected, and Windows cannot delete a file still open: there, those stay behind in the + // temporary directory. + await rm(directory, { recursive: true, force: true }).catch((error: NodeJS.ErrnoException) => { + if (process.platform !== "win32" || error.code !== "EBUSY") throw error; + }); +}); + +const fresh = (name: string): string => { + files += 1; + return join(directory, `${name}-${files}.db`); +}; + +const clientOf = (path: string): Client => { + const client = createClient({ url: `file:${path}` }); + clients.push(client); + return client; +}; + +interface SchemaObject { + readonly type: string; + readonly name: string; + readonly sql: string | null; +} + +const schemaOf = async (client: Client): Promise => + ( + await client.execute("SELECT type, name, sql FROM sqlite_master WHERE name NOT LIKE 'sqlite_%'") + ).rows.map(({ type, name, sql }) => ({ + type: String(type), + name: String(name), + sql: sql === null ? null : String(sql), + })); + +// In place, in one write transaction through the app's own engine, as pg_restore --clean +// --single-transaction does: the app's connections see the database before the restore or after +// it, never in between. Every table, index and trigger goes, and the copy's come back with its rows. +const restoreInPlace = async (client: Client, copy: string): Promise => { + const source = clientOf(copy); + const copied = await schemaOf(source); + const tables = copied.filter(({ type }) => type === "table"); + const rows = await Promise.all( + tables.map(async ({ name }) => ({ + name, + rows: (await source.execute(`SELECT * FROM "${name}"`)).rows, + })), + ); + const live = (await schemaOf(client)).filter(({ type }) => type === "table"); + await client.batch( + [ + ...live.map(({ name }) => `DROP TABLE "${name}"`), + ...tables.flatMap(({ sql }) => (sql === null ? [] : [sql])), + ...rows.flatMap(({ name, rows: values }) => + values.map((row) => { + const columns = Object.keys(row); + return { + sql: `INSERT INTO "${name}" (${columns.map((column) => `"${column}"`).join(", ")}) VALUES (${columns.map(() => "?").join(", ")})`, + args: columns.map((column) => row[column] ?? null), + }; + }), + ), + // After the rows, so no trigger writes what the copy already holds. + ...copied.flatMap(({ type, sql }) => (type === "table" || sql === null ? [] : [sql])), + ], + "write", + ); +}; + +describe("restoring a libSQL file under a running app", () => { + restoreContract({ + hot: true, + create: async () => { + const path = fresh("store"); + const copy = fresh("backup"); + const client = clientOf(path); + return { + storage: sqlite({ path }), + forgetLedger: sqlite({ path: fresh("ledger") }), + // An online copy, as a backup tool takes it while the app runs. + backUp: async () => { + await client.execute(`VACUUM INTO '${copy}'`); + }, + restore: async () => { + await restoreInPlace(client, copy); + return sqlite({ path }); + }, + }; + }, + }); +}); + +describe("restoring a libSQL file beside the lost one", () => { + restoreContract({ + hot: false, + create: async () => { + let path = fresh("store"); + const copy = fresh("backup"); + return { + storage: sqlite({ path }), + forgetLedger: sqlite({ path: fresh("ledger") }), + backUp: async () => { + await clientOf(path).execute(`VACUUM INTO '${copy}'`); + }, + // A copy of the file, beside the old one, which Windows will not let go of while libSQL + // still holds it. + restore: async () => { + path = fresh("restored"); + await copyFile(copy, path); + return sqlite({ path }); + }, + }; + }, + }); +}); From b635cd637521d25ed77f59e230fdab58dc8dd742 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Tue, 10 Nov 2026 00:49:59 +0100 Subject: [PATCH 2/4] test(core): close the node:sqlite connections of the restore tests before removing their files --- packages/core/src/adapter/sqlite/restore.test.ts | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/packages/core/src/adapter/sqlite/restore.test.ts b/packages/core/src/adapter/sqlite/restore.test.ts index 14c20480..bff25a8e 100644 --- a/packages/core/src/adapter/sqlite/restore.test.ts +++ b/packages/core/src/adapter/sqlite/restore.test.ts @@ -9,7 +9,10 @@ import type { EventPayloadArgs, PayloadArgs } from "../../modules/payload.ts"; import type { Registry } from "../../modules/registry.ts"; import type { Adapter } from "../adapter.ts"; import { restoreContract } from "../testing/restore.contract.ts"; -import { createNodeSqliteAdapter } from "./node-sqlite.ts"; +import { + type NodeSqliteAdapter, + createNodeSqliteAdapter as openNodeSqlite, +} from "./node-sqlite.ts"; let directory: string; let files = 0; @@ -18,7 +21,18 @@ beforeAll(async () => { directory = await mkdtemp(join(tmpdir(), "bounda-restore-")); }); +// The adapter leaves its connection to whoever opened it, as a test helper: each is closed here, +// before the directory goes, since Windows cannot delete a file still open. +const opened: NodeSqliteAdapter[] = []; + +const createNodeSqliteAdapter = (path: string): NodeSqliteAdapter => { + const sqlite = openNodeSqlite(path); + opened.push(sqlite); + return sqlite; +}; + afterAll(async () => { + for (const { db } of opened.splice(0)) db.close(); await rm(directory, { recursive: true, force: true }); }); From 5e5c270ce84b5e77cc3fc2fe702f9766224bfbd0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Tue, 10 Nov 2026 08:06:27 +0100 Subject: [PATCH 3/4] Merge main (B1) into feat/restore-harness and cover partial forgetting and legal retention across a restore --- .changeset/restore-harness.md | 2 +- docs/src/content/docs/guides/backups.md | 2 + .../src/adapter/testing/restore.contract.ts | 123 +++++++++++++++++- 3 files changed, 125 insertions(+), 2 deletions(-) diff --git a/.changeset/restore-harness.md b/.changeset/restore-harness.md index 22bfecfd..35318a0c 100644 --- a/.changeset/restore-harness.md +++ b/.changeset/restore-harness.md @@ -2,4 +2,4 @@ "@bounda-dev/core": minor --- -The restore watch also watches the newest event each instance read, or committed as a forgetting, at its position: a copy that lost it is found at the next pass, even one taken in the five seconds before the generation moved, and an instance in the `web` role finds it as well. The dispatcher checks at the start of every pass, and a pass that finds a restore reconciles the forget ledger before anything is delivered or claimed, so a scheduled command of someone the copy brought back never runs. An app without a forget ledger logs an error when it finds a restore. `restoreContract` in `@bounda-dev/core/adapter/testing` is the end-to-end restore an adapter has to pass: persons write, a backup is taken, some are forgotten, the master key rotates, and the copy is restored with the app stopped and, when the engine can, running, also under an instance in the `web` role, without a forget ledger through `app.restored()`, with a forgetting the ledger recorded whose commit never happened, and with a strict ledger that cannot be read at the restored start; `expectForgotten` checks every person forgotten after the copy, and the app goes on. It runs on node:sqlite, libSQL, PostgreSQL and the Durable Object. The new guide "Backups and restores" has the steps for each engine and what is guaranteed. +The restore watch also watches the newest event each instance read, or committed as a forgetting, at its position: a copy that lost it is found at the next pass, even one taken in the five seconds before the generation moved, and an instance in the `web` role finds it as well. The dispatcher checks at the start of every pass, and a pass that finds a restore reconciles the forget ledger before anything is delivered or claimed, so a scheduled command of someone the copy brought back never runs. An app without a forget ledger logs an error when it finds a restore. `restoreContract` in `@bounda-dev/core/adapter/testing` is the end-to-end restore an adapter has to pass: persons write, a backup is taken, some are forgotten, the master key rotates, and the copy is restored with the app stopped and, when the engine can, running, also under an instance in the `web` role, without a forget ledger through `app.restored()`, with a forgetting the ledger recorded whose commit never happened, with a strict ledger that cannot be read at the restored start, and with a forgetting of some categories alone beside one of a person as a whole, where only those categories go again and what a legal obligation keeps stays readable; `expectForgotten` checks every person forgotten after the copy, and the app goes on. It runs on node:sqlite, libSQL, PostgreSQL and the Durable Object. The new guide "Backups and restores" has the steps for each engine and what is guaranteed. diff --git a/docs/src/content/docs/guides/backups.md b/docs/src/content/docs/guides/backups.md index cfa2b04c..71b1daa0 100644 --- a/docs/src/content/docs/guides/backups.md +++ b/docs/src/content/docs/guides/backups.md @@ -85,6 +85,8 @@ restored. letter and in the state of their processes, and none of their values in any row of the database. - **Their scheduled commands never run**, even when they fall due as the copy comes back. +- **A forgetting of some categories alone** goes again for those categories alone, and what a + [legal obligation](/guides/personal-data/#categories) keeps after a person is forgotten stays readable. - **A forgetting whose commit never happened**, recorded in the ledger by a process that stopped before committing it, is committed at the restored start too. - **A strict ledger that cannot be read** fails the restored start before it serves anything; the diff --git a/packages/core/src/adapter/testing/restore.contract.ts b/packages/core/src/adapter/testing/restore.contract.ts index 1634bf36..9c0fac5c 100644 --- a/packages/core/src/adapter/testing/restore.contract.ts +++ b/packages/core/src/adapter/testing/restore.contract.ts @@ -228,6 +228,68 @@ const registry = { }, } as unknown as Registry; +// A newsletter address under a consent, a tax id kept by law for six years, and a name in the +// default category. +const categorized = { + aggregates: { + customer: { + state: { initialState: {} }, + events: { + customerBilled: { + payload: ({ z, personal }: EventPayloadArgs) => + z.object({ + newsletter: personal(z.string(), { category: "marketing" }), + taxId: personal(z.string(), { category: "invoices" }), + name: personal(z.string()), + }), + evolve: () => ({}), + }, + }, + commands: { + billCustomer: { + module: { + payload: ({ z }: PayloadArgs) => + z.object({ + customerId: z.string(), + newsletter: z.string(), + taxId: z.string(), + name: z.string(), + }), + handler: ({ + command, + events, + }: { + command: { payload: { newsletter: string; taxId: string; name: string } }; + events: Builders; + }) => [ + events.customerBilled?.({ + newsletter: command.payload.newsletter, + taxId: command.payload.taxId, + name: command.payload.name, + }), + ], + }, + }, + }, + policies: {}, + processes: {}, + }, + }, + readModels: {}, + personalData: { + categories: { + marketing: { basis: "consent", retention: "730d" }, + invoices: { basis: "legal-obligation", retention: "6y" }, + }, + }, +} as unknown as Registry; + +const billOf = (person: string) => ({ + newsletter: `${person}.newsletter@example.com`, + taxId: `${person}-TAX-2026-0001`, + name: `${person} Billedname`, +}); + const START = new Date("2026-11-09T09:00:00.000Z"); // Longer than the scenario runs for, so no periodic reconciliation hides a restore the app missed. const RECONCILE_EVERY_MS = 24 * 60 * 60_000; @@ -592,11 +654,12 @@ export const restoreContract: RestoreContractFunction = ({ create, hot }) => { await write(first, "ana"); await target.backUp(); const written = await target.storage.createStorage(); - const key = await written.personalKeys.get("ana"); + const key = await written.personalKeys.get("ana", "default"); await written.close(); // What a forget records before it commits, left alone by a process that stopped in between. const intention = { personId: "ana", + categories: null, eventId: "019a6b1e-0000-7000-8000-00000000a7a0", forgottenAt: clock.now().toISOString(), reason: "requested" as const, @@ -637,5 +700,63 @@ export const restoreContract: RestoreContractFunction = ({ create, hot }) => { await expectForgottenIn(storage, restored, "ana"); await expectForgottenIn(storage, restored, "bob"); }); + + it("forgets again only the categories a partial forgetting named, and keeps readable what the law keeps", async () => { + const target = await create(); + const clock = createFixedClock(START); + const open = async (storage: Adapter) => { + const app = await createApp({ + registry: categorized, + config: { + storage, + runtime: { + personalData: { + keyWrapper: BEFORE_ROTATION, + forgetLedger: { store: target.forgetLedger, reconcileEvery: RECONCILE_EVERY_MS }, + }, + }, + }, + clock, + env, + }); + stops.push(() => app.stop()); + return app; + }; + const read = async (app: BoundaApp, person: string) => + (await app.exportPersonalData(person)).events[0]?.data; + const first = await open(target.storage); + const commands = first.commands as unknown as Commands; + for (const person of ["ana", "bob"]) { + await commands.billCustomer?.({ customerId: person, ...billOf(person) }); + } + await target.backUp(); + // Ana withdraws her consent to the newsletter; bob is forgotten as a whole. + const anaForgotten = (await first.forget("ana", { categories: ["marketing"] })).eventIds; + const bobForgotten = (await first.forget("bob")).eventIds; + await first.runUntilIdle(); + await first.stop(); + + const storage = await target.restore(); + const restored = await open(storage); + expect(await forgettings(restored, "ana")).toEqual(anaForgotten); + expect(await forgettings(restored, "bob")).toEqual(bobForgotten); + const ana = billOf("ana"); + const bob = billOf("bob"); + expect(await read(restored, "ana")).toEqual({ + newsletter: null, + taxId: ana.taxId, + name: ana.name, + }); + expect(await read(restored, "bob")).toEqual({ + newsletter: null, + taxId: bob.taxId, + name: null, + }); + await expectForgotten(storage, "ana", { + categories: ["marketing"], + values: [ana.newsletter], + }); + await expectForgotten(storage, "bob", { values: [bob.newsletter, bob.name] }); + }); }); }; From c090188a6af497f4fb7a9a7b5d8aea7103fc0e29 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Tue, 10 Nov 2026 12:08:57 +0100 Subject: [PATCH 4/4] test(core): give each restore contract case a timeout of its own, for file databases on Windows --- .../src/adapter/testing/restore.contract.ts | 337 ++++++++++-------- 1 file changed, 184 insertions(+), 153 deletions(-) diff --git a/packages/core/src/adapter/testing/restore.contract.ts b/packages/core/src/adapter/testing/restore.contract.ts index 9c0fac5c..c21cd301 100644 --- a/packages/core/src/adapter/testing/restore.contract.ts +++ b/packages/core/src/adapter/testing/restore.contract.ts @@ -290,6 +290,11 @@ const billOf = (person: string) => ({ name: `${person} Billedname`, }); +// Each case starts and stops apps several times and commits hundreds of write transactions, each +// synced to disk: a file database on a Windows runner takes some seconds for what takes a few +// hundred milliseconds elsewhere, waiting on nothing but the disk. +const RESTORE_CASE_TIMEOUT_MS = 30_000; + const START = new Date("2026-11-09T09:00:00.000Z"); // Longer than the scenario runs for, so no periodic reconciliation hides a restore the app missed. const RECONCILE_EVERY_MS = 24 * 60 * 60_000; @@ -566,40 +571,49 @@ export const restoreContract: RestoreContractFunction = ({ create, hot }) => { } }; - it("forgets again at the start after a restore whoever was forgotten since the copy, everywhere, and goes on", async () => { - await scenario("at start"); - }); + it( + "forgets again at the start after a restore whoever was forgotten since the copy, everywhere, and goes on", + async () => { + await scenario("at start"); + }, + RESTORE_CASE_TIMEOUT_MS, + ); it.skipIf(!hot)( "forgets again, at the worker's next pass, whoever a restore under the running app brings back", async () => { await scenario("hot"); }, + RESTORE_CASE_TIMEOUT_MS, ); - it("reconciles at the start after a restore whatever the copy says of a failed reconciliation", async () => { - const target = await create(); - const clock = createFixedClock(START); - const { definition, state } = faultyLedger(target.forgetLedger); - const first = await start(target.storage, definition, BEFORE_ROTATION, clock); - await write(first, "ana"); - // A reconciliation fails, and its hold is in the copy. - state.failing = true; - clock.advance(RECONCILE_EVERY_MS); - await first.app.runUntilIdle(); - expect(await first.app.getLag()).toMatchObject({ forgetLedger: { reconciled: false } }); - await target.backUp(); - state.failing = false; - const anaForgotten = (await first.app.forget("ana")).eventIds; - await first.app.runUntilIdle(); - await first.app.stop(); - - const storage = await target.restore(); - const restored = await start(storage, definition, BEFORE_ROTATION, clock); - expect(await forgettings(restored.app, "ana")).toEqual(anaForgotten); - expect(await restored.app.getLag()).toMatchObject({ forgetLedger: { reconciled: true } }); - await expectForgottenIn(storage, restored, "ana"); - }); + it( + "reconciles at the start after a restore whatever the copy says of a failed reconciliation", + async () => { + const target = await create(); + const clock = createFixedClock(START); + const { definition, state } = faultyLedger(target.forgetLedger); + const first = await start(target.storage, definition, BEFORE_ROTATION, clock); + await write(first, "ana"); + // A reconciliation fails, and its hold is in the copy. + state.failing = true; + clock.advance(RECONCILE_EVERY_MS); + await first.app.runUntilIdle(); + expect(await first.app.getLag()).toMatchObject({ forgetLedger: { reconciled: false } }); + await target.backUp(); + state.failing = false; + const anaForgotten = (await first.app.forget("ana")).eventIds; + await first.app.runUntilIdle(); + await first.app.stop(); + + const storage = await target.restore(); + const restored = await start(storage, definition, BEFORE_ROTATION, clock); + expect(await forgettings(restored.app, "ana")).toEqual(anaForgotten); + expect(await restored.app.getLag()).toMatchObject({ forgetLedger: { reconciled: true } }); + await expectForgottenIn(storage, restored, "ana"); + }, + RESTORE_CASE_TIMEOUT_MS, + ); it.skipIf(!hot)( "forgets again whoever a restore under an instance that runs no worker brings back", @@ -623,140 +637,157 @@ export const restoreContract: RestoreContractFunction = ({ create, hot }) => { await expectForgottenIn(target.storage, web, "ana"); }); }, + RESTORE_CASE_TIMEOUT_MS, ); - it("starts over when an operator says a database without a forget ledger was restored, which bounda forget --from then forgets again", async () => { - const target = await create(); - const clock = createFixedClock(START); - const first = await start(target.storage, null, BEFORE_ROTATION, clock); - await write(first, "ana"); - await target.backUp(); - await first.app.forget("ana"); - await first.app.runUntilIdle(); - await first.app.stop(); - - const storage = await target.restore(); - const restored = await start(storage, null, BEFORE_ROTATION, clock); - // Nothing outside the database says it went back. - expect((await restored.app.status()).restoresFound).toBe(0); - await restored.app.restored(); - expect((await restored.app.status()).restoresFound).toBe(1); - expect(await restored.app.forgetAgain([{ personId: "ana" }])).toEqual(["ana"]); - await restored.app.runUntilIdle(); - await expectForgottenIn(storage, restored, "ana"); - }); - - it("commits at the restored start a forgetting the ledger recorded before a commit that never happened", async () => { - const target = await create(); - const clock = createFixedClock(START); - const { definition, opened } = faultyLedger(target.forgetLedger); - const first = await start(target.storage, definition, BEFORE_ROTATION, clock); - await write(first, "ana"); - await target.backUp(); - const written = await target.storage.createStorage(); - const key = await written.personalKeys.get("ana", "default"); - await written.close(); - // What a forget records before it commits, left alone by a process that stopped in between. - const intention = { - personId: "ana", - categories: null, - eventId: "019a6b1e-0000-7000-8000-00000000a7a0", - forgottenAt: clock.now().toISOString(), - reason: "requested" as const, - fingerprints: key === null ? [] : [key.fingerprint], - }; - await opened[0]?.append([intention]); - await first.app.stop(); - - const storage = await target.restore(); - const restored = await start(storage, definition, BEFORE_ROTATION, clock); - expect(await forgettings(restored.app, "ana")).toEqual([intention.eventId]); - await expectForgottenIn(storage, restored, "ana"); - }); + it( + "starts over when an operator says a database without a forget ledger was restored, which bounda forget --from then forgets again", + async () => { + const target = await create(); + const clock = createFixedClock(START); + const first = await start(target.storage, null, BEFORE_ROTATION, clock); + await write(first, "ana"); + await target.backUp(); + await first.app.forget("ana"); + await first.app.runUntilIdle(); + await first.app.stop(); + + const storage = await target.restore(); + const restored = await start(storage, null, BEFORE_ROTATION, clock); + // Nothing outside the database says it went back. + expect((await restored.app.status()).restoresFound).toBe(0); + await restored.app.restored(); + expect((await restored.app.status()).restoresFound).toBe(1); + expect(await restored.app.forgetAgain([{ personId: "ana" }])).toEqual(["ana"]); + await restored.app.runUntilIdle(); + await expectForgottenIn(storage, restored, "ana"); + }, + RESTORE_CASE_TIMEOUT_MS, + ); - it("does not start on a restored copy a strict ledger cannot read, and applies the whole restore once it can", async () => { - const target = await create(); - const clock = createFixedClock(START); - const { definition, state } = faultyLedger(target.forgetLedger); - const strict = { strict: true }; - const first = await start(target.storage, definition, BEFORE_ROTATION, clock, strict); - await write(first, "ana"); - await write(first, "bob"); - await target.backUp(); - const anaForgotten = (await first.app.forget("ana")).eventIds; - const bobForgotten = (await first.app.forget("bob")).eventIds; - await first.app.runUntilIdle(); - await first.app.stop(); + it( + "commits at the restored start a forgetting the ledger recorded before a commit that never happened", + async () => { + const target = await create(); + const clock = createFixedClock(START); + const { definition, opened } = faultyLedger(target.forgetLedger); + const first = await start(target.storage, definition, BEFORE_ROTATION, clock); + await write(first, "ana"); + await target.backUp(); + const written = await target.storage.createStorage(); + const key = await written.personalKeys.get("ana", "default"); + await written.close(); + // What a forget records before it commits, left alone by a process that stopped in between. + const intention = { + personId: "ana", + categories: null, + eventId: "019a6b1e-0000-7000-8000-00000000a7a0", + forgottenAt: clock.now().toISOString(), + reason: "requested" as const, + fingerprints: key === null ? [] : [key.fingerprint], + }; + await opened[0]?.append([intention]); + await first.app.stop(); + + const storage = await target.restore(); + const restored = await start(storage, definition, BEFORE_ROTATION, clock); + expect(await forgettings(restored.app, "ana")).toEqual([intention.eventId]); + await expectForgottenIn(storage, restored, "ana"); + }, + RESTORE_CASE_TIMEOUT_MS, + ); - const storage = await target.restore(); - state.failing = true; - await expect(start(storage, definition, BEFORE_ROTATION, clock, strict)).rejects.toThrow( - ForgetLedgerUnreconciledError, - ); - state.failing = false; - const restored = await start(storage, definition, BEFORE_ROTATION, clock, strict); - expect(await forgettings(restored.app, "ana")).toEqual(anaForgotten); - expect(await forgettings(restored.app, "bob")).toEqual(bobForgotten); - await expectForgottenIn(storage, restored, "ana"); - await expectForgottenIn(storage, restored, "bob"); - }); + it( + "does not start on a restored copy a strict ledger cannot read, and applies the whole restore once it can", + async () => { + const target = await create(); + const clock = createFixedClock(START); + const { definition, state } = faultyLedger(target.forgetLedger); + const strict = { strict: true }; + const first = await start(target.storage, definition, BEFORE_ROTATION, clock, strict); + await write(first, "ana"); + await write(first, "bob"); + await target.backUp(); + const anaForgotten = (await first.app.forget("ana")).eventIds; + const bobForgotten = (await first.app.forget("bob")).eventIds; + await first.app.runUntilIdle(); + await first.app.stop(); + + const storage = await target.restore(); + state.failing = true; + await expect(start(storage, definition, BEFORE_ROTATION, clock, strict)).rejects.toThrow( + ForgetLedgerUnreconciledError, + ); + state.failing = false; + const restored = await start(storage, definition, BEFORE_ROTATION, clock, strict); + expect(await forgettings(restored.app, "ana")).toEqual(anaForgotten); + expect(await forgettings(restored.app, "bob")).toEqual(bobForgotten); + await expectForgottenIn(storage, restored, "ana"); + await expectForgottenIn(storage, restored, "bob"); + }, + RESTORE_CASE_TIMEOUT_MS, + ); - it("forgets again only the categories a partial forgetting named, and keeps readable what the law keeps", async () => { - const target = await create(); - const clock = createFixedClock(START); - const open = async (storage: Adapter) => { - const app = await createApp({ - registry: categorized, - config: { - storage, - runtime: { - personalData: { - keyWrapper: BEFORE_ROTATION, - forgetLedger: { store: target.forgetLedger, reconcileEvery: RECONCILE_EVERY_MS }, + it( + "forgets again only the categories a partial forgetting named, and keeps readable what the law keeps", + async () => { + const target = await create(); + const clock = createFixedClock(START); + const open = async (storage: Adapter) => { + const app = await createApp({ + registry: categorized, + config: { + storage, + runtime: { + personalData: { + keyWrapper: BEFORE_ROTATION, + forgetLedger: { store: target.forgetLedger, reconcileEvery: RECONCILE_EVERY_MS }, + }, }, }, - }, - clock, - env, + clock, + env, + }); + stops.push(() => app.stop()); + return app; + }; + const read = async (app: BoundaApp, person: string) => + (await app.exportPersonalData(person)).events[0]?.data; + const first = await open(target.storage); + const commands = first.commands as unknown as Commands; + for (const person of ["ana", "bob"]) { + await commands.billCustomer?.({ customerId: person, ...billOf(person) }); + } + await target.backUp(); + // Ana withdraws her consent to the newsletter; bob is forgotten as a whole. + const anaForgotten = (await first.forget("ana", { categories: ["marketing"] })).eventIds; + const bobForgotten = (await first.forget("bob")).eventIds; + await first.runUntilIdle(); + await first.stop(); + + const storage = await target.restore(); + const restored = await open(storage); + expect(await forgettings(restored, "ana")).toEqual(anaForgotten); + expect(await forgettings(restored, "bob")).toEqual(bobForgotten); + const ana = billOf("ana"); + const bob = billOf("bob"); + expect(await read(restored, "ana")).toEqual({ + newsletter: null, + taxId: ana.taxId, + name: ana.name, }); - stops.push(() => app.stop()); - return app; - }; - const read = async (app: BoundaApp, person: string) => - (await app.exportPersonalData(person)).events[0]?.data; - const first = await open(target.storage); - const commands = first.commands as unknown as Commands; - for (const person of ["ana", "bob"]) { - await commands.billCustomer?.({ customerId: person, ...billOf(person) }); - } - await target.backUp(); - // Ana withdraws her consent to the newsletter; bob is forgotten as a whole. - const anaForgotten = (await first.forget("ana", { categories: ["marketing"] })).eventIds; - const bobForgotten = (await first.forget("bob")).eventIds; - await first.runUntilIdle(); - await first.stop(); - - const storage = await target.restore(); - const restored = await open(storage); - expect(await forgettings(restored, "ana")).toEqual(anaForgotten); - expect(await forgettings(restored, "bob")).toEqual(bobForgotten); - const ana = billOf("ana"); - const bob = billOf("bob"); - expect(await read(restored, "ana")).toEqual({ - newsletter: null, - taxId: ana.taxId, - name: ana.name, - }); - expect(await read(restored, "bob")).toEqual({ - newsletter: null, - taxId: bob.taxId, - name: null, - }); - await expectForgotten(storage, "ana", { - categories: ["marketing"], - values: [ana.newsletter], - }); - await expectForgotten(storage, "bob", { values: [bob.newsletter, bob.name] }); - }); + expect(await read(restored, "bob")).toEqual({ + newsletter: null, + taxId: bob.taxId, + name: null, + }); + await expectForgotten(storage, "ana", { + categories: ["marketing"], + values: [ana.newsletter], + }); + await expectForgotten(storage, "bob", { values: [bob.newsletter, bob.name] }); + }, + RESTORE_CASE_TIMEOUT_MS, + ); }); };