diff --git a/src/lib/server/docker.ts b/src/lib/server/docker.ts index c44ad06..ee18644 100644 --- a/src/lib/server/docker.ts +++ b/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, diff --git a/src/lib/server/node-readable-stream.ts b/src/lib/server/node-readable-stream.ts new file mode 100644 index 0000000..6d3150a --- /dev/null +++ b/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 { + return Readable.toWeb(source) as ReadableStream; +} diff --git a/tests/node-readable-stream.test.ts b/tests/node-readable-stream.test.ts new file mode 100644 index 0000000..cfa3f6d --- /dev/null +++ b/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((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'); + }); +});