Stream request body instead of buffering it in memory (#8084)

Co-authored-by: Matthew Phillips <matthew@matthewphillips.info>

packages/integrations/node/package.json

@@ -49,7 +49,7 @@
     "chai": "^4.3.7",
     "cheerio": "1.0.0-rc.12",
     "mocha": "^9.2.2",
-    "node-mocks-http": "^1.12.2",
+    "node-mocks-http": "^1.13.0",
     "undici": "^5.22.1"
   }
 }

packages/integrations/node/test/api-route.test.js

@@ -1,6 +1,7 @@
 import nodejs from '../dist/index.js';
 import { loadFixture, createRequestAndResponse } from './test-utils.js';
 import { expect } from 'chai';
+import crypto from 'node:crypto';
 
 describe('API routes', () => {
 	/** @type {import('./test-utils').Fixture} */
@@ -22,9 +23,11 @@ describe('API routes', () => {
 			url: '/recipes',
 		});
 
-		handler(req, res);
+		req.once('async_iterator', () => {
+			req.send(JSON.stringify({ id: 2 }));
+		});
 
-		req.send(JSON.stringify({ id: 2 }));
+		handler(req, res);
 
 		let [buffer] = await done;
 
@@ -43,11 +46,47 @@ describe('API routes', () => {
 			url: '/binary',
 		});
 
+		req.once('async_iterator', () => {
+			req.send(Buffer.from(new Uint8Array([1, 2, 3, 4, 5])));
+		});
+
 		handler(req, res);
-		req.send(Buffer.from(new Uint8Array([1, 2, 3, 4, 5])));
 
 		let [out] = await done;
 		let arr = Array.from(new Uint8Array(out.buffer));
 		expect(arr).to.deep.equal([5, 4, 3, 2, 1]);
 	});
+
+	it('Can post large binary data', async () => {
+		const { handler } = await import('./fixtures/api-route/dist/server/entry.mjs');
+
+		let { req, res, done } = createRequestAndResponse({
+			method: 'POST',
+			url: '/hash',
+		});
+
+		handler(req, res);
+
+		let expectedDigest = null;
+		req.once('async_iterator', () => {
+			// Send 256MB of garbage data in 256KB chunks. This should be fast (< 1sec).
+			let remainingBytes = 256 * 1024 * 1024;
+			const chunkSize = 256 * 1024;
+
+			const hash = crypto.createHash('sha256');
+			while (remainingBytes > 0) {
+				const size = Math.min(remainingBytes, chunkSize);
+				const chunk = Buffer.alloc(size, Math.floor(Math.random() * 256));
+				hash.update(chunk);
+				req.emit('data', chunk);
+				remainingBytes -= size;
+			}
+
+			req.emit('end');
+			expectedDigest = hash.digest();
+		});
+
+		let [out] = await done;
+		expect(new Uint8Array(out.buffer)).to.deep.equal(expectedDigest);
+	});
 });

packages/integrations/node/test/fixtures/api-route/src/pages/hash.ts

@@ -0,0 +1,16 @@
+import crypto from 'node:crypto';
+
+export async function post({ request }: { request: Request }) {
+	const hash = crypto.createHash('sha256');
+
+    const iterable = request.body as unknown as AsyncIterable<Uint8Array>;
+    for await (const chunk of iterable) {
+        hash.update(chunk);
+    }
+
+	return new Response(hash.digest(), {
+		headers: {
+			'Content-Type': 'application/octet-stream'
+		}
+	});
+}