context variables can leak out of asyncio.Task

[pmeier]

Bug report

Bug description:

This is a minimal reproducer for Kludex/uvicorn#2167. TL;DR when using asyncio as event loop, users of uvicorn were seeing polluted contexts after sending in large payloads. For those uvicorn paused reading on the main thread and resumed it in a task.

import asyncio
import contextvars
import sys

cvar1 = contextvars.ContextVar("cvar1")
cvar2 = contextvars.ContextVar("cvar2")
cvar3 = contextvars.ContextVar("cvar3")


def print_diagnostics(label):
    task = t.get_name() if (t := asyncio.current_task()) else None
    context = {c.name: v for c, v in contextvars.copy_context().items()}
    print(f"{label}: {task=}, {context=}\n{'-' * 80}")


class DemoProtocol(asyncio.Protocol):
    def __init__(self, on_conn_lost):
        self.transport = None
        self.on_conn_lost = on_conn_lost
        self.tasks = set()

    def connection_made(self, transport):
        print_diagnostics("connection_made")

        self.transport = transport

    def data_received(self, data):
        print_diagnostics("data_received")

        task = asyncio.create_task(self.asgi())
        self.tasks.add(task)
        task.add_done_callback(self.tasks.discard)

        self.transport.pause_reading()

    def connection_lost(self, exc):
        print_diagnostics("connection_lost")
        if not self.on_conn_lost.done():
            self.on_conn_lost.set_result(True)

    async def asgi(self):
        print_diagnostics("asgi start")

        cvar1.set(True)

        # make sure that we only resume after the pause
        # otherwise the resume does nothing
        while not self.transport._paused:
            await asyncio.sleep(0.1)

        cvar2.set(True)

        self.transport.resume_reading()

        cvar3.set(True)

        print_diagnostics("asgi end")


async def main():
    print(f"Python: {sys.version}\n{'-' * 80}")

    loop = asyncio.get_running_loop()
    on_conn_lost = loop.create_future()

    host, port = "127.0.0.1", 8888

    async with await loop.create_server(lambda: DemoProtocol(on_conn_lost), host, port):
        reader, writer = await asyncio.open_connection(host, port)
        writer.write(b"anything")
        await writer.drain()
        writer.close()
        await writer.wait_closed()
        await on_conn_lost


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

Python: 3.14.0 (main, Oct 14 2025, 21:27:55) [Clang 20.1.4 ]
--------------------------------------------------------------------------------
connection_made: task=None, context={}
--------------------------------------------------------------------------------
data_received: task=None, context={}
--------------------------------------------------------------------------------
asgi start: task='Task-4', context={}
--------------------------------------------------------------------------------
asgi end: task='Task-4', context={'cvar2': True, 'cvar3': True, 'cvar1': True}
--------------------------------------------------------------------------------
connection_lost: task=None, context={'cvar2': True, 'cvar1': True}
--------------------------------------------------------------------------------

The asgi task sets three context variables:

1. cvar1 is set at the beginning of the function, which may or may not be before the reading is paused.
2. cvar2 is set after the reading is paused.
3. cvar3 is set after the reading is resumed.

The context variables that have been set in the task before the reading is resumed (cvar1 and cvar2) leak out of the task into the main thread.

CPython versions tested on:

3.14

Operating systems tested on:

Linux