sqlalchemy: asyncpg terminate needs a delay

Describe the bug

Greetings, I came across a strange behavior: when you pass command_timeout param to asyncpg driver and the query is long enough to exceed it, there still will be an active session on PostgreSQL side

You may also check it by using select * from pg_stat_activity;

Optional link from https://docs.sqlalchemy.org which documents the behavior that is expected

No response

SQLAlchemy Version in Use

2.0.23

DBAPI (i.e. the database driver)

asyncpg

Database Vendor and Major Version

PostgreSQL 14

Python Version

3.10

Operating system

OSX

To Reproduce

import asyncio

import sqlalchemy as sa
from sqlalchemy import URL
from sqlalchemy.ext.asyncio import create_async_engine


async def test_asyncpg_with_command_timeout():
    engine = create_async_engine(
        URL.create(
            drivername='postgresql+asyncpg',
            host='localhost',
            port=5432,
            database='test',
            username='test',
            password='test',
        ),
        connect_args={
            'command_timeout': 2,
        },
    )

    try:
        async with engine.connect() as conn:
            await conn.execute(sa.text('select pg_sleep(:time)'), parameters={'time': 30})
    except asyncio.TimeoutError:
        print('gotcha')

    finally:
        await engine.dispose(close=True)


asyncio.run(test_asyncpg_with_command_timeout())

Error

2023-11-30 11:21:54.103 UTC [34] STATEMENT:  select pg_sleep($1)
2023-11-30 11:21:54.103 UTC [34] LOG:  execute __asyncpg_stmt_7__: select pg_sleep($1)
2023-11-30 11:21:54.103 UTC [34] DETAIL:  parameters: $1 = '30'
2023-11-30 11:22:24.115 UTC [34] LOG:  disconnection: session time: 0:00:30.126 user=test database=test host=192.168.65.1 port=25931
2023-11-30 11:22:24.115 UTC [34] DEBUG:  exit(0)

Additional context

I made my own investigation and the problem is because here https://github.com/MagicStack/asyncpg/blob/master/asyncpg/connect_utils.py#L1036 CancelledError is thrown, so asyncpg can not send CancelRequest message, but if I add a small delay here https://github.com/sqlalchemy/sqlalchemy/blob/main/lib/sqlalchemy/dialects/postgresql/asyncpg.py#L617 it will succeed. I know that it is not the solution, but maybe it will save some time