fix(simulation): reserve disk space before artifact import
This commit is contained in:
@@ -55,6 +55,7 @@ PROVIDER_JOB_STATES: Final = {
|
|||||||
}
|
}
|
||||||
PROVIDER_POLL_TIMEOUT_SECONDS: Final = 2 * 60 * 60 + 5 * 60
|
PROVIDER_POLL_TIMEOUT_SECONDS: Final = 2 * 60 * 60 + 5 * 60
|
||||||
PROVIDER_UNAVAILABLE_RETRY_LIMIT: Final = 6
|
PROVIDER_UNAVAILABLE_RETRY_LIMIT: Final = 6
|
||||||
|
IMPORT_DISK_RESERVE_BYTES: Final = 512 * 1024 * 1024
|
||||||
|
|
||||||
|
|
||||||
class SimulationProjectError(RuntimeError):
|
class SimulationProjectError(RuntimeError):
|
||||||
@@ -700,6 +701,7 @@ class SimulationProjectService:
|
|||||||
self._raise_if_cancelled(project_id)
|
self._raise_if_cancelled(project_id)
|
||||||
artifacts = _artifact_descriptors(result.get("artifacts"))
|
artifacts = _artifact_descriptors(result.get("artifacts"))
|
||||||
artifacts_root = self.store.artifacts_root(project_id)
|
artifacts_root = self.store.artifacts_root(project_id)
|
||||||
|
self._ensure_import_capacity(project_id, artifacts, artifacts_root)
|
||||||
for descriptor in artifacts:
|
for descriptor in artifacts:
|
||||||
self._raise_if_cancelled(project_id)
|
self._raise_if_cancelled(project_id)
|
||||||
_retry_provider_unavailable(
|
_retry_provider_unavailable(
|
||||||
@@ -747,6 +749,55 @@ class SimulationProjectService:
|
|||||||
self._queued_ids.add(project_id)
|
self._queued_ids.add(project_id)
|
||||||
self._condition.notify_all()
|
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:
|
def delete(self, project_id: str) -> None:
|
||||||
project = self.store.get(project_id)
|
project = self.store.get(project_id)
|
||||||
with self._condition:
|
with self._condition:
|
||||||
@@ -857,6 +908,36 @@ def _artifact_descriptors(value: object) -> list[dict[str, Any]]:
|
|||||||
return descriptors
|
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 _world_manifest(project_id: str, artifacts: list[dict[str, Any]]) -> dict[str, Any]:
|
||||||
def url_for(role: str) -> str | None:
|
def url_for(role: str) -> str | None:
|
||||||
descriptor = next((item for item in artifacts if item["role"] == role), None)
|
descriptor = next((item for item in artifacts if item["role"] == role), None)
|
||||||
|
|||||||
@@ -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"]
|
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(
|
def test_failed_local_project_reattaches_to_live_provider_job_without_rebuild(
|
||||||
tmp_path: Path,
|
tmp_path: Path,
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|||||||
Reference in New Issue
Block a user