From 0f286f0f42245c717b81d7ad3a239fddb475a5e5 Mon Sep 17 00:00:00 2001 From: Boris Tyshkevich Date: Thu, 6 Aug 2026 12:34:10 +0200 Subject: [PATCH 1/5] feat(#585 phase 1): add the ClickHouseTransport contract + current HTTP implementation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defines the narrow SQL Browser transport contract (src/net/clickhouse-transport.types.ts: ClickHouseTransport, TransportDeps, TransportRequest, StreamCallbacks) and puts the current custom HTTP transport behind it (src/net/clickhouse-http-transport.ts: createHttpTransport, plus chUrl/ChUrlOpts moved verbatim from ch-client.ts). A reusable contract-test-suite factory (tests/unit/clickhouse-transport-contract.ts) registers against createHttpTransport only, since no official implementation exists (ADR-0005 is Rejected) and Phases 2-4 do not proceed without a new decision. tests/unit/clickhouse-http-transport.test.ts covers the moved progress-line stream loop, including a split multi-byte UTF-8 character across byte chunks and per-chunk onLine-before-onChunk ordering. ch-client.ts is not yet wired to this seam in this commit — that lands next. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01LwFPT465eDJqYcRa8HGNLz --- src/net/clickhouse-http-transport.ts | 96 +++++++++++ src/net/clickhouse-transport.types.ts | 97 +++++++++++ tests/unit/clickhouse-http-transport.test.ts | 171 +++++++++++++++++++ tests/unit/clickhouse-transport-contract.ts | 163 ++++++++++++++++++ 4 files changed, 527 insertions(+) create mode 100644 src/net/clickhouse-http-transport.ts create mode 100644 src/net/clickhouse-transport.types.ts create mode 100644 tests/unit/clickhouse-http-transport.test.ts create mode 100644 tests/unit/clickhouse-transport-contract.ts diff --git a/src/net/clickhouse-http-transport.ts b/src/net/clickhouse-http-transport.ts new file mode 100644 index 00000000..475719d0 --- /dev/null +++ b/src/net/clickhouse-http-transport.ts @@ -0,0 +1,96 @@ +// Issue #585 Phase 1 — the current custom ClickHouse HTTP transport, +// re-seated behind the `ClickHouseTransport` contract (`clickhouse-transport.types.ts`). +// This is a pure move: `chUrl` (+ its `ChUrlOpts` parameter type) and the +// progress-line stream-read loop are relocated here verbatim from +// `ch-client.ts`, which re-imports/re-exports both so every existing importer +// keeps resolving. No behavior change; no product SQL; no auth/lifecycle +// policy (that stays app-side in `ch-client.ts`'s `authedFetch`). +// +// Ownership boundary: this file may depend only on `src/core` — never on +// `ch-client.ts`, `oauth.ts`, `oauth-config.ts`, `src/application/`, or +// `src/ui/`. `build/check-boundaries.mjs` enforces this mechanically. + +import type { ClickHouseTransport, StreamCallbacks, TransportDeps, TransportRequest } from './clickhouse-transport.types.js'; +import type { StreamLine } from '../core/stream.js'; + +/** `chUrl`'s query-string options. */ +export interface ChUrlOpts { + format?: string; + extra?: Record; + params?: Record; +} + +/** Build a ClickHouse HTTP URL with query-string options. Pure. */ +export function chUrl(origin: string, opts: ChUrlOpts = {}): string { + const format = opts.format || 'JSONStringsEachRowWithProgress'; + let url = origin + '?default_format=' + format + '&enable_http_compression=1'; + for (const [k, v] of Object.entries(opts.extra || {})) { + url += '&' + k + '=' + encodeURIComponent(v); + } + for (const [k, v] of Object.entries(opts.params || {})) { + url += '&' + k + '=' + encodeURIComponent(v); + } + return url; +} + +/** Drives the progress-bearing JSON-lines read loop: decode, line split, + * `JSON.parse` per line, trailing-buffer flush, malformed-line skip — + * byte-for-byte the loop formerly inlined in `runQuery`. `onLine` fires per + * parsed object, `onChunk` once per network chunk. A single `TextDecoder` + * used with `{ stream: true }` for the whole body (not per-chunk) so a + * multi-byte UTF-8 character split across two byte chunks still decodes + * correctly. */ +async function streamLines(body: ReadableStream, cbs: StreamCallbacks): Promise { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines[lines.length - 1]; + for (const line of lines.slice(0, -1)) { + if (!line) continue; + let json: StreamLine; + try { + json = JSON.parse(line); + } catch { + continue; + } + cbs.onLine && cbs.onLine(json); + } + cbs.onChunk && cbs.onChunk(); + } + if (buffer.trim()) { + try { + cbs.onLine && cbs.onLine(JSON.parse(buffer)); + } catch { + /* trailing partial line */ + } + } +} + +/** The current custom HTTP implementation of `ClickHouseTransport`. `deps`' + * accessors are read per-request (REQUIRED-PURE — see the contract's doc + * comment) so a live, mutable `origin`/`fetch` (e.g. `ConnectionSession`'s + * `chCtx`, mutated in place on sign-in) is always observed at its current + * value, never pinned to a stale snapshot (Adaptation A5). */ +export function createHttpTransport(deps: TransportDeps): ClickHouseTransport { + return { + async send(request: TransportRequest): Promise { + const url = chUrl(deps.origin(), { + format: request.defaultFormat, + extra: request.settings, + params: request.params, + }); + return deps.fetch()(url, { + method: 'POST', + body: request.sql, + headers: { Authorization: request.authorization }, + signal: request.signal, + }); + }, + streamLines, + }; +} diff --git a/src/net/clickhouse-transport.types.ts b/src/net/clickhouse-transport.types.ts new file mode 100644 index 00000000..09827020 --- /dev/null +++ b/src/net/clickhouse-transport.types.ts @@ -0,0 +1,97 @@ +// Issue #585 Phase 1 — the narrow SQL Browser ClickHouse-transport contract. +// Type-only (ADR-0002 phase-0 convention for seam contracts, hence the +// `.types.ts` suffix rather than the issue's suggested `clickhouse-transport.ts` +// — "exact names may follow repository conventions" per the issue). Puts the +// CURRENT custom HTTP implementation behind a contract a future official +// transport (Phase 2, gated on a new decision — ADR-0005 is Rejected) could +// also satisfy, without moving any product SQL or auth/lifecycle policy here. +// +// Ownership boundary: this file (and its implementation, +// `clickhouse-http-transport.ts`) may depend only on `src/core` (the narrow +// `StreamLine` type) — never on `ch-client.ts`, `oauth.ts`, `oauth-config.ts`, +// `src/application/`, or `src/ui/`, even type-only. `build/check-boundaries.mjs` +// enforces this mechanically (twin `RULES` entries for this file and the +// implementation file). + +import type { StreamLine } from '../core/stream.js'; + +/** What the transport is allowed to see of the environment. Deliberately + * excludes tokens, refresh, epochs, lifecycle callbacks: a transport + * implementation is compile-time incapable of ACQUIRING credentials or + * signaling lifecycle. (It still receives the resolved Authorization header + * per request — Adaptation A6 — so single-send/no-retry/no-caching discipline + * is contract- and test-enforced, not compiler-enforced.) Accessors, not + * snapshots — the live chCtx's origin is mutated in place on sign-in and the + * transport must observe the current value per request. + * + * REQUIRED-PURE: both accessors must be synchronous, side-effect-free plain + * property reads (production: `() => ctx.fetch` / `() => ctx.origin`). This + * matters because `send` evaluates them AFTER `authedFetch`'s final epoch + * fence and before the fetch itself; the type system cannot express purity, + * so — exactly like A6's single-send discipline — this rule is enforced by + * this doc comment and review, not by the compiler or the existing epoch + * race test (whose proof stops at the `send` invocation boundary). */ +export interface TransportDeps { + fetch(): typeof fetch; + origin(): string; +} + +/** One ClickHouse HTTP request, fully specified. No client-level defaults + * exist: `authorization` is the complete header value (scheme + credential), + * resolved by the caller (SQL Browser auth policy) for THIS request. */ +export interface TransportRequest { + /** Opaque SQL text. The transport never parses, rewrites, or appends to it + * (hard invariant 16: an authored FORMAT clause always wins over + * `defaultFormat` server-side, exactly as today). */ + sql: string; + /** Exact ClickHouse format name sent as `default_format`. */ + defaultFormat: string; + /** HTTP query-string settings (wait_end_of_query, max_result_rows, + * result_overflow_mode, add_http_cors_header, readonly, …) — the caller's + * policy decides which; the transport only serializes. */ + settings?: Record; + /** Query-string params riding alongside: native `param_*` parameters, + * `query_id`, `session_id`, `role` — today's exact wire vocabulary, + * unchanged (Adaptation A2). */ + params?: Record; + /** Complete Authorization header value. Never optional, never defaulted. */ + authorization: string; + signal?: AbortSignal; +} + +// No TransportResponse type in Phase 1 (Adaptation A3): `send` resolves with +// the NATIVE fetch `Response`. A structural subset would be assignable only in +// the direction Response -> subset, so authedFetch/exportQuery could not keep +// their `Promise` signatures without an unsafe cast. Native Response +// gives raw bytes (`body`, hard invariant 17) and `clone()` for authedFetch's +// non-destructive error-body peek for free. + +/** Callbacks driving `streamLines`' progress-bearing JSON-lines read loop. */ +export interface StreamCallbacks { + onLine?: (line: StreamLine) => void; + onChunk?: () => void; +} + +/** The SQL Browser transport contract. In Phase 1 exactly one implementation + * exists (`createHttpTransport`, `clickhouse-http-transport.ts`); a Phase 2 + * official-client implementation (does not proceed without a new decision) + * would satisfy the same contract. */ +export interface ClickHouseTransport { + /** POST one query; resolves at HTTP settlement (headers received) with the + * NATIVE fetch `Response` — Phase 1 defines no adapter-owned response type + * (Adaptation A3), which is what preserves `authedFetch`/`exportQuery`'s + * `Promise` signatures and `export-service.ts`'s + * `streamToFile(resp: Response, …)` consumer without casts. Exactly one + * fetch invocation (contract-suite-asserted, incl. on non-2xx — A6); no + * retry, no token read, no lifecycle callback, no error classification, no + * body consumption. HTTP error statuses resolve (they are responses); only + * network I/O failure / abort rejects. */ + send(request: TransportRequest): Promise; + /** Supported-stream mechanics for the progress-bearing JSON-lines formats: + * drives the read loop (decode, line split, JSON.parse, trailing-buffer + * flush, malformed-line skip), invoking onLine per parsed object and + * onChunk per network chunk — byte-for-byte the loop currently inlined in + * runQuery. Consuming a body is a caller decision made AFTER policy has + * classified the settled response. */ + streamLines(body: ReadableStream, cbs: StreamCallbacks): Promise; +} diff --git a/tests/unit/clickhouse-http-transport.test.ts b/tests/unit/clickhouse-http-transport.test.ts new file mode 100644 index 00000000..5fba628d --- /dev/null +++ b/tests/unit/clickhouse-http-transport.test.ts @@ -0,0 +1,171 @@ +import { describe, expect, it, vi } from 'vitest'; +import { chUrl, createHttpTransport } from '../../src/net/clickhouse-http-transport.js'; +import type { StreamLine } from '../../src/core/stream.js'; +import { runTransportContractSuite } from './clickhouse-transport-contract.js'; + +// Issue #585 Phase 1 — direct spec for the moved current-HTTP transport +// implementation. Registers the shared contract suite once (this is the +// ONLY implementation Phase 1 registers — see the suite factory's header +// comment), then covers the mechanics the shared suite deliberately leaves +// implementation-specific: `chUrl`'s exact URL shape and the moved +// progress-line stream loop, including two edge cases a manual move can +// silently alter (a split multi-byte UTF-8 character across byte chunks, +// and per-chunk onLine-before-onChunk ordering). + +runTransportContractSuite('createHttpTransport', createHttpTransport); + +function deps(fetchImpl: (url: string, init: RequestInit) => Response | Promise, origin = 'https://ch.example') { + const fetchMock = vi.fn(fetchImpl); + return { fetchMock, deps: { fetch: () => fetchMock as unknown as typeof fetch, origin: () => origin } }; +} + +// A stream that yields exactly the given byte chunks, in order — needed for +// the UTF-8-split case, which a whole-string-per-chunk helper is structurally +// incapable of producing. +function byteStream(chunks: Uint8Array[]): ReadableStream { + let i = 0; + return new ReadableStream({ + pull(controller) { + if (i < chunks.length) controller.enqueue(chunks[i++]); + else controller.close(); + }, + }); +} + +function stringStream(chunks: string[]): ReadableStream { + return byteStream(chunks.map((c) => new TextEncoder().encode(c))); +} + +describe('chUrl', () => { + it('uses default format and compression', () => { + expect(chUrl('https://o')).toBe('https://o?default_format=JSONStringsEachRowWithProgress&enable_http_compression=1'); + }); + it('applies format, extra and params', () => { + const url = chUrl('https://o', { format: 'JSON', extra: { wait_end_of_query: 1 }, params: { x: 'a b' } }); + expect(url).toContain('default_format=JSON'); + expect(url).toContain('wait_end_of_query=1'); + expect(url).toContain('x=a%20b'); + }); +}); + +describe('createHttpTransport().send — exact request shape', () => { + it('builds the exact literal URL from origin/format/settings/params, POSTs the SQL body, and sends the complete Authorization header', async () => { + const { fetchMock, deps: d } = deps(() => new Response('ok')); + const transport = createHttpTransport(d); + await transport.send({ + sql: 'SELECT 1', + defaultFormat: 'JSONCompact', + settings: { wait_end_of_query: 1 }, + params: { param_id: '5', query_id: 'q1', session_id: 's1', role: 'analyst' }, + authorization: 'Bearer tok', + }); + const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(url).toBe( + 'https://ch.example?default_format=JSONCompact&enable_http_compression=1' + + '&wait_end_of_query=1¶m_id=5&query_id=q1&session_id=s1&role=analyst', + ); + expect(init.method).toBe('POST'); + expect(init.body).toBe('SELECT 1'); + expect((init.headers as Record).Authorization).toBe('Bearer tok'); + }); + + it('threads the abort signal through to fetch', async () => { + const controller = new AbortController(); + const { fetchMock, deps: d } = deps(() => new Response('ok')); + const transport = createHttpTransport(d); + await transport.send({ sql: 'x', defaultFormat: 'JSON', authorization: 'Bearer t', signal: controller.signal }); + expect((fetchMock.mock.calls[0][1] as RequestInit).signal).toBe(controller.signal); + }); +}); + +describe('createHttpTransport().streamLines — moved progress-line loop mechanics', () => { + it('reassembles a line split across multiple chunks', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['{"row":{"a"', ':"1"}}\n']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ row: { a: '1' } }]); + }); + + it('skips empty lines between JSON objects', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['\n\n{"row":{"a":"1"}}\n\n']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ row: { a: '1' } }]); + }); + + it('skips a malformed JSON line without throwing', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['not json\n', '{"row":{"a":"1"}}\n']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ row: { a: '1' } }]); + }); + + it('flushes a valid trailing partial line (no terminating newline)', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['{"row":{"a":"1"}}']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ row: { a: '1' } }]); + }); + + it('discards a malformed trailing partial line without throwing', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['{bad trailing']); + await expect(createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) })).resolves.toBeUndefined(); + expect(lines).toEqual([]); + }); + + it('delivers an in-band exception line via onLine (hard invariant 10 feed-through)', async () => { + const { deps: d } = deps(() => new Response('ok')); + const lines: StreamLine[] = []; + const stream = stringStream(['{"exception":"DB::Exception: boom"}\n']); + await createHttpTransport(d).streamLines(stream, { onLine: (l) => lines.push(l) }); + expect(lines).toEqual([{ exception: 'DB::Exception: boom' }]); + }); + + it('calls onChunk once per network read', async () => { + const { deps: d } = deps(() => new Response('ok')); + let chunkCalls = 0; + const stream = stringStream(['{"row":{}}\n', '{"row":{}}\n', '{"row":{}}\n']); + await createHttpTransport(d).streamLines(stream, { onChunk: () => { chunkCalls++; } }); + expect(chunkCalls).toBe(3); + }); + + it('fires every onLine callback produced from a chunk before that chunk\'s onChunk (call-order)', async () => { + const { deps: d } = deps(() => new Response('ok')); + const order: string[] = []; + const stream = stringStream(['{"row":{"a":"1"}}\n{"row":{"a":"2"}}\n']); + await createHttpTransport(d).streamLines(stream, { + onLine: () => order.push('line'), + onChunk: () => order.push('chunk'), + }); + expect(order).toEqual(['line', 'line', 'chunk']); + }); + + it('tolerates entirely missing onLine/onChunk callbacks', async () => { + const { deps: d } = deps(() => new Response('ok')); + const stream = stringStream(['{"row":{}}\n', '{"row":{}}']); + await expect(createHttpTransport(d).streamLines(stream, {})).resolves.toBeUndefined(); + }); + + it('decodes a multi-byte UTF-8 character split across two byte chunks correctly (single TextDecoder with {stream:true})', async () => { + const { deps: d } = deps(() => new Response('ok')); + const line = '{"row":{"a":"€"}}\n'; // '€' — 3 UTF-8 bytes: 0xE2 0x82 0xAC + const fullBytes = new TextEncoder().encode(line); + let euroIdx = -1; + for (let i = 0; i < fullBytes.length - 2; i++) { + if (fullBytes[i] === 0xe2 && fullBytes[i + 1] === 0x82 && fullBytes[i + 2] === 0xac) { euroIdx = i; break; } + } + expect(euroIdx).toBeGreaterThan(-1); + const splitAt = euroIdx + 1; // split INSIDE the euro sign's 3-byte sequence + const chunk1 = fullBytes.slice(0, splitAt); + const chunk2 = fullBytes.slice(splitAt); + const lines: StreamLine[] = []; + await createHttpTransport(d).streamLines(byteStream([chunk1, chunk2]), { onLine: (l) => lines.push(l) }); + expect((lines[0].row as Record).a).toBe('€'); + }); +}); diff --git a/tests/unit/clickhouse-transport-contract.ts b/tests/unit/clickhouse-transport-contract.ts new file mode 100644 index 00000000..cf6e95f5 --- /dev/null +++ b/tests/unit/clickhouse-transport-contract.ts @@ -0,0 +1,163 @@ +// Issue #585 Phase 1 — reusable `ClickHouseTransport` contract-test-suite +// FACTORY. Deliberately NOT itself a `.test.ts` spec: vitest's `include` glob +// (`tests/vitest.config.ts`) is `tests/unit/**/*.test.{js,ts}`, so this file +// is never discovered directly — a real spec imports and calls +// `runTransportContractSuite(name, makeTransport)` to register it against one +// concrete implementation. +// +// Phase 1 registers this suite against `createHttpTransport` ONLY +// (`clickhouse-http-transport.test.ts`), because no official implementation +// exists yet and Phases 2-4 do not proceed without a new decision +// (docs/ADR-0005-clickhouse-web-client.md is Rejected). A future Phase 2 +// transport would import this same factory and re-run it unchanged against +// its own `makeTransport` — "structurally ready", not implementation-neutral: +// today's contract is current-HTTP-specific (native `Response` return, +// complete pre-resolved `authorization` header — Adaptations A3/A6), so this +// suite proves today's ONE implementation satisfies the contract, not that +// the contract already generalizes to a hypothetical second one. + +import { describe, expect, it, vi } from 'vitest'; +import type { ClickHouseTransport, TransportDeps, TransportRequest } from '../../src/net/clickhouse-transport.types.js'; + +type FetchImpl = (url: string, init: RequestInit) => Response | Promise; +type HeadersRecord = Record; + +/** A concrete `ClickHouseTransport` implementation under test, built from a + * `TransportDeps` this factory constructs and controls (so cases can flip + * `setOrigin` mid-test — Adaptation A5 / sabotage case 2 — and inspect every + * call the implementation made to the stub fetch). */ +export type MakeTransport = (deps: TransportDeps) => ClickHouseTransport; + +function baseRequest(overrides: Partial = {}): TransportRequest { + return { + sql: 'SELECT 1', + defaultFormat: 'JSON', + authorization: 'Bearer tok', + ...overrides, + }; +} + +export function runTransportContractSuite(name: string, makeTransport: MakeTransport): void { + describe(`ClickHouseTransport contract — ${name}`, () => { + function harness(fetchImpl: FetchImpl) { + const fetchMock = vi.fn(fetchImpl); + let origin = 'https://ch.example'; + const transport = makeTransport({ + fetch: () => fetchMock as unknown as typeof fetch, + origin: () => origin, + }); + return { transport, fetchMock, setOrigin: (o: string) => { origin = o; } }; + } + + it('serializes default_format, settings, params, and the complete Authorization header into one POST whose body is the SQL byte-identical', async () => { + const { transport, fetchMock } = harness(() => new Response('ok')); + await transport.send(baseRequest({ + sql: 'SELECT 1 FORMAT CSV', // an authored FORMAT clause — never appended to again + defaultFormat: 'JSONCompact', + settings: { wait_end_of_query: 1 }, + params: { param_id: '5', query_id: 'q1', session_id: 's1', role: 'analyst' }, + authorization: 'Bearer secret-token', + })); + expect(fetchMock).toHaveBeenCalledTimes(1); + const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(url).toContain('default_format=JSONCompact'); + expect(url).toContain('enable_http_compression=1'); + expect(url).toContain('wait_end_of_query=1'); + expect(url).toContain('param_id=5'); + expect(url).toContain('query_id=q1'); + expect(url).toContain('session_id=s1'); + expect(url).toContain('role=analyst'); + expect(init.method).toBe('POST'); + expect(init.body).toBe('SELECT 1 FORMAT CSV'); + expect((init.headers as HeadersRecord).Authorization).toBe('Bearer secret-token'); + }); + + it('omits settings/params from the URL when absent (no client-level defaults)', async () => { + const { transport, fetchMock } = harness(() => new Response('ok')); + await transport.send(baseRequest()); + const url = fetchMock.mock.calls[0][0] as string; + expect(url).toBe('https://ch.example?default_format=JSON&enable_http_compression=1'); + }); + + it('carries each send\'s own authorization value with no state cached between sends', async () => { + const { transport, fetchMock } = harness(() => new Response('ok')); + await transport.send(baseRequest({ authorization: 'Bearer first' })); + await transport.send(baseRequest({ authorization: 'Bearer second' })); + expect((fetchMock.mock.calls[0][1] as RequestInit).headers as unknown as HeadersRecord).toEqual({ Authorization: 'Bearer first' }); + expect((fetchMock.mock.calls[1][1] as RequestInit).headers as unknown as HeadersRecord).toEqual({ Authorization: 'Bearer second' }); + }); + + it('invokes the stub fetch exactly once per send — no internal retry or header caching — on a 2xx response', async () => { + const { transport, fetchMock } = harness(() => new Response('ok', { status: 200 })); + await transport.send(baseRequest()); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('invokes the stub fetch exactly once per send on a non-2xx response too', async () => { + const { transport, fetchMock } = harness(() => new Response('denied', { status: 403 })); + await transport.send(baseRequest()); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('resolves (never throws) on a non-2xx response, with the body reaching the caller byte-identical', async () => { + const { transport } = harness(() => new Response('{"exception":"Code: 60. DB::Exception: table not found"}', { status: 500 })); + const resp = await transport.send(baseRequest()); + expect(resp.status).toBe(500); + expect(await resp.text()).toBe('{"exception":"Code: 60. DB::Exception: table not found"}'); + }); + + it('rejects with the network/abort failure rather than resolving, for an aborted signal', async () => { + const controller = new AbortController(); + controller.abort(); + const { transport } = harness((_url, init) => { + if ((init.signal as AbortSignal | undefined)?.aborted) { + const err = Object.assign(new Error('aborted'), { name: 'AbortError' }); + return Promise.reject(err); + } + return Promise.resolve(new Response('ok')); + }); + await expect(transport.send(baseRequest({ signal: controller.signal }))).rejects.toMatchObject({ name: 'AbortError' }); + }); + + it('surfaces a mid-stream abort from streamLines rather than swallowing it', async () => { + const { transport } = harness(() => new Response('ok')); + const abortError = Object.assign(new Error('aborted'), { name: 'AbortError' }); + const stream = new ReadableStream({ + start(controller) { controller.error(abortError); }, + }); + await expect(transport.streamLines(stream, {})).rejects.toBe(abortError); + }); + + it('preserves response status and arbitrary headers, including X-ClickHouse-Summary', async () => { + const { transport } = harness(() => new Response('ok', { + status: 200, + headers: { 'X-ClickHouse-Summary': '{"read_rows":"1"}' }, + })); + const resp = await transport.send(baseRequest()); + expect(resp.status).toBe(200); + expect(resp.headers.get('X-ClickHouse-Summary')).toBe('{"read_rows":"1"}'); + }); + + it('never consumes the response body inside send — bodyUsed stays false and the raw bytes are still readable', async () => { + const bytes = new TextEncoder().encode('raw-bytes'); + const stream = new ReadableStream({ + start(controller) { controller.enqueue(bytes); controller.close(); }, + }); + const { transport } = harness(() => new Response(stream, { status: 200 })); + const resp = await transport.send(baseRequest()); + expect(resp.bodyUsed).toBe(false); + const reader = resp.body!.getReader(); + const { value } = await reader.read(); + expect(value).toEqual(bytes); + }); + + it('reads deps.origin() live per request instead of snapshotting it at construction time (Adaptation A5)', async () => { + const { transport, fetchMock, setOrigin } = harness(() => new Response('ok')); + await transport.send(baseRequest()); + setOrigin('https://new-cluster.example'); + await transport.send(baseRequest()); + expect(fetchMock.mock.calls[0][0]).toContain('https://ch.example'); + expect(fetchMock.mock.calls[1][0]).toContain('https://new-cluster.example'); + }); + }); +} From 9e093deb15b0597d297e8af610f405ca84fc3912 Mon Sep 17 00:00:00 2001 From: Boris Tyshkevich Date: Thu, 6 Aug 2026 12:34:23 +0200 Subject: [PATCH 2/5] refactor(#585 phase 1): route ch-client.ts through the transport seam authedFetch's internals now delegate through the transport (transportFor(ctx) := createHttpTransport({ fetch: () => ctx.fetch, origin: () => ctx.origin })) instead of calling chUrl + ctx.fetch directly. This is authedFetch's one exported-signature change (Claimed item 3): it now takes a structured Omit instead of a prebuilt URL string, and snapshots the incoming request's settings/params synchronously at entry, before its first await (ctx.getToken()) -- one centralized aliasing defense for every caller present and future, replacing chUrl's old pre-await URL serialization as the capture point. queryJson/runQuery/exportQuery/killQueryWithLease all delegate through the transport; runQuery's inline stream-parsing loop is deleted in favor of transport.streamLines. killQueryWithLease builds a one-shot transport directly from the frozen lease (never transportFor/authedFetch), preserving hard invariant 8 (no mutable-auth reads during cleanup). ChCtx is unchanged -- no transport field, no runtime switch; ch-client.ts re-exports chUrl, ChUrlOpts, and the new contract types so every existing importer (including tests/spike/clickhouse-client/current-adapter.ts) keeps resolving. tests/unit/ch-client.test.ts: ~25 direct authedFetch call sites updated mechanically to the new signature (no assertions changed); added invocation-time settings/params capture tests (mutation during a pending getToken and during a pending one-refresh retry), a live-origin-authority test through the production transportFor wiring, and a killQueryWithLease sabotage-guard test proving getToken/refresh are never read even if the lease object happens to carry them. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01LwFPT465eDJqYcRa8HGNLz --- src/net/ch-client.ts | 136 +++++++++++++++++------------------ tests/unit/ch-client.test.ts | 134 +++++++++++++++++++++++++++------- 2 files changed, 177 insertions(+), 93 deletions(-) diff --git a/src/net/ch-client.ts b/src/net/ch-client.ts index 08add363..577d65d3 100644 --- a/src/net/ch-client.ts +++ b/src/net/ch-client.ts @@ -12,6 +12,19 @@ import type { StreamLine } from '../core/stream.js'; import { parseAstTables, buildSchemaGraph, externalDbs } from '../core/schema-graph.js'; import type { SchemaGraphTableRow, SchemaGraphDictRow } from '../core/schema-graph.js'; import { sqlString } from '../core/format.js'; +// Issue #585 Phase 1 — the transport seam. `chUrl` moved verbatim to +// `clickhouse-http-transport.ts`; re-exported here (with its `ChUrlOpts` +// parameter type) so every existing importer — including +// `tests/spike/clickhouse-client/current-adapter.ts` — keeps resolving. The +// generic request-construction/fetch/stream mechanics live in +// `createHttpTransport`; this module keeps every auth/epoch/retry policy, +// product operation, and `ChCtx` exactly as before, delegating through the +// transport instead of calling `chUrl`/`ctx.fetch` directly. +import { chUrl, createHttpTransport } from './clickhouse-http-transport.js'; +import type { TransportRequest } from './clickhouse-transport.types.js'; +export { chUrl }; +export type { ChUrlOpts } from './clickhouse-http-transport.js'; +export type { ClickHouseTransport, StreamCallbacks, TransportDeps, TransportRequest } from './clickhouse-transport.types.js'; // ── Injected ctx seam ──────────────────────────────────────────────────────── @@ -101,33 +114,41 @@ export interface ChJsonResult> { data?: T[]; } -/** `chUrl`'s query-string options. */ -export interface ChUrlOpts { - format?: string; - extra?: Record; - params?: Record; -} - -/** Build a ClickHouse HTTP URL with query-string options. Pure. */ -export function chUrl(origin: string, opts: ChUrlOpts = {}): string { - const format = opts.format || 'JSONStringsEachRowWithProgress'; - let url = origin + '?default_format=' + format + '&enable_http_compression=1'; - for (const [k, v] of Object.entries(opts.extra || {})) { - url += '&' + k + '=' + encodeURIComponent(v); - } - for (const [k, v] of Object.entries(opts.params || {})) { - url += '&' + k + '=' + encodeURIComponent(v); - } - return url; +/** Delegates unconditionally to the single current transport implementation. + * `deps.fetch`/`deps.origin` are accessors reading the LIVE mutable `ctx` + * fields per request (never a snapshot) — `ctx.origin` is mutated in place on + * sign-in (`connection-session.ts`), so a request issued after that mutation + * must observe the new value. `ChCtx` itself gains no new field: there is no + * production runtime switch, only this one unconditional wiring (an + * injectable composition seam is introduced only when a second + * implementation actually exists — Phase 2, which requires a new decision). */ +function transportFor(ctx: ChCtx) { + return createHttpTransport({ fetch: () => ctx.fetch, origin: () => ctx.origin }); } /** - * POST `sql` to ClickHouse with one automatic token-refresh retry. Resolves to - * the raw Response. Throws Error('signed out') after calling ctx.onSignedOut() - * when authentication cannot be recovered. + * POST `request.sql` to ClickHouse with one automatic token-refresh retry. + * Resolves to the raw Response. Throws Error('signed out') after calling + * ctx.onSignedOut() when authentication cannot be recovered. `request` omits + * `authorization` — this function resolves the credential for THIS request + * (and its retry) itself; every other `TransportRequest` field is the + * caller's request, unchanged. */ -export async function authedFetch(ctx: ChCtx, url: string, sql: string, signal?: AbortSignal): Promise { +export async function authedFetch(ctx: ChCtx, request: Omit): Promise { const requestEpoch = ctx.currentEpoch?.(); + // Centralized aliasing defense (review finding folded in, pass-5 revision): + // snapshot the incoming request's settings/params synchronously HERE, at + // entry, before the first await (`ctx.getToken()`) — one mechanism for + // every present and future caller, rather than per-call-site defensive + // spreads. This preserves today's invocation-time capture (today `chUrl` + // serializes both records into the URL string synchronously, before this + // function's first await), so a caller that retains and mutates either + // record while a token/refresh await is pending cannot change the request + // this function already committed to sending — on the initial attempt AND + // the one-refresh retry alike. + const settings = request.settings ? { ...request.settings } : undefined; + const params = request.params ? { ...request.params } : undefined; + const { sql, defaultFormat, signal } = request; const token = await ctx.getToken(); // getToken may have awaited a sign-in/sign-out replacement. Its credential // belongs to that replacement and this request must not send it. @@ -141,19 +162,19 @@ export async function authedFetch(ctx: ChCtx, url: string, sql: string, signal?: // ctx.authHeader(token) lets the app pick the scheme (Bearer vs Basic); // default to Bearer so the seam stays optional. const authHeader = ctx.authHeader || ((t: string) => 'Bearer ' + t); + const transport = transportFor(ctx); for (;;) { let resp: Response; try { // Fence every attempt immediately before the injected side effect. A // retry must never send a replacement session's newly-read credential. + // (Precision: `transport.send` internally evaluates the REQUIRED-PURE + // `deps.origin()`/`deps.fetch()` accessors and builds the URL AFTER + // this fence, immediately before the fetch itself — see + // `clickhouse-transport.types.ts`'s `TransportDeps` doc comment.) const authorization = authHeader(bearer); if (!isCurrentEpoch(ctx, requestEpoch)) throw staleEpochAbort(); - resp = await ctx.fetch(url, { - method: 'POST', - body: sql, - headers: { Authorization: authorization }, - signal, - }); + resp = await transport.send({ sql, defaultFormat, settings, params, authorization, signal }); } catch (e) { // Only a rejected fetch is a transport failure. HTTP failures are normal // responses and caller cancellation is deliberately invisible here. @@ -222,7 +243,7 @@ export async function queryJson>( extra?: Record, params?: Record, ): Promise> { - const resp = await authedFetch(ctx, chUrl(ctx.origin, { format: 'JSON', extra, params }), sql, signal); + const resp = await authedFetch(ctx, { sql, defaultFormat: 'JSON', settings: extra, params, signal }); if (!resp.ok) throw new Error(parseExceptionText(await resp.text())); return resp.json(); } @@ -329,10 +350,14 @@ export async function killQueryWithLease( ): Promise { if (!queryId) return; try { - await lease.fetch(chUrl(lease.origin, { format: 'JSON' }), { - method: 'POST', - body: 'KILL QUERY WHERE query_id = ' + sqlString(queryId) + ' ASYNC', - headers: { Authorization: lease.authorization }, + // A one-shot transport built directly from the frozen lease — never + // `transportFor(ctx)` / `authedFetch` — so cleanup reads no mutable auth, + // token, or refresh state (hard invariant 8). + const transport = createHttpTransport({ fetch: () => lease.fetch, origin: () => lease.origin }); + await transport.send({ + sql: 'KILL QUERY WHERE query_id = ' + sqlString(queryId) + ' ASYNC', + defaultFormat: 'JSON', + authorization: lease.authorization, }); } catch { /* best-effort */ } } @@ -966,11 +991,12 @@ export interface ExportQueryOptions { */ export async function exportQuery(ctx: ChCtx, sql: string, opts: ExportQueryOptions = {}): Promise { const { queryId, signal, format, params } = opts; - const url = chUrl(ctx.origin, { - format: format || 'TabSeparatedWithNames', + const resp = await authedFetch(ctx, { + sql, + defaultFormat: format || 'TabSeparatedWithNames', params: { ...(queryId ? { query_id: queryId } : {}), ...(params || {}) }, + signal, }); - const resp = await authedFetch(ctx, url, sql, signal); if (!resp.ok) throw new Error(parseExceptionText(await resp.text())); return resp; } @@ -1039,20 +1065,21 @@ export async function runQuery(ctx: ChCtx, sql: string, o: RunQueryOptions = {}) const cap: Record = (o.resultRowLimit ?? 0) > 0 ? { max_result_rows: o.resultRowLimit!, result_overflow_mode: 'break' } : {}; - const url = chUrl(ctx.origin, { - format: fmtParam, + const resp = await authedFetch(ctx, { + sql, + defaultFormat: fmtParam, // wait_end_of_query buffers the whole response server-side so the HTTP // status reflects errors — but it defeats progressive streaming (first rows // wait for the query to finish: ~16s vs ~0.5s on a 1.3M-row scan). Keep it // only for raw modes (read whole anyway); the streaming Table path drops it // and surfaces mid-stream errors via the in-band `exception` line instead. - extra: { ...(isStreaming ? {} : { wait_end_of_query: 1 }), ...cap, add_http_cors_header: 1 }, + settings: { ...(isStreaming ? {} : { wait_end_of_query: 1 }), ...cap, add_http_cors_header: 1 }, // Tagging the request with a query_id lets Cancel issue KILL QUERY for it. // Caller-supplied params (o.params) ride alongside — e.g. multiquery SELECTs // add max_result_rows / result_overflow_mode to cap the result server-side. params: { ...(o.queryId ? { query_id: o.queryId } : {}), ...(o.params || {}) }, + signal: o.signal, }); - const resp = await authedFetch(ctx, url, sql, o.signal); if (!resp.ok) { return { error: parseExceptionText(await resp.text()) }; @@ -1060,33 +1087,6 @@ export async function runQuery(ctx: ChCtx, sql: string, o: RunQueryOptions = {}) if (!isStreaming) { return { raw: await resp.text() }; } - const reader = resp.body!.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split('\n'); - buffer = lines[lines.length - 1]; - for (const line of lines.slice(0, -1)) { - if (!line) continue; - let json: StreamLine; - try { - json = JSON.parse(line); - } catch { - continue; - } - o.onLine && o.onLine(json); - } - o.onChunk && o.onChunk(); - } - if (buffer.trim()) { - try { - o.onLine && o.onLine(JSON.parse(buffer)); - } catch { - /* trailing partial line */ - } - } + await transportFor(ctx).streamLines(resp.body!, { onLine: o.onLine, onChunk: o.onChunk }); return { streamed: true }; } diff --git a/tests/unit/ch-client.test.ts b/tests/unit/ch-client.test.ts index 721551f3..6bb6824b 100644 --- a/tests/unit/ch-client.test.ts +++ b/tests/unit/ch-client.test.ts @@ -113,14 +113,14 @@ describe('chUrl', () => { describe('authedFetch', () => { it('throws + signals out when no token', async () => { const ctx = ctxWith(() => jsonResp({}), { getToken: async () => null }); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toThrow('not signed in'); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toThrow('not signed in'); expect(ctx.onSignedOut).toHaveBeenCalled(); }); it('cancels without signaling auth loss when a missing-token request becomes stale', async () => { let epoch = 1; const token = deferred(); const ctx = ctxWith(() => jsonResp({}), { currentEpoch: () => epoch, getToken: () => token.promise }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); epoch = 2; token.resolve(null); await expect(pending).rejects.toMatchObject({ name: 'AbortError' }); @@ -137,7 +137,7 @@ describe('authedFetch', () => { onTransportConnected, onTransportOffline, }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); epoch = 2; token.resolve('replacement-token'); await expect(pending).rejects.toMatchObject({ name: 'AbortError' }); @@ -155,27 +155,27 @@ describe('authedFetch', () => { return 'Bearer replacement-' + token; }, }); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toMatchObject({ name: 'AbortError' }); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toMatchObject({ name: 'AbortError' }); expect(ctx.fetchMock).not.toHaveBeenCalled(); expect(ctx.onSignedOut).not.toHaveBeenCalled(); }); it('returns the response on success', async () => { const ctx = ctxWith(async () => jsonResp({ ok: 1 })); - const r = await authedFetch(ctx, 'u', 'sql'); + const r = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(r.ok).toBe(true); expect(ctx.fetchMock.mock.calls[0][1].headers.Authorization).toBe('Bearer tok'); }); it('reports a successful current transport connection when lifecycle hooks are supplied', async () => { const onTransportConnected = vi.fn(); const ctx = ctxWith(async () => jsonResp({ ok: 1 }), { currentEpoch: () => 4, onTransportConnected }); - await authedFetch(ctx, 'u', 'sql'); + await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(onTransportConnected).toHaveBeenCalledTimes(1); }); it('reports a non-abort fetch rejection as transport-offline', async () => { const failure = new Error('network unavailable'); const onTransportOffline = vi.fn(); const ctx = ctxWith(async () => { throw failure; }, { currentEpoch: () => 4, onTransportOffline }); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toBe(failure); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toBe(failure); expect(onTransportOffline).toHaveBeenCalledWith(failure); }); it('does not report a stale fetch rejection as transport-offline', async () => { @@ -188,7 +188,7 @@ describe('authedFetch', () => { fetchStarted.resolve(); return rejectedFetch.promise; }, { currentEpoch: () => epoch, onTransportOffline }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await fetchStarted.promise; epoch = 2; rejectedFetch.reject(failure); @@ -198,18 +198,18 @@ describe('authedFetch', () => { it('does not report HTTP failures or caller cancellation as transport-offline', async () => { const onTransportOffline = vi.fn(); const httpCtx = ctxWith(async () => textResp('server error', false, 500), { onTransportOffline }); - await authedFetch(httpCtx, 'u', 'sql'); + await authedFetch(httpCtx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); const controller = new AbortController(); controller.abort(); const abortCtx = ctxWith(async () => { throw new Error('cancelled request'); }, { onTransportOffline }); - await expect(authedFetch(abortCtx, 'u', 'sql', controller.signal)).rejects.toThrow('cancelled request'); + await expect(authedFetch(abortCtx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress', signal: controller.signal })).rejects.toThrow('cancelled request'); expect(onTransportOffline).not.toHaveBeenCalled(); }); it('does not report an AbortError rejection as transport-offline', async () => { const onTransportOffline = vi.fn(); const abortError = Object.assign(new Error('cancelled request'), { name: 'AbortError' }); const ctx = ctxWith(async () => { throw abortError; }, { onTransportOffline }); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toBe(abortError); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toBe(abortError); expect(onTransportOffline).not.toHaveBeenCalled(); }); it('refreshes once on 401 then retries', async () => { @@ -218,7 +218,7 @@ describe('authedFetch', () => { refresh: vi.fn(async () => true), getToken: vi.fn(async () => (n === 0 ? 'old' : 'new')), }); - const r = await authedFetch(ctx, 'u', 'sql'); + const r = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(r.ok).toBe(true); expect(ctx.refresh).toHaveBeenCalledTimes(1); }); @@ -227,7 +227,7 @@ describe('authedFetch', () => { async () => textResp('Code: 516. DB::Exception: Authentication failed', false, 403), { refresh: async () => false }, ); - await expect(authedFetch(ctx, 'u', 'sql')).rejects.toThrow('signed out'); + await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toThrow('signed out'); expect(ctx.onSignedOut).toHaveBeenCalledTimes(1); // Not overridden by `over` in this test, so it's still ctxWith's default // `vi.fn()` — the cast just recovers `.mock` from the (over-widened, since @@ -239,7 +239,7 @@ describe('authedFetch', () => { }); it('marks the ctx authenticated on a successful response', async () => { const ctx = ctxWith(async () => jsonResp({ ok: 1 })); - await authedFetch(ctx, 'u', 'sql'); + await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(ctx.authConfirmed).toBe(true); }); it('fences a stale successful response from lifecycle and authentication state', async () => { @@ -251,7 +251,7 @@ describe('authedFetch', () => { fetchStarted.resolve(); return response.promise; }, { currentEpoch: () => epoch, onTransportConnected }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await fetchStarted.promise; epoch = 2; response.resolve(jsonResp({ ok: 1 })); @@ -272,7 +272,7 @@ describe('authedFetch', () => { refresh: vi.fn(async () => true), onTransportOffline, }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await fetchStarted.promise; epoch = 2; response.resolve(jsonResp({}, false, 401)); @@ -298,7 +298,7 @@ describe('authedFetch', () => { currentEpoch: () => epoch, refresh: vi.fn(async () => true), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await bodyStarted.promise; epoch = 2; body.resolve('jwt::token_verification_exception'); @@ -318,7 +318,7 @@ describe('authedFetch', () => { return refresh.promise; }), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await refreshStarted.promise; expect(ctx.refresh).toHaveBeenCalledTimes(1); epoch = 2; @@ -342,7 +342,7 @@ describe('authedFetch', () => { return freshToken.promise; }), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await freshTokenStarted.promise; epoch = 2; freshToken.resolve('new'); @@ -361,7 +361,7 @@ describe('authedFetch', () => { return refresh.promise; }), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await refreshStarted.promise; epoch = 2; refresh.resolve(false); @@ -385,7 +385,7 @@ describe('authedFetch', () => { currentEpoch: () => epoch, refresh: vi.fn(async () => false), }); - const pending = authedFetch(ctx, 'u', 'sql'); + const pending = authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); await textStarted.promise; epoch = 2; denialText.resolve('Code: 516. Authentication failed'); @@ -396,7 +396,7 @@ describe('authedFetch', () => { // e.g. SHOW CREATE USER → HTTP 403 / UNKNOWN_USER, mid-session. const ctx = ctxWith(async () => textResp('Code: 192. DB::Exception: There is no user x', false, 403), { authConfirmed: true }); - const resp = await authedFetch(ctx, 'u', 'sql'); + const resp = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(resp.status).toBe(403); expect(ctx.onSignedOut).not.toHaveBeenCalled(); expect(ctx.refresh).not.toHaveBeenCalled(); @@ -407,19 +407,19 @@ describe('authedFetch', () => { async () => (n++ === 0 ? textResp('jwt::token_verification_exception', false, 500) : jsonResp({ ok: 1 })), { refresh: vi.fn(async () => true) }, ); - const r = await authedFetch(ctx, 'u', 'sql'); + const r = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(r.ok).toBe(true); }); it('returns a non-auth error response unchanged', async () => { const ctx = ctxWith(async () => textResp('syntax error', false, 400)); - const r = await authedFetch(ctx, 'u', 'sql'); + const r = await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(r.status).toBe(400); }); it('uses a provided authHeader (e.g. Basic) instead of Bearer', async () => { const ctx = ctxWith(async () => jsonResp({ ok: 1 }), { authHeader: (t) => 'Basic ' + t.toUpperCase(), }); - await authedFetch(ctx, 'u', 'sql'); + await authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' }); expect(ctx.fetchMock.mock.calls[0][1].headers.Authorization).toBe('Basic TOK'); }); }); @@ -442,6 +442,76 @@ describe('queryJson', () => { }); }); +// Issue #585 Phase 1 — `authedFetch`'s centralized entry snapshot (pass-5 +// revision): `extra`/`params` are captured synchronously the instant +// `authedFetch` is entered, before its first await (`ctx.getToken()`), +// preserving today's pre-await `chUrl` serialization timing under ONE +// mechanism instead of per-call-site defensive spreads. `queryJson` is the +// caller that historically forwarded these by reference +// (`ch-client.ts:218-228` pre-refactor), so these cases exercise it directly. +describe('queryJson — invocation-time settings/params capture (authedFetch\'s entry snapshot)', () => { + it('captures extra/params synchronously at entry — mutating the caller\'s live objects while token acquisition is pending does not reach the initial request', async () => { + const token = deferred(); + const extra: Record = { readonly: 2 }; + const params: Record = { param_id: '1' }; + const ctx = ctxWith(async () => jsonResp({ data: [] }), { getToken: () => token.promise }); + const pending = queryJson(ctx, 'SELECT 1', undefined, extra, params); + // Mutate the caller's own objects while the token await is still pending. + extra.readonly = 999; + params.param_id = 'mutated'; + token.resolve('tok'); + await pending; + const url = ctx.fetchMock.mock.calls[0][0]; + expect(url).toContain('readonly=2'); + expect(url).toContain('param_id=1'); + expect(url).not.toContain('readonly=999'); + expect(url).not.toContain('mutated'); + }); + + it('keeps the invocation-time snapshot across the one-refresh retry — mutating the caller\'s objects during the pending refresh does not reach the retried request', async () => { + let n = 0; + const extra: Record = { readonly: 2 }; + const params: Record = { param_id: '1' }; + const refreshStarted = deferred(); + const refresh = deferred(); + const ctx = ctxWith(async () => (n++ === 0 ? jsonResp({}, false, 401) : jsonResp({ data: [] })), { + refresh: vi.fn(async () => { + refreshStarted.resolve(); + return refresh.promise; + }), + }); + const pending = queryJson(ctx, 'SELECT 1', undefined, extra, params); + await refreshStarted.promise; + // Mutate while the refresh (and the post-refresh getToken it gates) is pending. + extra.readonly = 999; + params.param_id = 'mutated'; + refresh.resolve(true); + await pending; + expect(ctx.fetchMock).toHaveBeenCalledTimes(2); + const secondUrl = ctx.fetchMock.mock.calls[1][0]; + expect(secondUrl).toContain('readonly=2'); + expect(secondUrl).toContain('param_id=1'); + expect(secondUrl).not.toContain('readonly=999'); + expect(secondUrl).not.toContain('mutated'); + }); +}); + +// Issue #585 Phase 1, Adaptation A5 — `TransportDeps.origin()` is an +// accessor read live per request, never snapshotted; the live, mutable +// `ctx.origin` (mutated in place on sign-in — `connection-session.ts`) stays +// authoritative through `ch-client.ts`'s own `transportFor` wiring, not just +// in `createHttpTransport`'s isolated unit tests. +describe('transportFor — live origin authority through production wiring (Adaptation A5)', () => { + it('reads ctx.origin live per request; mutating it between two calls changes the next request\'s origin, never a stale snapshot', async () => { + const ctx = ctxWith(async () => jsonResp({ data: [] })); + await queryJson(ctx, 'SELECT 1'); + ctx.origin = 'https://new-cluster.example'; + await queryJson(ctx, 'SELECT 1'); + expect(ctx.fetchMock.mock.calls[0][0]).toContain('https://ch.example'); + expect(ctx.fetchMock.mock.calls[1][0]).toContain('https://new-cluster.example'); + }); +}); + describe('loadServerVersion', () => { it('returns the version string', async () => { const ctx = ctxWith(async () => jsonResp({ data: [{ v: '26.3.1', u: 1 }] })); @@ -1001,6 +1071,20 @@ describe('killQueryWithLease', () => { await killQueryWithLease(frozen, null, sqlString); expect(fetchMock).not.toHaveBeenCalled(); }); + + it('never reads getToken/refresh even if the lease object happens to carry them (hard invariant 8, defense in depth)', async () => { + const { lease: frozen } = leaseFetch(async () => jsonResp({ data: [] })); + const getToken = vi.fn(); + const refresh = vi.fn(); + // `AuthenticatedCancellationLease` has exactly 4 fields; this widens the + // VALUE (not killQueryWithLease's declared parameter type) to prove the + // one-shot lease-scoped transport genuinely never calls these, rather + // than merely never being GIVEN them. + const leaseWithExtras = Object.freeze({ ...frozen, getToken, refresh }); + await killQueryWithLease(leaseWithExtras, 'scope-q', sqlString); + expect(getToken).not.toHaveBeenCalled(); + expect(refresh).not.toHaveBeenCalled(); + }); }); describe('exportQuery', () => { From c34e00bf6bd87d14cf3f66c92f1449dc353bdffe Mon Sep 17 00:00:00 2001 From: Boris Tyshkevich Date: Thu, 6 Aug 2026 12:38:55 +0200 Subject: [PATCH 3/5] test(#585 phase 1): add architecture boundary checks for the transport seam build/check-boundaries.mjs: bans any src/** file from importing the bare @clickhouse/client-web specifier, with a single allowlist entry for the future official transport file (src/net/clickhouse-web-transport.ts, which does not exist yet -- ADR-0005 is Rejected and Phases 2-4 do not proceed without a new decision). Adds twin RULES entries forbidding both clickhouse-http-transport.ts and the type-only clickhouse-transport.types.ts from relatively importing ch-client.ts/oauth.ts/oauth-config.ts/ src/application//src/ui, closing the gap where a per-dir rule naming only the implementation file would leave the sibling contract file unguarded (check-boundaries.mjs matches import type too). tests/unit/client-web-spike-policy.test.js: mirrors the bare-specifier ban as a coverage-gated unit test, since this environment's ignore-scripts=true means a bare `npm test` never runs check:arch (CLAUDE.md hard rule 4). An independently implemented scanner, not a shared one -- the accepted two-scanner drift risk tests/unit/dashboard-boundaries.test.js already documents for its own mirror of the same script. Every forbidden category (ch-client.ts, oauth.ts, oauth-config.ts, src/application, src/ui) was individually sabotaged against both the implementation and contract files during implementation and confirmed to fail check:arch; a client-web import in ch-client.ts was confirmed to fail both check:arch and the mirrored unit test. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01LwFPT465eDJqYcRa8HGNLz --- build/check-boundaries.mjs | 42 +++++++++++++++ tests/unit/ch-client.test.ts | 29 ++++++++++- tests/unit/client-web-spike-policy.test.js | 59 +++++++++++++++++++++- 3 files changed, 128 insertions(+), 2 deletions(-) diff --git a/build/check-boundaries.mjs b/build/check-boundaries.mjs index 18df03df..fd048eb0 100644 --- a/build/check-boundaries.mjs +++ b/build/check-boundaries.mjs @@ -111,6 +111,26 @@ const RULES = [ except: ['src/ui/dnd-mime.js', 'src/ui/dnd-mime.ts', 'src/ui/dom.js', 'src/ui/dom.ts'], why: 'issue #60/#313: the editor layer is a leaf — UI actions are injected via app callbacks, not imported', }, + // Issue #585 Phase 1: the generic ClickHouse transport (and its type-only + // contract) is a low-level leaf that must not reach auth/application + // policy or UI, even type-only (the checker's own header comment: `import + // type` counts too). Two entries, not one, because collectFiles(ruleDir) + // matches per-`dir` — a single-file rule naming only the implementation + // file would leave the sibling contract file unguarded. `forbidden` targets + // are resolved repo-relative paths (never raw specifier strings like + // './ch-client.ts'), matching how the checker resolves and compares them. + { + dir: 'src/net/clickhouse-http-transport.ts', + forbidden: ['src/net/ch-client.ts', 'src/net/oauth.ts', + 'src/net/oauth-config.ts', 'src/application', 'src/ui'], + why: 'issue #585 Phase 1: the generic transport cannot reach auth/application policy or UI', + }, + { + dir: 'src/net/clickhouse-transport.types.ts', + forbidden: ['src/net/ch-client.ts', 'src/net/oauth.ts', + 'src/net/oauth-config.ts', 'src/application', 'src/ui'], + why: 'issue #585 Phase 1: the transport contract must not couple to auth/application policy or UI, even type-only', + }, ]; function collectFiles(target) { @@ -225,6 +245,28 @@ for (const file of collectFiles(path.join(repoRoot, 'src'))) { } } +// Issue #585 Phase 1: no file under src/** may import the official +// `@clickhouse/client-web` package (a bare specifier — the `RULES` loop above +// skips those, `if (!spec.startsWith('.')) continue;`, hence this separate +// block). ADR-0005 (docs/ADR-0005-clickhouse-web-client.md) is Rejected, so +// Phases 2-4 (the official-client cutover) do not proceed without a new +// decision — today this bans the import everywhere in `src/`. The single +// allowlist entry names the FUTURE official transport file (does not exist +// yet); the rule is written so it activates correctly the moment that file is +// born, rather than needing a second edit here. +const CLIENT_WEB_SPECIFIER = '@clickhouse/client-web'; +const CLIENT_WEB_ALLOWLIST = new Set(['src/net/clickhouse-web-transport.ts']); +for (const file of collectFiles(path.join(repoRoot, 'src'))) { + const relFile = path.relative(repoRoot, file).split(path.sep).join('/'); + checkedFiles += 1; + const source = fs.readFileSync(file, 'utf8'); + for (const spec of extractSpecifiers(source)) { + if (spec !== CLIENT_WEB_SPECIFIER && !spec.startsWith(`${CLIENT_WEB_SPECIFIER}/`)) continue; + if (CLIENT_WEB_ALLOWLIST.has(relFile)) continue; + violations.push(`${relFile} → ${spec} (issue #585 Phase 1: only the future official transport file may import @clickhouse/client-web — ADR-0005 is Rejected, Phases 2-4 do not proceed without a new decision)`); + } +} + if (violations.length) { console.error('check-boundaries: architecture violations:'); for (const line of violations) console.error(` ${line}`); diff --git a/tests/unit/ch-client.test.ts b/tests/unit/ch-client.test.ts index 6bb6824b..5362dd22 100644 --- a/tests/unit/ch-client.test.ts +++ b/tests/unit/ch-client.test.ts @@ -502,7 +502,7 @@ describe('queryJson — invocation-time settings/params capture (authedFetch\'s // authoritative through `ch-client.ts`'s own `transportFor` wiring, not just // in `createHttpTransport`'s isolated unit tests. describe('transportFor — live origin authority through production wiring (Adaptation A5)', () => { - it('reads ctx.origin live per request; mutating it between two calls changes the next request\'s origin, never a stale snapshot', async () => { + it('reads ctx.origin live per request across two independent authedFetch calls', async () => { const ctx = ctxWith(async () => jsonResp({ data: [] })); await queryJson(ctx, 'SELECT 1'); ctx.origin = 'https://new-cluster.example'; @@ -510,6 +510,33 @@ describe('transportFor — live origin authority through production wiring (Adap expect(ctx.fetchMock.mock.calls[0][0]).toContain('https://ch.example'); expect(ctx.fetchMock.mock.calls[1][0]).toContain('https://new-cluster.example'); }); + + // The case above reconstructs a transport per authedFetch call, so it can't + // by itself distinguish a live-read `deps.origin()` from one snapshotted at + // `createHttpTransport` construction time (sabotage case 2). This one uses + // the SAME transport instance across authedFetch's one-refresh retry loop + // (transportFor(ctx) is called once per authedFetch invocation, before the + // retry loop) to prove the origin read is live per SEND, not per + // construction. + it('reads ctx.origin live per send within one authedFetch retry cycle, not once at transport construction', async () => { + let n = 0; + const refreshStarted = deferred(); + const refresh = deferred(); + const ctx = ctxWith(async () => (n++ === 0 ? jsonResp({}, false, 401) : jsonResp({ data: [] })), { + refresh: vi.fn(async () => { + refreshStarted.resolve(); + return refresh.promise; + }), + }); + const pending = queryJson(ctx, 'SELECT 1'); + await refreshStarted.promise; + ctx.origin = 'https://new-cluster.example'; + refresh.resolve(true); + await pending; + expect(ctx.fetchMock).toHaveBeenCalledTimes(2); + expect(ctx.fetchMock.mock.calls[0][0]).toContain('https://ch.example'); + expect(ctx.fetchMock.mock.calls[1][0]).toContain('https://new-cluster.example'); + }); }); describe('loadServerVersion', () => { diff --git a/tests/unit/client-web-spike-policy.test.js b/tests/unit/client-web-spike-policy.test.js index f64fae7e..b612fc04 100644 --- a/tests/unit/client-web-spike-policy.test.js +++ b/tests/unit/client-web-spike-policy.test.js @@ -24,7 +24,7 @@ import { describe, expect, it } from 'vitest'; import { mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; -import { dirname, extname, join, relative, resolve } from 'node:path'; +import { dirname, extname, join, relative, resolve, sep } from 'node:path'; import { fileURLToPath } from 'node:url'; import { buildArtifact } from '../../build/build.mjs'; import { @@ -181,6 +181,63 @@ describe('no CDN or remote-URL import in spike sources', () => { }); }); +// Issue #585 Phase 1 — mirrors build/check-boundaries.mjs's bare-specifier +// ban on `@clickhouse/client-web` under `src/**`. This environment's npm +// config sets `ignore-scripts=true`, so a bare `npm test` never runs +// `pretest`/`check:arch` (CLAUDE.md hard rule 4) — mirroring the ban here +// means the coverage-gated suite enforces it too. This is an INDEPENDENTLY +// implemented scanner, not a shared one: `check-boundaries.mjs` exports +// nothing today, so unifying the two behind one exported scanner would be a +// real improvement but is a cross-cutting tooling change out of scope for +// this pure-refactor PR (tracked separately, `inbox`-labeled, per CLAUDE.md +// "Working discipline") — the two scanners can drift, the same accepted risk +// tests/unit/dashboard-boundaries.test.js already documents for its own +// mirror of the same script. Belt and braces: metafile (build-graph truth, +// tested above), source scan (this — fast, IDE-time), `check:arch` (gate/CI). +const SRC_DIR = resolve(projectRoot, 'src'); +const CLIENT_WEB_IMPORT_PATTERNS = [ + /\bimport\s+[\w*{}\s,]+\s+from\s*['"](@clickhouse\/client-web(?:\/[^'"]*)?)['"]/g, + /\bexport\s+[\w*{}\s,]+\s+from\s*['"](@clickhouse\/client-web(?:\/[^'"]*)?)['"]/g, + /\bimport\s*['"](@clickhouse\/client-web(?:\/[^'"]*)?)['"]/g, + /\bimport\s*\(\s*['"](@clickhouse\/client-web(?:\/[^'"]*)?)['"]/g, +]; +// Names the FUTURE official transport file (does not exist yet) — activates +// correctly the moment it is born, same as build/check-boundaries.mjs's own +// allowlist. +const CLIENT_WEB_ALLOWLIST = new Set(['src/net/clickhouse-web-transport.ts']); + +async function collectSrcSourceFiles(dir) { + const entries = await readdir(dir, { withFileTypes: true }); + const files = []; + for (const entry of entries) { + const full = join(dir, entry.name); + if (entry.isDirectory()) { + files.push(...await collectSrcSourceFiles(full)); + } else if (['.ts', '.tsx', '.js', '.mjs'].includes(extname(entry.name))) { + files.push(full); + } + } + return files; +} + +describe('no @clickhouse/client-web import anywhere under src/ (mirrors build/check-boundaries.mjs)', () => { + it('scans every src/** source file for a bare @clickhouse/client-web import, outside the not-yet-existing official-transport allowlist', async () => { + const files = await collectSrcSourceFiles(SRC_DIR); + expect(files.length).toBeGreaterThan(0); + const offenders = []; + for (const file of files) { + const relFile = relative(projectRoot, file).split(sep).join('/'); + if (CLIENT_WEB_ALLOWLIST.has(relFile)) continue; + const text = await readFile(file, 'utf8'); + for (const pattern of CLIENT_WEB_IMPORT_PATTERNS) { + pattern.lastIndex = 0; + if (pattern.test(text)) offenders.push(relFile); + } + } + expect(offenders).toEqual([]); + }); +}); + // Plan §31 "Wiki reconciliation": "The evidence validator and normal unit // test must reject a mismatch among: ADR Status; machine-readable decision; // wiki status text; wiki ADR link." This sub-task only owns the normal- From 1621166e6f440c78a64ecba34edf796b7abc3ca1 Mon Sep 17 00:00:00 2001 From: Boris Tyshkevich Date: Thu, 6 Aug 2026 12:40:33 +0200 Subject: [PATCH 4/5] docs(#585 phase 1): reconcile CHANGELOG, ADR-0005, ARCHITECTURE, and wiki Records Phase 1 (transport seam, no behavior change) as landed: CHANGELOG.md [Unreleased], an ADR-0005 addendum (Status stays Rejected -- this only adds a "Phase 1 addendum (landed)" subsection, matching the ADR's own prescription for a Rejected outcome), docs/ARCHITECTURE.md's query- execution section, and .wiki/Architecture.md + .wiki/Source-Map.md + .wiki/Decisions-and-Roadmap.md. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01LwFPT465eDJqYcRa8HGNLz --- .wiki/Architecture.md | 6 +++++- .wiki/Decisions-and-Roadmap.md | 10 +++++++--- .wiki/Source-Map.md | 4 +++- CHANGELOG.md | 23 +++++++++++++++++++++++ docs/ADR-0005-clickhouse-web-client.md | 24 ++++++++++++++++++++++++ docs/ARCHITECTURE.md | 22 ++++++++++++++++++++++ 6 files changed, 84 insertions(+), 5 deletions(-) diff --git a/.wiki/Architecture.md b/.wiki/Architecture.md index e3765733..a911dffd 100644 --- a/.wiki/Architecture.md +++ b/.wiki/Architecture.md @@ -51,7 +51,11 @@ module mocking. ## Query path 1. The editor/controller prepares SQL and typed parameters. -2. `src/net/ch-client.js` sends the HTTP request with injected auth/fetch context. +2. `src/net/ch-client.js` sends the HTTP request with injected auth/fetch context, + delegating generic request construction and stream mechanics through a narrow + transport contract (`src/net/clickhouse-transport.types.js` + + `src/net/clickhouse-http-transport.js`, #585 Phase 1) — auth/epoch/retry policy + stays in `ch-client.js`. 3. `JSONStringsEachRowWithProgress` is folded line by line by pure stream logic. 4. Results resolve through the panel registry to table, chart, logs, KPI, filter, text, or graph-oriented renderers. diff --git a/.wiki/Decisions-and-Roadmap.md b/.wiki/Decisions-and-Roadmap.md index cbfa9263..ced6db90 100644 --- a/.wiki/Decisions-and-Roadmap.md +++ b/.wiki/Decisions-and-Roadmap.md @@ -130,9 +130,13 @@ Two roadmap tracks are current: 0 completed as a fully valid outcome — **no production cutover occurred**, and `src/net/ch-client.ts` (the current custom transport) **remains authoritative**. Phase 1 (separating application policy from the concrete - transport implementation) remains useful and unblocked on its own merits; - Phases 2–4 (production adoption, cutover, and custom-transport deletion) do - not proceed without a new decision. + transport implementation) — **landed** on `wip/585-phase1-transport-seam`: + `src/net/clickhouse-transport.types.ts` (the `ClickHouseTransport` contract) + + `src/net/clickhouse-http-transport.ts` (`createHttpTransport`, the current + implementation behind it) put the existing transport behind a narrow seam + with zero behavior change; `ch-client.ts` keeps every auth/epoch/retry + policy and product operation. Phases 2–4 (production adoption, cutover, and + custom-transport deletion) do not proceed without a new decision. Re-read GitHub before acting because issue state can change; a MERGED PR is not proof its code is on `main` (see the reset above). diff --git a/.wiki/Source-Map.md b/.wiki/Source-Map.md index 1c231812..23845600 100644 --- a/.wiki/Source-Map.md +++ b/.wiki/Source-Map.md @@ -16,7 +16,9 @@ Back to [[Home]]. Related: [[Architecture]], [[Product-and-Features]]. | `src/dashboard/application/dashboard-repaint-plan.js` | pure repaint-decision arbitration extracted from `ui/dashboard.js`'s `renderDashboard` effect (#589) | | `src/ui/dashboard-tile-gestures.js` | Dashboard corner-drag resize, Command/Ctrl-drag reorder, and modifier-cue controller, extracted from `ui/dashboard.js` behind an injected `TileGestureDeps` seam (#589) | | `src/state.js` | signals-backed state model and persistence operations | -| `src/net/ch-client.js` | ClickHouse HTTP execution and schema calls | +| `src/net/ch-client.js` | ClickHouse HTTP execution and schema calls; auth/epoch/retry policy, product operations, `ChCtx` (#585 Phase 1: generic request/stream mechanics delegate through the transport seam below) | +| `src/net/clickhouse-transport.types.js` | Type-only `ClickHouseTransport` contract (`send`/`streamLines`, `TransportDeps`, `TransportRequest`) (#585 Phase 1) | +| `src/net/clickhouse-http-transport.js` | `createHttpTransport` — the current custom HTTP implementation of that contract, plus `chUrl`/`ChUrlOpts` (#585 Phase 1) | | `src/net/oauth.js` | OAuth flow/token exchange | | `src/editor/editor-port.js` | SQL editor contract and safe no-op port | | `src/editor/codemirror-adapter.js` | SQL CodeMirror 6 adapter | diff --git a/CHANGELOG.md b/CHANGELOG.md index ae7b31bd..b56afa6d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,29 @@ auto-generated per-PR notes; this file is the curated, human-readable history. consistent comment/blank-stripped metric). Production ClickHouse transport behavior is unchanged — `src/net/ch-client.ts` remains authoritative, and no production cutover occurred. +- **#585 Phase 1: internal transport seam, no behavior change.** Defined a + narrow SQL Browser ClickHouse-transport contract + (`src/net/clickhouse-transport.types.ts`: `ClickHouseTransport`, + `TransportDeps`, `TransportRequest`, `StreamCallbacks`) and put the current + custom HTTP implementation behind it + (`src/net/clickhouse-http-transport.ts`: `createHttpTransport`, plus + `chUrl`/`ChUrlOpts` moved verbatim from `ch-client.ts` and re-exported + unchanged). `ch-client.ts`'s auth/epoch/retry policy, product operations, + and `ChCtx` are otherwise untouched — no new `ChCtx` field, no runtime + transport switch. `authedFetch` is the one exported-signature change (its + only importers are `ch-client.ts` internals and its own unit test): it now + takes a structured request instead of a prebuilt URL string, and snapshots + the caller's `settings`/`params` synchronously at entry (before its first + await) as a single centralized aliasing defense. Added a reusable + contract-test-suite factory (`tests/unit/clickhouse-transport-contract.ts`), + registered against the current implementation only — Phases 2–4 (an + official-client cutover) remain out of scope and do not proceed without a + new decision, per ADR-0005 (Rejected). `build/check-boundaries.mjs` gains a + bare-specifier ban on `@clickhouse/client-web` anywhere under `src/**` + (mirrored as a coverage-gated unit test) and twin rules keeping the + transport implementation and its type-only contract from reaching + auth/application policy or UI, even type-only. No user-visible or + production-behavior change; bundle size delta ≈ 0 (pure code movement). - **Opt-in ChatGPT-authored `/ship` planning.** `/ship --planner chatgpt` keeps the existing Fable-authored workflow as the default, but lets ChatGPT own complete plan drafts and revisions while Fable/high performs repository-grounded diff --git a/docs/ADR-0005-clickhouse-web-client.md b/docs/ADR-0005-clickhouse-web-client.md index f7ea23cf..1e4760f9 100644 --- a/docs/ADR-0005-clickhouse-web-client.md +++ b/docs/ADR-0005-clickhouse-web-client.md @@ -630,6 +630,30 @@ teardown; permanent live-ClickHouse CI integration; general supported-browser documentation (tracked separately by #71); a permanent second transport path. +### Phase 1 addendum (landed) + +Phase 1 — "establish the transport seam without behavior change" — landed on +`wip/585-phase1-transport-seam`, per this ADR's own prescription for a +Rejected outcome ("retain the current transport with the layer separation +from Phase 1 remains useful independent of this decision"). It defines +`src/net/clickhouse-transport.types.ts` (the `ClickHouseTransport` contract: +`send`/`streamLines`, `TransportDeps`, `TransportRequest`, `StreamCallbacks`) +and `src/net/clickhouse-http-transport.ts` (`createHttpTransport` — the +current custom HTTP implementation behind that contract, plus `chUrl`/ +`ChUrlOpts` moved verbatim from `ch-client.ts`). `ch-client.ts`'s auth/epoch/ +retry policy, product operations, and `ChCtx` are otherwise unchanged — no +new `ChCtx` field, no runtime transport switch; `authedFetch` is the one +exported-signature change, scoped to its own unit test (no production +importer exists outside `ch-client.ts` itself). A reusable contract-test- +suite factory (`tests/unit/clickhouse-transport-contract.ts`) registers +against the current implementation only — Phases 2–4 remain exactly as +scoped above and do not proceed without a new decision. `build/ +check-boundaries.mjs` gained a bare-specifier ban on `@clickhouse/client-web` +under `src/**` (mirrored as a coverage-gated unit test), naming +`src/net/clickhouse-web-transport.ts` as the single allowlisted (not yet +existing) future official-transport file. No user-visible or production- +behavior change; bundle size delta ≈ 0. + ## Reproduction commands ```sh diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index d492572d..90a8a2a1 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -199,6 +199,28 @@ folded via the pure `applyStreamLine`; a single automatic token refresh on 401/403/`token_verification_exception` (before `authConfirmed` flips, an auth failure signs out; after, it is a query error). +### Transport seam (#585 Phase 1) + +Generic request construction and stream mechanics are split out behind a +narrow contract: `net/clickhouse-transport.types.ts` declares +`ClickHouseTransport` (`send`/`streamLines`), `TransportDeps` (`fetch`/ +`origin` accessors — read live per request, never snapshotted, since the +live `chCtx.origin` is mutated in place on sign-in), and `TransportRequest`. +`net/clickhouse-http-transport.ts`'s `createHttpTransport` is the current +custom HTTP implementation of that contract — `chUrl`/`ChUrlOpts` live there +now, re-exported unchanged from `ch-client.ts`. `ch-client.ts` keeps every +auth/epoch/retry/lifecycle policy (`authedFetch`), product operation, and +`ChCtx` exactly as before; a module-private `transportFor(ctx)` delegates +unconditionally to `createHttpTransport` — `ChCtx` gained no field and there +is no runtime transport switch. `authedFetch` snapshots the caller's +`settings`/`params` synchronously at entry, before its first await, as one +centralized defense against a caller mutating those objects while a +token/refresh await is pending. A reusable contract-test-suite factory +(`tests/unit/clickhouse-transport-contract.ts`) registers against this one +implementation; a future official-client implementation (ADR-0005 is +Rejected; Phases 2–4 do not proceed without a new decision) would satisfy the +same contract and reuse the same suite. + ## Build `build/build.mjs` runs esbuild (bundle + minify, IIFE), inlines the script and From 7cfa1ed82bd50ed01a56bb14bbd714ed3d646ee8 Mon Sep 17 00:00:00 2001 From: Boris Tyshkevich Date: Thu, 6 Aug 2026 13:19:08 +0200 Subject: [PATCH 5/5] fix(#585): address review pass 2 findings authedFetch awaited ctx.getToken() before transport.send() built the request URL, so a chUrl/encodeURIComponent throw (e.g. a lone UTF-16 surrogate in settings/params) was misclassified as a transport failure: onTransportOffline fired and a token had already been read, even though fetch was never invoked. Pre-refactor, every caller built the URL before calling authedFetch, so the same throw never reached token acquisition or the offline-classifying catch. Restore that ordering: authedFetch now calls chUrl once, eagerly and discarded, before its first await, reproducing the exact original failure shape (no token read, no fetch, no onTransportOffline call). transport.send still builds the real URL again at actual send time against the live ctx.origin, unaffected since origin is concatenated, never encodeURIComponent-encoded. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01LwFPT465eDJqYcRa8HGNLz --- src/net/ch-client.ts | 15 +++++++++++++++ tests/unit/ch-client.test.ts | 12 ++++++++++++ 2 files changed, 27 insertions(+) diff --git a/src/net/ch-client.ts b/src/net/ch-client.ts index 577d65d3..61f27e97 100644 --- a/src/net/ch-client.ts +++ b/src/net/ch-client.ts @@ -149,6 +149,21 @@ export async function authedFetch(ctx: ChCtx, request: Omit { await expect(authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress' })).rejects.toBe(abortError); expect(onTransportOffline).not.toHaveBeenCalled(); }); + it('rejects a malformed URL param synchronously without a token read, a fetch, or an offline signal (review pass 2, P1)', async () => { + const onTransportOffline = vi.fn(); + const ctx = ctxWith(async () => jsonResp({ ok: 1 }), { onTransportOffline }); + // A lone UTF-16 surrogate makes `chUrl`'s `encodeURIComponent` throw a + // URIError — request-preparation failure, not a network failure. + await expect( + authedFetch(ctx, { sql: 'sql', defaultFormat: 'JSONStringsEachRowWithProgress', params: { x: '\ud800' } }), + ).rejects.toBeInstanceOf(URIError); + expect(ctx.getToken).not.toHaveBeenCalled(); + expect(ctx.fetchMock).not.toHaveBeenCalled(); + expect(onTransportOffline).not.toHaveBeenCalled(); + }); it('refreshes once on 401 then retries', async () => { let n = 0; const ctx = ctxWith(async () => (n++ === 0 ? jsonResp({}, false, 401) : jsonResp({ ok: 1 })), {