redis_task_queue 0.10.1: neither the worker nor the client ever reconnected after Redis dropped

I maintain redis_task_queue, a small Redis-backed task queue for server-side Dart (Asynq/Sidekiq-shaped: enqueue from a request, process in a worker, retry on failure, dead-letter what still fails). Up through 0.10.0 it had a bug that only shows up after the code has been running for a while: if the connection to Redis ever dropped, neither the producer side (QueueClient) nor the consumer side (Worker) ever reconnected. They just stayed broken, forever, even after Redis was healthy again.

0.10.1, out now, fixes this. Here is what was wrong, why it never turned up in development, and why the fix has to be an actual reconnect rather than a try/catch.

The two ways this broke

QueueClient.connect() and Worker.connect() each open one TCP connection to Redis and keep it for as long as the object lives. That is by design: the README tells you to "keep one client around and reuse it" for QueueClient, and Worker.run() is meant to poll forever until you call stop(). Both are long-lived on purpose.

The underlying redis package (package:redis) holds that one socket for the life of the connection object and never redials it. Once the socket dies (Redis restarts, a managed-Redis provider fails over, a proxy in front of Redis closes an idle connection), every future call on that same connection object fails. Before 0.10.1, redis_task_queue had no code path that opened a new connection to replace the dead one. _command was a final Command field on both QueueClient and Worker, so there was not even a field to reassign if you had wanted to patch it from outside.

I reproduced both failure modes against a real Redis instance, severing the TCP connection mid-run:

• Worker.run() throws an unhandled stream is closed exception straight out of the bare await worker.run(); shown in the README's "Process" example. The run loop is gone. Nothing is polling anymore.
• QueueClient, reused across calls the way the README recommends, fails on every subsequent call on the same client, well after Redis is reachable again. The first failure after the drop threw stream is closed; the very next call on the same client threw a different error, Bad state: StreamSink is closed. The redis package's failures for a dead socket are not even consistent in shape, which matters for the fix below.

Here is an actual run against a real Redis, before the fix, using the package's own README usage pattern (a client reused across calls, a worker's run() awaited with no try/catch):

[baseline] worker processed first task before any drop: true
--- severing the connection ---
[right after sever] QueueClient.enqueue (reused client) -> id=null error=stream is closed
[right after sever] worker processed the second task within 6s: false
[right after sever] worker.run() already threw an unhandled exception: true stream is closed
[1s later, Redis long since reachable again] QueueClient.enqueue (same client, 3rd call) -> id=null error=Bad state: StreamSink is closed
[1s later] worker processed the third task: false
[done] worker.run() completed normally (no exception escaped): false

Redis was back up for the entire "1s later" block. It did not matter. The client and the worker were both done.

Why this never shows up locally

In local development you start redis-server once, at the top of your session, and it just runs. You never restart it mid-session, so the one code path that never worked, redialing after a drop, never runs either. The bug is invisible in exactly the environment most people test in.

It surfaces the first time something outside your control restarts Redis under a process that has actually been running for a while. A Redis version bump during a maintenance window, a managed-Redis failover, a proxy or load balancer closing an idle connection: these are routine production events, not rare ones. A worker that has been up for three weeks hits one of them eventually, and before 0.10.1, that is where it stayed: not backed off, not retrying, just done, usually with nothing more visible than a stack trace in a log nobody is watching at 3am.

The part that makes this worse than a plain crash: the README's own advice, "keep one client around and reuse it" for QueueClient, run this forever until stop() for Worker, is exactly the pattern that turns a ten-second Redis restart into a permanent outage. A short-lived client that reconnects on every call would never have hit this. Long-lived is the point of both classes, and long-lived is what made a ten-second blip fatal.

Why wrapping the call site in try/catch does not fix this

The obvious first reaction is to catch the exception. It does not work, and it is worth being specific about why, because the reason is different for the two classes.

For QueueClient, catching the error at the call site stops the exception from propagating, but it does nothing about the dead socket sitting inside the client:

import 'package:redis_task_queue/redis_task_queue.dart';

Future<void> main() async {
  final client = await QueueClient.connect();

  // Stops the crash. Does not fix anything.
  try {
    await client.enqueue(Task('email:welcome', {'user_id': '42'}));
  } catch (e) {
    print('enqueue failed: $e');
  }
}

The next call on the same client hits the same dead socket and fails too. Wrap every call site like this and a loud crash becomes a silent one: nothing is being enqueued, and nothing tells you, for as long as the process runs.

For Worker it is worse, because run() is a loop, not a single call. Catching around it does not resume the loop, it ends it:

import 'package:redis_task_queue/redis_task_queue.dart';

Future<void> main() async {
  final worker = await Worker.connect();
  worker.handle('email:welcome', (task, context) async {});

  // Doesn't fix it either.
  while (true) {
    try {
      await worker.run();
    } catch (_) {
      // run() has already exited by the time we get here.
    }
  }
}

