From c83b26b5564059e9c3c0198620bc2e7127e9eef6 Mon Sep 17 00:00:00 2001 From: Will Miao Date: Tue, 25 Aug 2026 17:38:01 +0800 Subject: [PATCH] fix(delete): run startup reconciliation walk off the event loop --- py/services/pending_delete_service.py | 37 ++++++++++++++++--- tests/services/test_pending_delete_service.py | 28 ++++++++++++++ 2 files changed, 59 insertions(+), 6 deletions(-) diff --git a/py/services/pending_delete_service.py b/py/services/pending_delete_service.py index fb56954f..e92f8804 100644 --- a/py/services/pending_delete_service.py +++ b/py/services/pending_delete_service.py @@ -500,10 +500,36 @@ class PendingDeleteService: QUARANTINE them (preserving the pre-registry sweep semantics). The walk only descends into dirs literally named ``.lm-pending-delete``, so false positives are structurally limited. + + The filesystem walk itself runs in a worker thread so a large or slow + library cannot block the event loop at startup; only the (rare) batch + registration awaits run on the loop. + """ + roots = await self._get_all_model_roots() + loop = asyncio.get_event_loop() + staging_parents = await loop.run_in_executor( + None, # Use default thread pool + self._collect_staging_parents, # Run the tree walk off the loop + roots, + ) + for staging_parent in staging_parents: + await self._register_batch_candidates(staging_parent) + + def _collect_staging_parents(self, roots: Sequence[str]) -> List[str]: + """Walk every model root and return its staging-parent dirs. + + Pure synchronous filesystem discovery with no awaits: walks with + ``followlinks=True, topdown=True``, prunes symlink cycles via a + per-root ``visited`` realpath set (realpath is used ONLY for this + dedup set - the returned paths are the unresolved business paths), + filters out :func:`_is_excluded_dir` dirs, and collects every dir + named ``.lm-pending-delete`` (including the case where a model root + itself is one). Results are returned in walk order. """ from .model_scanner import _is_excluded_dir - for root in await self._get_all_model_roots(): + staging_parents: List[str] = [] + for root in roots: if not os.path.isdir(root): continue visited: Set[str] = set() @@ -518,21 +544,20 @@ class PendingDeleteService: visited.add(real_dir) if os.path.basename(dirpath) == PENDING_DELETE_DIR_NAME: # The current dir IS a staging parent (reachable only when - # a model root itself is one): register its batches. - await self._register_batch_candidates(dirpath) + # a model root itself is one): collect its batches. + staging_parents.append(dirpath) dirnames[:] = [] continue next_dirs: List[str] = [] for name in dirnames: if name == PENDING_DELETE_DIR_NAME: - await self._register_batch_candidates( - os.path.join(dirpath, name) - ) + staging_parents.append(os.path.join(dirpath, name)) elif _is_excluded_dir(name): continue else: next_dirs.append(name) dirnames[:] = next_dirs + return staging_parents async def _register_batch_candidates(self, staging_parent: str) -> None: """Register every non-orphaned batch subdir of a staging parent.""" diff --git a/tests/services/test_pending_delete_service.py b/tests/services/test_pending_delete_service.py index ba23bdd7..7561ba19 100644 --- a/tests/services/test_pending_delete_service.py +++ b/tests/services/test_pending_delete_service.py @@ -16,6 +16,7 @@ import errno import json import os import shutil +import threading import time from collections.abc import Iterator from pathlib import Path @@ -1683,6 +1684,33 @@ async def test_reg_g_reconciliation_finds_external_batches( assert service._known_batch_dirs.get("ext-fresh") == str(fresh_dir) +# (g2) reconciliation runs the filesystem walk off the event loop so a large +# or slow library cannot block startup +async def test_reg_g2_reconciliation_walk_runs_in_worker_thread( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "loras" + root.mkdir() + (root / PENDING_DELETE_DIR_NAME).mkdir() + await _register_model_root(monkeypatch, lora_roots=[root]) + + loop_thread = threading.get_ident() + walk_threads: List[int] = [] + real_walk = os.walk + + def _recording_walk(*args: Any, **kwargs: Any): + walk_threads.append(threading.get_ident()) + return real_walk(*args, **kwargs) + + monkeypatch.setattr(os, "walk", _recording_walk) + + service = await PendingDeleteService.get_instance() + await service._reconcile_scan_roots() + + assert walk_threads, "reconciliation never walked the model roots" + assert all(thread_id != loop_thread for thread_id in walk_threads) + + # (h) _find_batch_dir with cleared registry locates + registers (restart sim) async def test_reg_h_find_batch_dir_restart_simulation( tmp_path: Path, monkeypatch: pytest.MonkeyPatch