diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py index 68353b1d..dbbe01ab 100644 --- a/src/apify_client/http_clients/_impit.py +++ b/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 diff --git a/tests/unit/test_http_clients.py b/tests/unit/test_http_clients.py index 34236887..10dddede 100644 --- a/tests/unit/test_http_clients.py +++ b/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'),