dox_queue

Background jobs for the Dox framework: Redis, database and sync drivers, retries, delays and workers.

A queue runs slow work in the background. The request answers at once. A worker process does the work later.

Use it to send emails, call slow APIs, make reports or resize images. The dox_queue package gives you jobs, three drivers (Redis, database and sync), retries, delays and a worker.

Install

dart pub add dox_queue

dox_queue has a builder. Your app already runs build_runner for dox_builder, so nothing else is needed.

How it works

1. You write a job class with @Job().
2. dart run build_runner build writes lib/generated/jobs.g.dart. It knows how to save and load each job.
3. Your code calls dispatch(). The job is saved in Redis or in the database.
4. A worker (dart run bin/queue.dart) takes the job and runs its handle() method.

Set up

1. Config

lib/config/queue.dart:

import 'package:dox_queue/dox_queue.dart';

import '../generated/jobs.g.dart';

QueueConfig queueConfig = QueueConfig(
  /// sync, database or redis
  driver: 'database',

  /// every @Job class, from lib/generated/jobs.g.dart
  jobs: jobRegistry,
);

Option	Default	What it does
driver	sync	sync, database or redis
jobs	empty	jobRegistry from lib/generated/jobs.g.dart
redis	localhost:6379	RedisOptions(host: ..., port: ..., password: ...) for the redis driver
prefix	dox_queue:	the start of every Redis key
connection	default	the dox_orm connection name, for the database driver
retryAfter	90 seconds	when a worker does not finish a job in this time, another worker can take it
workInProcess	false	run a worker inside the web server, see Small apps
workerQueues	['default']	the queues of the in-process worker
workerConcurrency	1	jobs at once in the in-process worker
workerTimeoutGrace	off	opt-in: after a timeout, the in-process worker waits up to this long for the old run (see Retries)

Keep retryAfter bigger than your longest job retryAfter is how Dox finds the job of a crashed worker. A job that runs longer than retryAfter looks crashed, so a second worker runs it again.

2. Provider

The provider sets up the queue in every isolate. Put it in lib/ so the generated container finds it.

lib/providers/queue_provider.dart:

import 'package:dox_queue/dox_queue.dart';

import '../config/queue.dart';

/// Sets up the queue. Dox finds every DoxProvider class,
/// see lib/container.g.dart.
class AppQueueProvider extends QueueProvider {
  AppQueueProvider() : super(queueConfig);
}

3. Worker file

bin/queue.dart:

import 'dart:io';

import 'package:dox_queue/dox_queue.dart';
import 'package:my_app/config/app.dart';

/// run queued jobs: `dart run bin/queue.dart [work] [flags]`.
/// the `dox` CLI runs this file for its queue commands.
Future<void> main(List<String> args) async {
  exitCode = await QueueWorker(app: appConfig).run(args);
}

The worker starts the providers of your app (database, cache, queue). It does not start the HTTP server.

4. Tables (database driver only)

Run dox queue:table. It writes a migration in db/migration that makes the jobs and failed_jobs tables. Then run dox migrate.

dox queue:table
dox migrate

queue:table is a package command of dox_queue. It does not overwrite: when a create_queue_tables migration is already there, it prints so and stops.

The migration it writes looks like this. You can also write it by hand. QueueTables makes the two tables.

db/migration/2026_10_25_000000_create_queue_tables.dart:

// ignore_for_file: file_names
import 'package:dox_migration/dox_migration.dart';
import 'package:dox_queue/dox_queue.dart';

/// The jobs and failed_jobs tables of dox_queue.
class CreateQueueTables extends Migration {
  @override
  Future<void> up(Schema schema) async {
    await QueueTables.up(schema);
  }

  @override
  Future<void> down(Schema schema) async {
    await QueueTables.down(schema);
  }
}

Write a job

lib/app/jobs/send_welcome_email.job.dart:

import 'package:dox_core/dox_core.dart';
import 'package:dox_queue/dox_queue.dart';

@Job()
class SendWelcomeEmail extends DoxJob {
  SendWelcomeEmail({required this.userId});

  final int userId;

  /// run up to 3 times
  @override
  int get tries => 3;

  /// wait 10 seconds before the 2nd try, 60 seconds before the 3rd
  @override
  List<Duration> get backoff => const <Duration>[
    Duration(seconds: 10),
    Duration(seconds: 60),
  ];

