From 95f69dc85e1a634c63486f592f1892b07eef196c Mon Sep 17 00:00:00 2001
From: Sanjay Santhanam <Sanjays2402@users.noreply.github.com>
Date: Wed, 9 Sep 2026 22:16:07 -0700
Subject: [PATCH] fix: close previous response before retrying in
 AsyncAuthorizedSession

In `AsyncAuthorizedSession.request`, the retry loop reassigned `response`
across retry attempts without closing the previous response. If the
payload had not reached EOF before the next retry fired, aiohttp kept
the socket checked out of the connector pool. Close (awaiting if
awaitable) any previous response before issuing the next attempt,
matching the cleanup pattern already used in the 401 recovery path.

Fixes googleapis/google-cloud-python#18315
---
 .../google/auth/aio/transport/sessions.py     | 12 +++++++
 .../tests/transport/aio/test_sessions.py      | 35 +++++++++++++++++++
 2 files changed, 47 insertions(+)

diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index f2ced1280e50..7acf8915ba1e 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -340,7 +340,19 @@ async def request(
                 actual_timeout = float(timeout)
             # Workaround issue in python 3.9 related to code coverage by adding `# pragma: no branch`
             # See https://github.com/googleapis/gapic-generator-python/pull/1174#issuecomment-1025132372
+            response = None
             async for _ in retries:  # pragma: no branch
+                if response is not None and hasattr(response, "close"):
+                    # Release the previous response before retrying so an
+                    # unread body does not keep its connection checked out
+                    # of the connector pool.
+                    try:
+                        res = response.close()
+                        if inspect.isawaitable(res):
+                            await res
+                    except Exception:
+                        pass
+
                 response = await with_timeout(
                     self._auth_request(
                         url, method, data, request_headers, actual_timeout, **kwargs
diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index ecdb60e4af5a..f312e3a117cb 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -287,6 +287,41 @@ async def test_request_max_retries(self, retry_status):
             await authed_session.request("GET", self.TEST_URL)
             assert auth_request.call_count == DEFAULT_MAX_RETRY_ATTEMPTS
 
+    @pytest.mark.asyncio
+    async def test_request_closes_previous_response_before_retry(self):
+        retry_response = MockResponse(status_code=503)
+        success_response = MockResponse(status_code=200)
+
+        class _SequenceMockRequest(MockRequest):
+            def __init__(self, responses):
+                super().__init__(response=None)
+                self._responses = responses
+
+            async def __call__(self, *args, **kwargs):
+                self.call_count += 1
+                return self._responses[self.call_count - 1]
+
+        auth_request = _SequenceMockRequest([retry_response, success_response])
+        with patch("asyncio.sleep", return_value=None):
+            authed_session = sessions.AsyncAuthorizedSession(
+                self.credentials, auth_request
+            )
+            response = await authed_session.request(
+                "GET",
+                self.TEST_URL,
+                max_allowed_time=float("inf"),
+                total_attempts=2,
+            )
+            assert response is success_response
+            assert auth_request.call_count == 2
+            # The initial retryable response must be closed before the retry
+            # so its connection is released back to the pool.
+            assert retry_response._close
+            # The final response is left open for the caller to close.
+            assert not success_response._close
+
+        await authed_session.close()
+
     @pytest.mark.asyncio
     async def test_http_get_method_success(self):
         expected_payload = b"content is retrieved."