Even this outer retry loop does not help, because it is still the same Worker instance with the same dead connection. Before 0.10.1 there was no way to swap that connection out, so calling run() again fails on its very first Redis command, immediately, and loops right back into the catch block: a tight, silent, CPU-and-log-spinning failure loop instead of a clean recovery.

What actually has to happen: hold onto the host and port so you can dial a fresh connection, open one, retry the specific command that just failed against it, and, specific to a worker's poll loop, back off before hammering a Redis that might still be down and rerun crash recovery before resuming polling, since a brand-new socket has no memory of what was still in flight when the old one died. A caught exception does not do any of that on its own.

What 0.10.1 actually does

Both classes now hold the host and port they connected with, and _command (the connection handle) is mutable instead of final:

-  QueueClient._(this._command, this._keys) : _idPrefix = _randomPrefix();
+  QueueClient._(this._command, this._keys, this._host, this._port)
+      : _idPrefix = _randomPrefix();

-  final Command _command;
+  Command _command;
   final Keys _keys;
+  final String _host;
+  final int _port;

Every Redis call in both classes now goes through a _send helper that tries the current connection, and on any failure at all, redials and retries once:

+  static Future<Command> _dial(String host, int port) =>
+      RedisConnection().connect(host, port);
+
+  Future<dynamic> _send(List<Object> command) async {
+    try {
+      return await _command.send_object(command);
+    } catch (_) {
+      _command = await _dial(_host, _port);
+      return await _command.send_object(command);
+    }
+  }

"Any failure at all" is deliberate, not sloppy. The redis package's errors for a dead socket are not a consistent type: a bare String, a StateError, sometimes a real SocketException. Pattern-matching specific exception types would mean guessing at every shape a dead socket can fail in and missing the next one. Treating any failure as "the connection might be dead, try a fresh one" does not need to know the shape.

Every public method that used to call _command.send_object(...) directly now calls _send(...) instead:

     if (processAt == null && processIn == null) {
-      await _command.send_object(['LPUSH', _keys.pending(queue), env.encode()]);
+      await _send(['LPUSH', _keys.pending(queue), env.encode()]);
       return id;
     }

For QueueClient, that is the whole fix: a call whose retry also fails still throws, because at that point Redis is actually down, not just blipped, and the caller needs to know that. For Worker, run()'s poll loop goes further. Here is the current loop, excerpted from worker.dart:

while (_running) {
  try {
    await _promoteDue();
    final claimed = await _claim(weighted, topQueue);
    if (claimed == null) continue;
    await _process(claimed);
  } catch (_) {
    // Every call above already went through _send, which reconnects and
    // retries once on its own; getting here means that retry failed too,
    // so the connection is still down, not just blipped.
    if (!_running) break;
    await Future<void>.delayed(_backoffBase);
    try {
      await _reconnect();
      await _recoverOrphans();
    } catch (_) {
      // Still down; the next pass through this loop retries again.
    }
  }
}

When _send's own retry also fails, the loop does not let the exception propagate out of run(). It backs off by backoffBase (the same backoff already used for task retries), reconnects, and reruns _recoverOrphans(), the same recovery that runs when a worker first starts, before going back to polling. That last part matters on its own: a fresh socket has no memory of what this worker had claimed onto its in-flight list when the old one died, so treating "just reconnected" the same as "just started" is what stops a task from being stuck there silently.

Same underlying bug, different shape of fix on each side, because the two classes are shaped differently. QueueClient is a set of individual calls, so retry-and-rethrow is the whole story. Worker is a loop that is supposed to survive forever, so the loop itself has to absorb the failure and keep going.

Upgrading needs no code changes

connect() takes the same arguments it always did, on both classes. If your connection to Redis never drops, 0.10.1 behaves exactly like 0.10.0. The only change in behavior is what happens when it does drop: a delay measured in the backoff, instead of a process that quietly stopped doing anything until someone noticed and restarted it.

The fix shipped with a new test, test/connection_drop_test.dart, which runs a real Redis behind a small TCP proxy and severs every socket the proxy is holding to simulate the drop. I ran it, unmodified, against the pre-fix code and the post-fix code as a check on the fix itself, not just the diff:

[baseline] worker processed first task before any drop: true
--- severing the connection ---
[right after sever] QueueClient.enqueue (reused client) -> id=...-2 error=none
[right after sever] worker processed the second task within 6s: true
[right after sever] worker.run() already threw an unhandled exception: false
[1s later, Redis long since reachable again] QueueClient.enqueue (same client, 3rd call) -> id=...-3 error=none
[1s later] worker processed the third task: true
[done] worker.run() completed normally (no exception escaped): true

Same scenario, same client and worker objects, no code on the calling side changed between the two runs. If you are running redis_task_queue against anything other than a Redis instance you know will never restart, 0.10.1 is worth pulling in.