durable_queue #

A durable task queue for Flutter and Dart: persistent, retryable tasks that survive crashes, restarts, and going offline.

Demo on an Android phone: photos are queued, the app goes to the home screen, the uploads still reach the server, and the server is told about each one when the app is opened again

Full videos on real phones: iPhone (78 s) · Android (103 s). Photos are queued, the app goes to the home screen, and every upload still reaches the server, including ones whose first try the server failed on purpose.

▶ Watch the 10-minute explainer on YouTube · Documentation site

The problem #

Apps constantly start work that must not be lost: uploading a photo, sending a message, syncing an edit. The first attempt often fails because the network drops, the server returns 503, or the user closes the app mid-request. A try/catch around the call handles one failure, but then you have to decide:

where the work lives until it can be retried, and whether it survives a restart;
how long to wait between attempts, and when to give up;
how to stop a double tap from uploading the same photo twice;
how many uploads may run at once, and which ones go first;
what happens to step 3 when step 2 fails.

durable_queue answers those questions in one small, framework-free engine. Work is described as data (a serializable task), stored through a pluggable storage interface, and executed by a worker that owns retries, concurrency, ordering, and recovery.

Features #

Tasks are data, so they can be stored and rebuilt after a restart
Task versions: tasks stored by an older app release are migrated after an update
Fixed and exponential retry with optional jitter, and a retryIf filter
Concurrency limits, for the queue and per task type, and priorities
Scheduled tasks: start at a given time or after a delay
Recurring tasks: run a task every few hours, across restarts
Timeouts per task type, and tasks that expire if they have not started in time
Rate limits per task type, or per key such as an API host
Gates: tasks that need the network wait while offline, without using attempts
Task dependencies, chains, and groups
Deduplication and idempotency keys
Cooperative cancellation, pause, resume, and stop
Hold and release one task, for example to let a user pause one upload
drain: run what is ready within a time limit, for background jobs
Hand-off: a task can wait for work the operating system finishes, such as a background upload
Lifecycle events, queries by status or group, counts per status, and waitFor one task
Progress reporting for long tasks, kept in memory, for progress bars
A dead-letter queue: failed tasks stay stored with their errors until you retry or remove them
Ready-made metrics (counts, success rate, run and wait times per task type) and logging
Startup recovery of tasks interrupted by a crash
An injectable clock, so retry logic is testable without waiting
Pure Dart: no HTTP client, database, Flutter, or state-management dependency

Install #

dependencies:
  durable_queue: ^0.5.12

Packages #

durable_queue is the core: pure Dart, with no database, HTTP client, or Flutter dependency. Integrations are separate packages from the same repository, so you add only what you use.

Package	What it adds
`durable_queue`	The engine: tasks, retries, priorities, dependencies, rate limits, gates.
`durable_queue_sqlite`	SQLite storage that survives restarts, on Android, iOS, macOS, Windows, Linux, and the web.
`durable_queue_hive`	Hive storage on `hive_ce`, for apps that already use Hive, on Android, iOS, macOS, Windows, Linux, and the web.
`durable_queue_drift`	drift storage: tasks live in your app's drift database, on Android, iOS, macOS, Windows, Linux, and the web.
`durable_queue_encryption`	Encryption at rest: wraps any storage and encrypts task payloads, error messages, deduplication keys, and groups with AES-256-GCM.
`durable_queue_dio`	Dio integration: retry rules for HTTP errors, `Retry-After`, idempotency headers, cancellation, queued HTTP requests.
`durable_queue_flutter`	Flutter integration: start the queue with the app, pause it in the background, and rebuild widgets from stored tasks.
`durable_queue_connectivity`	A network gate for Flutter on `connectivity_plus`, with an optional reachability check against your server.
`durable_queue_transfer`	Uploads and downloads that keep going after the app is minimised or closed, run by the operating system through `background_downloader`.
`durable_queue_workmanager`	Background jobs on Android and iOS: runs the queue from Workmanager while the app is away, with a lease so the app and the job never run it at once.

Need an integration that is not listed? Ask in Discussions.

Quick start #

For a complete Flutter app you can paste and run, with SQLite storage, offline waiting, and a live list of tasks, see the quick start page. The core on its own:

import 'package:durable_queue/durable_queue.dart';

final class UploadPhotoTask extends DurableTask {
  UploadPhotoTask({required this.path});

  final String path;

  @override
  String get type => 'upload_photo';

  @override
  Map<String, dynamic> toJson() => {'path': path};

