asyncio: cancelling a task that performs insert operations in parallel can raise unexpected errors #11761

dominik-schwabe · 2024-08-19T17:34:18Z

dominik-schwabe
Aug 19, 2024

Describe the bug

When using asyncio to insert entries in parallel, cancelling the insert task can lead to unexpected errors that are related to connection issues in that special case.
I suppose this is a bug because I expect it to perform cleanup that leaves the connections in a consistent state without raising an error and I couldn't find something at https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html that documents this behavior as expected.

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

No response

SQLAlchemy Version in Use

2.0.32

DBAPI (i.e. the database driver)

asyncpg

Database Vendor and Major Version

PostgreSQL 16

Python Version

3.12

Operating system

Linux

To Reproduce

from __future__ import annotations

import asyncio
from collections.abc import Coroutine, Sequence
from contextvars import ContextVar
from pathlib import Path
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, MappedAsDataclass, mapped_column

# --------------- BOILERPLATE START ---------------

SESSION_CTX: ContextVar[async_sessionmaker[AsyncSession]] = ContextVar("SESSION_CTX")


class Base(MappedAsDataclass, DeclarativeBase):
    pass


class Data(Base):
    __tablename__ = "data"

    id: Mapped[int] = mapped_column(init=False, primary_key=True)


async def run_with_db(coro: Coroutine[Any, Any, Any], database_url: str):
    engine = create_async_engine(database_url, echo=False)
    try:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.drop_all)
            await conn.run_sync(Data.metadata.create_all)
        SESSION_CTX.set(async_sessionmaker(engine, expire_on_commit=False))
        return await coro
    finally:
        await engine.dispose()


async def cancel_and_propagate(tasks: Sequence[asyncio.Task[Any]]):
    "cancels all tasks and propagates potential errors that are not CancelErrors"
    for task in tasks:
        task.cancel()
    # the following is just convenience to make sure the program finishes when an error get raised in a task
    await asyncio.wait(tasks)
    for task in tasks:
        if not task.cancelled():
            exc = task.exception()
            if exc is not None:
                raise exc


# ---------------- BOILERPLATE END ----------------


async def insert_new():
    sessionmaker = SESSION_CTX.get()
    async with sessionmaker() as session:
        session.add(Data())
        await session.commit()


async def insert_many():
    "insert a bunch of entries in parallel"
    while True:
        tasks = [asyncio.create_task(insert_new()) for _ in range(2)]
        try:
            await asyncio.gather(*tasks)
        finally:
            await cancel_and_propagate(tasks)  # proper cleanup


async def task_insert_many():
    "create a task that runs insert_many, wait 1 second and kill the task"
    while True:
        task = asyncio.create_task(insert_many())
        try:
            await asyncio.sleep(1)
        finally:
            await cancel_and_propagate([task])


# DATABASE_URL = f"sqlite+aiosqlite:///{Path('~/.cache/async-cancel-debug.db').expanduser()}"
DATABASE_URL = "postgresql+asyncpg://debug:debug@localhost:5432/async-cancel-debug"


asyncio.run(run_with_db(task_insert_many(), DATABASE_URL))
# this should fail fairly quickly with the following error:
# - sqlite3: sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no active connection
# - postgresql: sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) <class 'asyncpg.exceptions._base.InterfaceError'>: cannot call Transaction.rollback(): the underlying connection is closed

# asyncio.run(run_with_db(insert_many(), DATABASE_URL))  # this runs without errors

Error

Exception terminating connection <AdaptedConnection <asyncpg.connection.Connection object at 0x7683fa5a5310>>
Traceback (most recent call last):
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/pool/base.py", line 374, in _close_connection
    self._dialect.do_terminate(connection)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 1100, in do_terminate
    dbapi_connection.terminate()
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 879, in terminate
    self.await_(self._connection.close(timeout=2))
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 132, in await_only
    return current.parent.switch(awaitable)  # type: ignore[no-any-return,attr-defined] # noqa: E501
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 196, in greenlet_spawn
    value = await result
            ^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/asyncpg/connection.py", line 1466, in close
    await self._protocol.close(timeout)
  File "asyncpg/protocol/protocol.pyx", line 617, in close
