[Core] Add explicit cleanup for quarantined HWR artifacts

Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>

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

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))

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:
+            with self._cleanup_locks(identity.key):
-                build_lock = FileLock(
+                self._repair_quarantine_transitions_locked(identity.key)
-                    self._build_lock_path(identity.key),
+                self._quarantine_locked(identity.key, reason="cleanup")
-                    exclusive=True,
+                self._remove_cleanup_tombstones_locked(identity.key)
-                    deadline=None,
+                self._remove_deny_locked(identity.key)
-                    nonblocking=True,
+        except HostWeightError as exc:
-                )
+            return exc.failure
-            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)
         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] = []