diff --git a/src/k1link/simulation/projects.py b/src/k1link/simulation/projects.py index a78807c..a50ab2a 100644 --- a/src/k1link/simulation/projects.py +++ b/src/k1link/simulation/projects.py @@ -55,6 +55,7 @@ PROVIDER_JOB_STATES: Final = { } PROVIDER_POLL_TIMEOUT_SECONDS: Final = 2 * 60 * 60 + 5 * 60 PROVIDER_UNAVAILABLE_RETRY_LIMIT: Final = 6 +IMPORT_DISK_RESERVE_BYTES: Final = 512 * 1024 * 1024 class SimulationProjectError(RuntimeError): @@ -700,6 +701,7 @@ class SimulationProjectService: self._raise_if_cancelled(project_id) artifacts = _artifact_descriptors(result.get("artifacts")) artifacts_root = self.store.artifacts_root(project_id) + self._ensure_import_capacity(project_id, artifacts, artifacts_root) for descriptor in artifacts: self._raise_if_cancelled(project_id) _retry_provider_unavailable( @@ -747,6 +749,55 @@ class SimulationProjectService: self._queued_ids.add(project_id) self._condition.notify_all() + def _ensure_import_capacity( + self, + project_id: str, + artifacts: list[dict[str, Any]], + artifacts_root: Path, + ) -> None: + required_bytes = _remaining_import_bytes(artifacts_root, artifacts) + available_bytes = shutil.disk_usage(artifacts_root).free + if available_bytes >= required_bytes: + return + project = self.store.get(project_id) + source = project.get("source") + provider = project.get("provider") + if ( + isinstance(source, dict) + and source.get("kind") == "archive" + and isinstance(provider, dict) + and isinstance(provider.get("job_id"), str) + ): + source_files = source.get("files") + if isinstance(source_files, list) and len(source_files) == 1: + source_file = source_files[0] + if isinstance(source_file, dict): + logical_path = source_file.get("logical_path") + byte_length = source_file.get("byte_length") + if isinstance(logical_path, str) and isinstance(byte_length, int): + source_path = _confined_path( + self.store.source_root(project_id), + logical_path, + ) + try: + source_stat = source_path.stat() + except OSError: + source_stat = None + if ( + source_stat is not None + and source_path.is_file() + and not source_path.is_symlink() + and source_stat.st_size == byte_length + ): + source_path.unlink() + available_bytes = shutil.disk_usage(artifacts_root).free + if available_bytes < required_bytes: + raise SimulationProjectError( + "Недостаточно места для импорта Gaussian-мира: " + f"нужно {_human_bytes(required_bytes)}, " + f"доступно {_human_bytes(available_bytes)}." + ) + def delete(self, project_id: str) -> None: project = self.store.get(project_id) with self._condition: @@ -857,6 +908,36 @@ def _artifact_descriptors(value: object) -> list[dict[str, Any]]: return descriptors +def _remaining_import_bytes( + artifacts_root: Path, + artifacts: list[dict[str, Any]], +) -> int: + missing_bytes = 0 + replacement_scratch_bytes = 0 + for descriptor in artifacts: + logical_path = str(descriptor["logical_path"]) + expected_bytes = int(descriptor["byte_length"]) + target = _confined_path(artifacts_root, logical_path) + try: + current = target.stat() + except OSError: + current = None + if ( + current is not None + and target.is_file() + and not target.is_symlink() + and current.st_size == expected_bytes + ): + replacement_scratch_bytes = max(replacement_scratch_bytes, expected_bytes) + else: + missing_bytes += expected_bytes + return missing_bytes + replacement_scratch_bytes + IMPORT_DISK_RESERVE_BYTES + + +def _human_bytes(value: int) -> str: + return f"{value / (1024**3):.2f} ГиБ" + + def _world_manifest(project_id: str, artifacts: list[dict[str, Any]]) -> dict[str, Any]: def url_for(role: str) -> str | None: descriptor = next((item for item in artifacts if item["role"] == role), None) diff --git a/tests/test_simulation_projects.py b/tests/test_simulation_projects.py index 2f746f2..a9be5a4 100644 --- a/tests/test_simulation_projects.py +++ b/tests/test_simulation_projects.py @@ -611,6 +611,63 @@ 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_import_evicts_only_worker_backed_archive_staging_when_disk_is_low( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + store = SimulationProjectStore(tmp_path) + project = store.create( + name="Worker-backed archive", + scene_type="outdoor", + source_kind="archive", + files=[{"logical_path": "scene.rar", "byte_length": 6}], + ) + source_file = project["source"]["files"][0] + store.append_upload( + project["project_id"], + source_file["file_id"], + offset=0, + payload=b"source", + ) + store.begin_build(project["project_id"]) + store.update_processing( + project["project_id"], + status="processing", + provider_job_id="gsp-20260826000000-deadbeef", + provider_state="ready", + ) + service = SimulationProjectService(store, provider_factory=lambda: None) + source_path = store.source_root(project["project_id"]) / "scene.rar" + + class _DiskUsage: + def __init__(self, free: int) -> None: + self.free = free + + monkeypatch.setattr( + "k1link.simulation.projects.shutil.disk_usage", + lambda _path: _DiskUsage(0 if source_path.exists() else 10 * 1024**3), + ) + artifacts = [ + { + "role": "preview", + "logical_path": "preview.sog", + "media_type": "application/octet-stream", + "sha256": "a" * 64, + "byte_length": 1024, + } + ] + + service._ensure_import_capacity( + project["project_id"], + artifacts, + store.artifacts_root(project["project_id"]), + ) + + assert not source_path.exists() + retained_metadata = store.get(project["project_id"])["source"] + assert retained_metadata["uploaded_byte_length"] == retained_metadata["total_byte_length"] + + def test_failed_local_project_reattaches_to_live_provider_job_without_rebuild( tmp_path: Path, ) -> None: