@gnldev/queue weekly downloads statistics

@gnldev/queue

Durable background-task queue + worker on top of the journal. Each job runs as a durable run → if the process crashes mid-job it resumes from the journal, and a side effect already recorded as done is not run again (at-most-once). acquireRunLock keeps two workers off the same job at once: a worker whose process is frozen (a GC pause, a VM migration, SIGSTOP) keeps its job however long, and a worker whose process died loses it at once. The process is asked through the storage's process liveness (SqliteStorage/PostgresStorage answer for themselves); without one, the lease's TTL decides.

Install: pnpm add @gnldev/queue — or use it from a repo clone: pnpm install && pnpm -r build.

npm i @gnldev/queue   # peer: @gnldev/durable

import { enqueue, createWorker } from '@gnldev/queue';
import { SqliteStorage } from '@gnldev/durable/sqlite';

const storage = new SqliteStorage('runs.db');

// Producer: idempotent enqueue (same id → single job). Takes the WORK store, not the run journal.
await enqueue(storage.work, 'send-email', { to: 'a@x.com' }, { id: 'email:order-1' });

// Consumer: retried on handler crashes (dead-letter after maxAttempts). Takes the whole storage —
// it needs the work store to lock jobs and the run journal to keep the side effect at-most-once.
const worker = createWorker(storage, {
  // A bare `await sendEmail(...)` here is AT-LEAST-once: a crash after the send but before the
  // job is acked reclaims the lock and calls the handler again. Route it through the job's durable
  // run (`ctx.run`, a runDurable on a stable runId), and the send happens once.
  'send-email': async (payload, ctx) => ctx.run({ model, tools, prompt: '...' }),
});
await worker.runOnce();   // or worker.start() (poll loop)

One queue: enqueue here, run here

createQueue is the two calls above in one: the handlers, and a worker that runs them, started in this process. Pass worker: false when the workers run elsewhere.

import { createQueue } from '@gnldev/queue';
import { SqliteStorage } from '@gnldev/durable/sqlite';
const storage = new SqliteStorage('runs.db');

const queue = createQueue(storage, { 'send-email': async (payload, ctx) => { /* ... */ } });
await queue.enqueue('send-email', { to: 'a@x.com' }, { id: 'email:order-1' });
// on shutdown: await queue.close();

A job still waiting after NO_WORKER_NOTICE_MS (10 s) with no worker polling its store is reported once per type (console.warn): enqueued where nothing consumes it, it would otherwise wait without a word.

Delays, retries, cancelling, concurrency

import { enqueue, createWorker, cancelJob } from '@gnldev/queue';
import { InMemoryStorage } from '@gnldev/durable';
declare function sendReminder(p: unknown, signal: AbortSignal): Promise<void>;
const storage = new InMemoryStorage();

// Not before a time (epoch ms or a Date). Stored in the job record, so it survives restarts.
const id = await enqueue(storage.work!, 'remind', { to: 'a' }, { runAt: Date.now() + 60 * 60_000 });

// Changed your mind: no worker starts it after this returns true.
await cancelJob(storage.work!, id);

createWorker(storage, { remind: async (p, ctx) => sendReminder(p, ctx.signal) }, {
  retryDelayMs: (attempt) => 1_000 * 2 ** (attempt - 1), // 1s, 2s, 4s… between failed attempts
  concurrency: 4, // this worker runs up to 4 jobs at once
  typeConcurrency: { remind: 2 }, // at most 2 `remind` jobs at once across ALL workers
});

• retryDelayMs (number or (attempt, err) => ms, default 0): after a failed attempt the job may not run again before now + delay. The time is stored with the job (qtry:<id>), so every worker honours it, also after a restart. The last attempt dead-letters without waiting.
• Attempts are counted when they start, so an attempt the process died in counts too: a handler that takes its worker down every time is dead-lettered after maxAttempts starts, not started forever. listJobs() reports attempts started (a job done on its first try shows 1).
• runAt: compared with the run journal's clock when it has one (now()), else the worker's. A waiting job does not hold up the jobs behind it. Re-enqueuing the same id keeps the first runAt.
• cancelJob(work, id) → true if it cancelled the job, false for an unknown or already finished one. A job that is running is not killed: its ctx.signal aborts on the worker's next heartbeat tick (ttlMs/3; never with heartbeat: false), and ctx.run passes that signal on. If the handler finishes anyway the job shows done; otherwise cancelled. retryJob does not take a cancelled job.
• concurrency (default 1): that many independent poll loops in one worker, after start(). runOnce/drain still run one job per call, and drain runs only what is eligible now.
• typeConcurrency: a per-type limit for the whole fleet, kept as that many leases in the run journal. A dead worker's slot frees after ttlMs. Give every worker the same numbers.
• listJobs shows status: 'cancelled' and, for a job that is waiting, notBefore.
• During a rolling deploy, a 0.9 worker on the same storage ignores runAt, retry delays and cancels.

Jobs for an end user

A job can say whose it is. The job's run belongs to that owner, and the handler does not pass it on: ctx.run starts the run as the owner.

import { enqueue, createWorker } from '@gnldev/queue';
import { InMemoryStorage, user } from '@gnldev/durable';
declare const model: import('@ai-sdk/provider').LanguageModelV4;

const storage = new InMemoryStorage();
await enqueue(storage.work!, 'weekly-summary', { week: 39 }, { caller: user('ayse', 'acme') });
// the same job, in the shorthand: { resourceId: 'ayse', orgId: 'acme' }…