  /// stop a try after 30 seconds
  @override
  Duration? get timeout => const Duration(seconds: 30);

  /// the queue of this job
  @override
  String get queue => 'emails';

  @override
  Future<void> handle() async {
    MailService mail = Dox().ioc.get<MailService>();
    await mail.sendWelcome(userId);
  }

  /// runs once, after the last try failed
  @override
  Future<void> failed(Object error, StackTrace stack) async {
    // tell someone, or clean up
  }
}

Run dart run build_runner build after you add or change a job.

All the settings are optional:

Setting	Default	What it does
tries	the worker --tries (1)	how many times to run the job before it fails
backoff	none (retry at once)	the wait before each retry. The last entry is used for the rest
timeout	the worker --timeout (60 seconds)	the max run time of one try
retryUntil	none	retry until this time, not tries times. Read once, when the job is dispatched
maxExceptions	none	fail after this many tries that threw an error. A timeout does not count
failOnTimeout	false	fail at once when a try times out, with no retry
queue	default	the queue to put the job on
failed()	does nothing	runs once, after the last try failed

What is saved

The parameters of the unnamed constructor are saved with the job. Each parameter needs a field or getter with the same name.

These types work:

• int, double, num, String, bool
• DateTime
• enums (saved by name)
• List and Map<String, ...> of these, also nested
• dynamic and Object (the value must be JSON)
• nullable versions of all of them

Any other type stops the build with a clear message.

Save an id, not an object Do not save a User. Save the userId, and load the user in handle(). The user may change between dispatch and run. The id gives you fresh data.

Encrypted jobs

Job data is saved as plain JSON in Redis or the jobs table. To hide it from anyone who can read the store, add ShouldBeEncrypted:

@Job()
class ChargeCard extends DoxJob with ShouldBeEncrypted {
  // ...
}

The data is then encrypted with AES-GCM and the app key (AppConfig.appKey). The worker needs the same key, or the old key in AppConfig.previousKeys. The job name, queue and id stay plain text. Jobs without the mixin are saved as before.

Services in a job

A job is made again from its saved data, so it can not get services in its constructor. Get them in handle() from the Dox container:

UserService users = Dox().ioc.get<UserService>();

The worker runs registerContainer, so every @Inject() class of your app is there.

Job names

The class name is saved with each job. When you rename a class, queued jobs of the old name can not be found. Keep the old name with @Job(name: 'SendWelcomeEmail').

Dispatch a job

await SendWelcomeEmail(userId: user.id).dispatch();

Both ways do the same:

await Queue.dispatch(SendWelcomeEmail(userId: user.id));

A delay, another queue:

await SendWelcomeEmail(userId: 1).dispatch(
  delay: Duration(minutes: 10),
  queue: 'emails',
);

Wait until a time:

await SendWelcomeEmail(userId: 1).dispatch(
  delayUntil: DateTime.utc(2030, 1, 1, 9),
);

Give delay or delayUntil, not both.

dispatch() returns the job id.

Dispatch only when a condition is true, or false:

await SendWelcomeEmail(userId: 1).dispatchIf(user.wantsEmail);
await SendWelcomeEmail(userId: 1).dispatchUnless(user.isBlocked);

They return null when the job was not dispatched.

Run a job now, in this process, with no queue:

await SendWelcomeEmail(userId: 1).dispatchSync();

Queue an email

You do not need a job class to send an email later. With dox_mailer, call Mail.queue():

await Mail.queue(message);
await Mail.later(Duration(minutes: 10), message, queue: 'emails');

dox_queue has a built-in job for it. It is not in lib/generated/jobs.g.dart, and you do not need to add it.

Deploy the worker together with the web app. An old worker does not know this job, so it fails the queued emails.

With encrypt: true, a send error is logged and saved as its type and code only, eg. RedactedMailError: MailException (code 550). Details are encrypted: .... The full error text, with the answer of the server, is encrypted with the app key (at most the first 64K characters, counted in UTF-16 code units). To read it, take a FailedJob from Queue.driver.failed() and call RedactedMailError.decrypt(job.error).

Run the worker

dart run bin/queue.dart

Commands:

Command	What it does
work	run jobs. This is the default
failed	list the failed jobs
retry <id>	put one failed job back on its queue
retry all	put all failed jobs back
forget <id>	delete one failed job
flush	delete all failed jobs
clear [queue]	delete the jobs of a queue. Default: default
restart	stop the running workers after their current jobs
prune-failed	delete failed jobs older than --hours=N (default 24)

