From 18bbf49c93683186c4ec770064a6f4f752e3dd65 Mon Sep 17 00:00:00 2001 From: Benoit TRAVERS Date: Wed, 12 Aug 2026 00:47:37 +0200 Subject: [PATCH 1/2] feat(examples): add the worker deployment and the examples index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The same ApplicationModule + PersistenceModule composition, booted under an in-memory queue worker instead of the oRPC server — with no change to order-application, order-infrastructure or the kernel. The same DuplicateOrder that is a typed CONFLICT over HTTP is a dead-letter here, and a Defect that is an INTERNAL_SERVER_ERROR over HTTP is worth another delivery on a queue. queueWorkerRuntime declares a non-empty needs of its own ([PlaceOrder, Logger], two of the three the module exports), pinned in both directions by needs-gate.test-d.ts. A delivery is the unit and the message id is the traceId, since a retried message is two units; the disposition is applied inside the unit, because acking a message and flushing a response are the same obligation. Adds examples/README.md as the index — the layering, the dependency direction, and the two runtimes being the only ones in the repo with a non-empty needs — plus links from the root README, and CLAUDE.md records that examples/ is part of the gate, that the Prisma client is generated at test time, and why oRPC is pinned to an exact beta. --- CLAUDE.md | 32 ++- README.md | 15 +- examples/README.md | 106 ++++++++ examples/order-api/README.md | 6 +- examples/order-worker/README.md | 136 ++++++++++ examples/order-worker/package.json | 34 +++ examples/order-worker/src/env.spec.ts | 61 +++++ examples/order-worker/src/env.ts | 53 ++++ examples/order-worker/src/index.ts | 13 + examples/order-worker/src/main.ts | 42 ++++ examples/order-worker/src/module.ts | 26 ++ .../order-worker/src/needs-gate.test-d.ts | 48 ++++ .../order-worker/src/queue-runtime.spec.ts | 151 ++++++++++++ examples/order-worker/src/queue-runtime.ts | 232 ++++++++++++++++++ examples/order-worker/src/queue.ts | 103 ++++++++ examples/order-worker/src/test-fixtures.ts | 185 ++++++++++++++ examples/order-worker/src/vitest.d.ts | 1 + examples/order-worker/tsconfig.json | 13 + examples/order-worker/tsconfig.test-d.json | 6 + examples/order-worker/vitest.config.ts | 9 + knip.json | 3 +- pnpm-lock.yaml | 43 ++++ 22 files changed, 1311 insertions(+), 7 deletions(-) create mode 100644 examples/README.md create mode 100644 examples/order-worker/README.md create mode 100644 examples/order-worker/package.json create mode 100644 examples/order-worker/src/env.spec.ts create mode 100644 examples/order-worker/src/env.ts create mode 100644 examples/order-worker/src/index.ts create mode 100644 examples/order-worker/src/main.ts create mode 100644 examples/order-worker/src/module.ts create mode 100644 examples/order-worker/src/needs-gate.test-d.ts create mode 100644 examples/order-worker/src/queue-runtime.spec.ts create mode 100644 examples/order-worker/src/queue-runtime.ts create mode 100644 examples/order-worker/src/queue.ts create mode 100644 examples/order-worker/src/test-fixtures.ts create mode 100644 examples/order-worker/src/vitest.d.ts create mode 100644 examples/order-worker/tsconfig.json create mode 100644 examples/order-worker/tsconfig.test-d.json create mode 100644 examples/order-worker/vitest.config.ts diff --git a/CLAUDE.md b/CLAUDE.md index b61a984..7d1b151 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -20,7 +20,10 @@ throws to callers: every fallible operation returns an [`unthrown`](https://github.com/btravstack/unthrown) `Result`. pnpm workspace + turbo monorepo. `packages/start` is the single published -package; there are no other workspaces yet. +package; `examples/` holds five private ones — a clean-architecture application +(`order-domain` → `order-application` → `order-infrastructure`) booted under two +different runtimes (`order-api`, `order-worker`). They are consumers, not +fixtures: they are part of the gate, and `examples/README.md` is their index. ## Commands @@ -60,6 +63,11 @@ hook). User-facing changes need a changeset. question of how two runtimes in one process share a drain deadline, or whose failure takes the process down. `StartOptions.runtime` is therefore a single value, not an array, and no future option should make it plural. + `examples/order-api` and `examples/order-worker` make this testable rather + than asserted: the same `ApplicationModule` + `PersistenceModule` + composition under two runtimes, with the same `DuplicateOrder` arriving as a + typed `CONFLICT` on one and as a dead-letter on the other — and neither + mapping anywhere near the kernel. 2. **Ambient carries DATA. The DI `Context` carries CAPABILITIES.** The kernel opens one `AsyncLocalStorage` store per unit holding a small, fixed record — @@ -559,6 +567,25 @@ Source layout (`packages/start/src/`), one concept per file: `ambient.ts` ## Toolchain & conventions +- **`examples/` is part of the gate, not a folder of illustrations.** All five + workspaces run under the same six commands as the kernel — 57 specs plus two + `needs-gate.test-d.ts` files — so an example that stops compiling, stops + linting or stops passing fails CI exactly as `packages/start` would. They are + also the only place a runtime with a **non-empty `needs`** meets a real + module, which is what exercises `start`'s phantom rest-tuple gate and + `RuntimeHost`'s `Context>` end to end. +- **The Prisma client is generated at test time, and there is nothing to + install.** `@btravstack/start-example-order-infrastructure`'s `test` and + `typecheck` scripts both begin with `prisma generate`, writing a gitignored + client into `src/generated`, and turbo's `test` / `typecheck` / `test:types` + tasks carry a `^generate` edge so a dependent workspace gets one too. The + database is SQLite **in memory** with the schema applied by hand — + deliberately no Docker, so `pnpm test` stays self-contained on any machine. +- **oRPC is pinned to an exact beta.** `@orpc/{client,contract,server}` sit at + `2.0.0-beta.23` in the catalog because oRPC v2's `latest` dist-tag is still + the **1.x** line, while `@unthrown/orpc` peers on `^2.0.0-beta`: an unpinned + range resolves 1.x and fails `strictPeerDependencies`. The exact beta is the + contract until v2 goes stable; raise it deliberately, not on a bot bump. - **Runtime dependencies: none.** `unthrown` and `@btravstack/di` are **peer** dependencies — the dual-copy hazard is real for both (di's port identity and unthrown's `isResult` each compare across copies). `node:` builtins only @@ -701,6 +728,9 @@ A sixth rule is about production code that tests keep honest: Shipped: the whole kernel — phase tracker, injectable clock, ambient record, unit registry, `Runtime` contract, `start`, draining, signals, uncaught handling, probes, `runMain`, the testing entry point, and the invariants suite. +Plus the five `examples/` workspaces: the clean-architecture application and its +**two** deployments, `order-api` (oRPC) and `order-worker` (an in-memory queue), +which together are the proof of Thesis #1. Deferred, deliberately: diff --git a/README.md b/README.md index f4271d8..311f046 100644 --- a/README.md +++ b/README.md @@ -134,6 +134,12 @@ and deploy independently, and it removes a whole class of design problem: there is never a question of how two runtimes in one process share a drain deadline, or whose failure takes the process down. +[`examples/`](./examples) proves this rather than asserting it: one +clean-architecture application, booted by an oRPC runtime and by a queue-worker +runtime, with the application and persistence layers unchanged between them — +and the same `DuplicateOrder` arriving as a typed `CONFLICT` on one and as a +dead-letter on the other. + ## The `Runtime` contract ```ts @@ -554,10 +560,11 @@ complete one. ## Documentation -See [`packages/start`](./packages/start) for the package README, and -[`CLAUDE.md`](./CLAUDE.md) for the authoritative spec: the theses, the -load-bearing invariants with the test that guards each, and the internal design -notes. +See [`packages/start`](./packages/start) for the package README, +[`examples/`](./examples) for a five-package clean-architecture application +booted under two different runtimes, and [`CLAUDE.md`](./CLAUDE.md) for the +authoritative spec: the theses, the load-bearing invariants with the test that +guards each, and the internal design notes. ## License diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..cea2cf4 --- /dev/null +++ b/examples/README.md @@ -0,0 +1,106 @@ +# Examples + +Five small packages that are **one application booted two ways**: a clean +architecture split across four layers, deployed once as an oRPC API and once as +a queue worker — and, at the same time, exercising `@btravstack/start` end to +end from a consumer's own workspace, `workspace:*` and all. + +| Package | Layer | Shows | +| ------------------------------------------------ | --------- | ------------------------------------------------------------------------------------------------------------------------------------------ | +| [`order-domain`](./order-domain) | domain | Entities and rules with no dependencies at all: branded fields, an `Entity.invariant` re-checked on every path, failures as values. | +| [`order-application`](./order-application) | use cases | Ports declared by the caller, interactors, and an `ApplicationModule` whose `OrderRepository` is deliberately an **unmet need**. | +| [`order-infrastructure`](./order-infrastructure) | adapters | A Prisma-backed repository over in-memory SQLite, translating P-codes into the domain's vocabulary and closing the application's one need. | +| [`order-api`](./order-api) | runtime | The first deployment: an oRPC router over `node:http`, a scope forked per request, and `Result` → `ORPCError`. | +| [`order-worker`](./order-worker) | runtime | The second deployment: an in-memory queue worker over the **same** composition, and `Result` → ack / retry / dead-letter. | + +## The layering, and which way the arrows point + +``` + order-api order-worker ← one runtime each; one process each + └───────────┬───────────────┘ + ▼ + order-infrastructure ← Prisma, SQLite, P-codes + │ provides OrderRepository + ▼ + order-application ← use cases, and the ports they declare + │ + ▼ + order-domain ← entities and rules; depends on nothing +``` + +Every arrow points **inwards**, and the one that looks like it goes the wrong +way is the whole idea: `order-infrastructure` imports `order-application`, +because the port it implements — `OrderRepository`, spelled in the domain's +vocabulary — is declared by the caller that needs it, not by the database that +happens to satisfy it. `ApplicationModule` therefore leaves that need **unmet**, +which is not documentation but a type: `Module.scoped(ApplicationModule, …)` +does not compile until an outer module provides one. + +## One application, two deployments + +`OrderApiModule` and `OrderWorkerModule` are the same three lines: + +```ts +imports: [ApplicationModule, PersistenceModule], +provides: [], +exports: [PlaceOrder, FindOrder, Logger], +``` + +Nothing in `order-application` or `order-infrastructure` differs between them, +and nothing could: the use cases return a `Result`, and what a `Result` means to +a transport is the transport's business. The kernel's headline claim — several +runtime _kinds_, one per process, over the same module — is proved here rather +than asserted, and the sharpest form of the proof is that **the same `Err` +becomes two different outcomes**: + +| unthrown | `order-api` | `order-worker` | +| ---------------------- | ----------------------- | --------------------------- | +| `Ok(order)` | the procedure's output | **ack** | +| `Err(InvalidQuantity)` | `INVALID_QUANTITY` | **dead-letter** | +| `Err(DuplicateOrder)` | `CONFLICT` | **dead-letter** | +| `Defect` | `INTERNAL_SERVER_ERROR` | **retry**, then dead-letter | + +The kernel appears in neither column. `RunUnit` hands a runtime the work's own +`Result` and stays out of what it means. + +## The only non-empty `needs` in the repo + +`orpcRuntime` declares `[PlaceOrder, FindOrder, Logger]` and +`queueWorkerRuntime` declares `[PlaceOrder, Logger]` — two of the three the +module exports, because a runtime declares what _it_ needs. They are the **only +runtimes in this repository with a non-empty `needs`**: the kernel's own +`testRuntime` needs nothing, so `start`'s phantom rest-tuple gate — and +`RuntimeHost`'s `Context>`, where a runtime names port +_classes_ while di parameterises contexts by port _instances_ — are exercised +against a real module here and nowhere else. + +Both directions are pinned, in `order-api/src/needs-gate.test-d.ts` and +`order-worker/src/needs-gate.test-d.ts`: the wired call is an ordinary +two-argument one, and a module one port short fails on **arity**, naming the +missing need. + +## Why these are tests, not just illustrations + +Each package reads as application code, and each is covered by real specs — 57 +of them, run by the repository's own `pnpm test`: + +```sh +pnpm install +pnpm test # every example's specs, alongside the kernel's own +pnpm typecheck # includes the compile-time-only guarantees pinned with @ts-expect-error +``` + +Nothing is faked at the boundaries that matter. `order-infrastructure` runs +against a real Prisma client over in-memory SQLite, so a `DuplicateOrder` comes +from an actual `UNIQUE` index raising an actual P2002. `order-api` runs a real +`node:http` server and a real oRPC client over it, so the collapse of a `Defect` +to `INTERNAL_SERVER_ERROR` happens where it really happens. **No Docker, and +nothing to install**: the Prisma client is generated by the `test` script +itself. + +Where a guarantee is compile-time only — an unmet port, a runtime's `needs` — +the assertion is a `@ts-expect-error` in a `*.test-d.ts` file, checked by `tsc` +rather than executed. + +Nothing here is published: every package is `"private": true` and depends on the +kernel via `workspace:*`. diff --git a/examples/order-api/README.md b/examples/order-api/README.md index 90e73ca..71e30d3 100644 --- a/examples/order-api/README.md +++ b/examples/order-api/README.md @@ -28,6 +28,10 @@ adapter in between: | `Err(error)` | a returned `ORPCError` | | `Defect` | `INTERNAL_SERVER_ERROR` | +None of it is the kernel's doing — which is what [`order-worker`](../order-worker) +demonstrates by folding the very same `Result` into ack / retry / dead-letter +over the very same composition root. + `handlerResult` performs that elimination, and the `mapErrCases` in front of it is the triage point — the boundary where the application's vocabulary stops: @@ -133,7 +137,7 @@ the server's `mapErrCases`. ## Running it ```bash -pnpm --filter @btravstack/start-example-order-api test # 8 runtime specs + 4 env specs +pnpm --filter @btravstack/start-example-order-api test # 15 runtime specs + 4 env specs ``` The specs run against a real HTTP server and a real oRPC client — genuine JSON diff --git a/examples/order-worker/README.md b/examples/order-worker/README.md new file mode 100644 index 0000000..05fd8ac --- /dev/null +++ b/examples/order-worker/README.md @@ -0,0 +1,136 @@ +# `@btravstack/start` example: the order worker + +The second deployment. The same application, the same persistence, the same +composition — consumed off a queue instead of served over HTTP. + +``` +src/queue.ts the broker, reduced to what a worker needs of one +src/queue-runtime.ts the Runtime: start / drain / stop, and the ack/retry/dead-letter mapping +src/module.ts OrderWorkerModule — the composition root +src/env.ts process.env validated through a schema, as a Result +src/main.ts the process: readEnv + start + runMain +src/test-fixtures.ts serve / queue / gate / tapped, as Vitest fixtures +``` + +## The point of this package + +`OrderWorkerModule` is `OrderApiModule` with a different name: + +```ts +export const OrderWorkerModule = Module("OrderWorker")({ + imports: [ApplicationModule, PersistenceModule], + provides: [], + exports: [PlaceOrder, FindOrder, Logger], +}); +``` + +Nothing in `order-application` or `order-infrastructure` changed to make this +work, and nothing could have: the use cases return a `Result`, and what a +`Result` means to a transport is the transport's business. **One process, one +runtime** is not a slogan the kernel makes you take on trust — it is two +composition roots, two `Runtime` values, and one application underneath. + +## The same `Err`, two transports + +`DuplicateOrder` is one value. Over HTTP there is a caller waiting to be told, +so it becomes a `CONFLICT` the client receives **as a value**. On a queue there +is no caller, so the message is **parked** for a human instead. + +| unthrown | oRPC (`order-api`) | queue (this package) | +| ---------------------- | ----------------------- | --------------------------- | +| `Ok(order)` | the procedure's output | **ack** | +| `Err(InvalidQuantity)` | `INVALID_QUANTITY` | **dead-letter** | +| `Err(DuplicateOrder)` | `CONFLICT` | **dead-letter** | +| `Defect` | `INTERNAL_SERVER_ERROR` | **retry**, then dead-letter | + +The third row is the sharp one and the fourth is its mirror. An unmodelled +failure is the infrastructure one — a dropped connection, a pool timeout — and +infrastructure comes back, so it is the one thing worth another delivery. A +modelled error never is: a redelivery would ask the same impossible thing again. + +```ts +ctx + .get(PlaceOrder) + .execute(job.orderId, job.quantity) + .match({ + ok: () => ack, + errCases: (matcher) => + matcher + .with(P.tag("InvalidQuantity"), (error) => deadLetter(error._tag)) + .with(P.tag("DuplicateOrder"), (error) => deadLetter(error._tag)), + defect: (cause) => retry(String(cause)), + }); +``` + +Every case is named — this repo bans `P._`, and there is no `.otherwise()`. A +new domain error is a compile error here **and** in `order-api/src/router.ts`, +at the two places that have to decide what it means. + +## Acking is flushing + +The disposition is applied **inside the unit**, exactly as the API writes its +response inside the unit: + +```ts +host.run(metaFor(delivery), (ctx, _signal) => + dispositionOf(ctx, delivery.job).flatMap((disposition) => + dispose(ctx.get(Logger), queue, delivery, maxAttempts, disposition), + ), +); +``` + +A unit closes the instant its `Result` settles, and an idle registry is the +kernel's permission to call `Serving.stop()`. Settling the message afterwards +would race that: the message would be neither acked nor requeued when the +process went away. Flushing a response and acking a message are the same +obligation, wearing different clothes. + +## A delivery is the unit; the message is the trace + +```ts +const metaFor = (delivery: Delivery): UnitMeta => ({ + kind: "job", + id: `${delivery.job.id}#${delivery.attempt}`, + traceId: delivery.job.id, +}); +``` + +`UnitMeta.id` must be unique per unit, and a message id is not one: a retried +message is delivered twice and is two units. So the **delivery** is the id, and +the message id becomes the `traceId` — which is exactly what `traceId` is for. +It is the correlation id, minted outside this process, and holding it steady +across attempts is what joins three deliveries into one trace. + +## `Serving.info` with no port in it + +```ts +const info = (await app.runtimeInfo()).get(); // { queue: "orders", concurrency: 1 } +``` + +The API publishes `{ port, prefix }` on the same channel. That is why `Info` is +the runtime's own type parameter rather than a port number baked into the +kernel: a queue consumer has none, and what an operator wants to know about one +is which queue it is on and how many messages it will take at a time. + +## Running it + +```bash +pnpm --filter @btravstack/start-example-order-worker test # 7 runtime specs + 4 env specs +pnpm --filter @btravstack/start-example-order-worker test:types # the needs gate +``` + +`src/needs-gate.test-d.ts` pins the compile-time half: `queueWorkerRuntime` +declares `[PlaceOrder, Logger]` — two of the three ports the module exports, +because a runtime declares what _it_ needs — and a module missing either fails +`start`'s arity gate before anything runs. + +Every helper the specs need is a Vitest fixture in `src/test-fixtures.ts`, so +each file opens on `describe` and each test names its dependencies in its own +parameter list. Shutting an app down is the `serve` fixture's job, which is why +no test here has a `try`/`finally`. + +`src/main.ts` is the process itself. It creates the queue, because this +example's broker is a plain in-memory object; a real deployment builds an AMQP +channel from the environment instead, and nothing above that line changes. Like +`order-api`'s, it is typechecked by the gate rather than executed by it — the +example packages are source-only, and every spec drives `start` directly. diff --git a/examples/order-worker/package.json b/examples/order-worker/package.json new file mode 100644 index 0000000..c854a42 --- /dev/null +++ b/examples/order-worker/package.json @@ -0,0 +1,34 @@ +{ + "name": "@btravstack/start-example-order-worker", + "private": true, + "description": "The second deployment of the clean-architecture example: the same application module, consumed by an in-memory queue worker instead of an oRPC server", + "license": "MIT", + "author": "Benoit TRAVERS ", + "type": "module", + "main": "./src/index.ts", + "exports": { + ".": "./src/index.ts" + }, + "scripts": { + "test": "vitest run", + "test:types": "tsc --noEmit -p tsconfig.test-d.json", + "typecheck": "tsc --noEmit && tsc --noEmit -p tsconfig.test-d.json" + }, + "dependencies": { + "@btravstack/di": "catalog:", + "@btravstack/start": "workspace:*", + "@btravstack/start-example-order-application": "workspace:*", + "@btravstack/start-example-order-domain": "workspace:*", + "@btravstack/start-example-order-infrastructure": "workspace:*", + "@unthrown/standard-schema": "catalog:", + "unthrown": "catalog:", + "zod": "catalog:" + }, + "devDependencies": { + "@btravstack/tsconfig": "catalog:", + "@types/node": "catalog:", + "@unthrown/vitest": "catalog:", + "typescript": "catalog:", + "vitest": "catalog:" + } +} diff --git a/examples/order-worker/src/env.spec.ts b/examples/order-worker/src/env.spec.ts new file mode 100644 index 0000000..9c47f26 --- /dev/null +++ b/examples/order-worker/src/env.spec.ts @@ -0,0 +1,61 @@ +import { P } from "unthrown"; +import { describe, expect, it } from "vitest"; + +import { describeEnvIssues, readEnv } from "./env.js"; + +describe("readEnv", () => { + it("falls back to the documented defaults when nothing is set", () => { + // GIVEN an environment with neither variable set + const source = {}; + + // WHEN it is validated + const env = readEnv(source); + + // THEN both carry their defaults, as numbers + expect(env).toBeOkWith({ PROBE_PORT: 9000, CONCURRENCY: 1 }); + }); + + it("reads what a deployment actually supplies", () => { + // GIVEN both set, as the strings an environment always holds + const source = { PROBE_PORT: "0", CONCURRENCY: "8" }; + + // WHEN it is validated + const env = readEnv(source); + + // THEN they arrive parsed, and `0` survives as the ephemeral bind it is + expect(env).toBeOkWith({ PROBE_PORT: 0, CONCURRENCY: 8 }); + }); + + it("reports a malformed value rather than consuming nothing at all", () => { + // GIVEN the values `Number()` would silently turn into `NaN` and `0` — and + // a concurrency of `0` is a worker that takes no messages + const source = { PROBE_PORT: "abc", CONCURRENCY: "" }; + + // WHEN it is validated + const env = readEnv(source); + + // THEN neither reaches the runtime: both are issues in the error channel, + // named and in order, asserted on the `Err` itself rather than behind an + // `env.isErr() &&` guard that evaluates to `false` when it does not hold + expect(env).toBeErrWith([ + expect.objectContaining({ path: ["PROBE_PORT"] }), + expect.objectContaining({ path: ["CONCURRENCY"] }), + ]); + }); + + it("rejects a concurrency no worker should be asked for", () => { + // GIVEN a number that parses but that nothing sensible would run + const source = { CONCURRENCY: "1000" }; + + // WHEN its error channel is folded into the message a deployment reads + const described = readEnv(source).match({ + ok: () => "WRONGLY ACCEPTED", + // oxlint-disable-next-line unthrown/no-catch-all-pattern -- `E` is one type (the schema's list of issues), not a union of cases to enumerate + errCases: (matcher) => matcher.with(P._, describeEnvIssues), + defect: () => "defect", + }); + + // THEN the ceiling is the schema's business, not the scheduler's + expect(described).toContain("CONCURRENCY: Too big"); + }); +}); diff --git a/examples/order-worker/src/env.ts b/examples/order-worker/src/env.ts new file mode 100644 index 0000000..468ebce --- /dev/null +++ b/examples/order-worker/src/env.ts @@ -0,0 +1,53 @@ +import { fromSchema, type SchemaIssues } from "@unthrown/standard-schema"; +import type { Result } from "unthrown"; +import { z } from "zod"; + +/** + * A whole number, read the way an environment variable actually arrives: as a + * string. + * + * `z.coerce.number()` would be shorter and wrong — it is `Number()` underneath, + * so `CONCURRENCY=abc` becomes `NaN` and `CONCURRENCY=` becomes `0`, a worker + * that consumes nothing at all. A digits-only string is the honest model, and + * anything else is a validation issue rather than a number nobody asked for. + */ +const wholeNumber = (fallback: number, min: number, max: number) => + z + .string() + .regex(/^\d+$/u, "must be a whole number of decimal digits") + .transform(Number) + .pipe(z.int().min(min).max(max)) + .default(fallback); + +const environment = z.object({ + PROBE_PORT: wholeNumber(9000, 0, 65_535), + CONCURRENCY: wholeNumber(1, 1, 64), +}); + +/** The validated environment: every field present, typed, and in range. */ +export type Env = z.infer; + +// `fromSchema` is CURRIED — it takes the schema and hands back the validator. +const validate = fromSchema(environment); + +/** + * Validates the process environment **as a value**. + * + * A schema's own `.parse()` throws, which `unthrown/no-throw` bans and which + * would contradict the example it appears in. `@unthrown/standard-schema` makes + * the issues the modeled `E`, so the entry point folds a bad environment the + * same way it folds any other anticipated failure. + */ +export const readEnv = (source: typeof process.env = process.env): Result => + validate(source); + +const nameOf = (segment: NonNullable[number]): string => + String(typeof segment === "object" ? segment.key : segment); + +/** One line per issue, each naming the variable it is about. */ +export const describeEnvIssues = (issues: SchemaIssues): string => + issues + .map( + (issue) => `${(issue.path ?? []).map(nameOf).join(".") || "(environment)"}: ${issue.message}`, + ) + .join("\n"); diff --git a/examples/order-worker/src/index.ts b/examples/order-worker/src/index.ts new file mode 100644 index 0000000..c4bdd78 --- /dev/null +++ b/examples/order-worker/src/index.ts @@ -0,0 +1,13 @@ +export { OrderWorkerModule } from "./module.js"; +export { + createOrderQueue, + type Delivery, + type OrderQueue, + type PlaceOrderJob, + type Settlement, +} from "./queue.js"; +export { + queueWorkerRuntime, + type OrderWorkerInfo, + type QueueWorkerOptions, +} from "./queue-runtime.js"; diff --git a/examples/order-worker/src/main.ts b/examples/order-worker/src/main.ts new file mode 100644 index 0000000..094c7ec --- /dev/null +++ b/examples/order-worker/src/main.ts @@ -0,0 +1,42 @@ +import { runMain, start } from "@btravstack/start"; +import { P } from "unthrown"; + +import { describeEnvIssues, readEnv, type Env } from "./env.js"; +import { OrderWorkerModule } from "./module.js"; +import { queueWorkerRuntime } from "./queue-runtime.js"; +import { createOrderQueue } from "./queue.js"; + +/** + * The second process, and — apart from the runtime it names — the same one + * `order-api/src/main.ts` is: validate the environment, build the graph, serve + * it, and turn the exit report into a process exit code. + * + * The queue is created here because this example's broker is a plain in-memory + * object; a real deployment builds an AMQP channel from the environment + * instead, and nothing above this line changes. Typechecked by the gate, not + * executed by it — the example packages are source-only, and every spec drives + * `start` directly. + */ +const work = (env: Env): Promise => + runMain( + start(OrderWorkerModule, { + runtime: queueWorkerRuntime({ + queue: createOrderQueue(), + concurrency: env.CONCURRENCY, + }), + probes: { port: env.PROBE_PORT }, + }), + ); + +/** sysexits(3) `EX_CONFIG`: the deployment is wrong, not the code. */ +const abort = (reason: string): void => { + process.stderr.write(`${reason}\n`); + process.exitCode = 78; +}; + +await readEnv().match({ + ok: work, + // oxlint-disable-next-line unthrown/no-catch-all-pattern -- `E` is the issues array: one type with no discriminant, so there is nothing to enumerate and the single arm IS the enumeration + errCases: (matcher) => matcher.with(P._, (issues) => abort(describeEnvIssues(issues))), + defect: (cause) => abort(`the environment could not be validated: ${String(cause)}`), +}); diff --git a/examples/order-worker/src/module.ts b/examples/order-worker/src/module.ts new file mode 100644 index 0000000..4e491d1 --- /dev/null +++ b/examples/order-worker/src/module.ts @@ -0,0 +1,26 @@ +import { Module } from "@btravstack/di"; +import { + ApplicationModule, + FindOrder, + Logger, + PlaceOrder, +} from "@btravstack/start-example-order-application"; +import { PersistenceModule } from "@btravstack/start-example-order-infrastructure"; + +/** + * The composition root of the second deployment — and, imports and exports + * alike, the same one `OrderApiModule` is. That is the claim this package + * exists to make: `ApplicationModule` and `PersistenceModule` are booted here + * unchanged, under a runtime that speaks a queue instead of HTTP. + * + * It is declared here rather than imported from `order-api` on purpose. Sharing + * the module would also share the API's oRPC dependency, and a worker + * deployment that installs a web server to reach its use cases would falsify + * the very thing this is demonstrating. Two processes, two composition roots, + * one application. + */ +export const OrderWorkerModule = Module("OrderWorker")({ + imports: [ApplicationModule, PersistenceModule], + provides: [], + exports: [PlaceOrder, FindOrder, Logger], +}); diff --git a/examples/order-worker/src/needs-gate.test-d.ts b/examples/order-worker/src/needs-gate.test-d.ts new file mode 100644 index 0000000..dbde6ac --- /dev/null +++ b/examples/order-worker/src/needs-gate.test-d.ts @@ -0,0 +1,48 @@ +/** + * The compile-time half of the second deployment: `queueWorkerRuntime` declares + * two ports in `needs`, and `start`'s phantom rest-tuple gate turns a module + * that does not export both into a call-site arity error. Type-checked by this + * package's `test:types` script, never executed. + * + * Together with `order-api`'s, this is what makes the claim testable rather + * than asserted: two runtimes with different, non-empty `needs`, both proven + * against the same application graph at the `start(...)` call site. + */ +import { Module } from "@btravstack/di"; +import { start } from "@btravstack/start"; +import { + ApplicationModule, + FindOrder, + Logger, + PlaceOrder, +} from "@btravstack/start-example-order-application"; +import { PersistenceModule } from "@btravstack/start-example-order-infrastructure"; + +import { OrderWorkerModule } from "./module.js"; +import { queueWorkerRuntime } from "./queue-runtime.js"; +import { createOrderQueue } from "./queue.js"; + +const options = { + runtime: queueWorkerRuntime({ queue: createOrderQueue() }), + signals: false, + probes: false, +} as const; + +// Positive: the composition root exports both ports the runtime needs (and a +// third it does not), so the gate collapses to an empty tuple and this is an +// ordinary two-argument call. +const _wired = start(OrderWorkerModule, options); + +// The same graph, one port short: `Logger` is provided (the interactors depend +// on it) but not exported, so it is not in the application context the runtime +// is handed. +const PartialWorker = Module("PartialWorker")({ + imports: [ApplicationModule, PersistenceModule], + provides: [], + exports: [PlaceOrder, FindOrder], +}); + +// Negative: the gate becomes a required two-element tuple naming the unmet need, +// and the call fails on arity. +// @ts-expect-error — UNSATISFIED RUNTIME NEEDS: the module does not export Logger. +const _missingLogger = start(PartialWorker, options); diff --git a/examples/order-worker/src/queue-runtime.spec.ts b/examples/order-worker/src/queue-runtime.spec.ts new file mode 100644 index 0000000..48ce086 --- /dev/null +++ b/examples/order-worker/src/queue-runtime.spec.ts @@ -0,0 +1,151 @@ +import { describe, expect, vi } from "vitest"; + +import { OrderWorkerModule } from "./module.js"; +import { it } from "./test-fixtures.js"; + +describe("queueWorkerRuntime", () => { + it("acks a job whose order the use case places", async ({ serve, queue, aJob }) => { + // GIVEN the same composition the API boots — `ApplicationModule` and + // `PersistenceModule`, unchanged — under a queue runtime instead + serve(OrderWorkerModule); + + // WHEN a job is published + // THEN it reached the use case behind the transport, and the message is done + await expect(queue.publish(aJob("job-1", "o-1", 2))).toBeOkWith({ + jobId: "job-1", + outcome: "acked", + attempts: 1, + }); + }); + + it("dead-letters the DuplicateOrder the API answers CONFLICT for", async ({ + serve, + queue, + aJob, + }) => { + // GIVEN an order already placed by a first job + serve(OrderWorkerModule); + + // WHEN a second job asks for the same order id — chained, so the first + // job's settlement is consumed and a failure there cannot be mistaken for + // the duplicate. A second job is a second unit, over the same database: + // the application scope is opened once, by the kernel. + const settled = await queue + .publish(aJob("job-1", "o-1", 2)) + .flatMap(() => queue.publish(aJob("job-2", "o-1", 2))); + + // THEN the load-bearing assertion of this whole example: the identical + // `Err` the oRPC runtime turns into an inferable CONFLICT is parked here, + // because a queue has no caller waiting to be told. One `Result`, two + // transports, and the kernel involved in neither mapping. + expect(settled).toBeOkWith({ + jobId: "job-2", + outcome: "dead-lettered", + reason: "DuplicateOrder", + attempts: 1, + }); + }); + + it("dead-letters a job the domain rejects, rather than redelivering it", async ({ + serve, + queue, + aJob, + }) => { + // GIVEN a quantity the domain invariant rejects + serve(OrderWorkerModule); + + // WHEN the job is published + // THEN it is parked on the first attempt: a redelivery would ask the same + // impossible thing again + await expect(queue.publish(aJob("job-1", "o-1", 0))).toBeOkWith({ + jobId: "job-1", + outcome: "dead-lettered", + reason: "InvalidQuantity", + attempts: 1, + }); + }); + + it("redelivers an unmodelled failure, then parks it once the attempts run out", async ({ + serve, + queue, + aJob, + unmodelled, + }) => { + // GIVEN a repository whose failure nobody modelled, so it is a `Defect` + serve(unmodelled); + + // WHEN a job reaches it + // THEN the third channel takes the third route: infrastructure comes back, + // so a defect is worth another delivery — three of them, and then the + // message is parked carrying the cause + await expect(queue.publish(aJob("job-1", "o-1", 1))).toBeOkWith({ + jobId: "job-1", + outcome: "dead-lettered", + reason: "Error: the database is on fire", + attempts: 3, + }); + }); + + it("publishes the queue it consumes on Serving.info", async ({ serve }) => { + // GIVEN a worker consuming the fixture's queue + const app = serve(OrderWorkerModule); + + // WHEN the kernel is asked what the runtime published about itself + const info = app.runtimeInfo(); + + // THEN the same channel the API publishes `{ port, prefix }` on carries a + // shape with no port in it at all + await expect(info).toBeOkWith({ queue: "orders", concurrency: 1 }); + }); + + it("runs each job in its own unit, with its own trace id", async ({ + serve, + queue, + aJob, + tapped, + }) => { + // GIVEN the real graph with the very `Logger` instance the use cases and + // the disposition write to + serve(tapped.worker); + + // WHEN two jobs are consumed — chained, so neither settlement is dropped + const settled = await queue + .publish(aJob("job-1", "o-1", 1)) + .flatMap(() => queue.publish(aJob("job-2", "o-2", 1))); + + // THEN two jobs, two interactor lines plus two disposition lines, each + // carrying the message id its delivery correlated to and never the + // out-of-unit `[-]`. The unit id is `job-1#1`, the trace id is `job-1`: + // a redelivery is a new unit and the same trace. + expect(settled.map(() => tapped.traces())).toBeOkWith([ + "[job-1]", + "[job-1]", + "[job-2]", + "[job-2]", + ]); + }); + + it("waits for the in-flight job while draining", async ({ serve, queue, aJob, gate }) => { + // GIVEN a job held open inside the repository + const app = serve(gate.worker); + const settled = queue.publish(aJob("job-1", "o-1", 1)); + await gate.arrived; + + // WHEN the drain starts and the job is released only once the phase moved. + // `vi.waitUntil` synchronises rather than asserts — the drain samples + // `inFlightAtStart` in the same synchronous turn that advances the phase, + // so releasing afterwards is what makes the report exact rather than racy. + app.requestDrain(); + await vi.waitUntil(() => app.phase() === "draining"); + gate.release(); + + // THEN the drain waited for the delivery to be acked — read through the + // settlement, so its own `Result` is consumed and a job that never settled + // could not be reported as completed + const report = await settled.flatMap(() => app.exited); + + expect(report).toBeOkWith( + expect.objectContaining({ drain: { inFlightAtStart: 1, completed: 1, abandoned: 0 } }), + ); + }); +}); diff --git a/examples/order-worker/src/queue-runtime.ts b/examples/order-worker/src/queue-runtime.ts new file mode 100644 index 0000000..814243c --- /dev/null +++ b/examples/order-worker/src/queue-runtime.ts @@ -0,0 +1,232 @@ +import type { Context, ServiceOf } from "@btravstack/di"; +import type { Runtime, RuntimeHost, Serving, UnitMeta } from "@btravstack/start"; +import { Logger, PlaceOrder } from "@btravstack/start-example-order-application"; +import { OkAsync, P, fromSafePromise, type AsyncResult } from "unthrown"; + +import type { Delivery, OrderQueue, PlaceOrderJob } from "./queue.js"; + +/** + * What the worker publishes about itself once it is consuming, read back + * through `RunningApp.runtimeInfo()`. + * + * Nothing like the API's `{ port, prefix }`, which is the point: `Serving.info` + * is the runtime's own shape and deliberately not modelled as a port number. A + * queue consumer has no port, and what an operator wants to know about one is + * which queue it is on and how many messages it will take at a time. + */ +export type OrderWorkerInfo = { readonly queue: string; readonly concurrency: number }; + +export type QueueWorkerOptions = { + readonly queue: OrderQueue; + /** How many deliveries may be in flight at once. Default `1`. */ + readonly concurrency?: number; + /** How many times a message may be delivered before it is parked. Default `3`. */ + readonly maxAttempts?: number; +}; + +/** + * The ports this runtime resolves out of the application context — two of the + * three the module exports, because `FindOrder` is not part of any job this + * worker consumes. A runtime declares what *it* needs, not what the module has. + * + * Non-empty on purpose: it is what makes `start`'s arity gate mean something. + * A module that does not export both fails to compile at the `start(...)` call, + * before anything runs (`src/needs-gate.test-d.ts` pins both directions). + */ +type WorkerNeeds = typeof PlaceOrder | typeof Logger; + +/** + * A `Runtime` consuming order jobs off a queue. + * + * - `start` subscribes to the queue and begins pumping. There is nothing to + * bind, so the error channel is empty here — a worker holding a real broker + * connection would report a failed connect as `Err(RuntimeStartFailed)`, + * exactly as the oRPC runtime reports a failed bind. + * - `Serving.drain` stops *claiming*. Deliveries already in flight are the + * kernel's to time out, and the kernel's deadline signal has nothing to + * cancel here. + * - `Serving.stop` is the same act with nothing left to add: an in-memory queue + * has no connection to close. + */ +export const queueWorkerRuntime = ( + options: QueueWorkerOptions, +): Runtime => ({ + name: "queue-worker", + needs: [PlaceOrder, Logger], + start: (host) => OkAsync(consume(host, options)), +}); + +/** + * How a delivery ends. This is the transport mapping, and the whole reason this + * package exists beside `order-api`: the same three channels, folded into a + * queue's vocabulary instead of HTTP's. + */ +type Disposition = + | { readonly kind: "ack" } + | { readonly kind: "dead-letter"; readonly reason: string } + | { readonly kind: "retry"; readonly reason: string }; + +const ack: Disposition = { kind: "ack" }; +const deadLetter = (reason: string): Disposition => ({ kind: "dead-letter", reason }); +const retry = (reason: string): Disposition => ({ kind: "retry", reason }); + +/** + * The triage point — the boundary where the application's vocabulary stops, + * and the mirror of `order-api`'s `mapErrCases` into `ORPCError` codes. + * + * The same `Err` lands somewhere else entirely. `DuplicateOrder` is a `CONFLICT` + * over HTTP because there is a caller waiting to be told; a queue has no caller, + * so the message is **parked** for a human instead. `InvalidQuantity` likewise: + * a redelivery would ask the same impossible thing again. What *is* worth + * another delivery is a `Defect` — an unmodelled failure is the infrastructure + * one, and infrastructure comes back. + * + * Every case is named. A new domain error is a compile error here, at the one + * place that has to decide what happens to the message. + */ +const dispositionOf = ( + ctx: Context>, + job: PlaceOrderJob, +): AsyncResult => + fromSafePromise( + ctx + .get(PlaceOrder) + .execute(job.orderId, job.quantity) + .match({ + ok: () => ack, + errCases: (matcher) => + matcher + .with(P.tag("InvalidQuantity"), (error) => deadLetter(error._tag)) + .with(P.tag("DuplicateOrder"), (error) => deadLetter(error._tag)), + defect: (cause) => retry(String(cause)), + }), + ); + +/** + * Applying the disposition, **inside the unit**. + * + * This is the queue-shaped form of the contract a runtime owes: a unit closes + * the instant its `Result` settles, and an idle registry is the kernel's + * permission to call `Serving.stop()`. Settling the message after the unit + * returned would race that — the message would be neither acked nor requeued + * when the process went away. Flushing a response and acking a message are the + * same obligation. + */ +const dispose = ( + logger: ServiceOf, + queue: OrderQueue, + delivery: Delivery, + maxAttempts: number, + disposition: Disposition, +): AsyncResult => { + const { job, attempt } = delivery; + + if (disposition.kind === "retry" && attempt < maxAttempts) { + logger.info(`job ${job.id} retried after attempt ${attempt}: ${disposition.reason}`); + queue.requeue(delivery); + return OkAsync(); + } + + logger.info(`job ${job.id} ${disposition.kind === "ack" ? "acked" : "dead-lettered"}`); + queue.settle( + disposition.kind === "ack" + ? { jobId: job.id, outcome: "acked", attempts: attempt } + : { jobId: job.id, outcome: "dead-lettered", reason: disposition.reason, attempts: attempt }, + ); + return OkAsync(); +}; + +const deliver = ( + host: RuntimeHost, + queue: OrderQueue, + delivery: Delivery, + maxAttempts: number, +): AsyncResult => + host.run(metaFor(delivery), (ctx, _signal) => + dispositionOf(ctx, delivery.job).flatMap((disposition) => + dispose(ctx.get(Logger), queue, delivery, maxAttempts, disposition), + ), + ); + +const consume = ( + host: RuntimeHost, + options: QueueWorkerOptions, +): Serving => { + const { queue } = options; + const concurrency = options.concurrency ?? 1; + const maxAttempts = options.maxAttempts ?? 3; + + let accepting = true; + let inFlight = 0; + + const pump = (): void => { + while (accepting && inFlight < concurrency) { + const delivery = queue.claim(); + if (delivery === undefined) return; + + inFlight += 1; + // The unit's outcome is FOLDED to a value here rather than dropped: + // `AsyncResult` has an empty *error* channel, but a `Defect` + // can still be present. + void deliver(host, queue, delivery, maxAttempts).match({ + ok: () => released(), + // Nothing can land in the error channel — a job's own failure became a + // disposition inside the unit — so the matcher has no case to name. + errCases: (matcher) => matcher, + // Reached only if the disposition machinery itself failed, which leaves + // the message neither acked nor requeued. Parking it is the one + // remaining courtesy: a producer waiting on it gets an answer. + defect: (cause) => { + queue.settle({ + jobId: delivery.job.id, + outcome: "dead-lettered", + reason: String(cause), + attempts: delivery.attempt, + }); + released(); + }, + }); + } + }; + + const released = (): void => { + inFlight -= 1; + pump(); + }; + + const unsubscribe = queue.subscribe(pump); + + const stopClaiming = (): void => { + accepting = false; + unsubscribe(); + }; + + pump(); + + return { + info: { queue: queue.name, concurrency }, + drain: (signal) => { + void signal; + stopClaiming(); + return OkAsync(); + }, + stop: () => { + stopClaiming(); + return OkAsync(); + }, + }; +}; + +/** + * `UnitMeta.id` must be unique per unit, and a *message* id is not: a retried + * message is delivered twice and is two units. The delivery is the unit, so the + * attempt is part of the id — and the message id becomes the `traceId`, which + * is exactly what `traceId` is for. It is the correlation id, minted outside + * this process, and it stays the same across every attempt, so all three + * deliveries of a message join up in the log. + */ +const metaFor = (delivery: Delivery): UnitMeta => ({ + kind: "job", + id: `${delivery.job.id}#${delivery.attempt}`, + traceId: delivery.job.id, +}); diff --git a/examples/order-worker/src/queue.ts b/examples/order-worker/src/queue.ts new file mode 100644 index 0000000..e421f55 --- /dev/null +++ b/examples/order-worker/src/queue.ts @@ -0,0 +1,103 @@ +import { fromSafePromise, type AsyncResult } from "unthrown"; + +/** + * One message: place this order, this many items. + * + * `id` is the **message** id, minted by whoever published it, and it is + * deliberately not the order id — a message can be delivered more than once, + * and correlating those deliveries is exactly what it is for. + */ +export type PlaceOrderJob = { + readonly id: string; + readonly orderId: string; + readonly quantity: number; +}; + +/** One **delivery** of a message: the job, and which attempt this is. */ +export type Delivery = { readonly job: PlaceOrderJob; readonly attempt: number }; + +/** + * How a message ended, once it is no longer the broker's problem. A retry is + * not here on purpose: it is not an ending, it is another delivery. + */ +export type Settlement = + | { readonly jobId: string; readonly outcome: "acked"; readonly attempts: number } + | { + readonly jobId: string; + readonly outcome: "dead-lettered"; + readonly reason: string; + readonly attempts: number; + }; + +/** + * The broker, reduced to the four things a worker actually needs of one — plus + * `publish`, which is the producer's half. + * + * It is a plain in-memory object rather than a di port because it is the + * *transport*, not an application capability: the runtime owns it exactly as + * the oRPC runtime owns its `node:http` server, and nothing in the application + * or persistence layers can name it. A real deployment swaps this for an AMQP + * channel and the runtime above is unchanged. + */ +export type OrderQueue = { + readonly name: string; + /** + * Publishes a job and resolves when it **settles** — acked or dead-lettered, + * however many deliveries that took. An `AsyncResult`, like every other async + * surface in this stack, with an empty error channel: a settlement always + * arrives, because the worker's attempt budget is finite. + */ + readonly publish: (job: PlaceOrderJob) => AsyncResult; + /** Consumer side: take the next delivery, or `undefined` if there is none. */ + readonly claim: () => Delivery | undefined; + /** Consumer side: hand a delivery back for one more attempt. */ + readonly requeue: (delivery: Delivery) => void; + /** Consumer side: this message is finished with. */ + readonly settle: (settlement: Settlement) => void; + /** Consumer side: wake up when there is something to claim. Unsubscribes. */ + readonly subscribe: (listener: () => void) => () => void; +}; + +export const createOrderQueue = (name = "orders"): OrderQueue => { + const pending: Delivery[] = []; + const settlements = new Map void>(); + const listeners = new Set<() => void>(); + + const notify = (): void => { + for (const listener of listeners) listener(); + }; + + return { + name, + + publish: (job) => + fromSafePromise( + // The executor runs synchronously, so the message is queued before + // `publish` returns — a worker already pumping picks it up on this + // very tick. + new Promise((resolve) => { + settlements.set(job.id, resolve); + pending.push({ job, attempt: 1 }); + notify(); + }), + ), + + claim: () => pending.shift(), + + requeue: (delivery) => { + pending.push({ job: delivery.job, attempt: delivery.attempt + 1 }); + notify(); + }, + + settle: (settlement) => { + const resolve = settlements.get(settlement.jobId); + settlements.delete(settlement.jobId); + if (resolve !== undefined) resolve(settlement); + }, + + subscribe: (listener) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + }; +}; diff --git a/examples/order-worker/src/test-fixtures.ts b/examples/order-worker/src/test-fixtures.ts new file mode 100644 index 0000000..7584067 --- /dev/null +++ b/examples/order-worker/src/test-fixtures.ts @@ -0,0 +1,185 @@ +import { Module, Port, Provider, type Scope, type ServiceOf } from "@btravstack/di"; +import { start, type RunningApp } from "@btravstack/start"; +import { + ApplicationModule, + FindOrder, + Logger, + OrderRepository, + PlaceOrder, +} from "@btravstack/start-example-order-application"; +import { OrderNotFound } from "@btravstack/start-example-order-domain"; +import { ErrAsync, fromSafePromise } from "unthrown"; +import { expect, test } from "vitest"; + +import { OrderWorkerModule } from "./module.js"; +import { queueWorkerRuntime, type OrderWorkerInfo } from "./queue-runtime.js"; +import { createOrderQueue, type OrderQueue, type PlaceOrderJob } from "./queue.js"; + +type App = RunningApp; + +/** + * `X` is pinned to the three ports the composition roots export rather than + * left generic: `start`'s needs gate is a phantom rest parameter proven at the + * call site, and no proof is available inside a helper generic in the module's + * own exports. The runtime needs only two of them. + */ +type WorkerPorts = PlaceOrder | FindOrder | Logger; + +type Serve = (module: Module) => App; + +/** A message id that is deliberately not the order id — see `PlaceOrderJob`. */ +const jobOf = (id: string, orderId: string, quantity: number): PlaceOrderJob => ({ + id, + orderId, + quantity, +}); + +const persistenceOf = (repository: ServiceOf) => + Module("StubPersistence")({ + provides: [Provider(OrderRepository)({ value: repository })], + exports: [OrderRepository], + }); + +/** + * A composition root shaped like the real one but with the repository swapped: + * same `ApplicationModule`, same runtime, same three exported ports, so the + * transport under test is unchanged. + */ +const workerWith = (repository: ServiceOf) => + Module("StubWorker")({ + imports: [ApplicationModule, persistenceOf(repository)], + provides: [], + exports: [PlaceOrder, FindOrder, Logger], + }); + +/** + * `start` hands the application context to the runtime alone, so a spec cannot + * reach `Logger` the way `Module.scoped` can. This publishes the very `Logger` + * service instance the use cases and the disposition write to. + */ +class LoggerTap extends Port("LoggerTap")<{ readonly lines: () => readonly string[] }> {} + +const tappedWorker = () => { + let read: () => readonly string[] = () => []; + + return { + worker: Module("TappedWorker")({ + imports: [OrderWorkerModule], + provides: [ + Provider(LoggerTap)([Logger], { + sync: (logger) => { + read = logger.lines; + return { lines: logger.lines }; + }, + }), + ], + exports: [PlaceOrder, FindOrder, Logger], + }), + traces: (): readonly string[] => read().map((line) => line.slice(0, line.indexOf("]") + 1)), + }; +}; + +/** + * A composition root whose repository fails in a way nobody modelled: no + * `qualify` triaged the rejection, so it is a defect — and a defect is what the + * worker retries. + */ +const unmodelledWorker = () => + workerWith({ + save: () => fromSafePromise(Promise.reject(new Error("the database is on fire"))), + find: (id) => ErrAsync(new OrderNotFound({ id })), + }); + +/** + * A repository whose `save` never settles until `release()` is called, and + * whose `arrived` promise reports the moment the job reached it. The drain spec + * turns on knowing a unit is genuinely in flight before the drain starts — + * polling a wall clock instead would be the flake. + */ +const gatedWorker = () => { + let entered!: () => void; + const arrived = new Promise((resolve) => { + entered = resolve; + }); + let release!: () => void; + const held = new Promise((resolve) => { + release = resolve; + }); + + return { + worker: workerWith({ + save: (order) => { + entered(); + return fromSafePromise(held.then(() => order)); + }, + find: (id) => ErrAsync(new OrderNotFound({ id })), + }), + arrived, + release: () => release(), + }; +}; + +export type WorkerFixtures = { + /** The very queue the runtime consumes, so a spec is the producer half. */ + readonly queue: OrderQueue; + /** + * Starts an app on that queue and registers its shutdown. The teardown runs + * even when the test fails, which is what a `try`/`finally` used to + * hand-roll — and it keeps the assertion those blocks carried: the app + * exited `Ok`. + */ + readonly serve: Serve; + readonly aJob: typeof jobOf; + readonly unmodelled: ReturnType; + readonly gate: ReturnType; + readonly tapped: ReturnType; +}; + +export const it = test.extend({ + // oxlint-disable-next-line no-empty-pattern -- Vitest fixtures require a destructuring pattern; this one depends on no other fixture + queue: async ({}, use) => { + await use(createOrderQueue()); + }, + + serve: async ({ queue }, use) => { + const shutdowns: (() => Promise)[] = []; + + const serve: Serve = (module) => { + const app = start(module, { + runtime: queueWorkerRuntime({ queue }), + signals: false, + probes: false, + preDrainDelayMs: 0, + }); + shutdowns.push(async () => { + app.stop(); + await expect(app.exited).toBeOk(); + }); + return app; + }; + + await use(serve); + + for (const shutdown of shutdowns) await shutdown(); + }, + + // oxlint-disable-next-line no-empty-pattern -- see above + aJob: async ({}, use) => { + await use(jobOf); + }, + + // oxlint-disable-next-line no-empty-pattern -- see above + unmodelled: async ({}, use) => { + await use(unmodelledWorker()); + }, + + // oxlint-disable-next-line no-empty-pattern -- see above + gate: async ({}, use) => { + await use(gatedWorker()); + }, + + // oxlint-disable-next-line no-empty-pattern -- see above + tapped: async ({}, use) => { + await use(tappedWorker()); + }, +}); diff --git a/examples/order-worker/src/vitest.d.ts b/examples/order-worker/src/vitest.d.ts new file mode 100644 index 0000000..ad36daf --- /dev/null +++ b/examples/order-worker/src/vitest.d.ts @@ -0,0 +1 @@ +import type {} from "@unthrown/vitest"; diff --git a/examples/order-worker/tsconfig.json b/examples/order-worker/tsconfig.json new file mode 100644 index 0000000..3faf372 --- /dev/null +++ b/examples/order-worker/tsconfig.json @@ -0,0 +1,13 @@ +{ + "extends": "@btravstack/tsconfig/base.json", + "compilerOptions": { + "noEmit": true, + "types": ["node"], + // Inherited from the persistence layer this package composes: the generated + // Prisma client imports its own files with explicit `.ts` extensions, and + // those files are part of this program too. `noEmit` makes that legal. + "allowImportingTsExtensions": true + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "src/**/*.test-d.ts"] +} diff --git a/examples/order-worker/tsconfig.test-d.json b/examples/order-worker/tsconfig.test-d.json new file mode 100644 index 0000000..619908b --- /dev/null +++ b/examples/order-worker/tsconfig.test-d.json @@ -0,0 +1,6 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { "noUnusedLocals": false, "noUnusedParameters": false }, + "include": ["src/**/*.test-d.ts"], + "exclude": ["node_modules"] +} diff --git a/examples/order-worker/vitest.config.ts b/examples/order-worker/vitest.config.ts new file mode 100644 index 0000000..fb76260 --- /dev/null +++ b/examples/order-worker/vitest.config.ts @@ -0,0 +1,9 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + environment: "node", + include: ["src/**/*.spec.ts"], + setupFiles: ["@unthrown/vitest"], + }, +}); diff --git a/knip.json b/knip.json index 5d17339..330da10 100644 --- a/knip.json +++ b/knip.json @@ -4,6 +4,7 @@ "ignore": ["**/*.test-d.ts"], "ignoreDependencies": ["@btravstack/lefthook", "@btravstack/oxlint"], "workspaces": { - "examples/order-api": { "entry": ["src/main.ts"] } + "examples/order-api": { "entry": ["src/main.ts"] }, + "examples/order-worker": { "entry": ["src/main.ts"] } } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4bdf253..e2ae80f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -298,6 +298,49 @@ importers: specifier: 'catalog:' version: 4.1.10(@types/node@26.1.2)(@vitest/coverage-v8@4.1.10)(jiti@2.7.0)(yaml@2.9.0) + examples/order-worker: + dependencies: + '@btravstack/di': + specifier: 'catalog:' + version: 0.1.0(unthrown@5.2.0) + '@btravstack/start': + specifier: workspace:* + version: link:../../packages/start + '@btravstack/start-example-order-application': + specifier: workspace:* + version: link:../order-application + '@btravstack/start-example-order-domain': + specifier: workspace:* + version: link:../order-domain + '@btravstack/start-example-order-infrastructure': + specifier: workspace:* + version: link:../order-infrastructure + '@unthrown/standard-schema': + specifier: 'catalog:' + version: 5.2.0 + unthrown: + specifier: 'catalog:' + version: 5.2.0 + zod: + specifier: 'catalog:' + version: 4.4.3 + devDependencies: + '@btravstack/tsconfig': + specifier: 'catalog:' + version: 0.2.0 + '@types/node': + specifier: 'catalog:' + version: 26.1.2 + '@unthrown/vitest': + specifier: 'catalog:' + version: 5.2.0(unthrown@5.2.0)(vitest@4.1.10) + typescript: + specifier: 'catalog:' + version: 7.0.2 + vitest: + specifier: 'catalog:' + version: 4.1.10(@types/node@26.1.2)(@vitest/coverage-v8@4.1.10)(jiti@2.7.0)(yaml@2.9.0) + packages/start: devDependencies: '@btravstack/di': From 0eba57fcecc373e74ca52c32e8666f506d2c14aa Mon Sep 17 00:00:00 2001 From: Benoit TRAVERS Date: Wed, 12 Aug 2026 01:09:23 +0200 Subject: [PATCH 2/2] docs(examples): state the precondition OrderQueue.publish resolves under MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `publish`'s docstring claimed a settlement always arrives "because the worker's attempt budget is finite". The budget bounds the redeliveries of a job a worker has *claimed*; nothing bounds the wait for a claim. With no worker running — or one that stopped or drained with jobs still queued — the promise never resolves and an awaited `publish` hangs. Say so, rather than inventing broker behaviour to satisfy the old sentence: settling still-pending messages on shutdown is the opposite of what a durable queue does, it cannot be honoured by the AMQP channel this example promises swaps in unchanged, and it would still not cover the no-worker case. `Serving.drain` gets the same treatment: it stops *claiming*, so a message not yet claimed stays in the queue unsettled and the drain does not wait for it. Two specs pin both halves, racing a publish against one macrotask turn via the new `withinATurn` fixture — a bound, since the delivery path holds no timer — so the legitimate never-settles state fails in a millisecond instead of timing out. 150 tests -> 152. --- CLAUDE.md | 2 +- examples/README.md | 2 +- examples/order-worker/README.md | 30 +++++++++++- .../order-worker/src/queue-runtime.spec.ts | 49 +++++++++++++++++++ examples/order-worker/src/queue-runtime.ts | 6 ++- examples/order-worker/src/queue.ts | 22 +++++++-- examples/order-worker/src/test-fixtures.ts | 43 ++++++++++++++-- 7 files changed, 143 insertions(+), 11 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 7d1b151..2bc13c8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -568,7 +568,7 @@ Source layout (`packages/start/src/`), one concept per file: `ambient.ts` ## Toolchain & conventions - **`examples/` is part of the gate, not a folder of illustrations.** All five - workspaces run under the same six commands as the kernel — 57 specs plus two + workspaces run under the same six commands as the kernel — 59 specs plus two `needs-gate.test-d.ts` files — so an example that stops compiling, stops linting or stops passing fails CI exactly as `packages/start` would. They are also the only place a runtime with a **non-empty `needs`** meets a real diff --git a/examples/README.md b/examples/README.md index cea2cf4..8ac42fa 100644 --- a/examples/README.md +++ b/examples/README.md @@ -81,7 +81,7 @@ missing need. ## Why these are tests, not just illustrations -Each package reads as application code, and each is covered by real specs — 57 +Each package reads as application code, and each is covered by real specs — 59 of them, run by the repository's own `pnpm test`: ```sh diff --git a/examples/order-worker/README.md b/examples/order-worker/README.md index 05fd8ac..574b5be 100644 --- a/examples/order-worker/README.md +++ b/examples/order-worker/README.md @@ -101,6 +101,34 @@ the message id becomes the `traceId` — which is exactly what `traceId` is for. It is the correlation id, minted outside this process, and holding it steady across attempts is what joins three deliveries into one trace. +## A publish resolves on a **worker**, not on a broker + +`OrderQueue.publish` hands the producer the consumer's outcome: + +```ts +await expect(queue.publish(aJob("job-1", "o-1", 2))).toBeOkWith({ + jobId: "job-1", + outcome: "acked", + attempts: 1, +}); +``` + +That is a test convenience, and it carries a precondition worth stating: a real +AMQP `publish` resolves on the **broker's** ack and the producer never learns +how the message ended. Here it resolves when a **running worker settles** the +job — so the attempt budget, which bounds the retries of a job a worker has +claimed, says nothing about a job being claimed at all. Publish with no worker +running, or leave a message behind when one drains, and the returned +`AsyncResult` **never settles**: awaiting it waits forever. + +That is not a bug in the queue — it is what a broker does with an unconsumed +message, and `Serving.drain` stopping at _claiming_ is the right shape (the +next worker on the queue takes it). It is a bug waiting to happen in a spec, so +two of them pin it, racing the publish against one macrotask turn rather than +awaiting it: _"never settles a job published with no worker running"_ and +_"leaves a job the drain never claimed unsettled, without waiting for it"_. Both +fail in a millisecond instead of hanging until Vitest's timeout. + ## `Serving.info` with no port in it ```ts @@ -115,7 +143,7 @@ is which queue it is on and how many messages it will take at a time. ## Running it ```bash -pnpm --filter @btravstack/start-example-order-worker test # 7 runtime specs + 4 env specs +pnpm --filter @btravstack/start-example-order-worker test # 9 runtime specs + 4 env specs pnpm --filter @btravstack/start-example-order-worker test:types # the needs gate ``` diff --git a/examples/order-worker/src/queue-runtime.spec.ts b/examples/order-worker/src/queue-runtime.spec.ts index 48ce086..c9876b1 100644 --- a/examples/order-worker/src/queue-runtime.spec.ts +++ b/examples/order-worker/src/queue-runtime.spec.ts @@ -125,6 +125,55 @@ describe("queueWorkerRuntime", () => { ]); }); + it("never settles a job published with no worker running", async ({ + queue, + aJob, + withinATurn, + }) => { + // GIVEN a job published with nothing serving — no `serve`, so nothing will + // ever claim it + const published = queue.publish(aJob("job-1", "o-1", 1)); + + // WHEN it is given a full turn to settle + const outcome = await withinATurn(published); + + // THEN it is still pending, which is the precondition `publish` documents: + // a settlement comes from a *worker*, and the attempt budget bounds the + // retries of a claimed job, not the wait for a claim. Awaiting it here + // would hang the suite — racing a turn is what makes it a failure instead + expect(outcome).toBe("unsettled"); + }); + + it("leaves a job the drain never claimed unsettled, without waiting for it", async ({ + serve, + queue, + aJob, + gate, + withinATurn, + }) => { + // GIVEN a worker at its concurrency limit — one delivery held open inside + // the repository, and a second job queued behind it + const app = serve(gate.worker); + const held = queue.publish(aJob("job-1", "o-1", 1)); + await gate.arrived; + const queued = queue.publish(aJob("job-2", "o-2", 1)); + + // WHEN the drain runs to completion, the in-flight delivery released only + // once the phase moved + app.requestDrain(); + await vi.waitUntil(() => app.phase() === "draining"); + gate.release(); + await vi.waitUntil(() => app.phase() === "exited"); + + // THEN the drain waited for the delivery it had claimed and not for the one + // it had not: `Serving.drain` stops claiming, so an unclaimed message stays + // in the queue for the next worker — and this producer waits forever + expect({ inFlight: await withinATurn(held), unclaimed: await withinATurn(queued) }).toEqual({ + inFlight: "settled", + unclaimed: "unsettled", + }); + }); + it("waits for the in-flight job while draining", async ({ serve, queue, aJob, gate }) => { // GIVEN a job held open inside the repository const app = serve(gate.worker); diff --git a/examples/order-worker/src/queue-runtime.ts b/examples/order-worker/src/queue-runtime.ts index 814243c..fba1fa9 100644 --- a/examples/order-worker/src/queue-runtime.ts +++ b/examples/order-worker/src/queue-runtime.ts @@ -44,7 +44,11 @@ type WorkerNeeds = typeof PlaceOrder | typeof Logger; * exactly as the oRPC runtime reports a failed bind. * - `Serving.drain` stops *claiming*. Deliveries already in flight are the * kernel's to time out, and the kernel's deadline signal has nothing to - * cancel here. + * cancel here. A message not yet claimed stays in the queue **unsettled**, + * and the drain does not wait for it: draining hands nothing back to a + * broker, it stops taking more, and the next worker on the queue takes it. + * So a producer awaiting that message's settlement — the convenience + * `OrderQueue.publish` offers — waits forever. * - `Serving.stop` is the same act with nothing left to add: an in-memory queue * has no connection to close. */ diff --git a/examples/order-worker/src/queue.ts b/examples/order-worker/src/queue.ts index e421f55..48976c2 100644 --- a/examples/order-worker/src/queue.ts +++ b/examples/order-worker/src/queue.ts @@ -42,10 +42,24 @@ export type Settlement = export type OrderQueue = { readonly name: string; /** - * Publishes a job and resolves when it **settles** — acked or dead-lettered, - * however many deliveries that took. An `AsyncResult`, like every other async - * surface in this stack, with an empty error channel: a settlement always - * arrives, because the worker's attempt budget is finite. + * Publishes a job and resolves when a **running worker settles** it — acked + * or dead-lettered, however many deliveries that took. + * + * That is a precondition, not a guarantee. The attempt budget bounds the + * retries of a job a worker has *claimed*; nothing bounds the wait for a + * claim. A job published with no worker running, or left in the queue when + * one stops or drains, is never settled and **awaiting it waits forever** — + * which is what a broker does with an unconsumed message, rather than a + * defect of this one. `queue-runtime.spec.ts` pins both halves. + * + * The empty error channel is honest about something narrower: publishing + * itself cannot fail, and a dead-letter is a settlement rather than an error. + * + * Resolving on the *consumer's* outcome is a deliberate test convenience: a + * real AMQP `publish` resolves on the broker's ack and the producer never + * learns how the message ended. It is what lets a spec be the producer half + * and assert the disposition in one `expect` — and it is why awaiting one is + * only ever safe under a serving worker. */ readonly publish: (job: PlaceOrderJob) => AsyncResult; /** Consumer side: take the next delivery, or `undefined` if there is none. */ diff --git a/examples/order-worker/src/test-fixtures.ts b/examples/order-worker/src/test-fixtures.ts index 7584067..e6cba07 100644 --- a/examples/order-worker/src/test-fixtures.ts +++ b/examples/order-worker/src/test-fixtures.ts @@ -8,12 +8,12 @@ import { PlaceOrder, } from "@btravstack/start-example-order-application"; import { OrderNotFound } from "@btravstack/start-example-order-domain"; -import { ErrAsync, fromSafePromise } from "unthrown"; +import { ErrAsync, fromSafePromise, type AsyncResult } from "unthrown"; import { expect, test } from "vitest"; import { OrderWorkerModule } from "./module.js"; import { queueWorkerRuntime, type OrderWorkerInfo } from "./queue-runtime.js"; -import { createOrderQueue, type OrderQueue, type PlaceOrderJob } from "./queue.js"; +import { createOrderQueue, type OrderQueue, type PlaceOrderJob, type Settlement } from "./queue.js"; type App = RunningApp; @@ -34,6 +34,30 @@ const jobOf = (id: string, orderId: string, quantity: number): PlaceOrderJob => quantity, }); +/** + * Reports whether a publish settled within one macrotask turn. + * + * `publish` only resolves when a running worker settles the job, so awaiting + * one that nobody will claim hangs the test until Vitest's timeout — a broken + * suite where the fact under test is a legitimate state. The turn boundary is + * a bound rather than a guess: the delivery path is microtasks end to end, with + * no timer anywhere in it, so anything a serving worker was going to settle has + * settled by the time the immediate fires. + */ +const settledWithinATurn = ( + published: AsyncResult, +): Promise<"settled" | "unsettled"> => + Promise.race([ + published.match({ + ok: () => "settled" as const, + errCases: (matcher) => matcher, + defect: () => "settled" as const, + }), + new Promise<"unsettled">((resolve) => { + setImmediate(() => resolve("unsettled")); + }), + ]); + const persistenceOf = (repository: ServiceOf) => Module("StubPersistence")({ provides: [Provider(OrderRepository)({ value: repository })], @@ -120,7 +144,13 @@ const gatedWorker = () => { }; export type WorkerFixtures = { - /** The very queue the runtime consumes, so a spec is the producer half. */ + /** + * The very queue the runtime consumes, so a spec is the producer half. + * + * A publish is awaited only under a `serve`d worker — that is the + * precondition `OrderQueue.publish` documents, and `settledWithinATurn` is + * how the two specs that go without one stay a failure instead of a hang. + */ readonly queue: OrderQueue; /** * Starts an app on that queue and registers its shutdown. The teardown runs @@ -130,6 +160,8 @@ export type WorkerFixtures = { */ readonly serve: Serve; readonly aJob: typeof jobOf; + /** Races a publish against one macrotask turn — see `settledWithinATurn`. */ + readonly withinATurn: typeof settledWithinATurn; readonly unmodelled: ReturnType; readonly gate: ReturnType; readonly tapped: ReturnType; @@ -168,6 +200,11 @@ export const it = test.extend({ await use(jobOf); }, + // oxlint-disable-next-line no-empty-pattern -- see above + withinATurn: async ({}, use) => { + await use(settledWithinATurn); + }, + // oxlint-disable-next-line no-empty-pattern -- see above unmodelled: async ({}, use) => { await use(unmodelledWorker());