nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-10-28 19:06:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(storage): retain history until target cleanup is confirmed (#7202)

Browse Source
This commit is contained in:
Aqr-K
2026-10-27 05:42:07 +08:00
committed by GitHub
parent 42edfd1da8
commit 71c2e9e3df
15 changed files with 302 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
app/chain/storage.py
View File

@@ -8,7 +8,9 @@ from typing import Any, Dict, List, Optional, Sequence, cast
from app.application.directory import DirectoryHelper
from app.chain.base import ChainBase
from app.runtime.log import logger
from app.schemas.exception import StorageQueryError
from app.schemas.file import FileItem as _SchemaFileItem
from app.schemas.file import FileItemQueryResult
class StorageChain(ChainBase):
@@ -123,16 +125,23 @@ class StorageChain(ChainBase):
path: Path,
) -> Optional[_SchemaFileItem]:
"""
严格查询文件项：确认不存在返回空，能力提供方 或 I/O 失败直接抛出。
严格查询文件项，只有实际查询回执中的空值才表示目标不存在。
:param storage: 目标存储标识
:param path: 存储内的文件或目录路径
:return: 确认存在的文件项或确认不存在时的 None
:raises StorageQueryError: 没有可确认状态的提供方或回执不合法时抛出
"""
return cast(
Optional[_SchemaFileItem],
self.run_module_strict(
"get_file_item",
storage=storage,
path=path,
),
)
result = self.run_module_strict("get_file_item_strict", storage=storage, path=path)
if isinstance(result, FileItemQueryResult):
return result.item
if result is not None:
raise StorageQueryError(f"存储 {storage} 的严格查询未返回有效回执")
# 旧插件仍能证明非空命中；其空返回值无法区分“不支持”和“不存在”。
legacy_item = self.run_module_strict("get_file_item", storage=storage, path=path)
if isinstance(legacy_item, _SchemaFileItem):
return legacy_item
raise StorageQueryError(f"存储 {storage} 没有能够确认目标状态的查询提供方")
def get_parent_item(self, fileitem: _SchemaFileItem) -> Optional[_SchemaFileItem]:
"""

33
app/chain/transfer/records.py
View File

@@ -564,6 +564,11 @@ class ManualHistoryMixin(_TransferOwnerBase):
用户已明确重新整理，缺失结算版本或停在重试、人工复核态的旧记录也必须
能放弃；普通重试仍由持久调度器沿原计划继续，有效租约拒绝任何清理。
目标状态查询失败时保留历史与失败计数，不把未知状态当作目标已消失。
:param history: 用户明确要求重新整理的历史快照
:param transfer_history_oper: 删除历史所用的仓储端口
:return: 是否完成旧历史清理及可向用户展示的失败原因
"""
task_id = getattr(history, "transfer_task_id", None)
if task_id:
@@ -582,16 +587,26 @@ class ManualHistoryMixin(_TransferOwnerBase):
if not isinstance(history.dest_fileitem, dict):
return False, "目标文件历史数据无效，请删除该整理记录后从文件管理重新整理"
dest_fileitem = FileItem.model_validate(history.dest_fileitem)
if not dest_fileitem.storage or not dest_fileitem.path:
return False, "目标文件历史数据缺少存储或路径，请检查该整理记录"
storage_chain = StorageChain()
if storage_chain.exists(dest_fileitem):
deleted = storage_chain.delete_media_file(dest_fileitem)
# 远端存储可能在删除请求返回前由并发任务或生命周期策略移除文件。
# 只要读回确认目标已不存在，就按幂等删除继续，不能留下永远无法重整的历史。
if not deleted and storage_chain.exists(dest_fileitem):
return False, (
f"旧目标文件 {dest_fileitem.path} 删除失败，请检查媒体库权限，"
"或手动删除该文件后重试"
)
try:
target = storage_chain.get_file_item_strict(
storage=dest_fileitem.storage, path=Path(dest_fileitem.path),
)
if target:
deleted = storage_chain.delete_media_file(dest_fileitem)
# 删除失败时，只有严格查询确认目标消失，才接受并发移除后的幂等成功。
if not deleted and storage_chain.get_file_item_strict(
storage=dest_fileitem.storage, path=Path(dest_fileitem.path),
):
return False, (
f"旧目标文件 {dest_fileitem.path} 删除失败，请检查媒体库权限，"
"或手动删除该文件后重试"
)
except Exception: # pylint: disable=broad-except
return False, f"无法确认旧目标文件 {dest_fileitem.path} 的状态，请检查存储后重试"
transfer_history_oper.delete(history.id)
# 删除记录是用户显式要求重来，失败计数一并清零，否则重整仍会受上一轮次数限制
clear_transfer_failures(history.src, history.src_storage)

