@leumas/jobs

The one durable job engine, and the queue that is Leumas Fabric's control plane. A job is a row in the dynamic jobs collection, so it survives a restart; a runner registered for its kind does the work; progress rides the automation SSE bus.

import { createJobEngine } from '@leumas/jobs';
import { createJobsStore, createJobsRouter } from '@leumas/jobs/surface';

const jobs = createJobEngine({ connector, stream, name: 'default', concurrency: 1 });
jobs.registerRunner('render', async (job, helpers) => {
  helpers.log('starting');                       // durable — kept on the row, capped
  await helpers.progress('encoding', 40);
  return { url: '…' };                           // returning marks it done; throwing marks it failed
}, { label: 'Render', group: 'Video', memberSafe: true, schedulable: true });

app.use('/api/jobs', createJobsRouter({ store: createJobsStore({ connector }), engines: () => [jobs], gate, isAdmin }));

The five things worth knowing before you touch it

1. Submitting is EXECUTION, and it is authorized per kind. POST /api/jobs denies by default: a kind is operator-only unless its manifest says memberSafe: true. capability reaches the whole ecosystem-cell registry, so it deliberately is not. jobs is also in protectedCollections — the router is the only write door, because resumePending() executes whatever it finds at the next boot.

2. waiting is a state, not a held slot. A runner that cannot start yet returns helpers.requeue(reason, afterMs): the row goes to waiting with the reason in words, the concurrency slot is released, and it is re-asked later. Blocking instead would deadlock the queue — at the default concurrency of 1, one unplaceable job would starve every placeable job behind it. wakeWaiting() re-asks everything immediately, which is what makes plugging a machine in drain the backlog in front of the person who plugged it in.

3. Events are ADDRESSED. push({ userId: owner, andOperators: true }) — the owner sees their own job and operators still see the whole queue. Unaddressed events on that bus reach operators ONLY, so for a while the one person who could not watch a job was the person who started it.

4. Two engines share this collection. The default one and the fleet's (FLEET_CONCURRENCY), kept apart only by kind-scoped resumePending. Registering the same kind on both makes them fight over the same row. describeKinds() reports which engine owns a kind, and at what width.

5. A row is normalized on read. normalizeJob() backfills startedAt/endedAt for rows written before those existed and marks them derived — so a surface renders "≈" and a percentile can exclude them, rather than reporting a queue latency of zero that never happened.

Guards

pnpm smoke:jobs runs the real engine over an in-memory connector: timestamps, the durable tail, the deadlock canary (a waiting job must not hold its slot), cancel disarming a pending re-ask, SSE addressing, and the four authorization refusals. pnpm smoke:fabric covers the fleet path.