#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    schema: z.object({ userId: z.text() }), 8    retry: { retries: 3 }, 9    handler: async ({ payload, attempt }) => {10      // send the welcome email for payload.userId11    },12  });13 14  // cron-mode - fires on a schedule, no payload15  digest = $job({16    cron: "0 8 * * *",17    handler: async () => {18      // build and send the daily digest19    },20  });21}

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    cron: "0 * * * *", 4    handler: async () => { 5      const due = await this.repository.findMany({ where: { due: true } }); 6      await this.remind.pushMany( 7        due.map((row) => ({ payload: { id: row.id } })), 8      ); 9    },10  });11 12  remind = $job({13    schema: z.object({ id: z.text() }),14    handler: async ({ payload }) => {15      /* ... */16    },17  });18}

#Pushing work

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

push() accepts a second options argument:

pushMany() takes an array of { payload, key?, delay?, priority?, scheduledAt? } and writes them in a batched INSERT.

#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 (record: "error", the queue-mode 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  schema: z.object({ sessionId: z.uuid() }), 3  retry: { 4    retries: 3, 5    backoff: { initial: [1, "minute"], factor: 4, max: [30, "minute"] }, 6  }, 7  handler: async ({ payload }) => { 8    await this.payments.sync(payload.sessionId); 9  },10});

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  schema: z.object({ 3    cartId: z.uuid(), 4    stage: z.enum(["remind", "remindAgain", "abandon"]).optional(), 5  }), 6  retry: { retries: 3, backoff: { initial: [1, "minute"], factor: 4 } }, 7  handler: async ({ payload, reschedule }) => { 8    if (!(await this.carts.isRecoverable(payload.cartId))) { 9      return; // converted or gone: the sequence ends here10    }11    switch (payload.stage ?? "remind") {12      case "remind":13        await this.mailer.remind(payload.cartId, 1);14        reschedule({15          delay: [23, "hour"],16          payload: { ...payload, stage: "remindAgain" },17        });18        return;19      case "remindAgain":20        await this.mailer.remind(payload.cartId, 2);21        reschedule({22          delay: [24, "hour"],23          payload: { ...payload, stage: "abandon" },24        });25        return;26      case "abandon":27        await this.carts.markAbandoned(payload.cartId);28    }29  },30});31 32// One push starts the sequence; the key makes a second push land on it.33await 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. record and keep only apply 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  schema: z.object({ id: z.text() }),3  timeout: [30, "seconds"],4  handler: async ({ payload, signal }) => {5    await fetch(`https://example.com/${payload.id}`, { signal });6  },7});

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, Redis) 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

Queue-mode jobs default to record: "error" - the pending row is written at push time and removed on success, so a healthy queue leaves no rows behind. Cron jobs default to record: "all" with one retained success so the admin "Last run" column is accurate.

Note the deliberate asymmetry: per-job keep.ok: 0 means never trim, while the global keepLastSuccess: 0 means delete on success.

Trim runs on its own cron (trimCron, hourly by default) and costs one grouped count for the whole tick, so a job whose buffer is already at its limit is never queried individually. A buffer that is over its limit is emptied back down in chunks, however far over it is; if one tick cannot finish the job it logs what is left and the next tick continues.

A cron whose buffer is exactly one row (the default) updates that row rather than inserting a new one for the trim to delete later. A */15 cron used to write 96 rows a day so that trim could remove 95 of them, purely to keep one timestamp current. Jobs keeping more than one row still insert, since there the rows are the history.

#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. What repeats is the priority ordering: while a backlog persists, newly arriving critical work is served before low work that has been waiting. That is what $job priority means, and it is the only thing it means.

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 takes the per-job lock, and so cannot overlap a running tick.

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.