diff --git a/.changeset/aggregate-snapshots.md b/.changeset/aggregate-snapshots.md new file mode 100644 index 00000000..6a534f6b --- /dev/null +++ b/.changeset/aggregate-snapshots.md @@ -0,0 +1,7 @@ +--- +"@bounda-dev/core": minor +"@bounda-dev/cli": minor +"@bounda-dev/postgresql": minor +--- + +Snapshots of an aggregate's state, opt-in per aggregate with `runtime.overrides..snapshot: { every }`. A `snapshot:` subscriber writes them in the background, and a command loads the snapshot and folds only the events after it. A snapshot is a cache of the fold and never the source of truth: it is valid only for the `stateFingerprint` the generator now writes to the registry, a digest of the modules that shape the state and of the type inferred for it, plus the configured `snapshot.version`; one that is missing, from other code, unreadable or past the stream's head is ignored, and a share of the loads (`snapshot.verify`, `0.01` by default) is checked against a fold of the whole stream, which drops a snapshot that folds to another state. An app from `createTestApp` checks every load and throws `SnapshotError`. The state is kept with its types (`Date`, `Map`, `Set`, `bigint`). Every adapter gains a `snapshots` table and `Storage.snapshotStore`; `load` with `fromVersion` in the SQL event stores now reads the stream's head before the events, bounded by it, so an append landing between the two reads can no longer report a head past the events returned. diff --git a/docs/src/content/docs/adapters/index.mdx b/docs/src/content/docs/adapters/index.mdx index 127abac5..ec385801 100644 --- a/docs/src/content/docs/adapters/index.mdx +++ b/docs/src/content/docs/adapters/index.mdx @@ -88,6 +88,7 @@ On first use the adapter creates its tables, prefixed with `bounda_` by default: | `bounda_inbox` | Which handler already completed for which event, so a retry does not run it again | | `bounda_scheduled_commands` | Scheduled commands, process deadlines and timeouts, and delayed policy runs | | `bounda_dead_letters` | Handler runs that gave up, with the error and the attempt count | +| `bounda_snapshots` | The last snapshot of every stream of an aggregate that takes them, with the fingerprint it is valid for | Read models get one table each, named after the read model in snake_case behind `rm_`, so no read model can take one of the names above: `orderSummary` becomes `bounda_rm_order_summary`. diff --git a/docs/src/content/docs/concepts/long-streams.md b/docs/src/content/docs/concepts/long-streams.md index 7e135415..69ebea3e 100644 --- a/docs/src/content/docs/concepts/long-streams.md +++ b/docs/src/content/docs/concepts/long-streams.md @@ -1,6 +1,6 @@ --- title: Long streams close their books -description: What to do in Bounda with a stream that keeps growing, an account or the stock of a SKU, why closing the books comes before snapshots, and why snapshots will only come with a validity the runtime checks itself. +description: What to do in Bounda with a stream that keeps growing, an account or the stock of a SKU, why closing the books comes before snapshots, and how snapshots work when a stream has nothing to close, with a validity the runtime checks itself. sidebar: order: 51 --- @@ -74,7 +74,7 @@ command folds that month alone, while the closed ones stay in the event store fo that shows the account across years. The same shape fits a SKU's stock with a stock-take per period, or a cashier's shift. -## Why there are no snapshots yet +## Snapshots A snapshot stores the folded state so that the next load starts from it instead of from the first event. It is a cache, and a cache can be stale. The question is what a stale one breaks. @@ -94,15 +94,74 @@ reference says to adjust on every deployment that changes the aggregate's state. inferred by the generator and carries no version a person maintains, and a version someone has to remember to bump is the bug waiting for the one deploy where they forget. -So the stance is: **snapshots only with a validity the runtime checks itself**. They are planned as -a cache of the fold and never the source of truth: one that is missing, unreadable or doubtful is -ignored, and the stream is folded from the start, with the same result. A snapshot will be valid -only for a fingerprint of the modules that shape the state, so that a change to any of them makes -the old ones unusable without anyone deciding so. `bounda rebuild` already works that way: a paused -rebuild resumes only under a digest of the same view and projections, and starts again otherwise. -There is no date and no API for snapshots yet. +So Bounda's snapshots are **a cache of the fold, never the source of truth, with a validity the +runtime checks itself**. One that is missing, written by other code, unreadable or doubtful is +ignored, and the stream is folded from the start, with the same result. They are opt-in per +aggregate, for the stream that has nothing to close. -## What snapshots would not do +### Opting in + +```ts +// bounda.config.ts +runtime: { + overrides: { + account: { snapshot: { every: 200 } }, + }, +}, +``` + +`every` is how many events a stream gathers past its last snapshot before the next one is written. +A subscriber named `snapshot:account` follows the global stream like a projection does and writes +the snapshots in the background, in the `worker` role or an app from `createTestApp` after +`runUntilIdle()`, so no command ever waits for one: a command loads the snapshot, folds the events +after it and appends as before. The first deploy with snapshots walks the whole history and writes +one for every stream long enough, so the streams that hurt today are covered without a migration. +`app.getLag()` reports the subscriber like any other. + +### What a snapshot is valid for + +The generator computes a fingerprint of every aggregate's state and writes it to the registry as +`stateFingerprint`: a digest of `state.ts`, of every event module with its upcasts, of every module +those reach through relative imports, and of the state type it inferred. A snapshot is written +under that fingerprint and used only under the same one, so a change to any of those files, a +reformat or a comment included, retires every snapshot of the aggregate at the next deploy: each +stream is folded from its first event once, and the subscriber writes a fresh snapshot at its next +event. Nobody decides when; the cost is one slow load per stream, and a wrong state is never +decided from. + +What the fingerprint cannot see is a module reached through a package or a path alias such as +`~/lib/money`, since the generator follows relative paths only. When one of those changes what the +state folds to, raise `snapshot.version`, which is part of what the snapshot is valid for too: + +```ts +account: { snapshot: { every: 200, version: 2 } }, +``` + +### What the runtime checks + +A snapshot is used only when its fingerprint is the aggregate's, it reads back, and the stream +reaches its version; one that reads badly or is past the stream's head is dropped. On top of that, +`snapshot.verify`, `0.01` by default, is the share of loads from a snapshot that are checked +against a fold of the whole stream: a snapshot that folds to another state is logged, dropped and +not used, and the command goes on with the state of the stream. Such a difference means the fold is +not a function of the events alone, an `evolve` reading the clock or a module variable, which is +worth knowing however the state is loaded. The `bounda.snapshots` counter says what every load +found: `used`, `verified`, `mismatch`, `missing`, `stale`, `unreadable` or `ahead` +([observability](/reference/observability/)). + +An app from `createTestApp` checks every load and throws `SnapshotError` instead of dropping the +snapshot, from the command that loaded it or from `runUntilIdle()` when the subscriber met it. A +test suite that runs its aggregates with `snapshot` configured therefore proves their folds +deterministic ([testing](/guides/testing/#snapshots-are-checked)). + +### What a snapshot holds + +The state is kept with its types: `Date`, `Map`, `Set`, `bigint`, `undefined`, `RegExp` and +nested plain objects and arrays come back as they were. A class instance or a function in the +state cannot be kept: the subscriber logs it and writes no snapshot for that stream, and a test app +throws `SnapshotError` from `runUntilIdle()`. Keep the state to data. + +## What snapshots do not do - **Speed up a rebuild.** A rebuild runs projections over the global stream; it never folds an aggregate, so an aggregate's snapshot saves it nothing. What would make a large rebuild faster @@ -111,6 +170,9 @@ There is no date and no API for snapshots yet. Object, with everything that hangs off it) has one writer at a time, and a snapshot saves a read, not a commit. More throughput comes from [one store per tenant](/concepts/how-it-runs/#the-way-out-one-store-per-tenant). +- **Replace closing the books.** A stream with a period has a natural end and a state the next + period starts from; a snapshot keeps a stream without one loadable, while the stream itself + keeps growing and every projection of it still reads it whole. ## How others do it @@ -134,8 +196,9 @@ There is no date and no API for snapshots yet. ## What stays outside -- **A stream with no period and many events.** If the business has nothing to close, the fold is - paid on every command until snapshots exist. +- **A stream with no period and many events.** If the business has nothing to close, a snapshot + keeps each command's fold to the events since the last one; a projection or a read of the stream + still goes through all of it. - **The moment between periods.** A command writes one stream, so the next ledger opens when the policy runs, after the closing commits, not in the same transaction. An entry posted in between is rejected and retried by its caller. @@ -145,9 +208,8 @@ There is no date and no API for snapshots yet. ## Where to read more -- [What is not there yet](/reference/limitations/), the short version of this - page, and [Changing an event's shape](/guides/changing-events/) for the upcasters every fold - runs. +- [`snapshot`](/reference/configuration/#per-aggregate) in the configuration reference, and + [Changing an event's shape](/guides/changing-events/) for the upcasters every fold runs. - [State](/guides/project-layout/#state) and [policies](/guides/project-layout/#policies-policies), for the modules used above. - Oskar Dudycz, diff --git a/docs/src/content/docs/guides/changing-events.md b/docs/src/content/docs/guides/changing-events.md index f3b6325e..034013a0 100644 --- a/docs/src/content/docs/guides/changing-events.md +++ b/docs/src/content/docs/guides/changing-events.md @@ -94,9 +94,10 @@ what keeps the chain straight. new event type and keep handling the old one; upcasters transform, they do not invent. Upcasters are cheap but not free: they run on every read of an old event. An aggregate with -thousands of old events re-runs the chain on every command, since Bounda has no snapshots yet -([what is not there yet](/reference/limitations/)), and a rebuild runs it over -the whole global stream once. +thousands of old events re-runs the chain on every command, unless it takes +[snapshots](/concepts/long-streams/#snapshots), which fold the old events once, and a rebuild runs +it over the whole global stream once. Adding an upcaster retires the aggregate's snapshots, as any +change to how its state is folded does. ## What is not covered yet diff --git a/docs/src/content/docs/guides/deployment.mdx b/docs/src/content/docs/guides/deployment.mdx index 9928d00a..ab14b3fa 100644 --- a/docs/src/content/docs/guides/deployment.mdx +++ b/docs/src/content/docs/guides/deployment.mdx @@ -290,6 +290,11 @@ events a batch carries. Rebuilds honour it too. `backoff` sets how long a subscriber whose batches fail is left alone before the next try: see [A projection that keeps failing](#a-projection-that-keeps-failing). +An aggregate whose streams grow without a period to close takes +[snapshots](/concepts/long-streams/#snapshots) with `runtime.overrides..snapshot`; a +subscriber named `snapshot:` writes them in the background, and shows in the lag like +any other. + ## Observability The runtime reports OpenTelemetry spans and metrics with no adapter to write: register an SDK diff --git a/docs/src/content/docs/guides/testing.mdx b/docs/src/content/docs/guides/testing.mdx index af27b4d6..f3f7688c 100644 --- a/docs/src/content/docs/guides/testing.mdx +++ b/docs/src/content/docs/guides/testing.mdx @@ -276,6 +276,16 @@ stopped reacting: expect((await app.getLag()).maxLag).toBe(0); ``` +## Snapshots are checked + +An aggregate configured with [`snapshot`](/concepts/long-streams/#snapshots) gets them in a test +app too: `runUntilIdle()` writes them, and every later command loads from one. A test app checks +every such load against a fold of the whole stream, where a production app checks a sample, and +throws `SnapshotError` instead of dropping the snapshot: from the command whose state differs, or +from `runUntilIdle()` when the subscriber met the difference or a state it cannot keep. A suite +that passes with `snapshot` configured has proved the aggregate's fold a function of its events +alone, and its state made of data. + ## What is worth testing - **The rules**, through commands: what is accepted, what is rejected and with which code, and diff --git a/docs/src/content/docs/reference/cli.md b/docs/src/content/docs/reference/cli.md index b40140da..086df5cb 100644 --- a/docs/src/content/docs/reference/cli.md +++ b/docs/src/content/docs/reference/cli.md @@ -17,8 +17,10 @@ aggregates without `state.ts`. ## `bounda generate` Reads the project layout and writes `.bounda/registry.ts`, `.bounda/register.d.ts`, -`.bounda/types.ts` and one `+types/.ts` next to every module. Files whose content did not -change are left alone; `+types` files whose module is gone are removed, and so are the folders +`.bounda/types.ts` and one `+types/.ts` next to every module. The registry carries, for every +aggregate, a `stateFingerprint` of the modules that shape its state and of the type inferred for +it, which its [snapshots](/concepts/long-streams/#what-a-snapshot-is-valid-for) are valid for. +Files whose content did not change are left alone; `+types` files whose module is gone are removed, and so are the folders their removal empties; a folder holding nothing but `+types` (and Finder's `.DS_Store`) is not taken for a module. diff --git a/docs/src/content/docs/reference/configuration.md b/docs/src/content/docs/reference/configuration.md index b2737c86..7cfeafdc 100644 --- a/docs/src/content/docs/reference/configuration.md +++ b/docs/src/content/docs/reference/configuration.md @@ -100,8 +100,20 @@ a stopped one left: see [delivery guarantees](/concepts/delivery-guarantees/#the ### Per aggregate - **`overrides`**, default `{}`. Settings for one aggregate, by aggregate name: `commands.timeout`, - and `policies` and `processes` with the keys above. What an override leaves out comes from the - app's settings. + `policies` and `processes` with the keys above, and `snapshot`. What an override leaves out comes + from the app's settings. +- **`overrides..snapshot`**, none by default. Opts the aggregate into + [snapshots](/concepts/long-streams/#snapshots) of its state, written in the background and loaded + from instead of the stream's first event: + - **`every`**, required. How many events a stream gathers past its last snapshot before the next + one is written. + - **`verify`**, default `0.01`. The share of the loads from a snapshot, 0 to 1, checked against a + fold of the whole stream. A snapshot that folds to another state is logged, dropped and not + used. An app from `createTestApp` checks every load and throws `SnapshotError` instead. + - **`version`**, default `1`. Part of what a snapshot is valid for, next to the + `stateFingerprint` the generator computes from the modules that shape the state. Raise it to + retire every snapshot of the aggregate when the fold changed through something the fingerprint + cannot see, such as a package or a module behind a path alias. A process can set its own lifetime in its `config`, `({ events }) => ({ startedBy: [...], timeout: "48h" })`, which wins over `processes.timeout` and diff --git a/docs/src/content/docs/reference/errors.md b/docs/src/content/docs/reference/errors.md index 262f6a9d..4b43cfa6 100644 --- a/docs/src/content/docs/reference/errors.md +++ b/docs/src/content/docs/reference/errors.md @@ -72,6 +72,11 @@ dead letter with `errorType: "retriable_exhausted"`. event, or its process failed on another step first. The letter stays `failed`. - **`REBUILD_SUPERSEDED`** (`RebuildSupersededError`; an operator's answer). Another rebuild of the same read model took over; this one stopped without writing. `readModel`. +- **`SNAPSHOT_FAILED`** (`SnapshotError`; only in an app from `createTestApp`). A + [snapshot](/concepts/long-streams/#snapshots) did not stand for its stream: the state folded from + it differs from the one folded from the first event, or the state cannot be serialized. Thrown by + the command that loaded it or by `runUntilIdle()`. Any other app logs it, drops the snapshot and + folds the stream from the start. The last three are answers to an operator, from `app.deadLetters`, `bounda dead-letters` and `bounda rebuild`, not failures of a handler. diff --git a/docs/src/content/docs/reference/limitations.md b/docs/src/content/docs/reference/limitations.md index 1507540b..6504dcf5 100644 --- a/docs/src/content/docs/reference/limitations.md +++ b/docs/src/content/docs/reference/limitations.md @@ -8,12 +8,6 @@ sidebar: Bounda is 0.x, and this is the honest list of what a production app might want and does not get today. Each item says why, so nobody discovers it the hard way: -- **Snapshots.** An aggregate is folded from its whole stream on every command. That is fine for - the hundreds of events per instance that Bounda's kind of app produces, and it is not fine for - hundreds of thousands. Snapshots are deliberately not built yet: the state is inferred and - carries no version, so a snapshot written by yesterday's `evolve` would silently be wrong after - today's deploy. They come with a versioning story or not at all; until then, close the books - of an aggregate that would grow forever ([Long streams](/concepts/long-streams/)). - **Changing the shape of a process's state.** A process keeps its state in its own lifecycle events, so a change to that shape has the same problem an event payload has, and no `state.upcast.ts` yet. See @@ -31,5 +25,6 @@ get today. Each item says why, so nobody discovers it the hard way: What a production app does get, and where it is explained: [rebuilding a read model](/guides/deployment/#rebuilding-a-read-model) without taking it offline, [dead letters with a way out](/guides/dead-letters/), [upcasters](/guides/changing-events/) for -events whose payload changed, [observability](/reference/observability/) through OpenTelemetry, and -a dispatcher that reacts in milliseconds on PostgreSQL. +events whose payload changed, [snapshots](/concepts/long-streams/#snapshots) whose validity the +runtime checks itself, [observability](/reference/observability/) through OpenTelemetry, and a +dispatcher that reacts in milliseconds on PostgreSQL. diff --git a/docs/src/content/docs/reference/observability.md b/docs/src/content/docs/reference/observability.md index 8f9e9888..85ebbd38 100644 --- a/docs/src/content/docs/reference/observability.md +++ b/docs/src/content/docs/reference/observability.md @@ -25,7 +25,7 @@ Everything is reported under the scope `@bounda-dev/core`. Spans: | Span | When | Attributes | | --- | --- | --- | | `bounda.command ` | a command is dispatched | `bounda.command.type`, `bounda.aggregate.type`, `bounda.aggregate.id`, `bounda.correlation_id`, `bounda.causation_id`, `bounda.outcome` (`stored`, `scheduled`, `rejected`), `bounda.event.count`; a rejection adds the event `bounda.command.rejected`, its code in `bounda.rejected` | -| `bounda.subscriber ` | the dispatcher hands a batch to a projection, the policy runner or the process runner; idle passes produce none | `bounda.subscriber`, `bounda.subscriber.kind`, `bounda.position.after`, `bounda.event.count`, `bounda.outcome` (`advanced`, `held`, `failed`, `moved`) | +| `bounda.subscriber ` | the dispatcher hands a batch to a projection, the policy runner, the process runner or a snapshot subscriber; idle passes produce none | `bounda.subscriber`, `bounda.subscriber.kind` (`projection`, `policy`, `process`, `snapshot`), `bounda.position.after`, `bounda.event.count`, `bounda.outcome` (`advanced`, `held`, `failed`, `moved`) | | `bounda.projection .` | a projection handles one event | `bounda.read_model`, `bounda.projection`, the event's id, type and aggregate, `bounda.correlation_id` | | `bounda.policy .` | a policy handler runs | `bounda.policy`, the event's id, type and aggregate, `bounda.correlation_id`, `bounda.attempt` | | `bounda.process .` | a process handler runs; `… at ` for an `at-.ts`, `… at timeout` for `at-timeout.ts` | `bounda.process`, the event's id, type and aggregate, `bounda.correlation_id`, `bounda.attempt`; for a deadline, `bounda.process`, the process's aggregate type and id and `bounda.correlation_id` | @@ -46,8 +46,11 @@ Metrics: | `bounda.dispatcher.lag` | observable gauge, events each subscriber is behind the head | `bounda.subscriber` | | `bounda.commands` | counter | `bounda.command.type`, `bounda.outcome` (`stored`, `scheduled`, `rejected`, `failed`) | | `bounda.dead_letters` | counter | `bounda.handler.kind`, `bounda.handler`, `bounda.outcome` (`terminal`, `retriable_exhausted`) | +| `bounda.snapshots` | counter, loads of an aggregate with [snapshots](/concepts/long-streams/#snapshots) | `bounda.aggregate.type`, `bounda.outcome` (`used`, `verified`, `mismatch`, `missing`, `stale`, `unreadable`, `ahead`) | The lag gauge is what to alert on: a subscriber whose lag grows is a projection or a policy that is failing or stuck, and `app.getLag()` returns the same numbers for a health endpoint, with what a failing subscriber is stuck on (see -[A projection that keeps failing](/guides/deployment/#a-projection-that-keeps-failing)). +[A projection that keeps failing](/guides/deployment/#a-projection-that-keeps-failing)). A +`mismatch` on `bounda.snapshots` is worth one too: a snapshot folded to a state the stream does not +give, so an `evolve` of that aggregate depends on something other than the events. diff --git a/packages/cli/src/generate/emit/emit.test.ts b/packages/cli/src/generate/emit/emit.test.ts index f9876956..0deefb17 100644 --- a/packages/cli/src/generate/emit/emit.test.ts +++ b/packages/cli/src/generate/emit/emit.test.ts @@ -3,6 +3,7 @@ import { tmpdir } from "node:os"; import { dirname, join, relative, resolve, sep } from "node:path"; import { describe, expect, it } from "vitest"; import { discoverProject } from "../discover.ts"; +import { fingerprintStates, readStateSources } from "../fingerprint.ts"; import type { ProjectModel } from "../model.ts"; import { emitProject } from "./index.ts"; import { importPath } from "./paths.ts"; @@ -34,7 +35,10 @@ const generatedFilesOnDisk = async (root: string): Promise => describe("emitProject on the order-app fixture (golden)", () => { it("reproduces every generated file of the fixture byte for byte", async () => { const model = await discoverProject({ root: fixtureRoot }); - const files = emitProject({ model }); + const files = emitProject({ + model, + fingerprints: fingerprintStates({ sources: await readStateSources(model) }), + }); if (updateGolden) { for (const file of files) { await mkdir(dirname(file.path), { recursive: true }); diff --git a/packages/cli/src/generate/emit/index.ts b/packages/cli/src/generate/emit/index.ts index d4f19320..f681ceff 100644 --- a/packages/cli/src/generate/emit/index.ts +++ b/packages/cli/src/generate/emit/index.ts @@ -11,17 +11,22 @@ export const GENERATED_DIRECTORY: string = ".bounda"; export interface EmitProjectArgs { readonly model: ProjectModel; readonly inferredStates?: Readonly>; + /** + * By aggregate name, from `fingerprintStates`; an aggregate without one gets no + * `stateFingerprint`, so it cannot take snapshots. + */ + readonly fingerprints?: Readonly>; } export interface EmitProjectFunction { (args: EmitProjectArgs): readonly GeneratedFile[]; } -export const emitProject: EmitProjectFunction = ({ model, inferredStates }) => { +export const emitProject: EmitProjectFunction = ({ model, inferredStates, fingerprints = {} }) => { const directory = join(model.root, GENERATED_DIRECTORY); const typesPath = join(directory, "types.ts"); return [ - emitRegistry({ model, path: join(directory, "registry.ts") }), + emitRegistry({ model, path: join(directory, "registry.ts"), fingerprints }), emitRegister({ path: join(directory, "register.d.ts") }), emitTypes({ model, diff --git a/packages/cli/src/generate/emit/registry.ts b/packages/cli/src/generate/emit/registry.ts index 5a73d625..9a94776f 100644 --- a/packages/cli/src/generate/emit/registry.ts +++ b/packages/cli/src/generate/emit/registry.ts @@ -180,11 +180,18 @@ const emitUpcasts = (aggregate: AggregateModel, aliases: Aliases, indent: string ]; }; -const emitAggregate = (aggregate: AggregateModel, aliases: Aliases): string => { +const emitAggregate = ( + aggregate: AggregateModel, + aliases: Aliases, + fingerprint: string | undefined, +): string => { const indent = " "; return [ ` ${aggregate.name}: {`, ...(aggregate.state === null ? [] : [`${indent}state: ${aliases.of(aggregate.state.path)},`]), + ...(fingerprint === undefined + ? [] + : [`${indent}stateFingerprint: ${JSON.stringify(fingerprint)},`]), `${indent}events: ${record(aggregate.events.map((event) => [event.key, aliases.of(event.path)]))},`, ...emitUpcasts(aggregate, aliases, indent), ...emitPorts(aggregate, aliases, indent), @@ -226,6 +233,10 @@ export interface EmitRegistryArgs { * Absolute path the registry will be written to; imports are relative to it. */ readonly path: string; + /** + * By aggregate name; left out for an aggregate without one. + */ + readonly fingerprints?: Readonly>; } export interface EmitRegistryFunction { @@ -236,7 +247,7 @@ export interface EmitRegistryFunction { * `.bounda/registry.ts`: one namespace import per module, a type import per port, and the * structured registry `createApp` consumes. */ -export const emitRegistry: EmitRegistryFunction = ({ model, path }) => { +export const emitRegistry: EmitRegistryFunction = ({ model, path, fingerprints = {} }) => { const aliases = resolveAliases(model, path); const hasPorts = [...model.aggregates, ...model.readModels].some( (owner) => owner.ports.length > 0, @@ -250,7 +261,9 @@ export const emitRegistry: EmitRegistryFunction = ({ model, path }) => { "export const registry = {", ...group( "aggregates", - model.aggregates.map((aggregate) => emitAggregate(aggregate, aliases)), + model.aggregates.map((aggregate) => + emitAggregate(aggregate, aliases, fingerprints[aggregate.name]), + ), ), ...group( "readModels", diff --git a/packages/cli/src/generate/fingerprint.test.ts b/packages/cli/src/generate/fingerprint.test.ts new file mode 100644 index 00000000..9541f1a7 --- /dev/null +++ b/packages/cli/src/generate/fingerprint.test.ts @@ -0,0 +1,251 @@ +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { afterAll, describe, expect, it } from "vitest"; +import { fingerprintStates, readStateSources } from "./fingerprint.ts"; +import type { AggregateModel, ProjectModel } from "./model.ts"; + +const temporary: string[] = []; + +afterAll(async () => { + await Promise.all(temporary.map((root) => rm(root, { recursive: true, force: true }))); +}); + +const project = async (files: Readonly>): Promise => { + const root = await mkdtemp(join(tmpdir(), "bounda-fingerprint-")); + temporary.push(root); + for (const [path, content] of Object.entries(files)) { + await mkdir(dirname(join(root, path)), { recursive: true }); + await writeFile(join(root, path), content); + } + const ref = (path: string) => ({ path: join(root, path), relativePath: path }); + const order: AggregateModel = { + name: "order", + directory: join(root, "app/domain/order"), + state: "app/domain/order/state.ts" in files ? ref("app/domain/order/state.ts") : null, + events: Object.keys(files) + .filter((path) => /^app\/domain\/order\/order-[a-z-]+\.ts$/.test(path)) + .map((path) => ({ + ...ref(path), + key: "event", + typeName: "Event", + upcasts: + `${path.slice(0, -3)}.upcast.ts` in files ? ref(`${path.slice(0, -3)}.upcast.ts`) : null, + })), + ports: [], + commands: [], + policies: [], + processes: [], + }; + return { root, appDir: "app", aggregates: [order], readModels: [] }; +}; + +const base = { + "app/domain/order/state.ts": "export const initialState = { total: 0 };\n", + "app/domain/order/order-placed.ts": [ + 'import type { Event } from "./+types/order-placed";', + 'import { sum } from "./pricing/sum.js";', + 'export * from "../shared/money";', + "export const evolve = ({ state, event }: Event.EvolveArgs) => ({ total: sum(state.total, event.payload.total) });", + "", + ].join("\n"), + "app/domain/order/order-placed.upcast.ts": + "export const upcasts = [(payload: unknown) => payload];\n", + "app/domain/order/pricing/sum.ts": + 'import { round } from "../../shared";\nexport const sum = (a: number, b: number) => round(a + b);\n', + "app/domain/shared/index.ts": "export const round = (value: number) => Math.round(value);\n", + "app/domain/shared/money.ts": "export type Money = number;\nexport const zero: Money = 0;\n", + "app/domain/order/+types/order-placed.ts": + "export namespace Event { export type EvolveArgs = never; }\n", +}; + +const fingerprint = async (files: Readonly>): Promise => + fingerprintStates({ sources: await readStateSources(await project(files)) }).order ?? ""; + +describe("readStateSources", () => { + it("collects the state's modules and what they import through relative paths, by path", async () => { + const sources = await readStateSources(await project(base)); + expect(sources.order?.map((source) => source.relativePath)).toEqual([ + "app/domain/order/order-placed.ts", + "app/domain/order/order-placed.upcast.ts", + "app/domain/order/pricing/sum.ts", + "app/domain/order/state.ts", + "app/domain/shared/index.ts", + "app/domain/shared/money.ts", + ]); + expect(sources.order?.find((source) => source.relativePath.endsWith("state.ts"))?.text).toBe( + base["app/domain/order/state.ts"], + ); + }); + + it("follows what a dynamic import and a type-carrying import name, not a type-only one", async () => { + const sources = await readStateSources( + await project({ + ...base, + "app/domain/order/order-placed.ts": [ + 'import type { Money } from "../shared/money";', + 'import { type Round, round } from "../shared";', + 'const lazy = () => import("./pricing/sum.js");', + "export const evolve = () => ({});", + "", + ].join("\n"), + }), + ); + expect(sources.order?.map((source) => source.relativePath)).toEqual([ + "app/domain/order/order-placed.ts", + "app/domain/order/order-placed.upcast.ts", + "app/domain/order/pricing/sum.ts", + "app/domain/order/state.ts", + "app/domain/shared/index.ts", + ]); + }); + + it("leaves out packages, aliases, imports in comments and files that do not exist", async () => { + const sources = await readStateSources( + await project({ + ...base, + "app/domain/order/order-placed.ts": [ + 'import { z } from "zod";', + 'import { money } from "~/lib/money";', + '/* import { sum } from "./pricing/sum"; */', + '// import { round } from "../shared";', + 'import { gone } from "./gone";', + "export const evolve = () => ({});", + "", + ].join("\n"), + }), + ); + expect(sources.order?.map((source) => source.relativePath)).toEqual([ + "app/domain/order/order-placed.ts", + "app/domain/order/order-placed.upcast.ts", + "app/domain/order/state.ts", + ]); + }); + + it("resolves a specifier as TypeScript does, to a source file only", async () => { + const sources = await readStateSources( + await project({ + ...base, + "app/domain/order/order-placed.ts": [ + 'import { a } from "./lib/a.jsx";', + 'import { b } from "./lib/b.mjs";', + 'import { c } from "./lib/c";', + 'import { d } from "./lib/d";', + 'import { sum } from "./pricing/sum.js";', + "export const evolve = () => ({});", + "", + ].join("\n"), + "app/domain/order/lib/a.tsx": "export const a = 1;\n", + "app/domain/order/lib/b.mts": "export const b = 2;\n", + "app/domain/order/lib/c.tsx": "export const c = 3;\n", + "app/domain/order/lib/d/index.ts": "export const d = 4;\n", + "app/domain/order/pricing/sum.js": "export const sum = () => 0;\n", + }), + ); + expect(sources.order?.map((source) => source.relativePath)).toEqual([ + "app/domain/order/lib/a.tsx", + "app/domain/order/lib/b.mts", + "app/domain/order/lib/c.tsx", + "app/domain/order/lib/d/index.ts", + "app/domain/order/order-placed.ts", + "app/domain/order/order-placed.upcast.ts", + "app/domain/order/pricing/sum.ts", + "app/domain/order/state.ts", + "app/domain/shared/index.ts", + ]); + }); + + it("never takes a package for a file of the same name, and reads a cycle once", async () => { + const sources = await readStateSources( + await project({ + ...base, + "app/domain/order/order-placed.ts": [ + 'import { zero } from "money";', + 'import { round } from "../shared";', + "export const evolve = () => ({});", + "", + ].join("\n"), + "app/domain/order/money.ts": "export const zero = 0;\n", + "app/domain/shared/index.ts": + 'import { zero } from "./money";\nexport const round = (value: number) => value + zero;\n', + "app/domain/shared/money.ts": + 'import { round } from "./index";\nexport const zero = round(0);\n', + }), + ); + expect(sources.order?.map((source) => source.relativePath)).toEqual([ + "app/domain/order/order-placed.ts", + "app/domain/order/order-placed.upcast.ts", + "app/domain/order/state.ts", + "app/domain/shared/index.ts", + "app/domain/shared/money.ts", + ]); + }); + + it("reads the same text whatever the line endings", async () => { + const crlf = base["app/domain/order/state.ts"].replaceAll("\n", "\r\n"); + const sources = await readStateSources( + await project({ ...base, "app/domain/order/state.ts": crlf }), + ); + expect(sources.order?.find((source) => source.relativePath.endsWith("state.ts"))?.text).toBe( + base["app/domain/order/state.ts"], + ); + }); +}); + +describe("fingerprintStates", () => { + it("gives sixteen hex digits, the same for the same sources", async () => { + const first = await fingerprint(base); + expect(first).toMatch(/^[0-9a-f]{16}$/); + expect(await fingerprint(base)).toBe(first); + }); + + it("changes with any module that shapes the state, imported ones and comments included", async () => { + const first = await fingerprint(base); + const changed = async (path: keyof typeof base, content: string) => + fingerprint({ ...base, [path]: content }); + expect( + await changed("app/domain/order/state.ts", "export const initialState = { total: 1 };\n"), + ).not.toBe(first); + expect( + await changed( + "app/domain/shared/index.ts", + "export const round = (value: number) => Math.floor(value);\n", + ), + ).not.toBe(first); + expect( + await changed( + "app/domain/order/order-placed.upcast.ts", + "// v1\nexport const upcasts = [(payload: unknown) => payload];\n", + ), + ).not.toBe(first); + expect( + await changed( + "app/domain/order/+types/order-placed.ts", + "export namespace Event { export type EvolveArgs = unknown; }\n", + ), + ).toBe(first); + }); + + it("takes the inferred state type in, for the aggregates that have one", async () => { + const sources = await readStateSources(await project(base)); + const plain = fingerprintStates({ sources }); + const inferred = fingerprintStates({ + sources, + inferredStates: { order: { inferred: "{ readonly total: number; }" } }, + }); + const other = fingerprintStates({ + sources, + inferredStates: { + order: { inferred: "{ readonly total: number; readonly paid: boolean; }" }, + }, + }); + expect(inferred.order).not.toBe(plain.order); + expect(other.order).not.toBe(inferred.order); + expect(fingerprintStates({ sources, inferredStates: { order: { inferred: null } } })).toEqual( + plain, + ); + expect(fingerprintStates({ sources, inferredStates: { payment: { inferred: "{}" } } })).toEqual( + plain, + ); + }); +}); diff --git a/packages/cli/src/generate/fingerprint.ts b/packages/cli/src/generate/fingerprint.ts new file mode 100644 index 00000000..15afbdd3 --- /dev/null +++ b/packages/cli/src/generate/fingerprint.ts @@ -0,0 +1,139 @@ +import { createHash } from "node:crypto"; +import { readFile, stat } from "node:fs/promises"; +import { dirname, join, relative, resolve } from "node:path"; +import type { StateTypeSource } from "./emit/types.ts"; +import type { AggregateModel, ProjectModel } from "./model.ts"; +import { compareNames } from "./naming.ts"; +import { isInGenerated } from "./write.ts"; + +export interface StateSource { + // Relative to the project root, with forward slashes. + readonly relativePath: string; + readonly text: string; +} + +/** + * By aggregate name: the modules that shape its state, by path. + */ +export type StateSources = Readonly>; + +const COMMENTS = /\/\*[\s\S]*?\*\/|^\s*\/\/.*$/gm; +// Static imports and re-exports with a specifier, `import type` left out, and dynamic imports. +const STATIC_IMPORT = /^\s*import\s+(?!type\s)(?:[^"'`;]*?\sfrom\s*)?["']([^"'\n]+)["']/gm; +const EXPORT_FROM = /^\s*export\s+(?!type\s)[^"'`;]*?\sfrom\s*["']([^"'\n]+)["']/gm; +const DYNAMIC_IMPORT = /\bimport\(\s*["']([^"'\n]+)["']\s*\)/g; + +const specifiersOf = (text: string): readonly string[] => { + const stripped = text.replace(COMMENTS, ""); + return [STATIC_IMPORT, EXPORT_FROM, DYNAMIC_IMPORT].flatMap((pattern) => + [...stripped.matchAll(pattern)].map((match) => match[1] ?? ""), + ); +}; + +const isFile = async (path: string): Promise => + stat(path).then( + (found) => found.isFile(), + () => false, + ); + +/** + * As TypeScript resolves a relative specifier to a source file: as written, `.js` for `.ts`, + * without an extension, or a folder's `index.ts`. `null` for one no file answers to. + */ +const resolveRelative = async (from: string, specifier: string): Promise => { + const base = resolve(dirname(from), specifier); + const candidates = [ + base, + base.replace(/\.(js|jsx|mjs)$/, (_match, extension: string) => + extension === "jsx" ? ".tsx" : extension === "mjs" ? ".mts" : ".ts", + ), + `${base}.ts`, + `${base}.tsx`, + join(base, "index.ts"), + ].filter((candidate) => /\.(ts|tsx|mts)$/.test(candidate)); + for (const candidate of candidates) { + if (await isFile(candidate)) return candidate; + } + return null; +}; + +const rootsOf = (aggregate: AggregateModel): readonly string[] => [ + ...(aggregate.state === null ? [] : [aggregate.state.path]), + ...aggregate.events.flatMap((event) => [ + event.path, + ...(event.upcasts === null ? [] : [event.upcasts.path]), + ]), +]; + +/** + * Every module reached from `roots` through relative imports, the generated files left out, + * since they are made from these same modules and what they import is types alone. A + * specifier that is not relative, a package or a path alias, is not followed. + */ +const walk = async (root: string, roots: readonly string[]): Promise => { + const texts = new Map(); + const pending = [...roots]; + for (let path = pending.shift(); path !== undefined; path = pending.shift()) { + if (texts.has(path)) continue; + const text = (await readFile(path, "utf8")).replaceAll("\r\n", "\n"); + texts.set(path, text); + for (const specifier of specifiersOf(text)) { + if (!specifier.startsWith("./") && !specifier.startsWith("../")) continue; + const resolved = await resolveRelative(path, specifier); + if (resolved === null || isInGenerated(resolved) || resolved.includes("node_modules")) { + continue; + } + pending.push(resolved); + } + } + return [...texts] + .map(([path, text]) => ({ + relativePath: relative(root, path).split("\\").join("/"), + text, + })) + .sort((left, right) => compareNames(left.relativePath, right.relativePath)); +}; + +export interface ReadStateSourcesFunction { + (model: ProjectModel): Promise; +} + +/** + * Reads, for every aggregate, the modules that shape its state: `state.ts`, its events with + * their upcasts, and whatever those import through relative paths, transitively. + */ +export const readStateSources: ReadStateSourcesFunction = async (model) => + Object.fromEntries( + await Promise.all( + model.aggregates.map( + async (aggregate) => [aggregate.name, await walk(model.root, rootsOf(aggregate))] as const, + ), + ), + ); + +export interface FingerprintStatesArgs { + readonly sources: StateSources; + /** + * By aggregate name; the type the generator inferred, when it did, is part of the fingerprint. + */ + readonly inferredStates?: Readonly>; +} + +export interface FingerprintStatesFunction { + (args: FingerprintStatesArgs): Readonly>; +} + +/** + * Sixteen hex digits per aggregate over the text of its state's modules, by path, and the state + * type inferred for it: the same on every machine for the same sources, and different as soon as + * any of them changes, formatting and comments included. + */ +export const fingerprintStates: FingerprintStatesFunction = ({ sources, inferredStates = {} }) => + Object.fromEntries( + Object.entries(sources).map(([aggregate, modules]) => { + const hash = createHash("sha256"); + for (const { relativePath, text } of modules) hash.update(`${relativePath}\n${text}\0`); + hash.update(inferredStates[aggregate]?.inferred ?? ""); + return [aggregate, hash.digest("hex").slice(0, 16)]; + }), + ); diff --git a/packages/cli/src/generate/generate.ts b/packages/cli/src/generate/generate.ts index b991754e..4043d143 100644 --- a/packages/cli/src/generate/generate.ts +++ b/packages/cli/src/generate/generate.ts @@ -4,6 +4,7 @@ import { discoverProject } from "./discover.ts"; import { emitProject, GENERATED_DIRECTORY } from "./emit/index.ts"; import type { GeneratedFile } from "./emit/paths.ts"; import type { StateTypeSource } from "./emit/types.ts"; +import { fingerprintStates, readStateSources } from "./fingerprint.ts"; import type { GenerateWarning, ProjectModel } from "./model.ts"; import { inferStates } from "./state/infer.ts"; import { removeOrphans, writeGeneratedFile, writeGeneratedFiles } from "./write.ts"; @@ -56,19 +57,28 @@ export const generateProject: GenerateProjectFunction = async ({ inferState = true, }) => { const { warnings, ...model } = await discoverProject({ root, appDir }); + const sources = await readStateSources(model); const typesPath = join(root, GENERATED_DIRECTORY, "types.ts"); - const typesBefore = await readFile(typesPath, "utf8").catch(() => null); - const firstPass = emitProject({ model }); + const registryPath = join(root, GENERATED_DIRECTORY, "registry.ts"); + const before = (path: string): Promise => readFile(path, "utf8").catch(() => null); + const typesBefore = await before(typesPath); + const registryBefore = await before(registryPath); + const firstPass = emitProject({ model, fingerprints: fingerprintStates({ sources }) }); const needsInference = inferState && model.aggregates.some((aggregate) => aggregate.state === null); - // Inference reads the first-pass types from memory, so an existing types.ts is written once, - // with its final content, or not at all: watchers never see the uninferred one. The project - // only has to find the file, which a first run writes. - const first = await writeGeneratedFiles( - needsInference && typesBefore !== null - ? firstPass.filter((file) => file.path !== typesPath) - : firstPass, + // Inference reads the first-pass types from memory, and the registry's fingerprints take the + // inferred types in, so each of the two, when it exists, is written once, with its final + // content, or not at all: watchers never see the uninferred one. The project only has to find + // the files, which a first run writes. + const deferred = new Set( + needsInference + ? [ + ...(typesBefore === null ? [] : [typesPath]), + ...(registryBefore === null ? [] : [registryPath]), + ] + : [], ); + const first = await writeGeneratedFiles(firstPass.filter((file) => !deferred.has(file.path))); const typesRender = (states: Readonly>): string => emitProject({ model, inferredStates: states }).find((file) => file.path === typesPath) ?.content ?? ""; @@ -81,16 +91,25 @@ export const generateProject: GenerateProjectFunction = async ({ renderTypes: typesRender, }) : { states: {}, warnings: [] }; - const files = emitProject({ model, inferredStates: inferred.states }); - const typesAfter = files.find((file) => file.path === typesPath)?.content ?? ""; + const files = emitProject({ + model, + inferredStates: inferred.states, + fingerprints: fingerprintStates({ sources, inferredStates: inferred.states }), + }); + const contentOf = (path: string): string => + files.find((file) => file.path === path)?.content ?? ""; + const typesAfter = contentOf(typesPath); + const registryAfter = contentOf(registryPath); await writeGeneratedFile(typesPath, typesAfter); + await writeGeneratedFile(registryPath, registryAfter); const removed = await removeOrphans({ appDirectory: join(root, appDir), keep: new Set(files.map((file) => file.path)), }); const written = [ - ...first.written.filter((path) => path !== typesPath), + ...first.written.filter((path) => path !== typesPath && path !== registryPath), ...(typesAfter === typesBefore ? [] : [typesPath]), + ...(registryAfter === registryBefore ? [] : [registryPath]), ].sort(); return { model, diff --git a/packages/cloudflare/test/storage-contracts.test.ts b/packages/cloudflare/test/storage-contracts.test.ts index 62b72331..69069b0a 100644 --- a/packages/cloudflare/test/storage-contracts.test.ts +++ b/packages/cloudflare/test/storage-contracts.test.ts @@ -12,6 +12,7 @@ import { readModelRebuildContract, readModelTransactionContract, schedulerContract, + snapshotStoreContract, storageTransactionContract, tableContract, viewContract, @@ -36,6 +37,7 @@ describe("the Durable Object adapter", () => { inboxLedgerContract({ create: async () => (await freshStorage()).inboxLedger }); deadLetterStoreContract({ create: async () => (await freshStorage()).deadLetterStore }); schedulerContract({ create: async () => (await freshStorage()).scheduler }); + snapshotStoreContract({ create: async () => (await freshStorage()).snapshotStore }); storageTransactionContract({ create: freshStorage }); tableContract({ create: async () => diff --git a/packages/core/README.md b/packages/core/README.md index 42422dfc..ebf00da3 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -50,8 +50,10 @@ swaps it in without taking the read model offline; `bounda dead-letters` lists, discards the handler runs that gave up; an `.upcast.ts` next to an event brings stored payloads of an older shape up to date as they are read. Every command, batch and handler run is an OpenTelemetry span with `bounda.correlation_id`, the lag of every subscriber is a gauge, and on -PostgreSQL the dispatcher is woken by `NOTIFY` instead of polling. Snapshots are not there yet, on -purpose; the [deployment guide](https://docs.bounda.dev/guides/deployment/) says why. +PostgreSQL the dispatcher is woken by `NOTIFY` instead of polling. An aggregate whose streams never +end takes snapshots, opt-in and valid only for a fingerprint of the code that folds the state, so +a deploy that changes the fold retires them on its own; +[Long streams](https://docs.bounda.dev/concepts/long-streams/) says how. ## Exports diff --git a/packages/core/package.json b/packages/core/package.json index 812330c3..fcb7675f 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -84,6 +84,7 @@ }, "dependencies": { "@opentelemetry/api": "catalog:", + "devalue": "catalog:", "uuid": "catalog:", "zod": "catalog:" }, diff --git a/packages/core/src/adapter/adapter.ts b/packages/core/src/adapter/adapter.ts index f987cff8..28a3fdb4 100644 --- a/packages/core/src/adapter/adapter.ts +++ b/packages/core/src/adapter/adapter.ts @@ -7,6 +7,7 @@ import type { EventNotifier } from "./storage/event-notifier.ts"; import type { EventStore } from "./storage/event-store.ts"; import type { InboxLedger } from "./storage/inbox-ledger.ts"; import type { Scheduler } from "./storage/scheduler.ts"; +import type { SnapshotStore } from "./storage/snapshot-store.ts"; import type { ReadClient, Table } from "./storage/table.ts"; /** @@ -32,6 +33,11 @@ export interface Storage { readonly inboxLedger: InboxLedger; readonly deadLetterStore: DeadLetterStore; readonly scheduler: Scheduler; + /** + * Written by the snapshot subscriber, outside any command's transaction, and read by every + * command of an aggregate with snapshots. + */ + readonly snapshotStore: SnapshotStore; /** * Only for a backend that can push new events; without it the dispatcher only polls. */ diff --git a/packages/core/src/adapter/index.ts b/packages/core/src/adapter/index.ts index 9fd61312..51b0d600 100644 --- a/packages/core/src/adapter/index.ts +++ b/packages/core/src/adapter/index.ts @@ -58,4 +58,5 @@ export type { ScheduledCommand, Scheduler, } from "./storage/scheduler.ts"; +export type { Snapshot, SnapshotStore } from "./storage/snapshot-store.ts"; export type { FindManyArgs, ReadClient, Table, TableOrder } from "./storage/table.ts"; diff --git a/packages/core/src/adapter/sqlite/adapter.ts b/packages/core/src/adapter/sqlite/adapter.ts index 979d9a0a..809b45db 100644 --- a/packages/core/src/adapter/sqlite/adapter.ts +++ b/packages/core/src/adapter/sqlite/adapter.ts @@ -12,6 +12,7 @@ import { createSqliteInboxLedger } from "./inbox-ledger.ts"; import { openSqliteReadModel, rebuildSqliteReadModel } from "./read-model.ts"; import { createSqliteScheduler } from "./scheduler.ts"; import { ensureStorageSchema, storageTablesFor } from "./schema.ts"; +import { createSqliteSnapshotStore } from "./snapshot-store.ts"; /** * One handle on a SQLite database: what the stores write through, and what queries receive as @@ -98,6 +99,7 @@ export const createSqliteAdapter: CreateSqliteAdapterFunction = < return { ...storesOver(db), checkpointStore: createSqliteCheckpointStore({ db, table: tables.checkpoints }), + snapshotStore: createSqliteSnapshotStore({ db, table: tables.snapshots }), transact: (work) => db.write((tx) => work(storesOver(boundTo(tx)))), close: release, }; diff --git a/packages/core/src/adapter/sqlite/event-store.ts b/packages/core/src/adapter/sqlite/event-store.ts index 26691630..b0156826 100644 --- a/packages/core/src/adapter/sqlite/event-store.ts +++ b/packages/core/src/adapter/sqlite/event-store.ts @@ -71,18 +71,27 @@ export const createSqliteEventStore: CreateSqliteEventStoreFunction = ({ db, tab } return { version: actualVersion + stored.length, events: stored }; }), + // A load from the first event reads one statement, whose last row is the head. From later + // on, the head is read first and the events bounded by it: outside a transaction, an append + // landing between the two reads would otherwise report a head past the events returned. load: async ({ aggregateType, aggregateId, fromVersion = 1 }) => { - const rows = await db.all( - `SELECT ${COLUMNS} FROM ${table} WHERE "aggregate_type" = ? AND "aggregate_id" = ? AND "version" >= ? ORDER BY "version"`, - [aggregateType, aggregateId, fromVersion], - ); - const events = rows.map(toStoredEvent); - const last = events.at(-1); - const version = - fromVersion <= 1 - ? (last?.version ?? 0) - : await currentVersion(db, aggregateType, aggregateId); - return { events, version }; + if (fromVersion <= 1) { + const rows = await db.all( + `SELECT ${COLUMNS} FROM ${table} WHERE "aggregate_type" = ? AND "aggregate_id" = ? ORDER BY "version"`, + [aggregateType, aggregateId], + ); + const events = rows.map(toStoredEvent); + return { events, version: events.at(-1)?.version ?? 0 }; + } + const version = await currentVersion(db, aggregateType, aggregateId); + const rows = + version < fromVersion + ? [] + : await db.all( + `SELECT ${COLUMNS} FROM ${table} WHERE "aggregate_type" = ? AND "aggregate_id" = ? AND "version" >= ? AND "version" <= ? ORDER BY "version"`, + [aggregateType, aggregateId, fromVersion, version], + ); + return { events: rows.map(toStoredEvent), version }; }, readAll: async ({ afterPosition, limit }) => ( diff --git a/packages/core/src/adapter/sqlite/schema.ts b/packages/core/src/adapter/sqlite/schema.ts index 87e29544..948e2621 100644 --- a/packages/core/src/adapter/sqlite/schema.ts +++ b/packages/core/src/adapter/sqlite/schema.ts @@ -8,6 +8,7 @@ export interface StorageTables { readonly inbox: string; readonly deadLetters: string; readonly scheduledCommands: string; + readonly snapshots: string; } export interface StorageTablesForFunction { @@ -20,6 +21,7 @@ export const storageTablesFor: StorageTablesForFunction = (prefix) => ({ inbox: quoteIdentifier(storageTableNameFor({ prefix, table: "inbox" })), deadLetters: quoteIdentifier(storageTableNameFor({ prefix, table: "deadLetters" })), scheduledCommands: quoteIdentifier(storageTableNameFor({ prefix, table: "scheduledCommands" })), + snapshots: quoteIdentifier(storageTableNameFor({ prefix, table: "snapshots" })), }); const indexName = (table: string, suffix: string): string => @@ -100,6 +102,14 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "claim_id" TEXT )`, `CREATE INDEX IF NOT EXISTS ${indexName(tables.scheduledCommands, "execute_at")} ON ${tables.scheduledCommands} ("execute_at")`, + `CREATE TABLE IF NOT EXISTS ${tables.snapshots} ( + "aggregate_type" TEXT NOT NULL, + "aggregate_id" TEXT NOT NULL, + "version" INTEGER NOT NULL, + "fingerprint" TEXT NOT NULL, + "state" TEXT NOT NULL, + PRIMARY KEY ("aggregate_type", "aggregate_id") + )`, ]; export interface EnsureStorageSchemaArgs { diff --git a/packages/core/src/adapter/sqlite/snapshot-store.ts b/packages/core/src/adapter/sqlite/snapshot-store.ts new file mode 100644 index 00000000..114ab0f9 --- /dev/null +++ b/packages/core/src/adapter/sqlite/snapshot-store.ts @@ -0,0 +1,39 @@ +import type { Snapshot, SnapshotStore } from "../index.ts"; +import type { SqlExecutor } from "../sql/sql-table.ts"; + +export interface CreateSqliteSnapshotStoreArgs { + readonly db: SqlExecutor; + readonly table: string; +} + +export interface CreateSqliteSnapshotStoreFunction { + (args: CreateSqliteSnapshotStoreArgs): SnapshotStore; +} + +const toSnapshot = (row: Record): Snapshot => ({ + aggregateType: String(row.aggregate_type), + aggregateId: String(row.aggregate_id), + version: Number(row.version), + fingerprint: String(row.fingerprint), + state: String(row.state), +}); + +export const createSqliteSnapshotStore: CreateSqliteSnapshotStoreFunction = ({ db, table }) => ({ + get: async ({ aggregateType, aggregateId }) => { + const [row] = await db.all( + `SELECT "aggregate_type", "aggregate_id", "version", "fingerprint", "state" FROM ${table} WHERE "aggregate_type" = ? AND "aggregate_id" = ?`, + [aggregateType, aggregateId], + ); + return row === undefined ? null : toSnapshot(row); + }, + put: ({ aggregateType, aggregateId, version, fingerprint, state }) => + db.run( + `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state") VALUES (?, ?, ?, ?, ?) ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state"`, + [aggregateType, aggregateId, version, fingerprint, state], + ), + remove: ({ aggregateType, aggregateId }) => + db.run(`DELETE FROM ${table} WHERE "aggregate_type" = ? AND "aggregate_id" = ?`, [ + aggregateType, + aggregateId, + ]), +}); diff --git a/packages/core/src/adapter/sqlite/sqlite.test.ts b/packages/core/src/adapter/sqlite/sqlite.test.ts index 39fd1817..75e80974 100644 --- a/packages/core/src/adapter/sqlite/sqlite.test.ts +++ b/packages/core/src/adapter/sqlite/sqlite.test.ts @@ -17,11 +17,14 @@ import { readModelRebuildContract, readModelTransactionContract, schedulerContract, + snapshotStoreContract, storageTransactionContract, tableContract, viewContract, } from "../testing/index.ts"; +import { createSqliteEventStore } from "./event-store.ts"; import { createSqliteAdapter } from "./index.ts"; +import { storageTablesFor } from "./schema.ts"; const bind = (params: readonly unknown[]): SQLInputValue[] => params.map((value) => (typeof value === "boolean" ? Number(value) : value)) as SQLInputValue[]; @@ -98,6 +101,7 @@ describe("the SQLite stores on node:sqlite", () => { inboxLedgerContract({ create: async () => (await storage()).inboxLedger }); deadLetterStoreContract({ create: async () => (await storage()).deadLetterStore }); schedulerContract({ create: async () => (await storage()).scheduler }); + snapshotStoreContract({ create: async () => (await storage()).snapshotStore }); storageTransactionContract({ create: storage }); tableContract({ create: async () => @@ -117,6 +121,73 @@ describe("the SQLite stores on node:sqlite", () => { }); }); +describe("the SQLite event store", () => { + it("never reports a version past the events a load from a later version returns", async () => { + const { adapter, db } = nodeSqlite(); + const storage = await adapter.createStorage({ logger: silentLogger }); + const stream = { aggregateType: "order", aggregateId: "1" }; + await storage.eventStore.append({ + ...stream, + expectedVersion: 0, + events: [1, 2].map((version) => pendingEvent({ aggregateId: "1", version })), + }); + // An append lands between the first and the second statement of the load. + let statements = 0; + const racing: SqlDatabase = { + run: async (sql, params) => { + db.prepare(sql).run(...bind(params)); + }, + all: async (sql, params) => { + statements += 1; + if (statements === 2) { + await storage.eventStore.append({ + ...stream, + expectedVersion: 2, + events: [pendingEvent({ aggregateId: "1", version: 3 })], + }); + } + return db.prepare(sql).all(...bind(params)) as Record[]; + }, + write: () => Promise.reject(new Error("not written here")), + }; + const store = createSqliteEventStore({ db: racing, table: storageTablesFor("bounda_").events }); + const loaded = await store.load({ ...stream, fromVersion: 2 }); + expect(loaded.events.map((event) => event.version)).toEqual([2]); + expect(loaded.version).toBe(2); + expect(statements).toBe(2); + }); + + it("reads a load from the first event, or from past the head, in one statement", async () => { + const { adapter, db } = nodeSqlite(); + const storage = await adapter.createStorage({ logger: silentLogger }); + const stream = { aggregateType: "order", aggregateId: "1" }; + await storage.eventStore.append({ + ...stream, + expectedVersion: 0, + events: [1, 2].map((version) => pendingEvent({ aggregateId: "1", version })), + }); + let statements = 0; + const counting: SqlDatabase = { + run: async () => {}, + all: async (sql, params) => { + statements += 1; + return db.prepare(sql).all(...bind(params)) as Record[]; + }, + write: () => Promise.reject(new Error("not written here")), + }; + const store = createSqliteEventStore({ + db: counting, + table: storageTablesFor("bounda_").events, + }); + expect(await store.load(stream)).toMatchObject({ version: 2 }); + expect(statements).toBe(1); + expect(await store.load({ ...stream, fromVersion: 1 })).toMatchObject({ version: 2 }); + expect(statements).toBe(2); + expect(await store.load({ ...stream, fromVersion: 3 })).toEqual({ events: [], version: 2 }); + expect(statements).toBe(3); + }); +}); + describe("createSqliteAdapter", () => { it("is an adapter with the name and options it was given", () => { const { adapter } = nodeSqlite(undefined, "x_"); @@ -242,6 +313,7 @@ describe("createSqliteAdapter", () => { "app_inbox", "app_rm_order_summary", "app_scheduled_commands", + "app_snapshots", ]); }); diff --git a/packages/core/src/adapter/storage/snapshot-store.ts b/packages/core/src/adapter/storage/snapshot-store.ts new file mode 100644 index 00000000..ca057102 --- /dev/null +++ b/packages/core/src/adapter/storage/snapshot-store.ts @@ -0,0 +1,22 @@ +import type { StreamIdentity } from "../../contracts/event.ts"; + +/** + * The state of one aggregate stream folded up to `version`, as the kernel serialized it, valid + * only for the `fingerprint` of the code that folded it. + */ +export interface Snapshot extends StreamIdentity { + readonly version: number; + readonly fingerprint: string; + readonly state: string; +} + +/** + * One snapshot per stream, the last one `put`: a cache of the fold, never the source of truth, + * so a kernel that finds one missing, stale or doubtful folds the stream from its first event + * instead. `get` is `null` for a stream without one. `remove` forgets a stream's snapshot. + */ +export interface SnapshotStore { + get(stream: StreamIdentity): Promise; + put(snapshot: Snapshot): Promise; + remove(stream: StreamIdentity): Promise; +} diff --git a/packages/core/src/adapter/testing/index.ts b/packages/core/src/adapter/testing/index.ts index 37317c73..46d50a36 100644 --- a/packages/core/src/adapter/testing/index.ts +++ b/packages/core/src/adapter/testing/index.ts @@ -36,6 +36,11 @@ export type { export { readModelTransactionContract } from "./read-model-transaction.contract.ts"; export type { SchedulerContractArgs, SchedulerContractFunction } from "./scheduler.contract.ts"; export { schedulerContract } from "./scheduler.contract.ts"; +export type { + SnapshotStoreContractArgs, + SnapshotStoreContractFunction, +} from "./snapshot-store.contract.ts"; +export { snapshotStoreContract } from "./snapshot-store.contract.ts"; export type { StorageTransactionContractArgs, StorageTransactionContractFunction, diff --git a/packages/core/src/adapter/testing/read-model-rebuild.contract.ts b/packages/core/src/adapter/testing/read-model-rebuild.contract.ts index e2ca3f89..7c5d68ec 100644 --- a/packages/core/src/adapter/testing/read-model-rebuild.contract.ts +++ b/packages/core/src/adapter/testing/read-model-rebuild.contract.ts @@ -376,7 +376,14 @@ export const readModelRebuildContract: ReadModelRebuildContractFunction = ({ expectedVersion: 0, events: [pendingEvent({ aggregateId: "1", version: 1 })], }); - for (const name of ["events", "checkpoints", "inbox", "deadLetters", "scheduledCommands"]) { + for (const name of [ + "events", + "checkpoints", + "inbox", + "deadLetters", + "scheduledCommands", + "snapshots", + ]) { const model = await adapter.createReadModel({ name, fields: contractFields, diff --git a/packages/core/src/adapter/testing/snapshot-store.contract.ts b/packages/core/src/adapter/testing/snapshot-store.contract.ts new file mode 100644 index 00000000..3c9bacc0 --- /dev/null +++ b/packages/core/src/adapter/testing/snapshot-store.contract.ts @@ -0,0 +1,72 @@ +import { beforeEach, describe, expect, it } from "vitest"; +import type { Snapshot, SnapshotStore } from "../storage/snapshot-store.ts"; + +export interface SnapshotStoreContractArgs { + readonly create: () => Promise; +} + +export interface SnapshotStoreContractFunction { + (args: SnapshotStoreContractArgs): void; +} + +const order42: Snapshot = { + aggregateType: "order", + aggregateId: "42", + version: 200, + fingerprint: "abc", + state: '[{"status":1,"total":2},"placed",19.5]', +}; + +/** + * The behaviour every snapshot store must exhibit. + */ +export const snapshotStoreContract: SnapshotStoreContractFunction = ({ create }) => { + describe("snapshot store contract", () => { + let store: SnapshotStore; + + beforeEach(async () => { + store = await create(); + }); + + it("has nothing for a stream never snapshotted", async () => { + expect(await store.get({ aggregateType: "order", aggregateId: "42" })).toBeNull(); + }); + + it("keeps one snapshot per stream, as it was put", async () => { + const account42: Snapshot = { + ...order42, + aggregateType: "account", + version: 7, + fingerprint: "def", + state: "[1]", + }; + await store.put(order42); + await store.put(account42); + expect(await store.get({ aggregateType: "order", aggregateId: "42" })).toEqual(order42); + expect(await store.get({ aggregateType: "account", aggregateId: "42" })).toEqual(account42); + expect(await store.get({ aggregateType: "order", aggregateId: "43" })).toBeNull(); + }); + + it("replaces a stream's snapshot with the last one put", async () => { + await store.put(order42); + const later: Snapshot = { ...order42, version: 400, fingerprint: "ghi", state: "[2]" }; + await store.put(later); + expect(await store.get({ aggregateType: "order", aggregateId: "42" })).toEqual(later); + const earlier: Snapshot = { ...order42, version: 100 }; + await store.put(earlier); + expect(await store.get({ aggregateType: "order", aggregateId: "42" })).toEqual(earlier); + }); + + it("forgets the snapshot of the stream it is asked to remove, and only that one", async () => { + await store.put(order42); + await store.put({ ...order42, aggregateId: "43" }); + await store.remove({ aggregateType: "order", aggregateId: "42" }); + await store.remove({ aggregateType: "order", aggregateId: "never" }); + expect(await store.get({ aggregateType: "order", aggregateId: "42" })).toBeNull(); + expect(await store.get({ aggregateType: "order", aggregateId: "43" })).toEqual({ + ...order42, + aggregateId: "43", + }); + }); + }); +}; diff --git a/packages/core/src/config/defaults.ts b/packages/core/src/config/defaults.ts index a82ad757..80c8300f 100644 --- a/packages/core/src/config/defaults.ts +++ b/packages/core/src/config/defaults.ts @@ -37,3 +37,12 @@ export const DEFAULT_BACKOFF_BASE_DELAY_MS: number = 1_000; export const DEFAULT_BACKOFF_MAX_DELAY_MS: number = 30_000; export const DEFAULT_CATCH_UP_TIMEOUT_MS: number = 2_000; export const DEFAULT_CATCH_UP_POLL_MS: number = 15; +/** + * The share of the loads from a snapshot checked against the whole stream when `snapshot.verify` + * is left out. + */ +export const DEFAULT_SNAPSHOT_VERIFY: number = 0.01; +/** + * The `snapshot.version` of an aggregate whose configuration sets none. + */ +export const DEFAULT_SNAPSHOT_VERSION: number = 1; diff --git a/packages/core/src/config/index.ts b/packages/core/src/config/index.ts index 8e1e7882..02a51713 100644 --- a/packages/core/src/config/index.ts +++ b/packages/core/src/config/index.ts @@ -1,3 +1,4 @@ +export { DEFAULT_SNAPSHOT_VERIFY, DEFAULT_SNAPSHOT_VERSION } from "./defaults.ts"; export type { DefineConfigFunction, WithExactPorts } from "./define-config.ts"; export { defineConfig } from "./define-config.ts"; export type { @@ -13,7 +14,9 @@ export type { PortsSection, ProcessesConfig, ResolvedConfig, + ResolvedSnapshotConfig, RetryConfig, RuntimeConfig, RuntimeRole, + SnapshotConfig, } from "./types.ts"; diff --git a/packages/core/src/config/schema.test.ts b/packages/core/src/config/schema.test.ts index 780fa540..8bcda240 100644 --- a/packages/core/src/config/schema.test.ts +++ b/packages/core/src/config/schema.test.ts @@ -119,6 +119,7 @@ describe("resolveConfig", () => { maxChainDepth: 3, }, processes: resolved.runtime.processes, + snapshot: null, }); expect(resolved.forAggregate("payment").commands).toEqual({ timeoutMs: 60_000 }); expect(resolved.forAggregate("payment").processes).toEqual({ @@ -129,10 +130,62 @@ describe("resolveConfig", () => { commands: { timeoutMs: 10_000 }, policies: resolved.runtime.policies, processes: resolved.runtime.processes, + snapshot: null, }); expect(resolved.forAggregate("customer").policies.maxChainDepth).toBe(10); }); + it("resolves the snapshot settings of an aggregate that opts in, and none for the rest", () => { + const resolved = resolveConfig({ + storage: sqlite, + runtime: { + overrides: { + order: { snapshot: { every: 200 } }, + payment: { snapshot: { every: 50, verify: 0, version: 2 }, commands: { timeout: 1 } }, + customer: {}, + }, + }, + }); + expect(resolved.forAggregate("order").snapshot).toEqual({ + every: 200, + verify: 0.01, + version: 1, + }); + expect(resolved.forAggregate("payment").snapshot).toEqual({ + every: 50, + verify: 0, + version: 2, + }); + expect(resolved.forAggregate("customer").snapshot).toBeNull(); + expect(resolved.forAggregate("ledger").snapshot).toBeNull(); + }); + + it("rejects malformed snapshot settings with their path", () => { + const snapshot = (value: unknown) => + message({ storage: sqlite, runtime: { overrides: { order: { snapshot: value } } } }); + expect(snapshot({})).toContain("runtime.overrides.order.snapshot.every:"); + expect(snapshot({ every: 0 })).toContain("runtime.overrides.order.snapshot.every:"); + expect(snapshot({ every: 1.5 })).toContain("runtime.overrides.order.snapshot.every:"); + expect(snapshot({ every: 10, verify: 2 })).toContain( + "runtime.overrides.order.snapshot.verify:", + ); + expect(snapshot({ every: 10, verify: -1 })).toContain( + "runtime.overrides.order.snapshot.verify:", + ); + expect(snapshot({ every: 10, version: 0 })).toContain( + "runtime.overrides.order.snapshot.version:", + ); + expect(snapshot({ every: 10, sample: 1 })).toMatch( + /runtime\.overrides\.order\.snapshot: .*sample/, + ); + expect(() => + resolveConfig({ + storage: sqlite, + runtime: { overrides: { order: { snapshot: { every: 10, verify: 1, version: 1 } } } }, + }), + ).not.toThrow(); + }); + it("rejects a storage that is not an adapter definition", () => { expect(message({ storage: { path: "./x.db" } })).toContain( "storage: Expected an adapter definition such as sqlite({ ... })", diff --git a/packages/core/src/config/schema.ts b/packages/core/src/config/schema.ts index 9245441d..a49ab3e3 100644 --- a/packages/core/src/config/schema.ts +++ b/packages/core/src/config/schema.ts @@ -15,6 +15,8 @@ import { DEFAULT_POLL_INTERVAL_MS, DEFAULT_PROCESSES, DEFAULT_PROJECTION_BATCH_TIME_MS, + DEFAULT_SNAPSHOT_VERIFY, + DEFAULT_SNAPSHOT_VERSION, } from "./defaults.ts"; import type { Config, @@ -24,6 +26,7 @@ import type { ResolvedPoliciesConfig, ResolvedProcessesConfig, ResolvedRetryConfig, + ResolvedSnapshotConfig, } from "./types.ts"; const duration = z.union([z.number(), z.string()]).transform((value, context) => { @@ -66,10 +69,17 @@ const processes = z.strictObject({ handlerTimeout: timeout.optional(), }); +const snapshot = z.strictObject({ + every: z.int().min(1), + verify: z.number().min(0).max(1).optional(), + version: z.int().min(1).optional(), +}); + const overrides = z.strictObject({ commands: commands.optional(), policies: policies.optional(), processes: processes.optional(), + snapshot: snapshot.optional(), }); const runtime = z.strictObject({ @@ -107,6 +117,7 @@ type ParsedRetry = z.output; type ParsedCommands = z.output; type ParsedPolicies = z.output; type ParsedProcesses = z.output; +type ParsedSnapshot = z.output; const resolveRetry = ( parsed: ParsedRetry | undefined, @@ -146,6 +157,15 @@ const resolveProcesses = ( handlerTimeoutMs: parsed?.handlerTimeout ?? base.handlerTimeoutMs, }); +const resolveSnapshot = (parsed: ParsedSnapshot | undefined): ResolvedSnapshotConfig | null => + parsed === undefined + ? null + : { + every: parsed.every, + verify: parsed.verify ?? DEFAULT_SNAPSHOT_VERIFY, + version: parsed.version ?? DEFAULT_SNAPSHOT_VERSION, + }; + const formatIssues = (issues: readonly z.core.$ZodIssue[]): string => issues.map((issue) => ` ${issue.path.join(".") || "(root)"}: ${issue.message}`).join("\n"); @@ -166,6 +186,7 @@ export const resolveConfig: ResolveConfigFunction = (config) => { commands: baseCommands, policies: basePolicies, processes: baseProcesses, + snapshot: null, }; const resolvedOverrides = Object.fromEntries( Object.entries(parsed.runtime?.overrides ?? {}).map(([aggregate, override]) => [ @@ -174,6 +195,7 @@ export const resolveConfig: ResolveConfigFunction = (config) => { commands: resolveCommands(override.commands, baseCommands), policies: resolvePolicies(override.policies, basePolicies), processes: resolveProcesses(override.processes, baseProcesses), + snapshot: resolveSnapshot(override.snapshot), } satisfies ResolvedAggregateRuntime, ]), ); diff --git a/packages/core/src/config/types.ts b/packages/core/src/config/types.ts index 3107f57b..2a9d0dfa 100644 --- a/packages/core/src/config/types.ts +++ b/packages/core/src/config/types.ts @@ -134,6 +134,31 @@ export interface CatchUpConfig { readonly pollInterval?: DurationInput; } +/** + * Snapshots of one aggregate's state: a cache of the fold a subscriber writes in the background, + * which a command's load starts from instead of the stream's first event. Never the source of + * truth: one that is missing, from other code or doubtful is ignored. + */ +export interface SnapshotConfig { + /** + * How many events a stream gathers past its last snapshot before the next one is written. + */ + readonly every: number; + /** + * The share of the loads from a snapshot, 0 to 1, checked against a fold of the whole stream. + * A snapshot that folds to a different state is logged, dropped and not used. Defaults to + * `DEFAULT_SNAPSHOT_VERIFY`; an app from `createTestApp` checks every load and throws instead. + */ + readonly verify?: number; + /** + * Part of what a snapshot is valid for, next to the fingerprint the generator computes from the + * modules that shape the state. Raise it to retire every snapshot of the aggregate when the + * fold changed through something the fingerprint cannot see, such as a dependency. Defaults to + * `DEFAULT_SNAPSHOT_VERSION`. + */ + readonly version?: number; +} + /** * Settings one aggregate may override. */ @@ -141,6 +166,10 @@ export interface AggregateOverrides { readonly commands?: AggregateCommandsConfig; readonly policies?: PoliciesConfig; readonly processes?: ProcessesConfig; + /** + * Opts the aggregate into snapshots; without it none are taken. + */ + readonly snapshot?: SnapshotConfig; } /** @@ -211,10 +240,20 @@ export interface ResolvedCommandsConfig { readonly timeoutMs: number; } +export interface ResolvedSnapshotConfig { + readonly every: number; + readonly verify: number; + readonly version: number; +} + export interface ResolvedAggregateRuntime { readonly commands: ResolvedCommandsConfig; readonly policies: ResolvedPoliciesConfig; readonly processes: ResolvedProcessesConfig; + /** + * `null` for an aggregate that takes no snapshots. + */ + readonly snapshot: ResolvedSnapshotConfig | null; } /** diff --git a/packages/core/src/contracts/errors.ts b/packages/core/src/contracts/errors.ts index 2c25e9d7..f7efaa81 100644 --- a/packages/core/src/contracts/errors.ts +++ b/packages/core/src/contracts/errors.ts @@ -220,6 +220,17 @@ export class ChainDepthExceededError extends BoundaError { } } +/** + * Thrown in an app from `createTestApp` when a snapshot does not stand for its stream: the state + * folded from it differs from the one folded from the first event, or the state cannot be + * serialized. Any other app logs it, drops the snapshot and folds the stream from the start. + */ +export class SnapshotError extends BoundaError { + constructor(message: string, options?: ErrorOptions) { + super("SNAPSHOT_FAILED", message, options); + } +} + /** * Thrown by a rebuild that another rebuild of the same read model has taken over. Only the most * recent rebuild writes: the older one stops at its next step and leaves its work to the newer. diff --git a/packages/core/src/contracts/index.ts b/packages/core/src/contracts/index.ts index 6b4bb703..ab00a390 100644 --- a/packages/core/src/contracts/index.ts +++ b/packages/core/src/contracts/index.ts @@ -36,6 +36,7 @@ export { RebuildSupersededError, type Rejection, ScheduledClaimLostError, + SnapshotError, ValidationError, type ValidationIssue, } from "./errors.ts"; diff --git a/packages/core/src/kernel/aggregate/build-aggregates.test.ts b/packages/core/src/kernel/aggregate/build-aggregates.test.ts index 192f620c..0351ea12 100644 --- a/packages/core/src/kernel/aggregate/build-aggregates.test.ts +++ b/packages/core/src/kernel/aggregate/build-aggregates.test.ts @@ -4,7 +4,7 @@ import { ConfigurationError } from "../../contracts/errors.ts"; import { memory } from "../../memory/index.ts"; import type { PayloadArgs } from "../../modules/payload.ts"; import type { Registry } from "../../modules/registry.ts"; -import { choosePorts, orderRegistry } from "../test-support.ts"; +import { choosePorts, orderAggregate, orderRegistry } from "../test-support.ts"; import { buildAggregates } from "./build-aggregates.ts"; import { foldState } from "./fold-state.ts"; @@ -67,6 +67,39 @@ describe("buildAggregates", () => { expect(byName.customer?.initialState).toEqual({}); }); + it("takes the snapshot settings of the configuration, valid for the registry's fingerprint and the version", () => { + const registry: Registry = { + aggregates: { order: { ...orderAggregate, stateFingerprint: "abc" } }, + readModels: {}, + }; + const config = resolveConfig({ + storage: memory(), + ports: { order: { notifier: "silent" } }, + runtime: { overrides: { order: { snapshot: { every: 50, verify: 0.5, version: 3 } } } }, + }); + expect(buildAggregates({ registry, ports, config }).byName.order?.snapshot).toEqual({ + every: 50, + verify: 0.5, + fingerprint: "abc:3", + }); + expect(buildAggregates({ registry, ports }).byName.order?.snapshot).toBeNull(); + const without = resolveConfig({ storage: memory(), ports: { order: { notifier: "silent" } } }); + expect(buildAggregates({ registry, ports, config: without }).byName.order?.snapshot).toBeNull(); + }); + + it("refuses snapshots for an aggregate whose registry has no state fingerprint", () => { + const config = resolveConfig({ + storage: memory(), + ports: { order: { notifier: "silent" } }, + runtime: { overrides: { order: { snapshot: { every: 50 } } } }, + }); + expect(() => buildAggregates({ registry: orderRegistry, ports, config })).toThrow( + new ConfigurationError( + 'runtime.overrides.order.snapshot: the registry has no state fingerprint for "order"; run bounda generate', + ), + ); + }); + it("rejects payload functions that do not return a schema", () => { const registry: Registry = { aggregates: { diff --git a/packages/core/src/kernel/aggregate/build-aggregates.ts b/packages/core/src/kernel/aggregate/build-aggregates.ts index 742673ba..46ac07bd 100644 --- a/packages/core/src/kernel/aggregate/build-aggregates.ts +++ b/packages/core/src/kernel/aggregate/build-aggregates.ts @@ -1,4 +1,5 @@ import { z } from "zod"; +import type { ResolvedConfig } from "../../config/types.ts"; import { ConfigurationError } from "../../contracts/errors.ts"; import { createEventBuilders } from "../../modules/event.ts"; import { capitalize } from "../../modules/naming.ts"; @@ -7,6 +8,7 @@ import type { AggregateEntry, Registry } from "../../modules/registry.ts"; import type { ModulePorts } from "../ports/ports.ts"; import type { AggregateRuntime, + AggregateSnapshotRuntime, AggregatesRuntime, CommandRuntime, EventRuntime, @@ -54,10 +56,30 @@ const buildCommands = (name: string, entry: AggregateEntry): Record { + const settings = config?.forAggregate(name).snapshot ?? null; + if (settings === null) return null; + if (entry.stateFingerprint === undefined) { + throw new ConfigurationError( + `runtime.overrides.${name}.snapshot: the registry has no state fingerprint for "${name}"; run bounda generate`, + ); + } + return { + every: settings.every, + verify: settings.verify, + fingerprint: `${entry.stateFingerprint}:${settings.version}`, + }; +}; + const buildAggregate = ( name: string, entry: AggregateEntry, ports: Readonly>, + config: ResolvedConfig | undefined, ): AggregateRuntime => { const events = buildEvents(name, entry); return { @@ -65,6 +87,7 @@ const buildAggregate = ( aggregateIdField: entry.state?.aggregateId ?? `${name}Id`, initialState: entry.state?.initialState ?? {}, opensWithBegin: Object.values(events).some((event) => event.begin !== null), + snapshot: buildSnapshot(name, entry, config), events, eventsByType: Object.fromEntries(Object.values(events).map((event) => [event.type, event])), eventBuilders: createEventBuilders(entry.events) as AggregateRuntime["eventBuilders"], @@ -79,6 +102,10 @@ export interface BuildAggregatesArgs { * What `createPorts` built; an aggregate missing here gets no ports. */ readonly ports: ModulePorts; + /** + * Where the snapshot settings come from; left out, no aggregate takes snapshots. + */ + readonly config?: ResolvedConfig; } export interface BuildAggregatesFunction { @@ -88,11 +115,11 @@ export interface BuildAggregatesFunction { /** * Command type names must be unique across aggregates, since `app.commands` is one flat namespace. */ -export const buildAggregates: BuildAggregatesFunction = ({ registry, ports }) => { +export const buildAggregates: BuildAggregatesFunction = ({ registry, ports, config }) => { const byName = Object.fromEntries( Object.entries(registry.aggregates).map(([name, entry]) => [ name, - buildAggregate(name, entry, ports[name] ?? {}), + buildAggregate(name, entry, ports[name] ?? {}, config), ]), ); const commandsByType: Record = {}; diff --git a/packages/core/src/kernel/aggregate/fold-state.test.ts b/packages/core/src/kernel/aggregate/fold-state.test.ts new file mode 100644 index 00000000..e0db32a0 --- /dev/null +++ b/packages/core/src/kernel/aggregate/fold-state.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; +import { pendingEvent } from "../../adapter/testing/fixtures.ts"; +import { resolveConfig } from "../../config/schema.ts"; +import { memory } from "../../memory/index.ts"; +import { choosePorts, orderRegistry } from "../test-support.ts"; +import { buildAggregates } from "./build-aggregates.ts"; +import { foldState } from "./fold-state.ts"; + +const ports = choosePorts( + orderRegistry, + resolveConfig({ storage: memory(), ports: { order: { notifier: "silent" } } }), +); +const aggregate = buildAggregates({ registry: orderRegistry, ports }).byName.order; +if (aggregate === undefined) throw new Error("no order aggregate"); + +const stored = (version: number, type: string, payload: unknown) => ({ + ...pendingEvent({ aggregateId: "1", version, type, payload }), + position: version, +}); + +describe("foldState from a snapshot", () => { + it("folds the events after it over its state, keeping how the aggregate opened", () => { + const from = { + state: { status: "placed", total: 5, extra: "kept" }, + created: true, + openedWithout: "OrderPlaced", + }; + const folded = foldState({ + aggregate, + events: [stored(3, "OrderPaid", { method: "card" })], + from, + }); + expect(folded).toEqual({ + state: { status: "paid", total: 5, extra: "kept" }, + created: true, + openedWithout: "OrderPlaced", + }); + expect(from.state).toEqual({ status: "placed", total: 5, extra: "kept" }); + }); + + it("is the snapshot itself when nothing came after it", () => { + const from = { state: { status: "placed", total: 5 }, created: true, openedWithout: null }; + expect(foldState({ aggregate, events: [], from })).toEqual(from); + }); +}); diff --git a/packages/core/src/kernel/aggregate/fold-state.ts b/packages/core/src/kernel/aggregate/fold-state.ts index b85108c9..0b8bdbc0 100644 --- a/packages/core/src/kernel/aggregate/fold-state.ts +++ b/packages/core/src/kernel/aggregate/fold-state.ts @@ -6,6 +6,11 @@ import type { AggregateRuntime, EventRuntime } from "./runtime.ts"; export interface FoldStateArgs { readonly aggregate: AggregateRuntime; readonly events: readonly StoredEvent[]; + /** + * Where to fold from, a snapshot's state: `events` are then the ones after it. Left out, the + * fold starts from the aggregate's initial state. + */ + readonly from?: FoldedState; } export interface FoldedState { @@ -51,10 +56,10 @@ const merged = ( * `evolve` after) gets the one it has, since the pipeline only stores such a stream if it predates * the event's `begin`. */ -export const foldState: FoldStateFunction = ({ aggregate, events }) => { - let state = aggregate.initialState; - let created = false; - let openedWithout: string | null = null; +export const foldState: FoldStateFunction = ({ aggregate, events, from }) => { + let state = from?.state ?? aggregate.initialState; + let created = from?.created ?? false; + let openedWithout: string | null = from?.openedWithout ?? null; for (const event of events) { if (event.metadata.system) continue; const runtime = runtimeOf(aggregate, event); diff --git a/packages/core/src/kernel/aggregate/load-aggregate.test.ts b/packages/core/src/kernel/aggregate/load-aggregate.test.ts new file mode 100644 index 00000000..fd12fbb4 --- /dev/null +++ b/packages/core/src/kernel/aggregate/load-aggregate.test.ts @@ -0,0 +1,266 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { EventStore } from "../../adapter/storage/event-store.ts"; +import type { Snapshot } from "../../adapter/storage/snapshot-store.ts"; +import { pendingEvent } from "../../adapter/testing/fixtures.ts"; +import { resolveConfig } from "../../config/schema.ts"; +import { SnapshotError } from "../../contracts/errors.ts"; +import { memory } from "../../memory/index.ts"; +import type { Registry } from "../../modules/registry.ts"; +import { ATTRIBUTES, METRICS } from "../telemetry.ts"; +import { type FakeTelemetry, installFakeTelemetry } from "../telemetry-fake.ts"; +import { choosePorts, createRecordingLogger, orderAggregate } from "../test-support.ts"; +import { buildAggregates } from "./build-aggregates.ts"; +import { createAggregateLoader } from "./load-aggregate.ts"; +import { serializeFold } from "./snapshot-codec.ts"; + +let telemetry: FakeTelemetry | undefined; + +afterEach(() => { + telemetry?.restore(); + telemetry = undefined; +}); + +interface SetUpArgs { + readonly every?: number; + readonly verify?: number; + readonly strict?: boolean; + readonly snapshots?: boolean; +} + +const stream = { aggregateType: "order", aggregateId: "1" }; +const paid = { status: "paid", total: 10 }; + +const setUp = async ({ + every = 2, + verify = 0, + strict = false, + snapshots = true, +}: SetUpArgs = {}) => { + telemetry = installFakeTelemetry(); + const evolves = { + placed: vi.fn(orderAggregate.events.orderPlaced.evolve), + paid: vi.fn(orderAggregate.events.orderPaid.evolve), + archived: vi.fn(orderAggregate.events.orderArchived.evolve), + }; + const registry: Registry = { + aggregates: { + order: { + ...orderAggregate, + stateFingerprint: "fp", + events: { + orderPlaced: { ...orderAggregate.events.orderPlaced, evolve: evolves.placed }, + orderPaid: { ...orderAggregate.events.orderPaid, evolve: evolves.paid }, + orderArchived: { evolve: evolves.archived }, + }, + }, + }, + readModels: {}, + }; + const adapter = memory(); + const config = resolveConfig({ + storage: adapter, + ports: { order: { notifier: "silent" } }, + runtime: snapshots ? { overrides: { order: { snapshot: { every, verify } } } } : {}, + }); + const aggregate = buildAggregates({ registry, ports: choosePorts(registry, config), config }) + .byName.order; + if (aggregate === undefined) throw new Error("no order aggregate"); + const storage = await adapter.createStorage({ logger: { ...createRecordingLogger().logger } }); + const snapshotStore = { + get: vi.fn(storage.snapshotStore.get), + put: vi.fn(storage.snapshotStore.put), + remove: vi.fn(storage.snapshotStore.remove), + }; + const { logger, entries } = createRecordingLogger(); + const loader = createAggregateLoader({ snapshotStore, logger, strict }); + const append = async (...events: readonly (readonly [number, string, unknown])[]) => { + await storage.eventStore.append({ + ...stream, + expectedVersion: (events[0]?.[0] ?? 1) - 1, + events: events.map(([version, type, payload]) => + pendingEvent({ aggregateId: "1", version, type, payload }), + ), + }); + }; + await append([1, "OrderPlaced", { total: 10 }], [2, "OrderPaid", { method: "card" }]); + const snapshot = (overrides: Partial = {}): Snapshot => ({ + ...stream, + version: 2, + fingerprint: "fp:1", + state: serializeFold({ state: paid, created: true, openedWithout: null }), + ...overrides, + }); + const load = (eventStore: EventStore = storage.eventStore) => + loader.load({ aggregate, aggregateId: "1", eventStore }); + const outcomes = () => + (telemetry?.counts ?? []) + .filter((count) => count.metric === METRICS.snapshots) + .map((count) => count.attributes[ATTRIBUTES.outcome]); + return { storage, snapshotStore, evolves, entries, append, snapshot, load, outcomes }; +}; + +describe("createAggregateLoader", () => { + it("folds an aggregate without snapshots from its first event, asking no store", async () => { + const { snapshotStore, load, outcomes } = await setUp({ snapshots: false }); + expect(await load()).toEqual({ + folded: { state: paid, created: true, openedWithout: null }, + version: 2, + }); + expect(snapshotStore.get).not.toHaveBeenCalled(); + expect(outcomes()).toEqual([]); + }); + + it("folds from the first event while a stream has no snapshot", async () => { + const { load, outcomes, evolves } = await setUp(); + expect(await load()).toMatchObject({ folded: { state: paid }, version: 2 }); + expect(evolves.placed).toHaveBeenCalledTimes(1); + expect(outcomes()).toEqual(["missing"]); + expect(telemetry?.counts[0]?.attributes).toEqual({ + [ATTRIBUTES.aggregateType]: "order", + [ATTRIBUTES.outcome]: "missing", + }); + expect(telemetry?.instruments).toContainEqual({ + metric: METRICS.snapshots, + description: "Loads of an aggregate with snapshots, by what its snapshot was worth", + unit: "{load}", + }); + }); + + it("loads from the snapshot only the events after it", async () => { + const { storage, snapshot, append, load, outcomes, evolves } = await setUp(); + await storage.snapshotStore.put(snapshot()); + await append([3, "OrderArchived", {}]); + expect(await load()).toEqual({ + folded: { state: paid, created: true, openedWithout: null }, + version: 3, + }); + expect(evolves.placed).not.toHaveBeenCalled(); + expect(evolves.paid).not.toHaveBeenCalled(); + expect(evolves.archived).toHaveBeenCalledTimes(1); + expect(outcomes()).toEqual(["used"]); + }); + + it("ignores, but keeps, a snapshot taken by other code", async () => { + const { storage, snapshot, load, outcomes, evolves } = await setUp(); + await storage.snapshotStore.put(snapshot({ fingerprint: "fp:2" })); + expect(await load()).toMatchObject({ folded: { state: paid }, version: 2 }); + expect(evolves.placed).toHaveBeenCalledTimes(1); + expect(outcomes()).toEqual(["stale"]); + expect(await storage.snapshotStore.get(stream)).toEqual(snapshot({ fingerprint: "fp:2" })); + }); + + it("drops a snapshot it cannot read and folds from the first event", async () => { + const { storage, snapshot, load, outcomes, entries } = await setUp(); + await storage.snapshotStore.put(snapshot({ state: "[1]" })); + expect(await load()).toMatchObject({ folded: { state: paid }, version: 2 }); + expect(outcomes()).toEqual(["unreadable"]); + expect(await storage.snapshotStore.get(stream)).toBeNull(); + expect(entries).toEqual([ + { + level: "warn", + message: "snapshot could not be read; the stream is folded from its first event", + fields: expect.objectContaining({ + ...stream, + version: 2, + message: "not a serialized fold", + }), + }, + ]); + }); + + it("drops a snapshot past the stream's head and folds from the first event", async () => { + const { storage, snapshot, load, outcomes, entries } = await setUp(); + await storage.snapshotStore.put(snapshot({ version: 5 })); + expect(await load()).toMatchObject({ folded: { state: paid }, version: 2 }); + expect(outcomes()).toEqual(["ahead"]); + expect(await storage.snapshotStore.get(stream)).toBeNull(); + expect(entries).toEqual([ + { + level: "warn", + message: "snapshot is past the stream's head; the stream is folded from its first event", + fields: { ...stream, version: 5, head: 2 }, + }, + ]); + }); + + it("checks the share of loads `verify` asks for against the whole stream", async () => { + const { storage, snapshot, load, outcomes } = await setUp({ verify: 0.5 }); + await storage.snapshotStore.put(snapshot()); + for (let turn = 0; turn < 4; turn += 1) { + expect(await load()).toMatchObject({ folded: { state: paid }, version: 2 }); + } + expect(outcomes()).toEqual(["used", "verified", "used", "verified"]); + }); + + it("checks every load in strict mode", async () => { + const { storage, snapshot, load, outcomes } = await setUp({ strict: true }); + await storage.snapshotStore.put(snapshot()); + await load(); + await load(); + expect(outcomes()).toEqual(["verified", "verified"]); + }); + + it("checks against the stream as it was loaded, not as it moved since", async () => { + const { storage, snapshot, append, load, outcomes } = await setUp({ verify: 1 }); + await storage.snapshotStore.put(snapshot()); + let moved = false; + const moving: EventStore = { + ...storage.eventStore, + load: async (args) => { + if (args.fromVersion === undefined && !moved) { + moved = true; + await append([3, "OrderPlaced", { total: 99 }]); + } + return storage.eventStore.load(args); + }, + }; + expect(await load(moving)).toEqual({ + folded: { state: paid, created: true, openedWithout: null }, + version: 2, + }); + expect(outcomes()).toEqual(["verified"]); + }); + + it("drops a snapshot that folds to another state and answers with the stream's", async () => { + const { storage, snapshot, append, load, outcomes, entries, evolves } = await setUp({ + verify: 1, + }); + await storage.snapshotStore.put( + snapshot({ + state: serializeFold({ state: { ...paid, total: 99 }, created: true, openedWithout: null }), + }), + ); + await append([3, "OrderArchived", {}]); + expect(await load()).toEqual({ + folded: { state: paid, created: true, openedWithout: null }, + version: 3, + }); + expect(evolves.placed).toHaveBeenCalledTimes(1); + expect(outcomes()).toEqual(["mismatch"]); + expect(await storage.snapshotStore.get(stream)).toBeNull(); + expect(entries).toEqual([ + { + level: "error", + message: "snapshot folds to a state that differs from the stream; dropped", + fields: { ...stream, version: 2, head: 3 }, + }, + ]); + }); + + it("throws on a snapshot that folds to another state in strict mode", async () => { + const { storage, snapshot, load, outcomes, entries } = await setUp({ strict: true }); + await storage.snapshotStore.put( + snapshot({ + state: serializeFold({ state: paid, created: true, openedWithout: "OrderPlaced" }), + }), + ); + await expect(load()).rejects.toThrow( + new SnapshotError( + "Snapshot of order:1 at version 2 folds to a state at version 2 that differs from the stream's; the fold is not a function of the events alone", + ), + ); + expect(outcomes()).toEqual(["mismatch"]); + expect(entries).toEqual([]); + expect(await storage.snapshotStore.get(stream)).not.toBeNull(); + }); +}); diff --git a/packages/core/src/kernel/aggregate/load-aggregate.ts b/packages/core/src/kernel/aggregate/load-aggregate.ts new file mode 100644 index 00000000..b2d0b83f --- /dev/null +++ b/packages/core/src/kernel/aggregate/load-aggregate.ts @@ -0,0 +1,177 @@ +import type { EventStore } from "../../adapter/storage/event-store.ts"; +import type { SnapshotStore } from "../../adapter/storage/snapshot-store.ts"; +import { SnapshotError } from "../../contracts/errors.ts"; +import type { StreamIdentity } from "../../contracts/event.ts"; +import type { Logger } from "../../contracts/logger.ts"; +import { errorDetails } from "../shared/retry.ts"; +import { ATTRIBUTES, METRICS, meter } from "../telemetry.ts"; +import { type FoldedState, foldState } from "./fold-state.ts"; +import type { AggregateRuntime } from "./runtime.ts"; +import { deserializeFold, foldsEqual } from "./snapshot-codec.ts"; + +export interface LoadAggregateArgs { + readonly aggregate: AggregateRuntime; + readonly aggregateId: string; + /** + * The store to read from: the storage's, or a unit of work's with the events it staged. + */ + readonly eventStore: EventStore; +} + +export interface LoadedAggregate { + readonly folded: FoldedState; + /** + * The stream's head, system events included: what an append expects. + */ + readonly version: number; +} + +/** + * Loads an aggregate's state: from its snapshot when it takes them and the snapshot is valid, + * from its first event otherwise. The loads of one aggregate share the verification sample. + */ +export interface AggregateLoader { + load(args: LoadAggregateArgs): Promise; +} + +export interface CreateAggregateLoaderArgs { + readonly snapshotStore: SnapshotStore; + readonly logger: Logger; + /** + * For an app from `createTestApp`: every load from a snapshot is checked, and one that folds + * to a different state throws `SnapshotError` instead of being dropped. + */ + readonly strict: boolean; +} + +export interface CreateAggregateLoaderFunction { + (args: CreateAggregateLoaderArgs): AggregateLoader; +} + +/** + * What a snapshot was worth to a load, as `bounda.snapshots` counts them. `used` and `verified` + * both loaded from it; only the second compared the result with the whole stream. + */ +export type SnapshotOutcome = + | "missing" + | "stale" + | "unreadable" + | "ahead" + | "used" + | "verified" + | "mismatch"; + +/** + * A snapshot is used only when its fingerprint is the aggregate's, it reads back and the stream + * reaches its version; otherwise the stream is folded from its first event, and one that read + * badly or is past the stream's head is dropped. A verified load folds the whole stream up to + * the version it loaded, so an append landing in between does not count as a difference. + */ +export const createAggregateLoader: CreateAggregateLoaderFunction = ({ + snapshotStore, + logger, + strict, +}) => { + const counter = meter().createCounter(METRICS.snapshots, { + description: "Loads of an aggregate with snapshots, by what its snapshot was worth", + unit: "{load}", + }); + // Credit per aggregate, so the share `verify` asks for is met exactly, not on average. + const credits = new Map(); + const due = (aggregate: AggregateRuntime, verify: number): boolean => { + if (strict) return true; + const credit = (credits.get(aggregate.name) ?? 0) + verify; + credits.set(aggregate.name, credit >= 1 ? credit - 1 : credit); + return credit >= 1; + }; + const count = (aggregate: AggregateRuntime, outcome: SnapshotOutcome): void => { + counter.add(1, { [ATTRIBUTES.aggregateType]: aggregate.name, [ATTRIBUTES.outcome]: outcome }); + }; + const fromStart = async ( + aggregate: AggregateRuntime, + stream: StreamIdentity, + eventStore: EventStore, + through = Number.POSITIVE_INFINITY, + ): Promise => { + const loaded = await eventStore.load(stream); + const events = loaded.events.filter((event) => event.version <= through); + return { + folded: foldState({ aggregate, events }), + version: Math.min(loaded.version, through), + }; + }; + const dropped = async ( + aggregate: AggregateRuntime, + stream: StreamIdentity, + eventStore: EventStore, + outcome: SnapshotOutcome, + message: string, + fields: Readonly>, + ): Promise => { + count(aggregate, outcome); + logger.warn(message, { ...stream, ...fields }); + await snapshotStore.remove(stream); + return fromStart(aggregate, stream, eventStore); + }; + + return { + load: async ({ aggregate, aggregateId, eventStore }) => { + const stream = { aggregateType: aggregate.name, aggregateId }; + const settings = aggregate.snapshot; + if (settings === null) return fromStart(aggregate, stream, eventStore); + const snapshot = await snapshotStore.get(stream); + if (snapshot === null) { + count(aggregate, "missing"); + return fromStart(aggregate, stream, eventStore); + } + if (snapshot.fingerprint !== settings.fingerprint) { + count(aggregate, "stale"); + return fromStart(aggregate, stream, eventStore); + } + let from: FoldedState; + try { + from = deserializeFold(snapshot.state); + } catch (error) { + return dropped( + aggregate, + stream, + eventStore, + "unreadable", + "snapshot could not be read; the stream is folded from its first event", + { version: snapshot.version, ...errorDetails(error) }, + ); + } + const loaded = await eventStore.load({ ...stream, fromVersion: snapshot.version + 1 }); + if (loaded.version < snapshot.version) { + return dropped( + aggregate, + stream, + eventStore, + "ahead", + "snapshot is past the stream's head; the stream is folded from its first event", + { version: snapshot.version, head: loaded.version }, + ); + } + const folded = foldState({ aggregate, events: loaded.events, from }); + if (!due(aggregate, settings.verify)) { + count(aggregate, "used"); + return { folded, version: loaded.version }; + } + const whole = await fromStart(aggregate, stream, eventStore, loaded.version); + if (foldsEqual(folded, whole.folded)) { + count(aggregate, "verified"); + return { folded, version: loaded.version }; + } + count(aggregate, "mismatch"); + const message = `Snapshot of ${aggregate.name}:${aggregateId} at version ${snapshot.version} folds to a state at version ${loaded.version} that differs from the stream's; the fold is not a function of the events alone`; + if (strict) throw new SnapshotError(message); + logger.error("snapshot folds to a state that differs from the stream; dropped", { + ...stream, + version: snapshot.version, + head: loaded.version, + }); + await snapshotStore.remove(stream); + return whole; + }, + }; +}; diff --git a/packages/core/src/kernel/aggregate/runtime.ts b/packages/core/src/kernel/aggregate/runtime.ts index 44c1adb1..445c288c 100644 --- a/packages/core/src/kernel/aggregate/runtime.ts +++ b/packages/core/src/kernel/aggregate/runtime.ts @@ -30,11 +30,22 @@ export interface CommandRuntime { | null; } +/** + * The snapshot settings of an aggregate that takes them, with the fingerprint its snapshots are + * valid for: the registry's state fingerprint and the configured version. + */ +export interface AggregateSnapshotRuntime { + readonly every: number; + readonly verify: number; + readonly fingerprint: string; +} + export interface AggregateRuntime { readonly name: string; readonly aggregateIdField: string; readonly initialState: object; readonly opensWithBegin: boolean; + readonly snapshot: AggregateSnapshotRuntime | null; /** * The chosen implementation of every port, spread into the arguments of every handler of the * aggregate: its commands, policies and processes. diff --git a/packages/core/src/kernel/aggregate/snapshot-codec.test.ts b/packages/core/src/kernel/aggregate/snapshot-codec.test.ts new file mode 100644 index 00000000..8af1e472 --- /dev/null +++ b/packages/core/src/kernel/aggregate/snapshot-codec.test.ts @@ -0,0 +1,160 @@ +import { describe, expect, it } from "vitest"; +import type { FoldedState } from "./fold-state.ts"; +import { deserializeFold, foldsEqual, serializeFold } from "./snapshot-codec.ts"; + +const fold = (state: object, rest: Partial = {}): FoldedState => ({ + state, + created: true, + openedWithout: null, + ...rest, +}); + +describe("serializeFold and deserializeFold", () => { + it("keeps what JSON loses, and the fold's flags", () => { + const folded = fold( + { + placedAt: new Date("2026-03-01T10:00:00.000Z"), + lines: new Map([["sku-1", { quantity: 2 }]]), + tags: new Set(["gift", "fragile"]), + total: 12n, + note: undefined, + gaps: [1, undefined, Number.NaN, Number.POSITIVE_INFINITY], + pattern: /^ord-/u, + }, + { created: false, openedWithout: "OrderPlaced" }, + ); + const restored = deserializeFold(serializeFold(folded)); + expect(restored).toEqual(folded); + expect(restored.state).not.toBe(folded.state); + expect(Reflect.get(restored.state, "placedAt")).toBeInstanceOf(Date); + expect(Reflect.get(restored.state, "lines")).toBeInstanceOf(Map); + expect(Reflect.get(restored.state, "tags")).toBeInstanceOf(Set); + expect(Object.hasOwn(restored.state, "note")).toBe(true); + }); + + it("refuses a state that cannot be kept", () => { + class Clock {} + expect(() => serializeFold(fold({ clock: new Clock() }))).toThrow( + "Cannot stringify arbitrary non-POJOs", + ); + expect(() => serializeFold(fold({ tick: () => 1 }))).toThrow(/function/); + }); + + it("refuses text that is not a serialized fold", () => { + expect(() => deserializeFold("not json")).toThrow(); + expect(() => deserializeFold("[1]")).toThrow("not a serialized fold"); + expect(() => deserializeFold(serializeFold(fold({})).replace("true", "1"))).toThrow( + "not a serialized fold", + ); + expect(() => + deserializeFold(JSON.stringify([{ state: 1, created: 2, openedWithout: 3 }, [], true, null])), + ).toThrow("not a serialized fold"); + expect(() => + deserializeFold(JSON.stringify([{ state: 1, created: 2, openedWithout: 3 }, {}, true, 4])), + ).toThrow("not a serialized fold"); + }); +}); + +describe("foldsEqual", () => { + it("compares states by structure, whatever the order of their fields", () => { + expect(foldsEqual(fold({ a: 1, b: [1, { c: 2 }] }), fold({ b: [1, { c: 2 }], a: 1 }))).toBe( + true, + ); + expect(foldsEqual(fold({ a: 1 }), fold({ a: 2 }))).toBe(false); + expect(foldsEqual(fold({ a: 1 }), fold({ a: 1, b: undefined }))).toBe(false); + expect(foldsEqual(fold({ a: 1, b: 2 }), fold({ a: 1, c: 2 }))).toBe(false); + expect(foldsEqual(fold({ a: [1, 2] }), fold({ a: { 0: 1, 1: 2 } }))).toBe(false); + expect(foldsEqual(fold({ a: null }), fold({ a: {} }))).toBe(false); + expect(foldsEqual(fold({ a: {} }), fold({ a: null }))).toBe(false); + expect(foldsEqual(fold({ a: 1 }), fold({ a: {} }))).toBe(false); + expect(foldsEqual(fold({ a: {} }), fold({ a: 1 }))).toBe(false); + expect(foldsEqual(fold({ a: Number.NaN }), fold({ a: Number.NaN }))).toBe(true); + }); + + it("tells a date, a map and a set from an object that looks like one", () => { + expect(foldsEqual(fold({ at: new Date(0) }), fold({ at: {} }))).toBe(false); + expect(foldsEqual(fold({ at: {} }), fold({ at: new Date(0) }))).toBe(false); + expect(foldsEqual(fold({ lines: { size: 0 } }), fold({ lines: new Map() }))).toBe(false); + expect(foldsEqual(fold({ lines: new Map() }), fold({ lines: { size: 0 } }))).toBe(false); + expect(foldsEqual(fold({ tags: { size: 0 } }), fold({ tags: new Set() }))).toBe(false); + expect(foldsEqual(fold({ tags: new Set() }), fold({ tags: { size: 0 } }))).toBe(false); + expect(foldsEqual(fold({ tags: new Set() }), fold({ tags: {} }))).toBe(false); + expect(foldsEqual(fold({ lines: new Map() }), fold({ lines: {} }))).toBe(false); + expect(foldsEqual(fold({ at: new Date(0) }), fold({ at: {} }))).toBe(false); + }); + + it("compares dates, maps and sets by value", () => { + const at = "2026-03-01T10:00:00.000Z"; + expect(foldsEqual(fold({ at: new Date(at) }), fold({ at: new Date(at) }))).toBe(true); + expect(foldsEqual(fold({ at: new Date(at) }), fold({ at: new Date(0) }))).toBe(false); + expect(foldsEqual(fold({ at: new Date(at) }), fold({ at }))).toBe(false); + expect( + foldsEqual( + fold({ lines: new Map([["a", { quantity: 1 }]]) }), + fold({ lines: new Map([["a", { quantity: 1 }]]) }), + ), + ).toBe(true); + expect( + foldsEqual( + fold({ lines: new Map([["a", { quantity: 1 }]]) }), + fold({ lines: new Map([["a", { quantity: 2 }]]) }), + ), + ).toBe(false); + expect( + foldsEqual(fold({ lines: new Map([["a", 1]]) }), fold({ lines: new Map([["b", 1]]) })), + ).toBe(false); + expect( + foldsEqual( + fold({ + lines: new Map([ + ["a", 1], + ["b", 2], + ]), + }), + fold({ + lines: new Map([ + ["a", 1], + ["b", 3], + ]), + }), + ), + ).toBe(false); + expect(foldsEqual(fold({ tags: new Set([1, 2]) }), fold({ tags: new Set([1, 3]) }))).toBe( + false, + ); + expect( + foldsEqual( + fold({ tags: new Set([{ id: 1 }, { id: 2 }]) }), + fold({ tags: new Set([{ id: 1 }, { id: 3 }]) }), + ), + ).toBe(false); + expect( + foldsEqual( + fold({ lines: new Map([["a", 1]]) }), + fold({ + lines: new Map([ + ["a", 1], + ["b", 2], + ]), + }), + ), + ).toBe(false); + expect(foldsEqual(fold({ lines: new Map() }), fold({ lines: {} }))).toBe(false); + expect( + foldsEqual( + fold({ tags: new Set([{ id: 1 }, "x"]) }), + fold({ tags: new Set(["x", { id: 1 }]) }), + ), + ).toBe(true); + expect( + foldsEqual(fold({ tags: new Set([{ id: 1 }]) }), fold({ tags: new Set([{ id: 2 }]) })), + ).toBe(false); + expect(foldsEqual(fold({ tags: new Set([1]) }), fold({ tags: new Set([1, 2]) }))).toBe(false); + expect(foldsEqual(fold({ tags: new Set() }), fold({ tags: [] }))).toBe(false); + }); + + it("tells folds apart by whether and how the aggregate opened", () => { + expect(foldsEqual(fold({}), fold({}, { created: false }))).toBe(false); + expect(foldsEqual(fold({}), fold({}, { openedWithout: "OrderPlaced" }))).toBe(false); + }); +}); diff --git a/packages/core/src/kernel/aggregate/snapshot-codec.ts b/packages/core/src/kernel/aggregate/snapshot-codec.ts new file mode 100644 index 00000000..fe8da324 --- /dev/null +++ b/packages/core/src/kernel/aggregate/snapshot-codec.ts @@ -0,0 +1,85 @@ +import { parse, stringify } from "devalue"; +import { isRecord } from "../shared/merge-fields.ts"; +import type { FoldedState } from "./fold-state.ts"; + +export interface SerializeFoldFunction { + (folded: FoldedState): string; +} + +/** + * Keeps what JSON loses: `Date`, `Map`, `Set`, `bigint`, `undefined` and the rest devalue + * handles. Throws for what cannot be kept, a class instance or a function. + */ +export const serializeFold: SerializeFoldFunction = (folded) => stringify(folded); + +export interface DeserializeFoldFunction { + (text: string): FoldedState; +} + +/** + * Throws for text `serializeFold` did not produce. + */ +export const deserializeFold: DeserializeFoldFunction = (text) => { + const value: unknown = parse(text); + if ( + !isRecord(value) || + !isRecord(value.state) || + typeof value.created !== "boolean" || + (value.openedWithout !== null && typeof value.openedWithout !== "string") + ) { + throw new TypeError("not a serialized fold"); + } + return { state: value.state, created: value.created, openedWithout: value.openedWithout }; +}; + +const sameEntries = (left: Map, right: Map): boolean => + left.size === right.size && + [...left].every(([key, value]) => right.has(key) && equal(value, right.get(key))); + +const sameMembers = (left: Set, right: Set): boolean => + left.size === right.size && + [...left].every( + (member) => right.has(member) || [...right].some((other) => equal(member, other)), + ); + +const sameFields = (left: object, right: object): boolean => { + const keys = Object.keys(left); + return ( + keys.length === Object.keys(right).length && + keys.every( + (key) => Object.hasOwn(right, key) && equal(Reflect.get(left, key), Reflect.get(right, key)), + ) + ); +}; + +const equal = (left: unknown, right: unknown): boolean => { + if (Object.is(left, right)) return true; + if (typeof left !== "object" || typeof right !== "object" || left === null || right === null) { + return false; + } + if (left instanceof Date || right instanceof Date) { + return ( + left instanceof Date && right instanceof Date && Object.is(left.getTime(), right.getTime()) + ); + } + if (left instanceof Map || right instanceof Map) { + return left instanceof Map && right instanceof Map && sameEntries(left, right); + } + if (left instanceof Set || right instanceof Set) { + return left instanceof Set && right instanceof Set && sameMembers(left, right); + } + if (Array.isArray(left) !== Array.isArray(right)) return false; + return sameFields(left, right); +}; + +export interface FoldsEqualFunction { + (left: FoldedState, right: FoldedState): boolean; +} + +/** + * Structural, so the order fields were set in does not count. + */ +export const foldsEqual: FoldsEqualFunction = (left, right) => + left.created === right.created && + left.openedWithout === right.openedWithout && + equal(left.state, right.state); diff --git a/packages/core/src/kernel/app.ts b/packages/core/src/kernel/app.ts index c0c6f224..ec548c7d 100644 --- a/packages/core/src/kernel/app.ts +++ b/packages/core/src/kernel/app.ts @@ -12,6 +12,7 @@ import type { CommandsFacade, QueriesFacade, Registry } from "../modules/registr import { validateRegistry } from "../modules/validate.ts"; import type { AppEnv, AppRegistry, EnvSection } from "../register/index.ts"; import { buildAggregates } from "./aggregate/build-aggregates.ts"; +import { createAggregateLoader } from "./aggregate/load-aggregate.ts"; import { withUpcasting } from "./aggregate/upcasting.ts"; import { createCommandsFacade } from "./command/facade.ts"; import { createCommandPipeline } from "./command/pipeline.ts"; @@ -38,6 +39,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 { createSnapshotSubscriber, takesSnapshots } from "./snapshot/subscriber.ts"; import { ATTRIBUTES, METRICS, meter } from "./telemetry.ts"; /** @@ -273,17 +275,23 @@ export const assembleApp: AssembleAppFunction = async ({ try { const written = await config.storage.createStorage({ logger }); opened.unshift(() => written.close()); - const aggregates = buildAggregates({ registry, ports: ports.byAggregate }); + const aggregates = buildAggregates({ registry, ports: ports.byAggregate, config }); const storage = { ...written, eventStore: withUpcasting({ eventStore: written.eventStore, aggregates }), }; + const loader = createAggregateLoader({ + snapshotStore: storage.snapshotStore, + logger, + strict: test !== undefined, + }); const readModels = await buildReadModels({ registry, config, logger }); opened.unshift(() => readModels.close()); // One list per `runUntilIdle` under way. const observers = new Set(); const pipeline = createCommandPipeline({ aggregates, + loader, eventStore: storage.eventStore, scheduler: storage.scheduler, config, @@ -346,6 +354,18 @@ export const assembleApp: AssembleAppFunction = async ({ }), ), ...following, + ...Object.values(aggregates.byName) + .filter(takesSnapshots) + .map((aggregate) => + createSnapshotSubscriber({ + aggregate, + eventStore: storage.eventStore, + snapshotStore: storage.snapshotStore, + loader, + logger, + ...(test === undefined ? {} : { strict: { onError: test.onSnapshotError } }), + }), + ), ], batchSize: config.runtime.dispatcher.batchSize, pollIntervalMs: config.runtime.dispatcher.pollIntervalMs, diff --git a/packages/core/src/kernel/command/pipeline.ts b/packages/core/src/kernel/command/pipeline.ts index fc06ad40..e12d1287 100644 --- a/packages/core/src/kernel/command/pipeline.ts +++ b/packages/core/src/kernel/command/pipeline.ts @@ -26,7 +26,7 @@ import type { NewEvent } from "../../contracts/event.ts"; import type { IdGenerator } from "../../contracts/ids.ts"; import type { Logger } from "../../contracts/logger.ts"; import type { CausationContext } from "../../contracts/metadata.ts"; -import { foldState } from "../aggregate/fold-state.ts"; +import type { AggregateLoader } from "../aggregate/load-aggregate.ts"; import type { AggregateRuntime, AggregatesRuntime, CommandRuntime } from "../aggregate/runtime.ts"; import { errorDetails } from "../shared/retry.ts"; import { withTimeout } from "../shared/timeout.ts"; @@ -78,6 +78,7 @@ export interface CommandPipeline { export interface CreateCommandPipelineArgs { readonly aggregates: AggregatesRuntime; + readonly loader: AggregateLoader; readonly eventStore: EventStore; readonly scheduler: Scheduler; readonly config: ResolvedConfig; @@ -268,6 +269,7 @@ const rejectOf = ( */ export const createCommandPipeline: CreateCommandPipelineFunction = ({ aggregates, + loader, eventStore, scheduler, config, @@ -288,11 +290,11 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ const attempts = config.runtime.commands.concurrencyRetries + 1; const { timeoutMs } = config.forAggregate(aggregate.name).commands; for (let attempt = 1; ; attempt += 1) { - const loaded = await store.load({ - aggregateType: aggregate.name, + const { folded, version } = await loader.load({ + aggregate, aggregateId: command.aggregateId, + eventStore: store, }); - const folded = foldState({ aggregate, events: loaded.events }); const stream = `${aggregate.name}:${command.aggregateId}`; if (folded.openedWithout !== null && !warnedOpenings.has(stream)) { warnedOpenings.add(stream); @@ -302,7 +304,7 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ eventType: folded.openedWithout, }); } - const state = { ...folded.state, id: command.aggregateId, version: loaded.version }; + const state = { ...folded.state, id: command.aggregateId, version }; const issued = new WeakSet(); const produced = (await withTimeout({ run: (signal) => @@ -333,7 +335,7 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ aggregate, command, produced ?? [], - loaded.version, + version, ids, clock.now().toISOString(), ); @@ -343,7 +345,7 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ scheduled: false, aggregateType: aggregate.name, aggregateId: command.aggregateId, - version: loaded.version, + version, eventIds: [], eventTypes: [], position: 0, @@ -353,7 +355,7 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ const appended = await store.append({ aggregateType: aggregate.name, aggregateId: command.aggregateId, - expectedVersion: loaded.version, + expectedVersion: version, events, }); return { diff --git a/packages/core/src/kernel/dispatch/delivery.ts b/packages/core/src/kernel/dispatch/delivery.ts index 00e93696..5ba51aa0 100644 --- a/packages/core/src/kernel/dispatch/delivery.ts +++ b/packages/core/src/kernel/dispatch/delivery.ts @@ -5,10 +5,10 @@ import { errorDetails } from "../shared/retry.ts"; import { ATTRIBUTES, traced } from "../telemetry.ts"; /** - * What a subscriber does with the events it receives: keep a read model up to date, run policies - * or drive processes. + * What a subscriber does with the events it receives: keep a read model up to date, run policies, + * drive processes or snapshot an aggregate's state. */ -export type SubscriberKind = "projection" | "policy" | "process"; +export type SubscriberKind = "projection" | "policy" | "process" | "snapshot"; /** * `process` resolves to how many leading events of the batch are done; the rest is delivered diff --git a/packages/core/src/kernel/ports/ports.ts b/packages/core/src/kernel/ports/ports.ts index c22fd6c1..f6086e6b 100644 --- a/packages/core/src/kernel/ports/ports.ts +++ b/packages/core/src/kernel/ports/ports.ts @@ -2,7 +2,7 @@ import { type PortOwner, selectImplementations } from "../../config/ports.ts"; import type { PortsConfig } from "../../config/types.ts"; import type { Clock } from "../../contracts/clock.ts"; -import { ConfigurationError } from "../../contracts/errors.ts"; +import { ConfigurationError, type SnapshotError } from "../../contracts/errors.ts"; import type { Logger } from "../../contracts/logger.ts"; import type { CreateArgs, PortModules } from "../../modules/port.ts"; import type { Registry } from "../../modules/registry.ts"; @@ -23,6 +23,11 @@ export type ModulePorts = Readonly ({ clock: new Clock() }) } }, + commands: {}, + policies: {}, + processes: {}, + }, + }, + readModels: {}, +}; + +interface SetUpArgs { + readonly aggregate?: "order" | "ticker"; + readonly strict?: boolean; +} + +const setUp = async ({ aggregate: name = "order", strict = false }: SetUpArgs = {}) => { + const adapter = memory(); + const config = resolveConfig({ + storage: adapter, + ports: { order: { notifier: "silent" } }, + runtime: { + overrides: { order: { snapshot: { every: 2 } }, ticker: { snapshot: { every: 1 } } }, + }, + }); + const aggregate = buildAggregates({ registry, ports: choosePorts(registry, config), config }) + .byName[name]; + if (aggregate === undefined || !takesSnapshots(aggregate)) throw new Error("no aggregate"); + const storage = await adapter.createStorage({ logger: createRecordingLogger().logger }); + const { logger, entries } = createRecordingLogger(); + const errors: SnapshotError[] = []; + const snapshotStore = { + ...storage.snapshotStore, + get: vi.fn(storage.snapshotStore.get), + put: vi.fn(storage.snapshotStore.put), + }; + const subscriber = createSnapshotSubscriber({ + aggregate, + eventStore: storage.eventStore, + snapshotStore, + loader: createAggregateLoader({ snapshotStore, logger, strict }), + logger, + ...(strict ? { strict: { onError: (error: SnapshotError) => errors.push(error) } } : {}), + }); + const append = async ( + aggregateType: string, + aggregateId: string, + ...events: readonly (readonly [number, string, unknown])[] + ) => { + await storage.eventStore.append({ + aggregateType, + aggregateId, + expectedVersion: (events[0]?.[0] ?? 1) - 1, + events: events.map(([version, type, payload]) => + pendingEvent({ aggregateType, aggregateId, version, type, payload }), + ), + }); + }; + const deliver = async (afterPosition = 0) => + subscriber.process(await storage.eventStore.readAll({ afterPosition, limit: 100 })); + const stored = async (aggregateId = "1") => + storage.snapshotStore.get({ aggregateType: name, aggregateId }); + return { subscriber, storage, snapshotStore, entries, errors, append, deliver, stored }; +}; + +describe("createSnapshotSubscriber", () => { + it("writes a snapshot of a stream that gathered `every` events, at the head it finds", async () => { + const { subscriber, snapshotStore, append, deliver, stored, entries } = await setUp(); + expect(subscriber).toMatchObject({ name: "snapshot:order", kind: "snapshot" }); + await append("order", "1", [1, "OrderPlaced", { total: 10 }]); + expect(await deliver()).toBe(1); + expect(await stored()).toBeNull(); + expect(snapshotStore.get).not.toHaveBeenCalled(); + await append("order", "1", [2, "OrderPaid", { method: "card" }]); + await append("order", "1", [3, "OrderArchived", {}]); + expect(await deliver(1)).toBe(2); + const snapshot = await stored(); + expect(snapshot).toMatchObject({ + aggregateType: "order", + aggregateId: "1", + version: 3, + fingerprint: "fp:1", + }); + expect(deserializeFold(snapshot?.state ?? "")).toEqual({ + state: { status: "paid", total: 10 }, + created: true, + openedWithout: null, + }); + expect(entries).toEqual([ + { + level: "debug", + message: "snapshot written", + fields: { aggregateType: "order", aggregateId: "1", version: 3 }, + }, + ]); + }); + + it("counts the events since the last snapshot of this code, per stream", async () => { + const { snapshotStore, storage, append, deliver, stored } = await setUp(); + await append( + "order", + "1", + [1, "OrderPlaced", { total: 10 }], + [2, "OrderPaid", { method: "card" }], + ); + await append("order", "2", [1, "OrderPlaced", { total: 7 }]); + await append("customer", "9", [1, "CustomerRegistered", {}], [2, "CustomerRegistered", {}]); + expect(await deliver()).toBe(5); + expect(snapshotStore.put).toHaveBeenCalledTimes(1); + expect(await stored("1")).toMatchObject({ version: 2 }); + expect(await stored("2")).toBeNull(); + expect(await stored("9")).toBeNull(); + await append("order", "1", [3, "OrderArchived", {}]); + await deliver(5); + expect(await stored("1")).toMatchObject({ version: 2 }); + await append("order", "1", [4, "OrderArchived", {}]); + await deliver(6); + expect(await stored("1")).toMatchObject({ version: 4 }); + await storage.snapshotStore.put({ + aggregateType: "order", + aggregateId: "1", + version: 4, + fingerprint: "fp:2", + state: serializeFold({ state: {}, created: true, openedWithout: null }), + }); + await append("order", "1", [5, "OrderArchived", {}]); + await deliver(7); + expect(await stored("1")).toMatchObject({ version: 5, fingerprint: "fp:1" }); + }); + + it("logs a state it cannot keep and goes on", async () => { + const { append, deliver, stored, entries } = await setUp({ aggregate: "ticker" }); + await append("ticker", "1", [1, "Ticked", {}]); + expect(await deliver()).toBe(1); + expect(await stored()).toBeNull(); + expect(entries).toEqual([ + { + level: "error", + message: "aggregate state cannot be snapshotted", + fields: expect.objectContaining({ + aggregateType: "ticker", + aggregateId: "1", + version: 1, + message: "Cannot stringify arbitrary non-POJOs", + }), + }, + ]); + }); + + it("throws on a state it cannot keep in strict mode, telling the app first", async () => { + const { append, deliver, stored, entries, errors } = await setUp({ + aggregate: "ticker", + strict: true, + }); + await append("ticker", "1", [1, "Ticked", {}]); + const error = await deliver().catch((thrown: unknown) => thrown); + expect(error).toBeInstanceOf(SnapshotError); + expect((error as SnapshotError).message).toBe( + "State of ticker:1 cannot be snapshotted: Cannot stringify arbitrary non-POJOs", + ); + expect(errors).toEqual([error]); + expect((error as SnapshotError).cause).toMatchObject({ name: "DevalueError" }); + expect(await stored()).toBeNull(); + expect(entries).toEqual([]); + }); + + it("lets any other failure of the batch through, telling the app nothing", async () => { + for (const strict of [true, false]) { + const { snapshotStore, append, deliver, errors } = await setUp({ strict }); + snapshotStore.get.mockRejectedValue(new Error("database down")); + await append( + "order", + "1", + [1, "OrderPlaced", { total: 10 }], + [2, "OrderPaid", { method: "card" }], + ); + await expect(deliver()).rejects.toThrow("database down"); + expect(errors).toEqual([]); + } + }); + + it("tells the app of a snapshot that folds to another state in strict mode", async () => { + const { storage, append, deliver, errors } = await setUp({ strict: true }); + await append( + "order", + "1", + [1, "OrderPlaced", { total: 10 }], + [2, "OrderPaid", { method: "card" }], + ); + await storage.snapshotStore.put({ + aggregateType: "order", + aggregateId: "1", + version: 2, + fingerprint: "fp:1", + state: serializeFold({ + state: { status: "paid", total: 99 }, + created: true, + openedWithout: null, + }), + }); + await append("order", "1", [3, "OrderArchived", {}], [4, "OrderArchived", {}]); + await expect(deliver(2)).rejects.toThrow(SnapshotError); + expect(errors).toHaveLength(1); + }); +}); diff --git a/packages/core/src/kernel/snapshot/subscriber.ts b/packages/core/src/kernel/snapshot/subscriber.ts new file mode 100644 index 00000000..bead4389 --- /dev/null +++ b/packages/core/src/kernel/snapshot/subscriber.ts @@ -0,0 +1,114 @@ +import type { EventStore } from "../../adapter/storage/event-store.ts"; +import type { SnapshotStore } from "../../adapter/storage/snapshot-store.ts"; +import { SnapshotError } from "../../contracts/errors.ts"; +import type { Logger } from "../../contracts/logger.ts"; +import type { AggregateLoader } from "../aggregate/load-aggregate.ts"; +import type { AggregateRuntime, AggregateSnapshotRuntime } from "../aggregate/runtime.ts"; +import { serializeFold } from "../aggregate/snapshot-codec.ts"; +import type { Subscriber } from "../dispatch/delivery.ts"; +import { errorDetails } from "../shared/retry.ts"; + +export interface SnapshottedAggregate extends AggregateRuntime { + readonly snapshot: AggregateSnapshotRuntime; +} + +export interface TakesSnapshotsFunction { + (aggregate: AggregateRuntime): aggregate is SnapshottedAggregate; +} + +export const takesSnapshots: TakesSnapshotsFunction = ( + aggregate, +): aggregate is SnapshottedAggregate => aggregate.snapshot !== null; + +export interface SnapshotSubscriberNameFunction { + (aggregate: string): string; +} + +export const snapshotSubscriberName: SnapshotSubscriberNameFunction = (aggregate) => + `snapshot:${aggregate}`; + +export interface CreateSnapshotSubscriberArgs { + readonly aggregate: SnapshottedAggregate; + readonly eventStore: EventStore; + readonly snapshotStore: SnapshotStore; + readonly loader: AggregateLoader; + readonly logger: Logger; + /** + * For an app from `createTestApp`: a state that cannot be serialized throws `SnapshotError` + * instead of being logged, and every `SnapshotError` the batch raises is reported before it + * fails the batch, since the dispatcher only logs a failed batch. + */ + readonly strict?: { + onError(error: SnapshotError): void; + }; +} + +export interface CreateSnapshotSubscriberFunction { + (args: CreateSnapshotSubscriberArgs): Subscriber; +} + +/** + * Follows the global stream for one aggregate and writes a snapshot of every stream that gathered + * `every` events since its last valid one, at the head the load finds: the write is outside any + * command and idempotent, so two instances taking the same batch do no harm. A state that cannot + * be serialized is logged and left for the next batch of that stream. + */ +export const createSnapshotSubscriber: CreateSnapshotSubscriberFunction = ({ + aggregate, + eventStore, + snapshotStore, + loader, + logger, + strict, +}) => { + const { every, fingerprint } = aggregate.snapshot; + + const snapshot = async (aggregateId: string): Promise => { + const stream = { aggregateType: aggregate.name, aggregateId }; + const { folded, version } = await loader.load({ aggregate, aggregateId, eventStore }); + let state: string; + try { + state = serializeFold(folded); + } catch (error) { + const details = errorDetails(error); + if (strict !== undefined) { + throw new SnapshotError( + `State of ${aggregate.name}:${aggregateId} cannot be snapshotted: ${details.message}`, + { cause: error }, + ); + } + logger.error("aggregate state cannot be snapshotted", { ...stream, version, ...details }); + return; + } + await snapshotStore.put({ ...stream, version, fingerprint, state }); + logger.debug("snapshot written", { ...stream, version }); + }; + + const process = async (heads: ReadonlyMap): Promise => { + for (const [aggregateId, head] of heads) { + if (head < every) continue; + const current = await snapshotStore.get({ aggregateType: aggregate.name, aggregateId }); + const since = current !== null && current.fingerprint === fingerprint ? current.version : 0; + if (head - since >= every) await snapshot(aggregateId); + } + }; + + return { + name: snapshotSubscriberName(aggregate.name), + kind: "snapshot", + process: async (events) => { + // Events of one stream come in version order, so the last one seen is its head. + const heads = new Map(); + for (const event of events) { + if (event.aggregateType === aggregate.name) heads.set(event.aggregateId, event.version); + } + try { + await process(heads); + } catch (error) { + if (error instanceof SnapshotError) strict?.onError(error); + throw error; + } + return events.length; + }, + }; +}; diff --git a/packages/core/src/kernel/telemetry.ts b/packages/core/src/kernel/telemetry.ts index 681aa641..a1252252 100644 --- a/packages/core/src/kernel/telemetry.ts +++ b/packages/core/src/kernel/telemetry.ts @@ -129,10 +129,12 @@ export const METRICS: { readonly lag: "bounda.dispatcher.lag"; readonly commands: "bounda.commands"; readonly deadLetters: "bounda.dead_letters"; + readonly snapshots: "bounda.snapshots"; } = { lag: "bounda.dispatcher.lag", commands: "bounda.commands", deadLetters: "bounda.dead_letters", + snapshots: "bounda.snapshots", }; export interface DeadLetteredArgs { diff --git a/packages/core/src/kernel/test-support.ts b/packages/core/src/kernel/test-support.ts index b60e4367..b387b034 100644 --- a/packages/core/src/kernel/test-support.ts +++ b/packages/core/src/kernel/test-support.ts @@ -12,6 +12,7 @@ import type { RejectFunction } from "../modules/command.ts"; import type { PayloadArgs } from "../modules/payload.ts"; import type { Registry } from "../modules/registry.ts"; import { buildAggregates } from "./aggregate/build-aggregates.ts"; +import { createAggregateLoader } from "./aggregate/load-aggregate.ts"; import type { AggregatesRuntime } from "./aggregate/runtime.ts"; import { createCommandPipeline } from "./command/pipeline.ts"; import { createPorts, type ModulePorts } from "./ports/ports.ts"; @@ -315,10 +316,16 @@ export const createKernelHarness: CreateKernelHarnessFunction = async ({ logger: silentLogger, clock, }); - const aggregates = buildAggregates({ registry, ports: byAggregate }); + const aggregates = buildAggregates({ registry, ports: byAggregate, config }); + const loader = createAggregateLoader({ + snapshotStore: storage.snapshotStore, + logger, + strict: true, + }); const rejections: CommandRejection[] = []; const pipeline = createCommandPipeline({ aggregates, + loader, eventStore: storage.eventStore, scheduler: storage.scheduler, config, diff --git a/packages/core/src/memory/index.ts b/packages/core/src/memory/index.ts index 210b6b22..4ad73e98 100644 --- a/packages/core/src/memory/index.ts +++ b/packages/core/src/memory/index.ts @@ -17,6 +17,7 @@ import { createMemoryEventNotifier } from "./event-notifier.ts"; import { createMemoryEventStore } from "./event-store.ts"; import { createKeptInboxLedger } from "./inbox-ledger.ts"; import { createKeptScheduler } from "./scheduler.ts"; +import { createMemorySnapshotStore } from "./snapshot-store.ts"; import { createMemoryReadClient, createMemoryTable, type MemoryTable } from "./table.ts"; import { createCheckpointJournal, @@ -110,6 +111,7 @@ export const memory: MemoryFunction = (options = {}) => { ...stores, notifier, checkpointStore, + snapshotStore: createMemorySnapshotStore(), transact: createMemoryStorageTransaction({ ...stores, entries: { diff --git a/packages/core/src/memory/memory.test.ts b/packages/core/src/memory/memory.test.ts index 79d0045f..4ea8e3ac 100644 --- a/packages/core/src/memory/memory.test.ts +++ b/packages/core/src/memory/memory.test.ts @@ -11,6 +11,7 @@ import { readModelRebuildContract, readModelTransactionContract, schedulerContract, + snapshotStoreContract, storageTransactionContract, tableContract, viewContract, @@ -29,6 +30,7 @@ describe("memory adapter", () => { inboxLedgerContract({ create: async () => (await storage()).inboxLedger }); deadLetterStoreContract({ create: async () => (await storage()).deadLetterStore }); schedulerContract({ create: async () => (await storage()).scheduler }); + snapshotStoreContract({ create: async () => (await storage()).snapshotStore }); storageTransactionContract({ create: storage }); tableContract({ create: async () => { diff --git a/packages/core/src/memory/snapshot-store.ts b/packages/core/src/memory/snapshot-store.ts new file mode 100644 index 00000000..5f7e5fdb --- /dev/null +++ b/packages/core/src/memory/snapshot-store.ts @@ -0,0 +1,19 @@ +import type { Snapshot, SnapshotStore } from "../adapter/storage/snapshot-store.ts"; +import { streamId } from "../contracts/event.ts"; + +export interface CreateMemorySnapshotStoreFunction { + (): SnapshotStore; +} + +export const createMemorySnapshotStore: CreateMemorySnapshotStoreFunction = () => { + const snapshots = new Map(); + return { + get: async (stream) => snapshots.get(streamId(stream)) ?? null, + put: async (snapshot) => { + snapshots.set(streamId(snapshot), snapshot); + }, + remove: async (stream) => { + snapshots.delete(streamId(stream)); + }, + }; +}; diff --git a/packages/core/src/modules/registry.ts b/packages/core/src/modules/registry.ts index 72205db9..cf0ad23c 100644 --- a/packages/core/src/modules/registry.ts +++ b/packages/core/src/modules/registry.ts @@ -20,6 +20,12 @@ import type { ViewModule } from "./view.ts"; export interface AggregateEntry { readonly state?: StateModule; + /** + * Written by the generator: a digest of the modules that shape the state and of the type it + * inferred, which a snapshot of the state is valid for. An aggregate configured with + * `snapshot` needs it. + */ + readonly stateFingerprint?: string; readonly events: EventModules; // Keyed like `events`; only events whose payload has changed shape have one. readonly upcasts?: Readonly>; diff --git a/packages/core/src/modules/validate.test.ts b/packages/core/src/modules/validate.test.ts index a54fcc4b..3519c4e6 100644 --- a/packages/core/src/modules/validate.test.ts +++ b/packages/core/src/modules/validate.test.ts @@ -144,6 +144,21 @@ describe("validateRegistry", () => { ); }); + it("takes a state fingerprint only with something in it", () => { + expect(() => + validateRegistry({ + ...validRegistry, + aggregates: { order: { ...order, stateFingerprint: "" } }, + }), + ).toThrow("aggregates.order.stateFingerprint: must not be empty"); + expect(() => + validateRegistry({ + ...validRegistry, + aggregates: { order: { ...order, stateFingerprint: "abc" } }, + }), + ).not.toThrow(); + }); + it("accepts a well-formed registry", () => { expect(() => validateRegistry(validRegistry)).not.toThrow(); }); diff --git a/packages/core/src/modules/validate.ts b/packages/core/src/modules/validate.ts index 50609baa..18c023a9 100644 --- a/packages/core/src/modules/validate.ts +++ b/packages/core/src/modules/validate.ts @@ -69,6 +69,9 @@ const validateAggregate = ( if (aggregate.state !== undefined && !isRecord(aggregate.state.initialState)) { problems.push({ path: `${base}.state`, message: 'export "initialState" must be an object' }); } + if (aggregate.stateFingerprint !== undefined && aggregate.stateFingerprint === "") { + problems.push({ path: `${base}.stateFingerprint`, message: "must not be empty" }); + } for (const [key, event] of Object.entries(aggregate.events)) { requireFolding(problems, event, `${base}.events.${key}`); optionalFunction(problems, event, `${base}.events.${key}`, "payload"); diff --git a/packages/core/src/testing/index.ts b/packages/core/src/testing/index.ts index 36da0643..8909fac2 100644 --- a/packages/core/src/testing/index.ts +++ b/packages/core/src/testing/index.ts @@ -1,7 +1,7 @@ import type { Adapter } from "../adapter/adapter.ts"; import type { Config } from "../config/types.ts"; import { createFixedClock, type FixedClock } from "../contracts/clock.ts"; -import type { ConfigurationError } from "../contracts/errors.ts"; +import type { ConfigurationError, SnapshotError } from "../contracts/errors.ts"; import { createSequentialIdGenerator, type IdGenerator } from "../contracts/ids.ts"; import { type Logger, silentLogger } from "../contracts/logger.ts"; import { assembleApp, type BoundaApp } from "../kernel/app.ts"; @@ -52,7 +52,8 @@ export type CreateTestAppArgs = { export interface TestApp { /** * The app, whose `runUntilIdle()` throws the `ConfigurationError` of a port the test left out - * once a handler has read it. + * once a handler has read it, and the `SnapshotError` of a snapshot that did not stand for its + * stream. */ readonly app: BoundaApp; /** @@ -92,7 +93,7 @@ export const createTestApp: CreateTestAppFunction = async ({ const ids = createSequentialIdGenerator(); // A reaction's failure stays in its dead letter, so the first port read without a value is // kept here for `runUntilIdle` to throw: the test fails saying what to pass. - let missing: ConfigurationError | undefined; + let missing: ConfigurationError | SnapshotError | undefined; const app = await assembleApp({ registry, // Required by the `Config` of a project whose ports need a choice, and never read here: @@ -109,6 +110,9 @@ export const createTestApp: CreateTestAppFunction = async ({ onMissing: (error) => { missing ??= error; }, + onSnapshotError: (error) => { + missing ??= error; + }, }, }); return { diff --git a/packages/core/src/testing/snapshots.test.ts b/packages/core/src/testing/snapshots.test.ts new file mode 100644 index 00000000..96c6050f --- /dev/null +++ b/packages/core/src/testing/snapshots.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, it, vi } from "vitest"; +import { SnapshotError } from "../contracts/errors.ts"; +import type { PayloadArgs } from "../modules/payload.ts"; +import type { Registry } from "../modules/registry.ts"; +import { createTestApp } from "./index.ts"; + +interface State { + readonly count: number; +} + +class Clock {} + +const counter = (evolve: (state: State) => object) => + ({ + aggregates: { + counter: { + state: { initialState: { count: 0 } satisfies State }, + stateFingerprint: "counter", + events: { incremented: { evolve: ({ state }: { state: State }) => evolve(state) } }, + commands: { + increment: { + module: { + payload: ({ z }: PayloadArgs) => z.object({ counterId: z.string() }), + handler: ({ events }: { events: { incremented: () => unknown } }) => [ + events.incremented(), + ], + }, + }, + read: { + module: { + payload: ({ z }: PayloadArgs) => z.object({ counterId: z.string() }), + handler: ({ state }: { state: State }) => { + seen.push(state.count); + return []; + }, + }, + }, + }, + policies: {}, + processes: {}, + }, + }, + readModels: {}, + }) as const satisfies Registry; + +const seen: number[] = []; +const config = { runtime: { overrides: { counter: { snapshot: { every: 2 } } } } }; + +describe("snapshots in a test app", () => { + it("writes them when the app runs until idle and loads the state from them, checking every load", async () => { + const evolve = vi.fn((state: State) => ({ count: state.count + 1 })); + const { app } = await createTestApp({ registry: counter(evolve), config }); + await app.commands.increment({ counterId: "c-1" }); + await app.commands.increment({ counterId: "c-1" }); + expect(evolve).toHaveBeenCalledTimes(1); + await app.runUntilIdle(); + const lag = await app.getLag(); + expect(lag.subscribers.map((entry) => entry.subscriber)).toEqual(["snapshot:counter"]); + expect(lag.maxLag).toBe(0); + // The subscriber folded the two events once; the next command folds none from the snapshot, + // and the check against the stream folds the two again. + evolve.mockClear(); + await app.commands.read({ counterId: "c-1" }); + expect(seen.at(-1)).toBe(2); + expect(evolve).toHaveBeenCalledTimes(2); + await app.stop(); + }); + + it("fails a command whose state from the snapshot is not the stream's", async () => { + let bonus = 0; + const { app } = await createTestApp({ + registry: counter((state) => ({ count: state.count + 1 + bonus })), + config, + }); + await app.commands.increment({ counterId: "c-1" }); + await app.commands.increment({ counterId: "c-1" }); + await app.runUntilIdle(); + bonus = 1; + await expect(app.commands.read({ counterId: "c-1" })).rejects.toThrow(SnapshotError); + await app.stop(); + }); + + it("makes runUntilIdle throw for a state that cannot be snapshotted", async () => { + const { app } = await createTestApp({ + registry: counter((state) => ({ count: state.count + 1, clock: new Clock() })), + config, + }); + await app.commands.increment({ counterId: "c-1" }); + await app.commands.increment({ counterId: "c-1" }); + await expect(app.runUntilIdle()).rejects.toThrow( + new SnapshotError( + "State of counter:c-1 cannot be snapshotted: Cannot stringify arbitrary non-POJOs", + ), + ); + await app.stop(); + }); +}); diff --git a/packages/core/test-types/config.test-d.ts b/packages/core/test-types/config.test-d.ts index 17809482..7e9ff0c8 100644 --- a/packages/core/test-types/config.test-d.ts +++ b/packages/core/test-types/config.test-d.ts @@ -48,6 +48,16 @@ describe("defineConfig", () => { }, }, }); + defineConfig({ + storage: sqlite, + ports, + runtime: { + overrides: { + order: { snapshot: { every: 200 } }, + customer: { snapshot: { every: 50, verify: 0.1, version: 2 } }, + }, + }, + }); defineConfig({ storage: sqlite, ports, @@ -145,6 +155,18 @@ describe("defineConfig", () => { }); // @ts-expect-error batch is not a role defineConfig({ storage: sqlite, ports, runtime: { role: "batch" } }); + defineConfig({ + storage: sqlite, + ports, + // @ts-expect-error every is a number of events + runtime: { overrides: { order: { snapshot: { every: "200" } } } }, + }); + defineConfig({ + storage: sqlite, + ports, + // @ts-expect-error snapshots are opted into per aggregate, not app-wide + runtime: { snapshot: { every: 200 } }, + }); }); it("requires storage to be an adapter definition", () => { diff --git a/packages/core/test-types/fixtures/order-app-inferred/.bounda/registry.ts b/packages/core/test-types/fixtures/order-app-inferred/.bounda/registry.ts index 301b590f..434fc5b1 100644 --- a/packages/core/test-types/fixtures/order-app-inferred/.bounda/registry.ts +++ b/packages/core/test-types/fixtures/order-app-inferred/.bounda/registry.ts @@ -9,6 +9,7 @@ import * as reminderSent from "../app/domain/order/reminder-sent.ts"; export const registry = { aggregates: { order: { + stateFingerprint: "aff0623b5e0c832f", events: { orderCancelled, orderPaid, orderPlaced, reminderSent }, commands: { payOrder: { module: payOrder }, placeOrder: { module: placeOrder } }, policies: {}, diff --git a/packages/core/test-types/fixtures/order-app/.bounda/registry.ts b/packages/core/test-types/fixtures/order-app/.bounda/registry.ts index 2e13e85b..2d33c8a7 100644 --- a/packages/core/test-types/fixtures/order-app/.bounda/registry.ts +++ b/packages/core/test-types/fixtures/order-app/.bounda/registry.ts @@ -41,6 +41,7 @@ export const registry = { aggregates: { customer: { state: customerState, + stateFingerprint: "c452b414dc30e1d1", events: { customerRegistered }, commands: { registerCustomer: { module: registerCustomer } }, policies: {}, @@ -48,6 +49,7 @@ export const registry = { }, order: { state: orderState, + stateFingerprint: "844a21009adced43", events: { orderCancelled, orderPaid, orderPlaced }, upcasts: { orderPlaced: orderPlacedUpcasts }, ports: { diff --git a/packages/core/test-types/inference.test-d.ts b/packages/core/test-types/inference.test-d.ts index 35eda922..b5965447 100644 --- a/packages/core/test-types/inference.test-d.ts +++ b/packages/core/test-types/inference.test-d.ts @@ -227,6 +227,14 @@ describe("ports", () => { expectTypeOf().not.toHaveProperty("rates"); }); + it("carries the state fingerprint the generator computed for every aggregate", () => { + expectTypeOf().toBeString(); + expectTypeOf().toBeString(); + expectTypeOf().not.toEqualTypeOf< + string | undefined + >(); + }); + it("leave the aggregate's other modules out of the registry and the events", () => { expectTypeOf().not.toHaveProperty("money"); expectTypeOf().not.toHaveProperty("money"); diff --git a/packages/postgresql/src/event-store.ts b/packages/postgresql/src/event-store.ts index 1071189a..5e8492f7 100644 --- a/packages/postgresql/src/event-store.ts +++ b/packages/postgresql/src/event-store.ts @@ -94,20 +94,27 @@ export const createPostgresqlEventStore: CreatePostgresqlEventStoreFunction = ({ } return { version: actualVersion + stored.length, events: stored }; }), + // A load from the first event reads one statement, whose last row is the head. From later + // on, the head is read first and the events bounded by it: outside a transaction, an append + // landing between the two reads would otherwise report a head past the events returned. load: async ({ aggregateType, aggregateId, fromVersion = 1 }) => { - // `version` is an integer column, and a version past its range, such as the one that asks - // for the version alone, must still compare instead of failing to bind. - const rows = await db.all( - `SELECT ${COLUMNS} FROM ${table} WHERE "aggregate_type" = $1 AND "aggregate_id" = $2 AND "version" >= $3::bigint ORDER BY "version"`, - [aggregateType, aggregateId, fromVersion], - ); - const events = rows.map(toStoredEvent); - const last = events.at(-1); - const version = - fromVersion <= 1 - ? (last?.version ?? 0) - : await currentVersion(db, aggregateType, aggregateId); - return { events, version }; + if (fromVersion <= 1) { + const rows = await db.all( + `SELECT ${COLUMNS} FROM ${table} WHERE "aggregate_type" = $1 AND "aggregate_id" = $2 ORDER BY "version"`, + [aggregateType, aggregateId], + ); + const events = rows.map(toStoredEvent); + return { events, version: events.at(-1)?.version ?? 0 }; + } + const version = await currentVersion(db, aggregateType, aggregateId); + const rows = + version < fromVersion + ? [] + : await db.all( + `SELECT ${COLUMNS} FROM ${table} WHERE "aggregate_type" = $1 AND "aggregate_id" = $2 AND "version" >= $3 AND "version" <= $4 ORDER BY "version"`, + [aggregateType, aggregateId, fromVersion, version], + ); + return { events: rows.map(toStoredEvent), version }; }, readAll: async ({ afterPosition, limit }) => ( diff --git a/packages/postgresql/src/index.ts b/packages/postgresql/src/index.ts index 907c88ef..b6d77d71 100644 --- a/packages/postgresql/src/index.ts +++ b/packages/postgresql/src/index.ts @@ -16,6 +16,7 @@ import { type PostgresqlOptions, resolvePostgresqlOptions } from "./options.ts"; import { openPostgresqlReadModel, rebuildPostgresqlReadModel } from "./read-model.ts"; import { createPostgresqlScheduler } from "./scheduler.ts"; import { ensureStorageSchema, storageTablesFor } from "./schema.ts"; +import { createPostgresqlSnapshotStore } from "./snapshot-store.ts"; /** * The adapter `postgresql(...)` returns: what `bounda.config.ts` holds under `storage` or @@ -112,6 +113,7 @@ export const postgresql: PostgresqlFunction = (options) => { ...storesOver(db), notifier: createPostgresqlEventNotifier({ sql, channel: tables.channel }), checkpointStore: createPostgresqlCheckpointStore({ db, table: tables.checkpoints }), + snapshotStore: createPostgresqlSnapshotStore({ db, table: tables.snapshots }), // The append lock comes first, before any row the work may lock: two transactions that // took the lock and a row in opposite orders would deadlock. transact: (work) => diff --git a/packages/postgresql/src/postgresql.test.ts b/packages/postgresql/src/postgresql.test.ts index 2512c3fe..5543eb7b 100644 --- a/packages/postgresql/src/postgresql.test.ts +++ b/packages/postgresql/src/postgresql.test.ts @@ -22,6 +22,7 @@ import { readModelRebuildContract, readModelTransactionContract, schedulerContract, + snapshotStoreContract, storageTransactionContract, tableContract, viewContract, @@ -30,6 +31,8 @@ import { createTestApp } from "@bounda-dev/core/testing"; import { PostgreSqlContainer, type StartedPostgreSqlContainer } from "@testcontainers/postgresql"; import postgres, { type Sql } from "postgres"; import { afterAll, describe, expect, it, vi } from "vitest"; +import { createPostgresqlDatabase } from "./database.ts"; +import { createPostgresqlEventStore } from "./event-store.ts"; import { type PostgresqlAdapter, type PostgresqlOptions, postgresql } from "./index.ts"; import { resolvePostgresqlOptions } from "./options.ts"; @@ -93,6 +96,7 @@ describe.skipIf(container === null)("postgresql adapter", () => { inboxLedgerContract({ create: async () => (await openStorage()).inboxLedger }); deadLetterStoreContract({ create: async () => (await openStorage()).deadLetterStore }); schedulerContract({ create: async () => (await openStorage()).scheduler }); + snapshotStoreContract({ create: async () => (await openStorage()).snapshotStore }); storageTransactionContract({ create: () => openStorage() }); it("takes the append lock as a transaction starts, before anything it writes, and lets go of it with the transaction", async () => { @@ -379,6 +383,51 @@ describe.skipIf(container === null)("postgresql adapter", () => { expect(await eventStore.lastPosition()).toBe(100); }); + it("never reports a version past the events a load from a later version returns", async () => { + const adapter = fresh(); + const storage = await openStorage(adapter); + const stream = { aggregateType: "order", aggregateId: "1" }; + await storage.eventStore.append({ + ...stream, + expectedVersion: 0, + events: [1, 2].map((version) => pendingEvent({ aggregateId: "1", version })), + }); + const { tablePrefix } = resolvePostgresqlOptions(adapter.options); + const sql = postgres(url, { max: 1, onnotice: () => undefined }); + closers.push(() => sql.end({ timeout: 5 })); + const db = createPostgresqlDatabase(sql); + // An append lands between the first and the second statement of the load. + let statements = 0; + const store = createPostgresqlEventStore({ + db: { + ...db, + all: async (statement, params) => { + statements += 1; + if (statements === 2) { + await storage.eventStore.append({ + ...stream, + expectedVersion: 2, + events: [pendingEvent({ aggregateId: "1", version: 3 })], + }); + } + return db.all(statement, params); + }, + }, + table: `"${tablePrefix}events"`, + lockKey: "unused", + channel: "unused", + }); + const loaded = await store.load({ ...stream, fromVersion: 2 }); + expect(loaded.events.map((event) => event.version)).toEqual([2]); + expect(loaded.version).toBe(2); + expect(statements).toBe(2); + // From the first event, or from past the head, one statement is enough. + expect(await store.load(stream)).toMatchObject({ version: 3 }); + expect(statements).toBe(3); + expect(await store.load({ ...stream, fromVersion: 4 })).toEqual({ events: [], version: 3 }); + expect(statements).toBe(4); + }); + it("keeps every table in the configured schema", async () => { const schema = `s_${run}`; const adapter = fresh({ schema }); @@ -396,6 +445,7 @@ describe.skipIf(container === null)("postgresql adapter", () => { "inbox", "rm_order_summary", "scheduled_commands", + "snapshots", ]); }); diff --git a/packages/postgresql/src/schema.ts b/packages/postgresql/src/schema.ts index b0373966..3c37f25a 100644 --- a/packages/postgresql/src/schema.ts +++ b/packages/postgresql/src/schema.ts @@ -13,6 +13,7 @@ export interface StorageTables { readonly inbox: string; readonly deadLetters: string; readonly scheduledCommands: string; + readonly snapshots: string; readonly appendLockKey: string; // The first key of the advisory locks read models take on the checkpoints table. readonly checkpointsLockKey: string; @@ -51,6 +52,7 @@ export const storageTablesFor: StorageTablesForFunction = ({ prefix, schema }) = inbox: quoteIdentifier(storageTableNameFor({ prefix, table: "inbox" })), deadLetters: quoteIdentifier(storageTableNameFor({ prefix, table: "deadLetters" })), scheduledCommands: quoteIdentifier(storageTableNameFor({ prefix, table: "scheduledCommands" })), + snapshots: quoteIdentifier(storageTableNameFor({ prefix, table: "snapshots" })), appendLockKey: `bounda:${schema}.${events}`, checkpointsLockKey: `bounda:${schema}.${checkpoints}`, channel, @@ -131,6 +133,14 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "claim_id" text )`, `CREATE INDEX IF NOT EXISTS ${indexName(tables.scheduledCommands, "execute_at")} ON ${tables.scheduledCommands} ("execute_at")`, + `CREATE TABLE IF NOT EXISTS ${tables.snapshots} ( + "aggregate_type" text NOT NULL, + "aggregate_id" text NOT NULL, + "version" integer NOT NULL, + "fingerprint" text NOT NULL, + "state" text NOT NULL, + PRIMARY KEY ("aggregate_type", "aggregate_id") + )`, ]; export interface EnsureStorageSchemaArgs { @@ -156,6 +166,7 @@ export const ensureStorageSchema: EnsureStorageSchemaFunction = async ({ db, sch tables.inbox, tables.deadLetters, tables.scheduledCommands, + tables.snapshots, indexName(tables.deadLetters, "status"), indexName(tables.scheduledCommands, "execute_at"), ].map(unquoted); diff --git a/packages/postgresql/src/snapshot-store.ts b/packages/postgresql/src/snapshot-store.ts new file mode 100644 index 00000000..2c8c3c00 --- /dev/null +++ b/packages/postgresql/src/snapshot-store.ts @@ -0,0 +1,42 @@ +import type { Snapshot, SnapshotStore } from "@bounda-dev/core/adapter"; +import type { SqlExecutor } from "@bounda-dev/core/adapter/sql"; + +export interface CreatePostgresqlSnapshotStoreArgs { + readonly db: SqlExecutor; + readonly table: string; +} + +export interface CreatePostgresqlSnapshotStoreFunction { + (args: CreatePostgresqlSnapshotStoreArgs): SnapshotStore; +} + +const toSnapshot = (row: Record): Snapshot => ({ + aggregateType: String(row.aggregate_type), + aggregateId: String(row.aggregate_id), + version: Number(row.version), + fingerprint: String(row.fingerprint), + state: String(row.state), +}); + +export const createPostgresqlSnapshotStore: CreatePostgresqlSnapshotStoreFunction = ({ + db, + table, +}) => ({ + get: async ({ aggregateType, aggregateId }) => { + const [row] = await db.all( + `SELECT "aggregate_type", "aggregate_id", "version", "fingerprint", "state" FROM ${table} WHERE "aggregate_type" = $1 AND "aggregate_id" = $2`, + [aggregateType, aggregateId], + ); + return row === undefined ? null : toSnapshot(row); + }, + put: ({ aggregateType, aggregateId, version, fingerprint, state }) => + db.run( + `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state") VALUES ($1, $2, $3, $4, $5) ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state"`, + [aggregateType, aggregateId, version, fingerprint, state], + ), + remove: ({ aggregateType, aggregateId }) => + db.run(`DELETE FROM ${table} WHERE "aggregate_type" = $1 AND "aggregate_id" = $2`, [ + aggregateType, + aggregateId, + ]), +}); diff --git a/packages/sqlite/src/sqlite.test.ts b/packages/sqlite/src/sqlite.test.ts index 815f276f..69a122b4 100644 --- a/packages/sqlite/src/sqlite.test.ts +++ b/packages/sqlite/src/sqlite.test.ts @@ -22,6 +22,7 @@ import { readModelRebuildContract, readModelTransactionContract, schedulerContract, + snapshotStoreContract, storageTransactionContract, tableContract, viewContract, @@ -41,6 +42,7 @@ describe("sqlite adapter in memory", () => { inboxLedgerContract({ create: async () => (await openStorage()).inboxLedger }); deadLetterStoreContract({ create: async () => (await openStorage()).deadLetterStore }); schedulerContract({ create: async () => (await openStorage()).scheduler }); + snapshotStoreContract({ create: async () => (await openStorage()).snapshotStore }); storageTransactionContract({ create: () => openStorage() }); tableContract({ create: async () => @@ -239,6 +241,7 @@ describe("sqlite adapter on a file", () => { "app_inbox", "app_rm_order_summary", "app_scheduled_commands", + "app_snapshots", ]); await readModel.close(); await second.close(); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 27cc3f27..5f538db6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -230,6 +230,9 @@ catalogs: commander: specifier: 15.0.0 version: 15.0.0 + devalue: + specifier: 6.0.3 + version: 6.0.3 isbot: specifier: 5.2.2 version: 5.2.2 @@ -519,6 +522,9 @@ importers: '@opentelemetry/api': specifier: 'catalog:' version: 1.10.0 + devalue: + specifier: 'catalog:' + version: 6.0.3 uuid: specifier: 'catalog:' version: 14.0.3 @@ -11266,7 +11272,7 @@ snapshots: esbuild: 0.28.2 fdir: 6.5.0(picomatch@4.0.7) picomatch: 4.0.7 - postcss: 8.5.28 + postcss: 8.5.31 rollup: 4.64.2 tinyglobby: 0.2.17 optionalDependencies: diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 01a7e266..f0b89e5f 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -29,6 +29,7 @@ catalog: "@vitest/coverage-v8": 5.0.5 astro: 7.4.1 commander: 15.0.0 + devalue: 6.0.3 isbot: 5.2.2 lefthook: 2.2.2 postgres: 3.4.9