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'); + }); +});