A Redis Task Queue in Dart: Weighted Polling, Dead Letters, and One Nasty Empty-Poll Bug

I recently built a small Redis-backed task queue for server-side Dart, because the ecosystem didn't have a maintained one. In a companion post I wrote about why — the gap on pub.dev, what I ported from Go's Asynq, and what I deliberately left out of a first version.

This post is the other half: how it actually works. It's about 360 lines of Dart, and the interesting parts are all in how three Redis commands get used. If you've ever wondered what's under a task queue, this is a small enough one to hold in your head.

flowchart LR
    P["producer"] -->|LPUSH| Q["queue list<br/>rtq:pending"]
    Q -->|"weighted BRPOP"| W["worker"]
    W -->|success| D["done"]
    W -->|throw| Rt["retry"]
    Rt -->|"attempt < max"| Q
    Rt -->|exhausted| DL[("dead-letter<br/>rtq:dead")]

The data model: a Task the user sees, an Envelope Redis stores

The user-facing unit is a Task — a type string plus a JSON payload:

class Task {
  Task(this.type, this.payload);
  final String type;                     // routes to a handler
  final Map<String, dynamic> payload;    // whatever the handler needs
}

The split is deliberate: the code that enqueues and the code that processes only agree on a string and a shape. They don't share a class hierarchy, and they can live in different processes.

But the Task isn't what gets stored. Redis stores an Envelope — the task plus the metadata the queue needs to run and retry it:

class Envelope {
  final String id;
  final Task task;
  final String queue;
  final int maxRetries;
  int attempt;             // incremented on each retry
  String encode() => jsonEncode({ /* id, task, queue, max_retries, attempt */ });
}

attempt is the only mutable field, and it's the whole retry state machine: bump it each failure, give up when it passes maxRetries. Keeping run-state on the envelope (not in a separate Redis key) means a job is one self-contained string — nothing to clean up, nothing to get out of sync.

Enqueue is one LPUSH

Producing a task is deliberately trivial — a single write:

Future<String> enqueue(Task task, {String queue = 'default', int maxRetries = 5}) async {
  final id = _newId();
  final env = Envelope(id: id, task: task, queue: queue, maxRetries: maxRetries);
  await _command.send_object(['LPUSH', _keys.pending(queue), env.encode()]);
  return id;
}

LPUSH pushes the encoded envelope onto the head of a Redis list. Because the worker pops from the tail (more on that next), head-push + tail-pop gives you FIFO ordering for free — first in, first out, which is what you want for a job queue. Since it's a single command, you keep one client alive for the app's lifetime and reuse it; there's nothing to pool.

The worker: weighted polling with one BRPOP

This is the part with actual design in it. The worker drains multiple queues, and it drains them by weight so a flood of low-priority work can't starve the important stuff.

The trick is almost embarrassingly simple. You expand the weights into a repeated list of keys:

List<String> _weightedOrder() {
  final order = <String>[];
  _queues.forEach((queue, weight) {
    for (var i = 0; i < weight; i++) order.add(queue);
  });
  return order;   // {'critical':6,'default':3,'low':1} -> [crit×6, default×3, low×1]
}

Then you hand that whole list to a single BRPOP:

final keys = order.map(_keys.pending).toList();
final res = await _command.send_object(['BRPOP', ...keys, '1']) as List;

BRPOP key1 key2 ... timeout is a blocking right-pop across many keys at once. It scans the keys left to right and returns from the first one that has an item. Because critical appears six times near the front of the list, it gets first refusal six times over before low gets a look — that's the weighting, and Redis does it atomically. No priority field, no sorting, no second data structure. The blocking part also means an idle worker sleeps in Redis instead of hot-looping, so an empty queue costs you nothing.

The bug: an empty poll doesn't return null

Here's the one that cost me real time, and it's the reason I don't trust a queue I haven't run idle.

BRPOP with a timeout has two outcomes. A hit returns [key, value]. A timeout — no job for a full second — returns nothing. I assumed "nothing" meant null. It doesn't:

// WRONG: this branch never runs
if (res == null) continue;
final env = Envelope.decode(res[1] as String);   // boom, every idle second

The Redis client hands back [null] on timeout — a one-element list whose single element is null, not a bare null. So res == null was never true, and every idle second the worker tried to read res[1] off a list of length one and threw. An idle worker — the normal state — was crashing on a schedule.

The fix is one line:

if (res.isEmpty || res.first == null) continue;
await _process(Envelope.decode(res[1] as String));

I only found it because a test spins up a real worker against a real Redis and lets it sit with an empty queue. Unit tests with a mocked client would have sailed right past it — the mock would have returned whatever I thought BRPOP returns, which was the wrong thing. The boring states (idle, empty, timed-out) are where the real bugs live.

Retry or dead-letter

Processing routes by type and treats any throw as a failure — including "no handler registered," so a wiring mistake surfaces loudly instead of silently dropping jobs:

Future<void> _process(Envelope env) async {
  final handler = _handlers[env.task.type];
  try {
    if (handler == null) throw StateError('no handler for "${env.task.type}"');
    await handler(env.task);
  } catch (_) {
    await _retryOrDeadLetter(env);
  }
}

And the retry path is where a job's life ends one of two ways:

Future<void> _retryOrDeadLetter(Envelope env) async {
  if (env.attempt >= env.maxRetries) {
    await _command.send_object(['LPUSH', _keys.deadLetter(), env.encode()]);
    return;                       // out of tries -> dead-letter list
  }
  env.attempt++;
  await _command.send_object(['LPUSH', _keys.pending(env.queue), env.encode()]); // back on its queue
}

Two honest caveats I put in the README rather than hide:

1. The re-enqueue is immediate — no backoff. A production version delays it with a sorted set scored by a "run-after" timestamp. I left that out of 0.1.0 to keep the core path readable; it's the first thing on the list to add.
2. A dead-letter list, not silent death. Once retries are exhausted the envelope goes to <prefix>:dead so you can inspect what failed, instead of the job vanishing or looping forever.

That's the whole thing

Three commands — LPUSH to produce, BRPOP to consume, LPUSH again to retry or dead-letter — plus a repeated-key trick for weighting and an envelope that carries its own retry count. No magic, which is the point: a queue small enough to read end to end is a queue you can actually trust in front of your request path.

The package is redis_task_queue on pub.dev, source on GitHub. If you want the decisions behind it — what I took from Asynq and what I refused to copy into a first version — that's in the companion post.