From 05d2db1f93e69813c3de1430e8ebf64dfb3abec8 Mon Sep 17 00:00:00 2001 From: Shawn Chen Date: Wed, 12 Aug 2026 21:14:53 +1200 Subject: [PATCH] Stop teardown landing on a stream fetch that is still running MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The v3 suite fails intermittently on macOS with "a session (path=…) is active; close it before using standalone query()" — a registry-state error in a file that never opened that session. Three of four CI runs on chdb-io/chdb-node#76 hit it, always on a macOS shard, never on Linux. index.js states the invariant that explains it: because the C ABI has no interrupt, a native op runs on its libuv thread to completion no matter when the JS promise settles, so every native op registers in pendingNativeOps and nothing destroys a connection while one is outstanding. Releasing a connection mid-op aborts the shared in-process engine for the rest of the process, and on some platforms leaves the worker blocked inside libchdb so its promise never settles. Streaming was the one path that never registered. queryAsync and Session.queryAsync go through withAbortTimeout, insert and raw insert through runInsert and wrapRawNative, all of which track; StreamFetch tracked nothing. So _drainPendingOps() reported quiet while a fetch was mid-flight and the suite's global afterEach closed the connection under it. Two more holes on the same principle came out of reading around it. cancel() destroys the stream handle synchronously via StreamCancel, and the fetch worker reads that handle with no lock — its `if (!st_ || st_->finished || !st_->handle)` guard can be invalidated between the check and chdb_stream_fetch_result. And close() ignored _activeStream entirely, so the handle outlived the connection it points into and the iterator's own finally would later cancel against a connection that no longer existed. All three now go through the one mechanism: the fetch is tracked, cancel() defers the destroy behind an in-flight fetch, and close() cancels the active stream before releasing the connection. The last two compose without extra ordering logic — cancel() registers its deferred destroy in pendingNativeOps, which close()'s own deferral already waits for, so the stream handle is always destroyed before the connection. The two new tests pick a query that needs a real per-row computation. The obvious choice, count() over numbers(), returns in ~15ms, which is short enough that both tests pass whether or not the races are handled — so they assert the precondition (a fetch is in flight when teardown lands) rather than trusting it, and a future change that makes the fetch fast fails the test instead of silently losing the coverage. Verified by running the full v3 suite with and without the index.js change: the same nine failures either way (Layer 3 arrow-input and parametrized streaming, which need a different engine build than this machine has), plus the two new tests failing only without it. Also corrects a comment in async-stress.test.ts claiming plain queryAsync is not tracked for drain. withAbortTimeout has tracked every async query since chdb-io/chdb-node#53; the stale note sent this investigation down the wrong path for a while. Co-Authored-By: Claude Opus 5 (1M context) --- index.js | 40 +++++++++++++- test/v3/async-stress.test.ts | 10 ++-- test/v3/stream-cancel-race.test.ts | 89 ++++++++++++++++++++++++++++++ 3 files changed, 133 insertions(+), 6 deletions(-) create mode 100644 test/v3/stream-cancel-race.test.ts diff --git a/index.js b/index.js index ec00ed5..71eb794 100644 --- a/index.js +++ b/index.js @@ -345,6 +345,7 @@ class ChdbQueryStream { this._format = format; this._signal = signal; this._closed = false; + this._inflight = null; // the fetch currently on a libuv thread, if any } get closed() { return this._closed; } @@ -352,15 +353,26 @@ class ChdbQueryStream { async *[Symbol.asyncIterator]() { try { while (true) { + // cancel() may have landed while the previous fetch was in flight. Issuing + // another one would race the destroy it deferred onto that fetch. + if (this._closed) break; if (this._signal && this._signal.aborted) { throw new ChdbAbortError('Stream aborted'); } let raw; try { - raw = await chdbNode.StreamFetch(this._handle); + // Tracked like every other native op (see withAbortTimeout): a fetch + // runs on a libuv thread against this stream's connection, so a + // Session.close() or a teardown that does not wait for it destroys the + // connection mid-op and aborts the engine for the whole process. + // Streaming was the one path that skipped tracking. + this._inflight = trackNative(chdbNode.StreamFetch(this._handle)); + raw = await this._inflight; } catch (e) { this._closed = true; throw new ChdbStreamError(asQueryError(e).message, { cause: e }); + } finally { + this._inflight = null; } if (raw.done) { this._closed = true; break; } yield new StreamChunk(raw.bytes, raw.numRows, this._format); @@ -386,7 +398,20 @@ class ChdbQueryStream { cancel() { if (this._closed) return; this._closed = true; - try { chdbNode.StreamCancel(this._handle); } catch (_) { /* best effort */ } + const destroy = () => { + try { chdbNode.StreamCancel(this._handle); } catch (_) { /* best effort */ } + }; + // StreamCancel destroys the stream handle synchronously, and the fetch worker + // reads that handle without holding a lock. Destroying it while a fetch is + // executing pulls it out from under the worker. Wait for the fetch, and + // register the wait so a Session.close() defers behind the destroy too. + if (this._inflight) { + const deferred = this._inflight.then(destroy, destroy); + pendingNativeOps.add(deferred); + deferred.finally(() => pendingNativeOps.delete(deferred)); + } else { + destroy(); + } } } @@ -656,6 +681,17 @@ class Session { for (const sig of SESSION_SIGNALS) process.removeListener(sig, this.#signalHandler); this.#signalHandler = null; } + // Release a stream still open on this connection before the connection goes. + // Its handle points into the connection, so the stream's own finally would + // later call StreamCancel against a connection that no longer exists. Doing + // it here also orders the two teardowns correctly: a cancel that has to wait + // for an in-flight fetch registers that wait in pendingNativeOps, which the + // deferral below then waits for, so the stream handle is always destroyed + // before the connection it lives in. + if (this._activeStream && !this._activeStream.closed) { + try { this._activeStream.cancel(); } catch (_) { /* best effort */ } + } + this._activeStream = null; const conn = this.connection; this.connection = null; const teardown = () => { diff --git a/test/v3/async-stress.test.ts b/test/v3/async-stress.test.ts index c4fd8fa..bdd5af5 100644 --- a/test/v3/async-stress.test.ts +++ b/test/v3/async-stress.test.ts @@ -11,10 +11,12 @@ const HEAVY = (n: number) => `SELECT count() FROM numbers(${n})` // The storm/race cases below run dozens of heavy queries in sequence and take // ~15-30s on a fast runner. Give them a generous timeout so a slow/loaded CI -// runner never hits the 30s default and kills a test mid-flight — an abandoned -// in-flight native op (plain queryAsync is not tracked for drain) would collide -// with the next test on the single in-process engine and abort it (code 236), -// cascading failures into every later file. +// runner never hits the 30s default and kills a test mid-flight — a test killed +// at its timeout is torn down without the global afterEach draining its in-flight +// native op, which then collides with the next test on the single in-process +// engine and aborts it (code 236), cascading failures into every later file. +// (Plain queryAsync is tracked for drain; withAbortTimeout tracks every async +// query. This comment used to claim it was not.) const STRESS_TIMEOUT_MS = 120_000 describe('async concurrency — correctness & no deadlock', () => { diff --git a/test/v3/stream-cancel-race.test.ts b/test/v3/stream-cancel-race.test.ts new file mode 100644 index 0000000..ce1d9e9 --- /dev/null +++ b/test/v3/stream-cancel-race.test.ts @@ -0,0 +1,89 @@ +import { describe, it, expect } from 'vitest' +import { Session, queryAsync } from '../../index.js' +// @ts-expect-error — internal test helper, not in the type surface +import { _drainPendingOps } from '../../index.js' + +// Teardown landing on a stream fetch that is still running. +// +// A fetch runs on a libuv thread against the session's connection. StreamCancel +// destroys the stream handle synchronously and CloseConnection destroys the +// connection, and the fetch worker reads both without holding a lock, so either +// one landing mid-fetch pulls the memory out from under the worker. libchdb +// answers that by aborting the shared in-process engine, which fails every later +// query in the process, and on macOS can leave the worker blocked inside +// chdb_stream_fetch_result so its promise never settles. +// +// All of it was reachable because streaming was the one native async path that +// never registered itself in pendingNativeOps, so neither close() nor the +// suite-wide teardown drain knew a fetch was in flight. +// +// One row that needs a real per-row computation, so the first fetch takes about a +// second and teardown lands well inside it. count() over numbers() returns in +// ~15ms — fast enough that these would pass whether the races are handled or not. +const SLOW_ONE_ROW = 'SELECT max(sipHash64(number)) FROM numbers(200000000)' + +// The macOS symptom was a fetch whose promise never settled. Bound the wait so +// that failure reads as this assertion rather than as a suite timeout. +async function settlesWithin(p: Promise, ms: number): Promise<'settled' | 'hung'> { + let timer: NodeJS.Timeout + const hung = new Promise<'hung'>((r) => { timer = setTimeout(() => r('hung'), ms) }) + const settled = p.then(() => 'settled' as const, () => 'settled' as const) + try { + return await Promise.race([settled, hung]) + } finally { + clearTimeout(timer!) + } +} + +describe('teardown that lands on an in-flight stream fetch', () => { + it('cancel() waits for the fetch instead of destroying the handle under it', async () => { + const s = new Session() + try { + const stream = s.queryStream(SLOW_ONE_ROW, { format: 'JSONEachRow' }) + const it = stream[Symbol.asyncIterator]() + + // next() runs the generator body up to the await, so the fetch is dispatched + // before it returns and is still running on the next line. Asserted rather + // than assumed: if a change ever makes the fetch settle first, this test + // would quietly stop covering the race it exists for. + const first = it.next() + expect((stream as unknown as { _inflight: unknown })._inflight).toBeTruthy() + + stream.cancel() + expect(await settlesWithin(first, 20_000)).toBe('settled') + await _drainPendingOps() // let the deferred destroy land + + // The engine survived, and the cancelled stream was released cleanly enough + // that the session takes another one. + const rows: number[] = [] + for await (const row of s + .queryStream('SELECT number AS n FROM numbers(3)') + .rows<{ n: number }>()) { + rows.push(row.n) + } + expect(rows).toEqual([0, 1, 2]) + } finally { + s.close() + } + }, 60_000) + + it('close() releases the stream and defers the connection until the fetch drains', async () => { + const s = new Session() + const stream = s.queryStream(SLOW_ONE_ROW, { format: 'JSONEachRow' }) + const it = stream[Symbol.asyncIterator]() + + const first = it.next() + expect((stream as unknown as { _inflight: unknown })._inflight).toBeTruthy() + + s.close() // must cancel the stream and hold the connection until the fetch ends + expect(await settlesWithin(first, 20_000)).toBe('settled') + await _drainPendingOps() + + // Standalone query() refuses to run while any session connection is still + // registered ("a session (path=…) is active"), so this answering proves both + // that the engine is alive and that the deferred teardown released the + // connection instead of leaking it into the rest of the process. + const r = await queryAsync('SELECT 1', { format: 'CSV' }) + expect(r.text().trim()).toBe('1') + }, 60_000) +})