From ed0460fc8b1a3b87156383838a06a69aaa67ffaa Mon Sep 17 00:00:00 2001 From: freeman Date: Wed, 30 Sep 2026 21:19:41 +0800 Subject: [PATCH] fix(smb): keep remote snapshots complete and recoverable --- app/modules/filemanager/storages/__init__.py | 47 ++++- app/modules/filemanager/storages/smb.py | 48 +++-- docs/rules/05-architecture.md | 6 + docs/testing.md | 17 ++ tests/test_monitor_snapshot_semantics.py | 4 + tests/test_smb_storage.py | 176 ++++++++++++++++++- tests/test_storage_strict_query_coverage.py | 8 + 7 files changed, 286 insertions(+), 20 deletions(-) diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py index c8d1b7904..64f7bf2f7 100644 --- a/app/modules/filemanager/storages/__init__.py +++ b/app/modules/filemanager/storages/__init__.py @@ -51,6 +51,11 @@ class StorageBase(metaclass=ABCMeta): """返回存储快照目录时间检查的默认开关。""" return True + @property + def snapshot_strict_query(self) -> bool: + """声明快照是否使用严格查询；未适配的存储沿用原查询合同。""" + return False + def __init__(self): self.storagehelper = StorageHelper() @@ -164,6 +169,10 @@ class StorageBase(metaclass=ABCMeta): """ pass + def list_strict(self, fileitem: _SchemaFileItem) -> List[_SchemaFileItem]: + """完整列举目录；空列表表示确认无子项，无法确认时抛出 StorageQueryError。""" + raise StorageQueryError(f"存储 {self.schema} 未实现严格目录查询: {fileitem.path}") + @abstractmethod def create_folder(self, fileitem: _SchemaFileItem, name: str) -> Optional[_SchemaFileItem]: """ @@ -286,15 +295,34 @@ class StorageBase(metaclass=ABCMeta): """ pass + def _snapshot_list(self, fileitem: _SchemaFileItem) -> Optional[List[_SchemaFileItem]]: + """按存储能力读取目录，严格快照不能把无结果当作可以跳过的子树。""" + if not self.snapshot_strict_query: + return self.list(fileitem) + items = self.list_strict(fileitem) + if items is None: + raise StorageQueryError(f"目录查询没有返回完整结果: {fileitem.path}") + return items + + def _snapshot_read_error(self, fileitem: _SchemaFileItem, error: Exception) -> None: + """严格快照传播读取故障；未适配的存储保留原来的子树容错行为。""" + if self.snapshot_strict_query: + if isinstance(error, StorageQueryError): + raise error + raise StorageQueryError(f"读取快照路径失败: {fileitem.path} - {error}") from error + logger.debug(f"Snapshot error for {fileitem.path}: {error}") + def snapshot(self, path: Path, last_snapshot_time: float = None, max_depth: int = 5, - previous_snapshot: Optional[Dict[str, Dict]] = None) -> Dict[str, Dict]: + previous_snapshot: Optional[Dict[str, Dict]] = None) -> Optional[Dict[str, Dict]]: """ 快照文件系统，输出所有层级文件信息（不含目录） :param path: 路径 :param last_snapshot_time: 上次快照时间，用于增量快照 :param max_depth: 最大递归深度，避免过深遍历 :param previous_snapshot: 上次完整快照，用于保留未变化目录并清理已删除文件 + :return: 有效快照（允许为空）；严格查询失败返回 None，调用方必须保留旧基线 """ + strict_query = self.snapshot_strict_query root_path = PurePosixPath(path.as_posix()) files_info = { file_path: file_info @@ -340,7 +368,7 @@ class StorageBase(metaclass=ABCMeta): return # 只有目录列表成功返回后才清理旧基线，查询异常时继续保留待下轮重试 - sub_files = self.list(_fileitm) + sub_files = self._snapshot_list(_fileitm) if sub_files is None: return sub_files = list(sub_files) @@ -357,12 +385,15 @@ class StorageBase(metaclass=ABCMeta): } except Exception as e: - logger.debug(f"Snapshot error for {_fileitm.path}: {e}") + self._snapshot_read_error(_fileitm, e) - fileitem = self.get_item(path) - if not fileitem: - return {} - - __snapshot_file(fileitem) + try: + fileitem = self.get_item_strict(path) if strict_query else self.get_item(path) + if not fileitem: + return {} + __snapshot_file(fileitem) + except StorageQueryError as err: + logger.warning(f"存储快照失败，保留上次基线: {self.schema}:{path} - {err}") + return None return files_info diff --git a/app/modules/filemanager/storages/smb.py b/app/modules/filemanager/storages/smb.py index 1bb3927cb..217ac06ae 100644 --- a/app/modules/filemanager/storages/smb.py +++ b/app/modules/filemanager/storages/smb.py @@ -47,6 +47,16 @@ class SMB(StorageBase, metaclass=WeakSingleton): # 文件块大小，默认10MB chunk_size = 10 * 1024 * 1024 + @property + def snapshot_check_folder_modtime(self) -> bool: + """SMB 目录时间不保证传播后代变化，快照必须遍历到配置的深度上限。""" + return False + + @property + def snapshot_strict_query(self) -> bool: + """SMB 快照必须区分真实空目录与读取失败，防止错误更新监控基线。""" + return True + def __init__(self): """加载当前 SMB 配置并尝试建立连接。""" super().__init__() @@ -202,10 +212,10 @@ class SMB(StorageBase, metaclass=WeakSingleton): return "/" + file_path[len(prefix):].replace("\\", "/") def _create_fileitem( - self, stat_result, file_path: str, name: str + self, stat_result, file_path: str, name: str, *, strict: bool = False ) -> _SchemaFileItem: """ - 创建文件项 + 创建文件项；严格查询不能把元数据读取失败回退成普通文件。 """ import smbclient try: @@ -247,6 +257,8 @@ class SMB(StorageBase, metaclass=WeakSingleton): modify_time=modify_time, ) except Exception as e: + if strict: + raise StorageQueryError(f"【SMB】创建文件项失败: {file_path} - {e}") from e logger.error(f"【SMB】创建文件项失败：{e}") # 返回基本的文件项信息 return _SchemaFileItem( @@ -282,9 +294,18 @@ class SMB(StorageBase, metaclass=WeakSingleton): self._connected = False return False - def list(self, fileitem: _SchemaFileItem) -> List[_SchemaFileItem]: + def list_strict(self, fileitem: _SchemaFileItem) -> List[_SchemaFileItem]: + """用于快照的完整目录查询，任一子项读取失败都拒绝返回部分结果。""" + try: + return self.list(fileitem, strict=True) + except StorageQueryError: + raise + except Exception as err: + raise StorageQueryError(f"【SMB】查询目录失败: {fileitem.path} - {err}") from err + + def list(self, fileitem: _SchemaFileItem, *, strict: bool = False) -> List[_SchemaFileItem]: """ - 浏览文件 + 浏览文件；strict 供快照查询传播异常，普通浏览沿用容错回退。 """ import smbclient from smbprotocol.exceptions import SMBException, SMBResponseException @@ -292,7 +313,9 @@ class SMB(StorageBase, metaclass=WeakSingleton): self._check_connection() if fileitem.type == "file": - item = self.detail(fileitem) + if not fileitem.path: + raise StorageQueryError("【SMB】文件查询缺少路径") + item = self.get_item_strict(Path(fileitem.path)) if strict else self.detail(fileitem) if item: return [item] return [] @@ -310,11 +333,10 @@ class SMB(StorageBase, metaclass=WeakSingleton): # 列出目录内容 try: entries = smbclient.listdir(smb_path) - except SMBResponseException as e: - logger.error(f"【SMB】列出目录失败: {smb_path} - {e}") - return [] - except SMBException as e: + except (SMBResponseException, SMBException) as e: logger.error(f"【SMB】列出目录失败: {smb_path} - {e}") + if strict: + raise return [] items = [] @@ -325,14 +347,18 @@ class SMB(StorageBase, metaclass=WeakSingleton): entry_path = f"{smb_path}\\{entry}" try: stat_result = smbclient.stat(entry_path) - item = self._create_fileitem(stat_result, entry_path, entry) + item = self._create_fileitem(stat_result, entry_path, entry, strict=strict) items.append(item) except Exception as e: + if strict: + raise logger.debug(f"【SMB】获取文件信息失败: {entry_path} - {e}") continue return items except Exception as e: + if strict: + raise logger.error(f"【SMB】列出文件失败: {e}") return [] @@ -453,7 +479,7 @@ class SMB(StorageBase, metaclass=WeakSingleton): if err.errno in (errno.ENOENT, errno.ENOTDIR): return None raise StorageQueryError(f"【SMB】查询文件项失败: {path} - {err}") from err - return self._create_fileitem(stat_result, smb_path, Path(path).name) + return self._create_fileitem(stat_result, smb_path, Path(path).name, strict=True) except StorageQueryError: raise except Exception as e: diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 21985c9da..e23732103 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -717,6 +717,12 @@ SMB 同存储整理由 `app.modules.filemanager.storages.smb` 持有协议调用 下载器返回的远程路径保留存储 URI，整理 Chain 通过 `StorageChain` 查询源文件， 不直接依赖 SMB SDK；断线后目标状态不明沿用持久步骤的保守恢复合同。 +SMB 目录快照不依赖祖先目录 mtime 推断后代变化，而在既有递归深度上限内逐层列举。 +存储通过 `snapshot_strict_query` 声明已实现 `get_item_strict` / `list_strict`； +严格快照的任一查询失败必须返回 `None`，由远程轮询器保留旧基线并重试。 +只有确认目录清空或根路径不存在时才返回空字典。SMB 普通文件浏览继续允许容错回退， +严格快照不能复用其部分结果；其他存储需按各自查询合同逐步接入。 + `app.modules.filemanager` is a lazy compatibility entrypoint. The concrete `FileManagerModule` implementation lives in `app.modules.filemanager.module`, while the historical capability path and class module identity remain diff --git a/docs/testing.md b/docs/testing.md index a77b0bdce..234103a24 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -82,6 +82,23 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)): > 注意：识别这类端到端流程往往不止一个外部出口。例如 TMDB 识别除了目录请求，链路层还会向 MP 服务器上报/查询「共享识别 API」——这类旁路出口必须一并打桩。用下文的 socket 探针确认确实零出站。 +### 存储与监控的边界回归 + +存储快照用例优先在 SDK 的目录列表、stat、类型查询和连接边界打桩，保留真实的路径映射、 +文件项构造、快照遍历及轮询逻辑。仅替换 `snapshot_storage` 的返回值可以验证轮询器， +但不能证明存储适配器正确区分了空目录与查询失败。 + +`tests/test_smb_storage.py` 提供以下离线场景： + +- 单共享和多共享下，祖先目录时间不变时仍发现后代新增、修改、保留旧 mtime 的文件移入和删除。 +- 根目录、子目录、单个文件或类型查询失败时，不提交空快照或部分快照。 +- 连接失效保留旧基线；恢复后发现新文件，再次轮询不重复派发。 +- 真正清空的目录和确认不存在的根路径可以提交空快照。 +- 完整遍历仍遵守递归深度上限，不因本次修复增加扫描频率。 + +这些场景只承诺已验证的 SMB 查询语义；为其他存储接入严格快照时，应在其 SDK/API +边界复现同样的状态转换，并评估目录时间语义、分页和请求成本。 + ## 自隔离：用了什么，就还原什么 用例若修改了**进程级状态**——`sys.modules` 桩、单例（`Singleton._instances`）、`lru_cache`、环境变量、`settings` 字段——必须在用例或模块结束时还原。pytest 一次性导入全部测试模块，未还原的污染会扩散到后续用例，产生“单独跑过、一起跑挂”的测不准现象。 diff --git a/tests/test_monitor_snapshot_semantics.py b/tests/test_monitor_snapshot_semantics.py index 69a1bc429..f005caee4 100644 --- a/tests/test_monitor_snapshot_semantics.py +++ b/tests/test_monitor_snapshot_semantics.py @@ -136,6 +136,8 @@ def test_storage_snapshot_reconciles_deleted_children_and_keeps_skipped_subtree( """ storage = MagicMock() storage.snapshot_check_folder_modtime = True + storage.snapshot_strict_query = False + storage._snapshot_list = lambda item: StorageBase._snapshot_list(storage, item) root = schemas.FileItem(storage="alist", type="dir", path="/mon/", name="mon", modify_time=200) kept_dir = schemas.FileItem( storage="alist", type="dir", path="/mon/keep/", name="keep", modify_time=50 @@ -163,6 +165,8 @@ def test_storage_snapshot_always_lists_monitor_root_for_deletions(): """ storage = MagicMock() storage.snapshot_check_folder_modtime = True + storage.snapshot_strict_query = False + storage._snapshot_list = lambda item: StorageBase._snapshot_list(storage, item) storage.get_item.return_value = schemas.FileItem( storage="alist", type="dir", path="/mon/", name="mon", modify_time=50 ) diff --git a/tests/test_smb_storage.py b/tests/test_smb_storage.py index 3a080d755..081704d13 100644 --- a/tests/test_smb_storage.py +++ b/tests/test_smb_storage.py @@ -1,4 +1,4 @@ -"""SMB 保存重连和同一共享内服务端整理的回归测试。""" +"""SMB 保存重连、服务端整理与远程快照恢复的回归测试。""" import errno from pathlib import Path @@ -12,6 +12,8 @@ from app.application.storage import StorageHelper from app.modules.filemanager.module import FileManagerModule from app.modules.filemanager.storages import smb as smb_module from app.modules.filemanager.transhandler import TransHandler +from app.monitor.poller import RemotePoller +from app.monitor.snapshot import SnapshotStore from app.schemas.file import FileItem @@ -332,3 +334,175 @@ def test_multi_share_usage_query_failure_remains_unavailable(multi_share, monkey monkeypatch.setattr(smbclient, "stat_volume", stat_volume) assert multi_share.storage.usage() is None stat_volume.assert_called_once_with(r"\\10.10.10.11\video") + + +@pytest.fixture +def snapshot_server(smb_storage, monkeypatch): + """仅替换 SMB SDK 的文件系统边界，保留真实目录映射、文件项构造和快照遍历。""" + storage = smb_storage.storage + smb_storage.state["conf"] = {"host": "smb.example", "share": "media"} + storage.init_storage() + tree = { + "/downloads": {"directory": True, "mtime": 50}, + "/downloads/Series": {"directory": True, "mtime": 50}, + "/downloads/Series/Season01": {"directory": True, "mtime": 50}, + "/downloads/Series/Season01/old.mkv": {"directory": False, "mtime": 100, "size": 10}, + } + fault = {"operation": None, "path": None} + + def relative(path): + return storage._relative_path(path).rstrip("/") + + def read_node(operation, path): + key = relative(path) + if fault == {"operation": operation, "path": key}: + raise OSError(errno.EIO, "temporary SMB read failure") + if key not in tree: + raise FileNotFoundError(errno.ENOENT, "missing") + return key, tree[key] + + def list_directory(path): + key, _node = read_node("list", path) + return [Path(item).name for item in tree if Path(item).parent.as_posix() == key] + + def stat_item(path): + _key, node = read_node("stat", path) + return SimpleNamespace(st_mtime=node["mtime"], st_size=node.get("size", 0)) + + def is_directory(path): + _key, node = read_node("type", path) + return node["directory"] + + smb_storage.client.listdir.side_effect = list_directory + monkeypatch.setattr(smbclient, "stat", Mock(side_effect=stat_item)) + monkeypatch.setattr(smbclient.path, "isdir", Mock(side_effect=is_directory)) + monkeypatch.setattr(smbclient.path, "exists", Mock(side_effect=lambda path: relative(path) in tree)) + return SimpleNamespace( + storage=storage, module=smb_storage.module, client=smb_storage.client, + tree=tree, fault=fault, root=Path("/downloads"), + ) + + +@pytest.mark.parametrize("multiple_shares", [False, True]) +@pytest.mark.parametrize("change", ["added", "modified", "moved_old", "deleted"]) +def test_snapshot_detects_descendant_changes_without_parent_mtime(snapshot_server, multiple_shares, change): + """SMB 祖先目录时间不变时，单共享与多共享仍能发现后代新增、修改、旧文件移入和删除。""" + state = snapshot_server + if multiple_shares: + state.storage._configure_shares({"shares": ["downloads", "library"]}) + old_path = "/downloads/Series/Season01/old.mkv" + new_path = "/downloads/Series/Season01/new.mkv" + baseline = state.storage.snapshot(state.root) + assert set(baseline) == {old_path} + + if change in {"added", "moved_old"}: + state.tree[new_path] = { + "directory": False, "mtime": 200 if change == "added" else 1, "size": 20, + } + elif change == "modified": + state.tree[old_path].update(size=20, mtime=200) + else: + del state.tree[old_path] + + snapshot = state.module.snapshot_storage( + "smb", state.root, last_snapshot_time=100, previous_snapshot=baseline, + ) + assert snapshot is not None + changes = SnapshotStore.compare(baseline, snapshot) + expected = {"added": [], "modified": []} + if change == "deleted": + assert old_path not in snapshot + else: + expected["added" if change == "moved_old" else change] = [ + new_path if change in {"added", "moved_old"} else old_path + ] + assert changes == expected + assert baseline[old_path]["size"] == 10 + + +@pytest.mark.parametrize( + ("operation", "path"), + [ + ("stat", "/downloads"), + ("list", "/downloads"), + ("list", "/downloads/Series/Season01"), + ("stat", "/downloads/Series/Season01/old.mkv"), + ("type", "/downloads/Series"), + ], +) +def test_snapshot_query_failure_is_not_an_empty_or_partial_tree(snapshot_server, operation, path): + """根目录、子目录或文件元数据读取失败时，整个监控路径都不能提交为有效快照。""" + state = snapshot_server + baseline = state.storage.snapshot(state.root) + state.fault.update(operation=operation, path=path) + + assert state.module.snapshot_storage( + "smb", state.root, last_snapshot_time=0, previous_snapshot=baseline, + ) is None + assert len(baseline) == 1 + + +def test_disconnected_snapshot_preserves_failure_semantics(snapshot_server): + """连接失效不能冒充目录已删除；普通文件浏览仍沿用空列表回退。""" + state = snapshot_server + state.storage._connected = False + + assert state.module.snapshot_storage("smb", state.root) is None + assert state.storage.list(FileItem(storage="smb", type="dir", path="/downloads")) == [] + + +@pytest.mark.parametrize("root_missing", [False, True]) +def test_confirmed_empty_snapshot_is_success(snapshot_server, root_missing): + """确认目录已清空或根目录已删除时应返回空快照，而不是失败。""" + state = snapshot_server + baseline = state.storage.snapshot(state.root) + if root_missing: + state.tree.clear() + else: + del state.tree["/downloads/Series/Season01/old.mkv"] + + assert state.module.snapshot_storage( + "smb", state.root, last_snapshot_time=100, previous_snapshot=baseline, + ) == {} + + +def test_snapshot_keeps_depth_limit(snapshot_server): + """遍历不因 mtime 剪枝，但仍遵守调用方的递归深度预算。""" + state = snapshot_server + state.client.listdir.reset_mock() + + assert state.storage.snapshot(state.root, max_depth=2) == {} + assert state.client.listdir.call_count == 2 + + +def test_poll_retries_failed_smb_scan_without_losing_baseline(snapshot_server, monkeypatch): + """SMB 断线时保留旧基线，恢复后发现保留旧 mtime 的移入文件且只派发一次。""" + state = snapshot_server + records = {} + cache = Mock() + cache.get.side_effect = lambda key, **_kwargs: records.get(key) + cache.set.side_effect = lambda key, value, **_kwargs: records.__setitem__(key, value) + store = SnapshotStore(cache=cache) + baseline = state.storage.snapshot(state.root) + assert store.save("smb", baseline, file_count=1, last_snapshot_time=100) + saved_baseline = store.load("smb") + dispatcher = Mock() + dispatcher.is_transfer_candidate_path.return_value = True + dispatcher.handle_file.return_value = True + chain = SimpleNamespace(snapshot_storage=state.module.snapshot_storage) + monkeypatch.setattr("app.monitor.poller.StorageChain", lambda: chain) + poller = RemotePoller(store=store, dispatcher=dispatcher) + new_path = "/downloads/Series/Season01/new.mkv" + state.tree[new_path] = {"directory": False, "mtime": 1, "size": 20} + state.fault.update(operation="list", path="/downloads/Series/Season01") + + assert poller.poll("smb", [state.root]) is None + assert store.load("smb") == saved_baseline + dispatcher.handle_file.assert_not_called() + + state.fault.update(operation=None, path=None) + assert poller.poll("smb", [state.root]) == 2 + dispatcher.handle_file.assert_called_once() + assert dispatcher.handle_file.call_args.kwargs["event_path"] == Path(new_path) + assert poller.poll("smb", [state.root]) == 2 + dispatcher.handle_file.assert_called_once() diff --git a/tests/test_storage_strict_query_coverage.py b/tests/test_storage_strict_query_coverage.py index eb7b9c7d2..1fcdd3aff 100644 --- a/tests/test_storage_strict_query_coverage.py +++ b/tests/test_storage_strict_query_coverage.py @@ -19,6 +19,14 @@ def test_base_get_item_strict_fails_conservatively(): StorageBase.get_item_strict(stub, Path("/media/示例.mkv")) +def test_base_list_strict_fails_conservatively(): + """未适配严格目录查询的存储不能把宽松查询的空结果当作完整快照。""" + stub = SimpleNamespace(schema="dummy") + + with pytest.raises(StorageQueryError): + StorageBase.list_strict(stub, SimpleNamespace(path="/media")) + + @pytest.mark.parametrize("module_name", BUILTIN_STORAGE_MODULES) def test_builtin_storage_overrides_strict_query(module_name): """每个在树存储都必须覆写严格查询，否则整理会被基类保守拒绝。"""