Refuse to open a session while the default connection is busy

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 |

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<void>;
+
 /**
  * Diagnostic version information for the package, the loaded libchdb, and the
  * current runtime.

index.js

@@ -439,28 +439,45 @@ 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 = {}) {
   if (!query) return Promise.resolve(emptyResult());
   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`:
 // row arrays -> inline VALUES; Buffer/Uint8Array/string + format -> raw
 // 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,
 };

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

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)
 })