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.