From 6f847fbea96239d062c9203c7fa06ba0b5a57de1 Mon Sep 17 00:00:00 2001 From: Mohammad Wahbeh Date: Tue, 11 Aug 2026 10:05:21 +0300 Subject: [PATCH 1/2] =?UTF-8?q?feat(core):=20add=20I/O=20contracts=20?= =?UTF-8?q?=E2=80=94=20ByteQueue,=20BufferedSource/Sink,=20views,=20TeeSin?= =?UTF-8?q?k.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .changeset/io-contracts.md | 5 + bun.lock | 3 + ...26-07-24-phase3a-io-contracts-checklist.md | 9 +- .../2026-07-24-phase3a-io-contracts-design.md | 10 +- package.json | 1 + packages/core/src/invariant.ts | 28 ++ packages/core/src/io/buffered-sink.test.ts | 187 ++++++++++ packages/core/src/io/buffered-sink.ts | 145 ++++++++ packages/core/src/io/buffered-source.test.ts | 160 +++++++++ .../core/src/io/buffered-source.text.test.ts | 130 +++++++ packages/core/src/io/buffered-source.ts | 332 ++++++++++++++++++ .../core/src/io/buffered-source.views.test.ts | 134 +++++++ packages/core/src/io/byte-queue.bench.ts | 35 ++ .../core/src/io/byte-queue.property.test.ts | 69 ++++ packages/core/src/io/byte-queue.test.ts | 238 +++++++++++++ packages/core/src/io/byte-queue.ts | 244 +++++++++++++ packages/core/src/io/errors.test.ts | 62 ++++ packages/core/src/io/errors.ts | 77 ++++ packages/core/src/io/factories.test.ts | 87 +++++ packages/core/src/io/factories.ts | 99 ++++++ packages/core/src/io/index.ts | 31 ++ packages/core/src/io/limits.test.ts | 23 ++ packages/core/src/io/limits.ts | 26 ++ packages/core/src/io/pump.test.ts | 45 +++ packages/core/src/io/pump.ts | 32 ++ packages/core/src/io/retention-window.test.ts | 110 ++++++ packages/core/src/io/retention-window.ts | 168 +++++++++ packages/core/src/io/tee-sink.test.ts | 213 +++++++++++ packages/core/src/io/tee-sink.ts | 113 ++++++ .../core/src/io/test-support/fake-stream.ts | 71 ++++ .../core/src/io/test-support/rejection.ts | 22 ++ packages/core/tsconfig.build.json | 4 +- tsconfig.base.json | 3 +- 33 files changed, 2909 insertions(+), 7 deletions(-) create mode 100644 .changeset/io-contracts.md create mode 100644 packages/core/src/invariant.ts create mode 100644 packages/core/src/io/buffered-sink.test.ts create mode 100644 packages/core/src/io/buffered-sink.ts create mode 100644 packages/core/src/io/buffered-source.test.ts create mode 100644 packages/core/src/io/buffered-source.text.test.ts create mode 100644 packages/core/src/io/buffered-source.ts create mode 100644 packages/core/src/io/buffered-source.views.test.ts create mode 100644 packages/core/src/io/byte-queue.bench.ts create mode 100644 packages/core/src/io/byte-queue.property.test.ts create mode 100644 packages/core/src/io/byte-queue.test.ts create mode 100644 packages/core/src/io/byte-queue.ts create mode 100644 packages/core/src/io/errors.test.ts create mode 100644 packages/core/src/io/errors.ts create mode 100644 packages/core/src/io/factories.test.ts create mode 100644 packages/core/src/io/factories.ts create mode 100644 packages/core/src/io/index.ts create mode 100644 packages/core/src/io/limits.test.ts create mode 100644 packages/core/src/io/limits.ts create mode 100644 packages/core/src/io/pump.test.ts create mode 100644 packages/core/src/io/pump.ts create mode 100644 packages/core/src/io/retention-window.test.ts create mode 100644 packages/core/src/io/retention-window.ts create mode 100644 packages/core/src/io/tee-sink.test.ts create mode 100644 packages/core/src/io/tee-sink.ts create mode 100644 packages/core/src/io/test-support/fake-stream.ts create mode 100644 packages/core/src/io/test-support/rejection.ts diff --git a/.changeset/io-contracts.md b/.changeset/io-contracts.md new file mode 100644 index 0000000..36054cf --- /dev/null +++ b/.changeset/io-contracts.md @@ -0,0 +1,5 @@ +--- +"@dexpace/core": patch +--- + +Internal: byte-streaming primitives for product-spec §5 (IO-1–IO-42). No public API change. diff --git a/bun.lock b/bun.lock index 5ea441b..c138d38 100644 --- a/bun.lock +++ b/bun.lock @@ -15,6 +15,7 @@ "fast-check": "^3", "globals": "^17.8.0", "gts": "^7", + "mitata": "^1", "publint": "^0.3", "typescript": "^5.8", "typescript-eslint": "^8", @@ -483,6 +484,8 @@ "minimist-options": ["minimist-options@4.1.0", "", { "dependencies": { "arrify": "^1.0.1", "is-plain-obj": "^1.1.0", "kind-of": "^6.0.3" } }, "sha512-Q4r8ghd80yhO/0j1O3B2BjweX3fiHg9cdOwjJd2J76Q135c+NDxGCqdYKQ1SKBuFfgWbAUzBfvYjPUEeNgqN1A=="], + "mitata": ["mitata@1.0.34", "", {}, "sha512-Mc3zrtNBKIMeHSCQ0XqRLo1vbdIx1wvFV9c8NJAiyho6AjNfMY8bVhbS12bwciUdd1t4rj8099CH3N3NFahaUA=="], + "mri": ["mri@1.2.0", "", {}, "sha512-tzzskb3bG8LvYGFF/mDTpq3jpI6Q9wc3LEmBaghu+DdCssd1FakN7Bc0hVNmEyGq1bq3RgfkCb3cmQLpNPOroA=="], "ms": ["ms@2.1.3", "", {}, "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA=="], diff --git a/docs/superpowers/plans/2026-07-24-phase3a-io-contracts-checklist.md b/docs/superpowers/plans/2026-07-24-phase3a-io-contracts-checklist.md index ff74490..066b0ed 100644 --- a/docs/superpowers/plans/2026-07-24-phase3a-io-contracts-checklist.md +++ b/docs/superpowers/plans/2026-07-24-phase3a-io-contracts-checklist.md @@ -35,7 +35,7 @@ requirement ID in `docs/product-spec/05-i-o-contracts.md`, as dispositioned by |---|---|---|---|---| | IO-11 | MUST | `exhausted()`, single-byte read, count-less read of all remaining (empty when exhausted) | ✅ | Task 6 | | IO-12 | MUST | Exact-count read returns exactly N or fails; never short | ✅ | Task 6, asserted across chunk boundaries and on the short path | -| IO-13 | MUST | UTF-8 and explicit-charset reads, with symmetric write-side encodings | ✅ (read) / ⚠️ (write, bounded) | Task 7 (read: any `TextDecoder` label, ISO-8859-1 round-trip per the requirement's own conformance note), Task 9 (write: **UTF-8 and ISO-8859-1 only**). `TextEncoder` is UTF-8-only and `SEAM-1` forbids an encoding dependency, so full symmetry is unreachable; any other label throws rather than silently re-encoding. Ledgered deviation | +| IO-13 | MUST | UTF-8 and explicit-charset reads, with symmetric write-side encodings | ✅ (read) / ⚠️ (write, bounded) | Task 7 (read: any `TextDecoder` label), Task 9 (write: **UTF-8 and ISO-8859-1 only**), plus two `fast-check` round-trip property tests in `buffered-sink.test.ts` — sink-out/source-back through UTF-8, and through ISO-8859-1 asserting one byte per code point, which is what distinguishes an honored charset from a silent UTF-8 re-encoding. `TeeSink`'s own `writeUtf8`/`writeString` are asserted to mirror the primary's exact encoded bytes and to refuse an unsupported label identically. `TextEncoder` is UTF-8-only and `SEAM-1` forbids an encoding dependency, so full symmetry is unreachable; any other label throws rather than silently re-encoding. Ledgered deviation | | IO-14 | MUST | Line read consumes the terminator; `\n` and `\r\n` both terminate; lone `\r` is content; final unterminated line as-is; absent when exhausted first | ✅ | Task 7, including a `fast-check` property test with **adversarially generated chunk boundaries**, so a terminator straddling two stream chunks is covered — the case the requirement's rationale names and hand-picked examples miss | | IO-15 | MUST | Skip advances exactly N, fails if fewer remain; `skip(0)` a no-op even at/after EOF | ✅ | Task 6 | | IO-16 | SHOULD | Read-only host-native byte-stream bridge; symmetric writable bridge; closing the bridge closes the owner | ✅ | Task 12. Host-native means `ReadableStream`/`WritableStream` for this port, per `sdk-design/03` §3.1 — no `node:` import; Task 13 Step 9 greps to enforce that | @@ -61,7 +61,7 @@ requirement ID in `docs/product-spec/05-i-o-contracts.md`, as dispositioned by | IO-26 | MUST | Tap capacity limit; default effectively unbounded; a limit of 0 mirrors nothing while forwarding everything | ✅ | Task 10 (`Number.POSITIVE_INFINITY` default, spelled as a value rather than a magic number); all three cases asserted | | IO-27 | MUST | Mirror BEFORE forwarding; clear staging even on a failed write so no stale bytes prepend | ✅ | Task 10, both clauses asserted; staging cleared in a `finally` so it holds on the throwing path | | IO-28 | MUST | No direct backing-buffer handle; attempting it fails, directing callers at the typed writes | ✅ | Task 10 (`get buffer(): never`) | -| IO-29 | MUST | Tee's own flush/close/emit forward to the PRIMARY only, leaving the tap intact | ✅ | Task 10, with snapshot-after-close asserted | +| IO-29 | MUST | Tee's own flush/close/emit forward to the PRIMARY only, leaving the tap intact | ✅ | Task 10. All three asserted: `close` with snapshot-after-close, and `flush`/`emit` both by returning the tee with the tap intact and — the observable proof they are not swallowed by the decorator — by rejecting with `ClosedResourceError` once the primary is closed, which only the primary can raise | ## 5.6 Provider factories, timeouts, and thread-safety @@ -80,10 +80,11 @@ requirement ID in `docs/product-spec/05-i-o-contracts.md`, as dispositioned by | Nothing enters the published API surface | Design decision (styleguide 10.3, Phase 2's `Serde` precedent) | ✅ | Task 13 Step 8 — `git diff --exit-code packages/core/etc/core.api.md` must produce no output. Mechanical proof, not a review promise | | No runtime dependency added | `SEAM-1` | ✅ | Task 13 Step 7 runs `verify:seam-1`; `mitata` is a root devDependency only | | No `node:` import in core | `sdk-design/03` §3.1, runtime-agnosticism | ✅ | Task 13 Step 9 greps `packages/core/src/` and fails on any match | -| Property tests where invariants exist | styleguide 11.5 | ✅ | Task 4 (`ByteQueue` ×4), Task 7 (`readUtf8Line`), Task 8 (views ×2), Task 10 (`TeeSink` wire payload) | +| Property tests where invariants exist | styleguide 11.5 | ✅ | Task 4 (`ByteQueue` ×4), Task 7 (`readUtf8Line`), Task 8 (views ×2), Task 9 (charset round-trips ×2), Task 10 (`TeeSink` wire payload) | +| Rejection assertions are awaited and attributable | styleguide 11.9 | ✅ | `test-support/rejection.ts`. bun types `.rejects.toThrow()` as `void`, so the plan's `await expect(…).rejects` form fails `@typescript-eslint/await-thenable`; the helper awaits the promise and returns the reason instead, with no `eslint-disable`. Ledgered | | Negative-space and cleanup assertions | styleguide 11.9, 13.9 | ✅ | Idempotent close (Tasks 4, 5, 6, 9), both IO-42 directions (Tasks 4, 6), parent-close invalidation (Task 8), failed-write tap capture (Task 10) | | Determinism — no fake clocks needed | styleguide 11.8 | ✅ | IO-40 means this layer owns no timer; every stream under test is built from an in-memory array | -| Fakes over `mock.module` | styleguide 11.3 | ✅ | Task 5's `test-support/fake-stream.ts`, excluded from the build via `tsconfig.build.json` | +| Fakes over `mock.module` | styleguide 11.3 | ✅ | Task 5's `test-support/fake-stream.ts` and `test-support/rejection.ts`, both excluded from the build via `tsconfig.build.json`'s `src/io/test-support/**` | | No type-level tests | styleguide 11.6 | ✅ (correctly absent) | 11.6 requires them for public generics and conditional types; this phase publishes neither. Stated rather than manufactured | | Committed baseline bench | styleguide 15.6 | ✅ | Task 13, `byte-queue.bench.ts`. Baseline only — no optimization applied, no 15.10 ledger notes, per 15.1/15.6's "do not tune ahead of a profile" | | 80% aggregate coverage floor | `NFR-5` | ✅ | Task 13 Step 7 | diff --git a/docs/superpowers/specs/2026-07-24-phase3a-io-contracts-design.md b/docs/superpowers/specs/2026-07-24-phase3a-io-contracts-design.md index a61994c..79906b3 100644 --- a/docs/superpowers/specs/2026-07-24-phase3a-io-contracts-design.md +++ b/docs/superpowers/specs/2026-07-24-phase3a-io-contracts-design.md @@ -399,6 +399,10 @@ layer where the temptation to dump the offending bytes into the message is stron | `TeeSink` as a sink decorator | `sdk-design/03` §3.1 phrasing | `TransformStream` queueing muddies `IO-27`'s mirror-before-forward ordering; §3.1's substantive point is untouched | | `IO-30` resolution half, `IO-39` not built | product-spec §5.6 | No registry exists — same class as `SEAM-5`–`SEAM-10` | | `IO-38` not applicable | product-spec §5.4 | The requirement is about a close on one thread invalidating a slice being read on another, so it presupposes an instance can reach a second thread. None can. **Class instances are not structured-cloneable at all** — `postMessage`/`structuredClone` preserve neither prototypes nor `#private` fields, so a `ByteQueue` or `BufferedSource` sent to a worker arrives as a plain object with no methods and no close state to observe. `BufferedSource` is doubly excluded: a `ReadableStreamDefaultReader` is neither cloneable nor transferable. A raw `ArrayBuffer` *can* be transferred, but it carries no close state and derives no slices, so the hazard has no subject | +| `"DOM.AsyncIterable"` added to `tsconfig.base.json`'s `lib` | Phase 2's `lib: ["ES2022", "DOM"]` baseline | `IO-16`'s `toReadableStream()` returns a `ReadableStream`, and asserting it with `for await (const chunk of …)` needs the async-iteration declarations, which TypeScript ships in a separate `lib` entry from `DOM`. Workspace-wide because the `lib` array is; no runtime effect and no new dependency (`SEAM-1` untouched), and the API report is unchanged. The alternative — driving the bridge test through `getReader()` — was rejected because async iteration is how a consumer will actually use the bridge, so the test should exercise that path | +| `packages/core/src/invariant.ts` created in this phase | The plan's prerequisite, which lists `invariant` as existing from Phase 1 | Phase 1 shipped `requireField` for HTTP-4's required-field message, not a general assertion primitive, so `invariant` (styleguide 5.6, 8.7) did not exist. `IO-3`, `IO-10`, and `IO-21` all need it, so it was added here as an `@internal` module with `InvariantViolation` as its own class. Nothing in `src/http/` was changed to route through it — Phase 1's `requireField` still owns HTTP-4's message | +| `ByteQueue.takeBytes` checks `MAX_BYTE_ARRAY_LENGTH` *before* the short-source check | The plan's Task 3 code, which checked size first | With the plan's ordering, an over-limit request on a short queue raised `EndOfStreamError`, hiding the real problem, and the plan's own `IO-9` test (`takeBytes(MAX + 1)` expects `AllocationLimitError`) could not pass. `IO-9`'s actionable-refusal requirement wins over reporting a size mismatch that is a consequence of the over-limit ask | +| Rejection assertions go through a `rejection()` test helper, not `await expect(…).rejects.toThrow(…)` | The plan's test code, which used `await expect(…).rejects` throughout | bun types `rejects` as `Matchers` whose `toThrow()` returns `void`, even though at run time it returns a promise. So the plan's form fails this repo's type-aware `@typescript-eslint/await-thenable` gate, and dropping the `await` to satisfy lint leaves the assertion racing test teardown — bun still fails the run, but the failure can attribute to a later test. `test-support/rejection.ts` awaits the promise, returns the rejection reason, and fails loudly if the promise resolves, so every assertion is awaited and attributable with no `eslint-disable` | | Write-side charsets limited to UTF-8 and ISO-8859-1 | `IO-13`'s "symmetric write-side encodings" | `TextEncoder` is UTF-8-only and `SEAM-1` forbids an encoding dependency. Read side stays fully general via `TextDecoder`; the write side covers the two encodings HTTP needs, and `IO-13`'s own conformance note names ISO-8859-1 as the non-UTF-8 case. Any other label throws rather than silently corrupting bytes. The `writeUtf8(begin, end)` substring-range overload is subsumed by `String.prototype.slice` at the call site | ## Testing @@ -416,7 +420,11 @@ invariant-bearing functions, and §5 is almost nothing else: covered; `IO-14`'s own rationale calls out surviving slice-window boundaries, and that is exactly the case hand-picked examples miss. - **`readString`/`writeString`** — round-trip through UTF-8 and through ISO-8859-1 (`IO-13`, whose conformance note - names a non-UTF-8 charset explicitly). + names a non-UTF-8 charset explicitly). The ISO-8859-1 generator excludes code points `0x80`–`0x9F`: the WHATWG + Encoding Standard maps the label `iso-8859-1` onto windows-1252, so `TextDecoder` returns U+20AC for `0x80` + rather than U+0080. That is the platform's asymmetry, not the sink's — the write side is a straight + code-point-to-byte map — and HTTP needs none of those C1 controls. Recorded here so Phase 9 does not read the + excluded band as an untested gap. - **View independence** — N views at arbitrary offsets and counts each read the same bytes a direct read at that window would, and no view's read advances another's cursor (`IO-19`, `IO-20`, `IO-23`). - **`TeeSink`** — for arbitrary write sequences and arbitrary tap caps, the primary receives the exact concatenation diff --git a/package.json b/package.json index 6aeafe9..9bfba61 100644 --- a/package.json +++ b/package.json @@ -16,6 +16,7 @@ "fast-check": "^3", "globals": "^17.8.0", "gts": "^7", + "mitata": "^1", "publint": "^0.3", "typescript": "^5.8", "typescript-eslint": "^8" diff --git a/packages/core/src/invariant.ts b/packages/core/src/invariant.ts new file mode 100644 index 0000000..bec6ce9 --- /dev/null +++ b/packages/core/src/invariant.ts @@ -0,0 +1,28 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/invariant.ts + +/** + * Thrown by {@link invariant} when a broken precondition or postcondition is detected. + * + * Its own class distinguishes a programmer error — a violated invariant — from an operational + * failure a caller might recover from (styleguide 5.6, 8.7). + * + * @internal + */ +export class InvariantViolation extends Error { + override readonly name = 'InvariantViolation'; +} + +/** + * The project's single sanctioned assertion primitive (styleguide 5.6). + * + * A TypeScript assertion function: after `invariant(x !== undefined, msg)`, `x` narrows to exclude + * `undefined` for the rest of the scope. Used for preconditions and postconditions — broken + * invariants, never operational failures a caller might recover from, which go through the typed + * error tree instead. + * + * @internal + */ +export function invariant(cond: unknown, msg: string): asserts cond { + if (!cond) throw new InvariantViolation(msg); +} diff --git a/packages/core/src/io/buffered-sink.test.ts b/packages/core/src/io/buffered-sink.test.ts new file mode 100644 index 0000000..a0e041c --- /dev/null +++ b/packages/core/src/io/buffered-sink.test.ts @@ -0,0 +1,187 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/buffered-sink.test.ts +// Exercises: IO-4 (exact head removal, no partial write), IO-5 (flush, closeable), +// IO-13 (symmetric write-side encodings), IO-18 (emit vs flush), IO-41 (idempotent close), +// IO-42 (rejects after close), IO-6 (wrapper owns the caller's stream) +import {describe, expect, test} from 'bun:test'; +import fc from 'fast-check'; +import {BufferedSink} from './buffered-sink.js'; +import {BufferedSource} from './buffered-source.js'; +import {ByteQueue} from './byte-queue.js'; +import {ClosedResourceError, EndOfStreamError} from './errors.js'; +import {collectingWritableStream} from './test-support/fake-stream.js'; +import {rejection} from './test-support/rejection.js'; + +const queueOf = (...values: number[]): ByteQueue => { + const queue = new ByteQueue(); + queue.writeBytes(Uint8Array.from(values)); + return queue; +}; + +describe('BufferedSink', () => { + test('IO-4: write removes exactly the requested count from the source head', async () => { + const {stream, written} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + const source = queueOf(1, 2, 3, 4); + await sink.write(source, 3); + await sink.close(); + expect([...written()]).toEqual([1, 2, 3]); + expect(source.size).toBe(1); + }); + + test('IO-4: writing more than the source holds throws and transfers nothing', async () => { + const {stream, written} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + const source = queueOf(1, 2); + expect(await rejection(sink.write(source, 3))).toBeInstanceOf( + EndOfStreamError, + ); + await sink.close(); + expect([...written()]).toEqual([]); + expect(source.size).toBe(2); + }); + + test('IO-13: writeUtf8 encodes non-ASCII text symmetrically with the read side', async () => { + const {stream, written} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + await sink.writeUtf8('héllo ☃'); + await sink.close(); + expect(new TextDecoder('utf-8').decode(written())).toBe('héllo ☃'); + }); + + test('IO-13: writeString encodes ISO-8859-1', async () => { + const {stream, written} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + await sink.writeString('hé', 'iso-8859-1'); + await sink.close(); + expect([...written()]).toEqual([0x68, 0xe9]); + }); + + test('IO-13: writeString rejects a code point ISO-8859-1 cannot represent', async () => { + const {stream} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + expect( + (await rejection(sink.writeString('☃', 'iso-8859-1'))).message, + ).toContain('code point 9731 is not representable in iso-8859-1'); + }); + + test('IO-13: writeString rejects a charset the write side cannot encode', async () => { + // TextEncoder is UTF-8-only and SEAM-1 forbids an encoding dependency, so the write side covers + // exactly UTF-8 and ISO-8859-1. Anything else throws rather than silently re-encoding as UTF-8. + const {stream} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + expect( + (await rejection(sink.writeString('x', 'shift_jis'))).message, + ).toContain( + 'unsupported write charset: shift_jis (only utf-8 and iso-8859-1 can be encoded)', + ); + }); +}); + +describe('BufferedSink charset round-trips (IO-13)', () => { + /** + * Code points 0x80–0x9F are deliberately outside this generator: the WHATWG Encoding Standard maps + * the label `iso-8859-1` onto windows-1252, so `TextDecoder` turns 0x80 into U+20AC rather than + * U+0080. That asymmetry is the platform's, not this sink's, and HTTP never needs those C1 controls; + * the round-trip holds across every byte outside that band. + */ + const latin1Codes = fc.array( + fc.oneof( + fc.integer({min: 0x00, max: 0x7f}), + fc.integer({min: 0xa0, max: 0xff}), + ), + {maxLength: 64}, + ); + + test('property: arbitrary text round-trips through the sink and back as UTF-8', async () => { + // Styleguide 11.5 names codecs explicitly, and IO-13's whole claim is that the write side is + // symmetric with the read side — a claim only a round-trip can check. + await fc.assert( + fc.asyncProperty( + fc.string({unit: 'grapheme', maxLength: 64}), + async text => { + const {stream, written} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + await sink.writeString(text, 'utf-8'); + await sink.close(); + const source = BufferedSource.overBytes(written()); + expect(await source.readString('utf-8')).toBe(text); + }, + ), + ); + }); + + test('property: arbitrary ISO-8859-1 text round-trips as one byte per code point', async () => { + // IO-13's own conformance note names ISO-8859-1 as the non-UTF-8 charset to round-trip. The + // one-byte-per-code-point assertion is what distinguishes an honored charset from a silent + // UTF-8 re-encoding, which would widen every code point above 0x7F to two bytes. + await fc.assert( + fc.asyncProperty(latin1Codes, async codes => { + const text = codes.map(code => String.fromCharCode(code)).join(''); + const {stream, written} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + await sink.writeString(text, 'iso-8859-1'); + await sink.close(); + expect([...written()]).toEqual(codes); + const source = BufferedSource.overBytes(written()); + expect(await source.readString('iso-8859-1')).toBe(text); + }), + ); + }); +}); + +describe('BufferedSink lifecycle (IO-18, IO-41, IO-42, IO-6)', () => { + test('IO-18: flush and emit both return the sink for chaining', async () => { + const {stream} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + expect(await sink.emit()).toBe(sink); + expect(await sink.flush()).toBe(sink); + await sink.close(); + }); + + test('IO-41: close is idempotent', async () => { + const {stream} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + await sink.close(); + await sink.close(); + expect(sink.closed).toBe(true); + }); + + test('IO-42: write, flush, and emit all reject after close', async () => { + const {stream} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + await sink.close(); + expect(await rejection(sink.write(queueOf(1), 1))).toBeInstanceOf( + ClosedResourceError, + ); + expect(await rejection(sink.flush())).toBeInstanceOf(ClosedResourceError); + expect(await rejection(sink.emit())).toBeInstanceOf(ClosedResourceError); + }); + + test('IO-6: closing the sink closes the caller stream it took ownership of', async () => { + const {stream, isClosed} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + await sink.close(); + expect(isClosed()).toBe(true); + }); +}); + +describe('BufferedSink host-native bridge (IO-16)', () => { + test('toWritableStream forwards written chunks', async () => { + const {stream, written} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + const bridge = sink.toWritableStream(); + const writer = bridge.getWriter(); + await writer.write(Uint8Array.from([1, 2])); + await writer.close(); + expect([...written()]).toEqual([1, 2]); + }); + + test('closing the bridge closes the sink', async () => { + const {stream} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + const writer = sink.toWritableStream().getWriter(); + await writer.close(); + expect(sink.closed).toBe(true); + }); +}); diff --git a/packages/core/src/io/buffered-sink.ts b/packages/core/src/io/buffered-sink.ts new file mode 100644 index 0000000..c47c389 --- /dev/null +++ b/packages/core/src/io/buffered-sink.ts @@ -0,0 +1,145 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/buffered-sink.ts +import {invariant} from '../invariant.js'; +import {ByteQueue} from './byte-queue.js'; +import {ClosedResourceError, IoError} from './errors.js'; + +/** + * A buffered byte sink over a `WritableStream` (IO-4, IO-5, IO-13, IO-18). + * + * Takes no `AbortSignal` and imposes no timeout (IO-40). Not safe for concurrent use (IO-37). + * + * @internal + */ +export class BufferedSink { + readonly #writer: WritableStreamDefaultWriter; + #closed = false; + + private constructor(writer: WritableStreamDefaultWriter) { + this.#writer = writer; + } + + /** Wrap a caller-supplied stream (IO-30). */ + static overStream(stream: WritableStream): BufferedSink { + return new BufferedSink(stream.getWriter()); + } + + get closed(): boolean { + return this.#closed; + } + + /** + * Remove exactly `count` bytes from `src`'s head and push them downstream (IO-4). Fails rather than + * writing a partial amount when `src` holds fewer. + */ + async write(src: ByteQueue, count: number): Promise { + assertCount(count); + this.#assertOpen(); + if (count === 0) return; + // takeBytes raises EndOfStreamError when the source is short, before anything reaches the wire. + const payload = src.takeBytes(count); + await this.#writer.write(payload); + } + + /** Encode and write UTF-8 text (IO-13). */ + async writeUtf8(text: string): Promise { + return this.writeString(text, 'utf-8'); + } + + /** + * Encode and write text with an explicit charset (IO-13). + * + * The write side supports UTF-8 and ISO-8859-1 only. `TextEncoder` is UTF-8-only — there is no + * `TextEncoder('iso-8859-1')` — and SEAM-1 forbids an encoding dependency, so full symmetry with the + * read side is not reachable. These are the two encodings HTTP needs, and IO-13's own conformance note + * names ISO-8859-1. Any other label throws rather than silently re-encoding as UTF-8, which would + * corrupt the bytes on the wire. + */ + async writeString(text: string, charset: string): Promise { + this.#assertOpen(); + await this.#writer.write(encodeText(text, charset)); + } + + /** IO-18: a full force-out toward the destination. */ + async flush(): Promise { + this.#assertOpen(); + await this.#writer.ready; + return this; + } + + /** IO-18: a cheap one-level handoff, distinguished from `flush`. */ + async emit(): Promise { + this.#assertOpen(); + return Promise.resolve(this); + } + + /** IO-5, IO-41: closeable and idempotent. */ + async close(): Promise { + if (this.#closed) return; + this.#closed = true; + await this.#writer.close(); + } + + /** + * A writable host-native byte-stream bridge (IO-16). Closing the bridge closes the sink. + */ + toWritableStream(): WritableStream { + return new WritableStream({ + write: async (chunk): Promise => { + const staging = new ByteQueue(); + staging.writeBytes(chunk); + await this.write(staging, staging.size); + }, + close: async (): Promise => { + await this.close(); + }, + abort: async (): Promise => { + await this.close(); + }, + }); + } + + /** IO-42: a stream-backed sink rejects writes, flushes, and emits after close. */ + #assertOpen(): void { + if (this.#closed) throw new ClosedResourceError('BufferedSink'); + } +} + +/** + * The single source of truth for write-side encoding (IO-13). + * + * Exported because `TeeSink` must mirror the exact bytes this sink will emit. A second copy there would + * be two implementations of one encoding rule, free to drift — the same DRY hazard that had the RFC 3986 + * encoder extracted in Phase 2. Keeping the charset *rejection* here too means `TeeSink` cannot + * accidentally accept a label the primary would refuse. + * + * ISO-8859-1 is a direct code-point-to-byte map for 0–255; anything above is not representable. + */ +export function encodeText(text: string, charset: string): Uint8Array { + const normalized = charset.toLowerCase(); + if (normalized === 'utf-8' || normalized === 'utf8') + return new TextEncoder().encode(text); + if (normalized !== 'iso-8859-1' && normalized !== 'latin1') { + throw new IoError( + `unsupported write charset: ${charset} (only utf-8 and iso-8859-1 can be encoded)`, + ); + } + const out = new Uint8Array(text.length); + for (let i = 0; i < text.length; i += 1) { + const code = text.charCodeAt(i); + if (code > 0xff) { + throw new IoError( + `code point ${String(code)} is not representable in ${charset}`, + ); + } + out[i] = code; + } + return out; +} + +function assertCount(count: number): void { + invariant( + Number.isInteger(count) && count >= 0, + `count must be a non-negative integer, got ${String(count)}`, + ); +} diff --git a/packages/core/src/io/buffered-source.test.ts b/packages/core/src/io/buffered-source.test.ts new file mode 100644 index 0000000..73e3d87 --- /dev/null +++ b/packages/core/src/io/buffered-source.test.ts @@ -0,0 +1,160 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/buffered-source.test.ts +// Exercises: IO-1 (read protocol), IO-2 (zero-count read), IO-3 (negative count), +// IO-11 (exhausted, single-byte read, remaining-bytes read), IO-12 (exact-count read), +// IO-15 (skip), IO-41 (idempotent close), IO-42 (stream-backed rejects after close), +// IO-6 (wrapper owns the caller's stream) +import {describe, expect, test} from 'bun:test'; +import {BufferedSource} from './buffered-source.js'; +import {ByteQueue} from './byte-queue.js'; +import {ClosedResourceError, EndOfStreamError} from './errors.js'; +import {END_OF_STREAM} from './limits.js'; +import {fakeReadableStream} from './test-support/fake-stream.js'; +import {rejection} from './test-support/rejection.js'; + +const bytes = (...values: number[]): Uint8Array => Uint8Array.from(values); + +const sourceOver = (...chunks: Uint8Array[]): BufferedSource => + BufferedSource.overStream(fakeReadableStream(chunks)); + +describe('BufferedSource core reads', () => { + test('IO-1: read appends to the destination tail and returns the transferred count', async () => { + const source = sourceOver(bytes(1, 2, 3)); + const dest = new ByteQueue(); + dest.writeBytes(bytes(9)); + expect(await source.read(dest, 2)).toBe(2); + expect([...dest.snapshot()]).toEqual([9, 1, 2]); + }); + + test('IO-1: read returns END_OF_STREAM once exhausted', async () => { + const source = sourceOver(bytes(1)); + const dest = new ByteQueue(); + expect(await source.read(dest, 4)).toBe(1); + expect(await source.read(dest, 4)).toBe(END_OF_STREAM); + }); + + test('IO-2: a zero-count read returns 0 on a fresh source', async () => { + const source = sourceOver(bytes(1)); + expect(await source.read(new ByteQueue(), 0)).toBe(0); + }); + + test('IO-2: a zero-count read returns 0 — not END_OF_STREAM — on an exhausted source', async () => { + const source = sourceOver(); + expect(await source.read(new ByteQueue(), 4)).toBe(END_OF_STREAM); + expect(await source.read(new ByteQueue(), 0)).toBe(0); + }); + + test('IO-3: a negative count is rejected before any I/O', async () => { + const source = sourceOver(bytes(1, 2)); + expect( + (await rejection(source.read(new ByteQueue(), -1))).message, + ).toContain('count must be a non-negative integer, got -1'); + }); + + test('IO-11: exhausted() is false while bytes remain and true once they do not', async () => { + const source = sourceOver(bytes(1)); + expect(await source.exhausted()).toBe(false); + await source.readBytes(); + expect(await source.exhausted()).toBe(true); + }); + + test('IO-11: readByte returns the next byte, then fails at end', async () => { + const source = sourceOver(bytes(7)); + expect(await source.readByte()).toBe(7); + expect(await rejection(source.readByte())).toBeInstanceOf(EndOfStreamError); + }); + + test('IO-11: readBytes returns all remaining bytes, and empty when already exhausted', async () => { + const source = sourceOver(bytes(1, 2), bytes(3)); + expect([...(await source.readBytes())]).toEqual([1, 2, 3]); + expect([...(await source.readBytes())]).toEqual([]); + }); + + test('IO-12: readExactly returns exactly the requested count across chunk boundaries', async () => { + const source = sourceOver(bytes(1), bytes(2, 3), bytes(4)); + expect([...(await source.readExactly(3))]).toEqual([1, 2, 3]); + }); + + test('IO-12: readExactly fails rather than returning a short result', async () => { + const source = sourceOver(bytes(1, 2)); + expect(await rejection(source.readExactly(3))).toBeInstanceOf( + EndOfStreamError, + ); + }); +}); + +describe('BufferedSource skip and lifecycle (IO-15, IO-41, IO-42, IO-6)', () => { + test('IO-15: skip advances past exactly the requested count', async () => { + const source = sourceOver(bytes(1, 2, 3, 4)); + await source.skip(2); + expect([...(await source.readBytes())]).toEqual([3, 4]); + }); + + test('IO-15: skip fails when fewer bytes remain', async () => { + const source = sourceOver(bytes(1, 2)); + expect(await rejection(source.skip(3))).toBeInstanceOf(EndOfStreamError); + }); + + test('IO-15: skip(0) is a no-op, even at and after end of stream', async () => { + const source = sourceOver(bytes(1)); + await source.skip(0); + await source.readBytes(); + await source.skip(0); + expect(await source.exhausted()).toBe(true); + }); + + test('IO-41: close is idempotent', async () => { + const source = sourceOver(bytes(1)); + await source.close(); + await source.close(); + expect(source.closed).toBe(true); + }); + + test('IO-42: a stream-backed source REJECTS reads after close', async () => { + // The opposite direction from ByteQueue, which stays readable. IO-42 names both as the + // inconsistency porters get wrong; both directions are asserted, here and in Task 4. + const source = sourceOver(bytes(1, 2)); + await source.close(); + expect(await rejection(source.read(new ByteQueue(), 1))).toBeInstanceOf( + ClosedResourceError, + ); + expect(await rejection(source.readBytes())).toBeInstanceOf( + ClosedResourceError, + ); + }); + + test('overBytes wraps a byte array as an independent copy', async () => { + const input = bytes(1, 2, 3); + const source = BufferedSource.overBytes(input); + input[0] = 99; + expect([...(await source.readBytes())]).toEqual([1, 2, 3]); + }); + + test('IO-6: closing the source cancels the caller stream it took ownership of', async () => { + let cancelled = false; + const source = BufferedSource.overStream( + fakeReadableStream([bytes(1)], () => { + cancelled = true; + }), + ); + await source.close(); + expect(cancelled).toBe(true); + }); +}); + +describe('BufferedSource host-native bridge (IO-16)', () => { + test('toReadableStream yields the remaining bytes', async () => { + const source = sourceOver(bytes(1, 2), bytes(3)); + const collected: number[] = []; + for await (const chunk of source.toReadableStream()) + collected.push(...chunk); + expect(collected).toEqual([1, 2, 3]); + }); + + test('closing the bridge closes the owning source', async () => { + const source = sourceOver(bytes(1, 2, 3)); + const stream = source.toReadableStream(); + await stream.cancel(); + expect(source.closed).toBe(true); + }); +}); diff --git a/packages/core/src/io/buffered-source.text.test.ts b/packages/core/src/io/buffered-source.text.test.ts new file mode 100644 index 0000000..f9cd97e --- /dev/null +++ b/packages/core/src/io/buffered-source.text.test.ts @@ -0,0 +1,130 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/buffered-source.text.test.ts +// Exercises: IO-13 (UTF-8 and explicit-charset decode), IO-14 (line reads: \n and \r\n terminators, +// lone \r stays content, final unterminated line returned as-is, undefined when exhausted first) +import {describe, expect, test} from 'bun:test'; +import fc from 'fast-check'; +import {BufferedSource} from './buffered-source.js'; +import {fakeReadableStream} from './test-support/fake-stream.js'; +import {rejection} from './test-support/rejection.js'; + +const utf8 = (text: string): Uint8Array => new TextEncoder().encode(text); + +const sourceOver = (...chunks: Uint8Array[]): BufferedSource => + BufferedSource.overStream(fakeReadableStream(chunks)); + +/** Split `bytes` at the given cut points, so a terminator can straddle a chunk boundary. */ +function chunkAt(bytes: Uint8Array, cuts: readonly number[]): Uint8Array[] { + const bounded = [ + ...new Set(cuts.filter(c => c > 0 && c < bytes.length)), + ].sort((a, b) => a - b); + const out: Uint8Array[] = []; + let previous = 0; + for (const cut of bounded) { + out.push(bytes.subarray(previous, cut)); + previous = cut; + } + // An empty trailing subarray (only possible when `bytes` itself is empty) would enqueue a zero-length + // chunk, which RetentionWindow correctly rejects as an IO-17 protocol violation — a stream signals + // end-of-stream via `done`, never via a 0-byte delivery. Omitting it here keeps the fixture itself + // protocol-clean. + const last = bytes.subarray(previous); + if (last.length > 0) out.push(last); + return out; +} + +describe('BufferedSource text reads (IO-13)', () => { + test('readUtf8 decodes non-ASCII text', async () => { + expect(await sourceOver(utf8('héllo ☃')).readUtf8()).toBe('héllo ☃'); + }); + + test('readUtf8 decodes across a chunk boundary that splits a multi-byte character', async () => { + const encoded = utf8('☃'); + const source = sourceOver(encoded.subarray(0, 1), encoded.subarray(1)); + expect(await source.readUtf8()).toBe('☃'); + }); + + test('readString decodes an explicit non-UTF-8 charset', async () => { + // 0xE9 is é in ISO-8859-1 and invalid alone in UTF-8 — so this only passes if the charset is honored. + const source = sourceOver(Uint8Array.from([0x68, 0xe9])); + expect(await source.readString('iso-8859-1')).toBe('hé'); + }); + + test('readString rejects an unknown charset label', async () => { + expect( + (await rejection(sourceOver(utf8('x')).readString('not-a-charset'))) + .message, + ).toContain('unsupported charset: not-a-charset'); + }); +}); + +describe('BufferedSource line reads (IO-14)', () => { + test('splits on \\n and consumes the terminator', async () => { + const source = sourceOver(utf8('one\ntwo\n')); + expect(await source.readUtf8Line()).toBe('one'); + expect(await source.readUtf8Line()).toBe('two'); + expect(await source.readUtf8Line()).toBeUndefined(); + }); + + test('treats \\r\\n as a terminator and strips both bytes', async () => { + const source = sourceOver(utf8('one\r\ntwo\r\n')); + expect(await source.readUtf8Line()).toBe('one'); + expect(await source.readUtf8Line()).toBe('two'); + }); + + test('keeps a lone \\r not followed by \\n as line content', async () => { + const source = sourceOver(utf8('a\rb\n')); + expect(await source.readUtf8Line()).toBe('a\rb'); + }); + + test('returns a final unterminated line as-is', async () => { + const source = sourceOver(utf8('one\ntwo')); + expect(await source.readUtf8Line()).toBe('one'); + expect(await source.readUtf8Line()).toBe('two'); + expect(await source.readUtf8Line()).toBeUndefined(); + }); + + test('returns undefined when exhausted before any byte', async () => { + expect(await sourceOver().readUtf8Line()).toBeUndefined(); + }); + + test('returns an empty string for an empty line', async () => { + const source = sourceOver(utf8('\nx\n')); + expect(await source.readUtf8Line()).toBe(''); + expect(await source.readUtf8Line()).toBe('x'); + }); + + test('property: lines round-trip across adversarial chunk boundaries', async () => { + // IO-14's rationale calls out surviving slice-window boundaries; hand-picked examples miss the case + // where \r and \n land in different chunks, so the cut points are generated. + await fc.assert( + fc.asyncProperty( + fc.array(fc.stringMatching(/^[a-z \r]*$/), {maxLength: 8}), + fc.constantFrom('\n', '\r\n'), + fc.array(fc.integer({min: 0, max: 64}), {maxLength: 8}), + async (lines, terminator, cuts) => { + const encoded = utf8( + lines.map(line => `${line}${terminator}`).join(''), + ); + const source = BufferedSource.overStream( + fakeReadableStream(chunkAt(encoded, cuts)), + ); + + const read: string[] = []; + for (;;) { + const line = await source.readUtf8Line(); + if (line === undefined) break; + read.push(line); + } + // A line-content trailing \r merges with an appended \n into \r\n and is stripped by the + // reader; with a \r\n terminator only the terminator's own \r is stripped, so a content \r + // survives. The oracle mirrors exactly that rule. + const expected = lines.map(line => + terminator === '\n' ? line.replace(/\r$/, '') : line, + ); + expect(read).toEqual(expected); + }, + ), + ); + }); +}); diff --git a/packages/core/src/io/buffered-source.ts b/packages/core/src/io/buffered-source.ts new file mode 100644 index 0000000..93158d2 --- /dev/null +++ b/packages/core/src/io/buffered-source.ts @@ -0,0 +1,332 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/buffered-source.ts +import {invariant} from '../invariant.js'; +import {ByteQueue} from './byte-queue.js'; +import { + AllocationLimitError, + ClosedResourceError, + EndOfStreamError, + IoError, +} from './errors.js'; +import {END_OF_STREAM, MAX_BYTE_ARRAY_LENGTH} from './limits.js'; +import {RetentionWindow, type Cursor} from './retention-window.js'; + +/** + * A buffered, non-blocking byte source over a `ReadableStream` (IO-11–IO-24). + * + * Peek and slice views are instances of this same class over the same `RetentionWindow`, differing only in + * their cursor, their byte budget, and whether they own the window. A second class would need either + * inheritance — which styleguide 6.4 reserves for `Error` hierarchies — or ten duplicated delegating + * methods. + * + * Takes no `AbortSignal` and imposes no timeout: IO-40 assigns deadlines and prompt cancellation of + * blocked I/O to the transport that owns the real socket. Not safe for concurrent use (IO-37). + * + * @internal + */ +export class BufferedSource { + readonly #window: RetentionWindow; + readonly #cursor: Cursor; + readonly #ownsWindow: boolean; + readonly #limit: number; + readonly #startedAt: number; + #closed = false; + + // eslint-disable-next-line max-params -- private, view-internal plumbing; peek()/slice() are the public entry points (0-2 params each) + private constructor( + window: RetentionWindow, + cursor: Cursor, + ownsWindow: boolean, + limit: number, + ) { + this.#window = window; + this.#cursor = cursor; + this.#ownsWindow = ownsWindow; + this.#limit = limit; + this.#startedAt = cursor.at; + } + + /** Wrap a caller-supplied stream (IO-30). */ + static overStream(stream: ReadableStream): BufferedSource { + const window = new RetentionWindow(stream.getReader()); + return new BufferedSource( + window, + window.register(0), + true, + Number.POSITIVE_INFINITY, + ); + } + + /** Wrap a byte array as an independent copy (IO-30). */ + static overBytes(bytes: Uint8Array): BufferedSource { + const copy = bytes.slice(); + return BufferedSource.overStream( + new ReadableStream({ + start(controller): void { + if (copy.length > 0) controller.enqueue(copy); + controller.close(); + }, + }), + ); + } + + get closed(): boolean { + return this.#closed; + } + + /** Read up to `count` bytes onto `dest`'s tail (IO-1, IO-2, IO-3). */ + async read(dest: ByteQueue, count: number): Promise { + assertCount(count); + this.#assertOpen(); + // IO-2 before any exhaustion determination — a zero-count read is 0, never END_OF_STREAM. + if (count === 0) return 0; + const want = Math.min(count, this.#remainingBudget()); + if (want <= 0) return END_OF_STREAM; + const available = await this.#window.pullThrough(this.#cursor.at + 1); + if (!available) return END_OF_STREAM; + return this.#window.readInto(this.#cursor, dest, want); + } + + /** True exactly when no more bytes are available (IO-11). */ + async exhausted(): Promise { + this.#assertOpen(); + if (this.#remainingBudget() <= 0) return true; + return !(await this.#window.pullThrough(this.#cursor.at + 1)); + } + + /** The next byte, or a failure at end of stream (IO-11). */ + async readByte(): Promise { + const [value] = await this.readExactly(1); + invariant(value !== undefined, 'readExactly(1) returned an empty array'); + return value; + } + + /** Every remaining byte; empty when already exhausted (IO-11). */ + async readBytes(): Promise { + this.#assertOpen(); + const staging = new ByteQueue(); + while ((await this.read(staging, READ_CHUNK)) !== END_OF_STREAM) { + // Drain to exhaustion; `read` already bounds each transfer and advances the cursor. + } + return staging.snapshot(); + } + + /** Exactly `count` bytes, or a failure — never a short result (IO-12). */ + async readExactly(count: number): Promise { + assertCount(count); + this.#assertOpen(); + // IO-9: refuse eagerly with an actionable error. Routing this through ByteQueue would raise + // EndOfStreamError instead, since takeBytes checks its size before it ever tries to allocate. + if (count > MAX_BYTE_ARRAY_LENGTH) { + throw new AllocationLimitError(count, MAX_BYTE_ARRAY_LENGTH); + } + const staging = new ByteQueue(); + while (staging.size < count) { + const read = await this.read(staging, count - staging.size); + if (read === END_OF_STREAM) + throw new EndOfStreamError(staging.size, count); + } + return staging.takeBytes(count); + } + + /** Decode `count` bytes (or every remaining byte) as UTF-8 (IO-13). */ + async readUtf8(count?: number): Promise { + return this.readString('utf-8', count); + } + + /** Decode `count` bytes (or every remaining byte) with an explicit charset (IO-13). */ + async readString(charset: string, count?: number): Promise { + this.#assertOpen(); + const decoder = decoderFor(charset); + const raw = + count === undefined + ? await this.readBytes() + : await this.readExactly(count); + return decoder.decode(raw); + } + + /** + * The next line as UTF-8, with its terminator consumed (IO-14). + * + * Both `\n` and `\r\n` terminate. A lone `\r` not followed by `\n` stays line content, which falls out + * of scanning only for `\n`. Returns the final unterminated line as-is, and `undefined` when the source + * is exhausted before any byte — `undefined` rather than the spec's language-agnostic "null", per + * styleguide 3.5. + * + * Scans with a NON-CONSUMING peek before reading, deliberately. Reading first and pushing back the + * over-read cannot work: every read advances this cursor and `RetentionWindow.readInto` then trims the + * queue head to the slowest cursor, so the bytes past the terminator are already discarded by the time + * anything could rewind over them. Peeking leaves the cursor still, so the bytes stay retained, and the + * subsequent `readExactly` consumes exactly the line plus its terminator. + */ + async readUtf8Line(): Promise { + this.#assertOpen(); + const at = await this.#scanForNewline(); + if (at === END_OF_STREAM) { + const rest = await this.readBytes(); + return rest.length === 0 + ? undefined + : new TextDecoder('utf-8').decode(rest); + } + const line = await this.readExactly(at + 1); + const end = at > 0 && line[at - 1] === CARRIAGE_RETURN ? at - 1 : at; + return new TextDecoder('utf-8').decode(line.subarray(0, end)); + } + + /** + * Offset of the next `\n` relative to this cursor, or `END_OF_STREAM` if the source ends first. + * Never advances the cursor. Retention grows by one line's length, which is what IO-14 requires and + * all it requires. + */ + async #scanForNewline(): Promise { + let searched = 0; + for (;;) { + const available = Math.min( + this.#window.availableFrom(this.#cursor), + this.#remainingBudget(), + ); + if (available > searched) { + const scanned = this.#window.peekBytes(this.#cursor, available); + const found = scanned.indexOf(NEWLINE, searched); + if (found >= 0) return found; + searched = scanned.length; + } + if (searched >= this.#remainingBudget()) return END_OF_STREAM; + if (!(await this.#window.pullThrough(this.#cursor.at + searched + 1))) + return END_OF_STREAM; + } + } + + /** + * A non-consuming view over the whole remaining source (IO-19). Reads from it never advance this + * source's cursor. + * + * Deliberately uncapped: §5 bounds nothing, and every buffering cap the product spec mandates lives in + * §6 (Phase 3b). See `RetentionWindow` for why a cap here would partially fail IO-19. + */ + peek(): BufferedSource { + this.#assertOpen(); + return new BufferedSource( + this.#window, + this.#window.register(this.#cursor.at), + false, + this.#remainingBudget(), + ); + } + + /** + * A non-consuming, length-bounded view exposing at most `count` bytes starting `offset` ahead of this + * cursor (IO-20). + * + * Offset overflow is detected LAZILY — an offset past the source size constructs fine and surfaces as + * an empty read (IO-21) — because callers may slice speculatively before the body length is known. A + * negative offset or count is rejected eagerly. A slice of a slice composes additively and caps at the + * outer slice's remaining budget (IO-23). + */ + slice(offset: number, count: number): BufferedSource { + invariant( + Number.isInteger(offset) && offset >= 0, + `offset must be a non-negative integer, got ${String(offset)}`, + ); + assertCount(count); + this.#assertOpen(); + const budget = Math.max( + 0, + Math.min(count, this.#remainingBudget() - offset), + ); + return new BufferedSource( + this.#window, + this.#window.register(this.#cursor.at + offset), + false, + budget, + ); + } + + /** Advance past exactly `count` bytes; `skip(0)` is a no-op even at end of stream (IO-15). */ + async skip(count: number): Promise { + assertCount(count); + this.#assertOpen(); + if (count === 0) return; + const staging = new ByteQueue(); + let skipped = 0; + while (skipped < count) { + const read = await this.read(staging, count - skipped); + if (read === END_OF_STREAM) throw new EndOfStreamError(skipped, count); + skipped += read; + staging.clear(); + } + } + + /** + * IO-41: idempotent. A view releases only its own cursor and never closes its parent or moves the + * parent's cursor (IO-22); the owning source closes the window, which invalidates every outstanding + * view. + */ + async close(): Promise { + if (this.#closed) return; + this.#closed = true; + if (this.#ownsWindow) this.#window.close(); + else this.#window.release(this.#cursor); + return Promise.resolve(); + } + + /** IO-42: a stream-backed source rejects reads after close, unlike an in-memory `ByteQueue`. */ + #assertOpen(): void { + if (this.#closed) throw new ClosedResourceError('BufferedSource'); + this.#window.assertUsable(); + } + + /** + * A read-only host-native byte-stream bridge (IO-16). Closing the bridge closes the owning source. + * + * For this port the host-native byte stream IS `ReadableStream` — that is `sdk-design/03` §3.1's whole + * premise, and it keeps core free of any `node:` import. A consumer wanting a Node `Readable` calls + * `Readable.fromWeb()` at their own edge. + */ + toReadableStream(): ReadableStream { + return new ReadableStream({ + pull: async (controller): Promise => { + const staging = new ByteQueue(); + const read = await this.read(staging, BRIDGE_CHUNK); + if (read === END_OF_STREAM) { + controller.close(); + await this.close(); + return; + } + controller.enqueue(staging.snapshot()); + }, + cancel: async (): Promise => { + await this.close(); + }, + }); + } + + #remainingBudget(): number { + if (this.#limit === Number.POSITIVE_INFINITY) + return Number.POSITIVE_INFINITY; + return Math.max(0, this.#limit - (this.#cursor.at - this.#startedAt)); + } +} + +/** How much a bulk drain asks for per iteration. Not a retention bound — `read` transfers, never buffers. */ +const READ_CHUNK = 16 * 1024; +const BRIDGE_CHUNK = 16 * 1024; +const NEWLINE = 0x0a; +const CARRIAGE_RETURN = 0x0d; + +function assertCount(count: number): void { + invariant( + Number.isInteger(count) && count >= 0, + `count must be a non-negative integer, got ${String(count)}`, + ); +} + +function decoderFor(charset: string): TextDecoder { + try { + return new TextDecoder(charset); + } catch (e: unknown) { + // A charset label reaching this layer is internal, so this is an argument error, not boundary + // data. Phase 3b's HTTP-42 owns the "unknown declared charset falls back to UTF-8" rule. + throw new IoError(`unsupported charset: ${charset}`, {cause: e}); + } +} diff --git a/packages/core/src/io/buffered-source.views.test.ts b/packages/core/src/io/buffered-source.views.test.ts new file mode 100644 index 0000000..846f06f --- /dev/null +++ b/packages/core/src/io/buffered-source.views.test.ts @@ -0,0 +1,134 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/buffered-source.views.test.ts +// Exercises: IO-19 (peek is non-consuming over the whole remaining source), IO-20 (bounded slice), +// IO-21 (lazy offset overflow, eager negative rejection), IO-22 (closing a slice does not close the +// parent; closing the parent invalidates slices), IO-23 (independence, additive composition), +// IO-24 (reading a closed slice is a state error, distinct from EOF) +import {describe, expect, test} from 'bun:test'; +import fc from 'fast-check'; +import {BufferedSource} from './buffered-source.js'; +import {ClosedResourceError} from './errors.js'; +import {fakeReadableStream} from './test-support/fake-stream.js'; +import {rejection} from './test-support/rejection.js'; + +const bytes = (...values: number[]): Uint8Array => Uint8Array.from(values); + +const sourceOver = (...chunks: Uint8Array[]): BufferedSource => + BufferedSource.overStream(fakeReadableStream(chunks)); + +describe('BufferedSource views', () => { + test('IO-19: reads from a peek do not advance the original cursor', async () => { + const source = sourceOver(bytes(1, 2, 3)); + const peek = source.peek(); + expect([...(await peek.readBytes())]).toEqual([1, 2, 3]); + expect([...(await source.readBytes())]).toEqual([1, 2, 3]); + }); + + test('IO-20: a slice exposes at most count bytes starting offset ahead', async () => { + const source = sourceOver(bytes(1, 2, 3, 4, 5)); + const slice = source.slice(1, 3); + expect([...(await slice.readBytes())]).toEqual([2, 3, 4]); + }); + + test('IO-20: reading past the window behaves as end-of-window, and never advances the parent', async () => { + const source = sourceOver(bytes(1, 2, 3)); + const slice = source.slice(0, 2); + expect([...(await slice.readBytes())]).toEqual([1, 2]); + expect([...(await slice.readBytes())]).toEqual([]); + expect([...(await source.readBytes())]).toEqual([1, 2, 3]); + }); + + test('IO-21: an offset past the source size succeeds at construction and reads as empty', async () => { + const source = sourceOver(bytes(1, 2)); + const slice = source.slice(100, 4); + expect([...(await slice.readBytes())]).toEqual([]); + }); + + test('IO-21: a negative offset or count is rejected eagerly at construction', () => { + const source = sourceOver(bytes(1, 2)); + expect(() => source.slice(-1, 2)).toThrow( + 'offset must be a non-negative integer, got -1', + ); + expect(() => source.slice(0, -2)).toThrow( + 'count must be a non-negative integer, got -2', + ); + }); +}); + +describe('BufferedSource view lifecycle and independence (IO-22, IO-23, IO-24)', () => { + test('IO-22: closing a slice neither closes the parent nor advances its cursor', async () => { + const source = sourceOver(bytes(1, 2, 3)); + const slice = source.slice(0, 2); + await slice.readBytes(); + await slice.close(); + expect(source.closed).toBe(false); + expect([...(await source.readBytes())]).toEqual([1, 2, 3]); + }); + + test('IO-22: closing the parent invalidates outstanding slices', async () => { + const source = sourceOver(bytes(1, 2, 3)); + const slice = source.slice(0, 2); + await source.close(); + expect(await rejection(slice.readBytes())).toBeInstanceOf( + ClosedResourceError, + ); + }); + + test('IO-24: reading an explicitly closed slice fails loudly, distinct from a normal EOF', async () => { + const source = sourceOver(bytes(1, 2, 3)); + const slice = source.slice(0, 2); + await slice.close(); + expect(await rejection(slice.readBytes())).toBeInstanceOf( + ClosedResourceError, + ); + }); + + test('IO-23: two slices of one source have independent cursors and budgets', async () => { + const source = sourceOver(bytes(1, 2, 3, 4)); + const first = source.slice(0, 2); + const second = source.slice(2, 2); + expect([...(await second.readBytes())]).toEqual([3, 4]); + expect([...(await first.readBytes())]).toEqual([1, 2]); + }); + + test('IO-23: a slice of a slice composes offsets additively and caps at the outer remainder', async () => { + const source = sourceOver(bytes(1, 2, 3, 4, 5, 6)); + const outer = source.slice(1, 4); // 2,3,4,5 + const inner = outer.slice(1, 10); // starts at 3, capped to 3 bytes: 3,4,5 + expect([...(await inner.readBytes())]).toEqual([3, 4, 5]); + }); +}); + +describe('BufferedSource view properties', () => { + test('property: an arbitrary slice reads exactly the bytes at its window', async () => { + await fc.assert( + fc.asyncProperty( + fc.uint8Array({minLength: 1, maxLength: 64}), + fc.integer({min: 0, max: 64}), + fc.integer({min: 0, max: 64}), + async (data, offset, count) => { + const source = BufferedSource.overStream(fakeReadableStream([data])); + const slice = source.slice(offset, count); + const expected = [...data.subarray(offset, offset + count)]; + expect([...(await slice.readBytes())]).toEqual(expected); + }, + ), + ); + }); + + test('property: no view read advances any other view or the parent', async () => { + await fc.assert( + fc.asyncProperty( + fc.uint8Array({minLength: 1, maxLength: 32}), + async data => { + const source = BufferedSource.overStream(fakeReadableStream([data])); + const first = source.peek(); + const second = source.peek(); + await first.readBytes(); + expect([...(await second.readBytes())]).toEqual([...data]); + expect([...(await source.readBytes())]).toEqual([...data]); + }, + ), + ); + }); +}); diff --git a/packages/core/src/io/byte-queue.bench.ts b/packages/core/src/io/byte-queue.bench.ts new file mode 100644 index 0000000..35214eb --- /dev/null +++ b/packages/core/src/io/byte-queue.bench.ts @@ -0,0 +1,35 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/byte-queue.bench.ts +// Baseline only — no optimization has been applied and none is justified yet (styleguide 15.1, 15.6: +// do not tune ahead of a profile). This exists so Phases 6 and 8 inherit a regression floor on the +// SDK's hottest data structure. mitata measures a warm JIT in isolation, not end-to-end throughput. +import {bench, run} from 'mitata'; +import {ByteQueue} from './byte-queue.js'; + +const SMALL = new Uint8Array(64).fill(1); +const LARGE = new Uint8Array(64 * 1024).fill(1); + +bench( + 'ByteQueue writeBytes x1000 small chunks (warm-JIT, not end-to-end)', + () => { + const queue = new ByteQueue(); + for (let i = 0; i < 1000; i += 1) queue.writeBytes(SMALL); + }, +); + +bench( + 'ByteQueue write-then-read round trip, 64 KiB (warm-JIT, not end-to-end)', + () => { + const source = new ByteQueue(); + source.writeBytes(LARGE); + source.read(new ByteQueue(), source.size); + }, +); + +bench('ByteQueue snapshot of 64 KiB (warm-JIT, not end-to-end)', () => { + const queue = new ByteQueue(); + queue.writeBytes(LARGE); + queue.snapshot(); +}); + +await run(); diff --git a/packages/core/src/io/byte-queue.property.test.ts b/packages/core/src/io/byte-queue.property.test.ts new file mode 100644 index 0000000..08847ec --- /dev/null +++ b/packages/core/src/io/byte-queue.property.test.ts @@ -0,0 +1,69 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/byte-queue.property.test.ts +// Exercises: IO-7 (FIFO order across arbitrary chunk splits), IO-8 (snapshot independence), +// IO-10 (copyTo is non-consuming) +import {describe, expect, test} from 'bun:test'; +import fc from 'fast-check'; +import {ByteQueue} from './byte-queue.js'; + +const chunks = fc.array(fc.uint8Array({maxLength: 32}), {maxLength: 16}); + +describe('ByteQueue properties', () => { + test('IO-7: writing arbitrary chunks then reading back preserves byte order exactly', () => { + fc.assert( + fc.property(chunks, input => { + const queue = new ByteQueue(); + for (const chunk of input) queue.writeBytes(chunk); + const expected = input.flatMap(chunk => [...chunk]); + expect(queue.size).toBe(expected.length); + expect([...queue.snapshot()]).toEqual(expected); + }), + ); + }); + + test('IO-7: reading in arbitrary increments yields the same bytes as reading all at once', () => { + fc.assert( + fc.property( + chunks, + fc.array(fc.integer({min: 0, max: 8}), {maxLength: 32}), + (input, steps) => { + const source = new ByteQueue(); + for (const chunk of input) source.writeBytes(chunk); + const expected = input.flatMap(chunk => [...chunk]); + + const dest = new ByteQueue(); + for (const step of steps) source.read(dest, step); + source.read(dest, source.size); + + expect([...dest.snapshot()]).toEqual(expected); + }, + ), + ); + }); + + test('IO-8: a snapshot is unaffected by later writes', () => { + fc.assert( + fc.property(chunks, fc.uint8Array({maxLength: 16}), (input, later) => { + const queue = new ByteQueue(); + for (const chunk of input) queue.writeBytes(chunk); + const before = queue.snapshot(); + const expected = [...before]; + queue.writeBytes(later); + expect([...before]).toEqual(expected); + }), + ); + }); + + test('IO-10: copyTo never changes the source size', () => { + fc.assert( + fc.property(chunks, fc.integer({min: 0, max: 16}), (input, offset) => { + const source = new ByteQueue(); + for (const chunk of input) source.writeBytes(chunk); + fc.pre(offset <= source.size); + const sizeBefore = source.size; + source.copyTo(new ByteQueue(), offset); + expect(source.size).toBe(sizeBefore); + }), + ); + }); +}); diff --git a/packages/core/src/io/byte-queue.test.ts b/packages/core/src/io/byte-queue.test.ts new file mode 100644 index 0000000..3407ddc --- /dev/null +++ b/packages/core/src/io/byte-queue.test.ts @@ -0,0 +1,238 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/byte-queue.test.ts +// Exercises: IO-1 (tail-append, transferred count, EOF sentinel), IO-2 (zero-count read), +// IO-3 (negative count rejected before any I/O), IO-4 (exact head removal, no partial write), +// IO-7 (FIFO buffer that is simultaneously source and sink) +import {describe, expect, test} from 'bun:test'; +import {ByteQueue} from './byte-queue.js'; +import {AllocationLimitError, EndOfStreamError} from './errors.js'; +import {END_OF_STREAM, MAX_BYTE_ARRAY_LENGTH} from './limits.js'; + +const bytes = (...values: number[]): Uint8Array => Uint8Array.from(values); + +const drain = (queue: ByteQueue): number[] => [...queue.snapshot()]; + +describe('ByteQueue read (IO-1, IO-2, IO-7)', () => { + test('starts empty', () => { + expect(new ByteQueue().size).toBe(0); + }); + + test('IO-7: bytes written through the sink surface read back through the source surface in order', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2, 3)); + source.writeBytes(bytes(4, 5)); + const dest = new ByteQueue(); + expect(source.read(dest, 5)).toBe(5); + expect(drain(dest)).toEqual([1, 2, 3, 4, 5]); + expect(source.size).toBe(0); + }); + + test('IO-1: read appends to the TAIL of a non-empty destination, never overwriting', () => { + const dest = new ByteQueue(); + dest.writeBytes(bytes(9, 9)); + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2)); + expect(source.read(dest, 2)).toBe(2); + expect(drain(dest)).toEqual([9, 9, 1, 2]); + }); + + test('IO-1: read never returns more than requested, and returns at least 1 when not exhausted', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2, 3, 4)); + const dest = new ByteQueue(); + expect(source.read(dest, 2)).toBe(2); + expect(source.size).toBe(2); + }); + + test('IO-1: read of a partial source returns what it has, then END_OF_STREAM', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2)); + const dest = new ByteQueue(); + expect(source.read(dest, 8)).toBe(2); + expect(source.read(dest, 8)).toBe(END_OF_STREAM); + }); + + test('IO-2: a zero-count read returns 0 on a non-empty source', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1)); + expect(source.read(new ByteQueue(), 0)).toBe(0); + }); + + test('IO-2: a zero-count read returns 0 — NOT end-of-stream — on an exhausted source', () => { + expect(new ByteQueue().read(new ByteQueue(), 0)).toBe(0); + }); + + test('IO-3: a negative count is rejected before any transfer', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2, 3)); + const dest = new ByteQueue(); + expect(() => source.read(dest, -1)).toThrow( + 'count must be a non-negative integer, got -1', + ); + expect(source.size).toBe(3); + expect(dest.size).toBe(0); + }); +}); + +describe('ByteQueue write (IO-3, IO-4)', () => { + test('IO-4: write removes exactly the requested count from the source HEAD, in order', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2, 3)); + const dest = new ByteQueue(); + dest.write(source, 3); + expect(source.size).toBe(0); + expect(drain(dest)).toEqual([1, 2, 3]); + }); + + test('IO-4: writing more than the source holds throws instead of writing partially', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2, 3)); + const dest = new ByteQueue(); + expect(() => { + dest.write(source, 4); + }).toThrow(EndOfStreamError); + expect(source.size).toBe(3); + expect(dest.size).toBe(0); + }); + + test('IO-3: write rejects a negative count', () => { + expect(() => { + new ByteQueue().write(new ByteQueue(), -2); + }).toThrow('count must be a non-negative integer, got -2'); + }); + + test('writeBytes copies, so mutating the caller input afterwards does not change the queue', () => { + const input = bytes(1, 2, 3); + const queue = new ByteQueue(); + queue.writeBytes(input); + input[0] = 99; + expect(drain(queue)).toEqual([1, 2, 3]); + }); + + test('a transfer that straddles chunk boundaries preserves order', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2)); + source.writeBytes(bytes(3, 4)); + source.writeBytes(bytes(5, 6)); + const dest = new ByteQueue(); + expect(source.read(dest, 3)).toBe(3); + expect(drain(dest)).toEqual([1, 2, 3]); + expect(source.size).toBe(3); + const rest = new ByteQueue(); + expect(source.read(rest, 3)).toBe(3); + expect(drain(rest)).toEqual([4, 5, 6]); + }); +}); + +describe('ByteQueue snapshot and copyTo (IO-8, IO-9, IO-10)', () => { + test('IO-8: snapshot does not consume or mutate', () => { + const queue = new ByteQueue(); + queue.writeBytes(bytes(1, 2, 3)); + expect([...queue.snapshot()]).toEqual([1, 2, 3]); + expect(queue.size).toBe(3); + }); + + test('IO-8: a snapshot is independent of later mutations, in both directions', () => { + const queue = new ByteQueue(); + queue.writeBytes(bytes(1, 2, 3)); + const first = queue.snapshot(); + queue.writeBytes(bytes(4)); + expect([...first]).toEqual([1, 2, 3]); + first[0] = 99; + expect([...queue.snapshot()]).toEqual([1, 2, 3, 4]); + }); + + test('IO-9: materializing past the limit fails with an actionable error, not an allocation crash', () => { + const queue = new ByteQueue(); + queue.writeBytes(bytes(1, 2, 3)); + expect(() => queue.takeBytes(MAX_BYTE_ARRAY_LENGTH + 1)).toThrow( + AllocationLimitError, + ); + }); + + test('IO-10: copyTo copies a window without consuming or mutating the source', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2, 3, 4, 5)); + const dest = new ByteQueue(); + source.copyTo(dest, 1, 3); + expect([...dest.snapshot()]).toEqual([2, 3, 4]); + expect(source.size).toBe(5); + }); + + test('IO-10: copyTo defaults to offset-through-end', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2, 3, 4)); + const dest = new ByteQueue(); + source.copyTo(dest, 2); + expect([...dest.snapshot()]).toEqual([3, 4]); + }); + + test('IO-10: copyTo rejects an out-of-range window', () => { + const source = new ByteQueue(); + source.writeBytes(bytes(1, 2, 3)); + expect(() => { + source.copyTo(new ByteQueue(), 2, 5); + }).toThrow('copy window 2..7 exceeds size 3'); + expect(() => { + source.copyTo(new ByteQueue(), -1); + }).toThrow('offset must be a non-negative integer, got -1'); + }); + + test('IO-10: clear discards every byte', () => { + const queue = new ByteQueue(); + queue.writeBytes(bytes(1, 2, 3)); + queue.clear(); + expect(queue.size).toBe(0); + expect([...queue.snapshot()]).toEqual([]); + }); +}); + +describe('ByteQueue takeBytes and skip', () => { + test('takeBytes consumes exactly the requested count', () => { + const queue = new ByteQueue(); + queue.writeBytes(bytes(1, 2, 3, 4)); + expect([...queue.takeBytes(2)]).toEqual([1, 2]); + expect(queue.size).toBe(2); + }); + + test('takeBytes past the end throws rather than returning short', () => { + const queue = new ByteQueue(); + queue.writeBytes(bytes(1, 2)); + expect(() => queue.takeBytes(3)).toThrow(EndOfStreamError); + }); + + test('skip discards from the head and returns how many it discarded', () => { + const queue = new ByteQueue(); + queue.writeBytes(bytes(1, 2, 3, 4)); + expect(queue.skip(2)).toBe(2); + expect([...queue.snapshot()]).toEqual([3, 4]); + expect(queue.skip(9)).toBe(2); + expect(queue.size).toBe(0); + }); +}); + +describe('ByteQueue close (IO-41, IO-42)', () => { + test('IO-41: close is idempotent — a second close does not throw', () => { + const queue = new ByteQueue(); + queue.close(); + expect(() => { + queue.close(); + }).not.toThrow(); + expect(queue.closed).toBe(true); + }); + + test('IO-42: a purely in-memory buffer stays readable and writable after close', () => { + // IO-42 carves this out explicitly, and Phase 3b depends on it: snapshot-after-close is how + // post-mortem body logging works. Making an in-memory buffer throw here is one of the two + // directions IO-42 names as the porter's trap; the other is Task 6's stream-backed source, which + // MUST reject after close. + const queue = new ByteQueue(); + queue.writeBytes(bytes(1, 2, 3)); + queue.close(); + expect([...queue.snapshot()]).toEqual([1, 2, 3]); + expect(() => { + queue.writeBytes(bytes(4)); + }).not.toThrow(); + expect(queue.read(new ByteQueue(), 1)).toBe(1); + }); +}); diff --git a/packages/core/src/io/byte-queue.ts b/packages/core/src/io/byte-queue.ts new file mode 100644 index 0000000..f261adc --- /dev/null +++ b/packages/core/src/io/byte-queue.ts @@ -0,0 +1,244 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/byte-queue.ts +import {invariant} from '../invariant.js'; +import {AllocationLimitError, EndOfStreamError} from './errors.js'; +import {END_OF_STREAM, MAX_BYTE_ARRAY_LENGTH} from './limits.js'; + +/** + * One node in the queue's chunk list. `bytes` is never mutated after the node is linked in, which is what + * makes zero-copy `subarray` transfers between queues safe; `start` is the first byte not yet consumed. + */ +interface Chunk { + readonly bytes: Uint8Array; + start: number; + next: Chunk | undefined; +} + +/** + * A FIFO byte queue that is simultaneously a source and a sink (IO-7). + * + * Synchronous throughout: pure memory has nothing to wait for, so making it async would allocate a Promise + * on the SDK's hottest data structure (styleguide 15.4) and force every downstream synchronous consumer to + * become async for no I/O reason. `BufferedSource`/`BufferedSink` are the async surfaces. + * + * Not safe for concurrent use (IO-37); callers serialize access. + * + * @internal + */ +export class ByteQueue { + #head: Chunk | undefined = undefined; + #tail: Chunk | undefined = undefined; + #size = 0; + #closed = false; + + /** Bytes currently held (IO-7). */ + get size(): number { + return this.#size; + } + + /** Whether `close()` has been called. */ + get closed(): boolean { + return this.#closed; + } + + /** + * Append an independent copy of `bytes` to the tail. The copy is what lets IO-30's byte-array-wrapping + * factory promise that mutating the caller's input afterwards does not change the source. + */ + writeBytes(bytes: Uint8Array): void { + if (bytes.length === 0) return; + this.#append(bytes.slice()); + } + + /** + * Move up to `count` bytes from this queue's head onto `dest`'s tail (IO-1). + * + * Returns the number transferred: at least 1 when `count` is positive and the queue is not exhausted, + * exactly 0 when `count` is 0, `END_OF_STREAM` at end, and never more than requested. + */ + read(dest: ByteQueue, count: number): number { + assertCount(count); + // IO-2 is checked BEFORE exhaustion, deliberately: a zero-count read returns 0 even on an exhausted + // queue, and must never collapse to END_OF_STREAM. Reordering these two lines breaks IO-2. + if (count === 0) return 0; + if (this.#size === 0) return END_OF_STREAM; + const take = Math.min(count, this.#size); + this.#moveTo(dest, take); + return take; + } + + /** + * Move exactly `count` bytes from `src`'s head onto this queue's tail (IO-4). Fails rather than + * transferring a partial amount when `src` holds fewer. + */ + write(src: ByteQueue, count: number): void { + assertCount(count); + if (src.#size < count) throw new EndOfStreamError(src.#size, count); + src.#moveTo(this, count); + } + + /** + * A fresh, independent copy of the current contents, without consuming or mutating (IO-8). Later + * mutations do not affect a returned snapshot, and vice versa. + */ + snapshot(): Uint8Array { + return this.#materialize(0, this.#size); + } + + /** + * Copy the window `[offset, offset + count)` into `dest` WITHOUT consuming or mutating this queue + * (IO-10). `count` defaults to "from offset through end". An out-of-range window is rejected. + */ + copyTo(dest: ByteQueue, offset: number, count?: number): void { + invariant( + Number.isInteger(offset) && offset >= 0, + `offset must be a non-negative integer, got ${String(offset)}`, + ); + const length = count ?? this.#size - offset; + assertCount(length); + invariant( + offset + length <= this.#size, + `copy window ${String(offset)}..${String(offset + length)} exceeds size ${String(this.#size)}`, + ); + if (length === 0) return; + dest.#append(this.#materialize(offset, length)); + } + + /** Consume and return exactly `count` bytes, failing rather than returning short. */ + takeBytes(count: number): Uint8Array { + assertCount(count); + // IO-9 before IO-4/IO-12's short-source check: an over-limit request is refused with an actionable + // AllocationLimitError even when the queue also happens to be short, rather than surfacing as an + // ordinary EndOfStreamError that hides the real problem. + if (count > MAX_BYTE_ARRAY_LENGTH) + throw new AllocationLimitError(count, MAX_BYTE_ARRAY_LENGTH); + if (count > this.#size) throw new EndOfStreamError(this.#size, count); + const out = this.#materialize(0, count); + this.#discard(count); + return out; + } + + /** Discard up to `count` bytes from the head; returns how many were actually discarded. */ + skip(count: number): number { + assertCount(count); + const dropped = Math.min(count, this.#size); + this.#discard(dropped); + return dropped; + } + + /** Discard every byte (IO-10). */ + clear(): void { + this.#head = undefined; + this.#tail = undefined; + this.#size = 0; + } + + /** + * Copy `count` bytes starting `offset` from the head into one contiguous array (IO-9-bounded). + * + * Parameter order matches `copyTo(dest, offset, count)` deliberately: two adjacent `number`s in + * opposite orders across two methods is exactly the transposition hazard styleguide 5.5 names. + */ + #materialize(offset: number, count: number): Uint8Array { + if (count > MAX_BYTE_ARRAY_LENGTH) + throw new AllocationLimitError(count, MAX_BYTE_ARRAY_LENGTH); + const out = allocate(count); + let skip = offset; + let at = 0; + for ( + let chunk = this.#head; + chunk !== undefined && at < count; + chunk = chunk.next + ) { + const available = chunk.bytes.length - chunk.start; + if (skip >= available) { + skip -= available; + continue; + } + const from = chunk.start + skip; + const take = Math.min(available - skip, count - at); + out.set(chunk.bytes.subarray(from, from + take), at); + at += take; + skip = 0; + } + return out; + } + + #discard(count: number): void { + let remaining = count; + while (remaining > 0) { + const head = this.#head; + invariant(head !== undefined, 'byte-queue underflow during discard'); + const take = Math.min(head.bytes.length - head.start, remaining); + head.start += take; + remaining -= take; + if (head.start === head.bytes.length) this.#dropHead(); + } + this.#size -= count; + } + + /** + * Mark this queue closed (IO-41 — idempotent, the underlying resource released at most once). + * + * Deliberately leaves the read/write surface usable: IO-42 exempts a purely in-memory buffer so that + * snapshot-after-close body logging still works. A queue owns no external resource, so there is nothing + * else to release here. Invalidating derived views is `RetentionWindow`'s job, not this class's — views + * are cursors over a window, never over a bare queue. + */ + close(): void { + this.#closed = true; + } + + /** Caller owns the source-side size accounting; `#dropHead` deliberately does not touch `#size`. */ + #moveTo(dest: ByteQueue, count: number): void { + let remaining = count; + while (remaining > 0) { + const head = this.#head; + invariant(head !== undefined, 'byte-queue underflow during move'); + const take = Math.min(head.bytes.length - head.start, remaining); + dest.#append(head.bytes.subarray(head.start, head.start + take)); + head.start += take; + remaining -= take; + if (head.start === head.bytes.length) this.#dropHead(); + } + this.#size -= count; + } + + #append(bytes: Uint8Array): void { + const chunk: Chunk = {bytes, start: 0, next: undefined}; + if (this.#tail === undefined) this.#head = chunk; + else this.#tail.next = chunk; + this.#tail = chunk; + this.#size += bytes.length; + } + + #dropHead(): void { + const head = this.#head; + invariant(head !== undefined, 'byte-queue drop with no head'); + this.#head = head.next; + if (this.#head === undefined) this.#tail = undefined; + } +} + +function assertCount(count: number): void { + invariant( + Number.isInteger(count) && count >= 0, + `count must be a non-negative integer, got ${String(count)}`, + ); +} + +/** + * IO-9's backstop. The eager `MAX_BYTE_ARRAY_LENGTH` check is deliberately conservative, so a host whose + * real ceiling is lower would otherwise surface a raw `RangeError` — exactly the "low-level allocation + * crash" IO-9 exists to prevent. + */ +function allocate(count: number): Uint8Array { + try { + return new Uint8Array(count); + } catch (e: unknown) { + if (e instanceof RangeError) { + throw new AllocationLimitError(count, MAX_BYTE_ARRAY_LENGTH, {cause: e}); + } + throw e; + } +} diff --git a/packages/core/src/io/errors.test.ts b/packages/core/src/io/errors.test.ts new file mode 100644 index 0000000..4839dee --- /dev/null +++ b/packages/core/src/io/errors.test.ts @@ -0,0 +1,62 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/errors.test.ts +// Exercises: IO-4/IO-11/IO-12/IO-15 (EndOfStreamError), IO-17 (SourceContractViolationError), +// IO-24/IO-42 (ClosedResourceError), IO-9 (AllocationLimitError) +import {describe, expect, test} from 'bun:test'; +import {DexpaceError} from '../http/errors.js'; +import { + AllocationLimitError, + ClosedResourceError, + EndOfStreamError, + IoError, + SourceContractViolationError, +} from './errors.js'; + +describe('IoError tree', () => { + test('IoError descends from DexpaceError', () => { + expect(new IoError('boom')).toBeInstanceOf(DexpaceError); + }); + + test('every leaf descends from IoError', () => { + expect(new EndOfStreamError(3, 8)).toBeInstanceOf(IoError); + expect(new SourceContractViolationError('zero read')).toBeInstanceOf( + IoError, + ); + expect(new ClosedResourceError('BufferedSource')).toBeInstanceOf(IoError); + expect(new AllocationLimitError(9, 8)).toBeInstanceOf(IoError); + }); + + test('each error sets name from its own constructor', () => { + expect(new EndOfStreamError(3, 8).name).toBe('EndOfStreamError'); + expect(new ClosedResourceError('ByteQueue').name).toBe( + 'ClosedResourceError', + ); + }); + + test('EndOfStreamError names delivered-of-requested as typed fields and in the message', () => { + const error = new EndOfStreamError(3, 8); + expect(error.delivered).toBe(3); + expect(error.requested).toBe(8); + expect(error.message).toBe('end of stream: delivered 3 of 8 bytes'); + }); + + test('ClosedResourceError names the resource and is distinct from end-of-stream', () => { + const error = new ClosedResourceError('BufferedSource'); + expect(error.message).toBe('BufferedSource is closed'); + expect(error).not.toBeInstanceOf(EndOfStreamError); + }); + + test('AllocationLimitError points at streaming alternatives', () => { + const error = new AllocationLimitError(5_000, 4_000); + expect(error.requested).toBe(5_000); + expect(error.limit).toBe(4_000); + expect(error.message).toBe( + 'cannot materialize 5000 bytes as one array (limit 4000); stream the body instead', + ); + }); + + test('cause chains through', () => { + const cause = new RangeError('array too large'); + expect(new AllocationLimitError(5, 4, {cause}).cause).toBe(cause); + }); +}); diff --git a/packages/core/src/io/errors.ts b/packages/core/src/io/errors.ts new file mode 100644 index 0000000..f66b4c9 --- /dev/null +++ b/packages/core/src/io/errors.ts @@ -0,0 +1,77 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/errors.ts +import {DexpaceError} from '../http/errors.js'; + +/** + * Root of the I/O error tree (product-spec §5). + * + * Error messages in this tree carry counts and limits, never buffer contents — these buffers hold request + * and response bodies, which routinely contain credentials and PII (styleguide 8.8). + * + * @internal + */ +export class IoError extends DexpaceError {} + +/** + * A source ended before delivering the requested number of bytes (IO-11, IO-12, IO-15), or a sink write + * found fewer bytes in its source buffer than requested (IO-4). + * + * @internal + */ +export class EndOfStreamError extends IoError { + readonly delivered: number; + readonly requested: number; + + constructor(delivered: number, requested: number, options?: ErrorOptions) { + super( + `end of stream: delivered ${String(delivered)} of ${String(requested)} bytes`, + options, + ); + this.delivered = delivered; + this.requested = requested; + } +} + +/** + * A foreign source violated the read protocol — most commonly by returning zero bytes for a positive + * requested count, which IO-17 requires be raised rather than tolerated as end-of-stream or spun on. + * + * @internal + */ +export class SourceContractViolationError extends IoError {} + +/** + * A closed source, sink, buffer, or view was used (IO-42), or a view outlived the parent that invalidated + * it (IO-22). Distinct from `EndOfStreamError` by requirement — IO-24 demands a closed view fail loudly + * with a state error rather than looking like a normal exhaustion. + * + * @internal + */ +export class ClosedResourceError extends IoError { + readonly resource: string; + + constructor(resource: string, options?: ErrorOptions) { + super(`${resource} is closed`, options); + this.resource = resource; + } +} + +/** + * A materialization would exceed the maximum single-array allocation (IO-9). The message points at the + * streaming alternative, as IO-9 requires. + * + * @internal + */ +export class AllocationLimitError extends IoError { + readonly requested: number; + readonly limit: number; + + constructor(requested: number, limit: number, options?: ErrorOptions) { + super( + `cannot materialize ${String(requested)} bytes as one array (limit ${String(limit)}); stream the body instead`, + options, + ); + this.requested = requested; + this.limit = limit; + } +} diff --git a/packages/core/src/io/factories.test.ts b/packages/core/src/io/factories.test.ts new file mode 100644 index 0000000..4fddf2e --- /dev/null +++ b/packages/core/src/io/factories.test.ts @@ -0,0 +1,87 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/factories.test.ts +// Exercises: IO-30 (factory half — fresh, independent, empty buffers; stream, byte-array, and +// foreign-primitive wrapping; the byte-array source is an independent copy), IO-17 (a primitive +// source returning 0 for a positive request fails loudly) +import {describe, expect, test} from 'bun:test'; +import {SourceContractViolationError} from './errors.js'; +import { + bufferedSinkOverPrimitive, + bufferedSinkOverStream, + bufferedSourceOverBytes, + bufferedSourceOverPrimitive, + bufferedSourceOverStream, + newByteQueue, +} from './factories.js'; +import { + collectingWritableStream, + fakeReadableStream, +} from './test-support/fake-stream.js'; +import {rejection} from './test-support/rejection.js'; + +describe('IO-30 factories', () => { + test('two buffers are distinct and both empty', () => { + const first = newByteQueue(); + const second = newByteQueue(); + expect(first).not.toBe(second); + expect(first.size).toBe(0); + expect(second.size).toBe(0); + }); + + test('buffers are independent — writing to one does not affect the other', () => { + const first = newByteQueue(); + const second = newByteQueue(); + first.writeBytes(Uint8Array.from([1, 2])); + expect(second.size).toBe(0); + }); + + test('wrapping a byte array then mutating the input leaves the source unchanged', async () => { + const input = Uint8Array.from([1, 2, 3]); + const source = bufferedSourceOverBytes(input); + input[0] = 99; + expect([...(await source.readBytes())]).toEqual([1, 2, 3]); + }); + + test('wrapping a caller stream produces a readable source', async () => { + const source = bufferedSourceOverStream( + fakeReadableStream([Uint8Array.from([7, 8])]), + ); + expect([...(await source.readBytes())]).toEqual([7, 8]); + }); + + test('wrapping a caller stream produces a writable sink', async () => { + const {stream, written} = collectingWritableStream(); + const sink = bufferedSinkOverStream(stream); + await sink.writeUtf8('hi'); + await sink.close(); + expect(new TextDecoder().decode(written())).toBe('hi'); + }); + + test('wrapping a foreign primitive source supplies the typed reads', async () => { + const backing = newByteQueue(); + backing.writeBytes(Uint8Array.from([1, 2, 3])); + const source = bufferedSourceOverPrimitive({ + read: (dest, count) => backing.read(dest, count), + }); + expect([...(await source.readBytes())]).toEqual([1, 2, 3]); + }); + + test('IO-17: a primitive source returning 0 for a positive request fails loudly', async () => { + const source = bufferedSourceOverPrimitive({read: () => 0}); + expect(await rejection(source.readBytes())).toBeInstanceOf( + SourceContractViolationError, + ); + }); + + test('wrapping a foreign primitive sink supplies the typed writes', async () => { + const collected = newByteQueue(); + const sink = bufferedSinkOverPrimitive({ + write: (src, count) => { + collected.write(src, count); + }, + }); + await sink.writeUtf8('hi'); + await sink.close(); + expect(new TextDecoder().decode(collected.snapshot())).toBe('hi'); + }); +}); diff --git a/packages/core/src/io/factories.ts b/packages/core/src/io/factories.ts new file mode 100644 index 0000000..8768e3c --- /dev/null +++ b/packages/core/src/io/factories.ts @@ -0,0 +1,99 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/factories.ts +import {BufferedSink} from './buffered-sink.js'; +import {BufferedSource} from './buffered-source.js'; +import {ByteQueue} from './byte-queue.js'; +import {SourceContractViolationError} from './errors.js'; +import {END_OF_STREAM} from './limits.js'; + +/** + * IO-30's factory half. Named free functions rather than a namespace object, so the module stays + * tree-shakeable (styleguide 10.1, 15.9). + * + * IO-30's provider-*resolution* half — install precedence, idempotent install, caching, warning, + * de-duplication, and the IO-31–IO-36 rules it defers to — is deliberately not built. There is one + * implementation, always present, requiring no installation call; `sdk-design/03` §3.1 derives this in + * full, and it is the same permanent simplification as SEAM-5–SEAM-10. + * + * @internal + */ + +/** A fresh, independent, empty buffer (IO-30). */ +export function newByteQueue(): ByteQueue { + return new ByteQueue(); +} + +/** Wrap a caller stream as a buffered source (IO-30). */ +export function bufferedSourceOverStream( + stream: ReadableStream, +): BufferedSource { + return BufferedSource.overStream(stream); +} + +/** Wrap a byte array as a buffered source over an independent copy (IO-30). */ +export function bufferedSourceOverBytes(bytes: Uint8Array): BufferedSource { + return BufferedSource.overBytes(bytes); +} + +/** Wrap a caller stream as a buffered sink (IO-30). */ +export function bufferedSinkOverStream( + stream: WritableStream, +): BufferedSink { + return BufferedSink.overStream(stream); +} + +/** + * The raw read protocol of IO-1 — append up to `count` bytes to `dest`'s tail, return the number + * transferred or `END_OF_STREAM` — with none of the typed reads, views, or line semantics. What a + * "foreign primitive" source implements. + */ +export interface PrimitiveSource { + read(dest: ByteQueue, count: number): Promise | number; +} + +/** The raw write protocol of IO-4 — remove exactly `count` bytes from `src`'s head, push downstream. */ +export interface PrimitiveSink { + write(src: ByteQueue, count: number): Promise | void; +} + +/** How much the primitive-source adapter asks for per pull. */ +const PRIMITIVE_CHUNK = 16 * 1024; + +/** Wrap a foreign primitive source with the typed buffered surface (IO-30). */ +export function bufferedSourceOverPrimitive( + source: PrimitiveSource, +): BufferedSource { + const staging = new ByteQueue(); + return BufferedSource.overStream( + new ReadableStream({ + async pull(controller): Promise { + const read = await source.read(staging, PRIMITIVE_CHUNK); + if (read === END_OF_STREAM) { + controller.close(); + return; + } + if (read === 0) { + // IO-17: a zero-byte read for a positive request is a source-contract violation — never + // tolerated as end-of-stream, never spun on. + throw new SourceContractViolationError( + 'foreign source returned 0 bytes for a positive request', + ); + } + controller.enqueue(staging.takeBytes(read)); + }, + }), + ); +} + +/** Wrap a foreign primitive sink with the typed buffered surface (IO-30). */ +export function bufferedSinkOverPrimitive(sink: PrimitiveSink): BufferedSink { + return BufferedSink.overStream( + new WritableStream({ + async write(chunk): Promise { + const staging = new ByteQueue(); + staging.writeBytes(chunk); + await sink.write(staging, staging.size); + }, + }), + ); +} diff --git a/packages/core/src/io/index.ts b/packages/core/src/io/index.ts new file mode 100644 index 0000000..1fcf241 --- /dev/null +++ b/packages/core/src/io/index.ts @@ -0,0 +1,31 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/index.ts +// Internal barrel for product-spec §5 (IO-1–IO-42). +// +// NOTHING here is re-exported from packages/core/src/index.ts. Every symbol is @internal, kept out of +// the api-extractor surface so Phase 3b can promote deliberately (styleguide 10.3) — or not at all, if +// it shapes BODY-1's write-to-sink around the platform's WritableStream instead of BufferedSink. +export {BufferedSink} from './buffered-sink.js'; +export {BufferedSource} from './buffered-source.js'; +export {ByteQueue} from './byte-queue.js'; +export { + AllocationLimitError, + ClosedResourceError, + EndOfStreamError, + IoError, + SourceContractViolationError, +} from './errors.js'; +export { + bufferedSinkOverPrimitive, + bufferedSinkOverStream, + bufferedSourceOverBytes, + bufferedSourceOverPrimitive, + bufferedSourceOverStream, + newByteQueue, + type PrimitiveSink, + type PrimitiveSource, +} from './factories.js'; +export {END_OF_STREAM, MAX_BYTE_ARRAY_LENGTH} from './limits.js'; +export {writeAll} from './pump.js'; +export {RetentionWindow, type Cursor} from './retention-window.js'; +export {TeeSink} from './tee-sink.js'; diff --git a/packages/core/src/io/limits.test.ts b/packages/core/src/io/limits.test.ts new file mode 100644 index 0000000..7f9aa30 --- /dev/null +++ b/packages/core/src/io/limits.test.ts @@ -0,0 +1,23 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/limits.test.ts +// Exercises: IO-1 (the end-of-stream sentinel), IO-9 (maximum single-array allocation) +import {describe, expect, test} from 'bun:test'; +import {END_OF_STREAM, MAX_BYTE_ARRAY_LENGTH} from './limits.js'; + +describe('limits', () => { + test('END_OF_STREAM is the -1 sentinel IO-1 specifies', () => { + expect(END_OF_STREAM).toBe(-1); + }); + + test('MAX_BYTE_ARRAY_LENGTH is a positive safe integer', () => { + expect(Number.isSafeInteger(MAX_BYTE_ARRAY_LENGTH)).toBe(true); + expect(MAX_BYTE_ARRAY_LENGTH).toBeGreaterThan(0); + }); + + test('a Uint8Array of MAX_BYTE_ARRAY_LENGTH is at or under what the host actually allows', () => { + // The constant is deliberately conservative across V8 and JavaScriptCore. This asserts we did not + // pick a number the current host cannot honor; the RangeError backstop in ByteQueue covers hosts + // whose real ceiling is lower still. + expect(MAX_BYTE_ARRAY_LENGTH).toBeLessThanOrEqual(2 ** 32 - 1); + }); +}); diff --git a/packages/core/src/io/limits.ts b/packages/core/src/io/limits.ts new file mode 100644 index 0000000..27520ff --- /dev/null +++ b/packages/core/src/io/limits.ts @@ -0,0 +1,26 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/limits.ts + +/** + * End-of-stream sentinel returned by every read (IO-1). + * + * The numeric protocol is kept spec-literal rather than modelled as `number | undefined`, because IO-2 + * (a zero-count read returns 0 and must NOT report end-of-stream) and, later, BODY-25 ("EOF is signaled + * only by the explicit sentinel") both reason over it. + * + * @internal + */ +export const END_OF_STREAM = -1; + +/** + * Largest byte count this package will attempt to materialize as one contiguous `Uint8Array` (IO-9). + * + * Deliberately conservative. Core is runtime-agnostic, so `node:buffer`'s constant is unavailable; V8 and + * JavaScriptCore disagree on the real ceiling and both have moved it, and rule 12.6 forbids probing at + * import time. 2 GiB − 1 is at or below every supported host's limit. Callers that exceed it get an + * actionable `AllocationLimitError` rather than a low-level allocation crash; a `RangeError` backstop at + * the allocation site covers any host whose real ceiling is lower still. + * + * @internal + */ +export const MAX_BYTE_ARRAY_LENGTH = 2 ** 31 - 1; diff --git a/packages/core/src/io/pump.test.ts b/packages/core/src/io/pump.test.ts new file mode 100644 index 0000000..80edb15 --- /dev/null +++ b/packages/core/src/io/pump.test.ts @@ -0,0 +1,45 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/pump.test.ts +// Exercises: IO-17 (pump to exhaustion, terminate only on the EOF sentinel, raise a zero-read for a +// positive request as a source-contract violation) +import {describe, expect, test} from 'bun:test'; +import {BufferedSink} from './buffered-sink.js'; +import {BufferedSource} from './buffered-source.js'; +import {SourceContractViolationError} from './errors.js'; +import {writeAll} from './pump.js'; +import { + collectingWritableStream, + fakeReadableStream, + protocolViolatingStream, +} from './test-support/fake-stream.js'; +import {rejection} from './test-support/rejection.js'; + +describe('writeAll (IO-17)', () => { + test('pumps the source to exhaustion and returns the total transferred', async () => { + const source = BufferedSource.overStream( + fakeReadableStream([Uint8Array.from([1, 2]), Uint8Array.from([3, 4, 5])]), + ); + const {stream, written} = collectingWritableStream(); + const sink = BufferedSink.overStream(stream); + + expect(await writeAll(source, sink)).toBe(5); + await sink.close(); + expect([...written()]).toEqual([1, 2, 3, 4, 5]); + }); + + test('an already-exhausted source transfers zero and does not hang', async () => { + const source = BufferedSource.overStream(fakeReadableStream([])); + const {stream} = collectingWritableStream(); + expect(await writeAll(source, BufferedSink.overStream(stream))).toBe(0); + }); + + test('a source returning zero bytes for a positive request is a contract violation', async () => { + // Never tolerated as end-of-stream, and never spun on forever — a misbehaving foreign source must + // fail loudly rather than hang or truncate a body. + const source = BufferedSource.overStream(protocolViolatingStream()); + const {stream} = collectingWritableStream(); + expect( + await rejection(writeAll(source, BufferedSink.overStream(stream))), + ).toBeInstanceOf(SourceContractViolationError); + }); +}); diff --git a/packages/core/src/io/pump.ts b/packages/core/src/io/pump.ts new file mode 100644 index 0000000..32ea133 --- /dev/null +++ b/packages/core/src/io/pump.ts @@ -0,0 +1,32 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/pump.ts +import type {BufferedSink} from './buffered-sink.js'; +import type {BufferedSource} from './buffered-source.js'; +import {ByteQueue} from './byte-queue.js'; +import {END_OF_STREAM} from './limits.js'; + +/** How much the pump asks for per iteration. */ +const PUMP_CHUNK = 16 * 1024; + +/** + * Pump `source` to exhaustion into `sink` and return the total bytes transferred (IO-17). + * + * Terminates only on the end-of-stream sentinel. A zero-byte read for a non-zero requested count is a + * source-contract violation raised by the source itself — never tolerated here as end-of-stream, and + * never spun on. + * + * @internal + */ +export async function writeAll( + source: BufferedSource, + sink: BufferedSink, +): Promise { + const staging = new ByteQueue(); + let total = 0; + for (;;) { + const read = await source.read(staging, PUMP_CHUNK); + if (read === END_OF_STREAM) return total; + await sink.write(staging, staging.size); + total += read; + } +} diff --git a/packages/core/src/io/retention-window.test.ts b/packages/core/src/io/retention-window.test.ts new file mode 100644 index 0000000..4888d8d --- /dev/null +++ b/packages/core/src/io/retention-window.test.ts @@ -0,0 +1,110 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/retention-window.test.ts +// Exercises: IO-19/IO-20 (non-consuming views), IO-22 (parent close invalidates views), +// IO-23 (mutually independent cursors), IO-24 (closed view fails loudly, distinct from EOF) +import {describe, expect, test} from 'bun:test'; +import {ByteQueue} from './byte-queue.js'; +import {ClosedResourceError} from './errors.js'; +import {RetentionWindow} from './retention-window.js'; +import {fakeReadableStream} from './test-support/fake-stream.js'; + +const bytes = (...values: number[]): Uint8Array => Uint8Array.from(values); + +const windowOver = (...chunks: Uint8Array[]): RetentionWindow => + new RetentionWindow(fakeReadableStream(chunks).getReader()); + +describe('RetentionWindow', () => { + test('pullThrough pulls until the requested logical offset is available', async () => { + const window = windowOver(bytes(1, 2), bytes(3, 4)); + expect(await window.pullThrough(3)).toBe(true); + expect(window.pulledThrough).toBeGreaterThanOrEqual(3); + }); + + test('pullThrough returns false once the stream is exhausted', async () => { + const window = windowOver(bytes(1, 2)); + expect(await window.pullThrough(5)).toBe(false); + expect(window.pulledThrough).toBe(2); + }); + + test('readInto advances only the cursor it is given', async () => { + const window = windowOver(bytes(1, 2, 3, 4)); + const first = window.register(0); + const second = window.register(0); + await window.pullThrough(4); + + const dest = new ByteQueue(); + expect(window.readInto(first, dest, 2)).toBe(2); + expect(first.at).toBe(2); + expect(second.at).toBe(0); + }); + + test('IO-23: two cursors read the same bytes independently', async () => { + const window = windowOver(bytes(1, 2, 3)); + const first = window.register(0); + const second = window.register(0); + await window.pullThrough(3); + + const a = new ByteQueue(); + const b = new ByteQueue(); + window.readInto(first, a, 3); + window.readInto(second, b, 3); + expect([...a.snapshot()]).toEqual([1, 2, 3]); + expect([...b.snapshot()]).toEqual([1, 2, 3]); + }); + + test('bytes behind the slowest cursor are trimmed, bytes at or ahead of it are retained', async () => { + const window = windowOver(bytes(1, 2, 3, 4)); + const fast = window.register(0); + const slow = window.register(0); + await window.pullThrough(4); + + window.readInto(fast, new ByteQueue(), 4); + expect(window.retainedBytes).toBe(4); // slow still needs all four + + window.readInto(slow, new ByteQueue(), 4); + expect(window.retainedBytes).toBe(0); // nobody needs them now + }); +}); + +describe('RetentionWindow trim, peek, and close', () => { + test('releasing a cursor lets the head trim forward', async () => { + const window = windowOver(bytes(1, 2, 3, 4)); + const fast = window.register(0); + const slow = window.register(0); + await window.pullThrough(4); + window.readInto(fast, new ByteQueue(), 4); + + window.release(slow); + expect(window.retainedBytes).toBe(0); + }); + + test('peekBytes materializes without advancing the cursor', async () => { + const window = windowOver(bytes(1, 2, 3)); + const cursor = window.register(0); + await window.pullThrough(3); + expect([...window.peekBytes(cursor, 2)]).toEqual([1, 2]); + expect(cursor.at).toBe(0); + }); + + test('IO-22/IO-24: after close, any cursor use throws ClosedResourceError, not an EOF', async () => { + const window = windowOver(bytes(1, 2, 3)); + const cursor = window.register(0); + await window.pullThrough(3); + window.close(); + + expect(() => { + window.assertUsable(); + }).toThrow(ClosedResourceError); + expect(() => window.readInto(cursor, new ByteQueue(), 1)).toThrow( + ClosedResourceError, + ); + }); + + test('IO-41: close is idempotent', () => { + const window = windowOver(bytes(1)); + window.close(); + expect(() => { + window.close(); + }).not.toThrow(); + }); +}); diff --git a/packages/core/src/io/retention-window.ts b/packages/core/src/io/retention-window.ts new file mode 100644 index 0000000..2d4ac4b --- /dev/null +++ b/packages/core/src/io/retention-window.ts @@ -0,0 +1,168 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/retention-window.ts +import {invariant} from '../invariant.js'; +import {ByteQueue} from './byte-queue.js'; +import {ClosedResourceError, SourceContractViolationError} from './errors.js'; + +/** + * A reader's position, as a logical offset into the whole stream. Two cursors over one window are + * mutually independent (IO-23): advancing one never moves another. + * + * @internal + */ +export interface Cursor { + at: number; +} + +/** + * The shared buffer behind a `BufferedSource` and all of its peek/slice views. + * + * Bytes are retained from `min(all live cursors)` forward and trimmed as the slowest cursor advances, so + * with no views outstanding retention collapses to the read size. There is deliberately **no cap** here: + * §5 bounds nothing, and every cap the product spec mandates (BODY-19, BODY-30/HTTP-52, BODY-34) sits in + * §6 and belongs to Phase 3b. A cap at this layer would bound the spread between the fastest and slowest + * cursor, which in the divergent case stops a view reaching the end and partially fails IO-19's MUST. + * + * Owns the stream reader, so a view — which owns no reader — can still pull through its parent's source. + * + * @internal + */ +export class RetentionWindow { + readonly #queue = new ByteQueue(); + readonly #cursors = new Set(); + readonly #reader: ReadableStreamDefaultReader | undefined; + #retainedFrom = 0; + #pulledThrough = 0; + #exhausted = false; + #closed = false; + + constructor(reader: ReadableStreamDefaultReader | undefined) { + this.#reader = reader; + this.#exhausted = reader === undefined; + } + + /** Logical offset one past the last byte pulled from the stream. */ + get pulledThrough(): number { + return this.#pulledThrough; + } + + /** Bytes currently held because some cursor may still need them. */ + get retainedBytes(): number { + return this.#queue.size; + } + + get closed(): boolean { + return this.#closed; + } + + /** Register a new cursor at a logical offset (IO-23 — its own cursor, independent of every other). */ + register(at: number): Cursor { + this.assertUsable(); + const cursor: Cursor = {at}; + this.#cursors.add(cursor); + return cursor; + } + + /** + * Drop a cursor and let the retained head trim forward (IO-22 — releasing a view neither closes the + * parent nor moves the parent's cursor). + */ + release(cursor: Cursor): void { + this.#cursors.delete(cursor); + if (!this.#closed) this.#trim(); + } + + /** + * Pull from the stream until `offset` bytes are available, or the stream ends. Returns false at end. + */ + async pullThrough(offset: number): Promise { + this.assertUsable(); + while (this.#pulledThrough < offset && !this.#exhausted) { + await this.#pullOnce(); + this.assertUsable(); + } + return this.#pulledThrough >= offset; + } + + /** Move up to `count` already-pulled bytes onto `dest`, advancing only `cursor`. */ + readInto(cursor: Cursor, dest: ByteQueue, count: number): number { + this.assertUsable(); + const take = Math.min(count, this.#pulledThrough - cursor.at); + if (take <= 0) return 0; + this.#queue.copyTo(dest, cursor.at - this.#retainedFrom, take); + cursor.at += take; + this.#trim(); + return take; + } + + /** Materialize up to `count` already-pulled bytes without advancing `cursor` (IO-19, IO-20). */ + peekBytes(cursor: Cursor, count: number): Uint8Array { + this.assertUsable(); + const take = Math.min(count, this.#pulledThrough - cursor.at); + if (take <= 0) return new Uint8Array(0); + const staging = new ByteQueue(); + this.#queue.copyTo(staging, cursor.at - this.#retainedFrom, take); + return staging.snapshot(); + } + + /** How many pulled bytes sit at or ahead of `cursor`. Does not pull and does not advance. */ + availableFrom(cursor: Cursor): number { + this.assertUsable(); + return Math.max(0, this.#pulledThrough - cursor.at); + } + + /** IO-24: a closed window fails loudly with a state error, never as a normal EOF. */ + assertUsable(): void { + if (this.#closed) throw new ClosedResourceError('BufferedSource'); + } + + /** + * IO-41: idempotent. IO-22: invalidates every outstanding view, so a later read from one fails loudly + * rather than returning stale bytes. + */ + close(): void { + if (this.#closed) return; + this.#closed = true; + this.#cursors.clear(); + this.#queue.clear(); + this.#queue.close(); + // The reader lock must be released even if the stream already errored; a rejection here would + // otherwise become an unhandled rejection on a teardown path. + void this.#reader?.cancel().catch(() => undefined); + } + + async #pullOnce(): Promise { + invariant(this.#reader !== undefined, 'pull on a window with no reader'); + const {done, value} = await this.#reader.read(); + if (done) { + this.#exhausted = true; + return; + } + // `done: false` narrows `value` to a defined `Uint8Array` per the Streams spec's discriminated + // union — no runtime check needed on top of what the type already guarantees. + if (value.length === 0) { + // IO-17: a zero-length delivery for an outstanding read is a source-contract violation, never + // end-of-stream and never something to spin on. + throw new SourceContractViolationError( + 'source delivered 0 bytes without signalling end of stream', + ); + } + this.#queue.writeBytes(value); + this.#pulledThrough += value.length; + } + + /** Drop everything no live cursor can still reach. */ + #trim(): void { + const low = this.#lowestCursor(); + const drop = low - this.#retainedFrom; + if (drop <= 0) return; + this.#queue.skip(drop); + this.#retainedFrom = low; + } + + #lowestCursor(): number { + let low = this.#pulledThrough; + for (const cursor of this.#cursors) low = Math.min(low, cursor.at); + return low; + } +} diff --git a/packages/core/src/io/tee-sink.test.ts b/packages/core/src/io/tee-sink.test.ts new file mode 100644 index 0000000..8759341 --- /dev/null +++ b/packages/core/src/io/tee-sink.test.ts @@ -0,0 +1,213 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/tee-sink.test.ts +// Exercises: IO-25 (mirror into a tap AND forward the full untruncated payload), +// IO-26 (tap capacity limit; unbounded default; a limit of 0 mirrors nothing), +// IO-27 (mirror BEFORE forwarding; staging cleared even on a failed write), +// IO-28 (no direct backing-buffer handle), IO-29 (flush/close/emit forward to the primary only), +// IO-42 (write after close rejects with the source intact), +// IO-13 (the tap mirrors the primary's exact encoded bytes, and refuses a label identically) +import {describe, expect, test} from 'bun:test'; +import fc from 'fast-check'; +import {BufferedSink} from './buffered-sink.js'; +import {ByteQueue} from './byte-queue.js'; +import {ClosedResourceError} from './errors.js'; +import {TeeSink} from './tee-sink.js'; +import { + collectingWritableStream, + failingWritableStream, +} from './test-support/fake-stream.js'; +import {rejection} from './test-support/rejection.js'; + +const queueOf = (bytes: Uint8Array): ByteQueue => { + const queue = new ByteQueue(); + queue.writeBytes(bytes); + return queue; +}; + +describe('TeeSink', () => { + test('IO-25: the primary receives the full payload and the tap mirrors it', async () => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.write(queueOf(Uint8Array.from([1, 2, 3])), 3); + await tee.close(); + expect([...written()]).toEqual([1, 2, 3]); + expect([...tee.snapshot()]).toEqual([1, 2, 3]); + }); + + test('IO-26: past the tap limit the tap stops copying but the primary still gets everything', async () => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream), 2); + await tee.write(queueOf(Uint8Array.from([1, 2, 3, 4, 5])), 5); + await tee.close(); + expect([...written()]).toEqual([1, 2, 3, 4, 5]); + expect([...tee.snapshot()]).toEqual([1, 2]); + }); + + test('IO-26: a limit of 0 mirrors nothing and forwards everything', async () => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream), 0); + await tee.write(queueOf(Uint8Array.from([1, 2, 3])), 3); + await tee.close(); + expect([...written()]).toEqual([1, 2, 3]); + expect([...tee.snapshot()]).toEqual([]); + }); + + test('IO-26: the default limit mirrors everything', async () => { + const {stream} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.write(queueOf(new Uint8Array(10_000).fill(7)), 10_000); + await tee.close(); + expect(tee.snapshot().length).toBe(10_000); + }); +}); + +describe('TeeSink mirror-before-forward and lifecycle (IO-27, IO-28, IO-29, IO-42)', () => { + test('IO-27: a failed primary write still captures the attempted bytes in the tap', async () => { + const tee = new TeeSink( + BufferedSink.overStream(failingWritableStream('primary down')), + ); + expect( + (await rejection(tee.write(queueOf(Uint8Array.from([1, 2, 3])), 3))) + .message, + ).toContain('primary down'); + await Promise.resolve(); + expect([...tee.snapshot()]).toEqual([1, 2, 3]); + }); + + test("IO-27: a write following a FAILED write does not prepend the failed write's bytes", async () => { + // The staging buffer is per-call, so this holds structurally — but the assertion has to actually + // drive the failure path to prove it, which is why the first sink is the failing one. + const failing = new TeeSink( + BufferedSink.overStream(failingWritableStream('primary down')), + ); + expect( + (await rejection(failing.write(queueOf(Uint8Array.from([1, 2])), 2))) + .message, + ).toContain('primary down'); + + const {stream, written} = collectingWritableStream(); + const good = new TeeSink(BufferedSink.overStream(stream)); + await good.write(queueOf(Uint8Array.from([3])), 1); + await good.close(); + expect([...written()]).toEqual([3]); + }); + + test('IO-27: consecutive successful writes concatenate without duplication or reordering', async () => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.write(queueOf(Uint8Array.from([1, 2])), 2); + await tee.write(queueOf(Uint8Array.from([3])), 1); + await tee.close(); + expect([...written()]).toEqual([1, 2, 3]); + }); +}); + +describe('TeeSink no-raw-buffer and close (IO-28, IO-29, IO-42)', () => { + test('IO-28: there is no direct backing-buffer handle', () => { + const {stream} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + expect(() => tee.buffer).toThrow( + 'TeeSink exposes no backing buffer; use the typed write methods', + ); + }); + + test('IO-29: close forwards to the primary and leaves the tap intact for later snapshotting', async () => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.write(queueOf(Uint8Array.from([1, 2])), 2); + await tee.close(); + expect([...written()]).toEqual([1, 2]); + expect([...tee.snapshot()]).toEqual([1, 2]); + }); + + test('IO-29: flush and emit return the tee and leave the tap intact', async () => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.write(queueOf(Uint8Array.from([1, 2])), 2); + expect(await tee.flush()).toBe(tee); + expect(await tee.emit()).toBe(tee); + expect([...tee.snapshot()]).toEqual([1, 2]); + await tee.close(); + expect([...written()]).toEqual([1, 2]); + }); + + test('IO-29: flush and emit really reach the primary — a closed primary makes both reject', async () => { + // The observable proof that neither is swallowed by the decorator: BufferedSink rejects a flush + // or emit after close (IO-42), so the rejection can only have come from the primary. + const {stream} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.close(); + expect(await rejection(tee.flush())).toBeInstanceOf(ClosedResourceError); + expect(await rejection(tee.emit())).toBeInstanceOf(ClosedResourceError); + }); +}); + +describe('TeeSink text writes (IO-13, IO-25)', () => { + test('IO-25: writeUtf8 forwards the encoded bytes and mirrors exactly those bytes', async () => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.writeUtf8('héllo ☃'); + await tee.close(); + expect([...tee.snapshot()]).toEqual([...written()]); + expect(new TextDecoder('utf-8').decode(written())).toBe('héllo ☃'); + }); + + test('IO-13: writeString mirrors the charset-encoded bytes, not a UTF-8 re-encoding', async () => { + // 'é' is one byte in ISO-8859-1 and two in UTF-8, so a tap that re-encoded would differ from the + // wire body — the exact divergence the single shared `encodeText` exists to prevent. + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.writeString('hé', 'iso-8859-1'); + await tee.close(); + expect([...written()]).toEqual([0x68, 0xe9]); + expect([...tee.snapshot()]).toEqual([0x68, 0xe9]); + }); + + test('IO-13: an unsupported charset is refused before anything is mirrored or forwarded', async () => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + expect( + (await rejection(tee.writeString('x', 'shift_jis'))).message, + ).toContain( + 'unsupported write charset: shift_jis (only utf-8 and iso-8859-1 can be encoded)', + ); + await tee.close(); + expect([...tee.snapshot()]).toEqual([]); + expect([...written()]).toEqual([]); + }); + + test('IO-42: write after close rejects and leaves the source intact', async () => { + const {stream} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream)); + await tee.close(); + const source = queueOf(Uint8Array.from([1, 2])); + expect(await rejection(tee.write(source, 2))).toBeInstanceOf( + ClosedResourceError, + ); + expect(source.size).toBe(2); + expect([...tee.snapshot()]).toEqual([]); + }); + + test('IO-25 property: the primary always receives the exact concatenation of every written byte', async () => { + // The single most important property in §5: logging never reduces the wire body, whatever the cap. + await fc.assert( + fc.asyncProperty( + fc.array(fc.uint8Array({maxLength: 32}), {maxLength: 8}), + fc.integer({min: 0, max: 64}), + async (writes, tapLimit) => { + const {stream, written} = collectingWritableStream(); + const tee = new TeeSink(BufferedSink.overStream(stream), tapLimit); + for (const chunk of writes) + await tee.write(queueOf(chunk), chunk.length); + await tee.close(); + + const expected = writes.flatMap(chunk => [...chunk]); + expect([...written()]).toEqual(expected); + expect(tee.snapshot().length).toBe( + Math.min(tapLimit, expected.length), + ); + }, + ), + ); + }); +}); diff --git a/packages/core/src/io/tee-sink.ts b/packages/core/src/io/tee-sink.ts new file mode 100644 index 0000000..129167c --- /dev/null +++ b/packages/core/src/io/tee-sink.ts @@ -0,0 +1,113 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/tee-sink.ts +import {invariant} from '../invariant.js'; +import {encodeText, type BufferedSink} from './buffered-sink.js'; +import {ByteQueue} from './byte-queue.js'; +import {ClosedResourceError, IoError} from './errors.js'; + +/** + * A sink that mirrors written bytes into a bounded in-memory tap while forwarding the full, untruncated + * payload to its primary (IO-25–IO-29). + * + * Built as a plain `BufferedSink` decorator rather than on `TransformStream`, which `sdk-design/03` §3.1 + * sketches: a `TransformStream`'s own queueing and backpressure semantics muddy IO-27's + * mirror-before-forward ordering, the clause most easily gotten wrong. §3.1's substantive point — that + * the platform's `ReadableStream.tee()` solves a different problem (duplicating a *readable* for two + * consumers, not mirroring a *sink's* writes) — is why no platform primitive is used at all. + * + * The tap has no cap by default. §5 bounds nothing; BODY-19 and BODY-34 set the real cap in Phase 3b. + * + * @internal + */ +export class TeeSink { + readonly #primary: BufferedSink; + readonly #tap = new ByteQueue(); + readonly #tapLimit: number; + + constructor( + primary: BufferedSink, + tapLimit: number = Number.POSITIVE_INFINITY, + ) { + invariant( + tapLimit >= 0, + `tapLimit must be non-negative, got ${String(tapLimit)}`, + ); + this.#primary = primary; + this.#tapLimit = tapLimit; + } + + /** + * IO-28: a raw buffer write would reach only the tap or only the primary and silently corrupt the wire + * body, so no such handle exists. + */ + get buffer(): never { + throw new IoError( + 'TeeSink exposes no backing buffer; use the typed write methods', + ); + } + + /** Mirror into the tap, then forward the full payload to the primary (IO-25, IO-27). */ + async write(src: ByteQueue, count: number): Promise { + // IO-42: reject before consuming from `src` or touching the tap, so a caller that catches the + // rejection still holds its bytes — matching BufferedSink, which rejects before takeBytes. + if (this.#primary.closed) throw new ClosedResourceError('TeeSink'); + const staging = new ByteQueue(); + staging.write(src, count); + // IO-27: mirror BEFORE forwarding, so a failed primary write still captures the attempted bytes. + this.#mirror(staging); + // IO-27: staging is drained by the forward, so a later write cannot prepend stale bytes; the + // `finally` guarantees that holds even when the primary throws. + try { + await this.#primary.write(staging, count); + } finally { + staging.clear(); + } + } + + /** Mirror and forward UTF-8 text (IO-25). */ + async writeUtf8(text: string): Promise { + return this.writeString(text, 'utf-8'); + } + + /** + * Mirror and forward text with an explicit charset (IO-25). + * + * Encodes once, through the sink's own `encodeText`, then routes the bytes down the normal `write` + * path. That guarantees the tap mirrors exactly the bytes the primary emits — not a UTF-8 re-encoding + * of them — and that an unsupported charset is refused identically on both sides. + */ + async writeString(text: string, charset: string): Promise { + const encoded = new ByteQueue(); + encoded.writeBytes(encodeText(text, charset)); + return this.write(encoded, encoded.size); + } + + /** A non-consuming copy of the tap's contents. */ + snapshot(): Uint8Array { + return this.#tap.snapshot(); + } + + /** IO-29: forwards to the PRIMARY only, leaving the tap intact. */ + async flush(): Promise { + await this.#primary.flush(); + return this; + } + + /** IO-29: forwards to the PRIMARY only, leaving the tap intact. */ + async emit(): Promise { + await this.#primary.emit(); + return this; + } + + /** IO-29: forwards to the PRIMARY only; the tap survives for later snapshotting. */ + async close(): Promise { + await this.#primary.close(); + } + + /** IO-26: copy until the cap is reached, then stop copying while the payload still forwards. */ + #mirror(staging: ByteQueue): void { + const room = this.#tapLimit - this.#tap.size; + if (room <= 0) return; + staging.copyTo(this.#tap, 0, Math.min(room, staging.size)); + } +} diff --git a/packages/core/src/io/test-support/fake-stream.ts b/packages/core/src/io/test-support/fake-stream.ts new file mode 100644 index 0000000..b7e2e06 --- /dev/null +++ b/packages/core/src/io/test-support/fake-stream.ts @@ -0,0 +1,71 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/test-support/fake-stream.ts +// Test-only. Excluded from the build (tsconfig.build.json) and never exported from any barrel. +// Styleguide 11.3: fake your own interfaces rather than reaching for mock.module. + +/** A readable stream that yields exactly the chunks given, at exactly those boundaries. */ +export function fakeReadableStream( + chunks: readonly Uint8Array[], + onCancel?: () => void, +): ReadableStream { + let index = 0; + return new ReadableStream({ + cancel(): void { + onCancel?.(); + }, + pull(controller): void { + if (index >= chunks.length) { + controller.close(); + return; + } + const chunk = chunks[index]; + index += 1; + if (chunk !== undefined) controller.enqueue(chunk); + }, + }); +} + +/** A readable stream that violates the read protocol by yielding an empty chunk (drives IO-17). */ +export function protocolViolatingStream(): ReadableStream { + return fakeReadableStream([new Uint8Array(0)]); +} + +/** A writable stream that accumulates everything written, for asserting the wire payload. */ +export function collectingWritableStream(): { + stream: WritableStream; + written: () => Uint8Array; + isClosed: () => boolean; +} { + const parts: Uint8Array[] = []; + let closed = false; + const stream = new WritableStream({ + write(chunk): void { + parts.push(chunk.slice()); + }, + close(): void { + closed = true; + }, + }); + const written = (): Uint8Array => { + const total = parts.reduce((sum, part) => sum + part.length, 0); + const out = new Uint8Array(total); + let at = 0; + for (const part of parts) { + out.set(part, at); + at += part.length; + } + return out; + }; + return {stream, written, isClosed: () => closed}; +} + +/** A writable stream whose first write rejects, for asserting failure-path behavior. */ +export function failingWritableStream( + message: string, +): WritableStream { + return new WritableStream({ + write(): never { + throw new Error(message); + }, + }); +} diff --git a/packages/core/src/io/test-support/rejection.ts b/packages/core/src/io/test-support/rejection.ts new file mode 100644 index 0000000..94dfb38 --- /dev/null +++ b/packages/core/src/io/test-support/rejection.ts @@ -0,0 +1,22 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/test-support/rejection.ts +// Test-only. Excluded from the build (tsconfig.build.json) and never exported from any barrel. + +/** + * Await a promise that must reject, and return the rejection reason. + * + * Why this exists rather than `expect(promise).rejects.toThrow(...)`: bun types `rejects` as + * `Matchers`, whose `toThrow()` returns `void` even though at run time it returns a promise. + * So `await`ing it fails `@typescript-eslint/await-thenable`, and omitting the `await` leaves the + * assertion racing test teardown — bun still fails the run, but the failure can attribute to a later + * test. Capturing the rejection keeps every failure awaited and attributable, with no lint suppression. + */ +export async function rejection(promise: Promise): Promise { + try { + await promise; + } catch (e: unknown) { + if (e instanceof Error) return e; + throw new Error(`expected an Error rejection, got ${typeof e}`); + } + throw new Error('expected the promise to reject, but it resolved'); +} diff --git a/packages/core/tsconfig.build.json b/packages/core/tsconfig.build.json index d39dc7e..a8c27f3 100644 --- a/packages/core/tsconfig.build.json +++ b/packages/core/tsconfig.build.json @@ -6,6 +6,8 @@ "sourceMap": true }, "exclude": [ - "src/**/*.test.ts" + "src/**/*.test.ts", + "src/**/*.bench.ts", + "src/io/test-support/**" ] } diff --git a/tsconfig.base.json b/tsconfig.base.json index 0c1d0df..7c876c0 100644 --- a/tsconfig.base.json +++ b/tsconfig.base.json @@ -6,7 +6,8 @@ "moduleResolution": "nodenext", "lib": [ "ES2022", - "DOM" + "DOM", + "DOM.AsyncIterable" ], "noUncheckedIndexedAccess": true, "exactOptionalPropertyTypes": true, From 3f7978a2a6fcba7612ffc565654dc7404aed9936 Mon Sep 17 00:00:00 2001 From: Mohammad Wahbeh Date: Tue, 11 Aug 2026 10:19:27 +0300 Subject: [PATCH 2/2] test(core): close per-file coverage gaps in invariant, io/errors, rejection helper. --- packages/core/src/invariant.test.ts | 29 ++++++++++++++ packages/core/src/invariant.ts | 5 ++- packages/core/src/io/errors.test.ts | 16 ++++++++ packages/core/src/io/errors.ts | 19 ++++++++- .../src/io/test-support/rejection.test.ts | 40 +++++++++++++++++++ 5 files changed, 106 insertions(+), 3 deletions(-) create mode 100644 packages/core/src/invariant.test.ts create mode 100644 packages/core/src/io/test-support/rejection.test.ts diff --git a/packages/core/src/invariant.test.ts b/packages/core/src/invariant.test.ts new file mode 100644 index 0000000..be3fd82 --- /dev/null +++ b/packages/core/src/invariant.test.ts @@ -0,0 +1,29 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/invariant.test.ts +// Exercises: the project's sole assertion primitive (styleguide 5.6) and its error class. +import {describe, expect, test} from 'bun:test'; +import {invariant, InvariantViolation} from './invariant.js'; + +describe('invariant', () => { + test('does not throw when the condition is truthy', () => { + expect(() => { + invariant(true, 'unreachable'); + }).not.toThrow(); + }); + + test('throws InvariantViolation with the given message when the condition is falsy', () => { + expect(() => { + invariant(false, 'broken precondition'); + }).toThrow(InvariantViolation); + expect(() => { + invariant(false, 'broken precondition'); + }).toThrow('broken precondition'); + }); + + test('InvariantViolation sets its name and descends from Error', () => { + const error = new InvariantViolation('boom'); + expect(error).toBeInstanceOf(Error); + expect(error.name).toBe('InvariantViolation'); + expect(error.message).toBe('boom'); + }); +}); diff --git a/packages/core/src/invariant.ts b/packages/core/src/invariant.ts index bec6ce9..f0da862 100644 --- a/packages/core/src/invariant.ts +++ b/packages/core/src/invariant.ts @@ -10,7 +10,10 @@ * @internal */ export class InvariantViolation extends Error { - override readonly name = 'InvariantViolation'; + constructor(msg: string) { + super(msg); + this.name = 'InvariantViolation'; + } } /** diff --git a/packages/core/src/io/errors.test.ts b/packages/core/src/io/errors.test.ts index 4839dee..4038132 100644 --- a/packages/core/src/io/errors.test.ts +++ b/packages/core/src/io/errors.test.ts @@ -59,4 +59,20 @@ describe('IoError tree', () => { const cause = new RangeError('array too large'); expect(new AllocationLimitError(5, 4, {cause}).cause).toBe(cause); }); + + test('EndOfStreamError chains a cause', () => { + const cause = new Error('underlying read failure'); + expect(new EndOfStreamError(1, 2, {cause}).cause).toBe(cause); + }); + + test('ClosedResourceError chains a cause', () => { + const cause = new Error('already closed'); + expect(new ClosedResourceError('ByteQueue', {cause}).cause).toBe(cause); + }); + + test('SourceContractViolationError carries its message and descends from IoError', () => { + const error = new SourceContractViolationError('returned zero bytes'); + expect(error.message).toBe('returned zero bytes'); + expect(error.name).toBe('SourceContractViolationError'); + }); }); diff --git a/packages/core/src/io/errors.ts b/packages/core/src/io/errors.ts index f66b4c9..bc53c0c 100644 --- a/packages/core/src/io/errors.ts +++ b/packages/core/src/io/errors.ts @@ -10,7 +10,15 @@ import {DexpaceError} from '../http/errors.js'; * * @internal */ -export class IoError extends DexpaceError {} +export class IoError extends DexpaceError { + // bun's coverage tool never marks a bodiless subclass's implicit constructor as covered + // (undercounts function coverage); an explicit forwarding constructor is instrumented + // correctly and keeps the file above the 80% function-coverage floor without changing behavior. + // eslint-disable-next-line @typescript-eslint/no-useless-constructor -- see comment above + constructor(message: string, options?: ErrorOptions) { + super(message, options); + } +} /** * A source ended before delivering the requested number of bytes (IO-11, IO-12, IO-15), or a sink write @@ -38,7 +46,14 @@ export class EndOfStreamError extends IoError { * * @internal */ -export class SourceContractViolationError extends IoError {} +export class SourceContractViolationError extends IoError { + // See IoError's constructor above: keeps this bodiless subclass registered for bun's + // function coverage. + // eslint-disable-next-line @typescript-eslint/no-useless-constructor -- see comment above + constructor(message: string, options?: ErrorOptions) { + super(message, options); + } +} /** * A closed source, sink, buffer, or view was used (IO-42), or a view outlived the parent that invalidated diff --git a/packages/core/src/io/test-support/rejection.test.ts b/packages/core/src/io/test-support/rejection.test.ts new file mode 100644 index 0000000..5c83f30 --- /dev/null +++ b/packages/core/src/io/test-support/rejection.test.ts @@ -0,0 +1,40 @@ +// SPDX-License-Identifier: MIT +// packages/core/src/io/test-support/rejection.test.ts +// Exercises the `rejection()` test helper's own failure paths, not covered by its many callers +// (which all reject with a real Error). +import {describe, expect, test} from 'bun:test'; +import {rejection} from './rejection.js'; + +describe('rejection', () => { + test('returns the rejection reason when the promise rejects with an Error', async () => { + const error = new Error('boom'); + expect(await rejection(Promise.reject(error))).toBe(error); + }); + + test('throws when the promise rejects with a non-Error value', async () => { + let caught: unknown; + try { + // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors -- exercising rejection()'s non-Error branch + await rejection(Promise.reject('not an error')); + } catch (e: unknown) { + caught = e; + } + expect(caught).toBeInstanceOf(Error); + expect((caught as Error).message).toBe( + 'expected an Error rejection, got string', + ); + }); + + test('throws when the promise resolves instead of rejecting', async () => { + let caught: unknown; + try { + await rejection(Promise.resolve('fine')); + } catch (e: unknown) { + caught = e; + } + expect(caught).toBeInstanceOf(Error); + expect((caught as Error).message).toBe( + 'expected the promise to reject, but it resolved', + ); + }); +});