aclosing: release the connection on break, error and cancellation

Build an async pool context manager and a consumer that releases the resource promptly on every exit path — completion, early break, exception and cancellation.

class Pool:
    def __init__(self, log: list[str]) -> None: ...   # also self.live = 0

    @asynccontextmanager
    async def acquire(self, name: str) -> AsyncIterator[str]: ...
        # live += 1, log "acquire:{name}", yield name,
        # finally: live -= 1, log "release:{name}"


async def stream_rows(pool: Pool, count: int) -> AsyncGenerator[int, None]: ...
    # async with pool.acquire("conn") as conn:
    #     for index in range(count):
    #         await asyncio.sleep(0)
    #         yield len(conn) * 100 + index


def solve(count: int, stop_after: int, mode: str) -> dict[str, object]: ...

solve is synchronous and runs the whole scenario with asyncio.run. The consumer iterates stream_rows inside aclosing, collecting rows, and when stop_after >= 0 and enough rows have arrived it behaves according to mode:

Returns {"seen": [...], "log": list(log), "outcome": str, "live": pool.live}. live must be 0 and the log must end with "release:conn" at the moment the dict is built — snapshot the log with list(log), do not hand back the live list.

Why the snapshot matters, and it is the whole lesson. Breaking out of an async for leaves the generator suspended at its yield, inside the async with. Its finally has not run. Finalisation for an async generator is not refcount-driven: it happens through the event loop’s shutdown_asyncgens hook — which asyncio.run calls after your coroutine returns. So a solution without aclosing produces a log that eventually contains "release:conn", just not yet, and pool.live is still 1 when the result is built. In a request handler that breaks early, that is a connection held until the process shuts down; under load the pool exhausts and the symptom is “timeouts acquiring a connection” with no slow queries anywhere.

aclosing(agen) awaits agen.aclose() on exit, which throws GeneratorExit into the suspended generator and runs its finally before the block returns.

Do not reach for ExitStack or closing. agen.aclose() is a coroutine and a synchronous __exit__ cannot await it — you get a coroutine object that is never run, plus a RuntimeWarning nobody reads. --strict catches the ExitStack version, because an async context manager does not satisfy AbstractContextManager.

Typing. stream_rows must be annotated AsyncGenerator[int, None], not AsyncIterator[int]: aclosing requires something with aclose(), and AsyncIterator does not have it. The @asynccontextmanager method, by contrast, is annotated AsyncIterator[str] — the decorator wants an async iterator, and it is what turns it into a context manager.