Flags of work:

Flag	Default	What it does
--queue=high,default	default	the queues to work. The first one goes first
--concurrency=4	1	jobs at once in this process
--once	off	run one job, then stop
--tries=3	1	tries of a job that does not set tries
--timeout=60	60	max run time in seconds of a job that does not set timeout
--sleep=3	3	seconds to wait when the queues are empty
--stop-when-empty	off	stop when the queues are empty
--timeout-grace=30	off	after a timeout, wait up to this many seconds for the old run to end before the next try
--backoff=1,5,10	none	seconds to wait before each retry of a job that does not set backoff
--max-jobs=1000	off	stop after this many jobs
--max-time=3600	off	stop after this many seconds
--memory=256	off	stop when the process uses more than this many MB

With --max-jobs, --max-time and --memory, the worker stops after its running jobs. Let your process manager start it again.

dart run bin/queue.dart work --queue=emails,default --concurrency=4

--concurrency runs jobs as async tasks on one thread. It helps for jobs that wait (email, HTTP, database). For CPU heavy jobs, run more worker processes.

Stop the worker

Press Ctrl+C, or send SIGTERM (docker stop does it). The worker takes no new job. It waits for the running jobs, then it stops. A second signal stops it at once.

Restart the worker after a deploy The worker keeps your code in memory. Restart it after each deploy, or it runs the old code.

dart run bin/queue.dart restart tells every running worker to stop after its current jobs. Your process manager then starts it with the new code. It writes a time to the app cache. Each worker reads it every 3 seconds. So the workers and the command must use the same cache store, eg. file on one machine, or redis. The memory store does not work for it.

Keep it running

Run the worker as its own process, next to the web server. For example with Docker Compose:

services:
  web:
    build: .
    command: dart run bin/server.dart
  worker:
    build: .
    command: dart run bin/queue.dart work --concurrency=4

Or with systemd, Supervisor or a second Fly.io process. Set it to restart when it stops.

Retries and failed jobs

When handle() throws:

1. The worker waits the backoff time and tries again, up to tries times.
2. After the last try, the job goes to the failed jobs, and failed() runs.

A timeout counts as a failed try.

A timeout does not stop the code Dart can not stop a running Future. After a timeout the try counts as failed, but handle() keeps running in the background until it ends. Write jobs that are safe to run twice.

So by default the next try can start while the old run still works. Then the job runs twice at the same time. To stop this, start the worker with --timeout-grace=SECONDS (or WorkerOptions(timeoutGrace: ...), or QueueConfig(workerTimeoutGrace: ...) for the in-process worker). After a timeout, the worker then waits for the old run to end, up to that time, before it tries again or fails the job. The result of the old run is ignored: the try still counts as failed. While the worker waits, it takes no other job. If the old run is still going when the grace time ends, the worker logs a warning and goes on. Keep timeout plus the grace time below retryAfter, or another worker can take the job again while this one waits. Then the double run still happens.

When a worker crashes, its job stays reserved. After retryAfter another worker takes it. This counts as a try. When there are no tries left, the job fails with a message that the worker did not finish.

Failed jobs are kept until you retry or delete them:

dart run bin/queue.dart failed
dart run bin/queue.dart retry 6f1c0e...
dart run bin/queue.dart retry all
dart run bin/queue.dart forget 6f1c0e...
dart run bin/queue.dart flush

Drivers

Driver	Where jobs are	Needs
sync	nowhere: dispatch() runs the job at once	nothing
database	the jobs and failed_jobs tables	Postgres, MySQL 8+, MariaDB 10.6+ or SQLite
redis	Redis lists and sorted sets	Redis

Database

Jobs are rows in the jobs table. Failed jobs are rows in failed_jobs.

• On Postgres, MySQL and MariaDB a worker takes a job with SELECT ... FOR UPDATE SKIP LOCKED. Many workers do not wait on each other.
• On SQLite a worker takes a job with one UPDATE ... RETURNING. SQLite runs one write at a time, so this is safe. It is good for small apps.

Use another connection of config/database.dart with QueueConfig(connection: 'postgres').

Redis

QueueConfig queueConfig = QueueConfig(
  driver: 'redis',
  jobs: jobRegistry,
  redis: RedisOptions(host: env.redisHost, port: 6379),
);

To read the Redis server from the env (or .env), use QueueConfig.redisFromEnv():

