Async file upload retry blocks the event loop #2938

import asyncio
import io
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from google.genai import _api_client, types


class Handler(BaseHTTPRequestHandler):
  attempts = 0

  def do_POST(self):
    type(self).attempts += 1
    self.rfile.read(int(self.headers['Content-Length']))
    self.send_response(200)
    if type(self).attempts == 2:
      self.send_header('X-Goog-Upload-Status', 'final')
    self.end_headers()

  def log_message(self, format, *args):
    pass


async def main():
  server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
  threading.Thread(target=server.serve_forever, daemon=True).start()
  base_url = f'http://127.0.0.1:{server.server_port}'
  client = _api_client.BaseApiClient(
      vertexai=False,
      api_key='test',
      http_options=types.HttpOptions(
          base_url=base_url,
          async_client_args={'trust_env': False},
      ),
  )
  client._use_aiohttp = lambda: False
  _api_client.INITIAL_RETRY_DELAY = 0.25
  started = time.monotonic()

  async def ticker():
    await asyncio.sleep(0.05)
    return time.monotonic() - started

  try:
    _, tick = await asyncio.gather(
        client._async_upload_fd(
            io.BytesIO(b'test'), f'{base_url}/upload', 4
        ),
        ticker(),
    )
    print(f'attempts={Handler.attempts}')
    print(f'ticker_delay={tick:.3f}s')
    print(f'event_loop_blocked={tick >= 0.20}')
  finally:
    await client.aclose()
    server.shutdown()
    server.server_close()


asyncio.run(main())