From 26c4627dcff64774b76d6891216f33fb008d6a94 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 21 Aug 2026 02:32:49 +0000 Subject: [PATCH 1/2] undici: implement dispatch(), close() and destroy() on the shim dispatchers Implements the undici Dispatcher API over Bun.fetch: dispatch() with both the legacy and the v7 controller handler interfaces, request() with the body mixin, close() and destroy() lifecycle, and a dispatcher-aware fetch() so miniflare can route requests through a custom dispatcher. Co-authored-by: Matthew Bell <33056264+matthew2564@users.noreply.github.com> --- src/js/thirdparty/undici.js | 849 +++++++++++++- test/js/first_party/undici/undici.test.ts | 1229 ++++++++++++++++++++- 2 files changed, 2055 insertions(+), 23 deletions(-) diff --git a/src/js/thirdparty/undici.js b/src/js/thirdparty/undici.js index 70523eacde0e..dc28a67b17cf 100644 --- a/src/js/thirdparty/undici.js +++ b/src/js/thirdparty/undici.js @@ -1,10 +1,14 @@ const EventEmitter = require("node:events"); +const { Readable } = require("node:stream"); +const { Buffer } = require("node:buffer"); const { _ReadableFromWeb: ReadableFromWeb } = require("internal/webstreams_adapters"); const ObjectCreate = Object.create; -const kEmptyObject = ObjectCreate(null); +const kEmptyObject = Object.freeze(ObjectCreate(null)); +// Captured at module load so tampering with globalThis later cannot break dispatch. +const { AbortController, ArrayBuffer, Blob, ReadableStream } = globalThis; -var fetch = Bun.fetch; +const nativeFetch = Bun.fetch; const bindings = $cpp("Undici.cpp", "createUndiciInternalBinding"); const Response = bindings[0]; const Request = bindings[1]; @@ -70,12 +74,11 @@ class BodyReadable extends ReadableFromWeb { } get bodyUsed() { - // return this.#response.bodyUsed; - return this.#bodyUsed; + return this.#bodyUsed || Readable.isDisturbed(this); } #consume() { - if (this.#bodyUsed) throw new TypeError("unusable"); + if (this.#bodyUsed || Readable.isDisturbed(this)) throw new TypeError("unusable"); this.#bodyUsed = true; } @@ -206,7 +209,7 @@ async function request( const followRedirects = maxRedirections != null && maxRedirections > 0; /** @type {Response} */ - const resp = await fetch(url, { + const resp = await nativeFetch(url, { signal, mode: "cors", method, @@ -254,17 +257,581 @@ class MockAgent { function mockErrors() {} -class Dispatcher extends EventEmitter {} -class Agent extends Dispatcher {} -class Pool extends Dispatcher { - request() {} +function appendHeader(headers, name, value) { + if (value === undefined || value === null) return; + name = String(name); + if ($isJSArray(value)) { + for (const v of value) appendHeader(headers, name, v); + return; + } + const existing = headers[name]; + if (existing === undefined) headers[name] = String(value); + else if ($isJSArray(existing)) existing.push(String(value)); + else headers[name] = [existing, String(value)]; +} + +function headersFromRawHeaders(rawHeaders) { + const headers = ObjectCreate(null); + for (let i = 0; i + 1 < rawHeaders.length; i += 2) { + appendHeader(headers, String(rawHeaders[i]).toLowerCase(), String(rawHeaders[i + 1])); + } + return headers; +} + +function parseOrigin(origin) { + const url = origin instanceof URL ? origin : new URL(String(origin)); + if (url.protocol !== "http:" && url.protocol !== "https:") + throw new InvalidArgumentError("Invalid URL protocol: the URL must start with `http:` or `https:`."); + if (url.pathname !== "/" || url.search || url.hash) throw new InvalidArgumentError("invalid url"); + return url; +} + +function headersFromDispatchOpts(headers) { + if (headers == null) return undefined; + // Entries form so fetch() appends repeated names as separate lines; a record init would String()-join array values. + const out = []; + const push = (name, value) => { + if (value === undefined || value === null) return; + if ($isJSArray(value)) { + for (const v of value) push(name, v); + return; + } + out.push([String(name), String(value)]); + }; + if ($isJSArray(headers)) { + if (headers.length > 0 && $isJSArray(headers[0])) { + for (const [name, value] of headers) push(name, value); + } else { + for (let i = 0; i + 1 < headers.length; i += 2) push(headers[i], headers[i + 1]); + } + } else if (typeof headers[Symbol.iterator] === "function") { + // Headers, Map, or any entries iterable; Object.keys() sees no own properties on these. + for (const [name, value] of headers) push(name, value); + } else { + for (const name of Object.keys(headers)) push(name, headers[name]); + } + return out; +} + +async function* iterableToByteChunks(iterable) { + for await (const chunk of iterable) { + yield typeof chunk === "string" ? Buffer.from(chunk) : chunk; + } +} + +function bodyFromDispatchOpts(body) { + if (body == null) return null; + if (typeof body === "string") return body; + if (body instanceof ArrayBuffer || ArrayBuffer.isView(body)) return body; + if (body instanceof Blob || body instanceof ReadableStream) return body; + if (body instanceof FormData || body instanceof URLSearchParams) return body; + // Stream Readables and (async) iterables instead of buffering them. + if (typeof body[Symbol.asyncIterator] === "function" || typeof body[Symbol.iterator] === "function") { + return Readable.toWeb(Readable.from(iterableToByteChunks(body))); + } + return body; +} + +// One fetch()-backed request driving legacy (onHeaders/onData) or v7 controller (onResponseStart/...) handler callbacks. +function fetchDispatch(origin, opts, handler, pending) { + const isControllerStyle = + typeof handler.onRequestStart === "function" || typeof handler.onResponseStart === "function"; + + const ac = new AbortController(); + let aborted = false; + let abortReason; + let paused = false; + let resumeResolve = null; + const resume = () => { + paused = false; + if (resumeResolve) { + const r = resumeResolve; + resumeResolve = null; + r(); + } + }; + + const abort = reason => { + if (aborted) return; + aborted = true; + abortReason = reason ?? new RequestAbortedError("Request aborted"); + ac.abort(abortReason); + // Wake the body loop if it is parked in a pause, so it observes the abort. + resume(); + }; + + const controller = { + abort, + pause() { + paused = true; + }, + resume, + get aborted() { + return aborted; + }, + get reason() { + return abortReason; + }, + get paused() { + return paused; + }, + }; + + // Tracked for close() drain and destroy() abort; done exists before handler callbacks run so a sync close() from onConnect still waits. + let resolveDone; + const entry = { abort, done: new Promise(r => (resolveDone = r)) }; + pending?.add(entry); + + (async () => { + const path = opts.path || "/"; + if (typeof path !== "string" || path.charCodeAt(0) !== 0x2f /* '/' */) { + throw new InvalidArgumentError("path must start with '/'"); + } + // Concatenate rather than URL-resolve so '//other.host/x' cannot change the request authority. + const base = origin instanceof URL ? origin : new URL(String(origin)); + const url = new URL(base.origin + path); + const { query } = opts; + if (query) { + if (path.includes("?") || path.includes("#")) { + throw new InvalidArgumentError('Query params cannot be passed when url already contains "?" or "#".'); + } + url.search = new URLSearchParams(query).toString(); + } + // The dispatch layer preserves method case, like undici. + const method = opts.method ? String(opts.method) : "GET"; + + if (isControllerStyle) handler.onRequestStart?.(controller, { __proto__: null }); + else handler.onConnect?.(abort); + if (aborted) throw abortReason; + + const body = bodyFromDispatchOpts(opts.body); + + const maxRedirections = opts.maxRedirections; + const followRedirects = typeof maxRedirections === "number" && maxRedirections > 0; + // Transport gap: fetch() rejects GET/HEAD bodies that undici's own client would send; the loud error beats dropping the body. + const resp = await nativeFetch(url, { + method, + headers: headersFromDispatchOpts(opts.headers), + body, + redirect: followRedirects ? "follow" : "manual", + maxRedirects: followRedirects ? maxRedirections : undefined, + signal: ac.signal, + keepalive: !opts.reset, + }); + + // fetch() already decompressed the body, so drop the encoding headers. + const responseHeaders = { __proto__: null, ...resp.headers.toJSON() }; + if (method.toUpperCase() !== "HEAD") { + delete responseHeaders["content-encoding"]; + delete responseHeaders["content-length"]; + } + + if (isControllerStyle) { + handler.onResponseStart?.(controller, resp.status, responseHeaders, resp.statusText); + } else if (typeof handler.onHeaders === "function") { + const rawHeaders = []; + for (const name in responseHeaders) { + const value = responseHeaders[name]; + if ($isJSArray(value)) { + for (const v of value) rawHeaders.push(Buffer.from(name), Buffer.from(v)); + } else { + rawHeaders.push(Buffer.from(name), Buffer.from(value)); + } + } + if (handler.onHeaders(resp.status, rawHeaders, resume, resp.statusText) === false) paused = true; + } + + const respBody = resp.body; + if (respBody) { + for await (const chunk of respBody) { + if (aborted) throw abortReason; + while (paused) { + await new Promise(r => { + resumeResolve = r; + }); + } + if (aborted) throw abortReason; + const buf = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength); + const ret = isControllerStyle ? handler.onResponseData?.(controller, buf) : handler.onData?.(buf); + if (ret === false) paused = true; + } + } + + // Covers abort() called from onHeaders on an empty body or from onData on the final chunk. + if (aborted) throw abortReason; + + if (isControllerStyle) handler.onResponseEnd?.(controller, { __proto__: null }); + else handler.onComplete?.([]); + })() + .catch(err => { + // Cancel the response body when a handler callback threw before the body loop. + if (!aborted) ac.abort(err); + try { + if (isControllerStyle && typeof handler.onResponseError === "function") + handler.onResponseError(controller, err); + else if (typeof handler.onError === "function") handler.onError(err); + else if (typeof handler.onResponseError === "function") handler.onResponseError(controller, err); + } catch { + // A throwing error callback must not become an unhandled rejection. + } + }) + .finally(() => { + pending?.delete(entry); + resolveDone(); + }); + + return true; +} + +// dispatcher.request() body: a Readable with undici's body-mixin methods. +class DispatchBodyReadable extends Readable { + #used = false; + #contentType; + + constructor(options, contentType) { + super(options); + this.#contentType = contentType; + } + + get bodyUsed() { + return this.#used || Readable.isDisturbed(this); + } + + async #consume() { + if (this.#used || Readable.isDisturbed(this)) throw new TypeError("unusable"); + this.#used = true; + const chunks = []; + for await (const chunk of this) chunks.push(chunk); + return Buffer.concat(chunks); + } + + async text() { + return (await this.#consume()).toString(); + } + + async json() { + return JSON.parse((await this.#consume()).toString()); + } + + async arrayBuffer() { + const buf = await this.#consume(); + return buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength); + } + + async bytes() { + return new Uint8Array(await this.#consume()); + } + + async blob() { + const buf = await this.#consume(); + return this.#contentType ? new Blob([buf], { type: String(this.#contentType) }) : new Blob([buf]); + } + + async formData() { + const buf = await this.#consume(); + const headers = this.#contentType ? { "content-type": String(this.#contentType) } : undefined; + return await new Response(buf, { headers }).formData(); + } + + // undici's discard idiom: read and drop the body, destroying past the limit. + async dump(opts) { + const limit = opts?.limit ?? 131072; + this.#used = true; + let read = 0; + try { + for await (const chunk of this) { + read += chunk.length; + if (read > limit) { + this.destroy(); + break; + } + } + } catch { + // dump() resolves regardless of how the body ends, like undici + } + } +} + +class Dispatcher extends EventEmitter { + dispatch() { + throw new Error("not implemented"); + } + + close() { + throw new Error("not implemented"); + } + + destroy() { + throw new Error("not implemented"); + } + + request(opts, callback) { + if (callback === undefined) { + return new Promise((resolve, reject) => { + this.request(opts, (err, data) => (err ? reject(err) : resolve(data))); + }); + } + if (typeof callback !== "function") throw new InvalidArgumentError("invalid callback"); + if (!opts || typeof opts !== "object") { + queueMicrotask(() => callback(new InvalidArgumentError("opts must be an object."), { opaque: opts?.opaque })); + return; + } + const opaque = opts.opaque ?? null; + + let body = null; + let resumeBody = null; + let abortBody = null; + let completed = false; + const reqBody = opts.body; + const signal = opts.signal; + let onSignalAbort = null; + const removeSignal = () => { + if (!onSignalAbort) return; + if (typeof signal.removeEventListener === "function") signal.removeEventListener("abort", onSignalAbort); + else if (typeof signal.removeListener === "function") signal.removeListener("abort", onSignalAbort); + onSignalAbort = null; + }; + const destroyRequestBody = err => { + if (reqBody && typeof reqBody.destroy === "function" && !reqBody.destroyed) { + if (typeof reqBody.on === "function") reqBody.on("error", () => {}); + reqBody.destroy(err); + } + }; + if (signal?.aborted) { + const err = signal.reason ?? new RequestAbortedError("Request aborted"); + destroyRequestBody(err); + queueMicrotask(() => callback(err, { opaque })); + return; + } + const trailers = ObjectCreate(null); + let context = null; + try { + this.dispatch(opts, { + onConnect: (abort, ctx) => { + // A late onConnect after a terminal callback must not re-register the signal listener. + if (completed) return; + abortBody = abort; + context = ctx ?? null; + // onConnect may fire once per redirect/retry hop; drop the previous hop's listener first. + removeSignal(); + // Wired here because DispatchOptions has no signal field; user dispatch() implementations never see it. + if (signal) { + onSignalAbort = () => abort(signal.reason); + if (signal.aborted) onSignalAbort(); + else if (typeof signal.addEventListener === "function") + signal.addEventListener("abort", onSignalAbort, { once: true }); + else if (typeof signal.on === "function") signal.on("abort", onSignalAbort); + } + }, + onHeaders: (statusCode, rawHeaders, resume, _statusText) => { + // 1xx informational responses precede the final onHeaders, like undici. + if (statusCode < 200) { + if (typeof opts.onInfo === "function") + opts.onInfo({ statusCode, headers: headersFromRawHeaders(rawHeaders) }); + return true; + } + // onHeaders after a terminal callback or a second final onHeaders violates the contract; ignore it. + if (completed || body !== null) return true; + resumeBody = resume; + const headers = headersFromRawHeaders(rawHeaders); + body = new DispatchBodyReadable( + { + read() { + resumeBody(); + }, + destroy(err, cb) { + // Early body.destroy() cancels the request so the dispatch loop is not left parked. + removeSignal(); + if (!completed) abortBody?.(err ?? undefined); + cb(err); + }, + }, + headers["content-type"], + ); + callback(null, { + statusCode, + headers, + body, + trailers, + opaque, + context, + }); + return true; + }, + // Copy (the dispatch contract only guarantees the chunk during the callback); drop chunks outside onHeaders..onComplete. + onData: chunk => (completed || body === null ? true : body.push(Buffer.from(chunk))), + onComplete: rawTrailers => { + if (completed) return; + completed = true; + removeSignal(); + if (body === null) { + const err = new TypeError( + "request completed without a response: onHeaders must be called before onComplete", + ); + destroyRequestBody(err); + callback(err, { opaque }); + return; + } + if (rawTrailers && rawTrailers.length) Object.assign(trailers, headersFromRawHeaders(rawTrailers)); + body.push(null); + }, + onError: err => { + if (completed) return; + completed = true; + removeSignal(); + destroyRequestBody(err); + if (body) body.destroy(err); + else callback(err, { opaque }); + }, + }); + } catch (err) { + removeSignal(); + destroyRequestBody(err); + if (completed) return; + // Terminal like onError: flips completed so callbacks a throwing dispatch() already scheduled are ignored. + completed = true; + if (body) body.destroy(err); + else callback(err, { opaque }); + } + } +} + +const kDispatch = Symbol("kDispatch"); +const kPending = Symbol("kPending"); + +class DispatcherBase extends Dispatcher { + #closed = false; + #destroyed = false; + + constructor() { + super(); + this[kPending] = new Set(); + } + + get closed() { + return this.#closed; + } + + get destroyed() { + return this.#destroyed; + } + + close(callback) { + if (callback === undefined) { + return new Promise((resolve, reject) => { + this.close((err, data) => (err ? reject(err) : resolve(data))); + }); + } + if (typeof callback !== "function") throw new InvalidArgumentError("invalid callback"); + if (this.#destroyed) { + queueMicrotask(() => callback(new ClientDestroyedError("The client is destroyed"), null)); + return; + } + this.#closed = true; + // Drain, then transition to destroyed, like undici's close().then(() => destroy()). + Promise.allSettled(Array.from(this[kPending], entry => entry.done)).then(() => { + this.#destroyed = true; + queueMicrotask(() => callback(null, null)); + }); + } + + destroy(err, callback) { + if (typeof err === "function") { + callback = err; + err = null; + } + if (callback === undefined) { + return new Promise((resolve, reject) => { + this.destroy(err, (e, data) => (e ? reject(e) : resolve(data))); + }); + } + if (typeof callback !== "function") throw new InvalidArgumentError("invalid callback"); + this.#destroyed = true; + this.#closed = true; + const reason = err ?? new ClientDestroyedError("The client is destroyed"); + for (const entry of this[kPending]) entry.abort(reason); + Promise.allSettled(Array.from(this[kPending], entry => entry.done)).then(() => + queueMicrotask(() => callback(null, null)), + ); + } + + dispatch(opts, handler) { + if (!handler || typeof handler !== "object") throw new InvalidArgumentError("handler must be an object"); + if (typeof handler.onError !== "function" && typeof handler.onResponseError !== "function") { + // Matches undici: without an error callback, async failures would be unobservable. + throw new InvalidArgumentError("invalid onError method"); + } + try { + if (!opts || typeof opts !== "object") throw new InvalidArgumentError("opts must be an object."); + if (this.#destroyed) throw new ClientDestroyedError("The client is destroyed"); + if (this.#closed) throw new ClientClosedError("The client is closed"); + return this[kDispatch](opts, handler); + } catch (err) { + // The guard above proves one of the two callbacks exists. + if (typeof handler.onError === "function") handler.onError(err); + else handler.onResponseError(null, err); + return false; + } + } + + [kDispatch]() { + notImplemented(); + } +} + +class Agent extends DispatcherBase { + constructor(_options) { + super(); + } + + [kDispatch](opts, handler) { + if (!opts.origin) throw new InvalidArgumentError("opts.origin must be a non-empty string or URL."); + return fetchDispatch(parseOrigin(opts.origin), opts, handler, this[kPending]); + } +} + +class Pool extends DispatcherBase { + #origin; + + constructor(origin, _options) { + super(); + if (origin == null) throw new InvalidArgumentError("Origin must be a string or URL."); + this.#origin = parseOrigin(origin); + } + + [kDispatch](opts, handler) { + return fetchDispatch(this.#origin, opts, handler, this[kPending]); + } } -class BalancedPool extends Dispatcher {} -class Client extends Dispatcher { - request() {} + +class BalancedPool extends DispatcherBase { + #upstreams; + + constructor(upstreams = [], _options) { + super(); + this.#upstreams = ($isJSArray(upstreams) ? upstreams : [upstreams]).map(parseOrigin); + } + + [kDispatch](opts, handler) { + const upstream = this.#upstreams[0]; + if (!upstream) throw new BalancedPoolMissingUpstreamError("No upstream has been added to the BalancedPool"); + return fetchDispatch(upstream, opts, handler, this[kPending]); + } } -class DispatcherBase extends EventEmitter {} +class Client extends DispatcherBase { + #origin; + + constructor(origin, _options) { + super(); + if (origin == null) throw new InvalidArgumentError("Origin must be a string or URL."); + this.#origin = parseOrigin(origin); + } + + [kDispatch](opts, handler) { + return fetchDispatch(this.#origin, opts, handler, this[kPending]); + } +} class ProxyAgent extends DispatcherBase { constructor() { @@ -278,9 +845,35 @@ class EnvHttpProxyAgent extends DispatcherBase { } } -class RetryAgent extends Dispatcher { - constructor() { +class RetryAgent extends DispatcherBase { + #agent; + + constructor(agent, _options) { super(); + if (!agent || typeof agent.dispatch !== "function") { + throw new InvalidArgumentError("Argument opts.agent must implement Agent"); + } + this.#agent = agent; + } + + [kDispatch](opts, handler) { + return this.#agent.dispatch(opts, handler); + } + + get closed() { + return this.#agent.closed; + } + + get destroyed() { + return this.#agent.destroyed; + } + + close(callback) { + return this.#agent.close(callback); + } + + destroy(err, callback) { + return this.#agent.destroy(err, callback); } } @@ -316,16 +909,46 @@ class BodyTimeoutError extends UndiciError {} class RequestContentLengthMismatchError extends UndiciError {} class ConnectTimeoutError extends UndiciError {} class ResponseStatusCodeError extends UndiciError {} -class InvalidArgumentError extends UndiciError {} +class InvalidArgumentError extends UndiciError { + constructor(message) { + super(message); + this.name = "InvalidArgumentError"; + this.code = "UND_ERR_INVALID_ARG"; + } +} class InvalidReturnValueError extends UndiciError {} -class RequestAbortedError extends AbortError {} -class ClientDestroyedError extends UndiciError {} -class ClientClosedError extends UndiciError {} +class RequestAbortedError extends AbortError { + constructor(message) { + super(message); + this.name = "AbortError"; + this.code = "UND_ERR_ABORTED"; + } +} +class ClientDestroyedError extends UndiciError { + constructor(message) { + super(message); + this.name = "ClientDestroyedError"; + this.code = "UND_ERR_DESTROYED"; + } +} +class ClientClosedError extends UndiciError { + constructor(message) { + super(message); + this.name = "ClientClosedError"; + this.code = "UND_ERR_CLOSED"; + } +} class InformationalError extends UndiciError {} class SocketError extends UndiciError {} class NotSupportedError extends UndiciError {} class ResponseContentLengthMismatchError extends UndiciError {} -class BalancedPoolMissingUpstreamError extends UndiciError {} +class BalancedPoolMissingUpstreamError extends UndiciError { + constructor(message) { + super(message); + this.name = "MissingUpstreamError"; + this.code = "UND_ERR_BPL_MISSING_UPSTREAM"; + } +} class ResponseExceededMaxSizeError extends UndiciError {} class RequestRetryError extends UndiciError {} class SecureProxyConnectionError extends UndiciError {} @@ -401,14 +1024,19 @@ function serializeAMimeType() { } let globalDispatcher; +let defaultGlobalAgent; // Add missing dispatcher functions function setGlobalDispatcher(dispatcher) { + if (!dispatcher || typeof dispatcher.dispatch !== "function") { + throw new InvalidArgumentError("Argument agent must implement Agent"); + } globalDispatcher = dispatcher; } function getGlobalDispatcher() { - return (globalDispatcher ??= new Dispatcher()); + // The lazy default lives apart from globalDispatcher so calling this never reroutes bare fetch(). + return globalDispatcher ?? (defaultGlobalAgent ??= new Agent()); } // Add missing origin functions @@ -441,6 +1069,183 @@ function buildConnector(_options = {}) { }; } +// fetch with { dispatcher } routes through dispatcher.dispatch(); miniflare relies on this to reach workerd. +function fetchViaDispatcher(dispatcher, input, init) { + let url, method, headers, body, signal, redirect; + // Input parsing failures reject the returned promise like WHATWG fetch, never throw synchronously. + try { + if (input instanceof Request) { + url = new URL(input.url); + method = init.method ?? input.method; + headers = init.headers ?? input.headers; + body = init.body ?? input.body; + signal = init.signal ?? input.signal; + redirect = init.redirect ?? input.redirect ?? "follow"; + } else { + url = input instanceof URL ? input : new URL(String(input)); + method = init.method ?? "GET"; + headers = init.headers; + body = init.body; + signal = init.signal; + redirect = init.redirect ?? "follow"; + } + if (headers instanceof Headers) headers = headers.toJSON(); + method = method ? String(method) : "GET"; + // WHATWG fetch normalizes only these six methods; others keep their case. + const upper = method.toUpperCase(); + if ( + upper === "DELETE" || + upper === "GET" || + upper === "HEAD" || + upper === "OPTIONS" || + upper === "POST" || + upper === "PUT" + ) { + method = upper; + } + // A pre-aborted signal rejects before the dispatcher is ever invoked, like WHATWG fetch. + if (signal?.aborted) throw signal.reason ?? new DOMException("The operation was aborted.", "AbortError"); + } catch (err) { + return Promise.reject(err); + } + + return new Promise((resolve, reject) => { + let resolved = false; + let streamController = null; + let resumeData = null; + let abortDispatch = null; + let onSignalAbort = null; + const removeSignal = () => { + if (!onSignalAbort) return; + if (typeof signal.removeEventListener === "function") signal.removeEventListener("abort", onSignalAbort); + else if (typeof signal.removeListener === "function") signal.removeListener("abort", onSignalAbort); + onSignalAbort = null; + }; + const routeError = err => { + removeSignal(); + if (!resolved) { + resolved = true; + // Null so a late onData cannot park the dispatcher on the orphaned stream's backpressure. + streamController = null; + reject(err); + } else if (streamController) { + streamController.error(err); + streamController = null; + } + }; + try { + dispatcher.dispatch( + { + origin: url.origin, + path: url.pathname + url.search, + method, + headers, + body, + maxRedirections: redirect === "follow" ? 20 : 0, + }, + { + onConnect: abort => { + // A late onConnect after the promise settled must not re-register the signal listener. + if (resolved) return; + abortDispatch = abort; + // onConnect may fire once per redirect/retry hop; drop the previous hop's listener first. + removeSignal(); + // Wired here because DispatchOptions has no signal field; user dispatch() implementations never see it. + if (signal) { + onSignalAbort = () => abort(signal.reason); + if (signal.aborted) onSignalAbort(); + else if (typeof signal.addEventListener === "function") + signal.addEventListener("abort", onSignalAbort, { once: true }); + else if (typeof signal.on === "function") signal.on("abort", onSignalAbort); + } + }, + onHeaders: (statusCode, rawHeaders, resume, statusText) => { + if (statusCode < 200) return true; + // onHeaders after the promise settled violates the dispatch contract; keep the first outcome. + if (resolved) return true; + if ( + redirect === "error" && + (statusCode === 301 || + statusCode === 302 || + statusCode === 303 || + statusCode === 307 || + statusCode === 308) + ) { + const err = new TypeError(`Redirect response '${statusCode}' received when redirect mode is 'error'`); + routeError(err); + abortDispatch?.(err); + return true; + } + resumeData = resume; + const responseHeaders = []; + for (let i = 0; i + 1 < rawHeaders.length; i += 2) { + responseHeaders.push([String(rawHeaders[i]), String(rawHeaders[i + 1])]); + } + let responseBody = null; + const nullBody = statusCode === 204 || statusCode === 205 || statusCode === 304; + if (!nullBody && method !== "HEAD") { + responseBody = new ReadableStream({ + start(controller) { + streamController = controller; + }, + pull() { + resumeData(); + }, + cancel(reason) { + removeSignal(); + streamController = null; + abortDispatch?.(reason instanceof Error ? reason : undefined); + }, + }); + } + // Built before flipping resolved so a non-constructible status rejects; url/redirected stay unset (constructor limitation). + const response = new Response(responseBody, { status: statusCode, statusText, headers: responseHeaders }); + resolved = true; + resolve(response); + return true; + }, + onData: chunk => { + if (!streamController) return true; + // Copy: the dispatch contract only guarantees the chunk during the callback. + streamController.enqueue(new Uint8Array(chunk)); + return streamController.desiredSize > 0; + }, + onComplete: () => { + removeSignal(); + if (!resolved) { + resolved = true; + reject( + new TypeError( + `fetch failed for ${url.origin}: the dispatcher completed without calling onHeaders; onHeaders must be called before onComplete`, + ), + ); + return; + } + streamController?.close(); + // Null so post-onComplete onData/onComplete become no-ops instead of throwing on a closed controller. + streamController = null; + }, + onError: routeError, + }, + ); + } catch (err) { + // A dispatcher that throws synchronously after resolving must error the body, not vanish into the executor. + routeError(err); + } + }); +} + +function fetch(input, init) { + // Raw module-local on purpose: nativeFetch stays the fast path until setGlobalDispatcher() installs one. + const dispatcher = init?.dispatcher ?? globalDispatcher; + if (dispatcher && typeof dispatcher.dispatch === "function") { + return fetchViaDispatcher(dispatcher, input, init ?? kEmptyObject); + } + return nativeFetch(input, init); +} +const { preconnect } = nativeFetch; +if (preconnect) fetch.preconnect = preconnect; + // Update the exports to match the exact structure const moduleExports = { Agent, diff --git a/test/js/first_party/undici/undici.test.ts b/test/js/first_party/undici/undici.test.ts index c966f50b52e7..7e54922639d8 100644 --- a/test/js/first_party/undici/undici.test.ts +++ b/test/js/first_party/undici/undici.test.ts @@ -1,6 +1,18 @@ import { afterAll, beforeAll, describe, expect, it } from "bun:test"; import { Readable } from "node:stream"; -import { request, fetch as undiciFetch } from "undici"; +import { + Agent, + Client, + Dispatcher, + Pool, + RetryAgent, + errors, + getGlobalDispatcher, + request, + fetch as undiciFetch, +} from "undici"; + +import { bunEnv, bunExe } from "harness"; import { createServer } from "../../../http-test-server"; @@ -215,6 +227,1221 @@ describe("undici", () => { // expect(json.form.foo).toBe("bar"); // }); }); + + describe("Dispatcher", () => { + // Drives dispatch() with the legacy handler interface and collects the response. + function dispatchLegacy(dispatcher: any, opts: any) { + return new Promise<{ statusCode: number; headers: Record; body: string }>((resolve, reject) => { + let statusCode = 0; + const headers: Record = {}; + const chunks: Buffer[] = []; + dispatcher.dispatch(opts, { + onConnect: () => {}, + onHeaders: (status: number, rawHeaders: Buffer[]) => { + statusCode = status; + for (let i = 0; i + 1 < rawHeaders.length; i += 2) { + headers[String(rawHeaders[i]).toLowerCase()] = String(rawHeaders[i + 1]); + } + return true; + }, + onData: (chunk: Buffer) => { + chunks.push(chunk); + return true; + }, + onComplete: () => resolve({ statusCode, headers, body: Buffer.concat(chunks).toString() }), + onError: reject, + }); + }); + } + + it("Pool exposes dispatch(), close() and destroy()", () => { + const pool = new Pool(hostUrl); + expect(typeof pool.dispatch).toBe("function"); + expect(typeof pool.close).toBe("function"); + expect(typeof pool.destroy).toBe("function"); + expect(typeof pool.request).toBe("function"); + }); + + // Resolving null is what upstream undici's close()/destroy() resolve to. + it("Agent.close() resolves null", async () => { + const agent = new Agent(); + expect(typeof agent.close).toBe("function"); + expect(await agent.close()).toBe(null); + }); + + it("Agent.destroy() resolves null", async () => { + const agent = new Agent(); + expect(typeof agent.destroy).toBe("function"); + expect(await agent.destroy()).toBe(null); + }); + + it("Pool.close() and destroy() resolve null", async () => { + const pool = new Pool(hostUrl); + expect(await pool.close()).toBe(null); + expect(await pool.destroy()).toBe(null); + }); + + it("Client.close() and destroy() resolve null", async () => { + const client = new Client(hostUrl); + expect(typeof client.close).toBe("function"); + expect(typeof client.destroy).toBe("function"); + expect(await client.close()).toBe(null); + expect(await client.destroy()).toBe(null); + }); + + it("Pool.dispatch performs a request with the legacy handler interface", async () => { + const pool = new Pool(hostUrl); + const res = await dispatchLegacy(pool, { path: "/get", method: "GET" }); + expect(res.statusCode).toBe(200); + expect(res.headers["content-type"]).toBe("application/json"); + expect(JSON.parse(res.body)).toEqual({ url: `${hostUrl}/get`, method: "GET" }); + await pool.close(); + }); + + it("Pool.dispatch performs a request with the controller handler interface", async () => { + const pool = new Pool(hostUrl); + const res = await new Promise<{ statusCode: number; headers: any; body: string; ended: boolean }>( + (resolve, reject) => { + let statusCode = 0; + let headers: any; + let started = false; + const chunks: Buffer[] = []; + pool.dispatch( + { path: "/post", method: "POST", body: "Hello world" }, + { + onRequestStart: () => { + started = true; + }, + onResponseStart: (_controller: any, status: number, responseHeaders: any) => { + statusCode = status; + headers = responseHeaders; + }, + onResponseData: (_controller: any, chunk: Buffer) => { + chunks.push(chunk); + }, + onResponseEnd: () => { + resolve({ statusCode, headers, body: Buffer.concat(chunks).toString(), ended: started }); + }, + onResponseError: (_controller: any, err: Error) => reject(err), + }, + ); + }, + ); + expect(res.ended).toBe(true); + expect(res.statusCode).toBe(201); + expect(res.headers["content-type"]).toBe("application/json"); + expect((JSON.parse(res.body) as { data: string }).data).toBe("Hello world"); + await pool.destroy(); + }); + + it("Pool.request body exposes the undici body mixin", async () => { + const pool = new Pool(hostUrl); + const { statusCode, body } = await pool.request({ path: "/get", method: "GET" }); + expect(statusCode).toBe(200); + expect(body.bodyUsed).toBe(false); + expect(await body.json()).toEqual({ url: `${hostUrl}/get`, method: "GET" }); + expect(body.bodyUsed).toBe(true); + await expect(body.json()).rejects.toThrow("unusable"); + await pool.close(); + }); + + it("request() body reports bodyUsed after direct iteration", async () => { + const pool = new Pool(hostUrl); + const { body } = await pool.request({ path: "/get", method: "GET" }); + for await (const chunk of body) { + // drain directly instead of via the mixin + } + expect(body.bodyUsed).toBe(true); + await expect(body.text()).rejects.toThrow("unusable"); + await pool.close(); + }); + + it("Pool.request honors opts.signal", async () => { + await using server = Bun.serve({ + port: 0, + fetch() { + // Never respond; the request can only finish by being aborted. + return new Promise(() => {}); + }, + }); + const pool = new Pool(`http://localhost:${server.port}`); + const ac = new AbortController(); + const pending = pool.request({ path: "/", method: "GET", signal: ac.signal }); + ac.abort(); + // The signal's reason (a DOMException named AbortError) propagates, like undici. + await expect(pending).rejects.toHaveProperty("name", "AbortError"); + await pool.destroy(); + }); + + it("request rejects a pre-aborted signal without dispatching", async () => { + let dispatched = false; + const dispatcher = new (class extends Dispatcher { + dispatch() { + dispatched = true; + return true; + } + })(); + const ac = new AbortController(); + ac.abort(); + await expect( + dispatcher.request({ origin: "http://localhost:1", path: "/", method: "GET", signal: ac.signal }), + ).rejects.toHaveProperty("name", "AbortError"); + expect(dispatched).toBe(false); + }); + + it("fetch with a dispatcher rejects a pre-aborted signal without dispatching", async () => { + let dispatched = false; + const dispatcher = { + dispatch() { + dispatched = true; + return true; + }, + }; + const ac = new AbortController(); + ac.abort(); + await expect(undiciFetch("http://localhost:1/", { dispatcher, signal: ac.signal } as any)).rejects.toHaveProperty( + "name", + "AbortError", + ); + expect(dispatched).toBe(false); + }); + + it("a throwing onHeaders routes the error to onError", async () => { + const pool = new Pool(hostUrl); + const boom = new Error("bad status"); + const err = await new Promise((resolve, reject) => { + pool.dispatch( + { path: "/get", method: "GET" }, + { + onConnect: () => {}, + onHeaders: () => { + throw boom; + }, + onData: () => reject(new Error("should not receive data")), + onComplete: () => reject(new Error("should not complete")), + onError: resolve, + }, + ); + }); + expect(err).toBe(boom); + await pool.close(); + }); + + it("Pool.request resolves with a readable body", async () => { + const pool = new Pool(hostUrl); + const { statusCode, headers, body } = await pool.request({ path: "/get", method: "GET" }); + expect(statusCode).toBe(200); + expect(headers["content-type"]).toBe("application/json"); + const chunks: Buffer[] = []; + for await (const chunk of body) chunks.push(chunk); + expect(JSON.parse(Buffer.concat(chunks).toString())).toEqual({ url: `${hostUrl}/get`, method: "GET" }); + await pool.close(); + }); + + it("Client.request sends a request body", async () => { + const client = new Client(hostUrl); + const { statusCode, body } = await client.request({ path: "/post", method: "POST", body: "ping" }); + expect(statusCode).toBe(201); + const chunks: Buffer[] = []; + for await (const chunk of body) chunks.push(chunk); + expect((JSON.parse(Buffer.concat(chunks).toString()) as { data: string }).data).toBe("ping"); + await client.close(); + }); + + it("a completed close() transitions to destroyed, like undici", async () => { + const pool = new Pool(hostUrl); + await pool.close(); + expect(pool.closed).toBe(true); + expect(pool.destroyed).toBe(true); + await expect(dispatchLegacy(pool, { path: "/get", method: "GET" })).rejects.toHaveProperty( + "code", + "UND_ERR_DESTROYED", + ); + await expect(pool.request({ path: "/get", method: "GET" })).rejects.toBeInstanceOf(errors.ClientDestroyedError); + }); + + it("destroy() resolves and later requests fail with ClientDestroyedError", async () => { + const pool = new Pool(hostUrl); + await pool.destroy(); + expect(pool.destroyed).toBe(true); + await expect(pool.request({ path: "/get", method: "GET" })).rejects.toHaveProperty("code", "UND_ERR_DESTROYED"); + }); + + it("dispatch without an error callback throws synchronously", () => { + const pool = new Pool(hostUrl); + expect(() => + pool.dispatch({ path: "/get", method: "GET" }, { + onHeaders: () => true, + onData: () => {}, + onComplete: () => {}, + } as any), + ).toThrow(errors.InvalidArgumentError); + }); + + it("dispatch keeps '//' paths on the configured origin", async () => { + const pool = new Pool(hostUrl); + const res = await dispatchLegacy(pool, { path: "//evil.example/x", method: "GET" }); + expect(res.statusCode).toBe(404); + expect(JSON.parse(res.body).url).toStartWith(hostUrl); + await pool.close(); + }); + + it("dispatch rejects opts.query combined with a path that already has one", async () => { + const pool = new Pool(hostUrl); + await expect(dispatchLegacy(pool, { path: "/get?a=1", method: "GET", query: { b: "2" } })).rejects.toHaveProperty( + "code", + "UND_ERR_INVALID_ARG", + ); + await pool.close(); + }); + + it("dispatch rejects paths that do not start with '/'", async () => { + const pool = new Pool(hostUrl); + await expect(dispatchLegacy(pool, { path: "http://evil.example/x", method: "GET" })).rejects.toHaveProperty( + "code", + "UND_ERR_INVALID_ARG", + ); + await pool.close(); + }); + + it("streams a node Readable request body", async () => { + const client = new Client(hostUrl); + const { statusCode, body } = await client.request({ + path: "/post", + method: "POST", + body: Readable.from(["pi", "ng"]), + }); + expect(statusCode).toBe(201); + expect(((await body.json()) as { data: string }).data).toBe("ping"); + await client.close(); + }); + + it("body.destroy() cancels the underlying request", async () => { + const cancelled = Promise.withResolvers(); + await using server = Bun.serve({ + port: 0, + fetch() { + const stream = new ReadableStream({ + pull(controller) { + controller.enqueue(new Uint8Array(1024)); + }, + cancel() { + cancelled.resolve(); + }, + }); + return new Response(stream, { headers: { "content-type": "application/octet-stream" } }); + }, + }); + const pool = new Pool(`http://localhost:${server.port}`); + const { body } = await pool.request({ path: "/", method: "GET" }); + // Read one chunk, then bail out; breaking destroys the body stream. + for await (const chunk of body) break; + await cancelled.promise; + await pool.destroy(); + }); + + it("close() waits for in-flight requests to finish", async () => { + const gate = Promise.withResolvers(); + await using server = Bun.serve({ + port: 0, + async fetch() { + await gate.promise; + return new Response("done"); + }, + }); + const pool = new Pool(`http://localhost:${server.port}`); + const events: string[] = []; + const completed = new Promise((resolve, reject) => { + pool.dispatch( + { path: "/", method: "GET" }, + { + onConnect: () => {}, + onHeaders: () => true, + onData: () => true, + onComplete: () => { + events.push("complete"); + resolve(); + }, + onError: reject, + }, + ); + }); + const closed = pool.close().then(() => { + events.push("closed"); + }); + gate.resolve(); + await Promise.all([completed, closed]); + expect(events).toEqual(["complete", "closed"]); + }); + + it("destroy() aborts in-flight requests with ClientDestroyedError", async () => { + await using server = Bun.serve({ + port: 0, + fetch() { + return new Promise(() => {}); + }, + }); + const pool = new Pool(`http://localhost:${server.port}`); + const errPromise = new Promise((resolve, reject) => { + pool.dispatch( + { path: "/", method: "GET" }, + { + onConnect: () => {}, + onHeaders: () => reject(new Error("should not receive headers")), + onData: () => {}, + onComplete: () => reject(new Error("should not complete")), + onError: resolve, + }, + ); + }); + await pool.destroy(); + const err = await errPromise; + expect(err.code).toBe("UND_ERR_DESTROYED"); + }); + + it("abort() from onData on the final chunk delivers onError, not onComplete", async () => { + const pool = new Pool(hostUrl); + let abortFn: ((reason?: Error) => void) | undefined; + const err = await new Promise((resolve, reject) => { + pool.dispatch( + { path: "/get", method: "GET" }, + { + onConnect: (abort: (reason?: Error) => void) => { + abortFn = abort; + }, + onHeaders: () => true, + onData: () => { + abortFn!(); + return true; + }, + onComplete: () => reject(new Error("should not complete")), + onError: resolve, + }, + ); + }); + expect(err.code).toBe("UND_ERR_ABORTED"); + await pool.destroy(); + }); + + it("fetch routes through init.dispatcher like miniflare", async () => { + await using target = Bun.serve({ + port: 0, + fetch: req => new Response("routed:" + new URL(req.url).pathname), + }); + const pool = new Pool(`http://localhost:${target.port}`); + let dispatched = 0; + // miniflare's pattern: a custom dispatcher that rewrites every request + // into its own Pool, ignoring the URL's authority. + const dispatcher = { + dispatch(opts: any, handler: any) { + dispatched++; + return pool.dispatch(opts, handler); + }, + }; + // Nothing listens on the URL's port; only dispatcher routing can answer. + const res = await undiciFetch("http://localhost:1/test?q=1", { dispatcher } as any); + expect(res.status).toBe(200); + expect(await res.text()).toBe("routed:/test"); + expect(dispatched).toBe(1); + await pool.close(); + }); + + it("fetch with init.dispatcher sends the request body", async () => { + await using target = Bun.serve({ + port: 0, + fetch: async req => new Response("echo:" + (await req.text())), + }); + const pool = new Pool(`http://localhost:${target.port}`); + const dispatcher = { + dispatch: (opts: any, handler: any) => pool.dispatch(opts, handler), + }; + const res = await undiciFetch("http://localhost:1/", { + method: "POST", + body: "hello", + dispatcher, + } as any); + expect(await res.text()).toBe("echo:hello"); + await pool.close(); + }); + + it("async failures reach onResponseError on legacy-shaped handlers without onError", async () => { + const pool = new Pool("http://127.0.0.1:1"); + const err = await new Promise((resolve, reject) => { + pool.dispatch( + { path: "/", method: "GET" }, + { + onHeaders: () => reject(new Error("should not receive headers")), + onData: () => {}, + onComplete: () => reject(new Error("should not complete")), + onResponseError: (_controller: any, e: Error) => resolve(e), + }, + ); + }); + expect(err.code).toBe("ConnectionRefused"); + await pool.destroy(); + }); + + it("request() destroys a stream body when dispatch fails", async () => { + const pool = new Pool(hostUrl); + await pool.close(); + const reqBody = Readable.from(["x"]); + await expect(pool.request({ path: "/post", method: "POST", body: reqBody })).rejects.toBeInstanceOf( + errors.ClientDestroyedError, + ); + expect(reqBody.destroyed).toBe(true); + }); + + it("request({ signal }) cancels through a user Dispatcher subclass", async () => { + class NeverResponds extends Dispatcher { + dispatch(_opts: any, handler: any) { + // A compliant dispatch() never reads opts.signal; it only hands out abort. + handler.onConnect((reason?: Error) => handler.onError(reason ?? new Error("aborted"))); + return true; + } + } + const ac = new AbortController(); + const pending = new NeverResponds().request({ path: "/", method: "GET", signal: ac.signal }); + ac.abort(); + await expect(pending).rejects.toHaveProperty("name", "AbortError"); + }); + + it("request() exposes trailers and context from the dispatcher", async () => { + class WithTrailers extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}, { some: "context" }); + handler.onHeaders(200, [], () => {}, "OK"); + handler.onData(Buffer.from("hi")); + handler.onComplete([Buffer.from("x-foo"), Buffer.from("bar")]); + return true; + } + } + const { trailers, context, body } = await new WithTrailers().request({ path: "/", method: "GET" }); + expect(await body.text()).toBe("hi"); + expect(context).toEqual({ some: "context" }); + expect(trailers).toEqual({ "x-foo": "bar" }); + }); + + it("request(cb) does not invoke a throwing callback twice", async () => { + const pool = new Pool(hostUrl); + await pool.close(); + let calls = 0; + expect(() => + pool.request({ path: "/", method: "GET" }, () => { + calls++; + throw new Error("user callback threw"); + }), + ).not.toThrow(); + expect(calls).toBe(1); + }); + + it("request({ signal }) registers a single abort listener", async () => { + const pool = new Pool(hostUrl); + let adds = 0; + const signal = { + aborted: false, + on: () => { + adds++; + }, + removeListener: () => {}, + }; + const { body } = await pool.request({ path: "/get", method: "GET", signal: signal as any }); + await body.text(); + expect(adds).toBe(1); + await pool.close(); + }); + + it("fetch with dispatcher normalizes only the WHATWG methods", async () => { + const seen: string[] = []; + const dispatcher = { + dispatch(opts: any, handler: any) { + seen.push(opts.method); + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + handler.onComplete([]); + return true; + }, + }; + await undiciFetch("http://localhost:1/", { method: "get", dispatcher } as any); + await undiciFetch("http://localhost:1/", { method: "patch", dispatcher } as any); + expect(seen).toEqual(["GET", "patch"]); + }); + + it("fetch honors redirect 'error' set on a Request input", async () => { + await using target = Bun.serve({ + port: 0, + fetch: () => new Response(null, { status: 302, headers: { location: "/next" } }), + }); + const pool = new Pool(`http://localhost:${target.port}`); + const dispatcher = { dispatch: (opts: any, handler: any) => pool.dispatch(opts, handler) }; + const req = new Request("http://localhost:1/", { redirect: "error" }); + await expect(undiciFetch(req, { dispatcher } as any)).rejects.toBeInstanceOf(TypeError); + await pool.destroy(); + }); + + it("getGlobalDispatcher() does not reroute bare fetch", async () => { + await using server = Bun.serve({ + port: 0, + fetch: () => new Response("native"), + }); + getGlobalDispatcher(); + const res = await undiciFetch(`http://localhost:${server.port}/x`); + // nativeFetch populates res.url; the shim dispatcher path cannot. + expect(res.url).toBe(`http://localhost:${server.port}/x`); + expect(await res.text()).toBe("native"); + }); + + it("request(cb) delivers opaque on the error path", async () => { + const pool = new Pool(hostUrl); + await pool.close(); + const { promise, resolve } = Promise.withResolvers<{ err: any; data: any }>(); + pool.request({ path: "/", method: "GET", opaque: { reqId: 42 } }, (err: any, data: any) => + resolve({ err, data }), + ); + const { err, data } = await promise; + expect(err.code).toBe("UND_ERR_DESTROYED"); + expect(data.opaque).toEqual({ reqId: 42 }); + }); + + it("request() rejects when the dispatcher completes without a response", async () => { + class CompletesEarly extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + // A stray chunk before onHeaders is dropped rather than crashing the handler. + handler.onData(Buffer.from("stray")); + handler.onComplete([]); + return true; + } + } + const reqBody = Readable.from(["x"]); + await expect(new CompletesEarly().request({ path: "/", method: "POST", body: reqBody })).rejects.toThrow( + "onHeaders must be called before onComplete", + ); + expect(reqBody.destroyed).toBe(true); + }); + + it("fetch rejects when the dispatcher completes without a response", async () => { + const dispatcher = { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onComplete([]); + return true; + }, + }; + await expect(undiciFetch("http://localhost:1/", { dispatcher } as any)).rejects.toThrow( + "fetch failed for http://localhost:1: the dispatcher completed without calling onHeaders", + ); + }); + + it("request() keeps the first body when onHeaders fires twice", async () => { + class DoubleHeaders extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + // A second final onHeaders violates the contract; the first body must keep receiving data. + handler.onHeaders(500, [], () => {}, "ERR"); + handler.onData(Buffer.from("hi")); + handler.onComplete([]); + return true; + } + } + const { statusCode, body } = await new DoubleHeaders().request({ path: "/", method: "GET" }); + expect(statusCode).toBe(200); + expect(await body.text()).toBe("hi"); + }); + + it("fetch keeps the first response when onHeaders fires twice", async () => { + const dispatcher = { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + handler.onHeaders(500, [], () => {}, "ERR"); + handler.onData(Buffer.from("hi")); + handler.onComplete([]); + return true; + }, + }; + const res = await undiciFetch("http://localhost:1/", { dispatcher } as any); + expect(res.status).toBe(200); + expect(await res.text()).toBe("hi"); + }); + + it("fetch with redirect 'error' ignores a late redirect onHeaders", async () => { + const dispatcher = { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + // A contract-violating second onHeaders must not error the delivered 200 body. + handler.onHeaders(302, [], () => {}, "Found"); + handler.onData(Buffer.from("hi")); + handler.onComplete([]); + return true; + }, + }; + const res = await undiciFetch("http://localhost:1/", { dispatcher, redirect: "error" } as any); + expect(res.status).toBe(200); + expect(await res.text()).toBe("hi"); + }); + + it("fetch ignores dispatcher callbacks after onComplete", async () => { + let lateError: unknown = null; + const dispatcher = { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + handler.onData(Buffer.from("hi")); + handler.onComplete([]); + // Late callbacks must be no-ops, not TypeErrors thrown back into the dispatcher. + try { + handler.onData(Buffer.from("late")); + handler.onComplete([]); + } catch (err) { + lateError = err; + } + return true; + }, + }; + const res = await undiciFetch("http://localhost:1/", { dispatcher } as any); + expect(await res.text()).toBe("hi"); + expect(lateError).toBe(null); + }); + + it("request(cb) ignores a terminal callback after onError", async () => { + const boom = new Error("boom"); + class TwoTerminals extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onError(boom); + // A second terminal callback must not invoke the user callback again. + handler.onComplete([]); + return true; + } + } + const calls: any[] = []; + await new Promise(resolve => { + new TwoTerminals().request({ path: "/", method: "GET" }, (err: any) => { + calls.push(err); + resolve(); + }); + }); + expect(calls).toEqual([boom]); + }); + + it("request() ignores onData after onComplete", async () => { + class LateData extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + handler.onData(Buffer.from("hi")); + handler.onComplete([]); + // A late chunk must not push after EOF and error the delivered body. + handler.onData(Buffer.from("late")); + return true; + } + } + const { body } = await new LateData().request({ path: "/", method: "GET" }); + expect(await body.text()).toBe("hi"); + }); + + it("request() ignores onConnect after a terminal callback", async () => { + let adds = 0; + let removes = 0; + const signal = { + aborted: false, + on: () => { + adds++; + }, + removeListener: () => { + removes++; + }, + }; + class LateConnect extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onError(new Error("boom")); + // A late onConnect must not re-register the abort listener with no cleanup path left. + handler.onConnect(() => {}); + return true; + } + } + await expect(new LateConnect().request({ path: "/", method: "GET", signal: signal as any })).rejects.toThrow( + "boom", + ); + expect(adds).toBe(removes); + }); + + it("fetch ignores onConnect after the dispatch settled", async () => { + let adds = 0; + let removes = 0; + const signal = { + aborted: false, + addEventListener: () => { + adds++; + }, + removeEventListener: () => { + removes++; + }, + }; + const dispatcher = { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onError(new Error("boom")); + handler.onConnect(() => {}); + return true; + }, + }; + await expect(undiciFetch("http://localhost:1/", { dispatcher, signal } as any)).rejects.toThrow("boom"); + expect(adds).toBe(removes); + }); + + it("request() ignores onHeaders after onError", async () => { + const boom = new Error("boom"); + class ErrorsFirst extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onError(boom); + handler.onHeaders(200, [], () => {}, "OK"); + return true; + } + } + const calls: any[] = []; + await new Promise(resolve => { + new ErrorsFirst().request({ path: "/", method: "GET" }, (err: any) => { + calls.push(err); + resolve(); + }); + }); + expect(calls).toEqual([boom]); + }); + + it("fetch with dispatcher rejects invalid URLs instead of throwing", async () => { + const dispatcher = { dispatch: () => true }; + await expect(undiciFetch("not a url", { dispatcher } as any)).rejects.toBeInstanceOf(TypeError); + }); + + it("fetch routes through the global dispatcher when none is passed", async () => { + // Spawned so installing a global dispatcher cannot leak into other tests. + await using target = Bun.serve({ + port: 0, + fetch: () => new Response("via-global"), + }); + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const { Pool, setGlobalDispatcher, fetch } = require("undici"); + const pool = new Pool("http://localhost:${target.port}"); + setGlobalDispatcher({ dispatch: (opts, handler) => pool.dispatch(opts, handler) }); + const res = await fetch("http://localhost:1/x"); + console.log(await res.text()); + await pool.close();`, + ], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("via-global\n"); + expect(exitCode).toBe(0); + }); + + it("request() body formData() parses using the response content-type", async () => { + const fd = new FormData(); + fd.append("foo", "bar"); + const encoded = new Response(fd); + const contentType = encoded.headers.get("content-type")!; + const bytes = await encoded.bytes(); + class FormDataDispatcher extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [Buffer.from("content-type"), Buffer.from(contentType)], () => {}, "OK"); + handler.onData(Buffer.from(bytes)); + handler.onComplete([]); + return true; + } + } + const { body } = await new FormDataDispatcher().request({ path: "/", method: "GET" }); + const parsed = await body.formData(); + expect(parsed.get("foo")).toBe("bar"); + }); + + it("request() body dump() discards the body", async () => { + const pool = new Pool(hostUrl); + const { body } = await pool.request({ path: "/get", method: "GET" }); + await body.dump(); + expect(body.bodyUsed).toBe(true); + await expect(body.text()).rejects.toThrow("unusable"); + await pool.close(); + }); + + it("constructors reject origins carrying a path, query, or hash", () => { + expect(() => new Pool("http://localhost:3000/api")).toThrow(errors.InvalidArgumentError); + expect(() => new Client("http://localhost:3000/?q=1")).toThrow(errors.InvalidArgumentError); + }); + + it("constructors reject non-http(s) origins", () => { + expect(() => new Pool("ws://localhost:3000")).toThrow(errors.InvalidArgumentError); + expect(() => new Client("file:///tmp")).toThrow(errors.InvalidArgumentError); + }); + + it("fetch with dispatcher copies body chunks from the dispatcher", async () => { + const buf = Buffer.alloc(4); + const dispatcher = { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + buf.write("AAAA"); + handler.onData(buf); + buf.write("BBBB"); + handler.onData(buf); + handler.onComplete([]); + return true; + }, + }; + const res = await undiciFetch("http://localhost:1/", { dispatcher } as any); + expect(await res.text()).toBe("AAAABBBB"); + }); + + it("request() body blob() carries the response content-type", async () => { + const pool = new Pool(hostUrl); + const { body } = await pool.request({ path: "/get", method: "GET" }); + const blob = await body.blob(); + expect(blob.type).toContain("application/json"); + await pool.close(); + }); + + it("a lowercase head method keeps content-length in the handler headers", async () => { + const pool = new Pool(hostUrl); + const res = await dispatchLegacy(pool, { path: "/head", method: "head" }); + expect(res.statusCode).toBe(200); + expect(res.headers["content-length"]).toBeDefined(); + await pool.close(); + }); + + it("repeated onConnect keeps signal listeners balanced", async () => { + let adds = 0; + let removes = 0; + const signal = { + aborted: false, + on: () => { + adds++; + }, + removeListener: () => { + removes++; + }, + }; + class TwoHops extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + handler.onData(Buffer.from("hi")); + handler.onComplete([]); + return true; + } + } + const { body } = await new TwoHops().request({ path: "/", method: "GET", signal: signal as any }); + await body.text(); + expect(adds).toBe(2); + expect(removes).toBe(2); + }); + + it("request() copies body chunks from the dispatcher", async () => { + const buf = Buffer.alloc(4); + class ReusesBuffer extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + buf.write("AAAA"); + handler.onData(buf); + buf.write("BBBB"); + handler.onData(buf); + handler.onComplete([]); + return true; + } + } + const { body } = await new ReusesBuffer().request({ path: "/", method: "GET" }); + expect(await body.text()).toBe("AAAABBBB"); + }); + + it("RetryAgent.close() closes the wrapped dispatcher", async () => { + const agent = new Agent(); + const retry = new RetryAgent(agent); + await retry.close(); + expect(agent.closed).toBe(true); + expect(retry.closed).toBe(true); + }); + + it("close() called from onConnect still waits for the request", async () => { + const pool = new Pool(hostUrl); + const events: string[] = []; + let closed: Promise | undefined; + const completed = new Promise((resolve, reject) => { + pool.dispatch( + { path: "/get", method: "GET" }, + { + onConnect: () => { + closed = pool.close().then(() => events.push("closed")); + }, + onHeaders: () => true, + onData: () => true, + onComplete: () => { + events.push("complete"); + resolve(); + }, + onError: reject, + }, + ); + }); + await Promise.all([completed, closed!]); + expect(events).toEqual(["complete", "closed"]); + }); + + it("request() skips 1xx informational responses", async () => { + class Informational extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(100, [], () => {}, "Continue"); + handler.onHeaders(200, [Buffer.from("content-type"), Buffer.from("text/plain")], () => {}, "OK"); + handler.onData(Buffer.from("hi")); + handler.onComplete([]); + return true; + } + } + const infos: number[] = []; + const { statusCode, body } = await new Informational().request({ + path: "/", + method: "GET", + onInfo: (info: { statusCode: number }) => infos.push(info.statusCode), + }); + expect(statusCode).toBe(200); + expect(await body.text()).toBe("hi"); + expect(infos).toEqual([100]); + }); + + it("fetch({ dispatcher, signal }) aborts through the handler", async () => { + const dispatcher = { + dispatch(_opts: any, handler: any) { + // A compliant dispatch() never reads opts.signal; it only hands out abort. + handler.onConnect((reason?: Error) => handler.onError(reason ?? new Error("aborted"))); + return true; + }, + }; + const ac = new AbortController(); + const pending = undiciFetch("http://localhost:1/", { dispatcher, signal: ac.signal } as any); + ac.abort(); + await expect(pending).rejects.toHaveProperty("name", "AbortError"); + }); + + it("fetch with dispatcher and redirect 'error' rejects on redirects", async () => { + await using target = Bun.serve({ + port: 0, + fetch: () => new Response(null, { status: 302, headers: { location: "/next" } }), + }); + const pool = new Pool(`http://localhost:${target.port}`); + const dispatcher = { dispatch: (opts: any, handler: any) => pool.dispatch(opts, handler) }; + await expect(undiciFetch("http://localhost:1/", { dispatcher, redirect: "error" } as any)).rejects.toBeInstanceOf( + TypeError, + ); + await pool.destroy(); + }); + + it("fetch with dispatcher rejects on a non-constructible status instead of hanging", async () => { + const dispatcher = { + dispatch(_opts: any, handler: any) { + // Route the onHeaders throw through onError, like the builtin dispatchers do. + queueMicrotask(() => { + try { + handler.onHeaders(600, [], () => {}, "Weird"); + } catch (e) { + handler.onError(e); + } + }); + handler.onConnect(() => {}); + return true; + }, + }; + await expect(undiciFetch("http://localhost:1/", { dispatcher } as any)).rejects.toBeInstanceOf(RangeError); + }); + + it("fetch does not park the dispatcher on onData after onHeaders threw", async () => { + const dataReturns: any[] = []; + const dispatcher = { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + try { + handler.onHeaders(600, [], () => {}, "Weird"); + } catch (e) { + handler.onError(e); + } + // Late chunks must be dropped, not enqueued into the orphaned stream until backpressure pauses us. + dataReturns.push(handler.onData(Buffer.from("x"))); + dataReturns.push(handler.onData(Buffer.from("y"))); + return true; + }, + }; + await expect(undiciFetch("http://localhost:1/", { dispatcher } as any)).rejects.toBeInstanceOf(RangeError); + expect(dataReturns).toEqual([true, true]); + }); + + it("dispatch sends headers given as Headers or Map instances", async () => { + const received: (string | null)[] = []; + await using server = Bun.serve({ + port: 0, + fetch(req) { + received.push(req.headers.get("authorization")); + return new Response("ok"); + }, + }); + const pool = new Pool(`http://localhost:${server.port}`); + await dispatchLegacy(pool, { path: "/", method: "GET", headers: new Headers({ authorization: "Bearer a" }) }); + await dispatchLegacy(pool, { path: "/", method: "GET", headers: new Map([["authorization", "Bearer b"]]) }); + expect(received).toEqual(["Bearer a", "Bearer b"]); + await pool.close(); + }); + + it("dispatch sends array-valued request headers as separate lines", async () => { + const seen = Promise.withResolvers(); + await using server = Bun.serve({ + port: 0, + fetch(req) { + seen.resolve(req.headers.get("cookie")); + return new Response("ok"); + }, + }); + const pool = new Pool(`http://localhost:${server.port}`); + await dispatchLegacy(pool, { path: "/", method: "GET", headers: { cookie: ["a=1", "b=2"] } }); + // Separate Cookie lines combine with '; ' like undici; a record init would send 'a=1,b=2'. + expect(await seen.promise).toBe("a=1; b=2"); + await pool.close(); + }); + + it("request(cb) ignores callbacks scheduled by a throwing dispatch()", async () => { + class ThrowsThenCallsBack extends Dispatcher { + dispatch(_opts: any, handler: any) { + queueMicrotask(() => { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + handler.onComplete([]); + }); + throw new Error("sync boom"); + } + } + const calls: any[] = []; + new ThrowsThenCallsBack().request({ path: "/", method: "GET" }, (err: any) => { + calls.push(err?.message); + }); + // Let the scheduled callbacks run before asserting the callback fired exactly once. + await new Promise(resolve => queueMicrotask(() => queueMicrotask(resolve))); + expect(calls).toEqual(["sync boom"]); + }); + + it("fetch with dispatcher errors the body when dispatch throws after headers", async () => { + const boom = new Error("post-headers boom"); + const dispatcher = { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onHeaders(200, [], () => {}, "OK"); + throw boom; + }, + }; + const res = await undiciFetch("http://localhost:1/", { dispatcher } as any); + expect(res.status).toBe(200); + await expect(res.text()).rejects.toBe(boom); + }); + + it("close(callback) invokes the callback", async () => { + const pool = new Pool(hostUrl); + const { promise, resolve, reject } = Promise.withResolvers(); + pool.close((err: Error | null) => (err ? reject(err) : resolve())); + await promise; + expect(pool.closed).toBe(true); + }); + + it("abort() while the body is paused delivers onError instead of hanging", async () => { + let pulls = 0; + await using server = Bun.serve({ + port: 0, + fetch() { + const stream = new ReadableStream({ + pull(controller) { + pulls++; + controller.enqueue(new Uint8Array(1024)); + }, + }); + return new Response(stream, { headers: { "content-type": "application/octet-stream" } }); + }, + }); + const pool = new Pool(`http://localhost:${server.port}`); + let abortFn: ((reason?: Error) => void) | undefined; + const err = await new Promise((resolve, reject) => { + pool.dispatch( + { path: "/", method: "GET" }, + { + onConnect: (abort: (reason?: Error) => void) => { + abortFn = abort; + }, + onHeaders: () => true, + // Pause after the first chunk so the body loop parks. + onData: () => false, + onComplete: () => reject(new Error("should not complete")), + onError: resolve, + }, + ); + (async () => { + // Wait until more chunks are in flight, so the paused loop is parked + // holding an undelivered chunk, then abort. + const deadline = Date.now() + 5_000; + while (pulls < 3 && Date.now() < deadline) await Bun.sleep(5); + if (pulls < 3) { + reject(new Error(`body never parked while paused; pulls=${pulls}`)); + return; + } + abortFn!(); + })(); + }); + expect(err.code).toBe("UND_ERR_ABORTED"); + await pool.destroy(); + }); + + it("aborting from onConnect rejects with UND_ERR_ABORTED", async () => { + const pool = new Pool(hostUrl); + const err = await new Promise((resolve, reject) => { + pool.dispatch( + { path: "/get", method: "GET" }, + { + onConnect: (abort: (reason?: Error) => void) => abort(), + onHeaders: () => reject(new Error("should not receive headers")), + onData: () => {}, + onComplete: () => reject(new Error("should not complete")), + onError: resolve, + }, + ); + }); + expect(err.code).toBe("UND_ERR_ABORTED"); + await pool.close(); + }); + + it("Agent dispatches using opts.origin", async () => { + const agent = new Agent(); + const res = await dispatchLegacy(agent, { origin: hostUrl, path: "/get", method: "GET" }); + expect(res.statusCode).toBe(200); + expect(JSON.parse(res.body)).toEqual({ url: `${hostUrl}/get`, method: "GET" }); + await agent.close(); + }); + + it("getGlobalDispatcher returns a functional dispatcher", async () => { + const dispatcher = getGlobalDispatcher(); + expect(typeof dispatcher.dispatch).toBe("function"); + expect(typeof dispatcher.close).toBe("function"); + expect(typeof dispatcher.destroy).toBe("function"); + const res = await dispatchLegacy(dispatcher, { origin: hostUrl, path: "/get", method: "GET" }); + expect(res.statusCode).toBe(200); + }); + + it("new Pool() without an origin throws InvalidArgumentError", () => { + expect(() => new (Pool as any)()).toThrow(errors.InvalidArgumentError); + }); + }); }); describe("undici.request maxRedirections", () => { From 01b6a718a7a2a8ef81f067288d1be7cd8d450030 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Fri, 21 Aug 2026 20:05:13 +0000 Subject: [PATCH 2/2] drop a late 1xx onHeaders after a terminal callback --- src/js/thirdparty/undici.js | 6 ++++-- test/js/first_party/undici/undici.test.ts | 17 +++++++++++++++++ 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/src/js/thirdparty/undici.js b/src/js/thirdparty/undici.js index dc28a67b17cf..f9c0a9492010 100644 --- a/src/js/thirdparty/undici.js +++ b/src/js/thirdparty/undici.js @@ -624,14 +624,16 @@ class Dispatcher extends EventEmitter { } }, onHeaders: (statusCode, rawHeaders, resume, _statusText) => { + // onHeaders after a terminal callback violates the contract; a late 1xx must not fire onInfo either. + if (completed) return true; // 1xx informational responses precede the final onHeaders, like undici. if (statusCode < 200) { if (typeof opts.onInfo === "function") opts.onInfo({ statusCode, headers: headersFromRawHeaders(rawHeaders) }); return true; } - // onHeaders after a terminal callback or a second final onHeaders violates the contract; ignore it. - if (completed || body !== null) return true; + // A second final onHeaders violates the contract; keep the first body. + if (body !== null) return true; resumeBody = resume; const headers = headersFromRawHeaders(rawHeaders); body = new DispatchBodyReadable( diff --git a/test/js/first_party/undici/undici.test.ts b/test/js/first_party/undici/undici.test.ts index 7e54922639d8..b29b78dacdcd 100644 --- a/test/js/first_party/undici/undici.test.ts +++ b/test/js/first_party/undici/undici.test.ts @@ -1013,6 +1013,23 @@ describe("undici", () => { expect(calls).toEqual([boom]); }); + it("request() does not fire onInfo for a 1xx after a terminal callback", async () => { + class InfoAfterError extends Dispatcher { + dispatch(_opts: any, handler: any) { + handler.onConnect(() => {}); + handler.onError(new Error("boom")); + // A late 1xx violates the contract and must not reach opts.onInfo. + handler.onHeaders(100, [], () => {}, "Continue"); + return true; + } + } + const infos: number[] = []; + await expect( + new InfoAfterError().request({ path: "/", method: "GET", onInfo: (i: any) => infos.push(i.statusCode) }), + ).rejects.toThrow("boom"); + expect(infos).toEqual([]); + }); + it("fetch with dispatcher rejects invalid URLs instead of throwing", async () => { const dispatcher = { dispatch: () => true }; await expect(undiciFetch("not a url", { dispatcher } as any)).rejects.toBeInstanceOf(TypeError);