fix(streams): remove quadratic buffering in TextLineStream

streams/text_line_stream.ts

@@ -65,45 +65,81 @@ export interface TextLineStreamOptions {
  * ```
  */
 export class TextLineStream extends TransformStream<string, string> {
-  #currentLine = "";
+  #pending: string[] = [];
+  #pendingTrailsCR = false;
 
   /**
    * Constructs a new instance.
    *
    * @param options Options for the stream.
    */
   constructor(options: TextLineStreamOptions = { allowCR: false }) {
+    const allowCR = options.allowCR ?? false;
     super({
       transform: (chars, controller) => {
-        chars = this.#currentLine + chars;
+        if (chars.length === 0) return;
 
-        while (true) {
-          const lfIndex = chars.indexOf("\n");
-          const crIndex = options.allowCR ? chars.indexOf("\r") : -1;
+        // Fast path: if no line can complete within this chunk, buffer it
+        // without rescanning the accumulated text. This keeps the total cost
+        // linear when a single line spans many chunks. (With `allowCR`, a
+        // buffered chunk-final "\r" resolves as soon as the next chunk
+        // arrives, so it forces processing.)
+        if (
+          !chars.includes("\n") &&
+          (!allowCR || (!this.#pendingTrailsCR && !chars.includes("\r")))
+        ) {
+          this.#pending.push(chars);
+          return;
+        }
+
+        this.#pending.push(chars);
+        const text = this.#pending.length === 1
+          ? this.#pending[0]!
+          : this.#pending.join("");
+        this.#pending.length = 0;
+
+        let start = 0;
+        let lfIndex = text.indexOf("\n");
+        let crIndex = allowCR ? text.indexOf("\r") : -1;
 
+        while (true) {
           if (
-            crIndex !== -1 && crIndex !== (chars.length - 1) &&
+            crIndex !== -1 && crIndex !== (text.length - 1) &&
             (lfIndex === -1 || (lfIndex - 1) > crIndex)
           ) {
-            controller.enqueue(chars.slice(0, crIndex));
-            chars = chars.slice(crIndex + 1);
+            controller.enqueue(text.slice(start, crIndex));
+            start = crIndex + 1;
+            crIndex = text.indexOf("\r", start);
             continue;
           }
 
           if (lfIndex === -1) break;
 
-          const endIndex = chars[lfIndex - 1] === "\r" ? lfIndex - 1 : lfIndex;
-          controller.enqueue(chars.slice(0, endIndex));
-          chars = chars.slice(lfIndex + 1);
+          const endIndex = text[lfIndex - 1] === "\r" ? lfIndex - 1 : lfIndex;
+          controller.enqueue(text.slice(start, endIndex));
+          start = lfIndex + 1;
+          lfIndex = text.indexOf("\n", start);
+          if (crIndex !== -1 && crIndex < start) {
+            crIndex = text.indexOf("\r", start);
+          }
         }
 
-        this.#currentLine = chars;
+        if (start < text.length) {
+          const leftover = start === 0 ? text : text.slice(start);
+          this.#pending.push(leftover);
+          this.#pendingTrailsCR = allowCR && leftover.endsWith("\r");
+        } else {
+          this.#pendingTrailsCR = false;
+        }
       },
       flush: (controller) => {
-        if (this.#currentLine === "") return;
-        const currentLine = options.allowCR && this.#currentLine.endsWith("\r")
-          ? this.#currentLine.slice(0, -1)
-          : this.#currentLine;
+        if (this.#pending.length === 0) return;
+        let currentLine = this.#pending.length === 1
+          ? this.#pending[0]!
+          : this.#pending.join("");
+        if (allowCR && currentLine.endsWith("\r")) {
+          currentLine = currentLine.slice(0, -1);
+        }
         controller.enqueue(currentLine);
       },
     });

streams/text_line_stream_test.ts

@@ -104,6 +104,35 @@ Deno.test("TextLineStream parses no final empty chunk with terminal newline", as
   ]);
 });
 
+Deno.test("TextLineStream parses a line spanning many chunks", async () => {
+  const longLine = "a".repeat(4096);
+  const chunks: string[] = [];
+  for (let i = 0; i < longLine.length; i += 7) {
+    chunks.push(longLine.slice(i, i + 7));
+  }
+  chunks.push("\nrest");
+
+  for (const allowCR of [false, true]) {
+    const stream = ReadableStream.from(chunks.slice())
+      .pipeThrough(new TextLineStream({ allowCR }));
+    assertEquals(await Array.fromAsync(stream), [longLine, "rest"]);
+  }
+});
+
+Deno.test("TextLineStream parses input with empty chunks", async () => {
+  const stream = ReadableStream.from(["", "abc\nd", "", "ef", ""])
+    .pipeThrough(new TextLineStream());
+
+  assertEquals(await Array.fromAsync(stream), ["abc", "def"]);
+});
+
+Deno.test("TextLineStream with `allowCR` parses a chunk-final `\\r` once the next chunk arrives", async () => {
+  const stream = ReadableStream.from(["abc\r", "def"])
+    .pipeThrough(new TextLineStream({ allowCR: true }));
+
+  assertEquals(await Array.fromAsync(stream), ["abc", "def"]);
+});
+
 Deno.test("TextLineStream parses no final empty chunk without terminal newline", async () => {
   const stream = ReadableStream.from([
     "abc\n",