Asyncpg conflicts with autocommit_block

[ionsome]

Describe the bug
Creating postgresql index concurrently in migrations using op.get_context().autocommit_block() may cause a program stuck if you have opened asyncpg transaction.

Expected behavior
The program shouldn't stuck and migrations must be applied

To Reproduce

I've created a separated repo with reproducible examples: https://github.com/ionsome/alembic-asyncpg-concurrently-poc

docker-compose up -d
python3 example1_async_apply.py

After this logs examples stucks.

2024-04-20 15:11:56,002 INFO sqlalchemy.engine.Engine [no key 0.00014s] ()
2024-04-20 15:11:56,012 INFO sqlalchemy.engine.Engine COMMIT
2024-04-20 15:11:56,019 INFO sqlalchemy.engine.Engine BEGIN (implicit; DBAPI should not BEGIN due to autocommit mode)
2024-04-20 15:11:56,020 INFO sqlalchemy.engine.Engine INSERT INTO table_1 VALUES(1, 'aboba')
2024-04-20 15:11:56,020 INFO sqlalchemy.engine.Engine [generated in 0.00014s] ()
2024-04-20 15:11:56,024 INFO sqlalchemy.engine.Engine SELECT 1
2024-04-20 15:11:56,024 INFO sqlalchemy.engine.Engine [generated in 0.00017s] ()
2024-04-20 15:11:56,026 INFO sqlalchemy.engine.Engine CREATE INDEX CONCURRENTLY ix_1 ON table_1 (data)
2024-04-20 15:11:56,026 INFO sqlalchemy.engine.Engine [generated in 0.00016s] ()

Versions.

• OS: macOS 14.2.1 (23C71), M1 Pro
• Python: 3.10
• Alembic: 1.13.1
• SQLAlchemy: 2.0.29
• Database: postgresql 16
• DBAPI: asyncpg 0.29.0, psycopg2 2.9.9

Have a nice day!

[CaselIT]

Hi,

I'm not sure how do you expect alembic to behave here. If the db requeires exclusive access on a table alembic can't automatically decide to terminate extra transactions.

If you want to do that you can issue a command yourself using execute

[ionsome]

How is a transaction in a different connection, that has only SELECT 1, related to exclusive access?
It looks like a bug because it works with psycopg2 (example 3 and example 4 in repo).

[ionsome]

Even this migration won't be applied if there is a another asyncpg connection from sqlalchemy:

def upgrade() -> None:
    op.create_table(
        'table_1',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('data', sa.String(), nullable=False),
    )
    
    op.execute('COMMIT')
    op.execute('CREATE INDEX CONCURRENTLY ix_1 ON table_1 (data)') # stucks
    op.execute('SELECT 1') # not reached

But if you change this line for psycopg2 connection or use raw asyncpg connection then everything will work.

[ionsome]

It looks like there is a some kind of mutex in sqlalchemy 🤔

[ionsome]

It's possible to reproduce the problem without alembic :)

import asyncio
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import text


async_engine = create_async_engine(
    'postgresql+asyncpg://user:password@127.0.0.1:5432/test_db',
    echo=True,
)

async def main() -> None:
    async with async_engine.connect() as connection1:
        await connection1.execute(
            text(
            '''CREATE TABLE IF NOT EXISTS table_1 (
                id INTEGER NOT NULL, 
                data VARCHAR NOT NULL
            )'''
            )
        )
        await connection1.commit()

    async with async_engine.connect() as connection1, \
        async_engine.connect() as connection2:
            await connection1.execute(text('SELECT 1'))
            await connection2.execution_options(isolation_level="AUTOCOMMIT")
            await connection2.execute(text('CREATE INDEX CONCURRENTLY ix_4 ON table_1 (data)'))  # stucks


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

[CaselIT]

this is an asyncpg thing. See sqlalchemy/sqlalchemy#11299 (comment) and sqlalchemy/sqlalchemy#11299 (comment)

TLDR: use psycopg if you want to run migration concurrently to other connections

[ionsome]

Yes, I think the same :)
For more context – I'm used to prevent concurrent applying of migrations by holding an advisory lock in a separate connection. I had a working solution based on synchronous sqlalchemy, but wanted to rewrite it and make asynchronous.