From 0848e3e250b21d41ce0f589e48bef59a8d00049e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Sun, 25 Oct 2026 00:58:56 +0200 Subject: [PATCH 1/2] feat(core): remember what a command dispatched with an idempotency key answered A request that times out on the way back, a form submitted twice or a webhook delivered again dispatches its command twice, and the aggregate either rejects a retry of something that went through or does its work twice. A dispatch with idempotencyKey records what it answered, in the transaction of the events it stores, and every later dispatch under the same key, command and payload answers the same without running, its rejection included; another command or payload under the key is refused. Keys live 24 hours by default and the scheduled-command worker purges them. The React Router middleware passes the Idempotency-Key header to the commands of the request. The in-memory adapter now reserves positions when a transaction stages events and shows readers only the contiguous committed prefix, so a transaction answers the positions its events commit with, as the SQL adapters do. --- .changeset/idempotent-commands.md | 18 + docs/src/content/docs/adapters/index.mdx | 1 + .../docs/concepts/delivery-guarantees.md | 2 + .../docs/guides/calling-the-outside-world.md | 4 +- docs/src/content/docs/guides/deployment.mdx | 2 +- .../docs/guides/idempotent-commands.md | 112 ++++++ docs/src/content/docs/guides/react-router.md | 16 +- .../content/docs/guides/reading-the-stream.md | 2 +- .../content/docs/reference/configuration.md | 4 + docs/src/content/docs/reference/errors.md | 8 + .../content/docs/reference/observability.md | 4 +- .../cloudflare/test/storage-contracts.test.ts | 2 + packages/core/src/adapter/adapter.ts | 5 +- packages/core/src/adapter/deferred-writes.ts | 26 +- packages/core/src/adapter/index.ts | 8 + packages/core/src/adapter/sqlite/adapter.ts | 5 + .../src/adapter/sqlite/idempotency-store.ts | 53 +++ packages/core/src/adapter/sqlite/schema.ts | 9 + .../core/src/adapter/sqlite/sqlite.test.ts | 3 + .../core/src/adapter/staged-event-store.ts | 30 +- .../src/adapter/storage/idempotency-store.ts | 59 +++ packages/core/src/adapter/testing/fixtures.ts | 19 +- .../testing/idempotency-store.contract.ts | 100 ++++++ packages/core/src/adapter/testing/index.ts | 8 +- .../testing/storage-transaction.contract.ts | 38 +- packages/core/src/config/defaults.ts | 1 + packages/core/src/config/schema.test.ts | 10 +- packages/core/src/config/schema.ts | 10 +- packages/core/src/config/types.ts | 11 +- packages/core/src/contracts/command.ts | 15 + packages/core/src/contracts/errors.ts | 40 +++ packages/core/src/contracts/index.ts | 4 + packages/core/src/kernel/app.test.ts | 16 + packages/core/src/kernel/app.ts | 3 +- .../src/kernel/command/fingerprint.test.ts | 55 +++ .../core/src/kernel/command/fingerprint.ts | 32 ++ .../src/kernel/command/idempotency.test.ts | 340 ++++++++++++++++++ packages/core/src/kernel/command/pipeline.ts | 265 ++++++++++---- .../core/src/kernel/process/parked.test.ts | 32 +- packages/core/src/kernel/reactive-harness.ts | 3 +- .../core/src/kernel/scheduler/worker.test.ts | 49 +++ packages/core/src/kernel/scheduler/worker.ts | 21 +- packages/core/src/kernel/shared/retry.ts | 1 + packages/core/src/kernel/test-support.ts | 3 +- .../src/kernel/unit-of-work/unit-of-work.ts | 5 +- packages/core/src/memory/event-store.ts | 102 ++++-- packages/core/src/memory/idempotency-store.ts | 51 +++ packages/core/src/memory/index.ts | 4 + packages/core/src/memory/memory.test.ts | 156 ++++++++ packages/core/src/memory/transaction.ts | 57 ++- packages/core/test-types/config.test-d.ts | 21 ++ packages/core/test-types/inference.test-d.ts | 7 + packages/postgresql/src/idempotency-store.ts | 60 ++++ packages/postgresql/src/index.ts | 5 + packages/postgresql/src/postgresql.test.ts | 3 + packages/postgresql/src/schema.ts | 11 + .../src/cloudflare/create-bounda.test.ts | 27 ++ packages/react-router/src/cloudflare/serve.ts | 3 +- .../react-router/src/create-bounda.test.ts | 33 ++ packages/react-router/src/create-bounda.ts | 8 +- packages/react-router/src/failure.test.ts | 13 + packages/react-router/src/failure.ts | 6 +- .../react-router/src/idempotency-key.test.ts | 81 +++++ packages/react-router/src/idempotency-key.ts | 31 ++ packages/sqlite/src/sqlite.test.ts | 1 + 65 files changed, 1973 insertions(+), 161 deletions(-) create mode 100644 .changeset/idempotent-commands.md create mode 100644 docs/src/content/docs/guides/idempotent-commands.md create mode 100644 packages/core/src/adapter/sqlite/idempotency-store.ts create mode 100644 packages/core/src/adapter/storage/idempotency-store.ts create mode 100644 packages/core/src/adapter/testing/idempotency-store.contract.ts create mode 100644 packages/core/src/kernel/command/fingerprint.test.ts create mode 100644 packages/core/src/kernel/command/fingerprint.ts create mode 100644 packages/core/src/kernel/command/idempotency.test.ts create mode 100644 packages/core/src/memory/idempotency-store.ts create mode 100644 packages/postgresql/src/idempotency-store.ts create mode 100644 packages/react-router/src/idempotency-key.test.ts create mode 100644 packages/react-router/src/idempotency-key.ts diff --git a/.changeset/idempotent-commands.md b/.changeset/idempotent-commands.md new file mode 100644 index 00000000..da3b74e1 --- /dev/null +++ b/.changeset/idempotent-commands.md @@ -0,0 +1,18 @@ +--- +"@bounda-dev/core": minor +"@bounda-dev/postgresql": minor +"@bounda-dev/react-router": minor +--- + +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` +(`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 +with, as the SQL adapters do. diff --git a/docs/src/content/docs/adapters/index.mdx b/docs/src/content/docs/adapters/index.mdx index 127abac5..ba2ca2e4 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_idempotency_keys` | What a command dispatched with an `idempotencyKey` answered, until the key expires | 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/delivery-guarantees.md b/docs/src/content/docs/concepts/delivery-guarantees.md index db65078c..7716aa1c 100644 --- a/docs/src/content/docs/concepts/delivery-guarantees.md +++ b/docs/src/content/docs/concepts/delivery-guarantees.md @@ -139,6 +139,8 @@ framework that is precise about it lands on the same line. Bounda draws it at th - [What the runtime promises](/guides/reacting-to-events/#what-the-runtime-promises), the same guarantees in the terms of a handler's code, and [retries and timeouts](/guides/reacting-to-events/#retries-and-timeouts). +- [Idempotent commands](/guides/idempotent-commands/), the same question at the door: a command + a client sends twice is answered once, under the key it names. - [What more instances do](/concepts/how-it-runs/#what-more-instances-do) and [a projection that keeps failing](/guides/deployment/#a-projection-that-keeps-failing). - [Your event store is your outbox](/concepts/event-store-as-outbox/) and diff --git a/docs/src/content/docs/guides/calling-the-outside-world.md b/docs/src/content/docs/guides/calling-the-outside-world.md index c19c7eb7..c2735a70 100644 --- a/docs/src/content/docs/guides/calling-the-outside-world.md +++ b/docs/src/content/docs/guides/calling-the-outside-world.md @@ -21,7 +21,9 @@ event, so the history says what the decision was based on. The handler receives `idempotencyKey`, the command's id, which stays the same across those runs (a scheduled command keeps the id it was scheduled with): pass it to a call the provider deduplicates, -such as creating that payment intent, so a second run does not create another. It also receives +such as creating that payment intent, so a second run does not create another. It is not the +`idempotencyKey` option of a dispatch, which keeps a command a client sends twice from running +twice ([idempotent commands](/guides/idempotent-commands/)). It also receives `signal`, which aborts when the run passes `runtime.commands.timeout`; pass it to what it calls outside (`fetch(url, { signal })`), so a provider that hangs fails the command instead of holding the request (see [Retries and timeouts](/guides/reacting-to-events/#retries-and-timeouts)). diff --git a/docs/src/content/docs/guides/deployment.mdx b/docs/src/content/docs/guides/deployment.mdx index 9928d00a..c2428912 100644 --- a/docs/src/content/docs/guides/deployment.mdx +++ b/docs/src/content/docs/guides/deployment.mdx @@ -2,7 +2,7 @@ title: Deployment description: Roles, one database, many instances, rebuilds and failing projections, tuning and observability. sidebar: - order: 9 + order: 10 --- import { Steps } from "@astrojs/starlight/components"; diff --git a/docs/src/content/docs/guides/idempotent-commands.md b/docs/src/content/docs/guides/idempotent-commands.md new file mode 100644 index 00000000..bb1b0ebc --- /dev/null +++ b/docs/src/content/docs/guides/idempotent-commands.md @@ -0,0 +1,112 @@ +--- +title: Idempotent commands +description: Let a client send a command again without running it twice, with the idempotencyKey option and the Idempotency-Key header. +sidebar: + order: 9 +--- + +A command can reach the app twice: the client retried after a timeout, the user clicked twice, a +webhook was delivered again. The aggregate rejects what it must (`placeOrder` again is +`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. + +## The `idempotencyKey` option + +```ts +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 +`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 +that failed, with a `ConcurrencyError` after its retries, a `HANDLER_TIMEOUT` or an error of the +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 +`IdempotencyKeyConflictError` (`IDEMPOTENCY_KEY_CONFLICT`) and runs nothing. The payload is +compared as JSON, keys in any order. + +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 +it. It is remembered for `runtime.commands.idempotencyKeyLifetime`, 24 hours by default; past it +the key is free for another command, and the scheduled-command worker purges the record from +`bounda_idempotency_keys` ([configuration](/reference/configuration/#commands)). + +This is not the `idempotencyKey` a command handler receives: that one is the command's own id, +for the provider the handler calls, the same across the runs a concurrency conflict causes +([calling the outside world](/guides/calling-the-outside-world/#deciding-then-acting)). The +option is for whoever dispatches the command, and reaches the handler as `command.metadata` +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: + +```tsx +// app/routes/register.tsx +export const action = async ({ request, context }: Route.ActionArgs) => { + const form = await request.formData(); + try { + // Carries the request's Idempotency-Key, when the client sent one. + await context.get(bounda).commands.registerUser({ userId: String(form.get("userId")) }); + } catch (error) { + return failure(error); + } + return redirect("/users"); +}; +``` + +The client makes the key when the user submits and sends it again when it retries, so a retry +answers as the first attempt did. With a fetcher: + +```ts +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. + +```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")); +``` + +`failure()` answers a conflict with a 422 ([errors from the domain](/guides/react-router/#errors-from-the-domain)). +The host never makes keys up: a request without the header dispatches as before. + +## On Cloudflare + +`connect(stub)` sends the option to the Durable Object with the command, where it works as +above; React Router on Cloudflare reads the header the same way. The JSON API of `createWorker` +takes `delay` and `correlationId` from the query string and no key. + +## In a policy or a process + +The commands a policy or process dispatches get the same ids on every retry of the run, so a +scheduled one is stored once and a provider sees one key +([what the runtime promises](/guides/reacting-to-events/#what-the-runtime-promises)); they rarely +need an `idempotencyKey` of their own. One they pass is recorded with the run's unit of work and +answered from it on a later run. + +## Elsewhere + +[Stripe's `Idempotency-Key`](https://docs.stripe.com/api/idempotent_requests) remembers the first +response for 24 hours and refuses the key with other parameters, which is where the lifetime and +the fingerprint come from; the IETF's +[Idempotency-Key header](https://datatracker.ietf.org/doc/draft-ietf-httpapi-idempotency-key-header/) +draft names the header and the 422 for a mismatched payload. diff --git a/docs/src/content/docs/guides/react-router.md b/docs/src/content/docs/guides/react-router.md index be7ece8a..d80b720c 100644 --- a/docs/src/content/docs/guides/react-router.md +++ b/docs/src/content/docs/guides/react-router.md @@ -106,9 +106,11 @@ 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 | -`failure` goes by the error's `code`, `VALIDATION_FAILED` or `DOMAIN_ERROR`, so it answers the -same for the refusals a Durable Object sends back on Cloudflare, which arrive as plain errors. +`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 +Cloudflare, which arrive as plain errors. Anything else is rethrown and reaches the route's `ErrorBoundary`. `Failure`, the shape of that data, is exported from `@bounda-dev/react-router`. @@ -129,6 +131,16 @@ The option takes any signal, such as `AbortSignal.timeout(2_000)` for one action sooner than the app's limit. On Cloudflare it only counts before the call leaves the Worker: RPC 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 +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/). + ## Reading what you just wrote Read models are updated by projections that run in the background, so a page reached right after diff --git a/docs/src/content/docs/guides/reading-the-stream.md b/docs/src/content/docs/guides/reading-the-stream.md index 642d73c1..c732fe54 100644 --- a/docs/src/content/docs/guides/reading-the-stream.md +++ b/docs/src/content/docs/guides/reading-the-stream.md @@ -2,7 +2,7 @@ title: Reading the global stream description: Read a store's events by position, as they were stored, for an operator's console, an audit trail or a page that follows the stream. sidebar: - order: 10 + order: 11 --- Every event a store commits takes the next position in its diff --git a/docs/src/content/docs/reference/configuration.md b/docs/src/content/docs/reference/configuration.md index b2737c86..0057515f 100644 --- a/docs/src/content/docs/reference/configuration.md +++ b/docs/src/content/docs/reference/configuration.md @@ -51,6 +51,10 @@ types refuse; wrap it with `asDuration(process.env.X ?? "1m")`, which checks it. - **`commands.concurrencyRetries`**, default `3`. How many times a command runs again when its append finds the stream moved, before a `ConcurrencyError` reaches the caller. Each run gets its own `timeout`. +- **`commands.idempotencyKeyLifetime`**, default `"24h"`. How long a command dispatched with an + `idempotencyKey` is remembered, so a dispatch with the same key answers the same. Past it the + key is free again, and the scheduled-command worker purges the record. See + [idempotent commands](/guides/idempotent-commands/). ### Policies diff --git a/docs/src/content/docs/reference/errors.md b/docs/src/content/docs/reference/errors.md index 262f6a9d..84f52340 100644 --- a/docs/src/content/docs/reference/errors.md +++ b/docs/src/content/docs/reference/errors.md @@ -52,6 +52,10 @@ dead letter with `errorType: "retriable_exhausted"`. the handler runs. - **`CHAIN_DEPTH_EXCEEDED`** (`ChainDepthExceededError`; terminal). A chain of reactions went deeper 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`. + 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 only exports `begin` goes on an aggregate that exists. Nothing is stored. @@ -65,6 +69,9 @@ dead letter with `errorType: "retriable_exhausted"`. that counts. - **`SCHEDULED_CLAIM_LOST`** (`ScheduledClaimLostError`; handled by the runtime). The same for a scheduled command, a delayed policy run or a deadline, or the command was cancelled. +- **`IDEMPOTENCY_KEY_TAKEN`** (`IdempotencyKeyTakenError`; handled by the runtime). Of two + dispatches racing for one `idempotencyKey`, the one that lost: what it stored rolls back, and it + answers what the key holds. - **`DEAD_LETTER_SETTLED`** (`DeadLetterSettledError`; an operator's answer). A dead letter was retried or discarded after another retry or discard had settled it. - **`DEAD_LETTER_NOT_RETRIABLE`** (`DeadLetterNotRetriableError`; an operator's answer). A dead @@ -84,6 +91,7 @@ The hosts map the codes a client can cause to a status: | --- | --- | --- | | `VALIDATION_FAILED` | 400, with `issues` | 400, with `issues` | | `DOMAIN_ERROR` | 409, with `rejected` | 409, with `rejected` | +| `IDEMPOTENCY_KEY_CONFLICT` | 500, code `INTERNAL`, logged | 422 | | `CONCURRENCY_CONFLICT` | 409 | rethrown | | `CHAIN_DEPTH_EXCEEDED` | 409 | rethrown | | `FOREIGN_REJECTION` | 500, code `INTERNAL`, logged | rethrown | diff --git a/docs/src/content/docs/reference/observability.md b/docs/src/content/docs/reference/observability.md index 8f9e9888..e7d0e61b 100644 --- a/docs/src/content/docs/reference/observability.md +++ b/docs/src/content/docs/reference/observability.md @@ -24,7 +24,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.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`, or `replayed` for a dispatch answered from its `idempotencyKey`), `bounda.event.count`; a rejection adds the event `bounda.command.rejected`, its code in `bounda.rejected`, replayed or not | | `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.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` | @@ -44,7 +44,7 @@ Metrics: | Metric | Kind | Attributes | | --- | --- | --- | | `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.commands` | counter | `bounda.command.type`, `bounda.outcome` (`stored`, `scheduled`, `rejected`, `replayed`, `failed`) | | `bounda.dead_letters` | counter | `bounda.handler.kind`, `bounda.handler`, `bounda.outcome` (`terminal`, `retriable_exhausted`) | The lag gauge is what to alert on: a subscriber whose lag grows is a projection or a policy that is diff --git a/packages/cloudflare/test/storage-contracts.test.ts b/packages/cloudflare/test/storage-contracts.test.ts index 62b72331..8d1ccfdd 100644 --- a/packages/cloudflare/test/storage-contracts.test.ts +++ b/packages/cloudflare/test/storage-contracts.test.ts @@ -8,6 +8,7 @@ import { contractFields, deadLetterStoreContract, eventStoreContract, + idempotencyStoreContract, inboxLedgerContract, readModelRebuildContract, readModelTransactionContract, @@ -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 }); + idempotencyStoreContract({ create: async () => (await freshStorage()).idempotencyStore }); storageTransactionContract({ create: freshStorage }); tableContract({ create: async () => diff --git a/packages/core/src/adapter/adapter.ts b/packages/core/src/adapter/adapter.ts index f987cff8..5a04fb33 100644 --- a/packages/core/src/adapter/adapter.ts +++ b/packages/core/src/adapter/adapter.ts @@ -5,6 +5,7 @@ import type { CheckpointStore } from "./storage/checkpoint-store.ts"; import type { DeadLetterStore } from "./storage/dead-letter-store.ts"; import type { EventNotifier } from "./storage/event-notifier.ts"; import type { EventStore } from "./storage/event-store.ts"; +import type { IdempotencyStore } from "./storage/idempotency-store.ts"; import type { InboxLedger } from "./storage/inbox-ledger.ts"; import type { Scheduler } from "./storage/scheduler.ts"; import type { ReadClient, Table } from "./storage/table.ts"; @@ -13,7 +14,7 @@ import type { ReadClient, Table } from "./storage/table.ts"; * The write-side stores bound to one open transaction: what `Storage.transact` runs its * work with. What the work writes through them commits together when it resolves and rolls * back together when it throws. `eventStore.load` sees the events the work appended; what the - * other three read back while the transaction is open is the store as committed, or the + * other four read back while the transaction is open is the store as committed, or the * transaction's own writes, depending on the adapter, so the work must not depend on it. */ export interface StorageTransaction { @@ -21,6 +22,7 @@ export interface StorageTransaction { readonly inboxLedger: InboxLedger; readonly deadLetterStore: DeadLetterStore; readonly scheduler: Scheduler; + readonly idempotencyStore: IdempotencyStore; } /** @@ -32,6 +34,7 @@ export interface Storage { readonly inboxLedger: InboxLedger; readonly deadLetterStore: DeadLetterStore; readonly scheduler: Scheduler; + readonly idempotencyStore: IdempotencyStore; /** * Only for a backend that can push new events; without it the dispatcher only polls. */ diff --git a/packages/core/src/adapter/deferred-writes.ts b/packages/core/src/adapter/deferred-writes.ts index fc997bc9..89f19eda 100644 --- a/packages/core/src/adapter/deferred-writes.ts +++ b/packages/core/src/adapter/deferred-writes.ts @@ -1,11 +1,12 @@ import type { StorageTransaction } from "./adapter.ts"; +import type { IdempotencyRecord } from "./storage/idempotency-store.ts"; /** * The write-side stores besides the event store. */ export type DeferredStores = Pick< StorageTransaction, - "inboxLedger" | "deadLetterStore" | "scheduler" + "inboxLedger" | "deadLetterStore" | "scheduler" | "idempotencyStore" >; export interface DeferredWriteTarget { @@ -18,14 +19,16 @@ export interface DeferredWriteTarget { DeferredStores["scheduler"], "schedule" | "cancel" | "complete" | "fail" | "defer" >; + readonly idempotencyStore: Pick; } /** * The stores with their writes held back: `stores` answer reads from `live` at once and record * every write in order, and `flush` runs the writes recorded so far against another set of the * same stores, a transaction's or the live ones. `add` answers the letter as filed, whatever the - * store holds under that id already. `tryClaim`, `claimDue` and both `renew` act at once, on - * `live`, since their answer decides what happens next. + * store holds under that id already; `get` on the idempotency store answers a record `put` + * recorded before anything `live` holds. `tryClaim`, `claimDue`, both `renew` and `purge` act at + * once, on `live`, since their answer decides what happens next. */ export interface DeferredWrites { readonly stores: DeferredStores; @@ -42,8 +45,14 @@ export interface DeferWritesFunction { type Recorded = (target: DeferredWriteTarget) => Promise; -export const deferWrites: DeferWritesFunction = ({ inboxLedger, deadLetterStore, scheduler }) => { +export const deferWrites: DeferWritesFunction = ({ + inboxLedger, + deadLetterStore, + scheduler, + idempotencyStore, +}) => { const recorded: Recorded[] = []; + const records = new Map(); const later = ( write: (target: DeferredWriteTarget, ...args: Args) => Promise, @@ -84,6 +93,15 @@ export const deferWrites: DeferWritesFunction = ({ inboxLedger, deadLetterStore, defer: later((target, args) => target.scheduler.defer(args)), renew: scheduler.renew, }, + idempotencyStore: { + get: async (args) => records.get(args.key) ?? idempotencyStore.get(args), + put: async (args) => { + const { now: _now, ...record } = args; + records.set(record.key, record); + recorded.push((target) => target.idempotencyStore.put(args)); + }, + purge: idempotencyStore.purge, + }, }, flush: async (target) => { for (const write of recorded) await write(target); diff --git a/packages/core/src/adapter/index.ts b/packages/core/src/adapter/index.ts index 9fd61312..9413b0e9 100644 --- a/packages/core/src/adapter/index.ts +++ b/packages/core/src/adapter/index.ts @@ -35,6 +35,14 @@ export type { PendingEvent, ReadAllArgs, } from "./storage/event-store.ts"; +export type { + DispatchOutcome, + GetIdempotencyRecordArgs, + IdempotencyRecord, + IdempotencyStore, + PurgeIdempotencyRecordsArgs, + PutIdempotencyRecordArgs, +} from "./storage/idempotency-store.ts"; export type { ClaimArgs, ClaimKey, diff --git a/packages/core/src/adapter/sqlite/adapter.ts b/packages/core/src/adapter/sqlite/adapter.ts index 979d9a0a..76816180 100644 --- a/packages/core/src/adapter/sqlite/adapter.ts +++ b/packages/core/src/adapter/sqlite/adapter.ts @@ -8,6 +8,7 @@ import type { SqlDatabase, SqlTransaction } from "../sql/database.ts"; import { createSqliteCheckpointStore } from "./checkpoint-store.ts"; import { createSqliteDeadLetterStore } from "./dead-letter-store.ts"; import { createSqliteEventStore } from "./event-store.ts"; +import { createSqliteIdempotencyStore } from "./idempotency-store.ts"; import { createSqliteInboxLedger } from "./inbox-ledger.ts"; import { openSqliteReadModel, rebuildSqliteReadModel } from "./read-model.ts"; import { createSqliteScheduler } from "./scheduler.ts"; @@ -87,6 +88,10 @@ export const createSqliteAdapter: CreateSqliteAdapterFunction = < inboxLedger: createSqliteInboxLedger({ db: database, table: tables.inbox }), deadLetterStore: createSqliteDeadLetterStore({ db: database, table: tables.deadLetters }), scheduler: createSqliteScheduler({ db: database, table: tables.scheduledCommands }), + idempotencyStore: createSqliteIdempotencyStore({ + db: database, + table: tables.idempotencyKeys, + }), }); // The stores over an open transaction: their statements join it, and a `write` of their own // runs inside it instead of opening another, which SQLite would refuse. diff --git a/packages/core/src/adapter/sqlite/idempotency-store.ts b/packages/core/src/adapter/sqlite/idempotency-store.ts new file mode 100644 index 00000000..17d84a41 --- /dev/null +++ b/packages/core/src/adapter/sqlite/idempotency-store.ts @@ -0,0 +1,53 @@ +import { IdempotencyKeyTakenError } from "../../contracts/errors.ts"; +import type { DispatchOutcome, IdempotencyRecord, IdempotencyStore } from "../index.ts"; +import type { SqlDatabase } from "../sql/database.ts"; + +export interface CreateSqliteIdempotencyStoreArgs { + readonly db: SqlDatabase; + readonly table: string; +} + +export interface CreateSqliteIdempotencyStoreFunction { + (args: CreateSqliteIdempotencyStoreArgs): IdempotencyStore; +} + +const toRecord = (row: Record): IdempotencyRecord => ({ + key: String(row.key), + fingerprint: String(row.fingerprint), + outcome: JSON.parse(String(row.outcome)) as DispatchOutcome, + expiresAt: String(row.expires_at), +}); + +export const createSqliteIdempotencyStore: CreateSqliteIdempotencyStoreFunction = ({ + db, + table, +}) => ({ + get: async ({ key, now }) => { + const [row] = await db.all( + `SELECT "key", "fingerprint", "outcome", "expires_at" FROM ${table} WHERE "key" = ? AND "expires_at" > ?`, + [key, now.toISOString()], + ); + return row === undefined ? null : toRecord(row); + }, + // One statement whose row comes back only when the key was free or expired, so two dispatches + // racing for it can never both record. + put: async ({ key, fingerprint, outcome, expiresAt, now }) => { + const rows = await db.all( + `INSERT INTO ${table} ("key", "fingerprint", "outcome", "expires_at") VALUES (?, ?, ?, ?) + ON CONFLICT ("key") DO UPDATE SET + "fingerprint" = excluded."fingerprint", + "outcome" = excluded."outcome", + "expires_at" = excluded."expires_at" + WHERE ${table}."expires_at" <= ? + RETURNING "key"`, + [key, fingerprint, JSON.stringify(outcome), expiresAt, now.toISOString()], + ); + if (rows.length === 0) throw new IdempotencyKeyTakenError(key); + }, + purge: async ({ now }) => { + const rows = await db.all(`DELETE FROM ${table} WHERE "expires_at" <= ? RETURNING "key"`, [ + now.toISOString(), + ]); + return rows.length; + }, +}); diff --git a/packages/core/src/adapter/sqlite/schema.ts b/packages/core/src/adapter/sqlite/schema.ts index 87e29544..7f337019 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 idempotencyKeys: 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" })), + idempotencyKeys: quoteIdentifier(storageTableNameFor({ prefix, table: "idempotencyKeys" })), }); const indexName = (table: string, suffix: string): string => @@ -100,6 +102,13 @@ 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.idempotencyKeys} ( + "key" TEXT PRIMARY KEY, + "fingerprint" TEXT NOT NULL, + "outcome" TEXT NOT NULL, + "expires_at" TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS ${indexName(tables.idempotencyKeys, "expires_at")} ON ${tables.idempotencyKeys} ("expires_at")`, ]; export interface EnsureStorageSchemaArgs { diff --git a/packages/core/src/adapter/sqlite/sqlite.test.ts b/packages/core/src/adapter/sqlite/sqlite.test.ts index 39fd1817..484dab17 100644 --- a/packages/core/src/adapter/sqlite/sqlite.test.ts +++ b/packages/core/src/adapter/sqlite/sqlite.test.ts @@ -12,6 +12,7 @@ import { contractFields, deadLetterStoreContract, eventStoreContract, + idempotencyStoreContract, inboxLedgerContract, pendingEvent, readModelRebuildContract, @@ -98,6 +99,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 }); + idempotencyStoreContract({ create: async () => (await storage()).idempotencyStore }); storageTransactionContract({ create: storage }); tableContract({ create: async () => @@ -239,6 +241,7 @@ describe("createSqliteAdapter", () => { "app_checkpoints", "app_dead_letters", "app_events", + "app_idempotency_keys", "app_inbox", "app_rm_order_summary", "app_scheduled_commands", diff --git a/packages/core/src/adapter/staged-event-store.ts b/packages/core/src/adapter/staged-event-store.ts index 1ece8ff9..408b090d 100644 --- a/packages/core/src/adapter/staged-event-store.ts +++ b/packages/core/src/adapter/staged-event-store.ts @@ -6,15 +6,31 @@ import type { AppendArgs, EventStore } from "./storage/event-store.ts"; * An event store that keeps what is appended to it aside: `load` answers with the base store's * events plus the ones staged for that stream, `append` stages with the version the stream has * in that view, and `readAll` and `lastPosition` read the base store alone, since staged events - * have no position yet. `batches` hands out what to append at commit, one batch per stream, each + * are not written yet. `batches` hands out what to append at commit, one batch per stream, each * expecting the version the stream had in the base store when the view first met it. */ export interface StagedEventStore extends EventStore { - batches(): readonly AppendArgs[]; + batches(): readonly StagedBatch[]; +} + +/** + * A batch to append at commit, with the position each of its events was staged with: what + * `nextPosition` gave, or 0. + */ +export interface StagedBatch extends AppendArgs { + readonly positions: readonly number[]; +} + +export interface CreateStagedEventStoreArgs { + /** + * Reserves the position a staged event will have once committed, for a base store that can + * hold one; without it, staged events answer with position 0. + */ + readonly nextPosition?: () => number; } export interface CreateStagedEventStoreFunction { - (base: EventStore): StagedEventStore; + (base: EventStore, args?: CreateStagedEventStoreArgs): StagedEventStore; } interface StagedStream { @@ -24,7 +40,10 @@ interface StagedStream { readonly events: StoredEvent[]; } -export const createStagedEventStore: CreateStagedEventStoreFunction = (base) => { +export const createStagedEventStore: CreateStagedEventStoreFunction = ( + base, + { nextPosition = () => 0 } = {}, +) => { const streams = new Map(); const stagedFor = async (aggregateType: string, aggregateId: string): Promise => { @@ -72,7 +91,7 @@ export const createStagedEventStore: CreateStagedEventStoreFunction = (base) => actualVersion, }); } - const stored = events.map((event) => ({ ...event, position: 0 })); + const stored = events.map((event) => ({ ...event, position: nextPosition() })); staged.events.push(...stored); return { version: actualVersion + stored.length, events: stored }; }, @@ -86,6 +105,7 @@ export const createStagedEventStore: CreateStagedEventStoreFunction = (base) => aggregateId, expectedVersion, events: events.map(({ position: _position, ...event }) => event), + positions: events.map((event) => event.position), })), }; }; diff --git a/packages/core/src/adapter/storage/idempotency-store.ts b/packages/core/src/adapter/storage/idempotency-store.ts new file mode 100644 index 00000000..0aecab50 --- /dev/null +++ b/packages/core/src/adapter/storage/idempotency-store.ts @@ -0,0 +1,59 @@ +import type { DispatchResult, RejectedDispatch } from "../../contracts/command.ts"; + +/** + * What a dispatch answered, as the idempotency store keeps it: the result of a command that was + * stored or scheduled, or its rejection. A dispatch that failed is not kept. + */ +export type DispatchOutcome = DispatchResult | RejectedDispatch; + +/** + * A dispatch remembered under its idempotency key: the fingerprint of the command it was, what it + * answered, and when the key may be used for another command. + */ +export interface IdempotencyRecord { + readonly key: string; + readonly fingerprint: string; + readonly outcome: DispatchOutcome; + readonly expiresAt: string; +} + +export interface GetIdempotencyRecordArgs { + readonly key: string; + /** + * The kernel clock's time: a record expired by then is as good as gone. + */ + readonly now: Date; +} + +export interface PutIdempotencyRecordArgs extends IdempotencyRecord { + /** + * The kernel clock's time: a record expired by then is replaced. + */ + readonly now: Date; +} + +export interface PurgeIdempotencyRecordsArgs { + readonly now: Date; +} + +/** + * Remembers what a command dispatched with an idempotency key answered, so the same key answers + * the same again. `put` must be atomic: of two dispatches racing for one key exactly one records + * its outcome, and the other rejects with `IdempotencyKeyTakenError`, which rolls back whatever + * the recording was part of. + */ +export interface IdempotencyStore { + /** + * The record under `key`, or `null` when there is none or it has expired. + */ + get(args: GetIdempotencyRecordArgs): Promise; + /** + * Records the outcome, taking the place of an expired record. Rejects with + * `IdempotencyKeyTakenError` when the key holds a record that has not expired. + */ + put(args: PutIdempotencyRecordArgs): Promise; + /** + * Removes the records expired by `now`, and resolves to how many. + */ + purge(args: PurgeIdempotencyRecordsArgs): Promise; +} diff --git a/packages/core/src/adapter/testing/fixtures.ts b/packages/core/src/adapter/testing/fixtures.ts index 2c1a4e13..a00a5fe6 100644 --- a/packages/core/src/adapter/testing/fixtures.ts +++ b/packages/core/src/adapter/testing/fixtures.ts @@ -1,4 +1,4 @@ -import type { NewCommand } from "../../contracts/command.ts"; +import type { NewCommand, StoredDispatch } from "../../contracts/command.ts"; import type { CausationContext, EventMetadata } from "../../contracts/metadata.ts"; import type { NewDeadLetter } from "../storage/dead-letter-store.ts"; import type { PendingEvent } from "../storage/event-store.ts"; @@ -89,3 +89,20 @@ export const testDeadLetter: TestDeadLetterFunction = (id) => ({ firstFailedAt: "2026-01-01T00:00:00.000Z", lastFailedAt: "2026-01-01T00:00:00.000Z", }); + +export interface TestOutcomeFunction { + (aggregateId: string): StoredDispatch; +} + +/** + * What a stored command answered, for idempotency store contract tests. + */ +export const testOutcome: TestOutcomeFunction = (aggregateId) => ({ + scheduled: false, + aggregateType: "order", + aggregateId, + version: 1, + eventIds: [`order-${aggregateId}-1`], + eventTypes: ["OrderPlaced"], + position: 1, +}); diff --git a/packages/core/src/adapter/testing/idempotency-store.contract.ts b/packages/core/src/adapter/testing/idempotency-store.contract.ts new file mode 100644 index 00000000..7ef74162 --- /dev/null +++ b/packages/core/src/adapter/testing/idempotency-store.contract.ts @@ -0,0 +1,100 @@ +import { beforeEach, describe, expect, it } from "vitest"; +import { IdempotencyKeyTakenError } from "../../contracts/errors.ts"; +import type { IdempotencyStore } from "../storage/idempotency-store.ts"; +import { testOutcome } from "./fixtures.ts"; + +export interface IdempotencyStoreContractArgs { + readonly create: () => Promise; +} + +export interface IdempotencyStoreContractFunction { + (args: IdempotencyStoreContractArgs): void; +} + +const now = new Date("2026-01-01T00:00:00.000Z"); +const later = (ms: number): Date => new Date(now.getTime() + ms); +const record = (key: string, expiresInMs = 60_000) => ({ + key, + fingerprint: `fp-${key}`, + outcome: testOutcome(key), + expiresAt: later(expiresInMs).toISOString(), +}); + +/** + * The behaviour every idempotency store must exhibit. + */ +export const idempotencyStoreContract: IdempotencyStoreContractFunction = ({ create }) => { + describe("idempotency store contract", () => { + let store: IdempotencyStore; + + beforeEach(async () => { + store = await create(); + }); + + it("answers a record as it was put, and null for a key it never saw", async () => { + await store.put({ ...record("k1"), now }); + expect(await store.get({ key: "k1", now })).toEqual(record("k1")); + expect(await store.get({ key: "k2", now })).toBeNull(); + }); + + it("keeps a rejection as well as a result", async () => { + const rejected = { + rejected: "AlreadyPlaced", + message: "Order already placed", + aggregateType: "order", + aggregateId: "1", + }; + await store.put({ ...record("k1"), outcome: rejected, now }); + expect((await store.get({ key: "k1", now }))?.outcome).toEqual(rejected); + }); + + it("refuses to record a key that holds a record, and keeps the first", async () => { + await store.put({ ...record("k1"), now }); + const refused = await store + .put({ ...record("k1"), fingerprint: "other", now: later(1_000) }) + .catch((error: unknown) => error); + expect(refused).toBeInstanceOf(IdempotencyKeyTakenError); + expect(refused).toMatchObject({ code: "IDEMPOTENCY_KEY_TAKEN", idempotencyKey: "k1" }); + expect((refused as Error).message).toContain("k1"); + expect(await store.get({ key: "k1", now })).toEqual(record("k1")); + }); + + it("gives exactly one winner to concurrent puts", async () => { + const outcomes = await Promise.allSettled( + Array.from({ length: 5 }, (_, index) => + store.put({ ...record("k1"), fingerprint: `fp-${index}`, now }), + ), + ); + expect(outcomes.filter((outcome) => outcome.status === "fulfilled")).toHaveLength(1); + expect( + outcomes.filter( + (outcome) => + outcome.status === "rejected" && outcome.reason instanceof IdempotencyKeyTakenError, + ), + ).toHaveLength(4); + }); + + it("treats an expired record as gone: unanswered, replaced, and purged", async () => { + await store.put({ ...record("k1", 1_000), now }); + expect(await store.get({ key: "k1", now: later(999) })).not.toBeNull(); + expect(await store.get({ key: "k1", now: later(1_000) })).toBeNull(); + await expect( + store.put({ ...record("k1", 1_000), fingerprint: "late", now: later(999) }), + ).rejects.toBeInstanceOf(IdempotencyKeyTakenError); + await store.put({ ...record("k1", 5_000), fingerprint: "again", now: later(1_000) }); + expect(await store.get({ key: "k1", now: later(1_000) })).toMatchObject({ + fingerprint: "again", + expiresAt: later(5_000).toISOString(), + }); + + await store.put({ ...record("k2", 2_000), now }); + await store.put({ ...record("k3", 10_000), now }); + expect(await store.purge({ now: later(2_000) })).toBe(1); + expect(await store.purge({ now: later(2_000) })).toBe(0); + expect(await store.get({ key: "k1", now: later(2_000) })).not.toBeNull(); + expect(await store.get({ key: "k3", now: later(2_000) })).not.toBeNull(); + expect(await store.purge({ now: later(10_000) })).toBe(2); + expect(await store.get({ key: "k3", now: later(2_000) })).toBeNull(); + }); + }); +}; diff --git a/packages/core/src/adapter/testing/index.ts b/packages/core/src/adapter/testing/index.ts index 37317c73..1d72c64d 100644 --- a/packages/core/src/adapter/testing/index.ts +++ b/packages/core/src/adapter/testing/index.ts @@ -14,8 +14,14 @@ export type { PendingEventArgs, PendingEventFunction, TestCommandFunction, + TestOutcomeFunction, } from "./fixtures.ts"; -export { pendingEvent, testCommand, testContext, testMetadata } from "./fixtures.ts"; +export { pendingEvent, testCommand, testContext, testMetadata, testOutcome } from "./fixtures.ts"; +export type { + IdempotencyStoreContractArgs, + IdempotencyStoreContractFunction, +} from "./idempotency-store.contract.ts"; +export { idempotencyStoreContract } from "./idempotency-store.contract.ts"; export type { InboxLedgerContractArgs, InboxLedgerContractFunction, diff --git a/packages/core/src/adapter/testing/storage-transaction.contract.ts b/packages/core/src/adapter/testing/storage-transaction.contract.ts index bc01cad7..7f4ca14d 100644 --- a/packages/core/src/adapter/testing/storage-transaction.contract.ts +++ b/packages/core/src/adapter/testing/storage-transaction.contract.ts @@ -2,10 +2,11 @@ import { beforeEach, describe, expect, it } from "vitest"; import { ConcurrencyError, DeadLetterSettledError, + IdempotencyKeyTakenError, ScheduledClaimLostError, } from "../../contracts/errors.ts"; import type { Storage, StorageTransaction } from "../adapter.ts"; -import { pendingEvent, testCommand, testContext, testDeadLetter } from "./fixtures.ts"; +import { pendingEvent, testCommand, testContext, testDeadLetter, testOutcome } from "./fixtures.ts"; export interface StorageTransactionContractArgs { readonly create: () => Promise; @@ -17,6 +18,13 @@ export interface StorageTransactionContractFunction { const now = new Date("2026-01-01T00:00:00.000Z"); const key = { handler: "order.p", eventId: "e1" }; +const record = { + key: "idem-1", + fingerprint: "fp", + outcome: testOutcome("1"), + expiresAt: new Date(now.getTime() + 60_000).toISOString(), + now, +}; /** * The behaviour every `Storage.transact` must exhibit. Call it inside a `describe` of the @@ -65,6 +73,7 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ status: "failed", }); await tx.inboxLedger.complete(key); + await tx.idempotencyStore.put(record); if (fail) throw new Error("boom"); }); @@ -73,9 +82,10 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ expect(await storage.scheduler.list()).toEqual([]); expect(await storage.deadLetterStore.count()).toBe(0); expect((await storage.inboxLedger.get(key))?.status).toBe("pending"); + expect(await storage.idempotencyStore.get({ key: record.key, now })).toBeNull(); }; - it("writes the events of several streams, the claim, the scheduler entry and the dead letter together", async () => { + it("writes the events of several streams, the claim, the scheduler entry, the dead letter and the idempotency record together", async () => { await everything(false); const all = await storage.eventStore.readAll({ afterPosition: 0, limit: 10 }); expect(all.map((event) => event.aggregateType)).toEqual(["order", "payment"]); @@ -85,6 +95,30 @@ export const storageTransactionContract: StorageTransactionContractFunction = ({ ]); expect(await storage.deadLetterStore.count()).toBe(1); expect((await storage.inboxLedger.get(key))?.status).toBe("succeeded"); + expect(await storage.idempotencyStore.get({ key: record.key, now })).toMatchObject({ + fingerprint: "fp", + }); + }); + + it("writes nothing when the idempotency key it records is already taken", async () => { + await storage.idempotencyStore.put({ ...record, fingerprint: "first" }); + await expect( + storage.transact(async (tx) => { + await tx.eventStore.append({ + aggregateType: "order", + aggregateId: "1", + expectedVersion: 0, + events: [pendingEvent({ aggregateId: "1", version: 1 })], + }); + await tx.inboxLedger.complete(key); + await tx.idempotencyStore.put(record); + }), + ).rejects.toBeInstanceOf(IdempotencyKeyTakenError); + expect(await storage.eventStore.lastPosition()).toBe(0); + expect((await storage.inboxLedger.get(key))?.status).toBe("pending"); + expect(await storage.idempotencyStore.get({ key: record.key, now })).toMatchObject({ + fingerprint: "first", + }); }); it("writes nothing when the work throws", async () => { diff --git a/packages/core/src/config/defaults.ts b/packages/core/src/config/defaults.ts index a82ad757..d257b0e6 100644 --- a/packages/core/src/config/defaults.ts +++ b/packages/core/src/config/defaults.ts @@ -29,6 +29,7 @@ export const DEFAULT_PROCESSES: ResolvedProcessesConfig = { }; export const DEFAULT_CONCURRENCY_RETRIES: number = 3; +export const DEFAULT_IDEMPOTENCY_KEY_LIFETIME_MS: number = 86_400_000; export const DEFAULT_POLL_INTERVAL_MS: number = 100; export const DEFAULT_IDLE_INTERVAL_MS: number = 30_000; export const DEFAULT_BATCH_SIZE: number = 100; diff --git a/packages/core/src/config/schema.test.ts b/packages/core/src/config/schema.test.ts index 780fa540..81402cec 100644 --- a/packages/core/src/config/schema.test.ts +++ b/packages/core/src/config/schema.test.ts @@ -33,7 +33,7 @@ describe("resolveConfig", () => { expect(resolved.ports).toEqual({}); expect(resolved.runtime).toEqual({ role: "all", - commands: { concurrencyRetries: 3, timeoutMs: 30_000 }, + commands: { concurrencyRetries: 3, timeoutMs: 30_000, idempotencyKeyLifetimeMs: 86_400_000 }, policies: { retry: { strategy: "exponential", maxAttempts: 3, baseDelayMs: 1_000, maxDelayMs: 30_000 }, timeoutMs: 30_000, @@ -62,7 +62,7 @@ describe("resolveConfig", () => { readModels: { "users-directory": sqlite }, runtime: { role: "worker", - commands: { concurrencyRetries: 0, timeout: "5s" }, + commands: { concurrencyRetries: 0, timeout: "5s", idempotencyKeyLifetime: "1h" }, policies: { retry: { strategy: "fixed", maxAttempts: 5, baseDelay: "2s" }, timeout: "1m" }, processes: { timeout: "48h", handlerTimeout: "2m" }, dispatcher: { @@ -77,7 +77,11 @@ describe("resolveConfig", () => { ports: { order: { inventory: "http" } }, }); expect(resolved.runtime.role).toBe("worker"); - expect(resolved.runtime.commands).toEqual({ concurrencyRetries: 0, timeoutMs: 5_000 }); + expect(resolved.runtime.commands).toEqual({ + concurrencyRetries: 0, + timeoutMs: 5_000, + idempotencyKeyLifetimeMs: 3_600_000, + }); expect(resolved.runtime.policies.retry).toEqual({ strategy: "fixed", maxAttempts: 5, diff --git a/packages/core/src/config/schema.ts b/packages/core/src/config/schema.ts index 9245441d..8ed21654 100644 --- a/packages/core/src/config/schema.ts +++ b/packages/core/src/config/schema.ts @@ -10,6 +10,7 @@ import { DEFAULT_CATCH_UP_TIMEOUT_MS, DEFAULT_COMMANDS, DEFAULT_CONCURRENCY_RETRIES, + DEFAULT_IDEMPOTENCY_KEY_LIFETIME_MS, DEFAULT_IDLE_INTERVAL_MS, DEFAULT_POLICIES, DEFAULT_POLL_INTERVAL_MS, @@ -74,7 +75,12 @@ const overrides = z.strictObject({ const runtime = z.strictObject({ role: z.enum(["web", "worker", "all"]).optional(), - commands: commands.extend({ concurrencyRetries: z.int().min(0).optional() }).optional(), + commands: commands + .extend({ + concurrencyRetries: z.int().min(0).optional(), + idempotencyKeyLifetime: timeout.optional(), + }) + .optional(), policies: policies.optional(), processes: processes.optional(), dispatcher: z @@ -187,6 +193,8 @@ export const resolveConfig: ResolveConfigFunction = (config) => { ...baseCommands, concurrencyRetries: parsed.runtime?.commands?.concurrencyRetries ?? DEFAULT_CONCURRENCY_RETRIES, + idempotencyKeyLifetimeMs: + parsed.runtime?.commands?.idempotencyKeyLifetime ?? DEFAULT_IDEMPOTENCY_KEY_LIFETIME_MS, }, policies: basePolicies, processes: baseProcesses, diff --git a/packages/core/src/config/types.ts b/packages/core/src/config/types.ts index 3107f57b..930e592f 100644 --- a/packages/core/src/config/types.ts +++ b/packages/core/src/config/types.ts @@ -78,6 +78,12 @@ export interface CommandsRuntimeConfig { */ readonly timeout?: DurationInput; readonly concurrencyRetries?: number; + /** + * How long a command dispatched with an `idempotencyKey` is remembered, so a dispatch with the + * same key answers the same. Past it the key is free again, and the scheduled-command worker + * purges the record. Defaults to 24 hours. + */ + readonly idempotencyKeyLifetime?: DurationInput; } /** @@ -226,7 +232,10 @@ export interface ResolvedConfig { readonly readModels: Readonly>; readonly runtime: { readonly role: RuntimeRole; - readonly commands: ResolvedCommandsConfig & { readonly concurrencyRetries: number }; + readonly commands: ResolvedCommandsConfig & { + readonly concurrencyRetries: number; + readonly idempotencyKeyLifetimeMs: number; + }; readonly policies: ResolvedPoliciesConfig; readonly processes: ResolvedProcessesConfig; readonly dispatcher: { diff --git a/packages/core/src/contracts/command.ts b/packages/core/src/contracts/command.ts index 1f4bce6d..bccae97a 100644 --- a/packages/core/src/contracts/command.ts +++ b/packages/core/src/contracts/command.ts @@ -20,6 +20,11 @@ export interface NewCommand { readonly aggregateId: string; } +/** + * The longest `idempotencyKey` a dispatch takes, as the providers that take one allow. + */ +export const MAX_IDEMPOTENCY_KEY_LENGTH: number = 255; + /** * `delay` schedules the command instead of executing it now. `correlationId` overrides the * generated correlation id, for a command that starts a new request from outside the runtime. @@ -27,6 +32,16 @@ export interface NewCommand { export interface DispatchOptions { readonly delay?: DurationInput; readonly correlationId?: string; + /** + * 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, + * so a retry runs it. Between 1 and `MAX_IDEMPOTENCY_KEY_LENGTH` characters. + */ + readonly idempotencyKey?: string; /** * Withdraws the command until its events start being stored: the handler's `signal` aborts and * the dispatch rejects with the signal's `reason`, storing nothing. Already aborted, the handler diff --git a/packages/core/src/contracts/errors.ts b/packages/core/src/contracts/errors.ts index 2c25e9d7..554f8f25 100644 --- a/packages/core/src/contracts/errors.ts +++ b/packages/core/src/contracts/errors.ts @@ -95,6 +95,46 @@ export class ValidationError extends BoundaError { } } +export interface IdempotencyKeyConflictErrorArgs { + readonly idempotencyKey: string; + readonly commandType: string; +} + +/** + * 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. + */ +export class IdempotencyKeyConflictError extends BoundaError { + readonly idempotencyKey: string; + + constructor({ idempotencyKey, commandType }: IdempotencyKeyConflictErrorArgs) { + super( + "IDEMPOTENCY_KEY_CONFLICT", + `Idempotency key "${idempotencyKey}" was already used for a command other than ${commandType} with this payload`, + ); + this.idempotencyKey = idempotencyKey; + } +} + +/** + * Thrown by an idempotency store asked to record a dispatch under a key that already holds one: + * a duplicate that ran beside the first lost the race. Whatever the recording was part of rolls + * back, and the dispatch answers what the key holds instead. + */ +export class IdempotencyKeyTakenError extends BoundaError { + readonly idempotencyKey: string; + + constructor(idempotencyKey: string) { + super( + "IDEMPOTENCY_KEY_TAKEN", + `Idempotency key "${idempotencyKey}" already holds the outcome of a dispatch`, + ); + this.idempotencyKey = idempotencyKey; + } +} + export interface ClaimLostErrorArgs { readonly handler: string; readonly eventId: string; diff --git a/packages/core/src/contracts/index.ts b/packages/core/src/contracts/index.ts index 6b4bb703..9b79990b 100644 --- a/packages/core/src/contracts/index.ts +++ b/packages/core/src/contracts/index.ts @@ -12,6 +12,7 @@ export type { ScheduledDispatch, StoredDispatch, } from "./command.ts"; +export { MAX_IDEMPOTENCY_KEY_LENGTH } from "./command.ts"; export type { AsDurationFunction, DurationInput, @@ -32,6 +33,9 @@ export { DeadLetterSettledError, type DeadLetterSettledErrorArgs, DomainError, + IdempotencyKeyConflictError, + type IdempotencyKeyConflictErrorArgs, + IdempotencyKeyTakenError, NotFoundError, RebuildSupersededError, type Rejection, diff --git a/packages/core/src/kernel/app.test.ts b/packages/core/src/kernel/app.test.ts index ac345de4..f2ff41fa 100644 --- a/packages/core/src/kernel/app.test.ts +++ b/packages/core/src/kernel/app.test.ts @@ -431,6 +431,22 @@ describe("createApp", () => { await fresh.stop(); }); + it("answers a command dispatched again under its idempotency key as the first time, read models included", async () => { + const { app } = await start("web"); + const fresh = readYourWrites(app); + const payload = { orderId: "o-1", total: 10 }; + const placed = await fresh.commands.placeOrder(payload, { idempotencyKey: "req-1" }); + expect(placed).toMatchObject({ version: 1, position: 1 }); + expect(await fresh.queries.getOrder({ orderId: "o-1" })).toMatchObject({ status: "placed" }); + expect(await fresh.commands.placeOrder(payload, { idempotencyKey: "req-1" })).toEqual(placed); + await expect( + fresh.commands.placeOrder({ orderId: "o-1", total: 11 }, { idempotencyKey: "req-1" }), + ).rejects.toMatchObject({ code: "IDEMPOTENCY_KEY_CONFLICT" }); + expect(sentMessages).toEqual(["placed o-1 v0"]); + expect((await app.events.read()).map((event) => event.position)).toEqual([1]); + await fresh.stop(); + }); + it("does not start background work in the web role but still serves commands and queries", async () => { const { app, clock } = await start("web"); expect(app.role).toBe("web"); diff --git a/packages/core/src/kernel/app.ts b/packages/core/src/kernel/app.ts index c0c6f224..3ef05605 100644 --- a/packages/core/src/kernel/app.ts +++ b/packages/core/src/kernel/app.ts @@ -284,8 +284,7 @@ export const assembleApp: AssembleAppFunction = async ({ const observers = new Set(); const pipeline = createCommandPipeline({ aggregates, - eventStore: storage.eventStore, - scheduler: storage.scheduler, + storage, config, ids, clock, diff --git a/packages/core/src/kernel/command/fingerprint.test.ts b/packages/core/src/kernel/command/fingerprint.test.ts new file mode 100644 index 00000000..7ff7006b --- /dev/null +++ b/packages/core/src/kernel/command/fingerprint.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, it } from "vitest"; +import { fingerprintOf } from "./fingerprint.ts"; + +describe("fingerprintOf", () => { + it("is a SHA-256 in hex of the type and the payload", async () => { + const fingerprint = await fingerprintOf({ type: "PlaceOrder", payload: { total: 1 } }); + expect(fingerprint).toBe("6d14c4deeefd3d85dac3648381f574fd5e93da3ec6bba0967f6b5857ca5b4187"); + expect(await fingerprintOf({ type: "PlaceOrder", payload: { total: 1 } })).toBe(fingerprint); + // A digest with a byte below 16, whose hex keeps its leading zero. + expect(await fingerprintOf({ type: "T", payload: {} })).toBe( + "780ee647afcb9daf8dcdb984331fadbe36e98c600af7973d9d2fff89f976f0e7", + ); + }); + + it("tells an array from an object with its indexes as keys", async () => { + expect(await fingerprintOf({ type: "T", payload: ["a"] })).not.toBe( + await fingerprintOf({ type: "T", payload: { 0: "a" } }), + ); + }); + + it("changes with the type and with the payload", async () => { + const base = await fingerprintOf({ type: "PlaceOrder", payload: { total: 1 } }); + expect(await fingerprintOf({ type: "PayOrder", payload: { total: 1 } })).not.toBe(base); + expect(await fingerprintOf({ type: "PlaceOrder", payload: { total: 2 } })).not.toBe(base); + expect(await fingerprintOf({ type: "PlaceOrder", payload: [{ total: 1 }] })).not.toBe(base); + }); + + it("ignores the order of keys, at every depth", async () => { + const one = await fingerprintOf({ + type: "T", + payload: { a: 1, b: { c: [{ d: 1, e: 2 }], f: 3 } }, + }); + const other = await fingerprintOf({ + type: "T", + payload: { b: { f: 3, c: [{ e: 2, d: 1 }] }, a: 1 }, + }); + expect(other).toBe(one); + expect( + await fingerprintOf({ type: "T", payload: { a: 1, b: { c: [{ e: 2, d: 1 }] } } }), + ).not.toBe(one); + }); + + it("takes dates as their JSON, and no payload as none", async () => { + const at = new Date("2026-01-01T00:00:00.000Z"); + expect(await fingerprintOf({ type: "T", payload: { at } })).toBe( + await fingerprintOf({ type: "T", payload: { at: at.toISOString() } }), + ); + expect(await fingerprintOf({ type: "T", payload: undefined })).toBe( + await fingerprintOf({ type: "T", payload: null }), + ); + expect(await fingerprintOf({ type: "T", payload: undefined })).not.toBe( + await fingerprintOf({ type: "T", payload: {} }), + ); + }); +}); diff --git a/packages/core/src/kernel/command/fingerprint.ts b/packages/core/src/kernel/command/fingerprint.ts new file mode 100644 index 00000000..6d3bd0c8 --- /dev/null +++ b/packages/core/src/kernel/command/fingerprint.ts @@ -0,0 +1,32 @@ +export interface FingerprintArgs { + readonly type: string; + readonly payload: unknown; +} + +export interface FingerprintFunction { + (args: FingerprintArgs): Promise; +} + +// JSON with the keys of every object in order, so two payloads that are the same object +// written in another order fingerprint alike. +const canonical = (value: unknown): unknown => { + if (Array.isArray(value)) return value.map(canonical); + if (typeof value !== "object" || value === null) return value; + if (value instanceof Date) return value; + return Object.fromEntries( + Object.keys(value as Record) + .sort() + .map((key) => [key, canonical((value as Record)[key])]), + ); +}; + +/** + * 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. + */ +export const fingerprintOf: FingerprintFunction = async ({ type, payload }) => { + const bytes = new TextEncoder().encode(JSON.stringify([type, canonical(payload) ?? null])); + 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 new file mode 100644 index 00000000..cd573187 --- /dev/null +++ b/packages/core/src/kernel/command/idempotency.test.ts @@ -0,0 +1,340 @@ +import { describe, expect, it, onTestFinished } from "vitest"; +import { + ConfigurationError, + DomainError, + IdempotencyKeyConflictError, + IdempotencyKeyTakenError, + ValidationError, +} from "../../contracts/errors.ts"; +import { classifyFailure } from "../shared/retry.ts"; +import { ATTRIBUTES } from "../telemetry.ts"; +import { installFakeTelemetry } from "../telemetry-fake.ts"; +import { + createKernelHarness, + createRecordingLogger, + type KernelHarness, + placeOrderKeys, + sentMessages, +} from "../test-support.ts"; +import { createUnitOfWork } from "../unit-of-work/unit-of-work.ts"; +import { fingerprintOf } from "./fingerprint.ts"; + +const now = new Date("2026-01-01T00:00:00.000Z"); +const placeOrder = { type: "PlaceOrder", payload: { orderId: "o-1", total: 42 } }; +const withKey = (idempotencyKey: string, extra = {}) => ({ + options: { idempotencyKey, ...extra }, +}); + +const storedTypes = async ({ storage }: KernelHarness): Promise => + (await storage.eventStore.readAll({ afterPosition: 0, limit: 100 })).map((event) => event.type); + +describe("a command dispatched with an idempotency key", () => { + it("runs once and answers every later dispatch with the same key and payload the same, without running", async () => { + const telemetry = installFakeTelemetry(); + onTestFinished(() => telemetry.restore()); + const harness = await createKernelHarness(); + const first = await harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1") }); + expect(first).toMatchObject({ scheduled: false, version: 1, position: 1 }); + harness.clock.advance(1_000); + const again = await harness.pipeline.dispatch({ + type: "PlaceOrder", + // The same payload written in another order. + payload: { total: 42, orderId: "o-1" }, + ...withKey("req-1"), + }); + expect(again).toEqual(first); + expect(sentMessages).toEqual(["placed o-1 v0"]); + expect(await storedTypes(harness)).toEqual(["OrderPlaced"]); + expect(await harness.storage.idempotencyStore.get({ key: "req-1", now })).toMatchObject({ + outcome: first, + expiresAt: "2026-01-02T00:00:00.000Z", + }); + expect(telemetry.counts.map((count) => count.attributes[ATTRIBUTES.outcome])).toEqual([ + "stored", + "replayed", + ]); + expect(telemetry.spans.at(-1)?.attributes[ATTRIBUTES.outcome]).toBe("replayed"); + }); + + it("remembers a rejection as well, and throws it again with its code and message", async () => { + const telemetry = installFakeTelemetry(); + onTestFinished(() => telemetry.restore()); + const harness = await createKernelHarness(); + await harness.pipeline.dispatch(placeOrder); + const rejection = () => + harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-2") }).catch((e: unknown) => e); + const first = await rejection(); + expect(first).toBeInstanceOf(DomainError); + expect(placeOrderKeys).toHaveLength(2); + const again = await rejection(); + expect(again).toBeInstanceOf(DomainError); + expect(again).toMatchObject({ rejected: "AlreadyPlaced", message: "Order already placed" }); + expect(again).not.toBe(first); + expect(placeOrderKeys).toHaveLength(2); + expect(await storedTypes(harness)).toEqual(["OrderPlaced"]); + expect(telemetry.spans.at(-1)?.attributes[ATTRIBUTES.outcome]).toBe("replayed"); + expect(telemetry.spanEvents.map((event) => event.attributes)).toEqual([ + { [ATTRIBUTES.rejected]: "AlreadyPlaced" }, + { [ATTRIBUTES.rejected]: "AlreadyPlaced" }, + ]); + }); + + it("remembers a command that stored nothing", async () => { + const harness = await createKernelHarness(); + const touch = { type: "TouchOrder", payload: { orderId: "o-1" }, ...withKey("req-3") }; + const first = await harness.pipeline.dispatch(touch); + expect(first).toMatchObject({ scheduled: false, version: 0, eventIds: [], position: 0 }); + expect(await harness.pipeline.dispatch(touch)).toEqual(first); + expect(await harness.storage.idempotencyStore.get({ key: "req-3", now })).not.toBeNull(); + }); + + it("refuses the same key for another command or another payload, and runs nothing", async () => { + const harness = await createKernelHarness(); + await harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1") }); + const otherPayload = await harness.pipeline + .dispatch({ type: "PlaceOrder", payload: { orderId: "o-1", total: 43 }, ...withKey("req-1") }) + .catch((error: unknown) => error); + expect(otherPayload).toBeInstanceOf(IdempotencyKeyConflictError); + expect(otherPayload).toMatchObject({ + code: "IDEMPOTENCY_KEY_CONFLICT", + idempotencyKey: "req-1", + }); + expect((otherPayload as Error).message).toContain('"req-1"'); + expect((otherPayload as Error).message).toContain("PlaceOrder"); + await expect( + harness.pipeline.dispatch({ + type: "TouchOrder", + payload: { orderId: "o-1" }, + ...withKey("req-1"), + }), + ).rejects.toBeInstanceOf(IdempotencyKeyConflictError); + expect(classifyFailure(otherPayload)).toBe("terminal"); + expect(placeOrderKeys).toHaveLength(1); + expect(await storedTypes(harness)).toEqual(["OrderPlaced"]); + }); + + it("gives two dispatches racing for one key the same answer, storing the events once", async () => { + const harness = await createKernelHarness(); + const results = await Promise.all([ + harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1") }), + harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1") }), + ]); + expect(results[1]).toEqual(results[0]); + expect(results[0]).toMatchObject({ version: 1, position: 1 }); + expect(await storedTypes(harness)).toEqual(["OrderPlaced"]); + }); + + it("records what a retry after a conflict answers, not what the lost attempt did", async () => { + const harness = await createKernelHarness(); + const transact = harness.storage.transact.bind(harness.storage); + let interfered = false; + harness.storage.transact = async (work) => { + if (!interfered) { + interfered = true; + await harness.pipeline.dispatch(placeOrder); + } + return transact(work); + }; + const first = await harness.pipeline + .dispatch({ ...placeOrder, ...withKey("req-1") }) + .catch((error: unknown) => error); + expect(first).toMatchObject({ rejected: "AlreadyPlaced" }); + expect(placeOrderKeys).toEqual(["id-1", "id-3", "id-1"]); + await expect( + harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1") }), + ).rejects.toMatchObject({ rejected: "AlreadyPlaced" }); + expect(placeOrderKeys).toHaveLength(3); + expect(await storedTypes(harness)).toEqual(["OrderPlaced"]); + }); + + it("answers what the key holds when it loses the race to record, and fails when the record has gone", async () => { + const telemetry = installFakeTelemetry(); + onTestFinished(() => telemetry.restore()); + const harness = await createKernelHarness(); + const { idempotencyStore } = harness.storage; + const winner = { + key: "req-1", + fingerprint: await fingerprintOf(placeOrder), + outcome: { + scheduled: false as const, + aggregateType: "order", + aggregateId: "o-1", + version: 7, + eventIds: ["elsewhere"], + eventTypes: ["OrderPlaced"], + position: 7, + }, + expiresAt: "2026-01-02T00:00:00.000Z", + now, + }; + // The winner records once this dispatch has looked the key up and found nothing. + const get = idempotencyStore.get; + let looked = 0; + harness.storage.idempotencyStore.get = async (args) => { + looked += 1; + const found = await get(args); + if (looked === 1) await idempotencyStore.put(winner); + return found; + }; + expect(await harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1") })).toEqual( + winner.outcome, + ); + expect(placeOrderKeys).toHaveLength(1); + expect(await storedTypes(harness)).toEqual([]); + expect(telemetry.counts.map((count) => count.attributes[ATTRIBUTES.outcome])).toEqual([ + "replayed", + ]); + + harness.storage.idempotencyStore.get = async () => null; + const lost = await harness.pipeline + .dispatch({ ...placeOrder, ...withKey("req-1") }) + .catch((error: unknown) => error); + expect(lost).toBeInstanceOf(IdempotencyKeyTakenError); + expect(telemetry.counts.at(-1)?.attributes[ATTRIBUTES.outcome]).toBe("failed"); + expect(await storedTypes(harness)).toEqual([]); + + // Any other failure stays a failure, whatever the key holds by then. + const broken = { type: "BreakOrder", payload: { orderId: "o-1" } }; + looked = 0; + harness.storage.idempotencyStore.get = async () => { + looked += 1; + return looked === 1 ? null : { ...winner, fingerprint: await fingerprintOf(broken) }; + }; + await expect( + harness.pipeline.dispatch({ ...broken, ...withKey("req-2") }), + ).rejects.toBeInstanceOf(ConfigurationError); + }); + + it("schedules a delayed command once under its key", async () => { + const harness = await createKernelHarness(); + const delayed = { ...placeOrder, ...withKey("req-1", { delay: "5m" }) }; + const first = await harness.pipeline.dispatch(delayed); + expect(first).toMatchObject({ scheduled: true, executeAt: "2026-01-01T00:05:00.000Z" }); + harness.clock.advance(1_000); + expect(await harness.pipeline.dispatch(delayed)).toEqual(first); + expect(await harness.storage.scheduler.list()).toHaveLength(1); + }); + + it("forgets a key once its lifetime has passed, and runs the command again", async () => { + const harness = await createKernelHarness({ + config: { runtime: { commands: { idempotencyKeyLifetime: "1h" } } }, + }); + const touch = { type: "TouchOrder", payload: { orderId: "o-1" }, ...withKey("req-1") }; + await harness.pipeline.dispatch(touch); + expect(await harness.storage.idempotencyStore.get({ key: "req-1", now })).toMatchObject({ + expiresAt: "2026-01-01T01:00:00.000Z", + }); + harness.clock.advance(3_600_000); + await harness.pipeline.dispatch({ ...placeOrder, ...withKey("req-1") }); + expect(await storedTypes(harness)).toEqual(["OrderPlaced"]); + expect( + await harness.storage.idempotencyStore.get({ key: "req-1", now: harness.clock.now() }), + ).toMatchObject({ outcome: { eventTypes: ["OrderPlaced"] } }); + }); + + it("does not remember a command that failed, so a retry runs it", async () => { + const harness = await createKernelHarness(); + const broken = { type: "BreakOrder", payload: { orderId: "o-1" }, ...withKey("req-1") }; + await expect(harness.pipeline.dispatch(broken)).rejects.toBeInstanceOf(ConfigurationError); + expect(await harness.storage.idempotencyStore.get({ key: "req-1", now })).toBeNull(); + await expect(harness.pipeline.dispatch(broken)).rejects.toBeInstanceOf(ConfigurationError); + }); + + it("refuses an empty key, or one too long, before running anything", async () => { + const harness = await createKernelHarness(); + for (const idempotencyKey of ["", "k".repeat(256), 7, ["k"]] as unknown as string[]) { + const refused = await harness.pipeline + .dispatch({ ...placeOrder, options: { idempotencyKey } }) + .catch((error: unknown) => error); + expect(refused).toBeInstanceOf(ValidationError); + expect(refused).toMatchObject({ + message: 'Command "PlaceOrder" has an invalid idempotency key', + issues: [{ path: ["idempotencyKey"], message: "Expected a string of 1 to 255 characters" }], + }); + } + expect(placeOrderKeys).toEqual([]); + await harness.pipeline.dispatch({ ...placeOrder, ...withKey("k".repeat(255)) }); + expect(placeOrderKeys).toHaveLength(1); + }); + + it("records a dispatch within a unit of work when the unit commits, and answers the next from it", async () => { + const { logger, entries } = createRecordingLogger(); + const harness = await createKernelHarness({ logger }); + const unit = createUnitOfWork({ storage: harness.storage }); + const first = await harness.pipeline.dispatchUnattended({ + ...placeOrder, + ...withKey("req-1"), + within: unit, + }); + expect(first).toMatchObject({ scheduled: false, version: 1, position: 0 }); + expect(await harness.storage.idempotencyStore.get({ key: "req-1", now })).toBeNull(); + await unit.commit(); + expect(await harness.storage.idempotencyStore.get({ key: "req-1", now })).toMatchObject({ + outcome: first, + }); + expect(await storedTypes(harness)).toEqual(["OrderPlaced"]); + + const again = createUnitOfWork({ storage: harness.storage }); + expect( + await harness.pipeline.dispatchUnattended({ + ...placeOrder, + ...withKey("req-1"), + within: again, + }), + ).toEqual(first); + expect(placeOrderKeys).toHaveLength(1); + + const rejecting = createUnitOfWork({ storage: harness.storage }); + const rejected = await harness.pipeline.dispatchUnattended({ + ...placeOrder, + ...withKey("req-2"), + within: rejecting, + }); + expect(rejected).toMatchObject({ rejected: "AlreadyPlaced" }); + await rejecting.commit(); + const replayed = await harness.pipeline.dispatchUnattended({ + ...placeOrder, + ...withKey("req-2"), + within: createUnitOfWork({ storage: harness.storage }), + }); + expect(replayed).toEqual(rejected); + expect(placeOrderKeys).toHaveLength(2); + expect(entries.filter((entry) => entry.message === "command rejected")).toHaveLength(2); + expect(harness.rejections).toHaveLength(1); + + const twice = createUnitOfWork({ storage: harness.storage }); + const once = await harness.pipeline.dispatchUnattended({ + type: "TouchOrder", + payload: { orderId: "o-9" }, + ...withKey("req-9"), + within: twice, + }); + expect( + await harness.pipeline.dispatchUnattended({ + type: "TouchOrder", + payload: { orderId: "o-9" }, + ...withKey("req-9"), + within: twice, + }), + ).toEqual(once); + await twice.commit(); + expect(await harness.storage.idempotencyStore.get({ key: "req-9", now })).toMatchObject({ + outcome: once, + }); + + const scheduling = createUnitOfWork({ storage: harness.storage }); + const delayed = await harness.pipeline.dispatchUnattended({ + ...placeOrder, + ...withKey("req-3", { delay: "5m" }), + within: scheduling, + }); + expect(delayed).toMatchObject({ scheduled: true }); + expect(await harness.storage.scheduler.list()).toEqual([]); + expect(await harness.storage.idempotencyStore.get({ key: "req-3", now })).toBeNull(); + await scheduling.commit(); + expect(await harness.storage.scheduler.list()).toHaveLength(1); + expect(await harness.storage.idempotencyStore.get({ key: "req-3", now })).toMatchObject({ + outcome: delayed, + }); + }); +}); diff --git a/packages/core/src/kernel/command/pipeline.ts b/packages/core/src/kernel/command/pipeline.ts index fc06ad40..407eb0f2 100644 --- a/packages/core/src/kernel/command/pipeline.ts +++ b/packages/core/src/kernel/command/pipeline.ts @@ -1,16 +1,23 @@ -import type { EventStore, PendingEvent } from "../../adapter/storage/event-store.ts"; +import type { Storage, StorageTransaction } from "../../adapter/adapter.ts"; +import type { AppendResult, EventStore, PendingEvent } from "../../adapter/storage/event-store.ts"; +import type { + DispatchOutcome, + PutIdempotencyRecordArgs, +} from "../../adapter/storage/idempotency-store.ts"; import type { Scheduler } from "../../adapter/storage/scheduler.ts"; import type { ResolvedConfig } from "../../config/types.ts"; import type { Clock } from "../../contracts/clock.ts"; -import type { - Command, - CommandRejection, - DispatchOptions, - DispatchResult, - RejectedDispatch, - StoredDispatch, +import { + type Command, + type CommandRejection, + type DispatchOptions, + type DispatchResult, + MAX_IDEMPOTENCY_KEY_LENGTH, + type RejectedDispatch, + type ScheduledDispatch, + type StoredDispatch, } from "../../contracts/command.ts"; -import { parseDuration } from "../../contracts/duration.ts"; +import { type DurationInput, parseDuration } from "../../contracts/duration.ts"; import { BoundaError, ChainDepthExceededError, @@ -18,6 +25,8 @@ import { ConfigurationError, CreationOrderError, DomainError, + IdempotencyKeyConflictError, + IdempotencyKeyTakenError, NotFoundError, rejectionOf, ValidationError, @@ -33,6 +42,7 @@ import { withTimeout } from "../shared/timeout.ts"; import { withPorts } from "../shared/with-ports.ts"; import { ATTRIBUTES, METRICS, meter, SPAN_EVENTS, traced } from "../telemetry.ts"; import type { UnitStores } from "../unit-of-work/unit-of-work.ts"; +import { fingerprintOf } from "./fingerprint.ts"; import { validatePayload } from "./validate.ts"; export interface DispatchArgs { @@ -78,8 +88,7 @@ export interface CommandPipeline { export interface CreateCommandPipelineArgs { readonly aggregates: AggregatesRuntime; - readonly eventStore: EventStore; - readonly scheduler: Scheduler; + readonly storage: Storage; readonly config: ResolvedConfig; readonly ids: IdGenerator; readonly clock: Clock; @@ -118,6 +127,39 @@ export const scheduledCommandId: ScheduledCommandIdFunction = (dedupeKey) => ? dedupeKey.slice(SCHEDULED_COMMAND_PREFIX.length) : undefined; +type DispatchStores = Pick; + +interface Idempotency { + readonly key: string; + readonly fingerprint: string; +} + +const checkIdempotencyKey = (type: string, key: unknown): string => { + if (typeof key === "string" && key.length > 0 && key.length <= MAX_IDEMPOTENCY_KEY_LENGTH) { + return key; + } + throw new ValidationError(`Command "${type}" has an invalid idempotency key`, [ + { + path: ["idempotencyKey"], + message: `Expected a string of 1 to ${MAX_IDEMPOTENCY_KEY_LENGTH} characters`, + }, + ]); +}; + +const storedDispatch = ( + aggregate: AggregateRuntime, + aggregateId: string, + appended: AppendResult, +): StoredDispatch => ({ + scheduled: false, + aggregateType: aggregate.name, + aggregateId, + version: appended.version, + eventIds: appended.events.map((event) => event.id), + eventTypes: appended.events.map((event) => event.type), + position: appended.events.at(-1)?.position ?? 0, +}); + const resolveAggregateId = (aggregate: AggregateRuntime, payload: unknown): string => { const record = payload as Record; const value = record[aggregate.aggregateIdField] ?? record.id; @@ -268,8 +310,7 @@ const rejectOf = ( */ export const createCommandPipeline: CreateCommandPipelineFunction = ({ aggregates, - eventStore, - scheduler, + storage, config, ids, clock, @@ -339,15 +380,10 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ ); checkCreationOrder(aggregate, command, folded.created, events); if (events.length === 0) { - return { - scheduled: false, - aggregateType: aggregate.name, - aggregateId: command.aggregateId, + return storedDispatch(aggregate, command.aggregateId, { version: loaded.version, - eventIds: [], - eventTypes: [], - position: 0, - }; + events: [], + }); } try { const appended = await store.append({ @@ -356,15 +392,7 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ expectedVersion: loaded.version, events, }); - return { - scheduled: false, - aggregateType: aggregate.name, - aggregateId: command.aggregateId, - version: appended.version, - eventIds: appended.events.map((event) => event.id), - eventTypes: appended.events.map((event) => event.type), - position: appended.events.at(-1)?.position ?? 0, - }; + return storedDispatch(aggregate, command.aggregateId, appended); } catch (error) { if (!(error instanceof ConcurrencyError) || attempt >= attempts) throw error; logger.debug("command retried after concurrency conflict", { type: command.type, attempt }); @@ -377,7 +405,10 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ unit: "{command}", }); - const count = (type: string, outcome: "stored" | "scheduled" | "rejected" | "failed"): void => { + const count = ( + type: string, + outcome: "stored" | "scheduled" | "rejected" | "replayed" | "failed", + ): void => { commands.add(1, { [ATTRIBUTES.commandType]: type, [ATTRIBUTES.outcome]: outcome }); }; @@ -394,13 +425,20 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ within: UnitStores | undefined, rejected: (error: DomainError, rejection: CommandRejection) => Rejected, ): Promise => { - const stores = within ?? { eventStore, scheduler }; + const stores: DispatchStores = within ?? storage; const entry = aggregates.commandsByType[type]; if (entry === undefined) throw new NotFoundError(`Unknown command "${type}"`); const { aggregate, command: runtime } = entry; const depth = context?.depth ?? 0; const maxDepth = config.forAggregate(aggregate.name).policies.maxChainDepth; if (depth > maxDepth) throw new ChainDepthExceededError(depth, maxDepth); + const idempotency: Idempotency | undefined = + options.idempotencyKey === undefined + ? undefined + : { + key: checkIdempotencyKey(type, options.idempotencyKey), + fingerprint: await fingerprintOf({ type, payload }), + }; const scheduled = options.delay !== undefined; const input = scheduled ? asStored(payload) : payload; @@ -427,6 +465,98 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ const signals = [reactionSignal, options.signal].filter((one) => one !== undefined); for (const signal of signals) signal.throwIfAborted(); + const schedule = async ( + scheduler: Scheduler, + delay: DurationInput, + ): Promise => { + const executeAt = new Date(clock.now().getTime() + parseDuration(delay)); + await scheduler.schedule({ + dedupeKey: scheduledCommandKey(commandId), + command: { type, payload: input, aggregateId }, + executeAt, + context: { + correlationId: command.metadata.correlationId, + causationId: command.metadata.causationId, + depth, + }, + }); + return { + scheduled: true, + aggregateType: aggregate.name, + aggregateId, + executeAt: executeAt.toISOString(), + }; + }; + + const recordOf = ( + { key, fingerprint }: Idempotency, + outcome: DispatchOutcome, + ): PutIdempotencyRecordArgs => { + const now = clock.now(); + return { + key, + fingerprint, + outcome, + expiresAt: new Date( + now.getTime() + config.runtime.commands.idempotencyKeyLifetimeMs, + ).toISOString(), + now, + }; + }; + + const outcomeOf = (answer: DispatchResult | DomainError): DispatchOutcome => + answer instanceof DomainError + ? { + rejected: answer.rejected, + message: answer.message, + aggregateType: aggregate.name, + aggregateId, + } + : answer; + + // With a key, what the command answers is recorded under it where its events go: in the unit + // of work, or in the transaction of the append that stores them, so a duplicate running + // beside it loses on the key and stores nothing. + const perform = async (): Promise => { + if (idempotency === undefined) { + return options.delay === undefined + ? execute(aggregate, runtime, command, stores.eventStore, signals) + : schedule(stores.scheduler, options.delay); + } + if (options.delay !== undefined) { + const delay = options.delay; + const scheduling = async (transaction: DispatchStores): Promise => { + const answer = await schedule(transaction.scheduler, delay); + await transaction.idempotencyStore.put(recordOf(idempotency, answer)); + return answer; + }; + return within === undefined ? storage.transact(scheduling) : scheduling(within); + } + if (within !== undefined) { + const answer = await execute(aggregate, runtime, command, within.eventStore, signals); + await within.idempotencyStore.put(recordOf(idempotency, outcomeOf(answer))); + return answer; + } + let recorded = false; + const recording: EventStore = { + ...storage.eventStore, + append: async (args) => { + const appended = await storage.transact(async (transaction) => { + const result = await transaction.eventStore.append(args); + await transaction.idempotencyStore.put( + recordOf(idempotency, storedDispatch(aggregate, aggregateId, result)), + ); + return result; + }); + recorded = true; + return appended; + }, + }; + const answer = await execute(aggregate, runtime, command, recording, signals); + if (!recorded) await storage.idempotencyStore.put(recordOf(idempotency, outcomeOf(answer))); + return answer; + }; + return traced({ name: `bounda.command ${type}`, attributes: { @@ -437,52 +567,65 @@ export const createCommandPipeline: CreateCommandPipelineFunction = ({ [ATTRIBUTES.causationId]: command.metadata.causationId, }, run: async (span): Promise => { - if (options.delay !== undefined) { - const executeAt = new Date(clock.now().getTime() + parseDuration(options.delay)); - await stores.scheduler.schedule({ - dedupeKey: scheduledCommandKey(commandId), - command: { type, payload: input, aggregateId }, - executeAt, - context: { - correlationId: command.metadata.correlationId, - causationId: command.metadata.causationId, - depth, - }, + const replay = async (): Promise => { + if (idempotency === undefined) return null; + const found = await stores.idempotencyStore.get({ + key: idempotency.key, + now: clock.now(), }); - span.setAttribute(ATTRIBUTES.outcome, "scheduled"); - count(type, "scheduled"); - return { - scheduled: true, - aggregateType: aggregate.name, - aggregateId, - executeAt: executeAt.toISOString(), - }; - } - let result: StoredDispatch | DomainError; + if (found === null) return null; + if (found.fingerprint !== idempotency.fingerprint) { + throw new IdempotencyKeyConflictError({ + idempotencyKey: idempotency.key, + commandType: type, + }); + } + span.setAttribute(ATTRIBUTES.outcome, "replayed"); + count(type, "replayed"); + const { outcome } = found; + if (!("rejected" in outcome)) return outcome; + span.addEvent(SPAN_EVENTS.commandRejected, { [ATTRIBUTES.rejected]: outcome.rejected }); + return rejected(new DomainError(rejectionOf(outcome.rejected, outcome.message)), { + type, + ...outcome, + }); + }; + const remembered = await replay(); + if (remembered !== null) return remembered; + let answer: DispatchResult | DomainError; try { - result = await execute(aggregate, runtime, command, stores.eventStore, signals); + answer = await perform(); } catch (error) { + if (error instanceof IdempotencyKeyTakenError) { + const taken = await replay(); + if (taken !== null) return taken; + } count(type, "failed"); throw error; } - if (result instanceof DomainError) { + if (answer instanceof DomainError) { span.setAttribute(ATTRIBUTES.outcome, "rejected"); - span.addEvent(SPAN_EVENTS.commandRejected, { [ATTRIBUTES.rejected]: result.rejected }); + span.addEvent(SPAN_EVENTS.commandRejected, { [ATTRIBUTES.rejected]: answer.rejected }); count(type, "rejected"); - return rejected(result, { + return rejected(answer, { type, - rejected: result.rejected, - message: result.message, + rejected: answer.rejected, + message: answer.message, aggregateType: aggregate.name, aggregateId, }); } + if (answer.scheduled) { + span.setAttribute(ATTRIBUTES.outcome, "scheduled"); + count(type, "scheduled"); + return answer; + } span.setAttributes({ [ATTRIBUTES.outcome]: "stored", - [ATTRIBUTES.eventCount]: result.eventIds.length, + [ATTRIBUTES.eventCount]: answer.eventIds.length, }); count(type, "stored"); - return result; + return answer; }, }); }; diff --git a/packages/core/src/kernel/process/parked.test.ts b/packages/core/src/kernel/process/parked.test.ts index 96e07792..06877d91 100644 --- a/packages/core/src/kernel/process/parked.test.ts +++ b/packages/core/src/kernel/process/parked.test.ts @@ -377,28 +377,16 @@ describe("events of a failed process", () => { const [letter] = await deadLetters.list(); const transact = harness.storage.transact; let arrived = false; - // The archive arrives once the first transaction of the retry that schedules has decided what - // to write, before it commits. - harness.storage.transact = (work) => - transact(async (tx) => { - let schedules = false; - const result = await work({ - ...tx, - scheduler: { - ...tx.scheduler, - schedule: (args) => { - schedules = true; - return tx.scheduler.schedule(args); - }, - }, - }); - if (schedules && !arrived) { - arrived = true; - await harness.pipeline.dispatch({ type: "ArchiveOrder", payload: { orderId: "o-1" } }); - await harness.dispatcher.runUntilIdle(); - } - return result; - }); + // The archive arrives once the retried step is committed, as the resume that follows it has + // decided what to write and is about to commit. + harness.storage.transact = async (work) => { + if (!arrived && (await types()).at(-1) === PROCESS_EVENTS.handled) { + arrived = true; + await harness.pipeline.dispatch({ type: "ArchiveOrder", payload: { orderId: "o-1" } }); + await harness.dispatcher.runUntilIdle(); + } + return transact(work); + }; await deadLetters.retry(letter?.id ?? ""); harness.storage.transact = transact; expect(arrived).toBe(true); diff --git a/packages/core/src/kernel/reactive-harness.ts b/packages/core/src/kernel/reactive-harness.ts index 5c5b352c..4e236128 100644 --- a/packages/core/src/kernel/reactive-harness.ts +++ b/packages/core/src/kernel/reactive-harness.ts @@ -86,8 +86,7 @@ export const createReactiveHarness: CreateReactiveHarnessFunction = async ({ const readModels = await buildReadModels({ registry, config, logger }); const pipeline = createCommandPipeline({ aggregates, - eventStore: storage.eventStore, - scheduler: storage.scheduler, + storage, config, ids, clock, diff --git a/packages/core/src/kernel/scheduler/worker.test.ts b/packages/core/src/kernel/scheduler/worker.test.ts index 7a7ccd1d..ba0c6c4b 100644 --- a/packages/core/src/kernel/scheduler/worker.test.ts +++ b/packages/core/src/kernel/scheduler/worker.test.ts @@ -21,6 +21,7 @@ import { sentMessages, slowJob, } from "../test-support.ts"; +import { IDEMPOTENCY_PURGE_INTERVAL_MS } from "./worker.ts"; const LOST_CLAIM = "scheduled command no longer holds its claim; this run wrote nothing"; @@ -52,6 +53,54 @@ const failingRegistry: Registry = { }; describe("scheduled command worker", () => { + it("purges the expired idempotency keys once per interval, and goes on when it cannot", async () => { + const { logger, entries } = createRecordingLogger(); + const harness = await createReactiveHarness({ registry: orderRegistry, logger }); + const { idempotencyStore } = harness.storage; + const { purge } = idempotencyStore; + const record = (key: string, expiresInMs: number) => ({ + key, + fingerprint: "fp", + outcome: { rejected: "Nope", message: "no", aggregateType: "order", aggregateId: "1" }, + expiresAt: new Date(harness.clock.now().getTime() + expiresInMs).toISOString(), + now: harness.clock.now(), + }); + await idempotencyStore.put(record("gone", -1)); + await idempotencyStore.put(record("kept", 30_000)); + expect(await harness.worker.runOnce()).toBe(0); + expect(await idempotencyStore.get({ key: "gone", now: harness.clock.now() })).toBeNull(); + expect(entries).toContainEqual({ + level: "debug", + message: "expired idempotency keys purged", + fields: { purged: 1 }, + }); + + harness.clock.advance(IDEMPOTENCY_PURGE_INTERVAL_MS - 1); + await harness.worker.runOnce(); + expect( + await harness.storage.idempotencyStore.get({ key: "kept", now: new Date(0) }), + ).not.toBeNull(); + harness.clock.advance(1); + harness.storage.idempotencyStore.purge = async () => { + throw new Error("down"); + }; + expect(await harness.worker.runOnce()).toBe(0); + expect(entries).toContainEqual({ + level: "error", + message: "expired idempotency keys could not be purged", + fields: expect.objectContaining({ message: "down" }), + }); + harness.storage.idempotencyStore.purge = purge; + harness.clock.advance(IDEMPOTENCY_PURGE_INTERVAL_MS); + await harness.worker.runOnce(); + expect( + await harness.storage.idempotencyStore.get({ key: "kept", now: new Date(0) }), + ).toBeNull(); + expect( + entries.filter((entry) => entry.message === "expired idempotency keys purged"), + ).toHaveLength(2); + }); + it("runs due commands with their stored context and completes them", async () => { const harness = await createReactiveHarness({ registry: orderRegistry }); sentMessages.length = 0; diff --git a/packages/core/src/kernel/scheduler/worker.ts b/packages/core/src/kernel/scheduler/worker.ts index 5292456c..fd28c31d 100644 --- a/packages/core/src/kernel/scheduler/worker.ts +++ b/packages/core/src/kernel/scheduler/worker.ts @@ -59,6 +59,11 @@ export interface CreateScheduledCommandWorkerFunction { const CLAIM_LIMIT = 50; +/** + * How often a worker that runs a pass purges the expired idempotency keys. + */ +export const IDEMPOTENCY_PURGE_INTERVAL_MS = 60_000; + /** * A renewal before a rerun that failed in the store: the store's failure, not the run's, so it is * neither retried nor dead-lettered as one. @@ -87,7 +92,8 @@ interface DeadlineWait { * claim's failure. A deadline first waits for the process runner, so an event that cancels it is * seen first; this is best effort, since the aggregates still check what it sends. Deadline * entries are never removed here: the process runner writes what comes next, so a crash in between - * leaves the entry to its lease instead of losing it. + * leaves the entry to its lease instead of losing it. Every `IDEMPOTENCY_PURGE_INTERVAL_MS` a pass + * also purges the idempotency keys that expired. */ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction = ({ storage, @@ -117,6 +123,18 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction const startWithinMs = timeoutMs / 2; const waits = new Map(); + let purgedAt = Number.NEGATIVE_INFINITY; + + const purgeExpiredKeys = async (now: Date): Promise => { + if (now.getTime() - purgedAt < IDEMPOTENCY_PURGE_INTERVAL_MS) return; + purgedAt = now.getTime(); + try { + const purged = await storage.idempotencyStore.purge({ now }); + if (purged > 0) logger.debug("expired idempotency keys purged", { purged }); + } catch (error) { + logger.error("expired idempotency keys could not be purged", errorDetails(error)); + } + }; const isDeadline = (entry: ScheduledCommand): boolean => entry.command.type === PROCESS_DEADLINE_COMMAND; @@ -399,6 +417,7 @@ export const createScheduledCommandWorker: CreateScheduledCommandWorkerFunction for (const key of waits.keys()) { if (!due.some((entry) => entry.dedupeKey === key)) waits.delete(key); } + await purgeExpiredKeys(claimedAt); if (unreadable !== undefined) throw unreadable.error; return due.length; }); diff --git a/packages/core/src/kernel/shared/retry.ts b/packages/core/src/kernel/shared/retry.ts index a8097f6a..7132e970 100644 --- a/packages/core/src/kernel/shared/retry.ts +++ b/packages/core/src/kernel/shared/retry.ts @@ -11,6 +11,7 @@ const TERMINAL_CODES: ReadonlySet = new Set([ "NOT_FOUND", "CHAIN_DEPTH_EXCEEDED", "CREATION_ORDER", + "IDEMPOTENCY_KEY_CONFLICT", ]); export interface ClassifyFailureFunction { diff --git a/packages/core/src/kernel/test-support.ts b/packages/core/src/kernel/test-support.ts index b60e4367..666be47c 100644 --- a/packages/core/src/kernel/test-support.ts +++ b/packages/core/src/kernel/test-support.ts @@ -319,8 +319,7 @@ export const createKernelHarness: CreateKernelHarnessFunction = async ({ const rejections: CommandRejection[] = []; const pipeline = createCommandPipeline({ aggregates, - eventStore: storage.eventStore, - scheduler: storage.scheduler, + storage, config, ids: createSequentialIdGenerator(), clock, diff --git a/packages/core/src/kernel/unit-of-work/unit-of-work.ts b/packages/core/src/kernel/unit-of-work/unit-of-work.ts index a8014c3e..84bf2960 100644 --- a/packages/core/src/kernel/unit-of-work/unit-of-work.ts +++ b/packages/core/src/kernel/unit-of-work/unit-of-work.ts @@ -6,7 +6,10 @@ import { ConcurrencyError } from "../../contracts/errors.ts"; /** * What a reaction's commands write through while its attempt runs. */ -export type UnitStores = Pick; +export type UnitStores = Pick< + UnitOfWork, + "eventStore" | "scheduler" | "idempotencyStore" | "afterCommit" +>; /** * What one attempt of a reaction changes in the store, held back until `commit`: the events its diff --git a/packages/core/src/memory/event-store.ts b/packages/core/src/memory/event-store.ts index a0b504de..a01d6d96 100644 --- a/packages/core/src/memory/event-store.ts +++ b/packages/core/src/memory/event-store.ts @@ -5,12 +5,32 @@ import { streamId } from "../contracts/event.ts"; import { toJson } from "./json-copy.ts"; export interface CreateMemoryEventStoreArgs { - // Receives the position of the last event stored. + // Receives the last position readers can see. readonly onAppend?: (position: number) => void; } +/** + * A batch to append whose events may already hold their positions, reserved while they were + * staged; an event at 0, or a batch without `positions`, takes the next one when written. + */ +export interface PositionedAppendArgs extends AppendArgs { + readonly positions?: readonly number[]; +} + export interface MemoryEventStore extends EventStore { - appendAll(batches: readonly AppendArgs[]): Promise; + appendAll(batches: readonly PositionedAppendArgs[]): Promise; + /** + * Takes the next position for an event a transaction stages. Readers stop short of it until + * the transaction writes it with `appendAll` or gives it back with `release`, so an event + * written meanwhile at a later position never overtakes it. + */ + reserve(): number; + /** + * Gives back positions a transaction reserved and will not write: the log's tail moves back + * before them, and whatever is left among written events becomes a gap readers skip, as a + * rolled-back sequence leaves on PostgreSQL. + */ + release(positions: readonly number[]): void; } export interface CreateMemoryEventStoreFunction { @@ -38,10 +58,27 @@ const restore = ({ payload, metadata, ...rest }: Kept): StoredEvent => ({ export const createMemoryEventStore: CreateMemoryEventStoreFunction = ({ onAppend } = {}) => { const streams = new Map(); - const global: Kept[] = []; + const log = new Map(); + const gaps = new Set(); + let reserved = 0; + // Every position up to here is written or a gap: what readers see. + let watermark = 0; + + const advance = (): void => { + const before = watermark; + while (log.has(watermark + 1) || gaps.delete(watermark + 1)) watermark += 1; + if (watermark > before) onAppend?.(watermark); + }; + + const reserve = (): number => { + reserved += 1; + return reserved; + }; // Atomic because nothing yields between the version checks and the writes. - const appendAll = async (batches: readonly AppendArgs[]): Promise => { + const appendAll = async ( + batches: readonly PositionedAppendArgs[], + ): Promise => { const versions = new Map(); for (const { aggregateType, aggregateId, expectedVersion, events } of batches) { const key = streamId({ aggregateType, aggregateId }); @@ -52,30 +89,45 @@ export const createMemoryEventStore: CreateMemoryEventStoreFunction = ({ onAppen versions.set(key, actualVersion + events.length); } // Every batch is kept before any is written, so one that JSON refuses appends nothing. - let position = global.length; - const prepared = batches.map(({ aggregateType, aggregateId, expectedVersion, events }) => { - const stored = events.map((event) => { - position += 1; - return { ...event, position }; - }); - return { + const prepared = batches.map( + ({ aggregateType, aggregateId, expectedVersion, events, positions }) => ({ key: streamId({ aggregateType, aggregateId }), - kept: stored.map(keep), - result: { version: expectedVersion + stored.length, events: stored }, + expectedVersion, + events, + positions, + kept: events.map((event) => keep({ ...event, position: 0 })), + }), + ); + const results = prepared.map(({ key, expectedVersion, events, positions, kept }) => { + const written = kept.map((event, index) => ({ + ...event, + position: positions?.[index] || reserve(), + })); + streams.set(key, [...(streams.get(key) ?? []), ...written]); + for (const event of written) log.set(event.position, event); + return { + version: expectedVersion + written.length, + events: events.map((event, index) => ({ + ...event, + position: written[index]?.position ?? 0, + })), }; }); - for (const { key, kept } of prepared) { - streams.set(key, [...(streams.get(key) ?? []), ...kept]); - global.push(...kept); - } - const results = prepared.map(({ result }) => result); - if (results.some((result) => result.events.length > 0)) onAppend?.(global.length); + advance(); return results; }; return { appendAll, append: async (args) => (await appendAll([args]))[0] as AppendResult, + reserve, + release: (positions) => { + for (const position of [...positions].sort((a, b) => b - a)) { + if (position === reserved) reserved -= 1; + else gaps.add(position); + } + advance(); + }, load: async ({ aggregateType, aggregateId, fromVersion = 1 }) => { const stream = streams.get(streamId({ aggregateType, aggregateId })) ?? []; return { @@ -83,8 +135,14 @@ export const createMemoryEventStore: CreateMemoryEventStoreFunction = ({ onAppen version: stream.length, }; }, - readAll: async ({ afterPosition, limit }) => - global.slice(afterPosition, afterPosition + limit).map(restore), - lastPosition: async () => global.length, + readAll: async ({ afterPosition, limit }) => { + const events: StoredEvent[] = []; + for (let at = afterPosition + 1; at <= watermark && events.length < limit; at += 1) { + const kept = log.get(at); + if (kept !== undefined) events.push(restore(kept)); + } + return events; + }, + lastPosition: async () => watermark, }; }; diff --git a/packages/core/src/memory/idempotency-store.ts b/packages/core/src/memory/idempotency-store.ts new file mode 100644 index 00000000..c0550dd9 --- /dev/null +++ b/packages/core/src/memory/idempotency-store.ts @@ -0,0 +1,51 @@ +import type { IdempotencyRecord, IdempotencyStore } from "../adapter/storage/idempotency-store.ts"; +import { IdempotencyKeyTakenError } from "../contracts/errors.ts"; +import { createStoreEntries, type WithEntries } from "./entries.ts"; +import { jsonCopy } from "./json-copy.ts"; + +export interface CreateMemoryIdempotencyStoreFunction { + (): IdempotencyStore; +} + +export interface CreateKeptIdempotencyStoreFunction { + (): WithEntries; +} + +const expired = (record: IdempotencyRecord, now: Date): boolean => + new Date(record.expiresAt).getTime() <= now.getTime(); + +const copy = (record: IdempotencyRecord): IdempotencyRecord => ({ + ...record, + outcome: jsonCopy(record.outcome), +}); + +export const createKeptIdempotencyStore: CreateKeptIdempotencyStoreFunction = () => { + const records = new Map(); + + const store: IdempotencyStore = { + get: async ({ key, now }) => { + const record = records.get(key); + return record === undefined || expired(record, now) ? null : copy(record); + }, + put: async ({ now, ...record }) => { + const existing = records.get(record.key); + if (existing !== undefined && !expired(existing, now)) { + throw new IdempotencyKeyTakenError(record.key); + } + records.set(record.key, copy(record)); + }, + purge: async ({ now }) => { + let purged = 0; + for (const [key, record] of records) { + if (!expired(record, now)) continue; + records.delete(key); + purged += 1; + } + return purged; + }, + }; + return { store, entries: createStoreEntries(records, (key: string) => key) }; +}; + +export const createMemoryIdempotencyStore: CreateMemoryIdempotencyStoreFunction = () => + createKeptIdempotencyStore().store; diff --git a/packages/core/src/memory/index.ts b/packages/core/src/memory/index.ts index 210b6b22..db63c3c2 100644 --- a/packages/core/src/memory/index.ts +++ b/packages/core/src/memory/index.ts @@ -15,6 +15,7 @@ import { createMemoryCheckpointStore } from "./checkpoint-store.ts"; import { createKeptDeadLetterStore } from "./dead-letter-store.ts"; import { createMemoryEventNotifier } from "./event-notifier.ts"; import { createMemoryEventStore } from "./event-store.ts"; +import { createKeptIdempotencyStore } from "./idempotency-store.ts"; import { createKeptInboxLedger } from "./inbox-ledger.ts"; import { createKeptScheduler } from "./scheduler.ts"; import { createMemoryReadClient, createMemoryTable, type MemoryTable } from "./table.ts"; @@ -100,11 +101,13 @@ export const memory: MemoryFunction = (options = {}) => { const inboxLedger = createKeptInboxLedger(); const deadLetterStore = createKeptDeadLetterStore(); const scheduler = createKeptScheduler(); + const idempotencyStore = createKeptIdempotencyStore(); const stores = { eventStore: createMemoryEventStore({ onAppend: notifier.notify }), inboxLedger: inboxLedger.store, deadLetterStore: deadLetterStore.store, scheduler: scheduler.store, + idempotencyStore: idempotencyStore.store, }; storage ??= { ...stores, @@ -116,6 +119,7 @@ export const memory: MemoryFunction = (options = {}) => { inboxLedger: inboxLedger.entries, deadLetterStore: deadLetterStore.entries, scheduler: scheduler.entries, + idempotencyStore: idempotencyStore.entries, }, }), close: async () => {}, diff --git a/packages/core/src/memory/memory.test.ts b/packages/core/src/memory/memory.test.ts index 79d0045f..579987f6 100644 --- a/packages/core/src/memory/memory.test.ts +++ b/packages/core/src/memory/memory.test.ts @@ -6,6 +6,7 @@ import { contractFields, deadLetterStoreContract, eventStoreContract, + idempotencyStoreContract, inboxLedgerContract, pendingEvent, readModelRebuildContract, @@ -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 }); + idempotencyStoreContract({ create: async () => (await storage()).idempotencyStore }); storageTransactionContract({ create: storage }); tableContract({ create: async () => { @@ -107,6 +109,152 @@ describe("memory adapter", () => { ).rejects.toMatchObject({ message: "undefined has no JSON to store" }); }); + it("answers, inside a transaction, the positions its events commit with", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + await storage.eventStore.append({ + aggregateType: "order", + aggregateId: "1", + expectedVersion: 0, + events: [pendingEvent({ aggregateId: "1", version: 1 })], + }); + const inside = await storage.transact(async (tx) => { + const order = await tx.eventStore.append({ + aggregateType: "order", + aggregateId: "1", + expectedVersion: 1, + events: [2, 3].map((version) => + pendingEvent({ aggregateId: "1", version, type: "OrderPaid" }), + ), + }); + const payment = await tx.eventStore.append({ + aggregateType: "payment", + aggregateId: "9", + expectedVersion: 0, + events: [pendingEvent({ aggregateType: "payment", aggregateId: "9", version: 1 })], + }); + expect(await tx.eventStore.lastPosition()).toBe(1); + return [...order.events, ...payment.events].map((event) => event.position); + }); + expect(inside).toEqual([2, 3, 4]); + expect( + (await storage.eventStore.readAll({ afterPosition: 0, limit: 10 })).map( + (event) => event.position, + ), + ).toEqual([1, 2, 3, 4]); + }); + + it("shows an event appended beside a transaction only once the transaction settles, in position order", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + const heard: (number | undefined)[] = []; + await (storage.notifier as NonNullable).subscribe((position) => { + heard.push(position); + }); + const beside = (version: number) => + storage.eventStore.append({ + aggregateType: "order", + aggregateId: "2", + expectedVersion: version - 1, + events: [pendingEvent({ aggregateId: "2", version })], + }); + const committing = Promise.withResolvers(); + const transaction = storage.transact(async (tx) => { + await tx.eventStore.append({ + aggregateType: "order", + aggregateId: "1", + expectedVersion: 0, + events: [pendingEvent({ aggregateId: "1", version: 1 })], + }); + await committing.promise; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect((await beside(1)).events.map((event) => event.position)).toEqual([2]); + expect(await storage.eventStore.lastPosition()).toBe(0); + expect(await storage.eventStore.readAll({ afterPosition: 0, limit: 10 })).toEqual([]); + expect(heard).toEqual([]); + committing.resolve(); + await transaction; + expect(await storage.eventStore.lastPosition()).toBe(2); + expect( + (await storage.eventStore.readAll({ afterPosition: 0, limit: 10 })).map((event) => [ + event.position, + event.aggregateId, + ]), + ).toEqual([ + [1, "1"], + [2, "2"], + ]); + expect(heard).toEqual([2]); + expect((await beside(2)).events.map((event) => event.position)).toEqual([3]); + expect(heard).toEqual([2, 3]); + }); + + it("gives back the positions of a transaction that fails: the tail when nothing came after, a gap readers skip otherwise", async () => { + const storage = await memory().createStorage({ logger: silentLogger }); + const heard: (number | undefined)[] = []; + await (storage.notifier as NonNullable).subscribe((position) => { + heard.push(position); + }); + const failing = (beside?: () => Promise) => + storage + .transact(async (tx) => { + await tx.eventStore.append({ + aggregateType: "order", + aggregateId: "1", + expectedVersion: 0, + events: [pendingEvent({ aggregateId: "1", version: 1 })], + }); + await beside?.(); + throw new Error("boom"); + }) + .catch(() => undefined); + await failing(); + expect( + ( + await storage.eventStore.append({ + aggregateType: "order", + aggregateId: "2", + expectedVersion: 0, + events: [pendingEvent({ aggregateId: "2", version: 1 })], + }) + ).events.map((event) => event.position), + ).toEqual([1]); + expect(heard).toEqual([1]); + + await failing(async () => { + await storage.eventStore.append({ + aggregateType: "order", + aggregateId: "3", + expectedVersion: 0, + events: [pendingEvent({ aggregateId: "3", version: 1 })], + }); + expect(await storage.eventStore.lastPosition()).toBe(1); + }); + expect(await storage.eventStore.lastPosition()).toBe(3); + expect(heard).toEqual([1, 3]); + expect( + (await storage.eventStore.readAll({ afterPosition: 0, limit: 10 })).map((event) => [ + event.position, + event.aggregateId, + ]), + ).toEqual([ + [1, "2"], + [3, "3"], + ]); + expect(await storage.eventStore.readAll({ afterPosition: 1, limit: 1 })).toMatchObject([ + { position: 3 }, + ]); + expect( + ( + await storage.eventStore.append({ + aggregateType: "order", + aggregateId: "4", + expectedVersion: 0, + events: [pendingEvent({ aggregateId: "4", version: 1 })], + }) + ).events.map((event) => event.position), + ).toEqual([4]); + }); + it("puts every store back when a write fails while a transaction is being applied", async () => { const storage = await memory().createStorage({ logger: silentLogger }); const key = { handler: "order.p", eventId: "e1" }; @@ -172,10 +320,18 @@ describe("memory adapter", () => { context, }); await tx.deadLetterStore.add(letter("lost")); + await tx.idempotencyStore.put({ + key: "lost", + fingerprint: "fp", + outcome: { rejected: "Nope", message: "no", aggregateType: "order", aggregateId: "1" }, + expiresAt: "2026-01-02T00:00:00.000Z", + now, + }); await tx.inboxLedger.fail({ ...key, error: "boom" }); await tx.inboxLedger.complete(key); }), ).rejects.toThrow("connection lost"); + expect(await storage.idempotencyStore.get({ key: "lost", now })).toBeNull(); expect(await storage.eventStore.lastPosition()).toBe(1); expect( (await storage.eventStore.load({ aggregateType: "order", aggregateId: "1" })).events.map( diff --git a/packages/core/src/memory/transaction.ts b/packages/core/src/memory/transaction.ts index ce7ed133..4a042f25 100644 --- a/packages/core/src/memory/transaction.ts +++ b/packages/core/src/memory/transaction.ts @@ -3,6 +3,7 @@ import { type DeferredWriteTarget, deferWrites } from "../adapter/deferred-write import { createStagedEventStore } from "../adapter/staged-event-store.ts"; import type { CheckpointStore } from "../adapter/storage/checkpoint-store.ts"; import type { DeadLetterStore } from "../adapter/storage/dead-letter-store.ts"; +import type { IdempotencyStore } from "../adapter/storage/idempotency-store.ts"; import type { ClaimKey, InboxLedger } from "../adapter/storage/inbox-ledger.ts"; import type { Scheduler } from "../adapter/storage/scheduler.ts"; import type { StoreEntries } from "./entries.ts"; @@ -13,10 +14,12 @@ export interface CreateMemoryStorageTransactionArgs { readonly inboxLedger: InboxLedger; readonly deadLetterStore: DeadLetterStore; readonly scheduler: Scheduler; + readonly idempotencyStore: IdempotencyStore; readonly entries: { readonly inboxLedger: StoreEntries; readonly deadLetterStore: StoreEntries; readonly scheduler: StoreEntries; + readonly idempotencyStore: StoreEntries; }; } @@ -35,6 +38,7 @@ const createUndoLog = ({ inboxLedger, deadLetterStore, scheduler, + idempotencyStore, entries, }: CreateUndoLogArgs): UndoLog => { const undos: (() => void)[] = []; @@ -76,6 +80,10 @@ const createUndoLog = ({ defer: (args) => tracked(entries.scheduler, args.claim.dedupeKey, () => scheduler.defer(args)), }, + idempotencyStore: { + put: (args) => + tracked(entries.idempotencyStore, args.key, () => idempotencyStore.put(args)), + }, }, undo: () => { for (const undo of undos.splice(0).reverse()) undo(); @@ -84,11 +92,14 @@ const createUndoLog = ({ }; /** - * `Storage.transact` for the memory stores. The work's writes are held back; transactions - * then commit one at a time, every stream appended in one synchronous run, so no reader sees one - * stream appended without the others. A failed commit appends nothing and puts back each entry it - * changed, unless someone changed it since. `tryClaim`, `claimDue`, both `renew` and writes made on - * the stores themselves do not wait for a commit, so they can see a write it may still put back. + * `Storage.transact` for the memory stores. The work's writes are held back, each event staged + * with the position it will have, reserved from the store, so the work answers the positions + * its appends will commit with; transactions then commit one at a time, every stream appended + * in one synchronous run, so no reader sees one stream appended without the others, nor an + * event written meanwhile past a position still reserved. A failed transaction appends nothing, + * gives its positions back and puts back each entry it changed, unless someone changed it since. + * `tryClaim`, `claimDue`, both `renew`, `purge` and writes made on the stores themselves do not + * wait for a commit, so they can see a write it may still put back. */ export const createMemoryStorageTransaction: CreateMemoryStorageTransactionFunction = ({ eventStore, @@ -96,22 +107,34 @@ export const createMemoryStorageTransaction: CreateMemoryStorageTransactionFunct }) => { const commits = createMemoryLocks(); return async (work) => { - const staged = createStagedEventStore(eventStore); + const reserved: number[] = []; + const staged = createStagedEventStore(eventStore, { + nextPosition: () => { + const position = eventStore.reserve(); + reserved.push(position); + return position; + }, + }); const { stores, flush } = deferWrites(live); - const result = await work({ eventStore: staged, ...stores }); - // A transaction committed from inside this commit, as from a store write, waits forever. - const release = await commits.acquire("commit", true); - const log = createUndoLog(live); try { - await flush(log.target); - await eventStore.appendAll(staged.batches()); - } catch (error) { - log.undo(); - throw error; + const result = await work({ eventStore: staged, ...stores }); + // A transaction committed from inside this commit, as from a store write, waits forever. + const release = await commits.acquire("commit", true); + const log = createUndoLog(live); + try { + await flush(log.target); + await eventStore.appendAll(staged.batches()); + } catch (error) { + log.undo(); + throw error; + } finally { + release?.(); + } + reserved.length = 0; + return result; } finally { - release?.(); + eventStore.release(reserved); } - return result; }; }; diff --git a/packages/core/test-types/config.test-d.ts b/packages/core/test-types/config.test-d.ts index 17809482..55bb26c8 100644 --- a/packages/core/test-types/config.test-d.ts +++ b/packages/core/test-types/config.test-d.ts @@ -48,6 +48,21 @@ describe("defineConfig", () => { }, }, }); + defineConfig({ + storage: sqlite, + ports, + runtime: { commands: { idempotencyKeyLifetime: "48h" } }, + }); + defineConfig({ + storage: sqlite, + ports, + runtime: { + commands: { + // @ts-expect-error "two days" is not a duration string + idempotencyKeyLifetime: "two days", + }, + }, + }); defineConfig({ storage: sqlite, ports, @@ -143,6 +158,12 @@ describe("defineConfig", () => { // @ts-expect-error concurrencyRetries is not overridden per aggregate runtime: { overrides: { order: { commands: { concurrencyRetries: 1 } } } }, }); + defineConfig({ + storage: sqlite, + ports, + // @ts-expect-error idempotencyKeyLifetime is not overridden per aggregate + runtime: { overrides: { order: { commands: { idempotencyKeyLifetime: "1h" } } } }, + }); // @ts-expect-error batch is not a role defineConfig({ storage: sqlite, ports, runtime: { role: "batch" } }); }); diff --git a/packages/core/test-types/inference.test-d.ts b/packages/core/test-types/inference.test-d.ts index 35eda922..da7425cf 100644 --- a/packages/core/test-types/inference.test-d.ts +++ b/packages/core/test-types/inference.test-d.ts @@ -544,6 +544,7 @@ describe("facades", () => { DispatchOptions | undefined >(); expectTypeOf().toEqualTypeOf(); + expectTypeOf().toEqualTypeOf(); expectTypeOf>().toEqualTypeOf>(); expectTypeOf().toEqualTypeOf(); expectTypeOf().toEqualTypeOf<{ @@ -577,6 +578,12 @@ describe("facades", () => { expectTypeOf( commands.payOrder(payment, { correlationId: "c-1" }), ).resolves.toEqualTypeOf(); + expectTypeOf( + commands.payOrder(payment, { idempotencyKey: "req-1" }), + ).resolves.toEqualTypeOf(); + expectTypeOf( + commands.payOrder(payment, { idempotencyKey: "req-1", delay: "1h" }), + ).resolves.toEqualTypeOf(); expectTypeOf( commands.payOrder(payment, { delay: "1h", signal }), ).resolves.toEqualTypeOf(); diff --git a/packages/postgresql/src/idempotency-store.ts b/packages/postgresql/src/idempotency-store.ts new file mode 100644 index 00000000..d74ebaeb --- /dev/null +++ b/packages/postgresql/src/idempotency-store.ts @@ -0,0 +1,60 @@ +import { IdempotencyKeyTakenError } from "@bounda-dev/core"; +import type { + DispatchOutcome, + IdempotencyRecord, + IdempotencyStore, +} from "@bounda-dev/core/adapter"; +import { postgresqlDialect } from "@bounda-dev/core/adapter/sql"; +import type { PostgresqlDatabase } from "./database.ts"; + +export interface CreatePostgresqlIdempotencyStoreArgs { + readonly db: PostgresqlDatabase; + readonly table: string; +} + +export interface CreatePostgresqlIdempotencyStoreFunction { + (args: CreatePostgresqlIdempotencyStoreArgs): IdempotencyStore; +} + +const toRecord = (row: Record): IdempotencyRecord => ({ + key: String(row.key), + fingerprint: String(row.fingerprint), + outcome: row.outcome as DispatchOutcome, + expiresAt: String(row.expires_at), +}); + +/** + * `put` is a single upsert: PostgreSQL locks the conflicting row, so of two dispatches racing for + * one key exactly one gets its row back. + */ +export const createPostgresqlIdempotencyStore: CreatePostgresqlIdempotencyStoreFunction = ({ + db, + table, +}) => ({ + get: async ({ key, now }) => { + const [row] = await db.all( + `SELECT "key", "fingerprint", "outcome", "expires_at" FROM ${table} WHERE "key" = $1 AND "expires_at" > $2`, + [key, now.toISOString()], + ); + return row === undefined ? null : toRecord(row); + }, + put: async ({ key, fingerprint, outcome, expiresAt, now }) => { + const rows = await db.all( + `INSERT INTO ${table} ("key", "fingerprint", "outcome", "expires_at") VALUES ($1, $2, $3, $4) + ON CONFLICT ("key") DO UPDATE SET + "fingerprint" = excluded."fingerprint", + "outcome" = excluded."outcome", + "expires_at" = excluded."expires_at" + WHERE ${table}."expires_at" <= $5 + RETURNING "key"`, + [key, fingerprint, postgresqlDialect.encode("json", outcome), expiresAt, now.toISOString()], + ); + if (rows.length === 0) throw new IdempotencyKeyTakenError(key); + }, + purge: async ({ now }) => { + const rows = await db.all(`DELETE FROM ${table} WHERE "expires_at" <= $1 RETURNING "key"`, [ + now.toISOString(), + ]); + return rows.length; + }, +}); diff --git a/packages/postgresql/src/index.ts b/packages/postgresql/src/index.ts index 907c88ef..731c61dc 100644 --- a/packages/postgresql/src/index.ts +++ b/packages/postgresql/src/index.ts @@ -11,6 +11,7 @@ import { createPostgresqlDatabase, type PostgresqlDatabase } from "./database.ts import { createPostgresqlDeadLetterStore } from "./dead-letter-store.ts"; import { createPostgresqlEventNotifier } from "./event-notifier.ts"; import { createPostgresqlEventStore } from "./event-store.ts"; +import { createPostgresqlIdempotencyStore } from "./idempotency-store.ts"; import { createPostgresqlInboxLedger } from "./inbox-ledger.ts"; import { type PostgresqlOptions, resolvePostgresqlOptions } from "./options.ts"; import { openPostgresqlReadModel, rebuildPostgresqlReadModel } from "./read-model.ts"; @@ -102,6 +103,10 @@ export const postgresql: PostgresqlFunction = (options) => { table: tables.deadLetters, }), scheduler: createPostgresqlScheduler({ db: database, table: tables.scheduledCommands }), + idempotencyStore: createPostgresqlIdempotencyStore({ + db: database, + table: tables.idempotencyKeys, + }), }); const boundTo = (tx: SqlTransaction): PostgresqlDatabase => ({ run: tx.run, diff --git a/packages/postgresql/src/postgresql.test.ts b/packages/postgresql/src/postgresql.test.ts index 2512c3fe..06c80a85 100644 --- a/packages/postgresql/src/postgresql.test.ts +++ b/packages/postgresql/src/postgresql.test.ts @@ -17,6 +17,7 @@ import { contractFields, deadLetterStoreContract, eventStoreContract, + idempotencyStoreContract, inboxLedgerContract, pendingEvent, readModelRebuildContract, @@ -93,6 +94,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 }); + idempotencyStoreContract({ create: async () => (await openStorage()).idempotencyStore }); 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 () => { @@ -393,6 +395,7 @@ describe.skipIf(container === null)("postgresql adapter", () => { "checkpoints", "dead_letters", "events", + "idempotency_keys", "inbox", "rm_order_summary", "scheduled_commands", diff --git a/packages/postgresql/src/schema.ts b/packages/postgresql/src/schema.ts index b0373966..8bf427ef 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 idempotencyKeys: 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" })), + idempotencyKeys: quoteIdentifier(storageTableNameFor({ prefix, table: "idempotencyKeys" })), appendLockKey: `bounda:${schema}.${events}`, checkpointsLockKey: `bounda:${schema}.${checkpoints}`, channel, @@ -131,6 +133,13 @@ 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.idempotencyKeys} ( + "key" text PRIMARY KEY, + "fingerprint" text NOT NULL, + "outcome" jsonb NOT NULL, + "expires_at" text NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS ${indexName(tables.idempotencyKeys, "expires_at")} ON ${tables.idempotencyKeys} ("expires_at")`, ]; export interface EnsureStorageSchemaArgs { @@ -156,8 +165,10 @@ export const ensureStorageSchema: EnsureStorageSchemaFunction = async ({ db, sch tables.inbox, tables.deadLetters, tables.scheduledCommands, + tables.idempotencyKeys, indexName(tables.deadLetters, "status"), indexName(tables.scheduledCommands, "execute_at"), + indexName(tables.idempotencyKeys, "expires_at"), ].map(unquoted); if (await relationsExist({ db, schema, names })) return; await inSchema({ diff --git a/packages/react-router/src/cloudflare/create-bounda.test.ts b/packages/react-router/src/cloudflare/create-bounda.test.ts index bb317bc2..95a1c65e 100644 --- a/packages/react-router/src/cloudflare/create-bounda.test.ts +++ b/packages/react-router/src/cloudflare/create-bounda.test.ts @@ -55,6 +55,33 @@ describe("createBounda on Cloudflare", () => { ]); }); + it("sends the request's Idempotency-Key header with every command of the request", async () => { + const store = storeNamespace(); + env.STORE = store.namespace; + const { bounda, boundaMiddleware } = createBounda({ + config: { storage: cloudflare() }, + tenant: () => "acme", + consistency: "eventual", + }); + const { context, args } = middlewareArgs(); + args.request.headers.set("Idempotency-Key", "req-1"); + await boundaMiddleware(args, async () => undefined); + const commands = context.get(bounda).commands; + await Reflect.get(commands, "placeOrder")({ orderId: "o-1" }, { correlationId: "c-1" }); + await Reflect.get(commands, "payOrder")({ orderId: "o-1" }, { idempotencyKey: "own" }); + expect(store.calls).toEqual([ + [ + "acme", + "command", + "placeOrder", + { orderId: "o-1" }, + { idempotencyKey: "req-1", correlationId: "c-1" }, + "eventual", + ], + ["acme", "command", "payOrder", { orderId: "o-1" }, { idempotencyKey: "own" }, "eventual"], + ]); + }); + it("refuses at once a consistency it does not know, or a Worker it cannot serve", () => { env.STORE = storeNamespace().namespace; expect(() => diff --git a/packages/react-router/src/cloudflare/serve.ts b/packages/react-router/src/cloudflare/serve.ts index 2eb0b32d..bdf39f05 100644 --- a/packages/react-router/src/cloudflare/serve.ts +++ b/packages/react-router/src/cloudflare/serve.ts @@ -1,6 +1,7 @@ import type { BoundaClient, Registry } from "@bounda-dev/core"; import { createContext } from "react-router"; import type { Bounda, BoundaMiddleware } from "../create-bounda.ts"; +import { withIdempotencyKey } from "../idempotency-key.ts"; import type { ClientOfFunction } from "./clients.ts"; export interface ServeFunction { @@ -23,7 +24,7 @@ export const serve: ServeFunction = ( }; const boundaMiddleware: BoundaMiddleware = async (args, next) => { - args.context.set(bounda, (await clients())(args)); + args.context.set(bounda, withIdempotencyKey((await clients())(args), args.request)); return next(); }; diff --git a/packages/react-router/src/create-bounda.test.ts b/packages/react-router/src/create-bounda.test.ts index 7113b17f..0883fab1 100644 --- a/packages/react-router/src/create-bounda.test.ts +++ b/packages/react-router/src/create-bounda.test.ts @@ -121,6 +121,39 @@ describe("createBounda", () => { expect(second.context.get(bounda)).toBe(fake.app); }); + it("passes the request's Idempotency-Key header to the commands it serves", async () => { + const options: unknown[] = []; + const app = { + commands: { + placeOrder: async (_payload: unknown, dispatch?: unknown) => { + options.push(dispatch); + return { scheduled: false }; + }, + }, + start: () => undefined, + stop: async () => undefined, + } as unknown as BoundaApp; + const { bounda, boundaMiddleware } = createBounda({ + consistency: "eventual", + key: uniqueKey(), + boot: async () => app, + }); + const keyed = middlewareArgs(); + keyed.args.request.headers.set("Idempotency-Key", "req-1"); + await boundaMiddleware(keyed.args, async () => undefined); + const plain = middlewareArgs(); + await boundaMiddleware(plain.args, async () => undefined); + const commandsOf = (context: RouterContextProvider) => + context.get(bounda).commands as unknown as { + placeOrder: (payload: unknown, options?: unknown) => Promise; + }; + 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(plain.context.get(bounda)).toBe(app); + }); + it("returns what next returns, awaiting it", async () => { const { boundaMiddleware } = createBounda({ consistency: "eventual", diff --git a/packages/react-router/src/create-bounda.ts b/packages/react-router/src/create-bounda.ts index fdc286a4..d31577e0 100644 --- a/packages/react-router/src/create-bounda.ts +++ b/packages/react-router/src/create-bounda.ts @@ -8,6 +8,7 @@ import { readYourWrites, } from "@bounda-dev/core"; import { createContext, type MiddlewareFunction, type RouterContext } from "react-router"; +import { withIdempotencyKey } from "./idempotency-key.ts"; /** * Boots the app the middleware serves. Called on the first request, and again on the next one @@ -39,7 +40,8 @@ export interface CreateBoundaArgs { /** * A React Router middleware that boots the app on the first request, starts it and puts it in - * the context of every request. Mount it in `root.tsx`: + * the context of every request, its commands carrying the request's `Idempotency-Key` header, + * when it has one, as their `idempotencyKey`. Mount it in `root.tsx`: * `export const middleware = [boundaMiddleware]`. */ export interface BoundaMiddleware { @@ -153,8 +155,8 @@ export const createBounda: CreateBoundaFunction = (key); void retire(slot); - const boundaMiddleware: BoundaMiddleware = async ({ context }, next) => { - context.set(bounda, await load(slot, bootApp, consistency)); + const boundaMiddleware: BoundaMiddleware = async ({ context, request }, next) => { + context.set(bounda, withIdempotencyKey(await load(slot, bootApp, consistency), request)); return next(); }; diff --git a/packages/react-router/src/failure.test.ts b/packages/react-router/src/failure.test.ts index b574c1f5..8c2212a5 100644 --- a/packages/react-router/src/failure.test.ts +++ b/packages/react-router/src/failure.test.ts @@ -1,6 +1,7 @@ import { ConcurrencyError, DomainError, + IdempotencyKeyConflictError, type PayloadArgs, type RejectFunction, type Rejection, @@ -61,6 +62,18 @@ describe("failure", () => { expect(answer.init).toEqual({ status: 409 }); }); + it("answers a reused idempotency key with a 422", () => { + const answer = failure( + new IdempotencyKeyConflictError({ idempotencyKey: "req-1", commandType: "PlaceOrder" }), + ); + expect(answer.data).toEqual({ + error: + 'Idempotency key "req-1" was already used for a command other than PlaceOrder with this payload', + issues: [], + }); + expect(answer.init).toEqual({ status: 422 }); + }); + it("answers the refusals a Durable Object sends back, which arrive as plain errors with their code", () => { const issues = [{ path: ["total"], message: "Expected number" }]; const invalid = failure( diff --git a/packages/react-router/src/failure.ts b/packages/react-router/src/failure.ts index 32028ba9..7a42c81a 100644 --- a/packages/react-router/src/failure.ts +++ b/packages/react-router/src/failure.ts @@ -18,7 +18,8 @@ export interface FailureFunction { /** * Turns what a command is expected to throw into an action's answer, for the form to show: a * `ValidationError` becomes a 400 with its issues, a `DomainError`, the command's rejection, a 409 - * with its code in `rejected`. It goes by the error's `code`, so the same refusals that come back + * with its code in `rejected`, and an `IdempotencyKeyConflictError`, a key the request reuses + * for another command, a 422. It goes by the error's `code`, so the same refusals that come back * from a Durable Object count too. Anything else is rethrown, for the route's `ErrorBoundary`. * * ```ts @@ -40,5 +41,8 @@ export const failure: FailureFunction = (error) => { const { message, rejected } = error as Error & { readonly rejected: string }; return data({ error: message, issues: [], rejected }, { status: 409 }); } + if (code === "IDEMPOTENCY_KEY_CONFLICT") { + return data({ error: (error as Error).message, issues: [] }, { status: 422 }); + } throw error; }; diff --git a/packages/react-router/src/idempotency-key.test.ts b/packages/react-router/src/idempotency-key.test.ts new file mode 100644 index 00000000..a1018212 --- /dev/null +++ b/packages/react-router/src/idempotency-key.test.ts @@ -0,0 +1,81 @@ +import type { BoundaClient, DispatchOptions } from "@bounda-dev/core"; +import { describe, expect, it } from "vitest"; +import { IDEMPOTENCY_KEY_HEADER, withIdempotencyKey } from "./idempotency-key.ts"; + +const request = (idempotencyKey?: string) => + new Request("http://localhost/", { + headers: idempotencyKey === undefined ? {} : { [IDEMPOTENCY_KEY_HEADER]: idempotencyKey }, + }); + +const recording = () => { + const calls: unknown[][] = []; + const client = { + commands: { + placeOrder: async (payload?: unknown, options?: DispatchOptions) => { + calls.push([payload, options]); + return { scheduled: false }; + }, + kind: "facade", + }, + queries: { getOrder: async () => null }, + } as unknown as BoundaClient; + return { client, calls }; +}; + +describe("withIdempotencyKey", () => { + it("hands the header to every command as its key, unless the caller passes one", async () => { + const { client, calls } = recording(); + const served = withIdempotencyKey(client, request("req-1")); + expect(served).not.toBe(client); + expect(served.queries).toBe(client.queries); + const commands = served.commands as unknown as Record; + expect(commands.kind).toBe("facade"); + const placeOrder = commands.placeOrder as ( + payload?: unknown, + options?: DispatchOptions, + ) => Promise; + await placeOrder({ orderId: "o-1" }); + await placeOrder({ orderId: "o-1" }, { correlationId: "c-1" }); + await placeOrder({ orderId: "o-1" }, { idempotencyKey: "mine" }); + await placeOrder({ orderId: "o-1" }, { + idempotencyKey: undefined, + } as unknown as DispatchOptions); + expect(calls).toEqual([ + [{ orderId: "o-1" }, { idempotencyKey: "req-1" }], + [{ orderId: "o-1" }, { correlationId: "c-1", idempotencyKey: "req-1" }], + [{ orderId: "o-1" }, { idempotencyKey: "mine" }], + [{ orderId: "o-1" }, { idempotencyKey: "req-1" }], + ]); + }); + + it("leaves a client alone whose request has no header, and passes an empty header on", async () => { + const { client, calls } = recording(); + expect(withIdempotencyKey(client, request())).toBe(client); + const served = withIdempotencyKey(client, request("")); + await ( + served.commands as unknown as { placeOrder: (payload: unknown) => Promise } + ).placeOrder({}); + expect(calls).toEqual([[{}, { idempotencyKey: "" }]]); + }); + + it("works on commands that are a proxy, as a Durable Object's client has", async () => { + const calls: unknown[][] = []; + const client = { + commands: new Proxy( + {}, + { + get: (_target, key) => + typeof key === "string" && key !== "then" + ? async (...args: unknown[]) => calls.push([key, ...args]) + : undefined, + }, + ), + } as unknown as BoundaClient; + const served = withIdempotencyKey(client, request("req-2")); + await ( + served.commands as unknown as Record Promise> + ).anything?.({ id: "1" }); + expect(calls).toEqual([["anything", { id: "1" }, { idempotencyKey: "req-2" }]]); + 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 new file mode 100644 index 00000000..91e3b37d --- /dev/null +++ b/packages/react-router/src/idempotency-key.ts @@ -0,0 +1,31 @@ +import type { BoundaClient, DispatchOptions, Registry } from "@bounda-dev/core"; + +export const IDEMPOTENCY_KEY_HEADER = "Idempotency-Key"; + +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. +export const withIdempotencyKey: WithIdempotencyKeyFunction = ( + client: BoundaClient, + request: Request, +): BoundaClient => { + const idempotencyKey = request.headers.get(IDEMPOTENCY_KEY_HEADER); + if (idempotencyKey === 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; + return (payload?: unknown, options?: DispatchOptions) => + Reflect.apply(member, target, [ + payload, + { ...options, idempotencyKey: options?.idempotencyKey ?? idempotencyKey }, + ]); + }, + }) as BoundaClient["commands"]; + return { ...client, commands }; +}; diff --git a/packages/sqlite/src/sqlite.test.ts b/packages/sqlite/src/sqlite.test.ts index 815f276f..ec9df346 100644 --- a/packages/sqlite/src/sqlite.test.ts +++ b/packages/sqlite/src/sqlite.test.ts @@ -236,6 +236,7 @@ describe("sqlite adapter on a file", () => { "app_checkpoints", "app_dead_letters", "app_events", + "app_idempotency_keys", "app_inbox", "app_rm_order_summary", "app_scheduled_commands", From 18fbc6dda96640e7e5efaac7f2ef127bbd9be233 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Be=C3=B1at=20Espi=C3=B1a?= Date: Sun, 25 Oct 2026 05:38:06 +0100 Subject: [PATCH 2/2] fix(ci): print the mutation table without a pipe head can break The job summary piped sed into head -60 under pipefail; once the table of a package grew past 60 lines, head closed the pipe while sed still wrote, sed failed with a broken pipe and the job failed after a green run. awk prints the first 60 lines of the table and reads the rest without a reader to lose. --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bd7e0f45..d1fd2b22 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -170,7 +170,7 @@ jobs: echo "## Mutation testing: ${{ matrix.package }}" echo echo '```text' - sed -n '/^-*|/,/^-*|$/p' mutation.log | head -60 + awk '/^-*\|/,/^-*\|$/ { if (n++ < 60) print }' mutation.log grep -E "Final mutation score" mutation.log | sed 's/.*INFO MutationTestReportHelper//' || true echo '```' } >> "$GITHUB_STEP_SUMMARY"