SQLAlchemy proper handling of deadlocks

Question

I am using the SQLAlchemies async engine with the following initialization and would like to recover from deadlocks. As I could not find any information about that in the docs, I decided to use the timeout parameter.

 from sqlalchemy.ext.asyncio import create_async_engine 

    engine = create_async_engine(
        database_url, 
        connect_args={'timeout': timeout}, 
        pool_size=16, 
        max_overflow=10
    )

The problem that I am facing now is that I could not find a proper way to recover from TimeoutErrors.

Attempt Number 1

    for i in range(max_retries):
        try:
            async with self.engine.connect() as conn:
                async with conn.begin() as transaction:
                     output = await conn.execute(query)

        except asyncio.exceptions.TimeoutError as te:
            logger.info(f"Connection timeout! Trying again. Attempt: {i + 1}/{max_retries}")
        else:
            return output

Causes the following error:

asyncpg.exceptions.ConnectionDoesNotExistError: connection was closed in the middle of operation

Attempt Number 2

for i in range(max_retries):
            try:
                async with self.engine.connect() as conn:
                    transaction = await conn.begin()
                    try:
                        output = await conn.execute(query)
                        await transaction.commit()

                    except:
                        await transaction.rollback()
                        await transaction.close()
                        continue
                    else:
                        await transaction.close()
                        return output

            except asyncio.exceptions.TimeoutError as te:
                logger.info(f"Connection timeout! Trying again. Attempt: {i + 1}/{max_retries}")
            else:
                return output

Causes the following error:

asyncio.exceptions.InvalidStateError: invalid state

So my question is, what should I do?

Edit

After further testing, I realized that the timeout parameter does not work. When the deadlock occurs SQLAlchemy stops to respond and asyncio fills the log with the following messages:

INFO:asyncio:poll 55776.892 ms took 55832.809 ms: timeout
INFO:asyncio:poll 50130.676 ms took 50181.180 ms: timeout
INFO:asyncio:poll 60008.691 ms took 60031.415 ms: timeout
INFO:asyncio:poll 29676.484 ms took 29706.735 ms: timeout
INFO:asyncio:poll 30080.565 ms took 30111.141 ms: timeout
INFO:asyncio:poll 60180.535 ms took 60234.353 ms: timeout

Interestingly the rest of the application is not affected by it and continues to work as expected.

More information

The problem also occurs when using MySQL instead of PostgreSQL. No select or update query is performed. It is sufficient to execute a burst of about 1000 inserts into about 300 different tables to cause this problem with a certain probability. In our application, the engine locks up after about 10 minutes of usage. The only way we have found to mitigate this issue is to kill and restart the python script every 5 minutes.

SQLAlchemy proper handling of deadlocks

0 Answers