kewa

A tiny durable job queue on SQLite. Enqueue with dedupe keys, claim with locks, retry with backoff, and shed load when the backlog spikes, all in one table, no broker to run.

kewa (queue) is the job-queue core pulled out of an always-on service that runs thousands of background jobs a day on a single SQLite file. It is deliberately unopinionated: you bring the job types and their handlers, and kewa owns the mechanics: durable state, dedupe-merge semantics, per-type retry backoff, worker lanes with a watchdog, and pressure-based load shedding.

Why

• One SQLite file, no broker. Jobs live in a jobs table in your own database (a file: path, :memory:, or a remote libsql/Turso URL). No Redis, no RabbitMQ, no separate service to keep alive. Every claim is a local indexed update.
• Durable and idempotent. A claimed job is locked with a lease; if a worker dies mid-job the lease expires and the job is reclaimed. Handlers are meant to be idempotent, so a redelivery is safe.
• Dedupe and debounce built in. Every job has a unique dedupe_key. Re-enqueuing merges instead of duplicating: priority takes the max, and run_after normally moves earlier (an urgent request pulls a scheduled job forward) or, with debounce, later (a burst of events collapses into one delayed run).
• Backpressure that does not starve. When the runnable backlog grows past a target depth, low-value jobs park as deferred_pressure and reactivate once there is headroom. Reserved lanes keep a small always-runnable reserve so no type is starved to zero under sustained load.

Install

npm install @aleju03/kewa

Requires Node 18+. Storage is @libsql/client, so a file: path, :memory:, or a remote Turso URL all work.

Quick start

import { JobQueue, WorkerRunner, createDb } from "@aleju03/kewa";

const db = await createDb({ url: "file:jobs.db" });
const queue = new JobQueue(db);
await queue.ensureSchema();

const runner = new WorkerRunner(queue, {
  lanes: [
    { name: "mail",  jobTypes: ["send_email"],   claimLimit: 4, intervalMs: 250 },
    { name: "media", jobTypes: ["resize_image"], claimLimit: 2, intervalMs: 500 },
  ],
});

// You bring the job types. kewa dispatches by job.type to what you register.
runner.registerHandler<{ to: string }>("send_email", async (job, { signal }) => {
  await sendEmail(job.payload.to, { signal });
});
runner.registerHandler<{ id: number }>("resize_image", async (job) => {
  await resize(job.payload.id);
});

const stop = runner.start();          // background lane timers begin claiming

// Enqueue from anywhere that has the queue:
await queue.enqueue("send_email", `welcome:${userId}`, { to: user.email }, { priority: 10 });

// later: stop();  and  db.close();

enqueue(type, dedupeKey, payload, options?) inserts or merges a job. registerHandler(type, fn) binds a handler; a job whose type has no handler fails with a clear error. Handlers receive the Job and a { signal } context (the watchdog aborts it on timeout).

The injection story: you bring the job types

kewa ships zero domain knowledge. There is no built-in list of job names, no hardcoded handlers. Two things stay yours:

1. The job types and their payloads. You pick the type strings and registerHandler a function for each. The payload is any JSON-serializable value, typed through the handler generic.
2. The tuning. Depth thresholds, which types are sheddable, per-type caps, and reserved lanes are all constructor options with generic defaults. Nothing is assumed about your workload.

const queue = new JobQueue(db, {
  targetDepth: 200,                       // trim the runnable pool back to this under pressure
  softPressureDepth: 160,                 // at/above this, sheddable types defer on enqueue
  sheddableTypes: ["rebuild_search_index"],
  typeCaps: { call_external_api: 20 },    // keep at most N of this type runnable at target depth
  reservedLanes: { nightly_report: 1 },   // a small always-runnable reserve, invisible to shared depth
});

Lanes

A WorkerRunner runs one or more lanes. Each lane polls on its own intervalMs, claims up to claimLimit jobs (optionally filtered to a set of jobTypes), and runs them concurrently. Separate lanes keep slow work from starving fast work: a wide interactive lane and a narrow claimLimit: 1 backfill lane share the same queue without fighting.

runner.pause();          // stop starting new jobs (in-flight jobs finish)
runner.resume();
runner.status();         // { paused, stopped, workerId, lanes: [{ name, activeJobs, ... }] }
await runner.runOnce();  // claim + run one batch synchronously (handy in tests)

Each job invocation is guarded by a watchdog (DEFAULT_JOB_WATCHDOG_MS, 10 minutes, or per-lane jobTimeoutMs). If a handler hangs past the ceiling, its lane is released so it keeps ticking, the job takes the normal fail-with-backoff path, and the handler's AbortSignal fires so cooperative handlers can stop at their next boundary.

Backoff

A handler that throws marks its job failed and stamps run_after in the future, so the job is retried after a delay, not immediately. The default is exponential: base 30s, doubling per attempt, capped at 60 minutes. Override it globally, per handler, or per error:

const runner = new WorkerRunner(queue, {
  retryDelayMs: 5_000,                                   // flat override for every type
  retryBackoff: { baseMs: 1_000, factor: 3, maxMs: 60_000 }, // or tune the exponential curve
});

runner.registerHandler("call_external_api", handler, { retryDelayMs: 60_000 }); // per handler

const runner2 = new WorkerRunner(queue, {
  // Classify an error into fail (default), defer (retry, no failure recorded),
  // or complete (give up and mark done), with an optional per-occurrence delay.
  errorClassifier: (error) => {
    if (error instanceof RateLimited) return { retryDelayMs: error.retryAfterMs };
    if (error instanceof NotFound)    return { action: "complete", reason: "gone" };
    return { action: "fail" };
  },
});

Observe the lifecycle with onJobEvent, which fires on started / done / failed / deferred:

new WorkerRunner(queue, {
  onJobEvent: (e) => log.info(e.status, { job: e.jobId, type: e.type, lane: e.lane, attempts: e.attempts }),
});

Pressure and shedding

await queue.pressure() returns a snapshot: current runnable depth, deferred count, the configured thresholds, and whether the queue is shedding. Jobs scheduled for the future (retry backoffs, self-rescheduling work) count as appointments, not backlog, so they never trip shedding on their own.

Shedding runs on every enqueue and reserved-lane refills run on every claim; a long-lived deployment should also call await queue.shedPressure() on a timer (say every 30-60s) so parked work reactivates even when nothing is being enqueued. Reserved-lane types are held at their reserve independently of the shared pool, guaranteeing a trickle of progress for every type under any pressure.

Other queue methods: complete(id), fail(id, error, retryDelayMs?), defer(id, retryDelayMs?), depth(), summary() (counts by status/type with the newest error), clearFailed(type?), and hasRunnableOutsideTypes(types).

Run the demo

npm install
npm run example   # builds a queue, registers handlers, runs two lanes, sheds load, retries a flaky job
npm test          # queue merge/backoff/shedding + runner end-to-end tests

License

MIT