From 7e34e2b4f70e5d24e4f9b6b831abf669b2f9cd8b Mon Sep 17 00:00:00 2001 From: Shawn Chen Date: Thu, 13 Aug 2026 11:16:53 +1200 Subject: [PATCH 1/3] Refuse to open a session while the default connection is busy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit main went red right after #77 with the failure that PR was meant to end, so the stream teardown was not the cause. The full log said what the earlier ones had not: Error: Hook timed out in 30000ms. ChdbQueryError: chdb: a session (path=…) is active; … Error: Worker exited unexpectedly The hook is the suite-wide afterEach, whose first statement is `await _drainPendingOps()`. It sat for the full 30s, so `_closeAllSessions()` never ran, the session stayed open, and every later test in the file reported it as active. Six failures, one cause: a native op that never settled. Only one teardown can produce that and has no gate. `new Session()` calls CreateConnection, which reaches acquire_session_conn, which — because libchdb binds one data directory per process and the in-memory default holds the slot with an empty path — calls chdb_close_conn on the default connection outright. The whole pendingNativeOps discipline, close()'s deferral and #77's stream tracking included, protects session connections. Nothing protected this one, and the constructor never consulted the registry at all. The test that failed does exactly this race sixty times per run, and its name claimed the race was safe. The constructor is synchronous and cannot wait, so it refuses and names what to await. That trades a hang thirty seconds and one test file away for an error at the call site that caused it. Counted per connection, not globally: a session-bound query in flight is no reason to reject a second session on the same path, which is allowed. Only the three standalone entry points mark the default connection busy. The decrement is registered before any caller's handler, so code that awaits its own query sees zero on the next line instead of racing the bookkeeping. The test asserted something that was never exercised. HEAVY(60_000_000) is `count() FROM numbers(6e7)`, which the engine answers from the range without scanning — 5ms, against sleeps of 0-3ms. The query had usually finished before the session was opened, so most runs never entered the race at all, which is why a genuinely broken invariant looked intermittent. It now uses a query that forces per-row work (~120ms), asserts the refusal happened rather than tolerating either outcome, and checks the query is unaffected and a session opens the moment it drains. Full v3 suite before and after: the same nine failures either way (Layer 3 arrow-input and parametrized streaming, which want a different engine build than this machine has). Nothing else in the suite opens a session while a standalone query is outstanding. Co-Authored-By: Claude Opus 5 (1M context) --- index.js | 40 +++++++++++++++++++++++++++++++---- test/v3/async-stress.test.ts | 41 ++++++++++++++++++++++++++++++------ 2 files changed, 70 insertions(+), 11 deletions(-) diff --git a/index.js b/index.js index 71eb794..143b3a8 100644 --- a/index.js +++ b/index.js @@ -439,11 +439,28 @@ function queryBind(query, args = {}, format = "CSV") { } } +// Ops running on the shared default connection, counted separately from +// pendingNativeOps. Opening a Session destroys that connection — libchdb binds +// one data directory per process, so the in-memory default has to yield — and +// destroying it mid-op is what leaves a worker blocked inside libchdb with its +// promise never settling. new Session() consults this to refuse instead. +// +// A counter, and decremented from a handler attached before any caller's, so a +// caller that awaits its own query sees zero on the next line rather than +// racing the bookkeeping. +let pendingDefaultOps = 0; +const releaseDefaultOp = () => { pendingDefaultOps--; }; +function trackDefault(nativePromise) { + pendingDefaultOps++; + nativePromise.then(releaseDefaultOp, releaseDefaultOp); + return nativePromise; +} + // v3 async (non-blocking) standalone query. opts: { format?, signal?, timeout? } function queryAsync(query, opts = {}) { if (!query) return Promise.resolve(emptyResult()); const { sql, format } = prepArrow(query, opts, "CSV"); - return withAbortTimeout(chdbNode.QueryAsync(sql, format), opts); + return withAbortTimeout(trackDefault(chdbNode.QueryAsync(sql, format)), opts); } function queryBindAsync(query, params = {}, opts = {}) { @@ -451,7 +468,7 @@ function queryBindAsync(query, params = {}, opts = {}) { const { sql, format } = prepArrow(query, opts, "CSV"); let bound; try { bound = formatParams(params); } catch (e) { return Promise.reject(e); } - return runExclusiveParam(globalParamChain, () => chdbNode.QueryAsync(sql, format, bound), opts); + return runExclusiveParam(globalParamChain, () => trackDefault(chdbNode.QueryAsync(sql, format, bound)), opts); } // v3 insert (default connection). Dispatches on the shape of `values`: @@ -459,8 +476,8 @@ function queryBindAsync(query, params = {}, opts = {}) { // passthrough; Readable/AsyncIterable + format -> backpressured stream insert. function insert(opts) { return dispatchInsert( - (sql) => chdbNode.QueryAsync(sql, "CSV"), - (prefix, buf, countLines) => chdbNode.InsertRawAsync(prefix, buf, countLines), + (sql) => trackDefault(chdbNode.QueryAsync(sql, "CSV")), + (prefix, buf, countLines) => trackDefault(chdbNode.InsertRawAsync(prefix, buf, countLines)), opts || {}); } @@ -533,6 +550,21 @@ class Session { #signalHandler = null; // opt-in signal handler, deregistered on close() constructor(path = "", opts = {}) { + // Opening a session destroys the shared default connection: libchdb binds + // one data directory per process, so the in-memory default has to yield. + // Destroying it while a query is still running on it is not survivable — + // the engine aborts for the rest of the process, and on macOS the worker + // can stay blocked inside libchdb so its promise never settles, which turns + // into an unexplained hang somewhere later. The constructor is synchronous + // and cannot wait, so it refuses and says what to await instead. + if (pendingDefaultOps > 0) { + throw new ChdbConnectionError( + `Cannot open a session while ${pendingDefaultOps} standalone ` + + `${pendingDefaultOps === 1 ? 'operation is' : 'operations are'} still running on the ` + + `default connection. Await them first: opening a session closes that ` + + `connection, and closing it mid-operation aborts the engine for the ` + + `whole process.`); + } if (path === "") { // Create a temporary directory this.path = mkdtempSync(join(os.tmpdir(), TMP_PREFIX)); diff --git a/test/v3/async-stress.test.ts b/test/v3/async-stress.test.ts index bdd5af5..abdec88 100644 --- a/test/v3/async-stress.test.ts +++ b/test/v3/async-stress.test.ts @@ -9,6 +9,14 @@ import { queryAsync, queryBindAsync, Session } from '../../index.js' const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)) const HEAVY = (n: number) => `SELECT count() FROM numbers(${n})` +// HEAVY is not heavy: the engine answers count() over numbers() from the range +// itself, so HEAVY(60_000_000) returns in about 5ms — less than the sleeps below +// it. Any case that needs a query to still be running when the next line +// executes has to force per-row work, or it tests nothing on a fast machine and +// only sometimes tests anything on a slow one. ~120ms, comfortably longer than +// the 0-3ms sleeps. +const SLOW = 'SELECT max(sipHash64(number)) FROM numbers(20000000)' + // 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 — a test killed @@ -118,17 +126,36 @@ describe('lifecycle race: close / registry mutation during an in-flight query', } }, STRESS_TIMEOUT_MS) - it('opening a new session while a default-conn query is in flight is safe (60x)', async () => { - for (let i = 0; i < 60; i++) { - const p = queryAsync(HEAVY(60_000_000), { format: 'CSV' }).then((r) => r.text().trim(), () => 'err') + // This used to assert that opening a session mid-query was safe. It is not: + // the session takes the process's one data directory, which destroys the + // default connection the query is running on, and the engine does not survive + // that. It aborts for the rest of the process, and on macOS the worker can + // stay blocked inside libchdb so the query's promise never settles — which + // surfaced far away, as the suite-wide afterEach drain timing out at 30s and + // every later test in the file reporting "a session is active". + // + // The constructor is synchronous and cannot wait for the query, so it refuses. + // An error naming what to await is worth more than a wait nobody asked for. + it('refuses to open a session while a default-conn query is in flight, and opens once it drains (12x)', async () => { + for (let i = 0; i < 12; i++) { + const p = queryAsync(SLOW, { format: 'CSV' }) await sleep(i % 4) - const s = new Session() + + let opened: Session | null = null try { - const r = await p - expect(r === '60000000' || r === 'err').toBe(true) + opened = new Session() + } catch (e) { + expect((e as Error).message).toMatch(/still running on the default connection/) } finally { - s.close() // close even if the assertion throws, or the session leaks into the next test + opened?.close() // if it did open, do not leak it into the next test } + expect(opened).toBeNull() // the refusal is the contract, not best-effort + + // The query itself is untouched by the refusal, and the session opens as + // soon as it has drained — no lingering state, no retry backoff needed. + expect((await p).text().trim()).toMatch(/^\d+$/) + const s = new Session() + s.close() } }, STRESS_TIMEOUT_MS) }) From 4f6903d1cd9a5ba383d5f1a5de182bc639e2d754 Mon Sep 17 00:00:00 2001 From: Shawn Chen Date: Thu, 13 Aug 2026 11:37:52 +1200 Subject: [PATCH 2/3] Give the refusal a way out, and cover the other path into it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The refusal told callers to await the operation. Two situations have nothing left to await, and probing found both. An aborted or timed-out call rejects immediately while the engine keeps computing — the C ABI has no interrupt. The caller's promise is already settled, so "await it first" is advice they cannot follow, and the only way through was to retry the constructor until it stopped throwing. The second is the answer to whether this is only about the default connection. It is not. Moving between data directories never destroys anything — the engine rejects a second directory rather than evicting the first — so there is no corruption there. But close() returns before the connection is really gone when an operation is still running on it, and the next session at a different path then gets "only one active data directory per process; close the current session", which is precisely what the caller just did. drainPending() waits for both: native operations still on a libuv thread, and connections whose destruction close() deferred behind them. It loops rather than snapshotting, because a deferred teardown can be registered while the drain is already running. _drainPendingOps keeps working as the name the test harness uses. The deferred-teardown case is now refused in JS, ahead of the engine, so the misleading advice never reaches the caller. The teardown counter had the bug the count exists to prevent. Decrementing it in a .finally on the outer promise put it one microtask behind the teardown it guards, so a caller resuming in that window was told a connection was still alive after it had already gone — which failed the 100x close-during-query stress case on the first full run. It now decrements in the same tick as the teardown. The same hazard in the other direction is why the default-connection counter releases before any caller's handler. README documents the constraint, both waits, and the behaviour change: code that opened a session without awaiting its standalone queries used to close the busy connection and now gets an error instead. Full v3 suite: the nine pre-existing failures, unchanged, and three new contract tests covering await-then-open, abort-then-drain, and close-then-switch-directory. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 35 ++++++++++++++ index.d.ts | 21 ++++++++ index.js | 62 ++++++++++++++++++++---- index.mjs | 1 + test/v3/session-open-contract.test.ts | 69 +++++++++++++++++++++++++++ 5 files changed, 178 insertions(+), 10 deletions(-) create mode 100644 test/v3/session-open-contract.test.ts diff --git a/README.md b/README.md index 9edba43..5a72aaf 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,41 @@ Errors are typed (`ChdbSyntaxError`, `ChdbQueryError`, `ChdbConnectionError`, `ChdbAbortError`, `ChdbTimeoutError`, …), each carrying `.code`, the ClickHouse `.clickhouseCode`, and `.cause`. +### One data directory at a time + +libchdb binds a single data directory per process, so opening a `Session` takes +the slot the stateless `query`/`queryAsync` calls were using and closes their +connection. A connection closed while an operation is still running on it aborts +the engine for the rest of the process, so `new Session()` refuses instead: + +```js +const p = queryAsync("SELECT max(sipHash64(number)) FROM numbers(20000000)"); +new Session(); // throws: 1 standalone operation is still running +await p; +new Session(); // fine +``` + +Awaiting your own promise is not always enough. An aborted or timed-out call +rejects immediately while the engine keeps computing, and `close()` returns +before the connection is really gone when an operation is still using it. +`drainPending()` waits for both: + +```js +ac.abort(); +await p.catch(() => {}); // rejected, but the engine is still working +await drainPending(); +const s = new Session("./data"); +``` + +Moving between directories works the same way: after `session.close()`, wait with +`drainPending()` before opening one at a different path. + +**Behaviour change.** Earlier versions did not refuse — they closed the busy +connection, which usually aborted the engine and on macOS could leave a query +whose promise never settled. Code that opened a session without awaiting its +standalone queries now gets an error at the call site instead of a failure +somewhere later. + ### Feature matrix | Capability | Status | diff --git a/index.d.ts b/index.d.ts index 70cd53a..dd0e0f1 100644 --- a/index.d.ts +++ b/index.d.ts @@ -353,6 +353,27 @@ export class Session { [Symbol.dispose](): void; } +/** + * Wait until nothing native is outstanding: queries and inserts still running on + * a libuv thread, and connections that `close()` deferred destroying behind them. + * + * Needed in the two cases where awaiting your own promise is not enough. An + * aborted or timed-out call rejects immediately while the engine keeps + * computing, so there is no promise left to wait on. And `close()` returns + * before the connection is really gone when an operation is still using it. In + * both cases `new Session()` refuses until this resolves. + * + * ```js + * const ac = new AbortController() + * const p = queryAsync(sql, { signal: ac.signal }) + * ac.abort() + * await p.catch(() => {}) // rejects at once; the engine is still computing + * await drainPending() // now the default connection is actually free + * const s = new Session('./data') + * ``` + */ +export function drainPending(): Promise; + /** * Diagnostic version information for the package, the loaded libchdb, and the * current runtime. diff --git a/index.js b/index.js index 143b3a8..8650b95 100644 --- a/index.js +++ b/index.js @@ -510,12 +510,33 @@ function trackNative(nativePromise) { return nativePromise; } -// Wait for every native op started before now to fully settle. Internal helper -// for test teardown: drained in the global afterEach before sessions are closed -// so an early-settled (aborted/timed-out) op stays local to the test that -// started it instead of poisoning the shared single-connection engine. +// Connections that close() has released logically but not yet destroyed, +// because ops were still running on them. The engine still counts them against +// the one-data-directory limit, so a session at a different path cannot open +// until they finish — and the error the engine gives for that says "close the +// current session", which the caller already did. +let pendingTeardowns = 0; + +// Wait until nothing native is outstanding: queries and inserts still running on +// a libuv thread, and connections whose destruction close() deferred behind them. +// +// Two situations need this and cannot be handled by awaiting your own promise. +// An aborted or timed-out call rejects immediately while the engine keeps +// computing — there is no promise left to wait on. And close() returns before +// the connection is actually gone when an op is still using it. In both cases +// the next `new Session()` refuses, and this is the wait that clears it. +// +// Loops rather than snapshotting: a deferred teardown can be registered while +// the drain is already running, and it has to be waited for too. +async function drainPending() { + while (pendingNativeOps.size > 0) { + await Promise.allSettled([...pendingNativeOps]); + } +} + +// Same thing under the name the test harness has always used. function _drainPendingOps() { - return Promise.allSettled([...pendingNativeOps]); + return drainPending(); } // Force-close every session still open in this process. Internal helper for @@ -561,9 +582,23 @@ class Session { throw new ChdbConnectionError( `Cannot open a session while ${pendingDefaultOps} standalone ` + `${pendingDefaultOps === 1 ? 'operation is' : 'operations are'} still running on the ` + - `default connection. Await them first: opening a session closes that ` + - `connection, and closing it mid-operation aborts the engine for the ` + - `whole process.`); + `default connection. Opening a session closes that connection, and closing ` + + `it mid-operation aborts the engine for the whole process. Await the ` + + `operation, or \`await drainPending()\` — an aborted or timed-out call ` + + `rejects straight away but the engine keeps computing, so there may be no ` + + `promise left to await.`); + } + // A session that close() could not destroy yet still holds the process's one + // data directory. Left to the engine this surfaces as "only one active data + // directory per process; close the current session", which is misleading + // advice for a caller who did exactly that. + if (pendingTeardowns > 0) { + throw new ChdbConnectionError( + `Cannot open a session yet: ${pendingTeardowns} closed ` + + `${pendingTeardowns === 1 ? 'session is' : 'sessions are'} still releasing ` + + `${pendingTeardowns === 1 ? 'its' : 'their'} connection, because operations ` + + `were running when close() was called. close() does not destroy a connection ` + + `out from under a running operation. \`await drainPending()\` first.`); } if (path === "") { // Create a temporary directory @@ -738,7 +773,14 @@ class Session { // deferred closes too, so a new session is never created before the prior // connection is fully released. if (conn && pendingNativeOps.size > 0) { - const deferred = Promise.allSettled([...pendingNativeOps]).then(teardown); + pendingTeardowns++; + // Decremented in the same tick as the teardown it guards, not in a later + // .finally: the count exists to answer "is this connection still alive", + // and a caller that resumes between the two would be told yes about a + // connection that is already gone. + const deferred = Promise.allSettled([...pendingNativeOps]).then(() => { + try { teardown(); } finally { pendingTeardowns--; } + }); pendingNativeOps.add(deferred); deferred.finally(() => pendingNativeOps.delete(deferred)); } else { @@ -826,7 +868,7 @@ function _arrowUnregister(connection, tableName) { module.exports = { query, queryBind, queryAsync, queryBindAsync, insert, - Session, version, + Session, version, drainPending, _closeAllSessions, _drainPendingOps, _arrowRegisterColumns, _arrowUnregister, }; diff --git a/index.mjs b/index.mjs index b67b235..30e8ad5 100644 --- a/index.mjs +++ b/index.mjs @@ -15,6 +15,7 @@ export const queryBindAsync = mod.queryBindAsync export const insert = mod.insert export const Session = mod.Session export const version = mod.version +export const drainPending = mod.drainPending // Layer 3: ChdbCompileError is the only net-new error class; export it as a // named ESM binding so it is catchable by class. The fluent builder surface diff --git a/test/v3/session-open-contract.test.ts b/test/v3/session-open-contract.test.ts new file mode 100644 index 0000000..9e43b9c --- /dev/null +++ b/test/v3/session-open-contract.test.ts @@ -0,0 +1,69 @@ +import { describe, it, expect } from 'vitest' +import { mkdtempSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Session, queryAsync, drainPending } from '../../index.js' + +// When new Session() refuses, and how the caller gets unstuck. +// +// libchdb binds one data directory per process. Opening a session therefore +// destroys the in-memory default connection, and a connection destroyed while an +// operation is still running on it aborts the engine for the rest of the process +// — on macOS the worker can stay blocked inside libchdb and never settle. The +// constructor is synchronous and cannot wait, so it refuses. +// +// Awaiting your own promise is not always possible. An aborted call rejects +// straight away while the engine keeps computing, and close() returns before the +// connection is really gone. drainPending() is the wait for both. + +// Forces per-row work so the query is still running on the next line. count() +// over numbers() is answered from the range in ~5ms and would make these pass +// without ever entering the state they exist to test. +const SLOW = 'SELECT max(sipHash64(number)) FROM numbers(20000000)' +const tmpDir = (tag: string) => mkdtempSync(join(tmpdir(), `chdb-contract-${tag}-`)) + +describe('new Session() while the default connection is busy', () => { + it('refuses, and opens after the query is awaited', async () => { + const p = queryAsync(SLOW, { format: 'CSV' }) + expect(() => new Session()).toThrow(/still running on the default connection/) + + await p + const s = new Session() + s.close() + }, 60_000) + + it('refuses after an abort, where there is no promise left to await', async () => { + const ac = new AbortController() + const p = queryAsync(SLOW, { format: 'CSV', signal: ac.signal }) + ac.abort() + await p.catch(() => {}) // rejects at once; the engine is still computing + + // The rejection is not the end of the operation, so the refusal stands and + // the caller has nothing of their own left to wait on. + expect(() => new Session()).toThrow(/drainPending/) + + await drainPending() + const s = new Session() + s.close() + }, 60_000) +}) + +describe('new Session() at a different path while a close is still landing', () => { + it('refuses with the close already made, and opens after draining', async () => { + const a = tmpDir('a') + const b = tmpDir('b') + + const first = new Session(a) + const q = first.queryAsync(SLOW, { format: 'CSV' }) + first.close() // returns, but cannot destroy the connection under the query + + // Left to the engine this is "only one active data directory per process; + // close the current session" — advice the caller has already followed. + expect(() => new Session(b)).toThrow(/still releasing/) + + await q.catch(() => {}) + await drainPending() + const second = new Session(b) + second.close() + }, 60_000) +}) From 20e7bfb1699c7d67f27323dc988f1ce3ce61f457 Mon Sep 17 00:00:00 2001 From: Shawn Chen Date: Thu, 13 Aug 2026 11:49:45 +1200 Subject: [PATCH 3/3] Refuse only the directory that actually conflicts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The teardown guard rejected every new session while any deferred teardown was draining, which is wrong for the case the engine explicitly supports: several connections to one data directory coexist, and a deferred teardown releases a connection, not the directory. Opening a second session at the same path was refused for no reason. Tracked by normalized directory now, and only a different one is refused. A fresh temp directory counts as different, which it is. A sweep over the microtask and timer boundaries around both counters found no false positive or false negative in 132 checks per run, over five runs including one under four-way CPU contention — but its close-then-open case only ever used a different path, so it went straight past this. The same-path case is now covered there and in the contract tests. The sweep can rule things out only along the dimensions it varies. Also audited what could produce a false negative, which is the failure that matters: a native op on the default connection that nothing counts. Every chdbNode call lives in index.js, so Layer 3 and the connection surface cannot bypass the bookkeeping, and of the five native paths that reach get_default_conn only QueryAsync and InsertRawAsync are asynchronous. Both are counted. The rest hold the JS thread, so no session can open underneath them. The README abort example referenced an AbortController and a drainPending import that were not in it. Both snippets now run as written; both were executed. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 13 ++++-- index.js | 58 +++++++++++++++++++-------- test/v3/session-open-contract.test.ts | 31 ++++++++++++++ 3 files changed, 83 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 5a72aaf..98b79d7 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,8 @@ connection. A connection closed while an operation is still running on it aborts the engine for the rest of the process, so `new Session()` refuses instead: ```js +const { queryAsync, Session, drainPending } = require("chdb"); + const p = queryAsync("SELECT max(sipHash64(number)) FROM numbers(20000000)"); new Session(); // throws: 1 standalone operation is still running await p; @@ -83,14 +85,19 @@ before the connection is really gone when an operation is still using it. `drainPending()` waits for both: ```js +const ac = new AbortController(); +const p = queryAsync("SELECT max(sipHash64(number)) FROM numbers(20000000)", { + signal: ac.signal, +}); ac.abort(); -await p.catch(() => {}); // rejected, but the engine is still working -await drainPending(); +await p.catch(() => {}); // rejected, but the engine is still computing +await drainPending(); // now the connection is actually free const s = new Session("./data"); ``` Moving between directories works the same way: after `session.close()`, wait with -`drainPending()` before opening one at a different path. +`drainPending()` before opening one at a different path. Opening another session +at the *same* path needs no wait — those connections coexist by design. **Behaviour change.** Earlier versions did not refuse — they closed the busy connection, which usually aborted the engine and on macOS could leave a query diff --git a/index.js b/index.js index 8650b95..c1201cf 100644 --- a/index.js +++ b/index.js @@ -512,10 +512,22 @@ function trackNative(nativePromise) { // Connections that close() has released logically but not yet destroyed, // because ops were still running on them. The engine still counts them against -// the one-data-directory limit, so a session at a different path cannot open +// the one-data-directory limit, so a session at a DIFFERENT path cannot open // until they finish — and the error the engine gives for that says "close the // current session", which the caller already did. -let pendingTeardowns = 0; +// +// Keyed by normalized data directory, because same-path sessions coexist: a +// teardown pending on one of them is no reason to refuse another at that path. +// Counted per key since several connections can share a directory. +const pendingTeardownPaths = new Map(); +function addPendingTeardown(key) { + pendingTeardownPaths.set(key, (pendingTeardownPaths.get(key) || 0) + 1); +} +function removePendingTeardown(key) { + const n = pendingTeardownPaths.get(key); + if (n === undefined) return; + if (n <= 1) pendingTeardownPaths.delete(key); else pendingTeardownPaths.set(key, n - 1); +} // Wait until nothing native is outstanding: queries and inserts still running on // a libuv thread, and connections whose destruction close() deferred behind them. @@ -589,16 +601,28 @@ class Session { `promise left to await.`); } // A session that close() could not destroy yet still holds the process's one - // data directory. Left to the engine this surfaces as "only one active data - // directory per process; close the current session", which is misleading - // advice for a caller who did exactly that. - if (pendingTeardowns > 0) { - throw new ChdbConnectionError( - `Cannot open a session yet: ${pendingTeardowns} closed ` + - `${pendingTeardowns === 1 ? 'session is' : 'sessions are'} still releasing ` + - `${pendingTeardowns === 1 ? 'its' : 'their'} connection, because operations ` + - `were running when close() was called. close() does not destroy a connection ` + - `out from under a running operation. \`await drainPending()\` first.`); + // data directory, so a session at a DIFFERENT directory cannot open until it + // lands. Left to the engine that surfaces as "only one active data directory + // per process; close the current session" — misleading advice for a caller + // who did exactly that. + // + // Same directory is unaffected: those connections coexist by design, and the + // deferred teardown releases one connection, not the directory. A blanket + // refusal here would reject a supported pattern. + if (pendingTeardownPaths.size > 0) { + // "" is never a real key, so a fresh temp directory always counts as + // different — which it is. + const requested = path === "" ? "" : resolvePath(path); + const blocking = [...pendingTeardownPaths.keys()].filter((k) => k !== requested); + if (blocking.length > 0) { + throw new ChdbConnectionError( + `Cannot open a session at ${requested || "a new temporary directory"} yet: ` + + `a closed session at ${blocking[0]} is still releasing its connection, ` + + `because operations were running when close() was called. close() does ` + + `not destroy a connection out from under a running operation, and the ` + + `engine binds one data directory per process. \`await drainPending()\` ` + + `first, or open at the same path, which is allowed.`); + } } if (path === "") { // Create a temporary directory @@ -619,6 +643,7 @@ class Session { // this.path is left as the caller passed it (public surface). try { const key = this.path ? resolvePath(this.path) : this.path; + this._key = key; // normalized directory, for the deferred-teardown bookkeeping this.connection = chdbNode.CreateConnection(key); } catch (e) { if (this.isTemp) { try { this.#removeTempDir(); } catch (_) {} } @@ -773,13 +798,14 @@ class Session { // deferred closes too, so a new session is never created before the prior // connection is fully released. if (conn && pendingNativeOps.size > 0) { - pendingTeardowns++; - // Decremented in the same tick as the teardown it guards, not in a later - // .finally: the count exists to answer "is this connection still alive", + const key = this._key; + addPendingTeardown(key); + // Removed in the same tick as the teardown it guards, not in a later + // .finally: the entry exists to answer "is this connection still alive", // and a caller that resumes between the two would be told yes about a // connection that is already gone. const deferred = Promise.allSettled([...pendingNativeOps]).then(() => { - try { teardown(); } finally { pendingTeardowns--; } + try { teardown(); } finally { removePendingTeardown(key); } }); pendingNativeOps.add(deferred); deferred.finally(() => pendingNativeOps.delete(deferred)); diff --git a/test/v3/session-open-contract.test.ts b/test/v3/session-open-contract.test.ts index 9e43b9c..9616031 100644 --- a/test/v3/session-open-contract.test.ts +++ b/test/v3/session-open-contract.test.ts @@ -32,6 +32,20 @@ describe('new Session() while the default connection is busy', () => { s.close() }, 60_000) + // The bookkeeping is a counter, so its accuracy is entirely a question of when + // it is decremented: a tick late and a caller who has awaited their own query + // is refused for a connection that is already free. The release is registered + // ahead of any caller's handler for that reason, and extra microtask hops here + // would catch a change that moved it behind one. + it('never refuses after the await, however many microtasks are in between', async () => { + for (const hops of [0, 1, 3, 8]) { + await queryAsync(SLOW, { format: 'CSV' }) + for (let i = 0; i < hops; i++) await Promise.resolve() + const s = new Session() + s.close() + } + }, 60_000) + it('refuses after an abort, where there is no promise left to await', async () => { const ac = new AbortController() const p = queryAsync(SLOW, { format: 'CSV', signal: ac.signal }) @@ -66,4 +80,21 @@ describe('new Session() at a different path while a close is still landing', () const second = new Session(b) second.close() }, 60_000) + + it('allows the same path, which needs no wait at all', async () => { + const a = tmpDir('same') + + const first = new Session(a) + const q = first.queryAsync(SLOW, { format: 'CSV' }) + first.close() // deferred, exactly as above + + // Connections to one directory coexist by design, and the deferred teardown + // releases a connection rather than the directory. Refusing here would + // reject a supported pattern for no reason. + const alongside = new Session(a) + alongside.close() + + await q.catch(() => {}) + await drainPending() + }, 60_000) })