google-api-core: AsyncBidiRpc.open() after the call ends leaves the previous request generator pending forever

Aperta
  • Qualcosa non va?
#19,030 1 commento 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@shaashankrg ci sta già lavorando.

Dal 10/11/2026.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
35/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
grpc, python
Ambito
backend-api-design

Direzione di ricerca

Start in google/api_core/bidi_async.py, where AsyncBidiRpc.open() reuses self._request_queue and _AsyncRequestQueueGenerator is created for each call, and close() puts a single None sentinel. The grpc side is grpc/aio/_call.py, where _StreamRequestMixin.cancel() only cancels the request poller when super().cancel() returns True. Run the repro in the issue on Python 3.12 and check that the pending _consume_request_iterator tasks drop to 0 after close(). Done means each previous request generator ends on reopen or close, with no pending-task warnings.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

triage me type: bug
Determine this is the right repository
  • I determined this is the correct repository in which to report this bug.
Summary of the issue

Context
We consume Pub/Sub StreamingPull through google.api_core.bidi_async.AsyncBidiRpc. When the call ends (server EOF or a retryable error), we call open() again on the same AsyncBidiRpc. Nothing in the class documents that it can't be reopened.

Expected Behavior:
Reopening (and later close()) ends the request generator of each previous call, so grpc's request poller tasks for finished calls exit.

Actual Behavior:
Each open() creates a new _AsyncRequestQueueGenerator on the same _request_queue and never ends the previous one. grpc's _consume_request_iterator task for the finished call stays blocked in queue.get(). call.cancel() returns False on a finished call, so grpc never cancels that poller. close() puts a single None on the queue, which ends at most one generator.

When the AsyncBidiRpc is garbage-collected, asyncio reports Task was destroyed but it is pending! for each leaked poller. On Python 3.12 the async-generator finalizer then fails with RuntimeError: aclose(): asynchronous generator is already running, reported as "Task exception was never retrieved". Applications whose loop exception handler treats task exceptions as fatal shut down. In our case, this restarted streaming consumers across our fleet.

API client name and version

google-api-core v2.44.0 (also reproduced on v2.43.0), grpcio 1.84.1

Reproduction steps: code

file: main.py

import asyncio
import gc

import grpc
from google.api_core import bidi_async

METHOD = "/repro.Echo/Stream"
EOFS = 3


async def stream(request_iterator, context):
    # End the call after the first request, as a server ending StreamingPull would.
    async for request in request_iterator:
        yield request
        return


class Handler(grpc.GenericRpcHandler):
    def service(self, details):
        if details.method != METHOD:
            return None
        return grpc.stream_stream_rpc_method_handler(stream)


def report(loop, context):
    exc = context.get("exception")
    print(f"loop exception handler: {context['message']!r} {exc!r}")


async def reproduce():
    asyncio.get_running_loop().set_exception_handler(report)
    server = grpc.aio.server()
    server.add_generic_rpc_handlers([Handler()])
    port = server.add_insecure_port("127.0.0.1:0")
    await server.start()

    channel = grpc.aio.insecure_channel(f"127.0.0.1:{port}")
    multicallable = channel.stream_stream(METHOD)

    async def start_rpc(requests, metadata=None):
        # Awaitable, like a generated async client's streaming method.
        return multicallable(requests, metadata=metadata)

    rpc = bidi_async.AsyncBidiRpc(start_rpc, initial_request=b"hello")
    for _ in range(EOFS):
        # Reopen the same AsyncBidiRpc after the server ends the call.
        await rpc.open()
        while await rpc.recv() is not grpc.aio.EOF:
            pass
        await asyncio.sleep(0.05)

    await rpc.close()
    await asyncio.sleep(0.05)
    pending = [
        t for t in asyncio.all_tasks() if "_consume_request_iterator" in repr(t.get_coro())
    ]
    print(f"request pollers still pending after close(): {len(pending)}")

    del rpc, pending
    gc.collect()
    await asyncio.sleep(0.05)
    gc.collect()
    await asyncio.sleep(0.05)

    await channel.close()
    await server.stop(None)


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

Run: pip install "google-api-core[grpc]==2.44.0" grpcio==1.84.1 && python main.py

Reproduction steps: supporting files

None. The repro starts its own local grpc.aio server, with raw-bytes messages and no protos.

Reproduction steps: actual results

Python 3.12.14:

request pollers still pending after close(): 2
loop exception handler: 'Task was destroyed but it is pending!' None
loop exception handler: 'Task was destroyed but it is pending!' None
loop exception handler: 'Task exception was never retrieved' RuntimeError('aclose(): asynchronous generator is already running')
loop exception handler: 'Task exception was never retrieved' RuntimeError('aclose(): asynchronous generator is already running')

Python 3.13.14 (pending tasks only, no aclose error):

request pollers still pending after close(): 2
loop exception handler: 'Task was destroyed but it is pending!' None
loop exception handler: 'Task was destroyed but it is pending!' None
Reproduction steps: expected results
request pollers still pending after close(): 0

No loop exception handler output.

OS & version + platform

Production: Debian (python:3.12-slim image) on GKE. The repro also fails on macOS 26 (arm64).

Python environment

Python 3.12.14 (also 3.13.14)

Python dependencies

Package Version

certifi 2026.10.23
cffi 2.1.1
charset-normalizer 3.5.2
cryptography 50.0.2
google-api-core 2.44.0
google-auth 2.62.1
googleapis-common-protos 1.76.0
grpcio 1.84.1
grpcio-status 1.84.1
idna 3.20
opentelemetry-api 1.45.2
proto-plus 1.29.0
protobuf 7.36.2
pyasn1 0.6.5
pyasn1_modules 0.4.2
pycparser 3.12
requests 2.34.2
typing_extensions 4.16.0
urllib3 2.8.0

Additional context
  • Relevant code: google/api_core/bidi_async.py. AsyncBidiRpc.open() reuses self._request_queue for every new _AsyncRequestQueueGenerator, and close() puts one None sentinel. On the grpc side, grpc/aio/_call.py _StreamRequestMixin.cancel() only cancels _async_request_poller when super().cancel() returns True, which is false once the call has finished.
  • Possible fixes: end the previous request generator in open() (for example, a sentinel on the old queue, or a fresh queue per open), or document that an AsyncBidiRpc must not be reopened after its call ends.
  • Workaround: create a new AsyncBidiRpc for each reopen. We do this now.
Lingua principale
Python
Stelle
5.4k
Fork
1.8k
Merge medio
2g 7h
PR unite (30g)
173

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di googleapis/google-cloud-python

Tutte le issue di googleapis/google-cloud-python

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.