Fix streaming: make mid-stream exception detection sound (no false positives, no event-loop hang)

packages/client-common/__tests__/unit/stream_utils.test.ts

@@ -1,5 +1,8 @@
 import { describe, it, expect } from "vitest";
-import { extractErrorAtTheEndOfChunk } from "../../src/index";
+import {
+  endsWithExceptionMarker,
+  extractErrorAtTheEndOfChunk,
+} from "../../src/index";
 
 describe("utils/stream", () => {
   const errMsg = "boom";
@@ -33,6 +36,124 @@ describe("utils/stream", () => {
     expect(err).toBeInstanceOf(Error);
     expect(err?.message).toContain("error in the stream");
   });
+
+  // Regression: a malformed trailer whose only newline sits *above* the
+  // error-length hint (e.g. a single long CRLF-terminated row, or a trailer
+  // truncated by a proxy) used to run the backward scan index below zero and
+  // spin forever, blocking the Node.js event loop (nothing throws, so the
+  // surrounding try/catch could not rescue it). It must now return a plain
+  // Error instead of hanging.
+  it("returns an error instead of hanging when there is no length delimiter", () => {
+    const chunk = new TextEncoder().encode("x".repeat(100) + "\r\n");
+
+    const err = extractErrorAtTheEndOfChunk(chunk, tag);
+    expect(err).toBeInstanceOf(Error);
+    expect(err.message).toContain("error in the stream");
+  }, 5000);
+});
+
+describe("utils/stream endsWithExceptionMarker", () => {
+  const tag = "abcdefghijklmnop";
+  const enc = (s: string) => new TextEncoder().encode(s);
+
+  // A binary payload (e.g. Parquet) that merely happens to end in a \r\n pair.
+  const binaryEndingInCRLF = () => {
+    const bytes = new Uint8Array(64);
+    for (let i = 0; i < bytes.length; i++) {
+      bytes[i] = (i * 7) % 251;
+    }
+    bytes[62] = 0x0d;
+    bytes[63] = 0x0a;
+    return bytes;
+  };
+
+  const cases: Array<{
+    name: string;
+    chunk: Uint8Array;
+    checkTag: string;
+    expected: boolean;
+  }> = [
+    {
+      name: "the exact end-of-stream trailer for the tag",
+      chunk: enc(`${tag}\r\n__exception__\r\n`),
+      checkTag: tag,
+      expected: true,
+    },
+    {
+      name: "a full exception trailer preceded by a body",
+      chunk: buildValidErrorChunk("boom", tag),
+      checkTag: tag,
+      expected: true,
+    },
+    {
+      name: "a successful CRLF-terminated CSV/TSV body",
+      chunk: enc("0\r\n1\r\n2\r\n"),
+      checkTag: tag,
+      expected: false,
+    },
+    {
+      name: "a binary body that merely ends in a \\r\\n pair",
+      chunk: binaryEndingInCRLF(),
+      checkTag: tag,
+      expected: false,
+    },
+    {
+      name: "a chunk shorter than the marker",
+      chunk: enc("\r\n"),
+      checkTag: tag,
+      expected: false,
+    },
+    {
+      name: "the __exception__ marker present but a different tag value",
+      chunk: enc(`ponmlkjihgfedcba\r\n__exception__\r\n`),
+      checkTag: tag,
+      expected: false,
+    },
+    {
+      // Full-length suffix, correct tag, but the two bytes after the tag are
+      // not the `\r\n` separator: a near-miss trailer must be rejected.
+      name: "the tag matches but the bytes after it are not a CRLF separator",
+      chunk: enc(`${tag}XX__exception__\r\n`),
+      checkTag: tag,
+      expected: false,
+    },
+    {
+      // Correct tag and `\r\n`, but the fixed marker bytes are wrong: without a
+      // literal `__exception__` this is not a trailer.
+      name: "the tag and CRLF match but the __exception__ marker bytes differ",
+      chunk: enc(`${tag}\r\n${"x".repeat("__exception__".length)}\r\n`),
+      checkTag: tag,
+      expected: false,
+    },
+    {
+      // Everything matches except the terminating newline — the closest
+      // possible near-miss to a real trailer must still be rejected.
+      name: "a full-length trailer whose terminating newline byte is corrupted",
+      chunk: enc(`${tag}\r\n__exception__\rX`),
+      checkTag: tag,
+      expected: false,
+    },
+    {
+      name: "a trailer exactly one byte too short (missing final newline)",
+      chunk: enc(`${tag}\r\n__exception__\r`),
+      checkTag: tag,
+      expected: false,
+    },
+    {
+      // Guards the implicit non-empty-tag precondition: the tag is the only
+      // discriminator, so an empty tag must never match the bare marker.
+      name: "an empty tag against a body ending in the bare marker",
+      chunk: enc(`garbage\r\n__exception__\r\n`),
+      checkTag: "",
+      expected: false,
+    },
+  ];
+
+  for (const { name, chunk, checkTag, expected } of cases) {
+    it(`returns ${expected} for ${name}`, () => {
+      expect(endsWithExceptionMarker(chunk, checkTag)).toBe(expected);
+    });
+  }
 });
 
 /**

packages/client-common/src/index.ts

@@ -172,6 +172,7 @@ export {
   isCredentialsAuth,
   isJWTAuth,
   extractErrorAtTheEndOfChunk,
+  endsWithExceptionMarker,
   CARET_RETURN,
 } from "./utils";
 export { LogWriter, DefaultLogger, type LogWriterParams } from "./logger";

packages/client-common/src/utils/stream.ts

@@ -37,9 +37,19 @@ export function extractErrorAtTheEndOfChunk(
       );
     }
 
+    // Scan backwards for the newline that delimits the error-length hint.
+    // The scan is floored at the start of the chunk: a malformed trailer (e.g.
+    // truncated by a proxy, with no newline below the hint) must not send the
+    // index negative and spin forever, which would block the event loop since
+    // `chunk[-1]` is `undefined` (never a newline) and nothing throws.
     do {
       --errMsgLenStartIdx;
-    } while (chunk[errMsgLenStartIdx] !== NEWLINE);
+    } while (errMsgLenStartIdx >= 0 && chunk[errMsgLenStartIdx] !== NEWLINE);
+    if (errMsgLenStartIdx < 0) {
+      return new Error(
+        "there was an error in the stream, but the last chunk is malformed",
+      );
+    }
 
     const textDecoder = new TextDecoder("utf-8");
 
@@ -68,3 +78,51 @@ export function extractErrorAtTheEndOfChunk(
     return err as Error;
   }
 }
+
+/**
+ * Sound discriminator for the mid-stream exception trailer.
+ *
+ * When an error occurs after ClickHouse (25.11+) has already started streaming
+ * a 200 response, it terminates the body with the exact byte sequence
+ * `<exceptionTag>\r\n__exception__\r\n`, where `exceptionTag` is the random
+ * per-response token echoed by the `x-clickhouse-exception-tag` response header.
+ * Returns `true` only when `chunk` ends with that sequence.
+ *
+ * Requiring both the fixed `__exception__` marker *and* the random per-response
+ * tag makes detection reliable, so a bare `\r\n` occurring inside a *successful*
+ * response body — binary formats such as Parquet, or CRLF-terminated CSV/TSV
+ * rows (`output_format_*_crlf_end_of_line`) — is not mistaken for an exception.
+ */
+export function endsWithExceptionMarker(
+  chunk: Uint8Array,
+  exceptionTag: string,
+): boolean {
+  // The random per-response tag is the discriminator; without it the check
+  // cannot be sound. Refuse to treat the chunk as an exception rather than
+  // fall back to a marker-only match (better to miss a degenerate tag-less
+  // trailer than to abort a successful stream on a stray `__exception__`).
+  if (exceptionTag.length === 0) {
+    return false;
+  }
+  // Suffix layout, from `chunk.length` backwards:
+  //   <exceptionTag> \r \n __exception__ \r \n
+  const suffixLength = exceptionTag.length + 2 + EXCEPTION_MARKER.length + 2;
+  if (chunk.length < suffixLength) {
+    return false;
+  }
+  let pos = chunk.length - suffixLength;
+  for (let i = 0; i < exceptionTag.length; i++) {
+    if (chunk[pos++] !== exceptionTag.charCodeAt(i)) {
+      return false;
+    }
+  }
+  if (chunk[pos++] !== CARET_RETURN || chunk[pos++] !== NEWLINE) {
+    return false;
+  }
+  for (let i = 0; i < EXCEPTION_MARKER.length; i++) {
+    if (chunk[pos++] !== EXCEPTION_MARKER.charCodeAt(i)) {
+      return false;
+    }
+  }
+  return chunk[pos++] === CARET_RETURN && chunk[pos] === NEWLINE;
+}

