fix: own the readiness poll instead of inheriting it

CHANGELOG.md

@@ -19,6 +19,21 @@ Changelog style — terse "bullet + brief why":
   **Breaking:**. See the released entries below for the target density.
 -->
 
+### Fixed
+
+- `HotdataSourceClient` now defines its own `_poll` rather than inheriting one from
+  `ManagedDatabaseClient`, which no longer provides it: the framework's managed
+  client waits on the query run instead of on the result body, and dropped the
+  shared helper with it. Both readiness waits here would otherwise raise
+  `AttributeError` on the next `hotdata-framework` release this pin accepts. The
+  local copy treats `interrupted` as terminal — it was absent, so an interrupted
+  run waited out the full 300s bound and then reported a timeout — and drops
+  `cancelled`, which this API does not send; an unrecognised status keeps waiting
+  and the timeout names it. The wait reads its own `_POLL_TIMEOUT_SECONDS` /
+  `_POLL_SLEEP_SECONDS` rather than the base class's constants, so it rents
+  nothing -- and under distinct names, which avoids shadowing the base values for
+  inherited waits.
+
 ## [0.15.0] - 2026-08-27
 
 ### Changed

src/hotdata_dlt_destination/source_client.py

@@ -44,13 +44,14 @@
 trusted the persisted result is taken regardless of cost: slower and complete
 beats faster and silently short.
 The count is a second request rather than folded into the readiness poll because
-the poll lives in the shared client, and duplicating its sync/async branching to
-save ~12% of the latency would risk the two drifting.
+the poll would then have to branch on sync/async the way the read above it does,
+and duplicating that to save ~12% of the latency would risk the two drifting.
 """
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, TypeVar
+import time
+from typing import TYPE_CHECKING, Any, Protocol, TypeVar
 
 import pyarrow as pa
 import pyarrow.types as pat
@@ -64,7 +65,7 @@
 from hotdata_framework.managed_client import ManagedDatabaseClient
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator
+    from collections.abc import Callable, Iterator
 
 # Rows per Arrow batch handed to dlt. Bounds peak memory for one batch; it does
 # not bound the query, whose size the caller controls with its own LIMIT.
@@ -75,6 +76,16 @@
 T = TypeVar("T", pa.Table, pa.RecordBatch)
 
 
+class _HasStatus(Protocol):
+    """What `_poll` needs off a run or a result: a status, and a reason."""
+
+    status: str
+    error_message: str | None
+
+
+_S = TypeVar("_S", bound=_HasStatus)
+
+
 class IncompleteReadError(RuntimeError):
     """A read returned fewer rows than the result holds.
 
@@ -320,6 +331,58 @@ def _resolve(self, response: object, *, database_id: str) -> tuple[str, int] | N
             "so whether it produced rows cannot be established"
         )
 
