diff --git a/src/apify_client/http_clients/_base.py b/src/apify_client/http_clients/_base.py index 566389eb..6911c16a 100644 --- a/src/apify_client/http_clients/_base.py +++ b/src/apify_client/http_clients/_base.py @@ -7,6 +7,7 @@ import random import sys import time +from contextlib import suppress from datetime import UTC, datetime, timedelta from http import HTTPStatus from importlib import metadata @@ -365,6 +366,22 @@ def _handle_request_exception(self, exc: Exception, *, stop_retrying: Callable[[ logger.debug('Exception is not retryable', exc_info=exc) stop_retrying() + @staticmethod + def _is_transient_transport_error( + exc: Exception, + *, + transport_errors: type[Exception] | tuple[type[Exception], ...], + permanent_errors: tuple[type[Exception], ...], + ) -> bool: + """Apply the shared retry policy to an exception raised by an HTTP library. + + Every error from the library's own hierarchy counts as transient except the permanently-failing types the + adapter lists. Retrying is the default because transports also report genuinely transient failures through + their generic base class, e.g. Impit raises a bare `impit.HTTPError` for a body that ends mid-chunk. Sharing + one policy across adapters keeps a change of transport from changing which failures are retried. + """ + return isinstance(exc, transport_errors) and not isinstance(exc, permanent_errors) + @docs_group('HTTP clients') class HttpClient(HttpClientBase): @@ -559,8 +576,14 @@ 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 the error can be raised properly. - response.read() + try: + response.read() + except Exception as exc: + with suppress(Exception): + response.close() + self._handle_request_exception(exc, stop_retrying=stop_retrying) + raise + raise ApifyApiError(response, attempt, method=method) @@ -764,6 +787,12 @@ 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 the error can be raised properly. - await response.aread() + try: + await response.aread() + except Exception as exc: + with suppress(Exception): + await response.aclose() + self._handle_request_exception(exc, stop_retrying=stop_retrying) + raise + raise ApifyApiError(response, attempt, method=method) diff --git a/src/apify_client/http_clients/_impit.py b/src/apify_client/http_clients/_impit.py index 0895ac0b..9fbfbedc 100644 --- a/src/apify_client/http_clients/_impit.py +++ b/src/apify_client/http_clients/_impit.py @@ -22,6 +22,17 @@ from apify_client._statistics import ClientStatistics from apify_client.http_compressors._base import HttpCompressor +_PERMANENT_ERRORS = ( + # A bad URL scheme or a request Impit itself rejects cannot succeed on a retry. + impit.UnsupportedProtocol, + impit.LocalProtocolError, + # An over-long redirect chain is a routing loop, which repeating the request cannot break. + impit.TooManyRedirects, + # Status codes are retried by `_make_request` based on the status itself, never as a transport error. + impit.HTTPStatusError, +) +"""Impit errors that a retry cannot fix. Everything else in the `impit.HTTPError` tree is treated as transient.""" + @docs_group('HTTP clients') class ImpitHttpClient(HttpClient): @@ -115,9 +126,11 @@ def send_request( @override def is_retryable_transport_error(self, exc: Exception) -> bool: - # All errors from Impit's own hierarchy count as transient - they represent transport-level failures - # (network issues, timeouts, protocol errors, body decoding errors) that are typically transient. - return isinstance(exc, impit.HTTPError) + return self._is_transient_transport_error( + exc, + transport_errors=impit.HTTPError, + permanent_errors=_PERMANENT_ERRORS, + ) @docs_group('HTTP clients') @@ -208,6 +221,8 @@ async def send_request( @override def is_retryable_transport_error(self, exc: Exception) -> bool: - # All errors from Impit's own hierarchy count as transient - they represent transport-level failures - # (network issues, timeouts, protocol errors, body decoding errors) that are typically transient. - return isinstance(exc, impit.HTTPError) + return self._is_transient_transport_error( + exc, + transport_errors=impit.HTTPError, + permanent_errors=_PERMANENT_ERRORS, + ) diff --git a/tests/unit/test_http_client_regressions.py b/tests/unit/test_http_client_regressions.py new file mode 100644 index 00000000..b1bedd93 --- /dev/null +++ b/tests/unit/test_http_client_regressions.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from datetime import timedelta +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock, Mock + +import impit +import pytest + +from apify_client.http_clients import ImpitHttpClient, ImpitHttpClientAsync + +if TYPE_CHECKING: + from _pytest.monkeypatch import MonkeyPatch + + +def successful_response() -> Mock: + return Mock(status_code=200) + + +def test_generic_http_error_is_retried(monkeypatch: MonkeyPatch) -> None: + """A bare Impit HTTPError from a truncated body remains retryable.""" + client = ImpitHttpClient(min_delay_between_retries=timedelta(0)) + send_request = Mock( + side_effect=[impit.HTTPError('unexpected EOF'), impit.HTTPError('unexpected EOF'), successful_response()] + ) + monkeypatch.setattr(client, 'send_request', send_request) + + response = client.call(method='GET', url='https://example.com') + + assert response.status_code == 200 + assert send_request.call_count == 3 + + +async def test_generic_http_error_is_retried_async(monkeypatch: MonkeyPatch) -> None: + """The async Impit adapter treats a bare HTTPError from a truncated body the same way.""" + client = ImpitHttpClientAsync(min_delay_between_retries=timedelta(0)) + send_request = AsyncMock( + side_effect=[impit.HTTPError('unexpected EOF'), impit.HTTPError('unexpected EOF'), successful_response()] + ) + monkeypatch.setattr(client, 'send_request', send_request) + + response = await client.call(method='GET', url='https://example.com') + + assert response.status_code == 200 + assert send_request.call_count == 3 + + +def test_permanent_transport_error_is_not_retried(monkeypatch: MonkeyPatch) -> None: + """A transport error a retry cannot fix must fail on the first attempt instead of burning the whole backoff.""" + client = ImpitHttpClient(min_delay_between_retries=timedelta(0)) + send_request = Mock(side_effect=impit.UnsupportedProtocol('unsupported scheme')) + monkeypatch.setattr(client, 'send_request', send_request) + + with pytest.raises(impit.UnsupportedProtocol): + client.call(method='GET', url='https://example.com') + + assert send_request.call_count == 1 + + +def test_error_response_read_failure_uses_transport_retry_policy(monkeypatch: MonkeyPatch) -> None: + """Failures while reading a streamed 5xx error body are classified and retried like send failures.""" + client = ImpitHttpClient(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) + ] + send_request = Mock(side_effect=responses) + monkeypatch.setattr(client, 'send_request', send_request) + + with pytest.raises(impit.ReadError): + client.call(method='GET', url='https://example.com', stream=True) + + assert send_request.call_count == 2 + for response in responses: + response.close.assert_called_once() + + +async def test_error_response_read_failure_uses_transport_retry_policy_async(monkeypatch: MonkeyPatch) -> None: + """The async base also classifies failures while buffering streamed error responses.""" + client = ImpitHttpClientAsync(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) + ] + send_request = AsyncMock(side_effect=responses) + monkeypatch.setattr(client, 'send_request', send_request) + + with pytest.raises(impit.ReadError): + await client.call(method='GET', url='https://example.com', stream=True) + + assert send_request.call_count == 2 + for response in responses: + response.aclose.assert_awaited_once() diff --git a/tests/unit/test_http_clients.py b/tests/unit/test_http_clients.py index ab4fd5a2..44aac1b7 100644 --- a/tests/unit/test_http_clients.py +++ b/tests/unit/test_http_clients.py @@ -337,22 +337,39 @@ def test_parse_params_mixed() -> None: } -TRANSPORT_ERRORS = ( +RETRYABLE_TRANSPORT_ERRORS = ( + # Impit raises a bare `HTTPError` for a body that ends mid-chunk, so even the generic base class is transient. impit.HTTPError, impit.TimeoutException, impit.NetworkError, impit.RemoteProtocolError, impit.DecodingError, + # A proxy rejecting the tunnel is often transient, e.g. one that is overloaded or rate-limiting. + impit.ProxyError, ) -"""Transport errors from Impit's own hierarchy, all classified as retryable.""" +"""Transport errors that must stay retryable.""" + +NON_RETRYABLE_TRANSPORT_ERRORS = ( + impit.UnsupportedProtocol, + impit.LocalProtocolError, + impit.TooManyRedirects, +) +"""Transport errors that a retry cannot fix, so they must fail on the first attempt. + +`impit.HTTPStatusError` belongs here too, but no built-in adapter ever raises it, because `_make_request` decides on +status codes from the response itself. +""" def test_builtin_http_client_retry_policy() -> None: - """Every transport-level Impit failure is classified as retryable, everything else is not.""" + """Transient transport failures are retried, and the ones a retry cannot fix stop the loop immediately.""" with ImpitHttpClient() as client: - for error_class in TRANSPORT_ERRORS: + for error_class in RETRYABLE_TRANSPORT_ERRORS: assert client.is_retryable_transport_error(error_class('test')), error_class.__name__ + for error_class in NON_RETRYABLE_TRANSPORT_ERRORS: + assert not client.is_retryable_transport_error(error_class('test')), error_class.__name__ + # `InvalidResponseBodyError` is retried by `_handle_request_exception`, not as a transport failure. assert not client.is_retryable_transport_error(InvalidResponseBodyError(Mock())) assert not client.is_retryable_transport_error(ValueError('test'))