diff --git a/.changeset/initial-kernel.md b/.changeset/initial-kernel.md index 2714ea7..b6efaa7 100644 --- a/.changeset/initial-kernel.md +++ b/.changeset/initial-kernel.md @@ -31,7 +31,7 @@ application scope on every path. - `runMain`, which turns an outcome into a process exit code (`0` / `1` / `2` / `70`) by setting `process.exitCode`. - `currentUnit()` over an `AsyncLocalStorage` record carrying - `{ unitId, traceId, tenantId, deadline }` — data, never capabilities. + `{ unitId, traceId, tenantId, deadline, signal }` — data, never capabilities. - A `@btravstack/testing` package with `testRuntime`, `createFakeClock` and `withApp`. - **Every async API returns an `AsyncResult`, never a bare `Promise`** — the diff --git a/.changeset/unit-signal.md b/.changeset/unit-signal.md new file mode 100644 index 0000000..b46a91f --- /dev/null +++ b/.changeset/unit-signal.md @@ -0,0 +1,37 @@ +--- +"@btravstack/core": minor +"@btravstack/temporal": patch +"@btravstack/amqp": patch +--- + +**`UnitRecord` gains `signal: AbortSignal`** — the ambient record is five +fields now, not four. It is the **very** controller the unit's work callback is +handed, not a copy: one abort, two ways to reach it, fired at the drain +deadline or at once on a path that skips the drain. + +The gap it closes: a middleware-shaped runtime opens its unit around a call it +does not own the arguments of. `@btravstack/temporal`'s `activityUnits` and +`@btravstack/amqp`'s `messageUnits` both hand the kernel a work callback that +_is_ the library's `next()`, so an activity or a handler had no parameter to +receive the signal through and the kernel's `drainTimeoutMs` was unobservable +from inside the work. Injecting a context the transport's contract does not +type was the alternative, and it is exactly the hidden-dependency shape `di` +exists to prevent, so the signal travels on the record instead — data about +this unit, like `deadline`, with nothing to substitute in a test. +`@btravstack/http` is unchanged: it still passes the same signal as its +handler's third parameter. + +What each transport does with it is the transport's own business, and both +examples are worked: + +- **`examples/order-amqp-worker`** answers a `RetryableError` when + `currentUnit()?.signal.aborted`, leaving the delivery un-acked so the broker + hands it to the next worker. This transport has no cancellation of its own — + a redelivery is recovery, not cancellation. +- **`examples/order-temporal-worker`**'s `ShippingService.arrange` fails as a + **defect**, which the platform retries on another worker. The contract's + `ShippingUnavailable` is a permanent no and would be the wrong error for "we + ran out of time". Temporal's `Context.current().cancellationSignal` is a + different clock — workflow-side cancellation, and worker shutdown after + `shutdownGraceTime` — so the two are honoured together rather than one + standing in for the other. diff --git a/CLAUDE.md b/CLAUDE.md index 6f4e26c..49cd569 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -93,12 +93,24 @@ hook). User-facing changes need a changeset. 2. **Ambient carries DATA. The DI `Context` carries CAPABILITIES.** The kernel opens one `AsyncLocalStorage` store per unit holding a small, fixed record — - `{ unitId, traceId, tenantId, deadline }` (`UnitRecord` in `units.ts`) — + `{ unitId, traceId, tenantId, deadline, signal }` (`UnitRecord` in + `units.ts`) — and nothing else. Services never go in it. The line holds because what `di` exists to prevent is hidden _dependencies_: code that secretly needs a collaborator it never declared and cannot be tested without it. A trace id is not a collaborator — no substitutability question, no test double, nothing to - swap. A repository pulled from an ambient store is the untestable coupling; a + swap. Nor is an `AbortSignal`: `signal` is the **very** controller the work + callback is handed — one abort, two ways to reach it — and it is on the + record because the callback is not always where the work is. A + middleware-shaped runtime (`@btravstack/temporal`, `@btravstack/amqp`) opens + the unit around a call it does not own the arguments of, so an activity or a + handler has no parameter to receive it through, and injecting a context the + contract does not type was the alternative and was rejected; + `@btravstack/http` passes the same signal as its handler's third parameter, + which is that signal by another route. A transport's own cancellation — + Temporal's `Context.current().cancellationSignal` — is a **different clock**, + not this one. A repository pulled from an ambient store is the untestable + coupling; a tenant id read by the Postgres adapter is not. Legitimate readers are infrastructure adapters only (logger, OTel exporter, database adapter); application code reading the store is meant to be a lint error, in the spirit @@ -403,6 +415,12 @@ UnitNeeds>`: `NO RUNTIME` when the module exports no runtime port, `Result` — the `Promise` arm is Thesis #6's second exception, since it exists to accept a caller's `async` handler. `UnitRegistry.awaitIdle()` returns `AsyncResult`. +- **`UnitRecord`** — the ambient record: `{ unitId, traceId, tenantId, deadline, +signal }`. `signal` is the same `AbortSignal` `UnitWork` receives as its + argument — aborted at the drain deadline, or at once on a path that skips the + drain (`abortAll`) — carried here so a runtime whose work callback is a + library's `next()` still reaches it. Guarded by `units.spec.ts` → _"carries + the work's own AbortSignal on the ambient record"_. - **`currentUnit()` → `UnitRecord | undefined`** — the ambient read. `undefined` outside a unit. - **`Clock` / `systemClock`** — `{ now, sleep(ms, signal?) }`, where `sleep` @@ -720,7 +738,16 @@ orderActivities, workflows, imports })`, the sugar importing the starter; sync })`, `AmqpModule("OrderAmqpWorker")({ contract, handlers: orderHandlers, … })`), with its outbox relay a resourceful provider of its own rather than - something layered onto the runtime. + something layered onto the runtime. Both are also where **honouring the + kernel's deadline through the ambient record** is worked: neither middleware + injects anything into the call — `next()` unchanged — so + `currentUnit()?.signal` is the only route to it, and what each answers when + it is aborted is the transport's own business. `order-amqp-worker`'s + `orderChanged` returns a `RetryableError`, leaving the delivery un-acked so + the broker hands it to the next worker; `order-temporal-worker`'s + `ShippingService.arrange` fails as a **defect**, which the platform retries + on another worker — the contract's `ShippingUnavailable` is a permanent no + and would be the wrong error for "we ran out of time". - **`examples/order-api` consumes `@btravstack/http` rather than hand-rolling a transport, and its HTTP stack is the package's ONE way: oRPC over its own node adapter, `@unthrown/orpc` at the boundary.** The router is a di-provided diff --git a/docs/examples/order-amqp-worker.md b/docs/examples/order-amqp-worker.md index f1a28a1..b703e3c 100644 --- a/docs/examples/order-amqp-worker.md +++ b/docs/examples/order-amqp-worker.md @@ -49,6 +49,13 @@ export const orderHandlers = AmqpHandlers(orderContract)([Logger], { sync: (logger) => ({ orderChanged: (message) => { const { id, payload } = message.payload; + if (currentUnit()?.signal.aborted === true) { + return ErrAsync( + new RetryableError( + `the drain deadline passed before order ${id} was notified`, + ), + ); + } logger.info( payload === null ? `order ${id} is gone — notifying` @@ -68,6 +75,14 @@ placement's `Err` never crosses the broker, only the committed fact does — which is why this deployment is absent from the `Err` table on the [overview](/examples/). +The `currentUnit()?.signal` guard is the deployment's one kernel touchpoint, +and it is how a handler honours the drain deadline at all: `messageUnits` +calls `next()` unchanged, so there is no parameter to receive a signal +through and the ambient record is the only route to it. Answering a +`RetryableError` leaves the delivery **un-acked**, so the broker hands it to +the next worker rather than this one finishing work nobody is waiting for. +See [Read the ambient unit from an adapter](/how-to/read-the-ambient-unit). + ## The relay: a resourceful provider with its own config The relay's one piece of configuration is a slice of this deployment's own, diff --git a/docs/examples/order-temporal-worker.md b/docs/examples/order-temporal-worker.md index 0e7dfa0..46cc84c 100644 --- a/docs/examples/order-temporal-worker.md +++ b/docs/examples/order-temporal-worker.md @@ -153,8 +153,35 @@ export const OrderTemporalWorker = TemporalModule("OrderTemporalWorker")({ The same `ApplicationModule` + `PersistenceModule` pair as the API, plus `FulfillmentModule` — the two external services as in-memory stand-ins that -always say yes, because what this deployment demonstrates is the -orchestration; the specs swap in twins that say no. `TemporalModule` imports +say yes to anything the drain still has time for, because what this deployment +demonstrates is the orchestration; the specs swap in twins that say no. +`ShippingService.arrange` is the exception, and the deployment's one kernel +touchpoint: + +```ts +arrange: (orderId) => + currentUnit()?.signal.aborted === true + ? fromSafePromise( + Promise.reject( + new Error( + `the drain deadline passed before shipping for ${orderId} was arranged`, + ), + ), + ) + : (logger.info(`arranged shipping for order ${orderId}`), OkAsync()), +``` + +An adapter is where reading the ambient record is legitimate, and here it is +the only route to the unit's `AbortSignal` at all: `activityUnits` calls +`next()` unchanged, so an activity has no parameter to receive one through. +Failing as a **defect** is the point — the platform retries that attempt on +another worker, where the contract's `ShippingUnavailable` is a permanent no +and would be the wrong answer to "we ran out of time". Temporal's own +`Context.current().cancellationSignal` is a different clock, firing on +`shutdownGraceTime`; the two are honoured together. See +[Read the ambient unit from an adapter](/how-to/read-the-ambient-unit). + +`TemporalModule` imports the starter (`TemporalRuntime`, `TemporalConfig` from `TEMPORAL_ADDRESS` / `TEMPORAL_NAMESPACE`, `TemporalConnection` as a resource of the graph), provides the activities and exports the runtime. `main.ts` is diff --git a/docs/explanation/ambient-vs-context.md b/docs/explanation/ambient-vs-context.md index f3442bc..8ad5181 100644 --- a/docs/explanation/ambient-vs-context.md +++ b/docs/explanation/ambient-vs-context.md @@ -1,6 +1,6 @@ --- title: Ambient data, injected capabilities -description: Why the kernel's AsyncLocalStorage record holds four fields of data and never a service, who is meant to read it, and the lint rule that does not exist yet. +description: Why the kernel's AsyncLocalStorage record holds five fields of data and never a service, who is meant to read it, and the lint rule that does not exist yet. --- # Ambient data, injected capabilities @@ -12,7 +12,7 @@ description: Why the kernel's AsyncLocalStorage record holds four fields of data > [`UnitMeta` and `UnitRecord`](/reference/core/runtime). The kernel opens one `AsyncLocalStorage` store per unit of work. It holds a -small, fixed record — `UnitRecord`, four fields — and nothing else: +small, fixed record — `UnitRecord`, five fields — and nothing else: ```ts type UnitRecord = { @@ -20,6 +20,7 @@ type UnitRecord = { readonly traceId: string; readonly tenantId: string | undefined; readonly deadline: number | undefined; + readonly signal: AbortSignal; }; ``` @@ -53,7 +54,7 @@ So the line is: **ambient carries data, the di `Context` carries capabilities**. A repository through the store is the untestable coupling. A tenant id read by the Postgres adapter is not. -## Why exactly these four +## Why exactly these five Each field is a fact the kernel either mints or is handed at `run`, and each is one an adapter genuinely needs without being able to declare it. @@ -71,10 +72,42 @@ one an adapter genuinely needs without being able to declare it. tagging a span — are exactly the readers the store is for. - `deadline` is a timestamp a runtime may pass so an adapter can budget a remote call against what is left. +- `signal` is the **very** `AbortSignal` the kernel hands the unit's work + callback — one controller, two ways to reach it — fired at the drain + deadline, or at once on a path that skips the drain. There is no `Map`, no `set()`, no way for application code to add a field. A record that could grow would be a service locator with a smaller name. +### Is a signal really data? + +It is the one field that looks like a capability, so it is worth saying why it +is not. The test this page uses everywhere else is substitutability: a +collaborator has an interface behind it, a test double to swap in, a +`deps` array entry it should have been declared through. An `AbortSignal` +has none of those. It is a fact about _this_ unit — "the process has stopped +waiting for you" — exactly as `deadline` is the fact "this is when it will". +Nothing about the code that reads it changes shape when it is absent; the +`?.` in `currentUnit()?.signal` is the whole of the fallback. + +The reason it is on the record at all is that **the work callback is not +always where the work is**. `@btravstack/http` opens the unit around its own +listener, so it passes the signal as the handler's third parameter and never +needs the record. `@btravstack/temporal` and `@btravstack/amqp` are +middleware-shaped: the kernel's work callback is the library's `next()`, and +an activity or a handler has no parameter to receive a signal through. The +alternative was injecting a context — an extra first argument the Temporal or +AMQP contract does not type — which is exactly the hidden-dependency shape +this page argues against, so it was rejected. A deadline nobody can observe is +not a deadline. + +A transport's own cancellation is a **different clock** and stays separate. +Temporal's `Context.current().cancellationSignal` fires on a workflow-side +cancellation, and on worker shutdown after `shutdownGraceTime`; AMQP has no +cancellation story at all, since an un-acked delivery is redelivered, which is +recovery rather than cancellation. The two are honoured together, not one +standing in for the other. + ## Who reads it Legitimate readers are **infrastructure adapters only** — the logger, the @@ -101,7 +134,9 @@ The shipped runtimes are written to that rule. `@btravstack/http` opens a unit per request, `@btravstack/temporal` one per activity attempt and `@btravstack/amqp` one per delivery, and each injects nothing — a handler is a closure over the services its provider declared, and the ambient record is what an -adapter underneath reads. +adapter underneath reads. For the two middleware-shaped ones the record is +also the only route to the unit's `AbortSignal`, since they call `next()` +unchanged; `@btravstack/http` passes the same signal as an argument instead. ## The lint rule that does not exist diff --git a/docs/explanation/draining-in-three-beats.md b/docs/explanation/draining-in-three-beats.md index b1ebd0a..e8a136c 100644 --- a/docs/explanation/draining-in-three-beats.md +++ b/docs/explanation/draining-in-three-beats.md @@ -81,7 +81,13 @@ connections, a Temporal worker finishing an activity — that window is wide. So `awaitIdle` is **sequenced behind** the runtime's `drain`, never alongside it. At the deadline, whatever is still open is aborted through each unit's -`AbortSignal` and counted: +`AbortSignal` — the one the kernel handed the work callback, and the one on +the unit's ambient record as `currentUnit()?.signal`, which are the same +object. Two routes matter because the callback is not always where the work +is: a middleware-shaped runtime hands the kernel a callback that is the +library's own `next()`, so a Temporal activity or an AMQP handler has no +parameter to receive a signal through and reads it off the record instead. +Whatever is aborted is then counted: ```ts type DrainReport = { diff --git a/docs/how-to/consume-amqp-messages.md b/docs/how-to/consume-amqp-messages.md index 8c5eb3a..dadb8bb 100644 --- a/docs/how-to/consume-amqp-messages.md +++ b/docs/how-to/consume-amqp-messages.md @@ -175,6 +175,31 @@ id. A delivery tag is not used as the id on purpose: tags are per-channel and restart at `1` after the silent reconnects `amqp-connection-manager` performs. An adapter reads the trace id from `currentUnit()`. +## Honouring the drain deadline + +The middleware calls `next()` unchanged, so a handler has no parameter to +receive the unit's `AbortSignal` through: `currentUnit()?.signal` is the only +route to it, and it is aborted at the kernel's `drainTimeoutMs`. + +```ts +orderChanged: (message) => { + const { id, payload } = message.payload; + if (currentUnit()?.signal.aborted === true) { + return ErrAsync( + new RetryableError( + `the drain deadline passed before order ${id} was notified`, + ), + ); + } + // … +}; +``` + +A `RetryableError` leaves the delivery **un-acked**, so the broker hands it to +the next worker — the transport's own answer to "this process stopped waiting". +There is no cancellation to defer to here: AMQP has none, and a redelivery is +recovery rather than cancellation. + ## The drain: one deadline `Serving.drain(signal)` calls `worker.close({ drainTimeoutMs: null })` — diff --git a/docs/how-to/read-the-ambient-unit.md b/docs/how-to/read-the-ambient-unit.md index 6e79111..ec17cf7 100644 --- a/docs/how-to/read-the-ambient-unit.md +++ b/docs/how-to/read-the-ambient-unit.md @@ -1,12 +1,13 @@ --- title: Read the ambient unit from an adapter -description: Stamp a trace id on every log line with currentUnit(), know which code may read it, and see how each starter fills the record. +description: Stamp a trace id on every log line with currentUnit(), honour the drain deadline from an activity or a handler, know which code may read it, and see how each starter fills the record. --- # Read the ambient unit from an adapter > **How-to.** Read the kernel's per-unit record — trace id, tenant id, -> deadline — from a logger, an exporter or a database adapter, without +> deadline, the unit's `AbortSignal` — from a logger, an exporter, a database +> adapter or an activity, without > threading it through every signature. For _why_ this is data and not a > capability, see [Ambient data, injected capabilities](/explanation/ambient-vs-context). @@ -22,23 +23,28 @@ type UnitRecord = { readonly traceId: string; // the correlation id — `UnitMeta.traceId`, defaulting to `UnitMeta.id` readonly tenantId: string | undefined; // `UnitMeta.tenantId`, if the runtime supplied one readonly deadline: number | undefined; // `UnitMeta.deadline`, if the runtime supplied one + readonly signal: AbortSignal; // the unit's own — the very one the work callback is handed }; ``` `unitId` tells two units apart and needs nothing from the runtime. `traceId` is the one that joins a line logged here to a trace that started elsewhere. `tenantId` and `deadline` are plain data the runtime may stamp; **no shipped -starter sets either today**. Cancellation is not the record's job — the -`AbortSignal` the kernel hands unit work is what fires at the drain deadline. +starter sets either today**. `signal` is always there: the kernel mints one +`AbortController` per unit, hands its signal to the work callback **and** puts +that same object on the record, so both routes see one abort — at the drain +deadline, or at once on a path that skips the drain. ## Who may read it -| Reader | Reads `currentUnit()`? | -| ------------------------------------------------- | ---------------------- | -| a logger adapter | yes | -| an OTel exporter or span processor | yes | -| a database adapter stamping `tenantId` on a query | yes | -| a use case, a domain service, a router procedure | **no** | +| Reader | Reads `currentUnit()`? | +| ---------------------------------------------------------- | ---------------------- | +| a logger adapter | yes | +| an OTel exporter or span processor | yes | +| a database adapter stamping `tenantId` on a query | yes | +| an adapter checking `signal` before an outbound call | yes | +| a Temporal activity or an AMQP handler honouring the drain | yes — see below | +| a use case, a domain service, a router procedure | **no** | The line holds because what di exists to prevent is hidden _dependencies_: code that secretly needs a collaborator it never declared. A trace id is not a @@ -106,6 +112,76 @@ A unit-scoped finaliser runs **while the unit is still open**, so a request's own trace id — `examples/order-api/src/request-scope.ts` relies on exactly that. +## Recipe: honour the drain deadline + +The record's `signal` is the unit's own. When the drain runs out of time the +kernel aborts every unit still open, and work that keeps going is work nobody +in this process is waiting for any more. + +Which route you take to the signal depends on the runtime's shape: + +| Runtime | Where the signal is | +| ---------------------- | ---------------------------------------------------------------------------- | +| `@btravstack/http` | the handler's third parameter — and `currentUnit()?.signal`, the same object | +| `@btravstack/temporal` | `currentUnit()?.signal` only | +| `@btravstack/amqp` | `currentUnit()?.signal` only | + +The two workers are middleware-shaped: the kernel's work callback is the +library's `next()`, so an activity or a handler has **no parameter** to receive +a signal through. Injecting a context the contract does not type was the +alternative, and it is the hidden-dependency shape this stack exists to avoid. + +What you answer when the signal has fired is the transport's business, not the +kernel's. On AMQP, an un-acked delivery goes back to the broker, so a +`RetryableError` hands the message to the next worker — +`examples/order-amqp-worker/src/handlers.ts`: + +```ts +export const orderHandlers = AmqpHandlers(orderContract)([Logger], { + sync: (logger) => ({ + orderChanged: (message) => { + const { id, payload } = message.payload; + if (currentUnit()?.signal.aborted === true) { + return ErrAsync( + new RetryableError( + `the drain deadline passed before order ${id} was notified`, + ), + ); + } + logger.info(`order ${id} placed — notifying`); + return OkAsync(); + }, + }), +}); +``` + +On Temporal, the platform retries an attempt that fails as a **defect** on +another worker, which is the right shape for "we ran out of time" — where the +contract's own `ShippingUnavailable` is a permanent no and would be the wrong +error. `examples/order-temporal-worker/src/fulfillment.ts`: + +```ts +Provider(ShippingService)([Logger], { + sync: (logger) => ({ + arrange: (orderId) => + currentUnit()?.signal.aborted === true + ? fromSafePromise( + Promise.reject( + new Error( + `the drain deadline passed before shipping for ${orderId} was arranged`, + ), + ), + ) + : (logger.info(`arranged shipping for order ${orderId}`), OkAsync()), + }), +}); +``` + +A transport's **own** cancellation is a different clock and stays separate. +Temporal's `Context.current().cancellationSignal` fires on a workflow-side +cancellation and on worker shutdown after `shutdownGraceTime`; AMQP has none at +all. Honour both where both exist — neither stands in for the other. + ## How each starter fills `traceId` Every shipped runtime mints `UnitMeta.id` fresh per unit and puts the diff --git a/docs/how-to/run-a-temporal-worker.md b/docs/how-to/run-a-temporal-worker.md index 6c05153..4b38c4d 100644 --- a/docs/how-to/run-a-temporal-worker.md +++ b/docs/how-to/run-a-temporal-worker.md @@ -232,6 +232,34 @@ and `traceId` is the workflow id, the correlation id minted outside this process and stable across every retry. An adapter reads either from `currentUnit()`; the middleware injects nothing into the activity itself. +## Honouring the drain deadline + +Because the middleware injects nothing, `currentUnit()?.signal` is the **only** +route to the unit's `AbortSignal` from inside an activity — there is no +parameter to receive one through, and adding a context the contract does not +type was the alternative. It is aborted at the kernel's `drainTimeoutMs`: + +```ts +arrange: (orderId) => + currentUnit()?.signal.aborted === true + ? fromSafePromise( + Promise.reject( + new Error( + `the drain deadline passed before shipping for ${orderId} was arranged`, + ), + ), + ) + : (logger.info(`arranged shipping for order ${orderId}`), OkAsync()), +``` + +Failing as a **defect** is deliberate: the platform retries that attempt on +another worker, which is what "we ran out of time here" means. A modeled +contract error — `ShippingUnavailable` — is a permanent no and would be wrong. + +Temporal's own `Context.current().cancellationSignal` is a **different clock**: +it fires on a workflow-side cancellation, and on worker shutdown after +`shutdownGraceTime`. Honour both; neither stands in for the other. + ## See also - [`@btravstack/temporal`](/reference/temporal) — options, ports, `TemporalInfo`, `WorkflowSource`. diff --git a/docs/how-to/tune-the-drain-for-kubernetes.md b/docs/how-to/tune-the-drain-for-kubernetes.md index 288d61a..5799917 100644 --- a/docs/how-to/tune-the-drain-for-kubernetes.md +++ b/docs/how-to/tune-the-drain-for-kubernetes.md @@ -139,6 +139,13 @@ An orchestrator reading `2` learns the pod stopped, but not cleanly. The second signal is the operator's escape hatch (and double Ctrl-C in development). Skipping the drain is a decision not to _wait_ for in-flight work, not to leave it running: every open unit is aborted before `stopping`. + +Aborted work only stops if something reads the abort. The unit's `AbortSignal` +reaches the work callback as an argument **and** rides the ambient record as +`currentUnit()?.signal` — the same object — which is what lets a +middleware-shaped runtime honour the deadline: a Temporal activity or an AMQP +handler has no parameter to receive one through. See +[Read the ambient unit from an adapter](/how-to/read-the-ambient-unit). `stop()` is for an embedder that wants out now; `requestDrain()` is the programmatic SIGTERM. diff --git a/docs/how-to/write-a-runtime.md b/docs/how-to/write-a-runtime.md index 92b8b50..fec5eab 100644 --- a/docs/how-to/write-a-runtime.md +++ b/docs/how-to/write-a-runtime.md @@ -184,6 +184,17 @@ const serveOne = ( }); ``` +::: tip If your work callback is someone else's `next()` +A middleware-shaped runtime opens the unit around a call it does not own the +arguments of, so the `signal` parameter above has nowhere to go — the work +callback _is_ the library's `next()`. The same signal is on the ambient record +as `currentUnit()?.signal`, which is how `@btravstack/temporal` and +`@btravstack/amqp` let an activity or a handler honour the deadline without an +injected context the transport's contract does not type. Pass it as a +parameter when you can, as `@btravstack/http` does; read it off the record +when you cannot. +::: + **2. `UnitMeta.id` is unique per unit, or you supply a `traceId`.** `traceId` defaults to `id`, so passing a _category_ — a route template like `"POST /orders"` — gives every request the same trace id and silently defeats diff --git a/docs/reference/amqp.md b/docs/reference/amqp.md index ebf5d12..b863ffd 100644 --- a/docs/reference/amqp.md +++ b/docs/reference/amqp.md @@ -156,7 +156,13 @@ unreachable broker at exit `1` rather than `70`. One unit per **delivery**, `kind: "delivery"`, opened by the starter's own `WorkerMiddleware`, which calls `next()` unchanged — it injects nothing, and -the handler's own `Result` is what the worker routes. +the handler's own `Result` is what the worker routes. The ambient +`currentUnit()` record is therefore the only route to the unit's +`AbortSignal` from inside a handler: `currentUnit()?.signal`, aborted at the +kernel's `drainTimeoutMs`. This transport has no cancellation of its own to +defer to — an un-acked delivery is redelivered, which is recovery, not +cancellation — so answering a `RetryableError` on an aborted signal is what +hands the message to the next worker. | `UnitMeta` field | Value | | ---------------- | ---------------------------------------------------------------------------------------------------------------- | diff --git a/docs/reference/core/runtime.md b/docs/reference/core/runtime.md index c47505c..8c2135f 100644 --- a/docs/reference/core/runtime.md +++ b/docs/reference/core/runtime.md @@ -60,7 +60,9 @@ type RunUnit = ( Submit one piece of work as a **unit**. The kernel counts it towards the drain, opens its ambient record, hands it an `AbortSignal` (fired at the drain -deadline, or at once when the drain is skipped) and gives the work's own +deadline, or at once when the drain is skipped — the same object is on the +record as `signal`, for a runtime whose work callback is a library's `next()`) +and gives the work's own `Result` **straight back** — mapping that outcome to a transport is the runtime's job. With a `unit` module, `ctx` is the forked context (`Context`), built before `work` runs and torn down after it @@ -144,6 +146,7 @@ type UnitRecord = { readonly traceId: string; readonly tenantId: string | undefined; readonly deadline: number | undefined; + readonly signal: AbortSignal; }; const currentUnit: () => UnitRecord | undefined; @@ -154,7 +157,7 @@ const currentUnit: () => UnitRecord | undefined; | `UnitMeta` | What a runtime says about one unit as it submits it. `kind` is the category (`"http"`, `"tick"`, `"job"`); `id` identifies **this** unit. `traceId` defaults to `id`. | | `UnitWork` | The work callback. The `Promise` arm exists to accept a caller's `async` handler — the one place the package accepts a bare `Promise` on purpose. Whatever `Result` it settles is what `run` hands back; a throw becomes a `Defect`. | | `UnitRegistry` | The kernel's own accounting, exposed as a type. `closed()` is monotonic; `awaitIdle()` answers about the registry at the instant it is called and is what beat 3 of the drain races. | -| `UnitRecord` | The ambient record, opened in an `AsyncLocalStorage` store for the unit's whole extent. `unitId` is minted per unit and always unique; `traceId` is the correlation id. | +| `UnitRecord` | The ambient record, opened in an `AsyncLocalStorage` store for the unit's whole extent. `unitId` is minted per unit and always unique; `traceId` is the correlation id; `signal` is the **same** `AbortSignal` `UnitWork` receives as its argument. | | `currentUnit()` | The ambient read; `undefined` outside a unit. Its legitimate readers are infrastructure adapters (a logger, an OTel exporter, a database adapter) — see [Read the ambient unit from an adapter](/how-to/read-the-ambient-unit). Not enforced by lint today. | ## `Clock` and `systemClock` diff --git a/docs/reference/glossary.md b/docs/reference/glossary.md index 46894f0..3101811 100644 --- a/docs/reference/glossary.md +++ b/docs/reference/glossary.md @@ -12,9 +12,12 @@ description: Short definitions of the terms used throughout the start documentat `AbortSignal` and counted in `DrainReport.abandoned` — the field the exit code keys on (`2`). See [ExitReport and DrainReport](/reference/core/exit-report). -**ambient record** — The small, fixed `UnitRecord` — `{ unitId, traceId, tenantId, deadline }` — +**ambient record** — The small, fixed `UnitRecord` — `{ unitId, traceId, tenantId, deadline, signal }` — the kernel opens in an `AsyncLocalStorage` store for a unit's whole extent, and -`currentUnit()` reads. It carries **data**, never services. See +`currentUnit()` reads. It carries **data**, never services; `signal` is the +very `AbortSignal` the unit's work callback is handed, so a middleware-shaped +runtime — a Temporal activity, an AMQP delivery — can still honour the drain +deadline. See [Ambient data, injected capabilities](/explanation/ambient-vs-context). **composition root** — The one module a process boots: it imports the application and a starter, @@ -108,5 +111,6 @@ a message property). Why `UnitMeta.id` must be unique per unit. See **unit / unit of work** — One piece of work a runtime submits through `host.run(meta, work)`: an HTTP request, an activity attempt, a delivery. The kernel counts it towards the -drain, hands it an `AbortSignal` and an ambient record, and hands its `Result` +drain, hands it an `AbortSignal` and an ambient record carrying that same +signal, and hands its `Result` straight back. See [The Runtime contract](/reference/core/runtime). diff --git a/docs/reference/temporal.md b/docs/reference/temporal.md index 06c1e7f..402e8cd 100644 --- a/docs/reference/temporal.md +++ b/docs/reference/temporal.md @@ -215,7 +215,11 @@ started polling it publishes `TemporalInfo`, `{ taskQueue, namespace }`, on One unit per activity **attempt**, `kind: "activity"`, opened by the starter's own `ActivityMiddleware`, which calls `next()` unchanged — it injects nothing, and the ambient `currentUnit()` record is what an adapter -reads the trace id from. +reads the trace id from, and the **only** route to the unit's `AbortSignal` +from inside an activity: `currentUnit()?.signal`, aborted at the kernel's +`drainTimeoutMs`. Temporal's `Context.current().cancellationSignal` is a +different clock — a workflow-side cancellation, and worker shutdown after +`shutdownGraceTime` — so the two are honoured together. | `UnitMeta` field | Value | | ---------------- | ----------------------------------------------------------------------------------------------------------- | diff --git a/examples/order-amqp-worker/src/handlers.ts b/examples/order-amqp-worker/src/handlers.ts index de4ecf7..d569efa 100644 --- a/examples/order-amqp-worker/src/handlers.ts +++ b/examples/order-amqp-worker/src/handlers.ts @@ -1,7 +1,9 @@ +import { RetryableError } from "@amqp-contract/worker"; import { AmqpHandlers } from "@btravstack/amqp"; +import { currentUnit } from "@btravstack/core"; import { orderContract } from "@btravstack/example-order-amqp-contract"; import { Logger } from "@btravstack/example-order-application"; -import { OkAsync } from "unthrown"; +import { ErrAsync, OkAsync } from "unthrown"; /** * The consuming half: the handlers record `orderContract` wants, one per @@ -22,11 +24,26 @@ import { OkAsync } from "unthrown"; * handler, one stream, and a reader that keeps its own copy of a subject * upserts on a payload and drops on a tombstone. There is no second message * type to declare, subscribe to, or keep ordered against this one. + * + * It also honours the kernel's deadline. `currentUnit()?.signal` is aborted + * when the drain runs out of time, and a delivery this process is no longer + * waiting for should not have a notification sent on its behalf: answering a + * `RetryableError` leaves the message un-acked, so the broker hands it to the + * next worker instead. The signal reaches a handler through the ambient + * record because the runtime is middleware-shaped — the kernel's work + * callback is the library's `next()`, and a handler has no parameter to + * receive one through (`@btravstack/http` passes it as an argument, which is + * the same signal by another route). */ export const orderHandlers = AmqpHandlers(orderContract)([Logger], { sync: (logger) => ({ orderChanged: (message) => { const { id, payload } = message.payload; + if (currentUnit()?.signal.aborted === true) { + return ErrAsync( + new RetryableError(`the drain deadline passed before order ${id} was notified`), + ); + } logger.info( payload === null ? `order ${id} is gone — notifying` diff --git a/examples/order-temporal-worker/src/fulfillment.ts b/examples/order-temporal-worker/src/fulfillment.ts index 0aec4da..88768ff 100644 --- a/examples/order-temporal-worker/src/fulfillment.ts +++ b/examples/order-temporal-worker/src/fulfillment.ts @@ -1,6 +1,7 @@ +import { currentUnit } from "@btravstack/core"; import { Module, Provider } from "@btravstack/di"; import { Logger, ShippingService, StockService } from "@btravstack/example-order-application"; -import { OkAsync } from "unthrown"; +import { OkAsync, fromSafePromise } from "unthrown"; /** * The two external services the saga orchestrates, as in-memory stand-ins. In @@ -12,6 +13,19 @@ import { OkAsync } from "unthrown"; * A module of its own so the swap is one import: the composition root takes * `FulfillmentModule`, a spec takes its own failing twin, and * `ApplicationModule` — which owns the ports — never knows the difference. + * + * `arrange` honours the kernel's deadline, and an adapter is where reading the + * ambient record is legitimate (thesis 2: it carries data about this unit, and + * a service is never in it). `currentUnit()?.signal` is aborted once the drain + * has run out of time, and an outbound call whose answer nobody in this + * process will read is not worth starting: the activity attempt fails as a + * **defect**, which the platform retries on another worker — the right shape + * for "we ran out of time", where the contract's `ShippingUnavailable` is a + * permanent no. The signal arrives on the record rather than as a parameter + * because the runtime is middleware-shaped: the kernel's work callback is + * Temporal's `next()`, and an activity has none to receive it through. + * Temporal's own `Context.current().cancellationSignal` is a different clock + * — it fires on `shutdownGraceTime` — so the two are honoured together. */ export const FulfillmentModule = Module("Fulfillment")({ provides: [ @@ -29,10 +43,16 @@ export const FulfillmentModule = Module("Fulfillment")({ }), Provider(ShippingService)([Logger], { sync: (logger) => ({ - arrange: (orderId) => { - logger.info(`arranged shipping for order ${orderId}`); - return OkAsync(); - }, + arrange: (orderId) => + currentUnit()?.signal.aborted === true + ? fromSafePromise( + Promise.reject( + new Error( + `the drain deadline passed before shipping for ${orderId} was arranged`, + ), + ), + ) + : (logger.info(`arranged shipping for order ${orderId}`), OkAsync()), }), }), ], diff --git a/packages/amqp/CLAUDE.md b/packages/amqp/CLAUDE.md index e5b0752..00d5b53 100644 --- a/packages/amqp/CLAUDE.md +++ b/packages/amqp/CLAUDE.md @@ -134,6 +134,18 @@ not a defect"` guards it). `create` never throws synchronously (its own leaves for the adapters that read it, and it is how the package's own suite observes the trace id (`seam` in `test-fixtures.ts` records `currentUnit()` inside the handler). +- **The kernel's per-unit `AbortSignal` rides that record too, and there is no + other route to it here.** `host.run` hands one to its work callback, and the + callback IS `next()` — a handler has no parameter to receive it through, and + injecting a context the contract does not type was the alternative and was + rejected. This transport also has no cancellation story of its own to fall + back on: an un-acked delivery is **redelivered**, which is recovery, not + cancellation. So a handler that must stop when the kernel stops waiting reads + `currentUnit()?.signal`, and what it answers is its own business — + `examples/order-amqp-worker`'s `orderChanged` returns a `RetryableError`, + leaving the delivery un-acked so the broker hands it to the next worker. + `amqp-runtime.spec.ts` → _"hands the handler the unit's own AbortSignal, + through the ambient record"_ pins it, off the `deadline` fixture's handler. - **A delivery tag is not a valid unit id.** Tags are per-**channel** and restart at `1` after a reconnect, which `amqp-connection-manager` performs silently underneath this worker — the one identifier that looks unique per @@ -177,7 +189,15 @@ null })` **raced against `signal`**, and `stop()` reuses whatever deadline **four** total attempts (first plus three retries), not the same count as Temporal's `maximumAttempts: 3`. - **The suite needs Docker** (`@amqp-contract/testing` boots one RabbitMQ per - run); its fixtures compose `AmqpModule("Consuming")({ contract: + run) and carries **8 specs** in `amqp-runtime.spec.ts`: one the published + info, one the unreachable broker, three the unit boundary (_"opens one + kernel unit per delivery"_, _"refuses a blank message id rather than tracing + every delivery to it"_, _"builds the handlers from the application's own + services"_) and three the drain (_"lets an in-flight delivery finish while + draining"_, _"hands the handler the unit's own AbortSignal, through the + ambient record"_ off the `deadline` fixture, _"releases the kernel at its own + deadline, not the library's own close timeout"_). + Its fixtures compose `AmqpModule("Consuming")({ contract: echoContract, handlers, url: amqpConnectionUrl, imports: [AppModule] })` with a provider per test from `AmqpHandlers(echoContract)` — from `Greeting`, or a value — so the module reads no environment, and hand it to diff --git a/packages/amqp/src/amqp-runtime.spec.ts b/packages/amqp/src/amqp-runtime.spec.ts index 941ece6..8a05611 100644 --- a/packages/amqp/src/amqp-runtime.spec.ts +++ b/packages/amqp/src/amqp-runtime.spec.ts @@ -120,6 +120,29 @@ describe("amqp", () => { ); }); + it("hands the handler the unit's own AbortSignal, through the ambient record", async ({ + serve, + deadline, + publishMessage, + }) => { + // GIVEN a delivery whose handler is waiting on `currentUnit()?.signal` — + // the only route to it here, since the middleware's work callback is + // `next()` and a handler has no parameter to receive one through + const app = await serve(deadline.handlers, { drainTimeoutMs: 100 }); + publishMessage({ exchange: "amqp-test", routingKey: "echo.requested" }, { value: "x" }); + await deadline.arrived; + + // WHEN the drain runs out of time for it + app.requestDrain(); + const report = await app.exited; + + // THEN the handler saw the abort, and the kernel reported the unit + // abandoned — one deadline, observable from inside the work + expect( + report.map((exit) => ({ abandoned: exit.drain?.abandoned, sawAbort: deadline.sawAbort() })), + ).toBeOkWith({ abandoned: 1, sawAbort: true }); + }); + it("releases the kernel at its own deadline, not the library's own close timeout", async ({ serve, gate, diff --git a/packages/amqp/src/message-units.ts b/packages/amqp/src/message-units.ts index 079b99a..69a37ee 100644 --- a/packages/amqp/src/message-units.ts +++ b/packages/amqp/src/message-units.ts @@ -9,6 +9,13 @@ import type { RuntimeHost, UnitMeta } from "@btravstack/core"; * what the unit leaves for the adapters that read it. `next()` unchanged is * the whole of the chain — the handler's own `Result` is what the worker * routes, and this package is transparent to it. + * + * **The kernel's per-unit `AbortSignal` rides that record too.** `host.run` + * hands one to its work callback, and this middleware's callback is `next()` + * — a handler has no parameter to receive it through, and this transport has + * no cancellation story of its own to fall back on (an un-acked delivery is + * redelivered, which is recovery, not cancellation). A handler that must stop + * when the kernel stops waiting reads `currentUnit()?.signal`. */ export const messageUnits = (host: RuntimeHost): WorkerMiddleware => diff --git a/packages/amqp/src/test-fixtures.ts b/packages/amqp/src/test-fixtures.ts index 82ecf24..e0d3f80 100644 --- a/packages/amqp/src/test-fixtures.ts +++ b/packages/amqp/src/test-fixtures.ts @@ -105,6 +105,60 @@ const seamOf = () => { }; }; +/** + * A handler that waits on the kernel's own per-unit signal — reached through + * `currentUnit()`, since a middleware-shaped runtime hands its work no + * parameter — and reports what it saw. `arrived` is the moment the delivery + * reached it, so a drain spec knows the unit is genuinely in flight. + */ +const deadlineHandler = () => { + let entered!: () => void; + const arrived = new Promise((resolve) => { + entered = resolve; + }); + let sawAbort: boolean | undefined; + + const handlers: EchoProvider = echoHandlers({ + value: { + echo: () => { + const signal = currentUnit()?.signal; + entered(); + return fromSafePromise( + new Promise((done) => { + // No record at all is the very regression this fixture exists to + // catch: settle at once so the spec fails on `sawAbort` rather + // than hanging until the suite's timeout, which would report a + // slow test instead of a missing signal. + if (signal === undefined) { + sawAbort = false; + done(); + return; + } + // An already-aborted signal never fires `abort` again — the same + // arm `whenAborted` carries in `amqp-runtime.ts`, and the reason + // a drain deadline of `0` would otherwise strand this delivery. + if (signal.aborted) { + sawAbort = true; + done(); + return; + } + signal.addEventListener( + "abort", + () => { + sawAbort = true; + done(); + }, + { once: true }, + ); + }), + ); + }, + }, + }); + + return { handlers, arrived, sawAbort: (): boolean | undefined => sawAbort }; +}; + /** * A handler that never finishes until `release()` is called, and whose * `arrived` promise reports the moment the delivery reached it. Both drain @@ -141,6 +195,8 @@ export type AmqpFixtures = { readonly serveBroken: () => Promise; readonly seam: ReturnType; readonly gate: ReturnType; + /** A handler that waits on the unit's own `AbortSignal`, read off the ambient record. */ + readonly deadline: ReturnType; }; // Annotated explicitly: TS2883 otherwise refuses to name the inferred type, @@ -177,4 +233,8 @@ export const it: TestAPI = amqpIt.extend { + await use(deadlineHandler()); + }, }); diff --git a/packages/core/CLAUDE.md b/packages/core/CLAUDE.md index 7dfd451..94732ae 100644 --- a/packages/core/CLAUDE.md +++ b/packages/core/CLAUDE.md @@ -42,7 +42,13 @@ that proves them, rather than duplicated). deadline"_. The abort comes from `registry.abortAll()`, not from the runtime honouring `Serving.drain(signal)` — `@btravstack/testing`'s `testRuntime` deliberately ignores that signal, which is what makes it a test of the - kernel. + kernel. **The same signal is on the ambient record**, so a runtime whose + work callback is a library's `next()` still reaches it: + `units.spec.ts` → _"carries the work's own AbortSignal on the ambient + record"_ asserts identity (`record.signal === the parameter`) and the abort + together, and `@btravstack/temporal`'s and `@btravstack/amqp`'s own + _"hands the activity/handler the unit's own AbortSignal, through the ambient + record"_ prove it end to end through a real transport. 5. **The application scope closes on every path.** `invariants.spec.ts` → _"5. the application scope closes on a startup failure"_; `start.spec.ts` → _"closes the application scope on a clean @@ -91,6 +97,13 @@ Beyond the nine: `units.spec.ts` → _"decrements even when the work throws"_. - **The ambient record does not leak between concurrent units.** `units.spec.ts` → _"does not leak between concurrent units"_. +- **The record's `signal` IS the work's own, not a copy.** One + `AbortController` per unit: `registry.run` hands `controller.signal` to + `work` and puts that same object on the `UnitRecord`, so `abortAll` — and + therefore `drainApp`'s deadline — is observable from both routes at once. A + second controller mirrored onto the record would drift on exactly the path + that matters. `units.spec.ts` → _"carries the work's own AbortSignal on the + ambient record"_. - **The phase tracker is monotonic.** `phase.spec.ts` → _"refuses to move backwards and reports nothing"_ and _"treats re-entering the same phase as a no-op"_. @@ -367,7 +380,13 @@ ConfigInvalid })` rather than widening `exited`'s error union for every sample and closes before the deadline. - **`abortAll` iterates the live `Set`,** so a unit started synchronously from an - abort listener is visited by the same pass. + abort listener is visited by the same pass. The `Set` holds the + `AbortController`s, and each one's `signal` is on both the work callback's + parameter list **and** the unit's ambient record, so one `abort()` is seen by + a runtime that takes the parameter (`@btravstack/http`) and by one that + cannot (`@btravstack/temporal`, `@btravstack/amqp`, whose work callback is + the library's `next()`). Do not mirror the record's `signal` onto a second + controller: the identity is what the guard asserts. - **`units.ts` uses `fromSafePromise`, not `fromPromise`.** The promise cannot reject — the work's own throw is caught by `flatMap`'s throw-to-defect net once diff --git a/packages/core/README.md b/packages/core/README.md index 9fe453b..53e91b8 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -127,7 +127,10 @@ is a module that exports no runtime port at all. - **A per-unit scope no handler manages** — pass a module as `StartOptions.unit` and the kernel forks it around every unit. - **An ambient record, not an ambient container** — `currentUnit()` reads - `{ unitId, traceId, tenantId, deadline }`; services never travel there. + `{ unitId, traceId, tenantId, deadline, signal }`; services never travel + there. `signal` is the very `AbortSignal` the unit's work callback is handed, + so a runtime whose work is a library's `next()` — a Temporal activity, an + AMQP delivery — can still honour the drain deadline. - **Nothing throws.** Every async surface is an [`unthrown`](https://github.com/btravstack/unthrown) `AsyncResult`; `runMain` sets `process.exitCode` — `0` clean, `1` a modeled startup error, diff --git a/packages/core/src/units.spec.ts b/packages/core/src/units.spec.ts index 7f5dbc9..ac3af53 100644 --- a/packages/core/src/units.spec.ts +++ b/packages/core/src/units.spec.ts @@ -1,4 +1,4 @@ -import { ErrAsync, Ok, OkAsync } from "unthrown"; +import { ErrAsync, Ok, OkAsync, type Result } from "unthrown"; import { describe, expect, it } from "vitest"; import { createUnitRegistry, currentUnit, runWithUnit } from "./units.js"; @@ -8,6 +8,7 @@ const record = { traceId: "t-1", tenantId: "acme", deadline: undefined, + signal: new AbortController().signal, } as const; describe("ambient unit record", () => { @@ -99,6 +100,31 @@ describe("createUnitRegistry", () => { expect(seen).toBeOkWith(expect.objectContaining({ tenantId: "acme" })); }); + it("carries the work's own AbortSignal on the ambient record", async () => { + // GIVEN a registry with one unit open, whose work reads the record + const registry = createUnitRegistry(); + let record: ReturnType; + let fromParameter: AbortSignal | undefined; + const running = registry.run(meta, (signal) => { + record = currentUnit(); + fromParameter = signal; + return new Promise>((settle) => { + signal.addEventListener("abort", () => settle(Ok()), { once: true }); + }); + }); + + // WHEN the drain deadline aborts every open unit + registry.abortAll(); + await running; + + // THEN the record carried the very signal the work was handed, aborted — + // which is what a middleware-shaped runtime has instead of a parameter + expect({ + same: record?.signal === fromParameter, + aborted: record?.signal.aborted, + }).toEqual({ same: true, aborted: true }); + }); + it("nests correctly through the registry", async () => { const registry = createUnitRegistry(); diff --git a/packages/core/src/units.ts b/packages/core/src/units.ts index 26b9048..780b9f9 100644 --- a/packages/core/src/units.ts +++ b/packages/core/src/units.ts @@ -2,11 +2,27 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { OkAsync, fromSafePromise, type AsyncResult, type Result } from "unthrown"; +/** + * What the kernel opens per unit and `currentUnit()` reads: a small, fixed + * record of **data about this unit**, and never a service. See the root + * `CLAUDE.md`'s thesis 2 for the line that draws. + * + * `signal` is the same `AbortSignal` the work callback receives — aborted at + * the drain deadline, or at once on a path that skips the drain. It is here + * because the callback is not always where the work is: a middleware-shaped + * runtime (`@btravstack/temporal`, `@btravstack/amqp`) opens the unit around + * a call it does not own the arguments of, so an activity or a handler has no + * parameter to receive it through. It is data, not a capability: there is + * nothing to substitute in a test, and a deadline nobody can observe is not a + * deadline. A transport's own cancellation — Temporal's + * `Context.current().cancellationSignal` — is a different clock, not this one. + */ export type UnitRecord = { readonly unitId: string; readonly traceId: string; readonly tenantId: string | undefined; readonly deadline: number | undefined; + readonly signal: AbortSignal; }; const storage = new AsyncLocalStorage(); @@ -90,6 +106,10 @@ export const createUnitRegistry = (): UnitRegistry => { traceId: meta.traceId ?? meta.id, tenantId: meta.tenantId, deadline: meta.deadline, + // The very signal `work` is handed below: one abort, two ways to + // reach it, so a runtime whose work is a callback it does not own the + // arguments of (a middleware) is not left without the deadline. + signal: controller.signal, }; // `fromSafePromise` is correct rather than `fromPromise`: the promise diff --git a/packages/temporal/CLAUDE.md b/packages/temporal/CLAUDE.md index b8ceea2..ba6effb 100644 --- a/packages/temporal/CLAUDE.md +++ b/packages/temporal/CLAUDE.md @@ -114,6 +114,20 @@ TemporalConfig, TemporalActivitiesPort as ActivitiesPortOf], { sync })` — is `temporal-contract`'s own `ActivityMiddleware`, imported: with the library a peer there is no structural copy, no cast and no `oxlint-disable` left in that file. +- **The kernel's per-unit `AbortSignal` reaches an activity through that + record, and only through it.** `host.run` hands one to its work callback, + and here the callback IS `next()` — an activity has no parameter to receive + it through, and giving it one would mean injecting a context the contract + does not type, which was the alternative and was rejected. So an activity + that must stop when the **kernel** stops waiting reads + `currentUnit()?.signal`, aborted at `drainTimeoutMs`. Temporal's own + `Context.current().cancellationSignal` is a **different clock** — a + workflow-side cancellation, and worker shutdown after `shutdownGraceTime` — + so the two are honoured together rather than one standing in for the other. + `examples/order-temporal-worker`'s `ShippingService.arrange` is the worked + answer: it fails as a **defect** on an aborted signal, which the platform + retries on another worker, where the contract's `ShippingUnavailable` is a + permanent no. - **`@temporal-contract/worker` and `@temporal-contract/contract` are peers** (and devDependencies, for the suite). A starter has real dependencies — it calls `declareActivitiesHandler` and types `contract` as a @@ -131,7 +145,7 @@ TemporalConfig, TemporalActivitiesPort as ActivitiesPortOf], { sync })` — - **Not included, deliberately**: `Result` → activity failure, which `declareActivitiesHandler` already owns. Doing it twice is what the removal of the raw-worker path was about. -- **`temporal-runtime.spec.ts` carries 12 specs.** Four are the starter's +- **`temporal-runtime.spec.ts` carries 13 specs.** Four are the starter's configuration (_"binds TEMPORAL_ADDRESS and TEMPORAL_NAMESPACE from the environment when nothing is pinned"_, _"pins what it is given and reads the rest from the environment"_, _"reads nothing from the environment when both @@ -145,7 +159,12 @@ TemporalConfig, TemporalActivitiesPort as ActivitiesPortOf], { sync })` — `currentUnit()?.traceId` from inside the attempt — the meta itself is no longer observable from outside the starter, and once a `traceId` is supplied the kernel never reads `meta.id` again; _"builds the activities from the - graph, closing over the services their provider declared"_), two the drain. + graph, closing over the services their provider declared"_), three the drain + (_"lets an in-flight activity finish while draining"_; _"hands the activity + the unit's own AbortSignal, through the ambient record"_, the `deadline` + fixture's activity waiting on `currentUnit()?.signal` and reporting + `sawAbort` alongside the report's `abandoned: 1`; _"releases the kernel at + its own deadline, not Temporal's"_). All boot through the `env` fixture (one `TestWorkflowEnvironment` per test) and `test-fixtures.ts`'s `compose`: `TemporalModule("Worker")({ contract: { ...echoContract, taskQueue }, activities: , diff --git a/packages/temporal/src/activity-units.ts b/packages/temporal/src/activity-units.ts index c52bffb..124da54 100644 --- a/packages/temporal/src/activity-units.ts +++ b/packages/temporal/src/activity-units.ts @@ -8,6 +8,17 @@ import { activityInfo } from "@temporalio/activity"; * declared, and the ambient `currentUnit()` record is there for an adapter that * wants the trace id. * + * **That includes the kernel's per-unit `AbortSignal`.** `host.run` hands one + * to its work callback, and this middleware's callback is `next()` — an + * activity has no parameter to receive it through, and giving it one would + * mean injecting a context the contract does not type. So it travels on the + * ambient record instead: `currentUnit()?.signal`, aborted at the kernel's + * `drainTimeoutMs`. Temporal's own `Context.current().cancellationSignal` is + * a **different clock** — it fires on a workflow-side cancellation and on + * worker shutdown after `shutdownGraceTime` — so an activity that must stop + * when the *kernel* stops waiting reads this one, and the two are honoured + * together rather than one standing in for the other. + * * There is deliberately no `Result`-unwrapping boundary: `declareActivitiesHandler` * owns the mapping from a settled `Result` to an activity failure, and the * kernel maps nothing to a transport. diff --git a/packages/temporal/src/temporal-runtime.spec.ts b/packages/temporal/src/temporal-runtime.spec.ts index 19c989c..5a8bc08 100644 --- a/packages/temporal/src/temporal-runtime.spec.ts +++ b/packages/temporal/src/temporal-runtime.spec.ts @@ -202,6 +202,37 @@ describe("temporal", () => { ); }); + it("hands the activity the unit's own AbortSignal, through the ambient record", async ({ + serve, + deadline, + }) => { + // GIVEN an activity waiting on `currentUnit()?.signal` — the only route to + // it here, since the middleware's work callback is `next()` and an + // activity has no parameter to receive one through. Temporal's own + // `Context.current().cancellationSignal` is a different clock: it fires on + // `shutdownGraceTime`, which this test never reaches. + const { app, client, taskQueue } = await serve({ + activities: deadline.activities, + drainTimeoutMs: 100, + }); + await client.workflow.start("runEcho", { + taskQueue, + workflowId: "wf-deadline-1", + args: ["x"], + }); + await deadline.arrived; + + // WHEN the drain runs out of time for it + app.requestDrain(); + const report = await app.exited; + + // THEN the activity saw the kernel's abort, and the unit is reported + // abandoned — one deadline, observable from inside the work + expect( + report.map((exit) => ({ abandoned: exit.drain?.abandoned, sawAbort: deadline.sawAbort() })), + ).toBeOkWith({ abandoned: 1, sawAbort: true }); + }); + it("releases the kernel at its own deadline, not Temporal's", async ({ serve, gate }) => { // GIVEN an activity that never finishes, and a drain with no time to give it const { app, client, taskQueue } = await serve({ diff --git a/packages/temporal/src/test-fixtures.ts b/packages/temporal/src/test-fixtures.ts index 7e77907..6a399c9 100644 --- a/packages/temporal/src/test-fixtures.ts +++ b/packages/temporal/src/test-fixtures.ts @@ -99,6 +99,66 @@ const contractSeamOf = () => { }; }; +/** + * An activity that waits on the kernel's own per-unit signal — reached through + * `currentUnit()`, since this runtime's work callback is Temporal's `next()` + * and an activity has no parameter to receive one through — and reports what + * it saw. `arrived` is the moment the attempt reached it, so a drain spec + * knows the unit is genuinely in flight. + */ +const deadlineOf = () => { + let entered!: () => void; + const arrived = new Promise((resolve) => { + entered = resolve; + }); + let sawAbort: boolean | undefined; + + return { + activities: EchoActivities({ + value: { + runEcho: { + echo: (value) => { + const signal = currentUnit()?.signal; + entered(); + return fromSafePromise( + new Promise((done) => { + // No record at all is the very regression this fixture exists + // to catch: settle at once so the spec fails on `sawAbort` + // rather than hanging until the suite's timeout, which would + // report a slow test instead of a missing signal. + if (signal === undefined) { + sawAbort = false; + done(value); + return; + } + // An already-aborted signal never fires `abort` again — the + // same arm `whenAborted` carries in `temporal-runtime.ts`, + // and the reason a drain deadline of `0` would otherwise + // strand this activity. + if (signal.aborted) { + sawAbort = true; + done(value); + return; + } + signal.addEventListener( + "abort", + () => { + sawAbort = true; + done(value); + }, + { once: true }, + ); + }), + ); + }, + }, + }, + }), + arrived, + sawAbort: (): boolean | undefined => sawAbort, + }; +}; + /** * The same wiring, but the activity resolves only once `release()` is called * and reports its arrival through `arrived`. The drain specs turn on knowing a @@ -187,6 +247,8 @@ export type TemporalFixtures = { readonly boot: Boot; readonly contractSeam: ReturnType; readonly gate: ReturnType; + /** An activity that waits on the unit's own `AbortSignal`, read off the ambient record. */ + readonly deadline: ReturnType; readonly configured: ReturnType; }; @@ -267,6 +329,10 @@ export const it = test.extend({ gate.release(); }, // oxlint-disable-next-line no-empty-pattern -- see above + deadline: async ({}, use) => { + await use(deadlineOf()); + }, + // oxlint-disable-next-line no-empty-pattern -- see above configured: async ({}, use) => { await use(configuredOf()); },