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

1import { $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:

1import { AlephaApiJobs } from "alepha/api/jobs";2 3alepha.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.

 1import { z } from "alepha"; 2import { $job } from "alepha/api/jobs"; 3  4class Emails { 5  // queue-mode - call push() to enqueue work 6  welcome = $job({ 7    name: "emails.send-welcome", 8    description: "Sends the welcome email to a new user.", 9    schema: z.object({ userId: z.text() }),10    retry: { retries: 3 },11    handler: async ({ payload, attempt }) => {12      // send the welcome email for payload.userId13    },14  });15 16  // cron-mode - fires on a schedule, no payload17  digest = $job({18    name: "emails.send-digest",19    description: "Builds and sends the daily digest.",20    cron: "0 8 * * *",21    handler: async () => {22      // build and send the daily digest23    },24  });25}

To run scheduled work over a set of payloads, compose the two: a cron job that pushes, and a queue job that handles.

 1class Reminders { 2  sweep = $job({ 3    name: "reminders.sweep", 4    description: "Pushes a reminder for every row that is due.", 5    cron: "0 * * * *", 6    handler: async () => { 7      const due = await this.repository.findMany({ where: { due: true } }); 8      await this.remind.pushMany( 9        due.map((row) => ({ payload: { id: row.id } })),10      );11    },12  });13 14  remind = $job({15    name: "reminders.send",16    description: "Sends one reminder.",17    schema: z.object({ id: z.text() }),18    handler: async ({ payload }) => {19      /* ... */20    },21  });22}

#Naming

Every job declares a name and a description.

• The name is <domain>.<action> 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.<domain>.<action> 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). Get the name right when you declare it.

#Pushing work

1const executionId = await this.welcome.push({ userId: "u1" });

push() accepts a second options argument:

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

 1reconcile = $job({ 2  name: "checkouts.reconcile", 3  description: "Settles a checkout the payment provider has confirmed.", 4  schema: z.object({ sessionId: z.uuid() }), 5  retry: { 6    retries: 3, 7    backoff: { initial: [1, "minute"], factor: 4, max: [30, "minute"] }, 8  }, 9  handler: async ({ payload }) => {10    await this.payments.sync(payload.sessionId);11  },12});

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:

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

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:

 1cartRecovery = $job({ 2  name: "carts.recover", 3  description: 4    "Reminds a buyer about an abandoned cart, then marks it abandoned.", 5  schema: z.object({ 6    cartId: z.uuid(), 7    stage: z.enum(["remind", "remindAgain", "abandon"]).optional(), 8  }), 9  retry: { retries: 3, backoff: { initial: [1, "minute"], factor: 4 } },10  handler: async ({ payload, reschedule }) => {11    if (!(await this.carts.isRecoverable(payload.cartId))) {12      return; // converted or gone: the sequence ends here13    }14    switch (payload.stage ?? "remind") {15      case "remind":16        await this.mailer.remind(payload.cartId, 1);17        reschedule({18          delay: [23, "hour"],19          payload: { ...payload, stage: "remindAgain" },20        });21        return;22      case "remindAgain":23        await this.mailer.remind(payload.cartId, 2);24        reschedule({25          delay: [24, "hour"],26          payload: { ...payload, stage: "abandon" },27        });28        return;29      case "abandon":30        await this.carts.markAbandoned(payload.cartId);31    }32  },33});34 35// One push starts the sequence; the key makes a second push land on it.36await 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 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.

1await myJob.push(payload, { inline: true });2// resolves -> the handler ran to completion, outbox row terminal3// 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.

1await myJob.push(payload); //                 async, retries per the policy2await 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:

1report = $job({2  name: "reports.build",3  description: "Builds one report.",4  schema: z.object({ id: z.text() }),5  timeout: [30, "seconds"],6  handler: async ({ payload, signal }) => {7    await fetch(`https://example.com/${payload.id}`, { signal });8  },9});

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.

1import { AlephaApiJobs, AlephaApiJobsQueue } from "alepha/api/jobs";2 3alepha.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.

 1import { $job } from "alepha/api/jobs"; 2  3class Reminders { 4  send = $job({ 5    name: "reminders.send-due", 6    description: "Sends the reminders that are due.", 7    cron: "0 0 * * *", 8    retention: { 9      ok: { last: 7 }, // the last 7 successes10      error: { days: 30 }, // 30 days of failures11    },12    handler: async () => {13      // send the reminders that are due14    },15  });16}

Each status takes { last?, days? } or false:

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:

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:

1import { jobConfig } from "alepha/api/jobs";2 3alepha.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:

1// Works - the store is set before anything reads it.2const alepha = Alepha.create();3alepha.store.mut(jobConfig, (c) => ({ ...c, sweepCron: "*/5 * * * *" }));4alepha.with(MyApp);5 6// Silently does nothing to the schedule.7const alepha = Alepha.create().with(MyApp);8alepha.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.

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

#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 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:

#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

  Copy

   1import { CronProvider } from "alepha/scheduler"; 2  3class Ticker { 4  protected readonly cron = $inject(CronProvider); 5  6  protected readonly setup = $hook({ 7    on: "start", 8    handler: () => { 9      this.cron.createCronJob("revalidate", "0 * * * *", async () => {10        // ...11      });12    },13  });14}

  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

  Copy

   1import { $interval } from "alepha/datetime"; 2  3class Poller { 4  poll = $interval({ 5    duration: [30, "seconds"], 6    handler: async () => { 7      // ... 8    }, 9  });10}

  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.