fix(load): measure poll-error tolerance in time, and name the job when giving up

Two review points, both on paths where the caller is left not knowing whether the
load landed.

TOLERANCE WAS A COUNT, WHICH MEANS NOTHING ON ITS OWN. Five consecutive failures
at the load's 2s interval is about eight seconds, and the thing being survived is a
gateway blip: a rolling restart or a load balancer reconverging serves 502s for
longer than that. Past it the poll aborted, the caller's retry re-submitted the
load, and the first job was still holding the table -- the door this tolerance
exists to close, just narrower than before.

A count is also the wrong unit, because what it means in wall-clock depends on
whichever `interval_s` the caller passed, and callers differ. Now a time bound:
give up after 120s of CONTINUOUS failure, reset on any success, still bounded by
the poll's own deadline. Failed checks also back off to 15s rather than retrying
every 2s -- whatever is serving 502s does not need the extra traffic.

THE GIVE-UP PATHS DROPPED THE JOB ID. `api_error_message(e)` is the gateway's
message ("502: Bad Gateway"), and the failed-job raise preferred the server's
message, which need not mention the id. Those are exactly the two paths where the
answer is unknown and the id is the only handle for asking. Both name the job now,
matching what `_poll_job`'s own TimeoutError already did.

The test for the tolerance drives a FAKE CLOCK advanced by the sleeps. With
`sleep` patched to a no-op and a real clock, it waited out the whole grace period
in wall time -- 120 seconds of a spinning loop to assert one message, and it
slowed the suite from 0.5s to 121s before I noticed. A test measuring a time bound
has to own the clock.

152 passed in 0.53s. `ruff check` clean on the touched files.

Author: anoop-narang

hotdata_framework/client.py

@@ -95,8 +95,23 @@
 # first place. Bounded rather than unbounded so a wedged job still surfaces.
 _LOAD_JOB_TIMEOUT_S = 3600.0
 
-# Consecutive failed status checks before a poll gives up.
-_JOB_POLL_MAX_CONSECUTIVE_ERRORS = 5
+# How long a poll tolerates CONTINUOUS status-check failure before giving up.
+#
+# Time, not a count: a count means whatever the caller's `interval_s` makes it,
+# and callers differ -- five checks is eight seconds at the load interval and
+# something else at the index one. The thing being survived is a gateway blip,
+# and a rolling restart or a load balancer reconverging routinely serves 502s for
+# longer than a few seconds. Too short and the poll aborts, the caller's retry
+# re-submits the load, and the original job is still holding the table -- the
+# door this tolerance exists to close.
+#
+# Still bounded by the poll's own deadline, so this only decides how a stretch of
+# failures ends, never how long the wait can be.
+_JOB_POLL_ERROR_GRACE_S = 120.0
+
+# Failed checks back off rather than hammering at `interval_s`: whatever is
+# serving 502s does not need the extra traffic.
+_JOB_POLL_ERROR_MAX_BACKOFF_S = 15.0
 
 
 @dataclass(frozen=True)
