diff --git a/app/modules/filemanager/fsproxy.py b/app/modules/filemanager/fsproxy.py index 6341af724..8a03756ce 100644 --- a/app/modules/filemanager/fsproxy.py +++ b/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 diff --git a/app/modules/filemanager/fsworker.py b/app/modules/filemanager/fsworker.py index 5b6a5a7c5..9124fdfbc 100644 --- a/app/modules/filemanager/fsworker.py +++ b/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, diff --git a/app/monitor/dispatcher.py b/app/monitor/dispatcher.py index fbfb652ff..227897d1e 100644 --- a/app/monitor/dispatcher.py +++ b/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: diff --git a/app/monitor/monitor.py b/app/monitor/monitor.py index bb71d7ef6..78fdcf61a 100644 --- a/app/monitor/monitor.py +++ b/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: 路径,天然按目录隔离。 + 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() diff --git a/app/monitor/syslimits.py b/app/monitor/syslimits.py index 87b202f64..694939b67 100644 --- a/app/monitor/syslimits.py +++ b/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: diff --git a/app/monitor/watcher.py b/app/monitor/watcher.py index 35103145e..b446b0038 100644 --- a/app/monitor/watcher.py +++ b/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}") diff --git a/tests/test_monitor_gap_recovery.py b/tests/test_monitor_gap_recovery.py index 3f9212ea0..082d074b2 100644 --- a/tests/test_monitor_gap_recovery.py +++ b/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 diff --git a/tests/test_monitor_mount_isolation.py b/tests/test_monitor_mount_isolation.py index 3da5851d0..1f34f799c 100644 --- a/tests/test_monitor_mount_isolation.py +++ b/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): diff --git a/tests/test_monitor_resilience.py b/tests/test_monitor_resilience.py index 68e15e24e..715610af9 100644 --- a/tests/test_monitor_resilience.py +++ b/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