fix(monitor): 消除探测饿死与重试饥饿导致的跨目录连坐 (#6291)

app/modules/filemanager/fsproxy.py

@@ -108,6 +108,15 @@ class FileSystemProxy:
         """
         return self._call("listdir", path=str(path))
 
+    def count_entries(self, path: Path, max_check: int = 10000) -> Dict[str, int]:
+        """
+        统计目录规模。整棵树的遍历在子进程内一次完成，超时可整体放弃。
+        :param path: 目标目录
+        :param max_check: 文件数上限，超过即提前结束
+        :return: {"file_count", "dir_count"}
+        """
+        return self._call("count_entries", path=str(path), max_check=max_check)
+
     def rename(self, src: Path, dst: Path) -> bool:
         """
         同一存储内重命名/移动。跨存储会抛 OSError(EXDEV)，由调用方走原有路径。
@@ -287,6 +296,14 @@ class FileSystemProxy:
             return True
         if op == "listdir":
             return sorted(os.listdir(payload["path"]))
+        if op == "count_entries":
+            file_count = dir_count = 0
+            for _, dirs, files in os.walk(payload["path"]):
+                file_count += len(files)
+                dir_count += len(dirs)
+                if file_count > (payload.get("max_check") or 10000):
+                    break
+            return {"file_count": file_count, "dir_count": dir_count}
         if op == "rename":
             os.rename(payload["src"], payload["dst"])
             return True

app/modules/filemanager/fsworker.py

@@ -94,6 +94,26 @@ def _copy(payload, emit):
     return {"copied": copied, "total": total}
 
 
+def _count_entries(payload, _emit):
+    """
+    统计目录下的文件与子目录数量，超过上限即提前结束。
+
+    放在子进程里做而不是逐层 listdir 走 IPC：递归遍历一棵大目录树会产生成千
+    上万次往返，代价不可接受；一次调用在子进程内跑完 os.walk，父进程只需对
+    这一次调用设超时即可整体放弃。
+    """
+    directory = payload["path"]
+    max_check = payload.get("max_check") or 10000
+    file_count = 0
+    dir_count = 0
+    for _, dirs, files in os.walk(directory):
+        file_count += len(files)
+        dir_count += len(dirs)
+        if file_count > max_check:
+            break
+    return {"file_count": file_count, "dir_count": dir_count}
+
+
 def _rename(payload, _emit):
     """
     同一存储内重命名/移动。
@@ -134,6 +154,7 @@ _HANDLERS = {
     "exists": _exists,
     "listdir": _listdir,
     "copy": _copy,
+    "count_entries": _count_entries,
     "rename": _rename,
     "unlink": _unlink,
     "rmtree": _rmtree,

app/monitor/dispatcher.py

@@ -13,6 +13,7 @@ from app.helper.transferhistory import (HistoryGateAction, describe_history_gate
                                         evaluate_history_gate, is_skip_action,
                                         max_failed_retries, resolve_history)
 from app.log import logger
+from app.modules.filemanager.fsproxy import fsproxy
 from app.schemas import FileItem
 from app.schemas.types import MediaType
 
@@ -230,8 +231,10 @@ class TransferDispatcher:
         :return: (文件大小, 修改时间, 文件是否仍然存在)；大小为 None 表示本次读取仍然失败
         """
         try:
-            file_stat = Path(event_path).stat()
-            return file_stat.st_size, file_stat.st_mtime, True
+            # 走可强杀的子进程：裸 stat 在挂死的挂载上永不返回，会把整个
+            # 待重试队列的驱动动作钉死，其他健康目录的重试项再也不会被消费
+            info = fsproxy.stat(Path(event_path))
+            return info["size"], info["mtime"], True
         except FileNotFoundError:
             return None, None, False
         except OSError as err:

app/monitor/monitor.py

@@ -71,6 +71,10 @@ class Monitor(ConfigReloadMixin, metaclass=SingletonClass):
         self._stable_cycles: Dict[str, int] = {}
         # 判定为挂载级故障、已暂停一切访问的监控目录（path -> 隔离状态）
         self._isolated: Dict[str, Dict[str, Any]] = {}
+        # 探测通过、等待下一周期重建的目录。不能靠 is_stalled 重新发现：
+        # 进入隔离前 __rebuild_watcher 已调用过 watcher.stop()，停止标志置位后
+        # is_stalled() 恒为 False，检测环节再也认不出它需要重建
+        self._pending_rebuild: Dict[str, Any] = {}
         # 触碰挂载的恢复动作执行器，把 block 型故障挡在看门狗线程之外
         self._recovery = RecoveryExecutor()
         # 定时服务
@@ -369,7 +373,10 @@ class Monitor(ConfigReloadMixin, metaclass=SingletonClass):
         with self._watcher_lock:
             watchers = list(self._watchers)
             isolated = set(self._isolated)
-        broken: List[LocalDirectoryWatcher] = []
+            # 探测已确认挂载恢复的目录，本轮直接送去重建
+            resumed = list(self._pending_rebuild.values())
+            self._pending_rebuild.clear()
+        broken: List[LocalDirectoryWatcher] = list(resumed)
         for watcher in watchers:
             key = str(watcher.watch_path)
             if key in isolated:
@@ -486,10 +493,14 @@ class Monitor(ConfigReloadMixin, metaclass=SingletonClass):
                 entry = self._isolated.pop(key, None)
             if not entry:
                 continue
-            logger.info(f"✓ 挂载探测通过，解除隔离并重建目录监控: {mon_path}")
+            logger.info(f"✓ 挂载探测通过，解除隔离并登记重建: {mon_path}")
             self.__clear_alert(mon_path, f"目录监控挂载已恢复响应，正在重建监控: {mon_path}")
-            # 重建内部会按 watcher 的最后心跳时间发起补偿扫描，补回隔离期间落地的文件
-            self.__rebuild_watcher(entry["watcher"])
+            # 只登记、不在此处重建。重建会触碰挂载，若挂载能应答探测却在重建时再次
+            # 挂死，这条探测线程将永不返回，全局 PROBE_KEY 从此恒为 BUSY——列表中
+            # 其余隔离目录连探测机会都没有了，「按目录隔离」的承诺就此失效。
+            # 交由下一周期的 __check_watchers 走 rebuild:<path> 路径，天然按目录隔离。
+            with self._watcher_lock:
+                self._pending_rebuild[key] = entry["watcher"]
 
     def __drive_pending(self):
         """
@@ -766,6 +777,7 @@ class Monitor(ConfigReloadMixin, metaclass=SingletonClass):
             self._restart_marks = {}
             self._stable_cycles = {}
             self._isolated = {}
+            self._pending_rebuild = {}
         # 已冻死的恢复线程无法回收，这里只是不再跟踪它们，避免重载后同名目录
         # 被残留记录误判为 BUSY 而永远拿不到重建机会
         self._recovery.clear()

app/monitor/syslimits.py

@@ -15,17 +15,17 @@ def count_directory_entries(directory: Path, max_check: int = 10000) -> Tuple[in
     :param max_check: 最大检查文件数量，避免长时间阻塞
     :return: (文件数量, 目录数量)
     """
-    file_count = 0
-    dir_count = 0
     try:
-        for _, dirs, files in os.walk(str(directory)):
-            file_count += len(files)
-            dir_count += len(dirs)
-            if file_count > max_check:
-                break
+        # 走可强杀的子进程：挂载挂死时 os.walk 永不返回，会把启动重试的
+        # 恢复动作永久钉死，进而饿死其他健康目录的待重试项
+        # 延迟导入：filemanager 包的 __init__ 会拖入整条 chain 依赖，
+        # 模块级导入会破坏 monitor 包的轻量加载
+        from app.modules.filemanager.fsproxy import fsproxy
+        result = fsproxy.count_entries(directory, max_check=max_check)
+        return result.get("file_count", 0), result.get("dir_count", 0)
     except Exception as err:
         logger.debug(f"统计目录规模失败: {err}")
-    return file_count, dir_count
+    return 0, 0
 
 
 def count_directory_files(directory: Path, max_check: int = 10000) -> int:

app/monitor/watcher.py

@@ -109,9 +109,14 @@ class LocalDirectoryWatcher:
         """
         启动本地目录监控线程。
         """
-        if not self._watch_path.exists():
-            raise FileNotFoundError(f"监控目录不存在: {self._watch_path}")
-        if not self._watch_path.is_dir():
+        # 走可强杀的子进程：这两行是事故中最先冻住的地方——挂载挂死时
+        # exists()/is_dir() 永不返回，重建监控的恢复动作就此永久悬挂。
+        # 经代理后超时会抛 OSError，由上层判定为挂载级故障并转入隔离
+        # 延迟导入：filemanager 包的 __init__ 会拖入整条 chain 依赖，
+        # 模块级导入会破坏 monitor 包的轻量加载
+        from app.modules.filemanager.fsproxy import fsproxy
+        info = fsproxy.stat(self._watch_path)
+        if not info["is_dir"]:
             raise NotADirectoryError(f"监控路径不是目录: {self._watch_path}")
         if self.is_alive():
             logger.info(f"本地目录监控已在运行中: {self._watch_path}")

tests/test_monitor_gap_recovery.py

@@ -30,6 +30,7 @@ def _build_monitor(handle_file: MagicMock = None):
     monitor._restart_marks = {}
     monitor._stable_cycles = {}
     monitor._isolated = {}
+    monitor._pending_rebuild = {}
     monitor._recovery = RecoveryExecutor()
     return monitor, dispatcher
 

tests/test_monitor_mount_isolation.py

@@ -40,6 +40,7 @@ def _build_monitor(monkeypatch, put_recorder=None):
     monitor._restart_marks = {}
     monitor._stable_cycles = {}
     monitor._isolated = {}
+    monitor._pending_rebuild = {}
     monitor._recovery = RecoveryExecutor()
     return monitor
 
@@ -201,17 +202,19 @@ def test_watchdog_survives_blocking_exists_in_real_rebuild(tmp_path, monkeypatch
     monitor._watchers = [watcher]
 
     release = threading.Event()
-    real_exists = Path.exists
 
-    def blocking_exists(self, *args, **kwargs):
+    def blocking_stat(path, *_args, **_kwargs):
         """
-        模拟 block 型挂载：对监控目录的 exists() 永不返回。
+        模拟 block 型挂载：监控目录的属性读取永不返回。
+        入口校验已改走子进程代理，因此在代理这一层注入阻塞。
         """
-        if self == tmp_path:
+        if Path(path) == tmp_path:
             release.wait()
-        return real_exists(self, *args, **kwargs)
+        return {"size": 0, "mtime": 0.0, "is_dir": True, "is_file": False}
 
-    monkeypatch.setattr(Path, "exists", blocking_exists)
+    monkeypatch.setattr(
+        "app.modules.filemanager.fsproxy.fsproxy.stat", blocking_stat
+    )
     monkeypatch.setattr("app.monitor.monitor.probe_path", lambda *_a, **_kw: False)
 
     try:
@@ -277,7 +280,15 @@ def test_isolated_directory_recovers_after_probe_succeeds(tmp_path, monkeypatch)
     assert _run_watchdog(monitor)
 
     assert str(tmp_path) not in monitor._isolated, "探测通过后没有解除隔离"
-    assert rebuilt == [watcher], "解除隔离后没有重建监控"
+    # 不得在探测线程里内联重建：重建会触碰挂载，若此时再次挂死，全局 PROBE_KEY
+    # 将永久 BUSY，其余隔离目录连探测机会都没有
+    assert rebuilt == [], "探测线程内联执行了重建，会饿死其他隔离目录"
+    assert str(tmp_path) in monitor._pending_rebuild, "没有登记待重建"
+
+    # 下一周期应把它送去按目录隔离的重建路径
+    assert _run_watchdog(monitor)
+    assert rebuilt == [watcher], "下一周期没有重建该监控"
+    assert not monitor._pending_rebuild, "重建后未清空待重建登记"
 
 
 def test_isolated_directory_stays_isolated_while_probe_fails(tmp_path, monkeypatch):

tests/test_monitor_resilience.py

@@ -140,6 +140,7 @@ def _build_monitor(monkeypatch, put_recorder):
     monitor._restart_marks = {}
     monitor._stable_cycles = {}
     monitor._isolated = {}
+    monitor._pending_rebuild = {}
     monitor._recovery = RecoveryExecutor()
     return monitor