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

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

`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.

* test(retry): cover the Retry-After wiring, and scope the transport claim

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.

Author: anoop-narang

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