New `asyncio.exceptions.CancelledError` from 4.5.2

[rra]

Version: 4.5.2
Platform: Python 3.11.2 on Debian unstable
Description:

Starting with 4.5.2, I'm seeing a new exception when attempting to write to Redis using the asyncio layer. The problem seems to be related to FastAPI: if I create a FastAPI app with any Starlette custom middlware, and then try to set a key in Redis using a separate connection pool created with redis.asyncio.from_url, the set request fails immediately with an uncaught asyncio.exceptions.CancelledError and the following backtrace:

.tox/py/lib/python3.11/site-packages/gafaelfawr/storage/base.py:140: in store
    await self._redis.set(key, encrypted_data, ex=lifetime)
.tox/py/lib/python3.11/site-packages/redis/asyncio/client.py:509: in execute_command
    conn = self.connection or await pool.get_connection(command_name, **options)
.tox/py/lib/python3.11/site-packages/redis/asyncio/connection.py:1408: in get_connection
    if await connection.can_read_destructive():
.tox/py/lib/python3.11/site-packages/redis/asyncio/connection.py:817: in can_read_destructive
    return await self._parser.can_read_destructive()
.tox/py/lib/python3.11/site-packages/redis/asyncio/connection.py:250: in can_read_destructive
    return await self._stream.read(1)
/usr/lib/python3.11/asyncio/streams.py:689: in read
    await self._wait_for_data('read')
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <StreamReader transport=<_SelectorSocketTransport fd=15 read=polling write=<idle, bufsize=0>>>
func_name = 'read'

    async def _wait_for_data(self, func_name):
        """Wait until feed_data() or feed_eof() is called.
    
        If stream was paused, automatically resume it.
        """
        # StreamReader uses a future to link the protocol feed_data() method
        # to a read coroutine. Running two read coroutines at the same time
        # would have an unexpected behaviour. It would not possible to know
        # which coroutine would get the next data.
        if self._waiter is not None:
            raise RuntimeError(
                f'{func_name}() called while another coroutine is '
                f'already waiting for incoming data')
    
        assert not self._eof, '_wait_for_data after EOF'
    
        # Waiting for data while paused will make deadlock, so prevent it.
        # This is essential for readexactly(n) for case when n > self._limit.
        if self._paused:
            self._paused = False
            self._transport.resume_reading()
    
        self._waiter = self._loop.create_future()
        try:
>           await self._waiter
E           asyncio.exceptions.CancelledError

/usr/lib/python3.11/asyncio/streams.py:522: CancelledError

GitHub Actions failure log: https://github.com/lsst-sqre/gafaelfawr/actions/runs/4473916874/jobs/7861846517

Reverting to redis 4.5.1 makes the problem disappear again, so it appears to be triggered by some change in 4.5.2.

Here's the shortest test case that I've come up with. It depends on FastAPI and Starlette but none of my code base. My guess is that something is happening during addition of the middleware that's breaking something about the asyncio work inside redis, but I'm not sure what that could be. Oddly, adding one of the standard Starlette middleware classes does not trigger this problem, only a custom middleware class derived from BaseHTTPMiddleware.

from collections.abc import Callable, Awaitable

import redis.asyncio
from fastapi import FastAPI, APIRouter, Request, Response
from httpx import AsyncClient
from starlette.middleware.base import BaseHTTPMiddleware