19
app/modules/filemanager/module.py
View File

@@ -22,7 +22,8 @@ from app.modules.filemanager.storages import StorageBase
from app.modules.filemanager.transhandler import TransHandler
from app.runtime.log import logger
from app.runtime.settings import get_runtime_setting
from app.schemas.file import FileItem, StorageUsage
from app.schemas.exception import StorageQueryError
from app.schemas.file import FileItem, FileItemQueryResult, StorageUsage
from app.schemas.mediaserver import ExistMediaInfo
from app.schemas.system import TransferDirectoryConf
from app.schemas.tmdb import TmdbEpisode
@@ -477,6 +478,22 @@ class FileManagerModule(_ModuleBase):
return None
return storage_oper.get_item_strict(path)
def get_file_item_strict(self, storage: str, path: Path) -> Optional[FileItemQueryResult]:
"""
为支持的存储返回明确查询回执，不支持时让调度器继续寻找其他提供方。
:param storage: 目标存储标识
:param path: 存储内的文件或目录路径
:return: 已完成查询的回执；不支持该存储时返回 None
:raises StorageQueryError: 已注册存储的适配器无法取得时抛出
"""
if storage not in self._support_storages:
return None
storage_oper = self.__get_storage_oper(storage)
if storage_oper is None:
raise StorageQueryError(f"存储 {storage} 的文件查询适配器不可用")
return FileItemQueryResult(item=storage_oper.get_item_strict(path))
def get_parent_item(self, fileitem: FileItem) -> Optional[FileItem]:
"""
通过对应存储查询上级目录。

8
app/runtime/extensions/module/contracts.py
View File

@@ -373,6 +373,14 @@ _METHOD_CONTRACTS = {
aggregation=ModuleResultAggregation.FIRST_NON_EMPTY,
required_parameters=("storage", "path"),
),
"get_file_item_strict": ModuleMethodContract(
family="storage",
input_contract="StorageItemRequest",
result_contract="FileItemQueryResult | None",
result_shape=ModuleResultShape.MODEL,
aggregation=ModuleResultAggregation.FIRST_NON_EMPTY,
required_parameters=("storage", "path"),
),
"get_folder": ModuleMethodContract(
family="storage",
input_contract="StorageFolderRequest",

1
app/schemas/exports.py
View File

@@ -181,6 +181,7 @@ SCHEMA_EXPORTS = {
'Field': ('app.schemas.mcp', 'Field'),
'FileContextSnapshot': ('app.schemas.event', 'FileContextSnapshot'),
'FileItem': ('app.schemas.workflow', 'FileItem'),
'FileItemQueryResult': ('app.schemas.file', 'FileItemQueryResult'),
'FileNameData': ('app.schemas.common', 'FileNameData'),
'FileURI': ('app.schemas.file', 'FileURI'),
'FilterRuleGroup': ('app.schemas.system', 'FilterRuleGroup'),

10
app/schemas/file.py
View File

@@ -84,6 +84,16 @@ class FileItem(FileURI):
return str(value) if value is not None else None
class FileItemQueryResult(BaseModel):
"""
声明提供方已实际查询目标存储，空文件项表示已确认不存在。
与模块方法返回 None（此提供方不支持目标存储）区分，避免未知态被用于删除决策。
"""
item: Optional[FileItem] = Field(..., description="实际查询结果；None 仅表示已确认文件或目录不存在")
class StorageUsage(BaseModel):
"""
存储空间使用情况。

