fix(load): submit a managed load as a job and poll, don't hold a request open

A load's duration scales with the data, so waiting for one inside a single HTTP
request makes it depend on every layer between the client and the engine staying
up for the whole thing -- CDN, gateway, and this client's own socket read
deadline. Any one of them ending the request loses the load: the server marks it
`abandoned` and DISCARDS work it had already done, while the table's write lock is
still held against the retry that follows. The retry then collides with that lock
(409 RESOURCE_LOCKED) and the two can alternate indefinitely without the load ever
completing.

Seen in production on a table whose load runs past five minutes: the same load
acquired, was abandoned at ~4m45s, and re-acquired, three times in a row, while
the caller retried 409s for twenty-five minutes. Small tables on the same
connection succeeded in under two seconds throughout, which is what ruled out the
lock itself being at fault.

The server already offers the fix and recommends it for this exact case -- "run
the load as a background job and return a job ID to poll instead of blocking until
it finishes. Recommended for large uploads, which can take longer than an HTTP
request should stay open." So loads now go out with `async` set.

`async_after_ms` keeps the common case unchanged. The server answers 200 with the
result when the load finishes inside the window and only defers to a job when it
does not, so nothing pays for polling that did not need it -- dlt's bookkeeping
tables settle in under a second and would otherwise gain a submit-then-poll round
trip each.

The polling budget is the load's own, deliberately not the 300s that queries and
results share: reusing that is what put a five-minute ceiling on loads to begin
with. Bounded rather than unbounded, so a wedged job still surfaces as a timeout.

`_poll_job` already existed for the index path, which has had this shape all
along; this reuses it rather than adding a second poller.

Worth noting for `append`: it is the one non-idempotent mode, and the reason a
lost response is dangerous there is that "did it land?" becomes unanswerable. A
job id outlives the request that created it, so that ambiguity is gone -- the
retry rule is unchanged here, but it no longer rests on the response surviving.

Verified against a live deployment: the server accepts the new fields, a fast load
still returns inline, and no polling happens when it does. The deferred path is
covered by unit tests -- reproducing it live needs a load slow enough to exceed
the window, which the reporting workload produces and a synthetic 300k-row load
(0.8s) does not.

Author: anoop-narang

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- fix(load): submit managed loads as a job and poll, instead of holding one request open
+
 
 ## [0.12.0] - 2026-08-11
 

hotdata_framework/client.py

@@ -24,6 +24,7 @@
 from hotdata.models.database_default_table_decl import DatabaseDefaultTableDecl
 from hotdata.models.index_info_response import IndexInfoResponse
 from hotdata.models.job_status_response import JobStatusResponse
+from hotdata.models.load_managed_table_response import LoadManagedTableResponse
 from hotdata.models.load_managed_table_request import LoadManagedTableRequest
 from hotdata.models.query_request import QueryRequest
 from hotdata.models.query_response import QueryResponse
@@ -81,6 +82,19 @@
 # Jobs have no "cancelled" state; "partially_succeeded" carries an error_message.
 _JOB_TERMINAL = frozenset({"succeeded", "partially_succeeded", "failed"})
 
+# How long a load may finish INLINE before the server hands back a job instead.
+# Small enough that a slow load stops holding a request open, large enough that
+# the overwhelming majority never become jobs at all: dlt's bookkeeping tables
+# (`_dlt_version`, `_dlt_loads`, `_dlt_pipeline_state`) settle in under a second,
+# and paying a submit-then-poll round trip for those would be a regression.
+_LOAD_INLINE_WAIT_MS = 10_000
+
+# A load's own polling budget. Deliberately NOT the 300s used for queries and
+# results: a load is the one operation here whose duration scales with the data,
+# and reusing the query budget is what put a five-minute ceiling on it in the
+# first place. Bounded rather than unbounded so a wedged job still surfaces.
+_LOAD_JOB_TIMEOUT_S = 3600.0
+
 
 @dataclass(frozen=True)
 class ResultSummary:
