fix(delete): run startup reconciliation walk off the event loop

This commit is contained in:
Will Miao
2026-08-25 17:38:01 +08:00
parent a202c666bc
commit c83b26b556
2 changed files with 59 additions and 6 deletions
+31 -6
View File
@@ -500,10 +500,36 @@ class PendingDeleteService:
QUARANTINE them (preserving the pre-registry sweep semantics). The QUARANTINE them (preserving the pre-registry sweep semantics). The
walk only descends into dirs literally named ``.lm-pending-delete``, walk only descends into dirs literally named ``.lm-pending-delete``,
so false positives are structurally limited. 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 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): if not os.path.isdir(root):
continue continue
visited: Set[str] = set() visited: Set[str] = set()
@@ -518,21 +544,20 @@ class PendingDeleteService:
visited.add(real_dir) visited.add(real_dir)
if os.path.basename(dirpath) == PENDING_DELETE_DIR_NAME: if os.path.basename(dirpath) == PENDING_DELETE_DIR_NAME:
# The current dir IS a staging parent (reachable only when # The current dir IS a staging parent (reachable only when
# a model root itself is one): register its batches. # a model root itself is one): collect its batches.
await self._register_batch_candidates(dirpath) staging_parents.append(dirpath)
dirnames[:] = [] dirnames[:] = []
continue continue
next_dirs: List[str] = [] next_dirs: List[str] = []
for name in dirnames: for name in dirnames:
if name == PENDING_DELETE_DIR_NAME: if name == PENDING_DELETE_DIR_NAME:
await self._register_batch_candidates( staging_parents.append(os.path.join(dirpath, name))
os.path.join(dirpath, name)
)
elif _is_excluded_dir(name): elif _is_excluded_dir(name):
continue continue
else: else:
next_dirs.append(name) next_dirs.append(name)
dirnames[:] = next_dirs dirnames[:] = next_dirs
return staging_parents
async def _register_batch_candidates(self, staging_parent: str) -> None: async def _register_batch_candidates(self, staging_parent: str) -> None:
"""Register every non-orphaned batch subdir of a staging parent.""" """Register every non-orphaned batch subdir of a staging parent."""
@@ -16,6 +16,7 @@ import errno
import json import json
import os import os
import shutil import shutil
import threading
import time import time
from collections.abc import Iterator from collections.abc import Iterator
from pathlib import Path 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) 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) # (h) _find_batch_dir with cleared registry locates + registers (restart sim)
async def test_reg_h_find_batch_dir_restart_simulation( async def test_reg_h_find_batch_dir_restart_simulation(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch tmp_path: Path, monkeypatch: pytest.MonkeyPatch