diff --git a/README.md b/README.md index 9edba43..98b79d7 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,48 @@ 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 { 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; +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 +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 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. 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 +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 71eb794..c1201cf 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 || {}); } @@ -493,12 +510,45 @@ 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. +// +// 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. +// +// 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 @@ -533,6 +583,47 @@ 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. 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, 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 this.path = mkdtempSync(join(os.tmpdir(), TMP_PREFIX)); @@ -552,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 (_) {} } @@ -706,7 +798,15 @@ 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); + 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 { removePendingTeardown(key); } + }); pendingNativeOps.add(deferred); deferred.finally(() => pendingNativeOps.delete(deferred)); } else { @@ -794,7 +894,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/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) }) diff --git a/test/v3/session-open-contract.test.ts b/test/v3/session-open-contract.test.ts new file mode 100644 index 0000000..9616031 --- /dev/null +++ b/test/v3/session-open-contract.test.ts @@ -0,0 +1,100 @@ +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) + + // 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 }) + 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) + + 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) +})