  static UploadPhotoTask fromJson(Map<String, dynamic> json) =>
      UploadPhotoTask(path: json['path'] as String);
}

Future<void> main() async {
  final queue = DurableQueue(
    storage: MemoryQueueStorage(), // use a persistent adapter in an app
    maxConcurrentTasks: 3,
  );

  queue.register<UploadPhotoTask>(
    type: 'upload_photo',
    decoder: UploadPhotoTask.fromJson,
    handler: (task, context) async {
      await photoApi.upload(task.path); // your code
    },
  );

  await queue.start();

  await queue.enqueue(
    UploadPhotoTask(path: '/storage/avatar.jpg'),
    retryPolicy: RetryPolicy.exponential(
      maxAttempts: 5,
      initialDelay: Duration(seconds: 1),
    ),
  );
}

enqueue returns once the task is stored. The worker runs it in the background of your process and retries it if the handler throws.

How the packages fit together #

The core engine decides when each task runs. Everything else plugs into one of three slots: a gate decides whether a task may start now, storage decides where tasks live, and a handler does the actual work. durable_queue_flutter connects the queue to your app's lifecycle and UI, and durable_queue_workmanager runs it in background jobs while the app is away.

                    Your Flutter or Dart app
         enqueue tasks · show status · react to results
                               │
  durable_queue_flutter ──────▶│  starts the queue with the app,
  (optional)                   │  pauses it in the background,
                               │  rebuilds widgets from tasks
                               │
  durable_queue_workmanager ──▶│  runs it in background jobs
  (optional)                   │  while the app is away
                               ▼
 ┌─────────────────────────────────────────────────────────────┐
 │                 durable_queue · core engine                 │
 │  retries · priorities · dependencies · rate limits · events │
 ├───────────────────┬───────────────────┬─────────────────────┤
 │ Gates             │ Storage           │ Handlers            │
 │ may a task start? │ where tasks live  │ what a task does    │
 └─────────▲─────────┴─────────▲─────────┴──────────▲──────────┘
           │                   │                    │
           │                   │                    │
 durable_queue_        durable_queue_      durable_queue_dio
 connectivity          sqlite, _hive,      HTTP requests, retry
 waits while offline   _drift              rules, Retry-After
                       survive restarts

                       durable_queue_      durable_queue_transfer
                       encryption          uploads, downloads the
                       encrypts what is    OS finishes in the
                       stored              background

 built in:             built in:           built in:
 ManualQueueGate       MemoryQueueStorage  any async function

Every package is optional. Start with the core and its built-in pieces, then swap in a package for the slot you need: SQLite or Hive when tasks must survive restarts, encryption when tasks hold personal data, the connectivity gate when work needs the network, Dio for HTTP, transfers for uploads and downloads that must finish after the app is closed. Your own gates, storage adapters, and handlers fit the same slots.

How it works #

            Your app (Flutter or Dart)
                       │
         enqueue(task, retry, priority, dependsOn, group)
                       │
                       ▼
           ┌───────────────────────┐
           │     QueueStorage      │  memory, file, SQLite, ...
           │ pending · waiting ·   │  (survives restarts with a
           │ retryScheduled · ...  │   persistent adapter)
           └───────────┬───────────┘
                       │  worker claims the next ready task:
                       │  • dependencies completed
                       │  • retry time reached
                       │  • highest priority, then oldest
                       │  • a concurrency slot is free
                       ▼
           ┌───────────────────────┐
           │   your task handler   │
           └───────────┬───────────┘
              returns  │  throws
          ┌────────────┴──────────────┐
          ▼                           ▼
     completed              retryIf + RetryPolicy
          │                  ┌────────┴─────────┐
          │                  ▼                  ▼
          │           retryScheduled         failed
          │          (backoff, then           (or cancelled
          │           claimed again)           if requested)
          │                                     │
          └────────────────┬────────────────────┘
                           ▼
          tasks waiting on it are released, cancelled,
          failed, or run, per their DependencyFailurePolicy

Every state change is written to storage before the matching event is emitted. On the next launch, start recovers tasks that were left running when the process died.

Real-world use cases #

Media upload. Retry network failures with backoff, fail fast on a rejected file, and send an idempotency key so the server never stores a photo twice. See example/image_upload.dart.
Offline-first edits. Queue each change while offline, deduplicate repeated saves of the same record, and let the queue drain when connectivity returns.
Multi-step workflows. Upload, then process, then notify, as a chain where a failed step cancels the rest. See example/orchestration.dart.
Background sync. Group related sync tasks so they can be listed or cancelled together on logout.
Priority work. Let a chat message jump ahead of analytics uploads that are already waiting.

