Background work: queue, scheduler & events

Not every agent run starts with an HTTP request. Something has to process a backlog, fire at 03:00, or react to an event — and each of those is a place where a crash can duplicate a side effect. These three packages put the same guarantee at those three boundaries.

None of them reimplements durability. A job handler typically calls runDurable, a trigger runs a durable workflow, and an event consumer is expected to do the same. What each package adds is the piece the journal alone cannot give: a lock so two workers do not take the same job, a fire record so a cron slot is not run twice, and an ack marker so an event is not marked done before it is.

enqueue writes the job, createWorker claims and runs it. A worker holds a lock with a TTL; if it crashes, the lock goes stale, another worker reclaims the job, the handler runs again — and because the handler resumes the same runId, the side effect still happens once. Failure is bounded by maxAttempts, after which the job becomes dead-letter and can be revived with retryJob.

Empty-poll backoff is on by default: when a poll claims nothing, the interval doubles up to a cap and resets the moment a job is claimed. That default exists because an audit found the alternative — a thousand consumers polling an empty queue at a constant interval produces tens of thousands of queries a second.

Jobs are taken fairly across organizations. Each organization's jobs are a lane, and a worker asks the store for the next job of the organization that has waited longest and locks it, in one store call (SKIP LOCKED on Postgres and MySQL, findOneAndUpdate on MongoDB, one writer on SQLite and libSQL). An empty poll is a constant number of queries, whatever the number of organizations. A store that still holds a 0.16 queue needs gnl upgrade before this version's workers start.

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

await enqueue(storage.work, 'reindex', { docId: 'd-9' });

// one handler per job type; ctx carries the job's own durable runId and journal
const worker = createWorker(
  storage,
  {
    reindex: async (payload, ctx) => {
      // a reclaim resumes this runId, it does not restart it
      await runDurable({ runId: ctx.runId, journal: ctx.journal, model, tools, prompt: payload.docId });
    },
  },
  { ttlMs: 30_000, maxAttempts: 3 },
);

worker.start();

enqueueMany(work, jobs) adds many jobs in one write — all or none. Each job is an object with type, payload and optional options, the same ones enqueue takes, and is checked as enqueue checks it: one invalid job refuses the call before anything is written. The ids come back in call order; an id already enqueued, or repeated in the call, is one job; maxDepth counts the call's own jobs. Fairness is that of enqueue one job after another, but the call takes each organization's turn once, not once per job — the way to put a large batch on one organization, such as a CSV import (30,000 jobs on Postgres in about 3 seconds).

listJobs returns a page: read .items, and nextCursor when there is one. Without orgId it lists only the jobs that have no organization, not every organization's. An operator who wants every job on the store uses listAllJobs: the jobs with no organization first, then each organization's lane in name order, each in enqueue order. A listing is fixed at its start — every job enqueued before its first page is listed exactly once, and one enqueued after it belongs to the next listing — so it ends even under producers that never stop.

import { enqueueMany, listJobs, listAllJobs } from '@gnldev/queue';

// a CSV import: one write, all or none; the ids come back in call order
const ids = await enqueueMany(
  storage.work,
  rows.map((row) => ({
    type: 'import-row',
    payload: row,
    options: { id: `import:${row.id}`, orgId: 'acme' },
  })),
);

// one organization's jobs, a page at a time
const page = await listJobs(storage.work, { orgId: 'acme', limit: 50 });
page.items;       // the jobs
page.nextCursor;  // pass it back as { cursor } for the next page

// an operator's view: every job on the store
const all = await listAllJobs(storage.work, { limit: 100 });

scheduleWorkflow registers a trigger: a one-shot at, a periodic every, or a 5-field cron (UTC, minute resolution). The trigger definition is immutable in the journal and only its state moves. Each fire takes a run lock and derives its runId from the fire count, so the same slot cannot start twice.

Time is treated as data: the next run time is resolved and then frozen into the journal, which is what keeps a replay from drifting with the wall clock.

misfire decides what a missed window means when no poller ran. 'latest' (the default) fires the last missed occurrence, once; 'earliest' the oldest, once; 'none' runs no missed occurrence; 'catchup' fires every missed occurrence, oldest first, one per poll. Under every value the schedule goes on from the first occurrence after now, on its planned grid, so no drift accumulates. A nightly report usually wants latest; a billing cycle usually wants catchup.

'catchup' has no limit by default — a day asleep on a minute cron is 1,440 fires. With maxCatchup: n only the n latest missed occurrences fire, and the trigger logs one warning naming how many were skipped and when the oldest one was. Every fire is told which occurrence it is for: ctx.scheduledFor in a workflow step (the slot on the schedule, not the time the poll ran), kept across retries and resumes.

Due triggers no longer wait for each other. Every due trigger fires at once, and concurrency (on createScheduler and pollScheduler) has no default limit — pass concurrency: n to cap it; a 10-second workflow used to hold the whole poll loop and cut a 2-second cron from 12 fires to 2. A fire's lock is taken, and its decision checked under it, with the store's clock, so a process with a skewed clock no longer starts the same occurrence twice.

import { scheduleWorkflow, createScheduler } from '@gnldev/scheduler';

