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