+    # This wait's own bounds. `_poll` reading two private attributes off a base
+    # class in another distribution is the coupling that broke these waits once
+    # already, so it reads its own -- but under names of their own rather than
+    # the base class's, which the first attempt reused. Matching those names
+    # would shadow them for every *inherited* method too, so a framework release
+    # that retuned either value would silently stop reaching this client.
+    _POLL_TIMEOUT_SECONDS = 300.0
+    _POLL_SLEEP_SECONDS = 0.4
+
+    # Statuses that end a wait without satisfying it. `interrupted` is terminal
+    # and belongs here -- omitting it made an interrupted run wait out the whole
+    # 300s bound and then report a timeout, which names the wrong problem.
+    # `cancelled`, which this poll used to list, is not a status the API sends.
+    _POLL_FAILURES = frozenset({"failed", "interrupted"})
+
+    def _poll(
+        self,
+        fetch: Callable[[], _S],
+        *,
+        is_ready: Callable[[_S], bool],
+        describe: str,
+    ) -> _S:
+        """Poll ``fetch`` until ``is_ready`` holds, or raise on failure/timeout.
+
+        Owned here rather than inherited. The base class used to provide this and
+        no longer does: it now waits on the query run instead, so the generic
+        helper had no second caller left there. This client still needs a wait of
+        its own shape -- retry per request, not per wait, and ``limit=0`` on the
+        readiness check -- so it keeps one, and a private method on a base class
+        in another package was never a safe thing to build on.
+
+        An unrecognised status keeps polling rather than ending the wait. Calling
+        an unknown status terminal would be easier to diagnose and much worse to
+        live with: one status added upstream would fail every read at once, where
+        waiting costs a single slow call. So the timeout names the status it last
+        saw, which is what makes an omission findable.
+        """
+        deadline = time.monotonic() + self._POLL_TIMEOUT_SECONDS
+        last_status: str | None = None
+        while time.monotonic() < deadline:
+            obj = fetch()
+            last_status = obj.status
+            if is_ready(obj):
+                return obj
+            if obj.status in self._POLL_FAILURES:
+                raise RuntimeError(obj.error_message or f"{describe} {obj.status}")
+            time.sleep(self._POLL_SLEEP_SECONDS)
+        raise TimeoutError(
+            f"{describe} timed out after {self._POLL_TIMEOUT_SECONDS}s "
+            f"(last status: {last_status})"
+        )
+
     def _await_result(self, result_id: str, *, database_id: str) -> tuple[str, int]:
         """Wait for a result to be ready, and take its row count off the way out.
 
@@ -338,9 +401,8 @@ def _await_result(self, result_id: str, *, database_id: str) -> tuple[str, int]:
         response because the total is a HEADER: reading it means the
         ``_with_http_info`` variant, whose wrapper puts the model one level down
         at ``.data``, and ``_poll`` inspects ``obj.status`` directly to spot a
-        failed or cancelled result. Passing the wrapper in breaks that check, so
-        the poll keeps the plain model and the header is fetched once, after
-        ready.
+        failed result. Passing the wrapper in breaks that check, so the poll
+        keeps the plain model and the header is fetched once, after ready.
         """
         results = ResultsApi(self._runtime.api)
         self._poll(

tests/test_client.py

@@ -292,7 +292,7 @@ def test_list_managed_tables_by_id(monkeypatch) -> None:
 def _patch_query_apis(monkeypatch, arrow_table, *, arrow_in_hotdata_client: bool) -> dict:
     from hotdata.models.query_response import QueryResponse as _QR
 
-    scopes: dict[str, list] = {"result": [], "arrow": []}
+    scopes: dict[str, list] = {"run": [], "result": [], "arrow": []}
 
     class FakeQueryApi:
         def __init__(self, api):
@@ -311,11 +311,28 @@ def query(self, request, *, x_database_id):
                 execution_time_ms=1,
             )
 
+    # Which endpoint answers "is it ready?" depends on the installed framework:
+    # up to 0.13 it polls `GET /results/{id}`, after that it waits on the query
+    # run. Both are faked so this suite does not silently bind to one of them --
+    # the assertion below accepts whichever was used.
+    class FakeQueryRunsApi:
+        def __init__(self, api):
+            pass
+
+        def get_query_run(self, query_run_id, *, x_database_id=None):
+            scopes["run"].append(x_database_id)
+            return SimpleNamespace(
+                status="succeeded",
+                result_id="r1",
+                error_message=None,
+                warning_message=None,
+            )
+
     class FakeResultsApi:
         def __init__(self, api):
             pass
 
-        def get_result(self, result_id, *, x_database_id=None):
+        def get_result(self, result_id, *, x_database_id=None, limit=None):
             scopes["result"].append(x_database_id)
             return SimpleNamespace(status="ready", result_id=result_id, error_message=None)
 
@@ -328,7 +345,12 @@ def get_result_arrow(self, result_id, *, x_database_id):
             return arrow_table
 
     monkeypatch.setattr("hotdata_framework.managed_client.QueryApi", FakeQueryApi)
