asyncpg.exceptions._base.InterfaceError: connection is closed

[Ishankoradia]

First check

• [X] I added a descriptive title to this issue.
• [X] I used the GitHub search to find a similar issue and didn't find it.
• [X] I searched the Prefect documentation for this issue.
• [X] I checked that this issue is related to Prefect and not one of its dependencies.

Bug summary

Prefect keeps losing its connection to its own database. Error trace is given below. This usually happens after a deployment is run. I am using prefect version 2.13.5 with a remote postgres database.

Reproduction

I have no idea how to reproduce. On my local it works but on our production setup with a remote database & multiple pipelines running; it keeps throwing this error. 
We have 3 workers serving 2 queues in 1 work pool.

Error

Traceback (most recent call last):
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 709, in _start_transaction
11|prefect |     self._transaction = self._connection.transaction(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/asyncpg/connection.py", line 274, in transaction
11|prefect |     self._check_open()
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/asyncpg/connection.py", line 1399, in _check_open
11|prefect |     raise exceptions.InterfaceError('connection is closed')
11|prefect | asyncpg.exceptions._base.InterfaceError: connection is closed
11|prefect | 
11|prefect | The above exception was the direct cause of the following exception:
11|prefect | 
11|prefect | Traceback (most recent call last):
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1900, in _execute_context
11|prefect |     self.dialect.do_execute(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
11|prefect |     cursor.execute(statement, parameters)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 479, in execute
11|prefect |     self._adapt_connection.await_(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 68, in await_only
11|prefect |     return current.driver.switch(awaitable)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 121, in greenlet_spawn
11|prefect |     value = await result
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 408, in _prepare_and_execute
11|prefect |     await adapt_connection._start_transaction()
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 716, in _start_transaction
11|prefect |     self._handle_exception(error)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 682, in _handle_exception
11|prefect |     raise translated_error from error
11|prefect | sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_dbapi.InterfaceError: <class 'asyncpg.exceptions._base.InterfaceError'>: connection is closed
11|prefect | 
11|prefect | The above exception was the direct cause of the following exception:
11|prefect | 
11|prefect | Traceback (most recent call last):
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/starlette/middleware/errors.py", line 162, in __call__
11|prefect |     await self.app(scope, receive, _send)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/starlette/middleware/gzip.py", line 24, in __call__
11|prefect |     await responder(scope, receive, send)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/starlette/middleware/gzip.py", line 44, in __call__
11|prefect |     await self.app(scope, receive, self.send_with_gzip)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
11|prefect |     raise exc
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
11|prefect |     await self.app(scope, receive, sender)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/prefect/_vendor/fastapi/middleware/asyncexitstack.py", line 20, in __call__
11|prefect |     raise e
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/prefect/_vendor/fastapi/middleware/asyncexitstack.py", line 17, in __call__
11|prefect |     await self.app(scope, receive, send)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/starlette/routing.py", line 718, in __call__
11|prefect |     await route.handle(scope, receive, send)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/starlette/routing.py", line 276, in handle
11|prefect |     await self.app(scope, receive, send)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/starlette/routing.py", line 66, in app
11|prefect |     response = await func(request)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/prefect/server/utilities/server.py", line 101, in handle_response_scoped_depends
11|prefect |     response = await default_handler(request)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/prefect/_vendor/fastapi/routing.py", line 251, in app
11|prefect |     raw_response = await run_endpoint_function(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/prefect/_vendor/fastapi/routing.py", line 177, in run_endpoint_function
11|prefect |     return await dependant.call(**values)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/prefect/server/api/logs.py", line 27, in create_logs
11|prefect |     await models.logs.create_logs(session=session, logs=batch)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/prefect/server/database/dependencies.py", line 119, in async_wrapper
11|prefect |     return await fn(*args, **kwargs)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/prefect/server/models/logs.py", line 45, in create_logs
11|prefect |     await session.execute(log_insert.values([log.dict() for log in logs]))
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/ext/asyncio/session.py", line 214, in execute
11|prefect |     result = await greenlet_spawn(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 126, in greenlet_spawn
11|prefect |     result = context.throw(*sys.exc_info())
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 1714, in execute
11|prefect |     result = conn._execute_20(statement, params or {}, execution_options)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_20
11|prefect |     return meth(self, args_10style, kwargs_10style, execution_options)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection
11|prefect |     return connection._execute_clauseelement(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1572, in _execute_clauseelement
11|prefect |     ret = self._execute_context(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1943, in _execute_context
11|prefect |     self._handle_dbapi_exception(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 2124, in _handle_dbapi_exception
11|prefect |     util.raise_(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
11|prefect |     raise exception
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1900, in _execute_context
11|prefect |     self.dialect.do_execute(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
11|prefect |     cursor.execute(statement, parameters)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 479, in execute
11|prefect |     self._adapt_connection.await_(
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 68, in await_only
11|prefect |     return current.driver.switch(awaitable)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 121, in greenlet_spawn
11|prefect |     value = await result
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 408, in _prepare_and_execute
11|prefect |     await adapt_connection._start_transaction()
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 716, in _start_transaction
11|prefect |     self._handle_exception(error)
11|prefect |   File "/home/ddp/prefect-proxy/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 682, in _handle_exception
11|prefect |     raise translated_error from error
11|prefect | sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) <class 'asyncpg.exceptions._base.InterfaceError'>: connection is closed
11|prefect | [SQL: INSERT INTO log (id, created, updated, name, level, flow_run_id, task_run_id, message, timestamp) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)]
11|prefect | [parameters: (UUID('18ec0dce-76aa-4abb-98f3-e00e1d3ae76b'), DateTime(2024, 6, 3, 16, 55, 42, 869041, tzinfo=Timezone('UTC')), DateTime(2024, 6, 3, 16, 55, 42, 869063, tzinfo=Timezone('UTC')), 'prefect.flow_runs.worker', 20, UUID('00351d7e-5bb7-468c-adce-4017c90ace31'), None, 'Process 606166 exited cleanly.', DateTime(2024, 6, 3, 16, 55, 41, 993709, tzinfo=Timezone('+00:00')))]
11|prefect | (Background on this error at: https://sqlalche.me/e/14/rvf5)

Versions

2.13.5

Additional context

Any guidance on how to approach this would be appreciated.