Compile undici from a built-in implementation (#1425)

This commit is contained in:
Chris Tate
2026-10-20 21:14:37 -05:00
committed by GitHub
37 changed files with 1580 additions and 175 deletions
+204 -69
View File
@@ -328,15 +328,18 @@ export class Client extends DispatcherBase {
}
__onTimeout(): void {
this.__timerCancel = null;
this.__timerRefresh = null;
const socket = this.__socket;
const parser = this.__parser;
// A body timeout that fires while the response is paused keeps its
// timer: resuming refreshes it, as undici's parser does.
if (this.__timerType === TIMEOUT_BODY && parser !== null && parser.paused) return;
this.__timerCancel = null;
this.__timerRefresh = null;
if (socket === null || parser === null) return;
if (this.__timerType === TIMEOUT_HEADERS) {
this.__destroySocket(socket, new HeadersTimeoutError());
} else if (this.__timerType === TIMEOUT_BODY) {
if (!parser.paused) this.__destroySocket(socket, new BodyTimeoutError());
this.__destroySocket(socket, new BodyTimeoutError());
} else if (this.__timerType === TIMEOUT_IDLE) {
this.__destroySocket(socket, new InformationalError("socket idle timeout"));
}
@@ -782,14 +785,50 @@ function errorCode(err: Error): string {
const CRLF = Buffer.from("\r\n");
/** The value of an ASCII hex digit, or -1. */
function hexValue(c: number): number {
// Chunk-size line states (llhttp's chunk_size* and chunk_extension* nodes).
const SIZE_FIRST = 0;
const SIZE_DIGITS = 1;
const SIZE_AFTER = 2;
const SIZE_EXT = 3;
const SIZE_EXT_NAME = 4;
const SIZE_EXT_VALUE = 5;
const SIZE_EXT_QUOTED = 6;
const SIZE_EXT_PAIR = 7;
const SIZE_EXT_QUOTED_DONE = 8;
const SIZE_CR = 9;
function hexDigit(c: number): number {
if (c >= 0x30 && c <= 0x39) return c - 0x30;
if (c >= 0x41 && c <= 0x46) return c - 0x37;
if (c >= 0x61 && c <= 0x66) return c - 0x57;
return -1;
}
/** tchar: ALPHA, DIGIT and !#$%&'*+-.^_`|~. */
const TOKEN_BYTES = tokenBytes();
function tokenBytes(): Uint8Array {
const table = new Uint8Array(256);
const marks = "!#$%&'*+-.^_`|~";
for (let i = 0; i < marks.length; i++) table[marks.charCodeAt(i)] = 1;
for (let c = 0x30; c <= 0x39; c++) table[c] = 1;
for (let c = 0x41; c <= 0x5a; c++) {
table[c] = 1;
table[c + 0x20] = 1;
}
return table;
}
/** qdtext: HTAB, SP and 0x21-0xff but '"' and '\\'. */
function isQuotedByte(c: number): boolean {
return c === 0x09 || c === 0x20 || (c >= 0x21 && c !== 0x22 && c !== 0x5c);
}
/** The byte after '\\' in a quoted-pair: HTAB, SP, VCHAR or obs-text. */
function isQuotedPairByte(c: number): boolean {
return c === 0x09 || c === 0x20 || (c >= 0x21 && c !== 0x7f);
}
/** An incremental HTTP/1.1 response parser (the subset of llhttp undici
* relies on): status line, raw header pairs, Content-Length, chunked and
* read-until-close bodies, trailers, and pause/resume back-pressure. */
@@ -816,14 +855,9 @@ class ResponseParser {
bytesRead: number;
trailers: Buffer[];
skipBody: boolean;
/** The chunk-size line read so far: its value, its digit count, whether
* an extension or whitespace follows the digits, and whether its CR was
* read. Extensions are skipped as they arrive, so the line keeps no
* bytes. */
sizeState: number;
sizeDigits: number;
chunkSize: number;
chunkDigits: number;
chunkExtension: boolean;
chunkCR: boolean;
constructor(client: Client, socket: Socket) {
this.client = client;
@@ -846,10 +880,9 @@ class ResponseParser {
this.bytesRead = 0;
this.trailers = [];
this.skipBody = false;
this.sizeState = SIZE_FIRST;
this.sizeDigits = 0;
this.chunkSize = 0;
this.chunkDigits = 0;
this.chunkExtension = false;
this.chunkCR = false;
}
reset(): void {
@@ -886,6 +919,7 @@ class ResponseParser {
resume(): void {
if (!this.paused) return;
if (this.client.__timerType === TIMEOUT_BODY) this.client.__refreshTimer();
this.paused = false;
this.socket.resume();
this.run();
@@ -903,18 +937,154 @@ class ResponseParser {
}
}
fail(message: string, code: string): void {
// undici 7.x wraps llhttp's reason and leaves the code unset (its
// createError indexes the error-name table with the error number);
// 5.x reports the reason and its HPE_ code.
const error =
/** A parse error with llhttp's reason: undici 7 puts it inside its
* protocol message and, as its error table maps names to numbers only,
* sets no code; undici 5 uses the reason as the message and the code. */
fail(reason: string, code: string): void {
const err =
this.client.__profile.major >= 7
? new HTTPParserError(`Response does not match the HTTP/1.1 protocol (${message})`)
: new HTTPParserError(message, code, "");
this.client.__onParseError(this.socket, error);
? new HTTPParserError(`Response does not match the HTTP/1.1 protocol (${reason})`)
: new HTTPParserError(reason, code, "");
this.client.__onParseError(this.socket, err);
this.state = 99;
}
/** Reads the chunk-size line one byte at a time as llhttp does, so a line
* that never ends holds no memory: hex digits (past 16 significant ones
* the size overflows), then extensions, checked and dropped, then CRLF.
* The 7.x profile follows llhttp 9 (undici 7), the 5.x profile llhttp 8
* in its lenient mode (undici 5). Returns true when the line is complete
* and chunkSize holds its size; false when it needs more bytes or failed. */
chunkSizeLine(): boolean {
const buf = this.buf;
if (this.sizeState === SIZE_FIRST) {
// The common line, hex digits then CRLF in this buffer, in one pass;
// any other line takes the byte loop below from its start.
let size = 0;
let digits = 0;
let j = 0;
for (; j < buf.length && j < 32; j++) {
const d = hexDigit(buf[j]!);
if (d === -1) break;
if (digits > 0 || d !== 0) {
digits++;
size = size * 16 + d;
}
}
if (j > 0 && digits <= 16 && j + 1 < buf.length && buf[j] === 0x0d && buf[j + 1] === 0x0a) {
this.chunkSize = size;
this.buf = buf.subarray(j + 2);
return true;
}
}
const v7 = this.client.__profile.major >= 7;
let i = 0;
while (i < buf.length) {
const c = buf[i]!;
const state = this.sizeState;
if (state === SIZE_FIRST || state === SIZE_DIGITS) {
const d = hexDigit(c);
if (d === -1) {
if (state === SIZE_FIRST) {
this.fail("Invalid character in chunk size", "INVALID_CHUNK_SIZE");
return false;
}
this.sizeState = SIZE_AFTER;
continue;
}
if (state === SIZE_FIRST) {
this.chunkSize = 0;
this.sizeDigits = 0;
this.sizeState = SIZE_DIGITS;
}
if (this.sizeDigits > 0 || d !== 0) {
if (this.sizeDigits === 16) {
this.fail("Chunk size overflow", "INVALID_CHUNK_SIZE");
return false;
}
this.sizeDigits++;
this.chunkSize = this.chunkSize * 16 + d;
}
} else if (state === SIZE_AFTER) {
if (c === 0x0d) this.sizeState = SIZE_CR;
else if (c === 0x3b) this.sizeState = SIZE_EXT;
else if (c === 0x0a && v7) {
this.fail("Missing expected CR after chunk size", "CR_EXPECTED");
return false;
} else {
this.fail("Invalid character in chunk size", "INVALID_CHUNK_SIZE");
return false;
}
} else if (state === SIZE_EXT) {
if (c === 0x20 || c === 0x0d) {
this.fail("Invalid character in chunk extensions", "STRICT");
return false;
}
this.sizeState = SIZE_EXT_NAME;
continue;
} else if (state === SIZE_EXT_NAME) {
if (c === 0x3d) this.sizeState = SIZE_EXT_VALUE;
else if (c === 0x3b) this.sizeState = SIZE_EXT;
else if (c === 0x0d) this.sizeState = SIZE_CR;
else if (c === 0x0a && v7) {
this.fail("Missing expected CR after chunk extension name", "CR_EXPECTED");
return false;
} else if (TOKEN_BYTES[c] !== 1) {
this.fail("Invalid character in chunk extensions name", "STRICT");
return false;
}
} else if (state === SIZE_EXT_VALUE) {
if (c === 0x22) this.sizeState = SIZE_EXT_QUOTED;
// llhttp 8 goes on with the chunk-size node after a value's ';'.
else if (c === 0x3b) this.sizeState = v7 ? SIZE_EXT : SIZE_AFTER;
else if (c === 0x0d) this.sizeState = SIZE_CR;
else if (c === 0x0a && v7) {
this.fail("Missing expected CR after chunk extension value", "CR_EXPECTED");
return false;
} else if (TOKEN_BYTES[c] !== 1) {
this.fail("Invalid character in chunk extensions value", "STRICT");
return false;
}
} else if (state === SIZE_EXT_QUOTED) {
if (c === 0x22) this.sizeState = SIZE_EXT_QUOTED_DONE;
else if (c === 0x5c && v7) this.sizeState = SIZE_EXT_PAIR;
else if (!isQuotedByte(c)) {
this.fail("Invalid character in chunk extensions quoted value", "STRICT");
return false;
}
} else if (state === SIZE_EXT_PAIR) {
if (!isQuotedPairByte(c)) {
this.fail("Invalid quoted-pair in chunk extensions quoted value", "STRICT");
return false;
}
this.sizeState = SIZE_EXT_QUOTED;
} else if (state === SIZE_EXT_QUOTED_DONE) {
if (c === 0x3b) this.sizeState = SIZE_EXT;
else if (c === 0x0d) this.sizeState = SIZE_CR;
else if (c === 0x0a && v7) {
this.fail("Missing expected CR after chunk extension value", "CR_EXPECTED");
return false;
} else {
this.fail("Invalid character in chunk extensions quote value", "STRICT");
return false;
}
} else {
// SIZE_CR: llhttp 9 needs the LF; llhttp 8's lenient mode skips
// whatever byte follows the CR.
if (c !== 0x0a && v7) {
this.fail("Expected LF after chunk size", "STRICT");
return false;
}
this.buf = buf.subarray(i + 1);
this.sizeState = SIZE_FIRST;
return true;
}
i++;
}
this.buf = Buffer.alloc(0);
return false;
}
line(): Buffer | null {
const idx = this.buf.indexOf(CRLF);
if (idx === -1) return null;
@@ -941,7 +1111,12 @@ class ResponseParser {
const text = line.toString("latin1");
const m = /^HTTP\/1\.([01]) (\d{3})(?: (.*))?$/.exec(text);
if (m === null) {
this.fail("Response does not match the HTTP/1.1 protocol", "INVALID_CONSTANT");
const v7 = this.client.__profile.major >= 7;
if (!text.startsWith("HTTP/")) {
this.fail(v7 ? "Expected HTTP/, RTSP/ or ICE/" : "Expected HTTP/", "INVALID_CONSTANT");
} else {
this.fail(v7 ? "Invalid status code" : "Invalid response status", "INVALID_STATUS");
}
return;
}
this.versionMinor = Number(m[1]);
@@ -1055,55 +1230,15 @@ class ResponseParser {
if (!this.complete()) return;
}
} else if (this.state === 3) {
// Chunk size line: hex digits, then optional whitespace and
// extensions, then CRLF. It is read byte by byte as it arrives.
if (this.buf.length === 0) {
if (this.more()) continue;
// Chunk size line; a partial one continues with the next arrival.
if (!this.chunkSizeLine()) {
if (this.state === 3 && this.more()) continue;
return;
}
let i = 0;
let done = false;
for (; i < this.buf.length && !done; i++) {
const c = this.buf[i]!;
if (this.chunkCR) {
if (c !== 0x0a) {
this.fail("Expected LF after chunk size", "STRICT");
return;
}
done = true;
} else if (c === 0x0d) {
if (this.chunkDigits === 0) {
this.fail("Invalid character in chunk size", "INVALID_CHUNK_SIZE");
return;
}
this.chunkCR = true;
} else if (this.chunkExtension) {
continue;
} else if (hexValue(c) >= 0) {
this.chunkSize = this.chunkSize * 16 + hexValue(c);
this.chunkDigits++;
if (this.chunkSize > Number.MAX_SAFE_INTEGER) {
this.fail("Chunk size overflow", "INVALID_CHUNK_SIZE");
return;
}
} else if (this.chunkDigits > 0 && (c === 0x3b || c === 0x20 || c === 0x09)) {
this.chunkExtension = true;
} else {
this.fail("Invalid character in chunk size", "INVALID_CHUNK_SIZE");
return;
}
}
this.buf = this.buf.subarray(i);
if (!done) continue;
const size = this.chunkSize;
this.chunkSize = 0;
this.chunkDigits = 0;
this.chunkExtension = false;
this.chunkCR = false;
if (size === 0) {
if (this.chunkSize === 0) {
this.state = 5;
} else {
this.remaining = size;
this.remaining = this.chunkSize;
this.state = 4;
}
} else if (this.state === 4) {
@@ -8,7 +8,7 @@ export declare namespace DiagnosticsChannel {
completed: boolean;
method?: string;
path: string;
headers: string;
headers: string | string[];
addHeader(key: string, value: string): Request;
}
interface Response {
@@ -38,6 +38,12 @@ export interface ResponseData {
const defaultProfile: Profile = { major: 7 };
export class Dispatcher extends EventEmitter {
// An explicit constructor: the implicit one forwards its arguments to
// EventEmitter's options-taking constructor, which stays a runtime fence.
constructor() {
super();
}
dispatch(_options: DispatchOptions, _handler: DispatchHandlers): boolean {
throw new Error("not implemented");
}
+277 -55
View File
@@ -2,14 +2,16 @@
// program-installed global dispatcher) it is the runtime's own fetch, which
// implements the same WHATWG algorithm. With one, the request goes through
// that dispatcher (its connector, lookup and pool), following undici's
// fetch for request headers, redirects and errors; the response body is
// buffered into a native Response.
// fetch for request headers, redirects and errors. The promise resolves
// when the response headers arrive, and the Response body streams from the
// connection with undici's back-pressure (see dispatchOnce).
//
// Known gaps: a Response built here reports `url` as "" and `redirected` as
// false (the runtime's Response has no settable URL yet), and https requests
// offer `gzip, deflate` rather than undici's `br, gzip, deflate` (the runtime
// has no Brotli decoder).
import { gunzipSync, inflateSync } from "node:zlib";
import type { Transform } from "node:stream";
import * as zlib from "node:zlib";
import type { Dispatcher } from "./dispatcher.ts";
import { getGlobalDispatcher, hasCustomGlobalDispatcher } from "./api.ts";
import type { DispatchHandlers, Profile } from "./types.ts";
@@ -115,7 +117,10 @@ interface RawResponse {
status: number;
statusText: string;
headers: string[];
body: Buffer;
/** null when the response has no body to expose: a null-body status or
* HEAD (its bytes, if any, are read and dropped), or a redirect fetch
* follows or rejects (the request is aborted, closing its connection). */
body: ReadableStream<Uint8Array> | null;
}
const redirectStatuses = [301, 302, 303, 307, 308];
@@ -125,7 +130,7 @@ export function undiciFetch(
input: string | URL,
init: UndiciRequestInit | undefined,
createGlobal: () => Dispatcher,
_profile: Profile,
profile: Profile,
): Promise<Response> {
const explicit = init !== undefined ? init.dispatcher : undefined;
if (explicit === undefined && !hasCustomGlobalDispatcher()) {
@@ -139,7 +144,7 @@ export function undiciFetch(
});
}
const dispatcher = explicit ?? getGlobalDispatcher(createGlobal);
return fetchThrough(dispatcher, input, init ?? {});
return fetchThrough(dispatcher, input, init ?? {}, profile);
}
function normalizeMethod(method: string): string {
@@ -153,6 +158,7 @@ async function fetchThrough(
dispatcher: Dispatcher,
input: string | URL,
init: UndiciRequestInit,
profile: Profile,
): Promise<Response> {
let url = typeof input === "string" ? new URL(input) : new URL(input.href);
let method = normalizeMethod(init.method ?? "GET");
@@ -197,7 +203,7 @@ async function fetchThrough(
let raw: RawResponse;
try {
raw = await dispatchOnce(dispatcher, url, method, pairs, body, signal);
raw = await dispatchOnce(dispatcher, url, method, pairs, body, signal, redirectMode, profile);
} catch (err) {
if (signal !== null && signal.aborted) throw abortReason(signal);
throw new TypeError("fetch failed", { cause: err });
@@ -243,15 +249,11 @@ async function fetchThrough(
continue;
}
const nullBody = nullBodyStatuses.includes(raw.status) || method === "HEAD";
return new Response(
nullBody ? null : decode(raw.body, responseHeaders.get("content-encoding")),
{
status: raw.status,
statusText: raw.statusText,
headers: responseHeaders,
},
);
return new Response(raw.body, {
status: raw.status,
statusText: raw.statusText,
headers: responseHeaders,
});
}
}
@@ -324,20 +326,28 @@ function headerListOf(
return list;
}
function decode(body: Buffer, encoding: string | null): Buffer {
if (encoding === null || body.length === 0) return body;
let out = body;
const codings = encoding
.toLowerCase()
.split(",")
.map((c) => c.trim())
.reverse();
/** undici 7's limit on a response's content codings, applied in both
* profiles: undici 5 has none, and builds one decoder per listed coding. */
const maxContentEncodings = 5;
/** Whether every coding has a decoder; otherwise the body passes through
* as sent. */
function decodable(codings: string[]): boolean {
if (codings.length === 0) return false;
for (const coding of codings) {
if (coding === "gzip" || coding === "x-gzip") out = gunzipSync(out);
else if (coding === "deflate") out = inflateSync(out);
else return body;
if (coding !== "gzip" && coding !== "x-gzip" && coding !== "deflate") return false;
}
return out;
return true;
}
/** A decoder for one coding. undici 7 decodes with Z_SYNC_FLUSH, which
* accepts a truncated stream; undici 5 uses zlib's defaults. */
function decoderFor(coding: string, profile: Profile): Transform {
const options =
profile.major >= 7
? { flush: zlib.constants.Z_SYNC_FLUSH, finishFlush: zlib.constants.Z_SYNC_FLUSH }
: {};
return coding === "deflate" ? zlib.createInflate(options) : zlib.createGunzip(options);
}
function abortReason(signal: AbortSignal): Error {
@@ -347,6 +357,16 @@ function abortReason(signal: AbortSignal): Error {
: new DOMException("This operation was aborted", "AbortError");
}
/** Unread bytes held beside the response stream before the connection
* pauses: Node's Readable highWaterMark, which undici's fetch body uses. */
const bodyHighWaterMark = 64 * 1024;
/** Dispatches one request and resolves when its response headers arrive.
* The body streams as undici's fetch reads one: up to bodyHighWaterMark
* unread bytes wait beside the stream, past that the connection (or the
* last decoder) pauses until the reader pulls again; cancelling the stream
* aborts the request, and an error after the headers errors the stream
* with TypeError('terminated'). */
function dispatchOnce(
dispatcher: Dispatcher,
url: URL,
@@ -354,50 +374,252 @@ function dispatchOnce(
headers: string[],
body: Buffer | null,
signal: AbortSignal | null,
redirectMode: "follow" | "manual" | "error",
profile: Profile,
): Promise<RawResponse> {
return new Promise<RawResponse>((resolve, reject) => {
let status = 0;
let statusText = "";
let rawHeaders: string[] = [];
const chunks: Buffer[] = [];
let settled = false;
let resolved = false;
let done = false;
let dump = false;
let abortFn: ((err?: Error) => void) | null = null;
const onAbort = (): void => {
// The response body: its stream controller, the unread bytes beside it,
// and how to restart the paused producer.
let controller: ReadableStreamDefaultController<Uint8Array> | null = null;
let hasBody = false;
const queue: Buffer[] = [];
let queued = 0;
let ended = false;
let closed = false;
let onDrain: (() => void) | null = null;
// The decoders, first-applied last; their listeners hold this scope, so
// releaseDecoders drops them once the body ends, fails or is cancelled.
let decoding = false;
let decoders: Transform[] = [];
let first: Transform | null = null;
let encodedBytes = 0;
let resumeConnection: (() => void) | null = null;
const abortRequest = (err: Error): void => {
const abort = abortFn;
if (abort !== null && signal !== null) abort(abortReason(signal));
if (abort !== null && !done) abort(err);
};
const onAbort = (): void => {
if (signal === null) return;
const reason = abortReason(signal);
// After the headers undici 7 errors the body with the signal's
// reason; undici 5 aborts its fetch controller without one, so the
// body gets the controller's default AbortError.
errorBody(
profile.major >= 7 ? reason : new DOMException("The operation was aborted.", "AbortError"),
);
abortRequest(reason);
};
const stopListening = (): void => {
if (signal !== null) signal.removeEventListener("abort", onAbort);
};
if (signal !== null) signal.addEventListener("abort", onAbort);
const finish = (err: Error | null): void => {
if (settled) return;
settled = true;
if (signal !== null) signal.removeEventListener("abort", onAbort);
if (err !== null) reject(err);
else resolve({ status, statusText, headers: rawHeaders, body: Buffer.concat(chunks) });
const releaseDecoders = (): void => {
const chain = decoders;
decoders = [];
first = null;
for (const decoder of chain) {
decoder.removeAllListeners();
decoder.destroy();
}
};
const closeBody = (): void => {
closed = true;
controller!.close();
stopListening();
};
const errorBody = (err: unknown): void => {
if (!hasBody || closed) return;
closed = true;
queue.length = 0;
queued = 0;
controller!.error(err);
stopListening();
releaseDecoders();
};
const endBody = (): void => {
if (closed) return;
ended = true;
if (queue.length === 0) closeBody();
};
/** Takes a chunk; false when the producer should pause until onDrain. */
const pushBody = (chunk: Buffer): boolean => {
if (closed) return true;
if (queue.length === 0 && (controller!.desiredSize ?? 0) > 0) {
controller!.enqueue(new Uint8Array(chunk));
return true;
}
queue.push(chunk);
queued += chunk.length;
return queued < bodyHighWaterMark;
};
const pullBody = (): void => {
if (closed) return;
const next = queue.shift();
if (next !== undefined) {
queued -= next.length;
controller!.enqueue(new Uint8Array(next));
}
if (queue.length === 0 && ended) {
closeBody();
return;
}
const drain = onDrain;
if (drain !== null && queued < bodyHighWaterMark) {
onDrain = null;
drain();
}
};
const cancelBody = (reason: unknown): void => {
if (closed) return;
closed = true;
queue.length = 0;
queued = 0;
stopListening();
releaseDecoders();
abortRequest(
reason instanceof Error
? reason
: new DOMException("The operation was aborted.", "AbortError"),
);
};
const failBody = (err: Error): void => {
errorBody(new TypeError("terminated", { cause: err }));
abortRequest(err);
};
const handler: DispatchHandlers = {
onConnect: (abort: (err?: Error) => void): void => {
abortFn = abort;
if (signal !== null && signal.aborted) abort(abortReason(signal));
},
onHeaders: (
statusCode: number,
raw: Buffer[],
_resume: () => void,
text: string,
): boolean => {
onHeaders: (statusCode: number, raw: Buffer[], resume: () => void, text: string): boolean => {
if (statusCode < 200) return true;
status = statusCode;
statusText = text;
rawHeaders = [];
for (const b of raw) rawHeaders.push(b.toString("latin1"));
resolved = true;
resumeConnection = resume;
const rawHeaders: string[] = [];
let location = false;
let encoding: string | null = null;
for (let i = 0; i + 1 < raw.length; i += 2) {
const name = raw[i]!.toString("latin1");
const value = raw[i + 1]!.toString("latin1");
rawHeaders.push(name, value);
const lower = name.toLowerCase();
if (lower === "location") location = true;
else if (lower === "content-encoding")
encoding = encoding === null ? value : `${encoding}, ${value}`;
}
const follows =
redirectStatuses.includes(statusCode) && location && redirectMode !== "manual";
if (follows || nullBodyStatuses.includes(statusCode) || method === "HEAD") {
dump = true;
stopListening();
resolve({ status: statusCode, statusText: text, headers: rawHeaders, body: null });
// A redirect that is followed or rejected never reads its body:
// abort the request so the connection is released instead of
// draining a body that may never end.
if (follows) abortRequest(new DOMException("The operation was aborted.", "AbortError"));
return true;
}
const codings =
encoding === null
? []
: encoding
.toLowerCase()
.split(",")
.map((c) => c.trim());
if (codings.length > maxContentEncodings) {
dump = true;
stopListening();
reject(
new Error(
`too many content-encodings in response: ${codings.length}, maximum allowed is ${maxContentEncodings}`,
),
);
abortRequest(new DOMException("The operation was aborted.", "AbortError"));
return true;
}
hasBody = true;
const stream = new ReadableStream<Uint8Array>({
start: (c: ReadableStreamDefaultController<Uint8Array>): void => {
controller = c;
},
pull: (): void => pullBody(),
cancel: (reason: unknown): void => cancelBody(reason),
});
if (decodable(codings)) {
// Decoders run last-applied coding first, piped into each other;
// the last one feeds the body.
decoding = true;
let last: Transform | null = null;
for (let i = codings.length - 1; i >= 0; i--) {
const decoder = decoderFor(codings[i]!, profile);
decoder.on("error", (err: Error) => failBody(err));
if (last === null) first = decoder;
else last.pipe(decoder);
decoders.push(decoder);
last = decoder;
}
const output = last!;
output.on("data", (chunk: Buffer) => {
if (!pushBody(chunk)) {
output.pause();
onDrain = (): void => {
output.resume();
};
}
});
output.on("end", () => {
releaseDecoders();
endBody();
});
first!.on("drain", () => resume());
}
resolve({ status: statusCode, statusText: text, headers: rawHeaders, body: stream });
return true;
},
onData: (chunk: Buffer): boolean => {
chunks.push(chunk);
return true;
if (dump || !hasBody || closed) return true;
if (decoding) {
const decoder = first;
if (decoder === null) return true;
encodedBytes += chunk.length;
return decoder.write(chunk);
}
if (pushBody(chunk)) return true;
onDrain = resumeConnection;
return false;
},
onComplete: (_trailers: string[] | null): void => {
done = true;
if (!hasBody) {
stopListening();
return;
}
// zlib rejects an empty stream; undici ends such a body unread.
const decoder = first;
if (decoder !== null && encodedBytes > 0) {
decoder.end();
} else {
releaseDecoders();
endBody();
}
},
onError: (err: Error): void => {
done = true;
if (!resolved) {
stopListening();
reject(err);
return;
}
errorBody(new TypeError("terminated", { cause: err }));
},
onComplete: (_trailers: string[] | null): void => finish(null),
onError: (err: Error): void => finish(err),
};
dispatcher.dispatch(
{
+34 -8
View File
@@ -32,8 +32,10 @@ export class Request {
error: Error | null;
abortFn: ((err?: Error) => void) | null;
handler: DispatchHandlers;
/** The request as diagnostics_channel subscribers see it: one boxed
* value per request, so its identity is stable across publications. */
profile: Profile;
/** The request as diagnostics_channel subscribers see it: one untyped
* object per request, so its identity is stable across publications.
* The request keeps its completed and headers fields current. */
diagnostic: unknown;
constructor(origin: string, opts: DispatchOptions, handler: DispatchHandlers, profile: Profile) {
@@ -91,6 +93,7 @@ export class Request {
}
this.completed = false;
this.profile = profile;
this.aborted = false;
this.idempotent = opts.idempotent ?? (method === "HEAD" || method === "GET");
this.blocking = opts.blocking ?? (profile.major >= 7 ? method !== "HEAD" : false);
@@ -131,18 +134,29 @@ export class Request {
}
diagnosticView(): unknown {
let view = this.diagnostic;
if (view === null) {
view = {
if (this.diagnostic === null) {
// addHeader adds a header as undici's Request.addHeader does (the
// tracing instrumentations' header injection).
const view: any = {
origin: this.origin,
completed: this.completed,
method: this.method,
path: this.path,
headers: this.headers,
headers: this.diagnosticHeaders(),
addHeader: (key: string, value: string): unknown => {
this.addHeader(key, value);
return view;
},
};
this.diagnostic = view;
}
return view;
return this.diagnostic;
}
/** The diagnostics view's headers: undici 7 publishes its header array,
* undici 5 the raw block. */
diagnosticHeaders(): string | string[] {
return this.profile.major >= 7 ? this.headerPairs.slice() : this.headers;
}
/** The DiagnosticsChannel.Request view's raw header block. */
@@ -155,7 +169,9 @@ export class Request {
}
addHeader(key: string, value: string): Request {
this.headerPairs.push(key, value);
processHeader(this, key, value, this.profile);
const view: any = this.diagnostic;
if (view !== null) view.headers = this.diagnosticHeaders();
return this;
}
@@ -207,6 +223,8 @@ export class Request {
onComplete(trailers: Buffer[]): void {
this.completed = true;
const view: any = this.diagnostic;
if (view !== null) view.completed = true;
if (channels.trailers.hasSubscribers)
channels.trailers.publish({ request: this.diagnosticView(), trailers });
if (this.handler.onComplete !== undefined) {
@@ -227,10 +245,18 @@ export class Request {
function processHeader(request: Request, key: string, val: string, profile: Profile): void {
const lower = key.toLowerCase();
// undici 7 checks the name and every value before the special cases;
// undici 5 checks the first Host value before storing it. Either way a
// CR or LF in a value never reaches the request head.
if (profile.major >= 7) {
if (!tokenRegExp.test(lower)) throw new InvalidArgumentError("invalid header key");
if (headerCharRegex.test(val)) throw new InvalidArgumentError(`invalid ${key} header`);
}
if (lower === "host") {
// The client writes the Host value into the request head as given.
if (headerCharRegex.test(val)) throw new InvalidArgumentError(`invalid ${key} header`);
if (request.host === null) {
if (headerCharRegex.test(val)) throw new InvalidArgumentError(`invalid ${key} header`);
request.host = val;
return;
}
@@ -35,7 +35,9 @@ function onTick(): void {
}
}
fastTimers.length = len;
// An idle tick stops; the next timer schedules a new one.
if (fastTimers.length !== 0) refreshTimeout();
else refreshTick = null;
}
function refreshTimeout(): void {
@@ -3797,8 +3797,20 @@ function lowerDynReceiverMethodCall(
)
)
return true;
// A standard-library member spelled as a function-typed property
// (undici-types declares Headers.append/forEach/... and the body mixin
// as `readonly append: (...) => void`; scriptc's own declarations use
// methods) is still the handle's native method: it takes the shared
// receiver dispatch below either way (which also calls a plain object's
// own member), so the lowering does not depend on which declaration
// file supplies the type.
const libMethod =
prop !== undefined &&
prop.declarations.length > 0 &&
prop.declarations.every((decl) => lowerer.isStdlibFile(decl.getSourceFile()));
return (
prop !== undefined &&
!libMethod &&
prop.declarations.some(
(decl) => ts.isPropertyDeclaration(decl) || ts.isPropertySignature(decl),
)
@@ -622,6 +622,20 @@ export function lowerFetchCall(lowerer: Lowerer, call: ts.CallExpression): IrExp
: lowerer.lowerExpr(argument);
stmts.push({ kind: "exprStmt", expr: discarded, loc: discarded.loc });
}
// A union input holding a URL arm (`string | URL`) converts through the
// checked-dynamic input path, which reads the URL's href at the call
// like the bare-URL arm below; narrowing it to a string would throw on
// the URL arm.
const unionWithUrl =
input.type.kind === "union" &&
(lowerer.unions.get(input.type.unionId)?.arms.some((arm) => arm.kind === "url") ?? false) &&
lowerer.dynConvertible(input.type);
const dynInput: IrExpr | null =
input.type.kind === "dyn"
? inputRef
: unionWithUrl
? { kind: "dynFrom", value: inputRef, type: DYN, loc: locOf(inputNode) }
: null;
const url: IrExpr =
input.type.kind === "url"
? {
@@ -631,13 +645,13 @@ export function lowerFetchCall(lowerer: Lowerer, call: ts.CallExpression): IrExp
type: STRING,
loc: locOf(inputNode),
}
: input.type.kind === "dyn"
? inputRef
: dynInput !== null
? dynInput
: lowerer.coerceInto(inputNode, inputRef, STRING);
const answer: IrExpr = {
kind: "libCall",
fn: input.type.kind === "dyn" ? "fetch.input" : "fetch.start",
args: [input.type.kind === "dyn" ? inputRef : url, initRef],
fn: dynInput !== null ? "fetch.input" : "fetch.start",
args: [url, initRef],
type: { kind: "promise", inner: DYN },
loc,
};
+10 -1
View File
@@ -4718,7 +4718,9 @@ function mapTypeInner(type: ts.Type, ctx: TypeMapperCtx): IrType | null {
// semantics would DEGRADE under the collapse — class instances,
// Maps/Sets, functions (closure identity, `x === String`
// narrowing), promises, regexes, generators, handles — keep their
// existing union homes and fences below. A pending recursive
// existing union homes and fences below. URL is the
// exception: its handle boxes by reference and keeps every member,
// so it collapses too (dynSubsumableUnionArm). A pending recursive
// placeholder cannot collapse (back-references already hold the
// union id); the degenerate recursive spelling keeps its fence.
if (
@@ -7704,6 +7706,13 @@ function dynSubsumableUnionArm(arm: IrType, ctx: TypeMapperCtx): boolean {
case "undefinedT":
case "nullT":
return true;
// URL boxes BY REFERENCE (its checked-dynamic handle keeps identity,
// reads and writes its members through the native accessors, answers
// `instanceof URL`, and checks back out to the typed value), so it
// rides the collapse beside fetch's Request/Response handles:
// `string | URL | Request`, fetch's own input type under @types/node.
case "url":
return true;
case "record":
case "array":
// `T | Promise<T>` with T bound to `unknown`: the checked-dynamic value
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/6251-net-connect-poll-phase-order.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/corpus/6252-fetch-input-url-unions.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/fixtures/node-types/fetch-handle-members.ts"
],
"diags": []
}
@@ -0,0 +1,6 @@
{
"order": [
"<repo>/tests/fixtures/node-types/fetch-input-unions.ts"
],
"diags": []
}
@@ -15,6 +15,7 @@
"<repo>/packages/compiler/shims/undici/core/fetch.ts",
"<repo>/packages/compiler/shims/undici/core/errors-v5.ts",
"<repo>/packages/compiler/shims/undici/v5.ts",
"<repo>/tests/fixtures/npm/cases/undici5/peers.ts",
"<repo>/tests/fixtures/npm/cases/undici5/main.ts"
],
"diags": []
@@ -15,6 +15,7 @@
"<repo>/packages/compiler/shims/undici/core/fetch.ts",
"<repo>/packages/compiler/shims/undici/core/errors-v7.ts",
"<repo>/packages/compiler/shims/undici/v7.ts",
"<repo>/tests/fixtures/npm/cases/undici7/peers.ts",
"<repo>/tests/fixtures/npm/cases/undici7/main.ts"
],
"diags": []
@@ -18,9 +18,12 @@
* program the check builds includes them), since real projects compile
* them with their own Node types.
*/
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import ts5 from "typescript5";
import { describe, expect, test } from "vitest";
import * as ad from "../src/frontend/ts7/adapter.js";
import { createProgram, getPreEmitDiagnostics } from "../src/frontend/ts7/program-adapter.js";
const repoRoot = join(import.meta.dirname, "../../..");
const shimRoot = join(repoRoot, "packages/compiler/shims/undici");
@@ -282,31 +285,29 @@ type _clientApi = ExpectNever<Exclude<keyof Public<shim.Client>, keyof real.Clie
`;
}
function check(files: Record<string, string>): string[] {
const options: ts5.CompilerOptions = {
/** The diagnostics of one program, checked by the TypeScript 7 checker. */
function check(name: string, text: string): string[] {
const dir = mkdtempSync(join(tmpdir(), "scriptc-undici-shim-types-"));
const file = join(dir, name);
writeFileSync(file, text);
const program = createProgram([file], {
strict: true,
noEmit: true,
skipLibCheck: true,
allowImportingTsExtensions: true,
module: ts5.ModuleKind.ESNext,
moduleResolution: ts5.ModuleResolutionKind.Bundler,
target: ts5.ScriptTarget.ES2022,
module: ad.ModuleKind.ESNext as number,
moduleResolution: ad.ModuleResolutionKind.Bundler as number,
target: ad.ScriptTarget.ES2022 as number,
lib: ["lib.es2023.d.ts"],
types: ["node"],
typeRoots: [join(repoRoot, "node_modules/@types")],
};
const host = ts5.createCompilerHost(options);
const readFile = host.readFile.bind(host);
const fileExists = host.fileExists.bind(host);
host.readFile = (name) => files[name] ?? readFile(name);
host.fileExists = (name) => name in files || fileExists(name);
const program = ts5.createProgram(Object.keys(files), options, host);
return ts5.getPreEmitDiagnostics(program).map((d) => {
const where = d.file
? `${d.file.fileName}:${d.file.getLineAndCharacterOfPosition(d.start ?? 0).line + 1}`
: "";
return `${where} ${ts5.flattenDiagnosticMessageText(d.messageText, "\n")}`;
});
try {
return getPreEmitDiagnostics(program).map((d) => `${d.fileName ?? ""}:${d.pos} ${d.text}`);
} finally {
program.dispose();
rmSync(dir, { recursive: true, force: true });
}
}
describe.each([5, 7])("undici %i.x shim types", (major) => {
@@ -314,18 +315,15 @@ describe.each([5, 7])("undici %i.x shim types", (major) => {
const shim = join(shimRoot, `v${major}.ts`);
test("a program written against undici type-checks against the real declarations", () => {
const file = join(repoRoot, `.undici-usage-real-${major}.ts`);
expect(check({ [file]: helpers + usage(major).replace(/UNDICI/g, real) })).toEqual([]);
expect(check("usage-real.ts", helpers + usage(major).replace(/UNDICI/g, real))).toEqual([]);
});
test("the same program type-checks against the shim", () => {
const file = join(repoRoot, `.undici-usage-shim-${major}.ts`);
expect(check({ [file]: helpers + usage(major).replace(/UNDICI/g, shim) })).toEqual([]);
expect(check("usage-shim.ts", helpers + usage(major).replace(/UNDICI/g, shim))).toEqual([]);
});
test("the shim's exports and records are a compatible subset", () => {
const file = join(repoRoot, `.undici-shape-${major}.ts`);
const text = helpers + shape(major).replace(/REAL/g, real).replace(/SHIM/g, shim);
expect(check({ [file]: text })).toEqual([]);
expect(check("shape.ts", text)).toEqual([]);
});
});
+29
View File
@@ -4130,6 +4130,25 @@ static int scr_workers_pollfd(void) {
* macrotask). */
bool scr_loop_has_ready(void) { return scr_ready_len > 0 || scr_nt_head != NULL; }
/* The poll phase runs once per loop iteration, like libuv's: the net unit
* drains the poller at most once between check phases. I/O that becomes
* ready while this iteration's callbacks run (a loopback connect started
* by one of them) waits for the next iteration, so immediates queued in
* the meantime fire first — Node's poll → check order. A full batch polls
* again without waiting, at most SCR_LOOP_IO_BATCHES times per phase (the
* count libuv's uv__io_poll uses); later events wait for the next
* iteration, so timers and immediates still run while sockets stay busy. */
enum { SCR_LOOP_IO_BATCHES = 48 };
static SCR_TL bool scr_loop_io_open = true;
static SCR_TL int scr_loop_io_batches = 0;
bool scr_loop_io_take(void) {
bool open = scr_loop_io_open;
scr_loop_io_open = false;
return open;
}
bool scr_loop_io_more(void) { return ++scr_loop_io_batches < SCR_LOOP_IO_BATCHES; }
void scr_loop_io_reopen(void) { scr_loop_io_open = true; }
/* The process.nextTick queue to exhaustion, ticks queued meanwhile
* included. false: a tick threw (uncaught, pending). */
static bool scr_loop_run_ticks(void) {
@@ -4607,6 +4626,7 @@ static bool scr_loop_run_pass(ScrPromise *top_level, bool first_checkpoint) {
* every macrotask; pending ticks or a woken fiber restart the turn
* first (the timer loop's break above), so the phase only runs on a
* drained checkpoint. */
bool check_ran = false;
if (scr_ready_len == 0 && scr_nt_head == NULL && scr_nunhandled == 0 && scr_pending_immediates > 0) {
size_t end = scr_nimmediates;
while (scr_immediates_head < end) {
@@ -4637,6 +4657,15 @@ static bool scr_loop_run_pass(ScrPromise *top_level, bool first_checkpoint) {
scr_immediates_head = 0;
scr_nimmediates = 0;
}
check_ran = true;
}
/* The next poll phase opens once this iteration's check phase ran (or
* had nothing to run). A check phase deferred for pending microtasks
* keeps it closed, so the next iteration reaches the check phase
* before polling again. */
if (check_ran || scr_pending_immediates == 0) {
scr_loop_io_open = true;
scr_loop_io_batches = 0;
}
}
return rejection_failed;
+21 -3
View File
@@ -5346,17 +5346,30 @@ static int scr_net_pollfd(void) {
return scr_net_poller != NULL ? scrp_poller_fd(scr_net_poller) : -1;
}
bool scr_loop_io_take(void);
bool scr_loop_io_more(void);
void scr_loop_io_reopen(void);
static void scr_net_dispatch(void) {
if (!scr_net_servers && !scr_net_socks &&
!scr_net_proto_pending()) {
return;
}
/* One poller drain per loop iteration (scr_loop_io_take); a full batch
* keeps draining within the phase's batch budget (scr_loop_io_more), like
* libuv's poll phase. Deferred emits (the sweep) still run on every
* call. */
bool may_drain = scr_loop_io_take();
for (;;) {
scr_net_sweep();
if (scr_exc_pending()) return;
if (scr_net_poller == NULL) return;
ScrPollerEvent evs[64];
int n = scrp_drain(scr_net_poller, evs, 64);
int n = 0;
if (may_drain) {
n = scrp_drain(scr_net_poller, evs, 64);
may_drain = n == 64 && scr_loop_io_more();
}
for (int i = 0; i < n; i++) {
void *udata = evs[i].udata;
if (!udata) continue;
@@ -5430,8 +5443,13 @@ static void scr_net_dispatch(void) {
}
if (scr_exc_pending()) return;
}
if (!scr_net_flags_pending()) return;
if (scr_loop_has_ready()) return; /* microtasks interleave first */
if (!scr_net_flags_pending() && !may_drain) return;
if (scr_loop_has_ready()) {
/* microtasks interleave first; a full batch's rest drains on the
* next call, still within this poll phase */
if (may_drain) scr_loop_io_reopen();
return;
}
}
}
-1
View File
@@ -30,7 +30,6 @@ const ALLOWED_TYPESCRIPT5_IMPORTS = new Set([
"packages/compiler/src/library/semantic-source.ts",
"packages/compiler/test/ts7/harness.ts",
"packages/compiler/test/ts7/resolver-parity.test.ts",
"packages/compiler/test/undici-shim-types.test.ts",
]);
function sourceFilesUnder(dir) {
@@ -0,0 +1,70 @@
// A connection started by a poll-phase callback completes in the NEXT loop
// iteration's poll phase: the callback's nextTick and promise jobs run
// first, then the check phase's setImmediate, then the new socket's
// 'connect' (libuv/Node order; loopback connects complete at once, so an
// event loop that polls again before the check phase reorders them). The
// second half is the same through an awaited promise, the shape of an HTTP
// client that dials a fresh connection after the previous response.
import * as net from "node:net";
const log: string[] = [];
const server = net.createServer((socket) => {
socket.on("data", (chunk: Buffer) => socket.write(chunk));
});
function dial(port: number, label: string): Promise<net.Socket> {
return new Promise((resolve) => {
const socket = net.connect({ port, host: "127.0.0.1" });
socket.on("connect", () => {
log.push(`connect ${label}`);
resolve(socket);
});
});
}
function roundTrip(socket: net.Socket, label: string): Promise<void> {
return new Promise((resolve) => {
socket.once("data", () => {
log.push(`data ${label}`);
setImmediate(() => log.push(`immediate after data ${label}`));
process.nextTick(() => log.push(`tick after data ${label}`));
Promise.resolve().then(() => log.push(`promise after data ${label}`));
resolve();
});
socket.write("ping");
});
}
server.listen(0, "127.0.0.1", () => {
const address = server.address();
const port = address !== null && typeof address === "object" ? address.port : 0;
const first = net.connect({ port, host: "127.0.0.1" });
first.on("connect", () => {
log.push("connect first");
first.write("ping");
});
first.once("data", () => {
log.push("data first");
setImmediate(() => log.push("immediate from data"));
process.nextTick(() => log.push("tick from data"));
Promise.resolve().then(() => log.push("promise from data"));
const second = net.connect({ port, host: "127.0.0.1" });
second.on("connect", async () => {
log.push("connect second");
second.destroy();
first.destroy();
// Awaited: each response's continuation dials the next connection.
const a = await dial(port, "a");
await roundTrip(a, "a");
log.push("continuation after a");
const b = await dial(port, "b");
await roundTrip(b, "b");
a.destroy();
b.destroy();
setImmediate(() => {
server.close();
console.log(log.join("\n"));
});
});
});
});
@@ -0,0 +1,48 @@
// `typeof globalThis.fetch` under @types/node takes `string | URL | Request`:
// unions holding URL handles beside fetch handles compile (URL boxes by
// reference into the checked-dynamic representation), and a custom fetch
// cast to that type through `unknown` keeps working, including an extra
// init member written through a record view (the dispatcher pattern).
import * as http from "node:http";
type FetchImpl = typeof globalThis.fetch;
interface DispatchingInit {
method?: string;
headers?: Record<string, string>;
dispatcher?: { name: string };
}
function customFetch(input: string | URL, init?: DispatchingInit): Promise<Response> {
console.log("custom fetch", typeof input === "string" ? "string" : "url", init?.dispatcher?.name ?? "none");
return fetch(input, { method: init?.method, headers: init?.headers });
}
const server = http.createServer((req, res) => res.end(`ok ${req.method} ${req.url}`));
server.listen(0, "127.0.0.1", async () => {
const address = server.address();
const port = address !== null && typeof address === "object" ? address.port : 0;
const impl: FetchImpl = customFetch as unknown as FetchImpl;
const init: RequestInit = { method: "POST" };
(init as Record<string, unknown>).dispatcher = { name: "pinned" };
const res = await impl(`http://127.0.0.1:${port}/a`, init);
console.log(res.status, await res.text());
const viaUrl = await impl(new URL(`http://127.0.0.1:${port}/b`));
console.log(viaUrl.status, await viaUrl.text());
const direct = await fetch(new URL(`http://127.0.0.1:${port}/c`));
console.log(direct.status, await direct.text());
server.close();
});
function describeInput(input: string | Request | URL): string {
if (typeof input === "string") return `string ${input}`;
if (input instanceof URL) return `url ${input.href} ${input.hostname}`;
return `request ${input.method} ${input.url}`;
}
console.log(describeInput("http://a.test/x"));
const u = new URL("http://b.test:8080/p?q=1");
console.log(describeInput(u));
console.log(describeInput(new Request("http://c.test/r", { method: "PUT" })));
const items: (string | URL | Request)[] = [u, "s"];
console.log(items.length, items[0] === u, String(items[0]));
let slot: string | URL | Request = u;
if (slot instanceof URL) { slot.pathname = "/changed"; }
console.log(u.href);
const params: string | Request | URL = new URL("http://d.test/?b=2");
console.log(typeof params === "string" ? params : params instanceof URL ? params.searchParams.get("b") : "req");
+32
View File
@@ -0,0 +1,32 @@
// undici-types (@types/node's fetch declarations) spells the Headers methods
// and the Request/Response body mixin as function-typed properties; they
// must lower to the same native handle methods as the shipped declarations.
async function main(): Promise<void> {
const h = new Headers([["b", "2"], ["a", "1"]]);
h.append("set-cookie", "x=1");
h.append("set-cookie", "y=2");
h.set("c", "3");
h.delete("b");
const seen: string[] = [];
h.forEach((value, name) => seen.push(`${name}=${value}`));
console.log(seen.join(","), h.get("a"), h.has("b"), h.getSetCookie().join("|"));
const r1 = new Response("body-text", { status: 202, statusText: "Acc", headers: h });
console.log(r1.status, r1.statusText, r1.ok, r1.headers.get("c"), r1.bodyUsed);
console.log(await r1.text(), r1.bodyUsed);
const r2 = new Response(JSON.stringify({ k: 1 }));
console.log(JSON.stringify(await r2.json()));
const r3 = new Response(new Uint8Array([1, 2, 3]));
console.log((await r3.arrayBuffer()).byteLength);
const r4 = new Response("abc");
console.log((await r4.bytes()).length);
const r5 = Response.json({ z: true }, { status: 200 });
console.log(r5.headers.get("content-type"), await r5.text());
const req = new Request("http://x.test/a?b=1", { method: "POST", body: "payload", headers: { q: "1" } });
console.log(req.method, req.url, req.headers.get("q"), await req.text());
const req2 = new Request("http://x.test/j", { method: "PUT", body: "{\"n\":2}" });
console.log(JSON.stringify(await req2.json()));
const ac = new AbortController();
ac.abort();
console.log(ac.signal.aborted);
}
main();
+48
View File
@@ -0,0 +1,48 @@
// `typeof globalThis.fetch` under @types/node takes `string | URL | Request`:
// unions holding URL handles beside fetch handles compile (URL boxes by
// reference into the checked-dynamic representation), and a custom fetch
// cast to that type through `unknown` keeps working, including an extra
// init member written through a record view (the dispatcher pattern).
import * as http from "node:http";
type FetchImpl = typeof globalThis.fetch;
interface DispatchingInit {
method?: string;
headers?: Record<string, string>;
dispatcher?: { name: string };
}
function customFetch(input: string | URL, init?: DispatchingInit): Promise<Response> {
console.log("custom fetch", typeof input === "string" ? "string" : "url", init?.dispatcher?.name ?? "none");
return fetch(input, { method: init?.method, headers: init?.headers });
}
const server = http.createServer((req, res) => res.end(`ok ${req.method} ${req.url}`));
server.listen(0, "127.0.0.1", async () => {
const address = server.address();
const port = address !== null && typeof address === "object" ? address.port : 0;
const impl: FetchImpl = customFetch as unknown as FetchImpl;
const init: RequestInit = { method: "POST" };
(init as Record<string, unknown>).dispatcher = { name: "pinned" };
const res = await impl(`http://127.0.0.1:${port}/a`, init);
console.log(res.status, await res.text());
const viaUrl = await impl(new URL(`http://127.0.0.1:${port}/b`));
console.log(viaUrl.status, await viaUrl.text());
const direct = await fetch(new URL(`http://127.0.0.1:${port}/c`));
console.log(direct.status, await direct.text());
server.close();
});
function describeInput(input: string | Request | URL): string {
if (typeof input === "string") return `string ${input}`;
if (input instanceof URL) return `url ${input.href} ${input.hostname}`;
return `request ${input.method} ${input.url}`;
}
console.log(describeInput("http://a.test/x"));
const u = new URL("http://b.test:8080/p?q=1");
console.log(describeInput(u));
console.log(describeInput(new Request("http://c.test/r", { method: "PUT" })));
const items: (string | URL | Request)[] = [u, "s"];
console.log(items.length, items[0] === u, String(items[0]));
let slot: string | URL | Request = u;
if (slot instanceof URL) { slot.pathname = "/changed"; }
console.log(u.href);
const params: string | Request | URL = new URL("http://d.test/?b=2");
console.log(typeof params === "string" ? params : params instanceof URL ? params.searchParams.get("b") : "req");
+2 -1
View File
@@ -18,6 +18,7 @@ import {
setGlobalDispatcher,
} from "undici";
import type { Dispatcher } from "undici";
import { checkPeers } from "./peers.ts";
const server = http.createServer((req, res) => {
const url = req.url ?? "/";
@@ -195,7 +196,6 @@ server.listen(0, "127.0.0.1", async () => {
for (let i = 0; i < 4; i++) {
const r = await request(`${origin}/r${i}`, { dispatcher: agent });
await r.body.text();
await turn();
}
console.log("connections opened:", agent.connections);
console.log("stats:", agent.statsFor(origin));
@@ -294,5 +294,6 @@ server.listen(0, "127.0.0.1", async () => {
await agent.close();
server.close();
await checkPeers();
console.log("done");
});
+183
View File
@@ -0,0 +1,183 @@
// Requests against a raw TCP peer that sends malformed or endless
// responses: header values with CR or LF never reach the wire, a
// chunk-size line is read as llhttp reads it (so an endless one holds no
// memory), and fetch() resolves at the headers with a body that streams,
// cancels and fails as undici's does.
import * as net from "node:net";
import type { Socket } from "node:net";
import { gzipSync } from "node:zlib";
import { Agent, errors, fetch, request } from "undici";
function describe(err: unknown): string {
// HTTPParserError extends Error, not UndiciError.
if (err instanceof errors.HTTPParserError) return `${err.name} ${err.code ?? ""} ${err.message}`;
if (err instanceof errors.UndiciError) return `${err.name} ${err.code} ${err.message}`;
if (err instanceof DOMException) return `${err.name} ${err.message}`;
if (err instanceof Error) {
const code = (err as NodeJS.ErrnoException).code ?? "";
const cause = err.cause !== undefined ? ` cause: ${describe(err.cause)}` : "";
return `${err.name} ${code} ${err.message}${cause}`;
}
return String(err);
}
const MiB = 1024 * 1024;
const text = "hello ".repeat(2000);
const packed = gzipSync(text);
// Chunked bodies whose chunk-size lines the client reads. Each is a
// complete response, or fails at the size line.
const chunkCases: [string, string][] = [
["plain", "5\r\nhello\r\n0\r\n\r\n"],
["extensions", '5;a=b;c="d e";f\r\nhello\r\n0\r\n\r\n'],
["quoted pair", '5;a="b\\"c"\r\nhello\r\n0\r\n\r\n'],
["value then extension", "5;a=b;c=d\r\nhello\r\n0\r\n\r\n"],
["space in name", "5;a b\r\nhello\r\n0\r\n\r\n"],
["empty extension", "5;\r\nhello\r\n0\r\n\r\n"],
["bad value", "5;a=b c\r\nhello\r\n0\r\n\r\n"],
["bad quoted value", '5;a="b\x01"\r\nhello\r\n0\r\n\r\n'],
["after quoted value", '5;a="b"c\r\nhello\r\n0\r\n\r\n'],
["LF after name", "5;a\nhello\r\n0\r\n\r\n"],
["LF after value", "5;a=b\nhello\r\n0\r\n\r\n"],
["space after size", "5 \r\nhello\r\n0\r\n\r\n"],
["space before size", " 5\r\nhello\r\n0\r\n\r\n"],
["empty size", "\r\nhello\r\n0\r\n\r\n"],
["not hex", "g\r\nhello\r\n0\r\n\r\n"],
["LF only", "5\nhello\r\n0\r\n\r\n"],
["CR then other byte", "5\rXhello\r\n0\r\n\r\n"],
["16 hex digits", `${"0".repeat(11)}00005\r\nhello\r\n0\r\n\r\n`],
["17 significant digits", "10000000000000000\r\nhello\r\n0\r\n\r\n"],
["1 MiB of leading zeros", `${"0".repeat(MiB)}5\r\nhello\r\n0\r\n\r\n`],
["1 MiB extension", `5;a=${"b".repeat(MiB)}\r\nhello\r\n0\r\n\r\n`],
["1 MiB of digits", `1${"0".repeat(MiB)}\r\nhello\r\n0\r\n\r\n`],
["1 MiB of spaces", `5${" ".repeat(MiB)}\r\nhello\r\n0\r\n\r\n`],
];
let chunkBody = "";
let closes = 0;
const peer = net.createServer((s: Socket) => {
s.on("error", () => {});
s.on("close", () => {
closes++;
});
let head = "";
s.on("data", (d: Buffer) => {
head += d.toString("latin1");
if (!head.includes("\r\n\r\n")) return;
const lines = head.split("\r\n");
const path = lines[0]!.split(" ")[1]!;
head = "";
const send = (status: string, fields: string, body: string): void => {
s.write(`HTTP/1.1 ${status}\r\n${fields}\r\n`);
s.write(body);
};
if (path === "/echo") {
const received = lines.slice(1).filter((l) => l !== "");
send("200 OK", `Content-Length: ${JSON.stringify(received).length}\r\n`, JSON.stringify(received));
} else if (path === "/chunked") {
send("200 OK", "Transfer-Encoding: chunked\r\n", chunkBody);
} else if (path === "/endless") {
s.write("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n");
const timer = setInterval(() => s.write("4\r\ndata\r\n"), 5);
s.on("close", () => clearInterval(timer));
} else if (path === "/gzip") {
s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${packed.length}\r\n\r\n`);
for (let i = 0; i < packed.length; i += 50) s.write(packed.subarray(i, i + 50));
} else if (path === "/gzip-truncated") {
const part = packed.subarray(0, packed.length - 12);
s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${part.length}\r\n\r\n`);
s.write(part);
} else if (path === "/gzip-empty") {
send("200 OK", "Content-Encoding: gzip\r\nContent-Length: 0\r\n", "");
} else if (path === "/gzip-bad") {
send("200 OK", "Content-Encoding: gzip\r\nContent-Length: 8\r\n", "not gzip");
} else if (path === "/cut") {
s.write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n0123456789");
setTimeout(() => s.destroy(), 20);
} else if (path === "/no-content") {
send("204 No Content", "", "");
} else {
send("200 OK", "Content-Length: 5\r\n", "plain");
}
});
});
export async function checkPeers(): Promise<void> {
await new Promise<void>((resolve) => peer.listen(0, "127.0.0.1", () => resolve()));
const origin = `http://127.0.0.1:${(peer.address() as net.AddressInfo).port}`;
const dispatcher = new Agent();
console.log("== peers: header values");
// undici 5.12 writes a Host value with CR or LF into the request head;
// the built-in implementation rejects it as undici 5.29 does
// (package-shims.test.ts covers that).
for (const [name, value] of [["x-test", "a\r\nX-Injected: 1"]]) {
try {
const res = await request(`${origin}/echo`, { dispatcher, headers: { [name!]: value! } });
console.log(JSON.stringify(value), "sent:", await res.body.text());
} catch (err) {
console.log(JSON.stringify(value), describe(err));
}
}
console.log("== peers: chunk-size lines");
for (const [label, body] of chunkCases) {
chunkBody = body;
try {
const res = await request(`${origin}/chunked`, { dispatcher });
console.log(label, res.statusCode, JSON.stringify(await res.body.text()));
} catch (err) {
console.log(label, describe(err));
}
}
console.log("== peers: fetch bodies");
const tryFetch = async (
label: string,
path: string,
read: (res: Response) => Promise<void>,
method = "GET",
signal: AbortSignal | null = null,
): Promise<void> => {
try {
const res = await fetch(`${origin}${path}`, { dispatcher, method, signal });
console.log(label, res.status, res.body === null ? "null body" : "body");
await read(res);
} catch (err) {
console.log(label, describe(err));
}
};
await tryFetch("endless", "/endless", async (res) => {
const reader = res.body!.getReader();
const first = await reader.read();
console.log("first chunk:", first.done, new TextDecoder().decode(first.value!).slice(0, 4));
const before = closes;
await reader.cancel();
await new Promise<void>((resolve) => setTimeout(resolve, 100));
console.log("cancel closed the connection:", closes > before);
});
await tryFetch("gzip", "/gzip", async (res) => console.log("text matches:", (await res.text()) === text));
await tryFetch("gzip-truncated", "/gzip-truncated", async (res) =>
console.log("length", (await res.text()).length),
);
await tryFetch("gzip-empty", "/gzip-empty", async (res) => console.log(JSON.stringify(await res.text())));
await tryFetch("gzip-bad", "/gzip-bad", async (res) => console.log(JSON.stringify(await res.text())));
await tryFetch("cut", "/cut", async (res) => console.log(JSON.stringify(await res.text())));
const controller = new AbortController();
await tryFetch(
"abort after headers",
"/endless",
async (res) => {
controller.abort();
console.log(await res.text());
},
"GET",
controller.signal,
);
await tryFetch("no-content", "/no-content", async (res) => console.log(JSON.stringify(await res.text())));
await tryFetch("head", "/plain", async (res) => console.log(JSON.stringify(await res.text())), "HEAD");
await dispatcher.close();
peer.close();
}
+61 -1
View File
@@ -20,6 +20,7 @@ import {
setGlobalDispatcher,
} from "undici";
import type { Dispatcher } from "undici";
import { checkPeers } from "./peers.ts";
const server = http.createServer((req, res) => {
const url = req.url ?? "/";
@@ -31,6 +32,12 @@ const server = http.createServer((req, res) => {
res.end("cookies");
} else if (url === "/slow") {
setTimeout(() => res.end("late"), 1500);
} else if (url === "/stall-body") {
// Part of a body that never ends.
res.writeHead(200, { "content-type": "text/plain" });
res.write("part");
} else if (url === "/trace") {
res.end(`x-trace ${req.headers["x-trace"] ?? "-"}`);
} else if (url === "/missing") {
res.statusCode = 404;
res.setHeader("content-type", "application/json");
@@ -71,6 +78,28 @@ const byRequest = new WeakMap<object, Telemetry>();
channel("undici:request:create").subscribe((message) => {
const { request } = message as { request: object };
if (current !== null) byRequest.set(request, current);
// Read as an untyped value, as a JavaScript instrumentation reads it.
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const traced: any = request;
if (traced.path === "/trace") {
// A tracing instrumentation's header injection.
console.log(` [diag] create completed ${traced.completed}, ${typeof traced.addHeader}`);
console.log(` [diag] addHeader returns the request: ${traced.addHeader("x-trace", "span-1") === request}`);
console.log(` [diag] headers ${JSON.stringify(traced.headers)}`);
try {
traced.addHeader("x-bad", "a\r\nb");
} catch (err) {
console.log(` [diag] invalid header: ${describe(err)}`);
}
tracedRequests.add(request);
}
});
const tracedRequests = new Set<object>();
channel("undici:request:trailers").subscribe((message) => {
const { request } = message as { request: object };
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const traced: any = request;
if (tracedRequests.has(request)) console.log(` [diag] trailers completed ${traced.completed}`);
});
channel("undici:request:headers").subscribe((message) => {
const { request, response } = message as {
@@ -201,7 +230,6 @@ server.listen(0, "127.0.0.1", async () => {
for (let i = 0; i < 4; i++) {
const r = await request(`${origin}/r${i}`, { dispatcher: agent });
await r.body.text();
await turn();
}
console.log("connections opened:", agent.connections);
console.log("stats:", agent.statsFor(origin));
@@ -323,6 +351,37 @@ server.listen(0, "127.0.0.1", async () => {
);
});
console.log("== diagnostics request headers");
const trace = await request(`${origin}/trace`, { dispatcher: agent });
console.log("traced body:", await trace.body.text());
console.log("== body timeout after a pause");
const stallAgent = new Agent();
await new Promise<void>((resolve) => {
stallAgent.dispatch(
{ origin, path: "/stall-body", method: "GET", bodyTimeout: 100 },
{
onRequestStart() {},
onResponseStart(controller, statusCode) {
// Paused past bodyTimeout, then resumed: the timeout applies again.
console.log("paused response", statusCode);
controller.pause();
setTimeout(() => controller.resume(), 1500);
},
onResponseData() {},
onResponseEnd() {
console.log("stalled body ended");
resolve();
},
onResponseError(_controller, err) {
console.log("stalled body:", describe(err));
resolve();
},
},
);
});
await stallAgent.close();
console.log("== fetch");
const viaAgent = await fetch(`${origin}/json`, { dispatcher: agent });
console.log("through agent:", viaAgent.status, viaAgent.statusText, viaAgent.headers.get("content-type"), await viaAgent.text());
@@ -364,5 +423,6 @@ server.listen(0, "127.0.0.1", async () => {
await agent.close();
server.close();
await checkPeers();
console.log("done");
});
+197
View File
@@ -0,0 +1,197 @@
// Requests against a raw TCP peer that sends malformed or endless
// responses: header values with CR or LF never reach the wire, a
// chunk-size line is read as llhttp reads it (so an endless one holds no
// memory), and fetch() resolves at the headers with a body that streams,
// cancels and fails as undici's does.
import * as net from "node:net";
import type { Socket } from "node:net";
import { gzipSync } from "node:zlib";
import { Agent, errors, fetch, request } from "undici";
function describe(err: unknown): string {
// HTTPParserError extends Error, not UndiciError.
if (err instanceof errors.HTTPParserError) return `${err.name} ${err.code ?? ""} ${err.message}`;
if (err instanceof errors.UndiciError) return `${err.name} ${err.code} ${err.message}`;
if (err instanceof DOMException) return `${err.name} ${err.message}`;
if (err instanceof Error) {
const code = (err as NodeJS.ErrnoException).code ?? "";
const cause = err.cause !== undefined ? ` cause: ${describe(err.cause)}` : "";
return `${err.name} ${code} ${err.message}${cause}`;
}
return String(err);
}
const MiB = 1024 * 1024;
const text = "hello ".repeat(2000);
const packed = gzipSync(text);
// Chunked bodies whose chunk-size lines the client reads. Each is a
// complete response, or fails at the size line.
const chunkCases: [string, string][] = [
["plain", "5\r\nhello\r\n0\r\n\r\n"],
["extensions", '5;a=b;c="d e";f\r\nhello\r\n0\r\n\r\n'],
["quoted pair", '5;a="b\\"c"\r\nhello\r\n0\r\n\r\n'],
["value then extension", "5;a=b;c=d\r\nhello\r\n0\r\n\r\n"],
["space in name", "5;a b\r\nhello\r\n0\r\n\r\n"],
["empty extension", "5;\r\nhello\r\n0\r\n\r\n"],
["bad value", "5;a=b c\r\nhello\r\n0\r\n\r\n"],
["bad quoted value", '5;a="b\x01"\r\nhello\r\n0\r\n\r\n'],
["after quoted value", '5;a="b"c\r\nhello\r\n0\r\n\r\n'],
["LF after name", "5;a\nhello\r\n0\r\n\r\n"],
["LF after value", "5;a=b\nhello\r\n0\r\n\r\n"],
["space after size", "5 \r\nhello\r\n0\r\n\r\n"],
["space before size", " 5\r\nhello\r\n0\r\n\r\n"],
["empty size", "\r\nhello\r\n0\r\n\r\n"],
["not hex", "g\r\nhello\r\n0\r\n\r\n"],
["LF only", "5\nhello\r\n0\r\n\r\n"],
["CR then other byte", "5\rXhello\r\n0\r\n\r\n"],
["16 hex digits", `${"0".repeat(11)}00005\r\nhello\r\n0\r\n\r\n`],
["17 significant digits", "10000000000000000\r\nhello\r\n0\r\n\r\n"],
["1 MiB of leading zeros", `${"0".repeat(MiB)}5\r\nhello\r\n0\r\n\r\n`],
["1 MiB extension", `5;a=${"b".repeat(MiB)}\r\nhello\r\n0\r\n\r\n`],
["1 MiB of digits", `1${"0".repeat(MiB)}\r\nhello\r\n0\r\n\r\n`],
["1 MiB of spaces", `5${" ".repeat(MiB)}\r\nhello\r\n0\r\n\r\n`],
];
let chunkBody = "";
let redirectSocket: Socket | null = null;
let closes = 0;
const peer = net.createServer((s: Socket) => {
s.on("error", () => {});
s.on("close", () => {
closes++;
});
let head = "";
s.on("data", (d: Buffer) => {
head += d.toString("latin1");
if (!head.includes("\r\n\r\n")) return;
const lines = head.split("\r\n");
const path = lines[0]!.split(" ")[1]!;
head = "";
const send = (status: string, fields: string, body: string): void => {
s.write(`HTTP/1.1 ${status}\r\n${fields}\r\n`);
s.write(body);
};
if (path === "/echo") {
const received = lines.slice(1).filter((l) => l !== "");
send("200 OK", `Content-Length: ${JSON.stringify(received).length}\r\n`, JSON.stringify(received));
} else if (path === "/chunked") {
send("200 OK", "Transfer-Encoding: chunked\r\n", chunkBody);
} else if (path === "/endless") {
s.write("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n");
const timer = setInterval(() => s.write("4\r\ndata\r\n"), 5);
s.on("close", () => clearInterval(timer));
} else if (path === "/gzip") {
s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${packed.length}\r\n\r\n`);
for (let i = 0; i < packed.length; i += 50) s.write(packed.subarray(i, i + 50));
} else if (path === "/gzip-truncated") {
const part = packed.subarray(0, packed.length - 12);
s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${part.length}\r\n\r\n`);
s.write(part);
} else if (path === "/gzip-empty") {
send("200 OK", "Content-Encoding: gzip\r\nContent-Length: 0\r\n", "");
} else if (path === "/gzip-bad") {
send("200 OK", "Content-Encoding: gzip\r\nContent-Length: 8\r\n", "not gzip");
} else if (path === "/cut") {
s.write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n0123456789");
setTimeout(() => s.destroy(), 20);
} else if (path === "/many-codings") {
send("200 OK", "Content-Encoding: gzip, gzip, gzip, gzip, gzip, gzip\r\nContent-Length: 2\r\n", "ok");
} else if (path === "/redirect-endless") {
redirectSocket = s;
s.write("HTTP/1.1 302 Found\r\nLocation: /plain\r\nTransfer-Encoding: chunked\r\n\r\n");
const timer = setInterval(() => s.write("4\r\ndata\r\n"), 5);
s.on("close", () => clearInterval(timer));
} else if (path === "/no-content") {
send("204 No Content", "", "");
} else {
send("200 OK", "Content-Length: 5\r\n", "plain");
}
});
});
export async function checkPeers(): Promise<void> {
await new Promise<void>((resolve) => peer.listen(0, "127.0.0.1", () => resolve()));
const origin = `http://127.0.0.1:${(peer.address() as net.AddressInfo).port}`;
const dispatcher = new Agent();
console.log("== peers: header values");
for (const [name, value] of [
["host", "a\r\nX-Injected: 1"],
["host", "a\nb"],
["x-test", "a\r\nX-Injected: 1"],
]) {
try {
const res = await request(`${origin}/echo`, { dispatcher, headers: { [name!]: value! } });
console.log(JSON.stringify(value), "sent:", await res.body.text());
} catch (err) {
console.log(JSON.stringify(value), describe(err));
}
}
console.log("== peers: chunk-size lines");
for (const [label, body] of chunkCases) {
chunkBody = body;
try {
const res = await request(`${origin}/chunked`, { dispatcher });
console.log(label, res.statusCode, JSON.stringify(await res.body.text()));
} catch (err) {
console.log(label, describe(err));
}
}
console.log("== peers: fetch bodies");
const tryFetch = async (
label: string,
path: string,
read: (res: Response) => Promise<void>,
method = "GET",
signal: AbortSignal | null = null,
): Promise<void> => {
try {
const res = await fetch(`${origin}${path}`, { dispatcher, method, signal });
console.log(label, res.status, res.body === null ? "null body" : "body");
await read(res);
} catch (err) {
console.log(label, describe(err));
}
};
await tryFetch("endless", "/endless", async (res) => {
const reader = res.body!.getReader();
const first = await reader.read();
console.log("first chunk:", first.done, new TextDecoder().decode(first.value!).slice(0, 4));
const before = closes;
await reader.cancel();
await new Promise<void>((resolve) => setTimeout(resolve, 100));
console.log("cancel closed the connection:", closes > before);
});
await tryFetch("gzip", "/gzip", async (res) => console.log("text matches:", (await res.text()) === text));
await tryFetch("gzip-truncated", "/gzip-truncated", async (res) =>
console.log("length", (await res.text()).length),
);
await tryFetch("gzip-empty", "/gzip-empty", async (res) => console.log(JSON.stringify(await res.text())));
await tryFetch("gzip-bad", "/gzip-bad", async (res) => console.log(JSON.stringify(await res.text())));
await tryFetch("cut", "/cut", async (res) => console.log(JSON.stringify(await res.text())));
const controller = new AbortController();
await tryFetch(
"abort after headers",
"/endless",
async (res) => {
controller.abort();
console.log(await res.text());
},
"GET",
controller.signal,
);
await tryFetch("many-codings", "/many-codings", async (res) => console.log(await res.text()));
await tryFetch("redirect-endless", "/redirect-endless", async (res) =>
console.log("followed:", await res.text()),
);
await tryFetch("no-content", "/no-content", async (res) => console.log(JSON.stringify(await res.text())));
await tryFetch("head", "/plain", async (res) => console.log(JSON.stringify(await res.text())), "HEAD");
if (redirectSocket !== null) (redirectSocket as Socket).destroy();
await dispatcher.close();
peer.close();
}
+29
View File
@@ -0,0 +1,29 @@
// Reads every response of undici-hostile-peer.mjs (port in argv[2]): the
// endless chunk-size lines through request(), the large and the gzipped
// bodies through fetch() with a pause after the first chunk. Memory stays
// bounded by the parser's and the body stream's buffers.
import { Agent, fetch, request } from "undici";
const MiB = 1024 * 1024;
const origin = `http://127.0.0.1:${process.argv[2]}`;
const dispatcher = new Agent();
for (const path of ["/extension", "/zeros"]) {
const res = await request(`${origin}${path}`, { dispatcher });
console.log(path, res.statusCode, await res.body.text());
}
for (const path of ["/big", "/bomb"]) {
const res = await fetch(`${origin}${path}`, { dispatcher });
const reader = res.body!.getReader();
const first = await reader.read();
let total = first.value!.length;
await new Promise<void>((resolve) => setTimeout(resolve, 500));
for (;;) {
const { done, value } = await reader.read();
if (done) break;
total += value!.length;
}
console.log(path, res.status, total / MiB, "MiB");
}
const rss = process.memoryUsage().rss;
console.log(rss < 48 * MiB ? "rss under 48 MiB" : `rss ${Math.round(rss / MiB)} MiB`);
await dispatcher.close();
+64
View File
@@ -0,0 +1,64 @@
// A peer for package-shims.test.ts: prints its port, then answers
// /extension a chunk whose size line carries 64 MiB of extension,
// /zeros a chunk size with 64 MiB of leading zeros,
// /big a 64 MiB body, as fast as the socket takes it,
// /bomb 64 MiB of zeros, gzipped,
// /redirect-endless a 302 to /redirect-closed whose body never ends,
// /redirect-closed "closed" once the last /redirect-endless connection
// closed (waiting up to a second), else "open",
// and exits when its stdin closes.
import * as net from "node:net";
import * as zlib from "node:zlib";
const MiB = 1024 * 1024;
const size = 64 * MiB;
const bomb = zlib.gzipSync(Buffer.alloc(size));
function stream(s, head, piece, tail) {
s.write(head);
let sent = 0;
const pump = () => {
while (sent < size) {
sent += piece.length;
if (!s.write(piece)) return void s.once("drain", pump);
}
s.write(tail);
};
pump();
}
let redirectClosed = Promise.resolve();
const server = net.createServer((s) => {
s.on("error", () => {});
let head = "";
s.on("data", (d) => {
head += d.toString("latin1");
if (!head.includes("\r\n\r\n")) return;
const path = head.split(" ")[1];
head = "";
const chunked = "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n";
if (path === "/extension") {
stream(s, `${chunked}5;a=`, Buffer.alloc(64 * 1024, 98), "\r\nhello\r\n0\r\n\r\n");
} else if (path === "/zeros") {
stream(s, chunked, Buffer.alloc(64 * 1024, 48), "5\r\nhello\r\n0\r\n\r\n");
} else if (path === "/big") {
stream(s, `HTTP/1.1 200 OK\r\nContent-Length: ${size}\r\n\r\n`, Buffer.alloc(64 * 1024, 97), "");
} else if (path === "/redirect-endless") {
redirectClosed = new Promise((resolve) => s.on("close", resolve));
s.write("HTTP/1.1 302 Found\r\nLocation: /redirect-closed\r\nTransfer-Encoding: chunked\r\n\r\n");
const timer = setInterval(() => s.write("4\r\ndata\r\n"), 5);
s.on("close", () => clearInterval(timer));
} else if (path === "/redirect-closed") {
const timeout = new Promise((resolve) => setTimeout(() => resolve("open"), 1000));
void Promise.race([redirectClosed.then(() => "closed"), timeout]).then((state) => {
s.write(`HTTP/1.1 200 OK\r\nContent-Length: ${state.length}\r\n\r\n${state}`);
});
} else {
s.write(`HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: ${bomb.length}\r\n\r\n`);
s.write(bomb);
}
});
});
server.listen(0, "127.0.0.1", () => console.log(server.address().port));
process.stdin.on("end", () => process.exit(0)).resume();
+19
View File
@@ -0,0 +1,19 @@
// Follows and rejects redirects whose bodies never end
// (undici-hostile-peer.mjs, port in argv[2]) through a pool of one
// connection. The discarded redirect's request is aborted, so its
// connection closes, the follow-up request gets a connection, and the peer
// reports the redirect's connection closed.
import { Agent, fetch } from "undici";
const origin = `http://127.0.0.1:${process.argv[2]}`;
const dispatcher = new Agent({ connections: 1 });
const followed = await fetch(`${origin}/redirect-endless`, { dispatcher });
console.log("follow", followed.status, await followed.text());
try {
await fetch(`${origin}/redirect-endless`, { dispatcher, redirect: "error" });
} catch (err) {
console.log("error mode", (err as Error).name, (err as Error).message);
}
const after = await fetch(`${origin}/redirect-closed`, { dispatcher });
console.log("after error mode", await after.text());
await dispatcher.close();
+26
View File
@@ -0,0 +1,26 @@
// A Host header value with CR or LF, through the undici 5.x profile: it
// throws before the request is written, as undici 5.29 does (5.12, the
// vendored fixture version, writes the value into the request head).
import * as http from "node:http";
import { Agent, errors, request } from "undici";
const server = http.createServer((_req, res) => {
res.statusCode = 204;
res.end();
});
server.listen(0, "127.0.0.1", async () => {
const origin = `http://127.0.0.1:${(server.address() as { port: number }).port}`;
const dispatcher = new Agent();
for (const host of ["ok.test", "a\r\nX-Injected: 1", "a\nb"]) {
try {
const res = await request(`${origin}/`, { dispatcher, headers: { host } });
await res.body.dump();
console.log(JSON.stringify(host), res.statusCode);
} catch (err) {
if (err instanceof errors.UndiciError) console.log(JSON.stringify(host), err.name, err.code, err.message);
else throw err;
}
}
await dispatcher.close();
server.close();
});
@@ -0,0 +1 @@
{ "plain": 2981, "san": 38628 }
@@ -0,0 +1 @@
{ "plain": 3462, "san": 72216 }
+8 -4
View File
@@ -85,10 +85,14 @@ test("@types/node 20 class-declared timer handles compile and match Node", async
});
if (!result.ok)
throw new Error(result.diagnostics.map((d) => `${d.code}: ${d.message}`).join("\n"));
const [node, native] = await Promise.all([
execFileAsync(process.execPath, [join(dir, "main.ts")], { cwd: dir, encoding: "utf8" }),
execFileAsync(result.binaryPath, [], { cwd: dir, encoding: "utf8" }),
]);
// One after the other, and the timer 200 ms out: under load Node's own
// startup can outlast a short timer, which then runs before the
// immediate.
const node = await execFileAsync(process.execPath, [join(dir, "main.ts")], {
cwd: dir,
encoding: "utf8",
});
const native = await execFileAsync(result.binaryPath, [], { cwd: dir, encoding: "utf8" });
expect(native.stdout).toBe(node.stdout);
expect(native.stdout).toBe("hasRef true\nimmediate\nfired true\n");
}, 180_000);
+6 -1
View File
@@ -111,6 +111,11 @@ const KNOWN: Record<string, Known> = {
const cases = shardSelect(npmCases(fixturesRoot), (c) => c.name);
/** Per-case timeouts past the default 120 s: the undici cases compile the
* built-in undici implementation's module graph, which a sanitized build
* on a CI runner takes minutes to compile. */
const TIMEOUTS: Record<string, number> = { undici5: 600_000, undici7: 600_000 };
interface RunResult {
stdout: Buffer;
stderr: Buffer;
@@ -233,7 +238,7 @@ describe(`npm native (${cases.length} programs${sanitize ? ", sanitized" : ""}${
}
await expectMatchesNode(c, built.binary);
},
120_000,
TIMEOUTS[c.name] ?? 120_000,
);
}
});
+81 -2
View File
@@ -9,12 +9,19 @@
* - a shimmed package that is not installed, or whose major version has no
* shim, resolves exactly as before (here: refused like any missing or
* uncompiled package);
* - the undici shims keep memory bounded against a peer that sends endless
* chunk-size lines and large or compressed bodies, and the 5.x profile
* rejects a Host value with CR or LF as undici 5.29 does (the vendored
* 5.12 writes it into the request);
* - fetch aborts a redirect it follows or rejects, so a redirect body that
* never ends does not keep its connection busy (undici 7 leaves that
* connection open with its body paused).
* - a CommonJS package's require() of a shimmed package (fs-extra, required
* by an installed package's JavaScript) compiles the shim, and matches
* Node running the real package.
*/
import { execFile } from "node:child_process";
import { cpSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { execFile, spawn } from "node:child_process";
import { copyFileSync, cpSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { promisify } from "node:util";
@@ -24,6 +31,7 @@ import { compile } from "@scriptc/compiler";
const execFileAsync = promisify(execFile);
const repoRoot = join(import.meta.dirname, "../..");
const casesRoot = join(repoRoot, "tests/fixtures/npm/cases");
const shimFixtures = join(repoRoot, "tests/fixtures/package-shims");
const fsExtraModules = join(repoRoot, "tests/corpus/fs-extra-package/node_modules");
const sanitize = process.env["SCRIPTC_SAN"] === "1";
const scratch = mkdtempSync(join(tmpdir(), "scriptc-package-shims-"));
@@ -55,6 +63,37 @@ function installEncoding(dir: string): void {
);
}
/** Compiles `entry` (a file in tests/fixtures/package-shims) as main.ts of
* a copy of the npm case `name`, which carries the package. */
async function buildInCase(name: string, entry: string): Promise<{ dir: string; binary: string }> {
const dir = copyCase(name, `${name}-${entry.replace(/\..*$/, "")}`);
copyFileSync(join(shimFixtures, entry), join(dir, "main.ts"));
const out = join(dir, ".scriptc");
const result = await compile(join(dir, "main.ts"), {
outPath: join(out, "program"),
outDir: out,
sanitize,
});
if (!result.ok) throw new Error(result.diagnostics.map((d) => d.message).join("\n"));
return { dir, binary: result.binaryPath };
}
/** Runs `binary <port>` against undici-hostile-peer.mjs. */
async function runAgainstPeer(binary: string, cwd: string): Promise<string> {
const peer = spawn(process.execPath, [join(shimFixtures, "undici-hostile-peer.mjs")], {
stdio: ["pipe", "pipe", "inherit"],
});
try {
const port = await new Promise<string>((resolve, reject) => {
peer.stdout.once("data", (d: Buffer) => resolve(d.toString().trim()));
peer.once("exit", (code) => reject(new Error(`peer exited with ${code}`)));
});
return await run(binary, [port], cwd);
} finally {
peer.stdin.end();
}
}
describe("package shims", () => {
test("an unloaded package is never loaded, even when installed", async () => {
const original = join(casesRoot, "unloaded-optional-dependency");
@@ -213,4 +252,44 @@ describe("package shims", () => {
expect(result.diagnostics.map((d) => d.message).join("\n")).not.toContain("shims/undici");
}
}, 180_000);
for (const name of ["undici7", "undici5"]) {
test(`${name}: endless chunk-size lines and large bodies stay in bounded memory`, async () => {
const { dir, binary } = await buildInCase(name, "undici-bounded.ts");
const out = await runAgainstPeer(binary, dir);
expect(out.split("\n").slice(0, 4)).toEqual([
"/extension 200 hello",
"/zeros 200 hello",
"/big 200 64 MiB",
"/bomb 200 64 MiB",
]);
// ASan's shadow memory and quarantine keep freed buffers mapped.
if (!sanitize) expect(out.split("\n")[4]).toBe("rss under 48 MiB");
}, 300_000);
}
test("undici7: fetch closes the connection of a redirect it follows or rejects", async () => {
const { dir, binary } = await buildInCase("undici7", "undici-redirect.ts");
expect(await runAgainstPeer(binary, dir)).toBe(
[
"follow 200 closed",
"error mode TypeError fetch failed",
"after error mode closed",
"",
].join("\n"),
);
}, 180_000);
test("undici5: a Host value with CR or LF throws before the request is written", async () => {
const { dir, binary } = await buildInCase("undici5", "undici5-host.ts");
// undici 5.29.0's output.
expect(await run(binary, [], dir)).toBe(
[
'"ok.test" 204',
'"a\\r\\nX-Injected: 1" InvalidArgumentError UND_ERR_INVALID_ARG invalid host header',
'"a\\nb" InvalidArgumentError UND_ERR_INVALID_ARG invalid host header',
"",
].join("\n"),
);
}, 180_000);
});
+42
View File
@@ -52,6 +52,48 @@ test("node-types: filesystem capacity supports typed number and BigInt results u
expect(comparableStderr(native.stderr)).toBe(node.stderr);
});
test("node-types: Headers, Request and Response members typed by undici-types run natively", async () => {
const outDir = outDirFor("fetch-handle-members");
const entry = join(nodeTypesDir, "fetch-handle-members.ts");
const result = await compile(entry, {
outPath: join(
outDir,
process.platform === "win32" ? "fetch-handle-members.exe" : "fetch-handle-members",
),
outDir,
sanitize,
});
expect(result.ok, !result.ok ? JSON.stringify(result.diagnostics, null, 2) : "").toBe(true);
if (!result.ok) return;
const [native, node] = await Promise.all([
execFileAsync(result.binaryPath),
execFileAsync(process.execPath, ["--disable-warning=MODULE_TYPELESS_PACKAGE_JSON", entry]),
]);
expect(native.stdout).toBe(node.stdout);
expect(comparableStderr(native.stderr)).toBe(node.stderr);
});
test("node-types: fetch's string | URL | Request input and a custom fetch cast to typeof fetch", async () => {
const outDir = outDirFor("fetch-input-unions");
const entry = join(nodeTypesDir, "fetch-input-unions.ts");
const result = await compile(entry, {
outPath: join(
outDir,
process.platform === "win32" ? "fetch-input-unions.exe" : "fetch-input-unions",
),
outDir,
sanitize,
});
expect(result.ok, !result.ok ? JSON.stringify(result.diagnostics, null, 2) : "").toBe(true);
if (!result.ok) return;
const [native, node] = await Promise.all([
execFileAsync(result.binaryPath),
execFileAsync(process.execPath, ["--disable-warning=MODULE_TYPELESS_PACKAGE_JSON", entry]),
]);
expect(native.stdout).toBe(node.stdout);
expect(comparableStderr(native.stderr)).toBe(node.stderr);
});
for (const [name, title] of [
[
"event-loop-metrics",