# @leumas/jobs

Leumas Jobs — the one durable, connector-backed job engine every long-running domain shares. Persists each job in the dynamic jobs collection (survives restart), runs kind-keyed runners with a...


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.

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


---
Source: shared/engines/jobs/README.md
Canonical: https://docs.leumas.tech/p/engines/jobs