From 270487f7cfe32c415d65604de39cdc3f253573a8 Mon Sep 17 00:00:00 2001 From: Anoop Narang Date: Thu, 27 Aug 2026 11:32:19 +0530 Subject: [PATCH 1/2] fix(load): retry append loads instead of running them at most once MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `append` was excluded from retries on the grounds that it is not idempotent: if the server commits the load but the response is lost, a retry would re-append the same rows. That is not how the API behaves. It keys a receipt on `upload_id`, and a re-POST of the same id replays the committed result rather than applying the load again. So the invariant that makes a retry safe is re-sending the same upload, not the mode. `ManagedDatabaseClient` stages once, in `upload_parquet`, outside the operation it retries, so that invariant holds for every mode — and a test now pins it, since a refactor moving the upload inside the retry would break it silently. The exclusion cost availability. The destination serialises writes per table and refuses rather than queues, so concurrent writers to one table are answered `409 RESOURCE_LOCKED`. An append had no budget to wait that out, whatever `max_retries` the caller had configured — the one shape that most needs patience was the one shape that had none. `HotdataClient.load_managed_table(file=...)` uploads inside the call and so does not hold the invariant. It is unwrapped and unaffected. Two supporting changes the above needs: Classify a 409 by its `error.code` rather than by the status alone. `CONFLICT` is terminal — the request cannot succeed as posted, so the previous behaviour spent the whole budget reaching the same answer. `RESOURCE_LOCKED` stays transient. A 409 carrying no error envelope is classified exactly as before. `HotdataError` now carries `status_code`, `code` and `retry_after_seconds`, because the message is flattened and truncated for readability and so cannot serve as a discriminator. Honour `Retry-After` and jitter the backoff. `Retry-After` is a floor on the ramp, capped like the ramp so a mistaken header cannot park an attempt for an hour; jitter of up to +50% is added on top and never subtracted, so a stated `Retry-After` is not undercut. Writers that collide on one table started together and would otherwise retry in lockstep and collide again. This lengthens a 20-attempt budget from 285s to roughly 316-405s. --- CHANGELOG.md | 41 ++++++++++ hotdata_framework/client.py | 8 +- hotdata_framework/errors.py | 116 +++++++++++++++++++++++++--- hotdata_framework/managed_client.py | 50 +++++++++--- tests/test_client.py | 5 +- tests/test_errors.py | 77 ++++++++++++++++++ tests/test_managed_client.py | 100 ++++++++++++++++++++++-- 7 files changed, 363 insertions(+), 34 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d36adcf..6fee6cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- fix(load): retry an `append` load instead of running it at most once. + + `append` was excluded from retries on the grounds that it is not idempotent: + if the server commits but the response is lost, a retry would duplicate rows. + That is not how the server behaves. It keys a receipt on `upload_id`, and a + re-POST of the same id replays the committed result instead of applying the + load again — so what makes a retry safe is re-sending the same upload, not + the mode. This client stages once, in `upload_parquet`, outside the retried + operation, so the invariant holds for every mode. + + The exclusion cost real availability. The destination serialises writes per + table and refuses rather than queues, so concurrent writers to one table get + `409 RESOURCE_LOCKED` — and an append had no budget to wait it out, whatever + `max_retries` the caller had configured. + + `HotdataClient.load_managed_table(file=...)` uploads inside the call and so + does not hold the invariant. It is unwrapped and unaffected. + +- fix(errors): classify a 409 by its `error.code` rather than by the status alone. + + `CONFLICT` is now terminal: it means the request cannot succeed as posted, so + the previous behaviour spent the entire retry budget arriving at the same + answer. `RESOURCE_LOCKED` stays transient. A 409 with no error envelope — a + failed query result, say — is classified as before. + +- fix(retry): honour `Retry-After`, and jitter the backoff. + + `Retry-After` is taken as a floor on the ramp, capped like the ramp so a bad + header cannot park an attempt for an hour. Jitter of up to +50% is added on + top and never subtracted, so a stated `Retry-After` is not undercut. Without + it, writers that collided on one table retry in lockstep and collide again. + + This lengthens a 20-attempt budget from 285s to roughly 316-405s. + +### Added + +- `HotdataError` carries `status_code`, `code` and `retry_after_seconds`. The + message is flattened and truncated for readability, so it could not serve as + a discriminator; these can. ## [0.12.1] - 2026-08-18 diff --git a/hotdata_framework/client.py b/hotdata_framework/client.py index 0946a47..ff6570d 100644 --- a/hotdata_framework/client.py +++ b/hotdata_framework/client.py @@ -985,9 +985,11 @@ def _load_response_from_job(self, job_id: str) -> LoadManagedTableResponse: durable state rather than from a connection that has to stay alive. That also gives a caller a handle: the job id is returned on `LoadManagedTableResult`, so "did it land?" is answerable after a lost - response. `append` stays non-retryable -- knowing the id makes the question - answerable, it does not make a blind re-submission safe, and that call is - the caller's to make. + response. That answer is a convenience rather than a precondition for + retrying: re-sending the same upload_id replays the server's receipt + instead of applying the load a second time, which is what makes a retry + safe in every mode. It stops being safe for a caller that re-stages the + upload, because a fresh upload id has no receipt to replay. `partially_succeeded` is terminal and carries a message, so it is raised rather than returned -- a caller asked for a table's contents to be diff --git a/hotdata_framework/errors.py b/hotdata_framework/errors.py index 89749b6..0f77e43 100644 --- a/hotdata_framework/errors.py +++ b/hotdata_framework/errors.py @@ -1,10 +1,41 @@ from __future__ import annotations +import json +from collections.abc import Mapping + from hotdata.rest import ApiException +# The API explains a 409 with a machine-readable code, and the two it sends +# mean opposite things to a retry policy. RESOURCE_LOCKED is a refusal taken +# before any work: the insert that would have created the unit of work lost a +# unique-constraint race, so nothing was claimed and nothing was written. +# CONFLICT is the opposite — the request cannot succeed as posted, so retrying +# spends the whole budget arriving at the same answer. +_TERMINAL_CONFLICT_CODE = "CONFLICT" + class HotdataError(RuntimeError): - pass + """An API failure, carrying what a retry policy needs to decide. + + The message cannot be the discriminator: it is flattened and truncated for + readability, so keying on it means substring-matching prose. ``status_code`` + and ``code`` are the machine-readable form of the same answer, and + ``retry_after_seconds`` is the server's own estimate of how long the + condition it just refused will last. + """ + + def __init__( + self, + message: str, + *, + status_code: int | None = None, + code: str | None = None, + retry_after_seconds: float | None = None, + ) -> None: + super().__init__(message) + self.status_code = status_code + self.code = code + self.retry_after_seconds = retry_after_seconds class HotdataTransientError(HotdataError): @@ -15,6 +46,71 @@ class HotdataTerminalError(HotdataError): pass +def _error_code(body: object) -> str | None: + """The ``error.code`` an API error envelope carries, if this body is one. + + Not every 409 comes from an endpoint that speaks the envelope — a failed + query result is reported as one and carries a result document instead — so + a missing code is ordinary, and callers fall back to the status. + """ + if not isinstance(body, (str, bytes, bytearray)): + return None + try: + parsed: object = json.loads(body) + except ValueError: + return None + if not isinstance(parsed, Mapping): + return None + error: object = parsed.get("error") + if not isinstance(error, Mapping): + return None + code: object = error.get("code") + return code if isinstance(code, str) else None + + +def _retry_after_seconds(headers: object) -> float | None: + """``Retry-After`` as a number of seconds, when the response states one. + + Only the delta-seconds form is read. That is what the API sends, and the + HTTP-date form would need a comparison against a server clock we do not + have to be worth anything. + """ + if not isinstance(headers, Mapping): + return None + raw: object = headers.get("Retry-After") + if raw is None: + # The SDK hands us urllib3's case-insensitive mapping and the API sends + # the header lower-cased, so the direct hit is what normally answers. + # Fall back for any plain dict that reaches us instead — a missed + # header is silent, and silence here reads as "the server asked for + # nothing". + raw = next((v for k, v in headers.items() if str(k).lower() == "retry-after"), None) + if raw is None: + return None + try: + seconds = float(str(raw).strip()) + except ValueError: + return None + return seconds if seconds >= 0 else None + + +def _error_class(status_code: int, code: str | None) -> type[HotdataError]: + if status_code == 409 and code == _TERMINAL_CONFLICT_CODE: + # The request cannot succeed as posted — an upload already consumed + # with nothing to replay, a receipt naming a different target, an + # incompatible column type. Every retry reaches the same 409. + return HotdataTerminalError + if status_code in (408, 409, 425, 429): + return HotdataTransientError + if status_code == 501: + # Not Implemented is a permanent capability gap (e.g. the storage + # backend cannot issue presigned URLs) — retrying cannot succeed. + return HotdataTerminalError + if 500 <= status_code <= 599: + return HotdataTransientError + return HotdataTerminalError + + def classify_sdk_error(error: Exception) -> HotdataError: if isinstance(error, TimeoutError): return HotdataTransientError(str(error)) @@ -25,16 +121,14 @@ def classify_sdk_error(error: Exception) -> HotdataError: message = f"{status_code}: {error.reason or 'unknown error'}" # The response body is where the API explains itself (e.g. which # header is missing) — without it "400: Bad Request" is undebuggable. - body = getattr(error, "body", None) + body: object = getattr(error, "body", None) if body: message = f"{message} — {' '.join(str(body).split())[:500]}" - if status_code in (408, 409, 425, 429): - return HotdataTransientError(message) - if status_code == 501: - # Not Implemented is a permanent capability gap (e.g. the storage - # backend cannot issue presigned URLs) — retrying cannot succeed. - return HotdataTerminalError(message) - if 500 <= status_code <= 599: - return HotdataTransientError(message) - return HotdataTerminalError(message) + code = _error_code(body) + return _error_class(status_code, code)( + message, + status_code=status_code, + code=code, + retry_after_seconds=_retry_after_seconds(getattr(error, "headers", None)), + ) return HotdataTerminalError(str(error)) diff --git a/hotdata_framework/managed_client.py b/hotdata_framework/managed_client.py index bf2ff69..ed4b9b6 100644 --- a/hotdata_framework/managed_client.py +++ b/hotdata_framework/managed_client.py @@ -7,6 +7,7 @@ from __future__ import annotations +import random import time from collections.abc import Callable from typing import Any, Protocol, TypeVar @@ -53,6 +54,10 @@ class ManagedDatabaseClient: _QUERY_TIMEOUT_SECONDS = 300.0 _POLL_INTERVAL_SECONDS = 0.4 _MAX_BACKOFF_SECONDS = 30.0 + # Spread as a fraction of the wait, added on top of it. Half an interval is + # enough to decorrelate writers that started together without materially + # changing how long the budget lasts. + _RETRY_JITTER_FRACTION = 0.5 def __init__( self, @@ -207,9 +212,16 @@ def load_managed_table( mode: ManagedLoadMode = "replace", key: list[str] | None = None, ) -> LoadManagedTableResult: - # append is the only non-idempotent mode: if the server commits the load - # but the response is lost, a retry re-appends the same rows. Run it - # at-most-once; every other mode is safe to retry. + # Retryable in every mode, append included. A retry re-sends the SAME + # upload_id, and the server keys a receipt on it: a replay returns the + # committed result rather than applying the load a second time. So the + # invariant that makes this safe is the upload id, not the mode — a + # caller that re-stages the upload between attempts mints a new id, + # loses the receipt, and a retried append would then duplicate rows. + # This client stages once, in upload_parquet, outside the operation + # retried here. `HotdataClient.load_managed_table(file=...)` uploads + # inside the call and so does not hold the invariant; it is unwrapped, + # and retrying an append through it is the caller's to justify. # # `key` is the merge key for delete/update/upsert loads: when set it is # matched per-load instead of a key declared at table creation. Omit it @@ -222,20 +234,40 @@ def load_managed_table( upload_id=upload_id, mode=mode, key=key, - ), - retryable=(mode != "append"), + ) ) - def _request_with_retry(self, operation: Callable[[], T], *, retryable: bool = True) -> T: - max_attempts = self._max_retries if retryable else 1 + def _request_with_retry(self, operation: Callable[[], T]) -> T: + max_attempts = self._max_retries for attempt in range(1, max_attempts + 1): try: return operation() except Exception as error: mapped_error = classify_sdk_error(error.__cause__ or error) if isinstance(mapped_error, HotdataTransientError) and attempt < max_attempts: - backoff = min(self._retry_backoff_seconds * attempt, self._MAX_BACKOFF_SECONDS) - time.sleep(backoff) + time.sleep(self._retry_delay(attempt, mapped_error.retry_after_seconds)) continue raise mapped_error from error raise RuntimeError("No retry attempts configured") + + def _retry_delay(self, attempt: int, retry_after_seconds: float | None) -> float: + """A linear ramp, floored by the server's Retry-After and spread by jitter. + + Retry-After is a floor rather than a replacement: it says how long the + condition just refused typically lasts, while the ramp is what gives up + eventually, and taking the larger of the two honours both. It is capped + like the ramp so a hostile or mistaken header cannot park an attempt for + an hour. + + Jitter is added on top and never subtracted, so a stated Retry-After is + not undercut. It matters because the callers that collide are the ones + that started together: writers refused by one table's lock would retry + in lockstep on an identical ramp and re-collide every time. + _MAX_BACKOFF_SECONDS caps the ramp, deliberately not the jitter above + it — clamping the total would flatten every late attempt onto the same + value and re-correlate exactly the waits that most need spreading. + """ + base = min(self._retry_backoff_seconds * attempt, self._MAX_BACKOFF_SECONDS) + if retry_after_seconds is not None: + base = max(base, min(retry_after_seconds, self._MAX_BACKOFF_SECONDS)) + return base * (1.0 + random.random() * self._RETRY_JITTER_FRACTION) diff --git a/tests/test_client.py b/tests/test_client.py index dfe1484..7ead972 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -841,9 +841,8 @@ def test_a_failed_load_job_names_the_job_alongside_the_server_message(): def test_a_deferred_load_returns_the_job_id_to_the_caller(): - """`append` stays non-retryable, so the id is the only handle a caller has to - answer "did it land?" after a lost response -- the same reason - CreateIndexResult carries one.""" + """The id is the handle a caller has to answer "did it land?" after a lost + response -- the same reason CreateIndexResult carries one.""" from hotdata.models.submit_job_response import SubmitJobResponse client = HotdataClient("k", "ws", host="https://api.hotdata.dev") diff --git a/tests/test_errors.py b/tests/test_errors.py index 3094187..c2c5de9 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -31,6 +31,83 @@ def test_classify_sdk_error_without_body_keeps_short_form() -> None: assert str(err) == "409: Conflict" +LOCKED = ( + '{"error":{"code":"RESOURCE_LOCKED","message":"another operation is already ' + 'running for conn:c1:public:_dlt_pipeline_state; retry shortly"}}' +) +CONFLICT = '{"error":{"code":"CONFLICT","message":"upload already consumed"}}' + + +def test_resource_locked_is_transient_and_names_itself() -> None: + """A lock refusal is taken before any work — the insert that would have + created the unit of work lost a unique-constraint race — so nothing was + claimed and a retry is safe.""" + err = classify_sdk_error(ApiException(status=409, reason="Conflict", body=LOCKED)) + assert isinstance(err, HotdataTransientError) + assert err.status_code == 409 + assert err.code == "RESOURCE_LOCKED" + + +def test_conflict_is_terminal_despite_being_a_409() -> None: + """A CONFLICT cannot succeed as posted, so retrying it spends the entire + budget to arrive at the same 409. Classifying every 409 as transient meant + permanent conflicts burned the full ramp before surfacing.""" + err = classify_sdk_error(ApiException(status=409, reason="Conflict", body=CONFLICT)) + assert isinstance(err, HotdataTerminalError) + assert err.code == "CONFLICT" + + +def test_a_409_that_is_not_an_error_envelope_stays_transient() -> None: + """Not every 409 comes from an endpoint that speaks the envelope: a failed + query result is reported as one and carries a result document. With no code + to read, the status decides, and the classification is unchanged.""" + body = '{"result_id":"rslt1","status":"failed","error_message":"query panicked"}' + err = classify_sdk_error(ApiException(status=409, reason="Conflict", body=body)) + assert isinstance(err, HotdataTransientError) + assert err.code is None + + +def _locked(headers: object) -> ApiException: + """A lock refusal carrying response headers. + + ``ApiException`` only populates ``headers`` from a real ``http_resp``, so a + hand-built one sets it after construction — the same attribute the SDK + assigns.""" + err = ApiException(status=409, reason="Conflict", body=LOCKED) + err.headers = headers + return err + + +def test_retry_after_is_read_from_the_response() -> None: + assert classify_sdk_error(_locked({"Retry-After": "5"})).retry_after_seconds == 5.0 + + +def test_retry_after_is_found_however_the_header_is_cased() -> None: + """The API sends it lower-cased. urllib3's mapping is case-insensitive so + the direct lookup normally answers, but a plain dict must not silently read + as "the server asked for nothing".""" + assert classify_sdk_error(_locked({"retry-after": "5"})).retry_after_seconds == 5.0 + + +def test_an_unparseable_retry_after_is_ignored_rather_than_fatal() -> None: + """Only the delta-seconds form is read. An HTTP-date would need a server + clock to be worth anything, and a malformed header must not become an + exception raised while classifying another exception.""" + stamp = "Wed, 21 Oct 2026 07:28:00 GMT" + assert classify_sdk_error(_locked({"Retry-After": stamp})).retry_after_seconds is None + + +def test_headers_that_are_not_a_mapping_are_ignored() -> None: + assert classify_sdk_error(_locked(object())).retry_after_seconds is None + + +def test_a_body_that_is_not_json_does_not_break_classification() -> None: + """A proxy or load balancer can answer with HTML the API never wrote.""" + err = classify_sdk_error(ApiException(status=409, reason="Conflict", body="nope")) + assert isinstance(err, HotdataTransientError) + assert err.code is None + + def test_classify_sdk_error_truncates_and_flattens_body() -> None: noisy = "x\n" * 1000 err = classify_sdk_error(ApiException(status=500, reason="ISE", body=noisy)) diff --git a/tests/test_managed_client.py b/tests/test_managed_client.py index e831734..5e089fe 100644 --- a/tests/test_managed_client.py +++ b/tests/test_managed_client.py @@ -158,9 +158,14 @@ def get_result_arrow(self, result_id: str, *, x_database_id: str) -> pa.Table: assert arrow_scopes == ["db1"] -def _load_recording_runtime(calls: list[str]) -> SimpleNamespace: +def _load_recording_runtime( + calls: list[str], uploads: list[str] | None = None +) -> SimpleNamespace: """A runtime whose ``load_managed_table`` records each mode and always fails - with a transient error, so retry behaviour is observable via ``calls``.""" + with a transient error, so retry behaviour is observable via ``calls``. + + ``uploads`` records the upload id each attempt was sent with — the invariant + that makes retrying safe, so it is worth being able to assert on.""" def load_managed_table( database: str, @@ -172,6 +177,8 @@ def load_managed_table( key: list[str] | None = None, ) -> SimpleNamespace: calls.append(mode) + if uploads is not None: + uploads.append(upload_id) raise TimeoutError("commit succeeded but response was lost") runtime = _fake_runtime() @@ -189,19 +196,42 @@ def _managed_client(max_retries: int) -> Any: ) -def test_append_load_runs_at_most_once(monkeypatch: pytest.MonkeyPatch) -> None: - """``append`` is not idempotent: retrying after a commit whose response was - lost would duplicate rows. A transient failure must surface immediately - without re-appending, even with retries budgeted.""" +def test_append_load_retries_like_every_other_mode(monkeypatch: pytest.MonkeyPatch) -> None: + """``append`` is retried, because the retry re-sends the same ``upload_id`` + and the server replays its receipt for that id instead of appending twice. + + The mode was never what made a retry unsafe, so excluding ``append`` bought + no safety and cost it the whole retry budget — which is the budget that + outlasts a table's write lock.""" monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None) calls: list[str] = [] - client = _managed_client(max_retries=8) + client = _managed_client(max_retries=3) client._runtime = _load_recording_runtime(calls) with pytest.raises(mc.HotdataTransientError): client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append") - assert calls == ["append"] # tried once, never retried + assert calls == ["append", "append", "append"] + + +def test_a_retried_load_re_sends_the_same_upload_id(monkeypatch: pytest.MonkeyPatch) -> None: + """The upload id is the invariant the safety of a retried append rests on: + the server keys its replay receipt on it, so a retry that re-staged the + upload would mint a new id, find no receipt, and duplicate the rows. + + Staging happens in ``upload_parquet``, outside the retried operation. This + pins that arrangement, which a refactor moving the upload inward would + silently break.""" + monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None) + calls: list[str] = [] + uploads: list[str] = [] + client = _managed_client(max_retries=4) + client._runtime = _load_recording_runtime(calls, uploads) + + with pytest.raises(mc.HotdataTransientError): + client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append") + + assert uploads == ["u1", "u1", "u1", "u1"] def test_idempotent_load_retries_on_transient(monkeypatch: pytest.MonkeyPatch) -> None: @@ -217,6 +247,60 @@ def test_idempotent_load_retries_on_transient(monkeypatch: pytest.MonkeyPatch) - assert calls == ["replace", "replace", "replace"] # retried up to max_retries +def test_retry_delay_floors_on_the_servers_retry_after(monkeypatch: pytest.MonkeyPatch) -> None: + """``Retry-After`` states how long the refused condition lasts; the ramp only + knows how many attempts are left. Taking the larger of the two respects both, + so an early attempt does not retry in 1.5s against a lock the server just + said would hold for 5.""" + monkeypatch.setattr(mc.random, "random", lambda: 0.0) + client = _managed_client(max_retries=20) + client._retry_backoff_seconds = 1.5 + + assert client._retry_delay(attempt=1, retry_after_seconds=5.0) == 5.0 + # Past the point the ramp overtakes it, the ramp wins and the floor is inert. + assert client._retry_delay(attempt=10, retry_after_seconds=5.0) == 15.0 + + +def test_retry_delay_only_ever_adds_jitter(monkeypatch: pytest.MonkeyPatch) -> None: + """Jitter is added, never subtracted, so a stated ``Retry-After`` is never + undercut — the point of spreading is to stop writers colliding, not to + retry sooner than the server asked.""" + client = _managed_client(max_retries=20) + client._retry_backoff_seconds = 1.5 + + monkeypatch.setattr(mc.random, "random", lambda: 1.0) + assert client._retry_delay(attempt=1, retry_after_seconds=5.0) == pytest.approx(7.5) + monkeypatch.setattr(mc.random, "random", lambda: 0.0) + assert client._retry_delay(attempt=1, retry_after_seconds=5.0) == 5.0 + + +def test_retry_delay_caps_the_ramp_and_the_floor_but_not_the_jitter() -> None: + """The cap bounds what the ramp and a server-stated floor can ask for. The + jitter deliberately sits above it: clamping the total would land every late + attempt on exactly _MAX_BACKOFF_SECONDS and re-correlate the waits that most + need spreading.""" + client = _managed_client(max_retries=20) + client._retry_backoff_seconds = 1.5 + cap = mc.ManagedDatabaseClient._MAX_BACKOFF_SECONDS + ceiling = cap * (1.0 + mc.ManagedDatabaseClient._RETRY_JITTER_FRACTION) + + # attempt 100 would ramp to 150s, and an hour-long Retry-After is refused too. + assert cap <= client._retry_delay(attempt=100, retry_after_seconds=None) <= ceiling + assert cap <= client._retry_delay(attempt=1, retry_after_seconds=3600.0) <= ceiling + + +def test_retry_delay_decorrelates_callers_that_started_together() -> None: + """Writers refused by one table's lock started together, so an unjittered + ramp has them re-collide on every attempt. Identical inputs must not produce + an identical wait.""" + client = _managed_client(max_retries=20) + client._retry_backoff_seconds = 1.5 + + delays = {client._retry_delay(attempt=3, retry_after_seconds=5.0) for _ in range(50)} + + assert len(delays) > 1 + + def test_load_managed_table_forwards_key(monkeypatch: pytest.MonkeyPatch) -> None: """A per-load ``key`` is passed straight through to the runtime client.""" monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None) From 6e7ab0ddab89fc90153fd087194e9d0699c10b69 Mon Sep 17 00:00:00 2001 From: Anoop Narang Date: Thu, 27 Aug 2026 12:02:58 +0530 Subject: [PATCH 2/2] test(retry): cover the Retry-After wiring, and scope the transport claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review points from #75. Nothing exercised the path that carries `Retry-After` off a classified error and into the sleep. `_retry_delay` was tested directly, and the retry-loop tests stub sleep with a lambda that discards its argument, so a regression passing `None` there — or swapping the two positional args — would have left every test green. Refuse a load the way the API refuses a contended table, record what sleep actually received, and assert the waits floor on the header. That also covers classify → transient → retry end to end, which was only covered per-piece, and the no-header case, so the floor is visibly the header's contribution rather than a hard-coded one. Also pin that a `CONFLICT` surfaces on the first attempt, which is the half of the 409 split that had no loop-level test. The README and `test_retry_policy`'s module docstring both stated "a load is not idempotent" without qualification. That remains true of the transport, which sees a method and a status and cannot know what it would be replaying — but unscoped it reads as repo-wide and contradicts the call-layer retry. Both now say which layer they mean. --- CHANGELOG.md | 5 ++ README.md | 2 +- tests/test_managed_client.py | 104 ++++++++++++++++++++++++++++++++++- tests/test_retry_policy.py | 14 +++-- 4 files changed, 117 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6fee6cd..861078d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 This lengthens a 20-attempt budget from 285s to roughly 316-405s. +- docs: scope the "a load is not idempotent" claim in the README and in + `test_retry_policy` to the transport layer, which is where it is still true + and where those two were always talking about. Left unscoped they read as + repo-wide and contradict the call-layer retry above. + ### Added - `HotdataError` carries `status_code`, `code` and `retry_after_seconds`. The diff --git a/README.md b/README.md index 01278f4..987a10c 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Runtime boundary and guarantees are defined in `CONTRACT.md`. - **Environment-driven client setup** — create clients from `HOTDATA_API_KEY`, optional `HOTDATA_API_URL`, and `HOTDATA_WORKSPACE`. - **Workspace resolution** — choose an explicit workspace from env, otherwise discover workspaces and select the active workspace or first available workspace. -- **HTTP resilience** — retry SQL execution on stale pooled sockets. Transport-level retries are the SDK's own default, which this package leaves in place so a non-idempotent request is never replayed on a response status. +- **HTTP resilience** — retry SQL execution on stale pooled sockets. Transport-level retries are the SDK's own default, which this package leaves in place so a request is never blindly replayed on a response status. That is a claim about the transport, which cannot know what it would be replaying. `ManagedDatabaseClient` retries at the call layer, which can: a managed load is safe to re-send because it carries the same `upload_id` and the API replays its receipt for that id rather than applying the load twice. - **SQL execution helper** — run SQL through `POST /v1/query`, poll async query runs when needed, and return a `QueryResult`. - **Result utilities** — convert query results to records, pandas DataFrames, or metadata dictionaries for adapter display layers. - **History helpers** — list recent results and query run history with normalized dataclasses. diff --git a/tests/test_managed_client.py b/tests/test_managed_client.py index 5e089fe..b7eed9b 100644 --- a/tests/test_managed_client.py +++ b/tests/test_managed_client.py @@ -8,8 +8,10 @@ import pyarrow as pa import pytest from hotdata.models.query_response import QueryResponse +from hotdata.rest import ApiException import hotdata_framework.managed_client as mc +from hotdata_framework.errors import HotdataTerminalError def _query_response(result_id: str) -> QueryResponse: @@ -158,9 +160,7 @@ def get_result_arrow(self, result_id: str, *, x_database_id: str) -> pa.Table: assert arrow_scopes == ["db1"] -def _load_recording_runtime( - calls: list[str], uploads: list[str] | None = None -) -> SimpleNamespace: +def _load_recording_runtime(calls: list[str], uploads: list[str] | None = None) -> SimpleNamespace: """A runtime whose ``load_managed_table`` records each mode and always fails with a transient error, so retry behaviour is observable via ``calls``. @@ -186,6 +186,32 @@ def load_managed_table( return runtime +def _lock_refusing_runtime(retry_after: str | None) -> SimpleNamespace: + """A runtime whose loads are refused the way the API refuses a contended + table: ``409 RESOURCE_LOCKED``, optionally carrying ``Retry-After``.""" + + def load_managed_table( + database: str, + table: str, + *, + schema: str, + upload_id: str, + mode: str, + key: list[str] | None = None, + ) -> SimpleNamespace: + error = ApiException( + status=409, + reason="Conflict", + body='{"error":{"code":"RESOURCE_LOCKED","message":"retry shortly"}}', + ) + error.headers = {"Retry-After": retry_after} if retry_after else {} + raise error + + runtime = _fake_runtime() + runtime.load_managed_table = load_managed_table + return runtime + + def _managed_client(max_retries: int) -> Any: return mc.ManagedDatabaseClient( api_key="k", @@ -247,6 +273,78 @@ def test_idempotent_load_retries_on_transient(monkeypatch: pytest.MonkeyPatch) - assert calls == ["replace", "replace", "replace"] # retried up to max_retries +def test_a_lock_refusal_is_retried_and_waits_the_header_out( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """End to end for the contended-table case: a `409 RESOURCE_LOCKED` is + classified transient, retried, and each wait honours the `Retry-After` the + refusal carried. + + Worth having as one test rather than three: `_retry_delay` is exercised + directly elsewhere, but nothing else covers the wiring that carries the + header off the error and into the sleep. The other retry tests stub sleep + with a lambda that discards its argument, so a regression passing `None` + here — or swapping the two positional arguments — would leave them green.""" + slept: list[float] = [] + monkeypatch.setattr(mc.time, "sleep", lambda seconds: slept.append(seconds)) + monkeypatch.setattr(mc.random, "random", lambda: 0.0) + client = _managed_client(max_retries=4) + client._retry_backoff_seconds = 1.5 + client._runtime = _lock_refusing_runtime(retry_after="5") + + with pytest.raises(mc.HotdataTransientError): + client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append") + + # Four attempts, so three waits — each floored on the header rather than + # taking the ramp's 1.5s / 3.0s / 4.5s. + assert slept == [5.0, 5.0, 5.0] + + +def test_a_lock_refusal_without_a_header_falls_back_to_the_ramp( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The floor is the header's contribution, not a hard-coded one: with no + header the ramp decides, which keeps the refusal path working against a + server that does not state a wait.""" + slept: list[float] = [] + monkeypatch.setattr(mc.time, "sleep", lambda seconds: slept.append(seconds)) + monkeypatch.setattr(mc.random, "random", lambda: 0.0) + client = _managed_client(max_retries=4) + client._retry_backoff_seconds = 1.5 + client._runtime = _lock_refusing_runtime(retry_after=None) + + with pytest.raises(mc.HotdataTransientError): + client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append") + + assert slept == [1.5, 3.0, 4.5] + + +def test_a_permanent_conflict_is_not_retried(monkeypatch: pytest.MonkeyPatch) -> None: + """A `CONFLICT` cannot succeed as posted, so it surfaces on the first + attempt instead of spending the budget to reach the same 409.""" + monkeypatch.setattr(mc.time, "sleep", lambda _seconds: None) + attempts: list[int] = [] + + def load_managed_table(*_args: object, **_kwargs: object) -> SimpleNamespace: + attempts.append(1) + error = ApiException( + status=409, + reason="Conflict", + body='{"error":{"code":"CONFLICT","message":"upload already consumed"}}', + ) + error.headers = {} + raise error + + client = _managed_client(max_retries=8) + client._runtime = _fake_runtime() + client._runtime.load_managed_table = load_managed_table + + with pytest.raises(HotdataTerminalError): + client.load_managed_table("db", "orders", schema="public", upload_id="u1", mode="append") + + assert len(attempts) == 1 + + def test_retry_delay_floors_on_the_servers_retry_after(monkeypatch: pytest.MonkeyPatch) -> None: """``Retry-After`` states how long the refused condition lasts; the ramp only knows how many attempts are left. Taking the larger of the two respects both, diff --git a/tests/test_retry_policy.py b/tests/test_retry_policy.py index fa26ab7..6566634 100644 --- a/tests/test_retry_policy.py +++ b/tests/test_retry_policy.py @@ -1,8 +1,14 @@ -"""A POST must never be replayed because of a response status. +"""A POST must never be replayed *by the transport* because of a response status. -A load is not idempotent: re-sending one that the server is still working on -collides with the write lock the first attempt holds, and the duplicate is -refused. The generated SDK already draws this line — ``hotdata._retry`` retries +Re-sending a load the server is still working on collides with the write lock +the first attempt holds, and the duplicate is refused — so a blind transport +replay buys nothing and spends an attempt. This is a claim about the transport, +which sees a method and a status and cannot know what it would be re-sending. +It is not a claim that loads must never be retried: ``ManagedDatabaseClient`` +retries them at the call layer, where the same ``upload_id`` goes back out and +the API replays its receipt for that id instead of applying the load twice. + +The generated SDK already draws this line — ``hotdata._retry`` retries a *pre-response* connection reset on any method (the stale pooled socket case, where the server did no work) while leaving read timeouts and status retries idempotent-only. This wrapper used to pass its own ``retries=`` into