InterfaceError: connection already closed

I'm pushing 500+ tasks per minute through Dramatiq as part of a Django app and occasionally (once every 1-2 days) I suddenly get hundreds instances of this error from the workers and the only way to fix it is to restart the worker process.

Sentry reports that it's triggered in both `before_process_message()` and `after_process_message()` when they both call `Task.tasks.create_or_update_from_message()`.

I have `CONN_MAX_AGE` set to `0` and database connections are pooled via PgBouncer.

Please let me know if I'm missing any other information.