From 951cdc5b27709c615367f44c6f74563f7faed240 Mon Sep 17 00:00:00 2001 From: Aqr-K <95741669+Aqr-K@users.noreply.github.com> Date: Fri, 23 Oct 2026 08:14:15 +0800 Subject: [PATCH] fix: arbitrate subscription identity and restore workflow state (#7147) * fix: arbitrate subscription identity and restore workflow state * test: preserve legacy empty-owner subscription transactions * fix: align shared atomic configuration method with lifecycle prerequisite --------- Co-authored-by: jxxghp --- app/api/endpoints/subscribe.py | 4 +- app/api/servarr.py | 6 +- app/application/servarr.py | 20 +- app/application/subscription/contract.py | 8 + app/application/subscription/mutation.py | 12 + app/application/subscription/query.py | 4 + app/application/subscription/write.py | 1 + app/application/workflow.py | 2 +- app/chain/workflow.py | 39 +- app/db/adapters/subscription.py | 8 + app/db/models/subscribe.py | 68 +++- app/db/oper/subscribe.py | 86 ++++- app/db/oper/subscriptionsearch.py | 24 ++ app/locales/en-US.json | 1 + app/monitor/monitor.py | 2 +- app/monitor/poller.py | 18 +- app/monitor/snapshot.py | 4 +- app/scheduler/reconcile.py | 4 +- app/startup/composition/context.py | 2 + app/startup/composition/subscription.py | 14 +- app/workflow/actions/__init__.py | 22 +- .../c9e2f6a04173_subscription_identity.py | 61 +++ docs/architecture-overview.md | 2 +- docs/mcp-api.md | 4 + scripts/generate_agent_skill_docs.py | 4 +- skills/database-operation/SKILL.md | 4 +- skills/moviepilot-api/SKILL.md | 10 +- .../architecture/dependency-baseline.json | 5 +- tests/test_builtin_skill_boundaries.py | 4 +- tests/test_sdk_queries.py | 14 +- tests/test_servarr_series_add.py | 7 +- tests/test_subscribe_filter_clear.py | 7 +- tests/test_subscribe_oper.py | 1 + tests/test_subscribe_write_path.py | 54 ++- ..._subscription_identity_review_migration.py | 132 +++++++ tests/test_subscription_legacy_compat.py | 1 + tests/test_subscription_query_service.py | 1 + tests/test_subscription_review_regressions.py | 359 ++++++++++++++++++ tests/test_subscription_review_retarget.py | 83 ++++ tests/test_workflow_review_context.py | 54 +++ tests/test_workflow_review_regressions.py | 225 +++++++++++ 41 files changed, 1279 insertions(+), 102 deletions(-) create mode 100644 database/versions/c9e2f6a04173_subscription_identity.py create mode 100644 tests/test_subscription_identity_review_migration.py create mode 100644 tests/test_subscription_review_regressions.py create mode 100644 tests/test_subscription_review_retarget.py create mode 100644 tests/test_workflow_review_context.py create mode 100644 tests/test_workflow_review_regressions.py diff --git a/app/api/endpoints/subscribe.py b/app/api/endpoints/subscribe.py index a721c73ec..a320f1fed 100644 --- a/app/api/endpoints/subscribe.py +++ b/app/api/endpoints/subscribe.py @@ -52,7 +52,7 @@ from app.application.configuration import ( get_api_runtime_config_snapshot, get_configured_system_config, ) -from app.application.subscription.contract import SubscriptionQueryPort +from app.application.subscription.contract import SubscriptionIdentityConflictError, SubscriptionQueryPort from app.application.subscription.delete import ( DeleteSubscribeCommand, SubscribeDeletionActor, @@ -382,6 +382,8 @@ async def update_subscribe( actor, existing=subscribe, ) + except SubscriptionIdentityConflictError: + return _SchemaResponse(success=False, message="目标媒体身份已有订阅，当前修改已取消") except ValueError as error: logger.error(f"订阅分类设置无效：{error}", exc_info=True) return _SchemaResponse( diff --git a/app/api/servarr.py b/app/api/servarr.py index 9edf90f90..df87be07f 100644 --- a/app/api/servarr.py +++ b/app/api/servarr.py @@ -841,20 +841,22 @@ async def arr_add_series( seasons = [(season, True) for season in (fallback_seasons or [1])] # 检查订阅是否存在 left_seasons = [] + existing_id = None for season, monitored in seasons: if not monitored: continue - subscribe = await subscriptions.exists( + subscribe = await subscriptions.get_existing( media_source=MediaSource.TMDB, media_id=str(tv.tmdbId), season=season, ) if subscribe: + existing_id = subscribe.id continue left_seasons.append(season) # 全部已存在订阅 if not left_seasons: - return _SchemaServarrIdResponse(id=1) + return _SchemaServarrIdResponse(id=existing_id or 0) # TMDB 身份完整时允许空标题交由识别链补全；年份统一为订阅写入合同的字符串。 subscribe_title = tv.title or "" diff --git a/app/application/servarr.py b/app/application/servarr.py index 13d16adf5..0511af51a 100644 --- a/app/application/servarr.py +++ b/app/application/servarr.py @@ -7,7 +7,7 @@ from __future__ import annotations from dataclasses import dataclass from typing import TYPE_CHECKING, Any, List, Optional, Protocol -from app.schemas.types import MediaSource +from app.schemas.types import MediaSource, MediaType if TYPE_CHECKING: from app.application.subscription.delete import AsyncUnitOfWork @@ -83,6 +83,7 @@ class ServarrAsyncSubscriptionRepository(Protocol): season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> Optional[Any]: """ 按媒体身份读取命中的订阅。 @@ -92,6 +93,7 @@ class ServarrAsyncSubscriptionRepository(Protocol): :param season: 电视剧季号，None 表示不附加季号过滤 :param episode_group: 剧集组标识，None 只匹配未设置剧集组的记录 :param music_type: 音乐实体类型过滤；None 不限定类型，recording 兼容未标记类型的旧单曲记录 + :param mtype: 媒体类型，防止电影和电视剧的同编号冲突 :return: 满足媒体身份及附加条件的订阅记录，不存在时为 None """ ... @@ -197,12 +199,26 @@ class ServarrSubscriptionService: """ 判断指定媒体身份和季是否已有订阅。 """ + return await self.get_existing(media_source=media_source, media_id=media_id, season=season) is not None + + async def get_existing( + self, *, media_source: MediaSource, media_id: str, season: Optional[int] = None, + ) -> Optional[ServarrSubscription]: + """ + 返回命中身份的实际订阅，供重复添加响应保留正确主键。 + + :param media_source: 订阅数据来源 + :param media_id: 来源原生标识 + :param season: 电视剧季号；None 时不筛选季 + :return: 已有订阅的稳定投影，未命中为 None + """ record = await self._async_repository.async_exists( media_source=media_source, media_id=media_id, season=season, + mtype=MediaType.TV.value if season is not None else MediaType.MOVIE.value, ) - return record is not None + return self._project(record) if record is not None else None async def delete(self, subscribe_id: int) -> bool: """ diff --git a/app/application/subscription/contract.py b/app/application/subscription/contract.py index 4e664b691..32f7bda9d 100644 --- a/app/application/subscription/contract.py +++ b/app/application/subscription/contract.py @@ -17,6 +17,13 @@ from app.schemas.common import JsonData from app.schemas.media import build_media_key, resolve_media_identity from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaSource, MediaType + +class SubscriptionIdentityConflictError(ValueError): + """ + 目标完整身份已属于另一订阅，当前修改必须回滚且不得继承其进度。 + """ + + _SUBSCRIPTION_FIELDS = frozenset( { "name", @@ -444,6 +451,7 @@ class SubscriptionIdentity: return { "media_source": str(self.media_source), "media_id": self.media_id, + "type": self.type, "season": self.season, "episode_group": self.episode_group, "music_type": self.music_type, diff --git a/app/application/subscription/mutation.py b/app/application/subscription/mutation.py index 374e7fdb1..c98efdb1a 100644 --- a/app/application/subscription/mutation.py +++ b/app/application/subscription/mutation.py @@ -52,6 +52,18 @@ def _normalized_subscription_patch( payload, media_type=subscribe.type, ) + identity_fields = ("type", "media_source", "media_id", "season", "episode_group", "music_type") + if any(field in payload and payload[field] != getattr(subscribe, field) for field in identity_fields): + # 旧目标的下载和洗版事实不属于新作品、季或剧集组，须与身份修改在同一事务清除。 + normalized.update({ + "note": [], "episode_priority": {}, "current_priority": None, + "lack_episode": payload.get("total_episode", subscribe.total_episode), + "last_update": None, "last_search": None, + "current_audio_format": None, "current_bitrate": None, + "current_bit_depth": None, "current_sample_rate": None, + }) + if subscribe.music_type == MUSIC_ENTITY_ALBUM: + normalized["downloaded_tracks"] = [] return SubscriptionPatch(cast(dict[str, JsonData], normalized)) diff --git a/app/application/subscription/query.py b/app/application/subscription/query.py index 4a36707df..84c51260c 100644 --- a/app/application/subscription/query.py +++ b/app/application/subscription/query.py @@ -34,6 +34,7 @@ class SubscriptionQueryService: "media_source", "media_id", "music_type", + "episode_group", } _LEGACY_ID_FIELDS: tuple[tuple[str, MediaSource], ...] = ( ("tmdbid", MediaSource.TMDB), @@ -248,6 +249,7 @@ class SubscriptionQueryService: SubscriptionIdentity( media_source=media_source, media_id=str(media_id), + type=mediainfo.type.value if mediainfo.type else None, music_type=getattr(mediainfo, "music_type", None) if mediainfo.type == MediaType.MUSIC else None, season=meta.begin_season if meta else None, episode_group=mediainfo.episode_group, @@ -308,11 +310,13 @@ class SubscriptionQueryService: return None raw_season = identity.get("season") raw_music_type = identity.get("music_type") + raw_episode_group = identity.get("episode_group") return self._repository.get_by( SubscriptionIdentity( media_source=MediaSource(raw_source), media_id=str(raw_id), type=raw_type, + episode_group=raw_episode_group if isinstance(raw_episode_group, str) else None, season=raw_season if isinstance(raw_season, int) else None, music_type=raw_music_type if isinstance(raw_music_type, str) else None, ) diff --git a/app/application/subscription/write.py b/app/application/subscription/write.py index f841a5c00..b5eb10654 100644 --- a/app/application/subscription/write.py +++ b/app/application/subscription/write.py @@ -448,6 +448,7 @@ def _translate( media_source=MediaSource(media_source), media_id=media_id, music_type=music_type, + type=mediainfo.type.value if mediainfo.type else None, season=raw_season if isinstance(raw_season, int) else None, episode_group=mediainfo.episode_group, ) diff --git a/app/application/workflow.py b/app/application/workflow.py index 35e9fc161..4b2fad7eb 100644 --- a/app/application/workflow.py +++ b/app/application/workflow.py @@ -940,7 +940,7 @@ class WorkflowMutationCommand: if ( not updated.trigger_type or updated.trigger_type == WORKFLOW_TRIGGER_TIMER - ) and updated.timer: + ) and updated.timer and getattr(updated, "state", None) != "P": self._add_timer(updated) self._refresh_event(updated) return WorkflowMutationResult(True, "更新成功") diff --git a/app/chain/workflow.py b/app/chain/workflow.py index 7a955dcbf..096fbb23b 100644 --- a/app/chain/workflow.py +++ b/app/chain/workflow.py @@ -3,8 +3,11 @@ """ import ast +import base64 +import binascii import copy import inspect +import pickletools import threading from collections import defaultdict, deque from concurrent.futures import Future @@ -12,6 +15,8 @@ from contextvars import Context, copy_context from dataclasses import is_dataclass from datetime import date, datetime from functools import partial +from io import BytesIO, StringIO +from pickle import HIGHEST_PROTOCOL from time import monotonic, sleep from typing import Any, Callable, List, Optional, Tuple @@ -393,7 +398,7 @@ class WorkflowExecutor: state = metadata.get("state") if isinstance(metadata, dict) else None if state == "completed": state = "success" - if state in ("running", "queued"): + if state in ("running", "queued", "failed", "skipped"): state = "pending" if not state and action_id in legacy_actions: state = "success" @@ -414,7 +419,7 @@ class WorkflowExecutor: if self.workflow.current_action: logger.info(f"工作流已执行动作：{self.workflow.current_action}") try: - if isinstance(self.workflow.context, dict) and self.workflow.context.get("content"): + if self.is_legacy_context(self.workflow.context): logger.warning(f"工作流 {self.workflow.name} 的上下文为已废弃的旧版格式，将从空上下文开始执行") elif isinstance(self.workflow.context, dict): context = ActionContext.model_validate(self.workflow.context) @@ -425,6 +430,32 @@ class WorkflowExecutor: context.node_outputs = outputs return context + @staticmethod + def is_legacy_context(context: Any) -> bool: + """ + 只识别单字段的旧 Base64 Pickle 包装，绝不反序列化载荷。 + + 当前文本 content 即使是唯一字段也属于合法结构，不能据其非空丢弃上下文。 + + :param context: 持久化的工作流上下文字典 + :return: 仅旧包装内含支持协议的完整指令流、栈合法且无尾部数据时为 True + """ + if not isinstance(context, dict) or set(context) != {"content"}: + return False + content = context.get("content") + if not isinstance(content, str): + return False + try: + payload = base64.b64decode(content, validate=True) + if len(payload) < 2 or payload[0] != 0x80 or not 2 <= payload[1] <= HIGHEST_PROTOCOL: + return False + stream = BytesIO(payload) + # 静态解析校验指令、memo 与栈，不导入载荷引用，也不执行其 callable。 + pickletools.dis(stream, out=StringIO()) + return stream.tell() == len(payload) + except (ValueError, binascii.Error): + return False + def ensure_context_partitions(self) -> None: """ 确保上下文具备新版分区结构，并把旧字段映射到 artifacts。 @@ -461,6 +492,8 @@ class WorkflowExecutor: now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') metadata = self.node_metadata.setdefault(action_id, {}) metadata["state"] = state + if state == "success": + self.errors.pop(action_id, None) metadata["attempt"] = self.node_attempts.get(action_id, metadata.get("attempt") or 0) if state == "running": metadata["started_at"] = now @@ -1030,7 +1063,7 @@ class WorkflowExecutor: value = item for part in identity.split("."): value = self.read_value(value, int(part) if part.isdigit() else part) - return self.make_hashable_identity(item) + return self.make_hashable_identity(value) @staticmethod def make_hashable_identity(value: Any) -> Any: diff --git a/app/db/adapters/subscription.py b/app/db/adapters/subscription.py index 85d6fe1c1..65c2fff7b 100644 --- a/app/db/adapters/subscription.py +++ b/app/db/adapters/subscription.py @@ -365,6 +365,7 @@ class TransactionalSubscriptionRepository(_TransactionalSubscriptionWriter): identity.season, identity.episode_group, identity.music_type, + identity.type, ) ) ) @@ -408,6 +409,7 @@ class TransactionalSubscriptionRepository(_TransactionalSubscriptionWriter): media_id=identity.media_id, season=identity.season, music_type=identity.music_type, + episode_group=identity.episode_group, ) ) is not None @@ -428,6 +430,7 @@ class TransactionalSubscriptionRepository(_TransactionalSubscriptionWriter): season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> Optional[SubscriptionSnapshot]: """ 异步按媒体身份、季号及可选剧集组读取命中的订阅快照。 @@ -443,6 +446,7 @@ class TransactionalSubscriptionRepository(_TransactionalSubscriptionWriter): season=season, episode_group=episode_group, music_type=music_type, + mtype=mtype, ) return _project_subscription(record) if record is not None else None @@ -712,6 +716,7 @@ class SessionSubscriptionRepository: identity.season, identity.episode_group, identity.music_type, + identity.type, ) ) @@ -744,6 +749,7 @@ class SessionSubscriptionRepository: media_id=identity.media_id, season=identity.season, music_type=identity.music_type, + episode_group=identity.episode_group, ) return _project_subscription(record) if record is not None else None @@ -779,6 +785,7 @@ class SessionSubscriptionRepository: season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> Optional[SubscriptionSnapshot]: """ 异步按媒体身份、季号及可选剧集组读取命中的订阅快照。 @@ -789,6 +796,7 @@ class SessionSubscriptionRepository: season=season, episode_group=episode_group, music_type=music_type, + mtype=mtype, ) return _project_subscription(record) if record is not None else None diff --git a/app/db/models/subscribe.py b/app/db/models/subscribe.py index 0bbc112f8..ee42d1b3e 100644 --- a/app/db/models/subscribe.py +++ b/app/db/models/subscribe.py @@ -4,11 +4,13 @@ 会话及事务由调用方管理，本模块不负责提交或回滚。 """ +import hashlib +import json import time from typing import Any, Optional, Self from uuid import uuid4 -from sqlalchemy import JSON, ColumnElement, Float, Index, Integer, String, or_, select +from sqlalchemy import JSON, ColumnElement, Float, Index, Integer, String, event, inspect, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, Session, mapped_column @@ -17,6 +19,27 @@ from app.db.models._constraints import media_identity_constraint from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource +def subscription_identity_key(values: dict[str, Any]) -> Optional[str]: + """ + 为完整订阅身份生成数据库仲裁键，保留用户名大小写及空值区别。 + + 空字符串 owner 沿用显式跳过查重的兼容合同；旧单曲的空 music_type 按 recording 处理。 + + :param values: 已规范化的持久化字段 + :return: SHA256 身份摘要；身份不完整或显式跳过 owner 查重时为 None + """ + if not values.get("media_source") or not values.get("media_id") or not values.get("type"): + return None + if values.get("username") == "": + return None + identity = [values.get(field) for field in ( + "username", "type", "media_source", "media_id", "season", "episode_group", "music_type", + )] + if identity[1] == "音乐" and not identity[-1]: + identity[-1] = MUSIC_ENTITY_RECORDING + return hashlib.sha256(json.dumps(identity, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest() + + class Subscribe(Base): """ 订阅表 @@ -25,6 +48,8 @@ class Subscribe(Base): id = get_id_column() # 不随整数主键复用的业务活动标识，完成时原样带入订阅历史。 activity_key: Mapped[Optional[str]] = mapped_column(String(32), default=lambda: uuid4().hex) + # 完整身份的数据库唯一仲裁键；迁移保留的旧重复记录为空，不删除其业务事实。 + identity_key: Mapped[Optional[str]] = mapped_column(String(64)) # 标题 name: Mapped[str] = mapped_column(String, nullable=False, index=True) # 年份 @@ -131,6 +156,7 @@ class Subscribe(Base): episode_group: Mapped[Optional[str]] = mapped_column(String) __table_args__ = ( + Index("uq_subscribe_identity_key", "identity_key", unique=True), media_identity_constraint("subscribe"), Index("ix_subscribe_type_date", "type", "date"), Index("ix_subscribe_media_identity", "media_source", "media_id"), @@ -164,6 +190,7 @@ class Subscribe(Base): season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> Optional[Self]: """ 按媒体身份、季号与剧集组查询已有订阅。 @@ -172,6 +199,8 @@ class Subscribe(Base): if condition is None: return None statement = select(cls).where(condition) + if mtype is not None: + statement = statement.where(cls.type == mtype) if season is not None: statement = statement.where(cls.season == season) subscription: Optional[Self] = db.execute(statement.where(cls.episode_group == episode_group)).scalars().first() @@ -186,6 +215,7 @@ class Subscribe(Base): season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> Optional[Self]: """ 异步按媒体身份、季号与剧集组查询已有订阅。 @@ -194,6 +224,8 @@ class Subscribe(Base): if condition is None: return None statement = select(cls).where(condition) + if mtype is not None: + statement = statement.where(cls.type == mtype) if season is not None: statement = statement.where(cls.season == season) result = await db.execute(statement.where(cls.episode_group == episode_group)) @@ -210,6 +242,7 @@ class Subscribe(Base): season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> Optional[Self]: """ 按订阅 owner、媒体身份、季号与剧集组查询订阅行。 @@ -220,6 +253,8 @@ class Subscribe(Base): if condition is None: return None statement = select(cls).where(cls.username == username, condition) + if mtype is not None: + statement = statement.where(cls.type == mtype) if season is not None: statement = statement.where(cls.season == season) subscription: Optional[Self] = db.execute(statement.where(cls.episode_group == episode_group)).scalars().first() @@ -235,6 +270,7 @@ class Subscribe(Base): season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> Optional[Self]: """ 异步按订阅 owner、媒体身份、季号与剧集组查询订阅行。 @@ -245,6 +281,8 @@ class Subscribe(Base): if condition is None: return None statement = select(cls).where(cls.username == username, condition) + if mtype is not None: + statement = statement.where(cls.type == mtype) if season is not None: statement = statement.where(cls.season == season) result = await db.execute(statement.where(cls.episode_group == episode_group)) @@ -371,6 +409,7 @@ class Subscribe(Base): media_id: str | None = None, season: Optional[int] = None, music_type: Optional[str] = None, + episode_group: Optional[str] = None, ) -> Optional[Self]: """ 根据条件查询订阅 @@ -378,7 +417,7 @@ class Subscribe(Base): condition = cls._identity_condition(media_source, media_id, music_type) if condition is None: return None - statement = select(cls).where(condition, cls.type == type) + statement = select(cls).where(condition, cls.type == type, cls.episode_group == episode_group) if season is not None: statement = statement.where(cls.season == season) subscription: Optional[Self] = db.execute(statement).scalars().first() @@ -393,6 +432,7 @@ class Subscribe(Base): media_id: str | None = None, season: Optional[int] = None, music_type: Optional[str] = None, + episode_group: Optional[str] = None, ) -> Optional[Self]: """ 根据条件查询订阅 @@ -400,7 +440,7 @@ class Subscribe(Base): condition = cls._identity_condition(media_source, media_id, music_type) if condition is None: return None - query = select(cls).filter(condition, cls.type == type) + query = select(cls).filter(condition, cls.type == type, cls.episode_group == episode_group) if season is not None: query = query.filter(cls.season == season) result = await db.execute(query) @@ -462,3 +502,25 @@ class Subscribe(Base): ) ) return list(result.scalars().all()) + + +def _assign_subscription_identity(_mapper: Any, _connection: Any, record: Subscribe) -> None: + """ + 为 ORM 新建记录填充身份键，涵盖兼容 SDK 的直接模型写入。 + """ + record.identity_key = subscription_identity_key(record.__dict__) + + +def _refresh_subscription_identity(_mapper: Any, _connection: Any, record: Subscribe) -> None: + """ + 仅身份字段改变时更新唯一键，保留迁移留下的旧重复记录及其原有进度。 + """ + state = inspect(record) + if any(state.attrs[field].history.has_changes() for field in ( + "username", "type", "media_source", "media_id", "season", "episode_group", "music_type", + )): + record.identity_key = subscription_identity_key(record.__dict__) + + +event.listen(Subscribe, "before_insert", _assign_subscription_identity) +event.listen(Subscribe, "before_update", _refresh_subscription_identity) diff --git a/app/db/oper/subscribe.py b/app/db/oper/subscribe.py index ed4e9177f..25c6e5694 100644 --- a/app/db/oper/subscribe.py +++ b/app/db/oper/subscribe.py @@ -14,12 +14,16 @@ from typing import Any, List, Optional, Tuple, cast from sqlalchemy import delete as sqlalchemy_delete from sqlalchemy import func, select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.dialects.sqlite import insert as sqlite_insert +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session +from sqlalchemy.sql.dml import Insert -from app.application.subscription.contract import SubscribeDeletionCandidate +from app.application.subscription.contract import SubscribeDeletionCandidate, SubscriptionIdentityConflictError from app.db.base import DbOper -from app.db.models.subscribe import Subscribe +from app.db.models.subscribe import Subscribe, subscription_identity_key from app.db.oper.query import ( descending, enum_values, @@ -136,12 +140,26 @@ def _persistable(payload: Mapping[str, JsonData]) -> dict[str, JsonData]: # 订阅上会存出三种形态，PG 上还会直接拒写 persistable["search_imdbid"] = 1 if persistable.get("search_imdbid") else 0 persistable["date"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + persistable["identity_key"] = subscription_identity_key(persistable) return persistable +def _insert_subscription(session: Session | AsyncSession, payload: Mapping[str, JsonData]) -> Insert: + """ + 以数据库唯一身份键仲裁并发新增，不中止调用方拥有的事务。 + """ + dialect = session.get_bind().dialect.name + insert = pg_insert if dialect == "postgresql" else sqlite_insert + if dialect not in {"postgresql", "sqlite"}: + raise RuntimeError(f"订阅新增不支持数据库方言：{dialect}") + return insert(Subscribe).values(**payload).on_conflict_do_nothing( + index_elements=[Subscribe.identity_key], + ).returning(Subscribe.id) + + def _lookup_values( identity: Mapping[str, JsonData], -) -> tuple[MediaSource | str | None, str | None, int | None, str | None, str | None]: +) -> tuple[MediaSource | str | None, str | None, int | None, str | None, str | None, str | None]: """ 把宽 JSON 身份收窄为订阅模型查询参数。 """ @@ -150,12 +168,14 @@ def _lookup_values( raw_season = identity.get("season") raw_episode_group = identity.get("episode_group") raw_music_type = identity.get("music_type") + raw_type = identity.get("type") return ( raw_source if isinstance(raw_source, str) else None, str(raw_id) if isinstance(raw_id, (str, int)) else None, raw_season if isinstance(raw_season, int) else None, raw_episode_group if isinstance(raw_episode_group, str) else None, raw_music_type if isinstance(raw_music_type, str) else None, + raw_type if isinstance(raw_type, str) else None, ) @@ -178,7 +198,7 @@ class SubscribeOper(DbOper): """ if username == "": return None - media_source, media_id, season, episode_group, music_type = _lookup_values(identity) + media_source, media_id, season, episode_group, music_type, mtype = _lookup_values(identity) if username: owned_subscription: Optional[Subscribe] = self._execute_sync_query( lambda session: Subscribe.exists_by_username( @@ -189,6 +209,7 @@ class SubscribeOper(DbOper): season=season, episode_group=episode_group, music_type=music_type, + mtype=mtype, ), ) return owned_subscription @@ -200,6 +221,7 @@ class SubscribeOper(DbOper): season=season, episode_group=episode_group, music_type=music_type, + mtype=mtype, ) ) return subscription @@ -223,7 +245,7 @@ class SubscribeOper(DbOper): """ if username == "": return None - media_source, media_id, season, episode_group, music_type = _lookup_values(identity) + media_source, media_id, season, episode_group, music_type, mtype = _lookup_values(identity) if username: owned_subscription: Optional[Subscribe] = await Subscribe.async_exists_by_username( session, @@ -233,6 +255,7 @@ class SubscribeOper(DbOper): season=season, episode_group=episode_group, music_type=music_type, + mtype=mtype, ) return owned_subscription subscription: Optional[Subscribe] = await Subscribe.async_exists( @@ -242,6 +265,7 @@ class SubscribeOper(DbOper): season=season, episode_group=episode_group, music_type=music_type, + mtype=mtype, ) return subscription @@ -258,6 +282,7 @@ class SubscribeOper(DbOper): """ if not isinstance(self._db, Session): raise RuntimeError("同步订阅新增需要调用方提供 Session") + identity = {**identity, "type": identity.get("type") or payload.get("type")} subscribe = self._exists(identity, username) if subscribe: return SubscribeStageResult( @@ -265,12 +290,14 @@ class SubscribeOper(DbOper): message="订阅已存在", created=False, ) - subscribe = Subscribe(**_persistable(payload)) - self._db.add(subscribe) - self._db.flush() - if not subscribe.id: - return SubscribeStageResult(0, "新增订阅失败", True) - return SubscribeStageResult(subscribe.id, "新增订阅成功", True) + values = _persistable(payload) + inserted = self._db.execute(_insert_subscription(self._db, values)).scalar_one_or_none() + if inserted is not None: + return SubscribeStageResult(inserted, "新增订阅成功", True) + existing = self._db.scalar(select(Subscribe).where(Subscribe.identity_key == values["identity_key"])) + if existing is None: + raise RuntimeError("订阅身份仲裁后未找到既有记录") + return SubscribeStageResult(existing.id, "订阅已存在", False) async def async_stage_add( self, @@ -283,6 +310,7 @@ class SubscribeOper(DbOper): """ if not isinstance(self._db, AsyncSession): raise RuntimeError("异步订阅新增需要调用方提供 AsyncSession") + identity = {**identity, "type": identity.get("type") or payload.get("type")} subscribe = await self._async_exists(identity, username) if subscribe: return SubscribeStageResult( @@ -290,12 +318,14 @@ class SubscribeOper(DbOper): message="订阅已存在", created=False, ) - subscribe = Subscribe(**_persistable(payload)) - self._db.add(subscribe) - await self._db.flush() - if not subscribe.id: - return SubscribeStageResult(0, "新增订阅失败", True) - return SubscribeStageResult(subscribe.id, "新增订阅成功", True) + values = _persistable(payload) + inserted = (await self._db.execute(_insert_subscription(self._db, values))).scalar_one_or_none() + if inserted is not None: + return SubscribeStageResult(inserted, "新增订阅成功", True) + existing = await self._db.scalar(select(Subscribe).where(Subscribe.identity_key == values["identity_key"])) + if existing is None: + raise RuntimeError("订阅身份仲裁后未找到既有记录") + return SubscribeStageResult(existing.id, "订阅已存在", False) def add( self, @@ -365,6 +395,7 @@ class SubscribeOper(DbOper): season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> bool: """ 按媒体身份、季号及可选剧集组判断订阅是否存在。 @@ -375,6 +406,7 @@ class SubscribeOper(DbOper): "music_type": music_type, "season": season, "episode_group": episode_group, + "type": mtype, } return bool(self._exists(identity_params, username=None)) @@ -385,6 +417,7 @@ class SubscribeOper(DbOper): season: Optional[int] = None, episode_group: Optional[str] = None, music_type: Optional[str] = None, + mtype: Optional[str] = None, ) -> Optional[Subscribe]: """ 异步按媒体身份、季号及可选剧集组读取命中的订阅。 @@ -396,6 +429,7 @@ class SubscribeOper(DbOper): "music_type": music_type, "season": season, "episode_group": episode_group, + "type": mtype, }, username=None, ) @@ -620,6 +654,7 @@ class SubscribeOper(DbOper): media_id: str, season: Optional[int] = None, music_type: Optional[str] = None, + episode_group: Optional[str] = None, ) -> Optional[Subscribe]: """ 根据条件查询订阅 @@ -632,6 +667,7 @@ class SubscribeOper(DbOper): media_id=media_id, season=season, music_type=music_type, + episode_group=episode_group, ) ) return subscription @@ -643,6 +679,7 @@ class SubscribeOper(DbOper): media_id: str, season: Optional[int] = None, music_type: Optional[str] = None, + episode_group: Optional[str] = None, ) -> Optional[Subscribe]: """ 根据条件查询订阅 @@ -657,6 +694,7 @@ class SubscribeOper(DbOper): media_id=media_id, season=season, music_type=music_type, + episode_group=episode_group, ) ), ) @@ -808,7 +846,12 @@ class SubscribeOper(DbOper): return None for key, value in _normalize_integer_flags(payload).items(): setattr(subscribe, key, value) - await self._db.flush() + try: + await self._db.flush() + except IntegrityError as error: + if "identity_key" in str(error.orig): + raise SubscriptionIdentityConflictError("目标身份已有订阅，未迁移旧目标进度") from error + raise return subscribe async def async_update_filter_groups(self, sid: int, filter_groups: List[str]) -> Optional[Subscribe]: @@ -828,6 +871,13 @@ class SubscribeOper(DbOper): if subscribe: payload = _normalize_integer_flags(payload) self._stage_update(subscribe, payload) + if isinstance(self._db, Session): + try: + self._db.flush() + except IntegrityError as error: + if "identity_key" in str(error.orig): + raise SubscriptionIdentityConflictError("目标身份已有订阅，未迁移旧目标进度") from error + raise return subscribe def list_by_username( diff --git a/app/db/oper/subscriptionsearch.py b/app/db/oper/subscriptionsearch.py index fadb7c59c..571e150f6 100644 --- a/app/db/oper/subscriptionsearch.py +++ b/app/db/oper/subscriptionsearch.py @@ -53,6 +53,7 @@ class SubscriptionSearchOper(DbOper): raise RuntimeError(f"订阅搜索入队不支持数据库方言：{dialect}") insert = pg_insert if dialect == "postgresql" else sqlite_insert now = utc_now_text() + self._finish_expired_cancellations(now) batch = SubscriptionSearchBatch( batch_id=uuid4().hex, source=source, @@ -154,6 +155,28 @@ class SubscriptionSearchOper(DbOper): active_batch_ids.insert(0, batch.batch_id) return batch, created, coalesced, tuple(dict.fromkeys(active_batch_ids)) + def _finish_expired_cancellations(self, now: str) -> None: + """ + 原子结算取消且租约已过期的运行任务，释放活动键和分页快照。 + + :param now: 当前 UTC 时间文本；未过期租约继续由原执行者持有 + """ + if not isinstance(self._db, Session): + raise RuntimeError("取消任务结算需要调用方提供同步 Session") + rows = self._db.execute(update(SubscriptionSearchTask).where( + SubscriptionSearchTask.state == "running", + SubscriptionSearchTask.cancel_requested == 1, + or_(SubscriptionSearchTask.lease_expires_at.is_(None), SubscriptionSearchTask.lease_expires_at <= now), + ).values( + state="cancelled", phase="cancelled", active_key=None, lease_owner=None, + lease_token=None, lease_expires_at=None, current_site_id=None, updated_at=now, finished_at=now, + ).returning(SubscriptionSearchTask.task_id, SubscriptionSearchTask.batch_id)).all() + if not rows: + return + self._db.execute(delete(SearchSession).where(SearchSession.task_id.in_([row[0] for row in rows]))) + for batch_id in {row[1] for row in rows}: + self._refresh_batch(batch_id, now=now, error=None) + def claim_next(self, *, owner: str, lease_seconds: int) -> Optional[SubscriptionSearchTask]: """ CAS 认领任务并清除旧等待提示，过期 running 任务保留站点游标恢复。 @@ -161,6 +184,7 @@ class SubscriptionSearchOper(DbOper): if not isinstance(self._db, Session): raise RuntimeError("订阅搜索认领需要调用方提供同步 Session") now = utc_now_text() + self._finish_expired_cancellations(now) fairness_before = ( datetime.now(timezone.utc) - timedelta(minutes=15) ).isoformat(timespec="seconds") diff --git a/app/locales/en-US.json b/app/locales/en-US.json index 78125c293..c29a43fc8 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -177,6 +177,7 @@ "media_source 和 media_id 必须同时提供": "media_source and media_id must be provided together", "新增订阅时必须同时提供有效的 media_source 和 media_id": "A valid media_source and media_id must be provided together when creating a subscription", "更新媒体身份时必须同时提供有效的 media_source 和 media_id": "A valid media_source and media_id must be provided together when updating media identity", + "目标媒体身份已有订阅，当前修改已取消": "A subscription already exists for the target media identity; this change was cancelled", "分类策略已被其他会话更新，请刷新后重试": "The classification policy was updated by another session. Refresh and try again", "分类策略已被其他会话更新，请重新加载或合并后再发布": "The classification policy was updated by another session. Reload or merge before publishing again", "分类策略校验失败": "Classification policy validation failed", diff --git a/app/monitor/monitor.py b/app/monitor/monitor.py index a20272c28..11eb8e81c 100644 --- a/app/monitor/monitor.py +++ b/app/monitor/monitor.py @@ -955,7 +955,7 @@ class Monitor(ConfigReloadMixin, metaclass=SingletonClass): try: current_job = self._scheduler.get_job(f"monitor_{storage}") if current_job and current_job.trigger.interval.total_seconds() / 60 != new_interval: - self._scheduler.modify_job( + self._scheduler.reschedule_job( f"monitor_{storage}", trigger='interval', minutes=new_interval diff --git a/app/monitor/poller.py b/app/monitor/poller.py index aa42f092b..3314a7e44 100644 --- a/app/monitor/poller.py +++ b/app/monitor/poller.py @@ -233,6 +233,13 @@ class RemotePoller: :param mon_path: 监控路径 :return: 是否成功 """ + with self._get_lock(storage): + return self._force_full_scan(storage, mon_path) + + def _force_full_scan(self, storage: str, mon_path: Path) -> bool: + """ + 在存储快照锁内扫描单根，防止与同存储轮询读改写互相覆盖。 + """ try: logger.info(f"开始强制全量扫描: {storage}:{mon_path}") @@ -281,8 +288,15 @@ class RemotePoller: logger.error(f"读取快照基线失败，已跳过落盘以避免覆盖其他监控路径: {storage}:{mon_path}") return False old_snapshot = old_snapshot_data.get('snapshot', {}) if old_snapshot_data else {} - current_snapshot = {**old_snapshot, **new_snapshot} - if not self._store.save(storage, current_snapshot, len(current_snapshot)): + replaced_paths = self._snapshot_for_path(old_snapshot, mon_path) + current_snapshot = { + path: value for path, value in old_snapshot.items() + if path not in replaced_paths + } + current_snapshot.update(new_snapshot) + # 单根扫描不能推进存储级游标，否则会越过其他根尚未读取的修改。 + cursor = old_snapshot_data.get("timestamp", 0) if old_snapshot_data else 0 + if not self._store.save(storage, current_snapshot, len(current_snapshot), snapshot_time=cursor): logger.error(f"保存快照基线失败，全量扫描未完成: {storage}:{mon_path}") return False diff --git a/app/monitor/snapshot.py b/app/monitor/snapshot.py index eff7eaf4c..05ca74c04 100644 --- a/app/monitor/snapshot.py +++ b/app/monitor/snapshot.py @@ -61,8 +61,8 @@ class SnapshotStore: last_snapshot_time or 0, max((item.get('modify_time', 0) for item in snapshot.values()), default=0) ) - if not snapshot_time: - snapshot_time = time.time() + if not snapshot_time: + snapshot_time = time.time() snapshot_data = { 'version': self.VERSION, 'timestamp': snapshot_time, diff --git a/app/scheduler/reconcile.py b/app/scheduler/reconcile.py index 6d512c904..d4bb59e01 100644 --- a/app/scheduler/reconcile.py +++ b/app/scheduler/reconcile.py @@ -385,6 +385,8 @@ class SchedulerReconcileOwner(_SchedulerOwnerBase): return # 移除该工作流的全部服务 self.remove_workflow_job(workflow) + if getattr(workflow, "state", None) == "P": + return # 添加工作流服务 with self._lock: try: @@ -400,7 +402,7 @@ class SchedulerReconcileOwner(_SchedulerOwnerBase): self._jobs[job_id] = job self._scheduler.add_job( self.start, - trigger=CronTrigger.from_crontab(workflow.timer), + trigger=CronTrigger.from_crontab(workflow.timer, timezone=self._scheduler.timezone), id=job_id, name=workflow.name, kwargs={"job_id": job_id, "workflow_id": workflow.id}, diff --git a/app/startup/composition/context.py b/app/startup/composition/context.py index 3e26d2b29..7a238b5d7 100644 --- a/app/startup/composition/context.py +++ b/app/startup/composition/context.py @@ -123,6 +123,7 @@ class RequestSubscriptionRepository(SessionSubscriptionPort, Protocol): season: int | None = None, episode_group: str | None = None, music_type: str | None = None, + mtype: str | None = None, ) -> SubscriptionSnapshot | None: """ 按媒体身份、季和剧集组异步读取命中的订阅快照。 @@ -132,6 +133,7 @@ class RequestSubscriptionRepository(SessionSubscriptionPort, Protocol): :param season: 限定的季号；None 表示不按季过滤 :param episode_group: 限定的剧集组；None 表示不按组过滤 :param music_type: 音乐媒体的细分类型；非音乐时为 None + :param mtype: 媒体类型；None 表示不按类型过滤 :return: 命中的订阅快照；不存在时为 None """ ... diff --git a/app/startup/composition/subscription.py b/app/startup/composition/subscription.py index 261879407..3c9f10285 100644 --- a/app/startup/composition/subscription.py +++ b/app/startup/composition/subscription.py @@ -45,6 +45,7 @@ from app.db.adapters.subscription import ( from app.db.session import SessionFactory, async_session_scope from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork from app.runtime.events import eventmanager +from app.runtime.tasks import get_task_registry from app.schemas.common import JsonData from app.schemas.types import EventType, SystemConfigKey @@ -98,7 +99,16 @@ async def async_rule_group_mutation_scope( """ 非阻塞等待跨入口锁，并构造共享 AsyncSession 的规则原子服务。 """ - await asyncio.to_thread(_reference_mutation_lock.acquire) + lock = _reference_mutation_lock + acquisition = get_task_registry().create( + asyncio.to_thread(lock.acquire), owner="subscription.reference-lock", cancel_on_shutdown=False, + ) + try: + await asyncio.shield(acquisition) + except asyncio.CancelledError: + # shield 保留线程的迟到结果，回调负责释放已经没有请求所有者的锁。 + acquisition.add_done_callback(lambda result: lock.release() if result.result() else None) + raise try: async with async_session_scope() as session: async def publish_async( @@ -116,7 +126,7 @@ async def async_rule_group_mutation_scope( publish=publish_async, ) finally: - _reference_mutation_lock.release() + lock.release() @contextmanager diff --git a/app/workflow/actions/__init__.py b/app/workflow/actions/__init__.py index e5f876542..aae0975e3 100644 --- a/app/workflow/actions/__init__.py +++ b/app/workflow/actions/__init__.py @@ -219,17 +219,21 @@ class BaseAction(ABC): 保存缓存 """ workflow_key = self._cache_key % workflow_id - workflow_cache = self.systemconfigoper.get(workflow_key) or {} - action_cache = workflow_cache.get(self._action_id) or [] - if isinstance(data, list): - for item in data: + + def merge(current: Any) -> tuple[None, Any]: + """ + 在持久化原子更新范围内保留其他动作及本动作已有的去重记录。 + """ + workflow_cache = current or {} + action_cache = workflow_cache.get(self._action_id) or [] + items: list[_CacheItem | str] = [*data] if isinstance(data, list) else [data] + for item in items: if item not in action_cache: action_cache.append(item) - else: - if data not in action_cache: - action_cache.append(data) - workflow_cache[self._action_id] = action_cache - self.systemconfigoper.set(workflow_key, workflow_cache) + workflow_cache[self._action_id] = action_cache + return None, workflow_cache + + self.systemconfigoper.update_atomically(workflow_key, merge) @abstractmethod def execute(self, workflow_id: int, params: dict[str, object], context: ActionContext) -> ActionContext: diff --git a/database/versions/c9e2f6a04173_subscription_identity.py b/database/versions/c9e2f6a04173_subscription_identity.py new file mode 100644 index 000000000..5233347c8 --- /dev/null +++ b/database/versions/c9e2f6a04173_subscription_identity.py @@ -0,0 +1,61 @@ +""" +为订阅完整身份增加并发唯一仲裁键，保留既有重复订阅及其运行事实。 +""" + +import hashlib +import json + +import sqlalchemy as sa +from alembic import op + +revision = "c9e2f6a04173" +down_revision = "b8d2f5a01962" +branch_labels = None +depends_on = None + + +def _identity_key(row) -> str | None: + """ + 使用本迁移冻结的身份算法，避免未来代码变化改变旧库升级结果。 + """ + if not row["media_source"] or not row["media_id"] or not row["type"] or row["username"] == "": + return None + identity = [row[field] for field in ( + "username", "type", "media_source", "media_id", "season", "episode_group", "music_type", + )] + if identity[1] == "音乐" and not identity[-1]: + identity[-1] = "recording" + return hashlib.sha256(json.dumps(identity, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest() + + +def upgrade() -> None: + """ + 每个既有身份的首行持有唯一键，其余旧重复行原样保留，不合并下载进度或历史。 + """ + bind = op.get_bind() + inspector = sa.inspect(bind) + if "identity_key" not in {column["name"] for column in inspector.get_columns("subscribe")}: + op.add_column("subscribe", sa.Column("identity_key", sa.String(64))) + if "uq_subscribe_identity_key" in {index["name"] for index in inspector.get_indexes("subscribe")}: + return + table = sa.Table("subscribe", sa.MetaData(), autoload_with=bind) + assigned = set() + for row in bind.execute(sa.select(table).order_by(table.c.id)).mappings(): + key = _identity_key(row) + canonical = key if key not in assigned else None + if key: + assigned.add(key) + if row["identity_key"] != canonical: + bind.execute(table.update().where(table.c.id == row["id"]).values(identity_key=canonical)) + op.create_index("uq_subscribe_identity_key", "subscribe", ["identity_key"], unique=True) + + +def downgrade() -> None: + """ + 只移除仲裁键，不删除或合并用户订阅数据。 + """ + inspector = sa.inspect(op.get_bind()) + if "uq_subscribe_identity_key" in {index["name"] for index in inspector.get_indexes("subscribe")}: + op.drop_index("uq_subscribe_identity_key", table_name="subscribe") + if "identity_key" in {column["name"] for column in inspector.get_columns("subscribe")}: + op.drop_column("subscribe", "identity_key") diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 7d3cd151d..1523e4896 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -817,7 +817,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1129 | -| 内部导入边 | 9,709 | +| 内部导入边 | 9,710 | | 非平凡 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 deb14f6dc..8238390d5 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -350,6 +350,10 @@ Emby、Jellyfin、Plex、Navidrome 的目录、条目、搜索、季集和活动 - 设置更新优先回传 `expected_revision`。发生冲突时重新读取并比较，不直接覆盖；`generic_write_allowed=false` 时使用返回的专用操作。 - 插件存在分身时，显式选择目标实例或使用已经配置的默认目标。遇到实例歧义、停用目标或插件门禁拒绝时，按错误处理，不猜测另一个实例。 - 参数省略、空字符串与 `null` 的含义由具体操作决定。例如更新订阅时，省略字段通常保留原值，允许清空的字段可使用空值；不要把整份读取结果不加筛选地回写。 +- 订阅身份包含用户名、媒体类型、来源、媒体 ID、季号、剧集组和音乐实体类型。同编号的电影与电视剧可分别订阅；按来源定位时保留剧集组。同用户并发创建同一目标会复用真实订阅 ID。 +- 更新订阅的作品、季号、剧集组或音乐实体后，旧目标的下载与洗版进度在同一事务清除，缺失集数按新总集数重新开始；显式清空剧集组也属于目标切换。目标身份已存在时取消该次修改，原有两条订阅及进度保持不变。 +- Servarr 重复添加已监控的剧集返回真实既有订阅 ID。搜索任务取消后若执行者丢失，租约到期会结算取消并释放活动占位，再次搜索可重新入队。 +- 工作流恢复会重试失败节点及因失败跳过的节点，成功节点继续复用结果；文本 `content` 会随其他上下文完整恢复。编辑暂停工作流不会重新注册定时器，Cron 使用宿主调度时区；并行分支保存去重缓存时合并各自记录。 - 自定义过滤规则的新增、修改或删除遇到并发快照冲突时返回业务失败。先重新读取规则集合，再重新确认修改，不用旧集合覆盖已成功提交的配置。 - `search.torrents` 可能执行多轮站点搜索，工具等待上限为 300 秒；客户端应为此设置足够的超时。分页只限制返回结果，不会缩短搜索过程。 diff --git a/scripts/generate_agent_skill_docs.py b/scripts/generate_agent_skill_docs.py index 309123bf7..a2058cd28 100644 --- a/scripts/generate_agent_skill_docs.py +++ b/scripts/generate_agent_skill_docs.py @@ -17,8 +17,8 @@ PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) -MOVIEPILOT_API_SKILL_VERSION = "61" -DATABASE_OPERATION_SKILL_VERSION = "13" +MOVIEPILOT_API_SKILL_VERSION = "62" +DATABASE_OPERATION_SKILL_VERSION = "14" DATABASE_TABLE_GUIDES: dict[str, tuple[str, str, str]] = { diff --git a/skills/database-operation/SKILL.md b/skills/database-operation/SKILL.md index 616279ac8..c0bd06913 100644 --- a/skills/database-operation/SKILL.md +++ b/skills/database-operation/SKILL.md @@ -1,6 +1,6 @@ --- name: database-operation -version: 13 +version: 14 description: >- Use this skill when you need to inspect, query, maintain, or carefully modify the MoviePilot database. This skill uses the bundled scripts/mp-db.py helper, @@ -265,7 +265,7 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123" - Purpose: Stores active movie, TV, or music subscriptions, filters, progress, and download targets. - Useful queries: Inspecting state, missing episodes/tracks, quality rules, site scope, and match progress. - Write boundary: Create, update, search, or delete through the subscription API to preserve state-machine consistency. -- Columns: `id`, `activity_key`, `name`, `year`, `type`, `search_interval`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `downloaded_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `lack_episode`, `note`, `state`, `last_search`, `last_update`, `date`, `username`, `sites`, `downloader`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `manual_total_episode`, `custom_words`, `media_category_id`, `media_category`, `filter_groups`, `episode_group` +- Columns: `id`, `activity_key`, `identity_key`, `name`, `year`, `type`, `search_interval`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `downloaded_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `lack_episode`, `note`, `state`, `last_search`, `last_update`, `date`, `username`, `sites`, `downloader`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `manual_total_episode`, `custom_words`, `media_category_id`, `media_category`, `filter_groups`, `episode_group` ### `subscribeactivity` - Purpose: Structured subscription business events and resumable search-round progress, recorded after the feature upgrade. diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index d3f1fe3a7..d86351f9d 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -1,6 +1,6 @@ --- name: moviepilot-api -version: 61 +version: 62 description: >- Use this skill for MoviePilot product operations such as media search, torrent search, downloads, subscriptions, library checks, sites, storage, workflows, @@ -340,6 +340,14 @@ token exchange downloads retain their existing credential rules. `read_tool_result` for the next page instead of repeating the operation. These receipt and result tools are internal to the Agent, not external MCP tools. +## Subscription Identity And Workflow Recovery + +- Preserve media type, source, media ID, season, episode group, and music entity type when finding or creating a subscription. Movies and TV series with the same numeric ID are distinct targets. Keep the episode group in source-based lookup. +- Concurrent creation for the same user and identity returns the existing subscription ID. An update to an existing target identity cancels that update; reread both subscriptions before deciding the next step. +- Changing the work, season, episode group, or music entity resets the previous target's download and upgrade progress. Missing episodes restart from the new total; clearing the episode group also changes the target. +- Servarr duplicate series additions return the real existing subscription ID. If a cancelled search loses its executor, lease expiry settles the cancellation and permits a new search. +- Workflow recovery retries failed nodes and nodes skipped because of failed dependencies while retaining successful node results. Editing a paused workflow keeps its timer disabled; Cron uses the host scheduler timezone. + ## Collection Counts And Pagination - Search SSE `manual_paging=true` is a Web-only page-by-page mode and adds no Agent diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 4dbafedfa..95c4c169f 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -959,8 +959,8 @@ "runtime_only": true } }, - "edge_count": 9709, - "edge_sha256": "d50236fde9c2615a7212c95348b872b2bae5a8513424c833bb4968a62d7d15c1", + "edge_count": 9710, + "edge_sha256": "488e191954d9ab1e56db00a899877988a36d9e5c94e1ddd6ee40257a3b7b434c", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -10116,6 +10116,7 @@ "app.startup.composition.subscription -> app.db.uow", "app.startup.composition.subscription -> app.runtime", "app.startup.composition.subscription -> app.runtime.events", + "app.startup.composition.subscription -> app.runtime.tasks", "app.startup.composition.subscription -> app.schemas", "app.startup.composition.subscription -> app.schemas.common", "app.startup.composition.subscription -> app.schemas.types", diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index ad78ef128..41f80634d 100644 --- a/tests/test_builtin_skill_boundaries.py +++ b/tests/test_builtin_skill_boundaries.py @@ -127,10 +127,10 @@ def test_modified_builtin_skills_have_incremented_versions() -> None: expected_versions = { "browser-use": "4", "command-dispatch": "2", - "database-operation": "13", + "database-operation": "14", "feedback-issue": "12", "anysearch": "3", - "moviepilot-api": "61", + "moviepilot-api": "62", "moviepilot-update": "7", "organize-files": "6", "transfer-failed-retry": "6", diff --git a/tests/test_sdk_queries.py b/tests/test_sdk_queries.py index 473486f55..9830d37d7 100644 --- a/tests/test_sdk_queries.py +++ b/tests/test_sdk_queries.py @@ -302,14 +302,6 @@ def test_sdk_applies_combined_filters_in_each_query_domain(db, query_sdk): season=2, episode_group="eg-a", ), - _subscribe( - "状态不符", - media_id="combo-sub", - state="R", - username="alice", - season=2, - episode_group="eg-a", - ), _subscribe( "用户不符", media_id="combo-sub", @@ -344,6 +336,12 @@ def test_sdk_applies_combined_filters_in_each_query_domain(db, query_sdk): episode_group="eg-a", ), ) + # 迁移保留的旧重复行没有身份键，仍须正确应用状态筛选。 + db.session.execute(SubscribeModel.__table__.insert().values( + name="状态不符", type=MediaType.TV.value, state="R", media_source=TMDB, + media_id="combo-sub", username="alice", season=2, episode_group="eg-a", identity_key=None, + )) + db.session.commit() subscribe_page = sdk.list_subscriptions( SubscriptionFilter( media_source=TMDB, diff --git a/tests/test_servarr_series_add.py b/tests/test_servarr_series_add.py index f633f685f..cdceace42 100644 --- a/tests/test_servarr_series_add.py +++ b/tests/test_servarr_series_add.py @@ -93,6 +93,7 @@ def _patch_chains(mediainfo=None, exists=None, add_result=(123, "")): subscribe_chain.async_add_batch = AsyncMock(return_value=add_result) subscriptions = MagicMock() subscriptions.exists = AsyncMock(return_value=bool(exists)) + subscriptions.get_existing = AsyncMock(return_value=exists) batch_writer = MagicMock() return patch( "app.api.servarr.MediaChain", @@ -190,7 +191,7 @@ def test_add_series_already_subscribed_returns_existing(): with media_patch, chain_patch: result = _run_add(tv, subscriptions, batch_writer) - assert result.id == 1 + assert result.id == 9 subscribe_chain.async_add_batch.assert_not_awaited() @@ -198,7 +199,7 @@ def test_add_series_existing_subscription_uses_async_repository(db): """ 真实请求仓储查到已有季订阅时，兼容端点应幂等返回而不抛异常。 """ - db.add( + existing = db.add( Subscribe( name="已存在的 Servarr 订阅", type=MediaType.TV.value, @@ -228,7 +229,7 @@ def test_add_series_existing_subscription_uses_async_repository(db): result = db.run_async_session(execute) - assert result.id == 1 + assert result.id == existing.id def test_add_series_identity_resolution_failure_returns_500(): diff --git a/tests/test_subscribe_filter_clear.py b/tests/test_subscribe_filter_clear.py index 9e4438ed4..93df7c930 100644 --- a/tests/test_subscribe_filter_clear.py +++ b/tests/test_subscribe_filter_clear.py @@ -224,12 +224,15 @@ def test_update_subscribe_persists_cleared_optional_settings(db, media_type: Med assert saved.type == media_type.value assert saved.media_source == MediaSource.TMDB.value assert saved.media_id == "6894" - assert (saved.season, saved.total_episode, saved.lack_episode, saved.manual_total_episode) == (1, 12, 7, 0) + # 清空剧集组会切换订阅身份，旧组下载进度不能延续到默认组。 + assert (saved.season, saved.total_episode, saved.lack_episode, saved.manual_total_episode) == (1, 12, 12, 0) assert saved.search_interval == 24 if media_type == MediaType.MUSIC: assert (saved.music_type, saved.total_tracks) == ("album", 12) assert len(published) == 1 - assert set(published[0]["fields"]) == set(OPTIONAL_SETTINGS) + assert set(OPTIONAL_SETTINGS).issubset(published[0]["fields"]) + assert "lack_episode" in published[0]["fields"] + assert published[0]["subscribe_info"]["lack_episode"] == 12 intent = db.session.query(OutboxMessage).filter_by(event_key=published[0]["idempotency_key"]).one() assert intent.status == "completed" assert all(intent.payload["subscribe_info"][field] is None for field in OPTIONAL_SETTINGS) diff --git a/tests/test_subscribe_oper.py b/tests/test_subscribe_oper.py index cebba279e..bdd4103e6 100644 --- a/tests/test_subscribe_oper.py +++ b/tests/test_subscribe_oper.py @@ -632,6 +632,7 @@ def test_subscribe_chain_exists_forwards_episode_group(): SubscriptionIdentity( media_source=MediaSource.TMDB, media_id=media.media_id, + type=MediaType.TV.value, music_type=None, season=1, episode_group="eg-1", diff --git a/tests/test_subscribe_write_path.py b/tests/test_subscribe_write_path.py index 452d41847..f390ee894 100644 --- a/tests/test_subscribe_write_path.py +++ b/tests/test_subscribe_write_path.py @@ -15,9 +15,9 @@ 亲和会在写入时把 ``2003`` 悄悄转成 ``'2003'``、把 ``True`` 转成 ``1``， 落库后的值对「写入路径自己有没有转」完全无感（已用变异验证确认：删掉 ``_normalize_year`` 后按落库值断言的用例全部照过）。所以这两类契约必须在 -建模那一刻、即 ``Subscribe(**kwargs)`` 的入参上断言，见 ``payloads`` 夹具。 -强转跟着订阅表的列走，因此仍留在 ``app/db/oper/subscribe.py``，夹具也就 -仍然钉在那个模块的 ``Subscribe`` 上。 +SQL 语句执行之前的绑定参数上断言，见 ``payloads`` 夹具。 +夹具监听 SQLAlchemy 的 before_execute，覆盖同步、异步实际插入路径， +同时保留真实写入，避免 SQLite 参数适配替实现完成强转。 同步与异步两条链路是两份逐字复制的实现，任何一条改了另一条没跟上都属于 真实缺陷，故每个字段契约都在两条链路上各断言一次。 @@ -25,6 +25,8 @@ import asyncio import pytest +from sqlalchemy import Engine, event +from sqlalchemy.sql.dml import Insert from app.application.subscription.write import add_subscribe, async_add_subscribe from app.db.adapters.subscription import TransactionalSubscriptionRepository @@ -99,40 +101,28 @@ def _row(db, subscribe_id: int) -> Subscribe: return Subscribe.get(db.session, subscribe_id) -class _SubscribeSpy: - """ - 记录建模入参并转交真实模型，保持写入路径仍然真的落库。 - """ - - def __init__(self, recorded: list): - self._recorded = recorded - - def __call__(self, **kwargs): - """ - 截获 ``Subscribe(**kwargs)`` 的入参后构造真实模型实例。 - """ - self._recorded.append(dict(kwargs)) - return Subscribe(**kwargs) - - def __getattr__(self, name): - """ - 查重用的类方法（exists / async_exists 等）原样透传给真实模型。 - """ - return getattr(Subscribe, name) - - @pytest.fixture -def payloads(monkeypatch): +def payloads(): """ - 捕获写入路径建模时的原始 kwargs，即落库**前**的值与类型。 + 记录真实插入语句在数据库驱动强转之前的绑定值，并保持实际事务提交。 - 只用于持久化类型强转这一类契约：SQLite 的类型亲和会在写入时替写入路径 - 把类型「修好」，落库后的值证明不了转换真的发生过；而这两步转换恰恰是为 - PostgreSQL 而写的，漏了只会在生产库上炸。 + SQLite 读回的类型无法证明实现已为 PostgreSQL 归一参数，因此检查 + SQLAlchemy before_execute 提供的语句，而不是替换订阅 ORM 模型。 """ recorded: list = [] - monkeypatch.setattr("app.db.oper.subscribe.Subscribe", _SubscribeSpy(recorded)) - return recorded + + def capture(_connection, statement, _multiparams, _params, _execution_options): + """ + 仅保存订阅插入语句的原始参数，其他表和查询不参与断言。 + """ + if isinstance(statement, Insert) and statement.table.name == "subscribe": + recorded.append(dict(statement.compile().params)) + + event.listen(Engine, "before_execute", capture) + try: + yield recorded + finally: + event.remove(Engine, "before_execute", capture) # 每个字段契约都在同步与异步两条链路上跑一遍：两份实现是逐字复制的， diff --git a/tests/test_subscription_identity_review_migration.py b/tests/test_subscription_identity_review_migration.py new file mode 100644 index 000000000..38ba953e6 --- /dev/null +++ b/tests/test_subscription_identity_review_migration.py @@ -0,0 +1,132 @@ +""" +订阅身份唯一仲裁的无损升级、回放和双数据库语句合同。 +""" + +import importlib +from types import SimpleNamespace + +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.dialects import postgresql +from sqlalchemy.schema import CreateIndex + +from app.db.models.subscribe import Subscribe +from app.db.oper.subscribe import SubscribeOper, _insert_subscription + + +def test_identity_migration_preserves_duplicate_progress_and_replays(monkeypatch): + """ + 旧重复行、下载进度与主键原样保存，回放和降级不删除业务事实。 + """ + engine = sa.create_engine("sqlite://") + migration = importlib.import_module("database.versions.c9e2f6a04173_subscription_identity") + with engine.begin() as connection: + connection.execute(sa.text("CREATE TABLE subscribe (id INTEGER PRIMARY KEY, username TEXT, type TEXT, " + "media_source TEXT, media_id TEXT, season INTEGER, episode_group TEXT, music_type TEXT, note TEXT)")) + for identifier, note in [(7, "[1]"), (8, "[2]")]: + connection.execute(sa.text("INSERT INTO subscribe VALUES (:id, 'alice', '电视剧', " + "'themoviedb', '123', 1, NULL, NULL, :note)"), {"id": identifier, "note": note}) + monkeypatch.setattr(migration, "op", Operations(MigrationContext.configure(connection))) + migration.upgrade() + rows = connection.execute(sa.text("SELECT id,note,identity_key FROM subscribe ORDER BY id")).all() + assert [(row[0], row[1]) for row in rows] == [(7, "[1]"), (8, "[2]")] + assert rows[0][2] and rows[1][2] is None + migration.upgrade() + assert connection.execute(sa.text("SELECT id,note,identity_key FROM subscribe ORDER BY id")).all() == rows + migration.downgrade() + migration.downgrade() + assert connection.execute(sa.text("SELECT id,note FROM subscribe ORDER BY id")).all() == [(7, "[1]"), (8, "[2]")] + engine.dispose() + + +def test_legacy_duplicate_remains_existing_after_canonical_deletion(tmp_path): + """ + canonical 删除后仍按真实身份找到旧重复行，不依赖仲裁键查询来判断已存在。 + """ + engine = sa.create_engine(f"sqlite:///{tmp_path / 'legacy.db'}") + Subscribe.__table__.create(engine) + values = {"name": "旧重复", "username": "alice", "type": "电视剧", "media_source": "themoviedb", + "media_id": "123", "season": 1} + with sa.orm.Session(engine) as session: + canonical = Subscribe(**values) + session.add(canonical) + session.commit() + canonical_id = canonical.id + with engine.begin() as connection: + connection.execute(Subscribe.__table__.insert().values(**values, identity_key=None)) + connection.execute(sa.delete(Subscribe).where(Subscribe.id == canonical_id)) + with sa.orm.Session(engine) as session: + result = SubscribeOper(session).stage_add(values, values, "alice") + session.commit() + assert not result.created + assert session.scalar(sa.select(sa.func.count()).select_from(Subscribe)) == 1 + engine.dispose() + + +def test_postgresql_identity_insert_uses_unique_conflict_arbiter(): + """ + PostgreSQL 使用原生 ON CONFLICT 身份仲裁，不使用进程锁代替跨进程唯一约束。 + """ + session = SimpleNamespace(get_bind=lambda: SimpleNamespace(dialect=postgresql.dialect())) + statement = str(_insert_subscription(session, {"name": "示例", "identity_key": "key"}).compile( + dialect=postgresql.dialect(), + )) + assert "ON CONFLICT (identity_key) DO NOTHING" in statement + index = next(index for index in Subscribe.__table__.indexes if index.name == "uq_subscribe_identity_key") + assert "CREATE UNIQUE INDEX uq_subscribe_identity_key" in str(CreateIndex(index).compile(dialect=postgresql.dialect())) + + +def test_identity_upgrade_chain_preserves_plugin_installation_snapshot(monkeypatch): + """ + G06 最终前置与订阅迁移连续升降级，保留安装日志及旧订阅事实。 + """ + from alembic.config import Config + from alembic.script import ScriptDirectory + + config = Config() + config.set_main_option("script_location", "database") + script = ScriptDirectory.from_config(config) + assert script.get_heads() == ["c9e2f6a04173"] + assert script.get_revision("c9e2f6a04173").down_revision == "b8d2f5a01962" + predecessor = importlib.import_module("database.versions.b8d2f5a01962_plugin_data_identity") + migration = importlib.import_module("database.versions.c9e2f6a04173_subscription_identity") + engine = sa.create_engine("sqlite://") + try: + with engine.begin() as connection: + connection.execute(sa.text("CREATE TABLE plugindata (id INTEGER PRIMARY KEY, plugin_id TEXT, " + "key TEXT, value TEXT)")) + connection.execute(sa.text("INSERT INTO plugindata VALUES (1,'review','key','keep'), " + "(2,'review','key','old duplicate')")) + connection.execute(sa.text("CREATE TABLE plugininstallation (id INTEGER PRIMARY KEY, phase TEXT)")) + connection.execute(sa.text("INSERT INTO plugininstallation VALUES (7,'preparing')")) + connection.execute(sa.text("CREATE TABLE subscribe (id INTEGER PRIMARY KEY, username TEXT, type TEXT, " + "media_source TEXT, media_id TEXT, season INTEGER, episode_group TEXT, music_type TEXT, note TEXT)")) + connection.execute(sa.text("INSERT INTO subscribe VALUES " + "(7,'alice','电视剧','themoviedb','123',1,NULL,NULL,'[1]'), " + "(8,'alice','电视剧','themoviedb','123',1,NULL,NULL,'[2]')")) + operations = Operations(MigrationContext.configure(connection)) + monkeypatch.setattr(predecessor, "op", operations) + monkeypatch.setattr(migration, "op", operations) + for _ in range(2): + predecessor.upgrade() + migration.upgrade() + assert connection.execute(sa.text("SELECT id,phase,host_enabled_before FROM plugininstallation")).all() == [ + (7, "preparing", None), + ] + assert connection.execute(sa.text("SELECT id,value FROM plugindata")).all() == [(1, "keep")] + columns = {column["name"]: column for column in sa.inspect(connection).get_columns("plugininstallation")} + assert columns["host_enabled_before"]["nullable"] is True + before = connection.execute(sa.text("SELECT id,note,identity_key FROM subscribe ORDER BY id")).all() + assert [(row[0], row[1]) for row in before] == [(7, "[1]"), (8, "[2]")] + migration.downgrade() + predecessor.downgrade() + assert "host_enabled_before" not in { + column["name"] for column in sa.inspect(connection).get_columns("plugininstallation") + } + assert connection.execute(sa.text("SELECT id,phase FROM plugininstallation")).all() == [(7, "preparing")] + predecessor.upgrade() + migration.upgrade() + assert connection.execute(sa.text("SELECT id,note,identity_key FROM subscribe ORDER BY id")).all() == before + finally: + engine.dispose() diff --git a/tests/test_subscription_legacy_compat.py b/tests/test_subscription_legacy_compat.py index 0c22347ee..e8ad672fb 100644 --- a/tests/test_subscription_legacy_compat.py +++ b/tests/test_subscription_legacy_compat.py @@ -124,6 +124,7 @@ def test_legacy_subscription_facade_unwraps_typed_write_contract(monkeypatch) -> "season": 1, "episode_group": None, "music_type": None, + "type": None, }, "payload": {"name": "兼容订阅"}, "username": "legacy-user", diff --git a/tests/test_subscription_query_service.py b/tests/test_subscription_query_service.py index ea35f1df8..76ddaad26 100644 --- a/tests/test_subscription_query_service.py +++ b/tests/test_subscription_query_service.py @@ -66,6 +66,7 @@ def test_subscription_query_service_builds_complete_exists_identity() -> None: SubscriptionIdentity( media_source=MediaSource.TMDB, media_id="123", + type=MediaType.TV.value, season=2, episode_group="group-1", ) diff --git a/tests/test_subscription_review_regressions.py b/tests/test_subscription_review_regressions.py new file mode 100644 index 000000000..5a48274c9 --- /dev/null +++ b/tests/test_subscription_review_regressions.py @@ -0,0 +1,359 @@ +""" +订阅身份、并发新增、过期取消搜索及切换目标的真实持久化回归。 +""" + +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timedelta, timezone +from threading import Barrier +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from sqlalchemy import create_engine, func, select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import sessionmaker + +from app.api.servarr import arr_add_series +from app.application.servarr import ServarrSubscriptionService +from app.application.subscription.contract import SubscriptionIdentityConflictError, SubscriptionSnapshot +from app.application.subscription.mutation import ( + SubscriptionActor, + SyncSubscriptionMutationService, + _normalized_subscription_patch, +) +from app.application.subscription.query import SubscriptionQueryService +from app.db.adapters.subscription import SessionSubscriptionRepository +from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository +from app.db.base import Base +from app.db.models.searchsession import SearchSession +from app.db.models.subscribe import Subscribe +from app.db.models.subscriptionsearch import SubscriptionSearchBatch, SubscriptionSearchTask +from app.db.oper.subscribe import SubscribeOper +from app.db.uow import SqlAlchemyUnitOfWork +from app.schemas.servarr import SonarrSeason, SonarrSeries +from app.schemas.types import MediaSource + + +@pytest.fixture +def database(tmp_path): + """ + 每次创建隔离 SQLite 及独立事务工厂，不使用真实业务库。 + """ + engine = create_engine(f"sqlite:///{tmp_path / 'review.db'}", connect_args={"timeout": 20}) + Base.metadata.create_all(engine, tables=[model.__table__ for model in ( + Subscribe, SubscriptionSearchBatch, SubscriptionSearchTask, SearchSession, + )]) + yield sessionmaker(engine, expire_on_commit=False) + engine.dispose() + + +def _identity(media_type="电影", group=None): + """ + 构造明确包含媒体类型与剧集组的订阅身份。 + """ + return {"media_source": "themoviedb", "media_id": "123", "type": media_type, + "season": 1, "episode_group": group} + + +def _payload(media_type="电影", group=None): + """ + 构造同一用户名的订阅写入字段。 + """ + return {**_identity(media_type, group), "name": "示例", "username": "alice"} + + +def test_empty_owner_keeps_legacy_duplicates_inside_caller_transaction(database): + """ + 旧空用户名不占唯一身份，允许重复暂存且全部随调用方回滚。 + """ + with database() as session: + oper = SubscribeOper(session) + payload = {**_payload(), "username": ""} + first = oper.stage_add(_identity(), payload, "") + second = oper.stage_add(_identity(), payload, "") + assert first.created and second.created + assert first.subscribe_id != second.subscribe_id + assert session.scalar(select(func.count()).select_from(Subscribe)) == 2 + assert all(row.identity_key is None for row in session.scalars(select(Subscribe))) + session.rollback() + with database() as session: + assert session.scalar(select(func.count()).select_from(Subscribe)) == 0 + + +def test_async_empty_owner_keeps_legacy_duplicates_inside_caller_transaction(db): + """ + 正式AsyncSession沿用空用户名兼容语义，不提前提交两条无唯一身份的记录。 + """ + db.watermark(Subscribe) + + async def execute(session): + """ + 在独立异步事务中暂存两个旧身份，再由调用方撤销。 + """ + oper = SubscribeOper(session) + payload = {**_payload(), "username": ""} + first = await oper.async_stage_add(_identity(), payload, "") + second = await oper.async_stage_add(_identity(), payload, "") + assert first.created and second.created + assert first.subscribe_id != second.subscribe_id + rows = list((await session.scalars(select(Subscribe).where( + Subscribe.id.in_([first.subscribe_id, second.subscribe_id]), + ))).all()) + assert len(rows) == 2 and all(row.identity_key is None for row in rows) + await session.rollback() + assert await session.get(Subscribe, first.subscribe_id) is None + assert await session.get(Subscribe, second.subscribe_id) is None + + db.run_async_session(execute) + + +def test_movie_and_tv_with_same_source_number_are_distinct(database): + """ + 同一来源的电影和电视剧编号相同也能分别订阅。 + """ + with database() as session: + oper = SubscribeOper(session) + movie = oper.stage_add(_identity(), _payload(), "alice") + tv = oper.stage_add(_identity("电视剧"), _payload("电视剧"), "alice") + session.commit() + assert movie.subscribe_id != tv.subscribe_id + assert movie.created and tv.created + + +@pytest.mark.asyncio +async def test_async_expired_identity_fields_cannot_bypass_unique_target(tmp_path): + """ + 异步会话的过期模型更新也不能把缺少加载字段误判成无身份记录。 + """ + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'expired-identity.db'}") + try: + async with engine.begin() as connection: + await connection.run_sync(Subscribe.__table__.create) + async with async_sessionmaker(engine, expire_on_commit=False)() as session: + existing = Subscribe(**_payload()) + changing = Subscribe(**{**_payload(), "media_id": "456"}) + session.add_all([existing, changing]) + await session.commit() + session.expire(changing, ["username", "type", "media_source", "season", "episode_group", "music_type"]) + changing.media_id = "123" + with pytest.raises(IntegrityError): + await session.flush() + await session.rollback() + assert await session.scalar(select(func.count()).select_from(Subscribe)) == 2 + assert {record.media_id for record in await session.scalars(select(Subscribe))} == {"123", "456"} + finally: + await engine.dispose() + + +def test_source_lookup_keeps_episode_group(database): + """ + 同一作品同季的不同剧集组定位到各自记录。 + """ + with database() as session: + first = Subscribe(**_payload("电视剧", "first")) + second = Subscribe(**_payload("电视剧", "second")) + session.add_all([first, second]) + session.commit() + query = SubscriptionQueryService(SessionSubscriptionRepository(session)) + record = query.get_by_source(_identity("电视剧", "second")) + assert record.id == second.id + + +@pytest.mark.asyncio +async def test_async_source_lookup_keeps_episode_group(tmp_path): + """ + 请求级异步来源查找也须精确匹配剧集组，不能读到默认组或首条记录。 + """ + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'groups.db'}") + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with engine.begin() as connection: + await connection.run_sync(Subscribe.__table__.create) + async with factory() as session: + first = Subscribe(**_payload("电视剧", "first")) + second = Subscribe(**_payload("电视剧", "second")) + session.add_all([first, second]) + await session.commit() + record = await Subscribe.async_get_by( + session, type="电视剧", media_source="themoviedb", media_id="123", season=1, + episode_group="second", + ) + assert record.id == second.id + finally: + await engine.dispose() + + +def test_concurrent_creation_has_one_identity_owner(database, monkeypatch): + """ + 两个并发事务对同一用户同一身份只创建一行并返回相同主键。 + """ + rendezvous = Barrier(2) + original_exists = SubscribeOper._exists + + def simultaneous_lookup(instance, *args, **kwargs): + """ + 强制两个事务都在对方插入前完成空结果读取。 + """ + result = original_exists(instance, *args, **kwargs) + rendezvous.wait(timeout=10) + return result + + monkeypatch.setattr(SubscribeOper, "_exists", simultaneous_lookup) + + def create(_index): + """ + 在独占 Session 中执行真实暂存和提交。 + """ + with database() as session: + result = SubscribeOper(session).stage_add(_identity(), _payload(), "alice") + session.commit() + return result.subscribe_id + + with ThreadPoolExecutor(max_workers=2) as pool: + ids = list(pool.map(create, range(2))) + with database() as session: + assert len(list(session.scalars(select(Subscribe)))) == 1 + assert ids[0] == ids[1] + + +@pytest.mark.asyncio +async def test_async_concurrent_creation_uses_same_database_identity(tmp_path, monkeypatch): + """ + 请求级 AsyncSession 并发新增也由同一数据库唯一键仲裁。 + """ + import asyncio + + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'async.db'}") + factory = async_sessionmaker(engine, expire_on_commit=False) + async with engine.begin() as connection: + await connection.run_sync(Subscribe.__table__.create) + ready = asyncio.Event() + count = 0 + original = SubscribeOper._async_exists + + async def lookup(instance, *args, **kwargs): + """ + 固定两个请求均先查重，再一起进入插入仲裁。 + """ + nonlocal count + record = await original(instance, *args, **kwargs) + count += 1 + if count == 2: + ready.set() + await asyncio.wait_for(ready.wait(), 10) + return record + + monkeypatch.setattr(SubscribeOper, "_async_exists", lookup) + + async def create(): + """ + 使用独占异步会话创建订阅并提交。 + """ + async with factory() as session: + result = await SubscribeOper(session).async_stage_add(_identity(), _payload(), "alice") + await session.commit() + return result.subscribe_id + + try: + ids = await asyncio.gather(create(), create()) + async with factory() as session: + assert await session.scalar(select(func.count()).select_from(Subscribe)) == 1 + assert ids[0] == ids[1] + finally: + await engine.dispose() + + +def test_conflicting_retarget_rolls_back_old_progress(database): + """ + 改成同用户已有身份时明确拒绝，旧记录与另一目标的下载事实均不变。 + """ + with database() as session: + first = Subscribe(**_payload("电视剧"), note=[1], episode_priority={"1": 100}) + second = Subscribe(**{**_payload("电视剧"), "media_id": "456"}, note=[2]) + session.add_all([first, second]) + session.commit() + first_id, second_id = first.id, second.id + service = SyncSubscriptionMutationService( + repository=SessionSubscriptionRepository(session), unit_of_work=SqlAlchemyUnitOfWork(session), + outbox=Mock(), dispatch_store=Mock(), publish_modified=Mock(), + ) + with pytest.raises(SubscriptionIdentityConflictError): + service.update(first_id, {"media_id": "456"}, SubscriptionActor("alice", False)) + assert session.get(Subscribe, first_id).media_id == "123" + assert session.get(Subscribe, first_id).note == [1] + assert session.get(Subscribe, second_id).note == [2] + + +@pytest.mark.parametrize("expired_fields", [None, ["username", "type", "media_source", "season", "episode_group", "music_type"]]) +def test_expired_orm_identity_still_rejects_conflicting_retarget(database, expired_fields): + """ + ORM 已过期属性不能让直接兼容写入跳过唯一身份约束。 + """ + with database() as session: + first = Subscribe(**_payload("电视剧")) + second = Subscribe(**{**_payload("电视剧"), "media_id": "456"}) + session.add_all([first, second]) + session.commit() + first_id = first.id + session.expire(first, expired_fields) + first.media_id = "456" + with pytest.raises(IntegrityError): + session.flush() + session.rollback() + assert session.get(Subscribe, first_id).media_id == "123" + + +def test_expired_cancelled_search_releases_active_key(database): + """ + 旧进程取消中的租约过期后应终结任务，允许同订阅重新入队。 + """ + repository = TransactionalSubscriptionSearchRepository(database) + batch = repository.enqueue(subscription_ids=(7,), source="manual", priority=100) + claimed = repository.claim_next(owner="lost-process") + assert repository.request_cancel(batch.batch.batch_id) + with database() as session: + session.execute(update(SubscriptionSearchTask).where( + SubscriptionSearchTask.task_id == claimed.task_id, + ).values(lease_expires_at=(datetime.now(timezone.utc) - timedelta(hours=1)).isoformat())) + session.commit() + assert repository.claim_next(owner="new-process") is None + fresh = repository.enqueue(subscription_ids=(7,), source="manual", priority=100) + assert fresh.created_count == 1 + assert repository.claim_next(owner="new-process").task_id != claimed.task_id + + +@pytest.mark.parametrize("changed", [{"media_id": "456"}, {"season": 2}, {"episode_group": "new"}]) +def test_retarget_clears_old_episode_progress(changed): + """ + 作品、季或剧集组变化不继承旧目标的下载进度和洗版优先级。 + """ + snapshot = SubscriptionSnapshot(id=7, name="旧作品", type="电视剧", media_source=MediaSource.TMDB, + media_id="123", season=1, note=[1], episode_priority={"1": 100}, current_priority=100, + total_episode=12, lack_episode=11) + values = _normalized_subscription_patch(snapshot, changed).to_payload() + assert values["note"] == [] + assert values["episode_priority"] == {} + assert values["current_priority"] is None + assert values["lack_episode"] == 12 + + +@pytest.mark.asyncio +async def test_servarr_duplicate_series_returns_existing_id(): + """ + 重复添加的兼容接口返回实际记录 ID，后续读取和删除不会误用 1。 + """ + record = SimpleNamespace(id=47, name="示例", year="2026", type="电视剧", season=1, + poster=None, media_source="themoviedb", media_id="123") + repository = Mock(async_exists=AsyncMock(return_value=record)) + service = ServarrSubscriptionService(async_repository=repository, sync_repository=Mock()) + result = await arr_add_series( + SonarrSeries(tmdbId=123, title="示例", seasons=[SonarrSeason(seasonNumber=1, monitored=True)]), + "fake-api-key", service, Mock(), + ) + assert result.id == 47 diff --git a/tests/test_subscription_review_retarget.py b/tests/test_subscription_review_retarget.py new file mode 100644 index 000000000..ef190e06a --- /dev/null +++ b/tests/test_subscription_review_retarget.py @@ -0,0 +1,83 @@ +""" +独立审查发现的订阅目标切换质量事实回归，使用真实事务与同事务outbox暂存。 +""" + +from unittest.mock import AsyncMock, Mock + +from app.application.subscription.mutation import ( + SubscriptionActor, + SubscriptionMutationService, + SyncSubscriptionMutationService, +) +from app.db.adapters.outbox import SqlAlchemyAsyncOutboxStager, SqlAlchemyOutboxStager +from app.db.adapters.subscription import SessionSubscriptionRepository +from app.db.models.outbox import OutboxMessage +from app.db.models.subscribe import Subscribe +from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork + +_QUALITY_FACTS = ("current_audio_format", "current_bitrate", "current_bit_depth", "current_sample_rate") + + +def _old_album(): + """ + 构造已完成部分下载且保存了旧作品音频质量的订阅。 + """ + return Subscribe(name="旧专辑", username="review", type="音乐", music_type="album", + media_source="musicbrainz", media_id="old-release", total_episode=10, lack_episode=8, + note=[1, 2], downloaded_tracks=["old-track"], current_priority=100, + current_audio_format="FLAC", current_bitrate=1411200, current_bit_depth=24, current_sample_rate=96000) + + +def test_sync_album_retarget_clears_old_quality_in_real_transaction(db): + """ + 切换作品后真实行、业务结果与持久事件都不得保留旧版本质量事实。 + """ + db.watermark(Subscribe, OutboxMessage) + album = db.add(_old_album()) + service = SyncSubscriptionMutationService( + SessionSubscriptionRepository(db.session), SqlAlchemyUnitOfWork(db.session), + SqlAlchemyOutboxStager(db.session), Mock(claim_by_event_key=Mock(return_value=None)), Mock()) + result = service.update(album.id, {"media_id": "new-release"}, SubscriptionActor("review", False)) + db.session.expire_all() + stored = db.session.get(Subscribe, album.id) + event = db.session.query(OutboxMessage).filter( + OutboxMessage.event_key.startswith(f"subscribe.modified:{album.id}:"), + ).one() + assert result.business_committed + assert stored.media_id == "new-release" + assert stored.downloaded_tracks == [] + for field in _QUALITY_FACTS: + assert getattr(stored, field) is None + assert getattr(result.snapshot, field) is None + assert event.payload["subscribe_info"][field] is None + + +def test_async_album_retarget_clears_old_quality_in_real_transaction(db): + """ + 请求级异步事务与同步入口保持同一质量清理和outbox提交合同。 + """ + db.watermark(Subscribe, OutboxMessage) + album = db.add(_old_album()) + + async def execute(session): + """ + 在真实AsyncSession中调用正式订阅修改服务。 + """ + service = SubscriptionMutationService( + SessionSubscriptionRepository(session), SqlAlchemyAsyncUnitOfWork(session), + SqlAlchemyAsyncOutboxStager(session), AsyncMock(claim_by_event_key=AsyncMock(return_value=None)), AsyncMock()) + return await service.update(album.id, {"media_id": "new-release"}, SubscriptionActor("review", False)) + + result = db.run_async_session(execute) + db.session.expire_all() + stored = db.session.get(Subscribe, album.id) + event = db.session.query(OutboxMessage).filter( + OutboxMessage.event_key.startswith(f"subscribe.modified:{album.id}:"), + ).one() + assert result.business_committed + assert stored.media_id == "new-release" + assert stored.downloaded_tracks == [] + for field in _QUALITY_FACTS: + assert getattr(stored, field) is None + assert getattr(result.snapshot, field) is None + assert event.payload["subscribe_info"][field] is None diff --git a/tests/test_workflow_review_context.py b/tests/test_workflow_review_context.py new file mode 100644 index 000000000..67720a88c --- /dev/null +++ b/tests/test_workflow_review_context.py @@ -0,0 +1,54 @@ +""" +工作流上下文兼容识别的独立边界：普通文本完整恢复，旧载荷只静态识别。 +""" + +import base64 +import pickle + +import pytest + +from app.chain import workflow as workflow_module +from tests.test_workflow_execution import _build_workflow, _FakeWorkflowManager + + +@pytest.mark.parametrize("content", [ + "gA==", + base64.b64encode(b"\x80\x05").decode(), + base64.b64encode(b"\x80\x05invalid").decode(), + base64.b64encode(b"\x80\x05N.trailing").decode(), + base64.b64encode(b"\x80\xffN.").decode(), + base64.b64encode(b"\x80\x04.").decode(), +]) +def test_pickle_like_plain_text_survives_real_context_restore(content, monkeypatch): + """ + 看似协议头的合法文本及损坏载荷不得被当成有效旧包装丢弃。 + """ + monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: _FakeWorkflowManager([])) + executor = workflow_module.WorkflowExecutor(_build_workflow(context={"content": content})) + try: + assert executor.context.content == content + finally: + executor.executor.shutdown_bounded(timeout=1) + + +@pytest.mark.parametrize("protocol", [2, 4, 5]) +def test_real_legacy_context_is_identified_without_loading(protocol, monkeypatch): + """ + 完整旧协议包装仍从空上下文恢复，识别流程不得执行反序列化。 + """ + content = base64.b64encode(pickle.dumps({"legacy": "旧上下文"}, protocol=protocol)).decode() + + def reject_load(*_args, **_kwargs): + """ + 任何尝试执行载荷的行为都直接使安全回归失败。 + """ + raise AssertionError("旧载荷不得反序列化") + + monkeypatch.setattr(pickle, "loads", reject_load) + monkeypatch.setattr(pickle, "load", reject_load) + monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: _FakeWorkflowManager([])) + executor = workflow_module.WorkflowExecutor(_build_workflow(context={"content": content})) + try: + assert executor.context.content is None + finally: + executor.executor.shutdown_bounded(timeout=1) diff --git a/tests/test_workflow_review_regressions.py b/tests/test_workflow_review_regressions.py new file mode 100644 index 000000000..a7ee9840d --- /dev/null +++ b/tests/test_workflow_review_regressions.py @@ -0,0 +1,225 @@ +""" +验证工作流恢复、并发缓存、调度和共享引用锁的审查回归。 +""" + +import asyncio +import copy +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.interval import IntervalTrigger + +from app.application.configuration import SystemConfigService +from app.chain import workflow as workflow_module +from app.monitor.monitor import Monitor +from app.monitor.poller import RemotePoller +from app.monitor.snapshot import SnapshotStore +from app.runtime.cache import MemoryBackend +from app.scheduler.reconcile import SchedulerReconcileOwner +from app.startup.composition import subscription as composition +from app.workflow.actions import BaseAction +from tests.test_workflow_execution import _build_workflow, _FakeWorkflowManager +from tests.test_workflow_mutation_command import _command, _workflow + + +@pytest.mark.asyncio +async def test_cancelled_reference_lock_waiter_releases_late_acquisition(monkeypatch): + """ + 被取消请求的工作线程迟到拿锁后仍应释放，后续请求可以继续修改。 + """ + lock = threading.Lock() + attempted = threading.Event() + acquired = threading.Event() + + class ObservedLock: + """ + 用事件观察真实线程的锁获取，不访问数据库。 + """ + + def acquire(self): + """ + 报告开始等待和成功获取，以固定取消时序。 + """ + attempted.set() + result = lock.acquire() + acquired.set() + return result + + def release(self): + """ + 释放真实锁。 + """ + lock.release() + + lock.acquire() + monkeypatch.setattr(composition, "_reference_mutation_lock", ObservedLock()) + + async def wait_for_scope(): + """ + 只进入引用修改作用域，取消后不得实际创建数据库会话。 + """ + async with composition.async_rule_group_mutation_scope(lambda _values: None): + raise AssertionError("取消后不应进入事务") + + task = asyncio.create_task(wait_for_scope()) + assert await asyncio.to_thread(attempted.wait, 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + lock.release() + assert await asyncio.to_thread(acquired.wait, 5) + await asyncio.sleep(0) + available = await asyncio.to_thread(lock.acquire, True, 1) + if available: + lock.release() + else: + lock.release() + assert available + + +def test_failed_workflow_node_retries_on_resume(monkeypatch): + """ + 恢复时重新执行失败动作，并重新计算原先因失败跳过的下游。 + """ + calls = [] + workflow = _build_workflow(execution_state={"nodes": {"A": {"state": "failed"}, "B": {"state": "skipped"}}}) + monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: _FakeWorkflowManager(calls)) + executor = workflow_module.WorkflowExecutor(workflow) + executor.execute() + assert calls == ["A", "B"] + assert executor.success + + +def test_append_unique_respects_nested_identity(): + """ + 身份相同的不同内容只保留第一条，身份不同的条目继续追加。 + """ + executor = workflow_module.WorkflowExecutor.__new__(workflow_module.WorkflowExecutor) + first = {"media": {"id": "123"}, "name": "初始"} + second = {"media": {"id": "123"}, "name": "修改"} + third = {"media": {"id": "456"}, "name": "其他"} + assert executor.append_unique_values([first], [second, third], "media.id") == [first, third] + + +@pytest.mark.parametrize("context", [{"content": "合法文本"}, {"content": "合法文本", "workflow_context": {"key": "value"}}]) +def test_current_text_context_survives_resume(context, monkeypatch): + """ + 当前结构的文本内容不是旧序列化载荷，不得丢失恢复数据。 + """ + monkeypatch.setattr(workflow_module, "get_workflow_manager", lambda: _FakeWorkflowManager([])) + executor = workflow_module.WorkflowExecutor(_build_workflow(context=context)) + try: + assert executor.context.content == "合法文本" + assert executor.context.workflow_context == context.get("workflow_context", {}) + finally: + executor.executor.shutdown_bounded(timeout=1) + + +def test_paused_workflow_edit_does_not_register_timer(): + """ + 修改暂停定义仍可保存，但不得重新启动定时执行。 + """ + workflow = _workflow() + workflow.state = "P" + command, dependencies = _command(workflow) + assert command.update({"id": 7, "description": "修改说明"}).success + dependencies["add_timer"].assert_not_called() + + +def test_monitor_interval_change_uses_real_scheduler(monkeypatch): + """ + 文件数增长后真实 APScheduler 的下轮触发间隔随之调整。 + """ + monitor = Monitor.__new__(Monitor) + monitor._scheduler = BackgroundScheduler() + monitor._scheduler.add_job(lambda: None, IntervalTrigger(minutes=1), id="monitor_remote") + monitor._poller = SimpleNamespace(poll=lambda **_kwargs: 1001) + monkeypatch.setattr(monitor, "_Monitor__accepting_work", lambda: True) + monitor.polling_observer("remote", [Path("/library")]) + assert monitor._scheduler.get_job("monitor_remote").trigger.interval.total_seconds() == ( + SnapshotStore.adjust_interval(1001) * 60 + ) + + +def test_workflow_cron_uses_scheduler_timezone(): + """ + Cron 触发器必须沿用调度器时区，不能隐式使用宿主时区。 + """ + scheduler = BackgroundScheduler(timezone="Asia/Shanghai") + owner = Mock(_scheduler=scheduler, _lock=threading.RLock(), _jobs={}) + owner.start = lambda **_kwargs: None + workflow = SimpleNamespace(id=7, name="时区", timer="0 8 * * *") + SchedulerReconcileOwner.update_workflow_job(owner, workflow) + assert str(scheduler.get_job("workflow-7").trigger.timezone) == "Asia/Shanghai" + + +@pytest.mark.parametrize("existing_baseline", [False, True]) +def test_single_root_full_scan_preserves_shared_incremental_cursor(monkeypatch, existing_baseline): + """ + 只扫描一个根不应让另一根尚未消费的修改被共享游标越过。 + """ + store = SnapshotStore(cache=MemoryBackend()) + if existing_baseline: + assert store.save("remote", {"/b/old.mkv": {"size": 1, "modify_time": 10}}, snapshot_time=10) + dispatcher = SimpleNamespace(is_transfer_candidate_path=lambda _path: True, handle_file=lambda **_kwargs: True) + poller = RemotePoller(store, dispatcher) + monkeypatch.setattr("app.monitor.poller.StorageChain", lambda: SimpleNamespace( + snapshot_storage=lambda **_kwargs: {"/a/new.mkv": {"size": 1, "modify_time": 100}}, + )) + assert poller.force_full_scan("remote", Path("/a")) + loaded, success = store.load_checked("remote") + assert success + assert loaded["timestamp"] == (10 if existing_baseline else 0) + if existing_baseline: + assert "/b/old.mkv" in loaded["snapshot"] + + +def test_parallel_action_cache_updates_are_atomic(): + """ + 并行动作通过同一持久原子更新入口合并，不能丢失已成功分支。 + """ + state = {} + lock = threading.Lock() + rendezvous = threading.Barrier(2) + + class Repository: + """ + 在真实线程中回放配置事务，分离读取会固定得到同一旧值。 + """ + + def get(self, key): + """ + 模拟两个分离读取同时读到旧快照。 + """ + value = copy.deepcopy(state.get(key)) + rendezvous.wait(timeout=5) + return value + + def set(self, key, value): + """ + 保存配置快照。 + """ + state[key] = copy.deepcopy(value) + + def update_atomically(self, key, mutation): + """ + 在同一锁内读取并提交一次配置变更。 + """ + with lock: + result, value = mutation(None, copy.deepcopy(state.get(key))) + state[key] = value + return result + + service = SystemConfigService(reader=Repository(), writer=Repository()) + actions = [] + for action_id in ["A", "B"]: + action = Mock(_action_id=action_id, _cache_key="WorkflowCache_%s", systemconfigoper=service) + actions.append(action) + with ThreadPoolExecutor(max_workers=2) as pool: + list(pool.map(lambda action: BaseAction.save_cache(action, 7, action._action_id), actions)) + assert state["WorkflowCache_7"] == {"A": ["A"], "B": ["B"]}