Node.js Workers and Backpressure

@xbbg/xqdb keeps socket I/O and q decoding off the JavaScript event loop. Each Q has:

• one socket to q,
• one native worker thread that owns the socket, and
• a first-in, first-out queue of commands waiting for that thread.

connect(), sync(), asyn(), receive(), and disconnect() add a command to the queue and return a promise. The worker runs commands one at a time, in call order, and resolves each promise on the JavaScript thread when its command finishes. Because one complete request and reply finishes before the next command starts, IPC messages from different calls never interleave.

Admission and XQDB_BACKPRESSURE

A connection holds one running command plus up to eight waiting commands. The JavaScript thread never waits for room in the queue. When all eight waiting slots are taken, the new call rejects immediately with an XqdbError whose code is XQDB_BACKPRESSURE. That command was never queued and never sent.

import { Q, XqdbError } from "@xbbg/xqdb";

const q = await Q.connect({ host: "localhost", port: 1800, timeout: 60_000 });
try {
  const slow = "{t:.z.p+0D00:00:01; while[.z.p<t]; 42}[]"; // about one second of work in q
  const calls = Array.from({ length: 12 }, () => q.sync(slow));
  const results = await Promise.allSettled(calls);
  const rejected = results.filter(
    (result) => result.status === "rejected" && result.reason instanceof XqdbError && result.reason.code === "XQDB_BACKPRESSURE",
  );
  console.log(rejected.length); // typically 3: one runs, eight wait, the rest are rejected
} finally {
  await q.disconnect();
}

Keep the number of unfinished calls per Q at nine or fewer. To run more requests at the same time, spread them over several connections. Each Q has its own socket and worker thread:

import { Q, type XqdbInput, type XqdbValue } from "@xbbg/xqdb";

const options = { host: "localhost", port: 1800 };
const pool = await Promise.all(Array.from({ length: 4 }, () => Q.connect(options)));
let next = 0;

function query(expression: string, ...args: XqdbInput[]): Promise<XqdbValue> {
  const q = pool[next++ % pool.length]!;
  return q.sync(expression, ...args);
}

try {
  const days = ["2024.01.02", "2024.01.03", "2024.01.04", "2024.01.05"];
  const counts = await Promise.all(days.map((day) => query(`count select from trade where date=${day}`)));
  console.log(counts);
} finally {
  await Promise.allSettled(pool.map((q) => q.disconnect()));
}

Release 0.1.8 has no built-in connection pool. Every open connection holds its own native threads and one q handle, so size pools to your workload.

Commands that wait

A command holds the worker until it finishes, and every command behind it waits:

• A long sync() delays every later call on that Q.
• A receive() waits until the server sends a message or the socket times out. Use a dedicated Q for subscriptions.
• disconnect() is a queued command too. It runs only after the commands ahead of it finish or time out.

The socket timeout (default 30 seconds) bounds each wait. See Timeouts and Retries.

Request limits

A request over one of these limits rejects with an XqdbError whose code is XQDB_CONVERSION:

Limit	Value
Arguments per call	8
Expression length	64 MiB of UTF-8
Arguments per call, after native snapshot accounting	64 MiB
Nesting depth of arrays and objects	64 levels

The 64 MiB argument limit counts the native copy of the arguments, including the Arrow IPC bytes of any Table or Vector. serializeAsIpcBytes6() applies the same limit to its value. Split larger uploads into several calls. Results have no fixed size limit. They are bounded by memory and by the 40-bit length field of the q header. A result too large for the machine rejects with an error instead of ending the process. See Protocol and Limits.

Helpers and cleanup

readBinary6() and serializeAsIpcBytes6() do not use a connection. They run as native async tasks on the libuv thread pool.

Call disconnect() when you are done with a Q, in a finally block or a shutdown hook. If a Q is garbage-collected, its native connector closes the socket and stops the worker thread, but only when the garbage collector runs. Explicit cleanup releases the q handle predictably.