Repository navigation
Postgres async streaming ended prematurely causes asyncio.exceptions.CancelledError: Cancelled by cancel scope. Connections left open #11128
Replies: 7 comments 63 replies
|
we have lots and lots of these problems with people that are using FastAPI. So you might want to also reach out to them as we really are overwhelmed with constant FastAPI-related issues. We dont maintain FastAPI on this end so to the degree that FastAPI can come up with correct patterns for using async database libraries with their framework, it would be very helpful if they can provide this. I would ask what this phrase means, "like it used to" - is this something that behaved differently in a previous SQLAlchemy version? the code in question has changed, however asyncpg's "terminate" most certainly does leave the connection still present on the PG server side for some period of time, that was the entire reason we changed that method to attempt calling close() first. Here's a program illustrating your code, with a CancelledError in the middle and it runs repeatedly. The program uses exactly one connection only if you look at your PostgreSQL status in a different terminal. The exception thrown inside the "with" block will result in the connection returned to the pool normally. it's not actually invalidated, you can turn on from __future__ import annotations
import asyncio
from typing import Optional
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.future import select
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
class Base(AsyncAttrs, DeclarativeBase):
pass
class A(Base):
__tablename__ = "a"
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[Optional[str]]
async def async_main():
"""Main program function."""
engine = create_async_engine(
"postgresql+asyncpg://scott:tiger@localhost/test",
echo=True, echo_pool='debug'
)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async_session = async_sessionmaker(engine, expire_on_commit=False)
async with async_session() as session:
async with session.begin():
session.add_all(
[A() for i in range(500)]
)
while True:
try:
async with async_session() as session:
statement = select(A)
result = await session.stream(statement)
stream_result = result.scalars()
i = 0
async for rows in stream_result.partitions(10):
for r in rows:
i += 1
print(f"Row: {r}")
if i == 39:
raise asyncio.CancelledError("cancelled...")
except asyncio.CancelledError:
message = "A database stream was cancelled"
print(message)
asyncio.run(async_main()) |
|
That will reopen the bug reported in #10717 |
|
@CaselIT I had the same reaction. I was hoping it would just be a simple fix of adding in the missing exception but when I was still getting the same stack trace in the original comment after switching it to |
|
please dont reopen the bug yet because we can't keep changing this back and forth until we have a reproducer. in #10717, we have a reproducer. that's why we made a change. |
|
Yes, there seems to be something else going on. I'm not sure if faskapi cancel is the same as your example above @JordanZimmitti can you create a reproduction example? Even if it has faskapi as a start it's better than nothing, but if you can modify the example by mike it would be super |
|
@CaselIT Yes, if I can figure out how to produce a simple example of it I'll definitely post it here. After further debugging, using my debugger I did notice that the example given to me by @zzzeek does not actually ever call this code until I manually stop the running process. which makes sense why it's working fine without any errors. def terminate(self):
if util.concurrency.in_greenlet():
# in a greenlet; this is the connection was invalidated
# case.
try:
# try to gracefully close; see #10717
# timeout added in asyncpg 0.14.0 December 2017
self.await_(self._connection.close(timeout=2))
except BaseException:
# in the case where we are recycling an old connection
# that may have already been disconnected, close() will
# fail with the above timeout. in this case, terminate
# the connection without any further waiting.
# see issue #8419
self._connection.terminate()
else:
# not in a greenlet; this is the gc cleanup case
self._connection.terminate()
self._started = FalseI'm not sure yet what FastAPI does when a streaming request is terminated that tells SQLAlchemy to call the terminate function above, or why |
|
This is a better example that does actually trigger a terminate: from __future__ import annotations
import asyncio
from typing import Optional
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.future import select
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
class Base(AsyncAttrs, DeclarativeBase):
pass
class A(Base):
__tablename__ = "a"
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[Optional[str]]
async def async_main():
"""Main program function."""
engine = create_async_engine(
"postgresql+asyncpg://scott:tiger@localhost/test",
echo_pool="debug",
connect_args={"server_settings": {"application_name": "foobar"}},
)
async def stat():
async with engine.connect() as conn:
r = await conn.exec_driver_sql(
"""
select
pid,
application_name,
usename,
state,
query
from
pg_stat_activity
where
usename = 'scott'
or application_name = 'foobar'"""
)
print(r.mappings().all())
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async_session = async_sessionmaker(engine, expire_on_commit=False)
async with async_session() as session:
async with session.begin():
session.add_all([A() for i in range(500)])
async def go(session):
statement = select(A)
result = await session.stream(statement)
stream_result = result.scalars()
i = 0
async for rows in stream_result.partitions(10):
for r in rows:
i += 1
if i == 39:
print("cancel at", i)
task.cancel()
print(engine.pool.status())
await stat()
print(engine.pool.status())
for i in range(25):
try:
async with async_session() as session:
task = asyncio.ensure_future(go(session))
await task
except asyncio.CancelledError:
message = "A database stream was cancelled"
print(message)
print(engine.pool.status())
await stat()
print(engine.pool.status())
input("press any key to exit")
asyncio.run(async_main())The issues is still not reproducible at all. The pool keeps reporting: The above will close the asyncpg connection using calling |
|
@CaselIT I use uvloop for async operations could there be something there that could cause SQLalchemy to handle things differently? Sadly I'm away so I haven't had the time to dig deeper and try to create a working example. I'm also not quite sure how to reproduce it in a simple isolated environment either. Very frustrating for me as well but I'll try to keep seeing what I can find to isolate this |
Don't know honestly. This does not seem to be a sqlalchemy thing, since as far as I can see it seem to be behaving correctly, meaning that it closes the connection correctly. Overall try the above script to check if you can reproduce the issue or not with it |
|
Will do, when I get back and am able, I'll try it, thanks! |
|
I'm sorry that I have not been able to continue this conversation sooner, as I was temporarily moved to a different team and needed to put this on pause for a bit. I do have some more information though that I think you guys will find interesting. After further experimentation, I figured out two small issues with the current implementation that are causing connections to not release successfully for me. The first and main reason that Postgres connections are left in an The second reason is that the FastAPI streaming response throws an Below is the current implementation and my proposed changes that fix the issue for me. I have not been able to produce any issues with these changes in my local testing, but I'm curious to get your perspectives on this and see if its something I can make a PR for Thanks!
|
|
I get where you're coming from as it is not fully clear to me either. Thanks for the suggestion and trying to work this through with me. Interestingly the |
I don't think that will be enough. It seems that calling close without rollback leaves the connection idle in transaction in asyncpg. Calling rollback before close does not have this behaviour. It's not clear why
I think you need to install the binary version for psycopg, by default I think it's a python only version. Not sure why. I think it's |
|
@CaselIT Sorry, I should have been more specific. I did install psycopg3[binary] but it's still not nearly as quick sadly.
This above statement is correct, even after adding I would love to find a resolution with If we are worried about messing with the def terminate(self):
if util.concurrency.in_greenlet():
# in a greenlet; this is the connection was invalidated
# case.
try:
# try to gracefully close; see #10717
# timeout added in asyncpg 0.14.0 December 2017
self.rollback() # CHANGE: add rollback call here and leave close the same
self.await_(self._connection.close(timeout=2))
except (
asyncio.CancelledError, # CHANGE: Add CancelledError in list of exceptions to catch
asyncio.TimeoutError,
OSError,
self.dbapi.asyncpg.PostgresError,
):
# in the case where we are recycling an old connection
# that may have already been disconnected, close() will
# fail with the above timeout. in this case, terminate
# the connection without any further waiting.
# see issue #8419
self._connection.terminate()
else:
# not in a greenlet; this is the gc cleanup case
self._connection.terminate()
self._started = FalseTo be honest If I have to reach out to the |
Interesting, I thought they were comparable. I'll try another version of the change above to at least try to make it emit cancelled error on close. @JordanZimmitti You could not create a standalone case even using fastapi and other libraries you are using? |
|
No nothing reproducible yet, next sprint at work I have a spike fully dedicated to this issue. For additional context, this issue exists with our CSV |
|
This morning with a fresh pair of eyes and some coffee, I have more information that I found during debugging that might help clear up some confusion I had and better explain what is going on. So my async def listen_for_disconnect(self, receive: Receive) -> None:
while True:
message = await receive()
if message["type"] == "http.disconnect":
break
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
async with anyio.create_task_group() as task_group:
async def wrap(func: typing.Callable[[], typing.Awaitable[None]]) -> None:
await func()
task_group.cancel_scope.cancel()
task_group.start_soon(wrap, partial(self.stream_response, send))
await wrap(partial(self.listen_for_disconnect, receive))
if self.background is not None:
await self.background()After If you add any line before For example, this change also fixes the problem for me: def terminate(self):
if util.concurrency.in_greenlet():
# in a greenlet; this is the connection was invalidated
# case.
try:
# try to gracefully close; see #10717
# timeout added in asyncpg 0.14.0 December 2017
self.await_(asyncio.sleep(0)) # COMMENT: Even sleeping for zero seconds is enough time for the CancelledError to be raised
self.await_(self._connection.close(timeout=2))
except (
asyncio.CancelledError,
asyncio.TimeoutError,
OSError,
self.dbapi.asyncpg.PostgresError,
):
# in the case where we are recycling an old connection
# that may have already been disconnected, close() will
# fail with the above timeout. in this case, terminate
# the connection without any further waiting.
# see issue #8419
self._connection.terminate()
else:
# not in a greenlet; this is the gc cleanup case
self._connection.terminate()
self._started = FalseThis has been a very interesting problem to assess. Thank you for working with me as I uncover new details |
|
a sleep is much better to add there. though it wouldnt guarantee that the cancelled error wouldnt happen in the close() regardless? |
|
@zzzeek Yeah I'm not sure what the failure rate would be if any since I can't test it in a production-like setting. But I can say that at least in my local testing with many streams going to the API and closing prematurely at the same time I have not seen it fail to close connections |
|
I started a timer before |
|
Update. I can reproduce the issue using anyio, since it seems to be doing things differently from asyncio. This is a reproducer: from __future__ import annotations
import asyncio
import anyio
from sqlalchemy.ext.asyncio import async_sessionmaker
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import select
from sqlalchemy import func
stmt = select(func.generate_series(1, 500).column_valued())
async def with_asyncio(async_session: async_sessionmaker):
async def go_asyncio():
async with async_session() as session:
result = await session.stream_scalars(stmt)
async for _ in result.partitions(10):
task.cancel()
for i in range(3):
try:
task = asyncio.ensure_future(go_asyncio())
await task
except asyncio.CancelledError:
print("A database stream was cancelled")
async def with_anyio(async_session: async_sessionmaker):
async def go_anyio():
async with async_session() as session:
result = await session.stream_scalars(stmt)
async for _ in result.partitions(10):
tg.cancel_scope.cancel()
for _ in range(3):
async with anyio.create_task_group() as tg:
tg.start_soon(go_anyio)
async def run():
await run_case("asyncio", with_asyncio)
await run_case("anyio", with_anyio)
async def run_case(name, callback):
engine = create_async_engine(
"postgresql+asyncpg://scott:tiger@localhost/test",
# echo_pool="debug",
connect_args={"server_settings": {"application_name": name}},
)
async def stat():
async with engine.connect() as conn:
r = await conn.exec_driver_sql(
f"""
select pid, application_name, usename, state, query
from pg_stat_activity
where usename = 'scott' or application_name = '{name}'"""
)
print(r.mappings().all())
async_session = async_sessionmaker(engine, expire_on_commit=False)
print("$$$ using", name)
await stat()
print(engine.pool.status())
await callback(async_session)
print('done running')
await asyncio.sleep(0)
await stat()
print(engine.pool.status())
await engine.dispose()
print("$$$" * 50)
asyncio.run(run())The above prints: More wired things seems to be this: replacing with this (I just added an async def go_anyio():
async with async_session() as session:
result = await session.stream_scalars(stmt)
async for _ in result.partitions(10):
tg.cancel_scope.cancel()
await asyncio.sleep(0)fixes this I have no clue who is at fault here, but I believe that the anyio behavior is surprising here. I don't like adding random sleeps to fix issues with 3rd party libraries, but this seems the easiest solution. |
|
likely not an asyncpg issue. i'd try to understrnad what's happening here and then likely look into anyio |
|
I'm not sure if the issue is of asyncpg or anyio here. I think it's more of anyio. import asyncpg
import asyncio
import anyio
async def run():
name = "test-thing"
async def go_raw_anyio():
conn = await asyncpg.connect(
user="scott", password="tiger", host="localhost", database="test"
)
_transaction = conn.transaction()
await _transaction.start()
cursor = await conn.cursor(
"SELECT anon_1 FROM "
"generate_series(1::INTEGER, 500::INTEGER) AS anon_1"
)
while True:
partition = await cursor.fetch(10)
if partition:
tg.cancel_scope.cancel()
else:
break
for _ in range(5):
async with anyio.create_task_group() as tg:
tg.start_soon(go_raw_anyio)
input("Press Enter to continue...") |
|
i guess what happens is the asyncpg.close() manages to communicate with the server to some extent when the sleep() is present, otherwise does not. |
|
I think its actually the opposite. Its when the close function gets called before the terminate function during the task being cancelled that causes the terminate function to not terminate the connection successfully Well I opened up an issue with asyncpg before I read these latest posts. I'll keep it open anyways maybe they will be able to look at this from a different angle or something |
the asyncio.sleep() hits the exception raise in both test scripts. that somehow makes things "work", so I agree somehow not hitting the .close() helps, and hitting the close() and then doing a raise causes problems. no difference between the sqlalchemy and non-sqlalchemy scripts though, the latter is a direct inlining of the steps SQLAlchemy's dialect takes. When the
|
|
This also kinda reproduse, without asyncpg, but:
import asyncio
import anyio
class MockThingy:
def __init__(self) -> None:
self.things = list(range(100))
async def fetch(self, num):
await asyncio.sleep(0.05)
ret = self.things[:num]
self.things[:num] = []
return ret
async def close(self):
await asyncio.sleep(0.5)
async def partitions(cursor, size):
try:
while True:
partition = await cursor.fetch(size)
if partition:
yield partition
else:
break
except BaseException as e:
print(f"Got exception {e}; attempting to close")
try:
await cursor.close()
except (asyncio.CancelledError, asyncio.TimeoutError, OSError) as e2:
print(
f"Got a second exception {e2} while trying to close, trying to terminate"
)
async def run():
loop = asyncio.get_running_loop()
async def go_asyncio():
thingy = MockThingy()
async for _ in partitions(thingy, 10):
loop.call_soon(task.cancel)
print("$" * 50, "\nasyncio only\n\n")
for i in range(3):
try:
task = asyncio.ensure_future(go_asyncio())
await task
except asyncio.CancelledError:
print("A database stream was cancelled")
print("$" * 50, "\n anyio no async generator\n\n")
async def go_raw_anyio_no_async_gen():
thingy = MockThingy()
try:
while True:
partition = await thingy.fetch(10)
if partition:
tg.cancel_scope.cancel()
else:
break
except BaseException as e:
print(f"Got exception {e}; attempting to close")
try:
await thingy.close()
except (asyncio.CancelledError, asyncio.TimeoutError, OSError) as e2:
print(
f"Got a second exception {e2} while trying to close, trying to terminate"
)
for _ in range(3):
async with anyio.create_task_group() as tg:
tg.start_soon(go_raw_anyio_no_async_gen)
print("$" * 50, "\n anyio async generator\n\n")
async def go_raw_anyio():
thingy = MockThingy()
async for _ in partitions(thingy, 10):
tg.cancel_scope.cancel()
for _ in range(3):
async with anyio.create_task_group() as tg:
tg.start_soon(go_raw_anyio)
input("Press Enter to continue...")
if __name__ == "__main__":
asyncio.run(run()) |
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Describe the bug
Hi, hoping someone can help me with my issue!
When I asynchronously stream data to a client via an API, the connection no longer closes correctly when the client closes the stream before it has fully completed. Now an error occurs (stack trace below) and the connection stays open in Postgres. Also, the pool does not recognize that the connection is still open so connections can quickly grow over the
max_overflowsetting. So far this issue persists for all versions starting at2.0.24 to latest (2.0.28 at time of writing).I'm hoping to get some insight on this issue. If more information is needed i'd be happy to supply more where I can.
Thanks in advance!
Below is a code snippet of how I am connecting to the Postgres db
Below is a code snippet of how I am using sqlalchemy to stream the data
After doing some digging, I figured out that the error is occurring here as a two-fold issue. Currently, the exception raised never actually gets handled and sent to
self._connection.terminate(). Also, even when I change it toBaseExceptionandself._connection.terminate()gets invoked, whatever error occurs fromself.await_(self._connection.close(timeout=2))stops the terminate function from closing the connection. like it used toIf I change the function back to this everything works fine:
Optional link from https://docs.sqlalchemy.org which documents the behavior that is expected
No response
SQLAlchemy Version in Use
SQLAlchemy 2.0.28
DBAPI (i.e. the database driver)
asyncpg
Database Vendor and Major Version
PostgreSQL 14
Python Version
3.11
Operating system
OSX/Linux
To Reproduce
# I do not have a small isolated working example of this yet as it is part of a bigger FastAPI projectError
Additional context
Other notable library versions used:
asyncpg = "^0.29.0"greenlet = "^3.0.3"psycopg2-binary = "^2.9.8"uvloop = "^0.19.0"All reactions