cairnq (TypeScript / Node)

SQLite-first, cross-language, storage-centered durable task runtime. The TypeScript SDK (Node ≥ 20). API and worker processes coordinate only through a shared SQLite file.

Both drivers are optional peers — install the one you use: npm i better-sqlite3 for the SQLite backend, npm i pg for Postgres. Importing cairnq loads neither, so a Postgres-only deployment builds no native module.

import { CairnQ, Worker } from "cairnq";

// Worker side — a handler always receives (ctx, payload).
const worker = Worker.sqlite("tasks.db", { queues: ["gpu"] });
worker.task("image.generate", async (ctx, payload) => {
  await ctx.progress(0.1, "starting");
  return { url: await generate(payload.prompt) };
});
await worker.serve(); // runs until SIGINT/SIGTERM, then closes the store

// API side
const tasks = CairnQ.sqlite("tasks.db");
const task = await tasks.submit("image.generate", { prompt }, {
  key: `user:${userId}:image:${requestId}`,
  queue: "gpu",
  conflict: "reuse",
});

Synchronous call (submit + wait):

import { TaskFailed, TaskTimeout } from "cairnq";

try {
  const result = await tasks.call("summary.create", { text }, { timeoutMs: 10_000 });
} catch (err) {
  if (err instanceof TaskFailed) log(err.code, err.message, err.retryable); // envelope fields
  else if (err instanceof TaskTimeout) {
    // The task keeps running — resume the wait instead of submitting again.
    const result = await tasks.wait(err.taskId, { timeoutMs: 60_000 });
    // …or tasks.waitByKey(key), from a process that never held the id.
  }
}

Inspect a task by id/key:

const task = await tasks.getByKey(key);
if (task?.status === "succeeded") use(task.result);

Optionally define a task once and share the symbol across both ends — no string drift, the editor finds every caller, and payload + result are fully typed:

import { defineTask } from "cairnq";

export const summarize = defineTask<{ text: string }, { summary: string }>("summarize");

worker.task(summarize, async (ctx, payload) => ({ summary: await run(payload.text) }));
const { summary } = await tasks.call(summarize, { text }); // typed result, no cast

Opt-in: every API still accepts a plain name string (cross-language callers use it).

Running it in production

const worker = Worker.sqlite("tasks.db", {
  concurrency: 4,          // handler calls at once (a batch call counts as one)
  retryBackoffMs: 1_000,   // window doubles per attempt, capped at retryBackoffMaxMs (30s),
                           // jittered over its upper half; 0 disables
  onError: (err, info) => log.warn({ err, ...info }), // claims/writes the loop survived
});

// Nothing else deletes rows, so give the client a retention cutoff — it sweeps
// terminal tasks in bounded batches for as long as the handle is open. Tiered
// retention is the same option in its wider forms: a per-status map, or a list
// of RetentionRule filtering by anything purge() can (queue, status, name).
const tasks = CairnQ.sqlite("tasks.db", { retention: 7 * 24 * 3600_000 });

A handler that does real side effects should bail out when it loses its lease — the task is already running on another worker and nothing it writes is recorded:

worker.task("long.job", async (ctx) => {
  const res = await fetch(url, { signal: ctx.signal }); // aborts on lease loss
  if (ctx.lostLease || (await ctx.canceled())) return;
});

Multi-host

Same code, Postgres instead of the file — CairnQ.postgres(dsn) / Worker.postgres(dsn). Requires the optional pg peer dependency (npm i pg).

Sharing the application's connection

Given a PgExecutor instead of a DSN, cairnq runs inside a session the application already has — no second driver, no second pool:

import { CairnQ, type PgExecutor } from "cairnq";

const executor: PgExecutor = { /* ~30 lines over your driver */ };
const tasks = CairnQ.postgres(executor);

schema puts cairnq's tables in a schema of their own: CairnQ.postgres(dsn, { schema: "cairnq" }) creates it if absent and sets search_path per connection. The protocol's SQL names no schema, so nothing else changes. With your own executor the search_path is yours to set, and schema becomes an assertion about where it lands.

Every process in a deployment must agree on it. A queue whose API and worker resolve to different schemas is two empty queues, and — because every migration is create table if not exists — both sides come up healthy, pass their protocol version check, and never see each other's tasks. cairnq refuses to connect where it can see that about to happen (SchemaMismatch); the Python SDK applies the same rule.

Two things an adapter must get right: int8 has to come back as a JS number (every cairnq *_ms is an epoch or a counter, all inside the safe range), and jsonb as an object rather than a string. An executor cairnq was handed is never closed by cairnq.

That shared session is also what lets a task's settlement commit together with the rows the task produced:

worker.task("render.document", async (ctx, payload) => {
  const rendered = await render(payload);
  return ctx.succeedIn(async (session) => {
    await db.withSession(session).insert(pages).values(rendered);
    return { pages: rendered.length }; // becomes the task's result
  });
});

Without it the two are separate transactions, and a crash between them leaves work durable while the task still reads as running — on retry, recomputed. If the lease turns out to be gone, the settlement matches no row and the caller's writes roll back with it.

The protocol (schema + canonical SQL) lives in ../cairnq-protocol and is shared verbatim with the Python SDK. See ../cairnq-protocol/PROTOCOL.md.