diff --git a/CHANGELOG.md b/CHANGELOG.md index d36adcf..861078d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,52 @@ 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. + +- 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 + 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/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/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..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,12 @@ 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() @@ -179,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", @@ -189,19 +222,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 +273,132 @@ 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, + 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) 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