Tasks #

A task is a JSON payload plus a type name, like UploadPhotoTask above. The queue stores toJson(), never a closure, and rebuilds the task with the registered decoder. Register each type before enqueuing it; the type passed to register must match the task's type getter.

The handler's context carries the task id, attempt number, attempt budget, idempotency key, and whether cancellation was requested. Handlers that do not need it can ignore it.

Changing a task's payload: versions and migration #

Tasks can wait in storage across app updates. When a new release changes what a task's payload looks like, give the type a version and a migrate function, so tasks stored by an older release still run:

// Release 1 stored {'path': '/photo.jpg'}.
// Release 2 reads {'filePath': ..., 'compress': ...}.
queue.register<UploadPhotoTask>(
  type: 'upload_photo',
  version: 2,
  migrate: (fromVersion, json) => switch (fromVersion) {
    1 => {'filePath': json['path'], 'compress': true},
    _ => json,
  },
  decoder: UploadPhotoTask.fromJson,
  handler: uploadPhoto,
);

Each task records the version it was enqueued with (StoredTask.payloadVersion; 1 for tasks stored before versions existed). Before an attempt of an older task, its payload goes through migrate, then the decoder; the stored payload itself is not rewritten. Handle every older version in the one function, for example with a switch as above. A migration that throws fails the task at once, like a payload that cannot be decoded, with an error naming both versions.

A task stored by a newer release than the one running, for example after a user installed an older build, is not run with code that cannot read it. It waits without using attempts, checked again every hour, until an app that knows its version runs the queue. Its lastFailure holds a TaskVersionException saying why.

Only raise version for changes the decoder cannot handle. Adding an optional field with a default needs no new version.

Retry #

RetryPolicy.none();

RetryPolicy.fixed(
  maxAttempts: 3,
  delay: Duration(seconds: 2),
);

RetryPolicy.exponential(
  maxAttempts: 5,
  initialDelay: Duration(seconds: 1),
  maxDelay: Duration(minutes: 1),
  jitter: true,
);

maxAttempts includes the first run. retryIf chooses which errors are eligible:

queue.register<UploadPhotoTask>(
  type: 'upload_photo',
  decoder: UploadPhotoTask.fromJson,
  handler: upload,
  retryIf: (error, stackTrace) => error is TemporaryUploadException,
);

The default retries handler errors until the attempt budget is spent. Decode failures are not retried.

When the error itself says how long to wait, such as an HTTP 429 or 503 with a Retry-After header, return that wait from retryDelay. It replaces the policy's delay for that retry; return null to let the policy decide:

queue.register<UploadPhotoTask>(
  type: 'upload_photo',
  decoder: UploadPhotoTask.fromJson,
  handler: upload,
  retryDelay: (error, stackTrace) =>
      error is RateLimitedException ? error.retryAfter : null,
);

The attempt still counts toward maxAttempts. Negative waits count as zero, and waits are capped at RetryPolicy.maxSchedulableDelay.

The core does not speak HTTP. durable_queue_dio provides ready-made retryIf and retryDelay rules for Dio, including Retry-After, plus a deferOn rule for network gates and per-host rate limits for queued requests.

Deduplication and idempotency #

await queue.enqueue(
  SyncUserTask(userId: '123'),
  deduplicationKey: 'sync-user:123',
  idempotencyKey: '123',
);

If a task with that deduplication key is already pending, waiting, running, or retryScheduled, enqueue returns the existing id and leaves it unchanged. A completed, failed, or cancelled task does not block the key.

The idempotency key is stored and exposed on TaskContext. The handler decides how to send it. The same key does not, by itself, collapse two tasks.

Scheduled tasks #

Give a task an earliest start time with runAt or delay:

await queue.enqueue(SendReminderTask(id), runAt: DateTime(2026, 11, 1, 9));
await queue.enqueue(SyncTask(), delay: Duration(minutes: 15));

The task is stored at once and starts when its time comes and a slot is free. While it waits it is pending, counts as active, and holds its deduplication key. A time in the past means "now". With dependsOn, the task starts once its dependencies have completed and its time has come. enqueueChain applies the time to the first step.

The queue only runs while your app runs: a task scheduled for 3 a.m. starts at 3 a.m. if the app is running then, or on the next launch otherwise.

Recurring tasks #

Run a task on a fixed interval with schedule. Call it on every launch, after register, as you do for task types:

