diff --git a/src/k1link/simulation/projects.py b/src/k1link/simulation/projects.py index 4db6700..a78807c 100644 --- a/src/k1link/simulation/projects.py +++ b/src/k1link/simulation/projects.py @@ -14,7 +14,7 @@ from collections import deque from collections.abc import Callable from contextlib import suppress from pathlib import Path -from typing import Any, Final, TypeVar +from typing import Any, Final from urllib.parse import quote from uuid import uuid4 @@ -55,7 +55,6 @@ PROVIDER_JOB_STATES: Final = { } PROVIDER_POLL_TIMEOUT_SECONDS: Final = 2 * 60 * 60 + 5 * 60 PROVIDER_UNAVAILABLE_RETRY_LIMIT: Final = 6 -_T = TypeVar("_T") class SimulationProjectError(RuntimeError): @@ -722,6 +721,8 @@ class SimulationProjectService: ) except _SimulationProcessingCancelled: pass + except GaussianPipelineUnavailableError: + self._requeue_provider_unavailable(project_id) except (GaussianPipelineGatewayError, SimulationProjectError, OSError) as exc: with suppress(SimulationProjectError): self.store.fail(project_id, str(exc)) @@ -731,6 +732,21 @@ class SimulationProjectService: if provider is not None: provider.close() + def _requeue_provider_unavailable(self, project_id: str) -> None: + """Keep a retained source pending while its worker transport is unavailable.""" + try: + self.store.update_processing(project_id, status="queued") + except SimulationProjectError: + return + with self._condition: + cancel = self._cancel_events.get(project_id) + if cancel is not None and cancel.is_set(): + return + if self._active_project_id == project_id and project_id not in self._queued_ids: + self._queue.append(project_id) + self._queued_ids.add(project_id) + self._condition.notify_all() + def delete(self, project_id: str) -> None: project = self.store.get(project_id) with self._condition: @@ -794,7 +810,7 @@ class SimulationProjectService: raise _SimulationProcessingCancelled(project_id) -def _retry_provider_unavailable(operation: Callable[[], _T]) -> _T: +def _retry_provider_unavailable[T](operation: Callable[[], T]) -> T: delay_seconds = 1.0 for attempt in range(PROVIDER_UNAVAILABLE_RETRY_LIMIT): try: diff --git a/tests/test_simulation_projects.py b/tests/test_simulation_projects.py index cb33426..2f746f2 100644 --- a/tests/test_simulation_projects.py +++ b/tests/test_simulation_projects.py @@ -3,6 +3,7 @@ from __future__ import annotations import json from pathlib import Path from threading import Event +from time import monotonic, sleep from typing import Any import pytest @@ -444,6 +445,55 @@ def test_service_queue_processes_projects_strictly_one_at_a_time(tmp_path: Path) assert order == [projects[0]["project_id"], projects[1]["project_id"]] +def test_service_keeps_retained_source_queued_across_temporary_worker_outage( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + store = SimulationProjectStore(tmp_path) + project = store.create( + name="Reconnect without browser reupload", + scene_type="outdoor", + source_kind="folder", + files=_folder_files(), + ) + _upload_all(store, project) + store.begin_build(project["project_id"]) + + class _ReconnectProvider(_ReadyProvider): + def __init__(self) -> None: + super().__init__() + self.capability_calls = 0 + + def capabilities(self) -> dict[str, object]: + self.capability_calls += 1 + if self.capability_calls == 1: + raise GaussianPipelineUnavailableError("temporary tunnel failure") + return super().capabilities() + + provider = _ReconnectProvider() + monkeypatch.setattr( + "k1link.simulation.projects.PROVIDER_UNAVAILABLE_RETRY_LIMIT", + 1, + ) + service = SimulationProjectService( + store, + provider_factory=lambda: provider, + ) # type: ignore[arg-type] + + service.enqueue(project["project_id"]) + + deadline = monotonic() + 1.0 + while store.get(project["project_id"])["status"] != "ready" and monotonic() < deadline: + sleep(0.01) + recovered = store.get(project["project_id"]) + assert recovered["status"] == "ready" + assert recovered["error"] is None + assert recovered["source"]["uploaded_byte_length"] == recovered["source"]["total_byte_length"] + assert provider.capability_calls == 2 + assert provider.upload_calls == 1 + assert provider.submit_calls == 1 + + def test_service_deletes_a_queued_project_before_worker_submission(tmp_path: Path) -> None: store = SimulationProjectStore(tmp_path) project = store.create(