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.

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();

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. 'skip' (the default) jumps to the next slot on the planned grid so no drift accumulates; 'catchup' fires each missed occurrence in turn, one per poll. A nightly report usually wants skip; a billing cycle usually wants catchup.

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

await scheduleWorkflow(journal, {
  name: 'nightly-report',
  cron: '0 3 * * *',       // 03:00 UTC
  misfire: 'skip',         // or 'catchup' to replay every missed slot
});

createScheduler(journal, runner).start();

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 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.