await queue.schedule(
  'sync-contacts',
  every: Duration(hours: 6),
  task: () => SyncContactsTask(),
  retryPolicy: RetryPolicy.exponential(
    maxAttempts: 3,
    initialDelay: Duration(minutes: 1),
  ),
);

One occurrence is stored at a time. It is an ordinary task, with retries, gates, and events. When it completes, fails, or is cancelled, the next occurrence is stored for the next slot. Slots are every apart, starting now or at firstRunAt, and survive restarts: calling schedule again on the next launch keeps the stored occurrence.

If slots were missed, for example while the app was closed, the task runs once as soon as possible and then continues on the original slots. Missed runs are not repeated.

Each occurrence has an idempotency key naming its slot, such as schedule:sync-contacts@2026-11-01T06:00:00.000Z, which a handler can send to a server so a repeated attempt is recognised. Occurrences are in the group schedule:sync-contacts, so getTasks(group:) shows the history; finished ones stay stored until you purge them.

unschedule('sync-contacts') cancels the waiting occurrence and stores no more. Remove the schedule call from the app too, or the next launch starts it again. As with any task, an occurrence only runs while the queue runs; for work while the app is closed, see durable_queue_workmanager.

Timeouts and expiry #

Give a task type a timeout so a hung request or a stuck handler cannot hold a slot forever:

queue.register<UploadTask>(
  type: 'upload',
  decoder: UploadTask.fromJson,
  handler: upload,
  timeout: Duration(minutes: 2),
);

An attempt that runs longer fails with TaskTimeoutException, and retryIf and the retry policy decide what happens next, as for any other error. Dart cannot stop a running function, so the handler keeps running and its result is ignored; context.whenCancellationRequested completes at the timeout, so a handler that listens for it (for example with a Dio CancelToken) stops its work. Time a transfer spends handed off to the operating system does not count.

Give a task an expiry with expiresAt or expiresAfter when it is pointless after a while, such as a "typing" notice or a one-time code:

await queue.enqueue(
  SendTypingTask(chatId),
  expiresAfter: Duration(seconds: 30),
);

A task that has not started by then is cancelled instead of started: it emits TaskCancelled, lastFailure holds a TaskExpiredException, and tasks that depend on it settle as for any cancellation. A running attempt is not interrupted, but no retry starts after the expiry. retry clears an expiry that has passed. enqueueChain applies the expiry to every step.

Priorities #

await queue.enqueue(SendMessageTask(text: 'hi'), priority: 10);
await queue.enqueue(UploadLogsTask(), priority: -5);

Among tasks that are ready to start, higher priority goes first; equal priorities keep enqueue order. The default is 0. Priority decides which task gets the next free slot. It never interrupts a handler that is already running.

Concurrency per task type #

maxConcurrentTasks limits how many handlers the queue runs at once. To limit one type within that, for example two uploads at a time while other work goes on, pass maxConcurrent to register:

final queue = DurableQueue(storage: storage, maxConcurrentTasks: 4);
queue.register<UploadPhotoTask>(
  type: 'upload_photo',
  decoder: UploadPhotoTask.fromJson,
  handler: uploadPhoto,
  maxConcurrent: 2,
);

Ready tasks of a full type wait, without using attempts, and start in their usual order as slots free; tasks of other types run meanwhile. A task handed off to outside work (see below) does not take a slot. After a restart, a task that was waiting for a slot starts within gateRecheckInterval (five minutes by default) if no slot of its type frees before then.

Dependencies and chains #

A task can wait for other tasks:

final upload = await queue.enqueue(UploadPhotoTask(path: path));
final resize = await queue.enqueue(
  ResizePhotoTask(path: path),
  dependsOn: [upload],
);

Until every dependency is completed, the dependent task is waiting and does not use a concurrency slot. If a dependency fails, is cancelled, or no longer exists, onDependencyFailure decides what happens:

`DependencyFailurePolicy`	Effect on the waiting task
`cancel` (default)	Becomes `cancelled` without running.
`fail`	Becomes `failed` without running.
`run`	Runs anyway once every dependency has finished. Good for cleanup.

The outcome cascades: if resize is cancelled, tasks waiting on resize follow their own policy. The task's lastFailure names the dependency that blocked it.

enqueueChain stores steps that run one after another, even when maxConcurrentTasks is greater than one:

final ids = await queue.enqueueChain([
  UploadPhotoTask(path: path),
  ResizePhotoTask(path: path),
  NotifyFriendsTask(path: path),
], group: 'photo:$path');

With the default policy, a failed step cancels the rest of the chain. Dependencies must already be stored when you enqueue, so cycles cannot be created.

