Pool.release() received invalid connection

[d3QUone]

Describe the bug

The situation:

• Run 20 ~simultaionius SELECT queries using asyncio.create_task
• Queries start failing with exceptions.InterfaceError error
• I except it works fine inside every coroutine

To Reproduce

Minimal code to reproduce the bug:

import asyncio
import uuid

from tortoise import Tortoise, transactions


async def run_one():
    async with transactions.in_transaction():
        incident_id = uuid.uuid4()
        # Any model, any query here 
        count = await Order.filter(incident_id=incident_id).count()
        print(f"found {count} by {incident_id}")


async def run_many():
    await Tortoise.init(
        config=...,
    )

    tasks = []
    for _ in range(20):
        task = asyncio.create_task(run_one())
        tasks.append(task)
    await asyncio.gather(*tasks)

    await Tortoise.close_connections()


if __name__ == '__main__':
    asyncio.run(run_many())

Exception:

Coroutine exception:
Traceback (most recent call last):
  File "file.py", line 32, in inner
    return await func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "file.py", line 35, in run_one
    async with transactions.in_transaction() as connection:
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "contrib/python/tortoise-orm/tortoise/backends/base/client.py", line 302, in __aexit__
    await self.connection._parent._pool.release(self.connection._connection)
  File "contrib/python/asyncpg/asyncpg/pool.py", line 884, in release
    raise exceptions.InterfaceError(
asyncpg.exceptions._base.InterfaceError: Pool.release() received invalid connection: <PoolConnectionProxy <asyncpg.connection.Connection object at 0x10ce2e990> 0x10cf49ae0> is not a member of this pool

Expected behavior

• It worked fine without any exception

Additional context

Versions:

asyncpg==0.30.0
tortoise-orm==0.22.2

How to fix: I've found a workaround, but I think it should be at least documented, at most – fixed:

Run "SELECT" and wait before running any business-logic code:

await Tortoise.init(...)

async with transactions.in_transaction() as connection:
    await connection.execute_query("SELECT 1;")

# Business-logic ...

[henadzit]

@d3QUone thank you for the detailed report! I couldn't reproduce the issue. Is there anything specific about how you configure the database connection or how the database server is configured? Could you provide the code for Order?

The runnable script that I used for testing:

import asyncio
import uuid

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


class Tournament2(Model):
    id = fields.UUIDField(primary_key=True, default=uuid.uuid4)


async def run_one():
    async with transactions.in_transaction():
        incident_id = uuid.uuid4()
        # Any model, any query here
        count = await Tournament2.filter(id=incident_id).count()
        print(f"found {count} by {incident_id}")


async def run_many():
    await Tortoise.init(
        db_url="asyncpg://postgres:[email protected]:5432",
        modules={"models": ["__main__"]},
    )
    await Tortoise.generate_schemas()

    tasks = []
    for _ in range(1000):
        task = asyncio.create_task(run_one())
        tasks.append(task)
    await asyncio.gather(*tasks)

    await Tortoise.close_connections()


if __name__ == "__main__":
    asyncio.run(run_many())

[andrewshkovskii]

I ran into a similar issue but with a nested transactions block.

The current connection is stored in the context. When you create a coroutine, the current context gets copied, and the coroutine runs in that context. So, when you have nested transaction blocks, the inner coroutine will re-use the same connection as the outer, which is logical. So now you have nested transactions using the same connection.

But the problem is on rollback.

    async def rollback(self) -> None:
        if self._finalized:
            raise exceptions.TransactionManagementError("Transaction already finalised")

        await self._connection.rollback()
        await self._connection.set_autocommit(True)
        self._finalized = True

When you try to set autocommit back to True on the connection already in a transaction, you will get an error:

psycopg.ProgrammingError: can't change 'autocommit' now: connection in transaction status INTRANS

I was able to reproduce the issue using this code:

from asyncio import gather, get_event_loop
from contextlib import asynccontextmanager
import contextvars
from tortoise.connection import connections
from tortoise import Tortoise
from tortoise.transactions import in_transaction, atomic

from tournaments.domains.buy_ins.models import BuyIn
from tournaments.entrypoints.tortoise_init import TORTOISE_CONFIG

loop = get_event_loop()

null_context = contextvars.Context()


async def outer_coro():
    async with in_transaction() as conn:
        print("outer_coro")
        b = await BuyIn.all().select_for_update().using_db(conn)
        try:
            # coros = [
            #     loop.create_task(inner_coro(), context=null_context.copy())
            #     for i in range(9)
            # ]
            coros = (inner_coro(), inner_coro(),
                     inner_coro(), inner_coro(),
                     inner_coro(), inner_coro())
            await gather(*coros)

        except Exception as exc:
            print(f"Inner error {exc!r}")
            raise Exception("error2") from exc

async def inner_coro():
    print("inner coro")
    async with in_transaction() as conn:
        b = await BuyIn.all()
        raise ValueError("Error")


async def main():
    await Tortoise.init(**TORTOISE_CONFIG)
    await outer_coro()

loop.run_until_complete(main())

The code will even stuck on one of the inner coroutines if you will try to open one more transaction, even if pool minsize is 50.

If you switch to creating tasks with new context -- the inner coroutines will use new connections all run fine.

Using the same connection in the nested coroutines is logical and right, but using rollback with resetting auto-commit causes this problem. Moreover, I can't find a proper public API to use a new connection/client for nested blocks. [related https://github.com//issues/1128]

[henadzit]

@andrewshkovskii, thank you for reporting! Although I think it's different issue. Would you mind opening a new issue with your report?

The code will even stuck on one of the inner coroutines if you will try to open one more transaction, even if pool minsize is 50.

I believe this is going to be fixed with #1810

[d3QUone]

@henadzit, hi! Thank you for your response!

I believe the model is pretty simple, actually I can reproduce it with any model in my project:

import enum

from tortoise import fields, models


class ServiceName(enum.StrEnum):
    METRICS = "metrics"
    NOTIFICATIONS = "notifications"


class OrderEventType(enum.StrEnum):
    START_INCIDENT = "start_incident"
    FINISH_INCIDENT = "finish_incident"


class OrderState(enum.StrEnum):
    NEW = "new"
    SENT = "sent"
    COMPLETED = "completed"


class Incident(models.Model):
    id = fields.UUIDField(
        primary_key=True,
    )


class Order(models.Model):
    id = fields.BigIntField(
        primary_key=True,
    )
    incident = fields.ForeignKeyField(
        model_name="models.Incident",
        related_name="orders",
        on_delete=fields.OnDelete.CASCADE,
    )
    created_at = fields.DatetimeField(
        auto_now_add=True,
    )
    updated_at = fields.DatetimeField(
        auto_now=True,
    )
    sent_at = fields.DatetimeField(
        null=True,
    )
    completed_at = fields.DatetimeField(
        null=True,
    )
    state = fields.CharEnumField(
        enum_type=OrderState,
        max_length=16,
        db_index=True,
    )
    order_type = fields.CharEnumField(
        enum_type=OrderEventType,
        max_length=64,
        db_index=True,
    )
    receiver = fields.CharEnumField(
        enum_type=ServiceName,
        max_length=64,
        db_index=True,
    )
    queue_message = fields.JSONField()

May be it related to connection settings. I've specified timeouts and pool-connections there:

{
   "connections":{
      "default":{
         "engine":"tortoise.backends.asyncpg",
         "credentials":{
            "application_name":"not_set",
            "minsize":10,
            "maxsize":20,
            "timeout":1,
            "command_timeout":1,
            "server_settings": {
                "statement_timeout": 1000,
            },
            "port":5732,
            "database":"db_local",
            "host":"127.0.0.1",
            "user":"admin",
            "password":"admin"
         }
      }
   },
   "apps":{
      "models":{
         "models":[
            "src.core.incidents.models",
            "aerich.models"
         ],
         "default_connection":"default"
      }
   },
   "use_tz":True,
   "timezone":"Europe/London"
}

[henadzit]

@d3QUone, I was finally able to reproduce the issue! Basically it happens because multiple tasks are trying to initialize connection pools, however, the connection pool should be one. The fix would be to introduce a lock or initialize the pool in Tortoise.init. Thank you for reporting!

@andrewshkovskii , it's quite a coincidence that at the time of your comment I worked on implementing transaction savepoints that seem to be addressing your concerns. Here's the PR.

[d3QUone]

I think initializing the pool inside Tortoise.init sounds correct, I don't think it should be "lazy", but with lock.