From f0ec03a079f5dbff3a1150e7c01136b485790e25 Mon Sep 17 00:00:00 2001 From: Mohammad Wahbeh Date: Thu, 27 Aug 2026 19:22:37 +0300 Subject: [PATCH] =?UTF-8?q?feat(core):=20phase=206b=20=E2=80=94=20Server-S?= =?UTF-8?q?ent=20Events=20(SSE-1..41).?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement the full Server-Sent Events subsystem in `@dexpace/core` per `docs/product-spec/13-server-sent-events-and-streaming.md` (SSE-1 through SSE-41). The implementation is strictly pull-based, single-pass, and zero-dependency: events are parsed 1:1 on demand as the consumer polls the stream, with no unbounded buffering and no auto-reconnection logic. @dexpace/core additions: - `SseEvent`, `makeSseEvent`: Immutable event representation with defensively copied and frozen `data` lines (`SSE-20`). Structural equality (`sseEventsEqual`), string representation (`sseEventToString`), and content predicate (`isSseEventEmpty`, where comments count as content per `SSE-22`). - `SseLineReader` (@internal): Hand-rolled line framing supporting `\n`, `\r`, and `\r\n` line terminators across chunk boundaries (`SSE-2`), start-only UTF-8 BOM stripping via non-consuming `peek()` lookahead (`SSE-12`), and an optional configurable line length cap (`maxLineBytes` / `SseLineTooLongError`, `SSE-19`). - `SseParser` (@internal): Single-pass state machine implementing WHATWG SSE grammar with the three reference spec deviations: comments captured and dispatched (`SSE-6`), permissive dispatch when any of the 5 fields are set (`SSE-13`), and EOF dispatch of pending fields (`SSE-14`). Ignores NUL in IDs (`SSE-9`), unknown fields (`SSE-7`), and non-digit / overflow retries (`SSE-11`). - `SseStream` / `sseStreamFrom`: Resource-owning single-pass AsyncGenerator facade (`SSE-18`, `SSE-23`–`SSE-32`). Guarantees exactly-once release of the underlying Response body and BufferedSource across all termination routes (clean EOF, explicit `close()`, early `break`, consumer error, mid-stream read error, or abort signal). Teardown promise memoization ensures concurrent `close()` awaits in-flight releases and propagates failures (`SSE-30`). In-flight reader teardown on close is mapped to `IoError` (`SSE-31`). Abort listeners are cleaned up on normal completion. - `typedSseStream`: Lazy per-element stream adapter passing raw event name and newline-joined data (`SSE-33`, `SSE-35`). Dispatches `MapperOutcome` union (`mapperValue`, `MAPPER_SKIP`, `MAPPER_DONE`, `SSE-34`). Releases stream resource before propagating any mapper error, attaching close failure as suppressed (`SSE-36`). Tooling, gates, and conformance: - `scripts/verify-sse-37.mjs` & `test:scripts`: Recursive AST/regex gate enforcing zero serde dependencies in core SSE (`SSE-37`) and no reconnect / `Last-Event-ID` paths (`SSE-38`). - `test/node-conformance/sse.test.mjs`: 8 new conformance test cases running over real Node Web Streams and TextDecoder under `node --test` across Node 20.3.0 and LTS (`test:node`). - `docs/superpowers/plans/2026-07-28-phase6b-sse-checklist.md`: Full requirement traceability checklist mapping all 41 SSE requirements to code and tests. - `docs/open-items.md`: Recorded Section I entries for `SSE-41` reactive adapter deferral (Phase 8b `@dexpace/rx`), line reader separation rationale (`IO-14` vs `SSE-2`), `Symbol.asyncDispose` runtime floor guard, and JavaScript hash equality. Gates verified: typecheck, lint, build, bun test (1,514 passing, 100% coverage on src/sse/*), api, lint:publish (publint + attw), verify:dual-consumption, verify:consumer-types, verify:seam-1, verify:sse-37, test:scripts (40 passing), verify:runtime-floor, test:node (87 passing), and audit. --- .changeset/phase6b-sse.md | 7 + .github/workflows/ci.yml | 3 + docs/open-items.md | 35 +- .../plans/2026-07-28-phase6b-sse-checklist.md | 85 +++ ...2026-07-23-nodejs-sdk-v1-roadmap-design.md | 3 +- package.json | 1 + packages/core/etc/core.api.md | 95 +++ packages/core/src/index.public.test.ts | 27 + packages/core/src/index.ts | 21 + packages/core/src/sse/errors.test.ts | 19 + packages/core/src/sse/errors.ts | 20 + packages/core/src/sse/event.test.ts | 92 +++ packages/core/src/sse/event.ts | 118 ++++ packages/core/src/sse/lifecycle.test.ts | 90 +++ .../core/src/sse/line-reader.property.test.ts | 50 ++ packages/core/src/sse/line-reader.test.ts | 180 ++++++ packages/core/src/sse/line-reader.ts | 150 +++++ packages/core/src/sse/parser.property.test.ts | 43 ++ packages/core/src/sse/parser.test.ts | 213 +++++++ packages/core/src/sse/parser.ts | 155 +++++ packages/core/src/sse/stream.test.ts | 553 ++++++++++++++++++ packages/core/src/sse/stream.ts | 317 ++++++++++ packages/core/src/sse/typed.test.ts | 157 +++++ packages/core/src/sse/typed.ts | 135 +++++ scripts/verify-sse-37.mjs | 121 ++++ scripts/verify-sse-37.test.mjs | 111 ++++ test/node-conformance/sse.test.mjs | 221 +++++++ 27 files changed, 3020 insertions(+), 2 deletions(-) create mode 100644 .changeset/phase6b-sse.md create mode 100644 docs/superpowers/plans/2026-07-28-phase6b-sse-checklist.md create mode 100644 packages/core/src/sse/errors.test.ts create mode 100644 packages/core/src/sse/errors.ts create mode 100644 packages/core/src/sse/event.test.ts create mode 100644 packages/core/src/sse/event.ts create mode 100644 packages/core/src/sse/lifecycle.test.ts create mode 100644 packages/core/src/sse/line-reader.property.test.ts create mode 100644 packages/core/src/sse/line-reader.test.ts create mode 100644 packages/core/src/sse/line-reader.ts create mode 100644 packages/core/src/sse/parser.property.test.ts create mode 100644 packages/core/src/sse/parser.test.ts create mode 100644 packages/core/src/sse/parser.ts create mode 100644 packages/core/src/sse/stream.test.ts create mode 100644 packages/core/src/sse/stream.ts create mode 100644 packages/core/src/sse/typed.test.ts create mode 100644 packages/core/src/sse/typed.ts create mode 100644 scripts/verify-sse-37.mjs create mode 100644 scripts/verify-sse-37.test.mjs create mode 100644 test/node-conformance/sse.test.mjs diff --git a/.changeset/phase6b-sse.md b/.changeset/phase6b-sse.md new file mode 100644 index 0000000..6498ff3 --- /dev/null +++ b/.changeset/phase6b-sse.md @@ -0,0 +1,7 @@ +--- +'@dexpace/core': minor +--- + +Add the SSE subsystem: `sseStreamFrom()`, `SseStream`, `typedSseStream()`, the `SseEvent` value and its +operations, and the `MapperOutcome` union. Pull-based with no read-ahead; no reconnection and no last-event-id +continuity, both of which remain the caller's responsibility. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fba100e..5a555b5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,6 +53,9 @@ jobs: - name: SEAM-1 zero-dependency check run: bun run verify:seam-1 + - name: Verify SSE-37/SSE-38 (no serde dependency, no reconnect path in core SSE) + run: bun run verify:sse-37 + - name: Runtime-floor consistency check run: bun run verify:runtime-floor diff --git a/docs/open-items.md b/docs/open-items.md index 63140d0..2ca2430 100644 --- a/docs/open-items.md +++ b/docs/open-items.md @@ -858,7 +858,8 @@ maintain it; out of scope for a review pass because it is Phase 2 surface. ### H13 — `test:scripts` runs in no CI job — **OPEN** `scripts/*.test.mjs` runs only under `bun run test:scripts`, and `.github/workflows/ci.yml` has no step that -invokes it. As of 6a that glob covers `scripts/knowledge.test.mjs` and the new `scripts/verify-seam-1.test.mjs`. +invokes it. As of 6b that glob covers `scripts/knowledge.test.mjs`, `scripts/verify-seam-1.test.mjs`, and +`scripts/verify-sse-37.test.mjs`. The script was named `test:knowledge` until the Phase 6a reader pass renamed it: the glob had outgrown the name the moment `verify-seam-1.test.mjs` landed, and both places that cite it had to explain the mismatch in @@ -1016,6 +1017,38 @@ between them and a red CI run, so they should be pinned the same way rather than --- +## Section I — Phase 6b (Server-Sent Events) + +### I1 — `SSE-41` reactive adapter deferred to Phase 8b (`@dexpace/rx`) — **SCHEDULED** (Phase 8b) + +`SSE-41` (MAY) describes a backpressure-honoring `Observable` view over an SSE stream. Phase 6b ships the +pull-based `AsyncGenerator` surface `SSE-39` mandates; the reactive view is a bridge package, and the roadmap +scopes `§18`'s async-runtime adapters to Phase 8b specifically (`@dexpace/rx`). + +### I2 — Hand-rolled `SseLineReader` vs `BufferedSource.readUtf8Line()` — **RECORDED** + +`BufferedSource.readUtf8Line()` (`IO-14`) treats `\n` and `\r\n` as terminators but keeps a lone `\r` as line +content. `SSE-2` requires the opposite: a lone `\r` terminates an SSE line by itself. Both contracts are +normative for their respective subsystems, so SSE frames its own lines in `src/sse/line-reader.ts` rather than +reshaping a frozen Phase 3a surface. Recorded so Phase 10's deviation review does not read the duplication as +accidental. + +### I3 — `[Symbol.asyncDispose]` runtime-guarded and omitted from `.d.ts` — **WATCH** + +Node 20.3 (the pinned floor verified by `verify:runtime-floor` and CI `node-conformance`) predates +`Symbol.asyncDispose` (which landed in Node 20.4). TypeScript does not polyfill the well-known symbol for a +library that declares the member, so declaring it on the interface would cause `.d.ts` compilation failures for +consumers on standard `ES2023` lib without `esnext.disposable`. `SseStream` therefore installs +`[Symbol.asyncDispose]` at run time only when the symbol exists, matching `Response` (HTTP-38). Becomes an +unconditional `implements AsyncDisposable` when `engines.node` moves past Node 20.4. + +### I4 — `SSE-21` hash equality is N/A in JavaScript — **RECORDED** + +`SSE-21` mentions value equality and hash. JavaScript does not have language-level hash maps keyed by object +value equality (`hashCode`); value equality is provided via `sseEventsEqual()` (`SSE-21`). + +--- + ## 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-phase6b-sse-checklist.md b/docs/superpowers/plans/2026-07-28-phase6b-sse-checklist.md new file mode 100644 index 0000000..474831e --- /dev/null +++ b/docs/superpowers/plans/2026-07-28-phase6b-sse-checklist.md @@ -0,0 +1,85 @@ +# Phase 6b — Server-Sent Events Implementation Plan — Checklist + +Verification of [2026-07-28-phase6b-sse.md](./2026-07-28-phase6b-sse.md) against every requirement ID in +`docs/product-spec/13-server-sent-events-and-streaming.md` (`SSE-1`–`SSE-41`), as dispositioned by +`docs/superpowers/specs/2026-07-28-phase6b-sse-design.md`. + +**Status: EXECUTED (2026-08-27).** Every task below is implemented and tested across 1,497 repository tests, 40 script tests, and 79 Node conformance tests. Deviations, deferrals, and design rationales are recorded in `docs/open-items.md` §I and the roadmap design. + +**Legend:** ✅ Implemented and tested — ✅(t) Satisfied by construction, with a test as the only possible evidence — ⏳ Deferred (named target phase) — N/A Not applicable in this port. + +--- + +## §13.1 — Event Model (`SSE-20`–`SSE-22`) + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| SSE-20 | MUST | Immutable `SseEvent` with defensively copied `data` lines | ✅ | Task 1 — `packages/core/src/sse/event.ts` (`makeSseEvent()`). `event.test.ts` asserts `Object.isFrozen(event)` and `Object.isFrozen(event.data)`. | +| SSE-21 | MUST | Value equality and string representation | ✅ | Task 1 — `packages/core/src/sse/event.ts` (`sseEventsEqual()`, `sseEventToString()`). Tested in `event.test.ts`. (Hash equality is N/A in JS, recorded in §I). | +| SSE-22 | MUST | `isSseEventEmpty` predicate (comment counts as content) | ✅ | Task 1 — `packages/core/src/sse/event.ts` (`isSseEventEmpty()`). Tested in `event.test.ts`. | + +--- + +## §13.2 — Line Framing, Parsing & Grammar (`SSE-1`–`SSE-19`) + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| SSE-1 | MUST | Dispatch on blank line & reset per-block accumulators | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-2 | MUST | `\n`, `\r`, and `\r\n` line framing | ✅ | Task 2 — `packages/core/src/sse/line-reader.ts`. Tested across split chunk boundaries in `line-reader.test.ts` and `line-reader.property.test.ts`. | +| SSE-3 | MUST | First-colon field splitting | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-4 | MUST | Present-but-empty recorded as `""`, distinct from absent (`undefined`) | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-5 | MUST | Single leading `U+0020` space stripped | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-6 | MUST | Leading `:` captures comment (latest wins) & dispatches | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-7 | MUST | Unknown fields silently ignored | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-8 | MUST | `data` lines accumulated in wire order as `string[]` | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-9 | MUST | `id` with `\u0000` dropped completely | ✅ | Task 3 — `packages/core/src/sse/parser.ts` & `event.ts`. Tested in `parser.test.ts` and `event.test.ts`. | +| SSE-10 | MUST | `event` not defaulted to `"message"` | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-11 | MUST | `retry` digits-only ASCII capped at `Number.MAX_SAFE_INTEGER` | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-12 | MUST | Leading UTF-8 BOM stripped via lookahead (`peek()`) once at stream start; later BOMs preserved | ✅ | Task 2 — `packages/core/src/sse/line-reader.ts`. Tested in `line-reader.test.ts`, `parser.test.ts`, and `test/node-conformance/sse.test.mjs`. | +| SSE-13 | MUST | Permissive dispatch (any of the 5 fields set emits event) | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-14 | MUST | EOF dispatch of pending fields | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-15 | MUST | Stable end sentinel (`SSE_END`) | ✅ | Task 2 + Task 3 — `packages/core/src/sse/line-reader.ts` & `parser.ts`. Tested in `line-reader.test.ts` and `parser.test.ts`. | +| SSE-16 | MUST | Single-pass; no `last-event-id` state retention across events | ✅ | Task 3 — `packages/core/src/sse/parser.ts`. Tested in `parser.test.ts`. | +| SSE-17 | MUST | Parser does not close or own `BufferedSource` | ✅ | Task 2 + Task 3 — `packages/core/src/sse/line-reader.ts` & `parser.ts`. Tested in `parser.test.ts`. | +| SSE-18 | MUST | Single-consumer model re-expressed as single-pass AsyncGenerator | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | +| SSE-19 | MUST | Configurable line cap (`maxLineBytes` / `SseLineTooLongError`) | ✅ | Task 2 — `packages/core/src/sse/line-reader.ts`. Tested in `line-reader.test.ts`. | + +--- + +## §13.3 — Stream Facade & Resource Management (`SSE-23`–`SSE-32`) + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| SSE-23 | MUST | Exactly-once resource release across all termination paths | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `lifecycle.test.ts` (6-path matrix). | +| SSE-24 | MUST | Clean stream termination automatically releases resource | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts` and `lifecycle.test.ts`. | +| SSE-25 | MUST | Partial consume release via iterator `.return()` / early `break` | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts` and `lifecycle.test.ts`. | +| SSE-26 | MUST | Re-iteration throws `SseStreamError` | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | +| SSE-27 | MUST | Post-close iteration throws `SseStreamError`; mid-pull close ends cleanly | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | +| SSE-28 | MUST | Idempotent `close()` | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | +| SSE-29 | MUST | Mid-stream failure releases resource first; close error suppressed | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | +| SSE-30 | MUST | Clean terminal release failure swallowed/reported out-of-band; explicit `close()` propagates | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | +| SSE-31 | MUST | Close during in-flight read mapped to `IoError` | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts` and `test/node-conformance/sse.test.mjs`. | +| SSE-32 | MUST | `sseStreamFrom` binds response body lifecycle; rejects bodyless response | ✅ | Task 7 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | + +--- + +## §13.4 — Typed Adapter (`SSE-33`–`SSE-36`) + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| SSE-33 | MUST | Typed adapter passes raw event name + newline-joined data | ✅ | Task 6 — `packages/core/src/sse/typed.ts`. Tested in `typed.test.ts`. | +| SSE-34 | MUST | `MapperOutcome` union (`mapperValue`, `MAPPER_SKIP`, `MAPPER_DONE`) | ✅ | Task 6 — `packages/core/src/sse/typed.ts`. Tested in `typed.test.ts`. | +| SSE-35 | MUST | Lazy per-element mapping | ✅ | Task 6 — `packages/core/src/sse/typed.ts`. Tested in `typed.test.ts`. | +| SSE-36 | MUST | Throwing mapper releases resource before propagating error | ✅ | Task 6 — `packages/core/src/sse/typed.ts`. Tested in `typed.test.ts`. | + +--- + +## §13.5 — Boundaries, Flow Control & Isolation (`SSE-37`–`SSE-41`) + +| ID | Level | Requirement gist | Status | Where | +|---|---|---|---|---| +| SSE-37 | MUST | Zero serde dependencies in core SSE | ✅ | Task 8 — `scripts/verify-sse-37.mjs`, asserted in CI and `test:scripts`. | +| SSE-38 | MUST | No reconnect or `Last-Event-ID` path in core SSE | ✅ | Task 8 — `scripts/verify-sse-37.mjs`, asserted in CI and `test:scripts`. | +| SSE-39 | MUST | Pull-based flow control (1:1 with consumer demand) | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | +| SSE-40 | MUST | Single-pass lazy view reusing reader | ✅ | Task 5 — `packages/core/src/sse/stream.ts`. Tested in `stream.test.ts`. | +| SSE-41 | MAY | Reactive adapter view (`Observable`) | ⏳ | Deferred to Phase 8b (`@dexpace/rx`). Recorded in §I and roadmap. | diff --git a/docs/superpowers/specs/2026-07-23-nodejs-sdk-v1-roadmap-design.md b/docs/superpowers/specs/2026-07-23-nodejs-sdk-v1-roadmap-design.md index e33e560..815ed76 100644 --- a/docs/superpowers/specs/2026-07-23-nodejs-sdk-v1-roadmap-design.md +++ b/docs/superpowers/specs/2026-07-23-nodejs-sdk-v1-roadmap-design.md @@ -150,7 +150,8 @@ permanent simplification, not a postponement. | Phase 4 split into 4a (Execution Context, `§7`) / 4b (recovery-chain primitives, `§8.2`) / 4c (stage-based pipeline, `§8.1`) | Phase 4 brainstorm | — | ~76 combined normative IDs, comparable to Phase 3's ~79 that forced its own 3a/3b split; each sub-phase gets its own brainstorm→spec→plan cycle. Dependency order: 4a first (contexts are the pipeline's own per-call correlation state), then 4b and 4c | | Phase 5 split into 5a (Retry, `§9`) / 5b (Redirect, `§10`) / 5c (Auth, `§11`) | Phase 5 brainstorm | — | 111 combined normative IDs — the largest single phase in the roadmap, well past the ~76–79 that already forced the Phase 3 and Phase 4 splits. Build order is forced by coupling, not just size: retry is independent of the other two; redirect owns the cross-origin marker `REDIR-11` defines and `AUTH-29` reads, so it must precede auth; the standard-resilience preset needs all three steps installed, so it closes 5c. Each sub-phase gets its own brainstorm→spec→plan cycle | | Phase 6 split into 6a (Serde, `§14`) / 6b (SSE, `§13`) / 6c (Pagination, `§12`) | Phase 6 brainstorm (2026-07-28) | — | 107 combined normative IDs (`PAGE` 36, `SSE` 41, `SERDE` 30), between the ~76–79 that forced the Phase 3 and Phase 4 splits and Phase 5's 111. Cut along the spec's own section boundaries because **the spec forbids the couplings that would cross them**: `SSE-37` (MUST) bars any serde dependency from core SSE, and `§12`'s preamble declares pagination serde-agnostic — so the cross-segment contract surface is empty by mandate, which is exactly the property whose absence caused the 5b/5c drift below. **No segment depends on another; the 6a→6b→6c order is convenience, not dependency**, and any sub-phase may execute out of order. 6a leads only because it scaffolds the workspace's second package and is the one segment that reshapes an already-published seam (`SEAM-21`); 6c trails because it is the most coupled to *earlier* phases (4c's `Runtime`, 5a's `StepContext.options`, 3b's `Response` body). Full rationale, per-segment ownership, and the collapsed-ID clusters in the [Phase 6 segmentation design](./2026-07-28-phase6-segmentation-design.md) | -| Collapsed-requirement disposition tables for Phase 6 — `PAGE-25`–`PAGE-33` (§12.9's async engine: this port has one async model, so the async generator *is* the engine), `SSE-18`/`SSE-31` (threading re-expressed against the event loop), `SERDE-8`/`SERDE-21`/`SERDE-22`/`SERDE-25`/`SERDE-26` (codec-engine configuration with no configurable engine to configure) | Phase 6 brainstorm | Each owning sub-phase's design (6c, 6b, 6a) | Same service 5a's `RECOV-17`–`RECOV-34` table performs: without a row-by-row disposition, a naive appendix-B sweep reads ~18 collapsed requirements as uncovered. The segmentation design identifies the clusters and what does **not** collapse inside each — notably `PAGE-26`/`PAGE-27`/`PAGE-32`'s close-exactly-once obligations (re-expressed as `finally`-block obligations on the single generator) and `SSE-31`'s close-during-in-flight-read branch, both of which stay real, testable work. **Note (Phase 9 design, 2026-07-28):** Phase 9's actual design scopes to `§19`/`§20` (`XCUT`/`NFR`) only — it does not re-verify `PAGE`/`SSE`/`SERDE` disposition, which stays each owning sub-phase's own responsibility as this row already states (6c, 6b, 6a respectively) | +| Collapsed-requirement disposition tables for Phase 6 — `PAGE-25`–`PAGE-33` (§12.9's async engine: this port has one async model, so the async generator *is* the engine), `SSE-18`/`SSE-31` (threading re-expressed against the event loop), `SERDE-8`/`SERDE-21`/`SERDE-22`/`SERDE-25`/`SERDE-26` (codec-engine configuration with no configurable engine to configure) | Phase 6 brainstorm | Each owning sub-phase's design (6c, **Resolved for 6b in Phase 6b design**, 6a) | Same service 5a's `RECOV-17`–`RECOV-34` table performs: without a row-by-row disposition, a naive appendix-B sweep reads ~18 collapsed requirements as uncovered. The segmentation design identifies the clusters and what does **not** collapse inside each — notably `PAGE-26`/`PAGE-27`/`PAGE-32`'s close-exactly-once obligations (re-expressed as `finally`-block obligations on the single generator), `SSE-18` re-expressed against the event loop, and `SSE-31`'s close-during-in-flight-read branch (re-expressed and tested, **not** collapsed), both documented in the [Phase 6b design](./2026-07-28-phase6b-sse-design.md). **Note (Phase 9 design, 2026-07-28):** Phase 9's actual design scopes to `§19`/`§20` (`XCUT`/`NFR`) only — it does not re-verify `PAGE`/`SSE`/`SERDE` disposition, which stays each owning sub-phase's own responsibility as this row already states (6c, 6b, 6a respectively) | +| 3a's `readUtf8Line()` is unusable for SSE (`IO-14` keeps a lone `\r` as content, `SSE-2` requires it to terminate) | Phase 6b | **Resolved in Phase 6b** — closed 2026-08-27 | 6b owns `src/sse/line-reader.ts` instead of reshaping a frozen Phase 3a surface. Recorded so Phase 10's deviation review does not read the duplication as accidental | | `sdk-design-nodejs/07` §7.1's item-view snippet closes the page *after* yielding its items; `PAGE-11` (MUST) requires closing *before* | Phase 6 brainstorm | **Phase 6c** (erratum against `sdk-design-nodejs/07` **and** `docs/knowledge/pagination.md`) | The 2026-07-28 plans review found the erratum was being written into `sdk-design-nodejs/07` only, while `docs/knowledge/pagination.md` carries the *same* wrong ordering in its Reference section directly beside the correct MUST in its Rules section. The knowledge corpus is the standing tie-breaker every later phase consults, so an erratum that skips it leaves the contradiction live; 6c's plan now amends both. Recorded because **the conformance test is weaker than the requirement**: the snippet's `finally` still passes `PAGE-11`'s stated check (an early `break` drives `.return()`, hence the close), so following the design doc ships a violation the appendix-B checklist would not catch. Resolution per the standing tie-breaker (normative spec + knowledge corpus win over an illustrative snippet): `PAGE-11` governs — copy items, close, *then* yield. Costs nothing, since materialized items survive close per `PAGE-2`. The snippet remains correct about the thing §7.1 is actually arguing (JavaScript's automatic `.return()`-on-abandon), just not about close ordering | | `PAGE-5`'s "strategy MUST read everything it needs from the response **synchronously** inside parse" | Phase 6 brainstorm | **Phase 6c** (design must state the re-expression) | Node has no synchronous body read, so the literal reading is unimplementable and `parse` returns a promise. Every part of the requirement's actual intent survives: single-use-body discipline, no retention of the response or its body past the call, no close, no mutation. Flagged so an async signature does not later read as an oversight or get "fixed" back toward a literal reading | | `SSE-41` — reactive SSE adapter (backpressure-honoring `Observable` view, fatal/non-fatal split, source-ownership documentation) | Phase 6 brainstorm | **Phase 8b** (`@dexpace/rx`) | `MAY`. 6b ships the pull-based `AsyncGenerator` surface `SSE-39` mandates; the reactive view is a bridge package, and the roadmap scopes `§18`'s async-runtime adapters to 8b specifically (not 8a's transports) as of the 2026-07-28 [Phase 8 segmentation design](./2026-07-28-phase8-segmentation-design.md). `sdk-design-nodejs/02` identifies RxJS's push-based `Observable` as the one async shape in the Node ecosystem worth bridging at all. Is `ASYNC-21` restated — the segmentation design's §5.2 names it 8b's marquee deliverable | diff --git a/package.json b/package.json index 8158555..923a852 100644 --- a/package.json +++ b/package.json @@ -51,6 +51,7 @@ "verify:dual-consumption": "node scripts/verify-dual-consumption.mjs", "verify:consumer-types": "node scripts/verify-consumer-types.mjs", "verify:seam-1": "node scripts/verify-seam-1.mjs", + "verify:sse-37": "node scripts/verify-sse-37.mjs", "verify:runtime-floor": "node scripts/verify-runtime-floor.mjs" } } diff --git a/packages/core/etc/core.api.md b/packages/core/etc/core.api.md index a10f5a5..73cc9cc 100644 --- a/packages/core/etc/core.api.md +++ b/packages/core/etc/core.api.md @@ -388,12 +388,37 @@ export function isPresent(tristate: Tristate): tristate is { // @public export function isSerdeError(e: unknown): e is SerializationError | DeserializationError; +// @public +export function isSseEventEmpty(event: SseEvent): boolean; + // @public export function isTimeoutSignal(signal: AbortSignal): boolean; // @public export function isTristate(value: unknown): value is Tristate; +// @public +export function makeSseEvent(fields: SseEventFields): SseEvent; + +// @public +export const MAPPER_DONE: MapperOutcome; + +// @public +export const MAPPER_SKIP: MapperOutcome; + +// @public +export type MapperOutcome = { + readonly kind: 'value'; + readonly value: T; +} | { + readonly kind: 'skip'; +} | { + readonly kind: 'done'; +}; + +// @public +export function mapperValue(value: T): MapperOutcome; + // @public export function materialize(body: Body_2): Promise; @@ -744,6 +769,73 @@ export interface Serializer { serializeToString(value: unknown): string; } +// @public +export interface SseEvent { + // (undocumented) + readonly comment: string | undefined; + readonly data: readonly string[]; + // (undocumented) + readonly event: string | undefined; + // (undocumented) + readonly id: string | undefined; + // (undocumented) + readonly retryMs: number | undefined; +} + +// @public +export interface SseEventFields { + // (undocumented) + readonly comment?: string | undefined; + // (undocumented) + readonly data?: readonly string[] | undefined; + // (undocumented) + readonly event?: string | undefined; + // (undocumented) + readonly id?: string | undefined; + // (undocumented) + readonly retryMs?: number | undefined; +} + +// @public +export function sseEventsEqual(a: SseEvent, b: SseEvent): boolean; + +// @public +export function sseEventToString(event: SseEvent): string; + +// @public +export class SseLineTooLongError extends DexpaceError { + constructor(limitBytes: number, options?: ErrorOptions); + readonly limitBytes: number; +} + +// @public +export type SseMapper = (eventName: string | undefined, joinedData: string) => MapperOutcome; + +// @public +export class SseStream implements AsyncIterable { + [Symbol.asyncIterator](): AsyncIterator; + close(): Promise; +} + +// @public +export class SseStreamError extends DexpaceError { + constructor(message: string, options?: ErrorOptions); +} + +// @public +export function sseStreamFrom(response: Response_2, options?: SseStreamFromOptions): SseStream; + +// @public +export interface SseStreamFromOptions extends SseStreamOptions { + readonly maxLineBytes?: number | undefined; + readonly signal?: AbortSignal | undefined; +} + +// @public +export interface SseStreamOptions { + readonly onReleaseFailure?: ((error: unknown) => void) | undefined; +} + // @public export type Stage = 'PRE_REDIRECT' | 'REDIRECT' | 'POST_REDIRECT' | 'PRE_RETRY' | 'RETRY' | 'POST_RETRY' | 'PRE_AUTH' | 'AUTH' | 'POST_AUTH' | 'PRE_LOGGING' | 'LOGGING' | 'POST_LOGGING' | 'PRE_SERDE' | 'SERDE' | 'POST_SERDE' | 'SEND'; @@ -871,6 +963,9 @@ export class TypedResponse { value(): Promise; } +// @public +export function typedSseStream(stream: SseStream, mapper: SseMapper): AsyncIterable; + // @public export class UrlConstructionError extends DomainModelError { } diff --git a/packages/core/src/index.public.test.ts b/packages/core/src/index.public.test.ts index 6e0c593..e8ca5ac 100644 --- a/packages/core/src/index.public.test.ts +++ b/packages/core/src/index.public.test.ts @@ -41,3 +41,30 @@ test('io/ is still not public — 3b froze that decision and 6a does not reopen expect(barrel).not.toHaveProperty(name); } }); + +test('the SSE surface is publicly importable', async () => { + const barrel = await import('./index.js'); + for (const name of [ + 'sseStreamFrom', + 'SseStream', + 'typedSseStream', + 'mapperValue', + 'MAPPER_SKIP', + 'MAPPER_DONE', + 'SseStreamError', + 'SseLineTooLongError', + 'makeSseEvent', + 'sseEventsEqual', + 'isSseEventEmpty', + 'sseEventToString', + ]) { + expect(barrel).toHaveProperty(name); + } +}); + +test('the SSE parser internals stay private — publishing them would publish a way to break SSE-17', async () => { + const barrel = await import('./index.js'); + for (const name of ['SseParser', 'SseLineReader']) { + expect(barrel).not.toHaveProperty(name); + } +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index d0e1739..2acbdfb 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -196,3 +196,24 @@ export { } from './serde/response-handlers.js'; export type {DecodeTarget} from './serde/response-handlers.js'; export {serdeBody} from './body/serde-body.js'; + +// SSE (Phase 6b). The parser and line reader stay internal: they are driven only through the facade, and +// exposing them would expose a way to violate SSE-17's non-ownership contract by accident. +export type {SseEvent, SseEventFields} from './sse/event.js'; +export { + isSseEventEmpty, + makeSseEvent, + sseEventToString, + sseEventsEqual, +} from './sse/event.js'; +export {SseLineTooLongError} from './sse/line-reader.js'; +export {SseStreamError} from './sse/errors.js'; +export {SseStream, sseStreamFrom} from './sse/stream.js'; +export type {SseStreamFromOptions, SseStreamOptions} from './sse/stream.js'; +export { + MAPPER_DONE, + MAPPER_SKIP, + mapperValue, + typedSseStream, +} from './sse/typed.js'; +export type {MapperOutcome, SseMapper} from './sse/typed.js'; diff --git a/packages/core/src/sse/errors.test.ts b/packages/core/src/sse/errors.test.ts new file mode 100644 index 0000000..48a2565 --- /dev/null +++ b/packages/core/src/sse/errors.test.ts @@ -0,0 +1,19 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/errors.test.ts +// Exercises: SSE-26/SSE-27 (loud failure on re-iteration or post-close iteration), SSE-32 (bodyless response). +import {expect, test} from 'bun:test'; +import {DexpaceError} from '../http/errors.js'; +import {SseStreamError} from './errors.js'; + +test('sits directly under DexpaceError (two-level tree)', () => { + expect(new SseStreamError('x')).toBeInstanceOf(DexpaceError); +}); + +test('name identifies the leaf in a stack trace', () => { + expect(new SseStreamError('x').name).toBe('SseStreamError'); +}); + +test('chains a cause when given one', () => { + const backing = new Error('root'); + expect(new SseStreamError('x', {cause: backing}).cause).toBe(backing); +}); diff --git a/packages/core/src/sse/errors.ts b/packages/core/src/sse/errors.ts new file mode 100644 index 0000000..4bce953 --- /dev/null +++ b/packages/core/src/sse/errors.ts @@ -0,0 +1,20 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/errors.ts +import {DexpaceError} from '../http/errors.js'; + +/** + * A misuse or precondition failure of the SSE stream facade: a second iterator on a single-pass stream + * (SSE-26), an iterator requested after close (SSE-27), or a stream opened over a response with no body + * (SSE-32). + * + * Distinct from `IoError`, which is a genuine read failure. This type always means the *caller* did something + * the contract forbids, or the *server* sent a response the contract cannot work with. + * + * @public + */ +export class SseStreamError extends DexpaceError { + // eslint-disable-next-line @typescript-eslint/no-useless-constructor -- explicit constructor required for Bun test function coverage instrumentation + constructor(message: string, options?: ErrorOptions) { + super(message, options); + } +} diff --git a/packages/core/src/sse/event.test.ts b/packages/core/src/sse/event.test.ts new file mode 100644 index 0000000..7219bd2 --- /dev/null +++ b/packages/core/src/sse/event.test.ts @@ -0,0 +1,92 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/event.test.ts +// Exercises: SSE-20 (immutable, defensively-copied data list), SSE-21 (structural equality, stable string form), +// SSE-22 (is-empty true only when all five fields are unset; a comment counts as content). +import {expect, test} from 'bun:test'; +import {InvariantViolation} from '../invariant.js'; +import { + isSseEventEmpty, + makeSseEvent, + sseEventToString, + sseEventsEqual, +} from './event.js'; + +test('the data list is defensively copied at construction (SSE-20)', () => { + const supplied = ['a', 'b']; + const event = makeSseEvent({data: supplied}); + supplied.push('c'); + expect(event.data).toEqual(['a', 'b']); +}); + +test('the event and its data list are frozen (SSE-20)', () => { + const event = makeSseEvent({data: ['a']}); + expect(Object.isFrozen(event)).toBe(true); + expect(Object.isFrozen(event.data)).toBe(true); +}); + +test('unset fields are undefined and data defaults to an empty list', () => { + const event = makeSseEvent({}); + expect(event.id).toBeUndefined(); + expect(event.event).toBeUndefined(); + expect(event.comment).toBeUndefined(); + expect(event.retryMs).toBeUndefined(); + expect(event.data).toEqual([]); +}); + +test('equality is structural over all five fields (SSE-21)', () => { + const a = makeSseEvent({ + id: '1', + event: 'ping', + data: ['x'], + comment: 'c', + retryMs: 5, + }); + const b = makeSseEvent({ + id: '1', + event: 'ping', + data: ['x'], + comment: 'c', + retryMs: 5, + }); + expect(sseEventsEqual(a, b)).toBe(true); +}); + +test('equality distinguishes present-but-empty from absent (SSE-4 seen through SSE-21)', () => { + expect(sseEventsEqual(makeSseEvent({event: ''}), makeSseEvent({}))).toBe( + false, + ); +}); + +test('equality is order-sensitive across the data list', () => { + expect( + sseEventsEqual( + makeSseEvent({data: ['a', 'b']}), + makeSseEvent({data: ['b', 'a']}), + ), + ).toBe(false); +}); + +test('the string form is stable and leaks no identity (SSE-21)', () => { + const rendered = sseEventToString(makeSseEvent({id: '1', data: ['x']})); + expect(rendered).toBe(sseEventToString(makeSseEvent({id: '1', data: ['x']}))); + expect(rendered).not.toMatch(/\[object|0x[0-9a-f]+/); +}); + +test('is-empty is true only when every field is unset (SSE-22)', () => { + expect(isSseEventEmpty(makeSseEvent({}))).toBe(true); + expect(isSseEventEmpty(makeSseEvent({data: ['']}))).toBe(false); + expect(isSseEventEmpty(makeSseEvent({event: ''}))).toBe(false); +}); + +test('a comment-only event is NOT empty — a comment counts as content (SSE-22)', () => { + expect(isSseEventEmpty(makeSseEvent({comment: 'keep-alive'}))).toBe(false); +}); + +test('a NUL-bearing id cannot be built into an event — SSE-9 drops it at the parser', () => { + expect(() => makeSseEvent({id: 'a\u0000b'})).toThrow(InvariantViolation); +}); + +test('a negative or non-integer retryMs cannot be built into an event (SSE-11)', () => { + expect(() => makeSseEvent({retryMs: -1})).toThrow(InvariantViolation); + expect(() => makeSseEvent({retryMs: 1.5})).toThrow(InvariantViolation); +}); diff --git a/packages/core/src/sse/event.ts b/packages/core/src/sse/event.ts new file mode 100644 index 0000000..59504e7 --- /dev/null +++ b/packages/core/src/sse/event.ts @@ -0,0 +1,118 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/event.ts +import {invariant} from '../invariant.js'; + +/** + * One parsed Server-Sent Event (SSE-20). + * + * A frozen plain object, not a class: it has no lifecycle, no invariant to maintain past construction, and no + * behavior — `styleguide/typescript/06` §6.3's test for a data structure rather than an object. Equality and + * rendering are therefore free functions in this module, not methods. + * + * `undefined` means the field was **absent** from the block. A field present with an empty value is `''`, and + * that distinction is load-bearing (SSE-4): an empty `event:` is a present empty event name, not a missing one. + * + * @public + */ +export interface SseEvent { + readonly id: string | undefined; + readonly event: string | undefined; + /** Raw per-line `data` values in wire order, never joined at this layer (SSE-8). */ + readonly data: readonly string[]; + readonly comment: string | undefined; + readonly retryMs: number | undefined; +} + +/** + * Fields used to construct an {@link SseEvent}. + * + * @public + */ +export interface SseEventFields { + readonly id?: string | undefined; + readonly event?: string | undefined; + readonly data?: readonly string[] | undefined; + readonly comment?: string | undefined; + readonly retryMs?: number | undefined; +} + +/** + * Construct a frozen event, defensively copying the data list so later mutation cannot reach inside (SSE-20). + * + * @throws InvariantViolation when a field carries a value the grammar can never produce — a NUL-bearing `id` + * (SSE-9 drops those at the parser) or a `retryMs` that is not a non-negative safe integer (SSE-11). Both are + * programmer errors, not stream conditions: the parser is the only production caller and it filters both. + * + * @public + */ +export function makeSseEvent(fields: SseEventFields): SseEvent { + // Positive and negative space on the two fields the grammar constrains: what must hold, and the impossible + // value that must be absent. + invariant( + fields.retryMs === undefined || + (Number.isSafeInteger(fields.retryMs) && fields.retryMs >= 0), + `retryMs must be a non-negative safe integer when set, got ${String(fields.retryMs)}`, + ); + invariant( + !fields.id?.includes('\u0000'), + 'an SSE id containing U+0000 must be dropped by the parser, never carried into an event (SSE-9)', + ); + + return Object.freeze({ + id: fields.id, + event: fields.event, + data: Object.freeze([...(fields.data ?? [])]), + comment: fields.comment, + retryMs: fields.retryMs, + }); +} + +/** + * Structural equality over all five fields, order-sensitive across `data` (SSE-21). + * + * @public + */ +export function sseEventsEqual(a: SseEvent, b: SseEvent): boolean { + return ( + a.id === b.id && + a.event === b.event && + a.comment === b.comment && + a.retryMs === b.retryMs && + a.data.length === b.data.length && + a.data.every((line, index) => line === b.data[index]) + ); +} + +/** + * True only when every field is unset or empty (SSE-22). + * + * A comment counts as content, so a comment-only event reports non-empty — that is the deliberate deviation from + * strict WHATWG this subsystem replicates, not an oversight. + * + * @public + */ +export function isSseEventEmpty(event: SseEvent): boolean { + return ( + event.id === undefined && + event.event === undefined && + event.comment === undefined && + event.retryMs === undefined && + event.data.length === 0 + ); +} + +/** + * Stable, identity-free rendering for logs and assertion messages (SSE-21). + * + * @public + */ +export function sseEventToString(event: SseEvent): string { + const parts = [ + `id=${event.id ?? ''}`, + `event=${event.event ?? ''}`, + `data=[${event.data.join('|')}]`, + `comment=${event.comment ?? ''}`, + `retryMs=${event.retryMs === undefined ? '' : String(event.retryMs)}`, + ]; + return `SseEvent(${parts.join(', ')})`; +} diff --git a/packages/core/src/sse/lifecycle.test.ts b/packages/core/src/sse/lifecycle.test.ts new file mode 100644 index 0000000..c352c4e --- /dev/null +++ b/packages/core/src/sse/lifecycle.test.ts @@ -0,0 +1,90 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/lifecycle.test.ts +// SSE-23: exactly one release across the stream's whole life, regardless of how it terminated. +import {expect, test} from 'bun:test'; +import {BufferedSource} from '../io/buffered-source.js'; +import {SseParser} from './parser.js'; +import {SseStream} from './stream.js'; +import {MAPPER_DONE, mapperValue, typedSseStream} from './typed.js'; + +function counted(text: string): {stream: SseStream; closes: () => number} { + let closeCount = 0; + const web = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + controller.close(); + }, + }); + const stream = new SseStream(new SseParser(BufferedSource.overStream(web)), { + close(): Promise { + closeCount += 1; + return Promise.resolve(); + }, + }); + return {stream, closes: () => closeCount}; +} + +const THREE_EVENTS = 'data: a\n\ndata: b\n\ndata: STOP\n\n'; + +test.each([ + [ + 'clean end of stream', + async (stream: SseStream) => { + for await (const event of stream) { + void event; + } + }, + ], + [ + 'explicit close with no iteration', + async (stream: SseStream) => { + await stream.close(); + }, + ], + [ + 'partial consume then explicit close', + async (stream: SseStream) => { + for await (const event of stream) { + void event; + break; + } + await stream.close(); + }, + ], + [ + 'early break alone', + async (stream: SseStream) => { + for await (const event of stream) { + void event; + break; + } + }, + ], + [ + 'consumer throws mid-iteration', + async (stream: SseStream) => { + try { + for await (const event of stream) { + void event; + throw new Error('consumer blew up'); + } + } catch { + /* expected */ + } + }, + ], + [ + 'typed mapper returns Done', + async (stream: SseStream) => { + for await (const value of typedSseStream(stream, (_n, d) => + d === 'STOP' ? MAPPER_DONE : mapperValue(d), + )) { + void value; + } + }, + ], +])('exactly one release: %s (SSE-23)', async (_name, terminate) => { + const {stream, closes} = counted(THREE_EVENTS); + await terminate(stream); + expect(closes()).toBe(1); +}); diff --git a/packages/core/src/sse/line-reader.property.test.ts b/packages/core/src/sse/line-reader.property.test.ts new file mode 100644 index 0000000..365a401 --- /dev/null +++ b/packages/core/src/sse/line-reader.property.test.ts @@ -0,0 +1,50 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/line-reader.property.test.ts +// The guarantee the carry buffer exists to provide: how bytes arrive must not change how lines come out. +import {test} from 'bun:test'; +import fc from 'fast-check'; +import {BufferedSource} from '../io/buffered-source.js'; +import {SSE_END, SseLineReader} from './line-reader.js'; + +const FIXTURE = 'data: a\r\ndata: b\rdata: c\n\nid: 7\ndata: tail'; + +async function linesFromChunks( + chunks: readonly Uint8Array[], +): Promise { + const stream = new ReadableStream({ + start(controller) { + for (const c of chunks) controller.enqueue(c); + controller.close(); + }, + }); + const reader = new SseLineReader(BufferedSource.overStream(stream)); + const lines: string[] = []; + for (;;) { + const line = await reader.nextLine(); + if (line === SSE_END) return lines; + lines.push(line); + } +} + +test('the line sequence is identical for every chunk split of the same bytes', async () => { + const all = new TextEncoder().encode(FIXTURE); + const expected = await linesFromChunks([all]); + + await fc.assert( + fc.asyncProperty( + fc.uniqueArray(fc.integer({min: 1, max: all.length - 1}), {maxLength: 4}), + async rawCuts => { + const cuts = [...rawCuts].sort((a, b) => a - b); + const chunks: Uint8Array[] = []; + let prev = 0; + for (const cut of cuts) { + chunks.push(all.slice(prev, cut)); + prev = cut; + } + chunks.push(all.slice(prev)); + const actual = await linesFromChunks(chunks); + return JSON.stringify(actual) === JSON.stringify(expected); + }, + ), + ); +}); diff --git a/packages/core/src/sse/line-reader.test.ts b/packages/core/src/sse/line-reader.test.ts new file mode 100644 index 0000000..2aae29c --- /dev/null +++ b/packages/core/src/sse/line-reader.test.ts @@ -0,0 +1,180 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/line-reader.test.ts +// Exercises: SSE-2 (LF, CR, CRLF; CRLF is one terminator; a lone CR terminates by itself), SSE-12 (one leading +// BOM consumed via lookahead, a later BOM preserved as data), SSE-14 (a final unterminated line is content), +// SSE-19 (optional line cap, off by default). +import {expect, test} from 'bun:test'; +import {BufferedSource} from '../io/buffered-source.js'; +import {SSE_END, SseLineReader, SseLineTooLongError} from './line-reader.js'; + +/** Build a BufferedSource over a byte stream delivered in the given chunks. */ +function sourceOf(chunks: readonly (string | Uint8Array)[]): BufferedSource { + const stream = new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue( + typeof chunk === 'string' ? new TextEncoder().encode(chunk) : chunk, + ); + } + controller.close(); + }, + }); + // 3a exposes no public constructor — `overStream` takes the ReadableStream itself and acquires the reader. + return BufferedSource.overStream(stream); +} + +async function drain(reader: SseLineReader): Promise { + const lines: string[] = []; + for (;;) { + const line = await reader.nextLine(); + if (line === SSE_END) return lines; + lines.push(line); + } +} + +test('LF terminates a line (SSE-2)', async () => { + expect(await drain(new SseLineReader(sourceOf(['a\nb\n'])))).toEqual([ + 'a', + 'b', + ]); +}); + +test('CRLF is a single terminator, not two (SSE-2)', async () => { + expect(await drain(new SseLineReader(sourceOf(['a\r\nb\r\n'])))).toEqual([ + 'a', + 'b', + ]); +}); + +test('a lone CR terminates a line by itself (SSE-2)', async () => { + expect(await drain(new SseLineReader(sourceOf(['a\rb\r'])))).toEqual([ + 'a', + 'b', + ]); +}); + +test('mixed terminators in one stream all work', async () => { + expect(await drain(new SseLineReader(sourceOf(['a\nb\r\nc\rd\n'])))).toEqual([ + 'a', + 'b', + 'c', + 'd', + ]); +}); + +test('a CR ending one chunk and an LF starting the next is ONE terminator', async () => { + // The framing bug this reader exists to avoid: a naive splitter emits a spurious empty line here, which in + // SSE means a spurious event dispatch. + expect(await drain(new SseLineReader(sourceOf(['a\r', '\nb\n'])))).toEqual([ + 'a', + 'b', + ]); +}); + +test('a CR at the very end of the stream still terminates its line', async () => { + expect(await drain(new SseLineReader(sourceOf(['a\r'])))).toEqual(['a']); +}); + +test('a final line with no terminator is returned as content (SSE-14)', async () => { + expect(await drain(new SseLineReader(sourceOf(['a\nb'])))).toEqual([ + 'a', + 'b', + ]); +}); + +test('an empty stream yields no lines', async () => { + expect(await drain(new SseLineReader(sourceOf([])))).toEqual([]); +}); + +test('blank lines are preserved — they are the dispatch boundary', async () => { + expect(await drain(new SseLineReader(sourceOf(['a\n\nb\n'])))).toEqual([ + 'a', + '', + 'b', + ]); +}); + +test('one leading BOM is consumed exactly once (SSE-12)', async () => { + expect(await drain(new SseLineReader(sourceOf(['data: x\n'])))).toEqual([ + 'data: x', + ]); +}); + +test('a non-BOM prefix survives the lookahead intact (SSE-12)', async () => { + expect(await drain(new SseLineReader(sourceOf(['data: x\n'])))).toEqual([ + 'data: x', + ]); +}); + +test('a BOM later in the stream is preserved as ordinary data (SSE-12)', async () => { + expect(await drain(new SseLineReader(sourceOf(['data: ab\n'])))).toEqual([ + 'data: ab', + ]); +}); + +test('a multi-byte character split across chunks decodes correctly', async () => { + const bytes = new TextEncoder().encode('data: ü\n'); + const split = bytes.indexOf(0xc3); + expect( + await drain( + new SseLineReader( + sourceOf([bytes.slice(0, split + 1), bytes.slice(split + 1)]), + ), + ), + ).toEqual(['data: ü']); +}); + +test('no line cap applies by default (SSE-19)', async () => { + const long = 'x'.repeat(100_000); + expect(await drain(new SseLineReader(sourceOf([`${long}\n`])))).toEqual([ + long, + ]); +}); + +test('an explicit cap rejects an oversized line (SSE-19)', () => { + const reader = new SseLineReader(sourceOf(['x'.repeat(50)]), 10); + expect(reader.nextLine()).rejects.toBeInstanceOf(SseLineTooLongError); +}); + +test('the end sentinel is stable — repeated pulls past EOF keep reporting the end', async () => { + // Not a duplicate of the parser's SSE-15 test. The parser has its own `#ended` guard that would mask a reader + // which kept answering; this asserts the reader itself terminates, because a reader that returns `''` forever + // is an infinite supply of SSE dispatch boundaries, and `drain()` above would never return. + const reader = new SseLineReader(sourceOf(['a\n'])); + expect(await reader.nextLine()).toBe('a'); + expect(await reader.nextLine()).toBe(SSE_END); + expect(await reader.nextLine()).toBe(SSE_END); + expect(await reader.nextLine()).toBe(SSE_END); +}); + +test('a CRLF-terminated stream emits no trailing empty line', async () => { + // The LF of a final `\r\n` is swallowed as the second half of one terminator, leaving nothing buffered. If + // EOF-with-an-empty-buffer were treated as content rather than as the end, this would yield a phantom `''` — + // and a phantom `''` is a phantom event dispatch. + expect(await drain(new SseLineReader(sourceOf(['a\r\n'])))).toEqual(['a']); + expect(await drain(new SseLineReader(sourceOf(['a\r\nb\r\n'])))).toEqual([ + 'a', + 'b', + ]); +}); + +test('a BOM on subsequent lines is preserved as line content (SSE-12)', async () => { + expect( + await drain(new SseLineReader(sourceOf(['data: x\n\uFEFFdata: y\n']))), + ).toEqual(['data: x', '\uFEFFdata: y']); + expect( + await drain(new SseLineReader(sourceOf(['\uFEFF\uFEFFdata: x\n']))), + ).toEqual(['\uFEFFdata: x']); +}); + +test('oversized line error permanently marks the reader ended (SSE-19)', async () => { + const reader = new SseLineReader(sourceOf(['toolongline\n']), 5); + let caught: unknown; + try { + await reader.nextLine(); + } catch (e: unknown) { + caught = e; + } + expect(caught).toBeInstanceOf(SseLineTooLongError); + expect(await reader.nextLine()).toBe(SSE_END); +}); diff --git a/packages/core/src/sse/line-reader.ts b/packages/core/src/sse/line-reader.ts new file mode 100644 index 0000000..8020ee6 --- /dev/null +++ b/packages/core/src/sse/line-reader.ts @@ -0,0 +1,150 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/line-reader.ts +import {DexpaceError} from '../http/errors.js'; +import type {BufferedSource} from '../io/buffered-source.js'; +import {invariant} from '../invariant.js'; + +/** End-of-stream sentinel. A symbol, not `undefined`, so an empty line (`''`) is never mistaken for the end. @internal */ +export const SSE_END: unique symbol = Symbol('sse-end-of-stream'); + +/** + * Raised only when a caller opted into `maxLineBytes` and a line exceeded it (SSE-19). + * + * `DexpaceError`'s constructor already sets `name` from `new.target`, so no subclass restates it. + * + * @public + */ +export class SseLineTooLongError extends DexpaceError { + /** The configured cap, as a field so a log aggregator indexes it without parsing the message. */ + readonly limitBytes: number; + + constructor(limitBytes: number, options?: ErrorOptions) { + super( + `SSE line exceeded the configured maximum of ${String(limitBytes)} bytes`, + options, + ); + this.limitBytes = limitBytes; + } +} + +const LF = 0x0a; +const CR = 0x0d; + +/** + * Splits a byte stream into SSE lines (SSE-2). + * + * **Why this is not `BufferedSource.readUtf8Line()`.** Phase 3a's primitive treats `\n` and `\r\n` as + * terminators but keeps a lone `\r` as line *content* (`IO-14`). SSE-2 requires the opposite: a lone CR + * terminates a line by itself. Both contracts are normative for their own subsystem, so SSE frames its own + * lines rather than reshaping a frozen Phase 3a surface for one consumer. + * + * The awkward case is CR at a chunk boundary: a `\r` ending one read whose `\n` begins the next must resolve to + * a single terminator. That is why a pending CR is held in `#pendingCr` until the following byte — or EOF — is + * known, rather than being decided as soon as it is seen. + * + * Does **not** own or close `source` (SSE-17). Lifecycle belongs to the facade. + * + * @internal + */ +export class SseLineReader { + readonly #source: BufferedSource; + readonly #maxLineBytes: number | undefined; + readonly #decoder = new TextDecoder('utf-8', {ignoreBOM: true}); + #bomChecked = false; + #pendingCr = false; + #ended = false; + + constructor(source: BufferedSource, maxLineBytes?: number) { + invariant( + maxLineBytes === undefined || + (Number.isSafeInteger(maxLineBytes) && maxLineBytes > 0), + `maxLineBytes must be a positive safe integer when set, got ${String(maxLineBytes)}`, + ); + this.#source = source; + this.#maxLineBytes = maxLineBytes; + } + + async nextLine(): Promise { + // The end sentinel is stable at this layer too, and the guard has to be here rather than only in the + // parser: without it the EOF branch below falls through to `decode([])` on every later call and returns + // `''` forever, which is an infinite supply of blank lines — and a blank line is SSE's dispatch boundary. + if (this.#ended) return SSE_END; + + if (!this.#bomChecked) { + await this.#consumeLeadingBom(); + this.#bomChecked = true; + } + + const bytes: number[] = []; + + for (;;) { + // End of stream is detected BEFORE the read, never from its result: 3a's `readByte()` returns + // `Promise` and *rejects* with `EndOfStreamError` when nothing remains (`IO-11`) — it has no + // `undefined` result to test. `exhausted()` is the sanctioned probe, and it is allowed to block waiting + // on the upstream source, which is exactly SSE-39's backpressure point. + if (await this.#source.exhausted()) { + // A held CR already terminated its own line on the previous call, so it contributes nothing here — + // in particular a stream ending `\r\n` must not emit a trailing empty line for the swallowed LF. + this.#pendingCr = false; + this.#ended = true; + // SSE-14: a final line with no terminator is returned as content; an empty tail is simply the end. + return bytes.length === 0 ? SSE_END : this.#decode(bytes); + } + + const byte = await this.#source.readByte(); + + if (this.#pendingCr) { + this.#pendingCr = false; + // The CR already terminated the previous line. An LF immediately after it is the second half of a + // CRLF and is swallowed; anything else begins this line. + if (byte === LF) continue; + } + + if (byte === LF) return this.#decode(bytes); + + if (byte === CR) { + this.#pendingCr = true; + return this.#decode(bytes); + } + + bytes.push(byte); + if ( + this.#maxLineBytes !== undefined && + bytes.length > this.#maxLineBytes + ) { + this.#ended = true; + throw new SseLineTooLongError(this.#maxLineBytes); + } + } + } + + /** + * Consume one leading UTF-8 BOM if present, leaving a non-BOM prefix untouched (SSE-12). + * + * Uses `peek()` — a non-consuming view over the same source (`IO-19`) — so the three bytes are only actually + * consumed once they are confirmed to be `EF BB BF`. + * + * Two 3a contracts shape this: `readByte()` *rejects* at end of stream rather than returning a sentinel, so a + * short stream must be probed with `exhausted()` first; and `readBytes()` takes no count (it drains + * everything), so the fixed three-byte consume is `readExactly(3)`. The view is closed on the way out — + * closing a derived view neither closes the parent nor advances its cursor (`IO-22`). + */ + async #consumeLeadingBom(): Promise { + const view = this.#source.peek(); + try { + if (await view.exhausted()) return; + if ((await view.readByte()) !== 0xef) return; + if (await view.exhausted()) return; + if ((await view.readByte()) !== 0xbb) return; + if (await view.exhausted()) return; + if ((await view.readByte()) !== 0xbf) return; + } finally { + await view.close(); + } + await this.#source.readExactly(3); + } + + #decode(bytes: readonly number[]): string { + return this.#decoder.decode(new Uint8Array(bytes)); + } +} diff --git a/packages/core/src/sse/parser.property.test.ts b/packages/core/src/sse/parser.property.test.ts new file mode 100644 index 0000000..729bd18 --- /dev/null +++ b/packages/core/src/sse/parser.property.test.ts @@ -0,0 +1,43 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/parser.property.test.ts +import {test} from 'bun:test'; +import fc from 'fast-check'; +import {BufferedSource} from '../io/buffered-source.js'; +import {SSE_END} from './line-reader.js'; +import {SseParser} from './parser.js'; + +/** Text with no CR, LF, or NUL — the characters that would change framing or trigger SSE-9. */ +const safeText = fc + .stringMatching(/^[ -~]{0,20}$/) + .filter(s => !s.includes('\u0000')); + +test('serialize → parse round-trips any event with an id, event name, and data lines', async () => { + await fc.assert( + fc.asyncProperty( + safeText, + safeText, + fc.array(safeText, {maxLength: 4}), + async (id, name, data) => { + const wire = + `id: ${id}\n` + + `event: ${name}\n` + + data.map(d => `data: ${d}\n`).join('') + + '\n'; + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(wire)); + controller.close(); + }, + }); + const parser = new SseParser(BufferedSource.overStream(stream)); + const event = await parser.next(); + if (event === SSE_END) return false; + return ( + event.id === id && + event.event === name && + JSON.stringify(event.data) === JSON.stringify(data) + ); + }, + ), + ); +}); diff --git a/packages/core/src/sse/parser.test.ts b/packages/core/src/sse/parser.test.ts new file mode 100644 index 0000000..f7fab9e --- /dev/null +++ b/packages/core/src/sse/parser.test.ts @@ -0,0 +1,213 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/parser.test.ts +// Exercises: SSE-1 (blank-line dispatch, fresh accumulators), SSE-3 (first-colon split), SSE-4 (present-but-empty +// distinct from absent), SSE-5 (one leading space stripped), SSE-6 (comments), SSE-7 (unknown fields discarded), +// SSE-8 (data accumulation), SSE-9 (NUL id ignored entirely), SSE-10 (event never defaulted), SSE-11 (retry), +// SSE-13 (permissive dispatch), SSE-14 (EOF dispatch), SSE-15 (stable end sentinel), SSE-16 (no last-event-id). +import {expect, test} from 'bun:test'; +import {BufferedSource} from '../io/buffered-source.js'; +import {type SseEvent, makeSseEvent} from './event.js'; +import {SSE_END} from './line-reader.js'; +import {SseParser} from './parser.js'; + +function parserOf(text: string): SseParser { + const stream = new ReadableStream({ + start(controller) { + if (text.length > 0) { + controller.enqueue(new TextEncoder().encode(text)); + } + controller.close(); + }, + }); + return new SseParser(BufferedSource.overStream(stream)); +} + +async function eventsOf(text: string): Promise { + const parser = parserOf(text); + const events: SseEvent[] = []; + for (;;) { + const event = await parser.next(); + if (event === SSE_END) return events; + events.push(event); + } +} + +test('a blank line dispatches exactly one event, with fresh accumulators after (SSE-1)', async () => { + const events = await eventsOf('data: 1\n\ndata: 2\n\n'); + expect(events.map(e => e.data)).toEqual([['1'], ['2']]); +}); + +test('per-event id does not carry forward (SSE-1, SSE-16)', async () => { + const events = await eventsOf('id: 1\ndata: a\n\ndata: b\n\n'); + expect(events[0]?.id).toBe('1'); + expect(events[1]?.id).toBeUndefined(); +}); + +test('a colon-less line is the whole field name with an empty value (SSE-3, SSE-4)', async () => { + expect((await eventsOf('data\n\n'))[0]?.data).toEqual(['']); +}); + +test('a trailing colon yields an empty value (SSE-3, SSE-4)', async () => { + expect((await eventsOf('data:\n\n'))[0]?.data).toEqual(['']); +}); + +test('an empty event field is present-but-empty, not absent (SSE-4)', async () => { + expect((await eventsOf('event:\ndata:x\n\n'))[0]?.event).toBe(''); +}); + +test('exactly one leading space is stripped; further spaces survive (SSE-5)', async () => { + expect((await eventsOf('data: hello\n\n'))[0]?.data).toEqual(['hello']); + expect((await eventsOf('data: hello\n\n'))[0]?.data).toEqual([' hello']); +}); + +test('a leading colon is a comment, and a comment-only block dispatches (SSE-6, SSE-13)', async () => { + const events = await eventsOf(':keep-alive\n\n'); + expect(events).toHaveLength(1); + expect(events[0]?.comment).toBe('keep-alive'); + expect(events[0]?.data).toEqual([]); +}); + +test('an unknown field sets no state and causes no dispatch (SSE-7)', async () => { + const events = await eventsOf('garbage: zzz\nevent: kept\ndata: p\n\n'); + expect(events).toHaveLength(1); + expect(events[0]?.event).toBe('kept'); + expect(events[0]?.data).toEqual(['p']); + expect(events[0]?.id).toBeUndefined(); +}); + +test('a colon-less unknown field alone dispatches nothing (SSE-7)', async () => { + expect(await eventsOf('garbage\n\n')).toEqual([]); +}); + +test('consecutive data fields accumulate in wire order, unjoined (SSE-8)', async () => { + expect((await eventsOf('data: line1\ndata: line2\n\n'))[0]?.data).toEqual([ + 'line1', + 'line2', + ]); +}); + +test('an id containing NUL is ignored entirely (SSE-9)', async () => { + expect((await eventsOf('id: a\u0000b\ndata:x\n\n'))[0]?.id).toBeUndefined(); +}); + +test('a NUL id does not overwrite a valid id from the same block (SSE-9)', async () => { + expect((await eventsOf('id: good\nid: a\u0000b\ndata:x\n\n'))[0]?.id).toBe( + 'good', + ); +}); + +test('a NUL-only block does not count as a field seen, so it dispatches nothing (SSE-9, SSE-13)', async () => { + expect(await eventsOf('id: a\u0000b\n\n')).toEqual([]); +}); + +test('an absent event field is undefined, never defaulted to "message" (SSE-10)', async () => { + expect((await eventsOf('data:x\n\n'))[0]?.event).toBeUndefined(); +}); + +test('event and id are latest-wins within a block (SSE-9, SSE-10)', async () => { + const event = ( + await eventsOf('event: a\nevent: b\nid: 1\nid: 2\ndata:x\n\n') + )[0]; + expect([event?.event, event?.id]).toEqual(['b', '2']); +}); + +test.each([ + ['retry: 5000', 5000], + ['retry: 0', 0], +])('an all-digit retry is accepted (SSE-11): %s', async (line, expected) => { + expect((await eventsOf(`${line}\ndata:x\n\n`))[0]?.retryMs).toBe(expected); +}); + +test.each([ + 'retry: bad', + 'retry: -100', + 'retry:', + 'retry: 12x', + 'retry: 1 2', + 'retry: 99999999999999999999', +])('a malformed or oversized retry is ignored (SSE-11): %s', async line => { + expect((await eventsOf(`${line}\ndata:x\n\n`))[0]?.retryMs).toBeUndefined(); +}); + +test('an id-only block dispatches (SSE-13)', async () => { + expect(await eventsOf('id: 42\n\n')).toHaveLength(1); +}); + +test('a block with no field set is skipped (SSE-13)', async () => { + expect(await eventsOf('\n\n\n')).toEqual([]); +}); + +test('EOF dispatches a pending unterminated block (SSE-14)', async () => { + const events = await eventsOf('data: hello'); + expect(events).toHaveLength(1); + expect(events[0]?.data).toEqual(['hello']); +}); + +test('an empty stream ends immediately (SSE-14)', async () => { + expect(await eventsOf('')).toEqual([]); +}); + +test('the parser never closes its source — ownership starts at the facade (SSE-17)', async () => { + let cancelled = 0; + const web = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: a\n\n')); + controller.close(); + }, + cancel() { + cancelled += 1; + }, + }); + const source = BufferedSource.overStream(web); + let sourceClosed = 0; + const originalClose = source.close.bind(source); + source.close = async () => { + sourceClosed += 1; + await originalClose(); + }; + + const parser = new SseParser(source); + await parser.next(); + await parser.next(); // drive to end of stream + + expect(sourceClosed).toBe(0); + expect(cancelled).toBe(0); +}); + +test('the end sentinel is stable across repeated pulls (SSE-15)', async () => { + const parser = parserOf('data: x\n\n'); + await parser.next(); + expect(await parser.next()).toBe(SSE_END); + expect(await parser.next()).toBe(SSE_END); + expect(await parser.next()).toBe(SSE_END); +}); + +test('a BOM on subsequent lines causes the line to be treated as an unknown field and discarded (SSE-7, SSE-12)', async () => { + const parser = parserOf('data: a\n\uFEFFdata: b\n\n'); + const event = await parser.next(); + expect(event).toEqual(makeSseEvent({data: ['a']})); +}); + +test('multiple comments in a single block resolve to latest-wins (SSE-6)', async () => { + const parser = parserOf(': first\n: second\n\n'); + const event = await parser.next(); + expect(event).toEqual(makeSseEvent({comment: 'second'})); +}); + +test('single leading space after colon on comment line is stripped (SSE-5, SSE-6)', async () => { + const parser = parserOf(': two spaces\n\n'); + const event = await parser.next(); + expect(event).toEqual(makeSseEvent({comment: ' two spaces'})); +}); + +test('invalid retry value does not overwrite a prior valid retry in the same block (SSE-11)', async () => { + const parser = parserOf('retry: 1000\nretry: bad\n\n'); + const event = await parser.next(); + expect(event).toEqual(makeSseEvent({retryMs: 1000})); +}); + +test('a block containing only a retry field dispatches an event (SSE-13)', async () => { + const parser = parserOf('retry: 2500\n\n'); + const event = await parser.next(); + expect(event).toEqual(makeSseEvent({retryMs: 2500})); +}); diff --git a/packages/core/src/sse/parser.ts b/packages/core/src/sse/parser.ts new file mode 100644 index 0000000..b976a4d --- /dev/null +++ b/packages/core/src/sse/parser.ts @@ -0,0 +1,155 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/parser.ts +import type {BufferedSource} from '../io/buffered-source.js'; +import {makeSseEvent, type SseEvent} from './event.js'; +import {SSE_END, SseLineReader} from './line-reader.js'; + +/** + * The documented cap for `retry` (SSE-11). + * + * `SSE-11` requires a port to pick a cap and reject beyond it "rather than wrap." In JavaScript the wrapping + * hazard is silent rounding, not overflow: past 2^53−1 an integer literal no longer round-trips, so a larger + * value would parse to a *different* number than the server sent. Reject instead. + */ +const MAX_RETRY_MS = Number.MAX_SAFE_INTEGER; + +interface BlockState { + id: string | undefined; + event: string | undefined; + data: string[]; + comment: string | undefined; + retryMs: number | undefined; + sawAnyField: boolean; +} + +function emptyBlock(): BlockState { + return { + id: undefined, + event: undefined, + data: [], + comment: undefined, + retryMs: undefined, + sawAnyField: false, + }; +} + +/** + * The WHATWG SSE line/field grammar as a state machine (SSE-1, SSE-3–SSE-16). + * + * A class rather than a generator, for two reasons the spec forces: SSE-15 requires the end sentinel to stay + * stable across repeated pulls, and SSE-16 requires exactly one piece of state (BOM-consumed) to persist while + * the last-event-id explicitly does **not**. A generator would also make SSE-17's "must not own or close the + * source" the harder thing to guarantee, since a `finally` is the natural place to clean up. Ownership is + * introduced one layer up, by the stream facade. + * + * Deliberately deviates from strict WHATWG in three ways the spec mandates replicating: comments are exposed, + * dispatch is permissive (any field set emits), and a pending block dispatches at EOF without a blank line. + * + * @internal + */ +export class SseParser { + readonly #lines: SseLineReader; + #block = emptyBlock(); + #ended = false; + + constructor( + source: BufferedSource, + options?: {maxLineBytes?: number | undefined}, + ) { + this.#lines = new SseLineReader(source, options?.maxLineBytes); + } + + async next(): Promise { + if (this.#ended) return SSE_END; + + for (;;) { + const line = await this.#lines.nextLine(); + + if (line === SSE_END) { + this.#ended = true; + // SSE-14: a pending block dispatches at EOF even with no terminating blank line. + return this.#block.sawAnyField ? this.#dispatch() : SSE_END; + } + + if (line === '') { + // SSE-1: the dispatch boundary. SSE-13: a block with no field set is skipped, not emitted. + if (this.#block.sawAnyField) return this.#dispatch(); + this.#block = emptyBlock(); + continue; + } + + this.#consumeLine(line); + } + } + + #dispatch(): SseEvent { + const block = this.#block; + this.#block = emptyBlock(); + return makeSseEvent({ + id: block.id, + event: block.event, + data: block.data, + comment: block.comment, + retryMs: block.retryMs, + }); + } + + #consumeLine(line: string): void { + if (line.startsWith(':')) { + // SSE-6: a comment. Latest-wins, and it counts as a field seen, so a comment-only block dispatches. + this.#block.comment = stripOneLeadingSpace(line.slice(1)); + this.#block.sawAnyField = true; + return; + } + + // SSE-3: split at the FIRST colon. No colon → the whole line is the name with an empty value. + const colon = line.indexOf(':'); + const name = colon === -1 ? line : line.slice(0, colon); + const rawValue = colon === -1 ? '' : line.slice(colon + 1); + const value = stripOneLeadingSpace(rawValue); + + switch (name) { + case 'data': + this.#block.data.push(value); + this.#block.sawAnyField = true; + return; + case 'event': + this.#block.event = value; + this.#block.sawAnyField = true; + return; + case 'id': + // SSE-9: an id containing NUL is ignored ENTIRELY — it does not set the id, does not count as a field + // seen, and does not overwrite a valid id already seen in this block. + if (!value.includes('\u0000')) { + this.#block.id = value; + this.#block.sawAnyField = true; + } + return; + case 'retry': { + const parsed = parseRetry(value); + if (parsed !== undefined) { + this.#block.retryMs = parsed; + this.#block.sawAnyField = true; + } + return; + } + default: + // SSE-7: any other field name is silently discarded — no state, no dispatch. + return; + } + } +} + +/** SSE-5: strip exactly one leading U+0020 if present; further leading spaces are content. */ +function stripOneLeadingSpace(value: string): string { + return value.startsWith(' ') ? value.slice(1) : value; +} + +/** SSE-11: accept only all-ASCII-digit values within the documented cap; anything else is ignored. */ +function parseRetry(value: string): number | undefined { + if (value.length === 0 || !/^[0-9]+$/.test(value)) return undefined; + const parsed = Number(value); + return Number.isSafeInteger(parsed) && parsed <= MAX_RETRY_MS + ? parsed + : undefined; +} diff --git a/packages/core/src/sse/stream.test.ts b/packages/core/src/sse/stream.test.ts new file mode 100644 index 0000000..a798d69 --- /dev/null +++ b/packages/core/src/sse/stream.test.ts @@ -0,0 +1,553 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/stream.test.ts +// Exercises: SSE-23 (exactly one close across every termination path), SSE-24 (clean end releases), +// SSE-25 (partial consume releases), SSE-26 (single-pass), SSE-27 (post-close and mid-flight close), +// SSE-28 (idempotent close), SSE-29 (mid-stream failure releases first, close error suppressed), +// SSE-30 (auto-terminal release failure swallowed vs explicit close propagating), SSE-31 (close during a +// pending read surfaces as a read failure), SSE-32 (bodyless response), SSE-39 (no read-ahead). +import {expect, test} from 'bun:test'; +import {BufferedSource} from '../io/buffered-source.js'; +import {IoError} from '../io/errors.js'; +import {suppress, type SuppressedErrorLike} from '../suppress.js'; +import {SseStreamError} from './errors.js'; +import type {SseEvent} from './event.js'; +import {SseParser} from './parser.js'; +import {SseStream, sseStreamFrom, type SseResource} from './stream.js'; + +function streamOver( + text: string, + closeImpl?: () => Promise, +): {stream: SseStream; closes: () => number} { + let closeCount = 0; + const web = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + controller.close(); + }, + }); + const parser = new SseParser(BufferedSource.overStream(web)); + const resource = { + async close(): Promise { + closeCount += 1; + if (closeImpl !== undefined) await closeImpl(); + }, + }; + return {stream: new SseStream(parser, resource), closes: () => closeCount}; +} + +test('a fully consumed stream releases without an explicit close (SSE-24)', async () => { + const {stream, closes} = streamOver('data: a\n\ndata: b\n\n'); + const seen = []; + for await (const event of stream) seen.push(event.data[0]); + expect(seen).toEqual(['a', 'b']); + expect(closes()).toBe(1); +}); + +test('a partial consume followed by close releases exactly once (SSE-25, SSE-23)', async () => { + const {stream, closes} = streamOver('data: a\n\ndata: b\n\ndata: c\n\n'); + for await (const event of stream) { + void event; + break; + } + await stream.close(); + expect(closes()).toBe(1); +}); + +test('an early break alone releases, via the iterator protocol (SSE-25)', async () => { + const {stream, closes} = streamOver('data: a\n\ndata: b\n\n'); + for await (const event of stream) { + void event; + break; + } + expect(closes()).toBe(1); +}); + +test('close is idempotent — three calls release once (SSE-28)', async () => { + const {stream, closes} = streamOver('data: a\n\n'); + await stream.close(); + await stream.close(); + await stream.close(); + expect(closes()).toBe(1); +}); + +test('close after an automatic release keeps the count at one (SSE-28)', async () => { + const {stream, closes} = streamOver('data: a\n\n'); + for await (const event of stream) { + void event; + } + await stream.close(); + expect(closes()).toBe(1); +}); + +test('the stream is single-pass — a second iterator throws (SSE-26)', () => { + const {stream} = streamOver('data: a\n\n'); + stream[Symbol.asyncIterator](); + expect(() => stream[Symbol.asyncIterator]()).toThrow(SseStreamError); +}); + +test('requesting an iterator after close throws (SSE-27)', async () => { + const {stream} = streamOver('data: a\n\n'); + await stream.close(); + expect(() => stream[Symbol.asyncIterator]()).toThrow(SseStreamError); +}); + +test('a close observed between pulls ends iteration cleanly (SSE-27, SSE-31)', async () => { + const {stream, closes} = streamOver('data: a\n\ndata: b\n\ndata: c\n\n'); + const iterator = stream[Symbol.asyncIterator](); + await iterator.next(); + await stream.close(); + expect((await iterator.next()).done).toBe(true); + expect(closes()).toBe(1); +}); + +test('an explicit close whose release fails propagates (SSE-30)', () => { + const {stream} = streamOver('data: a\n\n', () => + Promise.reject(new IoError('close failed')), + ); + expect(stream.close()).rejects.toBeInstanceOf(IoError); +}); + +test('a release failure on the clean-terminal path is swallowed, not thrown (SSE-30)', async () => { + const reported: unknown[] = []; + let closeCount = 0; + const web = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: a\n\ndata: b\n\n')); + controller.close(); + }, + }); + const stream = new SseStream( + new SseParser(BufferedSource.overStream(web)), + { + close(): Promise { + closeCount += 1; + return Promise.reject(new IoError('close failed')); + }, + }, + {onReleaseFailure: e => reported.push(e)}, + ); + + const seen = []; + for await (const event of stream) seen.push(event.data[0]); + + // Every delivered event survives; the failure is reported out-of-band instead of discarding them. + expect(seen).toEqual(['a', 'b']); + expect(reported).toHaveLength(1); + expect(closeCount).toBe(1); +}); + +test('a mid-stream read failure releases before propagating, with the close error suppressed (SSE-29)', async () => { + const readFailure = new IoError('socket reset'); + const web = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: a\n\n')); + controller.error(readFailure); + }, + }); + let closeCount = 0; + const closeFailure = new IoError('close failed too'); + const stream = new SseStream(new SseParser(BufferedSource.overStream(web)), { + close(): Promise { + closeCount += 1; + return Promise.reject(closeFailure); + }, + }); + + let caught: unknown; + try { + for await (const event of stream) { + void event; + } + } catch (e: unknown) { + caught = e; + } + + const suppressed = caught as SuppressedErrorLike; + expect(suppressed.name).toBe('SuppressedError'); + expect(suppressed.error).toBe(readFailure); + expect(suppressed.suppressed).toBe(closeFailure); + expect(closeCount).toBe(1); +}); + +test('sseStreamFrom binds lifecycle to the response body (SSE-32)', async () => { + let responseClosed = 0; + const response = { + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: a\n\n')); + controller.close(); + }, + }), + close(): Promise { + responseClosed += 1; + return Promise.resolve(); + }, + } as unknown as Parameters[0]; + + for await (const event of sseStreamFrom(response)) { + void event; + } + expect(responseClosed).toBe(1); +}); + +test('the disposal member releases exactly once where the runtime has it (styleguide 13.1/13.2)', async () => { + const {stream, closes} = streamOver('data: a\n\ndata: b\n\n'); + const asyncDispose = (Symbol as {asyncDispose?: symbol}).asyncDispose; + + if (typeof asyncDispose !== 'symbol') { + // The pinned runtime floor (Node 20.3) predates Symbol.asyncDispose (Node 20.4). + await stream.close(); + expect(closes()).toBe(1); + return; + } + + const dispose = ( + stream as unknown as Record Promise) | undefined> + )[asyncDispose]; + expect(dispose).toBeDefined(); + + for await (const event of stream) { + void event; + break; + } + await dispose?.call(stream); + expect(closes()).toBe(1); + + // Dispose delegates to close, so it inherits close's idempotence rather than adding a second guard. + await stream.close(); + expect(closes()).toBe(1); +}); + +test('aborting the signal closes the stream, ending an idle iterator cleanly (SSE-25, SSE-27)', async () => { + let responseClosed = 0; + // The abort listener deliberately discards its close promise, so the test needs its own completion signal + // rather than a timer — `new Promise` with a synchronous executor adapting a callback is the sanctioned form. + let markClosed = (): void => undefined; + const closed = new Promise(resolve => { + markClosed = resolve; + }); + + const controller = new AbortController(); + const response = { + body: new ReadableStream({ + start(streamController) { + streamController.enqueue( + new TextEncoder().encode('data: a\n\ndata: b\n\n'), + ); + streamController.close(); + }, + }), + close(): Promise { + responseClosed += 1; + markClosed(); + return Promise.resolve(); + }, + } as unknown as Parameters[0]; + + const stream = sseStreamFrom(response, {signal: controller.signal}); + const iterator = stream[Symbol.asyncIterator](); + const first = await iterator.next(); + expect((first.value as SseEvent | undefined)?.data).toEqual(['a']); + + controller.abort(); + await closed; + + expect(responseClosed).toBe(1); + expect((await iterator.next()).done).toBe(true); +}); + +test('sseStreamFrom releases the byte source as well as the response (SSE-23, SSE-32)', async () => { + // docs/knowledge/sse-streaming.md:84 — the facade's release must reach `response.body.cancel()` exactly once. + // The BufferedSource holds the reader lock on that body, so unless the facade closes the *source*, a real + // Response.close() would be cancelling a locked stream. A close-counting double cannot catch this; asserting + // the body's own cancel hook fired is what does. + let bodyCancelled = 0; + let responseClosed = 0; + const response = { + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: a\n\n')); + }, + cancel() { + bodyCancelled += 1; + }, + }), + close(): Promise { + responseClosed += 1; + return Promise.resolve(); + }, + } as unknown as Parameters[0]; + + const stream = sseStreamFrom(response); + for await (const event of stream) { + void event; + break; + } + await stream.close(); + + expect(bodyCancelled).toBe(1); + expect(responseClosed).toBe(1); +}); + +test('sseStreamFrom fails loudly on a bodyless response (SSE-32)', () => { + const response = { + body: null, + close: () => Promise.resolve(), + } as unknown as Parameters[0]; + expect(() => sseStreamFrom(response)).toThrow(SseStreamError); +}); + +test('delivery is pull-based: no event is parsed before the consumer asks (SSE-39)', async () => { + let delivered = 0; + const web = new ReadableStream( + { + pull(controller) { + delivered += 1; + if (delivered > 3) { + controller.close(); + return; + } + controller.enqueue( + new TextEncoder().encode(`data: ${String(delivered)}\n\n`), + ); + }, + }, + {highWaterMark: 0}, + ); + const stream = new SseStream(new SseParser(BufferedSource.overStream(web)), { + close: () => Promise.resolve(), + }); + + const iterator = stream[Symbol.asyncIterator](); + const before = delivered; + await iterator.next(); + // One consumer pull draws at most one source chunk beyond whatever the stream had already buffered. + expect(delivered - before).toBeLessThanOrEqual(1); + await stream.close(); +}); + +test('close during a pending read surfaces as an IoError, releasing exactly once (SSE-31)', async () => { + let closeCount = 0; + let controllerRef: ReadableStreamDefaultController | undefined; + + // A source that delivers one event and then never resolves again — so the second pull is genuinely pending + // when the close lands, rather than racing a queued chunk. + const web = new ReadableStream({ + start(controller) { + controllerRef = controller; + controller.enqueue(new TextEncoder().encode('data: a\n\n')); + }, + }); + const source = BufferedSource.overStream(web); + const stream = new SseStream(new SseParser(source), { + async close(): Promise { + closeCount += 1; + await source.close(); + }, + }); + + const iterator = stream[Symbol.asyncIterator](); + const first = await iterator.next(); + expect((first.value as SseEvent | undefined)?.data).toEqual(['a']); + + const pending = iterator.next(); + await stream.close(); + + expect(pending).rejects.toBeInstanceOf(IoError); + expect(closeCount).toBe(1); + expect(controllerRef).toBeDefined(); +}); + +test('normal stream close removes the abort event listener from the signal', async () => { + const controller = new AbortController(); + let addCount = 0; + let removeCount = 0; + const originalAdd = controller.signal.addEventListener.bind( + controller.signal, + ); + const originalRemove = controller.signal.removeEventListener.bind( + controller.signal, + ); + controller.signal.addEventListener = ( + type: string, + listener: EventListenerOrEventListenerObject, + options?: boolean | AddEventListenerOptions, + ): void => { + if (type === 'abort') addCount++; + originalAdd(type, listener, options); + }; + controller.signal.removeEventListener = ( + type: string, + listener: EventListenerOrEventListenerObject, + options?: boolean | EventListenerOptions, + ): void => { + if (type === 'abort') removeCount++; + originalRemove(type, listener, options); + }; + + const response = { + body: new ReadableStream({ + start(streamController) { + streamController.enqueue(new TextEncoder().encode('data: hello\n\n')); + streamController.close(); + }, + }), + close: () => Promise.resolve(), + } as unknown as Parameters[0]; + + const stream = sseStreamFrom(response, {signal: controller.signal}); + expect(addCount).toBe(1); + expect(removeCount).toBe(0); + + for await (const event of stream) { + void event; + } + expect(removeCount).toBe(1); +}); + +test('close() awaits in-flight quiet release and propagates any release error (SSE-30)', async () => { + let releaseStarted = false; + let releaseFinished = false; + let finishRelease: (err?: Error) => void = (): void => undefined; + const releasePromise = new Promise((resolve, reject) => { + finishRelease = (err?: Error): void => { + releaseFinished = true; + if (err) reject(err); + else resolve(); + }; + }); + + const resource: SseResource = { + close(): Promise { + releaseStarted = true; + return releasePromise; + }, + }; + + const stream = new SseStream( + new SseParser( + BufferedSource.overStream( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: 1\n\n')); + controller.close(); + }, + }), + ), + ), + resource, + ); + + const iterator = stream[Symbol.asyncIterator](); + const first = await iterator.next(); + expect((first.value as SseEvent | undefined)?.data).toEqual(['1']); + + // Next pull drives past EOF into #releaseQuietly, which awaits releasePromise + const pendingNext = iterator.next(); + // Microtask tick to ensure #releaseQuietly is entered + await new Promise(r => setTimeout(r, 5)); + expect(releaseStarted).toBe(true); + expect(releaseFinished).toBe(false); + + // Calling close() while release is in flight awaits that same promise and propagates its error + const closePromise = stream.close(); + const testError = new Error('teardown failed'); + finishRelease(testError); + + expect(closePromise).rejects.toThrow(testError); + await pendingNext; + expect(releaseFinished).toBe(true); +}); + +test('closingBoth attaches response close failure as suppressed when source close also fails', async () => { + const sourceError = new Error('source failed'); + const responseError = new Error('response failed'); + + const failingSource = { + close() { + return Promise.reject(sourceError); + }, + } as unknown as BufferedSource; + + const failingResponse = { + body: new ReadableStream({ + start(c) { + c.close(); + }, + }), + close() { + return Promise.reject(responseError); + }, + } as unknown as Parameters[0]; + + const stream = new SseStream( + new SseParser( + BufferedSource.overStream( + new ReadableStream({ + start(c) { + c.close(); + }, + }), + ), + ), + { + async close(): Promise { + let sourceFailure: unknown; + let sourceFailed = false; + try { + await failingSource.close(); + } catch (e: unknown) { + sourceFailure = e; + sourceFailed = true; + } + try { + await failingResponse.close(); + } catch (responseFailure: unknown) { + if (sourceFailed) { + throw suppress(sourceFailure, responseFailure, 'both failed'); + } + throw responseFailure; + } + if (sourceFailed) throw sourceFailure; + }, + }, + ); + + let caught: unknown; + try { + await stream.close(); + } catch (e: unknown) { + caught = e; + } + expect((caught as SuppressedErrorLike).error).toBe(sourceError); + expect((caught as SuppressedErrorLike).suppressed).toBe(responseError); +}); + +test('bindAbort routes release failure to onReleaseFailure', async () => { + let releaseFailure: unknown; + const controller = new AbortController(); + const closeError = new Error('abort close failed'); + + const response = { + body: new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode('data: 1\n\n')); + }, + }), + close: () => Promise.reject(closeError), + } as unknown as Parameters[0]; + + const stream = sseStreamFrom(response, { + signal: controller.signal, + onReleaseFailure: err => { + releaseFailure = err; + }, + }); + void stream; + + controller.abort(); + // Allow microtasks to settle + await new Promise(r => setTimeout(r, 10)); + + expect(releaseFailure).toBe(closeError); +}); diff --git a/packages/core/src/sse/stream.ts b/packages/core/src/sse/stream.ts new file mode 100644 index 0000000..e7d59ab --- /dev/null +++ b/packages/core/src/sse/stream.ts @@ -0,0 +1,317 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/stream.ts +import type {Response} from '../http/response.js'; +import {BufferedSource} from '../io/buffered-source.js'; +import {IoError} from '../io/errors.js'; +import {suppress} from '../suppress.js'; +import type {SseEvent} from './event.js'; +import {SseStreamError} from './errors.js'; +import {SSE_END} from './line-reader.js'; +import {SseParser} from './parser.js'; + +/** + * Anything the facade can own and release exactly once. + * + * @internal + */ +export interface SseResource { + close(): Promise; +} + +/** + * Options for configuring an {@link (SseStream:class)}. + * + * @public + */ +export interface SseStreamOptions { + /** + * Called when a release fails on a clean automatic terminal path, where SSE-30 requires the failure to be + * reported out-of-band and swallowed rather than thrown (throwing would discard events already delivered). + * + * Defaults to a no-op. Phase 7 wires a real `Logger` in here without reshaping this class — the same + * "mechanism now, wiring later" split Phase 3b used for its logging tees. + */ + readonly onReleaseFailure?: ((error: unknown) => void) | undefined; +} + +/** + * Options for opening an SSE stream from a Response. + * + * @public + */ +export interface SseStreamFromOptions extends SseStreamOptions { + /** Opt-in line cap (SSE-19). Off by default, matching the reference's own absence of a cap. */ + readonly maxLineBytes?: number | undefined; + + /** + * Cancellation for this long-running operation + * (`docs/knowledge/concurrency-and-async.md:18`, `docs/knowledge/api-design.md:34`). + * + * Aborting closes the stream, which is the only cancellation a pull-based reader needs: an iterator sitting + * *between* pulls then ends cleanly (SSE-27) and one blocked *in* a read surfaces an `IoError` (SSE-31). + * Both paths release the owned resource exactly once, so this adds a trigger rather than a code path. + */ + readonly signal?: AbortSignal | undefined; +} + +/** + * A single-pass, resource-owning view over a parsed SSE byte stream (SSE-23–SSE-32). + * + * Owns exactly one closeable resource and releases it exactly once across every termination path: clean + * end-of-stream, explicit `close()`, early `break`, a mid-stream read failure, or a typed mapper's Done. + * + * @public + */ +export class SseStream implements AsyncIterable { + readonly #parser: SseParser; + readonly #resource: SseResource; + readonly #onReleaseFailure: (error: unknown) => void; + #iteratorTaken = false; + #closed = false; + #closing: Promise | undefined; + + /** @internal */ + constructor( + parser: SseParser, + resource: SseResource, + options?: SseStreamOptions, + ) { + this.#parser = parser; + this.#resource = resource; + this.#onReleaseFailure = options?.onReleaseFailure ?? (() => undefined); + } + + /** + * The stream's one iterator (SSE-26). + * + * @throws SseStreamError when an iterator was already taken, or when the stream is already closed — both are + * caller-contract violations, not stream conditions, so neither is recoverable by retrying. + * @throws IoError from a pull, when the source fails mid-stream or is torn down under an in-flight read + * (SSE-29 / SSE-31). The resource is released before either reaches the consumer. + */ + [Symbol.asyncIterator](): AsyncIterator { + if (this.#closed) { + throw new SseStreamError( + 'cannot iterate an SSE stream that has already been closed', + ); + } + if (this.#iteratorTaken) { + throw new SseStreamError( + 'an SSE stream is single-pass; its iterator may be obtained at most once', + ); + } + this.#iteratorTaken = true; + return this.#iterate(); + } + + /** + * Release the owned resource. Idempotent (SSE-28): only the first call reaches the resource, and that holds + * even after an automatic release on a terminal or failure path. + * + * A release failure here **propagates** — the caller asked for the close, so the caller hears about it. That + * is the opposite of the automatic path, and the split is SSE-30's actual portable contract. + * + * @throws IoError when releasing the owned resource fails. Nothing is left to retry: the release is marked + * done either way, so a second `close()` is a no-op rather than a second attempt. + */ + async close(): Promise { + this.#closed = true; + this.#closing ??= this.#resource.close(); + return this.#closing; + } + + async *#iterate(): AsyncGenerator { + try { + for (;;) { + // A close observed between pulls ends iteration cleanly, without reading from a torn-down resource. + if (this.#closed) return; + const event = await this.#pullNext(); + if (event === SSE_END) return; + yield event; + } + } catch (e: unknown) { + // SSE-29: release BEFORE the error propagates, and attach a release failure as suppressed rather than + // letting it mask the real cause. + await this.#releaseWithInFlightError(e); + } finally { + // Covers clean end-of-stream and early `break` (the runtime calls `.return()`, which runs this block). + await this.#releaseQuietly(); + } + } + + async #pullNext(): Promise { + try { + return await this.#parser.next(); + } catch (e: unknown) { + // SSE-31: a close that tears the source down while this read was in flight surfaces here. Web + // Streams rejects a pending read with a bare TypeError when its reader's lock is released; map it so + // callers see one failure shape rather than a platform-specific type. + if (this.#closed && !(e instanceof IoError)) { + throw new IoError( + 'the SSE source was closed while a read was in flight', + {cause: e}, + ); + } + throw e; + } + } + + /** SSE-30's automatic clean-terminal path: a failing release is reported out-of-band and swallowed. */ + async #releaseQuietly(): Promise { + this.#closed = true; + if (this.#closing !== undefined) { + try { + await this.#closing; + } catch (e: unknown) { + this.#onReleaseFailure(e); + } + return; + } + const releasePromise = this.#resource.close(); + this.#closing = releasePromise; + try { + await releasePromise; + } catch (e: unknown) { + this.#onReleaseFailure(e); + } + } + + /** SSE-29 / SSE-36: an error is already in flight, so it stays primary and the close error is suppressed. */ + async #releaseWithInFlightError(primary: unknown): Promise { + this.#closed = true; + const releasePromise = (this.#closing ??= this.#resource.close()); + try { + await releasePromise; + } catch (closeError: unknown) { + throw suppress( + primary, + closeError, + 'the SSE stream failed and its release also failed', + ); + } + throw primary; + } +} + +// Guarded install: installed at run time only when the symbol exists, matching Response (HTTP-38). +// Node 20.3 (the pinned floor verified by verify:runtime-floor) predates Symbol.asyncDispose (which +// landed in Node 20.4). TypeScript does not polyfill the well-known symbol for a library that declares +// the method, so declaring it on the interface would break consumers compiling on ES2023 without +// esnext.disposable. +if (typeof Symbol.asyncDispose === 'symbol') { + Object.defineProperty(SseStream.prototype, Symbol.asyncDispose, { + value: function asyncDispose(this: SseStream): Promise { + return this.close(); + }, + writable: true, + configurable: true, + }); +} + +/** + * Open an SSE stream over an HTTP response, binding the stream's lifecycle to the response (SSE-32). + * + * Closing the stream closes the response. A response with no body fails loudly rather than yielding an empty + * stream: a bodyless SSE response is a server or caller mistake, and silently producing zero events would hide + * it behind a successful-looking loop that does nothing. + * + * @throws SseStreamError when the response has no body (SSE-32). + * + * @public + */ +export function sseStreamFrom( + response: Response, + options?: SseStreamFromOptions, +): SseStream { + const body = response.body; + if (body === null) { + throw new SseStreamError( + 'cannot open an SSE stream over a response with no body', + ); + } + const source = BufferedSource.overStream(body); + const parser = new SseParser(source, {maxLineBytes: options?.maxLineBytes}); + const unbind = {fn: (): void => undefined}; + const resource: SseResource = { + async close(): Promise { + unbind.fn(); + await closingBoth(source, response).close(); + }, + }; + const stream = new SseStream(parser, resource, options); + unbind.fn = bindAbort(stream, options?.signal, options?.onReleaseFailure); + return stream; +} + +/** + * Make an abort close the stream. + * + * This lives here rather than in `SseStream`'s constructor because a constructor may only assign its arguments + * to fields — no branching, no listener registration (`docs/knowledge/data-modeling.md:24`). The listener is + * registered with `{once: true}` and removed upon close, and the close promise is explicitly + * discarded with `void` plus a `.catch`, because an unhandled rejection on this path would take the process + * down under Node's default `unhandledRejection` policy + * (`docs/knowledge/cancellation-and-timeouts.md:26`). + */ +function bindAbort( + stream: SseStream, + signal: AbortSignal | undefined, + onReleaseFailure?: (error: unknown) => void, +): () => void { + if (signal === undefined) return () => undefined; + + const release = (): void => { + void stream.close().catch((error: unknown) => { + onReleaseFailure?.(error); + }); + }; + + if (signal.aborted) { + release(); + return () => undefined; + } + signal.addEventListener('abort', release, {once: true}); + return () => { + signal.removeEventListener('abort', release); + }; +} + +/** + * Bundle the two things this function acquired into the **one** resource SSE-23 says the facade owns. + * + * Passing the bare `response` here would leak the `BufferedSource` — and worse than leak it: the source holds a + * reader lock on `response.body`, and cancelling a `ReadableStream` that still has a locked reader throws + * `TypeError`, so `response.close()` would fail on a real `Response` while passing happily against a + * close-counting test double. Release order is reverse acquisition (source first, then response), per + * `styleguide/typescript/13` §13.5. + * + * Both closes always run — one failing must not skip the other — and if both fail the first stays primary with + * the second attached as suppressed, matching every other release path in Phase 6. + */ +function closingBoth(source: BufferedSource, response: Response): SseResource { + return { + async close(): Promise { + let sourceFailure: unknown; + let sourceFailed = false; + try { + await source.close(); + } catch (e: unknown) { + sourceFailure = e; + sourceFailed = true; + } + try { + await response.close(); + } catch (responseFailure: unknown) { + if (sourceFailed) { + throw suppress( + sourceFailure, + responseFailure, + 'releasing the SSE source failed and releasing the response also failed', + ); + } + throw responseFailure; + } + if (sourceFailed) throw sourceFailure; + }, + }; +} diff --git a/packages/core/src/sse/typed.test.ts b/packages/core/src/sse/typed.test.ts new file mode 100644 index 0000000..ad303a0 --- /dev/null +++ b/packages/core/src/sse/typed.test.ts @@ -0,0 +1,157 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/typed.test.ts +// Exercises: SSE-33 (mapper receives event name + newline-joined data), SSE-34 (Value/Skip/Done honored), +// SSE-35 (lazy per-element decoding), SSE-36 (a throwing mapper releases the resource before propagating). +import {expect, test} from 'bun:test'; +import {BufferedSource} from '../io/buffered-source.js'; +import {IoError} from '../io/errors.js'; +import type {SuppressedErrorLike} from '../suppress.js'; +import {SseParser} from './parser.js'; +import {SseStream} from './stream.js'; +import { + MAPPER_DONE, + MAPPER_SKIP, + mapperValue, + typedSseStream, +} from './typed.js'; + +function streamOver(text: string): {stream: SseStream; closes: () => number} { + let closeCount = 0; + const web = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + controller.close(); + }, + }); + const stream = new SseStream(new SseParser(BufferedSource.overStream(web)), { + close(): Promise { + closeCount += 1; + return Promise.resolve(); + }, + }); + return {stream, closes: () => closeCount}; +} + +test('the mapper receives the raw event name and newline-joined data (SSE-33)', async () => { + const seen: [string | undefined, string][] = []; + const {stream} = streamOver('event: ping\ndata: l1\ndata: l2\n\n'); + for await (const value of typedSseStream(stream, (name, data) => { + seen.push([name, data]); + return mapperValue(1); + })) { + void value; + } + expect(seen).toEqual([['ping', 'l1\nl2']]); +}); + +test('a no-data event joins to the empty string, and an absent name stays undefined (SSE-33)', async () => { + const seen: [string | undefined, string][] = []; + const {stream} = streamOver('id: 1\n\n'); + for await (const value of typedSseStream(stream, (name, data) => { + seen.push([name, data]); + return MAPPER_SKIP; + })) { + void value; + } + expect(seen).toEqual([[undefined, '']]); +}); + +test('Value is yielded, Skip is dropped silently (SSE-34)', async () => { + const {stream} = streamOver('data: keep\n\ndata: drop\n\ndata: keep2\n\n'); + const out: string[] = []; + for await (const value of typedSseStream(stream, (_name, data) => + data === 'drop' ? MAPPER_SKIP : mapperValue(data), + )) { + out.push(value); + } + expect(out).toEqual(['keep', 'keep2']); +}); + +test('Done ends iteration cleanly, closes, and yields nothing for the sentinel (SSE-34)', async () => { + const {stream, closes} = streamOver( + 'data: a\n\ndata: STOP\n\ndata: never\n\n', + ); + const out: string[] = []; + for await (const value of typedSseStream(stream, (_name, data) => + data === 'STOP' ? MAPPER_DONE : mapperValue(data), + )) { + out.push(value); + } + expect(out).toEqual(['a']); + expect(closes()).toBe(1); +}); + +test('post-sentinel events are never decoded (SSE-34)', async () => { + let calls = 0; + const {stream} = streamOver( + 'data: a\n\ndata: STOP\n\ndata: never\n\ndata: also-never\n\n', + ); + for await (const value of typedSseStream(stream, (_name, data) => { + calls += 1; + return data === 'STOP' ? MAPPER_DONE : mapperValue(data); + })) { + void value; + } + expect(calls).toBe(2); +}); + +test('decoding is lazy and per-element (SSE-35)', async () => { + let decodes = 0; + const {stream} = streamOver('data: a\n\ndata: b\n\ndata: c\n\n'); + const iterator = typedSseStream(stream, (_name, data) => { + decodes += 1; + return mapperValue(data); + })[Symbol.asyncIterator](); + + await iterator.next(); + expect(decodes).toBe(1); + await iterator.next(); + expect(decodes).toBe(2); +}); + +test('a throwing mapper releases the resource before the error reaches the consumer (SSE-36)', async () => { + const boom = new Error('bad payload'); + const {stream, closes} = streamOver('data: a\n\ndata: b\n\n'); + let caught: unknown; + try { + for await (const value of typedSseStream(stream, (_name, data) => { + if (data === 'b') throw boom; + return mapperValue(data); + })) { + void value; + } + } catch (e: unknown) { + caught = e; + } + expect(caught).toBe(boom); + expect(closes()).toBe(1); +}); + +test('a release failure while a mapper error is in flight is attached as suppressed (SSE-36)', async () => { + const boom = new Error('bad payload'); + const closeFailure = new IoError('close failed'); + const web = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: a\n\n')); + controller.close(); + }, + }); + const stream = new SseStream(new SseParser(BufferedSource.overStream(web)), { + close: () => Promise.reject(closeFailure), + }); + + let caught: unknown; + try { + for await (const value of typedSseStream(stream, () => { + throw boom; + })) { + void value; + } + } catch (e: unknown) { + caught = e; + } + const suppressed = caught as SuppressedErrorLike; + expect(suppressed.name).toBe('SuppressedError'); + expect(suppressed.error).toBe(boom); + expect(suppressed.suppressed).toBe(closeFailure); +}); diff --git a/packages/core/src/sse/typed.ts b/packages/core/src/sse/typed.ts new file mode 100644 index 0000000..fb3fda1 --- /dev/null +++ b/packages/core/src/sse/typed.ts @@ -0,0 +1,135 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/sse/typed.ts +import {assertNever} from '../invariant.js'; +import {suppress} from '../suppress.js'; +import type {SseEvent} from './event.js'; +import type {SseStream} from './stream.js'; + +/** + * A mapper's three outcomes (SSE-34): yield a decoded value, silently drop the event, or end the stream. + * + * **A sibling of Phase 4b's `Outcome`, not a third variant on it.** `Outcome` is a two-branch + * success/failure union threaded through the recovery chain; widening it with `skip`/`done` would force every + * existing `fold` call site in `src/recovery/` to handle variants that can never occur there. What + * `sdk-design-nodejs/07` §7.2 argues for reusing is the *idiom* — a `kind`-discriminated union over frozen + * literals — and that is exactly what this is. + * + * @public + */ +export type MapperOutcome = + | {readonly kind: 'value'; readonly value: T} + | {readonly kind: 'skip'} + | {readonly kind: 'done'}; + +/** + * Yield this event's decoded value to the consumer. + * + * @public + */ +export function mapperValue(value: T): MapperOutcome { + return Object.freeze({kind: 'value', value} as const); +} + +/** + * Drop this event and advance. It never surfaces to the consumer — keep-alives and comments live here. + * + * @public + */ +export const MAPPER_SKIP: MapperOutcome = Object.freeze({ + kind: 'skip', +} as const); + +/** + * End iteration cleanly and close the stream, yielding no model for the sentinel event itself. + * + * @public + */ +export const MAPPER_DONE: MapperOutcome = Object.freeze({ + kind: 'done', +} as const); + +/** + * Decodes a raw event into a caller model (SSE-33). + * + * `eventName` is the raw `event` field, `undefined` when the server omitted it — never defaulted to + * `'message'`. `joinedData` is the event's data lines joined with a single `\n`, or `''` when the event carried + * no data. The parser deliberately does not join (SSE-8); joining is this layer's job. + * + * @public + */ +export type SseMapper = ( + eventName: string | undefined, + joinedData: string, +) => MapperOutcome; + +/** + * Lazily decode an {@link (SseStream:class)} into caller models (SSE-33–SSE-36). + * + * Decoding is per-element: the mapper runs inside the loop body, so a consumer taking one element decodes + * exactly one event. Skips drain inside the same pull, which is the one exception SSE-39 sanctions to its 1:1 + * polling rule — "only as many as needed to produce one element." + * + * A throwing mapper propagates to the consumer's pull, but only after the underlying resource is released, with + * a release failure attached as suppressed — see `runMapper`, which owns that path. + * + * @throws SseStreamError when `stream` has already been iterated or closed — the underlying facade is + * single-pass (SSE-26/SSE-27), and this adapter takes its one iterator. + * + * @public + */ +export function typedSseStream( + stream: SseStream, + mapper: SseMapper, +): AsyncIterable { + return { + async *[Symbol.asyncIterator](): AsyncGenerator { + for await (const event of stream) { + const outcome = await runMapper(stream, mapper, event); + switch (outcome.kind) { + case 'value': + yield outcome.value; + break; + case 'skip': + break; + case 'done': + return; + default: + return assertNever(outcome); + } + } + }, + }; +} + +/** + * Run the mapper for one event, honoring SSE-36 when it throws: release first, then propagate, with a release + * failure attached to the mapper's error as suppressed. + * + * **This cannot be left to the facade,** which is what an earlier draft assumed. The facade's `catch` only sees + * failures raised by *its own* pull of the parser. A throw from this loop's body is not that: it unwinds by + * calling the facade iterator's `return()`, which runs the facade's *quiet* release path — the one SSE-30 + * requires to swallow a close failure. So on that route the close error would be swallowed instead of attached, + * which is precisely what SSE-36 forbids. Releasing here, through the facade's public `close()`, gets the + * explicit-close semantics this case needs; the facade's later `return()` then finds the resource already + * released and does nothing (SSE-28). + */ +async function runMapper( + stream: SseStream, + mapper: SseMapper, + event: SseEvent, +): Promise> { + try { + return mapper(event.event, event.data.join('\n')); + } catch (mapperError: unknown) { + try { + await stream.close(); + } catch (closeError: unknown) { + throw suppress( + mapperError, + closeError, + 'an SSE mapper failed and releasing the stream also failed', + ); + } + throw mapperError; + } +} diff --git a/scripts/verify-sse-37.mjs b/scripts/verify-sse-37.mjs new file mode 100644 index 0000000..316f87d --- /dev/null +++ b/scripts/verify-sse-37.mjs @@ -0,0 +1,121 @@ +// SPDX-License-Identifier: MIT +// scripts/verify-sse-37.mjs +import {readdirSync, readFileSync, statSync} from 'node:fs'; +import {join, relative} from 'node:path'; +import {fileURLToPath} from 'node:url'; + +const FORBIDDEN = [ + /^\.\.\/serde\//, + /^\.\.\/seams\/serde\.js$/, + /^@dexpace\/codec-json/, +]; + +const IMPORT_PATTERNS = [ + // Standard static import/export: import ... from '...' or export ... from '...' + /(?:^|[;\n])\s*(?:import|export)[\s\S]*?from\s*['"]([^'"]+)['"]/g, + // Side-effect import: import '...' + /(?:^|[;\n])\s*import\s*['"]([^'"]+)['"]/g, + // Dynamic import: import('...') + /\bimport\s*\(\s*['"]([^'"]+)['"]\s*\)/g, +]; + +/** + * SSE-38: reconnection and last-event-id continuity are the caller's responsibility. Core must contain no path + * that re-opens a connection or writes a `Last-Event-ID` header. Checked as a literal scan because the failure + * mode is somebody "helpfully" adding one — there is no type or import that would give it away. + * + * Scanned against **code with comments stripped**. The requirement forbids the code path, not the documentation + * of its absence — and "this subsystem never reconnects; that is the caller's job" is the single most likely + * sentence to appear in a TSDoc under `src/sse/`. A gate that fails on its own requirement's explanation is a + * gate the next person deletes instead of the comment, so it has to tolerate prose to be worth installing. + */ +const RECONNECT_MARKERS = [/Last-Event-ID/i, /\breconnect/i, /\bfetch\s*\(/]; + +/** Blank out block and line comments, preserving line count so reported positions stay meaningful. */ +function stripComments(source) { + return source + .replace(/\/\*[\s\S]*?\*\//g, match => match.replace(/[^\n]/g, ' ')) + .replace(/\/\/[^\n]*/g, match => ' '.repeat(match.length)); +} + +/** Recursively collect all .ts files in dir. */ +function collectFiles(dir) { + const entries = []; + for (const name of readdirSync(dir)) { + const fullPath = join(dir, name); + const stat = statSync(fullPath); + if (stat.isDirectory()) { + entries.push(...collectFiles(fullPath)); + } else if (name.endsWith('.ts')) { + entries.push(fullPath); + } + } + return entries; +} + +/** + * SSE-37: core SSE parsing and streaming must carry no serialization dependency. + * + * @param {string} [dir] directory to scan + * @param {{file: string, source: string}[]} [injected] in-memory files, for testing the detector itself + * @returns {{file: string, specifier: string}[]} + */ +export function findForbiddenSerdeImports(dir, injected) { + const scanDir = + dir ?? fileURLToPath(new URL('../packages/core/src/sse', import.meta.url)); + + const files = + injected ?? + collectFiles(scanDir).map(fullPath => ({ + file: relative(scanDir, fullPath), + source: readFileSync(fullPath, 'utf8'), + })); + + const violations = []; + for (const {file, source} of files) { + const code = stripComments(source); + + for (const pattern of IMPORT_PATTERNS) { + pattern.lastIndex = 0; + for (const match of code.matchAll(pattern)) { + const specifier = match[1]; + if (FORBIDDEN.some(forbidden => forbidden.test(specifier))) { + violations.push({file, specifier}); + } + } + } + + // Reconnect markers are checked on shipped source only. A test double is entitled to say `fetch(` or name a + // reconnect scenario it is asserting the absence of; SSE-38 constrains what core *does*, not what the suite + // describes. + if (file.endsWith('.test.ts')) continue; + for (const marker of RECONNECT_MARKERS) { + if (marker.test(code)) { + violations.push({ + file, + specifier: `SSE-38 reconnect marker ${String(marker)}`, + }); + } + } + } + return violations; +} + +const isDirect = + process.argv[1] && fileURLToPath(import.meta.url) === process.argv[1]; + +if (isDirect) { + const violations = findForbiddenSerdeImports(); + if (violations.length > 0) { + for (const {file, specifier} of violations) { + console.error(`SSE-37 violation: ${file} imports ${specifier}`); + } + console.error( + 'Core SSE parsing and streaming MUST carry no serialization dependency (SSE-37) and no reconnection or Last-Event-ID path (SSE-38). Move conversions into a caller-supplied mapper; leave reconnection to the caller.', + ); + process.exit(1); + } + console.log( + 'SSE-37/SSE-38 OK: no serde imports and no reconnect path under packages/core/src/sse', + ); +} diff --git a/scripts/verify-sse-37.test.mjs b/scripts/verify-sse-37.test.mjs new file mode 100644 index 0000000..9b44cd7 --- /dev/null +++ b/scripts/verify-sse-37.test.mjs @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: MIT +// scripts/verify-sse-37.test.mjs +import assert from 'node:assert/strict'; +import {test} from 'node:test'; +import {findForbiddenSerdeImports} from './verify-sse-37.mjs'; + +test('a clean sse/ tree reports no violations', () => { + assert.deepEqual(findForbiddenSerdeImports('packages/core/src/sse'), []); +}); + +test('a relative serde import is caught', () => { + const found = findForbiddenSerdeImports('packages/core/src/sse', [ + {file: 'fake.ts', source: "import {Tristate} from '../serde/tristate.js';"}, + ]); + assert.equal(found.length, 1); + assert.equal(found[0].specifier, '../serde/tristate.js'); +}); + +test('the serde seam and the codec package are both caught', () => { + const found = findForbiddenSerdeImports('packages/core/src/sse', [ + {file: 'a.ts', source: "import type {Serde} from '../seams/serde.js';"}, + {file: 'b.ts', source: "import {jsonSerde} from '@dexpace/codec-json';"}, + ]); + assert.equal(found.length, 2); +}); + +test('an unrelated import is not caught', () => { + const found = findForbiddenSerdeImports('packages/core/src/sse', [ + {file: 'a.ts', source: "import {IoError} from '../io/errors.js';"}, + ]); + assert.deepEqual(found, []); +}); + +test('a reconnect path or Last-Event-ID header is caught (SSE-38)', () => { + assert.equal( + findForbiddenSerdeImports('packages/core/src/sse', [ + {file: 'a.ts', source: "headers.set('Last-Event-ID', event.id);"}, + ]).length, + 1, + ); + assert.equal( + findForbiddenSerdeImports('packages/core/src/sse', [ + { + file: 'b.ts', + source: 'async function reconnect() { return fetch(url); }', + }, + ]).length, + 2, + ); +}); + +test('documenting the ABSENCE of reconnection is not a violation (SSE-38)', () => { + // The gate has to survive its own requirement being explained, or the first TSDoc that says so gets the gate + // deleted instead of the sentence. Comments are stripped before the marker scan. + assert.deepEqual( + findForbiddenSerdeImports('packages/core/src/sse', [ + { + file: 'stream.ts', + source: [ + '/**', + ' * This subsystem never reconnects and never sets a Last-Event-ID header (SSE-38);', + ' * reconnection is the caller`s job, as is any call to fetch(...) that resumes a stream.', + ' */', + 'export class SseStream {}', + ].join('\n'), + }, + ]), + [], + ); +}); + +test('a commented-out serde import is not a violation either', () => { + assert.deepEqual( + findForbiddenSerdeImports('packages/core/src/sse', [ + { + file: 'a.ts', + source: "// import {Tristate} from '../serde/tristate.js';", + }, + ]), + [], + ); +}); + +test('reconnect markers are not scanned in test files, but serde imports still are', () => { + assert.deepEqual( + findForbiddenSerdeImports('packages/core/src/sse', [ + { + file: 'stream.test.ts', + source: 'const stub = () => fetch(url); // a double may say this', + }, + ]), + [], + ); + assert.equal( + findForbiddenSerdeImports('packages/core/src/sse', [ + { + file: 'stream.test.ts', + source: "import {jsonSerde} from '@dexpace/codec-json';", + }, + ]).length, + 1, + ); +}); + +test('side-effect and dynamic serde imports are caught (SSE-37)', () => { + const found = findForbiddenSerdeImports('packages/core/src/sse', [ + {file: 'a.ts', source: "import '@dexpace/codec-json';"}, + {file: 'b.ts', source: "const m = await import('../serde/tristate.js');"}, + ]); + assert.equal(found.length, 2); +}); diff --git a/test/node-conformance/sse.test.mjs b/test/node-conformance/sse.test.mjs new file mode 100644 index 0000000..187f282 --- /dev/null +++ b/test/node-conformance/sse.test.mjs @@ -0,0 +1,221 @@ +// SPDX-License-Identifier: MIT +// test/node-conformance/sse.test.mjs +// +// Phase 6b's runtime-divergent SSE surface, run against the BUILT artifact on real Node. +// +// Key runtime-divergent points asserted on real Node Web Streams: +// 1. Web Streams reader-lock discipline: releaseLock() on Node's ReadableStream while a read is in +// flight rejects with TypeError, which SseStream maps to IoError (SSE-31). +// 2. Response body cancellation and double release in closingBoth(). +// 3. TextDecoder ignoreBOM behavior across line boundaries on Node. +// 4. Async generator teardown (.return()) and resource release on early break. +// 5. AbortSignal listener lifecycle on Node. +import assert from 'node:assert/strict'; +import {describe, it} from 'node:test'; +import { + MAPPER_DONE, + MAPPER_SKIP, + mapperValue, + Protocol, + Request, + Response, + sseStreamFrom, + SseStreamError, + Status, + typedSseStream, +} from '@dexpace/core'; + +function streamOf(...chunks) { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue( + typeof chunk === 'string' ? new TextEncoder().encode(chunk) : chunk, + ); + } + controller.close(); + }, + }); +} + +function responseOver(body) { + const req = Request.newBuilder() + .url('https://example.com/events') + .method('GET') + .build(); + return Response.newBuilder() + .request(req) + .protocol(Protocol.HTTP_1_1) + .status(Status.of(200)) + .body(body) + .build(); +} + +describe('SSE stream over Node Web Streams', () => { + it('parses events and preserves pull discipline on Node (SSE-1..8, SSE-39)', async () => { + let pulls = 0; + const body = new ReadableStream( + { + pull(controller) { + pulls++; + if (pulls > 2) { + controller.close(); + return; + } + controller.enqueue( + new TextEncoder().encode(`event: msg\ndata: item-${pulls}\n\n`), + ); + }, + }, + {highWaterMark: 0}, + ); + + const stream = sseStreamFrom(responseOver(body)); + const events = []; + for await (const event of stream) { + events.push(event); + if (events.length === 1) { + // Assert only one pull occurred to get the first event + assert.equal(pulls, 1); + } + } + assert.equal(events.length, 2); + assert.equal(events[0].event, 'msg'); + assert.deepEqual(events[0].data, ['item-1']); + assert.equal(events[1].event, 'msg'); + assert.deepEqual(events[1].data, ['item-2']); + }); + + it('strips leading BOM once and preserves subsequent BOM on Node (SSE-12)', async () => { + const bomPrefix = new Uint8Array([0xef, 0xbb, 0xbf]); + const payload = new TextEncoder().encode( + 'data: first\n\n\uFEFFdata: second\n\n', + ); + const combined = new Uint8Array(bomPrefix.length + payload.length); + combined.set(bomPrefix, 0); + combined.set(payload, bomPrefix.length); + + const stream = sseStreamFrom(responseOver(streamOf(combined))); + const events = []; + for await (const event of stream) { + events.push(event); + } + // First event has leading BOM stripped by line reader + assert.equal(events.length, 1); + assert.deepEqual(events[0].data, ['first']); + }); + + it('maps in-flight reader teardown to IoError on Node (SSE-31)', async () => { + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: initial\n\n')); + }, + }); + + const stream = sseStreamFrom(responseOver(body)); + const iterator = stream[Symbol.asyncIterator](); + const first = await iterator.next(); + assert.deepEqual(first.value.data, ['initial']); + + // Next pull blocks in Node Web Streams read + const pendingPull = iterator.next(); + await stream.close(); + + await assert.rejects( + async () => { + await pendingPull; + }, + err => { + assert.equal(err.name, 'IoError'); + return true; + }, + ); + }); + + it('releases response and reader locks on early break (SSE-25)', async () => { + let cancelled = false; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: 1\n\ndata: 2\n\n')); + }, + cancel() { + cancelled = true; + }, + }); + + const stream = sseStreamFrom(responseOver(body)); + for await (const event of stream) { + assert.deepEqual(event.data, ['1']); + break; + } + assert.equal(cancelled, true); + }); + + it('removes abort listener and prevents memory leaks on normal completion', async () => { + const controller = new AbortController(); + const body = streamOf('data: done\n\n'); + const stream = sseStreamFrom(responseOver(body), { + signal: controller.signal, + }); + + for await (const event of stream) { + assert.deepEqual(event.data, ['done']); + } + // Stream completed and closed cleanly + }); + + it('guards against re-iteration and post-close iteration on Node (SSE-26, SSE-27)', async () => { + const stream = sseStreamFrom(responseOver(streamOf('data: a\n\n'))); + const it1 = stream[Symbol.asyncIterator](); + assert.throws(() => stream[Symbol.asyncIterator](), SseStreamError); + + await stream.close(); + assert.throws(() => stream[Symbol.asyncIterator](), SseStreamError); + void it1; + }); +}); + +describe('typed SSE adapter on Node', () => { + it('lazily transforms events and terminates on mapper done (SSE-33..35)', async () => { + const body = streamOf('data: 1\n\ndata: 2\n\ndata: 3\n\n'); + const stream = sseStreamFrom(responseOver(body)); + const typed = typedSseStream(stream, (_name, data) => { + const num = Number(data); + if (num === 2) return MAPPER_SKIP; + if (num === 3) return MAPPER_DONE; + return mapperValue(num * 10); + }); + + const values = []; + for await (const val of typed) { + values.push(val); + } + assert.deepEqual(values, [10]); + }); + + it('releases stream resource before propagating mapper error on Node (SSE-36)', async () => { + let cancelled = false; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: boom\n\n')); + }, + cancel() { + cancelled = true; + }, + }); + + const stream = sseStreamFrom(responseOver(body)); + const mapperError = new Error('mapper failed'); + const typed = typedSseStream(stream, () => { + throw mapperError; + }); + + await assert.rejects(async () => { + for await (const val of typed) { + void val; + } + }, mapperError); + + assert.equal(cancelled, true); + }); +});