diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 763ad6ed..6e487cce 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -10,7 +10,7 @@ "plugins": [ { "name": "kbagent", - "version": "0.84.2", + "version": "0.84.3", "source": "./plugins/kbagent", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "category": "development" diff --git a/plugins/kbagent/.claude-plugin/plugin.json b/plugins/kbagent/.claude-plugin/plugin.json index fce2a941..8e8f47d1 100644 --- a/plugins/kbagent/.claude-plugin/plugin.json +++ b/plugins/kbagent/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "kbagent", - "version": "0.84.2", + "version": "0.84.3", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "author": { "name": "Keboola", diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index 84f75c4c..74ccedbf 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -3633,6 +3633,43 @@ Four things a coding agent will otherwise get wrong: does not accept a Storage API token -- it is the still-open primary ask of issue #594. Do not imply this command covers billing/invoice history; tell the user it is out of reach from the CLI today. + +## A Storage job that failed instantly used to report success (since v0.84.3) + +Every async Storage operation -- table import/export, `create-table`, +`delete-table`, `truncate-table`, snapshot create/restore, dev-branch +create/delete, bucket share, workspace table load -- enqueues a Storage job and +waits on the same shared poller. Before v0.84.3 that poller checked for a +terminal state in **two** places, and they had drifted: the check on each +*polled* body raised `STORAGE_JOB_FAILED` on `status: error`, but the check on +the *initial* response body (the one the enqueuing POST/PUT/DELETE returns) +handed the job straight back to the caller. + +So when the Storage API failed **fast** -- answering with an already-terminal +error instead of a queued `waiting` job -- the failure was swallowed. Every call +site returns either that job or `job.get("results", {})`, and an error job +carries no `results`, so the operation came back as an **empty success**: exit +code 0, no error output, nothing actually done. Jobs that failed after at least +one poll always raised correctly, which is why this never showed up in normal +use -- only fast failures were affected. + +- **Since v0.84.3** the terminal state is evaluated once, at the top of the poll + loop, so the initial body and every polled body travel identical code. A fast + failure now raises `STORAGE_JOB_FAILED` like any other failed job. +- **If you have a script that treated an empty result as success**, it will now + see the error it should always have seen. That is the intended change; check + for `if not result:` style guards written against the old behaviour. +- **Nothing else moved**: wait budgets (`STORAGE_JOB_MAX_WAIT` 60 s, + `IMPORT_JOB_MAX_WAIT` / `EXPORT_JOB_MAX_WAIT` 600 s), poll cadence, the + up-to-one-interval deadline overshoot, error messages and exit codes are + byte-identical. +- **Known narrowness, unchanged by the fix**: the poller recognises only + `success` and `error`. Any other terminal status the Storage API might report + would exhaust the whole budget and surface as `STORAGE_JOB_TIMEOUT` + (`retryable: true`) rather than a failure -- so a wait that times out suspiciously + fast against a small budget is worth reading as "unknown status", not "still + running". + ## `config update --set 'state...'` is now a hard error, not a silent no-op (since v0.84.2) Before v0.84.2, `--set` on `config update` / `config row-update` applied every diff --git a/pyproject.toml b/pyproject.toml index 7b867968..d18aec42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "keboola-cli" -version = "0.84.2" +version = "0.84.3" description = "AI-friendly CLI for managing Keboola projects" readme = "README.md" requires-python = ">=3.12" diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index 427f1e79..7e8a42e1 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -24,6 +24,31 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { + "0.84.3": [ + "Fix: a Storage job that failed instantly no longer reports success. " + "`_wait_for_storage_job`, the poller every async Storage operation waits on, checked " + "for a terminal state in two places -- once on the response body the caller hands in, " + "once on each polled body -- and the two had drifted: the polled check raised " + "`STORAGE_JOB_FAILED` on `status: error`, the initial check handed the job straight " + "back. So whenever the Storage API failed FAST (answering with an already-terminal " + "error instead of a queued `waiting` job) the failure was swallowed. All 19 call " + "sites in `client/` either return that job or its `results` -- and an error job " + "carries no `results` -- so a failed table import, dev-branch create/delete, " + "snapshot, bucket share or table export came back as an empty success: exit 0, no " + "error, nothing done. Only jobs that failed after at least one poll ever raised. " + "The poller now evaluates the terminal state ONCE, at the top of the loop, so the " + "caller's body and every polled body travel identical code and the two checks cannot " + "drift apart again. Wait budgets, poll cadence, overshoot, error messages and exit " + "codes are unchanged -- the only behaviour that moves is that a fast failure now " + "raises instead of returning. Scripts that treated an empty result as success will " + "now see the error they should always have seen. Also adds the poller's first tests " + "(`TestWaitForStorageJob`): it had none of its own, and neither the fast-fail path " + "nor the timeout path was covered anywhere. Hardened the failure message extraction " + "at the same time: a job whose `error` field is a plain string (or an int, as the " + "Metastore once returned) used to raise `AttributeError` -- a traceback instead of a " + "clean `STORAGE_JOB_FAILED` exit -- and a string error's text is now used as the " + "message instead of being discarded.", + ], "0.84.2": [ "New: `kbagent config clone` duplicates a configuration WHOLE (closes #587). " "`--project P --component-id C --config-id ID --name N [--target-project P2] " diff --git a/src/keboola_agent_cli/client/_core.py b/src/keboola_agent_cli/client/_core.py index 8c38774c..c8712f7c 100644 --- a/src/keboola_agent_cli/client/_core.py +++ b/src/keboola_agent_cli/client/_core.py @@ -22,6 +22,32 @@ from ..stream_client import StreamClient +def _storage_job_error_message(job: dict[str, Any]) -> str: + """Best-effort human message out of a failed Storage job's ``error`` field. + + Written tolerantly on purpose. An API ``error`` field is not reliably a + dict in this codebase's experience: the Metastore once answered with an + int (``{"error": 422}``), which is why ``BaseHttpClient._raise_api_error`` + accepts ``error`` only when it is a non-empty string; the Queue poller + guards its own ``result`` with ``isinstance``; and + ``_extract_query_job_error`` handles strings, dicts and unknown shapes. + Assuming a dict here would turn a failed job into an ``AttributeError`` + traceback instead of ``STORAGE_JOB_FAILED`` -- and since the terminal + check now also sees the caller's initial response body, that shape would + arrive from one more direction than before. + """ + error = job.get("error") + if isinstance(error, dict): + message = error.get("message") + # A dict without a usable message falls through to the generic text + # rather than rendering "None" or a raw dict repr at the user. + if isinstance(message, str) and message: + return message + elif isinstance(error, str) and error: + return error + return "Storage job failed" + + class _CoreClient(BaseHttpClient): """Shared plumbing base for the Keboola client mixins.""" @@ -195,8 +221,30 @@ def _wait_for_storage_job( ) -> dict[str, Any]: """Poll a Storage API job until it reaches a terminal state. + The terminal state is evaluated in ONE place for the caller's initial + body and for every polled body alike -- the loop checks before it + fetches. Keep it that way: this used to be two checks (an early return + before the loop plus a second check inside it) and they drifted, so an + already-terminal ERROR initial body was returned as-is instead of + raising. Every call site either returns the job or its ``results``, so + that surfaced as a silent empty success. + + The check-then-fetch *shape* matches the sibling pollers + ``wait_for_queue_job`` / ``wait_for_query_job``; the behaviour does not, + and deliberately so -- this is not a parity claim. Two differences worth + knowing: this poller recognises only ``success`` and ``error``, so any + other terminal status the Storage API might report would exhaust the + whole budget and surface as ``STORAGE_JOB_TIMEOUT`` (the queue poller + keys off ``isFinished`` and ends on any terminal state), and the sleep + here is not capped to the remaining budget, so a wait overshoots its + deadline by up to one poll interval. Both predate this restructure. + Args: - job: Initial job response from POST/DELETE. + job: Initial job response from the request that enqueued the job + (POST, PUT or DELETE -- e.g. ``change_sharing_type`` enqueues + with PUT). May already be terminal (the Storage API can fail + fast, never returning ``waiting``), in which case no request + is made at all. max_wait: Maximum seconds to wait (default: STORAGE_JOB_MAX_WAIT). Returns: @@ -206,25 +254,25 @@ def _wait_for_storage_job( KeboolaApiError: If the job fails or times out. """ job_id = job.get("id") - if job.get("status") in ("success", "error"): - return job - deadline = time.monotonic() + max_wait - while time.monotonic() < deadline: - time.sleep(STORAGE_JOB_POLL_INTERVAL) - response = self._request("GET", f"/v2/storage/jobs/{job_id}") - job = response.json() + while True: status = job.get("status") if status == "success": return job if status == "error": - error_msg = job.get("error", {}).get("message", "Storage job failed") raise KeboolaApiError( - message=error_msg, + message=_storage_job_error_message(job), status_code=500, error_code=ErrorCode.STORAGE_JOB_FAILED, retryable=False, ) + # Checked before sleeping, so an exhausted budget never costs a + # poll interval. + if time.monotonic() >= deadline: + break + time.sleep(STORAGE_JOB_POLL_INTERVAL) + job = self._request("GET", f"/v2/storage/jobs/{job_id}").json() + raise KeboolaApiError( message=f"Storage job {job_id} did not complete within {max_wait}s", status_code=504, diff --git a/tests/test_client.py b/tests/test_client.py index 84472270..d43dcdcd 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -3,7 +3,7 @@ import contextlib import json import logging -from typing import SupportsIndex +from typing import Any, SupportsIndex from unittest.mock import patch from urllib.parse import parse_qs, quote @@ -11,8 +11,8 @@ import pytest from keboola_agent_cli.client import KeboolaClient -from keboola_agent_cli.constants import MAX_RETRIES -from keboola_agent_cli.errors import KeboolaApiError +from keboola_agent_cli.constants import MAX_RETRIES, STORAGE_JOB_POLL_INTERVAL +from keboola_agent_cli.errors import ErrorCode, KeboolaApiError def _noop_sleep(seconds: SupportsIndex | float, /) -> None: @@ -2473,6 +2473,26 @@ def test_load_workspace_tables_preserve_with_branch(self, httpx_mock) -> None: _BASE = "https://connection.keboola.com" +def _mk_client() -> KeboolaClient: + """Client on the shared _BASE/_TOKEN pair, for the poller test classes.""" + return KeboolaClient(stack_url=_BASE, token=_TOKEN) + + +class _Omit: + """Sentinel: leave the key out entirely, as distinct from setting it to None. + + Spelling "absent" as ``None`` would collapse the two shapes, and ``None`` is + the more interesting of them -- ``{"error": None}`` is one of the two bodies + that made the pre-0.84.3 extraction raise AttributeError. + """ + + def __repr__(self) -> str: + return "" + + +_OMIT = _Omit() + + class TestPrepareFileUpload: """Tests for KeboolaClient.prepare_file_upload().""" @@ -3339,12 +3359,157 @@ def test_empty_string_passes_through(self) -> None: assert unwrap("") == "" +class TestWaitForStorageJob: + """Tests for _wait_for_storage_job -- terminal states, fast paths, deadline. + + The shared Storage-job poller is reached by 19 call sites across ``client/`` + (storage tables 16x, dev branches 2x, workspaces 1x) but had no tests of its + own; its contracts are pinned here so they cannot drift apart again. + """ + + def test_already_successful_body_returns_without_polling(self, httpx_mock) -> None: + """A terminal-success initial body is returned as-is, with no HTTP call.""" + with _mk_client() as client: + job = client._wait_for_storage_job({"id": 1, "status": "success", "results": {"x": 1}}) + + assert job["results"] == {"x": 1} + # Asserted explicitly rather than left to pytest_httpx's teardown: an + # unmatched request surfaces as httpx.TimeoutException, which + # _do_request catches and retries with real sleeps -- a regression + # would take seconds and report a confusing KeboolaApiError. + assert httpx_mock.get_requests() == [] + + def test_already_failed_body_raises(self, httpx_mock) -> None: + """A terminal-error initial body raises STORAGE_JOB_FAILED, never returns. + + The initial body is the POST/PUT/DELETE response the caller hands in. A + Storage API fast fail (terminal straight away, never ``waiting``) must + not reach the caller as a normal return value -- every call site either + returns the job or ``job.get("results", {})``, so a silently returned + error job becomes an empty success. + """ + with _mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: + client._wait_for_storage_job({"id": 1, "status": "error", "error": {"message": "boom"}}) + + assert exc_info.value.error_code == ErrorCode.STORAGE_JOB_FAILED + assert "boom" in str(exc_info.value) + assert httpx_mock.get_requests() == [] + + def test_timeout_raises_storage_job_timeout(self, httpx_mock, monkeypatch) -> None: + """max_wait=0 with a non-terminal body raises immediately -- no sleep, no HTTP. + + The empty ``sleeps`` list is the load-bearing assertion: it pins that the + deadline is checked BEFORE the sleep, so an exhausted budget never costs + a poll interval. Asserting only the error code would not -- moving the + check after the sleep still breaks before fetching, so the test would + pass a second slower instead of failing (measured). + """ + sleeps: list[float] = [] + monkeypatch.setattr("keboola_agent_cli.client.time.sleep", sleeps.append) + + with _mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: + client._wait_for_storage_job({"id": 1, "status": "waiting"}, max_wait=0) + + assert exc_info.value.error_code == ErrorCode.STORAGE_JOB_TIMEOUT + assert sleeps == [] + assert httpx_mock.get_requests() == [] + + def test_budget_below_one_interval_still_polls_once(self, httpx_mock, monkeypatch) -> None: + """A budget shorter than the poll interval buys exactly one poll. + + The other half of "poll counts are unchanged": the deadline is checked + before the sleep but not against the sleep's length, so a 0.5 s budget + overshoots by up to one full interval and polls once. Pinned so the + overshoot cannot be silently traded away. + """ + sleeps: list[float] = [] + monkeypatch.setattr("keboola_agent_cli.client.time.sleep", sleeps.append) + httpx_mock.add_response( + url=f"{_BASE}/v2/storage/jobs/1", + method="GET", + json={"id": 1, "status": "success", "results": {"ok": True}}, + ) + + with _mk_client() as client: + job = client._wait_for_storage_job({"id": 1, "status": "waiting"}, max_wait=0.5) + + assert job["results"] == {"ok": True} + assert sleeps == [STORAGE_JOB_POLL_INTERVAL] + assert len(httpx_mock.get_requests()) == 1 + + def test_polled_error_raises(self, httpx_mock, monkeypatch) -> None: + """A non-terminal initial body is polled; a polled error raises.""" + monkeypatch.setattr("keboola_agent_cli.client.time.sleep", _noop_sleep) + httpx_mock.add_response( + url=f"{_BASE}/v2/storage/jobs/1", + method="GET", + json={"id": 1, "status": "error", "error": {"message": "polled boom"}}, + ) + + with _mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: + client._wait_for_storage_job({"id": 1, "status": "waiting"}) + + assert exc_info.value.error_code == ErrorCode.STORAGE_JOB_FAILED + assert "polled boom" in str(exc_info.value) + + def test_string_error_field_still_raises_with_its_text(self, httpx_mock) -> None: + """An `error` that is a plain string is used as the message, not dereferenced. + + The field is not reliably a dict: the Metastore once answered with an int + (see BaseHttpClient._raise_api_error), the Queue poller isinstance-guards + its own `result`, and _extract_query_job_error handles both shapes. + Assuming a dict here would raise AttributeError -- a traceback instead of + a clean STORAGE_JOB_FAILED exit -- and the terminal check now sees the + caller's initial body too, so the shape can arrive from one more + direction than before. + """ + with _mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: + client._wait_for_storage_job({"id": 1, "status": "error", "error": "plain text boom"}) + + assert exc_info.value.error_code == ErrorCode.STORAGE_JOB_FAILED + assert "plain text boom" in str(exc_info.value) + + def test_unusable_error_field_falls_back_to_generic_message(self, httpx_mock) -> None: + """A missing / null / non-string / message-less `error` yields the generic text.""" + for error_field in ({}, {"message": ""}, 422, None, ["boom"], _OMIT): + job: dict[str, Any] = {"id": 1, "status": "error"} + if error_field is not _OMIT: + job["error"] = error_field + with _mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: + client._wait_for_storage_job(job) + + assert exc_info.value.error_code == ErrorCode.STORAGE_JOB_FAILED, error_field + assert "Storage job failed" in str(exc_info.value), error_field + + def test_polled_success_returns_the_polled_body(self, httpx_mock, monkeypatch) -> None: + """A non-terminal initial body is polled until success; results come back. + + The normal happy path. It was covered only incidentally before (via + `storage truncate-table`, whose fixture returns a terminal body and so + never reaches the loop at all). + """ + monkeypatch.setattr("keboola_agent_cli.client.time.sleep", _noop_sleep) + httpx_mock.add_response( + url=f"{_BASE}/v2/storage/jobs/1", + method="GET", + json={"id": 1, "status": "processing"}, + ) + httpx_mock.add_response( + url=f"{_BASE}/v2/storage/jobs/1", + method="GET", + json={"id": 1, "status": "success", "results": {"rows": 7}}, + ) + + with _mk_client() as client: + job = client._wait_for_storage_job({"id": 1, "status": "waiting"}) + + assert job["results"] == {"rows": 7} + assert len(httpx_mock.get_requests()) == 2 + + class TestWaitForQueueJob: """Tests for wait_for_queue_job -- strategy dispatch, deadline, failure.""" - def _mk_client(self): - return KeboolaClient(stack_url=_BASE, token=_TOKEN) - def test_wait_success_on_first_poll(self, httpx_mock, monkeypatch) -> None: """Finished job returns on the first poll; no sleep needed.""" httpx_mock.add_response( @@ -3355,7 +3520,7 @@ def test_wait_success_on_first_poll(self, httpx_mock, monkeypatch) -> None: sleeps: list[float] = [] monkeypatch.setattr("keboola_agent_cli.client.time.sleep", lambda s: sleeps.append(s)) - with self._mk_client() as client: + with _mk_client() as client: job = client.wait_for_queue_job("job-1", max_wait=60.0) assert job["status"] == "success" @@ -3377,7 +3542,7 @@ def test_wait_honors_exponential_intervals(self, httpx_mock, monkeypatch) -> Non sleeps: list[float] = [] monkeypatch.setattr("keboola_agent_cli.client.time.sleep", lambda s: sleeps.append(s)) - with self._mk_client() as client: + with _mk_client() as client: client.wait_for_queue_job("job-2", max_wait=600.0, poll_strategy="exponential") # Two polls -> two sleeps; both at the 2s phase of the curve. @@ -3400,7 +3565,7 @@ def test_wait_honors_fixed_strategy(self, httpx_mock, monkeypatch) -> None: sleeps: list[float] = [] monkeypatch.setattr("keboola_agent_cli.client.time.sleep", lambda s: sleeps.append(s)) - with self._mk_client() as client: + with _mk_client() as client: client.wait_for_queue_job("job-3", max_wait=600.0, poll_strategy="fixed") assert sleeps == [STORAGE_JOB_POLL_INTERVAL] @@ -3408,7 +3573,7 @@ def test_wait_honors_fixed_strategy(self, httpx_mock, monkeypatch) -> None: def test_wait_rejects_unknown_strategy(self) -> None: """Invalid strategy raises ValueError before any network call.""" with ( - self._mk_client() as client, + _mk_client() as client, pytest.raises(ValueError, match="Invalid poll_strategy"), ): client.wait_for_queue_job("job-4", poll_strategy="linear") @@ -3427,7 +3592,7 @@ def test_wait_raises_on_status_error(self, httpx_mock, monkeypatch) -> None: ) monkeypatch.setattr("keboola_agent_cli.client.time.sleep", lambda s: None) - with self._mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: + with _mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: client.wait_for_queue_job("bad-1", max_wait=60.0) assert exc_info.value.error_code == "QUEUE_JOB_FAILED" @@ -3456,7 +3621,7 @@ def fake_monotonic() -> float: monkeypatch.setattr("keboola_agent_cli.client.time.monotonic", fake_monotonic) monkeypatch.setattr("keboola_agent_cli.client.time.sleep", lambda s: None) - with self._mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: + with _mk_client() as client, pytest.raises(KeboolaApiError) as exc_info: client.wait_for_queue_job("slow-1", max_wait=5.0) assert exc_info.value.error_code == "QUEUE_JOB_TIMEOUT" @@ -3491,7 +3656,7 @@ def fake_monotonic() -> float: sleeps: list[float] = [] monkeypatch.setattr("keboola_agent_cli.client.time.sleep", lambda s: sleeps.append(s)) - with self._mk_client() as client, pytest.raises(KeboolaApiError): + with _mk_client() as client, pytest.raises(KeboolaApiError): client.wait_for_queue_job("deadline-1", max_wait=100.0) assert sleeps == [1.0] # trimmed from 2.0 to 1.0 diff --git a/uv.lock b/uv.lock index c08b7997..9ee669cb 100644 --- a/uv.lock +++ b/uv.lock @@ -590,7 +590,7 @@ wheels = [ [[package]] name = "keboola-cli" -version = "0.84.2" +version = "0.84.3" source = { editable = "." } dependencies = [ { name = "croniter" },