Jobs: how does work run after the call, or on a schedule?

A job is work that should not happen inside the call that caused it — sending a message, calling a slow provider, rebuilding a report — or that happens on a schedule. Jobs live in the database (dw_job, dw_recurring_job), so they survive restarts, join the transaction that enqueues them, and are claimed by exactly one worker even when several processes run.

Declaring jobs

A job has two halves. What it is — its name and how its payload travels as JSON — is a DwJobKind<P>, a constant the places that enqueue it import. How it runs is a DwQueuedJob<P> in the job list of the feature it belongs to, often built from a service instance the enqueuing code never sees. Both live in the feature's <feature>_jobs.dart:

// lib/src/invoices/invoices_jobs.dart
abstract final class InvoicesJobs {
  static const send = DwJobKind<({int invoiceId})>(
    'invoices.send',
    encode: _encode,
    decode: _decode,
  );

  static Map<String, Object?> _encode(({int invoiceId}) p) => {'invoiceId': p.invoiceId};
  static ({int invoiceId}) _decode(Map<String, Object?> json) =>
      (invoiceId: json['invoiceId']! as int);
}

final invoicesJobs = <DwJobDefinition>[
  DwQueuedJob(
    InvoicesJobs.send,
    handle: (ctx, p) async => ctx.log.info('sending invoice ${p.invoiceId}'),
  ),
  DwRecurringJob(
    'invoices.mark_overdue',
    every: const Duration(hours: 1),
    handle: (ctx) => ctx.db.invoices.updateWhere(
      where: (t) =>
          t.dueAt.lt(ctx.now) & t.paidAt.isNull() & t.overdue.equals(false),
      set: (t) => [t.overdue.set(true)],
    ),
  ),
];

passed as that feature's DwServerFeature(jobs: invoicesJobs) — there is no app-wide job list. The kinds' class is <Feature>Jobs, the list <feature>Jobs, and a job's name '<feature>.<snake_case>'. The example's bookings/bookings_jobs.dart is a worked one: BookSession enqueues a reminder two hours before the session, and the job decides when it runs whether the booking is still active.

DwJobKind<P>(name, {encode, decode}) — the one place a payload is spelled as a map: encode runs at the enqueue, decode before the handler, and both sides see P (a record, a class, int). DwJobKind.withoutPayload(name) is a DwJobKind<void>, enqueued with null.

DwQueuedJob<P>(kind, {handle, transactional, maxAttempts, backoff, lease}) — a job that runs once per enqueue:

Parameter	Default	Meaning
handle	required	Future<void> Function(DwCallContext ctx, P payload)
transactional	true	run the handler in the transaction that claimed the row (below)
maxAttempts	5	attempts before the job is marked failed
backoff	dwDefaultJobBackoff	the delay before the next attempt, by attempt number: 10 s, 20 s, 40 s … capped at one hour
lease	5 min	how long a non-transactional job is reserved for its worker

DwRecurringJob(name, {every, handle}) — a job that runs every every, with handle: Future<void> Function(DwCallContext ctx).