packages/client-node/CHANGELOG.md

@@ -15,8 +15,10 @@
 ## Bug fixes
 
 - Fixed `Array(Date)` / `Array(Date32)` query-parameter binding (and other temporal element types nested in arrays, tuples, and maps). A JS `Date` inside a container was serialized as a bare Unix timestamp (e.g. `[1683244800]`), which the server's `Array(Date)` element parser rejects (`CANNOT_PARSE_INPUT_ASSERTION_FAILED`). Container-nested `Date` values are now emitted as a quoted UTC date string (e.g. `['2023-05-05']`), the one encoding every temporal element type accepts. Note: a `Date` used inside `Array(DateTime)` / `Array(DateTime64)` is now bound at day precision (the time-of-day is dropped), since date-only is the only form `Array(Date)` accepts; scalar `Date` / `DateTime` binding is unchanged. ([#947])
+- Fixed mid-stream exception detection so a bare `\r\n` in a _successful_ response body is no longer mistaken for an error. The detector now confirms the actual `__exception__` trailer — the fixed marker plus the random per-response `x-clickhouse-exception-tag` token — at the end of the chunk before aborting, instead of firing on any `\r`-before-`\n` pair. This unbreaks streaming binary formats such as `Parquet` and CRLF-terminated `CSV` / `TSV` (`output_format_*_crlf_end_of_line`) against ClickHouse 25.11+. The trailer-length scan is now bounded as well, so a malformed or proxy-truncated trailer returns an error instead of hanging the event loop. ([#975])
 
 [#947]: https://github.com/ClickHouse/clickhouse-js/pull/947
+[#975]: https://github.com/ClickHouse/clickhouse-js/pull/975
 
 # 1.23.1
 

packages/client-node/__tests__/unit/node_exception_tag.test.ts

@@ -0,0 +1,114 @@
+import { describe, it, expect } from "vitest";
+import type { DataFormat } from "@clickhouse/client-common";
+import { Readable } from "stream";
+import { ResultSet } from "../../src";
+import { guid } from "../../../client-common/__tests__/utils/guid";
+
+// Regression coverage for the in-band mid-stream exception detector. When an
+// error occurs after a 200 response has started streaming, ClickHouse (25.11+)
+// terminates the body with `<tag>\r\n__exception__\r\n`, echoing the random
+// per-response token from the `x-clickhouse-exception-tag` header. The detector
+// must fire ONLY on that real trailer, never on a stray `\r\n` in a successful
+// body (binary Parquet, or CRLF-terminated CSV/TSV rows).
+describe("[Node.js] mid-stream exception tag detection", () => {
+  const tag = "abcdefghijklmnop";
+
+  function makeResultSet(chunks: Buffer[], format: DataFormat = "CSV") {
+    return ResultSet.instance({
+      stream: Readable.from(chunks),
+      format,
+      query_id: guid(),
+      log_error: () => undefined,
+      response_headers: { "x-clickhouse-exception-tag": tag },
+    });
+  }
+
+  async function collectRowText(
+    rs: ReturnType<typeof makeResultSet>,
+  ): Promise<string[]> {
+    const rows: string[] = [];
+    for await (const chunk of rs.stream()) {
+      for (const row of chunk) {
+        rows.push(row.text);
+      }
+    }
+    return rows;
+  }
+
+  it("streams a successful CRLF-terminated CSV body to completion", async () => {
+    const rs = makeResultSet([Buffer.from("0\r\n1\r\n2\r\n")]);
+    const rows = await collectRowText(rs);
+    expect(rows).toHaveLength(3);
+  });
+
+  it("streams a binary body containing \\r\\n to completion", async () => {
+    const parquetish = Buffer.from([
+      0x50,
+      0x41,
+      0x52,
+      0x31, // "PAR1"
+      0x0d,
+      0x0a, // stray \r\n
+      0x00,
+      0x01,
+      0x02,
+      0x03,
+      0x0d,
+      0x0a, // stray \r\n
+      0xff,
+      0xfe,
+      0xfd,
+    ]);
+    await expect(
+      collectRowText(makeResultSet([parquetish], "Parquet")),
+    ).resolves.toBeInstanceOf(Array);
+  });
+
+  it("still surfaces a genuine mid-stream exception with the real message", async () => {
+    const errMsg =
+      "Code: 395. DB::Exception: Value passed to 'throwIf' function is non-zero: " +
+      "while executing 'FUNCTION throwIf(equals(number, 3))'. " +
+      "(FUNCTION_THROW_IF_VALUE_IS_NON_ZERO) (version 26.5.1.882)";
+    const body =
+      "0\n1\n2\n" +
+      "\r\n__exception__\r\n" +
+      tag +
+      "\r\n" +
+      errMsg +
+      "\n" +
+      (errMsg.length + 1) +
+      " " +
+      tag +
+      "\r\n__exception__\r\n";
+    await expect(
+      collectRowText(makeResultSet([Buffer.from(body, "latin1")])),
+    ).rejects.toThrow("Value passed to 'throwIf' function is non-zero");
+  });
+
+  // With output_format_*_crlf_end_of_line the row terminator is itself `\r\n`,
+  // so the `\r`-before-`\n` pre-filter matches at the FIRST row rather than at
+  // the trailer. Detection must still surface the genuine server error
+  // (extractErrorAtTheEndOfChunk always parses the trailer at the end of the
+  // chunk, independent of which newline triggered the check) — not a bogus
+  // row-keyed error — and must not hang.
+  it("surfaces the real exception message when preceding rows are CRLF-terminated", async () => {
+    const errMsg =
+      "Code: 395. DB::Exception: Value passed to 'throwIf' function is non-zero: " +
+      "while executing 'FUNCTION throwIf(equals(number, 3))'. " +
+      "(FUNCTION_THROW_IF_VALUE_IS_NON_ZERO) (version 26.5.1.882)";
+    const body =
+      "0\r\n1\r\n2\r\n" +
+      "\r\n__exception__\r\n" +
+      tag +
+      "\r\n" +
+      errMsg +
+      "\n" +
+      (errMsg.length + 1) +
+      " " +
+      tag +
+      "\r\n__exception__\r\n";
+    await expect(
+      collectRowText(makeResultSet([Buffer.from(body, "latin1")])),
+    ).rejects.toThrow("Value passed to 'throwIf' function is non-zero");
+  });
+});

packages/client-node/src/result_set.ts

@@ -11,6 +11,7 @@ import type {
 } from "./common/index";
 import {
   extractErrorAtTheEndOfChunk,
+  endsWithExceptionMarker,
   defaultJSONHandling,
   EXCEPTION_TAG_HEADER_NAME,
   CARET_RETURN,
@@ -247,11 +248,16 @@ export class ResultSet<
             }
             break;
           } else {
-            // Check for exception in the chunk (only after 25.11)
+            // Check for a mid-stream exception trailer (only after 25.11).
+            // The `\r`-before-`\n` heuristic is a cheap pre-filter; detection is
+            // only confirmed once the chunk actually ends with the exception
+            // marker, so a stray `\r\n` in a successful response body (binary
+            // Parquet, CRLF CSV/TSV rows) is no longer a false positive.
             if (
               exceptionTag !== undefined &&
               idx >= 1 &&
-              chunk[idx - 1] === CARET_RETURN
+              chunk[idx - 1] === CARET_RETURN &&
+              endsWithExceptionMarker(chunk, exceptionTag)
             ) {
               return callback(extractErrorAtTheEndOfChunk(chunk, exceptionTag));
             }

packages/client-web/CHANGELOG.md

@@ -15,8 +15,10 @@
 ## Bug fixes
 
 - Fixed `Array(Date)` / `Array(Date32)` query-parameter binding (and other temporal element types nested in arrays, tuples, and maps). A JS `Date` inside a container was serialized as a bare Unix timestamp (e.g. `[1683244800]`), which the server's `Array(Date)` element parser rejects (`CANNOT_PARSE_INPUT_ASSERTION_FAILED`). Container-nested `Date` values are now emitted as a quoted UTC date string (e.g. `['2023-05-05']`), the one encoding every temporal element type accepts. Note: a `Date` used inside `Array(DateTime)` / `Array(DateTime64)` is now bound at day precision (the time-of-day is dropped), since date-only is the only form `Array(Date)` accepts; scalar `Date` / `DateTime` binding is unchanged. ([#947])
+- Fixed mid-stream exception detection so a bare `\r\n` in a _successful_ response body is no longer mistaken for an error. The detector now confirms the actual `__exception__` trailer — the fixed marker plus the random per-response `x-clickhouse-exception-tag` token — at the end of the chunk before aborting, instead of firing on any `\r`-before-`\n` pair. This unbreaks streaming binary formats such as `Parquet` and CRLF-terminated `CSV` / `TSV` (`output_format_*_crlf_end_of_line`) against ClickHouse 25.11+. The trailer-length scan is now bounded as well, so a malformed or proxy-truncated trailer returns an error instead of hanging the event loop. ([#975])
 
 [#947]: https://github.com/ClickHouse/clickhouse-js/pull/947
+[#975]: https://github.com/ClickHouse/clickhouse-js/pull/975
 
 # 1.23.1