Tortoise never recovers from Postgres downtime if transactions are used

[henadzit]

Describe the bug

I ran into this while figuring out how connections are managed in Tortoise. Basically when transactions are used and Postgres is going down, Tortoise never recovers after Postgres restarts - the transactions continue to fail with the same error as when Postgres was down. It happens with both asyncpg and psycopg.

Psycopg error logs:

Traceback (most recent call last):
  File "/hidden/tortoise-orm/issues/issue_1598_asyncpg_reconnect.py", line 35, in test
    except Exception as e:
  File "/hidden/tortoise-orm/tortoise/models.py", line 1157, in create
    await instance.save(using_db=db, force_create=True)
  File "/hidden/tortoise-orm/tortoise/models.py", line 948, in save
    await executor.execute_insert(self)
  File "/hidden/tortoise-orm/tortoise/backends/base/executor.py", line 205, in execute_insert
    insert_result = await self.db.execute_insert(self.insert_query, values)
  File "/hidden/tortoise-orm/tortoise/backends/psycopg/client.py", line 99, in execute_insert
    inserted, rows = await self.execute_query(query, values)
  File "/hidden/tortoise-orm/tortoise/backends/base_postgres/client.py", line 41, in _translate_exceptions
    return await self._translate_exceptions(func, *args, **kwargs)
  File "/hidden/tortoise-orm/tortoise/backends/psycopg/client.py", line 166, in _translate_exceptions
    return await func(self, *args, **kwargs)
  File "/hidden/tortoise-orm/tortoise/backends/psycopg/client.py", line 129, in execute_query
    async with connection.cursor(row_factory=row_factory) as cursor:
  File "/hidden/tortoise-orm/.venv/lib/python3.8/site-packages/psycopg/connection_async.py", line 229, in cursor
    self._check_connection_ok()
  File "/hidden/tortoise-orm/.venv/lib/python3.8/site-packages/psycopg/_connection_base.py", line 524, in _check_connection_ok
    raise e.OperationalError("the connection is closed")
psycopg.OperationalError: the connection is closed

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/hidden/tortoise-orm/issues/issue_1598_asyncpg_reconnect.py", line 36, in test
    print("ERROR", e)
  File "/hidden/tortoise-orm/tortoise/backends/base/client.py", line 325, in __aexit__
    await self.connection.rollback()
  File "/hidden/tortoise-orm/tortoise/backends/psycopg/client.py", line 226, in rollback
    await self._connection.rollback()
  File "/hidden/tortoise-orm/.venv/lib/python3.8/site-packages/psycopg/connection_async.py", line 278, in rollback
    await self.wait(self._rollback_gen())
  File "/hidden/tortoise-orm/.venv/lib/python3.8/site-packages/psycopg/connection_async.py", line 414, in wait
    return await waiting.wait_async(gen, self.pgconn.socket, interval=interval)
  File "psycopg_binary/pq/pgconn.pyx", line 204, in psycopg_binary.pq.PGconn.socket.__get__
psycopg.OperationalError: the connection is lost

asyncpg error logs:

ERROR cannot call Transaction.rollback(): the underlying connection is closed
Traceback (most recent call last):
  File "/hidden/tortoise-orm/issues/issue_1598_asyncpg_reconnect.py", line 33, in test
    await SomeModel.create(field1="Some value")
  File "/hidden/tortoise-orm/tortoise/models.py", line 1157, in create
    await instance.save(using_db=db, force_create=True)
  File "/hidden/tortoise-orm/tortoise/models.py", line 948, in save
    await executor.execute_insert(self)
  File "/hidden/tortoise-orm/tortoise/backends/base/executor.py", line 205, in execute_insert
    insert_result = await self.db.execute_insert(self.insert_query, values)
  File "/hidden/tortoise-orm/tortoise/backends/base_postgres/client.py", line 41, in _translate_exceptions
    return await self._translate_exceptions(func, *args, **kwargs)
  File "/hidden/tortoise-orm/tortoise/backends/asyncpg/client.py", line 87, in _translate_exceptions
    return await func(self, *args, **kwargs)
  File "/hidden/tortoise-orm/tortoise/backends/asyncpg/client.py", line 113, in execute_insert
    return await connection.fetchrow(query, *values)
  File "/hidden/tortoise-orm/.venv/lib/python3.8/site-packages/asyncpg/pool.py", line 51, in call_con_method
    raise exceptions.InterfaceError(
asyncpg.exceptions._base.InterfaceError: cannot call Connection.fetchrow(): connection has been released back to the pool

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/hidden/tortoise-orm/issues/issue_1598_asyncpg_reconnect.py", line 34, in test
    await asyncio.sleep(2 + 10 * random.random())
  File "/hidden/tortoise-orm/tortoise/backends/base/client.py", line 325, in __aexit__
    await self.connection.rollback()
  File "/hidden/tortoise-orm/tortoise/backends/asyncpg/client.py", line 198, in rollback
    await self.transaction.rollback()
  File "/hidden/tortoise-orm/.venv/lib/python3.8/site-packages/asyncpg/connresource.py", line 19, in _check
    self._check_conn_validity(meth.__name__)
  File "/hidden/tortoise-orm/.venv/lib/python3.8/site-packages/asyncpg/connresource.py", line 41, in _check_conn_validity
    raise exceptions.InterfaceError(
asyncpg.exceptions._base.InterfaceError: cannot call Transaction.rollback(): the underlying connection is closed

To Reproduce

1. Start Postgres and run the snippet below
2. Stop Postgres, wait until errors appear in the log
3. Start Postgres

Code

import asyncio
import logging
import sys
import random
import traceback

from tortoise import Tortoise, fields, run_async, transactions
from tortoise.models import Model

fmt = logging.Formatter(
fmt="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
sh = logging.StreamHandler(sys.stdout)
sh.setLevel(logging.DEBUG)
sh.setFormatter(fmt)

logger_tortoise = logging.getLogger("tortoise")
logger_tortoise.setLevel(logging.DEBUG)
logger_tortoise.addHandler(sh)

class SomeModel(Model):
field1 = fields.CharField(max_length=10)

async def test():
while True:
try:
async with transactions.in_transaction():
await asyncio.sleep(5 * random.random())
await SomeModel.create(field1="Some value")
await asyncio.sleep(2 + 10 * random.random())
except Exception as e:
print("ERROR", e)
print(traceback.format_exc())

async def run():
await Tortoise.init(
db_url="psycopg://postgres:123456@127.0.0.1:5432/db",
modules={"models": ["main"]},
)
await Tortoise.generate_schemas()

await asyncio.gather(*(test() for _ in range(10)))

if name == "main":
run_async(run())

Expected behavior

Tortoise should recover.