SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix(simulation): resume live provider builds

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-26 15:46:46 +03:00
parent 5a9738d6aa
commit a98ad929d9
2 changed files with 147 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/simulation/projects.py
+61 -9
View File
@@ -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 не вернул артефакты сцены.")
tests/test_simulation_projects.py
+86
View File
@@ -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: