Drop the unreachable arms of the busy-page backoff

Nothing on the read path sends Retry-After: the API answers with a bare
FastAPI HTTPException and the ingress rate limiter adds no headers either, so
that branch was shaping a delay no server asks for. The ceiling was unreachable
too — four doublings from half a second top out at four, jitter included.

What is left is the backoff itself. Name the jitter fraction rather than
spelling the band inline, and cover the growth the constants describe. The
give-up test now exhausts the budget on a later page, so it also pins that an
already delivered chunk stands.

Signed-off-by: Honglin Cao <hocao@nvidia.com>

Author: V2arK

centml/sdk/api.py

@@ -33,12 +33,15 @@
 # caller's events within this generous margin of the boundary can be re-delivered.
 LOG_DEDUP_RETENTION_MS = 300_000
 # The log read path is rate limited upstream on a bucket shared by every caller, and the
-# API reports a saturated bucket the same way it reports a sick store: HTTP 503. The
-# budget spans a few seconds, long enough to outlast a bucket refill without looking hung.
+# API reports a saturated bucket the same way it reports a sick store: HTTP 503. Doubling
+# from half a second spends about seven seconds over the budget below — long enough to
+# outlast a bucket refill without looking hung.
 LOG_BUSY_STATUS = 503
 LOG_RETRY_ATTEMPTS = 5
 LOG_RETRY_BASE_SECONDS = 0.5
-LOG_RETRY_MAX_SECONDS = 8.0
+# Fraction of each backoff to vary it by: the bucket is shared, so clients backing off in
+# lockstep would re-collide on it every round.
+LOG_RETRY_JITTER = 0.25
 
 
 def _with_busy_retry(fetch_page):
@@ -51,14 +54,8 @@ def _with_busy_retry(fetch_page):
         except ApiException as exc:
             if exc.status != LOG_BUSY_STATUS:
                 raise
-            retry_after = exc.headers.get("Retry-After") if exc.headers else None
-            if retry_after is not None and retry_after.isdigit():
-                delay = min(float(retry_after), LOG_RETRY_MAX_SECONDS)
-            else:
-                # jitter: the bucket is shared, so unjittered clients re-collide on it
-                backoff = min(LOG_RETRY_BASE_SECONDS * 2**attempt, LOG_RETRY_MAX_SECONDS)
-                delay = backoff * (0.75 + random.random() * 0.5)
-            time.sleep(delay)
+            backoff = LOG_RETRY_BASE_SECONDS * 2**attempt
+            time.sleep(backoff * (1 + random.uniform(-LOG_RETRY_JITTER, LOG_RETRY_JITTER)))
     # the last attempt propagates whatever it raises
     return fetch_page()
 
@@ -457,17 +454,16 @@ def fetch_logs(
         in ascending (timestamp, id) order.
 
         A page request the store answers as busy (HTTP 503) is retried with
-        exponential backoff and jitter, honouring Retry-After when the server
-        sends one; the anchor survives the retry, so an upstream rate limit costs
-        a pause rather than the stream.
-
-        If a page request fails for any other reason, or the retries run out, the
-        iterator raises and, like any generator, cannot be resumed — but every
-        chunk already yielded is complete and none is left half-built. Resume
-        with a new fetch_logs whose start_time is the
-        last delivered event's timestamp: bounds are inclusive, so the only lines
-        delivered again are the ones sharing that millisecond, which the caller
-        already holds.
+        exponential backoff and jitter; the anchor survives the retry, so an
+        upstream rate limit costs a pause rather than the stream.
+
+        If a page request fails for any other reason, or the retries run out,
+        the iterator raises and, like any generator, cannot be resumed — but
+        every chunk already yielded is complete and none is left half-built.
+        Resume with a new fetch_logs whose start_time is the last delivered
+        event's timestamp: bounds are inclusive, so the only lines delivered
+        again are the ones sharing that millisecond, which the caller already
+        holds.
         """
         if not 1 <= chunk_size <= MAX_LOG_PAGE_LINES:
             raise ValueError(

tests/test_sdk_api.py

@@ -17,6 +17,8 @@
 from centml.sdk.api import (
     LOG_DEDUP_RETENTION_MS,
     LOG_RETRY_ATTEMPTS,
+    LOG_RETRY_BASE_SECONDS,
+    LOG_RETRY_JITTER,
     MAX_LOG_PAGE_LINES,
     CentMLClient,
     DeploymentLogSession,
@@ -1031,42 +1033,50 @@ def test_fetch_logs_survives_a_busy_store_without_losing_its_anchor():
 
 def test_fetch_logs_gives_up_on_a_busy_store_after_the_retry_budget():
     api = MagicMock()
-    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = ApiException(status=503)
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [
+        _log_page(_log_event("1-a", 1000))
+    ] + [ApiException(status=503)] * LOG_RETRY_ATTEMPTS
     client = CentMLClient(api)
 
+    stream = client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000, chunk_size=1)
+    assert [e.id for e in next(stream)] == ["1-a"]
+
     with patch("centml.sdk.api.time.sleep") as sleep:
         with pytest.raises(ApiException):
-            next(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000))
+            next(stream)
 
-    assert api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.call_count == LOG_RETRY_ATTEMPTS
+    # The whole budget goes on the one page, and the chunk already delivered stands.
     assert sleep.call_count == LOG_RETRY_ATTEMPTS - 1
 
 
-def test_fetch_logs_does_not_retry_a_request_the_store_rejects():
+def test_fetch_logs_backs_off_further_on_each_busy_answer():
     api = MagicMock()
-    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = ApiException(status=404)
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [ApiException(status=503)] * (
+        LOG_RETRY_ATTEMPTS - 1
+    ) + [_log_page()]
     client = CentMLClient(api)
 
     with patch("centml.sdk.api.time.sleep") as sleep:
-        with pytest.raises(ApiException):
-            next(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000))
+        assert _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000)) == []
 
-    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_called_once()
-    sleep.assert_not_called()
+    waited = [call.args[0] for call in sleep.call_args_list]
+    assert len(waited) == LOG_RETRY_ATTEMPTS - 1
+    for attempt, seconds in enumerate(waited):
+        nominal = LOG_RETRY_BASE_SECONDS * 2**attempt
+        assert nominal * (1 - LOG_RETRY_JITTER) <= seconds <= nominal * (1 + LOG_RETRY_JITTER)
 
 
-def test_fetch_logs_waits_the_retry_after_the_store_asks_for():
-    busy = ApiException(status=503)
-    busy.headers = {"Retry-After": "2"}
+def test_fetch_logs_does_not_retry_a_request_the_store_rejects():
     api = MagicMock()
-    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = [busy, _log_page()]
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.side_effect = ApiException(status=404)
     client = CentMLClient(api)
 
     with patch("centml.sdk.api.time.sleep") as sleep:
-        assert _flatten(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000)) == []
+        with pytest.raises(ApiException):
+            next(client.fetch_logs(123, 2, pod="pod-a", start_time=1, end_time=10_000))
 
-    # Honoured verbatim, not jittered: the server named the delay.
-    sleep.assert_called_once_with(2.0)
+    api.get_deployment_logs_v4_logs_deployment_id_revision_number_get.assert_called_once()
+    sleep.assert_not_called()
 
 
 def test_fetch_logs_is_lazy_until_the_first_next():