fix docker stream backpressure (#1445)

src/lib/server/docker.ts

@@ -14,6 +14,7 @@ import { Readable } from 'node:stream';
 import * as tls from 'node:tls';
 import { createHash } from 'node:crypto';
 import { pumpWebStreamToWritable } from './stream-pump';
+import { toWebReadableStream } from './node-readable-stream';
 import { computeRequestTimeoutMs } from './backups/request-timeout';
 import { helperWaitDeadline } from './helper-wait-core';
 import type { Environment } from './db';
@@ -467,14 +468,7 @@ export function httpsAgentRequest(
 			}
 
 			if (streaming) {
-				const readable = new ReadableStream({
-					start(controller) {
-						res.on('data', (chunk: Buffer) => controller.enqueue(new Uint8Array(chunk)));
-						res.on('end', () => controller.close());
-						res.on('error', (err) => controller.error(err));
-					},
-					cancel() { res.destroy(); }
-				});
+				const readable = toWebReadableStream(res);
 				resolve(new Response(readable, { status, statusText, headers }));
 			} else {
 				const chunks: Buffer[] = [];
@@ -792,22 +786,7 @@ export function unixSocketStreamRequest(
 				}
 			}
 
-			const readable = new ReadableStream({
-				start(controller) {
-					res.on('data', (chunk: Buffer) => {
-						controller.enqueue(new Uint8Array(chunk));
-					});
-					res.on('end', () => {
-						controller.close();
-					});
-					res.on('error', (err) => {
-						controller.error(err);
-					});
-				},
-				cancel() {
-					res.destroy();
-				}
-			});
+			const readable = toWebReadableStream(res);
 
 			resolve(new Response(readable, {
 				status: res.statusCode || 200,

src/lib/server/node-readable-stream.ts

@@ -0,0 +1,9 @@
+import { Readable } from 'node:stream';
+
+/**
+ * Convert a Node readable to a Web stream while preserving Node backpressure.
+ * Cancellation and source errors are forwarded by Node's native adapter.
+ */
+export function toWebReadableStream(source: Readable): ReadableStream<Uint8Array> {
+	return Readable.toWeb(source) as ReadableStream<Uint8Array>;
+}

tests/node-readable-stream.test.ts

@@ -0,0 +1,75 @@
+import { describe, expect, test } from 'bun:test';
+import { PassThrough, Readable } from 'node:stream';
+import { toWebReadableStream } from '../src/lib/server/node-readable-stream';
+
+const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+describe('toWebReadableStream', () => {
+	test('keeps a fast producer bounded behind a slow consumer', async () => {
+		const totalChunks = 128;
+		const chunkSize = 64 * 1024;
+		let produced = 0;
+
+		const source = new Readable({
+			highWaterMark: 16 * 1024,
+			read() {
+				if (produced >= totalChunks) {
+					this.push(null);
+					return;
+				}
+
+				produced++;
+				setImmediate(() => {
+					if (!this.destroyed) this.push(Buffer.alloc(chunkSize));
+				});
+			}
+		});
+
+		const reader = toWebReadableStream(source).getReader();
+		const first = await reader.read();
+		expect(first.done).toBe(false);
+		expect(first.value?.byteLength).toBe(chunkSize);
+
+		await delay(50);
+		expect(produced).toBeLessThanOrEqual(4);
+
+		let receivedBytes = first.value?.byteLength ?? 0;
+		while (true) {
+			const { done, value } = await reader.read();
+			if (done) break;
+			receivedBytes += value.byteLength;
+		}
+
+		expect(produced).toBe(totalChunks);
+		expect(receivedBytes).toBe(totalChunks * chunkSize);
+	});
+
+	test('destroys the Node stream when the Web reader is cancelled', async () => {
+		const source = new PassThrough();
+		const reader = toWebReadableStream(source).getReader();
+		source.write(Buffer.from('chunk'));
+		await reader.read();
+
+		let cancelError: Error | undefined;
+		source.once('error', (error) => {
+			cancelError = error;
+		});
+		const closed = new Promise<void>((resolve) => source.once('close', resolve));
+
+		await reader.cancel(new Error('cancelled'));
+		await closed;
+
+		expect(source.destroyed).toBe(true);
+		expect(cancelError?.message).toBe('cancelled');
+	});
+
+	test('forwards Node stream errors to the Web reader', async () => {
+		const source = new PassThrough();
+		const reader = toWebReadableStream(source).getReader();
+		const pendingRead = reader.read();
+
+		source.destroy(new Error('source failed'));
+
+		await expect(pendingRead).rejects.toThrow('source failed');
+	});
+});