diff --git a/packages/client-common/__tests__/unit/stream_utils.test.ts b/packages/client-common/__tests__/unit/stream_utils.test.ts index 1e1b9f9df..5fac939fa 100644 --- a/packages/client-common/__tests__/unit/stream_utils.test.ts +++ b/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); + }); + } }); /** diff --git a/packages/client-common/src/index.ts b/packages/client-common/src/index.ts index 2a751ae7c..aca1bce1e 100644 --- a/packages/client-common/src/index.ts +++ b/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"; diff --git a/packages/client-common/src/utils/stream.ts b/packages/client-common/src/utils/stream.ts index 4023d8c39..f2666299d 100644 --- a/packages/client-common/src/utils/stream.ts +++ b/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 + * `\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: + // \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; +} diff --git a/packages/client-node/CHANGELOG.md b/packages/client-node/CHANGELOG.md index 435f9b73f..14dde3d20 100644 --- a/packages/client-node/CHANGELOG.md +++ b/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 diff --git a/packages/client-node/__tests__/unit/node_exception_tag.test.ts b/packages/client-node/__tests__/unit/node_exception_tag.test.ts new file mode 100644 index 000000000..4f82c7898 --- /dev/null +++ b/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 `\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, + ): Promise { + 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"); + }); +}); diff --git a/packages/client-node/src/result_set.ts b/packages/client-node/src/result_set.ts index cb9f827eb..dfd269563 100644 --- a/packages/client-node/src/result_set.ts +++ b/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)); } diff --git a/packages/client-web/CHANGELOG.md b/packages/client-web/CHANGELOG.md index 12fb839c6..f52fd1bff 100644 --- a/packages/client-web/CHANGELOG.md +++ b/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 diff --git a/packages/client-web/__tests__/unit/web_exception_tag.test.ts b/packages/client-web/__tests__/unit/web_exception_tag.test.ts new file mode 100644 index 000000000..fdeb930d5 --- /dev/null +++ b/packages/client-web/__tests__/unit/web_exception_tag.test.ts @@ -0,0 +1,122 @@ +import { describe, it, expect } from "vitest"; +import type { DataFormat } from "@clickhouse/client-common"; +import { guid } from "@test/utils"; +import { ResultSet } from "../../src"; + +// 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 `\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("[Web] mid-stream exception tag detection", () => { + const tag = "abcdefghijklmnop"; + + function makeResultSet(chunks: Uint8Array[], format: DataFormat = "CSV") { + return new ResultSet( + new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(chunk); + } + controller.close(); + }, + }), + format, + guid(), + { "x-clickhouse-exception-tag": tag }, + ); + } + + async function collectRowText( + rs: ReturnType, + ): Promise { + const rows: string[] = []; + const reader = rs.stream().getReader(); + while (true) { + const { done, value } = await reader.read(); + if (done) break; + for (const row of value) { + rows.push(row.text); + } + } + return rows; + } + + it("streams a successful CRLF-terminated CSV body to completion", async () => { + const rs = makeResultSet([new TextEncoder().encode("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 = new Uint8Array([ + 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([new TextEncoder().encode(body)])), + ).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([new TextEncoder().encode(body)])), + ).rejects.toThrow("Value passed to 'throwIf' function is non-zero"); + }); +}); diff --git a/packages/client-web/src/result_set.ts b/packages/client-web/src/result_set.ts index d4f7fd338..33f474dab 100644 --- a/packages/client-web/src/result_set.ts +++ b/packages/client-web/src/result_set.ts @@ -12,6 +12,7 @@ import type { import { CARET_RETURN, extractErrorAtTheEndOfChunk, + endsWithExceptionMarker, recordSpanError, } from "./common/index"; import { @@ -195,11 +196,16 @@ export class ResultSet< } else { let bytesToDecode: Uint8Array; - // 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) ) { const err = extractErrorAtTheEndOfChunk(chunk, exceptionTag); this.finishSpan(err);