diff --git a/docs/technical/reconcile_windows_case_fallback_todo.md b/docs/technical/reconcile_windows_case_fallback_todo.md index c6d18f75..2e7df171 100644 --- a/docs/technical/reconcile_windows_case_fallback_todo.md +++ b/docs/technical/reconcile_windows_case_fallback_todo.md @@ -1,7 +1,7 @@ -# Reconcile 的 Windows 大小写回退分支 - 待验证清单 +# Reconcile 的 Windows 大小写回退分支 -> **状态**: 待 Windows 环境验证 | **创建日期**: 2026-09-11 -> **相关文件**: `py/services/model_scanner.py` (`ModelScanner._reconcile_cache`) +> **状态**: 已改为 O(1) 索引（2026-10-06），不再需要 Windows 机器验证 | **创建日期**: 2026-09-11 +> **相关文件**: `py/services/model_scanner.py`（`_walk_root_for_reconcile` / `_CachedPathLookups`） > **相关历史**: #871 (`76ee59cd`, 路径重叠去重)、#1108 (按文件夹扫描的需求) --- @@ -9,84 +9,70 @@ ## 背景 Refresh 按钮走的是 `_reconcile_cache()`（快速增量对账）。2026-09-11 做了一轮性能优化，把两处"预防性"的 -realpath 全量遍历改成按需触发（详见下方"已完成"）。优化后，一次零变更 Refresh 在 5 万文件库上从 -~1400 ms 降到 ~120 ms。 - -清理过程中发现**唯一一处遗留的可疑点**：Windows 专属的大小写不敏感回退分支。它无法在 Linux 上验证， -因此单独记录，留待 Windows 机器上确认。 +realpath 全量遍历改成按需触发。清理时留下的唯一可疑点，是 Windows 专属的大小写不敏感回退分支：它排在 +精确匹配和 realpath 别名匹配之后，只有**未命中**的文件才会走到，但一旦走到就是 O(文件数 × 缓存条目数)。 --- -## 待验证分支（现状） +## 现状（已改造） -`py/services/model_scanner.py` 中 `_reconcile_cache()` 的 walk 循环内： +分支语义保持不变，查找改为 O(1)： ```python -# Try case-insensitive match on Windows -if os.name == 'nt': - lower_path = file_path.lower() - matched = False - for cached_path in cached_paths: # 每个未命中文件都全量扫一遍缓存 - if cached_path.lower() == lower_path: - found_paths.add(cached_path) - matched = True - break - if matched: - continue +if _CASE_INSENSITIVE_PATHS: # 模块级常量，默认 os.name == "nt" + cached_case_match = lookups.match_casefold_path(file_path) ``` -它排在精确匹配（`file_path in cached_paths`）和 realpath 别名匹配之后，只有**未命中**的文件才会走到。 +`lookups` 是 `_CachedPathLookups`：小写索引 `{lower(cached_path): cached_path}` 在第一次未命中时构建一次 +（与 realpath 别名索引同样懒构建，并用 `threading.Lock` 护住——walk 现在跑在工作线程里），之后每次未命中 +只做一次字典查询。 -### 为什么可疑 - -1. **可能不可达**：Windows 上 `os.path.realpath()` 会返回磁盘上的真实大小写，因此"缓存路径大小写与磁盘 - 不一致"的情形，理论上已经被上一步的 realpath 别名匹配覆盖。若如此，这段就是纯冗余代码。 -2. **一旦可达就是 O(N×M)**：每个未命中文件都要遍历全部 `cached_paths` 做小写比较。若某种路径写法让 - 整个库都变成"未命中"（例如缓存里的盘符/大小写形式与 walk 结果系统性不一致），一次 Refresh 会退化 - 成 文件数 × 缓存条目数 次字符串比较，比真实 IO 还贵。 -3. **没有测试覆盖**：`tests/services/test_model_scanner.py` 没有任何针对该分支的用例（它在 Linux 上 - 被 `os.name == 'nt'` 短路，无法覆盖）。 +`_CASE_INSENSITIVE_PATHS` 提成模块级常量的原因：这条分支在 Linux 上原本被 `os.name == 'nt'` 短路、没有任何 +测试覆盖；现在测试可以 monkeypatch 该常量，在 Linux 上真实执行这条分支。 --- ## 待办 -- [ ] **验证可达性**：在 Windows 上构造"缓存路径与磁盘真实大小写不一致"的场景，确认 realpath 别名匹配 - 是否已经命中，即上面的 `if os.name == 'nt'` 分支是否还有进入的必要。 -- [ ] **若不可达 / 冗余**：删除该分支，并在删除处留注释说明 realpath 已覆盖大小写归一（附验证记录）。 -- [ ] **若可达**：保留语义但改成 O(1)——预先构建一次 `lower_path -> cached_path` 映射（与 - `cached_real_paths` 同样按需、懒构建），把内层全量扫描换成一次字典查询。 -- [ ] **补一个 Windows-only 的回归测试**（`pytest.mark.skipif(os.name != "nt", ...)`），锁定最终结论。 -- [ ] 把验证结论回填到本文件，并同步更新状态行。 +- [x] **消除 O(N×M)**：改成懒构建的小写索引，查询降为 O(1)。 +- [x] **补回归测试**：`tests/services/test_model_scanner.py::test_reconcile_case_fold_fallback_is_indexed_not_linear` + （缓存路径与磁盘仅大小写不同、realpath 别名不命中 → 断言条目保留、不重新处理、索引只构建一次）。 +- [x] 回填本文件。 +- [ ] （可选，纯代码瘦身）在 Windows 上确认 realpath 别名匹配是否已覆盖全部情形；若确认该分支不可达， + 可以整体删掉这一层。它现在的成本已经可以忽略，删除不再是性能问题。 --- -## 验证方法（Windows） +## 验证方法（可选，Windows） -1. **构造不一致的大小写**：让缓存里的 `file_path` 与磁盘实际路径大小写不同（例如改过盘符/目录大小写， - 或从另一台机器迁移了 `settings.json` 与持久化缓存），然后在 UI 点 Refresh。 -2. **看后端日志判据**： - - 若 realpath 已覆盖 → 日志应显示 `Cache reconciliation completed in X seconds. Added 0, removed 0 models.`， - 且**没有** `Found N new files to process` / `Processing `。 - - 若回退分支在起作用 → 同样应该是 `Added 0, removed 0`（因为 `found_paths` 被补上），这是"分支可达" - 的证据；反之若出现大量 `Processing ...` 并重新 hash，说明连回退分支也没命中，问题更严重 - （缓存路径被当成了新文件 + 旧条目被删）。 -3. **跑测试**：`python -m pytest tests/services/test_model_scanner.py -k reconcile`（该文件在 Windows 上会 - 真实执行 `os.name == 'nt'` 分支）。 -4. **量化**：如果需要，可在 `_reconcile_cache` 里临时插桩统计该分支的进入次数与内层迭代次数，确认是否为 0。 +1. 构造"缓存路径与磁盘真实大小写不一致"的场景（改过盘符大小写、迁移过 `settings.json` 与持久化缓存），点 Refresh。 +2. 日志判据：`Cache reconciliation completed in X seconds. Added 0, removed 0 models.`，且**没有** + `Found N new files to process` / `Processing `。 +3. 跑测试：`python -m pytest tests/services/test_model_scanner.py -k reconcile`。 --- -## 已完成（本轮优化，供对照） +## 已完成（历史，供对照） -同一次清理里已经落地并验证的部分（Linux，5 万文件库）： +2026-09-11 那轮清理里在 Linux（5 万文件库）验证过的部分： - `cached_real_paths` 别名映射改为**首次未命中时**懒构建（原来每次 Refresh 都对全部缓存条目算一次 realpath）。 - 每个文件的 `realpath` 移到精确命中检查**之后**（原来对每个文件都算，命中即丢弃）。 - `get_model_roots()` 在新增文件处理阶段只快照一次（原来每个新文件重读一次）。 -- 全量去重 pass 加了 O(1) 前置判断（`cached_size_before != len(cached_paths) or total_added > 0`）， - 零变更且缓存干净时跳过；快照本身含重复路径时仍会自愈。 +- 全量去重 pass 加了 O(1) 前置判断（`cached_size_before != len(cached_paths) or total_added > 0`）。 结果：零变更 Refresh 5 万文件 **~1400 ms → ~120 ms**；根目录顺序/符号链接别名翻转场景仍是 -`re-processed=0`（不重新读 metadata、不重新 hash）。测试：`tests/services/test_model_scanner.py` -47 项、全量后端 2567 项全部通过。 +`re-processed=0`。 + +--- + +## 2026-10-06 追加：walk 阶段的并发与进度 + +同一次改动还做了两件事（与大小写分支无关，但都动到了同一段 walk 循环，故一并记录）： + +- walk 循环抽成同步函数 `_walk_root_for_reconcile()`，并按设备分组 + （`_root_device_key()` / `_group_roots_by_device()`）在工作线程里并行执行：同一设备内的 root 仍按配置顺序 + 串行（保证目录认领与去重的确定性），不同设备之间才并行。结果回到事件循环后按**配置的 root 顺序**合并， + 因此"同一个文件可达多条业务路径时谁胜出"与并发完成顺序无关。 +- walk 阶段按 root 广播进度：`_ReconcileWalkTracker` 以「该 root 的缓存条目数」为权重估算进度（真实文件数 + 只有走完才知道），进度条把 walk 记为 0-50%，新增文件处理阶段顺延为 50-99%，两段之间不会回退。 diff --git a/locales/de.json b/locales/de.json index 465bf26f..f7aee277 100644 --- a/locales/de.json +++ b/locales/de.json @@ -73,7 +73,8 @@ "lessThanMinute": "Weniger als eine Minute verbleibend", "minutes": "~{minutes} Min. verbleibend", "hours": "~{hours} Std. {minutes} Min. verbleibend" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/locales/en.json b/locales/en.json index c29ade8e..86f504d1 100644 --- a/locales/en.json +++ b/locales/en.json @@ -73,7 +73,8 @@ "lessThanMinute": "Less than a minute remaining", "minutes": "~{minutes} min remaining", "hours": "~{hours} hr {minutes} min remaining" - } + }, + "walkFiles": "{count} files" } }, "onboarding": { diff --git a/locales/es.json b/locales/es.json index 21285964..0a0b2ba4 100644 --- a/locales/es.json +++ b/locales/es.json @@ -73,7 +73,8 @@ "lessThanMinute": "Queda menos de un minuto", "minutes": "Quedan ~{minutes} min", "hours": "Quedan ~{hours} h {minutes} min" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/locales/fr.json b/locales/fr.json index 366e35d8..e5e6b251 100644 --- a/locales/fr.json +++ b/locales/fr.json @@ -73,7 +73,8 @@ "lessThanMinute": "Moins d’une minute restante", "minutes": "~{minutes} min restantes", "hours": "~{hours} h {minutes} min restantes" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/locales/he.json b/locales/he.json index 8103085f..7813caea 100644 --- a/locales/he.json +++ b/locales/he.json @@ -73,7 +73,8 @@ "lessThanMinute": "נותרה פחות מדקה", "minutes": "נותרו ~{minutes} דקות", "hours": "נותרו ~{hours} שעות ו-{minutes} דקות" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/locales/ja.json b/locales/ja.json index 722caea0..85d45581 100644 --- a/locales/ja.json +++ b/locales/ja.json @@ -73,7 +73,8 @@ "lessThanMinute": "残り1分未満", "minutes": "残り約 {minutes} 分", "hours": "残り約 {hours} 時間 {minutes} 分" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/locales/ko.json b/locales/ko.json index d8549300..04a226a9 100644 --- a/locales/ko.json +++ b/locales/ko.json @@ -73,7 +73,8 @@ "lessThanMinute": "남은 시간 1분 미만", "minutes": "약 {minutes}분 남음", "hours": "약 {hours}시간 {minutes}분 남음" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/locales/ru.json b/locales/ru.json index 7de2cb90..5537bff4 100644 --- a/locales/ru.json +++ b/locales/ru.json @@ -73,7 +73,8 @@ "lessThanMinute": "Осталось меньше минуты", "minutes": "Осталось ~{minutes} мин", "hours": "Осталось ~{hours} ч {minutes} мин" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/locales/zh-CN.json b/locales/zh-CN.json index ce63542a..f76dbc12 100644 --- a/locales/zh-CN.json +++ b/locales/zh-CN.json @@ -73,7 +73,8 @@ "lessThanMinute": "剩余时间不到一分钟", "minutes": "剩余约 {minutes} 分钟", "hours": "剩余约 {hours} 小时 {minutes} 分钟" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/locales/zh-TW.json b/locales/zh-TW.json index 9695e40b..8aee04d7 100644 --- a/locales/zh-TW.json +++ b/locales/zh-TW.json @@ -73,7 +73,8 @@ "lessThanMinute": "剩餘時間不到一分鐘", "minutes": "剩餘約 {minutes} 分鐘", "hours": "剩餘約 {hours} 小時 {minutes} 分鐘" - } + }, + "walkFiles": "[TODO: Translate] {count} files" } }, "onboarding": { diff --git a/py/services/model_scanner.py b/py/services/model_scanner.py index 7724e1cb..5ee78291 100644 --- a/py/services/model_scanner.py +++ b/py/services/model_scanner.py @@ -2,10 +2,11 @@ import json import os import logging import asyncio +import threading import time import shutil -from contextlib import asynccontextmanager -from dataclasses import dataclass +from contextlib import asynccontextmanager, suppress +from dataclasses import dataclass, field from typing import Any, Awaitable, Callable, Dict, List, Mapping, Optional, Sequence, Set, Tuple, Type, Union, cast from ..utils.models import BaseModelMetadata, autov3_from_civitai_files @@ -89,6 +90,388 @@ PAGE_TYPE_MAP = { 'other': 'other', } +# Case-insensitive cache matching is a Windows-only safety net (NTFS treats +# "Model.txt" and "model.TXT" as the same file). Module-level so the branch can +# be exercised on case-sensitive filesystems in tests. +_CASE_INSENSITIVE_PATHS: bool = os.name == "nt" + +# Cadence of walk-phase progress broadcasts during a reconcile. +_WALK_PROGRESS_INTERVAL_SECONDS = 0.5 + +# Share of the progress bar owned by the reconcile walk. The new-file pass +# takes the rest (50-99) so the bar never jumps backwards between phases. +_WALK_PROGRESS_SHARE = 50 + +# How many files a walk worker may scan between two progress reports. +_WALK_PROGRESS_REPORT_EVERY = 256 + + +def _new_file_pass_progress(processed: int, total: int) -> int: + """Bar percentage for the new-file pass (walk share .. 99).""" + if total <= 0: + return _WALK_PROGRESS_SHARE + ratio = min(max(processed / total, 0.0), 1.0) + return _WALK_PROGRESS_SHARE + min(49, int(ratio * 49)) + + +def _root_display_label(root: str) -> str: + """Short label identifying a model root in progress messages.""" + drive, _tail = os.path.splitdrive(root) + if drive: + return drive + normalized = root.rstrip("/\\") + return os.path.basename(normalized) or root + + +def _normalized_root_prefix(root: str) -> str: + """Forward-slash root prefix used to attribute cached paths to a root.""" + prefix = os.path.normpath(root).replace(os.sep, "/") + if not prefix.endswith("/"): + prefix += "/" + return prefix.lower() if _CASE_INSENSITIVE_PATHS else prefix + + +def _count_cached_entries_per_root( + cached_paths: Set[str], roots: Sequence[str] +) -> Dict[str, int]: + """Attribute cached entries to model roots (longest prefix wins). + + Used as the walk-workload weight for progress reporting: the walk itself is + what discovers the real file count, so the cached entry count is the only + estimate available up front. + """ + counts: Dict[str, int] = {root: 0 for root in roots} + if not roots: + return counts + + prefixes = sorted( + ((_normalized_root_prefix(root), root) for root in roots), + key=lambda item: len(item[0]), + reverse=True, + ) + case_insensitive = _CASE_INSENSITIVE_PATHS + for path in cached_paths: + candidate = path.lower() if case_insensitive else path + for prefix, root in prefixes: + if candidate.startswith(prefix): + counts[root] += 1 + break + return counts + + +def _root_device_key(root: str) -> str: + """Group key identifying the storage device that holds a root. + + Roots sharing a device must be walked sequentially so directory claims stay + deterministic (configured order wins), while roots on different devices are + independent and can be walked by parallel workers. + """ + drive, _tail = os.path.splitdrive(root) + if drive: + return drive.lower() + try: + return f"dev:{os.stat(root).st_dev}" + except OSError: + return f"path:{os.path.normpath(root)}" + + +def _group_roots_by_device(roots: Sequence[str]) -> List[List[str]]: + """Group roots by device, preserving the configured root order. + + Roots on distinct devices cannot alias unless a symlink crosses devices, so + walking them in parallel cannot disturb the directory-claim order in any + realistic layout (and even then the realpath alias index keeps the cached + business path stable). + """ + groups: Dict[str, List[str]] = {} + for root in roots: + groups.setdefault(_root_device_key(root), []).append(root) + return list(groups.values()) + + +def _build_realpath_index(cached_paths: Set[str]) -> Dict[str, str]: + """Map realpath -> cached business path (overlapping roots / symlinks).""" + index: Dict[str, str] = {} + for cached_path in cached_paths: + try: + index.setdefault(os.path.realpath(cached_path), cached_path) + except Exception: + continue + return index + + +def _build_casefold_index(cached_paths: Set[str]) -> Dict[str, str]: + """Map lower-cased path -> cached business path (Windows case fallback).""" + index: Dict[str, str] = {} + for cached_path in cached_paths: + index.setdefault(cached_path.lower(), cached_path) + return index + + +class _CachedPathLookups: + """Lazily built path indexes over the cached paths, safe for workers. + + Both indexes are only ever consulted for walk *misses*, so building them on + first use keeps a no-change refresh free of per-cached-entry realpath work. + The case-fold index replaces the former linear scan over every cached path, + which was O(files x cached entries) whenever it was reached. + """ + + def __init__(self, cached_paths: Set[str]) -> None: + self.cached_paths = cached_paths + self._lock = threading.Lock() + self._realpath_index: Optional[Dict[str, str]] = None + self._casefold_index: Optional[Dict[str, str]] = None + + def match_real_path(self, real_path: str) -> Optional[str]: + index = self._realpath_index + if index is None: + with self._lock: + if self._realpath_index is None: + self._realpath_index = _build_realpath_index(self.cached_paths) + index = self._realpath_index + return index.get(real_path) + + def match_casefold_path(self, file_path: str) -> Optional[str]: + index = self._casefold_index + if index is None: + with self._lock: + if self._casefold_index is None: + self._casefold_index = _build_casefold_index(self.cached_paths) + index = self._casefold_index + return index.get(file_path.lower()) + + +class _RealDirClaims: + """Thread-safe claim set for real directory paths. + + Mirrors the historical global ``visited_real_paths`` set: a directory + reachable through several roots (overlapping roots, symlinked aliases) must + only be walked once, otherwise the same physical file could end up in the + library under two business paths. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self._claimed: Set[str] = set() + + def claim(self, real_path: str) -> bool: + with self._lock: + if real_path in self._claimed: + return False + self._claimed.add(real_path) + return True + + +class _ReconcileWalkTracker: + """Thread-safe progress bookkeeping for the reconcile walk. + + Per-root progress is weighted by the number of cached entries under that + root: the walk is what discovers the real file count, so the cached count + is the only workload estimate available before it runs. Roots with no + cached entries contribute a single weight unit and are reported only once + they finish. + """ + + def __init__(self, roots: Sequence[str], expected: Mapping[str, int]) -> None: + self._lock = threading.Lock() + self._roots: List[str] = list(roots) + self._labels = {root: _root_display_label(root) for root in self._roots} + self._expected = { + root: max(int(expected.get(root, 0) or 0), 0) for root in self._roots + } + self._weight = { + root: max(self._expected[root], 1) for root in self._roots + } + self._seen = {root: 0 for root in self._roots} + self._done = {root: False for root in self._roots} + self._active: Set[str] = set() + + def start_root(self, root: str) -> None: + with self._lock: + if root in self._seen: + self._active.add(root) + + def report(self, root: str, files_seen: int) -> None: + with self._lock: + if root in self._seen: + self._seen[root] = max(int(files_seen), 0) + + def finish_root(self, root: str) -> None: + with self._lock: + self._done[root] = True + self._active.discard(root) + + def snapshot(self) -> Optional[Dict[str, Any]]: + """Return the current payload extras plus the bar percentage.""" + with self._lock: + if not self._roots: + return None + + total_weight = sum(self._weight.values()) or 1 + completed = 0.0 + files_seen = 0 + roots_done = 0 + for root in self._roots: + files_seen += self._seen[root] + if self._done[root]: + completed += self._weight[root] + roots_done += 1 + continue + expected = self._expected[root] + if expected > 0: + completed += self._weight[root] * min( + self._seen[root] / expected, 0.99 + ) + + active = [root for root in self._roots if root in self._active] + if active: + current = max(active, key=lambda root: self._seen[root]) + else: + current = self._roots[-1] + + expected_total = sum(self._expected.values()) + ratio = min(completed / total_weight, 1.0) + return { + 'progress': min(_WALK_PROGRESS_SHARE, int(ratio * _WALK_PROGRESS_SHARE)), + # processed/total feed the client-side ETA; the total is a lower + # bound and never below the files already seen, so the estimate + # stays coherent when the library grew since the last scan. + 'processed': files_seen, + 'total': max(expected_total, files_seen), + 'files_seen': files_seen, + 'roots_total': len(self._roots), + 'roots_done': roots_done, + 'active_roots': [self._labels[root] for root in active][:4], + 'current_name': self._labels[current], + } + + +@dataclass +class _RootWalkResult: + """Outcome of walking a single model root (no scanner state touched).""" + + root_path: str + found_paths: Set[str] = field(default_factory=set) + stale_paths: List[str] = field(default_factory=list) + # (business path, real path) pairs for files that are neither cached nor + # excluded; the caller claims real paths in configured root order. + new_candidates: List[Tuple[str, str]] = field(default_factory=list) + discovered_folders: Set[str] = field(default_factory=set) + files_seen: int = 0 + cancelled: bool = False + + +def _walk_root_for_reconcile( + *, + root_path: str, + file_extensions: Set[str], + cached_paths: Set[str], + path_to_item: Mapping[str, Dict[str, Any]], + lookups: _CachedPathLookups, + dir_claims: _RealDirClaims, + excluded_models: Set[str], + is_cancelled: Callable[[], bool], + report_progress: Callable[[int], None], +) -> _RootWalkResult: + """Walk one model root and classify every model file found. + + Runs synchronously (worker thread) and never mutates scanner state: all + cache updates happen on the event loop once every root has been walked. + """ + result = _RootWalkResult(root_path=root_path) + stale_seen: Set[str] = set() + files_since_report = 0 + + def mark_stale_if_needed(cached_path: str) -> None: + """Queue a cached path for file_name repair when it drifted.""" + if cached_path in stale_seen: + return + item = path_to_item.get(cached_path) + if item is None: + return + if item.get("file_name") == _file_name_stem(cached_path): + return + stale_seen.add(cached_path) + result.stale_paths.append(cached_path) + + for root, dirnames, files in os.walk(root_path, followlinks=True): + dirnames[:] = [d for d in dirnames if not _is_excluded_dir(d)] + + real_root = os.path.realpath(root) + if not dir_claims.claim(real_root): + continue + + # Record every visited directory (including empty ones) so the folder + # tree stays accurate without a live walk. + rel_dir = os.path.relpath( + os.path.abspath(root), os.path.abspath(root_path) + ).replace(os.path.sep, "/") + if rel_dir != "." and not _is_hidden_relative_path(rel_dir): + result.discovered_folders.add(rel_dir) + + for file in files: + ext = os.path.splitext(file)[1].lower() + if ext not in file_extensions: + continue + + result.files_seen += 1 + files_since_report += 1 + if files_since_report >= _WALK_PROGRESS_REPORT_EVERY: + # A flat library (thousands of files in one directory) would + # otherwise report nothing until that directory ends. + files_since_report = 0 + report_progress(result.files_seen) + + # Construct paths exactly as they would be in cache + file_path = os.path.join(root, file).replace(os.sep, '/') + + if file_path in cached_paths: + result.found_paths.add(file_path) + mark_stale_if_needed(file_path) + continue + + # Only a cache miss needs the physical path, so the realpath + # syscalls are paid per changed file rather than per file in the + # library. + real_file_path = os.path.realpath(os.path.join(root, file)) + + cached_real_match = lookups.match_real_path(real_file_path) + if cached_real_match: + result.found_paths.add(cached_real_match) + mark_stale_if_needed(cached_real_match) + continue + + if file_path in excluded_models: + continue + + # Windows: a cached path may differ from the walk result only by + # case. O(1) via the lower-cased index (was a full scan of the + # cached paths per miss). + if _CASE_INSENSITIVE_PATHS: + cached_case_match = lookups.match_casefold_path(file_path) + if cached_case_match: + result.found_paths.add(cached_case_match) + mark_stale_if_needed(cached_case_match) + continue + + # Not cached yet; the caller claims the real path in root order so + # a file reachable through several roots is only added once. + result.new_candidates.append((file_path, real_file_path)) + + if files_since_report: + # Publish per directory so a library spread over many directories + # reports as it progresses, not only at the end of the walk. + files_since_report = 0 + report_progress(result.files_seen) + + if is_cancelled(): + result.cancelled = True + break + + report_progress(result.files_seen) + return result + def _is_pending_delete_path(path: str) -> bool: """Return True when any path component is the pending-delete staging dir.""" @@ -1082,129 +1465,76 @@ class ModelScanner: cached_paths = {item['file_path'] for item in self._cache.raw_data} path_to_item = {item['file_path']: item for item in self._cache.raw_data} - # physical path -> cached business path, for the alias case where the - # same file is reachable under a different path than the cached one - # (overlapping roots / symlink layout changes): keep the existing - # entry instead of delete + re-add (which would re-read metadata and - # re-hash every file). Built lazily on the first miss, because a - # realpath per cached entry is ~half the cost of a no-change - # reconcile and the map is only ever consulted for misses. - cached_real_paths: Optional[Dict[str, str]] = None + # Every configured root that is currently reachable. A root that is + # missing (drive switched off, unmounted share) is skipped, so its + # cached entries are reported as missing below. + roots: List[str] = [] + seen_roots: Set[str] = set() + for root_path in self.get_model_roots(): + if not root_path or root_path in seen_roots: + continue + if not os.path.exists(root_path): + continue + seen_roots.add(root_path) + roots.append(root_path) - def lookup_cached_real_path(real_path: str) -> Optional[str]: - nonlocal cached_real_paths - if cached_real_paths is None: - cached_real_paths = {} - for cached_path in cached_paths: - try: - cached_real_paths.setdefault(os.path.realpath(cached_path), cached_path) - except Exception: - continue - return cached_real_paths.get(real_path) - - # Track found files and new files - found_paths = set() - new_files = [] + # Roots on different devices are walked by parallel workers (a cold + # or slow drive then no longer serializes the others); roots sharing + # a device stay sequential so directory claims remain deterministic. + tracker = _ReconcileWalkTracker( + roots, _count_cached_entries_per_root(cached_paths, roots) + ) + walk_results = await self._walk_roots_for_reconcile( + roots=roots, + tracker=tracker, + cached_paths=cached_paths, + path_to_item=path_to_item, + ) + + # Final walk snapshot: the bar reaches the walk share, then the + # new-file pass continues from there. + await self._broadcast_walk_progress(tracker) + + if self.is_cancelled(): + logger.info(f"{self.model_type.capitalize()} Scanner: Reconcile scan cancelled") + await self._broadcast_scan_progress( + 'cancelled', 'reconcile_scan', 0, False, + elapsed_seconds=time.time() - start_time, + ) + return + + # Merge the per-root results in configured root order: which + # business path wins a file reachable through several roots (and + # which cached entry counts as found) must not depend on the order + # the workers happened to finish in. + found_paths: Set[str] = set() + discovered_folders: Set[str] = set() # Cached entries whose stored file_name no longer matches the file - # on disk (e.g. dotted stems truncated by the legacy .civitai.info + # on disk (e.g. dotted stems truncated by the legacy .civitai_info # migration, issue #1112). Repaired in place after the walk; the # list stays empty on a clean library, so a no-change reconcile # only pays one string compare per cached file. stale_paths: List[str] = [] stale_seen: Set[str] = set() + new_files: List[str] = [] + discovered_real_files: Set[str] = set() - def mark_stale_if_needed(cached_path: str) -> None: - """Queue a cached path for file_name repair when it drifted.""" - if cached_path in stale_seen: - return - item = path_to_item.get(cached_path) - if item is None: - return - if item.get("file_name") == _file_name_stem(cached_path): - return - stale_seen.add(cached_path) - stale_paths.append(cached_path) - - visited_real_paths = set() - discovered_real_files = set() - discovered_folders: Set[str] = set() - - # Scan all model roots - for root_path in self.get_model_roots(): - if not os.path.exists(root_path): - continue - - # Recursively scan directory - for root, dirnames, files in os.walk(root_path, followlinks=True): - dirnames[:] = [d for d in dirnames if not _is_excluded_dir(d)] - real_root = os.path.realpath(root) - if real_root in visited_real_paths: + for result in walk_results: + found_paths |= result.found_paths + discovered_folders |= result.discovered_folders + for cached_path in result.stale_paths: + if cached_path in stale_seen: continue - visited_real_paths.add(real_root) + stale_seen.add(cached_path) + stale_paths.append(cached_path) - # Record every visited directory (including empty ones) so - # the folder tree stays accurate without a live walk. - rel_dir = os.path.relpath( - os.path.abspath(root), os.path.abspath(root_path) - ).replace(os.path.sep, "/") - if rel_dir != "." and not _is_hidden_relative_path(rel_dir): - discovered_folders.add(rel_dir) - - for file in files: - ext = os.path.splitext(file)[1].lower() - if ext in self.file_extensions: - # Construct paths exactly as they would be in cache - file_path = os.path.join(root, file).replace(os.sep, '/') - - # Check if this file is already in cache - if file_path in cached_paths: - found_paths.add(file_path) - mark_stale_if_needed(file_path) - continue - - # Only a cache miss needs the physical path, so the - # realpath syscalls are paid per changed file rather - # than per file in the library. - real_file_path = os.path.realpath(os.path.join(root, file)) - - cached_real_match = lookup_cached_real_path(real_file_path) - if cached_real_match: - found_paths.add(cached_real_match) - mark_stale_if_needed(cached_real_match) - continue - - if file_path in self._excluded_models: - continue - - # Try case-insensitive match on Windows - if os.name == 'nt': - lower_path = file_path.lower() - matched = False - for cached_path in cached_paths: - if cached_path.lower() == lower_path: - found_paths.add(cached_path) - mark_stale_if_needed(cached_path) - matched = True - break - if matched: - continue - - if real_file_path in discovered_real_files: - continue - - discovered_real_files.add(real_file_path) - # This is a new file to process - new_files.append(file_path) - - # Yield control periodically - await asyncio.sleep(0) - if self.is_cancelled(): - logger.info(f"{self.model_type.capitalize()} Scanner: Reconcile scan cancelled") - await self._broadcast_scan_progress( - 'cancelled', 'reconcile_scan', 0, False, - elapsed_seconds=time.time() - start_time, - ) - return + for result in walk_results: + for file_path, real_file_path in result.new_candidates: + if real_file_path in discovered_real_files: + continue + discovered_real_files.add(real_file_path) + # This is a new file to process + new_files.append(file_path) # Process new files in batches total_added = 0 @@ -1281,7 +1611,7 @@ class ModelScanner: last_progress_time = current_time await self._broadcast_scan_progress( 'processing', 'process_new', - min(99, int(1 + (processed_new / total_new) * 98)), False, + _new_file_pass_progress(processed_new, total_new), False, processed=processed_new, total=total_new, current_name=os.path.basename(path), ) @@ -1290,7 +1620,7 @@ class ModelScanner: logger.info(f"{self.model_type.capitalize()} Scanner: Reconcile processing cancelled") await self._broadcast_scan_progress( 'cancelled', 'process_new', - min(99, int(1 + (processed_new / total_new) * 98)), False, + _new_file_pass_progress(processed_new, total_new), False, elapsed_seconds=time.time() - start_time, ) return @@ -1444,7 +1774,117 @@ class ModelScanner: finally: self._is_initializing = False # Unset flag self.bump_cache_version() - + + async def _walk_roots_for_reconcile( + self, + *, + roots: List[str], + tracker: _ReconcileWalkTracker, + cached_paths: Set[str], + path_to_item: Mapping[str, Dict[str, Any]], + ) -> List[_RootWalkResult]: + """Walk every root off the event loop and return results in root order. + + One worker per device: cold or slow drives no longer serialize the + others, and the event loop stays free for the whole walk (previously it + was re-entered once per directory). Results are re-ordered to the + configured root order so the merge below is deterministic. + """ + if not roots: + return [] + + lookups = _CachedPathLookups(cached_paths) + dir_claims = _RealDirClaims() + excluded_models = set(self._excluded_models) + loop = asyncio.get_running_loop() + + monitor = asyncio.create_task(self._monitor_walk_progress(tracker)) + try: + futures = [ + loop.run_in_executor( + None, + self._walk_root_group_sync, + group, + tracker, + lookups, + dir_claims, + excluded_models, + path_to_item, + ) + for group in _group_roots_by_device(roots) + ] + outcomes = await asyncio.gather(*futures, return_exceptions=True) + finally: + monitor.cancel() + with suppress(asyncio.CancelledError, Exception): + await monitor + + for outcome in outcomes: + if isinstance(outcome, BaseException): + raise outcome + + by_root: Dict[str, _RootWalkResult] = {} + for outcome in outcomes: + for result in cast(List[_RootWalkResult], outcome): + by_root[result.root_path] = result + return [by_root[root] for root in roots if root in by_root] + + def _walk_root_group_sync( + self, + roots: List[str], + tracker: _ReconcileWalkTracker, + lookups: _CachedPathLookups, + dir_claims: _RealDirClaims, + excluded_models: Set[str], + path_to_item: Mapping[str, Dict[str, Any]], + ) -> List[_RootWalkResult]: + """Walk the roots of one device sequentially (worker-thread entry point).""" + results: List[_RootWalkResult] = [] + for root_path in roots: + if self.is_cancelled(): + break + + tracker.start_root(root_path) + + def report_progress(files_seen: int, _root: str = root_path) -> None: + tracker.report(_root, files_seen) + + try: + result = _walk_root_for_reconcile( + root_path=root_path, + file_extensions=self.file_extensions, + cached_paths=lookups.cached_paths, + path_to_item=path_to_item, + lookups=lookups, + dir_claims=dir_claims, + excluded_models=excluded_models, + is_cancelled=self.is_cancelled, + report_progress=report_progress, + ) + finally: + tracker.finish_root(root_path) + + results.append(result) + if result.cancelled: + break + return results + + async def _monitor_walk_progress(self, tracker: _ReconcileWalkTracker) -> None: + """Broadcast walk progress while the root workers are running.""" + while True: + await asyncio.sleep(_WALK_PROGRESS_INTERVAL_SECONDS) + await self._broadcast_walk_progress(tracker) + + async def _broadcast_walk_progress(self, tracker: _ReconcileWalkTracker) -> None: + """Emit one walk-phase progress message (best effort).""" + snapshot = tracker.snapshot() + if not snapshot: + return + progress = int(snapshot.pop('progress', 0)) + await self._broadcast_scan_progress( + 'processing', 'reconcile_scan', progress, False, **snapshot + ) + def is_initializing(self) -> bool: """Check if the scanner is currently initializing""" return self._is_initializing diff --git a/static/js/api/baseModelApi.js b/static/js/api/baseModelApi.js index 5030869d..d477b123 100644 --- a/static/js/api/baseModelApi.js +++ b/static/js/api/baseModelApi.js @@ -532,17 +532,48 @@ export class BaseModelApiClient { ); const etaTracker = createScanEtaTracker(); let ws = null; + let lastProgressStage = null; const handleScanProgress = (data) => { if (typeof data.progress === 'number') { state.loadingManager.setProgress(data.progress); } + // The per-file rate of one stage says nothing about the next one: + // the walk phase counts files while the new-file pass processes + // them, so a carried-over average would produce a nonsense ETA. + if (data.stage && data.stage !== lastProgressStage) { + lastProgressStage = data.stage; + etaTracker.reset(); + } let statusText = translate( `common.scanProgress.stages.${data.stage}`, { total: data.total }, data.stage || '' ); - if (data.status === 'processing' && data.total > 0) { + if (data.status === 'processing' && data.stage === 'reconcile_scan') { + // Walk phase: the real file count is only known once the walk + // finishes, so report the files checked so far and the roots + // being walked instead of a processed/total ratio. + const filesSeen = Number(data.files_seen) || 0; + if (filesSeen > 0) { + const roots = Array.isArray(data.active_roots) + ? data.active_roots.filter(Boolean) + : []; + const formattedFiles = filesSeen.toLocaleString(); + const filesText = translate( + 'common.scanProgress.walkFiles', + { count: formattedFiles }, + `${formattedFiles} files` + ); + statusText += roots.length + ? ` ${roots.join(', ')} (${filesText})` + : ` (${filesText})`; + const etaText = etaTracker.update(data.processed, data.total); + if (etaText) { + statusText += ` | ${etaText}`; + } + } + } else if (data.status === 'processing' && data.total > 0) { statusText += ` (${data.processed}/${data.total})`; if (data.current_name) { statusText += ` ${data.current_name}`; diff --git a/static/js/utils/scanEtaUtils.js b/static/js/utils/scanEtaUtils.js index 7af13e07..54852938 100644 --- a/static/js/utils/scanEtaUtils.js +++ b/static/js/utils/scanEtaUtils.js @@ -22,7 +22,7 @@ export function formatScanRemainingTime(remainingMs) { * Create an ETA tracker for scan progress. Uses an exponential moving * average (0.7/0.3) over the observed per-file processing time, mirroring * the estimator in components/initialization.js. - * @returns {{ update: (processed: number, total: number) => (string|null) }} + * @returns {{ update: (processed: number, total: number) => (string|null), reset: () => void }} */ export function createScanEtaTracker() { let startTime = null; @@ -56,6 +56,16 @@ export function createScanEtaTracker() { return translate('initialization.estimatingTime', {}, 'Estimating time...'); } return formatScanRemainingTime((total - lastProcessed) * averageMsPerFile); + }, + + /** + * Drop all samples. Call when the scan moves to another stage: the + * per-file rate of one stage says nothing about the next one. + */ + reset() { + startTime = null; + lastProcessed = 0; + averageMsPerFile = null; } }; } diff --git a/tests/frontend/api/baseModelApi.refresh.test.js b/tests/frontend/api/baseModelApi.refresh.test.js index 7eb9c508..55724efe 100644 --- a/tests/frontend/api/baseModelApi.refresh.test.js +++ b/tests/frontend/api/baseModelApi.refresh.test.js @@ -304,6 +304,94 @@ describe('BaseModelApiClient.refreshModels scan progress', () => { await promise; }); + it('shows the walked roots and file count during the reconcile walk', async () => { + const fetchControl = mockFetchPending(); + const client = await createClient(); + const { promise, socket } = await startRefresh(client); + + socket.emit({ + type: 'scan_progress', + status: 'processing', + stage: 'reconcile_scan', + model_type: 'lora', + pageType: 'loras', + full_rebuild: false, + progress: 12, + processed: 1234, + total: 2000, + files_seen: 1234, + roots_total: 3, + roots_done: 0, + active_roots: ['G:', 'Y:'], + current_name: 'G:', + }); + + expect(setProgressMock).toHaveBeenCalledWith(12); + const walkStatus = setStatusMock.mock.calls.at(-1)[0]; + expect(walkStatus).toContain('G:, Y:'); + expect(walkStatus).toContain('1,234 files'); + // No processed/total ratio: the real file count is unknown mid-walk. + expect(walkStatus).not.toContain('(1234/2000)'); + expect(walkStatus).toContain('Estimating time...'); + + // Walk finished: no ETA once the counters meet the estimate. + socket.emit({ + type: 'scan_progress', + status: 'processing', + stage: 'reconcile_scan', + model_type: 'lora', + full_rebuild: false, + progress: 50, + processed: 2000, + total: 2000, + files_seen: 2000, + active_roots: [], + current_name: 'Y:', + }); + + const finalWalkStatus = setStatusMock.mock.calls.at(-1)[0]; + expect(finalWalkStatus).toContain('2,000 files'); + expect(finalWalkStatus).not.toContain('Estimating time...'); + + fetchControl.resolveOk(); + await promise; + }); + + it('drops the ETA samples when the scan moves to another stage', async () => { + const fetchControl = mockFetchPending(); + let now = 1000; + vi.spyOn(Date, 'now').mockImplementation(() => now); + + const client = await createClient(); + const { promise, socket } = await startRefresh(client); + + const emit = (stage, processed, total, extra = {}) => socket.emit({ + type: 'scan_progress', + status: 'processing', + stage, + model_type: 'lora', + full_rebuild: false, + progress: 50, + processed, + total, + ...extra, + }); + + emit('process_models', 1, 10); + now = 101000; + emit('process_models', 2, 10); + expect(setStatusMock.mock.calls.at(-1)[0]).toContain('~7 min remaining'); + + // Same counters on the walk stage: without the reset the old per-file rate + // (50s/file for 2 files) would be reused and produce a huge ETA. + now = 102000; + emit('reconcile_scan', 2, 100, { files_seen: 2, active_roots: ['G:'] }); + expect(setStatusMock.mock.calls.at(-1)[0]).toContain('Estimating time...'); + + fetchControl.resolveOk(); + await promise; + }); + it('shows the cancelled toast when the server reports cancellation', async () => { const fetchControl = mockFetchPending(); const client = await createClient(); diff --git a/tests/services/test_model_scanner.py b/tests/services/test_model_scanner.py index 95778f24..3ba31da4 100644 --- a/tests/services/test_model_scanner.py +++ b/tests/services/test_model_scanner.py @@ -16,7 +16,13 @@ import pytest from py.services import model_scanner from py.services.model_cache import ModelCache from py.services.model_hash_index import ModelHashIndex -from py.services.model_scanner import CacheBuildResult, ModelScanner +from py.services.model_scanner import ( + CacheBuildResult, + ModelScanner, + _ReconcileWalkTracker, + _count_cached_entries_per_root, + _group_roots_by_device, +) from py.services.pending_delete_service import ( PENDING_DELETE_DIR_NAME, PENDING_DELETE_TTL_SECONDS, @@ -2049,3 +2055,304 @@ async def test_reconcile_cache_broadcasts_error(tmp_path: Path, monkeypatch): assert messages[0]["status"] == "started" assert messages[-1]["status"] == "error" assert messages[-1]["error"] == "walk failed" + + +# --- reconcile walk: device grouping, parallel workers, walk progress ------- + + +def test_reconcile_walk_tracker_reports_weighted_progress(): + tracker = _ReconcileWalkTracker(["/mnt/g", "/mnt/y"], {"/mnt/g": 100, "/mnt/y": 0}) + + snapshot = tracker.snapshot() + assert snapshot is not None + assert snapshot["progress"] == 0 + assert snapshot["files_seen"] == 0 + assert snapshot["roots_total"] == 2 + assert snapshot["roots_done"] == 0 + assert snapshot["active_roots"] == [] + # No files seen yet: the total never drops below what was walked. + assert snapshot["total"] == 100 + + tracker.start_root("/mnt/g") + tracker.report("/mnt/g", 50) + snapshot = tracker.snapshot() + # Half of the 100-weight root, out of 101 total weight -> 24.75% of 50. + assert snapshot["progress"] == 24 + assert snapshot["files_seen"] == 50 + assert snapshot["active_roots"] == ["g"] + assert snapshot["current_name"] == "g" + + # A root with no cached entries contributes one weight unit and only counts + # once it is finished. + tracker.finish_root("/mnt/g") + tracker.start_root("/mnt/y") + tracker.report("/mnt/y", 10) + snapshot = tracker.snapshot() + assert snapshot["roots_done"] == 1 + assert snapshot["progress"] == int((100 / 101) * 50) + assert snapshot["active_roots"] == ["y"] + + tracker.finish_root("/mnt/y") + snapshot = tracker.snapshot() + assert snapshot["progress"] == 50 + assert snapshot["roots_done"] == 2 + assert snapshot["files_seen"] == 60 + # The estimate is the cached entry count, so it also covers new files. + assert snapshot["total"] == 100 + + # A library that grew past the estimate raises the reported total instead of + # overshooting it (the client ETA divides by it). + tracker.start_root("/mnt/g") + tracker.report("/mnt/g", 150) + assert tracker.snapshot()["total"] == 160 + + assert _ReconcileWalkTracker([], {}).snapshot() is None + + +def test_reconcile_walk_tracker_snapshot_is_json_serializable(): + """The snapshot goes straight into a WebSocket payload.""" + tracker = _ReconcileWalkTracker(["/mnt/g"], {"/mnt/g": 2}) + tracker.start_root("/mnt/g") + tracker.report("/mnt/g", 1) + + snapshot = tracker.snapshot() + assert snapshot is not None + assert json.loads(json.dumps(snapshot)) == snapshot + + +def test_group_roots_by_device_keeps_configured_order(tmp_path: Path, monkeypatch): + device_by_root = {"a": "dev1", "b": "dev1", "c": "dev2"} + monkeypatch.setattr( + model_scanner, "_root_device_key", lambda root: device_by_root[root] + ) + + assert _group_roots_by_device(["a", "b", "c"]) == [["a", "b"], ["c"]] + + monkeypatch.undo() + + # Real layout: roots under one temp dir share a device and stay sequential. + first = tmp_path / "one" + second = tmp_path / "two" + third = tmp_path / "three" + for path in (first, second, third): + path.mkdir() + assert _group_roots_by_device([str(first), str(second), str(third)]) == [ + [str(first), str(second), str(third)] + ] + + +def test_count_cached_entries_per_root_uses_longest_prefix(): + counts = _count_cached_entries_per_root( + { + "/m/loras/a.txt", + "/m/loras/sub/b.txt", + "/m/loras2/c.txt", + "/somewhere/else/d.txt", + }, + ["/m/loras", "/m/loras2"], + ) + + assert counts == {"/m/loras": 2, "/m/loras2": 1} + + +@pytest.mark.asyncio +async def test_reconcile_case_fold_fallback_is_indexed_not_linear( + tmp_path: Path, monkeypatch +): + """The Windows case fallback must resolve through a lower-cased index built + once, not by scanning every cached path per miss (was O(files x cached)).""" + root = tmp_path / "loras" + root.mkdir() + for name in ("one", "two", "three"): + (root / f"{name}.txt").write_text(name, encoding="utf-8") + + scanner = DummyScanner(root) + await scanner._initialize_cache() + + # Rewrite the cached paths to a case variant that does not exist on disk: + # the realpath alias index cannot match it, so only the case-fold fallback + # can keep these entries. + for item in scanner._cache.raw_data: + stem = os.path.splitext(os.path.basename(item["file_path"]))[0] + item["file_name"] = stem.upper() + item["file_path"] = _normalize_path(root / f"{stem.upper()}.TXT") + + monkeypatch.setattr(model_scanner, "_CASE_INSENSITIVE_PATHS", True) + + builds = 0 + real_build = model_scanner._build_casefold_index + + def _counting_build(cached_paths): + nonlocal builds + builds += 1 + return real_build(cached_paths) + + monkeypatch.setattr(model_scanner, "_build_casefold_index", _counting_build) + + processed: List[str] = [] + + async def _record_process(file_path, root_path, *args, **kwargs): + processed.append(file_path) + return await DummyScanner._process_model_file( + scanner, file_path, root_path, *args, **kwargs + ) + + scanner._process_model_file = _record_process # type: ignore[method-assign] + + await scanner._reconcile_cache() + + cache = await scanner.get_cached_data() + assert {item["file_path"] for item in cache.raw_data} == { + _normalize_path(root / "ONE.TXT"), + _normalize_path(root / "TWO.TXT"), + _normalize_path(root / "THREE.TXT"), + } + assert processed == [] + assert builds == 1 + + +@pytest.mark.asyncio +async def test_reconcile_walks_different_devices_in_parallel(tmp_path: Path, monkeypatch): + """Roots on different devices are walked concurrently, and their results are + still merged in configured root order (not completion order).""" + root_a = tmp_path / "a" + root_b = tmp_path / "b" + root_a.mkdir() + root_b.mkdir() + + scanner = MultiRootDummyScanner([root_a, root_b]) + await scanner._initialize_cache() + + monkeypatch.setattr( + model_scanner, + "_root_device_key", + lambda root: "dev-a" if str(root) == str(root_a) else "dev-b", + ) + + barrier = threading.Barrier(2, timeout=5) + overlap_violations: List[str] = [] + real_walk = os.walk + + def _walk_with_barrier(path, *args, **kwargs): + def _generate(): + first = True + for entry in real_walk(path, *args, **kwargs): + if first: + first = False + try: + barrier.wait() + except threading.BrokenBarrierError: + overlap_violations.append(str(path)) + if str(path) == str(root_a): + # Root A is first in configured order but finishes last. + time.sleep(0.05) + yield entry + + return _generate() + + monkeypatch.setattr(model_scanner.os, "walk", _walk_with_barrier) + + new_a = root_a / "a_new.txt" + new_a.write_text("a", encoding="utf-8") + new_b = root_b / "b_new.txt" + new_b.write_text("b", encoding="utf-8") + + processed: List[str] = [] + + async def _record_process(file_path, root_path, *args, **kwargs): + processed.append(file_path) + return await DummyScanner._process_model_file( + scanner, file_path, root_path, *args, **kwargs + ) + + scanner._process_model_file = _record_process # type: ignore[method-assign] + + await scanner._reconcile_cache() + + assert overlap_violations == [], "roots on different devices were not walked concurrently" + assert processed == [ + _normalize_path(new_a), + _normalize_path(new_b), + ] + + +@pytest.mark.asyncio +async def test_reconcile_walk_broadcasts_per_root_progress(tmp_path: Path, monkeypatch): + root = tmp_path / "loras" + root.mkdir() + for i in range(4): + sub = root / f"dir{i}" + sub.mkdir() + (sub / "one.txt").write_text("x", encoding="utf-8") + + scanner = DummyScanner(root) + await scanner._initialize_cache() + + ws_stub = RecordingWebSocketManager() + monkeypatch.setattr(model_scanner, "ws_manager", ws_stub) + monkeypatch.setattr(model_scanner, "_WALK_PROGRESS_INTERVAL_SECONDS", 0.01) + + real_walk = os.walk + + def _slow_walk(path, *args, **kwargs): + def _generate(): + for entry in real_walk(path, *args, **kwargs): + time.sleep(0.05) + yield entry + + return _generate() + + monkeypatch.setattr(model_scanner.os, "walk", _slow_walk) + + await scanner._reconcile_cache() + + walk_messages = [ + message + for message in ws_stub.broadcasts + if message["stage"] == "reconcile_scan" and message["status"] == "processing" + ] + # More than the final snapshot => the monitor ticked during the walk. + assert len(walk_messages) >= 2 + + # Intermediate ticks name the root being walked. + assert any(message["active_roots"] == ["loras"] for message in walk_messages) + + final = walk_messages[-1] + assert final["files_seen"] == 4 + assert final["processed"] == final["files_seen"] + assert final["total"] >= final["files_seen"] + assert final["roots_total"] == 1 + assert final["roots_done"] == 1 + assert final["active_roots"] == [] + assert final["current_name"] == "loras" + assert 0 < final["progress"] <= 50 + assert final["full_rebuild"] is False + assert walk_messages[0]["progress"] <= final["progress"] + + +@pytest.mark.asyncio +async def test_reconcile_prunes_entries_of_an_offline_root(tmp_path: Path): + """A root that is gone (drive switched off) is skipped by the walk, so its + cached entries are reported as missing. Locked in because it is the + documented consequence of refreshing with a drive powered down.""" + root_a = tmp_path / "a" + root_b = tmp_path / "b" + root_a.mkdir() + root_b.mkdir() + file_a = root_a / "a0.txt" + file_a.write_text("a", encoding="utf-8") + file_b = root_b / "b0.txt" + file_b.write_text("b", encoding="utf-8") + + scanner = MultiRootDummyScanner([root_a, root_b]) + await scanner._initialize_cache() + assert len(scanner._cache.raw_data) == 2 + + root_b.rename(tmp_path / "b_offline") + + await scanner._reconcile_cache() + + remaining = {item["file_path"] for item in scanner._cache.raw_data} + assert remaining == {_normalize_path(file_a)} + assert scanner._hash_index.get_path("hash-b0") is None + assert scanner._hash_index.get_path("hash-a0") == _normalize_path(file_a)