elements
Manual Jobs

Jobs

elements man jobs Read as markdown

A job is a unit of asynchronous work that runs on a worker process pulled from a queue. Jobs exist to take long-running or expensive work off your app's web request path. The web server stays responsive while a separate worker pool churns through the queue in the background. Scale the worker pool to scale throughput.

Elements ships with jobs and cron built in. Most applications need them: emails to send, payments to charge, exports to generate, digests to deliver. The project server boots a job server automatically, both locally and on every machine after deploy.

The job queue lives in the same Postgres database as the rest of your app, in the elements.jobs table. Because the queue is in your database, enqueueing a job inside tx() joins the surrounding transaction. The job row and your other writes commit together or not at all. There is no second system that can drift out of sync with your data.

Jobs are scheduled. Schedule one immediately and the next available worker picks it up, or schedule one for a future time with a plain-English time string.

new SendWelcomeJob({ userId, to, name }).schedule();
new SendFollowupEmailJob({ userId }).schedule("in 3 days");
new RetryPaymentJob({ orderId }).schedule("in 1h");
new ExpirePromoJob({ promoId }).schedule("tomorrow at 8am");
new WeeklyReminderJob({ userId }).schedule("next monday");

Cron handles the other half of background work: things that run on a schedule, not in response to a user action. Send a daily digest at 9am, score posts every five minutes, archive old rows nightly. app.cron(...) is declared in the project's root index.ts and runs on whichever app machine is currently the leader. Postgres arbitrates the leader election so exactly one machine fires each tick.

Cron and jobs compose. The cron tick is the trigger; a job does the work. A typical cron callback is one line: enqueue a job and return. That keeps the cron loop free to fire the next tick on time, and lets the work itself benefit from job retries, timeouts, and worker-pool scaling.

app.cron("every day at 9am", "daily digest", () => new SendDailyDigestJob().schedule());
app.cron("every 5m", "rank posts", () => new RankPostsJob().schedule());
app.cron("every day at 2am", "archive old rows", () => new ArchiveOldRowsJob().schedule());

At a Glance

// app/jobs/send-welcome/index.ts
import { Job, email } from "@elements/app";
import WelcomeEmail from "#app/emails/welcome";

export interface SendWelcomeJobFields {
  userId: string;
  to: string;
  name: string;
}

export class SendWelcomeJob extends Job<SendWelcomeJobFields> {
  static maxAttempts = 5;
  static timeoutMs = 60_000;

  run() {
    let { to, name } = this.fields;
    email({
      to,
      subject: "welcome",
      body: new WelcomeEmail({ name }),
    });
  }
}

Schedule it from an @rpc. Use tx() to make the user insert and the job enqueue atomic:

// app/shared/services/users.ts
import { tx, sql, ValidationError } from "@elements/app";
import { SendWelcomeJob } from "#app/jobs/send-welcome";

interface SignupForm {
  name: string;
  email: string;
  password: string;
}

/** @rpc */
export function signup(form: SignupForm): User {
  if (!form.email.includes("@")) {
    throw new ValidationError("invalid email");
  }
  if (!sql(`select 1 from users where email = ${form.email}`).empty()) {
    throw new ValidationError("email taken");
  }

  return tx(() => {
    let user = sql<User>(`
      insert into users (name, email, passwordHash)
      values (${form.name}, ${form.email}, crypt(${form.password}, gen_salt('bf', 12)))
      returning *
    `).firstOrThrow("insert returned no row");

    new SendWelcomeJob({ userId: user.id, to: user.email, name: user.name }).schedule();

    return user;
  });
}

What just happened:

  • The user row and the elements.jobs row land in the same transaction.
  • If the transaction rolls back (validation failure, db error, anything thrown), there's no half-state. No user without a job. No job without a user.
  • The job's pg_notify fires only on commit. Workers can't pick up a job whose enqueuing transaction hasn't committed.

The Job Class

export abstract class Job<TFields extends object = {}> {
  static maxAttempts: number;     // default 3
  static timeoutMs: number;       // default 30000
  static priority: number;        // default 100, 1 = top, higher = lower

  id: string;                     // uuid, set by Elements
  fields: TFields;                // typed payload
  scratch: string;                // absolute path to a per-job scratch dir

  abstract run(): void;
  schedule(arg?: string | JobScheduleArg): string;    // sync-style; scheduleAsync returns Promise<string>
}

Override run() with the work. Throws cause the job to be retried with exponential backoff, up to maxAttempts times.

Per-class statics override the defaults:

export class HeavyImportJob extends Job<{ url: string }> {
  static maxAttempts = 1;
  static timeoutMs = 600_000;
  static priority = 50;

  run() { /* ... */ }
}

this.fields is the typed payload. this.id is the job uuid. this.scratch is an absolute path to a private directory created before run() and removed after (success or crash).

Scheduling

new SendWelcomeJob({ to, name }).schedule();
new SendWelcomeJob({ to, name }).schedule("in 1h");
new SendWelcomeJob({ to, name }).schedule({ at: "tomorrow at 8am" });
new SendWelcomeJob({ to, name }).schedule({ priority: 1 });
new SendWelcomeJob({ to, name }).schedule({
  at: "next monday",
  idempotencyKey: `welcome:${userId}`,
});

schedule() returns the job's uuid (a string).

JobScheduleArg:

  • at: when to become eligible. Time string. Omit for immediate.
  • priority: 1-N. 1 = top. Default 100. Overrides static priority.
  • idempotencyKey: dedupe key. At most one row per non-null key. Re-scheduling with the same key returns the existing id.

