From 98c377dfc308996da2ab80bbe4c232ba04ef8c38 Mon Sep 17 00:00:00 2001 From: Elliott Johnson Date: Mon, 5 Oct 2026 15:43:26 -0600 Subject: [PATCH 1/6] feat: add `unevalStream` --- .changeset/uneval-stream.md | 5 + README.md | 42 +++ src/index.js | 1 + src/types.ts | 26 ++ src/uneval-stream.js | 528 +++++++++++++++++++++++++++ test/README.md | 2 + test/helpers/stream.js | 277 ++++++++++++++ test/helpers/subprocess.js | 27 ++ test/stream-primitive-growth.test.js | 202 ++++++++++ test/stream-reference-growth.test.js | 382 +++++++++++++++++++ test/stream-regressions.test.js | 152 ++++++++ test/stream-retention.test.js | 306 ++++++++++++++++ test/stream.test.js | 425 +++++++++++++++++++++ 13 files changed, 2375 insertions(+) create mode 100644 .changeset/uneval-stream.md create mode 100644 src/uneval-stream.js create mode 100644 test/helpers/stream.js create mode 100644 test/helpers/subprocess.js create mode 100644 test/stream-primitive-growth.test.js create mode 100644 test/stream-reference-growth.test.js create mode 100644 test/stream-regressions.test.js create mode 100644 test/stream-retention.test.js create mode 100644 test/stream.test.js diff --git a/.changeset/uneval-stream.md b/.changeset/uneval-stream.md new file mode 100644 index 00000000..2f9594bf --- /dev/null +++ b/.changeset/uneval-stream.md @@ -0,0 +1,5 @@ +--- +'devalue': minor +--- + +feat: Add `unevalStream` for streaming asynchronous values diff --git a/README.md b/README.md index e92d8814..85ecb5bc 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ Like `JSON.stringify`, but handles - `Temporal` - custom types via replacers, reducers and revivers - promises (via `stringifyAsync`) +- streamed promises (via `unevalStream`) Try it out [here](https://svelte.dev/repl/138d70def7a748ce9eda736ef1c71239?version=3.49.0). @@ -87,6 +88,47 @@ devalue.parse(stringified); // { quick: 'data', slow: { ... } } Promises are awaited and their resolved values are serialized. The output format is identical to `stringify`, so `parse` and `unflatten` work unchanged. +### `unevalStream` + +`unevalStream` is a streaming version of `uneval`. It synchronously returns a `head` expression that recreates the value with every promise _pending_, and a `tail` of statement blocks that settle those promises as they settle on the server: + +```js +const { head, tail } = devalue.unevalStream({ + quick: 'data', + slow: fetch('/api/slow').then((r) => r.json()) +}); + +res.write(``); +for await (const block of tail) res.write(``); +``` + +Evaluate `head` once, then each block once, in order, in the same realm. Concatenating them into a single script also works. A settled value may contain further promises, which later blocks settle. + +- every promise is pending in `head`, regardless of timing +- object identity is preserved across `head` and every block +- the replacer is called at most once per object for the whole stream + +#### Options + +| Option | Description | +| --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `id` | Session key (returned as `id`). Random by default; pass a stable value when `head` must be byte-identical across runs. | +| `scope` | Trusted, assignable expression holding the client session table. Defaults to `globalThis.__d`. | +| `signal` | An `AbortSignal` that stops the tail. Pending and later `next()` calls reject once with `signal.reason`; if the signal is already aborted, or aborts during the head emission, `unevalStream` throws `signal.reason`. | +| `onerror` | `(error, value) => void`, called when an outcome can't be serialized. The client promise rejects with a generic `Error`. | + +#### Constraints + +Don't mutate the input until the tail has finished, and don't mutate the revived value until every block has been evaluated — blocks address existing objects by path. + +Only native promises are streamed. Other thenables need a replacer that rebuilds them around a native promise: + +```js +devalue.unevalStream(value, (v, js) => { + if (v instanceof MyThenable) return js`new MyThenable(${Promise.resolve(v)})`; +}); +``` + ### `unflatten` In the case where devalued data is one part of a larger JSON string, `unflatten` allows you to revive just the bit you need: diff --git a/src/index.js b/src/index.js index bdb3dba1..b34b49e4 100644 --- a/src/index.js +++ b/src/index.js @@ -1,4 +1,5 @@ export { uneval } from './uneval.js'; +export { unevalStream } from './uneval-stream.js'; export { parse, unflatten } from './parse.js'; export { stringify, stringifyAsync } from './stringify.js'; export { diff --git a/src/types.ts b/src/types.ts index 7740adaa..1b6ce5a1 100644 --- a/src/types.ts +++ b/src/types.ts @@ -64,6 +64,32 @@ export type UnevalReplacer = ( js: JavaScriptTag ) => JavaScriptFragment | false | null | void; +/** Options for `unevalStream`. */ +export interface UnevalStreamOptions { + /** Session key. Pass a stable value when heads must be byte-identical across runs. */ + id?: string; + /** Trusted assignable expression holding the client session table. Default `globalThis.__d`. */ + scope?: string; + /** + * Stops the tail; pending and later `next()` calls reject once with `signal.reason`. + * If the signal is already aborted, or aborts during the head emission, `unevalStream` + * throws `signal.reason`. + */ + signal?: AbortSignal; + /** Receives outcomes that could not be serialized; the client promise rejects with a generic Error. */ + onerror?: (error: unknown, value: unknown) => void; +} + +/** The result of `unevalStream`. */ +export interface UnevalStreamResult { + /** A JavaScript expression that recreates the value, with every promise pending. */ + head: string; + /** Statement blocks that settle the promises created by `head` and earlier blocks. */ + tail: AsyncIterableIterator; + /** The session key. */ + id: string; +} + /** * The introspection/extraction operations `stringify` performs on the value * being serialized. Every dynamic operation — property reads, prototype diff --git a/src/uneval-stream.js b/src/uneval-stream.js new file mode 100644 index 00000000..5ab74a37 --- /dev/null +++ b/src/uneval-stream.js @@ -0,0 +1,528 @@ +/** + * @import { UnevalReplacer, UnevalStreamOptions, UnevalStreamResult } from './types'; + */ +import { js, JavaScriptSource } from './javascript-source.js'; +import { analyze } from './uneval.js'; +import { call_replacer, get_type, stringify_string } from './utils.js'; + +// `o` is reserved even when unused, since names are allocated before the binding is chosen +const RESERVED = ['s', 'n', 'o']; +const LOCAL_BINDING_COST = '(()=>{let o=s.o;return })()'.length; +const FAILED = 'new Error("devalue: failed to serialize asynchronous value")'; +const DONE = Object.freeze({ done: /** @type {true} */ (true), value: undefined }); +const identifier = /^[_$a-zA-Z][_$a-zA-Z0-9]*$/; +const then = Promise.prototype.then; +const noop = () => {}; +// Bounds UTF-8 serialized property/bracket text, independently of numeric slot width. +const MAX_PATH_COST = 128; + +/** + * Turn a value containing promises into a JavaScript expression (`head`) that + * recreates it with pending promises, and a stream of statement blocks (`tail`) + * that settle them as they settle on the server. + * @param {any} value + * @param {UnevalReplacer} [replacer] + * @param {UnevalStreamOptions} [options] + * @returns {UnevalStreamResult} + */ +export function unevalStream(value, replacer, options = {}) { + if (options.signal?.aborted) throw options.signal.reason; + const scope = options.scope ?? 'globalThis.__d'; + const id = options.id ?? random_id(); + const key = stringify_string(id); + const session = new Session(replacer, options, `${scope}[${key}]`); + const head = start(session, value, scope, key); + return { head, tail: session, id }; +} + +/** + * Emits the head for a new session. Assigned inside `Session` so it can reach private state. + * @type {(session: Session, value: any, scope: string, key: string) => string} + */ +let start; + +/** Persistent request state is owned here, never by the input-bearing setup scope. */ +class Session { + static { + start = (session, value, scope, key) => session.#start(value, scope, key); + } + + /** + * @param {UnevalReplacer | undefined} replacer + * @param {UnevalStreamOptions} options + * @param {string} table + */ + constructor(replacer, options, table) { + this.#replacer = replacer; + this.#onerror = options.onerror; + this.#signal = options.signal; + this.#table = table; + this.#on_abort = this.#abort_stream.bind(this); + } + + /** @type {UnevalReplacer | undefined} */ + #replacer; + /** @type {UnevalStreamOptions['onerror']} */ + #onerror; + /** @type {AbortSignal | undefined} */ + #signal; + /** @type {string} */ + #table; + /** @type {() => void} */ + #on_abort; + + /** + * @param {any} value + * @param {string} scope + * @param {string} key + * @returns {string} + */ + #start(value, scope, key) { + this.#signal?.addEventListener('abort', this.#on_abort); + + let head; + try { + head = this.#emit_value(value, true); + } catch (error) { + this.#finish(); + throw error; + } + + // Aborted mid-emission: the head is dead on arrival, so fail like a pre-aborted signal + if (this.#abort) throw this.#abort.reason; + + if (this.#pending === 0) { + this.#finish(); + return head; + } + + /** + * Wrap the head in an IIFE that sets up this stream's client-side state: + * - `n` is the shared `scope` table, created with a null prototype if missing + * - `s` is this stream's entry in `n`, keyed by `id`: + * - `s.o` holds slots for objects and primitives that later tail blocks reference + * - `s.p` holds the `[resolve, reject]` pair for each pending promise + * - `s.d(i)` creates deferred pending promise `i`, saving its resolvers in `s.p`. The + * no-op `catch` prevents unhandled rejections before user code attaches handlers + * - `s.r(i, k, v)` settles promise `i` with `v`, resolving when `k` is 0 and + * rejecting when `k` is 1, then drops its resolvers + * The head expression is then evaluated with `s` in scope + */ + return ( + `(()=>{let n=${scope}||(${scope}={__proto__:null}),s=(n[${key}]={o:[],p:[]});` + + `s.d=(i)=>{let p=new Promise((a,b)=>{s.p[i]=[a,b]});p.catch(()=>{});return p};` + + `s.r=(i,k,v)=>{s.p[i][k](v);delete s.p[i]};` + + `return ${head}})()` + ); + } + + /** + * Replacer results retained until termination, so each object is seen once + * @type {Map} + */ + #replacer_cache = new Map(); + + /** + * Where each object that already exists on the client lives + * @type {Map} + */ + #known = new Map(); + + /** + * Sometimes, paths are long enough that it makes sense to give them their own slot. + * When this occurs, we map the location to that slot. + * @type {Map} + */ + #aliases = new Map(); + + /** + * This is like `#aliases`, but for primitives. We can vastly reduce the wire size of payloads + * containing repeated strings or bigints (whose textual length is essentially unbounded) by saving them + * in a slot, which lets us reuse them with very short syntax. + * @type {Map} + */ + #primitives = new Map(); + + /** + * Promise settlements not yet emitted in a tail block + * @type {Array<{ index: number, ok: boolean, value: unknown }>} + */ + #queue = []; + + #next_slot = 0; + #next_promise = 0; + /** The number of pending promises that have been sent to the client but not yet resolved */ + #pending = 0; + #done = false; + + /** + * Clearing `current` when the stream ends prevents our pending promises from retaining no-longer-relevant state. + * Without this indirection, the callbacks passed to `then`/`enqueue` could strongly-reference our cache data or + * other unrelated stuff, which then can't be garbage-collected until the promise resolves. By passing them a pointer, + * we enable ourselves to null out that pointer when the stream is aborted or otherwise ends, which allows GC to occur + * even if there are latent promises. + * @type {{ current: Session | null }} + */ + #delivery = { current: this }; + + /** + * In order to fulfill the AsyncIterableIterator contract (and avoid accidentally sending results out of order), + * we use a chain of promises to force subsequent calls to `#next` to resolve in the original call order. + */ + #chain = Promise.resolve(); + + /** + * Resumes a `next()` waiting for a settlement or for the session to finish + * @type {(value?: unknown) => void} + */ + #wake = noop; + + /** @type {{ reason: unknown } | null} */ + #abort = null; + + /** + * @param {any} value + * @param {boolean} head + */ + #emit_value(value, head) { + const known = this.#known; + const replacer_cache = this.#replacer_cache; + const aliases = this.#aliases; + const primitives = this.#primitives; + const replacer = this.#replacer; + const slot_start = this.#next_slot; + const promise_start = this.#next_promise; + + /** @type {Map} */ + const locations = new Map(); + /** @type {Map} */ + const retained = new Map(); + /** @type {Map} */ + const fresh = new Map(); + /** @type {Map} */ + const staged = new Map(); + /** @type {Array<{ slot: number, location: Location }>} */ + const assignments = []; + + /** @type {Map} */ + const staged_primitives = new Map(); + /** @type {Array<{ slot: number, literal: string }>} */ + const primitive_assignments = []; + + try { + const analysis = analyze(value, { + known: (thing) => known.get(thing), + enter: (thing, parent, key) => { + let up = null; + let segment = ''; + if (parent !== undefined) { + up = /** @type {Location} */ (locations.get(parent)); + if (typeof key === 'number') segment = `[${key}]`; + else if (key !== undefined && key.length <= MAX_PATH_COST) segment = prop(key); + } + const cost = path_cost(segment); + if (up && segment && up.cost + cost <= MAX_PATH_COST) { + locations.set(thing, { up, key: segment, cost: up.cost + cost }); + } else { + const slot = this.#next_slot++; + locations.set(thing, { up: null, key: slot, cost: 0 }); + // The root is assigned around the whole expression instead. + if (thing !== value) retained.set(thing, slot); + } + }, + replace: (thing) => { + if (replacer) { + let source = replacer_cache.get(thing); + if (source === undefined) { + source = call_replacer(thing, replacer); + if (!this.#done) replacer_cache.set(thing, source); + } + if (source) return source; + } + if (get_type(thing) !== 'Promise' || !is_promise(thing)) return null; + const index = this.#next_promise++; + fresh.set(thing, index); + return JavaScriptSource.from(js`s.d(${index})`); + }, + reserved: RESERVED + }); + + // a head without promises needs no session at all + if (head && fresh.size === 0) return analysis.render(); + + this.#plan_aliases(analysis.references.values(), staged, assignments); + + // Without a replacer there are no free custom identifiers to shadow, so a local + // `o` can stand in for `s.o` when the uses it shortens pay for the wrapper. The + // root, each retained object and reference, and each alias (twice) use a slot. + const slot_uses = 1 + retained.size + analysis.references.size + 2 * assignments.length; + const binding = !replacer && slot_uses * 2 > LOCAL_BINDING_COST ? 'o' : 's.o'; + + let code = analysis.render({ + reference: (location) => expression(location, aliases, binding, staged), + retain: (thing) => { + const slot = retained.get(thing); + return slot === undefined ? undefined : `${binding}[${slot}]`; + }, + primitive: (thing, literal) => { + let slot = primitives.get(thing) ?? staged_primitives.get(thing); + if (slot === undefined) { + slot = this.#next_slot++; + staged_primitives.set(thing, slot); + primitive_assignments.push({ slot, literal }); + } + return `${binding}[${slot}]`; + } + }); + + const root = locations.get(value); + if (root !== undefined) code = `(${binding}[${root.key}]=${code})`; + if (assignments.length) { + /** @type {Map} */ + const initialized = new Map(); + const initializers = assignments.map(({ slot, location }) => { + const prefix = expression(location, aliases, binding, initialized); + initialized.set(location, slot); + return `${binding}[${slot}]=${prefix}`; + }); + code = `(${initializers.join(',')},${code})`; + } + if (binding === 'o') code = `(()=>{let o=s.o;return ${code}})()`; + + if (primitive_assignments.length) { + const initializers = primitive_assignments.map( + ({ slot, literal }) => `s.o[${slot}]=${literal}` + ); + code = `(${initializers.join(',')},${code})`; + } + // Trusted callbacks can cancel synchronously during emission. Never publish + // request state or attach new delivery after terminal cleanup. + if (!this.#done) { + for (const [thing, slot] of staged_primitives) primitives.set(thing, slot); + for (const [location, slot] of staged) aliases.set(location, slot); + for (const [thing, location] of locations) known.set(thing, location); + for (const [promise, index] of fresh) { + this.#pending += 1; + Session.#subscribe(promise, index, this.#delivery); + } + } + + return code; + } catch (error) { + this.#next_slot = slot_start; + this.#next_promise = promise_start; + throw error; + } + } + + /** + * Gives a slot to each shared path prefix that is used often enough to pay for + * its initialization. Each existing object is rendered once, because the emitter + * hoists repeated uses. Shared ancestors count as well as leaves, in + * ancestor-first order. Every walk is bounded by MAX_PATH_COST, not by the depth + * of the input graph. + * @param {Iterable} references + * @param {Map} staged + * @param {Array<{ slot: number, location: Location }>} assignments + */ + #plan_aliases(references, staged, assignments) { + const aliases = this.#aliases; + /** @type {Map} */ + const demand = new Map(); + for (const location of references) { + const path = []; + for (let node = location; node.up && !aliases.has(node); node = node.up) path.push(node); + for (const node of path.reverse()) demand.set(node, (demand.get(node) ?? 0) + 1); + } + for (const [location, count] of demand) { + const prefix = expression(location, aliases, 's.o', staged); + const slot = `s.o[${this.#next_slot}]`; + const prefix_cost = path_cost(prefix); + // Includes initialization, assignment, comma, and the first alias wrapper. + const assignment_cost = prefix_cost + slot.length + 2 + (assignments.length ? 0 : 2); + if (count * (prefix_cost - slot.length) > assignment_cost) { + assignments.push({ slot: this.#next_slot, location }); + staged.set(location, this.#next_slot++); + } + } + } + + /** + * @param {number} index + * @param {boolean} ok + * @param {unknown} value + */ + #enqueue(index, ok, value) { + if (this.#done) return; + this.#queue.push({ index, ok, value }); + this.#wake(); + } + + /** + * @param {number} index + * @param {boolean} ok + * @param {unknown} value + */ + #settle(index, ok, value) { + try { + return `s.r(${index},${ok ? 0 : 1},${this.#emit_value(value, false)})`; + } catch (error) { + try { + this.#onerror?.(error, value); + } catch { + // a failing error handler must not break the session + } + return `s.r(${index},1,${FAILED})`; + } + } + + #finish() { + if (this.#done) return; + this.#done = true; + this.#delivery.current = null; + this.#signal?.removeEventListener('abort', this.#on_abort); + this.#known.clear(); + this.#replacer_cache.clear(); + this.#aliases.clear(); + this.#primitives.clear(); + this.#queue.length = 0; + this.#replacer = undefined; + this.#onerror = undefined; + this.#signal = undefined; + this.#on_abort = noop; + const wake = this.#wake; + this.#wake = noop; + wake(); + } + + #abort_stream() { + if (this.#done) return; + this.#abort = { reason: this.#signal?.reason }; + this.#finish(); + } + + /** @returns {Promise>} */ + async #step() { + while (!this.#done && this.#queue.length === 0) { + await new Promise((resolve) => (this.#wake = resolve)); + this.#wake = noop; + } + + if (this.#abort) { + const { reason } = this.#abort; + this.#abort = null; + throw reason; + } + + if (this.#done) return DONE; + + const statements = []; + for (const { index, ok, value } of this.#queue.splice(0)) { + this.#pending -= 1; + statements.push(this.#settle(index, ok, value)); + } + + const complete = this.#pending === 0; + if (complete) { + statements.push(`delete ${this.#table}`); + } + + const value = `((s)=>{${statements.join(';')}})(${this.#table});`; + if (complete) this.#finish(); + return { done: false, value }; + } + + /** @returns {Promise>} */ + next() { + const result = this.#chain.then(this.#step.bind(this)); + this.#chain = result.then(noop, noop); + return result; + } + + /** @returns {Promise>} */ + return() { + this.#finish(); + return Promise.resolve(DONE); + } + + [Symbol.asyncIterator]() { + return this; + } + + /** + * Native reactions own only this detachable cell and the numeric index. + * @param {object} promise + * @param {number} index + * @param {{ current: Session | null }} delivery + */ + static #subscribe(promise, index, delivery) { + /** + * @param {boolean} ok + * @param {unknown} value + */ + const deliver = (ok, value) => { + const session = delivery.current; + if (session) session.#enqueue(index, ok, value); + }; + then.call( + promise, + (value) => deliver(true, value), + (error) => deliver(false, error) + ); + } +} + +/** @param {object} thing */ +function is_promise(thing) { + try { + then.call(thing, noop, noop); + return true; + } catch { + return false; + } +} + +/** + * A slot (`up` is null and `key` is its expression) or a property of another location + * Property keys are already serialized, bounded segments; root keys are slot indices. + * @typedef {{ up: Location | null, key: string | number, cost: number }} Location + */ + +/** + * @param {Location} location + * @param {Map} aliases + * @param {string} [binding] + * @param {Map} [staged] + * @returns {string} + */ +function expression(location, aliases, binding = 's.o', staged) { + const segments = []; + while (location.up !== null && !aliases.has(location) && !staged?.has(location)) { + segments.push(location.key); + location = location.up; + } + return `${binding}[${staged?.get(location) ?? aliases.get(location) ?? location.key}]${segments.reverse().join('')}`; +} + +/** + * Count UTF-8 bytes of an already bounded segment or path. Never called on an + * unbounded property key; slot width is the only uncapped part of a path. + * @param {string} source + */ +function path_cost(source) { + let cost = 0; + for (const character of source) { + const code = /** @type {number} */ (character.codePointAt(0)); + cost += code < 0x80 ? 1 : code < 0x800 ? 2 : code < 0x10000 ? 3 : 4; + } + return cost; +} + +/** @param {string} key */ +function prop(key) { + return identifier.test(key) ? `.${key}` : `[${stringify_string(key)}]`; +} + +function random_id() { + return Math.random().toString(36).slice(2, 10); +} diff --git a/test/README.md b/test/README.md index 49b7f2da..f5baec20 100644 --- a/test/README.md +++ b/test/README.md @@ -11,4 +11,6 @@ Unit tests are discovered from `src/**/*.test.js` and `test/**/*.test.js`. Brows The base64 and string-escaping tests include independent decoder and UTF-8 transport checks. Keep these independent from encoder round trips so correlated encoding and decoding regressions cannot hide each other. +`unevalStream` tests go through `expect_roundtrip(make, { replacer, globals, orders })` in `test/helpers/stream.js`. `make` receives a schedule whose `later(value)` and `later_reject(reason)` create promises that the harness settles in several orders (FIFO, LIFO, all at once, seeded shuffle). For each order, the harness evaluates the head and every tail block in fresh VM realms, both separately and concatenated into one script. It awaits every revived promise and checks structure and identity with `compare_topology`. It also asserts that the head is the same for every order, that the replacer was called once per object identity, and that the client session was deleted. New cases should be rows in the existing tables. Don't assert exact `head` strings except in escaping regressions. + Use `test.each` for independent input cases and `describe.each` for suites or nested case matrices. Wrap array-valued inputs in object rows so Vitest does not spread them into callback arguments, and create mutable values inside each test. Loops within a test are appropriate for building fixtures, checking one result's contents, or comparing growth across input sizes. diff --git a/test/helpers/stream.js b/test/helpers/stream.js new file mode 100644 index 00000000..9b6dcab9 --- /dev/null +++ b/test/helpers/stream.js @@ -0,0 +1,277 @@ +import vm from 'node:vm'; +import { expect } from 'vitest'; +import { unevalStream } from '../../src/index.js'; + +const DEFAULT_ORDERS = ['fifo', 'lifo', 'all-at-once', 'seeded-shuffle']; + +/** + * A fresh VM realm, with optional injected globals (e.g. classes used by a replacer) + * @param {Record} [globals] + */ +export function client(globals = {}) { + const context = vm.createContext({ ...globals }); + context.globalThis = context; + return { + context, + /** @param {string} source */ + head: (source) => vm.runInContext(`(${source})`, context), + /** @param {string} source */ + block: (source) => vm.runInContext(source, context), + /** + * Evaluates the head and every block as one script + * @param {string} head + * @param {string[]} blocks + */ + combined: (head, blocks) => + vm.runInContext(`(function(){const root=(${head});${blocks.join('')}return root})()`, context) + }; +} + +/** @param {number} seed */ +export function seeded_random(seed) { + let state = seed >>> 0; + return () => { + state ^= state << 13; + state ^= state >>> 17; + state ^= state << 5; + return (state >>> 0) / 0x1_0000_0000; + }; +} + +/** + * @template T + * @param {T[]} values + * @param {() => number} random + */ +export function shuffle(values, random) { + for (let i = values.length - 1; i > 0; i -= 1) { + const j = Math.floor(random() * (i + 1)); + [values[i], values[j]] = [values[j], values[i]]; + } + return values; +} + +const turn = () => new Promise((resolve) => setImmediate(resolve)); + +/** + * `later(value, ms?)` / `later_reject(reason, ms?)` create promises that settle when + * `settle_in(order)` runs, or after `ms` milliseconds once it starts if `ms` is given + */ +export function create_schedule() { + /** @type {Array<() => void>} */ + const ordered = []; + /** @type {Array<{ settle: () => void, ms: number }>} */ + const timed = []; + + /** + * @param {boolean} ok + * @param {unknown} value + * @param {number} [ms] + */ + function create(ok, value, ms) { + const { promise, resolve, reject } = Promise.withResolvers(); + // rejections nested in outcomes are only observed once the outcome is emitted + promise.catch(() => {}); + const settle = () => (ok ? resolve(value) : reject(value)); + if (ms === undefined) ordered.push(settle); + else timed.push({ settle, ms }); + return promise; + } + + return { + /** @param {unknown} value @param {number} [ms] */ + later: (value, ms) => create(true, value, ms), + /** @param {unknown} reason @param {number} [ms] */ + later_reject: (reason, ms) => create(false, reason, ms), + /** @param {string} order */ + async settle_in(order) { + const timers = timed.map(({ settle, ms }) => + new Promise((r) => setTimeout(r, ms)).then(settle) + ); + const settles = + order === 'lifo' + ? ordered.slice().reverse() + : order === 'seeded-shuffle' + ? shuffle(ordered.slice(), seeded_random(ordered.length + 1)) + : ordered; + if (order === 'all-at-once') { + for (const settle of settles) settle(); + } else { + for (const settle of settles) { + settle(); + await turn(); + } + } + await Promise.all(timers); + } + }; +} + +/** @param {unknown} value */ +const type_of = (value) => Object.prototype.toString.call(value).slice(8, -1); + +/** @param {unknown} value */ +const is_object = (value) => + value !== null && (typeof value === 'object' || typeof value === 'function'); + +/** @param {PromiseLike} promise */ +const outcome = (promise) => + Promise.resolve(promise).then( + (value) => ({ ok: true, value }), + (value) => ({ ok: false, value }) + ); + +/** @param {ArrayBufferLike} buffer @param {number} [offset] @param {number} [length] */ +const bytes = (buffer, offset, length) => Array.from(new Uint8Array(buffer, offset, length)); + +/** + * Compares a server value with its revived client counterpart: structure, prototypes, + * key order, and identity (repeated server objects must map to one client object, and + * distinct ones must not collapse). Promises match when their settled outcomes match. + * @param {any} server + * @param {any} revived + * @param {string} [label] + * @param {Map} [server_to_client] + * @param {WeakMap} [client_to_server] + * @returns {Promise} + */ +export async function compare_topology( + server, + revived, + label = 'root', + server_to_client = new Map(), + client_to_server = new WeakMap() +) { + if (!is_object(server)) { + expect(Object.is(revived, server), `${label}: primitive values differ`).toBe(true); + return; + } + if (server_to_client.has(server)) { + expect(revived, `${label}: repeated identity differs`).toBe(server_to_client.get(server)); + return; + } + expect(is_object(revived), `${label}: expected an object`).toBe(true); + expect(client_to_server.has(revived), `${label}: distinct server nodes collapsed`).toBe(false); + server_to_client.set(server, revived); + client_to_server.set(revived, server); + + /** @param {unknown} s @param {unknown} c @param {string} suffix */ + const recurse = (s, c, suffix) => + compare_topology(s, c, label + suffix, server_to_client, client_to_server); + + const type = type_of(server); + expect(type_of(revived), `${label}: type differs`).toBe(type); + + if (type === 'Promise') { + const [s, c] = await Promise.all([outcome(server), outcome(revived)]); + expect(c.ok, `${label}: settlement kind differs`).toBe(s.ok); + return recurse(s.value, c.value, s.ok ? '' : ''); + } + if (['String', 'Number', 'Boolean', 'BigInt', 'Date'].includes(type)) { + expect(Object.is(revived.valueOf(), server.valueOf()), `${label}: boxed value differs`).toBe( + true + ); + return; + } + if (['RegExp', 'URL', 'URLSearchParams'].includes(type)) { + expect(String(revived), `${label}: value differs`).toBe(String(server)); + return; + } + if (type === 'ArrayBuffer') { + expect(bytes(revived), `${label}: buffer bytes differ`).toEqual(bytes(server)); + return; + } + if (ArrayBuffer.isView(server)) { + expect(bytes(revived.buffer, revived.byteOffset, revived.byteLength)).toEqual( + bytes(server.buffer, server.byteOffset, server.byteLength) + ); + // a Node Buffer's pooled backing store is never sent + if (Buffer.isBuffer(server)) return; + expect(revived.byteOffset, `${label}: view offset differs`).toBe(server.byteOffset); + return recurse(server.buffer, revived.buffer, '.buffer'); + } + if (type === 'Map' || type === 'Set') { + const s = Array.from(server.entries()); + const c = Array.from(revived.entries()); + expect(c.length, `${label}: size differs`).toBe(s.length); + for (let i = 0; i < s.length; i += 1) { + await recurse(s[i][0], c[i][0], `.key[${i}]`); + if (type === 'Map') await recurse(s[i][1], c[i][1], `.value[${i}]`); + } + return; + } + + const proto = Object.getPrototypeOf(server); + const revived_proto = Object.getPrototypeOf(revived); + if (proto === null || proto === Object.prototype || proto === Array.prototype) { + expect(revived_proto?.constructor?.name ?? null, `${label}: prototype differs`).toBe( + proto?.constructor.name ?? null + ); + } else { + expect(revived_proto, `${label}: class differs`).toBe(proto); + } + if (Array.isArray(server)) expect(revived.length, `${label}: length differs`).toBe(server.length); + + const keys = Object.keys(server); + expect(Object.keys(revived), `${label}: keys differ`).toEqual(keys); + for (const key of keys) await recurse(server[key], revived[key], `[${JSON.stringify(key)}]`); +} + +/** + * Serializes `make(schedule)` once per settlement order, evaluates the result in fresh + * realms (block by block, and concatenated), and checks everything the contract promises. + * @param {(schedule: ReturnType) => unknown} make + * @param {{ + * replacer?: import('../../src/types.js').UnevalReplacer, + * globals?: Record, + * orders?: string[] + * }} [options] + */ +export async function expect_roundtrip(make, { replacer, globals, orders = DEFAULT_ORDERS } = {}) { + /** @type {Set} */ + const heads = new Set(); + /** @type {{ head: string, blocks: string[], root: unknown }} */ + let result = { head: '', blocks: [], root: undefined }; + + for (const order of orders) { + const schedule = create_schedule(); + const value = make(schedule); + + /** @type {Map} */ + const calls = new Map(); + /** @type {import('../../src/types.js').UnevalReplacer} */ + const counted = (v, js) => { + calls.set(v, (calls.get(v) ?? 0) + 1); + return replacer?.(v, js); + }; + + const { head, tail } = unevalStream(value, counted, { id: 'roundtrip' }); + heads.add(head); + + /** @type {string[]} */ + const blocks = []; + const consumed = (async () => { + for await (const block of tail) blocks.push(block); + })(); + await schedule.settle_in(order); + await consumed; + + const separate = client(globals); + const root = separate.head(head); + for (const block of blocks) separate.block(block); + await compare_topology(value, root, `${order} (separate)`); + + const concatenated = client(globals); + await compare_topology(value, concatenated.combined(head, blocks), `${order} (concatenated)`); + + for (const [v, n] of calls) expect(n, `${order}: replacer calls for ${type_of(v)}`).toBe(1); + for (const { context } of [separate, concatenated]) { + expect(Object.keys(context.__d ?? {}), `${order}: session not deleted`).toEqual([]); + } + + result = { head, blocks, root }; + } + + expect(heads.size, 'head depends on settlement order').toBe(1); + return result; +} diff --git a/test/helpers/subprocess.js b/test/helpers/subprocess.js new file mode 100644 index 00000000..8c2cde6f --- /dev/null +++ b/test/helpers/subprocess.js @@ -0,0 +1,27 @@ +import { execFileSync } from 'node:child_process'; + +/** + * Keep strict unhandled-rejection failures (and explicit GC) out of the Vitest process. + * Fixtures are stringified, so they must not capture anything from their module. + * @param {(helpers: any) => unknown} fixture + * @param {{ expose_gc?: boolean, data?: unknown }} [options] + */ +export function in_subprocess(fixture, { expose_gc = false, data } = {}) { + execFileSync( + process.execPath, + [ + '--unhandled-rejections=strict', + ...(expose_gc ? ['--expose-gc', '--max-old-space-size=128'] : []), + '--input-type=module', + '--eval', + ` + import assert from 'node:assert/strict'; + import vm from 'node:vm'; + import { setImmediate as turn, setTimeout as delay } from 'node:timers/promises'; + import { parse, stringify, stringifyAsync, unevalStream } from ${JSON.stringify(new URL('../../src/index.js', import.meta.url).href)}; + await (${fixture})({ assert, vm, turn, delay, parse, stringify, stringifyAsync, unevalStream, data: ${JSON.stringify(data)} }); + ` + ], + { timeout: 5000, stdio: 'pipe' } + ); +} diff --git a/test/stream-primitive-growth.test.js b/test/stream-primitive-growth.test.js new file mode 100644 index 00000000..287528f9 --- /dev/null +++ b/test/stream-primitive-growth.test.js @@ -0,0 +1,202 @@ +import { setImmediate as turn } from 'node:timers/promises'; +import { expect, test } from 'vitest'; +import { parse, stringify, uneval, unevalStream } from '../src/index.js'; +import { client } from './helpers/stream.js'; + +const primitives = [ + { kind: 'string', make: (n) => 'x'.repeat(n * 8) }, + { kind: 'escaped string', make: (n) => '\n"\\\0\u2028\u2029'.repeat(n) }, + { kind: 'bigint', make: (n) => BigInt('9'.repeat(n * 4)) } +]; + +test.each( + ['batch', 'separate'].flatMap((mode) => primitives.map((primitive) => ({ mode, ...primitive }))) +)('$mode: repeated $kind settlements grow linearly', async ({ mode, make }) => { + const sizes = []; + for (const n of [128, 256]) { + const primitive = make(n); + const encoded = stringify(Array(n).fill(primitive)); + const values = parse(encoded); + const controlled = mode === 'separate' ? values.map(() => Promise.withResolvers()) : []; + // These are distinct native promises, not repeated references to one promise. + const promises = values.map((value, i) => + mode === 'separate' ? controlled[i].promise : Promise.resolve(value) + ); + expect(new Set(promises).size).toBe(n); + const stream = unevalStream(promises); + const c = client(); + const root = c.head(stream.head); + let source = stream.head; + const consuming = (async () => { + for await (const block of stream.tail) { + source += block; + c.block(block); + } + })(); + if (mode === 'separate') { + for (let i = 0; i < n; i += 1) { + controlled[i].resolve(values[i]); + await turn(); + } + } + await consuming; + + expect(root).toHaveLength(n); + expect(new Set(root).size).toBe(n); + for (const revived of await Promise.all(root)) expect(revived).toBe(primitive); + sizes.push({ n, input: encoded.length, output: source.length }); + } + + // All VM values are verified before checking budgets; do not prescribe batching. + expect + .soft(sizes[1].output / sizes[0].output, 'doubling the input must not quadruple output') + .toBeLessThan(2.5); + for (const { n, input, output } of sizes) { + expect + .soft(output, `n=${n}: output exceeds the linear input budget`) + .toBeLessThan(input * 8 + n * 192 + 4096); + } +}); + +test('control: primitive interning preserves custom cycles across discarded renderings', async () => { + class Cyclic { + constructor(text, meta) { + this.text = text; + this.meta = meta; + } + } + const text = '\n"\\\0\u2028\u2029'.repeat(128); + const value = new Cyclic(text, {}); + value.meta.parent = value; + const first = Promise.withResolvers(); + const second = Promise.withResolvers(); + const stream = unevalStream({ first: first.promise, second: second.promise }, (v, js) => + v instanceof Cyclic ? js`new Cyclic(${v.text},${v.meta})` : undefined + ); + const c = client({ Cyclic }); + const root = c.head(stream.head); + + // Rendering text before meta's cycle can discard an outer constructor rendering. + // Its primitive-slot assignment must not be mistaken for an emitted assignment. + first.resolve(value); + c.block((await stream.tail.next()).value); + const revived = await root.first; + expect(revived).toBeInstanceOf(Cyclic); + expect(revived.text).toBe(text); + expect(revived.meta.parent).toBe(revived); + expect(revived.meta.parent.meta).toBe(revived.meta); + + second.resolve(text); + for await (const block of stream.tail) c.block(block); + expect(await root.second).toBe(text); +}); + +test('control: a failed emission does not poison a later settlement of the same primitive', async () => { + const text = '\n"\\\0\u2028\u2029'.repeat(128); + const first = Promise.withResolvers(); + const second = Promise.withResolvers(); + const stream = unevalStream({ first: first.promise, second: second.promise }); + const c = client(); + const root = c.head(stream.head); + + // The eligible primitive is encountered before the unsupported late field. + first.resolve({ text, unsupported: Symbol('unsupported late property') }); + c.block((await stream.tail.next()).value); + await expect(root.first).rejects.toThrow('devalue: failed to serialize asynchronous value'); + second.resolve(text); + for await (const block of stream.tail) c.block(block); + expect(await root.second).toBe(text); +}); + +test('a rendering failure discards staged primitives without losing earlier committed slots', async () => { + const committed = 'committed'.repeat(128); + const staged = 'staged'.repeat(128); + const first = Promise.withResolvers(); + const second = Promise.withResolvers(); + const stream = unevalStream({ committed, first: first.promise, second: second.promise }); + const c = client(); + const root = c.head(stream.head); + let reads = 0; + first.resolve({ + staged, + get late() { + if (++reads > 1) throw new Error('failure after primitive rendering'); + return null; + } + }); + c.block((await stream.tail.next()).value); + expect(reads).toBe(2); + await expect(root.first).rejects.toThrow('devalue: failed to serialize asynchronous value'); + second.resolve({ committed, staged }); + for await (const block of stream.tail) c.block(block); + expect(await root.second).toEqual({ committed, staged }); +}); + +test.each(primitives)('head $kind slots are reused by tails', async ({ make }) => { + const primitive = make(128); + const deferred = Promise.withResolvers(); + // Repetition exercises local primitive declarations as well as direct renders. + const stream = unevalStream({ values: [primitive, primitive], later: deferred.promise }); + const c = client(); + const root = c.head(stream.head); + expect(root.values).toEqual([primitive, primitive]); + deferred.resolve(primitive); + let source = ''; + for await (const block of stream.tail) { + source += block; + c.block(block); + } + expect(await root.later).toBe(primitive); + expect(source.length).toBeLessThan(primitive.toString().length / 2); +}); + +test('short strings and inexpensive bigints do not acquire primitive slots', async () => { + const short = '\n'.repeat(127); // Eligibility uses string length, not escaped length. + const bigint = BigInt('9'.repeat(126)); // 127 characters including the n suffix. + const deferred = Promise.withResolvers(); + const stream = unevalStream({ short, bigint, later: deferred.promise }); + const c = client(); + const root = c.head(stream.head); + const session = c.context.__d[stream.id]; + expect( + session.o.filter((value) => typeof value === 'string' || typeof value === 'bigint') + ).toEqual([]); + deferred.resolve({ short, bigint }); + for await (const block of stream.tail) c.block(block); + expect(await root.later).toEqual({ short, bigint }); + expect( + session.o.filter((value) => typeof value === 'string' || typeof value === 'bigint') + ).toEqual([]); +}); + +test.each(primitives)('a no-promise $kind head preserves standalone output', async ({ make }) => { + const primitive = make(128); + const value = { primitive, repeated: [primitive, primitive] }; + const stream = unevalStream(value); + expect(stream.head).toBe(uneval(value)); + const c = client(); + expect(c.head(stream.head)).toEqual(value); + expect(c.context.__d).toBeUndefined(); + expect(await stream.tail.next()).toEqual({ done: true, value: undefined }); +}); + +test('primitive initialization composes with reference aliases and alternate slot bindings', async () => { + const key = 'path'.repeat(20); + const nodes = Array.from({ length: 32 }, (_, i) => ({ i })); + const text = 'primitive'.repeat(128); + const first = Promise.withResolvers(); + const second = Promise.withResolvers(); + const stream = unevalStream({ [key]: nodes, first: first.promise, second: second.promise }); + const c = client(); + const root = c.head(stream.head); + first.resolve({ refs: nodes.slice(), values: [text, text] }); + c.block((await stream.tail.next()).value); + const revived = await root.first; + expect(revived.values).toEqual([text, text]); + for (let i = 0; i < nodes.length; i += 1) expect(revived.refs[i]).toBe(root[key][i]); + second.resolve({ refs: nodes.slice(), text }); + for await (const block of stream.tail) c.block(block); + const reused = await root.second; + expect(reused.text).toBe(text); + for (let i = 0; i < nodes.length; i += 1) expect(reused.refs[i]).toBe(revived.refs[i]); +}); diff --git a/test/stream-reference-growth.test.js b/test/stream-reference-growth.test.js new file mode 100644 index 00000000..848140c1 --- /dev/null +++ b/test/stream-reference-growth.test.js @@ -0,0 +1,382 @@ +import { expect, test } from 'vitest'; +import { stringify, uneval, unevalStream } from '../src/index.js'; +import { client, compare_topology, expect_roundtrip } from './helpers/stream.js'; + +/** + * @param {'long key' | 'deep chain' | 'short flat key'} kind + * @param {'head' | 'earlier tail'} origin + */ +async function reference_sizes(kind, origin) { + const sizes = []; + for (const n of [128, 256]) { + const nodes = Array.from({ length: n }, (_, i) => ({ i })); + let data; + let key; + if (kind === 'deep chain') { + for (let i = 0; i < nodes.length - 1; i += 1) nodes[i].next = nodes[i + 1]; + data = nodes[0]; + } else { + key = kind === 'long key' ? ''.repeat(n) : 'nodes'; + data = { [key]: nodes }; + } + // Include the fresh outcome array in the linear input budget, without promises. + const outcome = nodes.slice(); + const encoded = stringify({ data, outcome }); + const first = origin === 'earlier tail' ? Promise.withResolvers() : null; + const second = origin === 'earlier tail' ? Promise.withResolvers() : null; + const stream = unevalStream({ + data: first ? first.promise : data, + outcome: second ? second.promise : Promise.resolve(outcome) + }); + const c = client(); + const root = c.head(stream.head); + let source = stream.head; + let revived_data = root.data; + if (first && second) { + first.resolve(data); + const block = await stream.tail.next(); + expect(block.done).toBe(false); + source += block.value; + c.block(block.value); + revived_data = await root.data; + // Only settle the fresh shallow array after its ancestors exist on the client. + second.resolve(outcome); + } + for await (const block of stream.tail) { + source += block; + c.block(block); + } + + const revived = await root.outcome; + expect(revived).toHaveLength(n); + if (kind !== 'deep chain') expect(revived).not.toBe(revived_data[key]); + let node = kind === 'deep chain' ? revived_data : undefined; + for (let i = 0; i < n; i += 1) { + expect(revived[i]).toBe(kind === 'deep chain' ? node : revived_data[key][i]); + expect(revived[i].i).toBe(i); + if (kind === 'deep chain') node = node.next; + } + expect(new Set(revived).size).toBe(n); + sizes.push({ n, input: encoded.length, output: source.length }); + } + return sizes; +} + +test.each( + ['head', 'earlier tail'].flatMap((origin) => + ['long key', 'deep chain', 'short flat key'].map((kind) => ({ origin, kind })) + ) +)( + '$origin: $kind references grow linearly across a later promise outcome', + async ({ origin, kind }) => { + // Identity checks for both sizes run before any output-size assertions. + const sizes = await reference_sizes(kind, origin); + expect + .soft(sizes[1].output / sizes[0].output, 'doubling the input must not quadruple output') + .toBeLessThan(2.5); + for (const { n, input, output } of sizes) { + // Allow escaping and generous per-node session/slot overhead, not repeated paths. + expect + .soft(output, `n=${n}: output exceeds the linear input budget`) + .toBeLessThan(input * 8 + n * 192 + 4096); + } + } +); + +test('cheap single-use and repeated leaf paths do not create lazy slots', async () => { + const leaf = { value: 1 }; + const stream = unevalStream({ leaf, outcome: Promise.resolve([leaf, leaf]) }, undefined, { + id: 'cheap' + }); + const c = client(); + const root = c.head(stream.head); + const slots = c.context.__d.cheap.o; + expect(slots).toHaveLength(1); + const blocks = []; + for await (const block of stream.tail) { + blocks.push(block); + c.block(block); + } + const outcome = await root.outcome; + expect(outcome[0]).toBe(root.leaf); + expect(outcome[1]).toBe(root.leaf); + // Only the new outcome array is retained, not its already-known leaf. + expect(slots).toHaveLength(2); + expect(blocks.join('')).toContain('s.o[0].leaf'); + expect(blocks.join('')).not.toContain('let o='); +}); + +test.each([15, 16])('prefix length %i accounts for alias wrapper punctuation', async (length) => { + const key = 'x'.repeat(length); + const nodes = [{ i: 0 }, { i: 1 }]; + const stream = unevalStream( + { [key]: nodes, outcome: Promise.resolve(nodes.slice()) }, + undefined, + { + id: 'cost' + } + ); + const c = client(); + const root = c.head(stream.head); + const slots = c.context.__d.cost.o; + let tail = ''; + for await (const block of stream.tail) { + tail += block; + c.block(block); + } + const outcome = await root.outcome; + for (let i = 0; i < nodes.length; i += 1) expect(outcome[i]).toBe(root[key][i]); + // At 15 the alias breaks even; at 16 it saves exactly one byte, including + // the first comma-expression wrapper. Neither scope pays for a local binding. + expect(slots).toHaveLength(length === 15 ? 2 : 3); + expect(tail.split(key)).toHaveLength(length === 15 ? 3 : 2); + expect(tail).not.toContain('let o='); +}); + +test('one long but bounded leaf read avoids an unprofitable alias', async () => { + const key = 'x'.repeat(80); + const leaf = { value: 1 }; + const stream = unevalStream({ [key]: leaf, outcome: Promise.resolve(leaf) }, undefined, { + id: 'single' + }); + const c = client(); + const root = c.head(stream.head); + const slots = c.context.__d.single.o; + for await (const block of stream.tail) c.block(block); + expect(await root.outcome).toBe(root[key]); + expect(slots).toHaveLength(1); +}); + +test('profitable distinct descendants share an alias and a numeric local slot table', async () => { + const key = 'x'.repeat(80); + const nodes = Array.from({ length: 32 }, (_, i) => ({ i })); + const stream = unevalStream( + { [key]: nodes, outcome: Promise.resolve(nodes.slice()) }, + undefined, + { + id: 'shared' + } + ); + const c = client(); + const root = c.head(stream.head); + const slots = c.context.__d.shared.o; + expect(slots).toHaveLength(1); // below the ceiling: no speculative head slots + let source = stream.head; + let tail = ''; + for await (const block of stream.tail) { + source += block; + tail += block; + c.block(block); + } + const outcome = await root.outcome; + for (let i = 0; i < nodes.length; i += 1) expect(outcome[i]).toBe(root[key][i]); + expect(tail.split(key)).toHaveLength(2); // one bounded alias initialization + expect(tail).toContain('let o=s.o;'); + expect(tail).toMatch(/o\[\d+\]=o\[\d+\]/); + expect(Object.keys(slots)).toEqual(Array.from({ length: slots.length }, (_, i) => String(i))); + expect(source.length).toBeLessThan(2500); +}); + +test.each(['escaped segment', 'unicode segment', 'enormous segment', 'independent chains'])( + '%s forces ceiling anchors even for a single later read', + async (kind) => { + const leaf = { value: 42 }; + let data = leaf; + if (kind === 'independent chains') { + for (let i = 0; i < 80; i += 1) data = { next: data }; + data = { left: data, right: structuredClone(data) }; + } else { + const key = + kind === 'unicode segment' + ? 'é'.repeat(64) + : '<'.repeat(kind === 'escaped segment' ? 24 : 4096); + data = { [key]: leaf }; + } + const stream = unevalStream({ data, outcome: Promise.resolve(leaf) }, undefined, { + id: 'ceiling' + }); + const c = client(); + const root = c.head(stream.head); + const slots = c.context.__d.ceiling.o; + const before = slots.length; + expect(before).toBeGreaterThan(1); + let tail = ''; + for await (const block of stream.tail) { + tail += block; + c.block(block); + } + let revived = root.data; + if (kind === 'independent chains') { + expect(before).toBeGreaterThanOrEqual(7); + revived = revived.left; + for (let i = 0; i < 80; i += 1) revived = revived.next; + } else revived = revived[Object.keys(data)[0]]; + expect(await root.outcome).toBe(revived); + expect(slots).toHaveLength(before); // no profitable alias or new object + expect(tail.length).toBeLessThan(256); + expect(tail).not.toContain('\\u003C'); + if (kind === 'enormous segment') { + // Includes the unavoidable original key literal, not just the tail. + expect(stream.head.length + tail.length).toBeLessThan(stringify(data).length * 8 + 4096); + } + } +); + +test('no-promise heads remain byte-compatible even when paths exceed the ceiling', async () => { + const data = { ['<'.repeat(4096)]: { value: 1 } }; + const stream = unevalStream(data); + expect(stream.head).toBe(uneval(data)); + expect(stream.head).not.toContain('s.o'); + expect(await stream.tail.next()).toEqual({ done: true, value: undefined }); +}); + +test('a failed emission rolls back forced slots, lazy aliases and promise indices', async () => { + const key = 'x'.repeat(80); + const nodes = Array.from({ length: 32 }, (_, i) => ({ i })); + const first = Promise.withResolvers(); + const second = Promise.withResolvers(); + const nested = Promise.resolve('nested'); + const errors = []; + const stream = unevalStream( + { [key]: nodes, first: first.promise, second: second.promise }, + undefined, + { id: 'rollback', onerror: (error) => errors.push(error) } + ); + const c = client(); + const root = c.head(stream.head); + const slots = c.context.__d.rollback.o; + // Fail during rendering (the second read), after demand planning has staged aliases. + let reads = 0; + const failing = { + refs: nodes.slice(), + ['y'.repeat(256)]: { nested }, + get bad() { + if (++reads === 2) throw new Error('render failed'); + return 1; + } + }; + first.resolve(failing); + const failed = await stream.tail.next(); + c.block(failed.value); + await expect(root.first).rejects.toThrow('failed to serialize'); + expect(errors).toHaveLength(1); + expect(slots).toHaveLength(1); + second.resolve({ refs: nodes.slice(), nested }); + let tail = ''; + for await (const block of stream.tail) { + tail += block; + c.block(block); + } + const outcome = await root.second; + for (let i = 0; i < nodes.length; i += 1) expect(outcome.refs[i]).toBe(root[key][i]); + expect(await outcome.nested).toBe('nested'); + expect(tail).toContain('s.d(2)'); + expect(Object.keys(slots)).toEqual(Array.from({ length: slots.length }, (_, i) => String(i))); +}); + +test('successful aliases in one block survive a later failing settlement', async () => { + const key = 'x'.repeat(80); + const nodes = Array.from({ length: 32 }, (_, i) => ({ i })); + const promises = Array.from({ length: 3 }, () => Promise.withResolvers()); + const stream = unevalStream({ [key]: nodes, outcomes: promises.map((p) => p.promise) }); + const c = client(); + const root = c.head(stream.head); + promises[0].resolve(nodes.slice()); + promises[1].resolve({ bad: () => {} }); + await Promise.resolve(); + const block = await stream.tail.next(); + c.block(block.value); + const outcome = await root.outcomes[0]; + await expect(root.outcomes[1]).rejects.toThrow('failed to serialize'); + promises[2].resolve(nodes.slice()); + for await (const block of stream.tail) c.block(block); + const reused = await root.outcomes[2]; + for (let i = 0; i < nodes.length; i += 1) { + expect(outcome[i]).toBe(root[key][i]); + expect(reused[i]).toBe(outcome[i]); + } +}); + +test('forced anchors preserve cycles, sparse arrays and binary sharing without pooled bytes', async () => { + const buffer = new ArrayBuffer(16); + new Uint8Array(buffer).set([1, 2, 3, 4]); + const pooled = Buffer.allocUnsafe(3); + pooled.set([7, 8, 9]); + const sparse = []; + sparse.length = 0xffffffff; + sparse[4000000000] = { value: 1 }; + const data = { + buffer, + view: new Uint16Array(buffer, 2, 3), + other: new Uint8Array(buffer, 1, 5), + data_view: new DataView(buffer, 3, 5), + pooled, + sparse, + set: new Set(), + map: new Map() + }; + data.self = data; + data.set.add(data); + data.map.set(data, sparse); + const key = '<'.repeat(24); + const value = { [key]: data, outcome: Promise.resolve([data, buffer, data.view, sparse]) }; + const stream = unevalStream(value); + const c = client(); + const root = c.head(stream.head); + let source = stream.head; + for await (const block of stream.tail) { + source += block; + c.block(block); + } + await compare_topology(value, root); + expect(root[key].pooled.buffer.byteLength).toBe(3); + expect(source.length).toBeLessThan(3000); +}); + +test('compact unshared typed-array emission is not changed by speculative buffer retention', async () => { + const view = new Float64Array([1.5, -0, 2.5]); + const stream = unevalStream({ view, outcome: Promise.resolve(view) }); + expect(stream.head).toContain('new Float64Array([1.5,-0,2.5])'); + expect(stream.head).not.toContain('new Uint8Array'); + const c = client(); + const root = c.head(stream.head); + for await (const block of stream.tail) c.block(block); + expect(await root.outcome).toBe(root.view); +}); + +test.each(['typed view', 'DataView'])( + '%s preserves its buffer when the buffer path independently crosses the ceiling', + async (kind) => { + const buffer = new ArrayBuffer(24); + new Uint8Array(buffer).set([1, 2, 3, 4, 5, 6, 7, 8]); + const view = + kind === 'typed view' ? new Float64Array(buffer, 8, 1) : new DataView(buffer, 3, 5); + const key = 'x'.repeat(124); + const value = { [key]: view, outcome: Promise.resolve([view, buffer]) }; + const stream = unevalStream(value); + const c = client(); + const root = c.head(stream.head); + expect(stream.head).toContain('new Uint8Array'); + for await (const block of stream.tail) c.block(block); + await compare_topology(value, root); + expect((await root.outcome)[1]).toBe(root[key].buffer); + } +); + +test('replaced containers retain identity without shadowing free replacer identifiers', async () => { + class Box { + constructor(value) { + this.value = value; + } + } + const key = 'x'.repeat(80); + const nodes = Array.from({ length: 32 }, (_, i) => new Box({ i })); + const result = await expect_roundtrip( + (schedule) => ({ [key]: nodes, outcome: schedule.later(nodes.slice()) }), + { + replacer: (value, js) => value instanceof Box && js`new o(${value.value})`, + globals: { o: Box } + } + ); + expect(result.blocks.join('')).not.toContain('let o='); +}); diff --git a/test/stream-regressions.test.js b/test/stream-regressions.test.js new file mode 100644 index 00000000..f6c9efaf --- /dev/null +++ b/test/stream-regressions.test.js @@ -0,0 +1,152 @@ +import { expect, test } from 'vitest'; +import { parse, stringify, unevalStream } from '../src/index.js'; +import { client, expect_roundtrip } from './helpers/stream.js'; + +/** @param {string} source */ +function utf8_transport(source) { + const bytes = new TextEncoder().encode(source); + const decoder = new TextDecoder(); + let reconstructed = ''; + for (const byte of bytes) reconstructed += decoder.decode(Uint8Array.of(byte), { stream: true }); + return reconstructed + decoder.decode(); +} + +test('survives UTF-8 byte boundaries and separate or concatenated VM evaluation', async () => { + const resolved = Promise.withResolvers(); + const failed = Promise.withResolvers(); + const shared = { label: 'shared-😀' }; + const reason = 'late rejection 😀 \udfff'; + const id = 'transport-001-😀--\ud800'; + const value = { + numeric: { 0: '0', 12: '12', '001': '001' }, + lone: '\ud800', + pair: '😀', + closing: '', + shared, + resolved: resolved.promise, + rejected: failed.promise + }; + const result = unevalStream(value, undefined, { id }); + expect(result.id).toBe(id); + resolved.resolve({ shared, again: shared }); + const sources = [result.head, /** @type {string} */ ((await result.tail.next()).value)]; + failed.reject(reason); + for await (const block of result.tail) sources.push(block); + expect(sources.length).toBe(3); + for (const source of sources) expect(utf8_transport(source)).toBe(source); + expect(sources.join('')).not.toMatch(/<\/script/i); + + /** @param {any} root */ + async function verify(root) { + expect(Object.keys(root.numeric)).toEqual(['0', '12', '001']); + expect(root.lone).toBe('\ud800'); + expect(root.pair).toBe('😀'); + expect(root.closing).toBe(value.closing); + const outcome = await root.resolved; + expect(outcome.shared).toBe(root.shared); + expect(outcome.again).toBe(root.shared); + await expect(root.rejected).rejects.toBe(reason); + } + + const [head, ...blocks] = sources.map(utf8_transport); + const separate = client(); + const root = separate.head(head); + for (const block of blocks) separate.block(block); + await verify(root); + await verify(client().combined(head, blocks)); +}); + +/** + * @param {unknown} value + * @param {'head' | 'tail'} mode + * @param {import('../src/types.js').UnevalReplacer} [replacer] + * @param {Record} [globals] + */ +async function roundtrip(value, mode, replacer, globals) { + const { head, blocks, root } = await expect_roundtrip( + ({ later }) => (mode === 'head' ? value : later(value)), + { replacer, globals, orders: ['fifo'] } + ); + return { restored: mode === 'head' ? root : await root, source: head + blocks.join('') }; +} + +class Box { + /** @param {unknown} value */ + constructor(value) { + this.value = value; + } +} + +test.each( + ['head', 'tail'].flatMap((mode) => + [ + { kind: 'string', make: (/** @type {number} */ n) => 'x'.repeat(n) }, + { + kind: 'escaped string', + make: (/** @type {number} */ n) => '\n"\\\0\u2028\u2029'.repeat(n) + }, + { kind: 'bigint', make: (/** @type {number} */ n) => BigInt('9'.repeat(n)) } + ].map((row) => ({ mode: /** @type {'head' | 'tail'} */ (mode), ...row })) + ) +)('$mode: repeated $kind output grows linearly', async ({ mode, make }) => { + let previous = 0; + for (const n of [2000, 4000]) { + const encoded = stringify(Array(n).fill(make(n))); + const { source } = await roundtrip(parse(encoded), mode); + expect(source.length).toBeLessThan(encoded.length * 4); + if (previous) expect(source.length).toBeLessThan(previous * 2.1); + previous = source.length; + expect(source).not.toContain('<'); + } +}); + +test.each( + ['head', 'tail'].flatMap((mode) => + ['inline', 'shared', 'cyclic', 'holes', 'custom-cycle'].map((kind) => ({ + mode: /** @type {'head' | 'tail'} */ (mode), + kind + })) + ) +)('$mode: $kind sparse arrays avoid eager allocation', async ({ mode, kind }) => { + // eager allocation would need ~20GB; parse creates dictionary-mode arrays + const length = 1_000_000; + const arrays = Array.from({ length: 2500 }, () => parse(`[[-7,${length},0,1],42]`)); + for (const array of arrays) { + if (kind === 'holes') delete array[0]; + else + array[1] = kind === 'cyclic' ? array : kind === 'custom-cycle' ? new Box(array) : undefined; + } + const value = kind === 'shared' ? [arrays, arrays.slice()] : arrays; + const { restored } = await roundtrip( + value, + mode, + (v, js) => (v instanceof Box ? js`new Box(${v.value})` : undefined), + { Box } + ); + const result = kind === 'shared' ? restored[0] : restored; + for (const i of [0, arrays.length - 1]) { + expect(result[i].length).toBe(length); + expect(Object.getOwnPropertyNames(result[i])).toEqual( + kind === 'holes' ? ['length'] : ['0', '1', 'length'] + ); + } +}); + +test.each([{ mode: /** @type {const} */ ('head') }, { mode: /** @type {const} */ ('tail') }])( + '$mode: Buffers preserve identity without exposing their shared pool', + async ({ mode }) => { + const allocation = Buffer.allocUnsafe(128).fill(255); + const first = allocation.subarray(8, 11); + const second = allocation.subarray(24, 26); + first.set([1, 2, 3]); + second.set([4, 5]); + /** @type {Record} */ + const value = { first, again: first, second }; + value.self = value; + const { restored, source } = await roundtrip(value, mode); + expect(restored.first.buffer.byteLength).toBe(3); + expect(restored.second.buffer.byteLength).toBe(2); + expect(restored.first.buffer).not.toBe(restored.second.buffer); + expect(source).not.toContain('255'); + } +); diff --git a/test/stream-retention.test.js b/test/stream-retention.test.js new file mode 100644 index 00000000..ed4e30ef --- /dev/null +++ b/test/stream-retention.test.js @@ -0,0 +1,306 @@ +import { test } from 'vitest'; +import { in_subprocess } from './helpers/subprocess.js'; + +async function retention({ assert, turn, unevalStream, data: mode }) { + assert.equal(typeof global.gc, 'function'); + + function create_root() { + const source = Promise.withResolvers(); + const graph = { nodes: Array.from({ length: 1000 }, (_, i) => ({ i })) }; + graph.self = graph; + if (mode !== 'plain replacer') graph.pending = source.promise; + const weak = new WeakRef(graph); + const node_weak = new WeakRef(graph.nodes[500]); + const controller = new AbortController(); + const result = + mode === 'sanity' + ? null + : unevalStream(graph, mode === 'plain replacer' ? () => undefined : undefined, { + signal: controller.signal + }); + // Only the live control keeps an explicit input reference. This does not require + // active streams to use strong caches rather than an otherwise valid WeakMap. + return { + source, + weak, + node_weak, + controller, + result, + live_graph: mode === 'live' ? graph : null + }; + } + + async function cancel(fixture) { + const next = fixture.result.tail.next(); + if (mode.startsWith('return')) { + assert.deepEqual(await fixture.result.tail.return(), { done: true, value: undefined }); + assert.deepEqual(await next, { done: true, value: undefined }); + } else { + const reason = new Error('abort stream'); + fixture.controller.abort(reason); + await assert.rejects(next, (error) => error === reason); + } + if (!mode.endsWith('retained')) fixture.result = null; + fixture.controller = null; + } + + const fixture = create_root(); + let settled = false; + fixture.source.promise.then( + () => { + settled = true; + }, + () => { + settled = true; + } + ); + if (mode.startsWith('return') || mode.startsWith('abort')) { + await cancel(fixture); + } else if (mode === 'complete') { + fixture.source.resolve(42); + for await (const block of fixture.result.tail) assert.equal(typeof block, 'string'); + } else if (mode === 'plain replacer') { + assert.deepEqual(await fixture.result.tail.next(), { done: true, value: undefined }); + } + + // A deref during this loop would pin the target for the current job. Check only + // after repeated GCs in distinct jobs, including a turn after WeakRef creation. + for (let i = 0; i < 20; i += 1) { + await turn(); + global.gc(); + } + assert.ok(fixture.source.promise instanceof Promise, 'source remains externally held'); + assert.equal(settled, mode === 'complete', 'cancelled sources must remain pending'); + if (mode === 'live') { + assert.ok(fixture.result.head.length); + assert.equal(fixture.weak.deref(), fixture.live_graph); + assert.ok(fixture.node_weak.deref()); + } else { + if (mode === 'complete' || mode === 'plain replacer' || mode.endsWith('retained')) { + assert.ok(fixture.result.head.length, 'the finished result remains externally held'); + } + assert.ok(fixture.weak.deref() === undefined, `${mode}: stream retained the input root`); + assert.ok(fixture.node_weak.deref() === undefined, `${mode}: stream retained an input node`); + } + // Only after proving collection with the source still held and pending, show + // that cancellation did not cancel the job and late delivery stays detached. + if (mode.startsWith('return') || mode.startsWith('abort')) { + fixture.source.resolve(42); + assert.equal(await fixture.source.promise, 42); + if (fixture.result) { + assert.deepEqual(await fixture.result.tail.next(), { done: true, value: undefined }); + } + } +} + +test.each(['return', 'abort', 'return retained', 'abort retained'])( + '%s releases the graph while its native source stays pending', + (mode) => { + in_subprocess(retention, { expose_gc: true, data: mode }); + } +); + +test('natural completion releases the input graph with the result retained', () => { + in_subprocess(retention, { expose_gc: true, data: 'complete' }); +}); + +test('a no-promise head with a replacer releases the graph with the result retained', () => { + in_subprocess(retention, { expose_gc: true, data: 'plain replacer' }); +}); + +test('control: the same graph collects without a stream retaining it', () => { + in_subprocess(retention, { expose_gc: true, data: 'sanity' }); +}); + +test('control: a live stream with an externally held input keeps its graph', () => { + in_subprocess(retention, { expose_gc: true, data: 'live' }); +}); + +async function callback_retention({ assert, vm, turn, unevalStream, data: callback }) { + assert.equal(typeof global.gc, 'function'); + let reports = 0; + + function create_result() { + const graph = { nodes: Array.from({ length: 1000 }, (_, i) => ({ i })), visits: 0 }; + const weak = new WeakRef(graph); + const replacer = () => { + graph.visits += 1; + return undefined; + }; + const onerror = () => { + reports += graph.nodes.length; + }; + const options = callback === 'onerror' ? { onerror } : {}; + // The graph is callback-only data, never an input node or cached replacement. + const result = unevalStream( + callback === 'replacer' ? { value: 42 } : Promise.resolve(42), + callback === 'replacer' ? replacer : undefined, + options + ); + if (callback === 'replacer') assert.ok(graph.visits > 0); + // No caller-held graph, callback or options reference escapes this helper. + return { weak, result }; + } + + const fixture = create_result(); + const context = vm.createContext({}); + const root = vm.runInContext(`(${fixture.result.head})`, context); + for await (const block of fixture.result.tail) vm.runInContext(block, context); + if (callback === 'replacer') assert.equal(root.value, 42); + else assert.equal(await root, 42); + assert.equal(reports, 0, 'supported outcomes must not invoke onerror'); + + // No reporter is active; a retained terminal result must release its callbacks. + // As above, do not deref until all GCs have run in separate jobs. + for (let i = 0; i < 20; i += 1) { + await turn(); + global.gc(); + } + assert.ok(fixture.result.head.length, 'the terminal result remains externally held'); + assert.ok( + fixture.weak.deref() === undefined, + `terminal result retained the ${callback} callback's otherwise unreferenced graph` + ); +} + +test.each(['replacer', 'onerror'])( + 'a retained terminal result releases graph-capturing %s callbacks', + (callback) => in_subprocess(callback_retention, { expose_gc: true, data: callback }) +); + +test('control: a retained terminal result does not retain an unrelated helper graph', () => { + in_subprocess(callback_retention, { expose_gc: true, data: 'none' }); +}); + +test('a live stream retains identities until the final delivery', () => { + in_subprocess( + async ({ assert, vm, turn, unevalStream }) => { + function create_result() { + const source = Promise.withResolvers(); + const graph = { node: {}, pending: source.promise }; + return { source, weak: new WeakRef(graph), result: unevalStream(graph) }; + } + const fixture = create_result(); + const context = vm.createContext({}); + const root = vm.runInContext(`(${fixture.result.head})`, context); + for (let i = 0; i < 20; i += 1) { + await turn(); + global.gc(); + } + const graph = fixture.weak.deref(); + assert.ok(graph, 'the active session must retain identities without caller-held input'); + fixture.source.resolve({ root: graph, node: graph.node }); + for await (const block of fixture.result.tail) vm.runInContext(block, context); + const delivered = await root.pending; + assert.equal(delivered.root, root); + assert.equal(delivered.node, root.node); + assert.equal(Object.keys(context.__d).length, 0, 'final block deletes the client session'); + }, + { expose_gc: true } + ); +}); + +test.each(['return', 'abort'])( + '%s releases queued outcomes and wakes ordered concurrent readers', + (mode) => { + in_subprocess( + async ({ assert, turn, unevalStream, data: mode }) => { + function create_result(signal) { + const queued = Promise.withResolvers(); + const pending = Promise.withResolvers(); + const graph = { node: {} }; + const result = unevalStream( + { queued: queued.promise, pending: pending.promise }, + undefined, + { signal } + ); + queued.resolve(graph); + return { result, pending, weak: new WeakRef(graph) }; + } + const controller = new AbortController(); + const fixture = create_result(controller.signal); + // Use a separate pending stream to check ordered wakeup and exact reason, + // while the first stream has an undelivered graph-bearing outcome. + const waiting = unevalStream(fixture.pending.promise, undefined, { + signal: controller.signal + }); + const readers = [waiting.tail.next(), waiting.tail.next(), waiting.tail.next()]; + await turn(); + if (mode === 'abort') { + const reason = { message: 'exact abort reason' }; + controller.abort(reason); + await assert.rejects(readers[0], (error) => error === reason); + } else { + await fixture.result.tail.return(); + await waiting.tail.return(); + assert.deepEqual(await readers[0], { done: true, value: undefined }); + } + for (const reader of readers.slice(1)) { + assert.deepEqual(await reader, { done: true, value: undefined }); + } + if (mode === 'abort') { + await assert.rejects( + fixture.result.tail.next(), + (error) => error === controller.signal.reason + ); + } + for (let i = 0; i < 20; i += 1) { + await turn(); + global.gc(); + } + assert.ok(fixture.weak.deref() === undefined, 'terminal queue releases its outcome'); + assert.ok(fixture.pending.promise instanceof Promise, 'source is still externally held'); + fixture.pending.resolve(42); + assert.equal(await fixture.pending.promise, 42); + assert.deepEqual(await waiting.tail.next(), { done: true, value: undefined }); + }, + { expose_gc: true, data: mode } + ); + } +); + +test.each(['return', 'abort'])( + 'synchronous %s during a replacer cannot republish terminal state', + (mode) => { + in_subprocess( + async ({ assert, turn, unevalStream, data: mode }) => { + function create_result() { + const source = Promise.withResolvers(); + const pending = Promise.withResolvers(); + const controller = new AbortController(); + const graph = { node: {}, pending: pending.promise, text: 'long text '.repeat(100) }; + let result; + result = unevalStream( + source.promise, + (thing) => { + if (thing === graph) { + if (mode === 'return') result.tail.return(); + else controller.abort(0); + } + return undefined; + }, + { signal: controller.signal } + ); + source.resolve(graph); + return { result, pending, weak: new WeakRef(graph) }; + } + const fixture = create_result(); + await fixture.result.tail.next(); + if (mode === 'abort') { + await assert.rejects(fixture.result.tail.next(), (error) => error === 0); + } + assert.deepEqual(await fixture.result.tail.next(), { done: true, value: undefined }); + for (let i = 0; i < 20; i += 1) { + await turn(); + global.gc(); + } + assert.ok(fixture.result.head.length, 'terminal result stays reachable'); + assert.ok(fixture.weak.deref() === undefined, 'emission did not republish the graph'); + assert.ok(fixture.pending.promise instanceof Promise, 'nested source is still held'); + fixture.pending.resolve(42); + assert.equal(await fixture.pending.promise, 42); + }, + { expose_gc: true, data: mode } + ); + } +); diff --git a/test/stream.test.js b/test/stream.test.js new file mode 100644 index 00000000..612407fc --- /dev/null +++ b/test/stream.test.js @@ -0,0 +1,425 @@ +import vm from 'node:vm'; +import { expect, test, vi } from 'vitest'; +import { unevalStream } from '../src/index.js'; +import { client, expect_roundtrip, seeded_random, shuffle } from './helpers/stream.js'; + +test.each([ + { + name: 'cycle', + make: () => { + const a = {}; + a.self = a; + return a; + } + }, + { + name: 'head → outcome identity', + make: ({ later }) => { + const s = {}; + return { s, p: later(s) }; + } + }, + { + name: 'outcome → outcome identity', + make: ({ later }) => { + const s = {}; + return [later(s), later({ s })]; + } + }, + { name: 'nested promises', make: ({ later }) => ({ p: later({ q: later({ r: later(1) }) }) }) }, + { + name: 'cycle through outcome', + make: ({ later }) => { + const root = {}; + root.p = later({ root }); + return root; + } + }, + { + name: 'rejection identity', + make: ({ later_reject }) => { + const e = { code: 1 }; + return { e, p: later_reject(e) }; + } + }, + { + name: 'map key identity', + make: ({ later }) => { + const k = {}; + return { m: new Map([[k, 1]]), p: later(k) }; + } + }, + { + name: 'set member identity', + make: ({ later }) => { + const k = {}; + return { s: new Set([k]), p: later(k) }; + } + }, + { + name: 'typed view buffer identity', + make: ({ later }) => { + const b = new ArrayBuffer(4); + return { v: new Uint8Array(b), p: later(b) }; + } + }, + { + name: 'sparse + null-proto', + make: ({ later }) => { + const a = Array(5); + a[3] = 'x'; + return { p: later(Object.assign(Object.create(null), { a })) }; + } + }, + { name: 'promise inside map value', make: ({ later }) => ({ m: new Map([['k', later(2)]]) }) }, + { + name: 'same promise twice', + make: ({ later }) => { + const p = later(1); + return [p, p, { p }]; + } + } +])('$name', ({ make }) => expect_roundtrip(make)); + +class Box { + /** @param {unknown} value */ + constructor(value) { + this.value = value; + } +} + +class Job { + /** @param {Promise} completion */ + constructor(completion) { + this.completion = completion; + } +} + +class Cyclic { + /** @param {unknown} [meta] */ + constructor(meta) { + this.meta = meta; + } +} + +class MyThenable { + /** @param {Promise} promise */ + constructor(promise) { + this.promise = promise; + } + /** @type {Promise['then']} */ + then(resolve, reject) { + return this.promise.then(resolve, reject); + } +} + +/** @type {import('../src/types.js').UnevalReplacer} */ +const replacer = (v, js) => { + if (v instanceof Box) return js`new Box(${v.value})`; + if (v instanceof Job) return js`new Job(${v.completion})`; + if (v instanceof Cyclic) return js`new Cyclic(${v.meta})`; + if (v instanceof MyThenable) return js`new MyThenable(${Promise.resolve(v)})`; +}; + +test.each([ + { + name: 'replaced object in head', + make: ({ later }) => { + const b = new Box({ x: 1 }); + return { b, p: later(b) }; + } + }, + { name: 'replaced object in outcome', make: ({ later }) => ({ p: later(new Box({ x: 1 })) }) }, + { name: 'replaced object wrapping a promise', make: ({ later }) => [new Job(later({ ok: 1 }))] }, + { + name: 'replaced object whose hole is a head object', + make: ({ later }) => { + const s = {}; + return { s, p: later(new Box(s)) }; + } + }, + { + name: 'cycle through a replaced object', + make: ({ later }) => { + const c = new Cyclic(); + c.meta = { parent: c }; + return { c, p: later(c.meta), q: later(new Cyclic([1])) }; + } + }, + { + name: 'custom thenable rebuilt on the client', + make: ({ later }) => { + const t = new MyThenable(later({ y: 2 })); + return { t, again: t, p: later(t) }; + } + } +])('replacer: $name', ({ make }) => + expect_roundtrip(make, { replacer, globals: { Box, Job, Cyclic, MyThenable } }) +); + +class MatrixWrapper { + /** @param {string} kind @param {unknown} value */ + constructor(kind, value) { + this.kind = kind; + this.value = value; + } +} + +/** + * @param {number} seed + * @param {(value: unknown) => Promise} later + */ +function build_matrix_case(seed, later) { + const random = seeded_random(seed); + const nodes = Array.from({ length: 8 }, (_, index) => ({ + index, + label: `node-${seed}-${index}`, + /** @type {unknown} */ + next: null + })); + for (let i = 0; i < nodes.length; i += 1) { + nodes[i].next = nodes[(i + 1 + Math.floor(random() * 3)) % nodes.length]; + } + + const shared = nodes[Math.floor(random() * nodes.length)]; + const distinct = { index: shared.index, label: shared.label, next: shared.next }; + const null_object = Object.assign(Object.create(null), { kind: 'null', shared, distinct }); + const sparse = Array(7); + sparse[1] = shared; + sparse[4] = null_object; + const buffer = new ArrayBuffer(12); + const bytes = new Uint8Array(buffer); + for (let i = 0; i < bytes.length; i += 1) bytes[i] = (seed * 17 + i * 29) & 255; + const view = new Uint16Array(buffer, 2, 4); + const wrapped = new MatrixWrapper(`wrapper-${seed}`, shared); + const map = new Map( + shuffle( + [ + [shared, view], + [null_object, sparse], + [distinct, buffer] + ], + random + ) + ); + const set = new Set(shuffle([view, shared, null_object, distinct], random)); + + const outcome_a = { + kind: 'outcome-a', + shared, + wrapped, + map: new Map([[shared, nodes[(seed + 3) % nodes.length]]]) + }; + const outcome_b = Object.assign(Object.create(null), { + kind: 'outcome-b', + items: [outcome_a, shared, view], + set: new Set([distinct, outcome_a, buffer]), + inner: later(new MatrixWrapper('late', nodes[seed % nodes.length])) + }); + const entries = shuffle( + [ + ['seed', seed], + ['numeric', { 0: '0', 12: '12', '001': '001' }], + ['nodes', nodes], + ['shared', shared], + ['distinct', distinct], + ['null_object', null_object], + ['sparse', sparse], + ['map', map], + ['set', set], + ['buffer', buffer], + ['view', view], + ['wrapped', wrapped], + ['promises', [outcome_a, outcome_b, outcome_a].map((o) => later(o))] + ], + random + ); + return Object.fromEntries(entries); +} + +test.each(Array.from({ length: 20 }, (_, i) => ({ seed: i + 1 })))( + 'random graph (seed $seed)', + ({ seed }) => + expect_roundtrip(({ later }) => build_matrix_case(seed, later), { + replacer: (v, js) => + v instanceof MatrixWrapper ? js`new MatrixWrapper(${v.kind},${v.value})` : undefined, + globals: { MatrixWrapper } + }) +); + +test('recognizes only branded native promises across realms and subclasses', async () => { + class SubPromise extends Promise {} + await expect_roundtrip(() => [ + Promise.resolve(1), + vm.runInNewContext('Promise.resolve(2)'), + SubPromise.resolve(3) + ]); + + expect(() => unevalStream({ [Symbol.toStringTag]: 'Promise', then() {} })).toThrow( + /Cannot stringify/ + ); +}); + +test('calls the replacer once per identity, even after a failed outcome', async () => { + class X { + constructor() { + this.value = 42; + } + } + const x = new X(); + /** @type {unknown[]} */ + const seen = []; + const onerror = vi.fn(); + + const a = Promise.withResolvers(); + const b = Promise.withResolvers(); + const stream = unevalStream( + { p1: a.promise, p2: b.promise }, + (value, js) => { + seen.push(value); + if (value instanceof X) return js`{value:${value.value}}`; + }, + { onerror } + ); + + const c = client(); + const root = c.head(stream.head); + + a.resolve({ x, fn: () => {} }); + c.block(/** @type {string} */ ((await stream.tail.next()).value)); + b.resolve(x); + c.block(/** @type {string} */ ((await stream.tail.next()).value)); + + expect(seen.filter((v) => v === x)).toHaveLength(1); + await expect(root.p1).rejects.toThrow('devalue: failed to serialize asynchronous value'); + expect(await root.p2).toEqual({ value: 42 }); + expect(onerror).toHaveBeenCalledOnce(); + expect((await stream.tail.next()).done).toBe(true); +}); + +test('head is deterministic regardless of settlement timing', () => { + const settled = unevalStream({ a: [1], p: Promise.resolve(1) }, undefined, { id: 'x' }); + const unsettled = unevalStream({ a: [1], p: new Promise(() => {}) }, undefined, { id: 'x' }); + + expect(settled.head).toBe(unsettled.head); +}); + +test('folds settled promises via the replacer', async () => { + const unhandled = vi.fn(); + process.on('unhandledRejection', unhandled); + + try { + const shared = { x: 1 }; + const ok = Promise.resolve(shared); + const bad = Promise.reject(shared); + bad.catch(() => {}); + + /** @type {Map} */ + const settled = new Map([ + [ok, { ok: true, value: shared }], + [bad, { ok: false, reason: shared }] + ]); + /** + * This is just a simple helper to create a rejected promise that doesn't throw an unhandled rejection + * @type {(reason: unknown, js: import('../src/types.js').JavaScriptTag) => any} + */ + const rejected = (reason, js) => js`(p => (p.catch(() => {}), p))(Promise.reject(${reason}))`; + + const { root, blocks } = await expect_roundtrip(() => ({ shared, ok, bad }), { + replacer: (v, js) => { + const r = settled.get(v); + if (r) return r.ok ? js`Promise.resolve(${r.value})` : rejected(r.reason, js); + }, + orders: ['fifo'] + }); + + expect(blocks).toHaveLength(0); + await new Promise((r) => setTimeout(r, 10)); + expect(unhandled).not.toHaveBeenCalled(); + const revived = /** @type {any} */ (root); + expect(await revived.ok).toBe(revived.shared); + } finally { + process.off('unhandledRejection', unhandled); + } +}); + +test('emits a plain expression when there are no promises', async () => { + const stream = unevalStream({ x: 1 }); + const c = client(); + + expect(stream.head).not.toContain('__d'); + expect(c.head(stream.head)).toEqual({ x: 1 }); + expect((await stream.tail.next()).done).toBe(true); + expect(c.context.__d).toBeUndefined(); +}); + +test('return() completes a pending next()', async () => { + const stream = unevalStream({ p: new Promise(() => {}) }); + const next = stream.tail.next(); + + expect(await stream.tail.return()).toEqual({ done: true, value: undefined }); + expect(await next).toEqual({ done: true, value: undefined }); +}); + +test('abort rejects a pending next() with the exact reason', async () => { + const controller = new AbortController(); + const stream = unevalStream({ p: new Promise(() => {}) }, undefined, { + signal: controller.signal + }); + const next = stream.tail.next(); + controller.abort(0); + + await expect(next).rejects.toBe(0); + expect(await stream.tail.next()).toEqual({ done: true, value: undefined }); +}); + +test.each([{ promises: true }, { promises: false }])( + 'abort during head emission throws the exact reason (promises: $promises)', + ({ promises }) => { + const controller = new AbortController(); + const reason = new Error('abort during head'); + const value = { + ...(promises ? { p: new Promise(() => {}) } : {}), + get tripwire() { + controller.abort(reason); + return 1; + } + }; + + let result; + let thrown; + try { + result = unevalStream(value, undefined, { signal: controller.signal }); + } catch (error) { + thrown = error; + } + + expect(thrown).toBe(reason); + expect(result).toBeUndefined(); + } +); + +test('concurrent next() calls settle in order', async () => { + const a = Promise.withResolvers(); + const b = Promise.withResolvers(); + const stream = unevalStream({ a: a.promise, b: b.promise }); + const c = client(); + const root = c.head(stream.head); + + /** @type {string[]} */ + const order = []; + const first = stream.tail.next().then((r) => (order.push('first'), r)); + const second = stream.tail.next().then((r) => (order.push('second'), r)); + const third = stream.tail.next().then((r) => (order.push('third'), r)); + + b.resolve('b'); + await new Promise((r) => setTimeout(r, 0)); + a.resolve('a'); + + const results = await Promise.all([first, second, third]); + expect(order).toEqual(['first', 'second', 'third']); + expect(results.map((r) => r.done)).toEqual([false, false, true]); + for (const r of results.slice(0, 2)) c.block(/** @type {string} */ (r.value)); + expect(await root.a).toBe('a'); + expect(await root.b).toBe('b'); +}); From 10690582ddc83c0a73538f2da5918a7def7c85b6 Mon Sep 17 00:00:00 2001 From: Elliott Johnson Date: Thu, 8 Oct 2026 21:44:12 -0600 Subject: [PATCH 2/6] refactor: drop path aliases and the local slot binding from unevalStream --- src/uneval-stream.js | 89 +++------------------------- test/stream-reference-growth.test.js | 53 ++++------------- 2 files changed, 19 insertions(+), 123 deletions(-) diff --git a/src/uneval-stream.js b/src/uneval-stream.js index 5ab74a37..fcdde4d8 100644 --- a/src/uneval-stream.js +++ b/src/uneval-stream.js @@ -5,9 +5,7 @@ import { js, JavaScriptSource } from './javascript-source.js'; import { analyze } from './uneval.js'; import { call_replacer, get_type, stringify_string } from './utils.js'; -// `o` is reserved even when unused, since names are allocated before the binding is chosen -const RESERVED = ['s', 'n', 'o']; -const LOCAL_BINDING_COST = '(()=>{let o=s.o;return })()'.length; +const RESERVED = ['s', 'n']; const FAILED = 'new Error("devalue: failed to serialize asynchronous value")'; const DONE = Object.freeze({ done: /** @type {true} */ (true), value: undefined }); const identifier = /^[_$a-zA-Z][_$a-zA-Z0-9]*$/; @@ -129,14 +127,7 @@ class Session { #known = new Map(); /** - * Sometimes, paths are long enough that it makes sense to give them their own slot. - * When this occurs, we map the location to that slot. - * @type {Map} - */ - #aliases = new Map(); - - /** - * This is like `#aliases`, but for primitives. We can vastly reduce the wire size of payloads + * Slots for long strings and bigints. We can vastly reduce the wire size of payloads * containing repeated strings or bigints (whose textual length is essentially unbounded) by saving them * in a slot, which lets us reuse them with very short syntax. * @type {Map} @@ -187,7 +178,6 @@ class Session { #emit_value(value, head) { const known = this.#known; const replacer_cache = this.#replacer_cache; - const aliases = this.#aliases; const primitives = this.#primitives; const replacer = this.#replacer; const slot_start = this.#next_slot; @@ -199,10 +189,6 @@ class Session { const retained = new Map(); /** @type {Map} */ const fresh = new Map(); - /** @type {Map} */ - const staged = new Map(); - /** @type {Array<{ slot: number, location: Location }>} */ - const assignments = []; /** @type {Map} */ const staged_primitives = new Map(); @@ -250,19 +236,11 @@ class Session { // a head without promises needs no session at all if (head && fresh.size === 0) return analysis.render(); - this.#plan_aliases(analysis.references.values(), staged, assignments); - - // Without a replacer there are no free custom identifiers to shadow, so a local - // `o` can stand in for `s.o` when the uses it shortens pay for the wrapper. The - // root, each retained object and reference, and each alias (twice) use a slot. - const slot_uses = 1 + retained.size + analysis.references.size + 2 * assignments.length; - const binding = !replacer && slot_uses * 2 > LOCAL_BINDING_COST ? 'o' : 's.o'; - let code = analysis.render({ - reference: (location) => expression(location, aliases, binding, staged), + reference: expression, retain: (thing) => { const slot = retained.get(thing); - return slot === undefined ? undefined : `${binding}[${slot}]`; + return slot === undefined ? undefined : `s.o[${slot}]`; }, primitive: (thing, literal) => { let slot = primitives.get(thing) ?? staged_primitives.get(thing); @@ -271,24 +249,12 @@ class Session { staged_primitives.set(thing, slot); primitive_assignments.push({ slot, literal }); } - return `${binding}[${slot}]`; + return `s.o[${slot}]`; } }); const root = locations.get(value); - if (root !== undefined) code = `(${binding}[${root.key}]=${code})`; - if (assignments.length) { - /** @type {Map} */ - const initialized = new Map(); - const initializers = assignments.map(({ slot, location }) => { - const prefix = expression(location, aliases, binding, initialized); - initialized.set(location, slot); - return `${binding}[${slot}]=${prefix}`; - }); - code = `(${initializers.join(',')},${code})`; - } - if (binding === 'o') code = `(()=>{let o=s.o;return ${code}})()`; - + if (root !== undefined) code = `(s.o[${root.key}]=${code})`; if (primitive_assignments.length) { const initializers = primitive_assignments.map( ({ slot, literal }) => `s.o[${slot}]=${literal}` @@ -299,7 +265,6 @@ class Session { // request state or attach new delivery after terminal cleanup. if (!this.#done) { for (const [thing, slot] of staged_primitives) primitives.set(thing, slot); - for (const [location, slot] of staged) aliases.set(location, slot); for (const [thing, location] of locations) known.set(thing, location); for (const [promise, index] of fresh) { this.#pending += 1; @@ -315,38 +280,6 @@ class Session { } } - /** - * Gives a slot to each shared path prefix that is used often enough to pay for - * its initialization. Each existing object is rendered once, because the emitter - * hoists repeated uses. Shared ancestors count as well as leaves, in - * ancestor-first order. Every walk is bounded by MAX_PATH_COST, not by the depth - * of the input graph. - * @param {Iterable} references - * @param {Map} staged - * @param {Array<{ slot: number, location: Location }>} assignments - */ - #plan_aliases(references, staged, assignments) { - const aliases = this.#aliases; - /** @type {Map} */ - const demand = new Map(); - for (const location of references) { - const path = []; - for (let node = location; node.up && !aliases.has(node); node = node.up) path.push(node); - for (const node of path.reverse()) demand.set(node, (demand.get(node) ?? 0) + 1); - } - for (const [location, count] of demand) { - const prefix = expression(location, aliases, 's.o', staged); - const slot = `s.o[${this.#next_slot}]`; - const prefix_cost = path_cost(prefix); - // Includes initialization, assignment, comma, and the first alias wrapper. - const assignment_cost = prefix_cost + slot.length + 2 + (assignments.length ? 0 : 2); - if (count * (prefix_cost - slot.length) > assignment_cost) { - assignments.push({ slot: this.#next_slot, location }); - staged.set(location, this.#next_slot++); - } - } - } - /** * @param {number} index * @param {boolean} ok @@ -383,7 +316,6 @@ class Session { this.#signal?.removeEventListener('abort', this.#on_abort); this.#known.clear(); this.#replacer_cache.clear(); - this.#aliases.clear(); this.#primitives.clear(); this.#queue.length = 0; this.#replacer = undefined; @@ -490,18 +422,15 @@ function is_promise(thing) { /** * @param {Location} location - * @param {Map} aliases - * @param {string} [binding] - * @param {Map} [staged] * @returns {string} */ -function expression(location, aliases, binding = 's.o', staged) { +function expression(location) { const segments = []; - while (location.up !== null && !aliases.has(location) && !staged?.has(location)) { + while (location.up !== null) { segments.push(location.key); location = location.up; } - return `${binding}[${staged?.get(location) ?? aliases.get(location) ?? location.key}]${segments.reverse().join('')}`; + return `s.o[${location.key}]${segments.reverse().join('')}`; } /** diff --git a/test/stream-reference-growth.test.js b/test/stream-reference-growth.test.js index 848140c1..a8f2f3ae 100644 --- a/test/stream-reference-growth.test.js +++ b/test/stream-reference-growth.test.js @@ -103,37 +103,9 @@ test('cheap single-use and repeated leaf paths do not create lazy slots', async // Only the new outcome array is retained, not its already-known leaf. expect(slots).toHaveLength(2); expect(blocks.join('')).toContain('s.o[0].leaf'); - expect(blocks.join('')).not.toContain('let o='); }); -test.each([15, 16])('prefix length %i accounts for alias wrapper punctuation', async (length) => { - const key = 'x'.repeat(length); - const nodes = [{ i: 0 }, { i: 1 }]; - const stream = unevalStream( - { [key]: nodes, outcome: Promise.resolve(nodes.slice()) }, - undefined, - { - id: 'cost' - } - ); - const c = client(); - const root = c.head(stream.head); - const slots = c.context.__d.cost.o; - let tail = ''; - for await (const block of stream.tail) { - tail += block; - c.block(block); - } - const outcome = await root.outcome; - for (let i = 0; i < nodes.length; i += 1) expect(outcome[i]).toBe(root[key][i]); - // At 15 the alias breaks even; at 16 it saves exactly one byte, including - // the first comma-expression wrapper. Neither scope pays for a local binding. - expect(slots).toHaveLength(length === 15 ? 2 : 3); - expect(tail.split(key)).toHaveLength(length === 15 ? 3 : 2); - expect(tail).not.toContain('let o='); -}); - -test('one long but bounded leaf read avoids an unprofitable alias', async () => { +test('a long but bounded leaf read is repeated rather than given a slot', async () => { const key = 'x'.repeat(80); const leaf = { value: 1 }; const stream = unevalStream({ [key]: leaf, outcome: Promise.resolve(leaf) }, undefined, { @@ -147,7 +119,7 @@ test('one long but bounded leaf read avoids an unprofitable alias', async () => expect(slots).toHaveLength(1); }); -test('profitable distinct descendants share an alias and a numeric local slot table', async () => { +test('distinct descendants are reached by bounded paths from a numeric slot table', async () => { const key = 'x'.repeat(80); const nodes = Array.from({ length: 32 }, (_, i) => ({ i })); const stream = unevalStream( @@ -170,11 +142,10 @@ test('profitable distinct descendants share an alias and a numeric local slot ta } const outcome = await root.outcome; for (let i = 0; i < nodes.length; i += 1) expect(outcome[i]).toBe(root[key][i]); - expect(tail.split(key)).toHaveLength(2); // one bounded alias initialization - expect(tail).toContain('let o=s.o;'); - expect(tail).toMatch(/o\[\d+\]=o\[\d+\]/); + // Each reference repeats a path of at most MAX_PATH_COST bytes; compression absorbs the repeats. + expect(tail.split(key)).toHaveLength(nodes.length + 1); expect(Object.keys(slots)).toEqual(Array.from({ length: slots.length }, (_, i) => String(i))); - expect(source.length).toBeLessThan(2500); + expect(source.length).toBeLessThan(nodes.length * 128 + 1024); }); test.each(['escaped segment', 'unicode segment', 'enormous segment', 'independent chains'])( @@ -363,7 +334,7 @@ test.each(['typed view', 'DataView'])( } ); -test('replaced containers retain identity without shadowing free replacer identifiers', async () => { +test('replaced containers retain identity', async () => { class Box { constructor(value) { this.value = value; @@ -371,12 +342,8 @@ test('replaced containers retain identity without shadowing free replacer identi } const key = 'x'.repeat(80); const nodes = Array.from({ length: 32 }, (_, i) => new Box({ i })); - const result = await expect_roundtrip( - (schedule) => ({ [key]: nodes, outcome: schedule.later(nodes.slice()) }), - { - replacer: (value, js) => value instanceof Box && js`new o(${value.value})`, - globals: { o: Box } - } - ); - expect(result.blocks.join('')).not.toContain('let o='); + await expect_roundtrip((schedule) => ({ [key]: nodes, outcome: schedule.later(nodes.slice()) }), { + replacer: (value, js) => value instanceof Box && js`new Box(${value.value})`, + globals: { Box } + }); }); From 16576f71747a69ab9b1466c18350308177f91d1a Mon Sep 17 00:00:00 2001 From: Elliott Johnson Date: Fri, 9 Oct 2026 10:30:34 -0600 Subject: [PATCH 3/6] perf: build unevalStream paths lazily and cap them by length --- src/uneval-stream.js | 63 +++++++++++++--------------- test/stream-reference-growth.test.js | 4 +- 2 files changed, 31 insertions(+), 36 deletions(-) diff --git a/src/uneval-stream.js b/src/uneval-stream.js index fcdde4d8..6cb25a76 100644 --- a/src/uneval-stream.js +++ b/src/uneval-stream.js @@ -11,8 +11,9 @@ const DONE = Object.freeze({ done: /** @type {true} */ (true), value: undefined const identifier = /^[_$a-zA-Z][_$a-zA-Z0-9]*$/; const then = Promise.prototype.then; const noop = () => {}; -// Bounds UTF-8 serialized property/bracket text, independently of numeric slot width. -const MAX_PATH_COST = 128; +// Bounds the length of a path's property/bracket text (at most 3 UTF-8 bytes per unit), +// independently of numeric slot width. +const MAX_PATH_LENGTH = 128; /** * Turn a value containing promises into a JavaScript expression (`head`) that @@ -183,8 +184,9 @@ class Session { const slot_start = this.#next_slot; const promise_start = this.#next_promise; + // A failed head ends the session, so only tails need to stage their locations. /** @type {Map} */ - const locations = new Map(); + const locations = head ? known : new Map(); /** @type {Map} */ const retained = new Map(); /** @type {Map} */ @@ -197,24 +199,20 @@ class Session { try { const analysis = analyze(value, { - known: (thing) => known.get(thing), + known: head ? undefined : (thing) => known.get(thing), enter: (thing, parent, key) => { - let up = null; - let segment = ''; - if (parent !== undefined) { - up = /** @type {Location} */ (locations.get(parent)); - if (typeof key === 'number') segment = `[${key}]`; - else if (key !== undefined && key.length <= MAX_PATH_COST) segment = prop(key); - } - const cost = path_cost(segment); - if (up && segment && up.cost + cost <= MAX_PATH_COST) { - locations.set(thing, { up, key: segment, cost: up.cost + cost }); - } else { - const slot = this.#next_slot++; - locations.set(thing, { up: null, key: slot, cost: 0 }); - // The root is assigned around the whole expression instead. - if (thing !== value) retained.set(thing, slot); + if (parent !== undefined && key !== undefined) { + const up = /** @type {Location} */ (locations.get(parent)); + const length = up.length + segment_length(key); + if (length <= MAX_PATH_LENGTH) { + locations.set(thing, { up, key, length }); + return; + } } + const slot = this.#next_slot++; + locations.set(thing, { up: null, key: slot, length: 0 }); + // The root is assigned around the whole expression instead. + if (thing !== value) retained.set(thing, slot); }, replace: (thing) => { if (replacer) { @@ -265,7 +263,7 @@ class Session { // request state or attach new delivery after terminal cleanup. if (!this.#done) { for (const [thing, slot] of staged_primitives) primitives.set(thing, slot); - for (const [thing, location] of locations) known.set(thing, location); + if (!head) for (const [thing, location] of locations) known.set(thing, location); for (const [promise, index] of fresh) { this.#pending += 1; Session.#subscribe(promise, index, this.#delivery); @@ -415,9 +413,9 @@ function is_promise(thing) { } /** - * A slot (`up` is null and `key` is its expression) or a property of another location - * Property keys are already serialized, bounded segments; root keys are slot indices. - * @typedef {{ up: Location | null, key: string | number, cost: number }} Location + * A slot (`up` is null and `key` is its index) or a property `key` of another location. + * `length` is the length of the path text from the nearest slot. + * @typedef {{ up: Location | null, key: string | number, length: number }} Location */ /** @@ -427,24 +425,21 @@ function is_promise(thing) { function expression(location) { const segments = []; while (location.up !== null) { - segments.push(location.key); + const key = location.key; + segments.push(typeof key === 'number' ? `[${key}]` : prop(key)); location = location.up; } return `s.o[${location.key}]${segments.reverse().join('')}`; } /** - * Count UTF-8 bytes of an already bounded segment or path. Never called on an - * unbounded property key; slot width is the only uncapped part of a path. - * @param {string} source + * Length of the path segment for `key`, or Infinity past the cap + * @param {string | number} key */ -function path_cost(source) { - let cost = 0; - for (const character of source) { - const code = /** @type {number} */ (character.codePointAt(0)); - cost += code < 0x80 ? 1 : code < 0x800 ? 2 : code < 0x10000 ? 3 : 4; - } - return cost; +function segment_length(key) { + if (typeof key === 'number') return String(key).length + 2; + if (key.length > MAX_PATH_LENGTH) return Infinity; + return identifier.test(key) ? key.length + 1 : prop(key).length; } /** @param {string} key */ diff --git a/test/stream-reference-growth.test.js b/test/stream-reference-growth.test.js index a8f2f3ae..43d7704d 100644 --- a/test/stream-reference-growth.test.js +++ b/test/stream-reference-growth.test.js @@ -142,7 +142,7 @@ test('distinct descendants are reached by bounded paths from a numeric slot tabl } const outcome = await root.outcome; for (let i = 0; i < nodes.length; i += 1) expect(outcome[i]).toBe(root[key][i]); - // Each reference repeats a path of at most MAX_PATH_COST bytes; compression absorbs the repeats. + // Each reference repeats a path of at most MAX_PATH_LENGTH characters; compression absorbs the repeats. expect(tail.split(key)).toHaveLength(nodes.length + 1); expect(Object.keys(slots)).toEqual(Array.from({ length: slots.length }, (_, i) => String(i))); expect(source.length).toBeLessThan(nodes.length * 128 + 1024); @@ -159,7 +159,7 @@ test.each(['escaped segment', 'unicode segment', 'enormous segment', 'independen } else { const key = kind === 'unicode segment' - ? 'é'.repeat(64) + ? 'é'.repeat(129) : '<'.repeat(kind === 'escaped segment' ? 24 : 4096); data = { [key]: leaf }; } From 0ed84ac82aab3951dc878ded5313bd72ccf06071 Mon Sep 17 00:00:00 2001 From: Elliott Johnson Date: Fri, 9 Oct 2026 11:04:22 -0600 Subject: [PATCH 4/6] feat: replace unevalStream's onerror with transformError --- README.md | 12 +-- src/types.ts | 10 +- src/uneval-stream.js | 139 +++++++++++++++++++-------- test/helpers/stream.js | 6 +- test/stream-reference-growth.test.js | 8 +- test/stream-regressions.test.js | 2 +- test/stream-retention.test.js | 8 +- test/stream.test.js | 113 +++++++++++++++++++++- 8 files changed, 238 insertions(+), 60 deletions(-) diff --git a/README.md b/README.md index 85ecb5bc..357a0327 100644 --- a/README.md +++ b/README.md @@ -110,12 +110,12 @@ Evaluate `head` once, then each block once, in order, in the same realm. Concate #### Options -| Option | Description | -| --------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `id` | Session key (returned as `id`). Random by default; pass a stable value when `head` must be byte-identical across runs. | -| `scope` | Trusted, assignable expression holding the client session table. Defaults to `globalThis.__d`. | -| `signal` | An `AbortSignal` that stops the tail. Pending and later `next()` calls reject once with `signal.reason`; if the signal is already aborted, or aborts during the head emission, `unevalStream` throws `signal.reason`. | -| `onerror` | `(error, value) => void`, called when an outcome can't be serialized. The client promise rejects with a generic `Error`. | +| Option | Description | +| ---------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `id` | Session key (returned as `id`). Random by default; pass a stable value when `head` must be byte-identical across runs. | +| `scope` | Trusted, assignable expression holding the client session table. Defaults to `globalThis.__d`. | +| `signal` | An `AbortSignal` that stops the tail. Pending and later `next()` calls reject once with `signal.reason`; if the signal is already aborted, or aborts during the head emission, `unevalStream` throws `signal.reason`. | +| `transformError` | `(error) => unknown`, called with a rejection reason, or with the error that kept a resolved value from serializing. Its result (or what its returned promise resolves to) is what the client promise rejects with. As with a replacer, returning `undefined`, `null` or `false` keeps the default: a generic `Error`, so server errors never reach the client unless you opt in. Throwing, rejecting, or returning something that can't be serialized also falls back to it. | #### Constraints diff --git a/src/types.ts b/src/types.ts index 1b6ce5a1..5e22fa1c 100644 --- a/src/types.ts +++ b/src/types.ts @@ -76,8 +76,14 @@ export interface UnevalStreamOptions { * throws `signal.reason`. */ signal?: AbortSignal; - /** Receives outcomes that could not be serialized; the client promise rejects with a generic Error. */ - onerror?: (error: unknown, value: unknown) => void; + /** + * Turns a rejection reason, or the error that kept a resolved value from serializing, + * into the value the client promise rejects with. May return a promise. Like a + * replacer, returning `undefined`, `null` or `false` keeps the default: the client + * promise rejects with a generic Error. So does throwing, rejecting, or returning + * something that can't be serialized. + */ + transformError?: (error: unknown) => unknown; } /** The result of `unevalStream`. */ diff --git a/src/uneval-stream.js b/src/uneval-stream.js index 6cb25a76..3525889e 100644 --- a/src/uneval-stream.js +++ b/src/uneval-stream.js @@ -7,6 +7,8 @@ import { call_replacer, get_type, stringify_string } from './utils.js'; const RESERVED = ['s', 'n']; const FAILED = 'new Error("devalue: failed to serialize asynchronous value")'; +// Stands in for an outcome that is sent as FAILED rather than serialized +const FAILURE = Symbol(); const DONE = Object.freeze({ done: /** @type {true} */ (true), value: undefined }); const identifier = /^[_$a-zA-Z][_$a-zA-Z0-9]*$/; const then = Promise.prototype.then; @@ -53,7 +55,7 @@ class Session { */ constructor(replacer, options, table) { this.#replacer = replacer; - this.#onerror = options.onerror; + this.#transform_error = options.transformError; this.#signal = options.signal; this.#table = table; this.#on_abort = this.#abort_stream.bind(this); @@ -61,8 +63,8 @@ class Session { /** @type {UnevalReplacer | undefined} */ #replacer; - /** @type {UnevalStreamOptions['onerror']} */ - #onerror; + /** @type {UnevalStreamOptions['transformError']} */ + #transform_error; /** @type {AbortSignal | undefined} */ #signal; /** @type {string} */ @@ -295,15 +297,43 @@ class Session { * @param {unknown} value */ #settle(index, ok, value) { + if (value === FAILURE) return `s.r(${index},1,${FAILED})`; try { return `s.r(${index},${ok ? 0 : 1},${this.#emit_value(value, false)})`; } catch (error) { - try { - this.#onerror?.(error, value); - } catch { - // a failing error handler must not break the session - } - return `s.r(${index},1,${FAILED})`; + // A transformed error that can't be serialized is not transformed again. + if (!ok) return `s.r(${index},1,${FAILED})`; + this.#fail(index, error); + return null; + } + } + + /** + * Queues the rejection of promise `index` with `error` (its rejection reason, or the + * error that kept its value from serializing) passed through `transformError`. + * Without one, when it fails, or when it opts out like a replacer (`undefined`, + * `null` or `false`), the client receives FAILED. + * @param {number} index + * @param {unknown} error + */ + #fail(index, error) { + let result; + try { + result = this.#transform_error?.(error); + } catch { + return this.#enqueue(index, false, FAILURE); + } + + if (result instanceof Promise) { + // As in #subscribe, the reaction only holds the detachable delivery cell. + const delivery = this.#delivery; + then.call( + result, + (value) => Session.#deliver(delivery, index, false, or_failure(value)), + () => Session.#deliver(delivery, index, false, FAILURE) + ); + } else { + this.#enqueue(index, false, or_failure(result)); } } @@ -317,7 +347,7 @@ class Session { this.#primitives.clear(); this.#queue.length = 0; this.#replacer = undefined; - this.#onerror = undefined; + this.#transform_error = undefined; this.#signal = undefined; this.#on_abort = noop; const wake = this.#wake; @@ -333,33 +363,44 @@ class Session { /** @returns {Promise>} */ async #step() { - while (!this.#done && this.#queue.length === 0) { - await new Promise((resolve) => (this.#wake = resolve)); - this.#wake = noop; - } + for (;;) { + while (!this.#done && this.#queue.length === 0) { + await new Promise((resolve) => (this.#wake = resolve)); + this.#wake = noop; + } - if (this.#abort) { - const { reason } = this.#abort; - this.#abort = null; - throw reason; - } + if (this.#abort) { + const { reason } = this.#abort; + this.#abort = null; + throw reason; + } - if (this.#done) return DONE; + if (this.#done) return DONE; + + const statements = []; + // Synchronously transformed failures are queued behind and join this block. + const queue = this.#queue; + for (let i = 0; i < queue.length; i += 1) { + const { index, ok, value } = queue[i]; + const statement = this.#settle(index, ok, value); + if (statement === null) continue; + this.#pending -= 1; + statements.push(statement); + } + queue.length = 0; - const statements = []; - for (const { index, ok, value } of this.#queue.splice(0)) { - this.#pending -= 1; - statements.push(this.#settle(index, ok, value)); - } + // Every outcome is waiting on an asynchronous transformError + if (statements.length === 0) continue; - const complete = this.#pending === 0; - if (complete) { - statements.push(`delete ${this.#table}`); - } + const complete = this.#pending === 0; + if (complete) { + statements.push(`delete ${this.#table}`); + } - const value = `((s)=>{${statements.join(';')}})(${this.#table});`; - if (complete) this.#finish(); - return { done: false, value }; + const value = `((s)=>{${statements.join(';')}})(${this.#table});`; + if (complete) this.#finish(); + return { done: false, value }; + } } /** @returns {Promise>} */ @@ -386,20 +427,34 @@ class Session { * @param {{ current: Session | null }} delivery */ static #subscribe(promise, index, delivery) { - /** - * @param {boolean} ok - * @param {unknown} value - */ - const deliver = (ok, value) => { - const session = delivery.current; - if (session) session.#enqueue(index, ok, value); - }; then.call( promise, - (value) => deliver(true, value), - (error) => deliver(false, error) + (value) => Session.#deliver(delivery, index, true, value), + (error) => { + const session = delivery.current; + if (session) session.#fail(index, error); + } ); } + + /** + * @param {{ current: Session | null }} delivery + * @param {number} index + * @param {boolean} ok + * @param {unknown} value + */ + static #deliver(delivery, index, ok, value) { + const session = delivery.current; + if (session) session.#enqueue(index, ok, value); + } +} + +/** + * `transformError` opts out with the same values as a replacer + * @param {unknown} value + */ +function or_failure(value) { + return value === undefined || value === null || value === false ? FAILURE : value; } /** @param {object} thing */ diff --git a/test/helpers/stream.js b/test/helpers/stream.js index 9b6dcab9..36521cec 100644 --- a/test/helpers/stream.js +++ b/test/helpers/stream.js @@ -245,7 +245,11 @@ export async function expect_roundtrip(make, { replacer, globals, orders = DEFAU return replacer?.(v, js); }; - const { head, tail } = unevalStream(value, counted, { id: 'roundtrip' }); + // Rejections are checked like resolutions, so send their reasons as they are. + const { head, tail } = unevalStream(value, counted, { + id: 'roundtrip', + transformError: (error) => error + }); heads.add(head); /** @type {string[]} */ diff --git a/test/stream-reference-growth.test.js b/test/stream-reference-growth.test.js index 43d7704d..9718703c 100644 --- a/test/stream-reference-growth.test.js +++ b/test/stream-reference-growth.test.js @@ -211,7 +211,13 @@ test('a failed emission rolls back forced slots, lazy aliases and promise indice const stream = unevalStream( { [key]: nodes, first: first.promise, second: second.promise }, undefined, - { id: 'rollback', onerror: (error) => errors.push(error) } + { + id: 'rollback', + transformError: (error) => { + errors.push(error); + throw error; + } + } ); const c = client(); const root = c.head(stream.head); diff --git a/test/stream-regressions.test.js b/test/stream-regressions.test.js index f6c9efaf..945d59f2 100644 --- a/test/stream-regressions.test.js +++ b/test/stream-regressions.test.js @@ -26,7 +26,7 @@ test('survives UTF-8 byte boundaries and separate or concatenated VM evaluation' resolved: resolved.promise, rejected: failed.promise }; - const result = unevalStream(value, undefined, { id }); + const result = unevalStream(value, undefined, { id, transformError: (error) => error }); expect(result.id).toBe(id); resolved.resolve({ shared, again: shared }); const sources = [result.head, /** @type {string} */ ((await result.tail.next()).value)]; diff --git a/test/stream-retention.test.js b/test/stream-retention.test.js index ed4e30ef..ca1fa87a 100644 --- a/test/stream-retention.test.js +++ b/test/stream-retention.test.js @@ -127,10 +127,10 @@ async function callback_retention({ assert, vm, turn, unevalStream, data: callba graph.visits += 1; return undefined; }; - const onerror = () => { + const transformError = () => { reports += graph.nodes.length; }; - const options = callback === 'onerror' ? { onerror } : {}; + const options = callback === 'transformError' ? { transformError } : {}; // The graph is callback-only data, never an input node or cached replacement. const result = unevalStream( callback === 'replacer' ? { value: 42 } : Promise.resolve(42), @@ -148,7 +148,7 @@ async function callback_retention({ assert, vm, turn, unevalStream, data: callba for await (const block of fixture.result.tail) vm.runInContext(block, context); if (callback === 'replacer') assert.equal(root.value, 42); else assert.equal(await root, 42); - assert.equal(reports, 0, 'supported outcomes must not invoke onerror'); + assert.equal(reports, 0, 'supported outcomes must not invoke transformError'); // No reporter is active; a retained terminal result must release its callbacks. // As above, do not deref until all GCs have run in separate jobs. @@ -163,7 +163,7 @@ async function callback_retention({ assert, vm, turn, unevalStream, data: callba ); } -test.each(['replacer', 'onerror'])( +test.each(['replacer', 'transformError'])( 'a retained terminal result releases graph-capturing %s callbacks', (callback) => in_subprocess(callback_retention, { expose_gc: true, data: callback }) ); diff --git a/test/stream.test.js b/test/stream.test.js index 612407fc..05735151 100644 --- a/test/stream.test.js +++ b/test/stream.test.js @@ -269,7 +269,7 @@ test('calls the replacer once per identity, even after a failed outcome', async const x = new X(); /** @type {unknown[]} */ const seen = []; - const onerror = vi.fn(); + const transformError = vi.fn(() => undefined); const a = Promise.withResolvers(); const b = Promise.withResolvers(); @@ -279,7 +279,7 @@ test('calls the replacer once per identity, even after a failed outcome', async seen.push(value); if (value instanceof X) return js`{value:${value.value}}`; }, - { onerror } + { transformError } ); const c = client(); @@ -293,7 +293,7 @@ test('calls the replacer once per identity, even after a failed outcome', async expect(seen.filter((v) => v === x)).toHaveLength(1); await expect(root.p1).rejects.toThrow('devalue: failed to serialize asynchronous value'); expect(await root.p2).toEqual({ value: 42 }); - expect(onerror).toHaveBeenCalledOnce(); + expect(transformError).toHaveBeenCalledOnce(); expect((await stream.tail.next()).done).toBe(true); }); @@ -423,3 +423,110 @@ test('concurrent next() calls settle in order', async () => { expect(await root.a).toBe('a'); expect(await root.b).toBe('b'); }); + +test('rejection reasons never reach the client by default', async () => { + const stream = unevalStream({ + error: Promise.reject(new Error('db password=hunter2')), + object: Promise.reject({ secret: 'internal detail' }), + string: Promise.reject('stack trace') + }); + let source = stream.head; + const c = client(); + const root = c.head(stream.head); + for await (const block of stream.tail) { + source += block; + c.block(block); + } + for (const key of ['error', 'object', 'string']) { + await expect(root[key]).rejects.toThrow('devalue: failed to serialize asynchronous value'); + } + expect(source).not.toMatch(/hunter2|secret|internal|stack/); +}); + +test.each(['sync', 'async'])('transformError (%s) shapes rejections and failures', async (mode) => { + const shared = { shared: true }; + /** @param {unknown} error */ + const shape = (error) => ({ + message: error instanceof Error ? error.message : String(error), + shared + }); + const stream = unevalStream( + { + shared, + rejected: Promise.reject(new Error('nope')), + unserializable: Promise.resolve({ fn() {} }), + resolved: Promise.resolve(1) + }, + undefined, + { + transformError: (error) => (mode === 'sync' ? shape(error) : Promise.resolve(shape(error))) + } + ); + const c = client(); + const root = c.head(stream.head); + for await (const block of stream.tail) c.block(block); + + await expect(root.rejected).rejects.toEqual({ message: 'nope', shared: root.shared }); + const reason = await root.unserializable.catch((/** @type {any} */ e) => e); + expect(reason.message).toMatch(/Cannot stringify/); + expect(reason.shared).toBe(root.shared); + expect(await root.resolved).toBe(1); + expect(c.context.__d).toEqual({}); +}); + +test.each([ + [ + 'throws', + () => { + throw new Error('transform failed'); + } + ], + ['rejects', () => Promise.reject(new Error('transform failed'))], + ['returns something unserializable', () => ({ fn() {} })], + ['resolves to something unserializable', () => Promise.resolve(Symbol())], + ['returns undefined', () => undefined], + ['returns null', () => null], + ['returns false', () => false], + ['resolves to null', () => Promise.resolve(null)] +])('falls back to the generic error when transformError %s', async (_, transformError) => { + const stream = unevalStream({ p: Promise.reject(new Error('secret')) }, undefined, { + transformError + }); + const c = client(); + const root = c.head(stream.head); + let source = ''; + for await (const block of stream.tail) { + source += block; + c.block(block); + } + await expect(root.p).rejects.toThrow('devalue: failed to serialize asynchronous value'); + expect(source).not.toContain('secret'); + expect(c.context.__d).toEqual({}); +}); + +test('a pending async transformError does not hold back other settlements', async () => { + const transform = Promise.withResolvers(); + const stream = unevalStream( + { failed: Promise.reject(new Error('x')), ok: Promise.resolve(1) }, + undefined, + { transformError: () => transform.promise } + ); + const c = client(); + const root = c.head(stream.head); + c.block(/** @type {string} */ ((await stream.tail.next()).value)); + expect(await root.ok).toBe(1); + transform.resolve('shaped'); + c.block(/** @type {string} */ ((await stream.tail.next()).value)); + await expect(root.failed).rejects.toBe('shaped'); + expect((await stream.tail.next()).done).toBe(true); +}); + +test('return() during an async transformError ends the stream', async () => { + const stream = unevalStream({ p: Promise.reject(new Error('x')) }, undefined, { + transformError: () => new Promise(() => {}) + }); + const next = stream.tail.next(); + await new Promise((resolve) => setTimeout(resolve, 0)); + await stream.tail.return(); + expect(await next).toEqual({ done: true, value: undefined }); +}); From 0dddfcdf4945b823f771d4f45dd3eea2c6832bd3 Mon Sep 17 00:00:00 2001 From: Elliott Johnson Date: Tue, 13 Oct 2026 16:16:08 -0600 Subject: [PATCH 5/6] feat: reject pending client promises when unevalStream is aborted --- README.md | 2 +- src/types.ts | 7 +-- src/uneval-stream.js | 30 +++++++----- test/stream-retention.test.js | 31 ++++++------ test/stream.test.js | 89 +++++++++++++++++++++++++++++++++-- 5 files changed, 125 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 357a0327..a7ef8d9d 100644 --- a/README.md +++ b/README.md @@ -114,7 +114,7 @@ Evaluate `head` once, then each block once, in order, in the same realm. Concate | ---------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `id` | Session key (returned as `id`). Random by default; pass a stable value when `head` must be byte-identical across runs. | | `scope` | Trusted, assignable expression holding the client session table. Defaults to `globalThis.__d`. | -| `signal` | An `AbortSignal` that stops the tail. Pending and later `next()` calls reject once with `signal.reason`; if the signal is already aborted, or aborts during the head emission, `unevalStream` throws `signal.reason`. | +| `signal` | An `AbortSignal` that stops the tail. The next `next()` yields one final block that rejects every promise still pending on the client with a generic `Error` and deletes the client session, then the tail is done. If the signal is already aborted, or aborts during the head emission, `unevalStream` throws `signal.reason`. | | `transformError` | `(error) => unknown`, called with a rejection reason, or with the error that kept a resolved value from serializing. Its result (or what its returned promise resolves to) is what the client promise rejects with. As with a replacer, returning `undefined`, `null` or `false` keeps the default: a generic `Error`, so server errors never reach the client unless you opt in. Throwing, rejecting, or returning something that can't be serialized also falls back to it. | #### Constraints diff --git a/src/types.ts b/src/types.ts index 5e22fa1c..7b30c4f5 100644 --- a/src/types.ts +++ b/src/types.ts @@ -71,9 +71,10 @@ export interface UnevalStreamOptions { /** Trusted assignable expression holding the client session table. Default `globalThis.__d`. */ scope?: string; /** - * Stops the tail; pending and later `next()` calls reject once with `signal.reason`. - * If the signal is already aborted, or aborts during the head emission, `unevalStream` - * throws `signal.reason`. + * Stops the tail: the next `next()` yields a final block that rejects every promise + * still pending on the client with a generic Error and deletes the client session, + * then the tail is done. If the signal is already aborted, or aborts during the head + * emission, `unevalStream` throws `signal.reason`. */ signal?: AbortSignal; /** diff --git a/src/uneval-stream.js b/src/uneval-stream.js index 3525889e..dd7dd9db 100644 --- a/src/uneval-stream.js +++ b/src/uneval-stream.js @@ -7,6 +7,7 @@ import { call_replacer, get_type, stringify_string } from './utils.js'; const RESERVED = ['s', 'n']; const FAILED = 'new Error("devalue: failed to serialize asynchronous value")'; +const ABORTED = 'new Error("devalue: stream aborted")'; // Stands in for an outcome that is sent as FAILED rather than serialized const FAILURE = Symbol(); const DONE = Object.freeze({ done: /** @type {true} */ (true), value: undefined }); @@ -171,7 +172,11 @@ class Session { */ #wake = noop; - /** @type {{ reason: unknown } | null} */ + /** + * Set by an abort and cleared once the final block that rejects the remaining + * client promises has been emitted + * @type {{ reason: unknown } | null} + */ #abort = null; /** @@ -369,13 +374,7 @@ class Session { this.#wake = noop; } - if (this.#abort) { - const { reason } = this.#abort; - this.#abort = null; - throw reason; - } - - if (this.#done) return DONE; + if (this.#done && !this.#abort) return DONE; const statements = []; // Synchronously transformed failures are queued behind and join this block. @@ -389,10 +388,19 @@ class Session { } queue.length = 0; - // Every outcome is waiting on an asynchronous transformError - if (statements.length === 0) continue; + // Checked after settling, since a replacer can abort while this block is emitted. + // Rejects everything still pending on the client, including any promises this + // block itself created after the abort. + const aborted = this.#abort !== null; + if (aborted) { + this.#abort = null; + statements.push(`s.p.forEach((p)=>p[1](${ABORTED}))`); + } else if (statements.length === 0) { + // Every outcome is waiting on an asynchronous transformError + continue; + } - const complete = this.#pending === 0; + const complete = aborted || this.#pending === 0; if (complete) { statements.push(`delete ${this.#table}`); } diff --git a/test/stream-retention.test.js b/test/stream-retention.test.js index ca1fa87a..47a96962 100644 --- a/test/stream-retention.test.js +++ b/test/stream-retention.test.js @@ -36,9 +36,10 @@ async function retention({ assert, turn, unevalStream, data: mode }) { assert.deepEqual(await fixture.result.tail.return(), { done: true, value: undefined }); assert.deepEqual(await next, { done: true, value: undefined }); } else { - const reason = new Error('abort stream'); - fixture.controller.abort(reason); - await assert.rejects(next, (error) => error === reason); + fixture.controller.abort(new Error('abort stream')); + const last = await next; + assert.equal(last.done, false); + assert.match(last.value, /devalue: stream aborted/); } if (!mode.endsWith('retained')) fixture.result = null; fixture.controller = null; @@ -219,17 +220,18 @@ test.each(['return', 'abort'])( } const controller = new AbortController(); const fixture = create_result(controller.signal); - // Use a separate pending stream to check ordered wakeup and exact reason, - // while the first stream has an undelivered graph-bearing outcome. + // Use a separate pending stream to check ordered wakeup, while the first + // stream has an undelivered graph-bearing outcome. const waiting = unevalStream(fixture.pending.promise, undefined, { signal: controller.signal }); const readers = [waiting.tail.next(), waiting.tail.next(), waiting.tail.next()]; await turn(); if (mode === 'abort') { - const reason = { message: 'exact abort reason' }; - controller.abort(reason); - await assert.rejects(readers[0], (error) => error === reason); + controller.abort(); + const last = await readers[0]; + assert.equal(last.done, false); + assert.match(last.value, /devalue: stream aborted/); } else { await fixture.result.tail.return(); await waiting.tail.return(); @@ -239,10 +241,10 @@ test.each(['return', 'abort'])( assert.deepEqual(await reader, { done: true, value: undefined }); } if (mode === 'abort') { - await assert.rejects( - fixture.result.tail.next(), - (error) => error === controller.signal.reason - ); + const last = await fixture.result.tail.next(); + assert.equal(last.done, false); + assert.match(last.value, /devalue: stream aborted/); + assert.doesNotMatch(last.value, /node/, 'the queued outcome is dropped'); } for (let i = 0; i < 20; i += 1) { await turn(); @@ -285,9 +287,10 @@ test.each(['return', 'abort'])( return { result, pending, weak: new WeakRef(graph) }; } const fixture = create_result(); - await fixture.result.tail.next(); + const block = await fixture.result.tail.next(); if (mode === 'abort') { - await assert.rejects(fixture.result.tail.next(), (error) => error === 0); + // The block being emitted when the abort happened carries the final rejections + assert.match(block.value, /devalue: stream aborted/); } assert.deepEqual(await fixture.result.tail.next(), { done: true, value: undefined }); for (let i = 0; i < 20; i += 1) { diff --git a/test/stream.test.js b/test/stream.test.js index 05735151..0ae0fa93 100644 --- a/test/stream.test.js +++ b/test/stream.test.js @@ -361,15 +361,94 @@ test('return() completes a pending next()', async () => { expect(await next).toEqual({ done: true, value: undefined }); }); -test('abort rejects a pending next() with the exact reason', async () => { +test('abort emits a final block that rejects pending client promises', async () => { const controller = new AbortController(); - const stream = unevalStream({ p: new Promise(() => {}) }, undefined, { + const settled = Promise.withResolvers(); + const stream = unevalStream( + { a: settled.promise, b: new Promise(() => {}), c: new Promise(() => {}) }, + undefined, + { signal: controller.signal } + ); + const c = client(); + const root = c.head(stream.head); + + settled.resolve(1); + const first = await stream.tail.next(); + c.block(/** @type {string} */ (first.value)); + expect(await root.a).toBe(1); + + const next = stream.tail.next(); + const later = stream.tail.next(); + controller.abort(new Error('server-only reason')); + + const last = await next; + expect(last.done).toBe(false); + c.block(/** @type {string} */ (last.value)); + for (const p of [root.b, root.c]) { + await expect(p).rejects.toThrow('devalue: stream aborted'); + } + expect(last.value).not.toContain('server-only reason'); + expect(Object.keys(c.context.__d)).toEqual([]); + expect(await later).toEqual({ done: true, value: undefined }); + expect(await stream.tail.next()).toEqual({ done: true, value: undefined }); +}); + +test('abort with no pending next() emits the final block on the next read', async () => { + const controller = new AbortController(); + const queued = Promise.withResolvers(); + const stream = unevalStream({ q: queued.promise, p: new Promise(() => {}) }, undefined, { signal: controller.signal }); - const next = stream.tail.next(); - controller.abort(0); + const c = client(); + const root = c.head(stream.head); + + // An outcome queued but not yet emitted is rejected along with the rest. + queued.resolve(1); + await new Promise((r) => setTimeout(r, 0)); + controller.abort(); + + const last = await stream.tail.next(); + c.block(/** @type {string} */ (last.value)); + await expect(root.q).rejects.toThrow('devalue: stream aborted'); + await expect(root.p).rejects.toThrow('devalue: stream aborted'); + expect(Object.keys(c.context.__d)).toEqual([]); + expect(await stream.tail.next()).toEqual({ done: true, value: undefined }); +}); + +test('abort after completion or return() emits nothing', async () => { + for (const mode of ['complete', 'return']) { + const controller = new AbortController(); + const stream = unevalStream({ p: Promise.resolve(1) }, undefined, { + signal: controller.signal + }); + if (mode === 'complete') expect((await stream.tail.next()).done).toBe(false); + else await stream.tail.return(); + controller.abort(); + expect(await stream.tail.next()).toEqual({ done: true, value: undefined }); + } +}); + +test('abort during a tail block rejects the promises that block creates', async () => { + const controller = new AbortController(); + const source = Promise.withResolvers(); + const graph = { nested: new Promise(() => {}) }; + const stream = unevalStream( + source.promise, + (thing) => { + if (thing === graph) controller.abort(); + }, + { signal: controller.signal } + ); + const c = client(); + const root = c.head(stream.head); + + source.resolve(graph); + const block = await stream.tail.next(); + c.block(/** @type {string} */ (block.value)); - await expect(next).rejects.toBe(0); + const revived = await root; + await expect(revived.nested).rejects.toThrow('devalue: stream aborted'); + expect(Object.keys(c.context.__d)).toEqual([]); expect(await stream.tail.next()).toEqual({ done: true, value: undefined }); }); From 929fabc5cd17e060bb625976e0a62a6578a76cc5 Mon Sep 17 00:00:00 2001 From: Elliott Johnson Date: Tue, 13 Oct 2026 16:17:53 -0600 Subject: [PATCH 6/6] chore: use explicit .js extension in unevalStream's type imports --- src/uneval-stream.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/uneval-stream.js b/src/uneval-stream.js index dd7dd9db..c0496a19 100644 --- a/src/uneval-stream.js +++ b/src/uneval-stream.js @@ -1,5 +1,5 @@ /** - * @import { UnevalReplacer, UnevalStreamOptions, UnevalStreamResult } from './types'; + * @import { UnevalReplacer, UnevalStreamOptions, UnevalStreamResult } from './types.js'; */ import { js, JavaScriptSource } from './javascript-source.js'; import { analyze } from './uneval.js';