The server refuses to start on a name starting with dw. (the framework's), a name declared twice, a non-positive interval or fewer than one attempt. A payload is not a DTO of the protocol (D-016): a job is server-to-server, and its type never reaches the app's registry (D-092).

ctx.jobs.enqueue

Future<bool> enqueue<P>(
  DwJobKind<P> job,
  P payload, {
  DateTime? runAt,
  String? key,
});

await ctx.jobs.enqueue(InvoicesJobs.send, (invoiceId: invoice.id));

• The row is written through ctx.db, so an enqueue joins the enclosing transaction: inside a transactional command the job exists only if the command commits, and a refusal or a failure leaves no job behind. Outside a transaction it is written at once.
• runAt schedules it, an instant — ctx.now.add(…); ctx.now when omitted.
• key deduplicates: while a job with the same key is pending or running, another enqueue with it does nothing and answers false. A job that ran out of attempts no longer blocks its key. Once a job has succeeded its row is gone, so the key is free again — idempotence for work that must happen once ever belongs in the project's own rows.
• A kind the server declares no job for throws ArgumentError, and a payload that is not JSON fails jsonEncode, both at the call site rather than in a worker later. Recurring jobs cannot be enqueued.

How jobs run

Each server process runs DwServerSettings.jobWorkers workers (2 by default). A worker takes the next due recurring job, else the next due queued job, claiming its row with FOR UPDATE SKIP LOCKED — so any number of workers in any number of processes never run one row twice. It wakes on a LISTEN/NOTIFY signal sent when a job is enqueued (delivered at commit), at the next due time, and at least every jobPollInterval in case a notification was lost. jobWorkers: 0 runs no jobs in that process.

The context of a job has no caller: accountId, sessionKey and callerUtcOffset are null. It may publish; a transactional job's publications are delivered after it commits.

Time is the server's clock

Every due time of the queue — a job's default runAt, whether it is due, its retry after a failure, a non-transactional job's lease, a recurring job's next run — is read from the server's clock (DwAppServer(clock: …)), the same one a job's handler reads as ctx.now; not from the database's now(). A job is due exactly when ctx.now says so.

In a test that is the point: with a DwTestClock, moving the clock makes due whatever it passed and wakes the workers, so a reminder queued for tomorrow runs now, and its handler sees tomorrow:

final clock = DwTestClock(DateTime.utc(2026, 9, 29, 8));
// … the server built with `clock: clock`, a session booked for tomorrow at 10:00 …
clock.moveTo(DateTime.utc(2026, 9, 30, 10, 5)); // the queue fell behind
// the reminder runs, sees the session already started, and sends nothing

The example's test/src/core/push_test.dart does exactly this. A clock that stands still has a consequence: a retry after its backoff, a non-transactional job's expired lease and a recurring job's next run wait until the test moves the clock past them. What the framework stamps in the database itself — created_at, session keys and sign-in codes with their expiry, command outcomes — stays on the database's clock and keeps real time.

Modules that queue work of their own keep it on the same clock: push deliveries are due, leased, retried and finished by it (push delivery), so the job covering a delivery never runs before the delivery is due. A clock of a project's own is a DwServerClock — now(), and jumps, the event a server wakes its job executor on.

Transactional jobs (the default)

The handler runs in a savepoint of the transaction that claimed the row, and the row is deleted in that same transaction. So the job disappears exactly when its work commits: a crash mid-job rolls everything back and leaves the job pending, as if it never started.

A deadlock or a serialization failure in the handler re-runs it in place, up to three times in all, before an attempt is spent — the conflict rolls the savepoint back, so the retry starts clean. On any other throw, or a conflict lost three times, the savepoint rolls back, and the attempt, the error's text (last_error) and the next run time — backoff(attempt) from now — are written under the same row lock. Intermediate failures are logged as warnings. The last attempt sets failed_at and alerts (alerts). A failed row stays in dw_job for the operator — nothing retries it — until dw.cleanup removes it DwServerSettings.failedJobRetention (30 days) after it failed.

Non-transactional jobs

transactional: false is for handlers that call external services, which must not hold a transaction open for seconds or be undone by a rollback. The claim commits first, with the attempt counted and the row leased for lease; the handler then runs on the pool, and the row is deleted when it returns. A crash mid-job leaves the lease to expire, and the job runs again.

So a non-transactional job may run twice, and a handler running longer than its lease can run twice at once — another worker claims the row when the lease expires. Make such handlers idempotent, and give them a lease longer than they take.

Its ctx.transaction is retried on a deadlock or a serialization failure, like every top-level transaction (database): keep the external call outside it. What such a handler publishes inside a ctx.transaction goes out when that transaction commits, not when the job ends — a status ("analysing") reaches subscribers before the slow call it announces. Publications outside any transaction go when the handler returns.

Which attempt

ctx.job is the run a handler belongs to: attempt (from 1), maxAttempts and isLastAttempt. A handler that must tell the people waiting on it that it gave up does so when isLastAttempt is true and it is about to fail. The delay before the next attempt is DwQueuedJob(backoff: (attempt) => …); outside a job, ctx.job is null.

Recurring jobs

A recurring job's next run time lives in dw_recurring_job, so a restart neither skips nor repeats a run. At start the server syncs the table with the declared jobs: a shortened interval pulls the next run in, a lengthened one keeps the run already due, and a job no longer declared is removed.

A run executes in the transaction that claimed it. After a long outage the job runs once, not once per missed slot, and its next run is the first slot of its schedule after now. A deadlock or a serialization failure re-runs the handler in place, as for a queued job. Any other failure alerts, stores its text in last_error, and waits for the next slot: a recurring job has no retries — its next run is the retry.

Two versions at once

During a deployment the new server starts beside the old one on the same tables. A process claims only the jobs it declares: a recurring job the new version added — inserted already due — and a job of a kind only the new code enqueues wait for a process that declares them, instead of stopping the old worker or being marked failed by it. If the new version never takes over (its health check failed and the deployment stopped), such rows simply wait; the old server logs a warning at start for queued jobs no job of it declares.

The framework's jobs

Job	Kind	What it does
dw.cleanup	recurring, hourly	deletes command outcomes older than DwServerSettings.commandOutcomeRetention, code tickets past their use, session keys revoked more than a day ago, and jobs that failed longer ago than DwServerSettings.failedJobRetention
dw.files.cleanup	recurring, every DwFileStorage.cleanupInterval	removes unfinished uploads past their ticket and grace, object first (uploads)
dw.files.deleteObject	queued, non-transactional, 10 attempts	deletes a file's object after ctx.files.delete commits

The file jobs exist only when the server has files.

Related

• Handlers and the call context — ctx.jobs and transactions.
• Routes — a webhook that acknowledges at once and enqueues the work.
• Testing — servers, databases and jobs in tests.