fix: Fail fast on permanent transport errors and retry streamed error-read failures

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)

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,
+        )

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()

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'))