class DummyMiddleware(BaseHTTPMiddleware):
    async def dispatch(
        self,
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        return await call_next(request)


@pytest.mark.asyncio
async def test_redis() -> None:
    app = FastAPI()
    app.add_middleware(DummyMiddleware)

    @app.get("/")
    async def index() -> dict[str, str]:
        return {}

    async with AsyncClient(app=app, base_url="https://example.com/") as client:
        await client.get("/")
        pool = redis.asyncio.from_url("redis://localhost:6379/0")
        await pool.set("key", "value")

(If I had to guess, it would be the switch to asyncio.timeout, but I can't find evidence to support that. I know asyncio.timeout does raise CancelledError and then converts it to TimeoutError, but I couldn't find any place where that machinery shouldn't work correctly. This error happens immediately, so it doesn't seem to be a timeout and therefore I may be barking up the wrong tree.)

[rra]

Ah, Starlette's BaseHTTPMiddleware calls anyio.create_task_group(). I suspect that is related somehow, although I'm still not seeing exactly how. https://github.com/encode/starlette/blob/master/starlette/middleware/base.py#L39

[sileht]

I guess the CancellationError is raised here, where it should be an asyncio.Timeout.

[sileht]

asyncio check that asyncio.current_task().uncancel() == 0 before converting CancelledError into TimeoutError, so somehow this value is unexpected, not 0.

[sileht]

The thing is that this bug was always there, but with the legacy async_timeout lib, the bug was hidden because of this line:
https://github.com/aio-libs/async-timeout/blob/master/async_timeout/__init__.py#L210

All CancelledError was converted into TimeoutError even if the cancellation source was not the timeout.

[sileht]

Looks related to agronholm/anyio#374

[curtiscook]

FWIW: I'm experiencing the same issue. I've noticed that the connection does work, but seems to break if you await a function that's not Redis. We were able to reproduce this on the command line w/o starlette (even though we are using FastAPI)
given

def get_redis_pool() -> Redis:
    global _redis_pool

    if _redis_pool is None or _redis_pool.connection is None:
        _redis_pool = aioredis.from_url(REDIS_URL, max_connections=10)
    assert _redis_pool is not None
    return _redis_pool


async def get_key(key: str) -> Any:
    redis = get_redis_pool()
    return await redis.get(key)


async def set_key(key: str, value: Any, expire_seconds: Optional[int] = None) -> bool:
    redis = get_redis_pool()
    if expire_seconds is not None:
        exp = timedelta(seconds=expire_seconds)
        await redis.set(key, value, ex=exp)
    else:
        await redis.set(key, value)
    return True

This function will work fine:

async def foo():
	await get_key("key2")
	await set_key("key2", "value")

This will break:

async def foo():
	await get_key("key2")
	await bar()
	await set_key("key2", "value")

Might be related that bar uses HTTPX

[sileht]

Both starlette and httpx rely on anyio

[rra]

Yeah, this is looking to me like an anyio bug caused by use of task groups, presumably related to this admonition in the Python library documentation:

The asyncio components that enable structured concurrency, like asyncio.TaskGroup and asyncio.timeout(), are implemented using cancellation internally and might misbehave if a coroutine swallows asyncio.CancelledError.

I don't completely understand the mechanism yet, but enough keywords are lining up that this looks like it. @curtiscook's example is interesting, though, since I can't find a call to create_task_group in httpx or httpcore. Maybe it's somewhere else in httpx's dependencies, or maybe that bar function uses anyio?

[rra]

BTW, for those who are specifically having this issue with Starlette middleware, it looks like Starlette has rewritten their built-in middleware to avoid the use of BaseHTTPMiddleware at the cost of having to interact at a lower level with the request and response. That avoids creating a task group, and I wonder if that would make the problem go away. I'm considering rewriting our middleware similarly just as an experiment.

[decaz]

Looks like version 4.5.3 release didn't resolve the issue.

[bellini666]

FWIW: I'm experiencing the same issue. I've noticed that the connection does work, but seems to break if you await a function that's not Redis. We were able to reproduce this on the command line w/o starlette (even though we are using FastAPI) given

def get_redis_pool() -> Redis:
    global _redis_pool

    if _redis_pool is None or _redis_pool.connection is None:
        _redis_pool = aioredis.from_url(REDIS_URL, max_connections=10)
    assert _redis_pool is not None
    return _redis_pool


async def get_key(key: str) -> Any:
    redis = get_redis_pool()
    return await redis.get(key)


async def set_key(key: str, value: Any, expire_seconds: Optional[int] = None) -> bool:
    redis = get_redis_pool()
    if expire_seconds is not None:
        exp = timedelta(seconds=expire_seconds)
        await redis.set(key, value, ex=exp)
    else:
        await redis.set(key, value)
    return True

This function will work fine:

async def foo():
	await get_key("key2")
	await set_key("key2", "value")

This will break:

async def foo():
	await get_key("key2")
	await bar()
	await set_key("key2", "value")

Might be related that bar uses HTTPX

The same here! I have an async function that will sometimes call an external resource using httpx to validate the input. Everything works fine when httpx is not called, but when it is called then I get CancelledError.

This started happening after upgrading to 4.5.2/4.5.3 (not sure which one exactly since we upgraded from 4.5.1 directly)

[ice1x]

Seems like blocker for Redis isn't it?

[sileht]

This bug was always there:

• in < 4.5.2, because of aiohttp swallows asyncio.CancelledError during connection timeout aio-libs/async-timeout#229, the redis connection was just closed, then reopen.
• in >= 4.5.2, async-timeout has been replaced by asyncio.timeout, and now an unexpected CancelledError is raised.

[bellini666]

This bug was always there:

• in < 4.5.2, because of aiohttp swallows asyncio.CancelledError during connection timeout aio-libs/async-timeout#229, the redis connection was just closed, then reopen.
• in >= 4.5.2, async-timeout has been replaced by asyncio.timeout, and now an unexpected CancelledError is raised.

But why does CancelledError gets raised when doing some other async stuff in the same task? E.g. the httpx issue mentioned by me here and by @curtiscook here

I don't understand why CancelledError is being raised when calling a redis api just because I called httx before it (if I don't call httpx, CancelledError is not raised, but when calling, CancelledError` will be raised every time)

I worked around this error by wrapping my call to redis in an asyncio.shield, and although it should be harmless, it really looks like something wrong is happening here.

I'll try to create a minimum reproducible demo