From e5cf7d60e27c890ca6e4fbf0e2db69adf68d906c Mon Sep 17 00:00:00 2001 From: Volv G Date: Tue, 7 Jul 2026 19:35:33 -0700 Subject: [PATCH] feat(pipeline-run): configure submit recovery attempts Assisted-By: devx/12bfa114-4ee7-4929-a37c-98018f6b0f41 --- .../tangle-cli/src/tangle_cli/__init__.py | 2 +- .../src/tangle_cli/pipeline_run_manager.py | 31 +++++++- .../src/tangle_cli/pipeline_runner.py | 2 + .../src/tangle_cli/pipeline_runs_cli.py | 18 ++++- pyproject.toml | 2 +- tests/test_packaging.py | 3 +- tests/test_pipeline_runs_cli.py | 72 ++++++++++++++++--- uv.lock | 4 +- 8 files changed, 114 insertions(+), 20 deletions(-) diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index a1cb5f7..dcaea45 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.0" + __version__ = "0.1.1" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py index 56b47ed..1736d76 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py @@ -37,7 +37,8 @@ _EXECUTION_STATE_TIMINGS_METADATA_KEY = "execution_state_timings" _EXECUTION_STATE_TIMING_MONOTONIC_METADATA_KEY = "_execution_state_timing_monotonic" _SUBMISSION_ID_ANNOTATION_KEY = "tangle-cli/submission-id" -_SUBMIT_RECOVERY_BACKOFF_SECONDS = (0.5, 1.0, 2.0) +_SUBMIT_RECOVERY_BACKOFF_SECONDS = (0.5, 1.0, 2.0, 4.0, 8.0, 16.0) +_DEFAULT_SUBMIT_RECOVERY_ATTEMPTS = 2 class PipelineRunError(RuntimeError): @@ -1528,6 +1529,11 @@ def _submission_id_from_body(body: Mapping[str, Any]) -> str | None: submission_id = annotations.get(_SUBMISSION_ID_ANNOTATION_KEY) return str(submission_id) if submission_id else None + @staticmethod + def _submit_recovery_backoff_seconds(submit_recovery_attempts: int) -> tuple[float, ...]: + attempt_count = max(0, min(int(submit_recovery_attempts), len(_SUBMIT_RECOVERY_BACKOFF_SECONDS))) + return _SUBMIT_RECOVERY_BACKOFF_SECONDS[:attempt_count] + def _submitted_runs_for_submission_id(self, submission_id: str) -> list[dict[str, Any]]: query = { "and": [ @@ -1553,11 +1559,21 @@ def _recover_submitted_run_after_submit_error( self, *, submission_id: str | None, + submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, ) -> dict[str, Any] | None: if not submission_id: return None - total_lookup_attempts = len(_SUBMIT_RECOVERY_BACKOFF_SECONDS) - for lookup_attempt, delay_seconds in enumerate(_SUBMIT_RECOVERY_BACKOFF_SECONDS, start=1): + backoff_seconds = self._submit_recovery_backoff_seconds(submit_recovery_attempts) + total_lookup_attempts = len(backoff_seconds) + if total_lookup_attempts == 0: + self.logger.warn( + "Submit recovery lookup disabled " + f"({_SUBMISSION_ID_ANNOTATION_KEY}={submission_id}, " + f"submit_recovery_attempts={submit_recovery_attempts}); " + "resubmitting the same frozen body with preserved inputs." + ) + return None + for lookup_attempt, delay_seconds in enumerate(backoff_seconds, start=1): self.logger.info( "Waiting " f"{delay_seconds:g}s before checking whether failed submit already created a pipeline run " @@ -1652,6 +1668,7 @@ def _run_body_factory( timeout_clock: str = "monotonic", exit_on_first_failure: bool = False, metadata: dict[str, Any] | None = None, + submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, metadata_factory: Callable[ [int, PipelineRunContext | None, Exception | None], dict[str, Any] ] | None = None, @@ -1728,6 +1745,7 @@ def _run_body_factory( if reused_after_submit_failure: recovered_response = self._recover_submitted_run_after_submit_error( submission_id=self._submission_id_from_body(body), + submit_recovery_attempts=submit_recovery_attempts, ) if recovered_response is not None: response = self._adopt_submitted_run( @@ -1759,6 +1777,7 @@ def _run_body_factory( ) recovered_response = self._recover_submitted_run_after_submit_error( submission_id=submission_id_for_recovery, + submit_recovery_attempts=submit_recovery_attempts, ) if recovered_response is None: self.hooks.on_submit_error(submit_exc, context=context) @@ -1839,6 +1858,7 @@ def run_prepared_body( timeout_clock: str = "monotonic", exit_on_first_failure: bool = False, metadata: dict[str, Any] | None = None, + submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, ) -> dict[str, Any]: """Submit/wait/retry an already prepared submit body. @@ -1867,6 +1887,7 @@ def body_factory( timeout_clock=timeout_clock, exit_on_first_failure=exit_on_first_failure, metadata=metadata, + submit_recovery_attempts=submit_recovery_attempts, ) def run_pipeline_spec( @@ -1887,6 +1908,7 @@ def run_pipeline_spec( timeout_clock: str = "monotonic", exit_on_first_failure: bool = False, metadata: dict[str, Any] | None = None, + submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, ) -> dict[str, Any]: """Submit/wait/retry an already hydrated/validated in-memory spec.""" @@ -1916,6 +1938,7 @@ def body_factory( timeout_clock=timeout_clock, exit_on_first_failure=exit_on_first_failure, metadata=metadata, + submit_recovery_attempts=submit_recovery_attempts, ) def run_pipeline( @@ -1936,6 +1959,7 @@ def run_pipeline( timeout_clock: str = "monotonic", exit_on_first_failure: bool = False, metadata: dict[str, Any] | None = None, + submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, ) -> dict[str, Any]: """Submit (and optionally wait for) a pipeline with lifecycle hooks. @@ -1970,6 +1994,7 @@ def body_factory( timeout_clock=timeout_clock, exit_on_first_failure=exit_on_first_failure, metadata=metadata, + submit_recovery_attempts=submit_recovery_attempts, ) diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_runner.py b/packages/tangle-cli/src/tangle_cli/pipeline_runner.py index 2b50d39..d0580f8 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_runner.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_runner.py @@ -507,6 +507,7 @@ def run_pipeline( open_browser: bool = False, include_next_steps: bool = False, metadata: dict[str, Any] | None = None, + submit_recovery_attempts: int = 2, ) -> dict[str, Any]: """Run a pipeline path through generic preparation + lifecycle hooks. @@ -603,6 +604,7 @@ def metadata_factory( timeout_clock=timeout_clock, exit_on_first_failure=exit_on_first_failure, metadata_factory=metadata_factory, + submit_recovery_attempts=submit_recovery_attempts, ) context = result.get("context") attempt = context.attempt if isinstance(context, PipelineRunContext) else max(preparations) diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py index 8c67c84..2da0eb5 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py @@ -28,6 +28,7 @@ from .logger import Logger, logger_for_log_type from .pipeline_run_annotations import AnnotationManager from .pipeline_run_manager import ( + _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, PipelineRunError, PipelineRunHooks, PipelineRunManager, @@ -166,6 +167,15 @@ def pipeline_runs_submit( auth_header: AuthHeaderOption = None, header: HeaderOption = None, config: ConfigOption = None, + submit_recovery_attempts: Annotated[ + int, + Parameter( + help=( + "Number of post-failed-submit recovery lookups before resubmitting; " + "higher values wait longer for delayed run registration." + ) + ), + ] = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS, log_type: LogTypeOption = "console", ) -> None: """Hydrate and submit a local pipeline YAML file as a run.""" @@ -181,6 +191,7 @@ def pipeline_runs_submit( "run_as": (run_as, None), "trusted_source": (trusted_source, None), "trusted_hydration_cli": ("trusted_hydration_cli", trusted_hydration, None, False), + "submit_recovery_attempts": (submit_recovery_attempts, _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS), "log_type": (log_type, "console"), **api_arg_specs(base_url=base_url, token=token, auth_header=auth_header, header=header), } @@ -194,7 +205,12 @@ def action(manager: PipelineRunManager, args: ArgsContainer) -> dict[str, Any]: } if args.dry_run: return manager.build_submit_body(args.pipeline_path, **kwargs) - return manager.submit_pipeline(args.pipeline_path, **kwargs) + result = manager.run_pipeline( + args.pipeline_path, + **kwargs, + submit_recovery_attempts=args.submit_recovery_attempts, + ) + return result["response"] _run_manager_action(config, base_url, specs, action) diff --git a/pyproject.toml b/pyproject.toml index ce556a6..b04e037 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.0" +version = "0.1.1" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 9e0e4d2..6aa2593 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -14,7 +14,6 @@ from tangle_cli.openapi import codegen - _REPO_ROOT = Path(__file__).resolve().parents[1] @@ -179,7 +178,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")] assert not any(name.startswith("tangle_api/") for name in names) assert "tangle_cli/openapi/openapi.json" not in names - assert "Version: 0.1.0" in metadata + assert "Version: 0.1.1" in metadata assert "Requires-Dist: tangle-api==0.1.0" in requires_dist assert not any("extra == 'native'" in line for line in requires_dist) assert "Provides-Extra: native" in metadata diff --git a/tests/test_pipeline_runs_cli.py b/tests/test_pipeline_runs_cli.py index 7148023..f20249f 100644 --- a/tests/test_pipeline_runs_cli.py +++ b/tests/test_pipeline_runs_cli.py @@ -135,7 +135,9 @@ def test_pipeline_runs_help_exposes_run_commands_not_local_pipeline_commands(cap assert "diagram" not in output run_app(app, ["sdk", "pipeline-runs", "submit", "--help"]) - assert "--log-type" in capsys.readouterr().out + submit_help = capsys.readouterr().out + assert "--log-type" in submit_help + assert "--submit-recovery-attempts" in submit_help def test_pipeline_runs_submit_builds_create_payload(monkeypatch, tmp_path: Path, capsys): @@ -161,12 +163,62 @@ def test_pipeline_runs_submit_builds_create_payload(monkeypatch, tmp_path: Path, result = json.loads(capsys.readouterr().out) assert result == {"id": "run-1", "root_execution_id": "exec-1"} - assert fake_client.created[0]["annotations"] == {"team": "oss"} + assert fake_client.created[0]["annotations"]["team"] == "oss" + assert fake_client.created[0]["annotations"]["tangle-cli/submission-id"] root_task = fake_client.created[0]["root_task"] assert root_task["componentRef"]["spec"]["name"] == "Demo Pipeline" assert root_task["arguments"] == {"query": "default", "required": "value"} +def test_pipeline_runs_submit_uses_default_submit_recovery_attempts(monkeypatch, tmp_path: Path, capsys): + pipeline_path = _write_pipeline(tmp_path / "pipeline.yaml") + captured: dict[str, Any] = {} + + def fake_run_pipeline(self, pipeline_path, **kwargs): + del self, pipeline_path + captured.update(kwargs) + return {"response": {"id": "run-1"}} + + monkeypatch.setattr(PipelineRunManager, "run_pipeline", fake_run_pipeline) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FakeClient()) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit", str(pipeline_path), "--no-hydrate"]) + + assert json.loads(capsys.readouterr().out) == {"id": "run-1"} + assert captured["submit_recovery_attempts"] == pipeline_run_manager._DEFAULT_SUBMIT_RECOVERY_ATTEMPTS + + +def test_pipeline_runs_submit_accepts_submit_recovery_attempts(monkeypatch, tmp_path: Path, capsys): + pipeline_path = _write_pipeline(tmp_path / "pipeline.yaml") + captured: dict[str, Any] = {} + + def fake_run_pipeline(self, pipeline_path, **kwargs): + del self, pipeline_path + captured.update(kwargs) + return {"response": {"id": "run-1"}} + + monkeypatch.setattr(PipelineRunManager, "run_pipeline", fake_run_pipeline) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FakeClient()) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit", + str(pipeline_path), + "--no-hydrate", + "--submit-recovery-attempts", + "6", + ], + ) + + assert json.loads(capsys.readouterr().out) == {"id": "run-1"} + assert captured["submit_recovery_attempts"] == 6 + + def test_pipeline_runs_submit_accepts_export_config_args_and_hydrate(monkeypatch, tmp_path: Path): pipeline_path = _write_pipeline(tmp_path / "pipeline.yaml") config = tmp_path / "pipeline.config.yaml" @@ -1752,7 +1804,7 @@ def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]: def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: self.list_calls.append(kwargs) - if len(self.list_calls) <= 2: + if len(self.list_calls) < len(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS): return {"pipeline_runs": [], "next_page_token": None} return { "pipeline_runs": [{"id": "run-created", "root_execution_id": "exec-created"}], @@ -1763,13 +1815,13 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: manager = PipelineRunManager(client=client) body = {"root_task": {"componentRef": {"spec": {"name": "delayed-recovery"}}}} - result = manager.run_prepared_body(body) + result = manager.run_prepared_body(body, submit_recovery_attempts=6) assert result["response"]["id"] == "run-created" assert result["context"].metadata["recovered_after_submit_error"] is True assert len(client.created) == 1 - assert len(client.list_calls) == 3 - assert sleeps == [0.5, 1.0, 2.0] + assert len(client.list_calls) == len(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS) + assert sleeps == list(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS) def test_pipeline_runs_submit_failure_recovery_refuses_ambiguous_matches(monkeypatch) -> None: @@ -1869,8 +1921,8 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: assert client.created[0]["annotations"]["tangle-cli/submission-id"] == client.created[1]["annotations"][ "tangle-cli/submission-id" ] - assert len(client.list_calls) == 2 * len(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS) - expected_sleeps = list(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS) + expected_sleeps = list(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS[:2]) + assert len(client.list_calls) == 2 * len(expected_sleeps) assert sleeps == expected_sleeps + expected_sleeps @@ -1927,7 +1979,7 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: client = RecoverOnRetryClient() manager = PipelineRunner(client=client, hooks=hooks) - result = manager.run_pipeline(pipeline_path, hydrate=False, max_attempts=2) + result = manager.run_pipeline(pipeline_path, hydrate=False, max_attempts=2, submit_recovery_attempts=6) assert result["response"]["id"] == "run-created" assert result["context"].attempt == 2 @@ -1935,7 +1987,7 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]: assert hooks.prepare_run_arguments_calls == 1 assert len(client.created) == 1 assert len(client.list_calls) == len(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS) + 1 - assert sleeps == [0.5, 1.0, 2.0, 0.5] + assert sleeps == [*pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS, 0.5] assert events == [("before_retry", 2, None), ("after_retry_submit", 2, "run-created")] diff --git a/uv.lock b/uv.lock index ca1d97b..e622122 100644 --- a/uv.lock +++ b/uv.lock @@ -8,7 +8,7 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-06-29T15:21:31.720493Z" +exclude-newer = "2026-07-01T02:35:01.441363Z" exclude-newer-span = "P7D" [manifest] @@ -1786,7 +1786,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.0" +version = "0.1.1" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },