diff --git a/src/k1link/simulation/projects.py b/src/k1link/simulation/projects.py index 51dfb43..f25c07b 100644 --- a/src/k1link/simulation/projects.py +++ b/src/k1link/simulation/projects.py @@ -9,9 +9,10 @@ import re import shutil import threading import time +from collections.abc import Callable from contextlib import suppress from pathlib import Path -from typing import Any, Final +from typing import Any, Final, TypeVar from urllib.parse import quote from uuid import uuid4 @@ -20,6 +21,7 @@ from k1link.simulation.gaussian_pipeline_gateway import ( BUILD_REQUEST_SCHEMA, GaussianPipelineGateway, GaussianPipelineGatewayError, + GaussianPipelineUnavailableError, configured_gaussian_pipeline_gateway, discover_gaussian_source_bundle, ) @@ -46,6 +48,8 @@ PROVIDER_JOB_STATES: Final = { "failed", } PROVIDER_POLL_TIMEOUT_SECONDS: Final = 2 * 60 * 60 + 5 * 60 +PROVIDER_UNAVAILABLE_RETRY_LIMIT: Final = 6 +_T = TypeVar("_T") class SimulationProjectError(RuntimeError): @@ -301,6 +305,7 @@ class SimulationProjectStore: document["provider"]["progress"] = progress if bundle_sha256 is not None: document["source"]["bundle_sha256"] = bundle_sha256 + document["error"] = None document["updated_at_utc"] = utc_now_iso() self._write(document) return document @@ -437,6 +442,35 @@ class SimulationProjectService: def begin_build(self, project_id: str) -> dict[str, Any]: project = self.store.get(project_id) + if project.get("status") == "failed": + job_id = project["provider"].get("job_id") + provider_state = project["provider"].get("state") + if ( + isinstance(job_id, str) + and isinstance(provider_state, str) + and provider_state in PROVIDER_JOB_STATES - {"failed"} + ): + provider = self.provider_factory() + if provider is None: + raise SimulationProjectConflictError( + "Gaussian Pipeline недоступен для продолжения сборки." + ) + try: + job = _retry_provider_unavailable(lambda: provider.get_job(job_id)) + finally: + provider.close() + current_state = job.get("state") + if ( + isinstance(current_state, str) + and current_state in PROVIDER_JOB_STATES - {"failed"} + ): + return self.store.update_processing( + project_id, + status="processing", + provider_job_id=job_id, + provider_state=current_state, + progress=job.get("progress"), + ) if project.get("status") in {"failed", "ready"}: job_id = project["provider"].get("job_id") if isinstance(job_id, str): @@ -446,7 +480,7 @@ class SimulationProjectService: "Gaussian Pipeline недоступен для повторной сборки." ) try: - provider.delete_job(job_id) + _retry_provider_unavailable(lambda: provider.delete_job(job_id)) finally: provider.close() return self.store.begin_build(project_id) @@ -458,7 +492,7 @@ class SimulationProjectService: provider = self.provider_factory() if provider is None: raise SimulationProjectError("Gaussian Pipeline не настроен.") - provider.capabilities() + _retry_provider_unavailable(provider.capabilities) existing_job_id = project["provider"].get("job_id") if isinstance(existing_job_id, str): job_id = existing_job_id @@ -511,7 +545,7 @@ class SimulationProjectService: raise SimulationProjectError( "Gaussian Pipeline превысил лимит ожидания сборки." ) - job = provider.get_job(job_id) + job = _retry_provider_unavailable(lambda: provider.get_job(job_id)) state = job.get("state") if not isinstance(state, str) or state not in PROVIDER_JOB_STATES: raise SimulationProjectError( @@ -537,14 +571,19 @@ class SimulationProjectService: status="importing", provider_state="ready", ) - result = provider.get_result(job_id) + result = _retry_provider_unavailable(lambda: provider.get_result(job_id)) artifacts = _artifact_descriptors(result.get("artifacts")) artifacts_root = self.store.artifacts_root(project_id) for descriptor in artifacts: - provider.download_artifact( - job_id, - descriptor, - _confined_path(artifacts_root, str(descriptor["logical_path"])), + _retry_provider_unavailable( + lambda descriptor=descriptor: provider.download_artifact( + job_id, + descriptor, + _confined_path( + artifacts_root, + str(descriptor["logical_path"]), + ), + ) ) world_manifest = _world_manifest(project_id, artifacts) self.store.complete( @@ -576,6 +615,19 @@ class SimulationProjectService: self.store.delete(project_id) +def _retry_provider_unavailable(operation: Callable[[], _T]) -> _T: + delay_seconds = 1.0 + for attempt in range(PROVIDER_UNAVAILABLE_RETRY_LIMIT): + try: + return operation() + except GaussianPipelineUnavailableError: + if attempt + 1 >= PROVIDER_UNAVAILABLE_RETRY_LIMIT: + raise + time.sleep(delay_seconds) + delay_seconds = min(delay_seconds * 2.0, 10.0) + raise AssertionError("provider retry loop exhausted without returning or raising") + + def _artifact_descriptors(value: object) -> list[dict[str, Any]]: if not isinstance(value, list) or not value: raise SimulationProjectError("Gaussian Pipeline не вернул артефакты сцены.") diff --git a/tests/test_simulation_projects.py b/tests/test_simulation_projects.py index 9375f97..81f9f55 100644 --- a/tests/test_simulation_projects.py +++ b/tests/test_simulation_projects.py @@ -8,6 +8,7 @@ from fastapi import FastAPI from fastapi.testclient import TestClient from k1link.simulation.gaussian_pipeline_gateway import ( + GaussianPipelineUnavailableError, GaussianSourceBundleUpload, GaussianSourceMemberUpload, ) @@ -281,6 +282,91 @@ def test_failed_project_retries_from_retained_source_and_releases_old_job(tmp_pa assert queued["source"]["uploaded_byte_length"] == queued["source"]["total_byte_length"] +def test_failed_local_project_reattaches_to_live_provider_job_without_rebuild( + tmp_path: Path, +) -> None: + store = SimulationProjectStore(tmp_path) + project = store.create( + name="Reconnect scene", + scene_type="outdoor", + source_kind="folder", + files=_folder_files(), + ) + _upload_all(store, project) + store.begin_build(project["project_id"]) + store.update_processing( + project["project_id"], + status="processing", + provider_job_id="gsp-20260826000000-deadbeef", + provider_state="building_streamed_sog", + progress={"completed_steps": 3, "total_steps": 5}, + bundle_sha256="d" * 64, + ) + store.fail(project["project_id"], "temporary provider transport failure") + provider = _ReadyProvider() + service = SimulationProjectService( + store, + provider_factory=lambda: provider, + ) # type: ignore[arg-type] + + resumed = service.begin_build(project["project_id"]) + + assert resumed["status"] == "processing" + assert resumed["error"] is None + assert resumed["provider"]["job_id"] == "gsp-20260826000000-deadbeef" + assert resumed["provider"]["state"] == "ready" + assert provider.deleted == [] + assert provider.upload_calls == 0 + assert provider.submit_calls == 0 + + +def test_live_provider_reattach_retries_temporary_unavailability( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + store = SimulationProjectStore(tmp_path) + project = store.create( + name="Flaky tunnel scene", + scene_type="outdoor", + source_kind="folder", + files=_folder_files(), + ) + _upload_all(store, project) + store.begin_build(project["project_id"]) + store.update_processing( + project["project_id"], + status="processing", + provider_job_id="gsp-20260826000000-deadbeef", + provider_state="building_collision", + bundle_sha256="d" * 64, + ) + store.fail(project["project_id"], "temporary provider transport failure") + + class _FlakyProvider(_ReadyProvider): + def __init__(self) -> None: + super().__init__() + self.get_job_calls = 0 + + def get_job(self, job_id: str) -> dict[str, object]: + self.get_job_calls += 1 + if self.get_job_calls < 3: + raise GaussianPipelineUnavailableError("temporary tunnel failure") + return super().get_job(job_id) + + provider = _FlakyProvider() + monkeypatch.setattr("k1link.simulation.projects.time.sleep", lambda _delay: None) + service = SimulationProjectService( + store, + provider_factory=lambda: provider, + ) # type: ignore[arg-type] + + resumed = service.begin_build(project["project_id"]) + + assert resumed["status"] == "processing" + assert provider.get_job_calls == 3 + assert provider.deleted == [] + + def test_ready_project_rebuilds_from_retained_source_and_preserves_viewer_settings( tmp_path: Path, ) -> None: