diff --git a/app/adapters/system/fsproxy.py b/app/adapters/system/fsproxy.py index 085cc8fc6..60371e741 100644 --- a/app/adapters/system/fsproxy.py +++ b/app/adapters/system/fsproxy.py @@ -28,6 +28,8 @@ import threading from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Union, cast +import psutil + from app.runtime.log import logger from app.runtime.settings import get_runtime_setting @@ -118,6 +120,26 @@ class FileSystemProxy: """ return Path(self._call("realpath", path=str(path))) + def is_fuse_path(self, path: Path) -> bool: + """ + 按实际路径所属的最深挂载点识别 FUSE 文件系统。 + + 符号链接解析沿用代理超时边界。挂载表包含容器绑定挂载和虚拟文件系统, + 不缓存结果,避免重挂载后沿用旧类型。查询失败由调用方保留文件并复核。 + + :param path: 待核对的本地文件路径 + :return: 所属挂载类型为 fuse、fuseblk 或极空间 zfuse(含子类型)时返回 True + :raises OSError: 路径解析超时或挂载表读取失败 + """ + resolved = self.realpath(path) + partition = max( + (entry for entry in psutil.disk_partitions(all=True) + if resolved.is_relative_to(Path(entry.mountpoint))), + key=lambda entry: len(Path(entry.mountpoint).parts), + default=None, + ) + return partition is not None and partition.fstype.split(".", 1)[0] in {"fuse", "fuseblk", "zfuse"} + def listdir(self, path: Path) -> List[str]: """ 列出目录条目名。 diff --git a/app/modules/filemanager/transhandler.py b/app/modules/filemanager/transhandler.py index 271babbaa..bd4f3d46e 100644 --- a/app/modules/filemanager/transhandler.py +++ b/app/modules/filemanager/transhandler.py @@ -11,6 +11,7 @@ from typing import Any, Callable, List, Optional, Tuple, Union, cast from jinja2 import Template +from app.adapters.system.fsproxy import fsproxy from app.adapters.system.host import SystemUtils from app.application.audio import AudioMetadataHelper from app.application.classification.reference import ( @@ -844,7 +845,8 @@ class TransHandler: 清理前比对冻结文件的大小、类型及可用身份;证据不足时保留文件。 本地链接的查询沿用 stat 语义,硬链接和软链接均比较实际内容的身份。 - 本地必须核对修改时间,优先纳秒证据;同路径还核对状态时间,防止等长改写后还原 mtime。 + 本地必须核对修改时间,优先纳秒证据;同路径还核对状态时间。 + FUSE 可在写入收尾后单独更新状态时间,完整身份一致时按挂载类型兼容。 rename 合法改变状态时间,因此跨路径只比较修改时间。旧本地快照回退到秒级修改时间, 缺少时间凭据时转人工复核。云盘修改时间可能在上传后变化或缺失,不参与校验。 @@ -867,21 +869,44 @@ class TransHandler: if expected.fileid != current.fileid: raise TransferExecutionConflictError(f"文件 {expected.path} 已被替换,禁止清理") if expected.storage == "local": - expected_time: Optional[int | float] - current_time: Optional[int | float] - if expected.modify_time_ns is not None: - expected_time, current_time = expected.modify_time_ns, current.modify_time_ns - else: - expected_time, current_time = expected.modify_time, current.modify_time - if expected_time is None or current_time is None: - raise StorageQueryError(f"文件 {expected.path} 缺少修改时间身份凭据,需人工复核") - if expected_time != current_time: - raise TransferExecutionConflictError(f"文件 {expected.path} 的修改时间已变化,禁止清理") - if expected.path == current.path and expected.change_time_ns is not None: - if current.change_time_ns is None: - raise StorageQueryError(f"文件 {expected.path} 缺少状态时间身份凭据,需人工复核") - if expected.change_time_ns != current.change_time_ns: - raise TransferExecutionConflictError(f"文件 {expected.path} 的状态时间已变化,禁止清理") + TransHandler.__assert_local_file_version(expected, current) + + @staticmethod + def __assert_local_file_version(expected: FileItem, current: FileItem) -> None: + """ + 在大小、类型和稳定 ID 已核对后,检查本地文件的时间版本凭据。 + + 同路径状态时间可识别等长改写后还原 mtime 的变化。仅在稳定 ID 和纳秒修改时间 + 均一致且当前挂载确认为 FUSE 时,允许单独的 ctime 变化并记录告警。 + 旧快照凭据不足、挂载未知或查询失败时继续保留文件。 + + :param expected: 已完成基础身份核验的冻结本地文件 + :param current: 当前严格查询结果 + :raises TransferExecutionConflictError: 修改时间或不满足兼容条件的状态时间变化 + :raises StorageQueryError: 缺少冻结快照要求的时间凭据 + :raises OSError: FUSE 挂载识别失败 + """ + expected_time: Optional[int | float] + current_time: Optional[int | float] + if expected.modify_time_ns is not None: + expected_time, current_time = expected.modify_time_ns, current.modify_time_ns + else: + expected_time, current_time = expected.modify_time, current.modify_time + if expected_time is None or current_time is None: + raise StorageQueryError(f"文件 {expected.path} 缺少修改时间身份凭据,需人工复核") + if expected_time != current_time: + raise TransferExecutionConflictError(f"文件 {expected.path} 的修改时间已变化,禁止清理") + if expected.path != current.path or expected.change_time_ns is None: + return + if current.change_time_ns is None: + raise StorageQueryError(f"文件 {expected.path} 缺少状态时间身份凭据,需人工复核") + if expected.change_time_ns == current.change_time_ns: + return + if (expected.fileid and expected.modify_time_ns is not None + and fsproxy.is_fuse_path(Path(cast(str, current.path)))): + logger.warning(f"文件 {expected.path} 的 FUSE 状态时间已变化,大小、类型、文件 ID 和纳秒修改时间一致,继续清理") + return + raise TransferExecutionConflictError(f"文件 {expected.path} 的状态时间已变化,禁止清理") @classmethod def __inherit_shared_local_identity(cls, expected: FileItem, receipt: Optional[FileItem]) -> FileItem: diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 3dee77002..3ec30ab46 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -837,7 +837,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1164 | -| 内部导入边 | 10,123 | +| 内部导入边 | 10,124 | | 非平凡 SCC | 1(29 个模块的 TMDB vendor 精确 containment) | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51(精确 policy 准入,Application / Chain 均为 0) | diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 247fa1eb3..69f6b44d3 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -254,6 +254,8 @@ Discord 仅指定 `userid` 的定向通知使用私聊;无法私聊时返回 `FileItem` 的可选 `modify_time_ns` 和 `change_time_ns` 保存本地纳秒版本证据, JSON 使用十进制字符串避免浏览器数字精度丢失,客户端应原样保留查询返回值,不自行生成。 恢复清理发现本地源或目标版本变化时会保留副本并转人工复核; +对于确认为 FUSE 的本地挂载,同路径文件的大小、类型、稳定文件 ID 和纳秒修改时间均一致时, +单独的状态时间变化会记录告警并继续清理。缺少上述凭据或挂载查询失败时仍需复核。 云盘时间戳变化仍按大小与稳定文件标识核验。OpenList 中断或大小不符的下载不会作为移动成功结果删除源文件。 `storage.rename` 的 `recursive=true` 会识别并重命名所有子目录内媒体,随后重命名所选目录。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index c3f06210d..8283e9547 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -962,8 +962,8 @@ "runtime_only": true } }, - "edge_count": 10123, - "edge_sha256": "8d8cbb23d149ab386fb65d8766391d81b1f8e37484f52d4640f92c07c722167e", + "edge_count": 10124, + "edge_sha256": "3dac06b5d1ba3e44d6ed06f06c678f43767e1e8ee613cd4b7d2d09c07ccfd463", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -7753,6 +7753,7 @@ "app.modules.filemanager.storages.u115 -> app.schemas.types", "app.modules.filemanager.transhandler -> app.adapters", "app.modules.filemanager.transhandler -> app.adapters.system", + "app.modules.filemanager.transhandler -> app.adapters.system.fsproxy", "app.modules.filemanager.transhandler -> app.adapters.system.host", "app.modules.filemanager.transhandler -> app.application", "app.modules.filemanager.transhandler -> app.application.audio", diff --git a/tests/test_fs_proxy.py b/tests/test_fs_proxy.py index 04fead913..eedbfd08e 100644 --- a/tests/test_fs_proxy.py +++ b/tests/test_fs_proxy.py @@ -14,7 +14,10 @@ import subprocess import sys import time from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock +import psutil import pytest from app.adapters.system.fsproxy import FileSystemProxy, FileSystemTimeout @@ -63,6 +66,73 @@ def test_realpath_resolves_parent_alias_for_missing_target(tmp_path, proxy, monk assert proxy.realpath(alias / "new.mkv") == real.resolve() / "new.mkv" +@pytest.mark.parametrize("fstype, expected", [ + ("fuse", True), ("fuse.zfsv3", True), ("zfuse.zfsv3", True), + ("fuse.shfs", True), ("fuseblk", True), ("fuseblk.ntfs", True), + ("ext4", False), ("zfs", False), ("overlay", False), ("nfs", False), ("fusectl", False), +]) +def test_is_fuse_path_uses_innermost_mount(tmp_path, proxy, monkeypatch, fstype, expected): + """ + 识别容器实际绑定挂载的类型,目录前缀相似的另一挂载不能覆盖当前路径。 + """ + root = tmp_path.resolve() + partitions = [ + SimpleNamespace(mountpoint=str(root.parent), fstype="overlay"), + SimpleNamespace(mountpoint=str(root), fstype=fstype), + SimpleNamespace(mountpoint=str(root) + "-other", fstype="fuse.sshfs"), + ] + query = Mock(return_value=list(reversed(partitions))) + monkeypatch.setattr(psutil, "disk_partitions", query) + + assert proxy.is_fuse_path(root / "Movie.mkv") is expected + query.assert_called_once_with(all=True) + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_is_fuse_path_resolves_symlink_to_mount_with_spaces(tmp_path, proxy, monkeypatch, enabled): + """ + 含空格的挂载路径及目录别名按链接目标识别,代理关闭时沿用相同语义。 + """ + root = tmp_path.resolve() + mount = root / "fuse volume" + mount.mkdir() + alias = root / "alias" + alias.symlink_to(mount, target_is_directory=True) + partitions = [ + SimpleNamespace(mountpoint=str(root), fstype="ext4"), + SimpleNamespace(mountpoint=str(mount), fstype="zfuse.zfsv3"), + ] + monkeypatch.setattr(psutil, "disk_partitions", lambda **_kwargs: partitions) + monkeypatch.setattr(FileSystemProxy, "_enabled", staticmethod(lambda: enabled)) + + assert proxy.is_fuse_path(alias / "Movie.mkv") + + +def test_is_fuse_path_refreshes_mount_table_and_keeps_unknown_type(tmp_path, proxy, monkeypatch): + """ + 挂载类型改变立即生效,挂载表缺少匹配项时不能认定为 FUSE。 + """ + root = tmp_path.resolve() + partition = SimpleNamespace(mountpoint=str(root), fstype="zfuse.zfsv3") + query = Mock(side_effect=[[partition], [SimpleNamespace(mountpoint=str(root), fstype="ext4")], []]) + monkeypatch.setattr(psutil, "disk_partitions", query) + + assert proxy.is_fuse_path(root / "Movie.mkv") + assert not proxy.is_fuse_path(root / "Movie.mkv") + assert not proxy.is_fuse_path(root / "Movie.mkv") + assert query.call_count == 3 + + +def test_is_fuse_path_propagates_mount_query_failure(tmp_path, proxy, monkeypatch): + """ + 无法读取挂载表时保留查询错误,由整理层保留文件并进入未知态复核。 + """ + monkeypatch.setattr(psutil, "disk_partitions", Mock(side_effect=OSError("mount query failed"))) + + with pytest.raises(OSError, match="mount query failed"): + proxy.is_fuse_path(tmp_path) + + def test_stat_on_directory(tmp_path, proxy): """ 目录的 stat 要正确标记类型。 @@ -161,7 +231,7 @@ def test_rename_symlink_uses_link_device_not_target_device(tmp_path, proxy): # 核心价值:超时可放弃 # --------------------------------------------------------------------------- # -@pytest.mark.parametrize("operation", ["stat", "realpath"]) +@pytest.mark.parametrize("operation", ["stat", "realpath", "is_fuse_path"]) def test_timeout_raises_and_kills_worker(tmp_path, monkeypatch, operation): """ 这是整个代理存在的意义:挂载不返回时,调用方必须在有限时间内拿到异常, diff --git a/tests/test_transfer_fuse_identity.py b/tests/test_transfer_fuse_identity.py new file mode 100644 index 000000000..7c4548128 --- /dev/null +++ b/tests/test_transfer_fuse_identity.py @@ -0,0 +1,222 @@ +""" +验证 FUSE 延迟更新状态时间时的整理清理与文件身份保护。 +""" + +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import psutil +import pytest + +from app.application.transfer.execution import ( + TransferExecutionCommand, + TransferExecutionConflictError, + TransferExecutionState, + TransferManualReviewDecision, + TransferStepState, +) +from app.chain.transfer.execution import _TransferManualReviewRequired +from app.db.adapters.transfer.admission import TransactionalTransferAdmissionRepository +from app.domain.context import MediaInfo +from app.domain.meta.metavideo import MetaVideo +from app.modules.filemanager.storages.local import LocalStorage +from app.modules.filemanager.transhandler import TransHandler +from app.schemas.exception import StorageQueryError +from app.schemas.file import FileItem +from app.schemas.types import MediaType +from tests.test_transfer_safe_overwrite import _checkpoint, _durable_runner +from tests.test_transfer_safe_overwrite import repository_factory as _repository_factory + + +@pytest.fixture +def repository_factory(): + """ + 为完整整理流程提供临时 SQLite 步骤账本,并在用例后释放引擎。 + """ + yield from _repository_factory.__wrapped__() + + +@pytest.fixture +def fuse_mount(tmp_path, monkeypatch): + """ + 将用例临时目录报告为容器内的 zfuse 挂载,不触碰宿主实际挂载。 + """ + partitions = [SimpleNamespace(mountpoint=str(tmp_path.resolve()), fstype="zfuse.zfsv3")] + query = Mock(return_value=partitions) + monkeypatch.setattr(psutil, "disk_partitions", query) + return query + + +def test_fuse_cleanup_allows_only_ctime_drift(tmp_path, fuse_mount, monkeypatch): + """ + FUSE 同路径文件的完整身份一致时,单独变化的状态时间允许继续清理。 + """ + path = tmp_path / "source.mkv" + path.write_bytes(b"movie") + expected = object.__new__(LocalStorage).get_item_strict(path) + current = expected.model_copy(update={"change_time_ns": expected.change_time_ns + 5_000_000_000}) + warning = Mock() + monkeypatch.setattr("app.modules.filemanager.transhandler.logger.warning", warning) + + TransHandler._TransHandler__assert_file_identity(expected, current) + + fuse_mount.assert_called_once_with(all=True) + assert "FUSE 状态时间已变化" in warning.call_args.args[0] + + +@pytest.mark.parametrize("change", [ + {"size": 6}, {"type": "dir"}, {"fileid": "other"}, {"fileid": None}, + {"modify_time_ns": 1}, {"modify_time_ns": None}, {"change_time_ns": None}, +]) +def test_fuse_cleanup_rejects_other_identity_changes(tmp_path, fuse_mount, change): + """ + FUSE 的状态时间兼容不能接纳大小、类型、inode、修改时间变化或凭据丢失。 + """ + path = tmp_path / "source.mkv" + path.write_bytes(b"movie") + expected = object.__new__(LocalStorage).get_item_strict(path) + current = expected.model_copy(update={"change_time_ns": expected.change_time_ns + 1, **change}) + + with pytest.raises((TransferExecutionConflictError, StorageQueryError)): + TransHandler._TransHandler__assert_file_identity(expected, current) + + fuse_mount.assert_not_called() + + +@pytest.mark.parametrize("missing", ["fileid", "modify_time_ns"]) +def test_fuse_legacy_snapshot_without_strong_identity_keeps_ctime_check(tmp_path, fuse_mount, missing): + """ + 旧快照缺少稳定文件 ID 或纳秒修改时间时,状态时间变化仍需人工复核。 + """ + path = tmp_path / "source.mkv" + path.write_bytes(b"movie") + expected = object.__new__(LocalStorage).get_item_strict(path).model_copy(update={missing: None}) + current = expected.model_copy(update={"change_time_ns": expected.change_time_ns + 1}) + + with pytest.raises(TransferExecutionConflictError, match="状态时间已变化"): + TransHandler._TransHandler__assert_file_identity(expected, current) + + fuse_mount.assert_not_called() + + +@pytest.mark.parametrize("resume", ["fresh", "restart", "manual_review"]) +def test_move_cleans_fuse_source_after_upload_and_durable_resume( + tmp_path, monkeypatch, repository_factory, fuse_mount, resume, +): + """ + 上传后仅源 ctime 延迟变化时,首次执行、重启及人工确认后重试均完成删除且不重复上传。 + """ + source, target = tmp_path / "source.mkv", tmp_path / "library" / "Movie.mkv" + source.write_bytes(b"movie") + target.parent.mkdir() + local = object.__new__(LocalStorage) + query = local.get_item_strict + uploaded = False + + def query_source(path): + """ + 上传结束后回放底层单独更新源文件状态时间,大小和内容版本保持一致。 + """ + item = query(path) + if item is not None and Path(path) == source and uploaded: + return item.model_copy(update={"change_time_ns": item.change_time_ns + 5_000_000_000}) + return item + + def query_target(path): + """ + 以独立临时副本提供远端严格查询结果。 + """ + item = query(path) + return item.model_copy(update={"storage": "remote"}) if item else None + + def upload(_folder, path, _name): + """ + 写入完整目标,再使源文件的后续查询体现状态时间变化。 + """ + nonlocal uploaded + target.write_bytes(path.read_bytes()) + uploaded = True + return query_target(target) + + remote = Mock() + remote.get_folder.return_value = FileItem(storage="remote", path=str(target.parent), type="dir") + remote.get_item_strict.side_effect = query_target + remote.upload.side_effect = upload + monkeypatch.setattr(local, "get_item_strict", query_source) + checkpoint = _checkpoint(source, target, transfer_type="move", mode="never") + checkpoint = replace( + checkpoint, target_storage="remote", + items=(replace(checkpoint.items[0], target_storage="remote"),), + ) + repository = repository_factory(checkpoint) + runner = _durable_runner(repository, checkpoint) + handler = TransHandler() + monkeypatch.setattr(handler, "_TransHandler__intercept_transfer", lambda **_kwargs: (True, "")) + + def execute(step_runner): + """ + 从正常计划执行入口运行真实步骤及源删除。 + """ + return handler.execute_transfer_plan( + checkpoint, meta=MetaVideo("Movie.2026.mkv"), + mediainfo=MediaInfo(type=MediaType.MOVIE, title="Movie", year="2026"), + source_oper=local, target_oper=remote, step_runner=step_runner, + ) + + if resume == "restart": + class ProcessStopped(BaseException): + """ + 模拟上传回执落账后、源删除开始前进程退出。 + """ + + run = runner.run + + def stop_before_delete(**kwargs): + """ + 精确停在独立删除步骤之前,保留已成功的上传回执。 + """ + if kwargs["kind"] == "delete_move_source": + raise ProcessStopped() + return run(**kwargs) + + monkeypatch.setattr(runner, "run", stop_before_delete) + with pytest.raises(ProcessStopped): + execute(runner) + assert source.exists() and target.read_bytes() == b"movie" + runner = _durable_runner(repository, checkpoint) + + if resume == "manual_review": + partitions = fuse_mount.return_value + fuse_mount.return_value = [] + with pytest.raises(_TransferManualReviewRequired, match="状态时间已变化"): + execute(runner) + snapshot = repository.get_snapshot(task_id="safe-overwrite") + assert snapshot.state is TransferExecutionState.MANUAL_REVIEW + deletion = next(step for step in snapshot.steps if step.kind == "delete_move_source") + resolved = TransferExecutionCommand(repository).resolve_manual_review( + task_id="safe-overwrite", operation_id=deletion.operation_id, + decision=TransferManualReviewDecision.NOT_APPLIED, actor="admin", reason="确认源和已上传目标完整", + ) + assert resolved.state is TransferExecutionState.RETRY_WAIT + claimed = TransactionalTransferAdmissionRepository(repository._session_factory).claim_task( + task_id="safe-overwrite", owner_id="recovery", lease_seconds=120, + ) + runner = type(runner)( + task_id="safe-overwrite", lease_token=claimed.lease_token, + checkpoint_fingerprint=checkpoint.fingerprint, repository=repository, + ) + fuse_mount.return_value = partitions + fuse_mount.reset_mock() + + result = execute(runner) + + assert result.success + assert not source.exists() + assert target.read_bytes() == b"movie" + remote.upload.assert_called_once() + deletion = next(step for step in repository.get_snapshot(task_id="safe-overwrite").steps + if step.kind == "delete_move_source") + assert deletion.state is TransferStepState.SUCCEEDED + fuse_mount.assert_called_once_with(all=True)