fix: Classify and retry failures while reading a streamed error body

src/apify_client/http_clients/_impit.py

@@ -4,6 +4,7 @@
 import logging
 import random
 import time
+from contextlib import suppress
 from datetime import timedelta
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Any, TypeVar
@@ -246,8 +247,19 @@ def _make_request(
  logger.debug('Status code is not retryable', extra={'status_code': response.status_code})
  stop_retrying()
 
- # Read the response in case it is a stream, so we can raise the error properly.
- response.read()
+ # Read the response in case it is a stream, so we can raise the error properly. A failed read goes through
+ # the same classification as a failed send.
+ try:
+ response.read()
+ except Exception as exc:
+ logger.debug('Reading the error response failed', exc_info=exc)
+ with suppress(Exception):
+ response.close()
+ if not _is_retryable_error(exc):
+ logger.debug('Exception is not retryable', exc_info=exc)
+ stop_retrying()
+ raise
+
  raise ApifyApiError(response, attempt, method=method)
 
  @staticmethod
@@ -508,8 +520,19 @@ async def _make_request(
  logger.debug('Status code is not retryable', extra={'status_code': response.status_code})
  stop_retrying()
 
- # Read the response in case it is a stream, so we can raise the error properly.
- await response.aread()
+ # Read the response in case it is a stream, so we can raise the error properly. A failed read goes through
+ # the same classification as a failed send.
+ try:
+ await response.aread()
+ except Exception as exc:
+ logger.debug('Reading the error response failed', exc_info=exc)
+ with suppress(Exception):
+ await response.aclose()
+ if not _is_retryable_error(exc):
+ logger.debug('Exception is not retryable', exc_info=exc)
+ stop_retrying()
+ raise
+
  raise ApifyApiError(response, attempt, method=method)
 
  @staticmethod

tests/unit/test_http_clients.py

@@ -323,6 +323,83 @@ def test_is_retryable_error() -> None:
  assert not _is_retryable_error(Exception('test'))
 
 
+def test_error_response_read_failure_is_retried_and_closed() -> None:
+ """A failure while buffering a streamed error body is retried like a failed send, and the response is closed."""
+ client = ImpitHttpClient(token='test_token', max_retries=1, min_delay_between_retries=timedelta(0))
+ responses = [
+ Mock(status_code=500, read=Mock(side_effect=impit.ReadError('truncated')), close=Mock()) for _ in range(2)
+ ]
+ request = Mock(side_effect=responses)
+ client._impit_client = Mock(request=request)
+
+ with pytest.raises(impit.ReadError):
+ client.call(method='GET', url='https://api.test.com/endpoint', stream=True)
+
+ assert request.call_count == 2
+ for response in responses:
+ response.close.assert_called_once()
+
+
+async def test_error_response_read_failure_is_retried_and_closed_async() -> None:
+ """The async client also retries a failed buffering of a streamed error body and closes the response."""
+ client = ImpitHttpClientAsync(token='test_token', max_retries=1, min_delay_between_retries=timedelta(0))
+ responses = [
+ Mock(status_code=500, aread=AsyncMock(side_effect=impit.ReadError('truncated')), aclose=AsyncMock())
+ for _ in range(2)
+ ]
+ request = AsyncMock(side_effect=responses)
+ client._impit_async_client = Mock(request=request)
+
+ with pytest.raises(impit.ReadError):
+ await client.call(method='GET', url='https://api.test.com/endpoint', stream=True)
+
+ assert request.await_count == 2
+ for response in responses:
+ response.aclose.assert_awaited_once()
+
+
+def test_non_retryable_error_response_read_failure_stops_retrying() -> None:
+ """A read failure that is not a transport error stops the retry loop instead of being retried."""
+ client = ImpitHttpClient(token='test_token', min_delay_between_retries=timedelta(0))
+ response = Mock(status_code=500, read=Mock(side_effect=ValueError('broken response')), close=Mock())
+ request = Mock(return_value=response)
+ client._impit_client = Mock(request=request)
+
+ with pytest.raises(ValueError, match='broken response'):
+ client.call(method='GET', url='https://api.test.com/endpoint', stream=True)
+
+ request.assert_called_once()
+ response.close.assert_called_once()
+
+
+async def test_non_retryable_error_response_read_failure_stops_retrying_async() -> None:
+ """The async client also stops retrying when a read failure is not a transport error."""
+ client = ImpitHttpClientAsync(token='test_token', min_delay_between_retries=timedelta(0))
+ response = Mock(status_code=500, aread=AsyncMock(side_effect=ValueError('broken response')), aclose=AsyncMock())
+ request = AsyncMock(return_value=response)
+ client._impit_async_client = Mock(request=request)
+
+ with pytest.raises(ValueError, match='broken response'):
+ await client.call(method='GET', url='https://api.test.com/endpoint', stream=True)
+
+ request.assert_awaited_once()
+ response.aclose.assert_awaited_once()
+
+
+def test_error_response_read_failure_on_non_retryable_status_is_not_retried() -> None:
+ """A transient read failure on a status that is not retryable surfaces immediately instead of being retried."""
+ client = ImpitHttpClient(token='test_token', min_delay_between_retries=timedelta(0))
+ response = Mock(status_code=404, read=Mock(side_effect=impit.ReadError('truncated')), close=Mock())
+ request = Mock(return_value=response)
+ client._impit_client = Mock(request=request)
+
+ with pytest.raises(impit.ReadError):
+ client.call(method='GET', url='https://api.test.com/endpoint', stream=True)
+
+ request.assert_called_once()
+ response.close.assert_called_once()
+
+
 @pytest.fixture(
  params=[
  pytest.param((GzipHttpCompressor(), 'gzip', gzip.decompress), id='gzip'),