Dartvel queues and jobs: background work with retries

Send email, resize images and call slow APIs after the response, with retries.

// lib/jobs/welcome.dart
// The generated job types, without Flutter, so a worker can run it.
import '../dartvel_client/jobs.g.dart';
import 'package:dartvel_core/dartvel.dart';

@DVJob(queue: 'mail', maxAttempts: 5, backoffSeconds: 60)
class const _SendWelcomeEmail({required final String userId});

@DVJob.handler()
Future<void> _handleSendWelcomeEmail(SendWelcomeEmail job) =>
    sendWelcomeEmail(job.userId);

A handler that uses DV runs in the app only. Write it against dartvel_core so a worker can run it.

await SendWelcomeEmail(userId: 'user-1').dispatch();

// Override the declared settings for one dispatch.
await SendWelcomeEmail(userId: 'user-2').dispatch(
  queue: 'priority-mail',
  maxAttempts: 10,
);

// lib/backend/functions/signup.post.dart
import 'package:dartvel_core/dartvel.dart';

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

@DVBackendFunction()
Future<Map<String, Object?>> _signup(String userId) async {
  // The response goes out now. A worker sends the email.
  await SendWelcomeEmail(userId: userId).dispatch();
  return <String, Object?>{'queued': true};
}

Start the generated backend with these environment variables and it works jobs instead of serving HTTP.

DARTVEL_ROLE=worker
DARTVEL_QUEUE=mail,default
DATABASE_URL=postgres://app@db.internal/app

A failed job goes back on its queue until maxAttempts, then to dead letters. Only the SQS and Pub/Sub adapters wait out the backoff first.

dartvel queue work --queue mail --max-jobs 10

This runs your project's worker, so DATABASE_URL must be set in your shell. The failed, retry and flush subcommands still read the CLI's own empty queue, so use the calls below for those.

// A worker drains the queue. It is a process, not a line in a page:
//
//     dartvel queue work --queue mail --max-jobs 10
//
// What an application does is look at what failed.
final List<DVJobEnvelope<DVJobPayload>> dead = await DV.Jobs.deadLetters('mail');
for (final DVJobEnvelope<DVJobPayload> job in dead) {
  DV.log('${job.id} failed ${job.attempts} times: ${job.lastError}');
  await DV.Jobs.retry(job.id); // or DV.Jobs.discard(job.id)
}

With DATABASE_URL set, jobs are durable with no setup. The generated backend keeps them in that database, and every web, worker and cron process of the deployment shares them.

Generation writes how each job is encoded from its @DVJob class, so there is nothing to register.

Without DATABASE_URL, jobs stay in the memory of the process that dispatched them, except in a web-server binary, which keeps them in the SQLite file beside it.

Adapters for the other brokers exist, but nothing in pubspec.yaml selects one yet. See the status below.

dispatch() records the current tenant. The worker runs the handler inside that tenant, so tenant-scoped models and cache keys behave as they did for the request.

// A cron function is public. Every other generation input is private.
@DVBackendCron('0 3 * * *', catchUp: true)
Future<void> nightlyRollup() => rollUpYesterday();

With DATABASE_URL set, each occurrence is claimed in the database, so two cron processes do not both run it. Without one, a cron process refuses to start unless DARTVEL_SCHEDULE_LEASE=none says it is the only one.

Delayed jobs, exponential backoff, unique jobs and pausing a queue are not built.

The queue failed, retry and flush commands act on the CLI's own process queue.

Only the database is chosen from configuration. Redis, SQS, AMQP, Pub/Sub and Kafka have adapters, but no pubspec.yaml key selects one.

GETTING STARTED

APP

DATA

BACKEND

OPERATIONS

SHIPPING

REFERENCE

Queues and jobs

Declare a job and its handler

Dispatch a job

Run workers

Work a queue by hand

Where jobs are kept

Jobs keep their tenant

Schedule work with cron

Status