await scheduleWorkflow(journal, {
  name: 'nightly-report',
  cron: '0 3 * * *',       // 03:00 UTC
  misfire: 'latest',       // the default: after an outage, the last missed slot fires once
  // misfire: 'catchup', maxCatchup: 10,   // or every missed slot, capped at the 10 latest
});

createScheduler(journal, runner, { concurrency: 4 }).start(); // concurrency: no limit unless you cap it

// inside a workflow step: the occurrence this fire is for
//   ctx.scheduledFor   // epoch ms of the slot, not of the poll

emit appends to a log; each consumer processes events under its own ack marker, so N consumers each see every event. The marker is written with a CAS after the handler succeeds. That ordering is the whole design: if the handler throws or the process dies, the event is not lost and the next poll retries it.

The cost is stated rather than hidden: this is exactly-once marking with at-least-once delivery. A crash between a successful handler and its ack — or a concurrent poll race — can redeliver. Write the handler idempotently, or put runDurable / claim inside it, and the redelivery costs nothing.

import { emit, createConsumer } from '@gnldev/events';

await emit(storage.work, 'order.paid', { orderId: 'o-7' });

const consumer = createConsumer(
  storage.work,
  'order.paid',
  async (payload, meta) => {
    // the ack marker is written only after this returns — so make it idempotent
    await runDurable({ runId: `evt:${meta.id}`, journal, model, tools, prompt: payload.orderId });
  },
  { name: 'fulfilment' },   // ack markers are separated by this name (fan-out)
);

consumer.start();

A handler's right to run can end while it runs: the job is cancelled, another worker holds it now because the lease was lost, or it has held it as long as it may. All three reach the handler as one thing, ctx.signal, which aborts with an ExecutionRevokedError; its why says which one: canceled, lease-lost or timeout. It used to abort on a cancel only, and a worker that had lost its lease ran on beside the new holder.

maxHoldMs on a queue worker (default: no limit) is the time limit. When it is up, ctx.signal aborts with timeout, and the attempt counts as failed — retried, dead-lettered at maxAttempts. The lease is kept until the handler returns, so nothing runs beside it. An event handler gets the same through meta.signal (timeout, lease-lost), and createConsumer takes maxHoldMs too. A scheduler fire's runner gets opts.signal, aborted as lease-lost when a renewal answers that the lock is not its own, or three renewals in a row go unanswered; createGnl().runWorkflow stops the workflow at its next step boundary on it — it stops the workflow, it does not cancel it.

createWorker(
  storage,
  {
    reindex: async (payload, ctx) => {
      // ctx.signal.reason is an ExecutionRevokedError; its why is 'canceled' | 'lease-lost' | 'timeout'.
      // ctx.run passes the signal on to the run; other long work checks it itself.
      const res = await fetch(payload.url, { signal: ctx.signal });
      return ctx.run({ model, tools, prompt: await res.text() });
    },
  },
  { ttlMs: 30_000, maxAttempts: 3, maxHoldMs: 10 * 60_000 }, // a handler held past 10 minutes: timeout
);

The framework cannot stop code that ignores the signal. ctx.run passes it on to the run; other long work has to check it itself, and side effects belong inside ctx.run so a takeover does not repeat them. A handler still running after its lease was lost or its maxHoldMs ran out is warned about once (console.warn), ttlMs (queue), leaseMs (events) or lockTtlMs (scheduler) later. A test that compares an event's meta with toEqual now needs toMatchObject.

A consumer's cursor only advances past an event its handler accepted. So a handler that keeps throwing used to park the cursor forever: every event behind the failing one waited on a delivery that was never going to succeed. Not a lost message — a stopped topic.

A delivery is now quarantined after maxAttempts failures (default 8) and the cursor moves on. The attempts are spaced by retryDelayMs — exponential from 60s, doubling, capped at an hour — so the budget spans about two hours rather than burning out in a second at the poll interval. A retry budget is only a budget if it outlasts the kind of outage it exists for.

A not-yet-due event is skipped but still freezes the cursor: it is retryable, so nothing behind it may be marked as passed. Waiting is not giving up. maxAttempts: Infinity keeps that shape forever if you would rather never quarantine — other events still flow, but this consumer stays parked and re-scans from that point on every poll.

import { createConsumer, listDeadEvents, retryDeadEvent } from '@gnldev/events';

const consumer = createConsumer(storage.work, 'order.paid', handler, {
  name: 'fulfilment',
  maxAttempts: 8,        // quarantine after this many FAILED attempts (the default)
  retryDelayMs: 60_000,  // fixed spacing; omit for exponential 60s → 1h (the default)
});

// what is parked, and why
const dead = await listDeadEvents(storage.work, 'order.paid', 'fulfilment');
// [{ id: 'e-3', attempts: 8, error: 'ECONNREFUSED', at: 173... }]

// hand one back once the downstream is healthy again
await retryDeadEvent(storage.work, 'order.paid', 'fulfilment', 'e-3');

listDeadEvents(work, topic, consumer) shows what is parked and retryDeadEvent(work, topic, consumer, eventId) hands one back. Studio surfaces both, so an operator does not need a script. Addressing is by the triple (topic, consumer, id) — a topic fans out, so one event has one quarantine record per consumer and an id alone names several of them.