Keep stream deadlines from cancelling the caller between messages

[stefanvanburen]

With the async client, a server-streaming or bidi call with timeout_ms could cancel the caller's own task. If the deadline passed while the caller was doing its own work between messages (a slow async for body, or holding the iterator while doing something else), the caller got a bare CancelledError in unrelated code instead of ConnectError(DEADLINE_EXCEEDED), and the task was left with a pending cancellation that can confuse an enclosing TaskGroup or asyncio.timeout.

async for hat in client.make_similar_hats(Size(inches=10), timeout_ms=50):
    await asyncio.sleep(0.1)  # CancelledError raised here, not DEADLINE_EXCEEDED

The deadline is now enforced only while the client is waiting on the network, so it surfaces as DEADLINE_EXCEEDED from the iterator. The new test fails on main on 3.10 and 3.14. I looked at enabling ruff's preview ASYNC119 as a guard, but it also flags harmless code (_request_stream on the server holds only a validating reader across its yield), and ASYNC101, which targets this case exactly, isn't implemented in ruff, so the test is the guard.

[stefanvanburen]

(In draft because I still need to understand this a little more fully myself; feel free to review I suppose...)

[anuraaga]

I think the state machine is to eagerly close a stream that isn't being read with a separate async task - it feels too heavyweight to me and I think there is a better alternative on the server side.

I think this simpler form that makes sure timeout is only around the await in our code here and not around the response generator should work more simply enough for this PR

deadline = loop.time() + timeout_ms / 1000.0  # or None

async with contextlib.AsyncExitStack() as stack:
    async with asyncio_timeout_at(deadline):
        resp = await stack.enter_async_context(self._http_client.stream(...))
    ...
    it = aiter(resp.content)
    while True:
        async with asyncio_timeout_at(deadline):
            try:
                chunk = await anext(it)
            except StopAsyncIteration:
                break
        for message in reader.feed(bytes(chunk)):
            yield message
            await sleep(0)

[stefanvanburen]

Thanks, that's much simpler. I've dropped the close-at-deadline commit and gone with your version, plus one addition: a deadline check before each read. When a chunk is already buffered, anext(it) returns without suspending, so asyncio_timeout_at with a deadline that has already passed never fires and the stream runs to completion after the deadline. The regression test caught this with your snippet as-is (DID NOT RAISE ConnectError). With if deadline is not None and loop.time() >= deadline: raise TimeoutError at the top of the loop, it passes on 3.10 and 3.14. The error-response path never yields, so it keeps a single asyncio_timeout_at around its whole read.

On the server side: our server parses connect-timeout-ms / grpc-timeout into ctx.timeout_ms but doesn't end the call when it runs out. Is that what you had in mind? If so, I'll open an issue for it as a follow-up.

[anuraaga]

Yeah - to be honest I had thought about that a while ago but forgot about it, so the issue would have helped I guess 😅 The weird things that needed more thought

• How could conformance pass in the first place without a server timeout
• Relatively hard/annoying for WSGI, especially on Windows

But probably time to sort this out