# Background Jobs `$job` is the primitive for work that happens outside a request. It is backed by a database table (the _outbox_), which is what makes it durable: a push writes a row before anything runs, so a handler that throws - or a process that dies mid-flight - leaves a record that a reconciliation sweep picks back up. ```typescript check import { $job } from "alepha/api/jobs"; ``` `$job` lives under `alepha/api/` because it needs an ORM connection and ships an admin controller. Register the module alongside your other API modules: ```typescript import { AlephaApiJobs } from "alepha/api/jobs"; alepha.with(AlephaApiJobs); ``` ## Two modes, never both A job declares **either** `schema` (queue-mode, push-driven) **or** `cron` (cron-mode, schedule-driven). Declaring both is a configuration error. ```typescript check import { z } from "alepha"; import { $job } from "alepha/api/jobs"; class Emails { // queue-mode - call push() to enqueue work welcome = $job({ name: "emails.send-welcome", description: "Sends the welcome email to a new user.", schema: z.object({ userId: z.text() }), retry: { retries: 3 }, handler: async ({ payload, attempt }) => { // send the welcome email for payload.userId }, }); // cron-mode - fires on a schedule, no payload digest = $job({ name: "emails.send-digest", description: "Builds and sends the daily digest.", cron: "0 8 * * *", handler: async () => { // build and send the daily digest }, }); } ``` To run scheduled work _over a set of payloads_, compose the two: a cron job that pushes, and a queue job that handles. ```typescript class Reminders { sweep = $job({ name: "reminders.sweep", description: "Pushes a reminder for every row that is due.", cron: "0 * * * *", handler: async () => { const due = await this.repository.findMany({ where: { due: true } }); await this.remind.pushMany( due.map((row) => ({ payload: { id: row.id } })), ); }, }); remind = $job({ name: "reminders.send", description: "Sends one reminder.", schema: z.object({ id: z.text() }), handler: async ({ payload }) => { /* ... */ }, }); } ``` ## Naming Every job declares a `name` and a `description`. - The name is `.` in lowercase kebab-case: `estates.sweep-commands`, `quests.send-due-reminders`. The domain is the module or business area, a plural noun where natural; the action is a verb phrase that does not repeat it. - `system..` is reserved for jobs shipped from `packages/` (`alepha` and every `@alepha/*`), such as `system.notifications.send`, so a framework job never collides with an application's. An application never uses `system.`. - The description is one sentence saying what the job does, shown to operators in the admin, at most 255 characters. Neither is translated. The name is checked at registration, and `yarn check:conventions` checks that `system.` is used exactly where it belongs. **The name is the job's identity in `job_executions`.** Renaming a job loses its rows: its history, and any work queued or scheduled under the old name, are deleted at the next trim tick (see [A rename loses the history](#a-rename-loses-the-history)). Get the name right when you declare it. ## Pushing work ```typescript const executionId = await this.welcome.push({ userId: "u1" }); ``` `push()` accepts a second options argument: | Option | Type | Description | | ----------------- | -------------- | ----------------------------------------------------------------------------------------- | | `delay` | `DurationLike` | Run no earlier than now + delay | | `scheduledAt` | `Date` | Run no earlier than this instant | | `key` | `string` | Deduplication key - see the caveat below | | `inline` | `boolean` | Run now and wait, no retry - see [`inline`](#inline-when-a-retry-is-worse-than-a-failure) | | `triggeredBy` | `string` | Who pushed it (user id), shown in the admin | | `triggeredByName` | `string` | Display name for `triggeredBy` | `pushMany()` takes an array of `{ payload, key?, delay?, scheduledAt? }` and writes them in a batched INSERT. `trigger()` runs a job now, outside its schedule: a cron job fires its handler once (see [Multi-replica deployments](#multi-replica-deployments) for how it avoids overlapping a tick), a queue job takes `{ payload }` and is pushed like any other. Both accept `triggeredBy` / `triggeredByName`; it is what the admin's "run now" button calls. ### `key` dedups in-flight work, not completed work A push with a `key` returns the existing execution id instead of enqueueing a second row - but only while a row with that key still exists. On success the row is either deleted (a queue job keeps no successes by default) or updated with `key` set to `null`. Either way **the key is released once the job succeeds**. So `key` means _"don't enqueue this twice while it's still pending, running, or failing"_ - it is not _"run this at most once ever"_. If you need the stronger guarantee, enforce it in the handler against your own data. ## Retries Set `retry: { retries: n }`, optionally with `when: (error) => boolean` to retry only certain failures. A failed attempt is rescheduled with **exponential backoff and full jitter**: attempt _n_ waits a uniformly random time in `[0, min(retryBackoffMax, retryBackoffBase * 2^(n-1))]`, defaults 5 s and 30 min. The jitter matters at least as much as the curve - without it every retrying row in the system shares one `scheduledAt` and they all hit a struggling downstream together. A job can carry its own curve when the global one does not fit: `retry.backoff` replaces the base, the factor and the cap for that job. A settlement that polls a payment provider wants minutes between attempts, not seconds: ```typescript reconcile = $job({ name: "checkouts.reconcile", description: "Settles a checkout the payment provider has confirmed.", schema: z.object({ sessionId: z.uuid() }), retry: { retries: 3, backoff: { initial: [1, "minute"], factor: 4, max: [30, "minute"] }, }, handler: async ({ payload }) => { await this.payments.sync(payload.sessionId); }, }); ``` Attempt _n_ waits `initial * factor^(n-1)`, capped by `max` (default: the global `retryBackoffMax`), then jittered the same way: `jitter` is on by default and means the full jitter above. `jitter: false` gives the exact curve, which is what a test asserting on `scheduledAt` wants. The row's `scheduledAt` is the truth and the sweep is the backstop, so nothing can lose a retry. What varies by runtime is only **how soon** something looks at it: | Runtime | Dispatch | Retry lands | | ----------------------------------------- | --------------- | --------------------------------- | | Node, any dispatcher | direct or queue | at the backoff, on a local timer | | Cloudflare Workers + `AlephaApiJobsQueue` | queue | at the backoff, held by the queue | | Cloudflare Workers, no queue | direct | **next sweep tick** (see below) | The last row is a real limit, not an oversight. A timer armed after the response never fires on Workers - the isolate freezes once `waitUntil` settles - so direct mode there cannot arrange a wake-up at all and retries keep `sweepCron` granularity. Add `AlephaApiJobsQueue` if that matters, or use [`inline`](#inline-when-a-retry-is-worse-than-a-failure) for a payload that expires before the next tick. Cron-mode jobs without `retry` do not retry - the next tick is the retry. Cron jobs that _declare_ `retry` go through the outbox instead, which is useful for once-daily jobs where waiting a full day is not acceptable. ### What a transport does with a delay `$job` asks exactly one thing of a transport: **do not deliver before time T.** Durability, retry policy, attempt counting, dead-lettering and crash recovery are all already owned by the outbox, which is why the interface carries one optional argument rather than a broker abstraction. The rule every backend follows: > `delaySeconds` is an optimisation. The outbox row's `scheduledAt` is the > truth, and the sweep is the backstop. A backend that cannot honour a delay > must **decline to enqueue** rather than enqueue immediately. Declining is not the same as ignoring. For a push transport, ignoring a delay means _delivering now_, and for a retry that is worse than doing nothing at all: no backoff whatsoever against a downstream that has just failed. | Backend | Delay | | ----------------------- | ---------------------------------------------------------------------- | | Cloudflare Queues | native, clamped at its 12-hour ceiling | | Redis (`REDIS_URL` set) | a sorted set scored by due-time, promoted into the list when it is due | | in-memory | a due timestamp, filtered on pop | Every backend that ships with Alepha honours a delay, so nothing declines in practice today. The rule still matters: it is what a custom `QueueProvider` is held to, and "ignore it and deliver now" has to stay unavailable as an option. The Redis tier buys two things a local timer cannot. A delayed message is **server-side state**, so it survives the process that pushed it - a deploy inside the delay window used to drop every armed timer it was holding. And one sorted-set entry costs nothing per message, where one live `setTimeout` per delayed job is real heap on a large backlog. Its scan only runs when the list is empty, so a busy queue pays no extra round trip for the tier at all. `ZREM` is the claim, so two pollers racing the same due message agree on an owner instead of delivering it twice. ## Waiting between stages: `reschedule` Some work is a sequence with waits in it: remind after an hour, remind again a day later, give up after another. One job carries the whole sequence when its handler calls `reschedule()` instead of returning: the same execution row goes back to `scheduled` with a new `scheduledAt` and a new payload, keeps its `id` and its `key`, and its `attempt` starts over. A stage in the payload is what the handler switches on: ```typescript cartRecovery = $job({ name: "carts.recover", description: "Reminds a buyer about an abandoned cart, then marks it abandoned.", schema: z.object({ cartId: z.uuid(), stage: z.enum(["remind", "remindAgain", "abandon"]).optional(), }), retry: { retries: 3, backoff: { initial: [1, "minute"], factor: 4 } }, handler: async ({ payload, reschedule }) => { if (!(await this.carts.isRecoverable(payload.cartId))) { return; // converted or gone: the sequence ends here } switch (payload.stage ?? "remind") { case "remind": await this.mailer.remind(payload.cartId, 1); reschedule({ delay: [23, "hour"], payload: { ...payload, stage: "remindAgain" }, }); return; case "remindAgain": await this.mailer.remind(payload.cartId, 2); reschedule({ delay: [24, "hour"], payload: { ...payload, stage: "abandon" }, }); return; case "abandon": await this.carts.markAbandoned(payload.cartId); } }, }); // One push starts the sequence; the key makes a second push land on it. await this.cartRecovery.push({ cartId }, { key: cartId, delay: [1, "hour"] }); ``` The wait is persisted before any timer is armed, so a redeploy inside it loses nothing: the row's `scheduledAt` is the truth and the sweep is the backstop, exactly as for a delayed push. Each stage gets the job's full retry budget, because `attempt` resets. A durable loop is the same shape with an iteration counter instead of a stage: the handler does its round, then reschedules itself with `iteration + 1` until it reaches its limit. The rules that keep it honest: - `reschedule()` records an intent; the row is written when the handler resolves. A handler that **throws** after calling it takes the retry path on the old payload, and the intent is discarded. Called twice, the last call wins. - The write is guarded on `running`: a `cancel()` that lands while the handler is finishing wins, and nothing is dispatched. - The new payload is validated against the schema when `reschedule()` is called, so a bad one fails the run there rather than parking garbage. - A rescheduled run emits `job:end` but not `job:success`; the execution is not over. Retention only applies to the stage that completes. - It throws from a cron tick (no row to park) and from an [`inline`](#inline-when-a-retry-is-worse-than-a-failure) push (the caller is waiting for an outcome). ## `inline`: when a retry is worse than a failure Some payloads expire. A verification code lives 300 seconds by default while the sweep runs every 900, so a retried code is **guaranteed** to arrive after it expired: all three attempts produce garbage and the user meanwhile sees nothing at all. `inline` says: run the handler here, make me wait, and if it fails tell me. ```typescript await myJob.push(payload, { inline: true }); // resolves -> the handler ran to completion, outbox row terminal // rejects -> the handler failed, row terminal `error`, nothing retries it ``` No dispatcher, no queue, no `waitUntil`. Behaviour is identical with and without `AlephaApiJobsQueue`, because the flag bypasses `JobDispatcher` entirely. Two things it does, and it is worth separating them: 1. **The caller learns.** A password reset fails in front of the user, who can simply ask for another one, instead of quietly succeeding on attempt two with a code that no longer works. 2. **A failure is terminal, never `scheduled`.** This half holds even where the caller swallows the rejection, and it is the one that closes the expired-code problem: nothing will deliver that payload later. **Per push, not per job.** Declare it on the job as a default if you like, but the useful form is the call-site override, because one job usually sits behind many callers - `sendNotification` is the single job behind every notification. Whether you can afford to wait is a property of the call site. ```typescript await myJob.push(payload); // async, retries per the policy await myJob.push(payload, { inline: true }); // this one waits, and does not retry ``` On a job that declares `retry`, a per-push `inline` means _this execution does not retry_: one attempt, terminal on failure, thrown to you. Declaring `inline` and `retry` together **on the job** is rejected at registration, as is `inline` with `cron` (a tick has no caller to block) and `inline` on `pushMany`. Read the contract precisely: - **"Ran to completion" means the handler resolved.** For an email that is the provider accepting the message, not delivery to an inbox. - **Call it after the commit, not inside a transaction.** An email cannot be rolled back, so sending inside a transaction that later fails means mailing a code for a row that no longer exists. `inline` buys ordering, not transactionality. - **Never on a message addressed to somebody other than the caller.** Blocking a login or registration response on a mail to the account _owner_ turns response time into an account-enumeration oracle: a slow answer means the account existed. Alepha's own `registrationAttempt` and `accountLockout` notifications are excluded for exactly this reason. Not to be confused with `$notification`'s `critical`, which is a different property one layer up: that one means the recipient cannot opt out. ## Timeouts and cancellation `timeout` caps a single attempt. The handler receives an `AbortSignal` - pass it to anything that supports one, since Alepha cannot interrupt synchronous work: ```typescript report = $job({ name: "reports.build", description: "Builds one report.", schema: z.object({ id: z.text() }), timeout: [30, "seconds"], handler: async ({ payload, signal }) => { await fetch(`https://example.com/${payload.id}`, { signal }); }, }); ``` `await job.cancel(executionId)` cancels a pending, scheduled or running execution; pass `{ cancelledBy, cancelledByName }` to say who did, the admin shows it. `await job.cancelByKey(key)` cancels the execution parked under a `key` and returns its id, or `null` when nothing is parked. It is the disarm half of a keyed push: an order that pays cancels the cart's reminder sequence by the cart id. Only a `pending` or `scheduled` row is cancelled; a `running` one is left to finish, because a listener reacting to an event cannot know whether that event is the running handler's own doing (a reconciliation stage that settles a checkout emits the very event that would cancel it). The handler's own re-check at its next stage is the right place for that decision. ## Dispatch modes How a pushed execution reaches its handler depends on which modules are loaded: - **direct** (default): the handler runs in-process right after `push()` returns. The outbox row is the durability guarantee: if the process dies, the sweep re-dispatches. Best for single-instance Node and Cloudflare Workers, where standing up a broker is overkill. - **queue**: add `AlephaApiJobsQueue` and dispatch goes through a real broker (Cloudflare Queues on workerd, Redis on Node and Bun once `REDIS_URL` is set) so a worker pool consumes the work. ```typescript import { AlephaApiJobs, AlephaApiJobsQueue } from "alepha/api/jobs"; alepha.with(AlephaApiJobs).with(AlephaApiJobsQueue); ``` Both modes are at-least-once. Write handlers to be idempotent. ## Retention A job keeps a history of its runs in `job_executions`, and `retention` says how much of it, per status: `ok` for successes, `error` for failures, which cancelled runs follow. ```typescript check import { $job } from "alepha/api/jobs"; class Reminders { send = $job({ name: "reminders.send-due", description: "Sends the reminders that are due.", cron: "0 0 * * *", retention: { ok: { last: 7 }, // the last 7 successes error: { days: 30 }, // 30 days of failures }, handler: async () => { // send the reminders that are due }, }); } ``` Each status takes `{ last?, days? }` or `false`: | Rule | Keeps | | ------------------- | ------------------------------------------------------- | | `{ last: 7 }` | The newest 7 rows | | `{ days: 30 }` | Rows that completed within the last 30 days | | `{ last, days }` | Rows within both limits: a row goes when it breaks one | | `{ days: () => n }` | A window read at every trim tick, for a runtime setting | | `false` | Nothing: no row is written for that outcome | `last` is an integer of at least 1 and `days` is positive; a rule with neither is refused at registration. There is no "forever". A status the job leaves out takes a default, decided by how often the job runs. A cron's cadence is the shortest gap among its next 10 fire dates, so `0 9 * * 1-5` is a daily job: | Job | Successes | Failures | | ------------------------------- | ------------ | -------- | | cron, at least every 15 minutes | last 12 | 30 days | | cron, up to hourly | last 24 | 30 days | | cron, up to daily | last 7 | 30 days | | cron, slower | last 5 | 30 days | | queue | not recorded | 30 days | Successes are kept by count and failures by age: the last 12 runs of a `*/15` cron are three hours of history, and the last 7 of a daily one are a week. A queue job's successes are not recorded unless it says so, in days: for per-item work, "the last 10" means the last 10 items whatever the volume. For a cron the newest row of each status outlives its window, so a monthly job never looks like it never ran. Every kept run keeps the log entries it produced (up to `logMaxEntries`), whatever its outcome. Nothing is written while a run is in progress: a queue job's row says `running`, and a cron tick has no row until it ends. The trim runs on its own cron (`trimCron`, hourly by default), orders by `completedAt`, and costs one grouped query for the whole tick, so a status that is within its limits is never queried individually. A status over its limits is cut back in chunks, however far over it is; if one tick cannot finish, it logs what is left and the next tick continues. The effective rule of every job, defaults filled in, is in the admin's job list and in devtools, with where it came from. ### A rename loses the history The trim also deletes every row whose job name no job declares, whatever its status. The name is a job's identity in `job_executions`: rename a job and its old rows, pending and scheduled ones included, are gone at the next trim tick. Any process pointed at the same database with a different job set does the same to the jobs it does not declare, so give each application its own database. A container that declares no job at all purges nothing. ## Configuration Tune the `jobConfig` atom: ```typescript import { jobConfig } from "alepha/api/jobs"; alepha.store.mut(jobConfig, (c) => ({ ...c, sweepCron: "*/5 * * * *" })); ``` **Mutate before you wire the module.** A cron expression is read once, when the `$job` field initializes, and wiring a module injects its services immediately. A mut applied after the module is wired lands in the store but never reaches the already-registered cron - no error, no effect: ```typescript // Works - the store is set before anything reads it. const alepha = Alepha.create(); alepha.store.mut(jobConfig, (c) => ({ ...c, sweepCron: "*/5 * * * *" })); alepha.with(MyApp); // Silently does nothing to the schedule. const alepha = Alepha.create().with(MyApp); alepha.store.mut(jobConfig, (c) => ({ ...c, sweepCron: "*/5 * * * *" })); ``` Inside a `$module`, the `register()` hook runs before `imports[]` and `services[]`, so it is also a safe place to do this. | Key | Default | Description | | ---------------------- | -------------- | ----------------------------------------------------------------------------------------------------- | | `sweepCron` | `*/15 * * * *` | Reconciliation sweep - bounds retry latency | | `trimCron` | `0 * * * *` | Trim tick: purges unregistered jobs' rows, enforces retention | | `sweepBatchSize` | `200` | Rows one sweep phase reads per tick - see below | | `maxRedispatch` | `3` | Lost deliveries tolerated before a `pending` row is failed | | `retryBackoffBase` | `5000` | First retry's backoff ceiling (ms); doubles per attempt, full jitter | | `retryBackoffMax` | `1800000` | Ceiling for that curve (ms) | | `staleThreshold` | `300000` | Pending age (ms) before the sweep re-dispatches | | `runTimeout` | `1800000` | Running age (ms) before a crash is assumed | | `retention.errorDays` | `30` | Days of failures a job keeps when it declares no `retention.error` | | `retention.maxRows` | `1000` | Cap on rows one status keeps under a default rule; a declared rule is never capped | | `drainTimeout` | `30000` | Time (ms) to wait for in-flight jobs on shutdown | | `logMaxEntries` | `100` | Log lines captured per run | | `directMaxConcurrency` | `10` | Concurrent handlers in direct mode - what keeps a `pushMany` of thousands from exhausting the DB pool | ### The sweep is bounded Each sweep phase reads at most `sweepBatchSize` rows per tick and leaves the rest for the next one. This matters under exactly the conditions where the sweep is load-bearing: a downstream outage turns the entire retrying population into rows the sweep matches at once, and every row carries its payload and any captured logs. A phase that fills its batch logs that it did, so a backlog is visible in the logs rather than something you infer from a graph. Every phase's action moves the row out of the status that phase owns, so progress across ticks is guaranteed. Every phase serves its oldest rows first, so under a backlog work is picked up in the order it has been waiting. There is no priority: a job cannot jump the queue. A `pending` row whose delivery is lost is re-dispatched at most `maxRedispatch` times before it is failed. This is counted separately from `attempt`, which only moves when a worker actually claims the row: a payload that kills the process between dispatch and claim never increments `attempt`, so the retry policy would never end it. ### Sweeps owned by other modules Modules that ship their own crons expose them the same way. All default to `*/15 * * * *` so they collapse onto the jobs sweep's trigger instead of adding their own - which matters on Cloudflare, where each distinct expression costs a Cron Trigger. | Atom | Key | Default | Bounded by | | ---------------------------------------------- | ------------------------ | -------------- | ---------------------------------------------------- | | `paymentsConfig` (`alepha/api/payments`) | `expireStaleIntentsCron` | `*/15 * * * *` | The 30-minute intent cutoff | | `checkoutConfig` (`@alepha/commerce/checkout`) | `stockSweepCron` | `*/15 * * * *` | Nothing - `reserved()` excludes holds by `expiresAt` | ## Multi-replica deployments Cron-mode jobs take a distributed lock per tick (`lock: true` by default), so a fleet of replicas fires the handler once, not once per replica. This needs a real `LockProvider` - the default `MemoryLockProvider` is per-process. See [Bare metal deployment](/docs/guides-deployment-bare) for the setup. The unit that is claimed is the **schedule instant**, not the job. Every replica derives the same instant from the same cron expression, so the first one there claims it and the others stand down - including a replica whose clock lags by a few milliseconds and arrives after the handler has already finished. That claim is what makes "once per tick" hold for a job with `retry`, where the tick only writes an outbox row and is over in a millisecond. A manual `trigger()` is not a scheduled instant, so it is never suppressed by one. It still never overlaps a run of the same job already in progress, whether on this replica or, through the per-job lock, on another: it returns without running instead. It resolves either way, so a trigger from the admin that lands on a tick is a no-op, not a second run. With `retry`, the run it cannot overlap is the tick writing its outbox row; the handler runs afterwards from the outbox, outside that guard. `lock` has no effect on queue-mode and direct-mode jobs. Those serialize through the outbox `claim()` UPDATE-guard instead, which is always on. ## Events `$job` emits lifecycle events you can hook: | Event | Payload | | ------------- | ------------------------------ | | `job:begin` | `{ name, now, executionId }` | | `job:success` | `{ name, executionId }` | | `job:error` | `{ name, error, executionId }` | | `job:cancel` | `{ name, executionId }` | | `job:end` | `{ name, executionId }` | ## When not to use `$job` - **A bare periodic tick with no database.** `$job` needs an ORM connection. For a tick without one, register directly against the cron engine: ```typescript import { CronProvider } from "alepha/scheduler"; class Ticker { protected readonly cron = $inject(CronProvider); protected readonly setup = $hook({ on: "start", handler: () => { this.cron.createCronJob("revalidate", "0 * * * *", async () => { // ... }); }, }); } ``` You get the tick, but no distributed lock - on multiple replicas every replica fires - no run history, no retry and nothing in the admin UI. Reach for it only when a database is genuinely unavailable. - **A fixed interval rather than a schedule.** `$interval` from `alepha/datetime` runs a function every N units of time, starting when the container starts and stopping when it stops: ```typescript check import { $interval } from "alepha/datetime"; class Poller { poll = $interval({ duration: [30, "seconds"], handler: async () => { // ... }, }); } ``` It is `setInterval` with the container's lifecycle and the testable clock attached, so `travel()` advances it. A throwing tick is logged and the next tick still runs, which is the one thing raw `setInterval` gets dangerously wrong: an unhandled rejection there takes the process down. Same caveats as the cron engine above, plus one more. "Every 30 seconds" means every 30 seconds _of this process's uptime_, so a restart resets the phase and two replicas drift apart. Use it for things where that does not matter - refreshing an in-memory cache, emitting a gauge - and use `$job` for anything where a missed or doubled run is a business problem. - **Fan-out to many subscribers.** Use `$topic` / `$subscriber`, which is publish/subscribe rather than work distribution.