@@ -921,11 +936,14 @@ def _poll_job(
         jobs = self._jobs_api()
         deadline = time.monotonic() + timeout_s
         last: JobStatusResponse | None = None
-        consecutive_errors = 0
+        # When the current run of failures began, or None while checks succeed.
+        failing_since: float | None = None
+        error_backoff = interval_s
         while time.monotonic() < deadline:
             try:
                 last = jobs.get_job(job_id)
-                consecutive_errors = 0
+                failing_since = None
+                error_backoff = interval_s
             except ApiException as e:
                 # A failed STATUS CHECK is not a failed job. Aborting here throws
                 # away work that is still running, and the caller's retry then
@@ -937,10 +955,20 @@ def _poll_job(
                 # CONSECUTIVE failures are the signal: an isolated 502 is noise, a
                 # run of them means the API is gone and there is nothing to wait
                 # for. The poll's own deadline bounds the total wait regardless.
-                consecutive_errors += 1
-                if consecutive_errors >= _JOB_POLL_MAX_CONSECUTIVE_ERRORS:
-                    raise RuntimeError(api_error_message(e)) from e
-                time.sleep(interval_s)
+                now = time.monotonic()
+                if failing_since is None:
+                    failing_since = now
+                elif now - failing_since >= _JOB_POLL_ERROR_GRACE_S:
+                    # Name the job. This is one of the two paths where the caller
+                    # cannot tell whether the load landed, so the id is the only
+                    # thing that makes the question answerable -- and it is exactly
+                    # what a message like "502: Bad Gateway" leaves out.
+                    raise RuntimeError(
+                        f"Job {job_id} status checks failed for "
+                        f"{_JOB_POLL_ERROR_GRACE_S:.0f}s: {api_error_message(e)}"
+                    ) from e
+                time.sleep(error_backoff)
+                error_backoff = min(error_backoff * 2, _JOB_POLL_ERROR_MAX_BACKOFF_S)
                 continue
             if last.status in _JOB_TERMINAL:
                 return last
@@ -968,9 +996,11 @@ def _load_response_from_job(self, job_id: str) -> LoadManagedTableResponse:
         final = self._poll_job(job_id, timeout_s=_LOAD_JOB_TIMEOUT_S)
         status = enum_value(final.status)
         if status != "succeeded":
-            raise RuntimeError(
-                final.error_message or f"load job {job_id} finished {status}"
-            )
+            # The id goes in whether or not the server's message mentions it: the
+            # caller is being told the load did not succeed, and "which load" is
+            # the next thing it needs.
+            detail = final.error_message or f"finished {status}"
+            raise RuntimeError(f"load job {job_id} {status}: {detail}")
         # `result` is a oneOf wrapper today; tolerate the model arriving directly,
         # the same way the index path does rather than disagreeing with it.
         payload = getattr(final.result, "actual_instance", final.result)

tests/test_client.py

@@ -640,6 +640,7 @@ def test_a_load_job_gets_its_own_budget_not_the_query_one():
     the first place; a load is the one operation whose duration scales with the
     data."""
     from hotdata.models.submit_job_response import SubmitJobResponse
+
     from hotdata_framework.client import _LOAD_JOB_TIMEOUT_S
 
     assert _LOAD_JOB_TIMEOUT_S > 300.0
@@ -744,28 +745,99 @@ def get_job(self, job_id):
     assert calls["n"] == 3
 
 
-def test_a_run_of_failed_status_checks_still_gives_up():
-    """Tolerance is for blips, not for an API that has gone away."""
+def test_a_sustained_run_of_failed_status_checks_gives_up_naming_the_job():
+    """Tolerance is for blips, not for an API that has gone away -- and the message
+    has to name the job, because this is one of the two paths where the caller
+    cannot tell whether the load landed. `502: Bad Gateway` alone does not."""
     from hotdata.exceptions import ApiException
-    from hotdata_framework.client import _JOB_POLL_MAX_CONSECUTIVE_ERRORS
+
+    from hotdata_framework.client import _JOB_POLL_ERROR_GRACE_S
 
     client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
 
     class _Dead:
         def get_job(self, job_id):
             raise ApiException(status=502, reason="Bad Gateway")
 
+    # A FAKE CLOCK, advanced by the sleeps. The tolerance is measured in time, so
+    # a no-op `sleep` with a real clock would make this test wait out the whole
+    # grace period in wall time -- 120 seconds of a spinning loop to assert one
+    # message.
+    clock = {"t": 0.0}
     with (
         patch.object(client, "_jobs_api", return_value=_Dead()),
-        patch("time.sleep", lambda *_: None),
+        patch("time.monotonic", lambda: clock["t"]),
+        patch("time.sleep", lambda s: clock.__setitem__("t", clock["t"] + s)),
     ):
         try:
-            client._poll_job("jobs_1", timeout_s=600.0, interval_s=0.01)
-        except RuntimeError:
-            pass
+            client._poll_job("jobs_1", timeout_s=6000.0, interval_s=1.0)
+        except RuntimeError as e:
+            assert "jobs_1" in str(e), f"gave up without naming the job: {e}"
         else:
             raise AssertionError("polled forever against a dead API")
-    assert _JOB_POLL_MAX_CONSECUTIVE_ERRORS >= 2
+    assert clock["t"] >= _JOB_POLL_ERROR_GRACE_S, "gave up before the grace elapsed"
+    assert clock["t"] < 6000.0, "ran to the poll deadline instead of the grace"
+
+
+def test_the_error_tolerance_is_measured_in_time_not_checks():
+    """A count means whatever the caller's `interval_s` makes it, and callers
+    differ. The thing being survived is a gateway blip, which lasts seconds to
+    minutes regardless of how often we happen to ask."""
+    from hotdata_framework.client import _JOB_POLL_ERROR_GRACE_S
+
+    # long enough to outlast a rolling restart / LB reconverge, not seconds
+    assert _JOB_POLL_ERROR_GRACE_S >= 60.0
+
+
+def test_failed_status_checks_back_off_instead_of_hammering():
+    """Whatever is serving 502s does not need the extra traffic."""
+    from hotdata.exceptions import ApiException
+
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    sleeps: list[float] = []
+    calls = {"n": 0}
+    final = SimpleNamespace(status="succeeded", error_message=None,
+                            result=SimpleNamespace(actual_instance=_load_response()))
+
+    class _Flaky:
+        def get_job(self, job_id):
+            calls["n"] += 1
+            if calls["n"] < 5:
+                raise ApiException(status=502, reason="Bad Gateway")
+            return final
+
+    with (
+        patch.object(client, "_jobs_api", return_value=_Flaky()),
+        patch("time.sleep", lambda s: sleeps.append(s)),
+    ):
+        client._poll_job("jobs_1", timeout_s=600.0, interval_s=1.0)
+
+    failed_waits = sleeps[:4]
+    assert failed_waits == sorted(failed_waits), f"did not back off: {failed_waits}"
+    assert failed_waits[-1] > failed_waits[0]
+
+
+def test_a_failed_load_job_names_the_job_alongside_the_server_message():
+    from hotdata.models.submit_job_response import SubmitJobResponse
+
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    connections = _FakeConnectionsApi(responses=[
+        SubmitJobResponse(id="jobs_42", status="running", status_url="/v1/jobs/jobs_42"),
+    ])
+    final = SimpleNamespace(status="failed", error_message="disk full", result=None)
+
+    with (
+        patch.object(client, "_databases_api", return_value=_ForbiddenDatabasesApi()),
+        patch.object(client, "connections", return_value=connections),
+        patch.object(client, "_poll_job", return_value=final),
+    ):
+        try:
+            client.load_managed_table(db, "orders", schema="public", upload_id="up_1")
+        except RuntimeError as e:
+            assert "disk full" in str(e) and "jobs_42" in str(e), e
+        else:
+            raise AssertionError("a failed load job did not raise")
 
 
 def test_a_deferred_load_returns_the_job_id_to_the_caller():