perf(providers): assemble fragmented responses once per line (#911)

src/ghidra/GhidraResponseBuffer.test.ts

@@ -3,6 +3,55 @@ import { describe, expect, it } from "vitest";
 import { GhidraResponseBuffer } from "./GhidraResponseBuffer.js";
 
 describe("Ghidra response buffer", () => {
+  it("preserves fragmented lines, whitespace and a trailing partial response", () => {
+    const lines: string[] = [];
+    const buffer = new GhidraResponseBuffer({
+      onLine: (line) => lines.push(line),
+    });
+    buffer.push('  {"result":"雪');
+    buffer.push('🦊"}\r');
+    buffer.push('\n\n {"result":2}\n{"res');
+    expect(lines).toEqual(['{"result":"雪🦊"}', '{"result":2}']);
+    buffer.push('ult":3}');
+    expect(lines).toHaveLength(2);
+    buffer.push("\n");
+    expect(lines).toEqual([
+      '{"result":"雪🦊"}',
+      '{"result":2}',
+      '{"result":3}',
+    ]);
+  });
+
+  it("drops both pending fragments and remaining lines when reset during delivery", () => {
+    const lines: string[] = [];
+    const buffer = new GhidraResponseBuffer({
+      onLine: (line) => {
+        lines.push(line);
+        buffer.reset();
+      },
+    });
+    buffer.push("discarded partial");
+    buffer.reset();
+    buffer.push("first\nsecond\npartial");
+    buffer.push("discarded");
+    buffer.reset();
+    buffer.push("next\n");
+    expect(lines).toEqual(["first", "next"]);
+  });
+
+  it("accepts a large response arriving in many small socket fragments", () => {
+    const lines: string[] = [];
+    const buffer = new GhidraResponseBuffer({
+      onLine: (line) => lines.push(line),
+    });
+    const response = `{"result":"${"x".repeat(8 * 1024 * 1024)}"}`;
+    for (let offset = 0; offset < response.length; offset += 4096)
+      buffer.push(response.slice(offset, offset + 4096));
+    expect(lines).toEqual([]);
+    buffer.push("\n");
+    expect(lines).toEqual([response]);
+  });
+
   it("accepts complete responses larger than the former 64 MiB ceiling", () => {
     const lines: string[] = [];
     const buffer = new GhidraResponseBuffer({

src/ghidra/GhidraResponseBuffer.ts

@@ -7,6 +7,7 @@ export interface GhidraResponseBufferOptions {
 export class GhidraResponseBuffer {
   readonly #options: GhidraResponseBufferOptions;
   #buffer = "";
+  #fragments: string[] = [];
 
   constructor(options: GhidraResponseBufferOptions) {
     this.#options = options;
@@ -14,7 +15,10 @@ export class GhidraResponseBuffer {
 
   /** Consume one decoded socket chunk. */
   push(chunk: string): void {
-    this.#buffer += chunk;
+    this.#fragments.push(chunk);
+    if (!chunk.includes("\n")) return;
+    this.#buffer += this.#fragments.join("");
+    this.#fragments = [];
     let newline = this.#buffer.indexOf("\n");
     while (newline >= 0) {
       const encodedLine = this.#buffer.slice(0, newline);
@@ -28,5 +32,6 @@ export class GhidraResponseBuffer {
   /** Drop any incomplete line when a session closes. */
   reset(): void {
     this.#buffer = "";
+    this.#fragments = [];
   }
 }

src/hopper/HopperResponseStream.test.ts

@@ -3,6 +3,80 @@ import { describe, expect, it } from "vitest";
 import { HopperResponseStream } from "./HopperResponseStream.js";
 
 describe("Hopper response stream", () => {
+  it("preserves fragmented messages and drops partial bytes on reset", () => {
+    const messages: unknown[] = [];
+    const stream = new HopperResponseStream({
+      accept: (message) => {
+        messages.push(message);
+        return true;
+      },
+      hasQueued: () => false,
+      nextRequestId: () => 4,
+      abort: (message) => {
+        throw new Error(message);
+      },
+    });
+    stream.push("discarded");
+    stream.reset();
+    stream.push(' {"id":1,"result":"雪');
+    stream.push('🦊"}\r');
+    stream.push('\n\n{"id":2,"result":null}\n{"id":');
+    stream.push('3,"result":true}');
+    expect(messages).toEqual([
+      { id: 1, result: "雪🦊" },
+      { id: 2, result: null },
+    ]);
+    stream.push("\n");
+    expect(messages).toEqual([
+      { id: 1, result: "雪🦊" },
+      { id: 2, result: null },
+      { id: 3, result: true },
+    ]);
+  });
+
+  it("stops delivering queued lines when validation aborts and resets the stream", () => {
+    const messages: unknown[] = [];
+    const failures: string[] = [];
+    const stream = new HopperResponseStream({
+      accept: (message) => {
+        messages.push(message);
+        return true;
+      },
+      hasQueued: () => false,
+      nextRequestId: () => 2,
+      abort: (message) => {
+        failures.push(message);
+        stream.reset();
+      },
+    });
+    stream.push('malformed\n{"id":1,"result":true}\n');
+    expect(messages).toEqual([]);
+    expect(failures).toHaveLength(1);
+    stream.push('{"id":1,"result":false}\n');
+    expect(messages).toEqual([{ id: 1, result: false }]);
+  });
+
+  it("accepts a large response arriving in many small socket fragments", () => {
+    const messages: unknown[] = [];
+    const stream = new HopperResponseStream({
+      accept: (message) => {
+        messages.push(message);
+        return true;
+      },
+      hasQueued: () => false,
+      nextRequestId: () => 2,
+      abort: (message) => {
+        throw new Error(message);
+      },
+    });
+    const response = `{"id":1,"result":"${"x".repeat(8 * 1024 * 1024)}"}`;
+    for (let offset = 0; offset < response.length; offset += 4096)
+      stream.push(response.slice(offset, offset + 4096));
+    expect(messages).toEqual([]);
+    stream.push("\n");
+    expect(messages).toEqual([JSON.parse(response)]);
+  });
+
   it("accepts complete responses larger than the former 10 MiB ceiling", () => {
     const messages: unknown[] = [];
     const failures: string[] = [];

src/hopper/HopperResponseStream.ts

@@ -13,11 +13,15 @@ export interface HopperResponseStreamOptions {
 /** Incrementally validates Hopper NDJSON response frames. */
 export class HopperResponseStream {
   #buffer = "";
+  #fragments: string[] = [];
 
   constructor(readonly options: HopperResponseStreamOptions) {}
 
   push(chunk: string): void {
-    this.#buffer += chunk;
+    this.#fragments.push(chunk);
+    if (!chunk.includes("\n")) return;
+    this.#buffer += this.#fragments.join("");
+    this.#fragments = [];
     let newline = this.#buffer.indexOf("\n");
     while (newline >= 0) {
       const line = this.#buffer.slice(0, newline).trim();
@@ -29,6 +33,7 @@ export class HopperResponseStream {
 
   reset(): void {
     this.#buffer = "";
+    this.#fragments = [];
   }
 
   #acceptLine(line: string): boolean {