2
docs/architecture-overview.md
View File

@@ -817,7 +817,7 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
| Python 模块 | 1142 |
| 内部导入边 | 9,837 |
| 内部导入边 | 9,839 |
| 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） |
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
| Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） |

8
docs/rules/05-architecture.md
View File

@@ -804,6 +804,14 @@ SMB 同存储整理由 `app.modules.filemanager.storages.smb` 持有协议调用
本地文件 ID 使用设备号和 inode，验证不额外读取整文件计算哈希。
外部 legacy transfer provider 继续遵循原有兼容合同。
严格文件查询使用模块能力 `get_file_item_strict(storage, path)`。提供方不支持目标存储时返回
`None`，调度继续寻找下一提供方；真正完成查询时返回 `FileItemQueryResult(item=...)`，
其中 `item=None` 才能作为目标不存在的证据。存储不可用或查询失败必须抛出。
`StorageChain.get_file_item_strict` 仅接受明确回执；旧插件的普通能力非空命中继续兼容，
但空结果没有存储归属证明，不能用于删除历史等破坏性决策。旧插件如需声明“已确认不存在”，
应实现新能力并返回回执；普通 `get_file_item` 的 nullable 查询合同不变。
显式重新整理在清理旧目标前及删除失败后均使用严格查询，未知态保留历史和失败计数。
SMB 目录快照不依赖祖先目录 mtime 推断后代变化，而在既有递归深度上限内逐层列举。
存储通过 `snapshot_strict_query` 声明已实现 `get_item_strict` / `list_strict`；
严格快照的任一查询失败必须返回 `None`，由远程轮询器保留旧基线并重试。

6
docs/testing.md
View File

@@ -147,6 +147,12 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)):
115 快照不再按目录时间剪枝；扫描间隔、深度和接口限流不变，但每轮可能增加目录列表请求。
这些用例不代表 115 已接入严格查询或验证了查询失败时的快照原子性。
### 存储严格查询回执回归
`tests/test_storage_cleanup_receipts.py` 通过真实 StorageChain、模块调度器和 FileManager，
验证未知存储、未初始化模块、缺失模块和适配器不可用均保留整理历史。自定义存储提供方
放在插件优先路径或宿主前后顺序时，都须能传递明确的文件项/不存在回执；普通查询仍可返回空值。
## 种子下载重试回归
下载选集回归使用真实 qBittorrent/Transmission Module 与离线下载器端口，验证自定义识别词

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

@@ -959,8 +959,8 @@
"runtime_only": true
}
},
"edge_count": 9837,
"edge_sha256": "d51ff72057e0c293893d968ca72a5f76f33c0c89d4d05a31c87110ba3b2cfc56",
"edge_count": 9839,
"edge_sha256": "72918db401db3ef24d1360aa2f29169d114fa13bebf8e255e807cf38bc5a5082",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -5496,6 +5496,7 @@
"app.chain.storage -> app.runtime",
"app.chain.storage -> app.runtime.log",
"app.chain.storage -> app.schemas",
"app.chain.storage -> app.schemas.exception",
"app.chain.storage -> app.schemas.file",
"app.chain.subscribe.calendar -> app.chain",
"app.chain.subscribe.calendar -> app.chain.media",
@@ -7509,6 +7510,7 @@
"app.modules.filemanager.module -> app.runtime.log",
"app.modules.filemanager.module -> app.runtime.settings",
"app.modules.filemanager.module -> app.schemas",
"app.modules.filemanager.module -> app.schemas.exception",
"app.modules.filemanager.module -> app.schemas.file",
"app.modules.filemanager.module -> app.schemas.mediaserver",
"app.modules.filemanager.module -> app.schemas.system",

20
tests/fixtures/architecture/runtime-contract-baseline.json vendored
View File

