nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-15 12:50:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(transfer): tolerate ctime drift on FUSE mounts (#7397)

Browse Source
This commit is contained in:
jxxghp
2026-11-13 09:41:07 +08:00
committed by GitHub
parent 74dd42aaef
commit 647c5f4da1
7 changed files with 362 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
app/adapters/system/fsproxy.py
View File

@@ -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]:
"""
列出目录条目名。

57
app/modules/filemanager/transhandler.py
View File

@@ -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:

2
docs/architecture-overview.md
View File

@@ -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） |

2
docs/mcp-api.md
View File

@@ -254,6 +254,8 @@ Discord 仅指定 `userid` 的定向通知使用私聊；无法私聊时返回
`FileItem` 的可选 `modify_time_ns` 和 `change_time_ns` 保存本地纳秒版本证据，
JSON 使用十进制字符串避免浏览器数字精度丢失，客户端应原样保留查询返回值，不自行生成。
恢复清理发现本地源或目标版本变化时会保留副本并转人工复核；
对于确认为 FUSE 的本地挂载，同路径文件的大小、类型、稳定文件 ID 和纳秒修改时间均一致时，
单独的状态时间变化会记录告警并继续清理。缺少上述凭据或挂载查询失败时仍需复核。
云盘时间戳变化仍按大小与稳定文件标识核验。OpenList 中断或大小不符的下载不会作为移动成功结果删除源文件。
`storage.rename` 的 `recursive=true` 会识别并重命名所有子目录内媒体，随后重命名所选目录。

5
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -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",

72
tests/test_fs_proxy.py
View File

@@ -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):
"""
这是整个代理存在的意义：挂载不返回时，调用方必须在有限时间内拿到异常，

222
tests/test_transfer_fuse_identity.py Normal file
View File

@@ -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)