Undici (#5117)

* replace node-fetch with undici * use git version for now * detect and expose ReadableStream (and friends) * omit body where appropriate * always stream responses * delete connection header * specify host * lint * lint * update to undici latest * try serving large response from endpoint * update test * remove node-fetch * changeset * only install necessary polyfills * cancel aborted requests * update docs * update types - node streams are no longer supported * update test Co-authored-by: Ben McCann <[email protected]>
sveltejs · Jun 28, 2022 · 7ed3c95 · 7ed3c95
1 parent f5c8800
commit 7ed3c95
Show file tree

Hide file tree

Showing 15 changed files with 117 additions and 71 deletions.
diff --git a/.changeset/slimy-ways-stare.md b/.changeset/slimy-ways-stare.md
@@ -0,0 +1,5 @@
+---
+'@sveltejs/kit': patch
+---
+
+[breaking] use undici instead of node-fetch
diff --git a/documentation/docs/01-web-standards.md b/documentation/docs/01-web-standards.md
@@ -45,6 +45,10 @@ export function get(event) {
 }
 ```
 
+### Stream APIs
+
+Most of the time, your endpoints will return complete data, as in the `userAgent` example above. Sometimes, you may need to return a response that's too large to fit in memory in one go, or is delivered in chunks, and for this the platform provides [streams](https://developer.mozilla.org/en-US/docs/Web/API/Streams_API) — [ReadableStream](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream), [WritableStream](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream) and [TransformStream](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream).
+
 ### URL APIs
 
 URLs are represented by the [`URL`](https://developer.mozilla.org/en-US/docs/Web/API/URL) interface, which includes useful properties like `origin` and `pathname` (and, in the browser, `hash`). This interface shows up in various places — `event.url` in [hooks](/docs/hooks) and [endpoints](/docs/routing#endpoints), [`$page.url`](/docs/modules#$app-stores) in [pages](/docs/routing#pages), `from` and `to` in [`beforeNavigate` and `afterNavigate`](/docs/modules#$app-navigation) and so on.

diff --git a/packages/kit/package.json b/packages/kit/package.json
@@ -33,7 +33,6 @@
 		"locate-character": "^2.0.5",
 		"marked": "^4.0.16",
 		"mime": "^3.0.0",
-		"node-fetch": "^3.2.4",
 		"port-authority": "^1.2.0",
 		"rollup": "^2.75.3",
 		"selfsigned": "^2.0.1",
@@ -45,6 +44,7 @@
 		"svelte2tsx": "~0.5.10",
 		"tiny-glob": "^0.2.9",
 		"typescript": "^4.7.2",
+		"undici": "^5.4.0",
 		"uvu": "^0.5.3"
 	},
 	"peerDependencies": {

diff --git a/packages/kit/src/node/index.js b/packages/kit/src/node/index.js
@@ -1,4 +1,3 @@
-import { Readable } from 'stream';
 import * as set_cookie_parser from 'set-cookie-parser';
 
 /** @param {import('http').IncomingMessage} req */
@@ -64,6 +63,7 @@ export async function getRequest(base, req) {
 		delete headers[':authority'];
 		delete headers[':scheme'];
 	}
+
 	return new Request(base + req.url, {
 		method: req.method,
 		headers,
@@ -85,13 +85,38 @@ export async function setResponse(res, response) {
 
 	res.writeHead(response.status, headers);
 
-	if (response.body instanceof Readable) {
-		response.body.pipe(res);
-	} else {
-		if (response.body) {
-			res.write(new Uint8Array(await response.arrayBuffer()));
-		}
+	if (response.body) {
+		let cancelled = false;
+
+		const reader = response.body.getReader();
+
+		res.on('close', () => {
+			reader.cancel();
+			cancelled = true;
+		});
+
+		const next = async () => {
+			const { done, value } = await reader.read();
 
+			if (cancelled) return;
+
+			if (done) {
+				res.end();
+				return;
+			}
+
+			res.write(Buffer.from(value), (error) => {
+				if (error) {
+					console.error('Error writing stream', error);
+					res.end();
+				} else {
+					next();
+				}
+			});
+		};
+
+		next();
+	} else {
 		res.end();
 	}
 }
diff --git a/packages/kit/src/node/polyfills.js b/packages/kit/src/node/polyfills.js
@@ -1,4 +1,5 @@
-import fetch, { Response, Request, Headers } from 'node-fetch';
+import { fetch, Response, Request, Headers } from 'undici';
+import { ReadableStream, TransformStream, WritableStream } from 'stream/web';
 import { webcrypto as crypto } from 'crypto';
 
 /** @type {Record<string, any>} */
@@ -7,13 +8,17 @@ const globals = {
 	fetch,
 	Response,
 	Request,
-	Headers
+	Headers,
+	ReadableStream,
+	TransformStream,
+	WritableStream
 };
 
 // exported for dev/preview and node environments
 export function installPolyfills() {
 	for (const name in globals) {
-		// TODO use built-in fetch once https://github.com/nodejs/undici/issues/1262 is resolved
+		if (name in globalThis) continue;
+
 		Object.defineProperty(globalThis, name, {
 			enumerable: true,
 			configurable: true,

diff --git a/packages/kit/src/runtime/server/endpoint.js b/packages/kit/src/runtime/server/endpoint.js
@@ -21,6 +21,8 @@ const text_types = new Set([
 	'multipart/form-data'
 ]);
 
+const bodyless_status_codes = new Set([101, 204, 205, 304]);
+
 /**
  * Decides how the body should be parsed based on its mime type
  *
@@ -124,8 +126,11 @@ export async function render_endpoint(event, mod) {
 		}
 	}
 
-	return new Response(method !== 'head' ? normalized_body : undefined, {
-		status,
-		headers
-	});
+	return new Response(
+		method !== 'head' && !bodyless_status_codes.has(status) ? normalized_body : undefined,
+		{
+			status,
+			headers
+		}
+	);
 }
diff --git a/packages/kit/src/runtime/server/page/load_node.js b/packages/kit/src/runtime/server/page/load_node.js
@@ -250,6 +250,11 @@ export async function load_node({
 						if (cookie) opts.headers.set('cookie', cookie);
 					}
 
+					// we need to delete the connection header, as explained here:
+					// https://github.com/nodejs/undici/issues/1470#issuecomment-1140798467
+					// TODO this may be a case for being selective about which headers we let through
+					opts.headers.delete('connection');
+
 					const external_request = new Request(requested, /** @type {RequestInit} */ (opts));
 					response = await options.hooks.externalFetch.call(null, external_request);
 				}

diff --git a/packages/kit/src/runtime/server/utils.js b/packages/kit/src/runtime/server/utils.js
@@ -39,9 +39,13 @@ export function is_pojo(body) {
 	if (body) {
 		if (body instanceof Uint8Array) return false;
 
-		// body could be a node Readable, but we don't want to import
-		// node built-ins, so we use duck typing
-		if (body._readableState && typeof body.pipe === 'function') return false;
+		// if body is a node Readable, throw an error
+		// TODO remove this for 1.0
+		if (body._readableState && typeof body.pipe === 'function') {
+			throw new Error('Node streams are no longer supported — use a ReadableStream instead');
+		}
+
+		if (body instanceof ReadableStream) return false;
 
 		// similarly, it could be a web ReadableStream
 		if (typeof ReadableStream !== 'undefined' && body instanceof ReadableStream) return false;

diff --git a/packages/kit/src/utils/http.spec.js b/packages/kit/src/utils/http.spec.js
@@ -1,7 +1,9 @@
 import { test } from 'uvu';
 import * as assert from 'uvu/assert';
 import { to_headers } from './http.js';
-import { Headers } from 'node-fetch';
+import { Headers } from 'undici';
+
+// @ts-ignore
 globalThis.Headers = Headers;
 
 test('handle header string value', () => {

diff --git a/...ics/src/routes/load/large-response.svelte → ...c/routes/load/large-response/index.svelte b/...ics/src/routes/load/large-response.svelte → ...c/routes/load/large-response/index.svelte
@@ -1,7 +1,7 @@
 <script context="module">
 	/** @type {import('@sveltejs/kit').Load} */
-	export async function load({ url, fetch }) {
-		const res = await fetch(`http://localhost:${url.searchParams.get('port')}/large-response.json`);
+	export async function load({ fetch }) {
+		const res = await fetch('/load/large-response/text.txt');
 		const text = await res.text();
 
 		return {

diff --git a/packages/kit/test/apps/basics/src/routes/load/large-response/text.txt.js b/packages/kit/test/apps/basics/src/routes/load/large-response/text.txt.js
@@ -0,0 +1,23 @@
+const chunk_size = 50000;
+const chunk_count = 100;
+
+let chunk = '';
+for (let i = 0; i < chunk_size; i += 1) {
+	chunk += String(i % 10);
+}
+
+export function get() {
+	let i = 0;
+
+	return {
+		body: new ReadableStream({
+			pull: (controller) => {
+				if (i++ < chunk_count) {
+					controller.enqueue(chunk);
+				} else {
+					controller.close();
+				}
+			}
+		})
+	};
+}
diff --git a/packages/kit/test/apps/basics/test/test.js b/packages/kit/test/apps/basics/test/test.js
@@ -1360,45 +1360,8 @@ test.describe.parallel('Load', () => {
 	test('handles large responses', async ({ page }) => {
 		await page.goto('/load');
 
-		const chunk_size = 50000;
-		const chunk_count = 100;
-		const total_size = chunk_size * chunk_count;
-
-		let chunk = '';
-		for (let i = 0; i < chunk_size; i += 1) {
-			chunk += String(i % 10);
-		}
-
-		let times_responded = 0;
-
-		const { port, server } = await start_server(async (req, res) => {
-			if (req.url === '/large-response.json') {
-				times_responded += 1;
-
-				res.writeHead(200, {
-					'Access-Control-Allow-Origin': '*'
-				});
-
-				for (let i = 0; i < chunk_count; i += 1) {
-					if (!res.write(chunk)) {
-						await new Promise((fulfil) => {
-							res.once('drain', () => {
-								fulfil(undefined);
-							});
-						});
-					}
-				}
-
-				res.end();
-			}
-		});
-
-		await page.goto(`/load/large-response?port=${port}`);
-		expect(await page.textContent('h1')).toBe(`text.length is ${total_size}`);
-
-		expect(times_responded).toBe(1);
-
-		server.close();
+		await page.goto('/load/large-response');
+		expect(await page.textContent('h1')).toBe('text.length is 5000000');
 	});
 
 	test('handles external api', async ({ page }) => {

diff --git a/packages/kit/test/typings/endpoint.test.ts b/packages/kit/test/typings/endpoint.test.ts
@@ -51,14 +51,6 @@ export const readable_stream_body: RequestHandler = () => {
 	};
 };
 
-// valid - body instance of stream.Readable should be allowed
-export const stream_readable_body: RequestHandler = async () => {
-	const { Readable } = await import('stream');
-	return {
-		body: new Readable()
-	};
-};
-
 // valid - different header pairs should be allowed
 export const differential_headers_assignment: RequestHandler = () => {
 	if (Math.random() < 0.5) {
@@ -193,3 +185,11 @@ export const error_nested_instances: RequestHandler = () => {
 		body: { typed: new Uint8Array() }
 	};
 };
+
+// @ts-expect-error - instances cannot be nested
+export const stream_readable_body: RequestHandler = async () => {
+	const { Readable } = await import('stream');
+	return {
+		body: new Readable()
+	};
+};
diff --git a/packages/kit/types/index.d.ts b/packages/kit/types/index.d.ts
@@ -268,7 +268,7 @@ export interface ResolveOptions {
 	transformPage?: ({ html }: { html: string }) => MaybePromise<string>;
 }
 
-export type ResponseBody = JSONValue | Uint8Array | ReadableStream | import('stream').Readable;
+export type ResponseBody = JSONValue | Uint8Array | ReadableStream;
 
 export class Server {
 	constructor(manifest: SSRManifest);

diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml