From 17dcbd3d4f4b8486ddc7c32b0311c13724587779 Mon Sep 17 00:00:00 2001 From: Will Miao Date: Thu, 27 Aug 2026 19:32:17 +0800 Subject: [PATCH] fix(delete): merge delete batches manifest-only, never move files Bulk delete merged staged batches by physically moving each loser's files into the winner's batch dir with os.rename. Cross-volume bulks (winner and loser on different filesystems) always hit EXDEV, forcing a rollback and degrading to the batch_ids array with per-batch undo. Merge is now manifest-only: loser entries are appended to the winner's manifest with their staged paths unchanged, so staged files keep living in each model's own .lm-pending-delete/ dir (no data IO, no EXDEV). Loser dirs are recorded in the winner manifest's merged_sources and each loser manifest is stamped merged_into so its own purge timer, a post-restart sweep or a direct undo call no-op. A cross-volume bulk is one undoable batch again, and undo/purge clean up the loser dirs once the merged batch settles. --- py/services/model_scanner.py | 4 +- py/services/pending_delete_service.py | 195 +++++++++-------- py/services/recipes/persistence_service.py | 2 +- tests/services/test_model_scanner.py | 58 +++--- tests/services/test_pending_delete_service.py | 197 +++++++++--------- tests/services/test_recipe_persistence.py | 46 +++- 6 files changed, 278 insertions(+), 224 deletions(-) diff --git a/py/services/model_scanner.py b/py/services/model_scanner.py index 70f5ed1e..69a28b9f 100644 --- a/py/services/model_scanner.py +++ b/py/services/model_scanner.py @@ -2501,8 +2501,8 @@ class ModelScanner: }) # Merge every staged per-file batch into ONE undoable batch. On a - # merge failure (cross-volume EXDEV etc.) the response falls back - # to the constituent batch_ids array so the frontend can undo them + # merge failure (defensive) the response falls back to the + # constituent batch_ids array so the frontend can undo them # sequentially. batch_field: Dict[str, Any] = {} if batch_ids: diff --git a/py/services/pending_delete_service.py b/py/services/pending_delete_service.py index e92f8804..0b47fae5 100644 --- a/py/services/pending_delete_service.py +++ b/py/services/pending_delete_service.py @@ -274,17 +274,21 @@ class PendingDeleteService: async def merge_batches(self, batch_ids: Sequence[str]) -> Optional[str]: """Merge several batches into the first batch's manifest. - Winner is ``batch_ids[0]``. The staged files of losing batches are - MOVED (os.rename) into the winner's batch dir and their ``staged`` - paths rewritten in the merged manifest BEFORE any loser dir is - removed. ``expires_at`` is re-anchored to ``now + TTL`` at merge time - and a FRESH purge timer is armed for the winner. + Winner is ``batch_ids[0]``. Merging is MANIFEST-ONLY: staged files + are NEVER moved, so the merge is a pure metadata operation with zero + data IO and is inherently cross-volume safe (no EXDEV, no rollback). + Every loser's entries are appended to the winner's manifest with + their ``staged`` paths unchanged (files keep living in the loser's + own batch dir - the sibling-of-model staging location), each loser + dir is recorded in the winner manifest's ``merged_sources``, and each + loser manifest is stamped ``merged_into`` so its own purge timer, a + post-restart sweep or a direct undo call no-op. ``expires_at`` is + re-anchored to ``now + TTL`` at merge time and a FRESH purge timer is + armed for the winner. - On any move failure every already-moved file is moved BACK and the - original batch dirs/manifests are left intact; ``None`` is returned so - callers fall back to the ``batch_ids`` array contract. Cross-volume - merges hit EXDEV here - expected and fine (the fallback is the normal - path for those bulks). + Returns the winner id, or ``None`` when the winner batch cannot be + resolved (callers then fall back to the ``batch_ids`` array + contract). """ if not batch_ids: return None @@ -298,77 +302,68 @@ class PendingDeleteService: if winner_manifest is None: return None - # Track (entry, original_staged_path, loser_dir) for rollback. - moved: List[Tuple[Dict[str, Any], str, str]] = [] - processed_losers: List[Tuple[str, str]] = [] # (loser_id, loser_dir) - - try: - for loser_id in batch_ids[1:]: - loser_dir = await self._find_batch_dir(loser_id) - if not loser_dir or os.path.normpath(loser_dir) == os.path.normpath( - winner_dir - ): + # Build the merged manifest in memory: loser entries are appended + # with their staged paths UNCHANGED - no file moves, no IO, no + # EXDEV. Loser dirs remain as physical storage until the merged + # batch is undone or purged. + merged_sources: List[str] = [] + seen_loser_dirs: Set[str] = set() + for loser_id in batch_ids[1:]: + loser_dir = await self._find_batch_dir(loser_id) + if not loser_dir or os.path.normpath(loser_dir) == os.path.normpath( + winner_dir + ): + continue + loser_abs = os.path.abspath(loser_dir) + if loser_abs in seen_loser_dirs: + continue + seen_loser_dirs.add(loser_abs) + loser_manifest = self._read_manifest(loser_dir) + if loser_manifest is None: + # Corrupted loser: leave it for the sweep to quarantine. + continue + for entry in loser_manifest.get("entries") or []: + if entry.get("restored"): continue - loser_manifest = self._read_manifest(loser_dir) - if loser_manifest is None: - # Corrupted loser: leave it for the sweep to quarantine. + staged_path = entry.get("staged") + if not staged_path or not os.path.exists(staged_path): continue - for entry in loser_manifest.get("entries") or []: - if entry.get("restored"): - continue - staged_path = entry.get("staged") - if not staged_path or not os.path.exists(staged_path): - continue - new_staged = os.path.join( - winner_dir, os.path.basename(staged_path) - ) - if os.path.exists(new_staged): - # os.rename would silently overwrite the existing - # staged file on POSIX - never drop a staged file. - # Abort the merge so callers fall back to the - # batch_ids array contract. - raise OSError( - f"Merge collision: {os.path.basename(staged_path)} " - "already staged in winner batch" - ) - os.rename(staged_path, new_staged) - original_staged = entry["staged"] - entry["staged"] = os.path.abspath(new_staged) - winner_manifest["entries"].append(entry) - moved.append((entry, original_staged, loser_dir)) - processed_losers.append((loser_id, loser_dir)) - except OSError as exc: - logger.warning( - "Merge of %s failed after moving files: %s; rolling back", - list(batch_ids), - exc, - ) - self._rollback_merge_moves(moved) - return None + winner_manifest["entries"].append(entry) + merged_sources.append(loser_abs) - # Re-anchor expiry and persist the merged manifest atomically. + # Re-anchor expiry and persist the merged manifest atomically - it + # becomes the ONLY source of truth for every merged file, wherever + # it physically lives. winner_manifest["expires_at"] = ( int(time.time()) + PENDING_DELETE_TTL_SECONDS ) + if merged_sources: + winner_manifest["merged_sources"] = merged_sources try: self._write_manifest_atomic(winner_dir, winner_manifest) except OSError as exc: logger.warning( - "Failed to write merged manifest for %s: %s; rolling back", + "Failed to write merged manifest for %s: %s", winner_id, exc, ) - self._rollback_merge_moves(moved) return None - # All moves committed: remove loser dirs (must be empty by now) - # and drop them from the registry. Skipped losers (missing / - # corrupted / same-dir) stay registered so the sweep still - # quarantines them, exactly as before the registry existed. - for loser_id, loser_dir in processed_losers: - self._remove_manifest(loser_dir) - self._remove_empty_dir(loser_dir) - await self._forget_batch(loser_id) + # Stamp each loser manifest so its own purge timer / a later sweep + # / a direct undo call no-op: the winner owns those files from + # here on. Best-effort coordination; a failed stamp only risks the + # loser being swept at its own (earlier) expiry after a restart. + for loser_dir in merged_sources: + try: + self._mark_merged(loser_dir, winner_id) + except OSError as exc: # pragma: no cover - best-effort + logger.warning( + "Failed to mark merged loser %s: %s", loser_dir, exc + ) + + # Losers are no longer independently managed. + for loser_dir in merged_sources: + await self._forget_batch(os.path.basename(loser_dir)) await self._remember_batch(winner_id, winner_dir) # Arm a fresh purge timer for the winner with the re-anchored @@ -397,6 +392,16 @@ class PendingDeleteService: if manifest is None: raise ValueError(f"Manifest missing for batch {batch_id}") + merged_into = manifest.get("merged_into") + if merged_into: + # The batch was merged into another batch: its staged files + # are owned by the winner's manifest. Undo via the winner so + # the whole merged batch stays consistent. + raise ValueError( + f"Batch {batch_id} was merged into batch {merged_into}; " + "undo that batch instead" + ) + if manifest.get("state") == "restored": return self._undo_result(manifest) @@ -448,6 +453,10 @@ class PendingDeleteService: self._remove_manifest(batch_dir) self._remove_empty_dir(batch_dir) await self._forget_batch(batch_id) + # Clean up merged loser dirs (their staged files were restored + # above) and drop them from the registry too. + for loser_id in self._remove_merged_batch_dirs(manifest): + await self._forget_batch(loser_id) logger.info("Restored pending-delete batch %s", batch_id) return self._undo_result(manifest) @@ -779,25 +788,35 @@ class PendingDeleteService: "Failed to remove staged copy %s: %s", staged_path, exc ) - def _rollback_merge_moves( - self, moved: Sequence[Tuple[Dict[str, Any], str, str]] - ) -> None: - """Move already-merged files back to their original loser batch dirs.""" - for _entry, original_staged, _loser_dir in reversed(list(moved)): - current = _entry.get("staged") - if not current or not original_staged: + def _mark_merged(self, loser_dir: str, winner_id: str) -> None: + """Stamp ``merged_into`` on a loser manifest (best-effort). + + The stamp makes the loser's own purge timer, post-restart sweeps and + direct undo calls no-op, so the winner's merged batch stays the only + owner of the loser's staged files until it is undone or purged. + """ + loser_manifest = self._read_manifest(loser_dir) + if loser_manifest is None: + return + loser_manifest["merged_into"] = winner_id + self._write_manifest_atomic(loser_dir, loser_manifest) + + def _remove_merged_batch_dirs(self, manifest: Dict[str, Any]) -> List[str]: + """Remove merged loser batch dirs once their files were handled. + + Called after a merged batch has been fully undone or purged: each + loser manifest (stamped ``merged_into``) and its now-empty dir are + removed so the sweep never quarantines an orphaned staging dir. + Best-effort - returns the removed batch ids for registry cleanup. + """ + removed: List[str] = [] + for src in manifest.get("merged_sources") or []: + if not isinstance(src, str) or not src: continue - if not os.path.exists(current): - continue - try: - os.rename(current, original_staged) - except OSError as exc: # pragma: no cover - best-effort rollback - logger.warning( - "Failed to roll back merge move %s -> %s: %s", - current, - original_staged, - exc, - ) + self._remove_manifest(src) + self._remove_empty_dir(src) + removed.append(os.path.basename(src)) + return removed def _purge_batch_dir(self, batch_dir: str) -> bool: """Purge one batch dir. Returns True when the batch was purged/removed.""" @@ -811,6 +830,13 @@ class PendingDeleteService: self._quarantine_batch_dir(batch_dir) return True + if manifest.get("merged_into"): + # Merged into another batch: the winner owns these staged files. + # The loser's own purge timer / post-restart sweep must not remove + # them early (the winner re-anchored the merged expiry to give the + # whole bulk one undo window). + return False + if manifest.get("state") == "restored": return False @@ -842,6 +868,7 @@ class PendingDeleteService: self._remove_manifest(batch_dir) self._remove_empty_dir(batch_dir) + self._remove_merged_batch_dirs(manifest) return True def _quarantine_batch_dir(self, batch_dir: str) -> str: diff --git a/py/services/recipes/persistence_service.py b/py/services/recipes/persistence_service.py index 3147d48e..664b0dc3 100644 --- a/py/services/recipes/persistence_service.py +++ b/py/services/recipes/persistence_service.py @@ -533,7 +533,7 @@ class RecipePersistenceService: # Merge succeeded: one undo action covers the whole bulk. payload["batch_id"] = merged_batch_id else: - # Merge failure (e.g. cross-volume move): expose the constituent + # Merge unresolvable (defensive): expose the constituent # batches so the caller can undo them one at a time. payload["batch_ids"] = batch_ids else: diff --git a/tests/services/test_model_scanner.py b/tests/services/test_model_scanner.py index 03e8957c..0b002b62 100644 --- a/tests/services/test_model_scanner.py +++ b/tests/services/test_model_scanner.py @@ -1124,7 +1124,7 @@ async def test_bulk_delete_stages_two_files_into_single_batch(tmp_path: Path): assert result["total_deleted"] == 2 assert result["cache_updated"] is True - # ONE batch id, no batch_ids array, and both files staged in its dir. + # ONE batch id, no batch_ids array - the merge succeeded. assert "batch_id" in result assert "batch_ids" not in result batch_id = result["batch_id"] @@ -1132,15 +1132,28 @@ async def test_bulk_delete_stages_two_files_into_single_batch(tmp_path: Path): staging = root / PENDING_DELETE_DIR_NAME batch_dir = staging / batch_id assert batch_dir.is_dir() - assert (batch_dir / "one.txt").read_bytes() == b"one" - assert (batch_dir / "two.txt").read_bytes() == b"two" + manifest = json.loads((batch_dir / "manifest.json").read_text(encoding="utf-8")) + assert sorted(os.path.basename(e["staged"]) for e in manifest["entries"]) == [ + "one.txt", + "two.txt", + ] - # Loser batch dirs are removed by the merge - exactly one batch remains. + # Manifest-only merge: each staged file physically remains in its OWN + # batch dir (one.txt in the winner, two.txt in the loser storage dir) - + # no file was moved, so no cross-volume IO ever happens. + assert (batch_dir / "one.txt").read_bytes() == b"one" batch_dirs = [d.name for d in staging.iterdir() if d.is_dir()] - assert batch_dirs == [batch_id] + assert len(batch_dirs) == 2 + assert batch_id in batch_dirs + staged_files = { + f.name + for bid in batch_dirs + for f in (staging / bid).iterdir() + if f.is_file() and f.name != "manifest.json" + } + assert staged_files == {"one.txt", "two.txt"} # The manifest carries the winner's cache snapshot for later undo. - manifest = json.loads((batch_dir / "manifest.json").read_text(encoding="utf-8")) assert manifest["model_snapshot"]["file_path"] == str(first) # Originals gone; cache entries removed. @@ -1171,17 +1184,19 @@ async def test_bulk_delete_merged_manifest_reanchors_expiry(tmp_path: Path): # expires_at >= staging completion time + TTL (re-anchor assertion). assert manifest["expires_at"] >= after + PENDING_DELETE_TTL_SECONDS - 2 assert manifest["expires_at"] >= before + PENDING_DELETE_TTL_SECONDS - # Both files are entries of the merged manifest. + # Both files are entries of the merged manifest; manifest-only merge means + # each staged file stays where staging put it (still on disk, in its own + # batch dir). assert len(manifest["entries"]) == 2 + assert all(os.path.exists(entry["staged"]) for entry in manifest["entries"]) assert (batch_dir / "one.txt").exists() - assert (batch_dir / "two.txt").exists() @pytest.mark.asyncio async def test_bulk_delete_merge_failure_falls_back_to_batch_ids( tmp_path: Path, monkeypatch ): - """Merge move failure -> batch_ids array of the intact constituent batches.""" + """Merge unresolvable -> batch_ids array of the intact constituent batches.""" root = tmp_path / "loras" root.mkdir() first = root / "one.txt" @@ -1190,24 +1205,17 @@ async def test_bulk_delete_merge_failure_falls_back_to_batch_ids( second.write_text("two", encoding="utf-8") scanner = _make_bulk_scanner(root, [first, second]) - real_rename = os.rename - fail_next = {"enabled": True} + # Simulate a merge that cannot resolve the winner batch (the only real + # merge failure mode since the merge no longer moves files): the caller + # must fall back to returning the constituent batch_ids array. + from py.services.pending_delete_service import get_pending_delete_service - def flaky_merge_rename(src: str, dst: str) -> None: - # Fail only when moving between batch dirs (merge), never during - # staging (src is then the original path, outside .lm-pending-delete). - if ( - fail_next["enabled"] - and PENDING_DELETE_DIR_NAME in src - and PENDING_DELETE_DIR_NAME in dst - ): - fail_next["enabled"] = False - raise OSError("simulated merge failure") - return real_rename(src, dst) + service = await get_pending_delete_service() - monkeypatch.setattr( - "py.services.pending_delete_service.os.rename", flaky_merge_rename - ) + async def _merge_unresolvable(_ids) -> None: + return None + + monkeypatch.setattr(service, "merge_batches", _merge_unresolvable) result = await scanner.bulk_delete_models([str(first), str(second)]) diff --git a/tests/services/test_pending_delete_service.py b/tests/services/test_pending_delete_service.py index 7561ba19..865bfbd3 100644 --- a/tests/services/test_pending_delete_service.py +++ b/tests/services/test_pending_delete_service.py @@ -1,9 +1,10 @@ """Tests for :mod:`py.services.pending_delete_service`. Covers the staging service contract: stage (model + recipe), undo (with -partial-undo retry and occupied-path protection), merge (with rollback and a -fresh purge timer), purge (expired-only, quarantine of malformed batches, -per-file lock tolerance) and the scanner exclusion of the staging directory. +partial-undo retry and occupied-path protection), merge (manifest-only: no +file moves, cross-root safe, with a fresh purge timer), purge (expired-only, +quarantine of malformed batches, per-file lock tolerance) and the scanner +exclusion of the staging directory. Deterministic time control: no real sleeps - tests rewrite ``expires_at`` in the manifest or monkeypatch time functions instead. @@ -539,9 +540,12 @@ async def test_j_stale_timer_purge_batch_noop(tmp_path: Path, monkeypatch) -> No # --------------------------------------------------------------------------- -# (k) MERGE -> single manifest, re-anchored expiry, all files under winner +# (k) MERGE -> single manifest, re-anchored expiry, staged files stay in +# their OWN batch dirs (manifest-only merge: no moves, no IO, no EXDEV) # --------------------------------------------------------------------------- -async def test_k_merge_produces_single_manifest_and_moves_all_files(tmp_path: Path, monkeypatch) -> None: +async def test_k_merge_is_manifest_only_and_leaves_files_in_place( + tmp_path: Path, monkeypatch, +) -> None: root = tmp_path / "loras" root.mkdir() _spy_purge_timers(monkeypatch) @@ -582,20 +586,23 @@ async def test_k_merge_produces_single_manifest_and_moves_all_files(tmp_path: Pa assert manifest["batch_id"] == bid_a assert len(manifest["entries"]) == 3 assert before_merge + PENDING_DELETE_TTL_SECONDS - 2 <= manifest["expires_at"] <= before_merge + PENDING_DELETE_TTL_SECONDS + 2 + assert manifest["merged_sources"] == [str(loser_dir)] + # Entry staged paths were NOT rewritten: every file keeps living in its + # own batch dir, byte-identical. staged_paths = [entry["staged"] for entry in manifest["entries"]] assert len(staged_paths) == 3 - for staged in staged_paths: - assert str(staged).startswith(str(winner_dir)) - assert os.path.exists(staged) - - # Byte-compare: no file dropped. + assert all(os.path.exists(staged) for staged in staged_paths) assert (winner_dir / "alpha.safetensors").read_bytes() == b"alpha-data" assert (winner_dir / "alpha.metadata.json").read_bytes() == b"alpha-meta" - assert (winner_dir / "beta.safetensors").read_bytes() == b"beta-data" + assert not (winner_dir / "beta.safetensors").exists() + assert (loser_dir / "beta.safetensors").read_bytes() == b"beta-data" - # Loser batch dir removed (after being empty). - assert not loser_dir.exists() + # The loser dir is retained as physical storage and stamped merged_into so + # its own timer / sweep / direct undo no-op. + assert loser_dir.is_dir() + loser_manifest = json.loads((loser_dir / "manifest.json").read_text(encoding="utf-8")) + assert loser_manifest["merged_into"] == bid_a # --------------------------------------------------------------------------- @@ -669,28 +676,32 @@ async def test_k3_merge_then_purge_empties_and_removes_winner_dir(tmp_path: Path assert not winner_dir.exists() assert not (root / "alpha.safetensors").exists() assert not (root / "beta.safetensors").exists() + # The loser storage dir is cleaned up with the merged batch. + assert not (root / PENDING_DELETE_DIR_NAME / bid_b).exists() # --------------------------------------------------------------------------- -# (l) MERGE MOVE FAILURE -> rollback, all batches intact, sequential undo works +# (l) CROSS-ROOT MERGE (the real-world EXDEV case) -> manifest-only merge +# succeeds, files stay in their own roots, ONE undo restores everything # --------------------------------------------------------------------------- -async def test_l_merge_move_failure_rolls_back(tmp_path: Path, monkeypatch) -> None: - root = tmp_path / "loras" - root.mkdir() +async def test_l_cross_root_merge_succeeds_without_moving_files( + tmp_path: Path, monkeypatch, +) -> None: + root_a = tmp_path / "loras_a" + root_a.mkdir() + root_b = tmp_path / "loras_b" + root_b.mkdir() _spy_purge_timers(monkeypatch) service = await PendingDeleteService.get_instance() - a1 = root / "alpha.safetensors" - a1.write_bytes(b"alpha-data") - bid_a = await _stage_simple(service, root, "alpha") - # Loser has TWO files so a move fails after the first was already moved. - b1 = root / "beta.safetensors" + bid_a = await _stage_simple(service, root_a, "alpha") + b1 = root_b / "beta.safetensors" b1.write_bytes(b"beta-data") - b2 = root / "beta.metadata.json" + b2 = root_b / "beta.metadata.json" b2.write_bytes(b"beta-meta") bid_b = await service.stage_model_delete( - scanner=ScannerForStage([root]), - target_dir=str(root), + scanner=ScannerForStage([root_b]), + target_dir=str(root_b), file_name="beta", main_extension=".safetensors", original_file_path=str(b1), @@ -698,77 +709,60 @@ async def test_l_merge_move_failure_rolls_back(tmp_path: Path, monkeypatch) -> N ) assert bid_b is not None - real_rename = os.rename - calls = {"n": 0} - fail_next = {"enabled": True} + # The batches live under DIFFERENT roots (a real os.rename would raise + # EXDEV here); the manifest-only merge must not move a single byte. + assert await service.merge_batches([bid_a, bid_b]) == bid_a - def flaky_rename(src: str, dst: str) -> None: - calls["n"] += 1 - if fail_next["enabled"] and calls["n"] == 2: - raise OSError("simulated move failure") - return real_rename(src, dst) - - monkeypatch.setattr("py.services.pending_delete_service.os.rename", flaky_rename) - - result = await service.merge_batches([bid_a, bid_b]) - assert result is None - - # Already-moved file moved back; both batch dirs + manifests + files intact. - winner_dir = root / PENDING_DELETE_DIR_NAME / bid_a - loser_dir = root / PENDING_DELETE_DIR_NAME / bid_b - assert winner_dir.is_dir() - assert loser_dir.is_dir() - assert (winner_dir / "manifest.json").exists() - assert (loser_dir / "manifest.json").exists() + loser_dir = root_b / PENDING_DELETE_DIR_NAME / bid_b assert (loser_dir / "beta.safetensors").read_bytes() == b"beta-data" assert (loser_dir / "beta.metadata.json").read_bytes() == b"beta-meta" - assert (winner_dir / "alpha.safetensors").read_bytes() == b"alpha-data" + assert not (root_b / "beta.safetensors").exists() - # Sequential undo of each constituent batch restores every file. - fail_next["enabled"] = False + # ONE undo of the merged batch restores files living in both roots. await service.undo(bid_a) - await service.undo(bid_b) - assert a1.read_bytes() == b"alpha-data" - assert b1.read_bytes() == b"beta-data" - assert b2.read_bytes() == b"beta-meta" + assert (root_a / "alpha.safetensors").read_bytes() == b"alpha-data" + assert (root_b / "beta.safetensors").read_bytes() == b"beta-data" + assert (root_b / "beta.metadata.json").read_bytes() == b"beta-meta" + assert not (root_a / PENDING_DELETE_DIR_NAME / bid_a).exists() + assert not (root_b / PENDING_DELETE_DIR_NAME / bid_b).exists() # --------------------------------------------------------------------------- -# (l2) MERGE SAME-BASENAME COLLISION -> abort + rollback, never overwrite +# (l2) SAME-BASENAME MERGE -> files never move, so identical basenames cannot +# collide; ONE undo restores each file to its own folder # --------------------------------------------------------------------------- -async def test_l2_merge_basename_collision_aborts_without_dropping_files( - tmp_path: Path, monkeypatch +async def test_l2_merge_same_basename_keeps_both_files( + tmp_path: Path, monkeypatch, ) -> None: root = tmp_path / "loras" root.mkdir() - _spy_purge_timers(monkeypatch) - - service = await PendingDeleteService.get_instance() sub_a = root / "a" sub_a.mkdir() sub_b = root / "b" sub_b.mkdir() + _spy_purge_timers(monkeypatch) + + service = await PendingDeleteService.get_instance() # Two distinct files that share the same basename after staging. bid_a = await _stage_simple(service, sub_a, "model") bid_b = await _stage_simple(service, sub_b, "model") - result = await service.merge_batches([bid_a, bid_b]) - assert result is None + # Manifest-only merge never moves files, so the identical basenames cannot + # overwrite each other - the merge succeeds into ONE undoable batch. + assert await service.merge_batches([bid_a, bid_b]) == bid_a - # No file dropped: both staged files exist in their own batch dirs. + # Both staged files still exist in their own batch dirs, byte-identical. a_dir = sub_a / PENDING_DELETE_DIR_NAME / bid_a b_dir = sub_b / PENDING_DELETE_DIR_NAME / bid_b assert (a_dir / "model.safetensors").read_bytes() == b"model-data" assert (b_dir / "model.safetensors").read_bytes() == b"model-data" - assert a_dir.is_dir() and b_dir.is_dir() - assert (a_dir / "manifest.json").exists() - assert (b_dir / "manifest.json").exists() - # Sequential undo of each constituent batch restores every original. + # One undo restores every original, each into its own folder. await service.undo(bid_a) - await service.undo(bid_b) assert (sub_a / "model.safetensors").read_bytes() == b"model-data" assert (sub_b / "model.safetensors").read_bytes() == b"model-data" + assert not a_dir.exists() + assert not b_dir.exists() # --------------------------------------------------------------------------- @@ -1576,7 +1570,8 @@ async def test_reg_d_quarantine_removes_registry_entry(tmp_path: Path) -> None: assert (batch_dir.with_name(f"{batch_id}.orphaned")).is_dir() -# (e) merge success: winner present + losers removed; EXDEV-abort: unchanged +# (e) merge success: winner present + losers forgotten; unresolvable-winner +# abort: registry unchanged async def test_reg_e_merge_registry_lifecycle( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -1595,13 +1590,18 @@ async def test_reg_e_merge_registry_lifecycle( assert bid_b not in service._known_batch_dirs assert bid_c in service._known_batch_dirs - # EXDEV-abort: registry untouched. - def exdev_rename(src: str, dst: str) -> None: - raise OSError(errno.EXDEV, "Invalid cross-device link", src, dst) + # Merge with an unresolvable winner: registry untouched (the caller + # falls back to the batch_ids array contract). + real_find = service._find_batch_dir - monkeypatch.setattr("py.services.pending_delete_service.os.rename", exdev_rename) + async def _unresolvable(batch_id: str) -> Optional[str]: + if batch_id == bid_c: + return None + return await real_find(batch_id) + + monkeypatch.setattr(service, "_find_batch_dir", _unresolvable) before = dict(service._known_batch_dirs) - assert await service.merge_batches([bid_a, bid_c]) is None + assert await service.merge_batches([bid_c]) is None assert dict(service._known_batch_dirs) == before @@ -2267,41 +2267,34 @@ async def test_symlink4_folder_deleted_edge_forgets_stale_registry( assert not batch_dir.exists() -# (e) MERGE EXDEV-ABORT: a cross-volume merge abort leaves the registry -# untouched AND the constituent batches individually undoable - sequential -# undo after the abort restores every file. (The merge-success winner/loser -# registry half is covered by test_reg_e; this adds the post-abort undo -# proof.) -async def test_symlink5_merge_exdev_abort_registry_unchanged_then_sequential_undo( +# (e) CROSS-VOLUME MERGE (the real-world EXDEV case): the manifest-only +# merge succeeds across staging roots, the loser leaves the registry, and +# ONE undo of the merged batch restores every file - no sequential +# per-batch undo needed. +async def test_symlink5_cross_volume_merge_then_one_undo_restores_everything( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - root = tmp_path / "loras" - root.mkdir() + root_a = tmp_path / "vol_a" + root_a.mkdir() + root_b = tmp_path / "vol_b" + root_b.mkdir() _spy_purge_timers(monkeypatch) service = await PendingDeleteService.get_instance() - bid_a = await _stage_simple(service, root, "alpha") - bid_b = await _stage_simple(service, root, "beta") + bid_a = await _stage_simple(service, root_a, "alpha") + bid_b = await _stage_simple(service, root_b, "beta") assert set(service._known_batch_dirs) == {bid_a, bid_b} - real_rename = os.rename - fail_next = {"enabled": True} + # Files live on different roots (an os.rename across them would EXDEV); + # the manifest-only merge succeeds and the loser leaves the registry. + assert await service.merge_batches([bid_a, bid_b]) == bid_a + assert bid_a in service._known_batch_dirs + assert bid_b not in service._known_batch_dirs - def exdev_rename(src: str, dst: str) -> None: - if fail_next["enabled"]: - raise OSError(errno.EXDEV, "Invalid cross-device link", src, dst) - return real_rename(src, dst) - - monkeypatch.setattr("py.services.pending_delete_service.os.rename", exdev_rename) - - before = dict(service._known_batch_dirs) - assert await service.merge_batches([bid_a, bid_b]) is None - assert dict(service._known_batch_dirs) == before - - # Sequential undo of the constituents after the abort restores everything. - fail_next["enabled"] = False + # One undo of the merged batch restores files living on both "volumes". await service.undo(bid_a) - await service.undo(bid_b) - assert (root / "alpha.safetensors").read_bytes() == b"alpha-data" - assert (root / "beta.safetensors").read_bytes() == b"beta-data" - staging = root / PENDING_DELETE_DIR_NAME + assert (root_a / "alpha.safetensors").read_bytes() == b"alpha-data" + assert (root_b / "beta.safetensors").read_bytes() == b"beta-data" + assert not (root_a / PENDING_DELETE_DIR_NAME / bid_a).exists() + assert not (root_b / PENDING_DELETE_DIR_NAME / bid_b).exists() + staging = root_a / PENDING_DELETE_DIR_NAME assert not staging.exists() or not any(staging.iterdir()) diff --git a/tests/services/test_recipe_persistence.py b/tests/services/test_recipe_persistence.py index 3b2bca23..f56f60ca 100644 --- a/tests/services/test_recipe_persistence.py +++ b/tests/services/test_recipe_persistence.py @@ -223,8 +223,9 @@ async def test_delete_recipe_skips_missing_preview_image(tmp_path: Path) -> None # --------------------------------------------------------------------------- -# (4) bulk_delete with 2 ids -> single batch_id, one batch dir with both -# recipes, re-anchored expires_at in the merged manifest +# (4) bulk_delete with 2 ids -> single batch_id, merged manifest holds all +# recipes, re-anchored expires_at; manifest-only merge: each staged copy +# stays in ITS OWN batch dir (loser dir retained as storage) # --------------------------------------------------------------------------- async def test_bulk_delete_merges_into_single_batch(tmp_path: Path) -> None: scanner = RecipeScannerStub(tmp_path) @@ -245,12 +246,17 @@ async def test_bulk_delete_merges_into_single_batch(tmp_path: Path) -> None: batch_id = result.payload["batch_id"] assert batch_id is not None assert "batch_ids" not in result.payload - assert len(_batch_dirs()) == 1, "loser batch dir must be removed after merge" + # Manifest-only merge: the winner batch dir plus the loser storage dir + # (stamped merged_into) both remain. + assert len(_batch_dirs()) == 2 batch_dir = _staging_parent() / batch_id manifest = json.loads((batch_dir / "manifest.json").read_text(encoding="utf-8")) assert manifest["batch_id"] == batch_id assert len(manifest["entries"]) == 4 + loser_dir = _staging_parent() / next( + d for d in _batch_dirs() if d.name != batch_id + ) # Re-anchored expires_at: now + TTL at merge time (not the earlier of the # two staged expiries). Loose window avoids any timing flakiness. @@ -260,14 +266,27 @@ async def test_bulk_delete_merges_into_single_batch(tmp_path: Path) -> None: <= int(time.time()) + PENDING_DELETE_TTL_SECONDS + 2 ) - # Both recipes' files live under ONE batch dir, byte-identical to originals. + # Both recipes' staged copies live under their OWN batch dirs, byte- + # identical to the originals - no file was ever moved. + winner_files = { + f.name + for f in batch_dir.iterdir() + if f.is_file() and f.name != "manifest.json" + } + loser_files = { + f.name + for f in loser_dir.iterdir() + if f.is_file() and f.name != "manifest.json" + } + assert winner_files == {"ra.recipe.json", "ra.webp"} + assert loser_files == {"rb.recipe.json", "rb.webp"} assert (batch_dir / "ra.recipe.json").read_bytes() == json_a_bytes assert (batch_dir / "ra.webp").read_bytes() == image_a_bytes - assert (batch_dir / "rb.recipe.json").read_bytes() == json_b_bytes - assert (batch_dir / "rb.webp").read_bytes() == image_b_bytes + assert (loser_dir / "rb.recipe.json").read_bytes() == json_b_bytes + assert (loser_dir / "rb.webp").read_bytes() == image_b_bytes + assert manifest.get("merged_sources") == [str(loser_dir)] # Originals removed; both snapshots present. - # Originals removed; merged manifest holds the winner's recipe snapshot. assert not json_a.exists() assert not json_b.exists() assert manifest["recipe_snapshot"] in (data_a, data_b) @@ -288,10 +307,17 @@ async def test_bulk_delete_merge_failure_falls_back_to_batch_ids( scanner.register_recipe("ra", json_a) scanner.register_recipe("rb", json_b) - def failing_rename(src: str, dst: str) -> None: - raise OSError("simulated merge move failure") + # Simulate a merge that cannot resolve the winner batch (the only real + # failure mode since the merge no longer moves files): the caller must + # fall back to the constituent batch_ids array. + from py.services.pending_delete_service import get_pending_delete_service - monkeypatch.setattr("py.services.pending_delete_service.os.rename", failing_rename) + service = await get_pending_delete_service() + + async def _merge_unresolvable(_ids) -> None: + return None + + monkeypatch.setattr(service, "merge_batches", _merge_unresolvable) result = await _make_service().bulk_delete( recipe_scanner=scanner, recipe_ids=["ra", "rb"]