Groups #

await queue.enqueue(SyncContactsTask(), group: 'sync');
await queue.enqueue(SyncCalendarTask(), group: 'sync');

final syncTasks = await queue.getTasks(group: 'sync');
final cancelled = await queue.cancelGroup('sync');

A group is a label. cancelGroup cancels every active task in it, using the same rules as cancel, and returns how many tasks it affected. counts(group:), retryFailed(group:), and durable_queue_flutter's QueueGroupBuilder work on a group too.

To store many tasks at once, for example a sync batch, use enqueueAll. It validates every task first, stores them in one operation, and starts work once:

final ids = await queue.enqueueAll(
  [for (final id in productIds) SyncProductTask(id)],
  group: 'sync-products',
  retryPolicy: RetryPolicy.exponential(
    maxAttempts: 5,
    initialDelay: Duration(seconds: 2),
  ),
);

The options apply to every task; the tasks do not depend on each other (use enqueueChain for that). For a deduplication or idempotency key per task, call enqueue for each.

Rate limits #

Limit how often tasks start. Every attempt counts, retries included, because each one reaches the service you are protecting.

// At most 10 uploads per minute.
queue.register<UploadPhotoTask>(
  type: 'upload_photo',
  decoder: UploadPhotoTask.fromJson,
  handler: upload,
  rateLimit: RateLimit(10, per: Duration(minutes: 1)),
);

// At most 5 requests per second to each API host, across task types.
queue.addRateLimit(
  'per-host',
  RateLimit(5, per: Duration(seconds: 1)),
  types: {'upload_photo', 'send_message'},
  keyOf: (task) => Uri.parse(task.payload['url'] as String).host,
);

Up to count tasks may start back to back, then starts are spaced by per / count. A task that has to wait is deferred: it keeps its status, attempt count, and place in line, and gets a reserved slot, so it is rescheduled once rather than polled. Other task types keep running meanwhile. A task must satisfy every limit that applies to it. Limits live in memory and start empty after a restart.

When a service asks you to slow down, hold a limit so every task in that bucket waits, not only the one that got the answer:

queue.holdRateLimit('per-host', key: 'api.example.com', duration: Duration(seconds: 30));

After the hold, starts resume at the steady rate. durable_queue_dio does this for you when a server answers 429 or 503 with Retry-After.

Gates: waiting for the network and other conditions #

A gate is a condition a task type needs before it starts, such as network access. While a required gate is closed, matching tasks are deferred without using attempts, other work keeps running, and the deferred tasks start as soon as the gate opens.

final network = ManualQueueGate();
queue.addGate('network', network); // add gates before registering types that need them

queue.register<UploadPhotoTask>(
  type: 'upload_photo',
  decoder: UploadPhotoTask.fromJson,
  handler: upload,
  requires: {'network'},
  // A connection error means "offline", not "this upload failed".
  deferOn: (error, stackTrace) =>
      error is SocketException ? 'network' : null,
);

// Open and close it from whatever tracks connectivity in your app:
connectivity.onConnectivityChanged.listen((results) {
  network.set(!results.contains(ConnectivityResult.none));
});

Closing a gate does not cancel tasks that are already running.
deferOn maps a handler error to one of the type's gates. The task is deferred instead of failing, the attempt does not count, and the gate's recheck is called. The task tries again after gateErrorDelay (default 30 seconds), or as soon as the gate closes and reopens.
A gate that never reports opening is checked again every gateRecheckInterval (default five minutes).
Implement QueueGate for other conditions, such as Wi-Fi only, charging, or signed in.

A connection-type report (Wi-Fi, mobile) does not prove the internet is reachable, for example behind a captive portal. deferOn covers that case: the request fails, the task waits, and the gate can probe.

On Flutter, durable_queue_connectivity provides this gate ready-made: ConnectivityGate.create() on connectivity_plus, an optional HttpProbe against your own server, and deferOnConnectionError('network').

Cancellation and controls #

await queue.cancel(taskId);
await queue.pause();
await queue.resume();
await queue.stop();

Cancelling queued or waiting work marks it cancelled immediately. Cancelling a running task does not abort the future. The handler can check context.isCancellationRequested, or wait on context.whenCancellationRequested, and throw TaskCancelledException to stop. If it returns normally, the task is completed.

whenCancellationRequested completes the moment the task is cancelled, so it can abort work that accepts a cancel signal:

handler: (task, context) async {
  final token = CancelToken();
  context.whenCancellationRequested.then((_) => token.cancel());
  await dio.post(task.url, data: task.body, cancelToken: token);
},

