Preserve NDJSON records across decode errors

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: ea5daa0b-7839-42f0-9d63-4ca84c7d2b5c

Author: TheLarkInn

common/reviews/api/rush-reporter.api.md

@@ -261,9 +261,17 @@ export class NdjsonDecoder {
     flush(): unknown[];
 }
 
+// @beta
+export class NdjsonInvalidRecordError extends Error {
+    constructor(decodedRecords: readonly unknown[], cause: Error);
+    readonly cause: Error;
+    readonly decodedRecords: readonly unknown[];
+}
+
 // @beta
 export class NdjsonRecordTooLargeError extends Error {
-    constructor(maxRecordBytes: number);
+    constructor(maxRecordBytes: number, decodedRecords?: readonly unknown[]);
+    readonly decodedRecords: readonly unknown[];
     readonly maxRecordBytes: number;
 }
 

libraries/reporter/src/index.ts

@@ -77,7 +77,12 @@ export {
   isReporterProtocolCompatible
 } from './protocol/ReporterProtocol';
 export type { INdjsonOptions } from './protocol/Ndjson';
-export { NdjsonRecordTooLargeError, encodeNdjsonRecord, NdjsonDecoder } from './protocol/Ndjson';
+export {
+  NdjsonInvalidRecordError,
+  NdjsonRecordTooLargeError,
+  encodeNdjsonRecord,
+  NdjsonDecoder
+} from './protocol/Ndjson';
 export type {
   IReporterHello,
   IReporterHelloAck,

libraries/reporter/src/protocol/Ndjson.ts

@@ -14,17 +14,49 @@ export class NdjsonRecordTooLargeError extends Error {
    */
   public readonly maxRecordBytes: number;
 
-  public constructor(maxRecordBytes: number) {
+  /**
+   * Valid records completed by the decoder before it rejected the oversized record.
+   */
+  public readonly decodedRecords: readonly unknown[];
+
+  public constructor(maxRecordBytes: number, decodedRecords: readonly unknown[] = []) {
     super(`The NDJSON record exceeds the maximum size of ${maxRecordBytes} bytes.`);
     this.name = 'NdjsonRecordTooLargeError';
     this.maxRecordBytes = maxRecordBytes;
+    this.decodedRecords = [...decodedRecords];
 
     // Restore the prototype chain, which is broken when subclassing a built-in
     // and compiling to CommonJS.
     Object.setPrototypeOf(this, NdjsonRecordTooLargeError.prototype);
   }
 }
 
+/**
+ * Thrown when a completed NDJSON record is not valid JSON.
+ *
+ * @beta
+ */
+export class NdjsonInvalidRecordError extends Error {
+  /**
+   * Valid records completed before the malformed record.
+   */
+  public readonly decodedRecords: readonly unknown[];
+
+  /**
+   * The JSON parser error that caused this failure.
+   */
+  public readonly cause: Error;
+
+  public constructor(decodedRecords: readonly unknown[], cause: Error) {
+    super('The NDJSON record is not valid JSON.');
+    this.name = 'NdjsonInvalidRecordError';
+    this.decodedRecords = [...decodedRecords];
+    this.cause = cause;
+
+    Object.setPrototypeOf(this, NdjsonInvalidRecordError.prototype);
+  }
+}
+
 /**
  * Options controlling NDJSON record size enforcement.
  *
@@ -67,6 +99,9 @@ export function encodeNdjsonRecord(value: unknown, options?: INdjsonOptions): st
  * Call {@link NdjsonDecoder.decode} for each received chunk to obtain the
  * records completed by that chunk, then call {@link NdjsonDecoder.flush} once
  * the stream ends to obtain any trailing record that was not newline-terminated.
+ * If a later record fails, its error exposes earlier valid records through
+ * `decodedRecords`; the rejected completed line is consumed and any subsequent
+ * buffered lines remain available to a later call.
  *
  * @beta
  */
@@ -83,7 +118,8 @@ export class NdjsonDecoder {
    * Appends a chunk and returns any records it completed.
    *
    * @param chunk - a fragment of the NDJSON stream
-   * @throws NdjsonRecordTooLargeError if a record exceeds the limit
+   * @throws {@link NdjsonRecordTooLargeError} if a record exceeds the limit
+   * @throws {@link NdjsonInvalidRecordError} if a completed record is malformed
    */
   public decode(chunk: string): unknown[] {
     this._buffer += chunk;
@@ -99,7 +135,7 @@ export class NdjsonDecoder {
 
     // A partial line that already exceeds the limit can never become a valid record.
     if (Buffer.byteLength(this._buffer, 'utf8') > this._maxRecordBytes) {
-      throw new NdjsonRecordTooLargeError(this._maxRecordBytes);
+      throw new NdjsonRecordTooLargeError(this._maxRecordBytes, records);
     }
 
     return records;
@@ -108,7 +144,8 @@ export class NdjsonDecoder {
   /**
    * Returns any trailing record that was not newline-terminated and resets the buffer.
    *
-   * @throws NdjsonRecordTooLargeError if the trailing record exceeds the limit
+   * @throws {@link NdjsonRecordTooLargeError} if the trailing record exceeds the limit
+   * @throws {@link NdjsonInvalidRecordError} if the trailing record is malformed
    */
   public flush(): unknown[] {
     const records: unknown[] = [];
@@ -122,12 +159,18 @@ export class NdjsonDecoder {
 
   private _processLine(line: string, records: unknown[]): void {
     if (Buffer.byteLength(line, 'utf8') > this._maxRecordBytes) {
-      throw new NdjsonRecordTooLargeError(this._maxRecordBytes);
+      throw new NdjsonRecordTooLargeError(this._maxRecordBytes, records);
     }
     const trimmed: string = line.trim();
     if (trimmed.length === 0) {
       return;
     }
-    records.push(JSON.parse(trimmed));
+    try {
+      records.push(JSON.parse(trimmed));
+    } catch (error) {
+      const cause: Error =
+        error instanceof Error ? error : new Error('An unknown JSON parsing failure occurred.');
+      throw new NdjsonInvalidRecordError(records, cause);
+    }
   }
 }

libraries/reporter/src/test/Protocol.test.ts

@@ -7,6 +7,7 @@ import {
   isReporterProtocolCompatible,
   encodeNdjsonRecord,
   NdjsonDecoder,
+  NdjsonInvalidRecordError,
   NdjsonRecordTooLargeError,
   InvalidReporterHelloError,
   negotiateReporterHello,
@@ -71,6 +72,39 @@ describe('NDJSON encode/decode', () => {
     const decoder: NdjsonDecoder = new NdjsonDecoder({ maxRecordBytes: 10 });
     expect(() => decoder.decode('x'.repeat(50))).toThrow(NdjsonRecordTooLargeError);
   });
+
+  it('exposes valid records completed before a malformed later record', () => {
+    const decoder: NdjsonDecoder = new NdjsonDecoder();
+    let caught: unknown;
+    try {
+      decoder.decode('{"id":1}\nnot-json\n{"id":2}\n');
+    } catch (error) {
+      caught = error;
+    }
+
+    expect(caught).toBeInstanceOf(NdjsonInvalidRecordError);
+    if (!(caught instanceof NdjsonInvalidRecordError)) {
+      throw new Error('Expected an NdjsonInvalidRecordError.');
+    }
+    expect(caught.decodedRecords).toEqual([{ id: 1 }]);
+    expect(decoder.decode('')).toEqual([{ id: 2 }]);
+  });
+
+  it('exposes valid records completed before an oversized later record', () => {
+    const decoder: NdjsonDecoder = new NdjsonDecoder({ maxRecordBytes: 10 });
+    let caught: unknown;
+    try {
+      decoder.decode(`1\n${'"'}${'x'.repeat(20)}${'"'}\n`);
+    } catch (error) {
+      caught = error;
+    }
+
+    expect(caught).toBeInstanceOf(NdjsonRecordTooLargeError);
+    if (!(caught instanceof NdjsonRecordTooLargeError)) {
+      throw new Error('Expected an NdjsonRecordTooLargeError.');
+    }
+    expect(caught.decodedRecords).toEqual([1]);
+  });
 });
 
 describe('negotiateReporterHello', () => {