fix(simulation): retry builds across worker tunnel outages

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:

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(