Time strings for at:

in 5m, in 1h, in 3 days, in 2 weeks
next monday, next tuesday, ... (case-insensitive, full weekday names)
tomorrow, tomorrow at 8am
next week

Transactional Scheduling

schedule() runs INSERT INTO elements.jobs ... SELECT pg_notify(...). When called inside tx(), both statements join the surrounding transaction. The pg_notify fires on commit, so the worker only observes the new row after the enqueuing transaction commits. If the transaction rolls back, the row is never visible to anyone.

tx(() => {
  let order = sql<Order>(`insert into orders (...) values (...) returning *`).firstOrThrow("insert returned no row");
  new ChargeCardJob({ orderId: order.id }).schedule();
  new SendReceiptJob({ orderId: order.id }).schedule();
});

All three writes commit together or not at all.

Cancel

import { Job } from "@elements/app";

let id = new SendWelcomeJob({ ... }).schedule("in 1h");
let cancelled = Job.cancel(id);

Job.cancel(id) returns true if the job moved from pending to cancelled. Returns false if the job is already running, completed, failed, cancelled, or missing.

Retries

A throw from run() retries the job with exponential backoff up to maxAttempts times. After the last failure, the job lands in failed state. Failed rows stay in elements.jobs for inspection.

The timeoutMs static caps how long a single attempt can run. A worker that exceeds it is killed and the job retries. Set timeoutMs higher than the longest expected run, with margin.

Scratch

Each job invocation gets a private directory at this.scratch, created before run() and removed after run() returns or throws. Use it for temp downloads, intermediate files, and similar throwaway state.

File Layout

A job is a folder under app/jobs/<name>/ scaffolded with elements create job:

elements create job send-welcome

app/jobs/send-welcome/
  index.ts          # exports the Job class
  test.ts           # tests

The class is automatically tagged with the @job build tag, which sets the path static Elements needs to resolve and instantiate the class on dequeue.

Cron vs Jobs

  • Cron triggers on time; jobs trigger on code.
  • Cron has no persistence; jobs persist in elements.jobs.
  • Cron has no retries; jobs have maxAttempts with exponential backoff.
  • Cron has no typed payload; jobs have a typed fields object.
  • Cron runs on the elected leader app process; jobs run on a worker process.
  • Cron is for "fire this at a time." Jobs are for "do this work, durably."

Cron + Jobs Together

Keep the cron callback small. Enqueue a job and return.

// index.ts
app.cron("every 5m", "rank posts", () => new RankPostsJob().schedule());
app.cron("every day at 2am", "archive old rows", () => new ArchiveOldRowsJob().schedule());
app.cron("every monday at 9:30am", "weekly digest", () => new WeeklyDigestJob().schedule());

Why:

  • The cron loop runs in the app process. The work runs in the worker. Long-running work in a cron callback would block subsequent ticks.
  • Jobs have retries, timeouts, and idempotency keys. Cron callbacks don't.
  • Cron leader election and the job queue compose cleanly on the same db.

Retention

Elements owns retention of elements.* tables. Success rows in cron_runs and jobs are pruned to a 24-hour sliding window. Failed rows are kept forever by default so you never miss an intermittent error. Both windows are configurable per area in config.jsoc. Prune runs once per day in an off-peak window (02:00–04:00 utc) in chunked transactions that yield to live traffic. You never need to run delete on these tables yourself.

{
  jobs: {
    retention: {
      success: '24h',    // delete completed/cancelled jobs after 24h
      error:   'never',  // keep failed jobs forever
    },
  },
  crons: {
    retention: {
      success: '24h',
      error:   'never',
    },
  },
}

Values are duration strings ('24h', '7d', '30d') or the literal 'never'. The two areas are configured independently. Defaults are success: '24h', error: 'never'. Healthy apps have a near-zero error rate, so failed-row growth stays tiny. Keeping failed rows is the safer default: an intermittent failure you never see is the worse outcome.

Prune is implemented as chunked delete in batches of 1000 rows, each batch its own short transaction with a 500ms lock_timeout and 2s statement_timeout. A contended batch bails and retries on the next daily run. Live inserts never queue behind the delete. The prune cron entries are registered automatically as @elements/app:prune-cron-runs and @elements/app:prune-jobs and show up alongside your own entries in select * from elements.cron_runs.

Cron Leader Election

Only one app machine runs each cron tick. Every app process tries to acquire the tick lease via a single-row update on elements.cron_lease against the same minute boundary. Postgres arbitrates. Exactly one machine wins. Every other machine's update returns zero rows and does nothing.

Each cron entry's cronFiresAt(schedule, tickBoundary) decides whether it's due this minute. Due entries run sequentially within the leader. Each run is recorded in elements.cron_runs (state, duration, error if any).

Cron uses the database for leader election (elements.cron_lease) and run history (elements.cron_runs). app.cron(...) throws at app start if no database is configured.

The Worker

Worker processes connect to the same elements.jobs table. They pull rows, instantiate the job class via the path static the @job transform set, populate id, fields, and scratch, and call run(). Workers run independently. Add more workers (per machine or per cluster) to scale throughput.

Debugging

  • select * from elements.jobs where state = 'failed' order by updated_at desc;: failed jobs.
  • select * from elements.cron_runs order by ran_at desc limit 50;: cron run history.
  • Per-job scratch directories live under the worker's scratch root and are removed on completion.

Related

  • database: the transactional tx() boundary the queue rides on.
  • channel: the pub/sub layer for in-app realtime; jobs use a similar pg_notify/LISTEN pattern internally.
  • email: jobs are the right place to send transactional email.