feat(scanner): report and parallelize the reconcile walk

A regular Refresh walked every configured root sequentially, so a full
25 TB drive delayed the roots behind it, and the dialog sat on "Checking
for changes..." at 0 % for the whole walk with no way to tell it was
working. On a cold external drive the walk itself dominates the cost, so
the fix is to overlap the drives and to show what the walk is doing.

Backend (py/services/model_scanner.py):
* Extract the per-root walk into the synchronous _walk_root_for_reconcile()
  worker and merge its results on the event loop afterwards, in configured
  root order: which business path wins a file reachable through several
  roots must not depend on the order the workers happened to finish in.
* Group roots by device (_root_device_key: drive letter on Windows, st_dev
  on POSIX) and run one worker per device. Roots sharing a device stay
  sequential, so directory claims and the overlap dedup (#871, #1041) keep
  their configured-order semantics; different devices run in parallel.
* Track walk progress per root (_ReconcileWalkTracker), weighted by each
  root's cached entry count because the real file count is only known once
  the walk ends. The bar splits into walk (0-50 %) and new-file (50-99 %)
  phases so it never jumps backwards, and the walk broadcasts files seen,
  active roots and the ETA counters.
* Replace the Windows case-insensitive fallback -- a scan of every cached
  path per miss, i.e. O(files x cached) -- with a lazily built lower-cased
  index (_CachedPathLookups, also now guarding the realpath alias index for
  worker threads), and lift the os.name == "nt" gate into the module-level
  _CASE_INSENSITIVE_PATHS so the branch is testable off Windows.
* Excluded-model membership is a set lookup instead of a list scan.

Frontend:
* render the walk phase as "Checking for changes... <roots> (N files)" with
  the ETA, and reset the ETA tracker when the stage changes: the per-file
  rate of counting files says nothing about processing them.
* add common.scanProgress.walkFiles; the other locales keep the sanctioned
  [TODO: Translate] placeholder until the translation pass.

Verified: no-change reconcile over 10k files/500 dirs 101 ms and 10k/5000
dirs 230 ms (was 93/229 ms, within noise); a two-device sandbox walk runs
both roots concurrently, names them in the progress messages and finishes
with added=15, removed=0; 3665 passed, 7 skipped; frontend 1456 passed,
vue widgets 96 passed.
This commit is contained in:
Will Miao
2026-10-06 12:40:57 +08:00
parent 8cd53c20f8
commit 14e660e876
16 changed files with 1059 additions and 187 deletions
@@ -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 <path>`。
- 若回退分支在起作用 → 同样应该是 `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 <path>`。
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%,两段之间不会回退。
+2 -1
View File
@@ -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": {
+2 -1
View File
@@ -73,7 +73,8 @@
"lessThanMinute": "Less than a minute remaining",
"minutes": "~{minutes} min remaining",
"hours": "~{hours} hr {minutes} min remaining"
}
},
"walkFiles": "{count} files"
}
},
"onboarding": {
+2 -1
View File
@@ -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": {
+2 -1
View File
@@ -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": {
+2 -1
View File
@@ -73,7 +73,8 @@
"lessThanMinute": "נותרה פחות מדקה",
"minutes": "נותרו ~{minutes} דקות",
"hours": "נותרו ~{hours} שעות ו-{minutes} דקות"
}
},
"walkFiles": "[TODO: Translate] {count} files"
}
},
"onboarding": {
+2 -1
View File
@@ -73,7 +73,8 @@
"lessThanMinute": "残り1分未満",
"minutes": "残り約 {minutes} 分",
"hours": "残り約 {hours} 時間 {minutes} 分"
}
},
"walkFiles": "[TODO: Translate] {count} files"
}
},
"onboarding": {
+2 -1
View File
@@ -73,7 +73,8 @@
"lessThanMinute": "남은 시간 1분 미만",
"minutes": "약 {minutes}분 남음",
"hours": "약 {hours}시간 {minutes}분 남음"
}
},
"walkFiles": "[TODO: Translate] {count} files"
}
},
"onboarding": {
+2 -1
View File
@@ -73,7 +73,8 @@
"lessThanMinute": "Осталось меньше минуты",
"minutes": "Осталось ~{minutes} мин",
"hours": "Осталось ~{hours} ч {minutes} мин"
}
},
"walkFiles": "[TODO: Translate] {count} files"
}
},
"onboarding": {
+2 -1
View File
@@ -73,7 +73,8 @@
"lessThanMinute": "剩余时间不到一分钟",
"minutes": "剩余约 {minutes} 分钟",
"hours": "剩余约 {hours} 小时 {minutes} 分钟"
}
},
"walkFiles": "[TODO: Translate] {count} files"
}
},
"onboarding": {
+2 -1
View File
@@ -73,7 +73,8 @@
"lessThanMinute": "剩餘時間不到一分鐘",
"minutes": "剩餘約 {minutes} 分鐘",
"hours": "剩餘約 {hours} 小時 {minutes} 分鐘"
}
},
"walkFiles": "[TODO: Translate] {count} files"
}
},
"onboarding": {
+558 -118
View File
@@ -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
+32 -1
View File
@@ -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}`;
+11 -1
View File
@@ -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;
}
};
}
@@ -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();
+308 -1
View File
@@ -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)