@@ -418,10 +432,25 @@ def load_managed_table(
         else:
             assert file is not None
             resolved_upload_id = self.upload_parquet(file)
+        # ASKED FOR AS A JOB, not as a held-open request. A load's duration scales
+        # with the data, and a single request that must survive minutes has to
+        # survive every layer between here and the engine -- CDN, gateway, socket
+        # read timeout -- any one of which ends it. When it ends, the server logs
+        # the load `abandoned` and DISCARDS work it had already done, while the
+        # table's write lock is still held against the retry that follows; the
+        # retry then collides with it (409 RESOURCE_LOCKED) and the pair can spin
+        # indefinitely without the load ever completing. Observed in production on
+        # a table whose load runs past five minutes.
+        #
+        # `async_after_ms` keeps the common case unchanged: the server answers 200
+        # with the result if it finishes inside the window, and only falls back to
+        # a job when it does not. So nothing pays for polling that did not need it.
         request = LoadManagedTableRequest(
             mode=mode,
             upload_id=resolved_upload_id,
             key=key,
+            var_async=True,
+            async_after_ms=_LOAD_INLINE_WAIT_MS,
         )
         try:
             loaded = self.connections().load_managed_table(
@@ -432,6 +461,8 @@ def load_managed_table(
             )
         except ApiException as e:
             raise RuntimeError(api_error_message(e)) from e
+        if isinstance(loaded, SubmitJobResponse):
+            loaded = self._load_response_from_job(loaded.id)
         return LoadManagedTableResult(
             connection_id=loaded.connection_id,
             schema_name=loaded.schema_name,
@@ -893,6 +924,33 @@ def _poll_job(
             f"Job {job_id} did not finish within {timeout_s}s (last status: {last_status})"
         )
 
+    def _load_response_from_job(self, job_id: str) -> LoadManagedTableResponse:
+        """The result of a load the server chose to run as a job.
+
+        Polling replaces waiting on the request, so the outcome is read from
+        durable state rather than from a connection that has to stay alive. That
+        also removes the ambiguity `append` was made non-retryable for: a lost
+        response no longer leaves "did it land?" unanswerable, because the job id
+        outlives the request that created it.
+
+        `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
+        replaced or appended to, and "some of it" is not an answer it can use.
+        """
+        final = self._poll_job(job_id, timeout_s=_LOAD_JOB_TIMEOUT_S)
+        status = enum_value(final.status)
+        if status != "succeeded":
+            raise RuntimeError(
+                final.error_message or f"load job {job_id} finished {status}"
+            )
+        payload = final.result.actual_instance if final.result is not None else None
+        if not isinstance(payload, LoadManagedTableResponse):
+            raise RuntimeError(
+                f"load job {job_id} succeeded without a load result "
+                f"(got {type(payload).__name__})"
+            )
+        return payload
+
     def _wait_result_ready(
         self,
         result_id: str,

tests/test_client.py

@@ -37,11 +37,16 @@ def add_database_table(self, database_id, var_schema, request):
 
 
 class _FakeConnectionsApi:
-    def __init__(self) -> None:
+    def __init__(self, responses=None) -> None:
         self.load_calls: list[tuple[str, str, str]] = []
+        self.requests: list = []
+        self._responses = list(responses) if responses else None
 
     def load_managed_table(self, connection_id, schema, table, request):
         self.load_calls.append((connection_id, schema, table))
+        self.requests.append(request)
+        if self._responses:
+            return self._responses.pop(0)
         return SimpleNamespace(
             connection_id=connection_id,
             schema_name=schema,
@@ -544,3 +549,168 @@ def test_from_env_requires_an_api_key(monkeypatch: pytest.MonkeyPatch):
     monkeypatch.delenv("HOTDATA_API_KEY", raising=False)
     with pytest.raises(RuntimeError, match="HOTDATA_API_KEY"):
         HotdataClient.from_env()
+
+
+# --------------------------------------------------------------------------
+# A load is submitted as a job, not held open on one request
+# --------------------------------------------------------------------------
+
+
+def _load_response(rows=7):
+    from hotdata.models.load_managed_table_response import LoadManagedTableResponse
+
+    return LoadManagedTableResponse(
+        connection_id="conn_1", schema_name="public",
+        table_name="orders", row_count=rows,
+        arrow_schema_json="{}",
+    )
+
+
+def test_a_load_asks_for_a_job_with_an_inline_window():
+    """The request itself is the fix. A load whose duration scales with the data
+    must not depend on one HTTP request surviving minutes through every layer
+    between here and the engine; when such a request dies the server discards the
+    work and leaves the table locked against the retry."""
+    from hotdata_framework.client import _LOAD_INLINE_WAIT_MS
+
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    connections = _FakeConnectionsApi()
+
+    with (
+        patch.object(client, "_databases_api", return_value=_ForbiddenDatabasesApi()),
+        patch.object(client, "connections", return_value=connections),
+    ):
+        client.load_managed_table(db, "orders", schema="public", upload_id="up_1")
+
+    req = connections.requests[0]
+    assert req.var_async is True, "load was not submitted as a job"
+    assert req.async_after_ms == _LOAD_INLINE_WAIT_MS
+
+
+def test_a_load_that_finishes_inline_costs_no_polling():
+    """dlt's bookkeeping tables settle in under a second. Making those pay a
+    submit-then-poll round trip would be a regression, which is what
+    `async_after_ms` exists to prevent."""
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    connections = _FakeConnectionsApi()
+
+    def _forbidden_poll(*a, **k):
+        raise AssertionError("polled a load the server answered inline")
+
+    with (
+        patch.object(client, "_databases_api", return_value=_ForbiddenDatabasesApi()),
+        patch.object(client, "connections", return_value=connections),
+        patch.object(client, "_poll_job", _forbidden_poll),
+    ):
+        result = client.load_managed_table(db, "orders", schema="public", upload_id="up_1")
+
+    assert result.row_count == 3
+
+
+def test_a_load_the_server_defers_is_polled_to_completion():
+    """The 202 path: the result comes from durable job state rather than from a
+    connection that had to stay alive to carry it."""
+    from hotdata.models.submit_job_response import SubmitJobResponse
+
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    connections = _FakeConnectionsApi(responses=[
+        SubmitJobResponse(id="jobs_1", status="running", status_url="/v1/jobs/jobs_1"),
+    ])
+    final = SimpleNamespace(
+        status="succeeded", error_message=None,
+        result=SimpleNamespace(actual_instance=_load_response(rows=91)),
+    )
+
+    with (
+        patch.object(client, "_databases_api", return_value=_ForbiddenDatabasesApi()),
+        patch.object(client, "connections", return_value=connections),
+        patch.object(client, "_poll_job", return_value=final) as poll,
+    ):
+        result = client.load_managed_table(db, "orders", schema="public", upload_id="up_1")
+
+    assert result.row_count == 91
+    assert poll.call_args.args[0] == "jobs_1"
+
+
+def test_a_load_job_gets_its_own_budget_not_the_query_one():
+    """Reusing the 300s query budget is what put a five-minute ceiling on loads in
+    the first place; a load is the one operation whose duration scales with the
+    data."""
+    from hotdata.models.submit_job_response import SubmitJobResponse
+    from hotdata_framework.client import _LOAD_JOB_TIMEOUT_S
+
+    assert _LOAD_JOB_TIMEOUT_S > 300.0
+
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    connections = _FakeConnectionsApi(responses=[
+        SubmitJobResponse(id="jobs_1", status="running", status_url="/v1/jobs/jobs_1"),
+    ])
+    final = SimpleNamespace(
+        status="succeeded", error_message=None,
+        result=SimpleNamespace(actual_instance=_load_response()),
+    )
+
+    with (
+        patch.object(client, "_databases_api", return_value=_ForbiddenDatabasesApi()),
+        patch.object(client, "connections", return_value=connections),
+        patch.object(client, "_poll_job", return_value=final) as poll,
+    ):
+        client.load_managed_table(db, "orders", schema="public", upload_id="up_1")
+
+    assert poll.call_args.kwargs["timeout_s"] == _LOAD_JOB_TIMEOUT_S
+
+
+def test_a_failed_load_job_raises_with_the_server_message():
+    from hotdata.models.submit_job_response import SubmitJobResponse
+
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    connections = _FakeConnectionsApi(responses=[
+        SubmitJobResponse(id="jobs_1", status="running", status_url="/v1/jobs/jobs_1"),
+    ])
+    final = SimpleNamespace(status="failed", error_message="disk full", result=None)
+
+    with (
+        patch.object(client, "_databases_api", return_value=_ForbiddenDatabasesApi()),
+        patch.object(client, "connections", return_value=connections),
+        patch.object(client, "_poll_job", return_value=final),
+    ):
+        try:
+            client.load_managed_table(db, "orders", schema="public", upload_id="up_1")
+        except RuntimeError as e:
+            assert "disk full" in str(e), e
+        else:
+            raise AssertionError("a failed load job did not raise")
+
+
+def test_a_partially_succeeded_load_job_is_not_treated_as_success():
+    """A caller asked for a table's contents to be replaced or appended to;
+    "some of it" is not an answer it can use."""
+    from hotdata.models.submit_job_response import SubmitJobResponse
+
+    client = HotdataClient("k", "ws", host="https://api.hotdata.dev")
+    db = ManagedDatabase(id="db_1", description="mydb", default_connection_id="conn_1")
+    connections = _FakeConnectionsApi(responses=[
+        SubmitJobResponse(id="jobs_1", status="running", status_url="/v1/jobs/jobs_1"),
+    ])
+    final = SimpleNamespace(
+        status="partially_succeeded", error_message="3 rows rejected",
+        result=SimpleNamespace(actual_instance=_load_response()),
+    )
+
+    with (
+        patch.object(client, "_databases_api", return_value=_ForbiddenDatabasesApi()),
+        patch.object(client, "connections", return_value=connections),
+        patch.object(client, "_poll_job", return_value=final),
+    ):
+        try:
+            client.load_managed_table(db, "orders", schema="public", upload_id="up_1")
+        except RuntimeError as e:
+            assert "3 rows rejected" in str(e), e
+        else:
+            raise AssertionError("partially_succeeded was treated as success")
+