From 23b32c1d1c70ed5ddfc37913ef5f5540b52edfdd Mon Sep 17 00:00:00 2001 From: Mohammad Wahbeh Date: Sat, 29 Aug 2026 01:13:35 +0300 Subject: [PATCH] =?UTF-8?q?feat(rx):=20phase=208b=20=E2=80=94=20the=20RxJS?= =?UTF-8?q?=20async-runtime=20bridge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ships `@dexpace/rx`, exposing Phase 6b's `SseStream`/`typedSseStream` and Phase 6c's `Paginator` as RxJS `Observable`s, per docs/product-spec/18-asynchronous-runtime-adapter-contract.md (ASYNC-1..ASYNC-22), docs/product-spec/13-server-sent-events-and-streaming.md (SSE-41), and docs/superpowers/specs/2026-07-28-phase8b-async-runtime-design.md. New `packages/rx/src/`, the workspace's fifth package: - `sse.ts` — `sseEvents$`, `typedSse$` (SSE-41, SSE-33..SSE-36, ASYNC-21). Single-subscription, and deliberately so: `SseStream` wraps an already-open, single-use response body (BODY-14) and is itself single-pass (SSE-26), so a second `subscribe()` reaches the facade's own guard and surfaces `SseStreamError` through the error channel. The restriction is inherited and documented, not reimplemented — there is no honest way to make re-subscription mean anything without a second HTTP call this package does not make. - `pagination.ts` — `pageItems$`, `pages$` (PAGE-8). Cold and repeatable, the opposite asymmetry: `.items()`/`.pages()` build a fresh generator per call, and the wrapper defers that call to `[Symbol.asyncIterator]()` so each subscription drives an independent fetch sequence rather than reusing the first walk's exhausted generator. - `from-async-iterable.ts` — `fromAsyncIterable`, `@internal` (ASYNC-6, ASYNC-13, ASYNC-21). See below; this file is a deviation from the plan, not a component it called for. - `index.ts` — the four wrappers, nothing else. **The bridge is hand-written, against this phase's own instruction.** The design (§1) and plan (Global Constraints) both said: do not write an `AsyncIterable`→`Observable` pull loop, use RxJS's `from()`, and *prove* it satisfies each clause instead of assuming it. The proof failed on one. `rxjs@7.8.2`'s async-iterable path (`internal/observable/innerFrom.js`) is a bare `for await` that tests `subscriber.closed` only *after* a pull resolves, so unsubscribing while a pull is suspended reaches the source only if and when the server sends again. For pagination that is invisible; for SSE it is the common case — an idle event stream is permanently suspended, so `unsubscribe()` would leave the response body unreleased and the connection open indefinitely. That is ASYNC-6's bidirectional-cancellation clause unsatisfied, and SSE-30's release obligation with it. `fromAsyncIterable` is the same loop plus a teardown that releases the caller-supplied source and drives `iterator.return()` — release first, because closing the source is what settles the suspended pull an async generator's queued `return()` would otherwise sit behind. Scope is exactly the failing clause: no scheduler, no error re-wrapping, no retry, no buffering. The conformance suite's last case asserts the *defect* in RxJS's own `from()` alongside this module's fix, so it fails the day RxJS closes the gap and the module should be deleted. No RxJS scheduler appears in this package's production code, which is what keeps ASYNC-8..ASYNC-11 free: every pull runs inside the continuation chain that called `subscribe()`, exactly what `AsyncLocalStorage` tracks. A caller who adds `observeOn`/`subscribeOn` downstream reintroduces the boundary 7b's snapshot helper exists for, and the TSDoc says so rather than leaving them to find out from a missing trace ID. `rxjs` is a **required** peer (`optional: false`), unlike `pino`/`debug` in the logging adapters. Those peer on a library they never import — they are structural over `PinoLike`/`DebugLike` and work with anything of that shape. This package imports `Observable` unconditionally at module load, so an optional marking would suppress the one warning that catches a missing install and hand the consumer `ERR_MODULE_NOT_FOUND` instead. Peer rather than dependency for the usual reason: a duplicate copy breaks `Observable`/`Subscription` identity the way two classloaders break `instanceof`, the same hazard core's own peer-dedup guard exists to prevent. Tests: 25 across three colocated files, plus 8 Node-runtime cases in `test/node-conformance/rx-bridge.test.mjs`. The Node layer is not precautionary here — whether the release lands depends on Node's `ReadableStream.cancel()` settling a suspended read and on Node's async-generator `return()` queueing behind an in-flight `next()`, both independent implementations of Bun's. The conformance suite covers all four cancellation shapes (unsubscribe from inside `next()`, while a pull is suspended, before the first emission, and with a rejected release); ASYNC-21's poll-once-per-demand case uses a source that outlives demand — ten available, two taken, two pulled — since a generator yielding exactly what the subscriber consumes cannot tell one-pull-per-emission from a bridge that prefetches. Gate wiring: `typecheck`, `build`, `api`, and `lint:publish` extended to the new package; `verify-dual-consumption.mjs` drives a real `SseStream` through `sseEvents$` under plain `node`; `verify-consumer-types.mjs` compiles all four signatures against the built `.d.ts`. `verify:seam-1` and `verify:runtime-floor` pick the package up on their own. Open items registered in docs/open-items.md §M rather than left silent: M1 records the hand-written bridge and names its removal trigger; M2 flags that the checklist's eight 🚫 `ASYNC-*` rows collapse onto `TRANSPORT-*` requirements **no shipped package implements yet**, so an appendix-B sweep run between 8b and 8a does not count them as covered; M3 confirms at implementation time what the design predicted — ASYNC-18's scheduled-delay primitive is a full-port collapse, not an 8b scope boundary, since the as-built package contains no timer, scheduler, or backoff at all. --- .changeset/2026-08-28-async-runtime-bridge.md | 17 + bun.lock | 28 +- docs/open-items.md | 46 +++ ...6-07-28-phase8b-async-runtime-checklist.md | 74 ++++ .../plans/2026-07-28-phase8b-async-runtime.md | 46 +++ package.json | 10 +- packages/rx/README.md | 52 +++ packages/rx/api-extractor.json | 22 ++ packages/rx/etc/rx.api.md | 26 ++ packages/rx/package.json | 51 +++ .../from-async-iterable.conformance.test.ts | 331 ++++++++++++++++++ packages/rx/src/from-async-iterable.ts | 110 ++++++ packages/rx/src/index.ts | 11 + packages/rx/src/pagination.test.ts | 278 +++++++++++++++ packages/rx/src/pagination.ts | 39 +++ packages/rx/src/sse.test.ts | 205 +++++++++++ packages/rx/src/sse.ts | 60 ++++ packages/rx/tsconfig.build.json | 11 + packages/rx/tsconfig.json | 16 + scripts/verify-consumer-types.mjs | 24 ++ scripts/verify-dual-consumption.mjs | 38 +- test/node-conformance/README.md | 6 +- test/node-conformance/rx-bridge.test.mjs | 260 ++++++++++++++ 23 files changed, 1752 insertions(+), 9 deletions(-) create mode 100644 .changeset/2026-08-28-async-runtime-bridge.md create mode 100644 docs/superpowers/plans/2026-07-28-phase8b-async-runtime-checklist.md create mode 100644 packages/rx/README.md create mode 100644 packages/rx/api-extractor.json create mode 100644 packages/rx/etc/rx.api.md create mode 100644 packages/rx/package.json create mode 100644 packages/rx/src/from-async-iterable.conformance.test.ts create mode 100644 packages/rx/src/from-async-iterable.ts create mode 100644 packages/rx/src/index.ts create mode 100644 packages/rx/src/pagination.test.ts create mode 100644 packages/rx/src/pagination.ts create mode 100644 packages/rx/src/sse.test.ts create mode 100644 packages/rx/src/sse.ts create mode 100644 packages/rx/tsconfig.build.json create mode 100644 packages/rx/tsconfig.json create mode 100644 test/node-conformance/rx-bridge.test.mjs diff --git a/.changeset/2026-08-28-async-runtime-bridge.md b/.changeset/2026-08-28-async-runtime-bridge.md new file mode 100644 index 0000000..940fa46 --- /dev/null +++ b/.changeset/2026-08-28-async-runtime-bridge.md @@ -0,0 +1,17 @@ +--- +"@dexpace/rx": minor +--- + +Add `@dexpace/rx`, the RxJS async-runtime bridge (Phase 8b, `SSE-41` / the non-collapsed `ASYNC-*` subset): + +- `sseEvents$(stream)` and `typedSse$(stream, mapper)` — single-subscription `Observable` views of Phase 6b's + `SseStream` and `typedSseStream`. A second `subscribe()` surfaces `SseStream`'s own `SSE-26` guard through the + error channel rather than inventing a new restriction. +- `pageItems$(paginator)` and `pages$(paginator)` — cold, repeatable `Observable` views of Phase 6c's + `Paginator`, one independent fetch sequence per subscription (`PAGE-8`). +- Unsubscribing reaches the source even while a pull is suspended (`ASYNC-6`), so an idle SSE stream releases its + response body immediately instead of at the server's next event. This is the one clause RxJS's own + `from(asyncIterable)` does not satisfy, so the package ships a small internal bridge in its place; the + conformance suite pins both behaviors. +- Source errors reach the error channel unwrapped (`ASYNC-13`); no new error class. +- `rxjs` and `@dexpace/core` are peer dependencies, and the package has zero runtime dependencies (`SEAM-1`). diff --git a/bun.lock b/bun.lock index cdf64ef..9cb038d 100644 --- a/bun.lock +++ b/bun.lock @@ -12,6 +12,7 @@ "@dexpace/core": "workspace:*", "@dexpace/logging-debug": "workspace:*", "@dexpace/logging-pino": "workspace:*", + "@dexpace/rx": "workspace:*", "@dexpace/transport-conformance": "workspace:*", "@dexpace/transport-fetch": "workspace:*", "@dexpace/transport-shared": "workspace:*", @@ -25,6 +26,7 @@ "gts": "^7", "mitata": "^1", "publint": "^0.3", + "rxjs": "^7.8.0", "typescript": "catalog:", "typescript-eslint": "^8", }, @@ -100,6 +102,22 @@ "pino", ], }, + "packages/rx": { + "name": "@dexpace/rx", + "version": "0.0.0", + "devDependencies": { + "@dexpace/core": "workspace:*", + "@microsoft/api-extractor": "catalog:", + "expect-type": "catalog:", + "fast-check": "catalog:", + "rxjs": "^7.8.0", + "typescript": "catalog:", + }, + "peerDependencies": { + "@dexpace/core": "workspace:*", + "rxjs": "^7.8.0", + }, + }, "packages/transport-conformance": { "name": "@dexpace/transport-conformance", "version": "0.0.0", @@ -232,6 +250,8 @@ "@dexpace/logging-pino": ["@dexpace/logging-pino@workspace:packages/logging-pino"], + "@dexpace/rx": ["@dexpace/rx@workspace:packages/rx"], + "@dexpace/transport-conformance": ["@dexpace/transport-conformance@workspace:packages/transport-conformance"], "@dexpace/transport-fetch": ["@dexpace/transport-fetch@workspace:packages/transport-fetch"], @@ -750,7 +770,7 @@ "run-parallel": ["run-parallel@1.2.0", "", { "dependencies": { "queue-microtask": "^1.2.2" } }, "sha512-5l4VyZR86LZ/lDxZTR6jqL8AFE2S0IFLMP26AbjsLVADxHdhB/c0GUsH+y39UfCi3dzz8OlQuPmnaJOMoDHQBA=="], - "rxjs": ["rxjs@6.6.7", "", { "dependencies": { "tslib": "^1.9.0" } }, "sha512-hTdwr+7yYNIT5n4AMYp85KA6yw2Va0FLa3Rguvbpa4W3I5xynaBZo41cM3XM+4Q6fRMj3sBYIR1VAmZMXYJvRQ=="], + "rxjs": ["rxjs@7.8.2", "", { "dependencies": { "tslib": "^2.1.0" } }, "sha512-dhKf903U/PQZY6boNNtAGdWbG85WAbjT/1xYoZIC7FAY0yWapOBQVsVrDl58W86//e1VpMNBtRV4MaXfdMySFA=="], "sade": ["sade@1.8.1", "", { "dependencies": { "mri": "^1.1.0" } }, "sha512-xal3CZX1Xlo/k4ApwCFrHVACi9fBqJ7V+mwhBsuf/1IOKbBy098Fex+Wa/5QMubw09pSZ/u8EY8PWgevJsXp1A=="], @@ -828,7 +848,7 @@ "ts-declaration-location": ["ts-declaration-location@1.0.7", "", { "dependencies": { "picomatch": "^4.0.2" }, "peerDependencies": { "typescript": ">=4.0.0" } }, "sha512-EDyGAwH1gO0Ausm9gV6T2nUvBgXT5kGoCMJPllOaooZ+4VvJiKBdZE7wK18N1deEowhcUptS+5GXZK8U/fvpwA=="], - "tslib": ["tslib@1.14.1", "", {}, "sha512-Xni35NKzjgMrwevysHTCArtLDpPvye8zV/0E4EyYn43P7/7qvQwPh9BGkHewbMulVntbigmcT7rdX3BNo9wRJg=="], + "tslib": ["tslib@2.8.1", "", {}, "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w=="], "type-check": ["type-check@0.4.0", "", { "dependencies": { "prelude-ls": "^1.2.1" } }, "sha512-XleUoc9uwGXqjWwXaUTZAmzMcFZ5858QA2vvx1Ur5xIcixXIP+8LnFDgRplU30us6teqdlskFfu+ae4K79Ooew=="], @@ -936,6 +956,8 @@ "inquirer/ansi-escapes": ["ansi-escapes@4.3.2", "", { "dependencies": { "type-fest": "^0.21.3" } }, "sha512-gKXj5ALrKWQLsYG9jlTRmR/xKluxHV+Z9QEwNIgCfM1/uwPMCuzVVnh5mwTd+OuBZcwSIMbqssNWRm1lE51QaQ=="], + "inquirer/rxjs": ["rxjs@6.6.7", "", { "dependencies": { "tslib": "^1.9.0" } }, "sha512-hTdwr+7yYNIT5n4AMYp85KA6yw2Va0FLa3Rguvbpa4W3I5xynaBZo41cM3XM+4Q6fRMj3sBYIR1VAmZMXYJvRQ=="], + "js-yaml/argparse": ["argparse@2.0.1", "", {}, "sha512-8+9WqebbFzpX9OR+Wa6O29asIogeRMzcGtAINdpMHHyAg10f05aSFVBbcEqGf/PXw1EjAZ+q2/bEBg3DvurK3Q=="], "marked-terminal/chalk": ["chalk@5.6.2", "", {}, "sha512-7NzBL0rN6fMUW+f7A6Io4h40qQlG+xGmtMxfbnH/K7TAtt8JQWVQK+6g0UXKMeVJoyV5EkkNsErQ8pVD3bLHbA=="], @@ -982,6 +1004,8 @@ "inquirer/ansi-escapes/type-fest": ["type-fest@0.21.3", "", {}, "sha512-t0rzBq87m3fVcduHDUFhKmyyX+9eo6WQjZvf51Ea/M0Q7+T374Jp1aUiyUl0GKxp8M/OETVHSDvmkyPgvX+X2w=="], + "inquirer/rxjs/tslib": ["tslib@1.14.1", "", {}, "sha512-Xni35NKzjgMrwevysHTCArtLDpPvye8zV/0E4EyYn43P7/7qvQwPh9BGkHewbMulVntbigmcT7rdX3BNo9wRJg=="], + "read-pkg-up/find-up/locate-path": ["locate-path@5.0.0", "", { "dependencies": { "p-locate": "^4.1.0" } }, "sha512-t7hw9pI+WvuwNJXwk5zVHpyhIqzg2qTlklJOf0mVxGSbe3Fp2VieZcduNYjaLDoy6p9uGpQEGWG87WpMKlNq8g=="], "read-pkg/normalize-package-data/hosted-git-info": ["hosted-git-info@2.8.9", "", {}, "sha512-mxIDAb9Lsm6DoOJ7xH+5+X4y1LU/4Hi50L9C5sIswK3JzULS4bwk1FvjdBgvYR4bzT4tuUQiC15FE2f5HbLvYw=="], diff --git a/docs/open-items.md b/docs/open-items.md index 9cb281f..bcf9835 100644 --- a/docs/open-items.md +++ b/docs/open-items.md @@ -1413,6 +1413,52 @@ Recorded at implementation time. Verified against `docs/product-spec/15-instrume `PIPE-2` fixes the `LOGGING` pillar step inside `RETRY` and `REDIRECT` pipelines. Consequently, `startSpan('http.client.request')` and metric increments (`http.client.request.count`, `http.client.request.duration`) execute per HTTP transmission attempt/hop. The higher-level logical operation span and HTTP-tracer lifecycle are owned by Phase 8a / `OBS-29`. +## Section M — Phase 8b (Async-Runtime Bridge, `@dexpace/rx`) + +Recorded at implementation time. Verified against `docs/product-spec/18-asynchronous-runtime-adapter-contract.md` +(`ASYNC-1`..`ASYNC-22`) and `SSE-41`. + +### M1 — The `AsyncIterable`→`Observable` Bridge Is Hand-Written, Not `rxjs`'s `from()` — **RECORDED** (2026-08-28) + +8b's design (§1) and plan (Global Constraints) both instructed: do not hand-write the pull loop, use RxJS's own +`from(asyncIterable)`, and prove it satisfies `ASYNC-6`/`ASYNC-13`/`ASYNC-21` rather than assuming it. The proof +failed on one clause. `rxjs@7.8.2`'s async-iterable path tests `subscriber.closed` only *after* a pull resolves, +so unsubscribing while a pull is suspended never reaches the source. For pagination that is invisible; for SSE it +is the common case — an idle event stream is permanently suspended, so `unsubscribe()` would leave the response +body unreleased and the connection open until the server next sent something. + +Resolved through the fallback both documents pre-authorized, scoped to that clause alone: +`packages/rx/src/from-async-iterable.ts` (`@internal`) adds a teardown that releases the caller-supplied source +and drives `iterator.return()`, release first so a suspended pull settles before the queued generator return. +No scheduler, no error re-wrapping, no buffering. + +Full rationale in the plan's Self-Review. This is a deviation from the *plan's* implementation instruction, not +from the product spec — `ASYNC-6`/`ASYNC-21`/`SSE-41` are satisfied as written, and +`docs/sdk-design-nodejs/10-deliberate-deviations-from-the-reference-contract.md`'s Phase 8b rows are unaffected. + +**Trigger:** an RxJS release that closes the gap. `from-async-iterable.conformance.test.ts`'s last case asserts +the defect (`returns === 0` after an idle unsubscribe) and fails when it is fixed; at that point delete the +module and go back to `from()`. + +### M2 — `ASYNC-*` IDs Marked 🚫 Are Not Yet Satisfied Anywhere — **SCHEDULED** (Phase 8a) + +`ASYNC-1`, `-2`, `-5`, `-15`, `-16`, `-17`, `-20`, and `-22` collapse onto `TRANSPORT-*` twins (`TRANSPORT-23`, +`-21`, `-9`, `-15`/`-16`, `-29`, and the `SEAM-16` body-ownership invariant). The 8b checklist marks them 🚫 +"collapses onto Phase 8a," which is the correct disposition but reads, at a glance, like a closed row. **No +shipped package implements those `TRANSPORT-*` requirements yet** — 8a (`transport-fetch`/`transport-undici`) has +not executed. Recorded so an appendix-B sweep run between 8b and 8a does not count eight `MUST`s as covered. + +**Trigger:** Phase 8a landing. Its checklist owns the ✅ for each twin. + +### M3 — `ASYNC-18` Confirmed a Full-Port Collapse at Implementation Time — **RESOLVED** (2026-08-28) + +8b's design predicted that no adapter in this port needs a non-blocking scheduled-delay primitive, correcting the +segmentation design's narrower "8b-only scope boundary" framing. The as-built package confirms it: `@dexpace/rx` +contains no timer, no scheduler, and no backoff — the four wrappers only iterate what they are handed. SSE +reconnection stays caller-owned (`SSE-38`) and retry/backoff stays in 5a's engine. Already reflected in +`docs/sdk-design-nodejs/10-deliberate-deviations-from-the-reference-contract.md` Item 1; no further action. + + ## Maintaining this file Add an entry the moment a gap is found, not when it is fixed — the failure mode this file prevents is a diff --git a/docs/superpowers/plans/2026-07-28-phase8b-async-runtime-checklist.md b/docs/superpowers/plans/2026-07-28-phase8b-async-runtime-checklist.md new file mode 100644 index 0000000..121c48a --- /dev/null +++ b/docs/superpowers/plans/2026-07-28-phase8b-async-runtime-checklist.md @@ -0,0 +1,74 @@ +# Phase 8b — Async-Runtime Bridge — Checklist + +**Status: EXECUTED.** Every ✅ below names code and tests that exist on this branch, not a plan step. Verified +against `docs/product-spec/18-asynchronous-runtime-adapter-contract.md` (`ASYNC-1` through `ASYNC-22`), +`docs/product-spec/13-server-sent-events-and-streaming.md` (`SSE-41`), plus Task 1 through Task 4 deliverables, +package builds, and API reports. + +**One implementation deviation**, recorded in full in the plan's Self-Review: the `AsyncIterable`→`Observable` +bridge is `packages/rx/src/from-async-iterable.ts`, not RxJS's own `from()`, because `rxjs@7.8.2` does not reach +the source when a subscription is torn down while a pull is suspended — the `ASYNC-6` clause an idle SSE stream +depends on. Every row citing that module below is citing the reason it exists. The `🚫` rows that name Phase 8a +are **not satisfied yet**: they collapse onto `TRANSPORT-*` requirements no shipped package implements — see +`docs/open-items.md` §M2. + +**Legend:** ✅ Implemented and tested — 🚫 Not built (permanent simplification / collapse, named reason) — ⏳ Deferred +(named target phase) — N/A Not applicable in this port. + +## 18.1 Completion and failure delivery + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| ASYNC-1 | MUST | Single-value completion future delivers non-null Response on success, failure channel on no response | 🚫 | Collapses onto `TRANSPORT-23` (Phase 8a) — for `Transport`, `send()` returns `Promise` directly | +| ASYNC-2 | MUST | Construction-time failure via failure channel, not sync throw | 🚫 | Collapses onto `TRANSPORT-21` (Phase 8a) | + +## 18.2 Cancellation modes + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| ASYNC-3 | MUST | Cancel-with-interrupt vs without on worker thread | N/A | Node event loop has no worker-thread-pool transport model to interrupt (`SEAM-18` disposition) | +| ASYNC-4 | MUST | Ordered interrupt delivery preventing pooled-thread poisoning | N/A | Node event loop has no pooled worker threads to poison (`SEAM-18` disposition) | +| ASYNC-5 | MUST | Orphaned closeable result closed exactly once on race | 🚫 | Collapses onto `TRANSPORT-9` (Phase 8a) | +| ASYNC-6 | MUST | Bidirectional cancellation across adapter | ✅ | `packages/rx/src/from-async-iterable.ts`'s teardown (release the source, then `iterator.return()`). Asserted across all four paths — synchronous unsubscribe from inside `next()`, unsubscribe while a pull is suspended, unsubscribe before the first emission, and a rejected release — in `from-async-iterable.conformance.test.ts`, `sse.test.ts`, `pagination.test.ts`, and on real Node in `test/node-conformance/rx-bridge.test.mjs`. The same suite pins RxJS's native `from()` **failing** this clause | +| ASYNC-7 | SHOULD | Document interrupt-mode choice per adapter | N/A | Vacuous: no blocking worker thread calls to interrupt | + +## 18.3 Logging-context propagation + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| ASYNC-8 | SHOULD | Propagate logging context across thread/scheduler hops | ✅ | Node `AsyncLocalStorage` auto-propagation through promise chains/async iteration; the package installs no RxJS scheduler, which is what keeps the continuation chain intact. Stated in `sseEvents$`/`typedSse$` TSDoc, including the caller-introduced `observeOn`/`subscribeOn` boundary | +| ASYNC-9 | MUST | Save, install, restore logging context | ✅ | Node `AsyncLocalStorage` auto-propagation invariant | +| ASYNC-10 | MUST | Capture logging context at logical caller point | ✅ | Node `AsyncLocalStorage` captures per subscription at iteration pull time | +| ASYNC-11 | MUST | Safe when no logging context backend installed | ✅ | `AsyncLocalStorage` handles undefined store gracefully | +| ASYNC-12 | MUST | Explicit transfer at thread boundary where auto-inheritance absent | N/A | Single-threaded event loop; continuation-local storage auto-propagates | + +## 18.4 Error unwrapping and blocking bridge + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| ASYNC-13 | MUST | Unwrap async framework wrapper exceptions to original cause | ✅ | `packages/rx/src/from-async-iterable.ts` passes a thrown value straight to `subscriber.error`; asserted in `from-async-iterable.conformance.test.ts` (`RangeError` in, same `RangeError` out) and `sse.test.ts` (a throwing `SseMapper`) | +| ASYNC-14 | MUST | Async->sync blocking bridge honoring thread interruption | N/A | Inapplicable in Node — no blocking HTTP client bridge (`SEAM-18` disposition) | + +## 18.5 Lifecycle + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| ASYNC-15 | MUST | Close/dispose operation idempotent, ownership-aware, interrupt-safe | 🚫 | Owned by Phase 8a `TRANSPORT-15`/`16`; `@dexpace/rx` owns no background thread pools | +| ASYNC-16 | SHOULD | Graceful executor shutdown on close | 🚫 | Owned by Phase 8a | +| ASYNC-17 | SHOULD | No-op default close for lightweight/functional transports | 🚫 | Owned by Phase 8a (`transport-fetch`) | + +## 18.6 Delay, options, and streaming + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| ASYNC-18 | MUST | Non-blocking scheduled-delay primitive | N/A | Resolved N/A to this port: `@dexpace/rx` does no reconnection, retry, or backoff; SSE reconnection is caller-owned; pagination retry lives in pipeline layer | +| ASYNC-19 | MUST | Per-call request options threaded through overloads | N/A | Resolved N/A: `@dexpace/rx` wraps already-constructed `SseStream` / `Paginator` instances; does not initiate new HTTP calls | +| ASYNC-20 | MUST | Delivered Response body not closed on late future cancel | 🚫 | Restates `SEAM-16` / transport invariant owned by Phase 8a | +| ASYNC-21 | MUST | Reactive streaming adapter (SSE) honors backpressure, completes on end-of-source, propagates errors without swallowing, single-subscriber | ✅ | `packages/rx/src/sse.ts`: `sseEvents$`, `typedSse$`, over `from-async-iterable.ts`'s one-pull-per-emission loop. `from-async-iterable.conformance.test.ts` (poll-once-per-demand, complete-on-end, error passthrough), `sse.test.ts` (single-subscriber via `SSE-26`), `test/node-conformance/rx-bridge.test.mjs` | +| ASYNC-22 | MUST | Safe for concurrent calls | 🚫 | Collapses onto `TRANSPORT-29` (Phase 8a) | + +## 13.7 Server-Sent Events + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| SSE-41 | MAY | Reactive SSE adapter with fatal/non-fatal split and documented source ownership | ✅ | `packages/rx/src/sse.ts`: `sseEvents$`, `typedSse$`. Source ownership is documented on both functions and in `packages/rx/README.md` (the adapter closes the stream on unsubscribe; the caller owns reconnection). The fatal/non-fatal split collapses — JavaScript has no catchable-fatal tier, per the design doc's Deviation Ledger. Asserted in `sse.test.ts` and `from-async-iterable.conformance.test.ts` | diff --git a/docs/superpowers/plans/2026-07-28-phase8b-async-runtime.md b/docs/superpowers/plans/2026-07-28-phase8b-async-runtime.md index 9e65970..e8ee2ce 100644 --- a/docs/superpowers/plans/2026-07-28-phase8b-async-runtime.md +++ b/docs/superpowers/plans/2026-07-28-phase8b-async-runtime.md @@ -391,6 +391,52 @@ git commit -m "feat(rx): promote public barrel for @dexpace/rx" ## Self-Review +**Executed 2026-08-28.** All four tasks shipped; the full gate sequence is green. One deviation, recorded below +rather than left silent. + +### Deviation Ledger addition — the `AsyncIterable`→`Observable` bridge is hand-written + +**What the plan said.** Global Constraints: "Do not hand-write an `AsyncIterable`-to-`Observable` pull loop. Use +RxJS's own `from()`." Task 1 Step 3 named the escape hatch: if a conformance clause fails against the installed +RxJS, write the minimal wrapping `Observable` that closes *exactly* that clause and record it here. + +**What was found.** `rxjs@7.8.2`'s async-iterable path (`internal/observable/innerFrom.js`) is a bare `for await` +loop that tests `subscriber.closed` only *after* a pull resolves: + +```js +async function process(asyncIterable, subscriber) { + for await (const value of asyncIterable) { + subscriber.next(value); + if (subscriber.closed) return; + } + subscriber.complete(); +} +``` + +Unsubscribing while a pull is suspended therefore reaches the source only if and when the source produces +again. For pagination that is invisible (a page fetch always settles). For SSE it is the failure mode that +matters most: an idle event stream is *permanently* suspended on `next()`, so `subscription.unsubscribe()` +leaves the response body unreleased and the connection open until the server happens to send something. That is +`ASYNC-6`'s bidirectional-cancellation clause, unsatisfied — and `SSE-30`'s release obligation with it. + +**What was built.** `packages/rx/src/from-async-iterable.ts` (`@internal`, ~60 lines): the same pull loop, plus a +teardown that releases the caller-supplied source and drives `iterator.return()` on unsubscription. Release runs +*before* the iterator return, because closing the source is what settles the suspended pull that an async +generator's queued `return()` would otherwise sit behind. Scope is exactly the failing clause — no scheduler, no +error re-wrapping, no retry, no buffering. + +**How it stays honest.** `from-async-iterable.conformance.test.ts`'s last case asserts the *defect* in RxJS's own +`from()` (`returns` stays `0` after an idle unsubscribe) alongside this module's `1`. When a future RxJS closes +the gap that case fails, and the reviewer's instruction is in the file header: delete the module and go back to +`from()`. `test/node-conformance/rx-bridge.test.mjs` proves the same cancellation path on real Node, since +whether the release lands depends on Node's `ReadableStream.cancel()` and async-generator `return()` queueing. + +**Not a deviation from the product spec.** `ASYNC-6`/`ASYNC-21`/`SSE-41` are satisfied as written; the deviation +is from this plan's own implementation instruction, which named this outcome as an allowed one. Nothing is added +to `docs/sdk-design-nodejs/10-deliberate-deviations-from-the-reference-contract.md`, whose Phase 8b rows +(`ASYNC-21`'s fatal/non-fatal collapse, `ASYNC-18`'s full-port collapse) are unaffected. + + - [ ] Task 1's conformance suite passed against the installed RxJS version with no fallback needed — or, if a fallback was needed, it is scoped to exactly the failing clause and recorded in this section as a Deviation Ledger addition to the design doc. diff --git a/package.json b/package.json index 37e5201..add0c71 100644 --- a/package.json +++ b/package.json @@ -21,6 +21,7 @@ "@dexpace/core": "workspace:*", "@dexpace/logging-debug": "workspace:*", "@dexpace/logging-pino": "workspace:*", + "@dexpace/rx": "workspace:*", "@dexpace/transport-conformance": "workspace:*", "@dexpace/transport-fetch": "workspace:*", "@dexpace/transport-shared": "workspace:*", @@ -34,6 +35,7 @@ "gts": "^7", "mitata": "^1", "publint": "^0.3", + "rxjs": "^7.8.0", "typescript": "catalog:", "typescript-eslint": "^8" }, @@ -45,16 +47,16 @@ "fix": "bun run build:deps && gts fix .", "build:core": "tsc -b packages/core/tsconfig.build.json", "build:deps": "bun run build:core && tsc -p packages/transport-shared/tsconfig.build.json", - "typecheck": "bun run build:deps && tsc -p packages/core/tsconfig.json --noEmit && tsc -p packages/codec-json/tsconfig.json --noEmit && tsc -p packages/logging-pino/tsconfig.json --noEmit && tsc -p packages/logging-debug/tsconfig.json --noEmit && tsc -p packages/body-file/tsconfig.json --noEmit && tsc -p packages/transport-shared/tsconfig.json --noEmit && tsc -p packages/transport-conformance/tsconfig.json --noEmit && tsc -p packages/transport-fetch/tsconfig.json --noEmit && tsc -p packages/transport-undici/tsconfig.json --noEmit", + "typecheck": "bun run build:deps && tsc -p packages/core/tsconfig.json --noEmit && tsc -p packages/codec-json/tsconfig.json --noEmit && tsc -p packages/logging-pino/tsconfig.json --noEmit && tsc -p packages/logging-debug/tsconfig.json --noEmit && tsc -p packages/body-file/tsconfig.json --noEmit && tsc -p packages/transport-shared/tsconfig.json --noEmit && tsc -p packages/transport-conformance/tsconfig.json --noEmit && tsc -p packages/transport-fetch/tsconfig.json --noEmit && tsc -p packages/transport-undici/tsconfig.json --noEmit && tsc -p packages/rx/tsconfig.json --noEmit", "prebuild": "bun run --cwd packages/core prebuild", - "build": "bun run build:deps && tsc -p packages/codec-json/tsconfig.build.json && tsc -p packages/logging-pino/tsconfig.build.json && tsc -p packages/logging-debug/tsconfig.build.json && tsc -p packages/body-file/tsconfig.build.json && tsc -p packages/transport-fetch/tsconfig.build.json && tsc -p packages/transport-undici/tsconfig.build.json", + "build": "bun run build:deps && tsc -p packages/codec-json/tsconfig.build.json && tsc -p packages/logging-pino/tsconfig.build.json && tsc -p packages/logging-debug/tsconfig.build.json && tsc -p packages/body-file/tsconfig.build.json && tsc -p packages/transport-fetch/tsconfig.build.json && tsc -p packages/transport-undici/tsconfig.build.json && tsc -p packages/rx/tsconfig.build.json", "test": "bun test", "knowledge": "node scripts/knowledge.mjs", "test:scripts": "node --test 'scripts/*.test.mjs'", "test:node": "node --test test/node-conformance/*.test.mjs", "bench": "bun run packages/core/src/io/byte-queue.bench.ts", - "api": "cd packages/core && bun run api:ci && cd ../codec-json && bun run api:ci && cd ../logging-pino && bun run api:ci && cd ../logging-debug && bun run api:ci && cd ../body-file && bun run api:ci && cd ../transport-shared && bun run api:ci && cd ../transport-fetch && bun run api:ci && cd ../transport-undici && bun run api:ci", - "lint:publish": "publint packages/core && attw --pack packages/core --ignore-rules cjs-resolves-to-esm && publint packages/codec-json && attw --pack packages/codec-json --ignore-rules cjs-resolves-to-esm && publint packages/logging-pino && attw --pack packages/logging-pino --ignore-rules cjs-resolves-to-esm && publint packages/logging-debug && attw --pack packages/logging-debug --ignore-rules cjs-resolves-to-esm && publint packages/body-file && attw --pack packages/body-file --ignore-rules cjs-resolves-to-esm && publint packages/transport-shared && attw --pack packages/transport-shared --ignore-rules cjs-resolves-to-esm && publint packages/transport-fetch && attw --pack packages/transport-fetch --ignore-rules cjs-resolves-to-esm && publint packages/transport-undici && attw --pack packages/transport-undici --ignore-rules cjs-resolves-to-esm", + "api": "cd packages/core && bun run api:ci && cd ../codec-json && bun run api:ci && cd ../logging-pino && bun run api:ci && cd ../logging-debug && bun run api:ci && cd ../body-file && bun run api:ci && cd ../transport-shared && bun run api:ci && cd ../transport-fetch && bun run api:ci && cd ../transport-undici && bun run api:ci && cd ../rx && bun run api:ci", + "lint:publish": "publint packages/core && attw --pack packages/core --ignore-rules cjs-resolves-to-esm && publint packages/codec-json && attw --pack packages/codec-json --ignore-rules cjs-resolves-to-esm && publint packages/logging-pino && attw --pack packages/logging-pino --ignore-rules cjs-resolves-to-esm && publint packages/logging-debug && attw --pack packages/logging-debug --ignore-rules cjs-resolves-to-esm && publint packages/body-file && attw --pack packages/body-file --ignore-rules cjs-resolves-to-esm && publint packages/transport-shared && attw --pack packages/transport-shared --ignore-rules cjs-resolves-to-esm && publint packages/transport-fetch && attw --pack packages/transport-fetch --ignore-rules cjs-resolves-to-esm && publint packages/transport-undici && attw --pack packages/transport-undici --ignore-rules cjs-resolves-to-esm && publint packages/rx && attw --pack packages/rx --ignore-rules cjs-resolves-to-esm", "audit": "bun audit --audit-level=high --prod", "changeset": "node scripts/changeset.mjs", "verify:dual-consumption": "node scripts/verify-dual-consumption.mjs", diff --git a/packages/rx/README.md b/packages/rx/README.md new file mode 100644 index 0000000..e420bf9 --- /dev/null +++ b/packages/rx/README.md @@ -0,0 +1,52 @@ +# @dexpace/rx + +RxJS async-runtime bridge for the dexpace Node.js SDK. + +## Installation + +```bash +npm install @dexpace/rx rxjs +``` + +## Usage + +```typescript +import {sseEvents$, typedSse$, pageItems$, pages$} from '@dexpace/rx'; +import {sseStreamFrom, Paginator} from '@dexpace/core'; + +// Server-Sent Events +sseEvents$(sseStreamFrom(response)).subscribe({ + next: event => console.log(event.data), +}); + +// ...or decoded into your own models +typedSse$(sseStreamFrom(response), (eventName, data) => + eventName === 'done' + ? {kind: 'done'} + : {kind: 'value', value: JSON.parse(data)}, +).subscribe({next: model => console.log(model)}); + +// Pagination, item by item or page by page +pageItems$(paginator).subscribe({next: item => console.log(item)}); +pages$(paginator).subscribe({next: page => console.log(page.items)}); +``` + +## Two subscription models, on purpose + +`sseEvents$`/`typedSse$` are **single-subscription**. An `SseStream` wraps one already-open HTTP response body, +which is single-use (`BODY-14`) and single-pass (`SSE-26`); there is no honest way to make re-subscription +meaningful without a second HTTP call this package does not make. A second `subscribe()` reaches `SseStream`'s +own guard and surfaces its error through the `Observable`'s error channel. + +`pageItems$`/`pages$` are **cold and repeatable**. `Paginator.items()`/`.pages()` build a fresh generator per +call (`PAGE-8`), so each subscription drives an independent fetch sequence. + +Both release their source on `unsubscribe()`, including while idle — an SSE stream waiting on the next event is +closed immediately rather than when the server next sends something. + +## Notes + +- `rxjs` and `@dexpace/core` are **peer** dependencies. A duplicate copy of either would break the identity + checks (`Observable`/`Subscription`, and core's branded symbols) that a bundled copy silently defeats. +- This package installs no RxJS scheduler. Diagnostic context therefore propagates on its own through Node's + `AsyncLocalStorage`; a caller who adds `observeOn`/`subscribeOn` downstream owns reinstating it. diff --git a/packages/rx/api-extractor.json b/packages/rx/api-extractor.json new file mode 100644 index 0000000..455423c --- /dev/null +++ b/packages/rx/api-extractor.json @@ -0,0 +1,22 @@ +{ + "$schema": "https://developer.microsoft.com/json-schemas/api-extractor/v7/api-extractor.schema.json", + "mainEntryPointFilePath": "/dist/index.d.ts", + "apiReport": { + "enabled": true, + "reportFolder": "/etc/" + }, + "docModel": { + "enabled": false + }, + "dtsRollup": { + "enabled": false + }, + "messages": { + "extractorMessageReporting": { + "ae-forgotten-export": { + "logLevel": "error", + "addToApiReportFile": false + } + } + } +} diff --git a/packages/rx/etc/rx.api.md b/packages/rx/etc/rx.api.md new file mode 100644 index 0000000..f54d12b --- /dev/null +++ b/packages/rx/etc/rx.api.md @@ -0,0 +1,26 @@ +## API Report File for "@dexpace/rx" + +> Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/). + +```ts + +import type { Observable } from 'rxjs'; +import type { Page } from '@dexpace/core'; +import type { Paginator } from '@dexpace/core'; +import { SseEvent } from '@dexpace/core'; +import { SseMapper } from '@dexpace/core'; +import { SseStream } from '@dexpace/core'; + +// @public +export function pageItems$(paginator: Paginator): Observable; + +// @public +export function pages$(paginator: Paginator): Observable>; + +// @public +export function sseEvents$(stream: SseStream): Observable; + +// @public +export function typedSse$(stream: SseStream, mapper: SseMapper): Observable; + +``` diff --git a/packages/rx/package.json b/packages/rx/package.json new file mode 100644 index 0000000..7ad80ed --- /dev/null +++ b/packages/rx/package.json @@ -0,0 +1,51 @@ +{ + "name": "@dexpace/rx", + "version": "0.0.0", + "description": "RxJS async-runtime bridge for the dexpace Node.js SDK.", + "license": "MIT", + "type": "module", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "engines": { + "node": ">=20.3" + }, + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + } + }, + "files": [ + "dist" + ], + "sideEffects": false, + "dependencies": {}, + "peerDependencies": { + "@dexpace/core": "workspace:*", + "rxjs": "^7.8.0" + }, + "peerDependenciesMeta": { + "@dexpace/core": { + "optional": false + }, + "rxjs": { + "optional": false + } + }, + "devDependencies": { + "@dexpace/core": "workspace:*", + "@microsoft/api-extractor": "catalog:", + "expect-type": "catalog:", + "fast-check": "catalog:", + "rxjs": "^7.8.0", + "typescript": "catalog:" + }, + "scripts": { + "build": "tsc -p tsconfig.build.json", + "typecheck": "tsc --noEmit", + "test": "bun test", + "api:local": "api-extractor run --local", + "api:ci": "api-extractor run", + "prepublishOnly": "bun run build && bun run api:ci && publint . && attw --pack . --ignore-rules cjs-resolves-to-esm" + } +} diff --git a/packages/rx/src/from-async-iterable.conformance.test.ts b/packages/rx/src/from-async-iterable.conformance.test.ts new file mode 100644 index 0000000..f012ba8 --- /dev/null +++ b/packages/rx/src/from-async-iterable.conformance.test.ts @@ -0,0 +1,331 @@ +// SPDX-License-Identifier: MIT +// packages/rx/src/from-async-iterable.conformance.test.ts +// +// Exercises: ASYNC-21 (poll-once-per-demand under a synchronous subscriber, complete on end-of-source, +// propagate a source error as an error signal), ASYNC-13 (no wrapper exception around a thrown value), +// ASYNC-6 (unsubscribe reaches the source's .return() exactly once across synchronous and asynchronous paths). +// +// This suite tests fromAsyncIterable against a hand-built async generator / iterator test double, deliberately not +// SseStream/Paginator, to isolate "does the async-iterable bridge satisfy the contract" from "does 6b/6c's own close discipline +// work" (already proven in their own test suites). +// +// The final describe block is the one that runs against rxjs's OWN from(), not ours: it pins the single +// ASYNC-6 clause the native operator fails, which is the entire justification for this package shipping a +// hand-written bridge instead of the one-liner its plan called for. If that case ever fails, RxJS has closed +// the gap and `from-async-iterable.ts` should be deleted in favor of `from()`. +import {describe, expect, test} from 'bun:test'; +import {firstValueFrom, from, take, toArray} from 'rxjs'; +import {fromAsyncIterable} from './from-async-iterable.js'; + +async function* countTo( + n: number, + onReturn?: () => void, +): AsyncGenerator { + try { + for (let i = 1; i <= n; i++) { + await Promise.resolve(); + yield i; + } + } finally { + onReturn?.(); + } +} + +function makePendingIterableDouble( + onReturn?: () => void, +): AsyncIterable { + let returned = false; + return { + [Symbol.asyncIterator]() { + return { + next(): Promise> { + if (returned) { + return Promise.resolve({done: true, value: undefined}); + } + return new Promise>(() => { + // never settles + }); + }, + return(): Promise> { + returned = true; + onReturn?.(); + return Promise.resolve({done: true, value: undefined}); + }, + }; + }, + }; +} + +describe('fromAsyncIterable — ASYNC-21', () => { + test('polls the source once per emission, never ahead of demand', async () => { + let pulls = 0; + async function* spy(): AsyncGenerator { + for (let i = 1; i <= 10; i++) { + await Promise.resolve(); + pulls++; + yield i; + } + } + + // The source deliberately outlives demand. A generator yielding exactly as many values as the + // subscriber consumes cannot tell one-pull-per-emission apart from a bridge that prefetches -- it has + // nothing left to prefetch, so the assertion passes either way. Ten available, two taken, two pulled. + const values = await firstValueFrom( + fromAsyncIterable(spy()).pipe(take(2), toArray()), + ); + expect(values).toEqual([1, 2]); + expect(pulls).toBe(2); + }); + + test('completes the Observable when the source generator returns', async () => { + const values = await firstValueFrom( + fromAsyncIterable(countTo(2)).pipe(toArray()), + ); + expect(values).toEqual([1, 2]); + }); + + test('a source throw surfaces via the error channel with the original value, unwrapped', async () => { + async function* throwing(): AsyncGenerator { + await Promise.resolve(); + yield 1; + throw new RangeError('boom'); + } + const errors: unknown[] = []; + await new Promise(resolve => { + fromAsyncIterable(throwing()).subscribe({ + next() { + // ignore + }, + error(err: unknown) { + errors.push(err); + resolve(); + }, + }); + }); + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(RangeError); + expect((errors[0] as RangeError).message).toBe('boom'); // ASYNC-13: not wrapped in an RxJS-internal type + }); +}); + +describe('fromAsyncIterable — ASYNC-6 (synchronous & asynchronous cancellation)', () => { + test("unsubscribing mid-stream synchronously calls the generator's .return() exactly once", async () => { + let returns = 0; + const generator = countTo(5, () => { + returns++; + }); + await new Promise(resolve => { + const subscription = fromAsyncIterable(generator).subscribe({ + next(value) { + if (value === 2) { + subscription.unsubscribe(); + // allow the microtask queue to settle the generator's finally block + setTimeout(resolve, 10); + } + }, + }); + }); + expect(returns).toBe(1); + }); + + test('unsubscribing asynchronously while idle awaiting .next() calls .return() immediately', async () => { + let returns = 0; + let pushNextValue!: (val: number) => void; + + const stream: AsyncIterable = { + [Symbol.asyncIterator]() { + let isDone = false; + let pendingResolve: + ((res: IteratorResult) => void) | undefined; + pushNextValue = (val: number) => { + if (pendingResolve) { + const r = pendingResolve; + pendingResolve = undefined; + r({done: false, value: val}); + } + }; + return { + next() { + if (isDone) { + return Promise.resolve({done: true, value: undefined}); + } + return new Promise>(resolve => { + pendingResolve = resolve; + }); + }, + return() { + isDone = true; + returns++; + return Promise.resolve({done: true, value: undefined}); + }, + }; + }, + }; + + const received: number[] = []; + const subscription = fromAsyncIterable(stream).subscribe({ + next(value) { + received.push(value); + }, + }); + + // Push first value + pushNextValue(1); + await new Promise(r => setTimeout(r, 10)); + expect(received).toEqual([1]); + expect(returns).toBe(0); + + // Unsubscribe while waiting for second value + subscription.unsubscribe(); + expect(returns).toBe(1); + }); +}); + +describe('fromAsyncIterable — ASYNC-6 (edge release handling)', () => { + test('unsubscribing before first emission calls .return() immediately', () => { + let returns = 0; + const iterable = makePendingIterableDouble(() => { + returns++; + }); + + const subscription = fromAsyncIterable(iterable).subscribe({ + next() { + // ignore + }, + }); + + subscription.unsubscribe(); + expect(returns).toBe(1); + }); + + test('unsubscribing swallows release failures from close() or return() per ASYNC-21 / SSE-30', async () => { + const iterable = { + close(): Promise { + return Promise.reject(new Error('close failed')); + }, + [Symbol.asyncIterator]() { + return { + next(): Promise> { + return new Promise>(() => { + // never settles + }); + }, + return(): Promise> { + return Promise.reject(new Error('return failed')); + }, + }; + }, + }; + + const subscription = fromAsyncIterable(iterable).subscribe({ + next() { + // ignore + }, + }); + + // Should not throw unhandled rejection + subscription.unsubscribe(); + await new Promise(r => setTimeout(r, 20)); + }); +}); + +describe('fromAsyncIterable — ASYNC-6 (caller-supplied source release)', () => { + test('releases the source before returning the iterator, so a suspended pull can settle', async () => { + const order: string[] = []; + let settlePendingPull: (() => void) | undefined; + + const iterable: AsyncIterable = { + [Symbol.asyncIterator]() { + return { + next(): Promise> { + return new Promise>(resolve => { + settlePendingPull = () => { + resolve({done: true, value: undefined}); + }; + }); + }, + return(): Promise> { + order.push('return'); + return Promise.resolve({done: true, value: undefined}); + }, + }; + }, + }; + + const subscription = fromAsyncIterable(iterable, () => { + order.push('release'); + settlePendingPull?.(); + return Promise.resolve(); + }).subscribe({ + next() { + // ignore + }, + }); + + subscription.unsubscribe(); + await new Promise(r => setTimeout(r, 10)); + expect(order).toEqual(['release', 'return']); + }); + + test('a rejected release does not surface from unsubscribe()', () => { + const iterable: AsyncIterable = { + [Symbol.asyncIterator]() { + return { + next(): Promise> { + return new Promise>(() => { + // never settles + }); + }, + }; + }, + }; + + const subscription = fromAsyncIterable(iterable, () => + Promise.reject(new Error('release failed')), + ).subscribe({ + next() { + // ignore + }, + }); + + expect(() => { + subscription.unsubscribe(); + }).not.toThrow(); + }); +}); + +describe("rxjs's own from() — the ASYNC-6 gap this module exists to close", () => { + test('does NOT reach the source when unsubscribed while a pull is suspended', async () => { + let returns = 0; + const iterable = makePendingIterableDouble(() => { + returns++; + }); + + const subscription = from(iterable).subscribe({ + next() { + // ignore + }, + }); + subscription.unsubscribe(); + await new Promise(r => setTimeout(r, 20)); + + // Deliberately asserting the DEFECT, not the fix. rxjs 7's async-iterable path only tests + // subscriber.closed after a pull resolves, so an idle SSE stream is never released. When this + // expectation starts failing, `fromAsyncIterable` has become redundant -- see this file's header. + expect(returns).toBe(0); + + // The same source through this module's bridge is released immediately. + let bridgedReturns = 0; + const bridged = fromAsyncIterable( + makePendingIterableDouble(() => { + bridgedReturns++; + }), + ).subscribe({ + next() { + // ignore + }, + }); + bridged.unsubscribe(); + expect(bridgedReturns).toBe(1); + }); +}); diff --git a/packages/rx/src/from-async-iterable.ts b/packages/rx/src/from-async-iterable.ts new file mode 100644 index 0000000..516c9fb --- /dev/null +++ b/packages/rx/src/from-async-iterable.ts @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: MIT +// packages/rx/src/from-async-iterable.ts +import {Observable} from 'rxjs'; + +/** + * Drive an iterator's `return()` on a termination path, swallowing a release failure. + * + * Release is *quiet* here (`SSE-30`): the terminal signal the subscriber sees has already been decided by the + * time this runs, so a failure to release cannot preempt it. `SseStream` still reports the failure through its + * own `onReleaseFailure` hook — swallowing here drops nothing that layer records. + */ +async function returnQuietly(iterator: AsyncIterator): Promise { + if (typeof iterator.return !== 'function') { + return; + } + try { + await iterator.return(); + } catch { + // Quiet release per ASYNC-21 / SSE-30 -- see this function's own doc comment. + } +} + +/** As {@link returnQuietly}, for the caller-supplied source release. */ +async function releaseQuietly(release: () => Promise): Promise { + try { + await release(); + } catch { + // Quiet release per ASYNC-21 / SSE-30 -- see returnQuietly's doc comment. + } +} + +/** + * Converts an {@link AsyncIterable} into an RxJS {@link Observable}, attaching a finalizer that reaches the + * source on every termination path (`ASYNC-6`, `ASYNC-21`). + * + * **Why this is not `rxjs`'s own `from(asyncIterable)`.** RxJS 7's async-iterable path is a bare + * `for await` loop that tests `subscriber.closed` only *after* a pull resolves. Unsubscribing while a pull is + * suspended — the normal state of an idle SSE stream waiting for the next event — therefore reaches the source + * only if and when the server sends something, so the response body is never released and the connection is + * held open indefinitely. Verified against `rxjs@7.8.2`; pinned by + * `from-async-iterable.conformance.test.ts`'s "rxjs's own from()" case, which fails if a future RxJS closes the + * gap and makes this module redundant. + * + * On a termination path this runs `release` first and *then* returns the iterator: closing the source is what + * settles an in-flight pull, and an async generator's `return()` is queued behind that pull rather than + * preempting it. Cancellation is the case that ordering exists for — it is the only one where the pull may + * stay suspended indefinitely. `return()` runs exactly once across every path — early cancellation, + * end-of-source, and a source error alike. + * + * @param iterable - The source to bridge. Its iterator is taken once per subscription. + * @param release - Optional source-level release, run ahead of `iterator.return()`. RxJS runs a subscriber's + * finalizer on *every* termination, so this fires exactly once per subscription — on unsubscription, on + * end-of-source, and on a source error alike, not on cancellation alone. Pass only a release that tolerates + * being called after the source has already drained; `SseStream.close()` is idempotent (`SSE-28`), which is + * what makes the end-of-source call a no-op rather than a second release. + * + * @internal + */ +export function fromAsyncIterable( + iterable: AsyncIterable, + release?: () => Promise, +): Observable { + return new Observable(subscriber => { + const iterator = iterable[Symbol.asyncIterator](); + + // A function, not a bare `subscriber.closed` read: cancellation lands during an `await`, and TypeScript's + // narrowing would otherwise treat every re-check inside the loop as dead code. It is the opposite -- those + // re-checks are the whole point. + const cancelled = (): boolean => subscriber.closed; + + // The single owner of `ASYNC-6`'s exactly-once obligation. Cancellation and end-of-source both reach it, + // and whichever arrives first is the one that releases. + let returned = false; + const returnOnce = async (): Promise => { + if (returned) { + return; + } + returned = true; + await returnQuietly(iterator); + }; + + void (async (): Promise => { + try { + while (!cancelled()) { + const result = await iterator.next(); + if (result.done === true || cancelled()) { + break; + } + subscriber.next(result.value); + } + if (!cancelled()) { + subscriber.complete(); + } + } catch (err: unknown) { + if (!cancelled()) { + subscriber.error(err); + } + } finally { + await returnOnce(); + } + })(); + + return () => { + if (release !== undefined) { + void releaseQuietly(release); + } + void returnOnce(); + }; + }); +} diff --git a/packages/rx/src/index.ts b/packages/rx/src/index.ts new file mode 100644 index 0000000..3ef3c7c --- /dev/null +++ b/packages/rx/src/index.ts @@ -0,0 +1,11 @@ +// SPDX-License-Identifier: MIT +// packages/rx/src/index.ts + +/** + * RxJS async-runtime bridge for the dexpace Node.js SDK. + * + * @packageDocumentation + */ + +export {sseEvents$, typedSse$} from './sse.js'; +export {pageItems$, pages$} from './pagination.js'; diff --git a/packages/rx/src/pagination.test.ts b/packages/rx/src/pagination.test.ts new file mode 100644 index 0000000..93102c7 --- /dev/null +++ b/packages/rx/src/pagination.test.ts @@ -0,0 +1,278 @@ +// SPDX-License-Identifier: MIT +// packages/rx/src/pagination.test.ts +// +// Exercises: PAGE-8 (cold and repeatable: multiple subscriptions drive independent fetch sequences), +// PAGE-1 (emits every item across all pages in server order, pages$ yields whole pages), +// ASYNC-6 (unsubscribing mid-walk cancels the generator cleanly), +// ASYNC-13 (a walk failure reaches the error channel unwrapped, after the items already delivered). +import {describe, expect, test} from 'bun:test'; +import {firstValueFrom, toArray} from 'rxjs'; +import { + Paginator, + Protocol, + Request, + Response, + Status, + type PageInfo, + type PaginationStrategy, + type Transport, +} from '@dexpace/core'; +import {pageItems$, pages$} from './pagination.js'; + +function createMockTransport(): { + transport: Transport; + getSendCount: () => number; +} { + let sendCount = 0; + const transport: Transport = { + send(req: Request): Promise { + sendCount++; + return Promise.resolve( + Response.newBuilder() + .request(req) + .status(Status.of(200)) + .protocol(Protocol.HTTP_1_1) + .build(), + ); + }, + close(): Promise { + return Promise.resolve(); + }, + }; + return {transport, getSendCount: () => sendCount}; +} + +function createTwoPageStrategy(): PaginationStrategy { + return { + parse(_response: Response, template: Request) { + const page = Number(template.url.searchParams.get('page') ?? '1'); + const items = [`item_${String(page)}_1`, `item_${String(page)}_2`]; + if (page >= 2) { + return Promise.resolve({items, nextRequest: undefined}); + } + const nextUrl = new URL(template.url); + nextUrl.searchParams.set('page', String(page + 1)); + const nextRequest = Request.newBuilder() + .method(template.method) + .url(nextUrl) + .build(); + return Promise.resolve({items, nextRequest}); + }, + }; +} + +function createInitialRequest(): Request { + return Request.newBuilder() + .method('GET') + .url('https://api.example.com/items?page=1') + .build(); +} + +/** Serves page 1 normally, then fails the page-2 exchange, so a walk breaks mid-stream rather than at the head. */ +function createFailingTransport(failure: Error): Transport { + return { + send(req: Request): Promise { + if (Number(req.url.searchParams.get('page') ?? '1') >= 2) { + return Promise.reject(failure); + } + return Promise.resolve( + Response.newBuilder() + .request(req) + .status(Status.of(200)) + .protocol(Protocol.HTTP_1_1) + .build(), + ); + }, + close(): Promise { + return Promise.resolve(); + }, + }; +} + +describe('pageItems$', () => { + test('emits every item across all pages in order', async () => { + const {transport} = createMockTransport(); + const paginator = new Paginator({ + transport, + initialRequest: createInitialRequest(), + strategy: createTwoPageStrategy(), + }); + + const items = await firstValueFrom(pageItems$(paginator).pipe(toArray())); + expect(items).toEqual(['item_1_1', 'item_1_2', 'item_2_1', 'item_2_2']); + }); + + test('is cold and repeatable: two subscriptions each drive a fresh fetch sequence', async () => { + const {transport, getSendCount} = createMockTransport(); + const paginator = new Paginator({ + transport, + initialRequest: createInitialRequest(), + strategy: createTwoPageStrategy(), + }); + + const observable = pageItems$(paginator); + const firstWalk = await firstValueFrom(observable.pipe(toArray())); + expect(firstWalk).toEqual(['item_1_1', 'item_1_2', 'item_2_1', 'item_2_2']); + expect(getSendCount()).toBe(2); + + const secondWalk = await firstValueFrom(observable.pipe(toArray())); + expect(secondWalk).toEqual([ + 'item_1_1', + 'item_1_2', + 'item_2_1', + 'item_2_2', + ]); + expect(getSendCount()).toBe(4); // Re-fetched both pages on second subscription + }); + + test('unsubscribing mid-walk cancels the page iteration cleanly (ASYNC-6)', async () => { + const {transport, getSendCount} = createMockTransport(); + const paginator = new Paginator({ + transport, + initialRequest: createInitialRequest(), + strategy: createTwoPageStrategy(), + }); + + const items: string[] = []; + await new Promise(resolve => { + const subscription = pageItems$(paginator).subscribe({ + next(item) { + items.push(item); + if (items.length === 2) { + subscription.unsubscribe(); + resolve(); + } + }, + }); + }); + + expect(items).toEqual(['item_1_1', 'item_1_2']); + expect(getSendCount()).toBe(1); // Did not fetch page 2 + }); +}); + +describe('pages$', () => { + test('emits whole Page objects across all pages in order', async () => { + const {transport} = createMockTransport(); + const paginator = new Paginator({ + transport, + initialRequest: createInitialRequest(), + strategy: createTwoPageStrategy(), + }); + + const pages = await firstValueFrom(pages$(paginator).pipe(toArray())); + expect(pages).toHaveLength(2); + const p0 = pages[0]; + const p1 = pages[1]; + expect(p0).toBeDefined(); + expect(p1).toBeDefined(); + if (p0 === undefined || p1 === undefined) { + throw new Error('expected 2 pages'); + } + expect(p0.items).toEqual(['item_1_1', 'item_1_2']); + expect(p0.status.code).toBe(200); + expect(p1.items).toEqual(['item_2_1', 'item_2_2']); + expect(p1.status.code).toBe(200); + }); + + test('is cold and repeatable for whole pages', async () => { + const {transport, getSendCount} = createMockTransport(); + const paginator = new Paginator({ + transport, + initialRequest: createInitialRequest(), + strategy: createTwoPageStrategy(), + }); + + const observable = pages$(paginator); + await firstValueFrom(observable.pipe(toArray())); + expect(getSendCount()).toBe(2); + + await firstValueFrom(observable.pipe(toArray())); + expect(getSendCount()).toBe(4); + }); + + test('unsubscribing after page 1 cancels further fetches (ASYNC-6)', async () => { + const {transport, getSendCount} = createMockTransport(); + const paginator = new Paginator({ + transport, + initialRequest: createInitialRequest(), + strategy: createTwoPageStrategy(), + }); + + const received: unknown[] = []; + await new Promise(resolve => { + const subscription = pages$(paginator).subscribe({ + next(page) { + received.push(page); + subscription.unsubscribe(); + resolve(); + }, + }); + }); + + expect(received).toHaveLength(1); + expect(getSendCount()).toBe(1); + }); +}); + +describe('pagination error propagation (ASYNC-13)', () => { + test('a transport failure mid-walk reaches the error channel unwrapped, after the items already delivered', async () => { + const failure = new TypeError('the page-2 exchange failed'); + const paginator = new Paginator({ + transport: createFailingTransport(failure), + initialRequest: createInitialRequest(), + strategy: createTwoPageStrategy(), + }); + + const items: string[] = []; + const errors: unknown[] = []; + await new Promise(resolve => { + pageItems$(paginator).subscribe({ + next(item) { + items.push(item); + }, + error(err: unknown) { + errors.push(err); + resolve(); + }, + }); + }); + + // Page 1's items are not rolled back by page 2's failure -- the error is a terminal signal, not an undo. + expect(items).toEqual(['item_1_1', 'item_1_2']); + expect(errors).toHaveLength(1); + // The exact instance, not an RxJS-internal or PaginationError wrapper. + expect(errors[0]).toBe(failure); + }); + + test('a strategy failure reaches pages$ error channel unwrapped', async () => { + const failure = new RangeError('cannot parse this page'); + const {transport} = createMockTransport(); + const strategy: PaginationStrategy = { + parse(): Promise> { + return Promise.reject(failure); + }, + }; + const paginator = new Paginator({ + transport, + initialRequest: createInitialRequest(), + strategy, + }); + + const errors: unknown[] = []; + await new Promise(resolve => { + pages$(paginator).subscribe({ + next() { + // ignore + }, + error(err: unknown) { + errors.push(err); + resolve(); + }, + }); + }); + + expect(errors).toHaveLength(1); + expect(errors[0]).toBe(failure); + }); +}); diff --git a/packages/rx/src/pagination.ts b/packages/rx/src/pagination.ts new file mode 100644 index 0000000..11bd645 --- /dev/null +++ b/packages/rx/src/pagination.ts @@ -0,0 +1,39 @@ +// SPDX-License-Identifier: MIT +// packages/rx/src/pagination.ts +import type {Observable} from 'rxjs'; +import type {Page, Paginator} from '@dexpace/core'; +import {fromAsyncIterable} from './from-async-iterable.js'; + +/** + * Bridges a {@link @dexpace/core#Paginator}'s item stream to a cold, repeatable RxJS `Observable` (PAGE-8). + * + * Each subscription obtains a fresh iterator from `Paginator.items()`, driving an independent pagination sequence + * across all pages. + * + * @param paginator - The `Paginator` instance whose items to observe. + * @returns An `Observable` emitting items of type `T` in server order. + * + * @public + */ +export function pageItems$(paginator: Paginator): Observable { + return fromAsyncIterable({ + [Symbol.asyncIterator]: () => paginator.items()[Symbol.asyncIterator](), + }); +} + +/** + * Bridges a {@link @dexpace/core#Paginator}'s page stream to a cold, repeatable RxJS `Observable` (PAGE-8). + * + * Each subscription obtains a fresh iterator from `Paginator.pages()`, driving an independent pagination sequence + * yielding whole {@link @dexpace/core#Page} objects. + * + * @param paginator - The `Paginator` instance whose pages to observe. + * @returns An `Observable` emitting {@link @dexpace/core#Page} objects. + * + * @public + */ +export function pages$(paginator: Paginator): Observable> { + return fromAsyncIterable({ + [Symbol.asyncIterator]: () => paginator.pages()[Symbol.asyncIterator](), + }); +} diff --git a/packages/rx/src/sse.test.ts b/packages/rx/src/sse.test.ts new file mode 100644 index 0000000..03e4a71 --- /dev/null +++ b/packages/rx/src/sse.test.ts @@ -0,0 +1,205 @@ +// SPDX-License-Identifier: MIT +// packages/rx/src/sse.test.ts +// +// Exercises: SSE-41 (reactive adapter), SSE-26 (single-pass: second subscription fails loudly), +// SSE-33-36 (typed adapter mapping over reactive stream), ASYNC-21, ASYNC-6. +import {describe, expect, test} from 'bun:test'; +import {firstValueFrom, toArray} from 'rxjs'; +import { + Protocol, + Request, + Response, + SseStream, + SseStreamError, + sseStreamFrom, + Status, +} from '@dexpace/core'; +import {sseEvents$, typedSse$} from './sse.js'; + +/** Distinguishes "the promise resolved" from a rejection value that happens to be falsy. */ +const RESOLVED = Symbol('resolved'); + +/** + * Settles `promise` and hands back whatever it rejected with. + * + * `expect(p).rejects.toX()` is typed `void` under `bun:test`, so awaiting it trips `await-thenable` -- the same + * idiom `@dexpace/codec-json`'s `json-serde.test.ts` settled on. + */ +async function rejection(promise: Promise): Promise { + try { + await promise; + return RESOLVED; + } catch (e: unknown) { + return e; + } +} + +function makeSseStreamFixture(text: string): SseStream { + const request = Request.newBuilder() + .method('GET') + .url('https://example.com/events') + .build(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + controller.close(); + }, + }); + const response = Response.newBuilder() + .request(request) + .status(Status.of(200)) + .protocol(Protocol.HTTP_1_1) + .body(body) + .build(); + return sseStreamFrom(response); +} + +function makeUnclosedSseStream(text: string, onCancel?: () => void): SseStream { + const request = Request.newBuilder() + .method('GET') + .url('https://example.com/events') + .build(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + }, + cancel() { + onCancel?.(); + }, + }); + const response = Response.newBuilder() + .request(request) + .status(Status.of(200)) + .protocol(Protocol.HTTP_1_1) + .body(body) + .build(); + return sseStreamFrom(response); +} + +describe('sseEvents$', () => { + test('emits every parsed SseEvent in order and completes at end-of-stream', async () => { + const stream = makeSseStreamFixture('data: one\n\ndata: two\n\n'); + const events = await firstValueFrom(sseEvents$(stream).pipe(toArray())); + expect(events.map(e => e.data)).toEqual([['one'], ['two']]); + }); + + test('a second subscription fails loudly (SSE-26, inherited)', async () => { + const stream = makeSseStreamFixture('data: one\n\n'); + const observable = sseEvents$(stream); + await firstValueFrom(observable.pipe(toArray())); + + // SseStream's own single-pass guard, surfaced through the error channel rather than reimplemented. + expect( + await rejection(firstValueFrom(observable.pipe(toArray()))), + ).toBeInstanceOf(SseStreamError); + }); + + test('unsubscribing mid-stream synchronously releases the underlying stream resource', async () => { + let cancelCalled = false; + const stream = makeUnclosedSseStream( + 'data: one\n\ndata: two\n\ndata: three\n\n', + () => { + cancelCalled = true; + }, + ); + const subscription = sseEvents$(stream).subscribe({ + next(event) { + if (event.data[0] === 'one') { + subscription.unsubscribe(); + } + }, + }); + await new Promise(r => setTimeout(r, 20)); + expect(cancelCalled).toBe(true); + }); + + test('unsubscribing asynchronously while idle releases the underlying stream resource (ASYNC-6)', async () => { + let cancelCalled = false; + const stream = makeUnclosedSseStream('data: one\n\n', () => { + cancelCalled = true; + }); + const received: string[] = []; + const subscription = sseEvents$(stream).subscribe({ + next(event) { + const item = event.data[0]; + if (item !== undefined) { + received.push(item); + } + }, + }); + + await new Promise(r => setTimeout(r, 10)); + expect(received).toEqual(['one']); + expect(cancelCalled).toBe(false); + + subscription.unsubscribe(); + await new Promise(r => setTimeout(r, 20)); + expect(cancelCalled).toBe(true); + }); +}); + +describe('typedSse$', () => { + test('decodes events and honors Value, Skip, and Done outcomes', async () => { + const stream = makeSseStreamFixture( + ':ping\n\nevent: delta\ndata: {"num":1}\n\nevent: delta\ndata: {"num":2}\n\nevent: done\ndata: end\n\ndata: ignored\n\n', + ); + const observable = typedSse$(stream, (event, data) => { + if (event === undefined) return {kind: 'skip'}; + if (event === 'done') return {kind: 'done'}; + const parsed = JSON.parse(data) as {num: number}; + return {kind: 'value', value: parsed.num}; + }); + + const values = await firstValueFrom(observable.pipe(toArray())); + expect(values).toEqual([1, 2]); + }); + + test('a throwing mapper propagates error through the Observable error channel', async () => { + const stream = makeSseStreamFixture('data: invalid-json\n\n'); + const observable = typedSse$(stream, (_event, data) => { + if (data === 'invalid-json') { + throw new TypeError('invalid json payload'); + } + return {kind: 'value', value: data}; + }); + + const errors: unknown[] = []; + await new Promise(resolve => { + observable.subscribe({ + next() { + // ignore + }, + error(err: unknown) { + errors.push(err); + resolve(); + }, + }); + }); + + expect(errors).toHaveLength(1); + expect(errors[0]).toBeInstanceOf(TypeError); + expect((errors[0] as TypeError).message).toBe('invalid json payload'); + }); + + test('unsubscribing asynchronously from typedSse$ releases the underlying stream (ASYNC-6)', async () => { + let cancelCalled = false; + const stream = makeUnclosedSseStream('data: 100\n\n', () => { + cancelCalled = true; + }); + const subscription = typedSse$(stream, (_e, d) => ({ + kind: 'value', + value: Number(d), + })).subscribe({ + next() { + // ignore + }, + }); + + await new Promise(r => setTimeout(r, 10)); + expect(cancelCalled).toBe(false); + + subscription.unsubscribe(); + await new Promise(r => setTimeout(r, 20)); + expect(cancelCalled).toBe(true); + }); +}); diff --git a/packages/rx/src/sse.ts b/packages/rx/src/sse.ts new file mode 100644 index 0000000..e4a343f --- /dev/null +++ b/packages/rx/src/sse.ts @@ -0,0 +1,60 @@ +// SPDX-License-Identifier: MIT +// packages/rx/src/sse.ts +import type {Observable} from 'rxjs'; +import { + typedSseStream, + type SseEvent, + type SseMapper, + type SseStream, +} from '@dexpace/core'; +import {fromAsyncIterable} from './from-async-iterable.js'; + +/** + * Bridges an {@link @dexpace/core#SseStream} to an RxJS `Observable` (SSE-41, ASYNC-21). + * + * Single-subscription: `SseStream` wraps an already-open, single-use HTTP response body (BODY-14) and is itself + * single-pass (SSE-26) -- obtaining an iterator succeeds at most once and a second attempt fails loudly. + * Subscribing to the returned `Observable` a second time reaches `SseStream`'s own guard and surfaces an error + * through the `Observable`'s error channel. + * + * Unsubscribing closes the stream, releasing the response body even when no event is in flight (ASYNC-6). + * + * Diagnostic context propagates on its own (ASYNC-8–ASYNC-11): every pull runs inside the continuation chain + * that called `subscribe()`, which is exactly what Node's `AsyncLocalStorage` tracks. That holds only for the + * `Observable` returned here -- a caller who pipes it through an RxJS scheduler operator (`observeOn`, + * `subscribeOn`) hands each emission to a task outside that chain, and owns reinstating the context on the far + * side. This package installs no scheduler of its own, precisely so that boundary is never introduced behind + * the caller's back. + * + * @param stream - The `SseStream` instance to observe. + * @returns An `Observable` emitting parsed {@link @dexpace/core#SseEvent}s. + * + * @public + */ +export function sseEvents$(stream: SseStream): Observable { + return fromAsyncIterable(stream, () => stream.close()); +} + +/** + * Bridges an {@link @dexpace/core#SseStream} to a typed RxJS `Observable` via an {@link @dexpace/core#SseMapper} (SSE-41, ASYNC-21, SSE-33–SSE-36). + * + * Shares every note on {@link sseEvents$}: single-subscription, release-on-unsubscribe, and automatic + * diagnostic-context propagation through the unscheduled path. + * + * A throwing mapper reaches the `Observable`'s error channel unwrapped (ASYNC-13), after `typedSseStream` has + * released the stream (SSE-36). + * + * @param stream - The `SseStream` instance to observe. + * @param mapper - The mapper decoding raw SSE events into domain items, skips, or done sentinels. + * @returns An `Observable` emitting decoded items of type `T`. + * + * @public + */ +export function typedSse$( + stream: SseStream, + mapper: SseMapper, +): Observable { + return fromAsyncIterable(typedSseStream(stream, mapper), () => + stream.close(), + ); +} diff --git a/packages/rx/tsconfig.build.json b/packages/rx/tsconfig.build.json new file mode 100644 index 0000000..d39dc7e --- /dev/null +++ b/packages/rx/tsconfig.build.json @@ -0,0 +1,11 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "declaration": true, + "declarationMap": true, + "sourceMap": true + }, + "exclude": [ + "src/**/*.test.ts" + ] +} diff --git a/packages/rx/tsconfig.json b/packages/rx/tsconfig.json new file mode 100644 index 0000000..5c03021 --- /dev/null +++ b/packages/rx/tsconfig.json @@ -0,0 +1,16 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src", + "target": "ES2023", + "lib": [ + "ES2023", + "DOM", + "DOM.AsyncIterable" + ] + }, + "include": [ + "src/**/*.ts" + ] +} diff --git a/scripts/verify-consumer-types.mjs b/scripts/verify-consumer-types.mjs index 7f2af15..052a2bf 100644 --- a/scripts/verify-consumer-types.mjs +++ b/scripts/verify-consumer-types.mjs @@ -88,6 +88,7 @@ const builtTransportUndici = join( 'dist', 'index.js', ); +const builtRx = join(repoRoot, 'packages', 'rx', 'dist', 'index.js'); const tsc = join(repoRoot, 'node_modules', '.bin', 'tsc'); // Checked up front, not left to the catch below. A missing prerequisite reported through the @@ -106,6 +107,7 @@ for (const artifact of [ builtTransportShared, builtTransportFetch, builtTransportUndici, + builtRx, ]) { assert.ok( existsSync(artifact), @@ -270,6 +272,13 @@ import { undiciTransport, type UndiciTransportOptions, } from ${JSON.stringify(builtTransportUndici)}; +import { + pageItems$, + pages$, + sseEvents$, + typedSse$, +} from ${JSON.stringify(builtRx)}; +import type {Paginator, SseMapper, SseStream} from ${JSON.stringify(built)}; export function readBody(response: Response): Promise { @@ -535,6 +544,21 @@ export function transportAdapters( export function undiciAdapter(options: UndiciTransportOptions): Transport { return undiciTransport(options); } + +export function rxBridge( + stream: SseStream, + mapper: SseMapper, + paginator: Paginator, +): void { + const _e = sseEvents$(stream); + const _t = typedSse$(stream, mapper); + const _i = pageItems$(paginator); + const _p = pages$(paginator); + void _e; + void _t; + void _i; + void _p; +} `; const tsconfig = { diff --git a/scripts/verify-dual-consumption.mjs b/scripts/verify-dual-consumption.mjs index 875cd44..b3f34a4 100644 --- a/scripts/verify-dual-consumption.mjs +++ b/scripts/verify-dual-consumption.mjs @@ -6,7 +6,17 @@ // Phase 6a, when `@dexpace/codec-json` became the workspace's second package -- a check hard-coded // to one package silently stops covering the workspace the moment it grows. import assert from 'node:assert/strict'; -import {absent, Headers, present, serdeBody, Status} from '@dexpace/core'; +import { + absent, + Headers, + present, + Protocol, + Request, + Response, + serdeBody, + sseStreamFrom, + Status, +} from '@dexpace/core'; import {jsonSerde} from '@dexpace/codec-json'; import {createPinoLogger} from '@dexpace/logging-pino'; @@ -18,6 +28,8 @@ import { } from '@dexpace/transport-shared'; import {fetchTransport} from '@dexpace/transport-fetch'; import {undiciTransport} from '@dexpace/transport-undici'; +import {pageItems$, pages$, sseEvents$, typedSse$} from '@dexpace/rx'; +import {firstValueFrom, toArray} from 'rxjs'; assert.equal(Status.of(200).code, 200); assert.equal(Status.of(200).name, 'OK'); @@ -96,6 +108,30 @@ for (const transport of [fetchTransport(), undiciTransport()]) { assert.equal(typeof transport[Symbol.asyncDispose], 'function'); await transport.close(); } +// Exercise @dexpace/rx bridge +const req = Request.newBuilder() + .method('GET') + .url('https://api.test/events') + .build(); +const sseBody = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: hello\n\n')); + controller.close(); + }, +}); +const sseResp = Response.newBuilder() + .request(req) + .status(Status.of(200)) + .protocol(Protocol.HTTP_1_1) + .body(sseBody) + .build(); +const sseStream = sseStreamFrom(sseResp); +const events = await firstValueFrom(sseEvents$(sseStream).pipe(toArray())); +assert.equal(events.length, 1); +assert.deepEqual(events[0].data, ['hello']); +assert.equal(typeof typedSse$, 'function'); +assert.equal(typeof pageItems$, 'function'); +assert.equal(typeof pages$, 'function'); console.log( 'dual-consumption check passed: plain Node import resolved and executed all packages in workspace', diff --git a/test/node-conformance/README.md b/test/node-conformance/README.md index d22de30..6f23c3f 100644 --- a/test/node-conformance/README.md +++ b/test/node-conformance/README.md @@ -37,8 +37,9 @@ Node). ## Membership rule **A phase that touches a runtime-divergent surface adds a case here, not only to `bun test`** (§5.9:378). That -means Phase 4 (pipelines, where `NFR-11`'s async-framework-leak check lands) and Phase 8 (concrete -`fetch`/`undici` transports, where this stops being precautionary and becomes the point). +means Phase 4 (pipelines, where `NFR-11`'s async-framework-leak check lands) and Phase 8 — both halves: 8a's +concrete `fetch`/`undici` transports, where this stops being precautionary and becomes the point, and 8b's +RxJS bridge, whose entire reason for being hand-written is a cancellation path the runtime decides. ## Files @@ -49,3 +50,4 @@ means Phase 4 (pipelines, where `NFR-11`'s async-framework-leak check lands) and | `body-lifecycle.test.mjs` | Phase 3b's public body surface over real Node Web Streams — reader-lock discipline, `pipeTo` ownership, multipart framing, error-body buffering | | `transport.test.mjs` | Phase 8a's two concrete transports against a real `node:http` server on Node's own `fetch`/`undici`, `AbortSignal`, and Web Streams — redirect passthrough, timeout and no-response classification, a single-use streaming request body, lazy response bodies, `SEAM-16`'s abort-after-delivery rule, and concurrency | | `redirect.test.mjs` | Phase 5b's Location resolution on Node's own WHATWG `URL` parser (relative resolution, percent-encoding preservation, userinfo clearing, bracketed IPv6, which malformed forms throw versus resolve as a relative reference) plus `PIPE-40`'s per-hop close discipline over real Node Web Streams | +| `rx-bridge.test.mjs` | Phase 8b's `@dexpace/rx` cancellation path — unsubscribing an idle `sseEvents$`/`typedSse$` must reach the source, which depends on Node's `ReadableStream.cancel()` settling a suspended read and on Node's async-generator `return()` queueing behind an in-flight `next()`; plus `pages$`'s mid-walk page release | diff --git a/test/node-conformance/rx-bridge.test.mjs b/test/node-conformance/rx-bridge.test.mjs new file mode 100644 index 0000000..eb73364 --- /dev/null +++ b/test/node-conformance/rx-bridge.test.mjs @@ -0,0 +1,260 @@ +// SPDX-License-Identifier: MIT +// test/node-conformance/rx-bridge.test.mjs +// +// Phase 8b's runtime-divergent surface, run against the BUILT artifact on real Node. +// +// `@dexpace/rx` is four one-line wrappers over `fromAsyncIterable`, and the whole reason that module is +// hand-written rather than `rxjs`'s own `from()` is a cancellation path whose behavior is decided by the +// runtime, not by this package: +// 1. Unsubscribing while a pull is suspended must reach the source. Whether that release lands depends on +// Node's Web Streams `cancel()` and on Node's async-generator `return()` queueing behind an in-flight +// `next()` -- both independent implementations of Bun's, and the SSE idle case is exactly the state a +// long-lived event stream sits in almost all the time. +// 2. The release ordering the bridge relies on (close the source, THEN return the iterator) only settles a +// suspended pull if the runtime's `ReadableStream` cancellation rejects/resolves the pending read. +// 3. `pages$` unsubscribed mid-walk must close the in-hand page's response body (PAGE-11/PAGE-26) through +// the same generator-return path. +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; +import {firstValueFrom, toArray} from 'rxjs'; +import { + Paginator, + Protocol, + Request, + Response, + sseStreamFrom, + Status, +} from '@dexpace/core'; +import {pageItems$, pages$, sseEvents$, typedSse$} from '@dexpace/rx'; + +/** + * An SSE response whose body stays open after the given text: the reader is left suspended on the next pull, + * which is the idle state the cancellation cases below need. `cancel()` firing is the only sanctioned way to + * observe the release (`Response` instances are frozen, so a spy assignment throws). + */ +function openSseStream(text, onCancel) { + const request = Request.newBuilder() + .method('GET') + .url('https://example.com/events') + .build(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + }, + cancel() { + onCancel(); + }, + }); + const response = Response.newBuilder() + .request(request) + .status(Status.of(200)) + .protocol(Protocol.HTTP_1_1) + .body(body) + .build(); + return sseStreamFrom(response); +} + +function closedSseStream(text) { + const request = Request.newBuilder() + .method('GET') + .url('https://example.com/events') + .build(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + controller.close(); + }, + }); + const response = Response.newBuilder() + .request(request) + .status(Status.of(200)) + .protocol(Protocol.HTTP_1_1) + .body(body) + .build(); + return sseStreamFrom(response); +} + +const settle = () => new Promise(resolve => setTimeout(resolve, 20)); + +describe('sseEvents$ over Node Web Streams (SSE-41, ASYNC-21)', () => { + it('emits every parsed event in order and completes at end-of-stream', async () => { + const stream = closedSseStream('data: one\n\ndata: two\n\n'); + const events = await firstValueFrom(sseEvents$(stream).pipe(toArray())); + assert.deepEqual( + events.map(event => event.data), + [['one'], ['two']], + ); + }); + + it('releases the response body when unsubscribed while idle (ASYNC-6)', async () => { + let cancelled = 0; + const stream = openSseStream('data: one\n\n', () => { + cancelled += 1; + }); + + const received = []; + const subscription = sseEvents$(stream).subscribe({ + next: event => received.push(event.data[0]), + }); + + await settle(); + assert.deepEqual(received, ['one'], 'the first event should have arrived'); + assert.equal(cancelled, 0, 'an idle stream must stay open until cancelled'); + + subscription.unsubscribe(); + await settle(); + assert.equal(cancelled, 1, 'unsubscribing must release the response body'); + }); + + it('releases the response body when unsubscribed from inside next() (ASYNC-6)', async () => { + let cancelled = 0; + const stream = openSseStream('data: one\n\ndata: two\n\n', () => { + cancelled += 1; + }); + + const subscription = sseEvents$(stream).subscribe({ + next: () => { + subscription.unsubscribe(); + }, + }); + + await settle(); + assert.equal(cancelled, 1); + }); + + it('fails loudly on a second subscription (SSE-26, inherited)', async () => { + const stream = closedSseStream('data: one\n\n'); + const events$ = sseEvents$(stream); + await firstValueFrom(events$.pipe(toArray())); + await assert.rejects(() => firstValueFrom(events$.pipe(toArray()))); + }); +}); + +describe('typedSse$ over Node Web Streams (SSE-33..SSE-36)', () => { + it('decodes events and terminates on the mapper done sentinel', async () => { + const stream = closedSseStream( + 'event: delta\ndata: 1\n\nevent: delta\ndata: 2\n\nevent: end\ndata: x\n\n', + ); + const values = await firstValueFrom( + typedSse$(stream, (eventName, data) => + eventName === 'end' + ? {kind: 'done'} + : {kind: 'value', value: Number(data)}, + ).pipe(toArray()), + ); + assert.deepEqual(values, [1, 2]); + }); + + it('releases the response body when unsubscribed while idle (ASYNC-6)', async () => { + let cancelled = 0; + const stream = openSseStream('data: 100\n\n', () => { + cancelled += 1; + }); + + const subscription = typedSse$(stream, (_eventName, data) => ({ + kind: 'value', + value: Number(data), + })).subscribe({next: () => undefined}); + + await settle(); + assert.equal(cancelled, 0); + + subscription.unsubscribe(); + await settle(); + assert.equal(cancelled, 1); + }); +}); + +describe('pageItems$/pages$ over Node (PAGE-8, ASYNC-6)', () => { + const twoPages = () => { + const closed = []; + let sendCount = 0; + const transport = { + send(request) { + sendCount += 1; + const page = Number(request.url.searchParams.get('page') ?? '1'); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('{}')); + }, + cancel() { + closed.push(page); + }, + }); + return Promise.resolve( + Response.newBuilder() + .request(request) + .status(Status.of(200)) + .protocol(Protocol.HTTP_1_1) + .body(body) + .build(), + ); + }, + close: () => Promise.resolve(), + }; + const strategy = { + parse(_response, template) { + const page = Number(template.url.searchParams.get('page') ?? '1'); + const items = [`item_${page}_1`, `item_${page}_2`]; + if (page >= 2) { + return Promise.resolve({items, nextRequest: undefined}); + } + const nextUrl = new URL(template.url); + nextUrl.searchParams.set('page', String(page + 1)); + return Promise.resolve({ + items, + nextRequest: Request.newBuilder() + .method(template.method) + .url(nextUrl) + .build(), + }); + }, + }; + const paginator = new Paginator({ + transport, + initialRequest: Request.newBuilder() + .method('GET') + .url('https://api.example.com/items?page=1') + .build(), + strategy, + }); + return {paginator, closed, sendCount: () => sendCount}; + }; + + it('walks every page and is cold: a second subscription re-fetches', async () => { + const {paginator, sendCount} = twoPages(); + const items$ = pageItems$(paginator); + + assert.deepEqual(await firstValueFrom(items$.pipe(toArray())), [ + 'item_1_1', + 'item_1_2', + 'item_2_1', + 'item_2_2', + ]); + assert.equal(sendCount(), 2); + + await firstValueFrom(items$.pipe(toArray())); + assert.equal( + sendCount(), + 4, + 'PAGE-8: each subscription drives a fresh walk', + ); + }); + + it('closes the in-hand page body and stops fetching on unsubscribe (PAGE-11, ASYNC-6)', async () => { + const {paginator, closed, sendCount} = twoPages(); + + await new Promise(resolve => { + const subscription = pages$(paginator).subscribe({ + next: () => { + subscription.unsubscribe(); + resolve(); + }, + }); + }); + await settle(); + + assert.equal(sendCount(), 1, 'page 2 must never be requested'); + assert.deepEqual(closed, [1], 'page 1 body must be released'); + }); +});