fix(load): retry append loads instead of running them at most once

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
 

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.

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

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))

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)

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")

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="<html>nope</html>"))
+    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))