pause finishes handlers that have already started and does not start new ones. stop waits for those handlers and leaves everything else stored.

Hold and release one task #

pause and resume act on the whole queue. To let a user pause one task, such as one upload in a list, hold it:

await queue.hold(taskId);    // pause this one
await queue.release(taskId); // and carry on

A task that has not started stays in the queue but does not start. Its start or retry time is kept for release, and tasks that depend on it keep waiting.
A running task is asked to stop, as for cancel. When the handler then throws, the task goes back to the queue held, and the interrupted attempt does not count. A handler that finishes anyway completes the task.
A task handed off to outside work stays running and is reported on queue.holdChanges, so that work can pause. durable_queue_transfer pauses downloads that support it and restarts other transfers after release.

hold and release return false when there is nothing to change. A held task keeps its status, holds its deduplication key, and can still be cancelled; StoredTask.held tells a UI to show it as paused. Holding does not stop a task from expiring. Holds are stored, so they survive a restart. Hold changes are reported on queue.holdChanges, separately from queue.events, so existing switch statements over events keep compiling.

Retry, delete, and purge finished tasks #

Completed, failed, and cancelled tasks stay in storage until you remove them.

// Run a failed or cancelled task again with a fresh attempt budget.
await queue.retry(taskId);

// Remove one finished task.
await queue.delete(taskId);

// Remove completed tasks older than a week. Returns the number removed.
final removed = await queue.purge(
  statuses: {TaskStatus.completed},
  olderThan: Duration(days: 7),
);

The dead-letter queue #

Failed tasks are the queue's dead-letter queue. A task whose attempts run out, or whose error retryIf rejects, becomes failed and stays stored, with its last error in lastFailure, until you retry, delete, or purge it:

// Inspect them, for example on a support screen.
final dead = await queue.getTasks(status: TaskStatus.failed);
for (final task in dead) {
  print('${task.type} ${task.id}: ${task.lastFailure?.error}');
}

// Retry one, or all of them after a server outage.
await queue.retry(dead.first.id);
final retried = await queue.retryFailed(); // or retryFailed(group: 'uploads')

// Or drop old ones.
await queue.purge(statuses: {TaskStatus.failed}, olderThan: Duration(days: 30));

To keep the errors of earlier attempts as well, create the queue with failureHistory:

final queue = DurableQueue(storage: storage, failureHistory: 5);

Each task then keeps up to that many failures in StoredTask.failures, oldest first: the error, stack trace, time, and attempt number of each. The history is stored with the task, so it survives restarts and a retry. It is off by default. With durable_queue_encryption, use 0.1.2 or later, which encrypts the history too.

retry keeps the task id, payload, keys, priority, group, and dependencies, and can take a new retryPolicy. It refuses tasks that are still active, refuses when another active task already holds the same deduplication key, and refuses when a dependency did not complete (retry that dependency first). Tasks cancelled because a dependency failed are not revived automatically. delete and purge only touch terminal tasks, and never remove a task that a waiting task still depends on.

When the queue is no longer needed, close stops it and closes the events and storageErrors streams:

await queue.close();

Running in a limited time window #

drain starts the queue, runs what is ready, and stops when nothing is left to start or a time limit passes. Use it where the app gets a short window to work, such as a background job started by the operating system:

final result = await queue.drain(timeLimit: Duration(seconds: 25));
print('${result.completed} done, ${result.failed} failed, stopped: ${result.reason.name}');
// result.nextWakeAt: when the next scheduled task, retry, or deferral is due.

Tasks scheduled for later, retries, and tasks held by a gate or rate limit do not keep the drain going. Pass waitForScheduled: true to also wait for timed work due before the limit.
At the limit no new task starts; tasks already running finish first, so leave headroom for your longest task.
The queue must be idle when drain is called, and is idle again when it returns.

On Android and iOS, durable_queue_workmanager runs drain from Workmanager jobs. Its lease keeps the job and the app's own queue from running the same storage at once.

Work that finishes outside the queue #

Some work outlives the handler that starts it. An upload given to the operating system's background transfer service keeps going after the app is suspended or closed, and its result arrives later, maybe after a restart. A handler hands such a task off instead of waiting:

queue.register<UploadPhotoTask>(
  type: 'upload_photo',
  decoder: UploadPhotoTask.fromJson,
  handler: (task, context) async {
    final transferId = await startSystemUpload(task.path); // returns at once
    context.handOff(transferId);
  },
);

