Worker.consume_queue hides database errors and permanently disables the worker
When using the database backend, the worker can get into an infinite error loop (see attached file for a log, the error message repeats every second), if the database connection fails.
Worker.consume_queue calls self.backend.fetch_batch or self.backend._fetch_task. For the database backend, both of these use sync_to_async to talk to the database.
If the database connection fails, then fetch_batch or _fetch_task will throw (for example django.db.utils.OperationalError: the connection is lost). This exception is caught by the except Exception in consume_queue's loop. This handler does nothing but log the exception and sleep for 1 second. Then on the next run, the same broken connection will throw the same error. This prevents the worker from ever fetching tasks.
Solution:
consume_queue should be calling _async_close_old_connections, otherwise old connections never get closed if there are never any tasks.
A workaround is to set Django's CONN_HEALTH_CHECKS to true (especially when using Psycopg's pool), because that runs a health check on every connection returned from the pool.