-    monkeypatch.setattr("hotdata_framework.managed_client.ResultsApi", FakeResultsApi)
+    monkeypatch.setattr("hotdata_framework.managed_client.QueryRunsApi", FakeQueryRunsApi)
+    # `raising=False`: the symbol is gone from the module in later framework
+    # versions, which stop reading a result body to learn its status.
+    monkeypatch.setattr(
+        "hotdata_framework.managed_client.ResultsApi", FakeResultsApi, raising=False
+    )
     monkeypatch.setattr("hotdata_framework.managed_client.ArrowResultsApi", FakeArrowResultsApi)
     if arrow_in_hotdata_client:
         monkeypatch.setattr(
@@ -346,7 +368,7 @@ def test_execute_sql_carries_database_scope(monkeypatch) -> None:
     client = _client(_Runtime(), config=_cfg(database_id="db_99"))
     table = client.execute_sql('SELECT * FROM "default"."public"."spans"')
     assert table.num_rows == 2
-    assert scopes["result"] == ["db_99"]
+    assert (scopes["run"] or scopes["result"]) == ["db_99"]
     assert scopes["arrow"] == ["db_99"]
     client.close()
 
@@ -360,7 +382,7 @@ def test_fetch_table_carries_database_scope(monkeypatch) -> None:
     client = _client(rt, config=_cfg(database_id="db_42"))
     table = client.fetch_table(schema="public", table="orders")
     assert table is not None and table.num_rows == 1
-    assert scopes["result"] == ["db_42"]
+    assert (scopes["run"] or scopes["result"]) == ["db_42"]
     assert scopes["arrow"] == ["db_42"]
     client.close()
 

tests/test_e2e_inmemory.py

@@ -44,6 +44,9 @@ def __init__(self) -> None:
         # (db, schema, table) -> (partition_by, sorted_by) as declared at creation
         self.layouts: dict[tuple, tuple[list, list]] = {}
         self.tables: dict[tuple, pa.Table] = {}
+        # The id minted by the most recent query, so the query-run fake can name
+        # the result the run produced.
+        self.last_result_id: str | None = None
         self.uploads: dict[str, pa.Table] = {}
         self.results: dict[str, pa.Table] = {}
         self.load_counts: dict[tuple, int] = {}
@@ -210,27 +213,53 @@ def query(self, request, *, x_database_id):
         be = _ACTIVE["backend"]
         database = be.id_to_name[x_database_id]
         result = be.execute_sql(database, request.sql)
+        # The run is what names the result now, so the id has to be reachable
+        # from the run fake as well as from this reply.
+        rid = be.store_result(result)
+        be.last_result_id = rid
         return QueryResponse(
             columns=[],
             rows=[],
             row_count=result.num_rows,
             preview_row_count=0,
             truncated=False,
             nullable=[],
-            result_id=be.store_result(result),
+            result_id=rid,
             query_run_id="qr",
             execution_time_ms=1,
         )
 
 
 class _FakeResultsApi:
+    """Readiness, as frameworks up to 0.13 ask for it."""
+
     def __init__(self, api):
         pass
 
-    def get_result(self, result_id, *, x_database_id=None):
+    def get_result(self, result_id, *, x_database_id=None, limit=None):
         return SimpleNamespace(status="ready", result_id=result_id, error_message=None)
 
 
+class _FakeQueryRunsApi:
+    """Readiness, and the id of the result the run produced.
+
+    Later frameworks wait on the query run rather than polling the result body
+    for a status. Both fakes are installed so this suite works either side of
+    that change instead of binding to one framework version.
+    """
+
+    def __init__(self, api):
+        pass
+
+    def get_query_run(self, query_run_id, *, x_database_id=None):
+        return SimpleNamespace(
+            status="succeeded",
+            result_id=_ACTIVE["backend"].last_result_id,
+            error_message=None,
+            warning_message=None,
+        )
+
+
 class _FakeArrowResultsApi:
     def __init__(self, api):
         pass
@@ -265,7 +294,10 @@ def backend(monkeypatch):
     be = InMemoryBackend()
     _ACTIVE["backend"] = be
     monkeypatch.setattr(mc, "QueryApi", _FakeQueryApi)
-    monkeypatch.setattr(mc, "ResultsApi", _FakeResultsApi)
+    monkeypatch.setattr(mc, "QueryRunsApi", _FakeQueryRunsApi)
+    # `raising=False`: later frameworks drop this import, having stopped reading
+    # a result body to learn its status.
+    monkeypatch.setattr(mc, "ResultsApi", _FakeResultsApi, raising=False)
     monkeypatch.setattr(mc, "ArrowResultsApi", _FakeArrowResultsApi)
     # hotdata_client.py also references ArrowResultsApi (for execute_sql) and
     # DatabasesApi (for the id-first bind).

tests/test_source_client.py

@@ -1139,3 +1139,72 @@ def test_read_rows_returns_empty_for_an_unpersisted_but_verified_zero() -> None:
     client = FakeSourceClient(table=empty, truncated=False, result_id=None)
     assert client.read_rows("SELECT * FROM public.t", database_id=DB) == []
     assert client.fetch_calls == 0
+
+
+def test_the_readiness_poll_is_owned_here_not_inherited() -> None:
+    """`_poll` must be defined on this class, not merely reachable through it.
+
+    It used to live on `ManagedDatabaseClient`, and this client called it without
+    defining one. When that base stopped waiting on the result and started
+    waiting on the query run, the generic helper lost its second caller there and
+    was removed -- and both readiness waits here would have become
+    `AttributeError` on the next framework release that this package's pin
+    accepted. Nothing in this suite would have caught it, because the method
+    resolved fine against the version installed at the time.
+
+    Reaching into a private method on a base class from another distribution is
+    the thing that made that possible, so the assertion is about ownership rather
+    than about the attribute merely existing.
+    """
+    for name in ("_poll", "_POLL_TIMEOUT_SECONDS", "_POLL_SLEEP_SECONDS"):
+        assert name in HotdataSourceClient.__dict__, (
+            f"{name} must be defined on HotdataSourceClient; inheriting it from "
+            "hotdata-framework couples this package to a private name it does not own"
+        )
+
+
+def test_poll_stops_on_a_run_that_was_interrupted() -> None:
+    """`interrupted` is terminal, and was in neither status this poll knew.
+
+    An interrupted run therefore satisfied no branch, so the wait ran to its full
+    300-second bound and then reported a timeout -- naming the wrong problem, and
+    charging five minutes for it.
+    """
+    client = HotdataSourceClient.__new__(HotdataSourceClient)
+    calls: list[str] = []
+
+    def fetch() -> SimpleNamespace:
+        calls.append("poll")
+        return SimpleNamespace(status="interrupted", error_message="instance lost")
+
+    with pytest.raises(RuntimeError, match="instance lost"):
+        client._poll(fetch, is_ready=lambda r: r.status == "ready", describe="Result r1")
+
+    # One request, not a timeout's worth.
+    assert calls == ["poll"]
+
+
+def test_poll_keeps_waiting_on_an_unknown_status_and_names_it_on_timeout() -> None:
+    """An unrecognised status waits rather than ending the wait.
+
+    Treating an unknown status as terminal is the easier failure to diagnose and
+    much the worse one to suffer: a single status added upstream would fail every
+    read at once, where waiting costs one slow call. What makes the omission
+    findable is the timeout naming the status it last saw -- which is exactly
+    what was missing while `interrupted` went unrecognised.
+    """
+    client = HotdataSourceClient.__new__(HotdataSourceClient)
+
+    def fetch() -> SimpleNamespace:
+        return SimpleNamespace(status="something_new", error_message=None)
+
+    with (
+        pytest.MonkeyPatch.context() as mp,
+        pytest.raises(TimeoutError, match="something_new"),
+    ):
+        # Both constants belong to this class -- and are deliberately not named
+        # after the base class's, which these would otherwise shadow for every
+        # inherited wait as well.
+        mp.setattr(HotdataSourceClient, "_POLL_TIMEOUT_SECONDS", 0.05)
+        mp.setattr(HotdataSourceClient, "_POLL_SLEEP_SECONDS", 0.0)
+        client._poll(fetch, is_ready=lambda r: r.status == "ready", describe="Result r1")