From 61cff83c01a370e4a5e6641c286f990f1a5c11e0 Mon Sep 17 00:00:00 2001 From: Junyi Yao Date: Thu, 15 Oct 2026 19:37:19 +0000 Subject: [PATCH] fix(code-interpreter-js): make readLines linear on long single-line outputs readLines appended every decoded chunk to one string buffer and re-scanned/re-sliced the whole buffer on each read, so a long line with no newlines (base64 images, wide DataFrames, large JSON) cost O(n^2) and blocked the event loop (~10s for a 40MB line). Decode and scan only the new chunk, accumulate line fragments in an array, and join once per emitted line. Streaming TextDecoder usage, trailing-partial-line flush at done, CR preservation, and reader.releaseLock() semantics are unchanged. Fixes https://github.com/e2b-dev/E2B/issues/1990 Signed-off-by: Junyi Yao --- .changeset/great-lines-repeat.md | 5 ++ packages/code-interpreter-js/src/utils.ts | 41 +++++++---- .../code-interpreter-js/tests/utils.test.ts | 73 ++++++++++++++++--- 3 files changed, 94 insertions(+), 25 deletions(-) create mode 100644 .changeset/great-lines-repeat.md diff --git a/.changeset/great-lines-repeat.md b/.changeset/great-lines-repeat.md new file mode 100644 index 0000000000..095a0089fc --- /dev/null +++ b/.changeset/great-lines-repeat.md @@ -0,0 +1,5 @@ +--- +"@e2b/code-interpreter": patch +--- + +Fix `readLines` quadratic line reassembly when streaming long single-line outputs (e.g. base64 images, wide DataFrames, large JSON): decode and scan only each new chunk, accumulating line fragments in an array that is joined once per emitted line, instead of appending to and re-scanning a single string buffer on every chunk. diff --git a/packages/code-interpreter-js/src/utils.ts b/packages/code-interpreter-js/src/utils.ts index 64a2e1f6bc..ca5ccc8bdf 100644 --- a/packages/code-interpreter-js/src/utils.ts +++ b/packages/code-interpreter-js/src/utils.ts @@ -66,33 +66,44 @@ export function isConnectionClosedError(error: unknown): boolean { export async function* readLines(stream: ReadableStream) { const reader = stream.getReader() const decoder = new TextDecoder() - let buffer = '' + // Fragments of the line currently being assembled. Kept as an array so a + // long line (e.g. a base64 image or a wide DataFrame with no newlines) is + // joined once per emitted line instead of being re-copied on every chunk, + // which kept this generator O(n^2) in line length. + const fragments: string[] = [] try { while (true) { const { done, value } = await reader.read() if (value !== undefined) { - buffer += decoder.decode(value, { stream: true }) + // Decode only the new chunk: TextDecoder in streaming mode already + // carries over incomplete multi-byte sequences across chunks. + const text = decoder.decode(value, { stream: true }) + let start = 0 + let newlineIdx = text.indexOf('\n', start) + while (newlineIdx !== -1) { + fragments.push(text.slice(start, newlineIdx)) + yield fragments.join('') + fragments.length = 0 + start = newlineIdx + 1 + newlineIdx = text.indexOf('\n', start) + } + if (start < text.length) { + fragments.push(text.slice(start)) + } } if (done) { - buffer += decoder.decode() - if (buffer.length > 0) { - yield buffer + const trailing = decoder.decode() + if (trailing.length > 0) { + fragments.push(trailing) + } + if (fragments.length > 0) { + yield fragments.join('') } break } - - let newlineIdx = -1 - - do { - newlineIdx = buffer.indexOf('\n') - if (newlineIdx !== -1) { - yield buffer.slice(0, newlineIdx) - buffer = buffer.slice(newlineIdx + 1) - } - } while (newlineIdx !== -1) } } finally { reader.releaseLock() diff --git a/packages/code-interpreter-js/tests/utils.test.ts b/packages/code-interpreter-js/tests/utils.test.ts index 6efc7f03ad..320dbdb6ec 100644 --- a/packages/code-interpreter-js/tests/utils.test.ts +++ b/packages/code-interpreter-js/tests/utils.test.ts @@ -2,6 +2,27 @@ import { expect, test } from 'vitest' import { readLines } from '../src/utils' +function streamFromChunks(chunks: (string | Uint8Array)[]) { + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue( + typeof chunk === 'string' ? new TextEncoder().encode(chunk) : chunk + ) + } + controller.close() + }, + }) +} + +async function collectLines(stream: ReadableStream) { + const lines: string[] = [] + for await (const line of readLines(stream)) { + lines.push(line) + } + return lines +} + test.each([1, 2, 3, 4, 7])( 'preserves UTF-8 output split into %s-byte chunks', async (chunkSize) => { @@ -48,17 +69,49 @@ test.each(['', 'first\n\nlast\n', 'first\nlast', '🌍\n你好\n'])( ) test('flushes an incomplete UTF-8 sequence when the stream ends', async () => { - const stream = new ReadableStream({ - start(controller) { - controller.enqueue(new Uint8Array([0xe4, 0xbd])) - controller.close() - }, - }) - const lines: string[] = [] + const lines = await collectLines( + streamFromChunks([new Uint8Array([0xe4, 0xbd])]) + ) - for await (const line of readLines(stream)) { - lines.push(line) + expect(lines).toEqual(['\uFFFD']) +}) + +test('emits a long single line assembled from many small chunks intact', async () => { + // Regression test for https://github.com/e2b-dev/E2B/issues/1990: + // outputs like base64 images or wide DataFrames are one long line. + const line = 'x'.repeat(1024 * 1024) + const chunkSize = 1024 + const chunks: string[] = [] + for (let offset = 0; offset < line.length; offset += chunkSize) { + chunks.push(line.slice(offset, offset + chunkSize)) } + const lines = await collectLines(streamFromChunks(chunks)) - expect(lines).toEqual(['\uFFFD']) + expect(lines).toEqual([line]) +}) + +test('keeps \\r when splitting on \\n', async () => { + const lines = await collectLines(streamFromChunks(['a\r\nb\r\nc\r'])) + + expect(lines).toEqual(['a\r', 'b\r', 'c\r']) +}) + +test('splits a newline that straddles a chunk boundary', async () => { + const lines = await collectLines(streamFromChunks(['a\r', '\nb'])) + + expect(lines).toEqual(['a\r', 'b']) +}) + +test('yields nothing for an empty stream', async () => { + const lines = await collectLines(streamFromChunks([])) + + expect(lines).toEqual([]) +}) + +test('ignores leading empty chunks', async () => { + const lines = await collectLines( + streamFromChunks([new Uint8Array(0), 'a\n', new Uint8Array(0), 'b']) + ) + + expect(lines).toEqual(['a', 'b']) })