diff --git a/.github/workflows/common.yml b/.github/workflows/common.yml index c64eed93be..0d3c1d5820 100644 --- a/.github/workflows/common.yml +++ b/.github/workflows/common.yml @@ -35,4 +35,4 @@ jobs: target_file: 'packages/shared/common/dist/esm/index.mjs' package_name: '@launchdarkly/js-sdk-common' pr_number: ${{ github.event.number }} - size_limit: 30000 + size_limit: 31500 diff --git a/packages/sdk/shopify-oxygen/src/platform/OxygenRequests.ts b/packages/sdk/shopify-oxygen/src/platform/OxygenRequests.ts index d46b08cabb..58a65d2392 100644 --- a/packages/sdk/shopify-oxygen/src/platform/OxygenRequests.ts +++ b/packages/sdk/shopify-oxygen/src/platform/OxygenRequests.ts @@ -1,4 +1,4 @@ -import { NullEventSource } from '@launchdarkly/js-server-sdk-common'; +import { internal, NullEventSource } from '@launchdarkly/js-server-sdk-common'; import type { EventSource, EventSourceCapabilities, @@ -62,10 +62,10 @@ export default class OxygenRequests implements platform.Requests { // - Cache API not available // - Cache is not enabled per initialization options // - Not a GET request (for now, we mostly interested in caching the feature poll request) - return fetch(url, options); + return fetch(url, internal.toRequestInit(options)); } - const request = new Request(url, finalOptions); + const request = new Request(url, internal.toRequestInit(finalOptions)); const cachedResponse = await cache.match(request); @@ -81,7 +81,7 @@ export default class OxygenRequests implements platform.Requests { request: Request, cache: Cache, ): Promise { - const response = await fetch(url, options); + const response = await fetch(url, internal.toRequestInit(options)); // Only cache successful GET requests if (cache && response.ok && (!options.method || options.method === 'GET')) { diff --git a/packages/shared/common/__tests__/internal/fetch/StreamingResponse.test.ts b/packages/shared/common/__tests__/internal/fetch/StreamingResponse.test.ts new file mode 100644 index 0000000000..f7dffaf28c --- /dev/null +++ b/packages/shared/common/__tests__/internal/fetch/StreamingResponse.test.ts @@ -0,0 +1,337 @@ +import { PassThrough, Readable } from 'node:stream'; + +import { + createStreamingResponse, + RawHeaders, + StreamingResponseSource, +} from '../../../src/internal/fetch/StreamingResponse'; + +function utf8(text: string): Uint8Array { + return new TextEncoder().encode(text); +} + +function makeSource( + parts: Uint8Array[], + meta: { headers?: RawHeaders; statusCode?: number; statusMessage?: string } = {}, +): StreamingResponseSource { + let index = 0; + return { + headers: meta.headers ?? {}, + statusCode: meta.statusCode, + statusMessage: meta.statusMessage, + [Symbol.asyncIterator]: () => ({ + next: async () => { + if (index < parts.length) { + index += 1; + return { done: false, value: parts[index - 1] }; + } + return { done: true }; + }, + }), + }; +} + +function makeTextSource( + parts: string[], + meta: { headers?: RawHeaders; statusCode?: number; statusMessage?: string } = {}, +): StreamingResponseSource { + return makeSource(parts.map(utf8), meta); +} + +it('exposes the status code and status message', () => { + const res = createStreamingResponse( + makeTextSource([], { statusCode: 301, statusMessage: 'Moved Permanently' }), + ); + expect(res.status).toBe(301); + expect(res.statusText).toBe('Moved Permanently'); +}); + +it('defaults the status to 0 and the status message to an empty string when absent', () => { + const res = createStreamingResponse(makeTextSource([])); + expect(res.status).toBe(0); + expect(res.statusText).toBe(''); +}); + +it('wraps the raw headers', () => { + const res = createStreamingResponse( + makeTextSource([], { + headers: { + 'content-type': 'text/event-stream', + 'set-cookie': ['a=1', 'b=2'], + empty: undefined, + }, + statusCode: 200, + }), + ); + expect(res.headers.get('content-type')).toBe('text/event-stream'); + expect(res.headers.get('set-cookie')).toBe('a=1, b=2'); + expect(res.headers.get('missing')).toBeNull(); + expect(res.headers.has('content-type')).toBe(true); + expect(res.headers.has('missing')).toBe(false); + expect(Array.from(res.headers.entries())).toEqual([ + ['content-type', 'text/event-stream'], + ['set-cookie', 'a=1, b=2'], + ]); + const visited: [string, string][] = []; + res.headers.forEach?.((value, key) => visited.push([key, value])); + expect(visited).toEqual([ + ['content-type', 'text/event-stream'], + ['set-cookie', 'a=1, b=2'], + ]); +}); + +it('matches header names without case sensitivity', () => { + const res = createStreamingResponse( + makeTextSource([], { headers: { 'Content-Type': 'text/event-stream' }, statusCode: 200 }), + ); + expect(res.headers.get('content-type')).toBe('text/event-stream'); + expect(res.headers.get('CONTENT-TYPE')).toBe('text/event-stream'); + expect(res.headers.has('Content-Type')).toBe(true); + expect(Array.from(res.headers.keys())).toEqual(['content-type']); +}); + +it('joins the values of header names that differ only by case', () => { + const res = createStreamingResponse( + makeTextSource([], { + headers: { 'Content-Type': 'text/html', 'content-type': 'text/event-stream' }, + statusCode: 200, + }), + ); + expect(res.headers.get('content-type')).toBe('text/html, text/event-stream'); +}); + +it('reports a header without a value as absent', () => { + const res = createStreamingResponse( + makeTextSource([], { headers: { empty: undefined }, statusCode: 200 }), + ); + expect(res.headers.has('empty')).toBe(false); + expect(res.headers.get('empty')).toBeNull(); + expect(Array.from(res.headers.keys())).toEqual([]); +}); + +it('does not report names inherited from the object prototype as headers', () => { + const res = createStreamingResponse(makeTextSource([], { statusCode: 200 })); + expect(res.headers.get('constructor')).toBeNull(); + expect(res.headers.get('__proto__')).toBeNull(); + expect(res.headers.has('toString')).toBe(false); +}); + +it('hands out one chunk per read and then reports done', async () => { + const res = createStreamingResponse(makeTextSource(['first', 'second'])); + const reader = res.body.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + expect(first.value).toEqual(utf8('first')); + const second = await reader.read(); + expect(second.value).toEqual(utf8('second')); + const end = await reader.read(); + expect(end.done).toBe(true); +}); + +it('reads the remaining stream as text', async () => { + const res = createStreamingResponse(makeTextSource(['hello ', 'world'])); + await expect(res.text()).resolves.toBe('hello world'); +}); + +it('decodes a multi-byte character split across chunks', async () => { + const bytes = utf8('héllo €!'); + const res = createStreamingResponse( + makeSource([bytes.slice(0, 2), bytes.slice(2, 7), bytes.slice(7)]), + ); + await expect(res.text()).resolves.toBe('héllo €!'); +}); + +it('parses the remaining stream as JSON', async () => { + const res = createStreamingResponse(makeTextSource(['{"a":', '1}'])); + await expect(res.json()).resolves.toEqual({ a: 1 }); +}); + +it('decodes ArrayBuffer and DataView chunks in text', async () => { + const bytes = utf8('buffer view'); + const asBuffer = bytes.slice(0, 7).buffer as unknown as Uint8Array; + const tail = bytes.slice(7); + const asView = new DataView( + tail.buffer, + tail.byteOffset, + tail.byteLength, + ) as unknown as Uint8Array; + const res = createStreamingResponse(makeSource([asBuffer, asView])); + await expect(res.text()).resolves.toBe('buffer view'); +}); + +it('rejects a string chunk in text instead of corrupting the result', async () => { + const res = createStreamingResponse(makeSource(['data: hi\n\n' as unknown as Uint8Array])); + const err = await res.text().then( + () => undefined, + (e) => e, + ); + expect(err).toBeInstanceOf(TypeError); + expect(String(err)).toContain('BufferSource'); + // The chunk contents must never leak into the error message. + expect(String(err)).not.toContain('data: hi'); +}); + +it('destroys the source when text rejects a bad chunk', async () => { + // Readable.from on strings yields string chunks, which the decoder rejects. The + // rejection must release the connection, because text() owns the body. + const readable = Readable.from(['data: hi\n\n']); + const source: StreamingResponseSource = Object.assign(readable, { headers: {} }); + const res = createStreamingResponse(source); + await expect(res.text()).rejects.toThrow('BufferSource'); + expect(readable.destroyed).toBe(true); +}); + +it('keeps the fixed message when the source destroy throws during a bad-chunk release', async () => { + // The release calls inside cancel are guarded. A hostile destroy must not replace the + // fixed BufferSource message, and the iterator return must still run. + const destroySpy = jest.fn(() => { + throw new Error('destroy boom'); + }); + const returnSpy = jest.fn(async () => ({ done: true })); + const source: StreamingResponseSource = { + headers: {}, + destroy: destroySpy, + [Symbol.asyncIterator]: () => ({ + next: async () => ({ done: false, value: 'data: hi\n\n' as unknown as Uint8Array }), + return: returnSpy, + }), + }; + const res = createStreamingResponse(source); + await expect(res.text()).rejects.toThrow('BufferSource'); + expect(destroySpy).toHaveBeenCalledTimes(1); + expect(returnSpy).toHaveBeenCalledTimes(1); +}); + +it('throws a named error when the source has no async iterator', () => { + const source = { headers: {} } as unknown as StreamingResponseSource; + expect(() => createStreamingResponse(source)).toThrow( + 'the streaming response source does not provide an async iterator', + ); +}); + +it('releases the source when a read fails during text', async () => { + const destroySpy = jest.fn(); + const returnSpy = jest.fn(async () => ({ done: true })); + const source: StreamingResponseSource = { + headers: {}, + destroy: destroySpy, + [Symbol.asyncIterator]: () => ({ + next: async () => { + throw new Error('connection reset'); + }, + return: returnSpy, + }), + }; + const res = createStreamingResponse(source); + await expect(res.text()).rejects.toThrow('connection reset'); + expect(destroySpy).toHaveBeenCalledTimes(1); + expect(returnSpy).toHaveBeenCalledTimes(1); +}); + +it('rejects a pending read when the stream errors', async () => { + const erroring: StreamingResponseSource = { + headers: {}, + [Symbol.asyncIterator]: () => ({ + next: async () => { + throw new Error('boom'); + }, + }), + }; + const res = createStreamingResponse(erroring); + await expect(res.body.getReader().read()).rejects.toThrow('boom'); +}); + +it('releases the stream through the iterator return method on cancel', async () => { + const returnSpy = jest.fn(async () => ({ done: true })); + const source: StreamingResponseSource = { + headers: {}, + [Symbol.asyncIterator]: () => ({ + next: async () => ({ done: false, value: utf8('chunk') }), + return: returnSpy, + }), + }; + const res = createStreamingResponse(source); + await res.body.getReader().cancel?.(); + await res.body.cancel?.(); + expect(returnSpy).toHaveBeenCalledTimes(2); +}); + +it('tolerates cancel when the iterator has no return method', async () => { + const res = createStreamingResponse(makeTextSource(['chunk'])); + await expect(Promise.resolve(res.body.cancel?.())).resolves.toBeUndefined(); +}); + +it('still releases the iterator when destroy throws', async () => { + const returnSpy = jest.fn(async () => ({ done: true })); + const source: StreamingResponseSource = { + headers: {}, + destroy: () => { + throw new Error('destroy boom'); + }, + [Symbol.asyncIterator]: () => ({ + next: async () => ({ done: false, value: utf8('chunk') }), + return: returnSpy, + }), + }; + const res = createStreamingResponse(source); + await expect(Promise.resolve(res.body.cancel?.())).resolves.toBeUndefined(); + expect(returnSpy).toHaveBeenCalledTimes(1); +}); + +it('tolerates an iterator whose return resolves synchronously', async () => { + const destroySpy = jest.fn(); + const source: StreamingResponseSource = { + headers: {}, + destroy: destroySpy, + [Symbol.asyncIterator]: () => ({ + next: async () => ({ done: false, value: utf8('chunk') }), + // @ts-ignore deliberately violates the declared Promise-returning shape. + return: () => ({ done: true }), + }), + }; + const res = createStreamingResponse(source); + await expect(Promise.resolve(res.body.cancel?.())).resolves.toBeUndefined(); + expect(destroySpy).toHaveBeenCalledTimes(1); +}); + +it('accepts a real Node readable stream as the source', async () => { + const readable = Readable.from([Buffer.from('data: '), Buffer.from('hello\n\n')]); + const source: StreamingResponseSource = Object.assign(readable, { + headers: { 'content-type': 'text/event-stream' }, + statusCode: 200, + statusMessage: 'OK', + }); + const res = createStreamingResponse(source); + expect(res.status).toBe(200); + expect(res.headers.get('content-type')).toBe('text/event-stream'); + await expect(res.text()).resolves.toBe('data: hello\n\n'); +}); + +it('destroys a real Node readable stream when the caller cancels mid-stream', async () => { + const readable = Readable.from([Buffer.from('first'), Buffer.from('endless')]); + const source: StreamingResponseSource = Object.assign(readable, { headers: {} }); + const res = createStreamingResponse(source); + const reader = res.body.getReader(); + await reader.read(); + await reader.cancel?.(); + expect(readable.destroyed).toBe(true); +}); + +it('destroys a real Node readable stream when the caller cancels with a read in progress', async () => { + const stalled = new PassThrough(); + const source: StreamingResponseSource = Object.assign(stalled, { headers: {} }); + const res = createStreamingResponse(source); + const reader = res.body.getReader(); + const pending = reader.read(); + await reader.cancel?.(); + expect(stalled.destroyed).toBe(true); + await expect(pending).rejects.toThrow('Premature close'); +}); + +it('destroys a real Node readable stream when the caller cancels before reading', async () => { + const readable = Readable.from([Buffer.from('never read')]); + const source: StreamingResponseSource = Object.assign(readable, { headers: {} }); + const res = createStreamingResponse(source); + await res.body.cancel?.(); + expect(readable.destroyed).toBe(true); +}); diff --git a/packages/shared/common/src/api/platform/Requests.ts b/packages/shared/common/src/api/platform/Requests.ts index 22924e5a02..13d97cb792 100644 --- a/packages/shared/common/src/api/platform/Requests.ts +++ b/packages/shared/common/src/api/platform/Requests.ts @@ -51,6 +51,43 @@ export interface Headers { * @param name The name of the header to check. */ has(name: string): boolean; + + /** + * Executes the callback once for each header, with the value first. Optional so existing + * implementations remain valid. The streaming client reads response headers through this + * method, or through {@link Headers.entries} when this method is absent. + * + * https://developer.mozilla.org/en-US/docs/Web/API/Headers/forEach + */ + forEach?(callback: (value: string, key: string) => void): void; +} + +/** + * A reader over a response body stream. + */ +export interface BodyReader { + /** + * Resolves with the next chunk, or with `done: true` when the server ends the stream. + * Rejects when the connection terminates before the server ends the stream. + */ + read(): Promise<{ done: boolean; value?: Uint8Array }>; + + /** + * Releases the reader and its connection. + */ + cancel?(): Promise | void; +} + +/** + * A response body stream. + */ +export interface ResponseBody { + getReader(): BodyReader; + + /** + * Releases the body. + */ + cancel?(): Promise | void; } /** @@ -60,6 +97,22 @@ export interface Response { headers: Headers; status: number; + /** + * The HTTP status message of the response. Optional so existing implementations remain + * valid. + * + * https://developer.mozilla.org/en-US/docs/Web/API/Response/statusText + */ + statusText?: string; + + /** + * The response body stream. An implementation provides it for a request made with + * {@link Options.streaming}. It can also provide it for other requests. + * + * https://developer.mozilla.org/en-US/docs/Web/API/Response/body + */ + body?: ResponseBody | null; + /** * Read the response and provide it as a string. */ @@ -71,6 +124,22 @@ export interface Response { json(): Promise; } +/** + * The platform's own `AbortSignal` type when the platform declares one as a global + * variable. An implementation that passes {@link Options} to a native `fetch` then + * type-checks against its platform's `RequestInit`. A platform that declares + * `AbortSignal` as a lexical class resolves to the fallback instead. The fallback is + * the minimal structural subset an implementation consumes. It keeps these declarations + * valid on a platform without the global. + */ +export type AbortSignalLike = typeof globalThis extends { AbortSignal: { prototype: infer T } } + ? T + : { + readonly aborted: boolean; + addEventListener(type: 'abort', listener: () => void, options?: { once?: boolean }): void; + removeEventListener(type: 'abort', listener: () => void): void; + }; + export interface Options { headers?: Record; method?: string; @@ -86,6 +155,25 @@ export interface Options { * https://developer.mozilla.org/en-US/docs/Web/API/RequestInit#keepalive */ keepalive?: boolean; + + /** + * An abort signal for the request. When the signal aborts, the implementation stops the + * request and rejects any pending read. Platform support for this field is best effort. + * + * https://developer.mozilla.org/en-US/docs/Web/API/RequestInit#signal + */ + signal?: AbortSignalLike; + + /** + * True for a streaming request. The implementation does not buffer the response body, + * it does not request compressed content, and it ignores {@link Options.timeout}. The + * response exposes its body through + * {@link Response.body}. An SDK must use a fetch as a streaming transport only when the + * implementation honors this field. A streaming implementation must provide at least one + * release path. It honors {@link Options.signal}, or it provides the body cancel. It + * preferably provides both. + */ + streaming?: boolean; } export interface EventSourceCapabilities { diff --git a/packages/shared/common/src/internal/fetch/StreamingResponse.ts b/packages/shared/common/src/internal/fetch/StreamingResponse.ts new file mode 100644 index 0000000000..8bbd4f3ec3 --- /dev/null +++ b/packages/shared/common/src/internal/fetch/StreamingResponse.ts @@ -0,0 +1,196 @@ +import type { Headers, Response, ResponseBody } from '../../api/platform/Requests'; + +// Each platform that constructs a streaming response provides the TextDecoder global. This +// package compiles without the global type libraries, so this scoped declaration stands in +// for it. +declare const TextDecoder: new () => { + decode(input?: Uint8Array, options?: { stream?: boolean }): string; +}; + +/** + * The shape of raw HTTP headers: names mapped to a value, a list of values, or no value. + * A Node `IncomingHttpHeaders` object satisfies this shape. + */ +export type RawHeaders = Record; + +/** + * The source of response chunks. A platform provides the iterator from its own stream type. + * For example, a Node `IncomingMessage` provides a compatible iterator through its + * `Symbol.asyncIterator` method. + */ +export interface ResponseChunkIterator { + /** + * Resolves with the next chunk, or with `done: true` when the stream ends. Rejects when + * the connection terminates before the server ends the stream. A chunk delivered together + * with `done: true` is discarded, so the iterator must deliver the last chunk on its own. + */ + next(): Promise<{ done?: boolean; value?: Uint8Array }>; + + /** + * Releases the iterator. A Node `Readable` iterator queues this call behind a pending + * read, so this call alone does not release an idle stream. + */ + return?(): Promise<{ done?: boolean; value?: Uint8Array }>; +} + +/** + * The key of the chunk iterator on a streaming response source. A consumer's compile-time + * lib may not declare `Symbol.asyncIterator`. This conditional resolves to that symbol when + * the lib declares it, and to no key at all otherwise. That keeps these declarations valid + * for every consumer, like `AbortSignalLike` does for the abort signal. Without the symbol, + * the source type cannot require the iterator, and a source without a stream fails at run + * time instead of compile time. + */ +type AsyncIteratorKey = typeof Symbol extends { readonly asyncIterator: infer S } + ? S extends keyof any + ? S + : never + : never; + +/** + * The source of a streaming response: the metadata and the chunk stream. A Node + * `IncomingMessage` satisfies this shape, so a platform can pass its message directly. + */ +export type StreamingResponseSource = { + headers: RawHeaders; + statusCode?: number; + statusMessage?: string; + + /** + * Destroys the underlying stream. `cancel` uses it to release the connection even while + * a read is in progress. A Node `Readable` provides it. A source without it must release + * through the iterator `return` method; a source whose `return` queues behind a pending + * read, like a Node readable iterator, must provide `destroy` or `cancel` cannot release + * it mid-read. + */ + destroy?(): void; +} & { [K in AsyncIteratorKey]: () => ResponseChunkIterator }; + +/** + * Wraps the raw headers in the fetch `Headers` shape. Lookups are case-insensitive. + * Multi-value headers are joined with ", ", names that differ only by case merge into one + * joined value, and names without a value are skipped. + */ +function wrapRawHeaders(headers: RawHeaders): Headers { + const normalized = new Map(); + Object.keys(headers).forEach((key) => { + const val = headers[key]; + if (val !== undefined && val !== null) { + const name = key.toLowerCase(); + const joined = Array.isArray(val) ? val.join(', ') : val; + const existing = normalized.get(name); + normalized.set(name, existing === undefined ? joined : `${existing}, ${joined}`); + } + }); + + return { + get: (name: string) => normalized.get(name.toLowerCase()) ?? null, + keys: () => normalized.keys(), + values: () => normalized.values(), + entries: () => normalized.entries(), + forEach: (callback: (value: string, key: string) => void) => { + normalized.forEach((value, key) => callback(value, key)); + }, + has: (name: string) => normalized.has(name.toLowerCase()), + }; +} + +/** + * A response for a streaming request. Unlike the base response, the body and the status + * message are always present. + */ +export interface StreamingResponse extends Response { + body: ResponseBody; + statusText: string; +} + +/** + * Creates a response for a streaming request. The response does not buffer the body. Each + * read hands out the next chunk from the source iterator. The body does not enforce a + * consumer lock: every `getReader`, `text`, and `json` call reads from the same iterator, + * so the caller uses one consumer per response. The `cancel` methods on the body and the + * reader call every release function the source provides: they destroy the source when it + * has a `destroy` method, and they also call the iterator's `return` method when one + * exists. The body passes through as-is, so the caller decodes any content encoding first. + * + * @param source The raw platform response, for example a Node `IncomingMessage`. A missing + * status code maps to 0, and a missing status message maps to an empty string. + */ +export function createStreamingResponse(source: StreamingResponseSource): StreamingResponse { + // A consumer compiled without `Symbol.asyncIterator` in its lib can pass a source that + // lacks the iterator without a compile error. The check turns that mistake into a named + // failure instead of a bare member-access TypeError. + if (typeof source[Symbol.asyncIterator] !== 'function') { + throw new TypeError('the streaming response source does not provide an async iterator'); + } + const iterator = source[Symbol.asyncIterator](); + + const cancel = async (): Promise => { + // Destroy the stream first. That rejects a pending read and unblocks a queued return + // call. A Node readable iterator queues the return call behind a pending read, and on + // a stream that was never read it completes without a destroy, so the return call + // alone cannot release the connection. Do not await it for the same reason. Each + // release call is guarded on its own, so a release function that throws, or a return + // that resolves synchronously, does not skip the other release path or reject cancel. + try { + source.destroy?.(); + } catch { + // A throwing destroy has nothing left to release. + } + try { + Promise.resolve(iterator.return?.()).catch(() => {}); + } catch { + // A throwing return has nothing left to release. + } + }; + + async function text(): Promise { + const decoder = new TextDecoder(); + let result = ''; + try { + let next = await iterator.next(); + while (!next.done) { + if (next.value !== undefined) { + try { + result += decoder.decode(next.value, { stream: true }); + } catch { + // decode() rejects a chunk that is not a BufferSource. The fixed message keeps + // chunk contents out of error messages. + throw new TypeError('the source supplied a chunk that is not a BufferSource'); + } + } + // The stream provides the chunks in sequence, so the reads cannot run in parallel. + // oxlint-disable-next-line no-await-in-loop + next = await iterator.next(); + } + } catch (err) { + // text() owns the body, so any failed read must release the source itself. This + // covers a rejecting next() as well as a bad chunk. + cancel(); + throw err; + } + // The final decode flushes a trailing partial multi-byte sequence. + return result + decoder.decode(); + } + + return { + headers: wrapRawHeaders(source.headers), + status: source.statusCode ?? 0, + statusText: source.statusMessage ?? '', + body: { + getReader: () => ({ + read: async () => { + const next = await iterator.next(); + if (next.done) { + return { done: true }; + } + return { done: false, value: next.value }; + }, + cancel, + }), + cancel, + }, + text, + json: async () => JSON.parse(await text()), + }; +} diff --git a/packages/shared/common/src/internal/fetch/index.ts b/packages/shared/common/src/internal/fetch/index.ts new file mode 100644 index 0000000000..143577f299 --- /dev/null +++ b/packages/shared/common/src/internal/fetch/index.ts @@ -0,0 +1,2 @@ +export * from './StreamingResponse'; +export * from './toRequestInit'; diff --git a/packages/shared/common/src/internal/fetch/toRequestInit.ts b/packages/shared/common/src/internal/fetch/toRequestInit.ts new file mode 100644 index 0000000000..c8b6ccb3a0 --- /dev/null +++ b/packages/shared/common/src/internal/fetch/toRequestInit.ts @@ -0,0 +1,12 @@ +import { Options } from '../../api/platform/Requests'; + +/** + * Converts the request options into the init shape a platform's native `fetch` accepts. A + * platform can declare its `AbortSignal` as a lexical class, which `AbortSignalLike` + * cannot resolve to, so the `signal` member widens here. The caller must pass a signal + * created by the running platform. The widening then only restores the type the runtime + * already has. + */ +export function toRequestInit(options: Options): Omit & { signal?: any } { + return options; +} diff --git a/packages/shared/common/src/internal/index.ts b/packages/shared/common/src/internal/index.ts index 7c1630e1b0..64bc52d368 100644 --- a/packages/shared/common/src/internal/index.ts +++ b/packages/shared/common/src/internal/index.ts @@ -3,6 +3,7 @@ export * from './diagnostics'; export * from './evaluation'; export * from './events'; export * from './fdv2'; +export * from './fetch'; export * from './metadata'; export * from './plugins'; export * from './json'; diff --git a/packages/shared/common/tsconfig.json b/packages/shared/common/tsconfig.json index b5d080223b..5727fc7f07 100644 --- a/packages/shared/common/tsconfig.json +++ b/packages/shared/common/tsconfig.json @@ -3,7 +3,7 @@ "rootDir": "src", "outDir": "dist", "target": "ES2020", - "lib": ["es6"], + "lib": ["es6", "es2018.asynciterable"], "module": "ESNext", "moduleResolution": "node", "strict": true, diff --git a/packages/shared/eventsource/__tests__/EventSource.close.test.ts b/packages/shared/eventsource/__tests__/EventSource.close.test.ts index 101da940c2..4715fbce1c 100644 --- a/packages/shared/eventsource/__tests__/EventSource.close.test.ts +++ b/packages/shared/eventsource/__tests__/EventSource.close.test.ts @@ -129,6 +129,7 @@ it('aborts the request when the error filter declines a retry mid-stream', async status: 200, statusText: 'OK', headers: { + get: () => 'text/event-stream', forEach(callback: (value: string, key: string) => void): void { callback('text/event-stream', 'content-type'); }, @@ -237,6 +238,7 @@ it('closes cleanly and aborts the request when the error filter throws mid-strea status: 200, statusText: 'OK', headers: { + get: () => 'text/event-stream', forEach(callback: (value: string, key: string) => void): void { callback('text/event-stream', 'content-type'); }, diff --git a/packages/shared/eventsource/__tests__/EventSource.fetchInjection.test.ts b/packages/shared/eventsource/__tests__/EventSource.fetchInjection.test.ts index 31898d74d1..094be9774f 100644 --- a/packages/shared/eventsource/__tests__/EventSource.fetchInjection.test.ts +++ b/packages/shared/eventsource/__tests__/EventSource.fetchInjection.test.ts @@ -23,6 +23,25 @@ import { writeEvents, } from './helpers'; +/** + * Builds a FetchHeaders over a record, like the SDK platform adapters do: `get` folds case, and + * `forEach` reports the record's own casing. + */ +function recordHeaders(record: Record): FetchHeaders { + return { + get: (name: string) => { + const match = Object.keys(record).find((key) => key.toLowerCase() === name.toLowerCase()); + return match === undefined ? null : record[match]; + }, + forEach: (callback: (value: string, key: string) => void) => { + Object.entries(record).forEach(([key, value]) => callback(value, key)); + }, + }; +} + +/** The headers of an ordinary SSE response. */ +const sseHeaders = (): FetchHeaders => recordHeaders({ 'content-type': 'text/event-stream' }); + /** * A canned response whose body hands out the given chunks and then stays pending forever, like an * idle SSE connection. The shape matches what the Node SDK http/https adapters produce. @@ -33,11 +52,7 @@ function idleStreamResponse(chunks: string[], headers?: FetchHeaders): FetchLike return { status: 200, statusText: 'OK', - headers: headers ?? { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: headers ?? sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -176,7 +191,7 @@ it('reports a non-200 response from an injected fetch as an error', async () => const injected: FetchLike = async () => ({ status: 401, statusText: 'Unauthorized', - headers: { forEach: () => {} }, + headers: recordHeaders({}), body: null, }); const url = `http://localhost:${deliberatelyUnusedPort}/stream`; @@ -191,8 +206,59 @@ it('reports a non-200 response from an injected fetch as an error', async () => } }); -it('accepts a response whose transport supplies no headers at all', async () => { - const injected: FetchLike = async () => idleStreamResponse(['data: hello\n\n'], { forEach() {} }); +it('accepts a response whose headers iterate no entries', async () => { + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], { get: () => null, forEach() {} }); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const es = createEventSource(url, { fetch: injected }); + es.onerror = () => {}; + try { + const messages = startMessageQueue(es); + expect((await messages.take()).data).toEqual('hello'); + } finally { + es.close(); + } +}); + +it('opens the stream when the response headers expose no forEach', async () => { + // A transport without header iteration reports no headers. A 200 with no headers is accepted, + // so the stream must open rather than fail as unreadable. + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], { get: () => null }); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const es = createEventSource(url, { fetch: injected }); + es.onerror = () => {}; + try { + const messages = startMessageQueue(es); + expect((await messages.take()).data).toEqual('hello'); + } finally { + es.close(); + } +}); + +it('rejects a wrong declared content type from a transport without forEach', async () => { + // Content-type validation reads through get, so it works for a transport whose headers + // expose no iteration. + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], { get: () => 'application/json' }); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const errors = new AsyncQueue(); + const es = createEventSource(url, { fetch: injected, initialRetryDelayMillis: 60000 }); + es.onerror = (e) => errors.add(e); + try { + const err = await errors.take(); + expect(err?.message).toContain('unexpected Content-Type'); + } finally { + es.close(); + } +}); + +it('opens the stream when get returns undefined for an absent header', async () => { + // A sloppy transport can implement get as a plain record lookup, which returns undefined + // instead of null. That reads as no declared type, not as a failure. + const sloppy = { get: () => undefined }; + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], sloppy as unknown as FetchHeaders); const url = `http://localhost:${deliberatelyUnusedPort}/stream`; const es = createEventSource(url, { fetch: injected }); es.onerror = () => {}; @@ -204,6 +270,111 @@ it('accepts a response whose transport supplies no headers at all', async () => } }); +it('rejects a wrong content type that only the iterated headers reveal', async () => { + // A get that does not fold header-name case returns null for a present header. The + // iterated record closes that gate bypass. + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], { + get: () => null, + forEach(callback: (value: string, key: string) => void): void { + callback('text/html', 'Content-Type'); + }, + }); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const errors = new AsyncQueue(); + const es = createEventSource(url, { fetch: injected, initialRetryDelayMillis: 60000 }); + es.onerror = (e) => errors.add(e); + try { + const err = await errors.take(); + expect(err?.message).toContain('unexpected Content-Type'); + } finally { + es.close(); + } +}); + +it('rejects a wrong content type when get returns a non-string value', async () => { + // Only a string from get is a declared type. A hostile non-string result defers to the + // iterated record instead of skipping the gate. + const hostile = { + get: () => 12345, + forEach(callback: (value: string, key: string) => void): void { + callback('text/html', 'content-type'); + }, + }; + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], hostile as unknown as FetchHeaders); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const errors = new AsyncQueue(); + const es = createEventSource(url, { fetch: injected, initialRetryDelayMillis: 60000 }); + es.onerror = (e) => errors.add(e); + try { + const err = await errors.take(); + expect(err?.message).toContain('unexpected Content-Type'); + } finally { + es.close(); + } +}); + +it('opens the stream for a transport whose headers only expose forEach', async () => { + // The only legal headers shape before get was added. It must keep working through the + // iterated record. + const legacy = { + forEach(callback: (value: string, key: string) => void): void { + callback('text/event-stream', 'content-type'); + }, + }; + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], legacy as unknown as FetchHeaders); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const es = createEventSource(url, { fetch: injected }); + es.onerror = () => {}; + try { + const messages = startMessageQueue(es); + expect((await messages.take()).data).toEqual('hello'); + } finally { + es.close(); + } +}); + +it('rejects a wrong content type for a transport whose headers only expose forEach', async () => { + const legacy = { + forEach(callback: (value: string, key: string) => void): void { + callback('text/html', 'content-type'); + }, + }; + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], legacy as unknown as FetchHeaders); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const errors = new AsyncQueue(); + const es = createEventSource(url, { fetch: injected, initialRetryDelayMillis: 60000 }); + es.onerror = (e) => errors.add(e); + try { + const err = await errors.take(); + expect(err?.message).toContain('unexpected Content-Type'); + } finally { + es.close(); + } +}); + +it('reports an empty message for a non-200 response without a statusText', async () => { + const injected: FetchLike = async () => ({ + status: 503, + headers: { get: () => null }, + body: null, + }); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const errors = new AsyncQueue(); + const es = createEventSource(url, { fetch: injected, errorFilter: () => false }); + es.onerror = (e) => errors.add(e); + try { + const err = await errors.take(); + expect(err?.status).toEqual(503); + expect(err?.message).toEqual(''); + } finally { + es.close(); + } +}); + it('aborts the request when a 200 response has no body', async () => { let signal: AbortSignal | undefined; const injected: FetchLike = async (_url, init) => { @@ -211,7 +382,7 @@ it('aborts the request when a 200 response has no body', async () => { return { status: 200, statusText: 'OK', - headers: { forEach: () => {} }, + headers: recordHeaders({}), body: null, }; }; @@ -232,12 +403,10 @@ it('folds response header names to lowercase for a transport that reports wire c // A standard Headers object reports lowercase names; a raw-header transport reports the wire // case. Consumers look headers up by lowercase name, so both must produce the same shape. const injected: FetchLike = async () => - idleStreamResponse(['data: hello\n\n'], { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'Content-Type'); - callback('abc', 'X-LD-EnvId'); - }, - }); + idleStreamResponse( + ['data: hello\n\n'], + recordHeaders({ 'Content-Type': 'text/event-stream', 'X-LD-EnvId': 'abc' }), + ); const url = `http://localhost:${deliberatelyUnusedPort}/stream`; const es = createEventSource(url, { fetch: injected }); es.onerror = () => {}; @@ -258,11 +427,7 @@ it('cancels the body reader on close for a transport that ignores the abort sign const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => { let delivered = false; @@ -299,11 +464,7 @@ it('yields to timers while a transport resolves reads synchronously', async () = const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -341,11 +502,7 @@ it('treats a stream of empty chunks as dead when the read timeout elapses', asyn const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -386,11 +543,7 @@ it('does not create a reader when onopen closes the stream', async () => { const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader, cancel: bodyCancel }, }); const url = `http://localhost:${deliberatelyUnusedPort}/stream`; @@ -414,17 +567,13 @@ it('releases the unread body of a rejected response', async () => { { status: 503, statusText: 'Service Unavailable', - headers: { forEach: () => {} }, + headers: recordHeaders({}), body: { getReader: () => ({ read: () => new Promise(() => {}) }), cancel: bodyCancel }, }, { status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/html', 'content-type'); - }, - }, + headers: recordHeaders({ 'content-type': 'text/html' }), body: { getReader: () => ({ read: () => new Promise(() => {}) }), cancel: bodyCancel }, }, ]; @@ -464,7 +613,7 @@ it('releases the body of a response that resolves after close', async () => { resolveFetch!({ status: 200, statusText: 'OK', - headers: { forEach: () => {} }, + headers: recordHeaders({}), body: { getReader: () => ({ read: () => new Promise(() => {}) }), cancel: bodyCancel }, }); // The response reaches the stale callback on the microtask queue. The timer hop runs after it. @@ -482,11 +631,7 @@ it('delivers chunks a transport supplies as ArrayBuffer', async () => { const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -530,11 +675,7 @@ it('delivers chunks from an ArrayBuffer created in another realm', async () => { const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -573,11 +714,7 @@ it('reports the fallback message when a read rejection hides its own message', a return { status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -632,11 +769,7 @@ it('delivers chunks a transport supplies as SharedArrayBuffer', async () => { const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -672,11 +805,7 @@ it('delivers an ArrayBuffer chunk that carries its own string tag', async () => const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -708,11 +837,7 @@ it('reports an error for a chunk that forges the buffer brand', async () => { const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => ({ @@ -737,7 +862,7 @@ it('reports an error for a chunk that forges the buffer brand', async () => { it('keeps the status when a response statusText getter throws', async () => { const hostile = { status: 503, - headers: { forEach: () => {} }, + headers: recordHeaders({}), body: null, }; Object.defineProperty(hostile, 'statusText', { @@ -765,7 +890,7 @@ it('releases the request when a hostile response getter escapes the callback', a signal = init.signal; const hostile = { statusText: 'OK', - headers: { forEach: () => {} }, + headers: recordHeaders({}), body: null, }; Object.defineProperty(hostile, 'status', { @@ -790,12 +915,13 @@ it('releases the request when a hostile response getter escapes the callback', a }); it('fails a 200 response whose headers cannot be read', async () => { - // A forEach that delivers a wrong content type and then throws must not open the stream as a - // headerless success with its partial headers discarded. + // A forEach that throws must not open the stream as a headerless success with its partial + // headers discarded, even when get works and declares a valid content type. const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', headers: { + get: () => 'text/event-stream', forEach(callback: (value: string, key: string) => void): void { callback('text/html', 'content-type'); throw new Error('headers boom'); @@ -815,6 +941,53 @@ it('fails a 200 response whose headers cannot be read', async () => { } }); +it('opens through the iterated headers when get throws', async () => { + // A throwing get leaves the iterated record as the only content-type source. The stream + // must open through it instead of failing as unreadable. + const injected: FetchLike = async () => + idleStreamResponse(['data: hello\n\n'], { + get: () => { + throw new Error('get boom'); + }, + forEach(callback: (value: string, key: string) => void): void { + callback('text/event-stream', 'content-type'); + }, + }); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const es = createEventSource(url, { fetch: injected }); + es.onerror = () => {}; + try { + const messages = startMessageQueue(es); + expect((await messages.take()).data).toEqual('hello'); + } finally { + es.close(); + } +}); + +it('fails a 200 response whose headers expose no access method', async () => { + // With no usable get and no iteration, the content type cannot be validated at all. + const injected: FetchLike = async () => ({ + status: 200, + statusText: 'OK', + headers: { + get: () => { + throw new Error('get boom'); + }, + }, + body: { getReader: () => ({ read: () => new Promise(() => {}) }) }, + }); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const es = createEventSource(url, { fetch: injected, initialRetryDelayMillis: 60000 }); + const events = new AsyncQueue(); + es.addEventListener('open', () => events.add('open')); + es.onerror = (e) => events.add(`error:${e?.message}`); + try { + expect(await events.take()).toEqual('error:stream response headers could not be read'); + } finally { + es.close(); + } +}); + it('cancels the reader when getReader itself closes the stream', async () => { // The body getter and getReader() are the last caller code before the read loop. A close() // from inside either one must not leave an uncancelled reader on the closed stream. @@ -824,11 +997,7 @@ it('cancels the reader when getReader itself closes the stream', async () => { const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => { es?.close(); @@ -862,6 +1031,7 @@ it('stays closed when a transport accessor calls close during the response callb status: 200, statusText: 'OK', headers: { + get: () => 'text/event-stream', forEach(): void { es?.close(); }, @@ -894,7 +1064,7 @@ it('reads the response status once for the failure report', async () => { let reads = 0; const hostile = { statusText: 'Service Unavailable', - headers: { forEach: () => {} }, + headers: recordHeaders({}), body: null, }; Object.defineProperty(hostile, 'status', { @@ -924,6 +1094,34 @@ it('reads the response status once for the failure report', async () => { } }); +it('keeps the headers when a non-200 response has no get', async () => { + // A transport without get fails the content-type lookup. A failure response does not + // need the declared type, so the report must keep the headers that forEach collected, + // including the FDv1 fallback directive. + const headers = { + forEach(callback: (value: string, key: string) => void): void { + callback('true', 'x-ld-fd-fallback'); + }, + } as unknown as FetchHeaders; + const injected: FetchLike = async () => ({ + status: 503, + statusText: 'Service Unavailable', + headers, + body: null, + }); + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const errors = new AsyncQueue(); + const es = createEventSource(url, { fetch: injected, errorFilter: () => false }); + es.onerror = (e) => errors.add(e); + try { + const err = await errors.take(); + expect(err?.status).toEqual(503); + expect(err?.headers?.['x-ld-fd-fallback']).toEqual('true'); + } finally { + es.close(); + } +}); + it('keeps the status when a non-200 response headers forEach throws', async () => { // Without the headers fallback, the throw would escape to the fetch catch and the 401 would // be reported with no status, so a permanent auth failure would retry forever. @@ -931,6 +1129,7 @@ it('keeps the status when a non-200 response headers forEach throws', async () = status: 401, statusText: 'Unauthorized', headers: { + get: () => null, forEach(): void { throw new Error('headers boom'); }, @@ -949,6 +1148,52 @@ it('keeps the status when a non-200 response headers forEach throws', async () = } }); +it('keeps the status when the response headers getter throws', async () => { + // The headers read happens once, inside the parser's guard. A throwing getter must not + // erase the status from the failure report, or a permanent auth failure would retry forever. + const hostile = { status: 401, statusText: 'Unauthorized', body: null }; + Object.defineProperty(hostile, 'headers', { + get() { + throw new Error('headers boom'); + }, + }); + const injected: FetchLike = async () => hostile as unknown as FetchLikeResponse; + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const errors = new AsyncQueue(); + const es = createEventSource(url, { fetch: injected, errorFilter: () => false }); + es.onerror = (e) => errors.add(e); + try { + const err = await errors.take(); + expect(err?.status).toEqual(401); + } finally { + es.close(); + } +}); + +it('fails a 200 response whose headers getter throws', async () => { + const hostile = { + status: 200, + statusText: 'OK', + body: { getReader: () => ({ read: () => new Promise(() => {}) }) }, + }; + Object.defineProperty(hostile, 'headers', { + get() { + throw new Error('headers boom'); + }, + }); + const injected: FetchLike = async () => hostile as unknown as FetchLikeResponse; + const url = `http://localhost:${deliberatelyUnusedPort}/stream`; + const es = createEventSource(url, { fetch: injected, initialRetryDelayMillis: 60000 }); + const events = new AsyncQueue(); + es.addEventListener('open', () => events.add('open')); + es.onerror = (e) => events.add(`error:${e?.message}`); + try { + expect(await events.take()).toEqual('error:stream response headers could not be read'); + } finally { + es.close(); + } +}); + it('ignores a stale fetch rejection after a newer attempt is streaming', async () => { // A slow transport can reject the first attempt's promise long after the read timeout has // already failed over to a second attempt. The stale rejection must not release the live @@ -976,11 +1221,7 @@ it('ignores a stale fetch rejection after a newer attempt is streaming', async ( return Promise.resolve({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -1060,11 +1301,7 @@ it('reports an error instead of stalling when a transport supplies a non-buffer const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => ({ @@ -1097,11 +1334,7 @@ it('keeps the read timeout alive across a partial multi-byte chunk', async () => const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -1142,11 +1375,7 @@ it('releases the request when getReader throws an error whose message getter thr return { status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => { const hostile = {}; @@ -1185,11 +1414,7 @@ it('releases the request when getReader throws', async () => { return { status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => { throw new Error('stream locked'); @@ -1219,11 +1444,7 @@ it('close() tolerates a reader cancel that throws synchronously', async () => { const injected: FetchLike = async () => ({ status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => { let delivered = false; @@ -1263,11 +1484,7 @@ it('aborts the failed request when a retried read error schedules a reconnect', return { status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { @@ -1307,11 +1524,7 @@ it('recomputes the message origin when urlBuilder picks a new origin for a recon return { status: 200, statusText: 'OK', - headers: { - forEach(callback: (value: string, key: string) => void): void { - callback('text/event-stream', 'content-type'); - }, - }, + headers: sseHeaders(), body: { getReader: () => ({ read: async (): Promise<{ done: boolean; value?: Uint8Array }> => { diff --git a/packages/shared/eventsource/__tests__/EventSource.throughput.test.ts b/packages/shared/eventsource/__tests__/EventSource.throughput.test.ts index 4678789f6f..0d20f3f036 100644 --- a/packages/shared/eventsource/__tests__/EventSource.throughput.test.ts +++ b/packages/shared/eventsource/__tests__/EventSource.throughput.test.ts @@ -12,6 +12,7 @@ function singleChunkResponse(chunk: Uint8Array): FetchLikeResponse { status: 200, statusText: 'OK', headers: { + get: () => 'text/event-stream', forEach(callback: (value: string, key: string) => void): void { callback('text/event-stream', 'content-type'); }, diff --git a/packages/shared/eventsource/__tests__/httpHelpers.test.ts b/packages/shared/eventsource/__tests__/httpHelpers.test.ts index c0b643ec53..b4dd0844b2 100644 --- a/packages/shared/eventsource/__tests__/httpHelpers.test.ts +++ b/packages/shared/eventsource/__tests__/httpHelpers.test.ts @@ -1,4 +1,4 @@ -import { splitHeaderListValue } from '../src/httpHelpers'; +import { headersToObject, splitHeaderListValue } from '../src/httpHelpers'; it('splits a joined value on commas outside quotes', () => { expect(splitHeaderListValue('a, b, c')).toEqual(['a', ' b', ' c']); @@ -38,3 +38,28 @@ it('runs an unterminated quoted string to the end of the value', () => { it('keeps empty parts', () => { expect(splitHeaderListValue('a,,b,')).toEqual(['a', '', 'b', '']); }); + +it('returns null when the headers expose no iteration method', () => { + expect(headersToObject({ get: () => null })).toBeNull(); +}); + +it('collects headers through forEach and folds the names to lowercase', () => { + expect( + headersToObject({ + get: () => null, + forEach: (callback) => { + callback('text/event-stream', 'Content-Type'); + }, + }), + ).toEqual({ 'content-type': 'text/event-stream' }); +}); + +it('collects headers through entries when forEach is absent', () => { + const pairs: [string, string][] = [['Content-Type', 'text/event-stream']]; + expect( + headersToObject({ + get: () => null, + entries: () => pairs, + }), + ).toEqual({ 'content-type': 'text/event-stream' }); +}); diff --git a/packages/shared/eventsource/__tests__/typeCompatibility.test.ts b/packages/shared/eventsource/__tests__/typeCompatibility.test.ts index a7d7aed4d8..29474f6aed 100644 --- a/packages/shared/eventsource/__tests__/typeCompatibility.test.ts +++ b/packages/shared/eventsource/__tests__/typeCompatibility.test.ts @@ -3,11 +3,22 @@ import type { EventSource as PlatformEventSource, EventSourceInitDict as PlatformEventSourceInitDict, EventSourceRetryDelayStrategy as PlatformRetryDelayStrategy, + Headers as PlatformHeaders, + Options as PlatformOptions, + Requests as PlatformRequests, + Response as PlatformResponse, } from '@launchdarkly/js-sdk-common'; import { makeEvent, MessageEvent, RetryEvent } from '../src/Event'; import { createEventSource } from '../src/EventSource'; -import { EventSourceInitDict, RetryDelayStrategy } from '../src/types'; +import { + EventSourceInitDict, + FetchHeaders, + FetchLike, + FetchLikeOptions, + FetchLikeResponse, + RetryDelayStrategy, +} from '../src/types'; import { deliberatelyUnusedPort } from './helpers'; /** @@ -35,6 +46,46 @@ type UnsupportedPlatformOptions = Exclude< // eslint-disable-next-line @typescript-eslint/no-unused-vars const unsupportedOptionsCheck: UnsupportedPlatformOptions extends never ? true : false = true; +/** + * The transport seam. A platform `Requests.fetch` doubles as this package's injected `FetchLike` + * through structural typing, with no dependency between the packages in either direction. The + * checks below are the enforcement point for that contract. The first check is the contract + * itself. A platform fetch function is assignable to `FetchLike`. + */ +// eslint-disable-next-line @typescript-eslint/no-unused-vars +const platformFetchSatisfiesFetchLike: (f: PlatformRequests['fetch']) => FetchLike = (f) => f; + +/** + * True only when the two types are identical. The exception checks below use it so they fail + * when an exception is added and when a documented exception stops being one. + */ +type Equals = [A] extends [B] ? ([B] extends [A] ? true : false) : false; + +/** + * Every request member this package produces must be a declared platform option, so a platform + * implementation knows to handle it. `credentials` is the deliberate exception. A platform + * transport over a raw socket API has no equivalent and ignores it. + */ +type UnsupportedFetchOptionMembers = Exclude; +// eslint-disable-next-line @typescript-eslint/no-unused-vars +const fetchOptionMembersCheck: Equals = true; + +/** + * Every response member this package reads must be declarable on the platform response, so a + * platform implementation knows what to expose. `url` is the deliberate exception. The client + * falls back to the request url for the message origin. + */ +type UnsupportedFetchResponseMembers = Exclude; +// eslint-disable-next-line @typescript-eslint/no-unused-vars +const fetchResponseMembersCheck: Equals = true; + +/** + * Every header member this package reads must be declarable on the platform headers. + */ +type UnsupportedFetchHeaderMembers = Exclude; +// eslint-disable-next-line @typescript-eslint/no-unused-vars +const fetchHeaderMembersCheck: UnsupportedFetchHeaderMembers extends never ? true : false = true; + /** * The retry strategy seam. An SDK adapts a `RetryState` into the platform's * `EventSourceRetryDelayStrategy` and the platform adapter spreads it into this package's init diff --git a/packages/shared/eventsource/src/EventSource.ts b/packages/shared/eventsource/src/EventSource.ts index c158e04a7b..5c6331bf9b 100644 --- a/packages/shared/eventsource/src/EventSource.ts +++ b/packages/shared/eventsource/src/EventSource.ts @@ -16,8 +16,8 @@ import { import { bodylessMethods, defaultFetch, - headersToObject, INVALID_HEADER_VALUE_CHAR, + parseResponseHeaders, splitHeaderListValue, } from './httpHelpers'; import { createDefaultEventRegistry } from './listenerRegistry'; @@ -621,17 +621,14 @@ export function createEventSource( return; } - // A hostile headers object must not break the failure report. Without the fallback, a - // non-200 status would be lost with the throw, and the error filter could not classify - // the failure. - let responseHeaders: Record; - let headersUnreadable = false; - try { - responseHeaders = headersToObject(res.headers); - } catch { - responseHeaders = {}; - headersUnreadable = true; - } + // Every header access is guarded inside the parser, so a hostile headers object cannot + // break the failure report. Without that guard, a non-200 status would be lost with the + // throw, and the error filter could not classify the failure. + const { + headers: responseHeaders, + contentType, + unreadable: headersUnreadable, + } = parseResponseHeaders(res); // The status is read once. A getter that changes its answer between reads would skew the // classification, and a throw on a later read would skip the release below. @@ -663,9 +660,9 @@ export function createEventSource( } if (headersUnreadable) { - // On a success status the headers decide whether the body is a stream at all. A 200 - // whose headers cannot be read must fail and retry, not open as a headerless stream - // with its partially collected headers discarded. + // On a success status the headers feed the open event and the declared content type + // gates the stream. A 200 whose headers throw during the read must fail and retry, not + // open with its partially collected headers discarded. failOnce({ message: 'stream response headers could not be read' }); releaseBody(res); return; @@ -677,14 +674,13 @@ export function createEventSource( // transport can report no headers. // // The media type must match exactly. Parameters after it, such as a charset, are allowed. - // A standard Headers object comma-joins a duplicated Content-Type header, so each part of - // a joined value must declare the event-stream media type. This is stricter than the - // Fetch algorithm, which keeps only the last parsable part, and an empty or unparsable - // part also fails. + // A standard get comma-joins a duplicated Content-Type header, so each part of a joined + // value must declare the event-stream media type. This is stricter than the Fetch + // algorithm, which keeps only the last parsable part, and an empty or unparsable part + // also fails. // // trim() also strips NBSP, form feed, and vertical tab, which are not HTTP whitespace. // The extra leniency is harmless. - const contentType = responseHeaders['content-type']; if ( contentType !== undefined && !splitHeaderListValue(contentType).every( diff --git a/packages/shared/eventsource/src/httpHelpers.ts b/packages/shared/eventsource/src/httpHelpers.ts index d78149c58b..f31bcd8ff6 100644 --- a/packages/shared/eventsource/src/httpHelpers.ts +++ b/packages/shared/eventsource/src/httpHelpers.ts @@ -11,15 +11,34 @@ */ /** - * The response headers. `forEach` is the only member this package reads. + * The response headers. The member set mirrors the subset of the standard `Headers` class that + * this package reads. */ export interface FetchHeaders { - forEach(callback: (value: string, key: string) => void): void; + /** + * Returns the value of a header, or `null` when the response carries no such header. The + * lookup is case-insensitive, like the standard `Headers.get`. The client validates the + * declared content type through it. + */ + get(name: string): string | null; + + /** + * Executes the callback once for each header, with the value first. It is optional so a + * transport whose response headers expose no iteration can still satisfy the type. + */ + forEach?(callback: (value: string, key: string) => void): void; + + /** + * Returns an iterator over the header name-value pairs, like the standard + * `Headers.entries`. It is the iteration fallback when `forEach` is absent. + */ + entries?(): Iterable<[string, string]>; } /** * A reader over the response body stream. `read` resolves with the next chunk, or with - * `done: true` when the server ends the stream, and rejects when the connection drops. + * `done: true` when the server ends the stream, and rejects when the connection terminates + * before the server ends the stream. */ export interface FetchBodyReader { read(): Promise<{ done: boolean; value?: Uint8Array }>; @@ -50,7 +69,11 @@ export interface FetchResponseBody { */ export interface FetchLikeResponse { readonly status: number; - readonly statusText: string; + /** + * The HTTP status message. It is optional so a minimal injected transport can omit it. The + * client then reports an empty message for a failed response. + */ + readonly statusText?: string; readonly headers: FetchHeaders; readonly body?: FetchResponseBody | null; /** @@ -91,14 +114,92 @@ export const defaultFetch: FetchLike = (url, init) => globalThis.fetch(url, init /** The methods for which a `fetch()` request cannot have a body. */ export const bodylessMethods = ['GET', 'HEAD']; -export function headersToObject(headers: FetchHeaders): Record { +/** + * Collects the headers into a plain record. Returns `null` when the headers object provides + * no iteration method at all, so the caller can tell an unreadable header set from an empty + * one. A standard Headers object reports lowercase names, so names fold to lowercase. The + * fold gives a transport that reports wire-cased names the same shape. + */ +export function headersToObject(headers: FetchHeaders): Record | null { const result: Record = {}; - headers.forEach((value, key) => { - // A standard Headers object reports lowercase names. The fold gives a transport that reports - // wire-cased names the same shape, so a consumer can look a header up by its lowercase name. - result[key.toLowerCase()] = value; - }); - return result; + if (headers.forEach) { + headers.forEach((value, key) => { + result[key.toLowerCase()] = value; + }); + return result; + } + if (headers.entries) { + Array.from(headers.entries()).forEach(([key, value]) => { + result[key.toLowerCase()] = value; + }); + return result; + } + return null; +} + +/** + * The response headers in the shape the client consumes. `contentType` is the declared + * content type, or `undefined` when the response declares none. `unreadable` is true when + * no header access works at all. + */ +export interface ParsedResponseHeaders { + headers: Record; + contentType: string | undefined; + unreadable: boolean; +} + +/** + * Reads and validates the headers of a response. Every access to the headers object is + * guarded, so a hostile response cannot break the failure report with a throw. The declared + * content type prefers the `get` lookup and falls back to the iterated record. + */ +export function parseResponseHeaders(res: FetchLikeResponse): ParsedResponseHeaders { + let resHeaders: FetchHeaders | undefined; + try { + ({ headers: resHeaders } = res); + } catch { + resHeaders = undefined; + } + if (resHeaders === undefined || resHeaders === null) { + return { headers: {}, contentType: undefined, unreadable: true }; + } + + let headers: Record = {}; + let iterated = false; + let unreadable = false; + try { + const collected = headersToObject(resHeaders); + iterated = collected !== null; + headers = collected ?? {}; + } catch { + unreadable = true; + } + + // `getResult` holds what `get` returned. `getAnsweredAbsent` is true only when a working + // `get` reported the header as absent, which is the one case where a missing record is + // not a failure. + let getResult: unknown; + let getAnsweredAbsent = false; + try { + if (typeof resHeaders.get === 'function') { + getResult = resHeaders.get('content-type'); + getAnsweredAbsent = getResult === null || getResult === undefined; + } + } catch { + // A throwing get leaves the declared type to the iterated record. + getResult = undefined; + } + + let contentType: string | undefined; + if (typeof getResult === 'string') { + contentType = getResult; + } else if (iterated) { + contentType = headers['content-type']; + } else if (!getAnsweredAbsent) { + unreadable = true; + } + + return { headers, contentType, unreadable }; } /**