asyncio.exceptions.CancelledError
Traceback (most recent call last):
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 830, in _start_transaction
    self._transaction = self._connection.transaction(
                        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/asyncpg/connection.py", line 302, in transaction
    self._check_open()
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/asyncpg/connection.py", line 1533, in _check_open
    raise exceptions.InterfaceError('connection is closed')
asyncpg.exceptions._base.InterfaceError: connection is closed

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1967, in _exec_single_context
    self.dialect.do_execute(
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 941, in do_execute
    cursor.execute(statement, parameters)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 572, in execute
    self._adapt_connection.await_(
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 132, in await_only
    return current.parent.switch(awaitable)  # type: ignore[no-any-return,attr-defined] # noqa: E501
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 196, in greenlet_spawn
    value = await result
            ^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 508, in _prepare_and_execute
    await adapt_connection._start_transaction()
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 837, in _start_transaction
    self._handle_exception(error)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 784, in _handle_exception
    raise translated_error from error
sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_dbapi.InterfaceError: <class 'asyncpg.exceptions._base.InterfaceError'>: connection is closed

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/dominik/database-debug/experiments.py", line 86, in <module>
    asyncio.run(run_with_db(task_insert_many(), DATABASE_URL))
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/asyncio/runners.py", line 194, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/asyncio/base_events.py", line 685, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/home/dominik/database-debug/experiments.py", line 34, in run_with_db
    return await coro
           ^^^^^^^^^^
  File "/home/dominik/database-debug/experiments.py", line 79, in task_insert_many
    await cancel_and_propagate([task])
  File "/home/dominik/database-debug/experiments.py", line 49, in cancel_and_propagate
    raise exc
  File "/home/dominik/database-debug/experiments.py", line 69, in insert_many
    await cancel_and_propagate(tasks)  # proper cleanup
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/database-debug/experiments.py", line 49, in cancel_and_propagate
    raise exc
  File "/home/dominik/database-debug/experiments.py", line 67, in insert_many
    await asyncio.gather(*tasks)
  File "/home/dominik/database-debug/experiments.py", line 59, in insert_new
    await session.commit()
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/ext/asyncio/session.py", line 1009, in commit
    await greenlet_spawn(self.sync_session.commit)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 203, in greenlet_spawn
    result = context.switch(value)
             ^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 2028, in commit
    trans.commit(_to_root=True)
  File "<string>", line 2, in commit
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/state_changes.py", line 139, in _go
    ret_value = fn(self, *arg, **kw)
                ^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 1313, in commit
    self._prepare_impl()
  File "<string>", line 2, in _prepare_impl
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/state_changes.py", line 139, in _go
    ret_value = fn(self, *arg, **kw)
                ^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 1288, in _prepare_impl
    self.session.flush()
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4352, in flush
    self._flush(objects)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4487, in _flush
    with util.safe_reraise():
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/util/langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/session.py", line 4448, in _flush
    flush_context.execute()
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/unitofwork.py", line 466, in execute
    rec.execute(self)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/unitofwork.py", line 642, in execute
    util.preloaded.orm_persistence.save_obj(
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/persistence.py", line 93, in save_obj
    _emit_insert_statements(
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/orm/persistence.py", line 1233, in _emit_insert_statements
    result = connection.execute(
             ^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1418, in execute
    return meth(
           ^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/sql/elements.py", line 515, in _execute_on_connection
    return connection._execute_clauseelement(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1640, in _execute_clauseelement
    ret = self._execute_context(
          ^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1846, in _execute_context
    return self._exec_single_context(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1986, in _exec_single_context
    self._handle_dbapi_exception(
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 2355, in _handle_dbapi_exception
    raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1967, in _exec_single_context
    self.dialect.do_execute(
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/engine/default.py", line 941, in do_execute
    cursor.execute(statement, parameters)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 572, in execute
    self._adapt_connection.await_(
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 132, in await_only
    return current.parent.switch(awaitable)  # type: ignore[no-any-return,attr-defined] # noqa: E501
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py", line 196, in greenlet_spawn
    value = await result
            ^^^^^^^^^^^^
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 508, in _prepare_and_execute
    await adapt_connection._start_transaction()
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 837, in _start_transaction
    self._handle_exception(error)
  File "/home/dominik/.local/share/mise/installs/python/3.12.2/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 784, in _handle_exception
    raise translated_error from error
sqlalchemy.exc.InterfaceError: (sqlalchemy.dialects.postgresql.asyncpg.InterfaceError) <class 'asyncpg.exceptions._base.InterfaceError'>: connection is closed
[SQL: INSERT INTO data DEFAULT VALUES RETURNING data.id]
(Background on this error at: https://sqlalche.me/e/20/rvf5)

Additional context

This also happens when using sqlite with aiosqlite.

SQLAlchemy==2.0.32
asyncpg==0.29.0
aiosqlite==0.20.0

zzzeek · 2024-08-19T17:36:46Z

zzzeek
Aug 19, 2024
Maintainer

sorry what behavior do you expect? are you indicating that as the program continues after these exception raises that the connection pool is corrupted? cancelling a task is going to be fairly chaotic from any perspective I can think of

0 replies

zzzeek · 2024-08-19T17:37:51Z

zzzeek
Aug 19, 2024
Maintainer

for example I dont see the code trying to catch any CancelledError, so not clear what the desired behavior would be

0 replies

zzzeek · 2024-08-19T17:40:19Z

zzzeek
Aug 19, 2024
Maintainer

it is expected that the connection will be closed on a cancellation also, since the state of such a connection can't be guaranteed. this is a discussion and at best some extra notes in the docs

0 replies

dominik-schwabe · 2024-08-19T17:50:45Z

dominik-schwabe
Aug 19, 2024
Author

The function cancel_and_propagate cancels the tasks, waits for the tasks to finish and checks if all tasks were cancelled successfully. If that is the case it continues execution without any problems. If at least one task contains an error (e.g. sqlalchemy.exc.InterfaceError) it raises it so the program doesn't run forever. This is why it is not necessary to capture CancelledErrors.

The main issue is that the connection that the task used was somehow corrupted (or something) and is returned to the pool and when another tasks get this connection from the pool it raises an error when using this connection because it is corrupted. At least that is what I think that happens.

The reason why I think SQLAlchemy should handle cancellations correctly is that python 3.11 introduced TaskGroups and uncancel which can be used to build clean code that unfortunately relies a lot on cancellation and therefore proper cleanup.
Another example might be a fastapi.responses.StreamingResponse that cancels the underlying stream when a client disconnects which could lead to unexpected errors should there be an insert operations at the same moment.

Also I am not aware if this only happens when inserting or also when performing different operations.

10 replies

zzzeek Aug 19, 2024
Maintainer

I found #11128 so far, not sure if there are others

zzzeek Aug 19, 2024
Maintainer

and of course the original issue where it was improved in #10717

zzzeek Aug 19, 2024
Maintainer

yah same change was proposed at #11128 (comment)

let's refer to the discussion following that comment. in that case we could not be given a reproducer

zzzeek Aug 19, 2024
Maintainer

using psycopg seems consistent with swallowing the CancelledError

can you clarify this statement? we dont seem to have any terminate() implementation for psycopg async, this means the connection pool would just throw away the connection and produce a warning if CancelledError occurs, meaning a server side session would be left opened. can we confirm that?

CaselIT Aug 19, 2024
Maintainer

I haven't verified on the server if there are differences. I was referring only to the script log using echo_pool=True.
The pg behaviour should be checked

For psycopg we actually call close, that seems a sync method, similar to asyncpg terminate (don't know if they do the same thing).

zzzeek · 2024-09-02T15:15:52Z

zzzeek
Sep 2, 2024
Maintainer

#11821 will add CancelledError

0 replies

CaselIT · 2025-07-10T20:54:03Z

CaselIT
Jul 10, 2025
Maintainer

This fix caused the issue reported at #12727

The proposed fix will change the behavior of the provided script so that instead of continuing it will terminate with a variation of <class 'asyncpg.exceptions._base.InterfaceError'>: cannot call Transaction.rollback(): the underlying connection is closed

0 replies

asyncio: cancelling a task that performs insert operations in parallel can raise unexpected errors #11761

Uh oh!

dominik-schwabe Aug 19, 2024

Describe the bug

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

SQLAlchemy Version in Use

DBAPI (i.e. the database driver)

Database Vendor and Major Version

Python Version

Operating system

To Reproduce

Error

Additional context

Replies: 6 comments · 10 replies

Uh oh!

zzzeek Aug 19, 2024 Maintainer

Uh oh!

zzzeek Aug 19, 2024 Maintainer

Uh oh!

zzzeek Aug 19, 2024 Maintainer

Uh oh!

Uh oh!

dominik-schwabe Aug 19, 2024 Author

Uh oh!

zzzeek Aug 19, 2024 Maintainer

Uh oh!

zzzeek Aug 19, 2024 Maintainer

Uh oh!

zzzeek Aug 19, 2024 Maintainer

Uh oh!

zzzeek Aug 19, 2024 Maintainer

Uh oh!

Uh oh!

CaselIT Aug 19, 2024 Maintainer

Uh oh!

zzzeek Sep 2, 2024 Maintainer

Uh oh!

CaselIT Jul 10, 2025 Maintainer

dominik-schwabe
Aug 19, 2024

Replies: 6 comments 10 replies

zzzeek
Aug 19, 2024
Maintainer

zzzeek
Aug 19, 2024
Maintainer

zzzeek
Aug 19, 2024
Maintainer

dominik-schwabe
Aug 19, 2024
Author

zzzeek Aug 19, 2024
Maintainer

zzzeek Aug 19, 2024
Maintainer

zzzeek Aug 19, 2024
Maintainer

zzzeek Aug 19, 2024
Maintainer

CaselIT Aug 19, 2024
Maintainer

zzzeek
Sep 2, 2024
Maintainer

CaselIT
Jul 10, 2025
Maintainer