diff --git a/app/application/history.py b/app/application/history.py index c944b7f4a..a0b64bc28 100644 --- a/app/application/history.py +++ b/app/application/history.py @@ -3,7 +3,26 @@ from __future__ import annotations from dataclasses import asdict, dataclass from typing import Any, Callable, Dict, NoReturn, Optional, Protocol, Union -from app.application.configuration import TransferRetryConfig, get_transfer_retry_config +from app.application.history_retry import ( + HistoryGateAction, + _failed_retry_counts, + clear_transfer_failures, + coerce_fileid, + coerce_modify_time, + coerce_size, + describe_history_gate, + evaluate_history_gate, + failed_retry_count, + failed_retry_key, + file_fingerprint, + history_src_fingerprint, + history_src_size, + is_skip_action, + max_failed_retries, + next_failed_retry_count, + record_transfer_failure, + resolve_history, +) from app.application.historymutation import ( DownloadFileMutationRepository as DownloadFileMutationRepository, ) @@ -29,8 +48,6 @@ from app.application.transfer import history as history_projection from app.domain.context import MediaInfo, MusicInfo from app.domain.meta.metabase import MetaBase from app.foundation.text import cut as jieba_cut -from app.runtime.cache import TTLCache -from app.runtime.log import logger from app.schemas.common import JsonData from app.schemas.file import FileItem from app.schemas.history import ( @@ -41,19 +58,6 @@ from app.schemas.history import ( from app.schemas.transfer import TransferInfo from app.schemas.types import MediaSource -# 失败重试次数的合法区间。下界为 1：一次瞬时故障（网络抖动、TMDB 瞬断、移动失败） -# 不该让文件永久漏整理，所以不允许关闭重试；上界为 10：永远识别不出的文件重试再多 -# 也不会成功，只会重复推送失败通知，批量导入场景下会刷屏，所以不允许无限重试 -MIN_FAILED_RETRIES = 1 -MAX_FAILED_RETRIES = 10 - -# 同一源路径的连续整理失败状态。整理链在写失败历史时累计、整理成功或删除历史时清零， -# 查重闸只读不写，避免监控层与整理链对同一个事件重复计数。缓存值会同时保存文件指纹， -# 因此同一路径的新版本天然获得独立预算；内存缓存会随进程重启清空，Redis 后端则保留到 TTL 到期。 -FAILED_RETRY_TTL = 24 * 3600 -_failed_retry_counts = TTLCache(region="transfer_failed_retry", maxsize=5000, ttl=FAILED_RETRY_TTL) - - class _FrozenJsonDict(dict[str, JsonData]): """保留 JSON 字典读取与序列化行为，并拒绝常规原地修改。""" @@ -816,457 +820,6 @@ class HistoryQueryService: return result.replace("*", "%").replace("?", "_") -class HistoryGateAction: - """ - 整理历史查重闸的判定结果。 - - 监控分发（app/monitor/dispatcher.py）与整理链计划整理段（app/chain/transfer/plan.py） - 共用本模块，避免两处各写一套去重策略后互相对冲：上游放行的文件被下游按 - 「存在记录即拦」全额收回，等于放行逻辑完全失效。 - """ - # 没有整理记录 - PASS_NO_RECORD = "pass_no_record" - # 上次整理失败且重试次数未用尽，放行重试 - PASS_FAILED = "pass_failed" - # 上次整理失败但源文件已变为新版本，放行并重置该版本的重试预算 - PASS_FAILED_VERSION_CHANGED = "pass_failed_version_changed" - # 已整理成功但源文件已变化，放行交由 overwrite_mode 决断 - PASS_SIZE_CHANGED = "pass_size_changed" - # 上次整理失败且重试次数已用尽，跳过 - SKIP_RETRY_EXHAUSTED = "skip_retry_exhausted" - # 已整理成功且源文件未变化，跳过 - SKIP = "skip" - - -def is_skip_action(action: str) -> bool: - """ - 判断查重闸判定是否为跳过整理。 - :param action: HistoryGateAction 之一 - :return: True 表示跳过 - """ - return action in (HistoryGateAction.SKIP, HistoryGateAction.SKIP_RETRY_EXHAUSTED) - - -def max_failed_retries(config: TransferRetryConfig | None = None) -> int: - """ - 读取失败重试上限并钳制到合法区间。 - - 配置为负数、0 或超过上界时都会被钳制并记录 warn：关闭重试会让瞬时故障造成 - 永久漏件，无限重试会让永久失败的文件反复刷通知，两端都不接受。 - :return: 合法的最大重试次数 - """ - raw = (config or get_transfer_retry_config()).max_failed_retries - try: - value = int(raw) - except (TypeError, ValueError): - logger.warn(f"TRANSFER_MAX_FAILED_RETRIES 配置非法（{raw!r}），" - f"已回退为 {MIN_FAILED_RETRIES}") - return MIN_FAILED_RETRIES - if value < MIN_FAILED_RETRIES: - logger.warn(f"TRANSFER_MAX_FAILED_RETRIES 不能小于 {MIN_FAILED_RETRIES}" - f"（当前 {value}），已按 {MIN_FAILED_RETRIES} 处理") - return MIN_FAILED_RETRIES - if value > MAX_FAILED_RETRIES: - logger.warn(f"TRANSFER_MAX_FAILED_RETRIES 不能大于 {MAX_FAILED_RETRIES}" - f"（当前 {value}），已按 {MAX_FAILED_RETRIES} 处理") - return MAX_FAILED_RETRIES - return value - - -def failed_retry_key(src_path: Optional[str], storage: Optional[str] = None) -> Optional[str]: - """ - 生成失败重试计数的缓存键。 - :param src_path: 整理记录使用的源路径 - :param storage: 源存储 - :return: 缓存键，源路径为空时返回 None - """ - if not src_path: - return None - return f"{storage or 'local'}:{src_path}" - - -def coerce_modify_time(modify_time: Any) -> Optional[float]: - """ - 统一转换文件修改时间，无法转换时返回 None。 - :param modify_time: 原始修改时间值 - :return: 文件修改时间 - """ - if modify_time is None: - return None - try: - return float(modify_time) - except (TypeError, ValueError): - return None - - -def coerce_fileid(fileid: Any) -> Optional[str]: - """ - 统一转换文件唯一标识，空值视为不可比对。 - :param fileid: 原始文件唯一标识 - :return: 非空文件唯一标识 - """ - if fileid is None: - return None - value = str(fileid).strip() - return value or None - - -def file_fingerprint( - file_size: Any = None, - file_modify_time: Any = None, - fileid: Any = None, -) -> Dict[str, Any]: - """ - 生成用于区分同一路径文件版本的稳定指纹。 - - 大小是所有存储器都尽量提供的最小指纹；两端均有数据时，修改时间和文件 ID 还可 - 识别“同大小替换”。只保留可比较字段，避免缺失元数据把同一文件误判成新版本。 - :param file_size: 文件大小 - :param file_modify_time: 文件修改时间 - :param fileid: 存储器文件唯一标识 - :return: 非空且可比较的指纹字段 - """ - fingerprint: Dict[str, Any] = {} - size = coerce_size(file_size) - if size is not None: - fingerprint["size"] = size - modify_time = coerce_modify_time(file_modify_time) - if modify_time is not None: - fingerprint["modify_time"] = modify_time - normalized_fileid = coerce_fileid(fileid) - if normalized_fileid is not None: - fingerprint["fileid"] = normalized_fileid - return fingerprint - - -def _retry_state(value: Any) -> tuple[int, Dict[str, Any]]: - """将新旧缓存值统一转换为失败次数与文件指纹。""" - if isinstance(value, dict): - raw_count = value.get("count", 0) - raw_fingerprint = value.get("fingerprint") - else: - # 兼容已写入 Redis 或内存的旧整数计数；下次带指纹写入时会自动升级结构。 - raw_count = value - raw_fingerprint = None - try: - count = max(int(raw_count or 0), 0) - except (TypeError, ValueError): - count = 0 - fingerprint = ( - file_fingerprint( - file_size=raw_fingerprint.get("size"), - file_modify_time=raw_fingerprint.get("modify_time"), - fileid=raw_fingerprint.get("fileid"), - ) - if isinstance(raw_fingerprint, dict) - else {} - ) - return count, fingerprint - - -def _is_file_version_changed( - recorded_fingerprint: Dict[str, Any], - current_fingerprint: Dict[str, Any], -) -> bool: - """判断两个可比文件指纹是否指向不同版本。""" - for field in ("fileid", "modify_time", "size"): - recorded_value = recorded_fingerprint.get(field) - current_value = current_fingerprint.get(field) - if ( - recorded_value is not None - and current_value is not None - and recorded_value != current_value - ): - return True - return False - - -def failed_retry_count(src_path: Optional[str], storage: Optional[str] = None, - file_size: Any = None, file_modify_time: Any = None, - fileid: Any = None) -> int: - """ - 读取同一源路径已累计的连续整理失败次数。 - :param src_path: 整理记录使用的源路径 - :param storage: 源存储 - :param file_size: 当前文件大小 - :param file_modify_time: 当前文件修改时间 - :param fileid: 当前文件唯一标识 - :return: 当前文件版本已失败次数，无记录时为 0 - """ - key = failed_retry_key(src_path, storage) - if not key: - return 0 - count, recorded_fingerprint = _retry_state(_failed_retry_counts.get(key)) - current_fingerprint = file_fingerprint( - file_size=file_size, - file_modify_time=file_modify_time, - fileid=fileid, - ) - if ( - recorded_fingerprint - and current_fingerprint - and _is_file_version_changed(recorded_fingerprint, current_fingerprint) - ): - return 0 - return count - - -def record_transfer_failure(src_path: Optional[str], storage: Optional[str] = None, - file_size: Any = None, file_modify_time: Any = None, - fileid: Any = None) -> int: - """ - 累计一次整理失败。 - :param src_path: 整理记录使用的源路径 - :param storage: 源存储 - :param file_size: 当前文件大小 - :param file_modify_time: 当前文件修改时间 - :param fileid: 当前文件唯一标识 - :return: 当前文件版本累计后的失败次数 - """ - key = failed_retry_key(src_path, storage) - if not key: - return 0 - count, recorded_fingerprint = _retry_state(_failed_retry_counts.get(key)) - current_fingerprint = file_fingerprint( - file_size=file_size, - file_modify_time=file_modify_time, - fileid=fileid, - ) - if current_fingerprint and ( - not recorded_fingerprint - or _is_file_version_changed(recorded_fingerprint, current_fingerprint) - ): - count = 0 - count += 1 - if current_fingerprint: - _failed_retry_counts[key] = { - "count": count, - "fingerprint": current_fingerprint, - } - elif recorded_fingerprint: - _failed_retry_counts[key] = { - "count": count, - "fingerprint": recorded_fingerprint, - } - else: - _failed_retry_counts[key] = count - return count - - -def clear_transfer_failures(src_path: Optional[str], storage: Optional[str] = None) -> None: - """ - 清空同一源路径的失败计数。整理成功、或用户删除整理记录（显式要求重来）时调用。 - :param src_path: 整理记录使用的源路径 - :param storage: 源存储 - """ - key = failed_retry_key(src_path, storage) - if key: - # 缺省值必须是 0 而不是 None：CacheBackend.pop 把「default 为 None」当成「未提供 - # default」，键不存在时会抛 KeyError。整理成功路径上绝大多数文件从未失败过， - # 传 None 会让每一次首次成功整理都炸掉成功回调 - _failed_retry_counts.pop(key, 0) - - -def coerce_size(size: Any) -> Optional[int]: - """ - 统一转换文件大小，无法转换时返回 None（视为不可比对）。 - :param size: 原始大小值 - :return: 文件大小 - """ - if size is None: - return None - try: - return int(size) - except (TypeError, ValueError): - return None - - -def history_src_size(history: TransferHistorySnapshot) -> Optional[int]: - """ - 读取整理记录中的源文件大小。 - src_fileitem 是 JSON 列，历史数据可能为空、缺 size 键甚至不是字典， - 取不到时统一返回 None 交由调用方保守处理。 - :param history: 整理记录 - :return: 源文件大小，取不到时为 None - """ - return history_src_fingerprint(history).get("size") - - -def history_src_fingerprint(history: TransferHistorySnapshot) -> Dict[str, Any]: - """ - 读取整理记录中的源文件版本指纹。 - :param history: 整理记录 - :return: 源文件的可比较指纹字段 - """ - src_fileitem = getattr(history, "src_fileitem", None) - if not isinstance(src_fileitem, dict): - return {} - return file_fingerprint( - file_size=src_fileitem.get("size"), - file_modify_time=src_fileitem.get("modify_time"), - fileid=src_fileitem.get("fileid"), - ) - - -def resolve_history( - src_path: str, - storage: Optional[str] = None, - transfer_history_oper: Optional[TransferHistoryQueryPort] = None, -) -> Optional[TransferHistorySnapshot]: - """ - 查询源路径对应的整理记录。 - - 新表通过 (src, src_storage) 唯一索引保证单条记录；仍保留对成功记录的二次确认， - 兼容升级前可能残留的重复数据，避免把已整理成功的文件重复整理。查询异常不在 - 此处吞掉，由调用方按各自的重试策略处理。 - :param src_path: 整理记录使用的源路径 - :param storage: 存储 - :param transfer_history_oper: 兼容旧关键字的类型化仓储，未传时使用组合根实现 - :return: 命中的整理记录，未命中时为 None - """ - repository = transfer_history_oper or get_transfer_history_repository() - history = repository.get_by_src(src_path, storage=storage) - if history is not None and not history.status: - history = repository.get_success_by_src(src_path, storage=storage) or history - return history - - -def evaluate_history_gate(history: Optional[TransferHistorySnapshot], - file_size: Optional[float] = None, - file_modify_time: Optional[float] = None, - fileid: Optional[str] = None, - retry_count: Optional[int] = None) -> str: - """ - 依据整理历史判断本次是否跳过整理。 - - 成功记录不能简单地「存在即跳过」：同路径重新上传的新版本会因此没有机会走到 - 整理链的 overwrite_mode 判定，升级永远无法入库，故任一可比文件指纹变化时一律放行。 - 失败记录按文件版本使用有界重试：新版本先放行并在下一次失败时从 1 重新计数， - 同一版本未达上限时继续重试，让瞬时故障（网络/识别/移动）自愈；达到上限后跳过， - 避免永久失败的文件反复刷失败通知。 - :param history: 整理记录，未命中时为 None - :param file_size: 当前文件大小，蓝光目录等场景可能为 None - :param file_modify_time: 当前文件修改时间 - :param fileid: 当前文件唯一标识 - :param retry_count: 已累计的失败次数，None 表示按记录源路径实时查询 - :return: HistoryGateAction 之一 - """ - if history is None: - return HistoryGateAction.PASS_NO_RECORD - recorded_fingerprint = history_src_fingerprint(history) - current_fingerprint = file_fingerprint( - file_size=file_size, - file_modify_time=file_modify_time, - fileid=fileid, - ) - if not history.status: - if _is_file_version_changed(recorded_fingerprint, current_fingerprint): - return HistoryGateAction.PASS_FAILED_VERSION_CHANGED - if retry_count is None: - retry_count = failed_retry_count( - getattr(history, "src", None), - getattr(history, "src_storage", None), - file_size=file_size, - file_modify_time=file_modify_time, - fileid=fileid, - ) - if retry_count >= max_failed_retries(): - return HistoryGateAction.SKIP_RETRY_EXHAUSTED - # 监控事件是稀疏驱动的（落地事件/延迟重扫/补偿扫描），入口还有 TTL 去重兜底， - # 配合失败次数上限，重试频率与总量都可控 - return HistoryGateAction.PASS_FAILED - if _is_file_version_changed(recorded_fingerprint, current_fingerprint): - # 同路径换成了另一个版本（如升级为更高码率），是否覆盖交给整理链的 - # overwrite_mode 决断，查重闸不做替代判断 - return HistoryGateAction.PASS_SIZE_CHANGED - # 无法比对大小（蓝光目录、历史记录缺 size）时保守跳过，避免重复整理 - return HistoryGateAction.SKIP - - -def describe_history_gate(history: Optional[TransferHistorySnapshot], - file_size: Optional[float] = None, - file_modify_time: Optional[float] = None, - fileid: Optional[str] = None) -> str: - """ - 生成查重闸判定的可读说明，供日志定位「到底是哪条记录在拦」。 - :param history: 整理记录 - :param file_size: 当前文件大小 - :param file_modify_time: 当前文件修改时间 - :param fileid: 当前文件唯一标识 - :return: 说明文本 - """ - if history is None: - return "无整理记录" - recorded_fingerprint = history_src_fingerprint(history) - current_fingerprint = file_fingerprint( - file_size=file_size, - file_modify_time=file_modify_time, - fileid=fileid, - ) - if not history.status: - count = getattr(history, "retry_count", None) - if count is None: - count = failed_retry_count( - getattr(history, "src", None), - getattr(history, "src_storage", None), - file_size=file_size, - file_modify_time=file_modify_time, - fileid=fileid, - ) - if _is_file_version_changed(recorded_fingerprint, current_fingerprint): - return f"失败记录 #{history.id}，文件版本已变化，重试预算将重置" - return f"失败记录 #{history.id}，已重试 {count}/{max_failed_retries()} 次" - recorded_size = recorded_fingerprint.get("size") - current_size = current_fingerprint.get("size") - if recorded_size is None and current_size is None: - return f"成功记录 #{history.id}，大小不可比对" - return f"成功记录 #{history.id}，大小 {recorded_size} -> {current_size}" - - -def next_failed_retry_count( - history: Optional[TransferHistorySnapshot], - *, - src_path: Optional[str], - storage: Optional[str] = None, - file_size: Optional[float] = None, - file_modify_time: Optional[float] = None, - fileid: Optional[str] = None, -) -> int: - """ - 计算下一次失败应持久化的连续次数，合并缓存与数据库记录。 - - 进程重启会清空本地缓存，因此同一文件版本必须以历史中的 retry_count 续算； - 文件指纹已变化时则视为新版本，从第一次失败重新开始。 - :param history: 当前源路径对应的整理历史 - :param src_path: 当前源路径 - :param storage: 当前源存储 - :param file_size: 当前文件大小 - :param file_modify_time: 当前文件修改时间 - :param fileid: 当前文件唯一标识 - :return: 本次失败写入历史的连续次数 - """ - cached_count = failed_retry_count( - src_path, - storage, - file_size=file_size, - file_modify_time=file_modify_time, - fileid=fileid, - ) - persisted_count = 0 - if history is not None and not history.status: - history_retry_count = getattr(history, "retry_count", None) or 0 - gate_action = evaluate_history_gate( - history, - file_size=file_size, - file_modify_time=file_modify_time, - fileid=fileid, - retry_count=history_retry_count, - ) - if gate_action != HistoryGateAction.PASS_FAILED_VERSION_CHANGED: - persisted_count = max(history_retry_count, 0) - return max(cached_count, persisted_count) + 1 - - # 整理历史写入口保留仓储事务契约；领域对象的字段映射由 transfer.history 统一维护。 def add_transfer_success( diff --git a/app/application/history_retry.py b/app/application/history_retry.py new file mode 100644 index 000000000..ceba64b91 --- /dev/null +++ b/app/application/history_retry.py @@ -0,0 +1,357 @@ +"""整理历史查重闸与失败重试预算。""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Dict, Optional + +from app.application.configuration import TransferRetryConfig, get_transfer_retry_config +from app.runtime.cache import TTLCache +from app.runtime.log import logger + +if TYPE_CHECKING: + from app.application.history import TransferHistoryQueryPort, TransferHistorySnapshot + + +# 失败重试次数的合法区间，避免瞬时故障永久漏件或永久失败反复刷通知。 +MIN_FAILED_RETRIES = 1 +MAX_FAILED_RETRIES = 10 +FAILED_RETRY_TTL = 24 * 3600 + +# 缓存值会同时保存文件指纹，使同一路径的新版本获得独立重试预算。 +_failed_retry_counts = TTLCache( + region="transfer_failed_retry", + maxsize=5000, + ttl=FAILED_RETRY_TTL, +) + + +class HistoryGateAction: + """整理历史查重闸的判定结果。""" + + PASS_NO_RECORD = "pass_no_record" + PASS_FAILED = "pass_failed" + PASS_FAILED_VERSION_CHANGED = "pass_failed_version_changed" + PASS_SIZE_CHANGED = "pass_size_changed" + SKIP_RETRY_EXHAUSTED = "skip_retry_exhausted" + SKIP = "skip" + + +def is_skip_action(action: str) -> bool: + """判断查重闸判定是否为跳过整理。""" + return action in (HistoryGateAction.SKIP, HistoryGateAction.SKIP_RETRY_EXHAUSTED) + + +def max_failed_retries(config: TransferRetryConfig | None = None) -> int: + """读取失败重试上限并钳制到合法区间。""" + raw = (config or get_transfer_retry_config()).max_failed_retries + try: + value = int(raw) + except (TypeError, ValueError): + logger.warn( + f"TRANSFER_MAX_FAILED_RETRIES 配置非法（{raw!r}），" + f"已回退为 {MIN_FAILED_RETRIES}" + ) + return MIN_FAILED_RETRIES + if value < MIN_FAILED_RETRIES: + logger.warn( + f"TRANSFER_MAX_FAILED_RETRIES 不能小于 {MIN_FAILED_RETRIES}" + f"（当前 {value}），已按 {MIN_FAILED_RETRIES} 处理" + ) + return MIN_FAILED_RETRIES + if value > MAX_FAILED_RETRIES: + logger.warn( + f"TRANSFER_MAX_FAILED_RETRIES 不能大于 {MAX_FAILED_RETRIES}" + f"（当前 {value}），已按 {MAX_FAILED_RETRIES} 处理" + ) + return MAX_FAILED_RETRIES + return value + + +def failed_retry_key(src_path: Optional[str], storage: Optional[str] = None) -> Optional[str]: + """生成失败重试计数的缓存键。""" + if not src_path: + return None + return f"{storage or 'local'}:{src_path}" + + +def coerce_modify_time(modify_time: Any) -> Optional[float]: + """统一转换文件修改时间，无法转换时返回 None。""" + if modify_time is None: + return None + try: + return float(modify_time) + except (TypeError, ValueError): + return None + + +def coerce_fileid(fileid: Any) -> Optional[str]: + """统一转换文件唯一标识，空值视为不可比对。""" + if fileid is None: + return None + value = str(fileid).strip() + return value or None + + +def file_fingerprint( + file_size: Any = None, + file_modify_time: Any = None, + fileid: Any = None, +) -> Dict[str, Any]: + """生成用于区分同一路径文件版本的稳定指纹。""" + fingerprint: Dict[str, Any] = {} + size = coerce_size(file_size) + if size is not None: + fingerprint["size"] = size + modify_time = coerce_modify_time(file_modify_time) + if modify_time is not None: + fingerprint["modify_time"] = modify_time + normalized_fileid = coerce_fileid(fileid) + if normalized_fileid is not None: + fingerprint["fileid"] = normalized_fileid + return fingerprint + + +def _retry_state(value: Any) -> tuple[int, Dict[str, Any]]: + """将新旧缓存值统一转换为失败次数与文件指纹。""" + if isinstance(value, dict): + raw_count = value.get("count", 0) + raw_fingerprint = value.get("fingerprint") + else: + raw_count = value + raw_fingerprint = None + try: + count = max(int(raw_count or 0), 0) + except (TypeError, ValueError): + count = 0 + fingerprint = ( + file_fingerprint( + file_size=raw_fingerprint.get("size"), + file_modify_time=raw_fingerprint.get("modify_time"), + fileid=raw_fingerprint.get("fileid"), + ) + if isinstance(raw_fingerprint, dict) + else {} + ) + return count, fingerprint + + +def _is_file_version_changed( + recorded_fingerprint: Dict[str, Any], + current_fingerprint: Dict[str, Any], +) -> bool: + """判断两个可比文件指纹是否指向不同版本。""" + for field in ("fileid", "modify_time", "size"): + recorded_value = recorded_fingerprint.get(field) + current_value = current_fingerprint.get(field) + if ( + recorded_value is not None + and current_value is not None + and recorded_value != current_value + ): + return True + return False + + +def failed_retry_count(src_path: Optional[str], storage: Optional[str] = None, + file_size: Any = None, file_modify_time: Any = None, + fileid: Any = None) -> int: + """读取同一源路径已累计的连续整理失败次数。""" + key = failed_retry_key(src_path, storage) + if not key: + return 0 + count, recorded_fingerprint = _retry_state(_failed_retry_counts.get(key)) + current_fingerprint = file_fingerprint( + file_size=file_size, + file_modify_time=file_modify_time, + fileid=fileid, + ) + if ( + recorded_fingerprint + and current_fingerprint + and _is_file_version_changed(recorded_fingerprint, current_fingerprint) + ): + return 0 + return count + + +def record_transfer_failure(src_path: Optional[str], storage: Optional[str] = None, + file_size: Any = None, file_modify_time: Any = None, + fileid: Any = None) -> int: + """累计一次整理失败并返回当前文件版本的连续失败次数。""" + key = failed_retry_key(src_path, storage) + if not key: + return 0 + count, recorded_fingerprint = _retry_state(_failed_retry_counts.get(key)) + current_fingerprint = file_fingerprint( + file_size=file_size, + file_modify_time=file_modify_time, + fileid=fileid, + ) + if current_fingerprint and ( + not recorded_fingerprint + or _is_file_version_changed(recorded_fingerprint, current_fingerprint) + ): + count = 0 + count += 1 + if current_fingerprint: + _failed_retry_counts[key] = { + "count": count, + "fingerprint": current_fingerprint, + } + elif recorded_fingerprint: + _failed_retry_counts[key] = { + "count": count, + "fingerprint": recorded_fingerprint, + } + else: + _failed_retry_counts[key] = count + return count + + +def clear_transfer_failures(src_path: Optional[str], storage: Optional[str] = None) -> None: + """清空同一源路径的失败计数。""" + key = failed_retry_key(src_path, storage) + if key: + # CacheBackend.pop 的默认值必须显式为 0，避免未命中时抛 KeyError。 + _failed_retry_counts.pop(key, 0) + + +def coerce_size(size: Any) -> Optional[int]: + """统一转换文件大小，无法转换时返回 None。""" + if size is None: + return None + try: + return int(size) + except (TypeError, ValueError): + return None + + +def history_src_size(history: TransferHistorySnapshot) -> Optional[int]: + """读取整理记录中的源文件大小。""" + return history_src_fingerprint(history).get("size") + + +def history_src_fingerprint(history: TransferHistorySnapshot) -> Dict[str, Any]: + """读取整理记录中的源文件版本指纹。""" + src_fileitem = getattr(history, "src_fileitem", None) + if not isinstance(src_fileitem, dict): + return {} + return file_fingerprint( + file_size=src_fileitem.get("size"), + file_modify_time=src_fileitem.get("modify_time"), + fileid=src_fileitem.get("fileid"), + ) + + +def resolve_history( + src_path: str, + storage: Optional[str] = None, + transfer_history_oper: Optional[TransferHistoryQueryPort] = None, +) -> Optional[TransferHistorySnapshot]: + """查询源路径对应的整理记录，并优先返回成功记录。""" + # 延迟导入避免查重策略模块与历史仓储契约之间形成初始化环。 + from app.application.history import get_transfer_history_repository + + repository = transfer_history_oper or get_transfer_history_repository() + history = repository.get_by_src(src_path, storage=storage) + if history is not None and not history.status: + history = repository.get_success_by_src(src_path, storage=storage) or history + return history + + +def evaluate_history_gate(history: Optional[TransferHistorySnapshot], + file_size: Optional[float] = None, + file_modify_time: Optional[float] = None, + fileid: Optional[str] = None, + retry_count: Optional[int] = None) -> str: + """依据整理历史和文件版本判断本次是否跳过整理。""" + if history is None: + return HistoryGateAction.PASS_NO_RECORD + recorded_fingerprint = history_src_fingerprint(history) + current_fingerprint = file_fingerprint( + file_size=file_size, + file_modify_time=file_modify_time, + fileid=fileid, + ) + if not history.status: + if _is_file_version_changed(recorded_fingerprint, current_fingerprint): + return HistoryGateAction.PASS_FAILED_VERSION_CHANGED + if retry_count is None: + retry_count = failed_retry_count( + getattr(history, "src", None), + getattr(history, "src_storage", None), + file_size=file_size, + file_modify_time=file_modify_time, + fileid=fileid, + ) + if retry_count >= max_failed_retries(): + return HistoryGateAction.SKIP_RETRY_EXHAUSTED + return HistoryGateAction.PASS_FAILED + if _is_file_version_changed(recorded_fingerprint, current_fingerprint): + return HistoryGateAction.PASS_SIZE_CHANGED + return HistoryGateAction.SKIP + + +def describe_history_gate(history: Optional[TransferHistorySnapshot], + file_size: Optional[float] = None, + file_modify_time: Optional[float] = None, + fileid: Optional[str] = None) -> str: + """生成查重闸判定的可读说明，供日志定位拦截原因。""" + if history is None: + return "无整理记录" + recorded_fingerprint = history_src_fingerprint(history) + current_fingerprint = file_fingerprint( + file_size=file_size, + file_modify_time=file_modify_time, + fileid=fileid, + ) + if not history.status: + count = getattr(history, "retry_count", None) + if count is None: + count = failed_retry_count( + getattr(history, "src", None), + getattr(history, "src_storage", None), + file_size=file_size, + file_modify_time=file_modify_time, + fileid=fileid, + ) + if _is_file_version_changed(recorded_fingerprint, current_fingerprint): + return f"失败记录 #{history.id}，文件版本已变化，重试预算将重置" + return f"失败记录 #{history.id}，已重试 {count}/{max_failed_retries()} 次" + recorded_size = recorded_fingerprint.get("size") + current_size = current_fingerprint.get("size") + if recorded_size is None and current_size is None: + return f"成功记录 #{history.id}，大小不可比对" + return f"成功记录 #{history.id}，大小 {recorded_size} -> {current_size}" + + +def next_failed_retry_count( + history: Optional[TransferHistorySnapshot], + *, + src_path: Optional[str], + storage: Optional[str] = None, + file_size: Optional[float] = None, + file_modify_time: Optional[float] = None, + fileid: Optional[str] = None, +) -> int: + """合并缓存与历史记录，计算下一次失败应持久化的连续次数。""" + cached_count = failed_retry_count( + src_path, + storage, + file_size=file_size, + file_modify_time=file_modify_time, + fileid=fileid, + ) + persisted_count = 0 + if history is not None and not history.status: + history_retry_count = getattr(history, "retry_count", None) or 0 + gate_action = evaluate_history_gate( + history, + file_size=file_size, + file_modify_time=file_modify_time, + fileid=fileid, + retry_count=history_retry_count, + ) + if gate_action != HistoryGateAction.PASS_FAILED_VERSION_CHANGED: + persisted_count = max(history_retry_count, 0) + return max(cached_count, persisted_count) + 1 diff --git a/tests/test_transfer_history_architecture.py b/tests/test_transfer_history_architecture.py index 436bcfd39..753b9a0c5 100644 --- a/tests/test_transfer_history_architecture.py +++ b/tests/test_transfer_history_architecture.py @@ -6,6 +6,7 @@ from pathlib import Path PROJECT_ROOT = Path(__file__).parents[1] APP_ROOT = PROJECT_ROOT / "app" HISTORY_APPLICATION_PATH = APP_ROOT / "application" / "history.py" +HISTORY_RETRY_APPLICATION_PATH = APP_ROOT / "application" / "history_retry.py" QUERY_APPLICATION_PATH = APP_ROOT / "application" / "query.py" HISTORY_ADAPTER_PACKAGE = APP_ROOT / "db" / "adapters" / "history" CANONICAL_CONSUMER_PATHS = ( @@ -160,6 +161,57 @@ def test_transfer_history_retires_dynamic_writer_facade() -> None: assert violations == [] +def test_history_retry_policy_has_one_canonical_owner() -> None: + """失败重试与查重闸逻辑只能由独立模块实现，旧模块仅保留兼容导出。""" + retry_tree = ast.parse( + HISTORY_RETRY_APPLICATION_PATH.read_text(encoding="utf-8-sig"), + filename=str(HISTORY_RETRY_APPLICATION_PATH), + ) + history_tree = ast.parse( + HISTORY_APPLICATION_PATH.read_text(encoding="utf-8-sig"), + filename=str(HISTORY_APPLICATION_PATH), + ) + expected_class = "HistoryGateAction" + expected_functions = { + "is_skip_action", + "max_failed_retries", + "failed_retry_key", + "file_fingerprint", + "failed_retry_count", + "record_transfer_failure", + "clear_transfer_failures", + "resolve_history", + "evaluate_history_gate", + "describe_history_gate", + "next_failed_retry_count", + } + retry_classes = { + node.name + for node in retry_tree.body + if isinstance(node, ast.ClassDef) + } + retry_functions = { + node.name + for node in retry_tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + history_classes = { + node.name + for node in history_tree.body + if isinstance(node, ast.ClassDef) + } + history_functions = { + node.name + for node in history_tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + + assert expected_class in retry_classes + assert expected_functions <= retry_functions + assert expected_class not in history_classes + assert expected_functions.isdisjoint(history_functions) + + def test_canonical_consumers_do_not_import_raw_transfer_history_persistence() -> None: """宿主消费者不得导入整理历史 ORM、Oper 或旧动态 getter。""" forbidden_from_imports = {