QueueConfig queueConfig = QueueConfig(
  driver: 'redis',
  jobs: jobRegistry,
  redis: QueueConfig.redisFromEnv(),
);

It reads:

• REDIS_URL, like rediss://user:password@host:6380/0. rediss:// turns on TLS. When it is set, the keys below are not read.
• Else REDIS_HOST, REDIS_PORT, REDIS_USERNAME, REDIS_PASSWORD and REDIS_TLS (true turns on TLS).

A key that is not set keeps its default. With no keys it is localhost:6379. It is opt-in: when you leave out redis, the queue uses localhost:6379 and does not read the env.

Each queue has a list of ready jobs, a sorted set of delayed jobs and a sorted set of reserved jobs. A Lua script takes a job, so two workers never get the same one. Failed jobs are in the hash dox_queue:failed.

Sync

dispatch() runs the job at once and waits for it. There is no worker, no retry and no delay. When the job throws, failed() runs and dispatch() throws the error. Use it for tests and for local work.

Small apps: a worker in the web server

You can skip the worker process. Set workInProcess: true:

QueueConfig queueConfig = QueueConfig(
  driver: 'database',
  jobs: jobRegistry,
  workInProcess: true,
);

The first isolate of the web server then runs jobs too. When the server stops, it waits for the running jobs.

Heavy jobs slow down your API The jobs share the isolate with web requests. A long or CPU heavy job makes requests on that isolate slow. Use a worker process when your jobs are heavy or many.

Testing

Use the sync driver. dispatch() then runs the job at once, so you can check the result right after.

import 'package:dox_queue/dox_queue.dart';
import 'package:my_app/generated/jobs.g.dart';
import 'package:test/test.dart';

void main() {
  setUp(() {
    Queue.configure(QueueConfig(driver: 'sync', jobs: jobRegistry));
  });

  test('sends the welcome email', () async {
    await SendWelcomeEmail(userId: 1).dispatch();
    // check the email was sent
  });
}

To check that your code dispatched a job, without running the job, use Queue.fake(). It keeps every dispatched job in a list. It does not run or save them. It needs no Queue.configure().

import 'package:dox_queue/dox_queue.dart';
import 'package:test/test.dart';

void main() {
  late QueueFake queue;

  setUp(() {
    queue = Queue.fake();
  });

  tearDown(Queue.restore);

  test('signing up queues the welcome email', () async {
    await signUp(userId: 1);

    queue.assertPushed<SendWelcomeEmail>();
    queue.assertPushed<SendWelcomeEmail>(
      (SendWelcomeEmail job) => job.userId == 1,
    );
    queue.assertPushedTimes<SendWelcomeEmail>(1);
    queue.assertNotPushed<SendInvoice>();
  });
}

Method	Checks
assertPushed<T>([test])	a T job was pushed. With test, one of them passes it.
assertPushedTimes<T>(n)	exactly n T jobs were pushed
assertNotPushed<T>()	no T job was pushed
assertNothingPushed()	no job was pushed at all

A failed check throws an AssertionError that says what was pushed.

To read the jobs, use queue.pushed. Each item has the job, the queue name and the delay (null when there was none). queue.jobsOf<T>() gives the T jobs. queue.clear() forgets them.

• dispatchSync() is not faked. It still runs the job.
• Mail.queue() and Mail.later() of dox_mailer go through dispatch(), so the fake catches them too.
• Queue.restore() stops the fake.

To test a job alone, call handle():

await SendWelcomeEmail(userId: 1).handle();

To test the queue itself, use the database driver with SQLite :memory:, and run a worker that stops when the queue is empty:

await Worker(
  driver: Queue.driver,
  jobs: jobRegistry,
  options: WorkerOptions(stopWhenEmpty: true),
).run();

Development

The tests use SQLite always. They also use Postgres, MariaDB and Redis when their ports are set:

PG_PORT=5432 MYSQL_PORT=3306 REDIS_PORT=6379 dart test

See .github/workflows/ci.yaml for the container settings.

For local work against the Dox repo next to this one, add a pubspec_overrides.yaml (it is gitignored):

dependency_overrides:
  dox_core:
    path: ../dox/packages/dox-core

Release

Set version in pubspec.yaml, add it to CHANGELOG.md, merge to main, then push a v<version> tag. The Release workflow publishes the package to pub.dev.

License

MIT. See LICENSE.