// Later, when the transfer reports back (in this run of the app or a later one):
await queue.completeHandOff(taskId, reference: transferId);
await queue.failHandOff(taskId, error, reference: transferId); // retry rules apply

A handed-off task stays running with handOffReference set (StoredTask.isHandedOff). It does not hold a concurrency slot, start does not treat it as interrupted, and tasks that depend on it wait for its outcome.
completeHandOff completes it. failHandOff applies the type's retryIf, retryDelay, and deferOn rules and the retry policy, as if the handler had thrown; a retry runs the handler again, which starts the work again. A TaskCancelledException cancels it.
Both take the reference the handler gave and return false when it does not match the current attempt, or the task is no longer handed off, so a late or repeated report is harmless.
cancel stops waiting at once: the task becomes cancelled and a TaskCancelled event (wasRunning: true) tells whoever runs the outside work to stop it.
The reference is stored as soon as handOff is called, so a crash before the handler returns does not start the work twice.

durable_queue_transfer does this for uploads and downloads with background_downloader.

Observation #

queue.events.listen((event) {
  // TaskEnqueued (with runAt for scheduled tasks), TaskStarted,
  // TaskRetryScheduled, TaskDeferred,
  // TaskCompleted, TaskFailed, TaskCancelled
});

final failed = await queue.getTasks(status: TaskStatus.failed);

final counts = await queue.counts(group: 'uploads');
print('${counts[TaskStatus.pending]} uploads waiting');

final done = await queue.waitFor(taskId, timeout: Duration(minutes: 1));
if (done.status != TaskStatus.completed) print(done.lastFailure);

counts returns every status, with zero for the ones no task is in. waitFor returns the task's record once it is completed, failed, or cancelled (at once if it already is); it does not start the queue.

TaskDeferred says why a task waits (rateLimit, gateClosed, or gateError), which limit or gate is responsible, and until when.

The event stream is broadcast and does not replay events that happened before the listener subscribed.

Progress #

A handler can report how far it has got, for example while it uploads:

handler: (task, context) async {
  await uploadFile(
    task.path,
    onProgress: (sent, total) => context.reportProgress(sent / total),
  );
},

Read it with queue.progressOf(taskId), or listen to queue.progress; durable_queue_flutter's QueueProgressBuilder rebuilds a widget with it. Progress is kept in memory and never written to storage, so reporting often costs no database writes. It is cleared when an attempt ends and starts again with the next one; after a restart it is empty until the task reports again. Work that runs outside a handler, such as a handed-off upload, reports with queue.reportProgress(taskId, fraction); durable_queue_transfer does this for you.

Metrics and logging #

QueueMetrics and QueueLogger listen to the event stream for you:

final metrics = QueueMetrics(queue);
QueueLogger(queue, onRecord: (record) => debugPrint('$record'));

// later, for example once a day:
analytics.log('queue', metrics.snapshot().toJson());
metrics.reset();