@@ -6236,6 +6236,26 @@
"timeout_policy": "caller_budget",
"version": 1
},
"get_file_item_strict": {
"aggregation": "first_non_empty",
"error_policy": "isolate_provider",
"execution": "sync_or_async",
"family": "storage",
"input_contract": "StorageItemRequest",
"optional_parameters": [],
"plugin_short_circuit": true,
"public_to_plugins": true,
"required_parameters": [
"path",
"storage"
],
"result_contract": "FileItemQueryResult | None",
"result_shape": "model",
"supports_async": true,
"supports_sync": true,
"timeout_policy": "caller_budget",
"version": 1
},
"get_folder": {
"aggregation": "first_non_empty",
"error_policy": "isolate_provider",

6
tests/test_manual_transfer_history.py
View File

@@ -166,7 +166,7 @@ def _patch_transfer_planning(monkeypatch, chain, fileitem, history, planned, del
),
)
monkeypatch.setattr("app.chain.transfer.records.StorageChain", lambda: SimpleNamespace(
exists=lambda current_fileitem: True,
get_file_item_strict=lambda **_kwargs: fileitem,
delete_media_file=lambda current_fileitem: deleted.append(
("target", current_fileitem.path)
)
@@ -1020,9 +1020,9 @@ def test_manual_reorganize_accepts_old_target_disappearing_during_delete(monkeyp
transfer_task_id=None, download_hash=None, downloader=None, src=None, src_storage=None,
)
deleted: list[tuple[str, object]] = []
states = iter((True, False))
states = iter((old_dest, None))
monkeypatch.setattr("app.chain.transfer.records.StorageChain", lambda: SimpleNamespace(
exists=lambda _item: next(states),
get_file_item_strict=lambda **_kwargs: next(states),
delete_media_file=lambda _item: False,
))
state, message = chain._delete_manual_transfer_history(

4
tests/test_storage_chain_strict.py
View File

@@ -11,7 +11,7 @@ from app.chain.storage import StorageChain
from app.modules.filemanager.module import FileManagerModule
from app.modules.filemanager.storages.local import LocalStorage
from app.schemas.exception import StorageQueryError
from app.schemas.file import FileItem
from app.schemas.file import FileItem, FileItemQueryResult
def test_storage_chain_strict_query_distinguishes_absent_and_failure() -> None:
@@ -20,7 +20,7 @@ def test_storage_chain_strict_query_distinguishes_absent_and_failure() -> None:
"""
chain = object.__new__(StorageChain)
chain._module_dispatcher = Mock()
chain._module_dispatcher.dispatch_strict.return_value = None
chain._module_dispatcher.dispatch_strict.return_value = FileItemQueryResult(item=None)
assert chain.get_file_item_strict("plugin", Path("/missing.mkv")) is None

176
tests/test_storage_cleanup_receipts.py Normal file
View File

@@ -0,0 +1,176 @@
"""
验证存储严格查询回执与整理历史清理的未知状态和提供方回退。
"""
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock
import pytest
from app.chain.storage import StorageChain
from app.chain.transfer.records import ManualHistoryMixin
from app.modules.filemanager.module import FileManagerModule
from app.runtime.extensions.module.dispatcher import ModuleInvocationDispatcher
from app.schemas.exception import StorageQueryError
from app.schemas.file import FileItem, FileItemQueryResult
@pytest.mark.parametrize("failed_lookup", [0, 1])
def test_manual_cleanup_preserves_history_when_target_state_is_unknown(tmp_path, monkeypatch, failed_lookup):
"""
首次或删除失败后的严格查询出错时，不删除仍有目标的历史及失败计数。
"""
path = tmp_path / "target.mkv"
path.write_bytes(b"existing-media")
item = FileItem(storage="local", path=str(path), type="file")
storage = object.__new__(StorageChain)
calls = []
def query(strict):
"""
在存储模块边界回放普通查询吞错和严格查询抛错的同一故障。
"""
index = len(calls)
calls.append(index)
if index == failed_lookup:
if strict:
raise OSError("storage unavailable")
return None
return item
monkeypatch.setattr(storage, "run_module", lambda *_args, **_kwargs: query(False))
monkeypatch.setattr(storage, "run_module_strict", lambda *_args, **_kwargs: FileItemQueryResult(item=query(True)))
monkeypatch.setattr(storage, "delete_media_file", Mock(return_value=False))
monkeypatch.setattr("app.chain.transfer.records.StorageChain", lambda: storage)
cleared = Mock()
monkeypatch.setattr("app.chain.transfer.records.clear_transfer_failures", cleared)
repository = SimpleNamespace(delete=Mock())
history = SimpleNamespace(id=42, transfer_task_id=None, status=True, mode="copy",
dest_fileitem=item.model_dump(), src="/source", src_storage="local")
success, message = ManualHistoryMixin._delete_manual_transfer_history(SimpleNamespace(), history, repository)
assert not success
assert message
assert path.read_bytes() == b"existing-media"
repository.delete.assert_not_called()
cleared.assert_not_called()
@pytest.mark.parametrize("scenario", ["absent", "delete_success", "delete_failed"])
def test_manual_cleanup_requires_confirmed_target_cleanup(monkeypatch, scenario):
"""
明确不存在或删除成功才清历史；删除失败且目标仍在则保留历史和失败计数。
"""
item = FileItem(storage="local", path="/library/target.mkv", type="file")
storage = object.__new__(StorageChain)
monkeypatch.setattr(storage, "run_module_strict", Mock(return_value=FileItemQueryResult(item=None if scenario == "absent" else item)))
deletion = Mock(return_value=scenario == "delete_success")
monkeypatch.setattr(storage, "delete_media_file", deletion)
monkeypatch.setattr("app.chain.transfer.records.StorageChain", lambda: storage)
cleared = Mock()
monkeypatch.setattr("app.chain.transfer.records.clear_transfer_failures", cleared)
repository = SimpleNamespace(delete=Mock())
history = SimpleNamespace(id=42, transfer_task_id=None, status=True, mode="copy",
dest_fileitem=item.model_dump(), src="/source", src_storage="local")
success, _message = ManualHistoryMixin._delete_manual_transfer_history(SimpleNamespace(), history, repository)
assert success is (scenario != "delete_failed")
assert repository.delete.called is success
assert cleared.called is success
assert deletion.called is (scenario != "absent")
@pytest.mark.parametrize("availability", ["unknown", "uninitialized", "no-module", "no-adapter"])
def test_manual_cleanup_keeps_history_without_available_storage_provider(monkeypatch, availability):
"""
真实模块调度和 FileManager 对未知或未初始化存储不能伪造“目标不存在”。
"""
module = object.__new__(FileManagerModule)
module._support_storages = [] if availability == "uninitialized" else ["local"]
monkeypatch.setattr(module, "_FileManagerModule__get_storage_oper", lambda _storage: None)
modules = [] if availability == "no-module" else [module]
def running_modules(method):
"""
返回实现本次能力的宿主模块快照。
:param method: 正在派发的能力名称
:return: 只包含该能力实现方的模块列表
"""
return [provider for provider in modules if hasattr(provider, method)]
catalog = SimpleNamespace(get_running_modules=running_modules)
dispatcher = ModuleInvocationDispatcher(module_catalog=catalog,
plugin_catalog=SimpleNamespace(get_plugin_modules=lambda: {}), plugin_error_handler=Mock(),
system_error_handler=Mock(), rate_limit_handler=Mock())
storage = object.__new__(StorageChain)
storage._module_dispatcher = dispatcher
monkeypatch.setattr("app.chain.transfer.records.StorageChain", lambda: storage)
cleared = Mock()
monkeypatch.setattr("app.chain.transfer.records.clear_transfer_failures", cleared)
item = FileItem(storage="unknown" if availability == "unknown" else "local", path="/target", type="file")
history = SimpleNamespace(id=42, transfer_task_id=None, status=True, mode="copy",
dest_fileitem=item.model_dump(), src="/source", src_storage="local")
repository = SimpleNamespace(delete=Mock())
success, message = ManualHistoryMixin._delete_manual_transfer_history(SimpleNamespace(), history, repository)
assert not success
assert message
repository.delete.assert_not_called()
cleared.assert_not_called()
# 普通查询继续保留既有 nullable 合同，不把严格用途扩散到普通浏览请求。
assert storage.get_file_item(item.storage, Path(item.path)) is None
@pytest.mark.parametrize("placement", ["plugin", "before-host", "after-host"])
@pytest.mark.parametrize("found", [False, True])
def test_strict_storage_receipt_preserves_provider_fallback(monkeypatch, placement, found):
"""
插件或任意宿主排序的自定义存储回执都能接管，明确不存在不与不适用混淆。
"""
module = object.__new__(FileManagerModule)
module._support_storages = ["local"]
monkeypatch.setattr(module, "_FileManagerModule__get_storage_oper", lambda _storage: None)
expected = FileItem(storage="custom", path="/target") if found else None
provider = SimpleNamespace(get_priority=lambda: 1 if placement == "before-host" else 9,
get_name=lambda: "custom", get_file_item_strict=lambda **_kwargs: FileItemQueryResult(item=expected))
modules = [module] if placement == "plugin" else [module, provider]
plugins = {("custom", "custom"): {"get_file_item_strict": provider.get_file_item_strict}} if placement == "plugin" else {}
def running_modules(method):
"""
返回实现本次能力的宿主模块快照。
:param method: 正在派发的能力名称
:return: 只包含该能力实现方的模块列表
"""
return [entry for entry in modules if hasattr(entry, method)]
catalog = SimpleNamespace(get_running_modules=running_modules)
chain = object.__new__(StorageChain)
chain._module_dispatcher = ModuleInvocationDispatcher(module_catalog=catalog,
plugin_catalog=SimpleNamespace(get_plugin_modules=lambda: plugins), plugin_error_handler=Mock(),
system_error_handler=Mock(), rate_limit_handler=Mock())
assert chain.get_file_item_strict("custom", Path("/target")) == expected
def test_strict_storage_keeps_legacy_nonempty_plugin_lookup(monkeypatch):
"""
未实现新回执的旧插件仍可返回真实文件项，空结果则必须保守拒绝删除用途。
"""
expected = FileItem(storage="custom", path="/target")
lookup = Mock(return_value=expected)
catalog = SimpleNamespace(get_running_modules=lambda _method: [])
chain = object.__new__(StorageChain)
chain._module_dispatcher = ModuleInvocationDispatcher(module_catalog=catalog,
plugin_catalog=SimpleNamespace(get_plugin_modules=lambda: {("legacy", "legacy"): {"get_file_item": lookup}}),
plugin_error_handler=Mock(), system_error_handler=Mock(), rate_limit_handler=Mock())
assert chain.get_file_item_strict("custom", Path("/target")) == expected
lookup.return_value = None
with pytest.raises(StorageQueryError, match="没有能够确认目标状态"):
chain.get_file_item_strict("custom", Path("/target"))

5
tests/test_transfer_durable_retry_owner.py
View File

@@ -19,6 +19,7 @@ from app.db.base import Base
from app.db.models.transferexecutionstep import TransferExecutionStep
from app.db.models.transferhistory import TransferHistory
from app.db.models.transferpending import TransferPending
from app.schemas.file import FileItem
from app.schemas.types import MediaSource, MediaType, NotificationChannel
@@ -264,7 +265,7 @@ def test_durable_manual_cleanup_discards_task_and_removes_old_state(monkeypatch)
monkeypatch.setattr(
"app.chain.transfer.records.StorageChain",
lambda: SimpleNamespace(
exists=lambda _fileitem: True,
get_file_item_strict=lambda **kwargs: FileItem(storage=kwargs["storage"], path=str(kwargs["path"])),
delete_media_file=lambda fileitem: deleted.append(
("target", fileitem.path)
) or True,
@@ -425,7 +426,7 @@ def test_manual_replan_without_settlement_revision_obeys_current_lease(
effects.append(("history", history_id))
monkeypatch.setattr("app.chain.transfer.records.StorageChain", lambda: SimpleNamespace(
exists=lambda _fileitem: True, delete_media_file=remove_target,
get_file_item_strict=lambda **kwargs: FileItem(storage=kwargs["storage"], path=str(kwargs["path"])), delete_media_file=remove_target,
))
monkeypatch.setattr("app.chain.transfer.records.clear_transfer_failures", lambda *args: effects.append(args))
chain = object.__new__(TransferChain)