diff --git a/CLAUDE.md b/CLAUDE.md index 2119a878..70fdb3ca 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -309,7 +309,7 @@ kbagent search QUERY [--project NAME] [--type table|bucket|config|flow|data-app| kbagent job list [--project NAME] [--component-id ID] [--status STATUS] [--limit N] kbagent job detail --project NAME --job-id ID -kbagent job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N] +kbagent job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N] [--idempotency-key KEY] [--force-rerun] kbagent job terminate --project NAME (--job-id ID [--job-id ID ...] | --status any|created|waiting|processing [--component-id ID] [--config-id ID] [--branch ID] [--limit N]) [--dry-run] [--yes] kbagent storage buckets [--project NAME] [--branch ID] diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index b01012ac..da00d3e5 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -94,7 +94,7 @@ Requires a **super-admin** Manage API token (same kind as `org setup`). Same def ## Job History - `job list [--project NAME] [--component-id ID] [--config-id ID] [--status STATUS] [--limit N]` -- list jobs (default 50, max 500) - `job detail --project NAME --job-id ID` -- full job detail with timing and result message -- `job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N]` -- run a job, optionally wait for completion (branch-aware). For configs with linked `keboola.variables` (root-level `configuration.variables_id`), kbagent auto-resolves a `variableValuesId` so transformations bind to the deployed values row. `--variable-values-id` overrides; `--no-variables` skips resolution. `NO_VARIABLE_ROWS` when the linked variables config has zero rows -- fix via `kbagent config variables-set`. `--mode debug` (since v0.43.6) sets the Queue API job `mode` body field to `"debug"`: the component runs with the same configuration + inputs but the worker redirects output to a Storage File tagged `debug-` instead of writing to destination buckets. Use for dry-runs, reproducing a failing job on a production config without touching downstream tables, or harvesting the worker's output bytes (download via `storage file-download --tag debug-`) to feed into VCR fixtures or component test cases. Default `--mode run` is unchanged. Invalid values (`--mode anything-else`) exit 2 at the Click choice gate before any wire call. Under `--wait`, polls with an exponential curve (2s x 30 -> 5s x 48 -> 15s); `--poll-strategy fixed` keeps a constant 1s interval. On FAILED/WARNING/TERMINATED, the last `--log-tail-lines` events (default 200, **0 disables -- recommended for automation pipelines**) are attached as `logTail` in the JSON result (or `details.logTail` on errors). If `--timeout` expires, kbagent issues `kill_job` on the remote and exits **7** (`JOB_TIMEOUT_TERMINATED`) with the cancelled `details.job` + `details.logTail`; if the kill itself fails, exits **4** (`QUEUE_JOB_TIMEOUT`, `retryable=true`). Use jq pattern `.error.details.logTail? // .data.logTail? // []` to pick up the tail regardless of exit code. +- `job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N] [--idempotency-key KEY] [--force-rerun]` -- run a job, optionally wait for completion (branch-aware). `--idempotency-key KEY` (since v0.63.0) makes a replayed run safe: on a second `job run` with the same key, a prior still-running or non-failed job is returned (JSON gains `idempotent_replay: true`, human mode prints a note) instead of creating a duplicate side effect; a prior FAILED run is re-run. Dedup is **client-side** (the Queue API has no server idempotency token -- verified against the live spec) and persisted to `/job_idempotency.json`, so it is scoped to one machine. Reusing a key for a *different* component/config exits with `INVALID_ARGUMENT` rather than returning the wrong job; `--force-rerun` ignores the stored entry and always creates a fresh job. For configs with linked `keboola.variables` (root-level `configuration.variables_id`), kbagent auto-resolves a `variableValuesId` so transformations bind to the deployed values row. `--variable-values-id` overrides; `--no-variables` skips resolution. `NO_VARIABLE_ROWS` when the linked variables config has zero rows -- fix via `kbagent config variables-set`. `--mode debug` (since v0.43.6) sets the Queue API job `mode` body field to `"debug"`: the component runs with the same configuration + inputs but the worker redirects output to a Storage File tagged `debug-` instead of writing to destination buckets. Use for dry-runs, reproducing a failing job on a production config without touching downstream tables, or harvesting the worker's output bytes (download via `storage file-download --tag debug-`) to feed into VCR fixtures or component test cases. Default `--mode run` is unchanged. Invalid values (`--mode anything-else`) exit 2 at the Click choice gate before any wire call. Under `--wait`, polls with an exponential curve (2s x 30 -> 5s x 48 -> 15s); `--poll-strategy fixed` keeps a constant 1s interval. On FAILED/WARNING/TERMINATED, the last `--log-tail-lines` events (default 200, **0 disables -- recommended for automation pipelines**) are attached as `logTail` in the JSON result (or `details.logTail` on errors). If `--timeout` expires, kbagent issues `kill_job` on the remote and exits **7** (`JOB_TIMEOUT_TERMINATED`) with the cancelled `details.job` + `details.logTail`; if the kill itself fails, exits **4** (`QUEUE_JOB_TIMEOUT`, `retryable=true`). Use jq pattern `.error.details.logTail? // .data.logTail? // []` to pick up the tail regardless of exit code. - `job terminate --project NAME (--job-id ID [--job-id ...] | --status any|created|waiting|processing [--component-id ID] [--config-id ID] [--branch ID] [--limit N]) [--dry-run] [--yes]` -- kill running Queue API jobs. Use to stop runaway loops or clean up pile-ups from repeated `job run` calls. Two modes: by ID (single/batch) or by filter (`--status any` catches every killable state). Response partitions IDs into `killed / already_finished / not_found / failed`; safe to re-run idempotently. Kill is async -- poll `job detail` for `isFinished=true`. ## Storage diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index 56be166e..22a9ee37 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -2700,3 +2700,35 @@ hits the SYNCHRONOUS Storage endpoint (no job to poll). `--not-null` on a table that ALREADY HAS ROWS is rejected by the backend with an API error (not a local validation error) unless you also pass `--default` -- the existing rows need a value for the new non-null column. Add `--default` when the table is non-empty. + +### `job run --idempotency-key` is client-side dedup, scoped to one machine (since v0.63.0) + +The Keboola Queue API `POST /jobs` accepts **no** client-supplied idempotency / +dedup token -- verified against the live OpenAPI spec (v1.3.8) and the server +source (an internal `deduplicationId` exists but is daemon-only, never read from +the public create-job request). So `kbagent job run --idempotency-key KEY` +de-duplicates **client-side**: a `/job_idempotency.json` map of +`key -> prior job`. Consequences to know: +- Dedup only works where that file is shared -- it is per config-dir / per + machine. A replay from a *different* machine is NOT deduplicated. +- A prior run that is still running or finished non-failed is returned + (`idempotent_replay: true`); a prior FAILED run is re-run; `--force-rerun` + always creates fresh. +- Reusing a key for a *different* `--component-id`/`--config-id` exits + `INVALID_ARGUMENT` (it refuses to return the wrong job), not silently. +- In-process SDK users get the same via `Client.run_job(idempotency_key=..., + idempotency_store=JobIdempotencyStore(path))` -- the facade is config-dir-free, + so you must supply the store path. + +### The importable SDK is now typed (`py.typed` + result models) (since v0.63.0) + +`keboola_agent_cli` ships a PEP 561 `py.typed` marker, so `mypy`/`ty`/IDEs treat +the in-process library as typed. The high-traffic facade methods return pydantic +models (`JobResult`, `QueryResult`, `UploadTableResult`, `ConfigDetailResult`, +`SyncPushResult`) exported from the package root, instead of bare dicts. Every +model is `extra="allow"`, so a new backend field never raises -- the *named* +fields are the stable, semver-versioned surface and extras stay reachable via +attribute access / `model_dump()`. They also accept the raw API key or the +snake_case field name, so `JobResult.model_validate(service_dict)` works on a +service-layer dict directly. This is a typing/contract addition only; the dict +shapes returned by the service layer and the `--json` CLI output are unchanged. diff --git a/plugins/kbagent/skills/kbagent/references/library-workflow.md b/plugins/kbagent/skills/kbagent/references/library-workflow.md index f420fa93..ea95610b 100644 --- a/plugins/kbagent/skills/kbagent/references/library-workflow.md +++ b/plugins/kbagent/skills/kbagent/references/library-workflow.md @@ -13,10 +13,11 @@ shell operations, use the `kbagent` CLI. | Symbol | Purpose | |--------|---------| -| `Client(url, token, *, branch_id=None)` | Stateless entry point to one project; context manager | +| `Client(url, token, *, branch_id=None, idempotency_store=None)` | Stateless entry point to one project; context manager | | `Client.query(workspace_id, sql, *, transactional=False, limit=500)` | Run SQL in a workspace -> `list[dict]` | | `Client.query_result(workspace_id, sql, ...)` | Same, but typed -> `QueryResult` (columns + truncation) | -| `Client.run_job(component_id, config_id, *, wait=False, ...)` | Run a Queue job -> `JobResult` | +| `Client.run_job(component_id, config_id, *, wait=False, idempotency_key=None, ...)` | Run a Queue job -> `JobResult` (replay-safe with a key) | +| `JobIdempotencyStore(path)` | Client-side dedup map for replay-safe `run_job` | | `Client.config_detail(component_id, config_id, *, branch_id=None)` | One config's detail -> `ConfigDetailResult` | | `Client.upload_table(table_id, file_path, *, incremental=False, ...)` | Import a CSV into an existing table -> `UploadTableResult` | | `Client.files.upload(source, *, name=None, tags=None, permanent=False)` | Upload a path **or** bytes -> `FileEntry` | @@ -26,12 +27,12 @@ shell operations, use the `kbagent` CLI. | `Client.raw` | The underlying `KeboolaClient` for endpoints the facade omits | | `FileEntry` | Uniform file shape: `id, name, tags, created, size_bytes, is_permanent, raw` | -Everything exported from `keboola_agent_cli` (`Client`, `Files`, `FileEntry`, and -the typed result models `JobResult`, `QueryResult`, `UploadTableResult`, -`SyncPushResult`, `ConfigDetailResult`) is committed public API and follows -semver. Since 0.63.0 the package ships a **`py.typed`** marker (PEP 561), so -`mypy` / `ty` / IDEs treat the SDK as typed -- a contract change surfaces at -type-check time, not at runtime. +Everything exported from `keboola_agent_cli` (`Client`, `Files`, `FileEntry`, +`JobIdempotencyStore`, and the typed result models `JobResult`, `QueryResult`, +`UploadTableResult`, `SyncPushResult`, `ConfigDetailResult`) is committed public +API and follows semver. Since 0.63.0 the package ships a **`py.typed`** marker +(PEP 561), so `mypy` / `ty` / IDEs treat the SDK as typed -- a contract change +surfaces at type-check time, not at runtime. ## Typed result models @@ -56,6 +57,38 @@ are the stable surface, but anything else the API returns is preserved never raises. They also accept the raw API key *or* the snake_case field name, so `JobResult.model_validate(service_dict)` works directly on a service-layer dict. +## Replay-safe job runs (idempotency) + +An agentic orchestrator that replays a side-effecting build step after a crash +must not fire the same job twice. The Queue API has **no** server-side +idempotency token, so kbagent dedups client-side: give `run_job` an +`idempotency_key` and a `JobIdempotencyStore` (the facade is config-dir-free, so +*you* choose where the dedup map lives -- typically inside your resume-checkpoint +dir). + +```python +from keboola_agent_cli import Client, JobIdempotencyStore + +store = JobIdempotencyStore("/var/run/myapp/job_idempotency.json") +with Client(url=URL, token=TOKEN, idempotency_store=store) as kbc: + job = kbc.run_job("keboola.ex-db-snowflake", "12345", + idempotency_key="bootstrap-extract", wait=True) + if job.idempotent_replay: + ... # a prior run was returned -- no new job fired +``` + +- A prior run that is still running or finished **non-failed** is returned + (`job.idempotent_replay is True`); a prior **failed** run is re-run; + `force_rerun=True` always creates fresh. +- Reusing a key for a *different* component/config raises (it refuses to return + the wrong job). +- Dedup is scoped to the store file -- a replay from a machine that does not + share it is **not** deduplicated. +- Pass `idempotency_store=` per-call to override the constructor's; passing + `idempotency_key` with no store anywhere raises `ValueError` (the stateless + facade has no config-dir to default it to). The `kbagent job run + --idempotency-key` CLI path defaults the store to `/`. + ## Auth & construction Auth is the storage token you pass in (12-factor) -- nothing is persisted to disk. diff --git a/src/keboola_agent_cli/__init__.py b/src/keboola_agent_cli/__init__.py index ed815b0a..ad720294 100644 --- a/src/keboola_agent_cli/__init__.py +++ b/src/keboola_agent_cli/__init__.py @@ -11,6 +11,7 @@ SyncPushResult, UploadTableResult, ) +from .services.job_idempotency_store import JobIdempotencyStore try: __version__ = version(APP_NAME) @@ -22,6 +23,7 @@ "ConfigDetailResult", "FileEntry", "Files", + "JobIdempotencyStore", "JobResult", "QueryResult", "SyncPushResult", diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index b4131801..51c212ea 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -266,7 +266,7 @@ kbagent job detail --project NAME --job-id ID Full job detail including result message and timing. - kbagent job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N] + kbagent job run --project NAME --component-id ID --config-id ID [--row-id ID ...] [--wait] [--timeout N] [--branch ID] [--mode run|debug] [--variable-values-id ID] [--no-variables] [--poll-strategy exponential|fixed] [--log-tail-lines N] [--idempotency-key KEY] [--force-rerun] Run a Queue API job. --row-id selects specific config rows (repeatable; omit to run entire config). --wait polls until job finishes. --timeout sets max wait in seconds (default 300). Branch-aware. When the config has linked variables (configuration.variables_id), kbagent auto-resolves diff --git a/src/keboola_agent_cli/commands/job.py b/src/keboola_agent_cli/commands/job.py index f2111d0d..24e2d6f0 100644 --- a/src/keboola_agent_cli/commands/job.py +++ b/src/keboola_agent_cli/commands/job.py @@ -238,6 +238,23 @@ def job_run( f"{MAX_LOG_TAIL_LINES}." ), ), + idempotency_key: str | None = typer.Option( + None, + "--idempotency-key", + help=( + "Client-supplied de-duplication token (issue #427). On replay with " + "the same key, a prior still-running or non-failed job is returned " + "instead of creating a duplicate -- safe for resumed/retried build " + "steps. A prior FAILED run is re-run. Dedup is client-side (the " + "Queue API has no idempotency key) and scoped to this machine's " + "config-dir; reusing a key for a different component/config errors." + ), + ), + force_rerun: bool = typer.Option( + False, + "--force-rerun", + help="Ignore any stored --idempotency-key entry and always create a fresh job.", + ), ) -> None: """Run a job for a component configuration. @@ -328,6 +345,8 @@ def job_run( poll_strategy=poll_strategy, log_tail_lines=log_tail_lines, mode=mode, + idempotency_key=idempotency_key, + force_rerun=force_rerun, ) except ConfigError as exc: formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR) @@ -347,6 +366,11 @@ def job_run( if formatter.json_mode: formatter.output(result) else: + if result.get("idempotent_replay"): + formatter.console.print( + "[dim]Idempotency key matched a prior run -- returning the existing " + "job (no new job created).[/dim]" + ) resolved_id = result.get("resolvedVariableValuesId") if resolved_id: formatter.console.print(f"[dim]Bound variable values row: {escape(resolved_id)}[/dim]") diff --git a/src/keboola_agent_cli/constants.py b/src/keboola_agent_cli/constants.py index 083f7f27..68753fd2 100644 --- a/src/keboola_agent_cli/constants.py +++ b/src/keboola_agent_cli/constants.py @@ -115,6 +115,10 @@ # failures without touching production tables. VALID_JOB_MODES: frozenset[str] = frozenset({"run", "debug"}) DEFAULT_JOB_MODE: str = "run" +# Client-side job idempotency store (issue #427). Lives in the config-dir +# alongside config.json; maps idempotency_key -> prior job so a replayed +# `kbagent job run --idempotency-key` does not fire a duplicate side effect. +JOB_IDEMPOTENCY_FILENAME: str = "job_idempotency.json" # Default log-tail length surfaced on FAILED/WARNING/TERMINATED jobs. DEFAULT_LOG_TAIL_LINES: int = 200 # Upper bound to prevent accidentally pulling tens of thousands of events diff --git a/src/keboola_agent_cli/lib.py b/src/keboola_agent_cli/lib.py index ae7e7747..4affe4c4 100644 --- a/src/keboola_agent_cli/lib.py +++ b/src/keboola_agent_cli/lib.py @@ -23,6 +23,11 @@ construction (12-factor: read it from ``KBC_TOKEN`` yourself); nothing is persisted to disk. +For replay-safe job runs (issue #427), pass an ``idempotency_store`` (the facade +is config-dir-free, so the consumer supplies *where* to persist the dedup map -- +typically inside its own resume-checkpoint dir) and an ``idempotency_key`` per +``run_job``. + Everything exported here is committed public API and changes follow semver. For lower-level access (raw Queue/Storage endpoints) reach for the underlying :class:`KeboolaClient` via :attr:`Client.raw`. @@ -45,6 +50,7 @@ ) from .errors import ErrorCode, KeboolaApiError from .result_models import ConfigDetailResult, JobResult, QueryResult, UploadTableResult +from .services.job_idempotency_store import JobIdempotencyStore, run_idempotent_job logger = logging.getLogger(__name__) @@ -206,8 +212,9 @@ def delete(self, file_id: int) -> None: class Client: """Stateless in-process entry point to one Keboola project. - Holds nothing but the stack URL, token, and a single :class:`KeboolaClient` - (which carries the shared retry/backoff). No config-dir, no ``project add``. + Holds nothing but the stack URL, token, a single :class:`KeboolaClient` + (which carries the shared retry/backoff), and an optional idempotency store. + No config-dir, no ``project add``. Args: url: Stack URL, e.g. ``https://connection.keboola.com``. @@ -215,15 +222,27 @@ class Client: branch_id: Dev branch to scope every operation to. ``None`` (default) targets production: Storage Files use the production scope and :meth:`query` resolves the project's default branch on first use. + idempotency_store: Optional default :class:`JobIdempotencyStore` used by + :meth:`run_job` when an ``idempotency_key`` is given (issue #427). + The facade is config-dir-free, so the consumer decides where the + dedup map lives. A per-call ``idempotency_store`` overrides this. """ - def __init__(self, url: str, token: str, *, branch_id: int | None = None) -> None: + def __init__( + self, + url: str, + token: str, + *, + branch_id: int | None = None, + idempotency_store: JobIdempotencyStore | None = None, + ) -> None: if not url: raise ValueError("url is required") if not token: raise ValueError("token is required") self._client = KeboolaClient(stack_url=url, token=token) self._resolved_branch_id = branch_id + self._idempotency_store = idempotency_store self.files = Files(self._client, branch_id) @property @@ -359,6 +378,9 @@ def run_job( wait: bool = False, timeout: float = DEFAULT_JOB_RUN_TIMEOUT, poll_strategy: str = DEFAULT_POLL_STRATEGY, + idempotency_key: str | None = None, + force_rerun: bool = False, + idempotency_store: JobIdempotencyStore | None = None, ) -> JobResult: """Run a Queue API job and return a typed :class:`JobResult`. @@ -367,6 +389,14 @@ def run_job( this thin facade does **not** auto-resolve linked variable values; pass ``variable_values_id`` explicitly if the config needs a values row. + Idempotency (issue #427): pass ``idempotency_key`` (plus an + ``idempotency_store`` here or on the constructor) to make a replayed call + safe -- a prior still-running or non-failed job is returned (with + ``JobResult.idempotent_replay = True``) instead of creating a duplicate. + A prior *failed* run is re-run; ``force_rerun=True`` always creates a + fresh job. The Queue API has no server-side idempotency, so this is + client-side and scoped to the supplied store. + Args: component_id: Component to run, e.g. ``keboola.ex-db-snowflake``. config_id: Configuration ID to run. @@ -378,21 +408,51 @@ def run_job( wait: If True, poll until the job finishes or ``timeout`` elapses. timeout: Max seconds to wait (only used when ``wait=True``). poll_strategy: Wait cadence, one of ``VALID_POLL_STRATEGIES``. + idempotency_key: Client-supplied de-duplication token. + force_rerun: Ignore any stored entry for ``idempotency_key``. + idempotency_store: Per-call store override (else the constructor's). + + Raises: + ValueError: If ``idempotency_key`` is given but no store is available + (the stateless facade cannot invent a persistence location). """ effective_branch = branch_id if branch_id is not None else self._resolved_branch_id - job = self._client.create_job( + + def _create() -> dict[str, Any]: + return self._client.create_job( + component_id=component_id, + config_id=config_id, + config_row_ids=config_row_ids, + mode=mode, + branch_id=effective_branch, + variable_values_id=variable_values_id, + ) + + store = idempotency_store if idempotency_store is not None else self._idempotency_store + if idempotency_key and store is None: + raise ValueError( + "idempotency_key requires an idempotency_store -- pass one to " + "Client(...) or to run_job(...). The stateless facade has no " + "config-dir to default it to." + ) + + job, replayed = run_idempotent_job( + store=store, + key=idempotency_key, component_id=component_id, config_id=config_id, - config_row_ids=config_row_ids, - mode=mode, branch_id=effective_branch, - variable_values_id=variable_values_id, + force_rerun=force_rerun, + create=_create, + fetch=self._client.get_job_detail, ) job_id = str(job.get("id", "")) if wait and job_id: job = self._client.wait_for_queue_job( job_id, max_wait=timeout, poll_strategy=poll_strategy ) + if replayed: + job = {**job, "idempotent_replay": True} return JobResult.model_validate(job) def config_detail( diff --git a/src/keboola_agent_cli/result_models.py b/src/keboola_agent_cli/result_models.py index b138d2bc..dfc5230e 100644 --- a/src/keboola_agent_cli/result_models.py +++ b/src/keboola_agent_cli/result_models.py @@ -87,6 +87,14 @@ class JobResult(_ApiResultModel): validation_alias=AliasChoices("logTail", "log_tail"), description="Trailing job events surfaced on a non-success terminal state (wait mode).", ) + idempotent_replay: bool = Field( + default=False, + description=( + "True when this job was returned from a prior run via a matching " + "idempotency key (issue #427) rather than freshly created -- i.e. no " + "new side effect was fired." + ), + ) @property def succeeded(self) -> bool: diff --git a/src/keboola_agent_cli/services/job_idempotency_store.py b/src/keboola_agent_cli/services/job_idempotency_store.py new file mode 100644 index 00000000..4c92b0f0 --- /dev/null +++ b/src/keboola_agent_cli/services/job_idempotency_store.py @@ -0,0 +1,280 @@ +"""Client-side idempotency store for Queue job runs (issue #427). + +The Keboola Queue API ``POST /jobs`` accepts **no** client-supplied idempotency +key -- verified against the live spec (v1.3.8) and the server source: a +``deduplicationId`` exists internally but is daemon-only and never read from the +public create-job request. So an agentic orchestrator that replays a +side-effecting build step after a crash would create a *duplicate* job. + +This module closes that gap on the client side: a small persistent map of +``idempotency_key -> {job_id, component_id, config_id, branch_id}`` plus a +probe-before-create helper (:func:`run_idempotent_job`). A replayed call with the +same key returns the prior job instead of firing the side effect again. + +Persistence mirrors :class:`~keboola_agent_cli.config_store.ConfigStore`: an +atomic write under an exclusive ``fcntl`` lock with ``0600`` permissions, so +concurrent ``run_job`` calls keyed by different keys do not corrupt the file. + +Scope / limits: +- Dedup is **per store file** (per config-dir / per machine). A replay from a + *different* machine that does not share the store file is not deduplicated. + For DB-enforced, cross-machine dedup the Queue API would need to expose its + internal ``deduplicationId`` -- a separate upstream request. +- Policy: a prior job that is still running or finished non-failed is returned + (no new side effect); a prior job that *failed* is re-run (a replay after a + fix should make progress), unless the caller pins the prior job explicitly. +""" + +from __future__ import annotations + +import json +import logging +import os +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from ..errors import ErrorCode, KeboolaApiError + +logger = logging.getLogger(__name__) + +# fcntl is POSIX-only; on Windows we skip locking (same trade-off as ConfigStore). +try: + import fcntl + + _HAVE_FCNTL = True +except ImportError: # pragma: no cover - Windows + _HAVE_FCNTL = False + +_STORE_VERSION = 1 + +# Terminal statuses that mean "this run did not succeed" -> a replay re-runs it. +# Mirrors ``JobResult.failed``. ``warning`` is a soft-success terminal and is +# deliberately NOT here (we don't re-run a job that warned). +_FAILED_STATUSES: frozenset[str] = frozenset({"error", "terminated", "cancelled"}) + + +def _flock(fd: int, operation: int) -> None: + """Best-effort advisory lock; a no-op where fcntl is unavailable.""" + if _HAVE_FCNTL: + fcntl.flock(fd, operation) + + +@dataclass(frozen=True) +class JobIdempotencyEntry: + """One recorded job run, keyed by the caller's idempotency key.""" + + job_id: str + component_id: str + config_id: str + branch_id: int | None + created_at: str + + @classmethod + def _from_dict(cls, data: dict[str, Any]) -> JobIdempotencyEntry: + return cls( + job_id=str(data.get("job_id", "")), + component_id=str(data.get("component_id", "")), + config_id=str(data.get("config_id", "")), + branch_id=data.get("branch_id"), + created_at=str(data.get("created_at", "")), + ) + + def _to_dict(self) -> dict[str, Any]: + return { + "job_id": self.job_id, + "component_id": self.component_id, + "config_id": self.config_id, + "branch_id": self.branch_id, + "created_at": self.created_at, + } + + +class JobIdempotencyStore: + """A persistent ``idempotency_key -> JobIdempotencyEntry`` map. + + Construct with the path to the JSON state file. For the CLI / service path + this lives alongside ``config.json`` in the config-dir; an in-process SDK + consumer supplies its own path (typically inside its resume-checkpoint dir). + """ + + def __init__(self, path: Path) -> None: + self._path = Path(path) + + @property + def path(self) -> Path: + return self._path + + def _read(self) -> dict[str, Any]: + if not self._path.exists(): + return {"version": _STORE_VERSION, "entries": {}} + try: + raw = self._path.read_text(encoding="utf-8") + data = json.loads(raw) if raw.strip() else {} + except (OSError, json.JSONDecodeError) as exc: + # A corrupt dedup file must not wedge job runs: log and start fresh. + # Worst case is one missed dedup, never a crash on a side-effecting path. + logger.warning("Ignoring unreadable idempotency store %s: %s", self._path, exc) + return {"version": _STORE_VERSION, "entries": {}} + if not isinstance(data, dict): + return {"version": _STORE_VERSION, "entries": {}} + data.setdefault("entries", {}) + return data + + def lookup(self, key: str) -> JobIdempotencyEntry | None: + """Return the entry recorded for ``key``, or ``None`` if unseen.""" + entry = self._read().get("entries", {}).get(key) + if not isinstance(entry, dict): + return None + return JobIdempotencyEntry._from_dict(entry) + + def record( + self, + key: str, + *, + job_id: str, + component_id: str, + config_id: str, + branch_id: int | None = None, + ) -> JobIdempotencyEntry: + """Persist (or overwrite) the entry for ``key`` under an exclusive lock.""" + entry = JobIdempotencyEntry( + job_id=job_id, + component_id=component_id, + config_id=config_id, + branch_id=branch_id, + created_at=datetime.now(UTC).isoformat(), + ) + self._path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + lock_fd = os.open(str(self._path), os.O_RDONLY | os.O_CREAT, 0o600) + try: + _flock(lock_fd, fcntl.LOCK_EX if _HAVE_FCNTL else 0) + # Read-modify-write *inside* the lock so concurrent records (distinct + # keys) don't clobber each other. + data = self._read() + data["version"] = _STORE_VERSION + data.setdefault("entries", {})[key] = entry._to_dict() + tmp_path = self._path.with_suffix(".tmp") + fd = os.open(str(tmp_path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + json.dump(data, fh, indent=2) + os.replace(str(tmp_path), str(self._path)) + finally: + if _HAVE_FCNTL: + _flock(lock_fd, fcntl.LOCK_UN) + os.close(lock_fd) + return entry + + def forget(self, key: str) -> None: + """Drop the entry for ``key`` (e.g. to force a fresh run next time).""" + if not self._path.exists(): + return + lock_fd = os.open(str(self._path), os.O_RDONLY | os.O_CREAT, 0o600) + try: + _flock(lock_fd, fcntl.LOCK_EX if _HAVE_FCNTL else 0) + data = self._read() + if data.get("entries", {}).pop(key, None) is not None: + tmp_path = self._path.with_suffix(".tmp") + fd = os.open(str(tmp_path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + json.dump(data, fh, indent=2) + os.replace(str(tmp_path), str(self._path)) + finally: + if _HAVE_FCNTL: + _flock(lock_fd, fcntl.LOCK_UN) + os.close(lock_fd) + + +def _job_is_failed(job: dict[str, Any]) -> bool: + return str(job.get("status") or "") in _FAILED_STATUSES + + +def run_idempotent_job( + *, + store: JobIdempotencyStore | None, + key: str | None, + component_id: str, + config_id: str, + branch_id: int | None, + force_rerun: bool, + create: Callable[[], dict[str, Any]], + fetch: Callable[[str], dict[str, Any]], +) -> tuple[dict[str, Any], bool]: + """Probe-before-create dispatch for an idempotent job run. + + Returns ``(job, replayed)``. When ``key`` (and a ``store``) are given and a + prior non-failed run exists, that prior job is fetched and returned + (``replayed=True``) instead of creating a new one. A prior *failed* run, a + purged (404) prior job, or ``force_rerun=True`` falls through to a fresh + ``create()`` (which is then recorded under ``key``). + + Args: + store: The dedup store, or ``None`` to disable dedup entirely. + key: The caller's idempotency key, or ``None`` to disable dedup. + component_id / config_id: Identify the job; also used to detect a key + reused for a *different* job (raises rather than return a wrong job). + branch_id: Recorded with the entry (informational). + force_rerun: If True, always create a new job (and overwrite the entry). + create: Thunk that creates the job and returns its dict. + fetch: ``job_id -> job dict`` (Queue ``get_job_detail``); used to + resurrect a prior job's current state. + + Raises: + KeboolaApiError: If ``key`` was already recorded for a different + component/config (``ErrorCode.INVALID_ARGUMENT``). + """ + if not key or store is None: + return create(), False + + existing = store.lookup(key) + if existing is not None and not force_rerun: + if existing.component_id != component_id or existing.config_id != config_id: + raise KeboolaApiError( + message=( + f"Idempotency key {key!r} was already used for a different job " + f"({existing.component_id}/{existing.config_id}); refusing to reuse " + f"it for {component_id}/{config_id}. Use a distinct key, or " + f"--force-rerun to override." + ), + status_code=0, + error_code=ErrorCode.INVALID_ARGUMENT, + ) + prior = _safe_fetch(fetch, existing.job_id) + if prior is not None and not _job_is_failed(prior): + logger.info( + "idempotency: key %r -> returning prior job %s (status=%s)", + key, + existing.job_id, + prior.get("status"), + ) + return prior, True + # Prior run failed / was purged -> fall through and re-run. + + job = create() + store.record( + key, + job_id=str(job.get("id", "")), + component_id=component_id, + config_id=config_id, + branch_id=branch_id, + ) + return job, False + + +def _safe_fetch(fetch: Callable[[str], dict[str, Any]], job_id: str) -> dict[str, Any] | None: + """Fetch a prior job's state; treat a 404 (purged job) as "gone" -> re-run. + + Any *other* API error propagates: when we cannot determine the prior job's + state we must not silently create a duplicate side effect. + """ + if not job_id: + return None + try: + return fetch(job_id) + except KeboolaApiError as exc: + if exc.status_code == 404: + logger.info("idempotency: prior job %s not found (404); will re-run", job_id) + return None + raise diff --git a/src/keboola_agent_cli/services/job_service.py b/src/keboola_agent_cli/services/job_service.py index e0e72cdd..f3095c1a 100644 --- a/src/keboola_agent_cli/services/job_service.py +++ b/src/keboola_agent_cli/services/job_service.py @@ -14,6 +14,7 @@ DEFAULT_JOB_MODE, DEFAULT_LOG_TAIL_LINES, DEFAULT_POLL_STRATEGY, + JOB_IDEMPOTENCY_FILENAME, JOB_TERMINATE_GRACE_SECONDS, JOB_TERMINATE_POLL_INTERVAL, KILLABLE_JOB_STATUSES, @@ -23,6 +24,7 @@ from ..errors import ErrorCode, KeboolaApiError from ..models import ProjectConfig from .base import BaseService +from .job_idempotency_store import JobIdempotencyStore, run_idempotent_job logger = logging.getLogger(__name__) @@ -324,6 +326,8 @@ def run_job( poll_strategy: str = DEFAULT_POLL_STRATEGY, log_tail_lines: int = DEFAULT_LOG_TAIL_LINES, mode: str = DEFAULT_JOB_MODE, + idempotency_key: str | None = None, + force_rerun: bool = False, ) -> dict[str, Any]: """Create and optionally wait for a Queue API job. @@ -372,6 +376,17 @@ def run_job( ``debug-`` instead of into destination buckets, so the job can be safely re-run for diagnostics without touching production data. + idempotency_key: Optional client-supplied de-duplication token + (issue #427). When a prior run recorded under the same key is + still running or finished non-failed, that job is returned + instead of creating a duplicate -- so an interrupted, replayed + build step is safe. A prior *failed* run is re-run. The Queue + API has no server-side idempotency, so dedup is client-side and + scoped to the local config-dir store (per machine). Reusing a key + for a *different* component/config raises rather than return a + wrong job. + force_rerun: If True, ignore any recorded ``idempotency_key`` entry + and always create a fresh job (overwriting the stored entry). Returns: Job dict with ``project_alias``. If wait=True, returns the @@ -419,13 +434,26 @@ def run_job( branch_id=branch_id, ) - job = client.create_job( + def _create() -> dict[str, Any]: + return client.create_job( + component_id=component_id, + config_id=config_id, + config_row_ids=config_row_ids, + branch_id=branch_id, + variable_values_id=resolved_values_id, + mode=mode, + ) + + store = self._default_idempotency_store() if idempotency_key else None + job, replayed = run_idempotent_job( + store=store, + key=idempotency_key, component_id=component_id, config_id=config_id, - config_row_ids=config_row_ids, branch_id=branch_id, - variable_values_id=resolved_values_id, - mode=mode, + force_rerun=force_rerun, + create=_create, + fetch=client.get_job_detail, ) job_id = str(job.get("id", "")) @@ -456,8 +484,20 @@ def run_job( job["project_alias"] = alias if resolved_values_id: job["resolvedVariableValuesId"] = resolved_values_id + if replayed: + job["idempotent_replay"] = True return job + def _default_idempotency_store(self) -> JobIdempotencyStore: + """The config-dir-scoped dedup store for the CLI / service job-run path. + + In-process SDK consumers supply their own store (see + ``keboola_agent_cli.Client``); the CLI path persists alongside + ``config.json`` so a replayed `kbagent job run --idempotency-key` is + deduplicated across invocations on the same machine. + """ + return JobIdempotencyStore(self._config_store.config_dir / JOB_IDEMPOTENCY_FILENAME) + def _handle_wait_error( self, client: Any, diff --git a/tests/test_cli.py b/tests/test_cli.py index 67bdf32c..b6746e51 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3289,6 +3289,58 @@ def test_job_run_with_branch_flag(self, tmp_path: Path) -> None: call_kwargs[1].get("branch_id") == 789 if call_kwargs[1] else False ) + def test_job_run_idempotency_key_forwarded(self, tmp_path: Path) -> None: + """job run --idempotency-key K --force-rerun forwards both to run_job, + and a replayed result prints the dedup note in human mode (issue #427).""" + config_dir = tmp_path / "config" + config_dir.mkdir() + + store = _setup_config_test( + config_dir, + {"prod": {"token": "901-55555-fakeTestTokenDoNotUseXXXXXXXX"}}, + ) + + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.ProjectService") as MockProjService, + patch("keboola_agent_cli.cli.ConfigService") as MockCfgService, + patch("keboola_agent_cli.cli.JobService") as MockJobService, + ): + MockStore.return_value = store + job_service = MagicMock() + job_service.run_job.return_value = { + "id": 555, + "status": "waiting", + "idempotent_replay": True, + } + MockJobService.return_value = job_service + MockProjService.return_value = ProjectService(config_store=store) + MockCfgService.return_value = ConfigService(config_store=store) + + result = runner.invoke( + app, + [ + "job", + "run", + "--project", + "prod", + "--component-id", + "keboola.snowflake-transformation", + "--config-id", + "100", + "--idempotency-key", + "build-step-1", + "--force-rerun", + ], + ) + + assert result.exit_code == 0, f"Exit code {result.exit_code}: {result.output}" + call_kwargs = job_service.run_job.call_args.kwargs + assert call_kwargs.get("idempotency_key") == "build-step-1" + assert call_kwargs.get("force_rerun") is True + # Human mode surfaces the replay note when the result was deduplicated. + assert "prior run" in result.output + def test_job_run_active_branch_resolved(self, tmp_path: Path) -> None: """job run without --branch uses active_branch_id from config.""" config_dir = tmp_path / "config" diff --git a/tests/test_job_idempotency_store.py b/tests/test_job_idempotency_store.py new file mode 100644 index 00000000..da2ab1d0 --- /dev/null +++ b/tests/test_job_idempotency_store.py @@ -0,0 +1,250 @@ +"""Tests for the client-side job idempotency store + dispatch (issue #427). + +Covers the persistence (record/lookup/forget, atomic file, 0600, corrupt-file +tolerance) and the probe-before-create policy in ``run_idempotent_job``: +return-prior on non-failed, re-run on failed/404, force_rerun, and the +collision guard when a key is reused for a different component/config. +""" + +import json +import sys +from pathlib import Path +from typing import Any + +import pytest + +from keboola_agent_cli.errors import ErrorCode, KeboolaApiError +from keboola_agent_cli.services.job_idempotency_store import ( + JobIdempotencyEntry, + JobIdempotencyStore, + run_idempotent_job, +) + + +@pytest.fixture +def store(tmp_path: Path) -> JobIdempotencyStore: + return JobIdempotencyStore(tmp_path / "job_idempotency.json") + + +def _entry(store: JobIdempotencyStore, key: str) -> JobIdempotencyEntry: + """lookup() that asserts a hit -- keeps the assertions terse and type-clean.""" + got = store.lookup(key) + assert got is not None, f"expected a recorded entry for {key!r}" + return got + + +class _Create: + """A create() thunk returning a fixed job dict, recording its call count.""" + + def __init__(self, job: dict[str, Any]) -> None: + self._job = job + self.calls: list[int] = [] + + def __call__(self) -> dict[str, Any]: + self.calls.append(1) + return self._job + + +class TestStorePersistence: + def test_lookup_unseen_key_is_none(self, store: JobIdempotencyStore) -> None: + assert store.lookup("nope") is None + + def test_record_then_lookup(self, store: JobIdempotencyStore) -> None: + entry = store.record("K1", job_id="job-1", component_id="c", config_id="cfg", branch_id=7) + assert isinstance(entry, JobIdempotencyEntry) + got = _entry(store, "K1") + assert got.job_id == "job-1" + assert got.component_id == "c" + assert got.config_id == "cfg" + assert got.branch_id == 7 + assert got.created_at # ISO timestamp set + + def test_record_overwrites(self, store: JobIdempotencyStore) -> None: + store.record("K1", job_id="old", component_id="c", config_id="cfg") + store.record("K1", job_id="new", component_id="c", config_id="cfg") + assert _entry(store, "K1").job_id == "new" + + def test_persists_across_instances(self, tmp_path: Path) -> None: + path = tmp_path / "idem.json" + JobIdempotencyStore(path).record("K", job_id="j", component_id="c", config_id="cfg") + # A fresh store object (simulating a process restart) sees the entry. + assert _entry(JobIdempotencyStore(path), "K").job_id == "j" + + def test_forget(self, store: JobIdempotencyStore) -> None: + store.record("K", job_id="j", component_id="c", config_id="cfg") + store.forget("K") + assert store.lookup("K") is None + # forgetting an unknown key (or with no file yet) is a no-op, not an error + store.forget("missing") + + @pytest.mark.skipif(sys.platform == "win32", reason="POSIX file mode") + def test_file_is_0600(self, store: JobIdempotencyStore) -> None: + store.record("K", job_id="j", component_id="c", config_id="cfg") + assert (store.path.stat().st_mode & 0o777) == 0o600 + + def test_valid_json_on_disk(self, store: JobIdempotencyStore) -> None: + store.record("K", job_id="j", component_id="c", config_id="cfg", branch_id=None) + data = json.loads(store.path.read_text()) + assert data["version"] == 1 + assert data["entries"]["K"]["job_id"] == "j" + assert data["entries"]["K"]["branch_id"] is None + + def test_corrupt_file_is_tolerated(self, store: JobIdempotencyStore) -> None: + store.path.write_text("{ this is not valid json") + # A corrupt dedup file must never wedge a job run. + assert store.lookup("anything") is None + # and a subsequent record recovers the file + store.record("K", job_id="j", component_id="c", config_id="cfg") + assert _entry(store, "K").job_id == "j" + + def test_concurrent_distinct_keys_dont_clobber(self, store: JobIdempotencyStore) -> None: + # Read-modify-write under lock: recording many keys keeps them all. + for i in range(20): + store.record(f"K{i}", job_id=f"j{i}", component_id="c", config_id="cfg") + for i in range(20): + assert _entry(store, f"K{i}").job_id == f"j{i}" + + +class TestRunIdempotentJob: + def test_no_key_always_creates(self, store: JobIdempotencyStore) -> None: + create = _Create({"id": "j1", "status": "processing"}) + job, replayed = run_idempotent_job( + store=store, + key=None, + component_id="c", + config_id="cfg", + branch_id=None, + force_rerun=False, + create=create, + fetch=lambda jid: {}, + ) + assert job["id"] == "j1" and replayed is False + + def test_first_call_creates_and_records(self, store: JobIdempotencyStore) -> None: + create = _Create({"id": "j1", "status": "processing"}) + job, replayed = run_idempotent_job( + store=store, + key="K", + component_id="c", + config_id="cfg", + branch_id=5, + force_rerun=False, + create=create, + fetch=lambda jid: {}, + ) + assert job["id"] == "j1" and replayed is False + assert _entry(store, "K").job_id == "j1" + assert _entry(store, "K").branch_id == 5 + + @pytest.mark.parametrize("prior_status", ["processing", "waiting", "success", "warning"]) + def test_replay_returns_prior_when_not_failed( + self, store: JobIdempotencyStore, prior_status: str + ) -> None: + store.record("K", job_id="prior", component_id="c", config_id="cfg") + create = _Create({"id": "fresh", "status": "processing"}) + job, replayed = run_idempotent_job( + store=store, + key="K", + component_id="c", + config_id="cfg", + branch_id=None, + force_rerun=False, + create=create, + fetch=lambda jid: {"id": jid, "status": prior_status}, + ) + assert job["id"] == "prior" and replayed is True + assert create.calls == [] # no new job created + + @pytest.mark.parametrize("prior_status", ["error", "terminated", "cancelled"]) + def test_replay_reruns_when_prior_failed( + self, store: JobIdempotencyStore, prior_status: str + ) -> None: + store.record("K", job_id="prior", component_id="c", config_id="cfg") + create = _Create({"id": "fresh", "status": "processing"}) + job, replayed = run_idempotent_job( + store=store, + key="K", + component_id="c", + config_id="cfg", + branch_id=None, + force_rerun=False, + create=create, + fetch=lambda jid: {"id": jid, "status": prior_status}, + ) + assert job["id"] == "fresh" and replayed is False + assert _entry(store, "K").job_id == "fresh" # entry overwritten + + def test_force_rerun_ignores_non_failed_prior(self, store: JobIdempotencyStore) -> None: + store.record("K", job_id="prior", component_id="c", config_id="cfg") + create = _Create({"id": "fresh", "status": "processing"}) + fetched: list[str] = [] + + def fetch(jid: str) -> dict[str, Any]: + fetched.append(jid) + return {"id": jid, "status": "success"} + + job, replayed = run_idempotent_job( + store=store, + key="K", + component_id="c", + config_id="cfg", + branch_id=None, + force_rerun=True, + create=create, + fetch=fetch, + ) + assert job["id"] == "fresh" and replayed is False + assert fetched == [] # force_rerun short-circuits before fetching prior + + def test_collision_different_job_raises(self, store: JobIdempotencyStore) -> None: + store.record("K", job_id="prior", component_id="c", config_id="cfg") + with pytest.raises(KeboolaApiError) as exc: + run_idempotent_job( + store=store, + key="K", + component_id="OTHER", + config_id="cfg", + branch_id=None, + force_rerun=False, + create=_Create({"id": "x", "status": "processing"}), + fetch=lambda jid: {"id": jid, "status": "processing"}, + ) + assert exc.value.error_code == ErrorCode.INVALID_ARGUMENT + + def test_prior_404_reruns(self, store: JobIdempotencyStore) -> None: + store.record("K", job_id="gone", component_id="c", config_id="cfg") + + def fetch_404(jid: str) -> dict[str, Any]: + raise KeboolaApiError("nope", status_code=404, error_code=ErrorCode.NOT_FOUND) + + job, replayed = run_idempotent_job( + store=store, + key="K", + component_id="c", + config_id="cfg", + branch_id=None, + force_rerun=False, + create=_Create({"id": "fresh", "status": "processing"}), + fetch=fetch_404, + ) + assert job["id"] == "fresh" and replayed is False + + def test_prior_fetch_non_404_propagates(self, store: JobIdempotencyStore) -> None: + # A flaky fetch (e.g. 500) must NOT silently create a duplicate: propagate. + store.record("K", job_id="prior", component_id="c", config_id="cfg") + + def fetch_500(jid: str) -> dict[str, Any]: + raise KeboolaApiError("boom", status_code=500, error_code=ErrorCode.API_ERROR) + + with pytest.raises(KeboolaApiError) as exc: + run_idempotent_job( + store=store, + key="K", + component_id="c", + config_id="cfg", + branch_id=None, + force_rerun=False, + create=_Create({"id": "fresh", "status": "processing"}), + fetch=fetch_500, + ) + assert exc.value.status_code == 500 diff --git a/tests/test_lib.py b/tests/test_lib.py index ed6b8f87..c0345f68 100644 --- a/tests/test_lib.py +++ b/tests/test_lib.py @@ -17,6 +17,7 @@ ConfigDetailResult, FileEntry, Files, + JobIdempotencyStore, JobResult, QueryResult, UploadTableResult, @@ -406,3 +407,47 @@ def test_public_surface(self) -> None: "SyncPushResult", "ConfigDetailResult", } + + +class TestRunJobIdempotency: + """Facade run_job idempotency (issue #427).""" + + def test_key_without_store_raises(self, client: Client) -> None: + with pytest.raises(ValueError, match="idempotency_store"): + client.run_job("c", "cfg", idempotency_key="K") + + def test_replay_returns_prior_job(self, mock_kc: MagicMock, tmp_path: Path) -> None: + store = JobIdempotencyStore(tmp_path / "idem.json") + with patch("keboola_agent_cli.lib.KeboolaClient", return_value=mock_kc): + c = Client(url=STACK_URL, token=FAKE_TOKEN, idempotency_store=store) + mock_kc.create_job.return_value = {"id": "j1", "status": "processing"} + first = c.run_job("c", "cfg", idempotency_key="K") + assert first.id == "j1" and first.idempotent_replay is False + + mock_kc.get_job_detail.return_value = {"id": "j1", "status": "processing"} + second = c.run_job("c", "cfg", idempotency_key="K") + assert second.id == "j1" and second.idempotent_replay is True + mock_kc.create_job.assert_called_once() + + def test_per_call_store_override(self, mock_kc: MagicMock, tmp_path: Path) -> None: + store = JobIdempotencyStore(tmp_path / "idem.json") + with patch("keboola_agent_cli.lib.KeboolaClient", return_value=mock_kc): + c = Client(url=STACK_URL, token=FAKE_TOKEN) # no default store + mock_kc.create_job.return_value = {"id": "j1", "status": "processing"} + c.run_job("c", "cfg", idempotency_key="K", idempotency_store=store) + entry = store.lookup("K") + assert entry is not None and entry.job_id == "j1" + + def test_force_rerun_creates_fresh(self, mock_kc: MagicMock, tmp_path: Path) -> None: + store = JobIdempotencyStore(tmp_path / "idem.json") + with patch("keboola_agent_cli.lib.KeboolaClient", return_value=mock_kc): + c = Client(url=STACK_URL, token=FAKE_TOKEN, idempotency_store=store) + mock_kc.create_job.side_effect = [ + {"id": "j1", "status": "processing"}, + {"id": "j2", "status": "processing"}, + ] + mock_kc.get_job_detail.return_value = {"id": "j1", "status": "success"} + c.run_job("c", "cfg", idempotency_key="K") + second = c.run_job("c", "cfg", idempotency_key="K", force_rerun=True) + assert second.id == "j2" and second.idempotent_replay is False + assert mock_kc.create_job.call_count == 2 diff --git a/tests/test_services.py b/tests/test_services.py index 28e9d94a..fb76c71e 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -4569,3 +4569,107 @@ def factory(url: str, token: str) -> MagicMock: assert result[2]["status"] == "error" assert result[2]["error_code"] == "UNEXPECTED_ERROR" assert "Segfault simulation" in result[2]["error"] + + +class TestJobServiceRunJobIdempotency: + """JobService.run_job idempotency_key wiring (issue #427). + + Uses a real ConfigStore on a tmp config-dir, so the default dedup store + lands at ``/job_idempotency.json`` -- the same path the CLI uses. + """ + + @staticmethod + def _service(tmp_config_dir: Path, mock_client: MagicMock) -> JobService: + store = ConfigStore(config_dir=tmp_config_dir) + store.add_project( + "prod", + ProjectConfig( + stack_url="https://connection.keboola.com", + token="901-55555-fakeTestTokenDoNotUseXXXXXXXX", + project_name="Production", + project_id=1234, + ), + ) + return JobService(config_store=store, client_factory=lambda url, token: mock_client) + + def test_first_run_creates_and_persists_entry(self, tmp_config_dir: Path) -> None: + mock_client = MagicMock() + mock_client.create_job.return_value = {"id": "job-1", "status": "processing"} + service = self._service(tmp_config_dir, mock_client) + + result = service.run_job( + alias="prod", + component_id="keboola.ex-db-snowflake", + config_id="cfg-1", + no_variables=True, + idempotency_key="build-step-1", + ) + assert result["id"] == "job-1" + assert "idempotent_replay" not in result + mock_client.create_job.assert_called_once() + # Entry persisted to the config-dir store. + assert (tmp_config_dir / "job_idempotency.json").exists() + + def test_replay_returns_prior_without_recreating(self, tmp_config_dir: Path) -> None: + mock_client = MagicMock() + mock_client.create_job.return_value = {"id": "job-1", "status": "processing"} + mock_client.get_job_detail.return_value = {"id": "job-1", "status": "processing"} + service = self._service(tmp_config_dir, mock_client) + + first = service.run_job( + alias="prod", + component_id="keboola.ex-db-snowflake", + config_id="cfg-1", + no_variables=True, + idempotency_key="build-step-1", + ) + second = service.run_job( + alias="prod", + component_id="keboola.ex-db-snowflake", + config_id="cfg-1", + no_variables=True, + idempotency_key="build-step-1", + ) + assert first["id"] == "job-1" + assert second["id"] == "job-1" + assert second["idempotent_replay"] is True + # The side effect (create_job) fired exactly once across both calls. + mock_client.create_job.assert_called_once() + + def test_replay_reruns_when_prior_failed(self, tmp_config_dir: Path) -> None: + mock_client = MagicMock() + mock_client.create_job.side_effect = [ + {"id": "job-1", "status": "processing"}, + {"id": "job-2", "status": "processing"}, + ] + mock_client.get_job_detail.return_value = {"id": "job-1", "status": "error"} + service = self._service(tmp_config_dir, mock_client) + + service.run_job( + alias="prod", + component_id="c", + config_id="cfg", + no_variables=True, + idempotency_key="K", + ) + second = service.run_job( + alias="prod", + component_id="c", + config_id="cfg", + no_variables=True, + idempotency_key="K", + ) + assert second["id"] == "job-2" + assert "idempotent_replay" not in second + assert mock_client.create_job.call_count == 2 + + def test_no_key_never_dedups(self, tmp_config_dir: Path) -> None: + mock_client = MagicMock() + mock_client.create_job.return_value = {"id": "job-1", "status": "processing"} + service = self._service(tmp_config_dir, mock_client) + + service.run_job(alias="prod", component_id="c", config_id="cfg", no_variables=True) + service.run_job(alias="prod", component_id="c", config_id="cfg", no_variables=True) + assert mock_client.create_job.call_count == 2 + # No store file created when no key is used. + assert not (tmp_config_dir / "job_idempotency.json").exists()