diff --git a/docs/design/module/host_weight_runtime.md b/docs/design/module/host_weight_runtime.md index ac92711fc..8a159bb00 100644 --- a/docs/design/module/host_weight_runtime.md +++ b/docs/design/module/host_weight_runtime.md @@ -299,6 +299,43 @@ Explicit cleanup uses the locked move and retries exact-key `.cleanup.*` tombstones left by an interrupted removal; a later build performs the same tombstone reconciliation before producing replacement content. +Failure-quarantined entries remain until explicitly selected for removal; +ordinary `cleanup(identity)` does not remove that diagnostic history. +`FilesystemHostWeightStore.cleanup_quarantined(storage_name)` accepts one +quarantined inventory basename and preserves the current artifact, deny marker, +and other failure-quarantined entries. It takes the same nonblocking build and +exclusive artifact locks, so an active builder or lease returns a typed refusal. +Malformed names are rejected before mutation, and symlinks are not followed. + +Selected entries move through the existing `.cleanup.*` tombstone protocol. +Retrying the original name reconciles pending cleanup tombstones for that key, +even if the original entry disappeared during an earlier attempt. A missing +selection is otherwise an idempotent success. Existing cleanup intents for the +key may be completed along with the selection; other quarantine history is +retained. Parent synchronization is retried even when an earlier deletion +removed the last tombstone before its sync failed. + +For a configured store, operators can inspect and deliberately select an entry: + +```python +from vllm_omni.host_weight_runtime import ArtifactInventoryState, HostWeightError + +for entry in store.inspect_domain().inventory: + if entry.state is ArtifactInventoryState.QUARANTINED: + print(entry.storage_name, entry.size_bytes) + +selected_storage_name = input("Quarantined entry to remove: ") +failure = store.cleanup_quarantined(selected_storage_name) +if failure is not None: + raise HostWeightError(failure) +``` + +Use the same authoritative domain and capacity policy as the service. Review +inventory sizes and inspect again after cleanup before explicitly retrying a +build or startup. These are logical-byte, point-in-time observations; concurrent +activity can affect the difference and physical disk/RAM reclamation is not +implied. This operation adds no automatic retention policy or service retry. + An operational failure while inspecting a noncooperative competing publication is a retryable storage failure. The competing entry remains authoritative and is not mislabeled as corrupt or quarantined merely because its manifest could diff --git a/tests/host_weight_runtime/test_filesystem_store.py b/tests/host_weight_runtime/test_filesystem_store.py index 7fe716c5d..d4051fd0e 100644 --- a/tests/host_weight_runtime/test_filesystem_store.py +++ b/tests/host_weight_runtime/test_filesystem_store.py @@ -440,6 +440,22 @@ def _corrupt_test_payload(artifact: Path) -> None: os.chmod(artifact, 0o555) +def _quarantine_and_rebuild(store: FilesystemHostWeightStore, identity: WeightArtifactIdentity) -> Path: + before = set(store.quarantine_dir.iterdir()) + _corrupt_test_payload(store.artifacts_dir / identity.key) + result = store.get_or_build( + BuildRequest(identity), + FakeProducer(identity), + validation=ValidationLevel.FULL_CHECKSUM, + deadline=time.monotonic() + 10, + ) + assert result.status is StoreStatus.BUILT + assert result.lease is not None + result.lease.close() + (quarantined,) = set(store.quarantine_dir.iterdir()) - before + return quarantined + + def _build_process(root: str, counter: str, initial_lookup_barrier: object, result_queue: object) -> None: try: identity = _identity() @@ -2496,6 +2512,199 @@ def test_store_and_free_space_capacity_limits_fail_before_publication(tmp_path: assert not list(free_limited.artifacts_dir.iterdir()) +def test_cleanup_quarantined_preserves_current_artifact_other_entries_and_deny(tmp_path: Path) -> None: + store = _make_store(tmp_path / "store") + identity, artifact = _publish_test_artifact(store) + first = _quarantine_and_rebuild(store, identity) + second = _quarantine_and_rebuild(store, identity) + ready_inode = artifact.stat().st_ino + lock_inodes = {path.name: path.stat().st_ino for path in store.locks_dir.iterdir()} + + assert store.cleanup_quarantined(first.name) is None + assert set(store.quarantine_dir.iterdir()) == {second} + assert artifact.stat().st_ino == ready_inode + hit = store.lookup(identity, validation=ValidationLevel.FULL_CHECKSUM) + assert hit.lease is not None + assert torch.equal(hit.lease.tensors["layer.weight"], _weight()) + hit.lease.close() + + _corrupt_test_payload(artifact) + assert store.lookup(identity, validation=ValidationLevel.FULL_CHECKSUM).status is StoreStatus.INVALID + deny = store.deny_dir / f"{identity.key}.json" + deny_bytes = deny.read_bytes() + assert store.cleanup_quarantined(second.name) is None + assert deny.read_bytes() == deny_bytes + assert artifact.stat().st_ino == ready_inode + assert {path.name: path.stat().st_ino for path in store.locks_dir.iterdir()} == lock_inodes + assert store.cleanup_quarantined(first.name) is None + + +@pytest.mark.parametrize("active", ["build", "lease"]) +def test_cleanup_quarantined_refuses_active_key(tmp_path: Path, active: str) -> None: + store = _make_store(tmp_path / "store") + identity, _ = _publish_test_artifact(store) + quarantined = _quarantine_and_rebuild(store, identity) + owner: FileLock | HostWeightLease + if active == "build": + owner = FileLock(store._build_lock_path(identity.key), exclusive=True, deadline=None) + expected = FailureCode.ACTIVE_BUILD_TIMEOUT + else: + lease = store.lookup(identity, validation=ValidationLevel.FULL_CHECKSUM).lease + assert lease is not None + owner = lease + expected = FailureCode.ACTIVE_LEASE + try: + failure = store.cleanup_quarantined(quarantined.name) + assert failure is not None and failure.code is expected and failure.retryable + assert quarantined.is_dir() + finally: + owner.close() + assert store.cleanup_quarantined(quarantined.name) is None + + +@pytest.mark.parametrize("recovery", ["cleanup", "build"]) +def test_cleanup_quarantined_retries_post_move_removal_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, recovery: str +) -> None: + store = _make_store(tmp_path / "store") + identity, artifact = _publish_test_artifact(store) + quarantined = _quarantine_and_rebuild(store, identity) + original_rmtree = shutil.rmtree + + def fail_removal(path: str | os.PathLike[str]) -> None: + if Path(path).parent == store.quarantine_dir: + raise OSError(errno.EIO, "injected removal failure") + original_rmtree(path) + + with monkeypatch.context() as fault: + fault.setattr(shutil, "rmtree", fail_removal) + failure = store.cleanup_quarantined(quarantined.name) + assert failure is not None and failure.code is FailureCode.QUARANTINE_FAILED and failure.retryable + assert not quarantined.exists() + assert len(list(store.quarantine_dir.glob(f"{identity.key}.cleanup.*"))) == 1 + + if recovery == "cleanup": + assert store.cleanup_quarantined(quarantined.name) is None + else: + # A real build (not a warm hit) reconciles the same tombstone protocol. + _corrupt_test_payload(artifact) + result = store.get_or_build( + BuildRequest(identity), + FakeProducer(identity), + validation=ValidationLevel.FULL_CHECKSUM, + deadline=time.monotonic() + 10, + ) + assert result.status is StoreStatus.BUILT + assert result.lease is not None + assert torch.equal(result.lease.tensors["layer.weight"], _weight()) + result.lease.close() + assert not list(store.quarantine_dir.glob(f"{identity.key}.cleanup.*")) + + +@pytest.mark.parametrize("after_removal", [False, True]) +def test_cleanup_quarantined_retries_parent_sync_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, after_removal: bool +) -> None: + store = _make_store(tmp_path / "store") + identity, _ = _publish_test_artifact(store) + quarantined = _quarantine_and_rebuild(store, identity) + original_sync = filesystem_store_module._fsync_directory + + def fail_sync(path: Path) -> None: + if path == store.quarantine_dir and ( + not after_removal or not list(store.quarantine_dir.glob(f"{identity.key}.cleanup.*")) + ): + raise OSError(errno.EIO, "injected quarantine sync failure") + original_sync(path) + + with monkeypatch.context() as fault: + fault.setattr(filesystem_store_module, "_fsync_directory", fail_sync) + failure = store.cleanup_quarantined(quarantined.name) + assert failure is not None and failure.code is FailureCode.QUARANTINE_FAILED and failure.retryable + assert not quarantined.exists() + assert bool(list(store.quarantine_dir.glob(f"{identity.key}.cleanup.*"))) is not after_removal + + # An already absent selection must still retry the failed durability step. + with monkeypatch.context() as fault: + fault.setattr(filesystem_store_module, "_fsync_directory", fail_sync) + assert store.cleanup_quarantined(quarantined.name) is not None + assert store.cleanup_quarantined(quarantined.name) is None + assert not list(store.quarantine_dir.glob(f"{identity.key}.cleanup.*")) + + +@pytest.mark.parametrize("entry_kind", ["symlink", "file"]) +def test_cleanup_quarantined_rejects_non_directory_entry(tmp_path: Path, entry_kind: str) -> None: + store = _make_store(tmp_path / "store") + identity, artifact = _publish_test_artifact(store) + entry = store.quarantine_dir / f"{identity.key}.validation_failure.test" + if entry_kind == "symlink": + entry.symlink_to(artifact, target_is_directory=True) + else: + entry.write_text("not an artifact directory") + failure = store.cleanup_quarantined(entry.name) + assert failure is not None and failure.code is FailureCode.QUARANTINE_FAILED + assert not failure.retryable + assert entry.exists() + hit = store.lookup(identity, validation=ValidationLevel.FULL_CHECKSUM) + assert hit.lease is not None + assert torch.equal(hit.lease.tensors["layer.weight"], _weight()) + hit.lease.close() + + +@pytest.mark.parametrize( + "name", + ["../escape", "/tmp/escape", "f" * 64, "g" * 64 + ".x", "f" * 64 + ".", "f" * 64 + ".x/../y", "f" * 64 + ".x\0"], +) +def test_cleanup_quarantined_rejects_invalid_selection_before_mutation(tmp_path: Path, name: str) -> None: + store = _make_store(tmp_path / "store") + locks = set(store.locks_dir.iterdir()) + with pytest.raises(ValueError, match="inventory basename"): + store.cleanup_quarantined(name) + assert set(store.locks_dir.iterdir()) == locks + + +def test_cleanup_quarantined_restores_bounded_store_recovery_headroom(tmp_path: Path) -> None: + store = _make_store(tmp_path / "store", capacity=CapacityPolicy(max_store_bytes=16384)) + identity = _identity() + for _ in range(32): + result = store.get_or_build( + BuildRequest(identity), + FakeProducer(identity), + validation=ValidationLevel.FULL_CHECKSUM, + deadline=time.monotonic() + 10, + ) + if result.status is StoreStatus.FAILED: + break + assert result.lease is not None + result.lease.close() + _corrupt_test_payload(store.artifacts_dir / identity.key) + assert result.failure is not None and result.failure.code is FailureCode.STORE_LIMIT_EXCEEDED + quarantined = [ + entry for entry in store.inspect_domain().inventory if entry.state is ArtifactInventoryState.QUARANTINED + ] + assert len(quarantined) > 1 + assert store.cleanup(identity) is None + still_full = store.get_or_build( + BuildRequest(identity), + FakeProducer(identity), + validation=ValidationLevel.FULL_CHECKSUM, + deadline=time.monotonic() + 10, + ) + assert still_full.failure is not None and still_full.failure.code is FailureCode.STORE_LIMIT_EXCEEDED + + assert store.cleanup_quarantined(quarantined[0].storage_name) is None + recovered = store.get_or_build( + BuildRequest(identity), + FakeProducer(identity), + validation=ValidationLevel.FULL_CHECKSUM, + deadline=time.monotonic() + 10, + ) + assert recovered.status is StoreStatus.BUILT + assert recovered.lease is not None + assert torch.equal(recovered.lease.tensors["layer.weight"], _weight()) + recovered.lease.close() + + def test_domain_capacity_policy_is_authoritative(tmp_path: Path) -> None: root = tmp_path / "store" _make_store(root, capacity=CapacityPolicy(max_store_bytes=4096)) diff --git a/vllm_omni/host_weight_runtime/filesystem/store.py b/vllm_omni/host_weight_runtime/filesystem/store.py index 0f01e36da..40f7fe021 100644 --- a/vllm_omni/host_weight_runtime/filesystem/store.py +++ b/vllm_omni/host_weight_runtime/filesystem/store.py @@ -1466,43 +1466,45 @@ class FilesystemHostWeightStore: total += file_stat.st_size return total + @contextlib.contextmanager + def _cleanup_locks(self, key: str) -> Iterator[None]: + """Exclude cooperative builders and leases without waiting for them.""" + try: + build_lock = FileLock(self._build_lock_path(key), exclusive=True, deadline=None, nonblocking=True) + except FileLockTimeoutError as exc: + raise HostWeightError( + _failure( + ResolutionStage.LIFECYCLE, + FailureCode.ACTIVE_BUILD_TIMEOUT, + f"artifact {key} has an active builder", + retryable=True, + ) + ) from exc + with build_lock: + try: + artifact_lock = FileLock(self._artifact_lock_path(key), exclusive=True, deadline=None, nonblocking=True) + except FileLockTimeoutError as exc: + raise HostWeightError( + _failure( + ResolutionStage.LIFECYCLE, + FailureCode.ACTIVE_LEASE, + f"artifact {key} has an active lease", + retryable=True, + ) + ) from exc + with artifact_lock: + yield + def cleanup(self, identity: WeightArtifactIdentity) -> HostWeightFailure | None: """Explicitly remove an inactive artifact without waiting for leases.""" try: - try: - build_lock = FileLock( - self._build_lock_path(identity.key), - exclusive=True, - deadline=None, - nonblocking=True, - ) - except FileLockTimeoutError: - return _failure( - ResolutionStage.LIFECYCLE, - FailureCode.ACTIVE_BUILD_TIMEOUT, - f"artifact {identity.key} has an active builder", - retryable=True, - ) - with build_lock: - try: - artifact_lock = FileLock( - self._artifact_lock_path(identity.key), - exclusive=True, - deadline=None, - nonblocking=True, - ) - except FileLockTimeoutError: - return _failure( - ResolutionStage.LIFECYCLE, - FailureCode.ACTIVE_LEASE, - f"artifact {identity.key} has an active lease", - retryable=True, - ) - with artifact_lock: - self._repair_quarantine_transitions_locked(identity.key) - self._quarantine_locked(identity.key, reason="cleanup") - self._remove_cleanup_tombstones_locked(identity.key) - self._remove_deny_locked(identity.key) + with self._cleanup_locks(identity.key): + self._repair_quarantine_transitions_locked(identity.key) + self._quarantine_locked(identity.key, reason="cleanup") + self._remove_cleanup_tombstones_locked(identity.key) + self._remove_deny_locked(identity.key) + except HostWeightError as exc: + return exc.failure except OSError as exc: return _failure( ResolutionStage.LIFECYCLE, @@ -1512,6 +1514,50 @@ class FilesystemHostWeightStore: ) return None + def cleanup_quarantined(self, storage_name: str) -> HostWeightFailure | None: + """Remove one quarantine inventory entry, preserving the current artifact. + + A retry also finishes earlier cleanup tombstones for the same key. + Active builders or leases are reported rather than interrupted. + """ + key, separator, suffix = storage_name.partition(".") + if ( + not separator + or not suffix + or "\0" in storage_name + or _ARTIFACT_KEY_RE.fullmatch(key) is None + or Path(storage_name).name != storage_name + ): + raise ValueError("quarantined storage_name must be a single inventory basename with an artifact key") + try: + with self._cleanup_locks(key): + entry = self.quarantine_dir / storage_name + exists = _path_exists_without_following(entry) + if exists and not stat.S_ISDIR(entry.lstat().st_mode): + return _failure( + ResolutionStage.LIFECYCLE, + FailureCode.QUARANTINE_FAILED, + f"quarantine entry is not a directory: {storage_name}", + ) + self._repair_quarantine_transitions_locked(key) + if exists and not storage_name.startswith(f"{key}.cleanup."): + tombstone = self.quarantine_dir / f"{key}.cleanup.{uuid.uuid4().hex}" + _move_artifact_locked(entry, tombstone) + self._remove_cleanup_tombstones_locked(key) + # A previous attempt may have removed the tombstone but failed + # its parent sync. Retry that sync even when no entry remains. + _fsync_directory(self.quarantine_dir) + except HostWeightError as exc: + return exc.failure + except OSError as exc: + return _failure( + ResolutionStage.LIFECYCLE, + FailureCode.QUARANTINE_FAILED, + f"failed to remove quarantined entry {storage_name}: {exc}", + retryable=True, + ) + return None + def inspect_domain(self) -> DomainInspection: usage = shutil.disk_usage(self.root) inventory: list[ArtifactInventoryEntry] = []