A snapshot has, in total and per task type: enqueued, started, completed, failed, retried, cancelled, and deferred counts, the success rate, and the average and longest run time (from an attempt's start to its outcome) and wait time (from enqueue, or a later runAt, to the first attempt). Numbers are kept in memory from the moment QueueMetrics is created.

QueueLogger turns events, holds, and storage errors into one-line records with a level: debug for routine progress, info for outcomes, warning for retries and storage errors, and error for tasks that failed for good. Pass minLevel to drop routine records. Records name task ids and types, never payloads.

Worker storage errors are reported as values on queue.storageErrors and retried after storageRetryDelay (default one second). A failed result write retries persistence without running the handler again. Storage outages hold serialized operations, so enqueue, queries, cancellation, pause, and stop can wait until storage recovers. Failed startup recovery leaves the queue idle and allows another start() call.

Using it in Flutter #

The core package does not depend on Flutter. durable_queue_flutter connects a queue to the app:

Future<void> main() async {
  WidgetsFlutterBinding.ensureInitialized();
  final queue = DurableQueue(storage: await openStorage()); // e.g. durable_queue_sqlite
  queue.register<UploadPhotoTask>(
    type: 'upload_photo',
    decoder: UploadPhotoTask.fromJson,
    handler: (task, context) => api.upload(task.path),
  );
  // Starts the queue, pauses it in the background, resumes in the foreground.
  runApp(DurableQueueScope(queue: queue, child: const MyApp()));
}

// Anywhere in the UI:
QueueTaskBuilder(
  taskId: uploadId,
  builder: (context, task) => Text(task?.status.name ?? '...'),
);

Without that package, create the queue the same way, call await queue.start() before runApp, and build UI from queue.events, getTask, and getTasks.

The queue runs while your app's process runs. It does not wake a terminated app; it starts again on the next launch. To keep work going while the app is away, run it from background jobs with durable_queue_workmanager.

Storage #

The engine talks to storage only through QueueStorage, so you choose where tasks live.

Storage	Survives restart	Status
`MemoryQueueStorage`	No	Included. For tests and process-lifetime queues.
`JsonFileStorage`	Yes	Example in `example/json_file_storage.dart`, checked against the full storage contract. Rewrites one file per change, so it suits small queues.
`durable_queue_sqlite`	Yes	Separate package. Indexed, transactional SQLite storage for Android, iOS, macOS, Windows, Linux, and the web. Use 0.2.1 or later with `durable_queue` 0.5.
`durable_queue_drift`	Yes	Separate package. The same tables as `durable_queue_sqlite`, inside your app's drift database.
`durable_queue_hive`	Yes	Separate package. Hive storage on `hive_ce`, for apps that already use Hive.
Isar, others	Yes	Adapters are welcome as separate packages; say what you need in Discussions.

Adapters stay in separate packages so the core never forces a database on you.

To write an adapter, implement QueueStorage and run the shared contract tests against it:

final class MyStorage implements QueueStorage {
  // save, get, update, delete, getNextReady, getWaitingDependents, ...
}

// test/my_storage_test.dart
void main() => queueStorageContractTests(MyStorage.new);

queueStorageContractTests lives in storage_contract.dart in the repository's test-support package; copy it into your adapter's tests. STORAGE.md lists every method, ordering rule, and atomicity requirement.

When to use durable_queue #

	Manual `try`/`catch` + retry loop	Simple in-memory queue	durable_queue	OS background scheduler (e.g. Android WorkManager)
Work survives an app restart	No	No	Yes, with a persistent storage adapter	Yes
Runs while the app is terminated	No	No	No	Yes, within OS limits
Retry with backoff and a retry filter	You write it	Usually you write it	Built in	Varies by platform
Deduplication and idempotency keys	You write it	Usually you write it	Built in	Varies by platform
Dependencies, chains, priorities	You write it	Rarely	Built in	Varies by platform
Rate limits, waiting for the network	You write it	Rarely	Built in	Network constraints only
Testable without real waiting	Depends on your code	Depends	Yes (`FakeQueueClock`)	Limited
Platforms	Any	Any	Any Dart platform	One platform per API

Use durable_queue when work must outlive a failed request or an app restart, and the app is allowed to finish it the next time it runs. Use an OS scheduler when the work must run while the app is closed; the two can be combined, with the OS job calling start. A plain try/catch is enough for work that is fine to lose. This table compares capabilities only. For what the queue itself costs (storing, running, and recovering tasks with each storage), see the benchmarks.

Guarantees #

Execution is at least once. A crash after a side effect but before the completion write can run the task again.

start recovers a task left in running: that attempt counts, then the retry policy either schedules another try or fails the task. Handlers should be idempotent when a repeated side effect would be harmful.

When a task finishes, the tasks waiting on it are updated before its own result is written. A crash in between never leaves a task waiting forever behind a dependency that already finished. The worst case is that dependents were released or cancelled for an outcome that recovery then records again.

Persistence is not background execution. A terminated app stays terminated until something else launches it. After the next launch, call start again.

Do not put secrets in task payloads or exception text. The queue stores them as plain data. To encrypt payloads and error messages at rest, wrap your storage with durable_queue_encryption. To report a vulnerability, see the security policy.

The full list, including what is not promised and how it is tested on devices: Guarantees. Measured costs: Benchmarks.

Examples #

dart run example/main.dart           # one task, one retry
dart run example/image_upload.dart   # uploads that survive a restart
dart run example/orchestration.dart  # chains, dependencies, priorities
dart run example/rate_limits_and_gates.dart  # rate limits, offline gate

More recipes, from offline forms to background jobs: Cookbook. A complete Flutter app, killed mid-upload in CI: durable_queue_demo.

Contributing and feedback #

Questions, feature ideas, and storage-integration requests: GitHub Discussions.
Bugs: issue tracker.
Code and adapters: see CONTRIBUTING.md.
New contributors: good first issues.

If durable_queue saves you work, a like on pub.dev (the thumbs-up at the top of this page) or a star on GitHub helps other developers find it.

durable_queue 0.5.13 durable_queue: ^0.5.13 copied to clipboard

Metadata