feat(core): remember what a command dispatched with an idempotency key answered

.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.

.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"

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`.

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

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)).

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";

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.

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

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

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
 

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 |

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 <Type>` | 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 <Type>` | 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 <name>` | 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 <readModel>.<projection>` | a projection handles one event | `bounda.read_model`, `bounda.projection`, the event's id, type and aggregate, `bounda.correlation_id` |
 | `bounda.policy <aggregate>.<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

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 () =>

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,14 +14,15 @@ 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 {
   readonly eventStore: EventStore;
   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.
    */