From fc4d826138dce999eabd0613db9913f2d9983eb0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Sat, 31 Oct 2026 11:44:26 +0100 Subject: [PATCH 1/2] fix: close what the review of idempotency, snapshots and personal data found - createWorker no longer forgets anyone over HTTP: an irreversible operation has no place on an endpoint without authentication; apps call `forget` through `connect` behind their own. - The React Router host derives each command's key from the Idempotency-Key header and the command's name, so an action that dispatches two commands no longer conflicts on the second. - An idempotency key now also names the dispatch's delay, and fingerprints tell maps, sets and bigints apart instead of collapsing them or throwing. - A person whose fields were read in the clear needs no key for a snapshot or a remembered rejection to be written, so such streams are snapshotted again. - A snapshot never replaces a newer one of the same fingerprint. - PostgreSQL indexes `persons` so forgetting a person finds their rows without a full scan. - The personal-data guide says what forget cannot reach: scheduled payloads, dead letters, logs. --- .changeset/aggregate-snapshots.md | 2 +- .changeset/idempotent-commands.md | 13 +++--- .changeset/personal-data.md | 2 +- docs/src/content/docs/adapters/cloudflare.md | 7 +-- .../docs/guides/idempotent-commands.md | 36 +++++++++------- docs/src/content/docs/guides/personal-data.md | 6 +++ docs/src/content/docs/guides/react-router.md | 8 ++-- docs/src/content/docs/reference/errors.md | 3 +- packages/cloudflare/src/worker.ts | 21 +++------ packages/cloudflare/test/worker.test.ts | 12 +----- packages/core/src/adapter/sqlite/held-keys.ts | 9 ++-- .../src/adapter/sqlite/idempotency-store.ts | 4 +- .../core/src/adapter/sqlite/snapshot-store.ts | 4 +- .../src/adapter/storage/idempotency-store.ts | 6 +-- .../src/adapter/storage/snapshot-store.ts | 11 ++--- .../testing/idempotency-store.contract.ts | 2 +- .../testing/snapshot-store.contract.ts | 33 +++++++++++++- packages/core/src/contracts/command.ts | 6 +-- packages/core/src/contracts/errors.ts | 6 +-- .../src/kernel/command/fingerprint.test.ts | 37 ++++++++++++++++ .../core/src/kernel/command/fingerprint.ts | 30 ++++++++++--- .../src/kernel/command/idempotency.test.ts | 15 +++++++ packages/core/src/kernel/command/pipeline.ts | 6 ++- .../personal-data/personal-data.test.ts | 35 +++++++++++++++ .../core/src/memory/personal-key-store.ts | 4 +- packages/core/src/memory/snapshot-store.ts | 7 ++- packages/postgresql/src/held-keys.ts | 11 ++--- packages/postgresql/src/schema.ts | 5 +++ packages/postgresql/src/snapshot-store.ts | 2 +- .../src/cloudflare/create-bounda.test.ts | 5 ++- .../react-router/src/create-bounda.test.ts | 10 +++-- .../react-router/src/idempotency-key.test.ts | 43 ++++++++++++++++--- packages/react-router/src/idempotency-key.ts | 19 +++++--- 33 files changed, 304 insertions(+), 116 deletions(-) diff --git a/.changeset/aggregate-snapshots.md b/.changeset/aggregate-snapshots.md index 6a534f6b..3c3b7fd1 100644 --- a/.changeset/aggregate-snapshots.md +++ b/.changeset/aggregate-snapshots.md @@ -4,4 +4,4 @@ "@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. +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`, which keeps the newest snapshot of a fingerprint whatever order two writers put theirs in; `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/.changeset/idempotent-commands.md b/.changeset/idempotent-commands.md index da3b74e1..c0405a18 100644 --- a/.changeset/idempotent-commands.md +++ b/.changeset/idempotent-commands.md @@ -6,13 +6,14 @@ A command can be dispatched again without running twice: `commands.x(payload, { idempotencyKey })` remembers what the first dispatch under a key answered, in the same transaction as the events it -stores, and every later dispatch with the same key, command and payload answers the same, its -rejection included, without running the handler; two racing for one key store the events once. -The same key for another command or payload rejects with `IdempotencyKeyConflictError` +stores, and every later dispatch with the same key, command, payload and `delay` answers the same, +its rejection included, without running the handler; two racing for one key store the events once. +The same key for another command, payload or `delay` rejects with `IdempotencyKeyConflictError` (`IDEMPOTENCY_KEY_CONFLICT`). A key lives `runtime.commands.idempotencyKeyLifetime`, 24 hours by default, in a new `bounda_idempotency_keys` table of every adapter, which the scheduled-command worker purges. `Storage` and `StorageTransaction` gain an `idempotencyStore`, with its contract -test in `@bounda-dev/core/adapter/testing`. The React Router middleware passes a request's -`Idempotency-Key` header to every command of the request, and `failure()` answers a conflict with -a 422. The in-memory adapter now answers, inside a transaction, the positions its events commit +test in `@bounda-dev/core/adapter/testing`. The React Router middleware gives every command of a request +a key derived from its `Idempotency-Key` header and the command's name, `idempotencyKeyFor(header, +name)`, so an action that dispatches two commands keys each apart, and `failure()` answers a +conflict with a 422. The in-memory adapter now answers, inside a transaction, the positions its events commit with, as the SQL adapters do. diff --git a/.changeset/personal-data.md b/.changeset/personal-data.md index 9fb3f359..7caf1f68 100644 --- a/.changeset/personal-data.md +++ b/.changeset/personal-data.md @@ -7,4 +7,4 @@ "@bounda-dev/react-router": minor --- -Personal data in events, forgotten by crypto-shredding. An event's `payload` receives `personal` and `person` next to `z`: `personal(schema)` marks a field as personal data, stored encrypted (AES-256-GCM through WebCrypto, on Node and in `workerd`) with a key of the person the event names, `person(schema)` marks the field that names them, and without it the aggregate's id does. `app.forget(personId)`, on every `BoundaClient`, deletes the key and appends `PersonalDataForgotten` to the stream `bounda:`, in one transaction: from then on every personal field of theirs reads as `null`, typed `T | null` in every handler. A read model that projects an event with personal fields must handle it from `projections/bounda/personal-data-forgotten.ts`, which the generator checks in `.bounda/types.ts`; a policy may react to it from `policies/bounda/`. The adapters keep the keys, with their fingerprints, in a new `personal_keys` table, `Storage` and `StorageTransaction` gain `personalKeys`, and the storage contracts cover it. The Cloudflare Worker answers `POST /forget/`. The global stream shows the fields as stored, encrypted. A command that loaded a person's data before `forget` and appends after it fails with `ConcurrencyError` and is decided again, instead of writing their data back under the new key another command may have given them meanwhile: the state remembers the key each person's data was read under, and `AppendArgs` gains `persons` (`{ person, fingerprint }`) for this, which a store ignores. Snapshots and remembered rejections name the same and are written only while that key is still the person's. +Personal data in events, forgotten by crypto-shredding. An event's `payload` receives `personal` and `person` next to `z`: `personal(schema)` marks a field as personal data, stored encrypted (AES-256-GCM through WebCrypto, on Node and in `workerd`) with a key of the person the event names, `person(schema)` marks the field that names them, and without it the aggregate's id does. `app.forget(personId)`, on every `BoundaClient`, deletes the key and appends `PersonalDataForgotten` to the stream `bounda:`, in one transaction: from then on every personal field of theirs reads as `null`, typed `T | null` in every handler. A read model that projects an event with personal fields must handle it from `projections/bounda/personal-data-forgotten.ts`, which the generator checks in `.bounda/types.ts`; a policy may react to it from `policies/bounda/`. The adapters keep the keys, with their fingerprints, in a new `personal_keys` table, `Storage` and `StorageTransaction` gain `personalKeys`, and the storage contracts cover it. The global stream shows the fields as stored, encrypted. A command that loaded a person's data before `forget` and appends after it fails with `ConcurrencyError` and is decided again, instead of writing their data back under the new key another command may have given them meanwhile: the state remembers the key each person's data was read under, and `AppendArgs` gains `persons` (`{ person, fingerprint }`) for this, which a store ignores. Snapshots and remembered rejections name the same and are written only while that key is still the person's; in PostgreSQL a GIN index on their `persons` lets `forget` find them without reading every row. diff --git a/docs/src/content/docs/adapters/cloudflare.md b/docs/src/content/docs/adapters/cloudflare.md index 241131e9..ea36c318 100644 --- a/docs/src/content/docs/adapters/cloudflare.md +++ b/docs/src/content/docs/adapters/cloudflare.md @@ -149,14 +149,15 @@ global stream. | --- | --- | | `POST /commands/` with the payload as the body | The dispatch result, once the read models reflect it, or once its events are stored with `createWorker({ config, tenantOf, consistency: "eventual" })`. `?delay=10m` schedules it | | `POST /queries/` with the payload as the body | The query's result | -| `POST /forget/` | [Forgets](/guides/personal-data/) the person, answering like a command | Refusals come back as `{ "error": { "code", "message" } }` with the statuses in [errors over HTTP](/reference/errors/#over-http); anything else is a 500 whose message goes to the logs, not to the caller. -It has **no authentication** and no operator endpoint, on purpose: it is a starting point. An -app with users writes its own `fetch` and talks to a store with `connect`: +It has **no authentication** and no operator endpoint, on purpose: it is a starting point. For +the same reason it never [forgets](/guides/personal-data/) anyone, which cannot be undone. An app +with users writes its own `fetch` and talks to a store with `connect`, whose `forget` it calls +behind its own authentication: ```ts import { connect } from "@bounda-dev/cloudflare"; diff --git a/docs/src/content/docs/guides/idempotent-commands.md b/docs/src/content/docs/guides/idempotent-commands.md index afe1aff9..9d6e45dc 100644 --- a/docs/src/content/docs/guides/idempotent-commands.md +++ b/docs/src/content/docs/guides/idempotent-commands.md @@ -10,7 +10,9 @@ webhook was delivered again. The aggregate rejects what it must (`placeOrder` ag `AlreadyPlaced`), but a rejection is the wrong answer to a retry of a command that went through, and a command the aggregate takes twice, such as adding a line, does its work twice. An **idempotency key** names one dispatch, so a second arrival under it answers what the first did -instead of running again. +instead of running again. What the runtime already guarantees for the events and reactions inside +the app is in [delivery guarantees](/concepts/delivery-guarantees/); this page is about commands +that come from outside it. ## The `idempotencyKey` option @@ -19,8 +21,8 @@ const placed = await app.commands.placeOrder(payload, { idempotencyKey: key }); ``` The first dispatch under a key runs the command and remembers what it answered, in the same -transaction as the events it stores. Every later dispatch with the same key, the same command and -the same payload answers the same without running the handler: the same `StoredDispatch`, its +transaction as the events it stores. Every later dispatch with the same key, the same command, +the same payload and the same `delay` answers the same without running the handler: the same `StoredDispatch`, its `position` included, so [reading your own writes](/guides/deployment/#reading-your-own-writes) waits for the same read models; the same `ScheduledDispatch` for a command with `delay`, which is scheduled once; or the same rejection, thrown again as a `DomainError` with its code. A dispatch @@ -30,10 +32,11 @@ handler's, is not remembered, so the client's retry runs it. Two dispatches racing for one key both run; one records its outcome and the other loses on the key, rolls back whatever it stored, and answers what the key holds. The events land once. -The key names one command with one payload. The same key with another command, or with the same -command and another payload, is a bug of the caller, not a retry: the dispatch rejects with +The key names one command with one payload and one `delay`. The same key with another command, +with another payload, or scheduled when it ran at once (or the other way round, or with another +delay), is a bug of the caller, not a retry: the dispatch rejects with `IdempotencyKeyConflictError` (`IDEMPOTENCY_KEY_CONFLICT`) and runs nothing. The payload is -compared as JSON, keys in any order. +compared as JSON, keys in any order, with maps, sets and bigints compared by what they hold. A key is a string of 1 to 255 characters that the caller chooses, one per operation the client means to do once, such as a UUID made when the form is submitted and sent again on every retry of @@ -52,9 +55,10 @@ does not: a replay never runs it. ## The `Idempotency-Key` header in React Router -The `boundaMiddleware` reads the `Idempotency-Key` header of every request and passes it as -`idempotencyKey` to every command the request dispatches, on Node and on Cloudflare alike, so an -action has nothing to add: +The `boundaMiddleware` reads the `Idempotency-Key` header of every request and gives every command +the request dispatches an `idempotencyKey` of its own, `idempotencyKeyFor(header, commandName)`, on +Node and on Cloudflare alike, so an action has nothing to add, even one that dispatches two +commands: ```tsx // app/routes/register.tsx @@ -78,15 +82,17 @@ const key = crypto.randomUUID(); fetcher.submit(form, { method: "post", headers: { "Idempotency-Key": key } }); ``` -A key names one command, so an action that dispatches two under one header gets a conflict on the -second: derive one per command with `idempotencyKeyFor` from `@bounda-dev/core`, or pass -`idempotencyKey` yourself, which wins over the header. +A key names one command, so an action that dispatches the same command twice shares one key +between them: the second answers what the first did with the same payload, and conflicts with +another. Pass `idempotencyKey` yourself, which wins over the header, derived with +`idempotencyKeyFor` from `@bounda-dev/core`: ```ts const key = request.headers.get("Idempotency-Key"); -const keyed = (effect: string) => (key === null ? {} : { idempotencyKey: idempotencyKeyFor(key, effect) }); -await commands.reserveSeats(seats, keyed("reserve")); -await commands.placeOrder(order, keyed("place")); +for (const [index, seat] of seats.entries()) { + const keyed = key === null ? {} : { idempotencyKey: idempotencyKeyFor(key, `seat-${index}`) }; + await commands.reserveSeat(seat, keyed); +} ``` `failure()` answers a conflict with a 422 ([errors from the domain](/guides/react-router/#errors-from-the-domain)). diff --git a/docs/src/content/docs/guides/personal-data.md b/docs/src/content/docs/guides/personal-data.md index 74bd0c2e..533aa08f 100644 --- a/docs/src/content/docs/guides/personal-data.md +++ b/docs/src/content/docs/guides/personal-data.md @@ -101,6 +101,12 @@ them. What it cannot see is personal data that came in the command's payload, which is never marked: keep that out of a rejection's message and say what went wrong with its code and the ids involved. +Command payloads are never sealed anywhere, so `forget` does not reach what holds one: a command +scheduled with `delay` keeps its payload until it runs, a +[dead letter](/guides/dead-letters/) of a scheduled command keeps it until it is retried or +discarded, and your logs keep whatever you wrote to them. Pass ids in those payloads, not the +data itself: the handler reads the email from the state, where it is sealed and forgotten. + ## Handling it in a read model A read model that projects an event with personal fields has those fields in its rows. It must diff --git a/docs/src/content/docs/guides/react-router.md b/docs/src/content/docs/guides/react-router.md index d80b720c..d497ca84 100644 --- a/docs/src/content/docs/guides/react-router.md +++ b/docs/src/content/docs/guides/react-router.md @@ -106,7 +106,7 @@ when its handler rejects it, with the code in `rejected`. Return `failure(error) |---|---|---| | `ValidationError` | 400 | `{ error, issues }`, each issue with the `path` of the field and its `message` | | `DomainError` | 409 | `{ error, issues: [], rejected }`, the code the command declares | -| `IdempotencyKeyConflictError` | 422 | `{ error, issues: [] }`, the request's `Idempotency-Key` was used for another command | +| `IdempotencyKeyConflictError` | 422 | `{ error, issues: [] }`, the request's `Idempotency-Key` came back with another payload | `failure` goes by the error's `code`, `VALIDATION_FAILED`, `DOMAIN_ERROR` or `IDEMPOTENCY_KEY_CONFLICT`, so it answers the same for the refusals a Durable Object sends back on @@ -134,9 +134,9 @@ cannot carry it into the Durable Object, where the command runs to the end. ## Retrying a request safely A request that times out on the way back, or a form submitted twice, dispatches its command -twice. When the client sends an `Idempotency-Key` header, the middleware passes it as -`idempotencyKey` to every command the request dispatches, and the second arrival answers what the -first did, its rejection included, without running the command again. The host never makes a key +twice. When the client sends an `Idempotency-Key` header, the middleware gives every command the +request dispatches an `idempotencyKey` derived from it and the command's name, and the second +arrival answers what the first did, its rejection included, without running the command again. The host never makes a key up: a request without the header dispatches as before. What the key means, its lifetime, and an action that dispatches two commands under one header are in [idempotent commands](/guides/idempotent-commands/). diff --git a/docs/src/content/docs/reference/errors.md b/docs/src/content/docs/reference/errors.md index 908ebb1a..b456064d 100644 --- a/docs/src/content/docs/reference/errors.md +++ b/docs/src/content/docs/reference/errors.md @@ -56,7 +56,8 @@ dead letter with `errorType: "retriable_exhausted"`. than `runtime.policies.maxChainDepth`. `depth`, `maxDepth`. - **`IDEMPOTENCY_KEY_CONFLICT`** (`IdempotencyKeyConflictError`; terminal). A dispatch's `idempotencyKey` was already used, within `runtime.commands.idempotencyKeyLifetime`, for - another command or for the same command with another payload. Nothing runs. `idempotencyKey`. + another command or for the same command with another payload or `delay`. Nothing runs. + `idempotencyKey`. See [idempotent commands](/guides/idempotent-commands/). - **`CREATION_ORDER`** (`CreationOrderError`; terminal). A command returned events that do not fit whether its aggregate exists: the first event is not one that exports `begin`, or an event that diff --git a/packages/cloudflare/src/worker.ts b/packages/cloudflare/src/worker.ts index 48dc937c..9327bd31 100644 --- a/packages/cloudflare/src/worker.ts +++ b/packages/cloudflare/src/worker.ts @@ -74,16 +74,8 @@ const optionsOf = (url: URL): DispatchOptions => { }; const route = (pathname: string): { readonly kind: string; readonly name: string } | null => { - const match = - /^\/(commands|queries)\/([A-Za-z][A-Za-z0-9]*)\/?$/.exec(pathname) ?? - /^\/(forget)\/([^/]+)\/?$/.exec(pathname); - if (match === null) return null; - try { - return { kind: match[1] as string, name: decodeURIComponent(match[2] as string) }; - } catch { - // A malformed escape names no one. - return null; - } + const match = /^\/(commands|queries)\/([A-Za-z][A-Za-z0-9]*)\/?$/.exec(pathname); + return match === null ? null : { kind: match[1] as string, name: match[2] as string }; }; /** @@ -92,8 +84,7 @@ const route = (pathname: string): { readonly kind: string; readonly name: string * * - `POST /commands/` with the payload as the body, `?delay=10m` to schedule it, answers * the dispatch result, once the read models reflect it unless `consistency` is `"eventual"`; - * - `POST /queries/` with the payload as the body answers the query's result; - * - `POST /forget/` forgets a person, as `forget` does, and answers like a command. + * - `POST /queries/` with the payload as the body answers the query's result. * * Refusals come back as `{ error: { code, message } }`: 400 for a payload that fails validation, * with `issues`, 404 for an unknown command, query or row, 409 for a command's rejection, with its @@ -120,7 +111,7 @@ export const createWorker: CreateWorkerFunction = ({ } let payload: unknown; try { - payload = target.kind === "forget" ? undefined : await readPayload(request); + payload = await readPayload(request); } catch { return failure("INVALID_JSON", "The body is not JSON", 400); } @@ -136,9 +127,7 @@ export const createWorker: CreateWorkerFunction = ({ const result = await unwrap( target.kind === "commands" ? stub.command(target.name, payload, optionsOf(url), consistency) - : target.kind === "queries" - ? stub.query(target.name, payload) - : stub.forget(target.name, consistency), + : stub.query(target.name, payload), ); return json(result ?? null); } catch (error) { diff --git a/packages/cloudflare/test/worker.test.ts b/packages/cloudflare/test/worker.test.ts index 2fd38f57..a23b8a14 100644 --- a/packages/cloudflare/test/worker.test.ts +++ b/packages/cloudflare/test/worker.test.ts @@ -231,17 +231,7 @@ describe("createWorker", () => { }); }); - it("forgets a person through the binding, with the worker's consistency", async () => { - const byDefault = await throughSpy("/forget/ada%20b", ""); - expect(byDefault.status).toBe(200); - expect(byDefault.addressed).toEqual(["acme"]); - expect(byDefault.forgets).toEqual([["ada b", "read-your-writes"]]); - const eventual = await throughSpy("/forget/ada/", "{not json", "eventual"); - expect(eventual.status).toBe(200); - expect(eventual.forgets).toEqual([["ada", "eventual"]]); - }); - - it("only knows POST on its three routes", async () => { + it("only knows POST on its two routes", async () => { const root = await exports.default.fetch(new Request("https://bounda.test/")); expect(root.status).toBe(404); expect(await root.json()).toEqual({ error: { code: "NOT_FOUND", message: "No route for /" } }); diff --git a/packages/core/src/adapter/sqlite/held-keys.ts b/packages/core/src/adapter/sqlite/held-keys.ts index 96b21899..3f370514 100644 --- a/packages/core/src/adapter/sqlite/held-keys.ts +++ b/packages/core/src/adapter/sqlite/held-keys.ts @@ -3,12 +3,13 @@ export interface HeldKeysFunction { } /** - * A condition true when every person in the JSON array of the first parameter holds, in - * `personalKeys`, the key of the fingerprint named with them, or any key for a `null` one; the - * second parameter is the array's length. A person named under two keys never holds both. + * A condition true when every person in the JSON array of the first parameter named with a + * fingerprint holds, in `personalKeys`, the key of that fingerprint; the second parameter is the + * same array. A person named with `null` was read in the clear and needs no key. A person named + * under two keys never holds both. */ export const heldKeys: HeldKeysFunction = (personalKeys) => - `(SELECT count(*) FROM ${personalKeys} AS "k" JOIN json_each(?) AS "p" ON "k"."person" = json_extract("p"."value", '$.person') AND "k"."fingerprint" = coalesce(json_extract("p"."value", '$.fingerprint'), "k"."fingerprint")) = ?`; + `(SELECT count(*) FROM ${personalKeys} AS "k" JOIN json_each(?) AS "p" ON "k"."person" = json_extract("p"."value", '$.person') AND "k"."fingerprint" = json_extract("p"."value", '$.fingerprint')) = (SELECT count(*) FROM json_each(?) WHERE json_extract("value", '$.fingerprint') IS NOT NULL)`; export interface NamingFunction { (table: string): string; diff --git a/packages/core/src/adapter/sqlite/idempotency-store.ts b/packages/core/src/adapter/sqlite/idempotency-store.ts index 73b6ba29..82009bec 100644 --- a/packages/core/src/adapter/sqlite/idempotency-store.ts +++ b/packages/core/src/adapter/sqlite/idempotency-store.ts @@ -42,7 +42,7 @@ export const createSqliteIdempotencyStore: CreateSqliteIdempotencyStoreFunction return { get, // One statement whose row comes back only when the key was free or expired and every person - // the record names holds the key they were read under, so two dispatches racing for it can + // the record names with a fingerprint holds that key, so two dispatches racing for it can // never both record, nor one record beside a forget. No row back with no live record under // the key is the second case. put: async ({ key, fingerprint, outcome, expiresAt, now, persons }) => { @@ -63,7 +63,7 @@ export const createSqliteIdempotencyStore: CreateSqliteIdempotencyStoreFunction expiresAt, JSON.stringify(persons), JSON.stringify(persons), - persons.length, + JSON.stringify(persons), now.toISOString(), ], ); diff --git a/packages/core/src/adapter/sqlite/snapshot-store.ts b/packages/core/src/adapter/sqlite/snapshot-store.ts index 29f91832..0cdf1daa 100644 --- a/packages/core/src/adapter/sqlite/snapshot-store.ts +++ b/packages/core/src/adapter/sqlite/snapshot-store.ts @@ -37,7 +37,7 @@ export const createSqliteSnapshotStore: CreateSqliteSnapshotStoreFunction = ({ // One statement, so a forget that removed a key before it runs leaves it nothing to write. put: ({ aggregateType, aggregateId, version, fingerprint, state, persons }) => db.run( - `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state", "persons") SELECT ?, ?, ?, ?, ?, ? WHERE ${heldKeys(personalKeys)} ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state", "persons" = excluded."persons"`, + `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state", "persons") SELECT ?, ?, ?, ?, ?, ? WHERE ${heldKeys(personalKeys)} ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state", "persons" = excluded."persons" WHERE excluded."version" >= ${table}."version" OR excluded."fingerprint" <> ${table}."fingerprint"`, [ aggregateType, aggregateId, @@ -46,7 +46,7 @@ export const createSqliteSnapshotStore: CreateSqliteSnapshotStoreFunction = ({ state, JSON.stringify(persons), JSON.stringify(persons), - persons.length, + JSON.stringify(persons), ], ), remove: ({ aggregateType, aggregateId }) => diff --git a/packages/core/src/adapter/storage/idempotency-store.ts b/packages/core/src/adapter/storage/idempotency-store.ts index 39f0734e..b07e07b3 100644 --- a/packages/core/src/adapter/storage/idempotency-store.ts +++ b/packages/core/src/adapter/storage/idempotency-store.ts @@ -47,9 +47,9 @@ export interface PurgeIdempotencyRecordsArgs { * the recording was part of. * * A record that names people lives only as long as their keys, as a snapshot does: `put` records - * nothing unless every person it names holds the key of the fingerprint it names them with, or - * any key for a `null` one, checked in the same statement, and `forget` removes every record - * that names a person. + * nothing unless every person it names with a fingerprint holds the key of that fingerprint (one + * named with `null` was read in the clear and needs none), checked in the same statement, and + * `forget` removes every record that names a person. */ export interface IdempotencyStore { /** diff --git a/packages/core/src/adapter/storage/snapshot-store.ts b/packages/core/src/adapter/storage/snapshot-store.ts index 20a03ec5..6e328c98 100644 --- a/packages/core/src/adapter/storage/snapshot-store.ts +++ b/packages/core/src/adapter/storage/snapshot-store.ts @@ -20,11 +20,12 @@ export interface Snapshot extends StreamIdentity { * instead. `get` is `null` for a stream without one. `remove` forgets a stream's snapshot. * * A snapshot holds personal data in the clear, so it lives only as long as its persons' keys: - * `put` writes nothing unless every person it names holds, in the storage's `personalKeys`, the - * key of the fingerprint it names them with, or any key for a `null` one, checked in the same - * statement; and `forget` removes every snapshot that names a person. A transaction that removes - * a key and then forgets its snapshots leaves none behind, whatever `put` runs beside it, and a - * key given to the person afterwards is never the one a snapshot was read under. + * `put` writes nothing unless every person it names with a fingerprint holds, in the storage's + * `personalKeys`, the key of that fingerprint, checked in the same statement (one named with + * `null` was read in the clear and needs none); and `forget` removes every snapshot that names a + * person. A transaction that removes a key and then forgets its snapshots leaves none behind, + * whatever `put` runs beside it, and a key given to the person afterwards is never the one a + * snapshot was read under. A `put` older than the snapshot of the same fingerprint is ignored. */ export interface SnapshotStore { get(stream: StreamIdentity): Promise; diff --git a/packages/core/src/adapter/testing/idempotency-store.contract.ts b/packages/core/src/adapter/testing/idempotency-store.contract.ts index 8e08172e..741259e7 100644 --- a/packages/core/src/adapter/testing/idempotency-store.contract.ts +++ b/packages/core/src/adapter/testing/idempotency-store.contract.ts @@ -46,7 +46,7 @@ export const idempotencyStoreContract: IdempotencyStoreContractFunction = ({ cre }; it("records a record that names people only while each of them has a key", async () => { - const persons = [held("c-1", "1"), held("c-2", null)]; + const persons = [held("c-1", "1"), held("c-2", "2"), held("c-3", null)]; const named = { ...record("k1"), outcome: rejection, persons, now }; await keys.add("c-1", stored("1")); await store.put(named); diff --git a/packages/core/src/adapter/testing/snapshot-store.contract.ts b/packages/core/src/adapter/testing/snapshot-store.contract.ts index 14ba563f..cb33e457 100644 --- a/packages/core/src/adapter/testing/snapshot-store.contract.ts +++ b/packages/core/src/adapter/testing/snapshot-store.contract.ts @@ -55,7 +55,7 @@ export const snapshotStoreContract: SnapshotStoreContractFunction = ({ create }) expect(await store.get({ aggregateType: "order", aggregateId: "43" })).toBeNull(); }); - it("replaces a stream's snapshot with the last one put", async () => { + it("replaces a stream's snapshot with the last one put of another fingerprint", async () => { await store.put(order42); const later: Snapshot = { ...order42, version: 400, fingerprint: "ghi", state: "[2]" }; await store.put(later); @@ -65,8 +65,21 @@ export const snapshotStoreContract: SnapshotStoreContractFunction = ({ create }) expect(await store.get({ aggregateType: "order", aggregateId: "42" })).toEqual(earlier); }); + it("keeps the newest snapshot of a fingerprint, whatever order they are put in", async () => { + const newer: Snapshot = { ...order42, version: 400, state: "[4]" }; + await store.put(newer); + await store.put({ ...order42, version: 300, state: "[3]" }); + expect(await store.get(order42)).toEqual(newer); + const again: Snapshot = { ...newer, state: "[4 again]" }; + await store.put(again); + expect(await store.get(order42)).toEqual(again); + const newest: Snapshot = { ...order42, version: 500, state: "[5]" }; + await store.put(newest); + expect(await store.get(order42)).toEqual(newest); + }); + it("writes a snapshot that names people only while each of them has a key", async () => { - const ada: Snapshot = { ...order42, persons: [held("c-1", "1"), held("c-2", null)] }; + const ada: Snapshot = { ...order42, persons: [held("c-1", "1"), held("c-2", "2")] }; await store.put(ada); expect(await store.get(order42)).toBeNull(); await keys.add("c-1", stored("1")); @@ -80,6 +93,22 @@ export const snapshotStoreContract: SnapshotStoreContractFunction = ({ create }) expect(await store.get(order42)).toEqual(ada); }); + it("needs no key for a person whose fields were read in the clear", async () => { + const clear: Snapshot = { ...order42, persons: [held("c-1", null), held("c-2", null)] }; + await store.put(clear); + expect(await store.get(order42)).toEqual(clear); + const mixed: Snapshot = { + ...order42, + version: 300, + persons: [held("c-1", null), held("c-2", "2")], + }; + await store.put(mixed); + expect(await store.get(order42)).toEqual(clear); + await keys.add("c-2", stored("2")); + await store.put(mixed); + expect(await store.get(order42)).toEqual(mixed); + }); + it("writes nothing for a person named with a key other than the one they hold", async () => { const named = (...persons: KeyedPerson[]): Snapshot => ({ ...order42, persons }); await keys.add("c-1", stored("new")); diff --git a/packages/core/src/contracts/command.ts b/packages/core/src/contracts/command.ts index bccae97a..24bd1bfe 100644 --- a/packages/core/src/contracts/command.ts +++ b/packages/core/src/contracts/command.ts @@ -36,9 +36,9 @@ export interface DispatchOptions { * Makes the dispatch safe to repeat: a key the caller chose for this one command, such as the * `Idempotency-Key` header of the request that carries it. The first dispatch under a key runs * the command and remembers what it answered, for `runtime.commands.idempotencyKeyLifetime`; - * every later one with the same key, the same command and the same payload answers the same - * without running it, its rejection included. The same key with another command or another - * payload rejects with `IdempotencyKeyConflictError`. A dispatch that fails is not remembered, + * every later one with the same key, the same command, the same payload and the same `delay` + * answers the same without running it, its rejection included. The same key with any of those + * changed rejects with `IdempotencyKeyConflictError`. A dispatch that fails is not remembered, * so a retry runs it. Between 1 and `MAX_IDEMPOTENCY_KEY_LENGTH` characters. */ readonly idempotencyKey?: string; diff --git a/packages/core/src/contracts/errors.ts b/packages/core/src/contracts/errors.ts index d1018307..886fc68f 100644 --- a/packages/core/src/contracts/errors.ts +++ b/packages/core/src/contracts/errors.ts @@ -104,9 +104,9 @@ export interface IdempotencyKeyConflictErrorArgs { /** * Thrown by a dispatch whose `idempotencyKey` was already used for another command, or for the - * same command with another payload, within `runtime.commands.idempotencyKeyLifetime`. Nothing - * runs and nothing is stored: a key names one command with one payload, so a second one under it - * is a bug of the caller, not a retry. + * same command with another payload or `delay`, within `runtime.commands.idempotencyKeyLifetime`. + * Nothing runs and nothing is stored: a key names one command with one payload, so a second one + * under it is a bug of the caller, not a retry. */ export class IdempotencyKeyConflictError extends BoundaError { readonly idempotencyKey: string; diff --git a/packages/core/src/kernel/command/fingerprint.test.ts b/packages/core/src/kernel/command/fingerprint.test.ts index 7ff7006b..69536472 100644 --- a/packages/core/src/kernel/command/fingerprint.test.ts +++ b/packages/core/src/kernel/command/fingerprint.test.ts @@ -52,4 +52,41 @@ describe("fingerprintOf", () => { await fingerprintOf({ type: "T", payload: {} }), ); }); + + it("tells payloads apart by what their maps, sets and bigints hold, in any order", async () => { + const of = (payload: unknown) => fingerprintOf({ type: "T", payload }); + expect(await of({ items: new Map([[1, "a"]]) })).not.toBe( + await of({ items: new Map([[2, "b"]]) }), + ); + expect( + await of({ + items: new Map([ + [1, "a"], + [2, { y: 1, x: 2 }], + ]), + }), + ).toBe( + await of({ + items: new Map([ + [2, { x: 2, y: 1 }], + [1, "a"], + ]), + }), + ); + expect(await of({ tags: new Set(["a"]) })).not.toBe(await of({ tags: new Set(["b"]) })); + expect(await of({ tags: new Set(["a", "b"]) })).toBe(await of({ tags: new Set(["b", "a"]) })); + expect(await of({ items: new Map() })).not.toBe(await of({ items: {} })); + expect(await of({ items: new Set() })).not.toBe(await of({ items: new Map() })); + expect(await of({ amount: 10n })).not.toBe(await of({ amount: 11n })); + expect(await of({ amount: 10n })).not.toBe(await of({ amount: "10" })); + }); + + it("tells a scheduled command from an immediate one, and delays apart", async () => { + const now = await fingerprintOf({ type: "T", payload: { a: 1 } }); + const later = await fingerprintOf({ type: "T", payload: { a: 1 }, delayMs: 60_000 }); + expect(later).not.toBe(now); + expect(await fingerprintOf({ type: "T", payload: { a: 1 }, delayMs: 60_000 })).toBe(later); + expect(await fingerprintOf({ type: "T", payload: { a: 1 }, delayMs: 0 })).not.toBe(now); + expect(await fingerprintOf({ type: "T", payload: { a: 1 }, delayMs: 1 })).not.toBe(later); + }); }); diff --git a/packages/core/src/kernel/command/fingerprint.ts b/packages/core/src/kernel/command/fingerprint.ts index 6d3bd0c8..7ca4a0e9 100644 --- a/packages/core/src/kernel/command/fingerprint.ts +++ b/packages/core/src/kernel/command/fingerprint.ts @@ -1,6 +1,9 @@ export interface FingerprintArgs { readonly type: string; readonly payload: unknown; + // Scheduled dispatches only: what `delay` came to, so a scheduled command and an immediate one + // never share a key. + readonly delayMs?: number | undefined; } export interface FingerprintFunction { @@ -8,11 +11,24 @@ export interface FingerprintFunction { } // JSON with the keys of every object in order, so two payloads that are the same object -// written in another order fingerprint alike. +// written in another order fingerprint alike. What JSON cannot write, a `Map`, a `Set` or a +// `bigint`, goes in tagged, so payloads that differ there never fingerprint alike. const canonical = (value: unknown): unknown => { + if (typeof value === "bigint") return { $bigint: value.toString() }; if (Array.isArray(value)) return value.map(canonical); if (typeof value !== "object" || value === null) return value; if (value instanceof Date) return value; + // Each entry as JSON, sorted by code unit, so the order they were added in does not count. + if (value instanceof Map) { + return { + $map: [...value] + .map(([key, entry]) => JSON.stringify([canonical(key), canonical(entry)])) + .sort(), + }; + } + if (value instanceof Set) { + return { $set: [...value].map((member) => JSON.stringify([canonical(member)])).sort() }; + } return Object.fromEntries( Object.keys(value as Record) .sort() @@ -21,12 +37,14 @@ const canonical = (value: unknown): unknown => { }; /** - * What identifies a command under an idempotency key: the SHA-256 of its type and payload, as - * JSON, in hex. `undefined` and a missing payload fingerprint alike, as JSON cannot tell them - * apart either. + * What identifies a command under an idempotency key: the SHA-256 of its type, its payload and + * its delay, if any, as JSON, in hex. `undefined` and a missing payload fingerprint alike, as JSON + * cannot tell them apart either. */ -export const fingerprintOf: FingerprintFunction = async ({ type, payload }) => { - const bytes = new TextEncoder().encode(JSON.stringify([type, canonical(payload) ?? null])); +export const fingerprintOf: FingerprintFunction = async ({ type, payload, delayMs }) => { + const identity = [type, canonical(payload) ?? null]; + if (delayMs !== undefined) identity.push({ delayMs }); + const bytes = new TextEncoder().encode(JSON.stringify(identity)); const digest = await crypto.subtle.digest("SHA-256", bytes); return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""); }; diff --git a/packages/core/src/kernel/command/idempotency.test.ts b/packages/core/src/kernel/command/idempotency.test.ts index 2f8c9ddf..1f217876 100644 --- a/packages/core/src/kernel/command/idempotency.test.ts +++ b/packages/core/src/kernel/command/idempotency.test.ts @@ -217,6 +217,21 @@ describe("a command dispatched with an idempotency key", () => { expect(await harness.storage.scheduler.list()).toHaveLength(1); }); + it("refuses the key of a delayed command for the same command now, or with another delay", async () => { + const harness = await createKernelHarness(); + await harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1", { delay: "5m" }) }); + await expect( + harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1") }), + ).rejects.toBeInstanceOf(IdempotencyKeyConflictError); + await expect( + harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1", { delay: "10m" }) }), + ).rejects.toBeInstanceOf(IdempotencyKeyConflictError); + expect( + await harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1", { delay: 300_000 }) }), + ).toMatchObject({ scheduled: true }); + expect(await storedTypes(harness)).toEqual([]); + }); + it("forgets a key once its lifetime has passed, and runs the command again", async () => { const harness = await createKernelHarness({ config: { runtime: { commands: { idempotencyKeyLifetime: "1h" } } }, diff --git a/packages/core/src/kernel/command/pipeline.ts b/packages/core/src/kernel/command/pipeline.ts index f3b20346..4cddfb16 100644 --- a/packages/core/src/kernel/command/pipeline.ts +++ b/packages/core/src/kernel/command/pipeline.ts @@ -441,7 +441,11 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ ? undefined : { key: checkIdempotencyKey(type, options.idempotencyKey), - fingerprint: await fingerprintOf({ type, payload }), + fingerprint: await fingerprintOf({ + type, + payload, + delayMs: options.delay === undefined ? undefined : parseDuration(options.delay), + }), }; const scheduled = options.delay !== undefined; diff --git a/packages/core/src/kernel/personal-data/personal-data.test.ts b/packages/core/src/kernel/personal-data/personal-data.test.ts index 4f8c9988..983659ce 100644 --- a/packages/core/src/kernel/personal-data/personal-data.test.ts +++ b/packages/core/src/kernel/personal-data/personal-data.test.ts @@ -360,6 +360,41 @@ describe("personal data in an app", () => { await app.stop(); }); + it("snapshots a stream whose personal fields were stored in the clear, before they were marked", async () => { + const adapter = memory(); + const storage = await adapter.createStorage({ logger: silentLogger }); + const metadata = { correlationId: "c", causationId: "c", commandId: "c", depth: 0 }; + await storage.eventStore.append({ + aggregateType: "customer", + aggregateId: "c-9", + expectedVersion: 0, + events: [ + { + id: "e-1", + aggregateType: "customer", + aggregateId: "c-9", + version: 1, + type: "CustomerRegistered", + payload: { email: "old@example.com", name: "Old" }, + timestamp: "2026-01-01T00:00:00.000Z", + metadata: { ...metadata, schemaVersion: 1, system: false }, + }, + ], + }); + const { app } = await createTestApp({ ...snapshotted, adapter }); + await app.commands.greetCustomer({ customerId: "c-9" }); + await app.runUntilIdle(); + const stream = { aggregateType: "customer", aggregateId: "c-9" }; + expect(await storage.personalKeys.get("c-9")).toBeNull(); + expect(await storage.snapshotStore.get(stream)).toMatchObject({ + version: 2, + persons: [{ person: "c-9", fingerprint: null }], + }); + await app.forget("c-9"); + expect(await storage.snapshotStore.get(stream)).toBeNull(); + await app.stop(); + }); + it("forgets the rejections it remembered for a person, so a retry decides again without their data", async () => { const adapter = memory(); const storage = await adapter.createStorage({ logger: silentLogger }); diff --git a/packages/core/src/memory/personal-key-store.ts b/packages/core/src/memory/personal-key-store.ts index f7905ecc..4a8fa046 100644 --- a/packages/core/src/memory/personal-key-store.ts +++ b/packages/core/src/memory/personal-key-store.ts @@ -7,7 +7,7 @@ import { createStoreEntries, type WithEntries } from "./entries.ts"; export interface KeptPersonalKeyStore extends WithEntries { /** - * Whether the person holds the key of that fingerprint, or any key for a `null` one, answered + * Whether the person holds the key of that fingerprint, always for a `null` one, answered * at once: the memory snapshot store checks it in the same step as it writes, so no forget runs * between the two. */ @@ -23,7 +23,7 @@ export const createKeptPersonalKeyStore: CreateKeptPersonalKeyStoreFunction = () return { holds: ({ person, fingerprint }) => { const stored = keys.get(person); - return stored !== undefined && (fingerprint === null || stored.fingerprint === fingerprint); + return fingerprint === null || stored?.fingerprint === fingerprint; }, store: { get: async (person) => keys.get(person) ?? null, diff --git a/packages/core/src/memory/snapshot-store.ts b/packages/core/src/memory/snapshot-store.ts index a489f5d6..04528367 100644 --- a/packages/core/src/memory/snapshot-store.ts +++ b/packages/core/src/memory/snapshot-store.ts @@ -19,7 +19,12 @@ export const createMemorySnapshotStore: CreateMemorySnapshotStoreFunction = ({ h return { get: async (stream) => snapshots.get(streamId(stream)) ?? null, put: async (snapshot) => { - if (snapshot.persons.every(holds)) snapshots.set(streamId(snapshot), snapshot); + const current = snapshots.get(streamId(snapshot)); + const older = + current !== undefined && + current.fingerprint === snapshot.fingerprint && + current.version > snapshot.version; + if (!older && snapshot.persons.every(holds)) snapshots.set(streamId(snapshot), snapshot); }, remove: async (stream) => { snapshots.delete(streamId(stream)); diff --git a/packages/postgresql/src/held-keys.ts b/packages/postgresql/src/held-keys.ts index 9c37b0fc..2d274ab4 100644 --- a/packages/postgresql/src/held-keys.ts +++ b/packages/postgresql/src/held-keys.ts @@ -9,13 +9,14 @@ export interface HeldKeysFunction { } /** - * A condition true when every person in `persons` holds, in `personalKeys`, the key of the - * fingerprint named with them, or any key for a `null` one. A person named under two keys never - * holds both. The keys it finds stay locked until the transaction ends: a forget that removes one - * waits for the write, and then finds the row to remove. + * A condition true when every person in `persons` named with a fingerprint holds, in + * `personalKeys`, the key of that fingerprint. A person named with `null` was read in the clear + * and needs no key. A person named under two keys never holds both. The keys it finds stay locked + * until the transaction ends: a forget that removes one waits for the write, and then finds the + * row to remove. */ export const heldKeys: HeldKeysFunction = ({ personalKeys, persons }) => - `(SELECT count(*) FROM (SELECT 1 FROM ${personalKeys} AS "k" JOIN jsonb_to_recordset(${persons}::jsonb) AS "p"("person" text, "fingerprint" text) ON "k"."person" = "p"."person" AND "k"."fingerprint" = coalesce("p"."fingerprint", "k"."fingerprint") FOR SHARE OF "k") AS "held") = jsonb_array_length(${persons}::jsonb)`; + `(SELECT count(*) FROM (SELECT 1 FROM ${personalKeys} AS "k" JOIN jsonb_to_recordset(${persons}::jsonb) AS "p"("person" text, "fingerprint" text) ON "k"."person" = "p"."person" AND "k"."fingerprint" = "p"."fingerprint" FOR SHARE OF "k") AS "held") = (SELECT count(*) FROM jsonb_to_recordset(${persons}::jsonb) AS "p"("person" text, "fingerprint" text) WHERE "p"."fingerprint" IS NOT NULL)`; export interface NamingFunction { (person: string): string; diff --git a/packages/postgresql/src/schema.ts b/packages/postgresql/src/schema.ts index a780d624..db4545c6 100644 --- a/packages/postgresql/src/schema.ts +++ b/packages/postgresql/src/schema.ts @@ -159,6 +159,9 @@ export const storageSchemaStatements: StorageSchemaStatementsFunction = (tables) "persons" jsonb NOT NULL, PRIMARY KEY ("aggregate_type", "aggregate_id") )`, + // So forgetting a person finds the rows that name them without reading every row. + `CREATE INDEX IF NOT EXISTS ${indexName(tables.snapshots, "persons")} ON ${tables.snapshots} USING gin ("persons" jsonb_path_ops)`, + `CREATE INDEX IF NOT EXISTS ${indexName(tables.idempotencyKeys, "persons")} ON ${tables.idempotencyKeys} USING gin ("persons" jsonb_path_ops)`, ]; export interface EnsureStorageSchemaArgs { @@ -190,6 +193,8 @@ export const ensureStorageSchema: EnsureStorageSchemaFunction = async ({ db, sch indexName(tables.deadLetters, "status"), indexName(tables.scheduledCommands, "execute_at"), indexName(tables.idempotencyKeys, "expires_at"), + indexName(tables.snapshots, "persons"), + indexName(tables.idempotencyKeys, "persons"), ].map(unquoted); if (await relationsExist({ db, schema, names })) return; await inSchema({ diff --git a/packages/postgresql/src/snapshot-store.ts b/packages/postgresql/src/snapshot-store.ts index 1dd3d44b..377ca230 100644 --- a/packages/postgresql/src/snapshot-store.ts +++ b/packages/postgresql/src/snapshot-store.ts @@ -36,7 +36,7 @@ export const createPostgresqlSnapshotStore: CreatePostgresqlSnapshotStoreFunctio }, put: ({ aggregateType, aggregateId, version, fingerprint, state, persons }) => db.run( - `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state", "persons") SELECT $1, $2, $3, $4, $5, $6::jsonb WHERE ${heldKeys({ personalKeys, persons: "$6" })} ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state", "persons" = excluded."persons"`, + `INSERT INTO ${table} ("aggregate_type", "aggregate_id", "version", "fingerprint", "state", "persons") SELECT $1, $2, $3, $4, $5, $6::jsonb WHERE ${heldKeys({ personalKeys, persons: "$6" })} ON CONFLICT ("aggregate_type", "aggregate_id") DO UPDATE SET "version" = excluded."version", "fingerprint" = excluded."fingerprint", "state" = excluded."state", "persons" = excluded."persons" WHERE excluded."version" >= ${table}."version" OR excluded."fingerprint" <> ${table}."fingerprint"`, [ aggregateType, aggregateId, diff --git a/packages/react-router/src/cloudflare/create-bounda.test.ts b/packages/react-router/src/cloudflare/create-bounda.test.ts index 95a1c65e..97729fe0 100644 --- a/packages/react-router/src/cloudflare/create-bounda.test.ts +++ b/packages/react-router/src/cloudflare/create-bounda.test.ts @@ -4,6 +4,7 @@ import { type BoundaClient, ConfigurationError, type Consistency, + idempotencyKeyFor, } from "@bounda-dev/core"; import { type RouterContext, RouterContextProvider } from "react-router"; import { afterEach, describe, expect, expectTypeOf, it } from "vitest"; @@ -55,7 +56,7 @@ describe("createBounda on Cloudflare", () => { ]); }); - it("sends the request's Idempotency-Key header with every command of the request", async () => { + it("keys every command of the request from its Idempotency-Key header", async () => { const store = storeNamespace(); env.STORE = store.namespace; const { bounda, boundaMiddleware } = createBounda({ @@ -75,7 +76,7 @@ describe("createBounda on Cloudflare", () => { "command", "placeOrder", { orderId: "o-1" }, - { idempotencyKey: "req-1", correlationId: "c-1" }, + { idempotencyKey: idempotencyKeyFor("req-1", "placeOrder"), correlationId: "c-1" }, "eventual", ], ["acme", "command", "payOrder", { orderId: "o-1" }, { idempotencyKey: "own" }, "eventual"], diff --git a/packages/react-router/src/create-bounda.test.ts b/packages/react-router/src/create-bounda.test.ts index 0883fab1..c9137605 100644 --- a/packages/react-router/src/create-bounda.test.ts +++ b/packages/react-router/src/create-bounda.test.ts @@ -1,4 +1,4 @@ -import { type BoundaApp, ConfigurationError } from "@bounda-dev/core"; +import { type BoundaApp, ConfigurationError, idempotencyKeyFor } from "@bounda-dev/core"; import { RouterContextProvider } from "react-router"; import { describe, expect, it } from "vitest"; import { createBounda } from "./create-bounda.ts"; @@ -121,7 +121,7 @@ describe("createBounda", () => { expect(second.context.get(bounda)).toBe(fake.app); }); - it("passes the request's Idempotency-Key header to the commands it serves", async () => { + it("keys the commands it serves from the request's Idempotency-Key header", async () => { const options: unknown[] = []; const app = { commands: { @@ -150,7 +150,11 @@ describe("createBounda", () => { await commandsOf(keyed.context).placeOrder({ orderId: "o-1" }); await commandsOf(keyed.context).placeOrder({ orderId: "o-1" }, { idempotencyKey: "own" }); await commandsOf(plain.context).placeOrder({ orderId: "o-1" }); - expect(options).toEqual([{ idempotencyKey: "req-1" }, { idempotencyKey: "own" }, undefined]); + expect(options).toEqual([ + { idempotencyKey: idempotencyKeyFor("req-1", "placeOrder") }, + { idempotencyKey: "own" }, + undefined, + ]); expect(plain.context.get(bounda)).toBe(app); }); diff --git a/packages/react-router/src/idempotency-key.test.ts b/packages/react-router/src/idempotency-key.test.ts index a1018212..5d723ad4 100644 --- a/packages/react-router/src/idempotency-key.test.ts +++ b/packages/react-router/src/idempotency-key.test.ts @@ -1,4 +1,4 @@ -import type { BoundaClient, DispatchOptions } from "@bounda-dev/core"; +import { type BoundaClient, type DispatchOptions, idempotencyKeyFor } from "@bounda-dev/core"; import { describe, expect, it } from "vitest"; import { IDEMPOTENCY_KEY_HEADER, withIdempotencyKey } from "./idempotency-key.ts"; @@ -23,7 +23,7 @@ const recording = () => { }; describe("withIdempotencyKey", () => { - it("hands the header to every command as its key, unless the caller passes one", async () => { + it("hands every command a key of its own from the header, unless the caller passes one", async () => { const { client, calls } = recording(); const served = withIdempotencyKey(client, request("req-1")); expect(served).not.toBe(client); @@ -40,11 +40,40 @@ describe("withIdempotencyKey", () => { await placeOrder({ orderId: "o-1" }, { idempotencyKey: undefined, } as unknown as DispatchOptions); + const placed = idempotencyKeyFor("req-1", "placeOrder"); expect(calls).toEqual([ - [{ orderId: "o-1" }, { idempotencyKey: "req-1" }], - [{ orderId: "o-1" }, { correlationId: "c-1", idempotencyKey: "req-1" }], + [{ orderId: "o-1" }, { idempotencyKey: placed }], + [{ orderId: "o-1" }, { correlationId: "c-1", idempotencyKey: placed }], [{ orderId: "o-1" }, { idempotencyKey: "mine" }], - [{ orderId: "o-1" }, { idempotencyKey: "req-1" }], + [{ orderId: "o-1" }, { idempotencyKey: placed }], + ]); + }); + + it("keys two commands of one request apart, and one command alike on every request", async () => { + const calls: unknown[][] = []; + const client = { + commands: { + reserveSeats: async (_payload?: unknown, options?: DispatchOptions) => + calls.push(["reserveSeats", options?.idempotencyKey]), + placeOrder: async (_payload?: unknown, options?: DispatchOptions) => + calls.push(["placeOrder", options?.idempotencyKey]), + }, + } as unknown as BoundaClient; + for (const attempt of [1, 2]) { + const commands = withIdempotencyKey(client, request("req-3")).commands as unknown as Record< + string, + (payload: unknown) => Promise + >; + await commands.reserveSeats?.({ attempt }); + await commands.placeOrder?.({ attempt }); + } + const [reserve, place] = calls.map(([, key]) => key); + expect(reserve).not.toBe(place); + expect(calls).toEqual([ + ["reserveSeats", reserve], + ["placeOrder", place], + ["reserveSeats", reserve], + ["placeOrder", place], ]); }); @@ -75,7 +104,9 @@ describe("withIdempotencyKey", () => { await ( served.commands as unknown as Record Promise> ).anything?.({ id: "1" }); - expect(calls).toEqual([["anything", { id: "1" }, { idempotencyKey: "req-2" }]]); + expect(calls).toEqual([ + ["anything", { id: "1" }, { idempotencyKey: idempotencyKeyFor("req-2", "anything") }], + ]); expect((served.commands as unknown as { then?: unknown }).then).toBeUndefined(); }); }); diff --git a/packages/react-router/src/idempotency-key.ts b/packages/react-router/src/idempotency-key.ts index 91e3b37d..ccb9c98e 100644 --- a/packages/react-router/src/idempotency-key.ts +++ b/packages/react-router/src/idempotency-key.ts @@ -1,4 +1,9 @@ -import type { BoundaClient, DispatchOptions, Registry } from "@bounda-dev/core"; +import { + type BoundaClient, + type DispatchOptions, + idempotencyKeyFor, + type Registry, +} from "@bounda-dev/core"; export const IDEMPOTENCY_KEY_HEADER = "Idempotency-Key"; @@ -6,20 +11,22 @@ export interface WithIdempotencyKeyFunction { (client: BoundaClient, request: Request): BoundaClient; } -// The client whose commands carry the request's `Idempotency-Key` header as their -// `idempotencyKey`, unless the caller passes one of its own; the client as it is for a request -// without the header. +// The client whose commands carry a key derived from the request's `Idempotency-Key` header and +// the command's name as their `idempotencyKey`, unless the caller passes one of its own, so an +// action that dispatches two commands records each under its own key; the client as it is for a +// request without the header. An empty header is passed on as it is, for the dispatch to refuse. export const withIdempotencyKey: WithIdempotencyKeyFunction = ( client: BoundaClient, request: Request, ): BoundaClient => { - const idempotencyKey = request.headers.get(IDEMPOTENCY_KEY_HEADER); - if (idempotencyKey === null) return client; + const header = request.headers.get(IDEMPOTENCY_KEY_HEADER); + if (header === null) return client; // A proxy rather than a copy: on Cloudflare the commands are a proxy with no keys to copy. const commands = new Proxy(client.commands as object, { get: (target, key, receiver) => { const member: unknown = Reflect.get(target, key, receiver); if (typeof member !== "function") return member; + const idempotencyKey = header === "" ? header : idempotencyKeyFor(header, String(key)); return (payload?: unknown, options?: DispatchOptions) => Reflect.apply(member, target, [ payload, From c0f43b44d3d9f1d4c646e56034256dcf72777308 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Sat, 31 Oct 2026 11:55:48 +0100 Subject: [PATCH 2/2] fix(core): keep a rejection's message out of the runtime's log, as it may quote personal data --- docs/src/content/docs/guides/reacting-to-events.md | 5 +++-- packages/core/src/kernel/command/pipeline.ts | 4 +++- packages/core/src/kernel/policy/runner.test.ts | 1 - packages/core/src/kernel/scheduler/worker.test.ts | 8 +++----- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/docs/src/content/docs/guides/reacting-to-events.md b/docs/src/content/docs/guides/reacting-to-events.md index f7febd20..b5cc554a 100644 --- a/docs/src/content/docs/guides/reacting-to-events.md +++ b/docs/src/content/docs/guides/reacting-to-events.md @@ -261,8 +261,9 @@ if (paid.rejected === "NotOpen") { } ``` -A rejection nobody looks at is logged (`command rejected`, at `info`) and recorded on the command's -span; a test asserts the ones it expects from what +A rejection nobody looks at is logged (`command rejected`, at `info`, with its code and the +command's type and aggregate but not its message, which may quote personal data) and recorded on +the command's span; a test asserts the ones it expects from what [`runUntilIdle()`](/guides/testing/#rejections-inside-reactions) returns. A scheduled command that is rejected when it runs changes nothing in the same way. While the run lasts, only a failure rejects the `await`: a payload that does not validate, a concurrency conflict that outlasts its diff --git a/packages/core/src/kernel/command/pipeline.ts b/packages/core/src/kernel/command/pipeline.ts index 4cddfb16..5e04e782 100644 --- a/packages/core/src/kernel/command/pipeline.ts +++ b/packages/core/src/kernel/command/pipeline.ts @@ -657,7 +657,9 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ }, dispatchUnattended: ({ within, ...args }) => send(args, within, (_error, rejection) => { - logger.info("command rejected", { ...rejection }); + // Without its message, which may quote personal data from the state, past any forget. + const { message: _message, ...logged } = rejection; + logger.info("command rejected", logged); within.afterCommit(() => onRejection?.(rejection)); const { type: _type, ...answer } = rejection; return answer; diff --git a/packages/core/src/kernel/policy/runner.test.ts b/packages/core/src/kernel/policy/runner.test.ts index 22d6f0a9..22d2c7d1 100644 --- a/packages/core/src/kernel/policy/runner.test.ts +++ b/packages/core/src/kernel/policy/runner.test.ts @@ -718,7 +718,6 @@ describe("the commands of a policy run", () => { fields: { type: "PlaceOrder", rejected: "AlreadyPlaced", - message: "Order already placed", aggregateType: "order", aggregateId: "o-1", }, diff --git a/packages/core/src/kernel/scheduler/worker.test.ts b/packages/core/src/kernel/scheduler/worker.test.ts index acf2f44c..9d11d4a4 100644 --- a/packages/core/src/kernel/scheduler/worker.test.ts +++ b/packages/core/src/kernel/scheduler/worker.test.ts @@ -963,11 +963,9 @@ describe("scheduled command payload", () => { aggregateId: "n-1", }; expect(rejections).toEqual([rejection]); - expect(entries).toContainEqual({ - level: "info", - message: "command rejected", - fields: rejection, - }); + // The log leaves the message out: it may quote personal data. + const { message: _message, ...logged } = rejection; + expect(entries).toContainEqual({ level: "info", message: "command rejected", fields: logged }); expect(await app.deadLetters.list()).toEqual([]); expect((await app.runUntilIdle()).rejections).toEqual([]); await app.stop();