From 05c13f398e0d65f9cda2aafe4c82cf9853d6cf79 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 3 Jun 2026 16:12:25 +0000 Subject: [PATCH] Adopt in-band exception stream transformer as optional ResultSet plug --- .../unit/node_exception_stream.test.ts | 232 ++++++++++++++ packages/client-node/src/result_set.ts | 49 ++- .../client-node/src/utils/exception_stream.ts | 299 ++++++++++++++++++ packages/client-node/src/utils/index.ts | 1 + 4 files changed, 568 insertions(+), 13 deletions(-) create mode 100644 packages/client-node/__tests__/unit/node_exception_stream.test.ts create mode 100644 packages/client-node/src/utils/exception_stream.ts diff --git a/packages/client-node/__tests__/unit/node_exception_stream.test.ts b/packages/client-node/__tests__/unit/node_exception_stream.test.ts new file mode 100644 index 000000000..9fa0a5e6a --- /dev/null +++ b/packages/client-node/__tests__/unit/node_exception_stream.test.ts @@ -0,0 +1,232 @@ +import { Readable } from 'stream' +import { pipeline } from 'stream/promises' +import { describe, expect, it } from 'vitest' +import { ResultSet } from '../../src' +import { + ClickHouseException, + ClickHouseExceptionStream, + ClickHouseStreamError, +} from '../../src/utils/exception_stream' + +const CRLF = Buffer.from('\r\n') +const MARKER = Buffer.from('__exception__') +const TAG = 'dmrdfnujjqvszhav' + +const MSG = + "Code: 395. DB::Exception: Value passed to 'throwIf' function is non-zero. (FUNCTION_THROW_IF_VALUE_IS_NON_ZERO) (version 25.11.1.1)" + +function buildBlock(tag: string, message: string): Buffer { + const msg = Buffer.from(message, 'utf8') + return Buffer.concat([ + CRLF, + MARKER, + CRLF, + Buffer.from(tag), + CRLF, + msg, + CRLF, + Buffer.from(`${msg.length} ${tag}`), + CRLF, + MARKER, + CRLF, + ]) +} + +function chunked(buf: Buffer, size: number): Buffer[] { + const out: Buffer[] = [] + for (let i = 0; i < buf.length; i += size) out.push(buf.subarray(i, i + size)) + return out +} + +async function collect( + chunks: Buffer[], + tag = TAG, + opts: { throwOnException?: boolean } = {}, +): Promise<{ + data: Buffer + error: Error | null + exception: ClickHouseException | null +}> { + const parser = new ClickHouseExceptionStream({ + tag, + throwOnException: opts.throwOnException ?? true, + }) + const collected: Buffer[] = [] + parser.on('data', (c: Buffer) => collected.push(c)) + let error: Error | null = null + try { + await pipeline(Readable.from(chunks), parser) + } catch (e) { + error = e as Error + } + return { + data: Buffer.concat(collected), + error, + exception: parser.getException(), + } +} + +describe('ClickHouseExceptionStream', () => { + it('passes data through when there is no exception', async () => { + const body = Buffer.from('0,0\n0,0\n0,0\n') + const r = await collect(chunked(body, 7)) + expect(r.data.equals(body)).toBe(true) + expect(r.error).toBeNull() + expect(r.exception).toBeNull() + }) + + it('detects an exception delivered in a single chunk', async () => { + const data = Buffer.from('0,0\n0,0\n') + const body = Buffer.concat([data, buildBlock(TAG, MSG)]) + const r = await collect([body]) + expect(r.data.equals(data)).toBe(true) + expect(r.error).toBeInstanceOf(ClickHouseException) + expect(r.exception?.code).toBe(395) + expect(r.exception?.errorName).toBe('FUNCTION_THROW_IF_VALUE_IS_NON_ZERO') + expect(r.exception?.message).toBe(MSG) + }) + + it('detects an exception split across 1-byte chunks', async () => { + const data = Buffer.from('0,0\n0,0\n') + const body = Buffer.concat([data, buildBlock(TAG, MSG)]) + const r = await collect(chunked(body, 1)) + expect(r.data.equals(data)).toBe(true) + expect(r.exception?.code).toBe(395) + }) + + it('preserves large data (> 16 KiB) that forces early release', async () => { + const data = Buffer.from('x'.repeat(50_000)) + const body = Buffer.concat([data, buildBlock(TAG, MSG)]) + const r = await collect(chunked(body, 4096)) + expect(r.data.equals(data)).toBe(true) + expect(r.exception?.code).toBe(395) + }) + + it('extracts a message that itself contains the literal marker', async () => { + const trickyMsg = + 'Code: 62. DB::Exception: bad token __exception__ here. (SYNTAX_ERROR) (version 25.11.1.1)' + const data = Buffer.from('row1\n') + const body = Buffer.concat([data, buildBlock(TAG, trickyMsg)]) + const r = await collect([body]) + expect(r.data.equals(data)).toBe(true) + expect(r.exception?.message).toBe(trickyMsg) + expect(r.exception?.code).toBe(62) + }) + + it('treats a marker without a matching tag as plain data', async () => { + const body = Buffer.from('{"s":"__exception__ in a string value"}\n') + const r = await collect([body]) + expect(r.data.equals(body)).toBe(true) + expect(r.exception).toBeNull() + expect(r.error).toBeNull() + }) + + it('errors with ClickHouseStreamError on a truncated block', async () => { + const data = Buffer.from('0,0\n') + const partial = Buffer.concat([ + data, + CRLF, + MARKER, + CRLF, + Buffer.from(TAG), + CRLF, + Buffer.from('Code: 1. DB::Excep'), + ]) + const r = await collect(chunked(partial, 3)) + expect(r.error).toBeInstanceOf(ClickHouseStreamError) + }) + + it('ends cleanly and exposes the exception when throwOnException is false', async () => { + const data = Buffer.from('0,0\n') + const body = Buffer.concat([data, buildBlock(TAG, MSG)]) + const r = await collect([body], TAG, { throwOnException: false }) + expect(r.error).toBeNull() + expect(r.exception?.code).toBe(395) + expect(r.data.equals(data)).toBe(true) + }) + + it('emits a clickhouse-exception event', async () => { + const data = Buffer.from('0,0\n') + const body = Buffer.concat([data, buildBlock(TAG, MSG)]) + const parser = new ClickHouseExceptionStream({ + tag: TAG, + throwOnException: false, + }) + let eventEx: ClickHouseException | null = null + parser.on('clickhouse-exception', (ex: ClickHouseException) => { + eventEx = ex + }) + parser.on('data', () => {}) + await pipeline(Readable.from([body]), parser) + expect(eventEx).not.toBeNull() + expect((eventEx as unknown as ClickHouseException).code).toBe(395) + }) + + it('requires a tag', () => { + expect(() => new ClickHouseExceptionStream({ tag: '' })).toThrow(TypeError) + }) +}) + +describe('ResultSet optional exception stream', () => { + const headers = { 'x-clickhouse-exception-tag': TAG } + + it('surfaces an in-band exception when enabled', async () => { + const data = Buffer.from('{"a":1}\n{"a":2}\n') + const body = Buffer.concat([data, buildBlock(TAG, MSG)]) + const rs = ResultSet.instance({ + stream: Readable.from(chunked(body, 5)), + format: 'JSONEachRow', + query_id: 'test', + log_error: () => {}, + response_headers: headers, + exceptionStream: true, + }) + + const rows: unknown[] = [] + let error: Error | null = null + try { + for await (const batch of rs.stream()) { + for (const row of batch) rows.push(row.json()) + } + } catch (e) { + error = e as Error + } + expect(error).toBeInstanceOf(ClickHouseException) + expect((error as ClickHouseException).code).toBe(395) + }) + + it('passes data through when enabled and there is no exception', async () => { + const data = Buffer.from('{"a":1}\n{"a":2}\n') + const rs = ResultSet.instance({ + stream: Readable.from([data]), + format: 'JSONEachRow', + query_id: 'test', + log_error: () => {}, + response_headers: headers, + exceptionStream: true, + }) + + const rows: unknown[] = [] + for await (const batch of rs.stream()) { + for (const row of batch) rows.push(row.json()) + } + expect(rows).toEqual([{ a: 1 }, { a: 2 }]) + }) + + it('is disabled by default (transformer not plugged in)', async () => { + const data = Buffer.from('{"a":1}\n') + const rs = ResultSet.instance({ + stream: Readable.from([data]), + format: 'JSONEachRow', + query_id: 'test', + log_error: () => {}, + response_headers: headers, + }) + + const rows: unknown[] = [] + for await (const batch of rs.stream()) { + for (const row of batch) rows.push(row.json()) + } + expect(rows).toEqual([{ a: 1 }]) + }) +}) diff --git a/packages/client-node/src/result_set.ts b/packages/client-node/src/result_set.ts index 5f426275a..64ae4963f 100644 --- a/packages/client-node/src/result_set.ts +++ b/packages/client-node/src/result_set.ts @@ -22,6 +22,7 @@ import { Buffer } from 'buffer' import type { Readable, TransformCallback } from 'stream' import Stream, { Transform } from 'stream' import { getAsText } from './utils' +import { ClickHouseExceptionStream } from './utils/exception_stream' const NEWLINE = 0x0a as const @@ -57,6 +58,14 @@ export interface ResultSetOptions { log_error: (error: Error) => void response_headers: ResponseHeaders jsonHandling?: JSONHandling + /** + * When enabled, an optional {@link ClickHouseExceptionStream} transformer is + * plugged into the internal stream to parse in-band exception blocks that + * ClickHouse (25.11+) appends to the response body after a 200 status. Off by + * default; requires the `X-ClickHouse-Exception-Tag` response header to be + * present, otherwise the transformer is skipped. + */ + exceptionStream?: boolean } export class ResultSet< @@ -67,6 +76,7 @@ export class ResultSet< private readonly exceptionTag: string | undefined = undefined private readonly log_error: (error: Error) => void private readonly jsonHandling: JSONHandling + private readonly exceptionStream: boolean private _consumed = false constructor( @@ -82,6 +92,7 @@ export class ResultSet< log_error?: (error: Error) => void, _response_headers?: ResponseHeaders, jsonHandling?: JSONHandling, + exceptionStream?: boolean, ) { this.jsonHandling = { ...defaultJSONHandling, @@ -89,6 +100,7 @@ export class ResultSet< } // eslint-disable-next-line no-console this.log_error = log_error ?? ((err: Error) => console.error(err)) + this.exceptionStream = exceptionStream ?? false if (_response_headers !== undefined) { this.response_headers = Object.freeze(_response_headers) @@ -201,19 +213,28 @@ export class ResultSet< objectMode: true, }) - const pipeline = Stream.pipeline( - this.consume(), - toRows, - function pipelineCb(err) { - if ( - err && - err.name !== 'AbortError' && - err.message !== resultSetClosedMessage - ) { - logError(err) - } - }, - ) + const source = this.consume() + const pipelineCb = function pipelineCb(err: NodeJS.ErrnoException | null) { + if ( + err && + err.name !== 'AbortError' && + err.message !== resultSetClosedMessage + ) { + logError(err) + } + } + + // Optionally plug the in-band exception transformer into the internal stream. + // Only used when explicitly enabled and the server advertised an exception tag. + const pipeline = + this.exceptionStream && this.exceptionTag !== undefined + ? Stream.pipeline( + source, + new ClickHouseExceptionStream({ tag: this.exceptionTag }), + toRows, + pipelineCb, + ) + : Stream.pipeline(source, toRows, pipelineCb) return pipeline as any } @@ -240,6 +261,7 @@ export class ResultSet< log_error, response_headers, jsonHandling, + exceptionStream, }: ResultSetOptions): ResultSet { return new ResultSet( stream, @@ -248,6 +270,7 @@ export class ResultSet< log_error, response_headers, jsonHandling, + exceptionStream, ) } } diff --git a/packages/client-node/src/utils/exception_stream.ts b/packages/client-node/src/utils/exception_stream.ts new file mode 100644 index 000000000..f0227d639 --- /dev/null +++ b/packages/client-node/src/utils/exception_stream.ts @@ -0,0 +1,299 @@ +import { Buffer } from 'buffer' +import type { TransformCallback } from 'stream' +import { Transform } from 'stream' + +/** + * Parser for ClickHouse's in-band HTTP exception block. + * + * When an error happens after the HTTP 200 status and headers have already been + * sent, ClickHouse cannot change the status code, so (with the default + * `http_write_exception_in_output_format=0`) it appends a format-agnostic block + * to the end of the response body: + * + * \r\n + * __exception__\r\n + * \r\n + * \r\n + * \r\n + * __exception__\r\n + * + * `` is a 16-byte random tag, also sent up front in the + * `X-ClickHouse-Exception-Tag` response header. The whole block is capped at + * 16 KiB. Because the message length comes *after* the message and the block is + * self-delimited by the tag, the robust way to extract it is to retain a 16 KiB + * sliding tail of the (already decompressed) body and parse it backwards at end + * of stream. This module does exactly that. + * + * NB: the body is expected to already be decompressed by the time it reaches + * this transformer; the client handles `Content-Encoding` upstream. + */ + +const MARKER = Buffer.from('__exception__', 'ascii') +const CRLF = Buffer.from('\r\n', 'ascii') + +/** ClickHouse caps the entire in-band exception block at 16 KiB. */ +export const DEFAULT_MAX_BLOCK_SIZE = 16 * 1024 + +export interface ClickHouseExceptionInfo { + /** Numeric error code parsed from "Code: N." (null if unparseable). */ + code: number | null + /** Symbolic error name, e.g. FUNCTION_THROW_IF_VALUE_IS_NON_ZERO (null if absent). */ + errorName: string | null + /** Full decoded error message text. */ + message: string + /** Raw message bytes exactly as sent by the server. */ + raw: Buffer +} + +/** Thrown / surfaced when ClickHouse reported an error mid-stream. */ +export class ClickHouseException extends Error { + readonly code: number | null + readonly errorName: string | null + readonly raw: Buffer + + constructor(info: ClickHouseExceptionInfo) { + super(info.message) + this.name = 'ClickHouseException' + this.code = info.code + this.errorName = info.errorName + this.raw = info.raw + } +} + +/** Thrown when the stream itself is malformed or truncated mid-exception-block. */ +export class ClickHouseStreamError extends Error { + constructor(message: string) { + super(message) + this.name = 'ClickHouseStreamError' + } +} + +export interface ClickHouseExceptionStreamOptions { + /** Value of the `X-ClickHouse-Exception-Tag` response header. Required. */ + tag: string + /** Override the retained tail size. Defaults to 16 KiB; do not set below it. */ + maxBlockSize?: number + /** + * If true (default), the stream errors with a `ClickHouseException` when an + * in-band exception is found. If false, the stream ends cleanly and the + * exception is exposed via `getException()` and the `clickhouse-exception` + * event instead. + */ + throwOnException?: boolean +} + +/** + * A Transform that passes ClickHouse result bytes straight through and detects + * an in-band exception block at the end of the stream. + * + * Pipe the *decompressed* response body through it. On success it ends cleanly. + * On an in-band exception it (by default) errors with a `ClickHouseException`. + */ +export class ClickHouseExceptionStream extends Transform { + private readonly tag: Buffer + private readonly maxBlock: number + private readonly throwOnException: boolean + /** Opening delimiter, used as a reliable "a real block started" probe: \r\n__exception__\r\n */ + private readonly openPattern: Buffer + private buf: Buffer = Buffer.alloc(0) + private exception: ClickHouseException | null = null + + constructor(opts: ClickHouseExceptionStreamOptions) { + super() + if (!opts.tag) { + throw new TypeError( + 'tag is required (value of the X-ClickHouse-Exception-Tag header)', + ) + } + this.tag = Buffer.from(opts.tag, 'ascii') + this.maxBlock = Math.max( + opts.maxBlockSize ?? DEFAULT_MAX_BLOCK_SIZE, + DEFAULT_MAX_BLOCK_SIZE, + ) + this.throwOnException = opts.throwOnException ?? true + this.openPattern = Buffer.concat([CRLF, MARKER, CRLF, this.tag]) + } + + /** The parsed exception after the stream ends, or null on success. */ + getException(): ClickHouseException | null { + return this.exception + } + + override _transform( + chunk: Buffer, + _enc: BufferEncoding, + cb: TransformCallback, + ): void { + this.buf = this.buf.length === 0 ? chunk : Buffer.concat([this.buf, chunk]) + // Anything older than the last `maxBlock` bytes cannot be part of a + // <= maxBlock block sitting at the very end, so release it as confirmed data. + if (this.buf.length > this.maxBlock) { + const releaseLen = this.buf.length - this.maxBlock + const release = this.buf.subarray(0, releaseLen) + this.buf = this.buf.subarray(releaseLen) + this.push(release) + } + cb() + } + + override _flush(cb: TransformCallback): void { + let parsed: { + exception: ClickHouseException + dataPrefixEnd: number + } | null + try { + parsed = this.parseTail(this.buf) + } catch (err) { + cb(err as Error) + return + } + + if (!parsed) { + // No exception block: the whole retained tail is result data. + if (this.buf.length > 0) this.push(this.buf) + cb() + return + } + + // Emit any trailing result bytes that preceded the block, then surface the error. + if (parsed.dataPrefixEnd > 0) { + this.push(this.buf.subarray(0, parsed.dataPrefixEnd)) + } + this.exception = parsed.exception + this.emit('clickhouse-exception', parsed.exception) + cb(this.throwOnException ? parsed.exception : undefined) + } + + private parseTail( + tail: Buffer, + ): { exception: ClickHouseException; dataPrefixEnd: number } | null { + const hasOpening = tail.indexOf(this.openPattern) !== -1 + + // Helper: a failed structural check is "truncated/malformed" if a real block + // was started (opening delimiter + matching tag present), otherwise it just + // means there is no exception and everything is data. + const bail = (reason: string): null => { + if (hasOpening) throw new ClickHouseStreamError(reason) + return null + } + + // 1. Closing marker must be the last `__exception__`, followed only by an + // optional trailing CRLF. (A message containing the literal "__exception__" + // is fine: it sits before the meta line, so it is never the last one.) + const closeIdx = tail.lastIndexOf(MARKER) + if (closeIdx === -1) { + return hasOpening + ? bail( + 'ClickHouse exception block started but never terminated (truncated stream)', + ) + : null + } + const trailing = tail.subarray(closeIdx + MARKER.length) + if (trailing.length !== 0 && !trailing.equals(CRLF)) { + return bail( + 'Malformed ClickHouse exception block (unexpected bytes after closing marker)', + ) + } + + // 2. Closing marker is preceded by CRLF; before that the meta line " ". + if (closeIdx < 2 || !tail.subarray(closeIdx - 2, closeIdx).equals(CRLF)) { + return bail( + 'Malformed ClickHouse exception block (missing CRLF before closing marker)', + ) + } + const metaEnd = closeIdx - 2 + const crlfBeforeMeta = tail.lastIndexOf(CRLF, metaEnd - 1) + if (crlfBeforeMeta === -1) { + return bail('Malformed ClickHouse exception block (no meta line)') + } + + const metaLine = tail + .subarray(crlfBeforeMeta + CRLF.length, metaEnd) + .toString('ascii') + const sp = metaLine.indexOf(' ') + if (sp === -1) { + return bail('Malformed ClickHouse exception block (bad meta line)') + } + const lenStr = metaLine.slice(0, sp) + const tagStr = metaLine.slice(sp + 1) + if (tagStr !== this.tag.toString('ascii')) { + // Tag mismatch: this "__exception__" is not ours -> treat as data. + return null + } + if (!/^\d+$/.test(lenStr)) { + return bail('Malformed ClickHouse exception block (bad message length)') + } + const declaredLen = Number(lenStr) + + // 3. The message occupies exactly `declaredLen` bytes ending at the CRLF + // before the meta line. Use the length to delimit it unambiguously. + const messageEnd = crlfBeforeMeta + const messageStart = messageEnd - declaredLen + if (messageStart < 0) { + return bail( + 'Malformed ClickHouse exception block (length exceeds buffer)', + ) + } + const raw = tail.subarray(messageStart, messageEnd) + + // 4. Validate the opening framing: is preceded by \r\n, + // which is preceded by \r\n__exception__\r\n. + const afterOpenTag = messageStart - CRLF.length + if ( + afterOpenTag < 0 || + !tail.subarray(afterOpenTag, messageStart).equals(CRLF) + ) { + return bail( + 'Malformed ClickHouse exception block (missing CRLF after opening tag)', + ) + } + const openTagStart = afterOpenTag - this.tag.length + if ( + openTagStart < 0 || + !tail.subarray(openTagStart, afterOpenTag).equals(this.tag) + ) { + return bail('Malformed ClickHouse exception block (opening tag mismatch)') + } + const afterMarker = openTagStart - CRLF.length + if ( + afterMarker < 0 || + !tail.subarray(afterMarker, openTagStart).equals(CRLF) + ) { + return bail( + 'Malformed ClickHouse exception block (missing CRLF after opening marker)', + ) + } + const markerStart = afterMarker - MARKER.length + if ( + markerStart < 0 || + !tail.subarray(markerStart, afterMarker).equals(MARKER) + ) { + return bail( + 'Malformed ClickHouse exception block (missing opening marker)', + ) + } + + // The block's injected leading CRLF (if present) is a separator, not data. + let dataPrefixEnd = markerStart + if ( + markerStart >= CRLF.length && + tail.subarray(markerStart - CRLF.length, markerStart).equals(CRLF) + ) { + dataPrefixEnd = markerStart - CRLF.length + } + + return { exception: toException(raw), dataPrefixEnd } + } +} + +function toException(raw: Buffer): ClickHouseException { + const message = raw.toString('utf8') + const codeMatch = /^Code:\s*(\d+)\./.exec(message) + const nameMatch = /\(([A-Z][A-Z0-9_]+)\)/.exec(message) + return new ClickHouseException({ + code: codeMatch ? Number(codeMatch[1]) : null, + errorName: nameMatch ? nameMatch[1] : null, + message, + raw, + }) +} diff --git a/packages/client-node/src/utils/index.ts b/packages/client-node/src/utils/index.ts index d9fa48708..db6b37690 100644 --- a/packages/client-node/src/utils/index.ts +++ b/packages/client-node/src/utils/index.ts @@ -1,4 +1,5 @@ export * from './stream' +export * from './exception_stream' export * from './encoder' export * from './process' export * from './user_agent'