Add PREFECT_CLIENT_MAX_RETRIES for configuration of maximum HTTP re…

…quest retries (PrefectHQ#9735)

src/prefect/client/base.py

@@ -18,6 +18,7 @@
 from prefect.settings import (
     PREFECT_CLIENT_RETRY_EXTRA_CODES,
     PREFECT_CLIENT_RETRY_JITTER_FACTOR,
+    PREFECT_CLIENT_MAX_RETRIES,
 )
 from prefect.utilities.math import bounded_poisson_interval, clamped_poisson_interval
 
@@ -164,8 +165,6 @@ class PrefectHttpxClient(httpx.AsyncClient):
     [Configuring Cloudflare Rate Limiting](https://support.cloudflare.com/hc/en-us/articles/115001635128-Configuring-Rate-Limiting-from-UI)
     """
 
-    RETRY_MAX = 5
-
     async def _send_with_retry(
         self,
         request: Callable,
@@ -175,25 +174,25 @@ async def _send_with_retry(
         """
         Send a request and retry it if it fails.
 
-        Sends the provided request and retries it up to self.RETRY_MAX times if
-        the request either raises an exception listed in `retry_exceptions` or receives
-        a response with a status code listed in `retry_codes`.
+        Sends the provided request and retries it up to PREFECT_CLIENT_MAX_RETRIES times
+        if the request either raises an exception listed in `retry_exceptions` or
+        receives a response with a status code listed in `retry_codes`.
 
         Retries will be delayed based on either the retry header (preferred) or
         exponential backoff if a retry header is not provided.
         """
         try_count = 0
         response = None
 
-        while try_count <= self.RETRY_MAX:
+        while try_count <= PREFECT_CLIENT_MAX_RETRIES.value():
             try_count += 1
             retry_seconds = None
             exc_info = None
 
             try:
                 response = await request()
             except retry_exceptions:  # type: ignore
-                if try_count > self.RETRY_MAX:
+                if try_count > PREFECT_CLIENT_MAX_RETRIES.value():
                     raise
                 # Otherwise, we will ignore this error but capture the info for logging
                 exc_info = sys.exc_info()
@@ -233,7 +232,8 @@ async def _send_with_retry(
                     )
                 )
                 + f"Another attempt will be made in {retry_seconds}s. "
-                f"This is attempt {try_count}/{self.RETRY_MAX + 1}.",
+                "This is attempt"
+                f" {try_count}/{PREFECT_CLIENT_MAX_RETRIES.value() + 1}.",
                 exc_info=exc_info,
             )
             await anyio.sleep(retry_seconds)

src/prefect/settings.py

@@ -536,6 +536,18 @@ def default_cloud_ui_url(settings, value):
 made via HTTP/1.1.
 """
 
+
+PREFECT_CLIENT_MAX_RETRIES = Setting(int, default=5)
+"""
+The maximum number of retries to perform on failed HTTP requests.
+
+Defaults to 5.
+Set to 0 to disable retries.
+
+See `PREFECT_CLIENT_RETRY_EXTRA_CODES` for details on which HTTP status codes are 
+retried.
+"""
+
 PREFECT_CLIENT_RETRY_JITTER_FACTOR = Setting(float, default=0.2)
 """
 A value greater than or equal to zero to control the amount of jitter added to retried
@@ -545,6 +557,7 @@ def default_cloud_ui_url(settings, value):
 can affect retry lengths.
 """
 
+
 PREFECT_CLIENT_RETRY_EXTRA_CODES = Setting(
     str, default="", value_callback=status_codes_as_integers_in_range
 )

tests/client/test_base_client.py

@@ -10,6 +10,7 @@
 from prefect.settings import (
     PREFECT_CLIENT_RETRY_EXTRA_CODES,
     PREFECT_CLIENT_RETRY_JITTER_FACTOR,
+    PREFECT_CLIENT_MAX_RETRIES,
     temporary_settings,
 )
 from prefect.testing.utilities import AsyncMock
@@ -221,6 +222,33 @@ async def test_prefect_httpx_client_retries_up_to_five_times(
         # 5 retries + 1 first attempt
         assert base_client_send.call_count == 6
 
+    @pytest.mark.usefixtures("mock_anyio_sleep")
+    @pytest.mark.parametrize(
+        "response_or_exc",
+        [RESPONSE_429_RETRY_AFTER_0, httpx.RemoteProtocolError("test")],
+    )
+    async def test_prefect_httpx_client_respects_max_retry_setting(
+        self,
+        monkeypatch,
+        response_or_exc,
+    ):
+        client = PrefectHttpxClient()
+        base_client_send = AsyncMock()
+        monkeypatch.setattr(AsyncClient, "send", base_client_send)
+
+        # Return more than 10 retryable responses
+        base_client_send.side_effect = [response_or_exc] * 20
+
+        with pytest.raises(Exception):
+            with temporary_settings({PREFECT_CLIENT_MAX_RETRIES: 10}):
+                await client.post(
+                    url="fake.url/fake/route",
+                    data={"evenmorefake": "data"},
+                )
+
+        # 10 retries + 1 first attempt
+        assert base_client_send.call_count == 11
+
     @pytest.mark.usefixtures("mock_anyio_sleep")
     @pytest.mark.parametrize(
         "final_response,expected_error_type",