From 362f606751ae44e0a0919cf62a3041027f488801 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Wed, 26 Aug 2026 12:04:18 +0800 Subject: [PATCH] feat(governance): unify durable event retention --- app/application/configuration.py | 5 + app/application/maintenance.py | 108 ++++++++++- app/application/outbox.py | 12 +- app/application/subscription/complete.py | 3 +- app/chain/transfer.py | 178 +++++++++--------- app/db/adapters/outbox.py | 1 - app/db/maintenance.py | 121 ++++++++++++ app/db/models/agentchat.py | 1 + app/db/models/agenttaskrun.py | 1 + app/db/models/subscribehistory.py | 1 + app/runtime/config.py | 10 + app/runtime/event/contracts.py | 5 + app/scheduler.py | 7 +- app/startup/composition/configuration.py | 5 + app/startup/initializers/modules.py | 28 ++- database/versions/a6c8e2f4b1d3_3_0_11.py | 56 ++++++ .../adr/0007-background-action-reliability.md | 20 +- docs/architecture-overview.md | 8 +- docs/rules/05-architecture.md | 5 + docs/rules/10-data-and-persistent.md | 23 ++- .../architecture/dependency-baseline.json | 10 +- .../runtime-contract-baseline.json | 32 +++- tests/test_agent_task_run_migration.py | 14 +- tests/test_cleanup_index_migration.py | 75 ++++++++ tests/test_data_cleanup_growth_tables.py | 156 +++++++++++++++ tests/test_data_cleanup_service.py | 32 +++- tests/test_event_contracts.py | 1 + tests/test_outbox.py | 96 ++++++++++ tests/test_transfer_overwrite_declined.py | 55 +++++- 29 files changed, 948 insertions(+), 121 deletions(-) create mode 100644 database/versions/a6c8e2f4b1d3_3_0_11.py create mode 100644 tests/test_cleanup_index_migration.py create mode 100644 tests/test_data_cleanup_growth_tables.py diff --git a/app/application/configuration.py b/app/application/configuration.py index b1d9a46b0..43ce0c09d 100644 --- a/app/application/configuration.py +++ b/app/application/configuration.py @@ -167,6 +167,11 @@ class ChainRuntimeConfig: data_cleanup_site_userdata_days: Any = 0 data_cleanup_transfer_history_days: Any = 0 data_cleanup_download_failure_days: Any = 0 + data_cleanup_subscribe_history_days: Any = 0 + data_cleanup_agent_chat_days: Any = 0 + data_cleanup_agent_task_run_days: Any = 0 + data_cleanup_outbox_completed_days: Any = 0 + data_cleanup_outbox_dead_days: Any = 0 download_subtitle: bool = True lyrics_batch_timeout: int = 120 music_metadata_to_simplified: bool = True diff --git a/app/application/maintenance.py b/app/application/maintenance.py index c2da0beda..cb92b0b2e 100644 --- a/app/application/maintenance.py +++ b/app/application/maintenance.py @@ -6,7 +6,7 @@ import json from dataclasses import dataclass -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from typing import Any, Callable, ContextManager, Dict, Optional, Protocol from app.application.configuration import get_chain_runtime_config_snapshot @@ -26,6 +26,11 @@ class CleanupPolicy: site_userdata_days: int transfer_history_days: int download_failure_days: int + subscribe_history_days: int + agent_chat_days: int + agent_task_run_days: int + outbox_completed_days: int + outbox_dead_days: int @dataclass(frozen=True, slots=True) @@ -73,6 +78,26 @@ class CleanupRepository(Protocol): """删除已经过期的下载失败冷却记录。""" ... + def delete_subscribe_history(self, db: Any, cutoff: str, limit: int) -> int: + """删除早于截止时间的订阅历史。""" + ... + + def delete_agent_chats(self, db: Any, cutoff: str, limit: int) -> int: + """删除早于截止时间且未被任务引用的 Agent 会话。""" + ... + + def delete_agent_task_runs(self, db: Any, cutoff: str, limit: int) -> int: + """删除早于截止时间且不再承担恢复语义的 Agent 运行历史。""" + ... + + def delete_outbox_completed(self, db: Any, cutoff: str, limit: int) -> int: + """删除早于截止时间的 Outbox 已完成记录。""" + ... + + def delete_outbox_dead(self, db: Any, cutoff: str, limit: int) -> int: + """删除早于截止时间的 Outbox 死信记录。""" + ... + class CleanupUnitOfWork(Protocol): """数据维护每一批删除所需的最小事务能力。""" @@ -249,6 +274,29 @@ class DataCleanupService: policy.download_failure_days, "%Y-%m-%d %H:%M:%S", ) + subscribe_history_cutoff = self._cutoff( + started_at, + policy.subscribe_history_days, + "%Y-%m-%d %H:%M:%S", + ) + agent_chat_cutoff = self._cutoff( + started_at, + policy.agent_chat_days, + "%Y-%m-%d %H:%M:%S", + ) + agent_task_run_cutoff = self._cutoff( + started_at, + policy.agent_task_run_days, + "%Y-%m-%d %H:%M:%S", + ) + outbox_completed_cutoff = self._outbox_cutoff( + started_at, + policy.outbox_completed_days, + ) + outbox_dead_cutoff = self._outbox_cutoff( + started_at, + policy.outbox_dead_days, + ) return [ CleanupPlan( "message", @@ -296,6 +344,46 @@ class DataCleanupService: db, download_failure_cutoff, batch_size ), ), + CleanupPlan( + "subscribehistory", + policy.subscribe_history_days, + subscribe_history_cutoff, + lambda db: self._repository.delete_subscribe_history( + db, subscribe_history_cutoff, batch_size + ), + ), + CleanupPlan( + "agentchat", + policy.agent_chat_days, + agent_chat_cutoff, + lambda db: self._repository.delete_agent_chats( + db, agent_chat_cutoff, batch_size + ), + ), + CleanupPlan( + "agenttaskrun", + policy.agent_task_run_days, + agent_task_run_cutoff, + lambda db: self._repository.delete_agent_task_runs( + db, agent_task_run_cutoff, batch_size + ), + ), + CleanupPlan( + "outbox_completed", + policy.outbox_completed_days, + outbox_completed_cutoff, + lambda db: self._repository.delete_outbox_completed( + db, outbox_completed_cutoff, batch_size + ), + ), + CleanupPlan( + "outbox_dead", + policy.outbox_dead_days, + outbox_dead_cutoff, + lambda db: self._repository.delete_outbox_dead( + db, outbox_dead_cutoff, batch_size + ), + ), ] def _cleanup_in_batches( @@ -326,6 +414,19 @@ class DataCleanupService: """按兼容格式计算一个清理截止时间。""" return (started_at - timedelta(days=retention_days)).strftime(pattern) + @staticmethod + def _outbox_cutoff(started_at: datetime, retention_days: int) -> str: + """按 Outbox 的 UTC ISO 格式生成可排序截止时间。""" + aware_started_at = ( + started_at.astimezone() + if started_at.tzinfo is None + else started_at + ) + return ( + aware_started_at.astimezone(timezone.utc) + - timedelta(days=retention_days) + ).isoformat() + def read_cleanup_policy() -> CleanupPolicy: """读取并规范化当前数据清理配置，单次运行期间保持快照一致。""" @@ -337,6 +438,11 @@ def read_cleanup_policy() -> CleanupPolicy: site_userdata_days=_normalize_days(config.data_cleanup_site_userdata_days), transfer_history_days=_normalize_days(config.data_cleanup_transfer_history_days), download_failure_days=_normalize_days(config.data_cleanup_download_failure_days), + subscribe_history_days=_normalize_days(config.data_cleanup_subscribe_history_days), + agent_chat_days=_normalize_days(config.data_cleanup_agent_chat_days), + agent_task_run_days=_normalize_days(config.data_cleanup_agent_task_run_days), + outbox_completed_days=_normalize_days(config.data_cleanup_outbox_completed_days), + outbox_dead_days=_normalize_days(config.data_cleanup_outbox_dead_days), ) diff --git a/app/application/outbox.py b/app/application/outbox.py index ec156dbb7..793810d6f 100644 --- a/app/application/outbox.py +++ b/app/application/outbox.py @@ -17,18 +17,28 @@ T = TypeVar("T") SUBSCRIBE_ADDED_TOPIC = "subscribe.added" SUBSCRIBE_MODIFIED_TOPIC = "subscribe.modified" SUBSCRIBE_DELETED_TOPIC = "subscribe.deleted" +SUBSCRIBE_COMPLETED_TOPIC = "subscribe.complete" DOWNLOAD_ADDED_TOPIC = "download.added" TRANSFER_COMPLETED_TOPIC = "transfer.completed" TRANSFER_FAILED_TOPIC = "transfer.failed" +SUBTITLE_TRANSFER_COMPLETED_TOPIC = "transfer.subtitle.completed" +SUBTITLE_TRANSFER_FAILED_TOPIC = "transfer.subtitle.failed" +AUDIO_TRANSFER_COMPLETED_TOPIC = "transfer.audio.completed" +AUDIO_TRANSFER_FAILED_TOPIC = "transfer.audio.failed" OUTBOX_LEASE_SECONDS = 60 DURABLE_EVENT_TOPICS: Mapping[EventType, str] = MappingProxyType({ EventType.SubscribeAdded: SUBSCRIBE_ADDED_TOPIC, EventType.SubscribeModified: SUBSCRIBE_MODIFIED_TOPIC, EventType.SubscribeDeleted: SUBSCRIBE_DELETED_TOPIC, + EventType.SubscribeComplete: SUBSCRIBE_COMPLETED_TOPIC, EventType.DownloadAdded: DOWNLOAD_ADDED_TOPIC, EventType.TransferComplete: TRANSFER_COMPLETED_TOPIC, EventType.TransferFailed: TRANSFER_FAILED_TOPIC, + EventType.SubtitleTransferComplete: SUBTITLE_TRANSFER_COMPLETED_TOPIC, + EventType.SubtitleTransferFailed: SUBTITLE_TRANSFER_FAILED_TOPIC, + EventType.AudioTransferComplete: AUDIO_TRANSFER_COMPLETED_TOPIC, + EventType.AudioTransferFailed: AUDIO_TRANSFER_FAILED_TOPIC, }) @@ -96,7 +106,6 @@ class OutboxRepository(Protocol): ) -> None: """记录有限退避或 dead-letter 终态。""" - class AsyncOutboxTransaction(Protocol): """异步业务事务暂存并收口 durable intent 的最小端口。""" @@ -236,7 +245,6 @@ class OutboxDispatcher: """释放 dispatcher 工厂创建的短生命周期持久化资源。""" self._close() - _configured_dispatcher: Callable[[], OutboxDispatcher] | None = None diff --git a/app/application/subscription/complete.py b/app/application/subscription/complete.py index a3b3bfdbc..19e174fff 100644 --- a/app/application/subscription/complete.py +++ b/app/application/subscription/complete.py @@ -10,6 +10,7 @@ from typing import Any, Protocol from app.application.outbox import ( OUTBOX_LEASE_SECONDS, OutboxIntent, + SUBSCRIBE_COMPLETED_TOPIC, SyncOutboxTransaction, SyncUnitOfWork, ) @@ -77,7 +78,7 @@ class CompleteSubscriptionCommand: self._outbox.stage( OutboxIntent( event_key=event_key, - topic="subscribe.complete", + topic=SUBSCRIBE_COMPLETED_TOPIC, payload=event_payload, ), now, diff --git a/app/chain/transfer.py b/app/chain/transfer.py index f234569ff..9a84e3071 100755 --- a/app/chain/transfer.py +++ b/app/chain/transfer.py @@ -41,7 +41,14 @@ from app.application.history import ( is_skip_action, record_transfer_failure, ) -from app.application.outbox import TRANSFER_COMPLETED_TOPIC, TRANSFER_FAILED_TOPIC +from app.application.outbox import ( + AUDIO_TRANSFER_COMPLETED_TOPIC, + AUDIO_TRANSFER_FAILED_TOPIC, + SUBTITLE_TRANSFER_COMPLETED_TOPIC, + SUBTITLE_TRANSFER_FAILED_TOPIC, + TRANSFER_COMPLETED_TOPIC, + TRANSFER_FAILED_TOPIC, +) from app.application.transfer import ( FailedRetryScheduler, JobManager, @@ -146,6 +153,53 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo "transfer_history_id": history_id, } + def _durable_transfer_event( + self, + task: TransferTask, + *, + success: bool, + ) -> Optional[tuple[str, EventType]]: + """返回当前整理结果应持久化的 topic 与兼容事件类型。""" + if success: + if self._is_primary_media_file(task.fileitem, task.mediainfo): + return TRANSFER_COMPLETED_TOPIC, EventType.TransferComplete + if self._is_subtitle_file(task.fileitem): + return ( + SUBTITLE_TRANSFER_COMPLETED_TOPIC, + EventType.SubtitleTransferComplete, + ) + if self._is_audio_file(task.fileitem): + return AUDIO_TRANSFER_COMPLETED_TOPIC, EventType.AudioTransferComplete + return None + if self._is_media_file(task.fileitem): + return TRANSFER_FAILED_TOPIC, EventType.TransferFailed + if self._is_subtitle_file(task.fileitem): + return SUBTITLE_TRANSFER_FAILED_TOPIC, EventType.SubtitleTransferFailed + if self._is_audio_file(task.fileitem): + return AUDIO_TRANSFER_FAILED_TOPIC, EventType.AudioTransferFailed + return None + + def _publish_transfer_result( + self, + event_type: EventType, + payload: dict[str, Any], + ) -> None: + """显式分派整理结果，使运行契约可追踪每种事件的生产者。""" + if event_type is EventType.TransferComplete: + self.eventmanager.send_event(EventType.TransferComplete, payload) + elif event_type is EventType.TransferFailed: + self.eventmanager.send_event(EventType.TransferFailed, payload) + elif event_type is EventType.SubtitleTransferComplete: + self.eventmanager.send_event(EventType.SubtitleTransferComplete, payload) + elif event_type is EventType.SubtitleTransferFailed: + self.eventmanager.send_event(EventType.SubtitleTransferFailed, payload) + elif event_type is EventType.AudioTransferComplete: + self.eventmanager.send_event(EventType.AudioTransferComplete, payload) + elif event_type is EventType.AudioTransferFailed: + self.eventmanager.send_event(EventType.AudioTransferFailed, payload) + else: + raise ValueError(f"不支持的整理结果事件：{event_type}") + def __init__(self) -> None: """初始化文件整理处理链。""" super().__init__() @@ -479,14 +533,16 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo fileid=task.fileitem.fileid if task.fileitem else None, ) + durable_event = self._durable_transfer_event(task, success=False) durable_transfer_failed = bool( - getattr(self, "durable_event_writer", None) - and self._is_media_file(task.fileitem) + getattr(self, "durable_event_writer", None) and durable_event ) if durable_transfer_failed: + assert durable_event is not None + topic, event_type = durable_event event_payload = self._transfer_result_payload(task, transferinfo) history = self.durable_event_writer.transfer_result( - topic=TRANSFER_FAILED_TOPIC, + topic=topic, stage_history=lambda writer: add_transfer_fail( fileitem=task.fileitem, mode=transferinfo.transfer_type if transferinfo else "", @@ -498,9 +554,8 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo transfer_history_oper=writer, ), event_payload=event_payload, - publish=lambda payload: self.eventmanager.send_event( - EventType.TransferFailed, - payload, + publish=lambda payload: self._publish_transfer_result( + event_type, payload ), ) else: @@ -515,45 +570,15 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo transfer_history_oper=transferhis, ) - # 整理失败事件 - if self._is_media_file(task.fileitem): - if not durable_transfer_failed: - # 显式旧测试上下文仍走原始发送；正式上下文由 outbox writer 发布。 - self.eventmanager.send_event( - EventType.TransferFailed, - self._transfer_result_payload( - task, - transferinfo, - history.id if history else None, - ), - ) - elif self._is_subtitle_file(task.fileitem): - # 字幕整理失败事件 - self.eventmanager.send_event( - EventType.SubtitleTransferFailed, - { - "fileitem": task.fileitem, - "meta": task.meta, - "mediainfo": task.mediainfo, - "transferinfo": transferinfo, - "downloader": task.downloader, - "download_hash": task.download_hash, - "transfer_history_id": history.id if history else None, - }, - ) - elif self._is_audio_file(task.fileitem): - # 音频文件整理失败事件 - self.eventmanager.send_event( - EventType.AudioTransferFailed, - { - "fileitem": task.fileitem, - "meta": task.meta, - "mediainfo": task.mediainfo, - "transferinfo": transferinfo, - "downloader": task.downloader, - "download_hash": task.download_hash, - "transfer_history_id": history.id if history else None, - }, + if durable_event and not durable_transfer_failed: + # 显式旧测试上下文仍走原始发送；正式上下文由 outbox writer 发布。 + self._publish_transfer_result( + durable_event[1], + self._transfer_result_payload( + task, + transferinfo, + history.id if history else None, + ), ) self.queue_failed_transfer_notification( @@ -592,14 +617,16 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo task.fileitem.storage if task.fileitem else None, ) + durable_event = self._durable_transfer_event(task, success=True) durable_transfer_complete = bool( - getattr(self, "durable_event_writer", None) - and self._is_primary_media_file(task.fileitem, task.mediainfo) + getattr(self, "durable_event_writer", None) and durable_event ) if durable_transfer_complete: + assert durable_event is not None + topic, event_type = durable_event event_payload = self._transfer_result_payload(task, transferinfo) history = self.durable_event_writer.transfer_result( - topic=TRANSFER_COMPLETED_TOPIC, + topic=topic, stage_history=lambda writer: add_transfer_success( fileitem=task.fileitem, mode=transferinfo.transfer_type if transferinfo else "", @@ -611,9 +638,8 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo transfer_history_oper=writer, ), event_payload=event_payload, - publish=lambda payload: self.eventmanager.send_event( - EventType.TransferComplete, - payload, + publish=lambda payload: self._publish_transfer_result( + event_type, payload ), ) else: @@ -628,45 +654,15 @@ class TransferChain(FileFilterMixin, ScrapeBatchMixin, EpisodeFormatMixin, Histo transfer_history_oper=transferhis, ) - # task整理完成事件 - if self._is_primary_media_file(task.fileitem, task.mediainfo): - if not durable_transfer_complete: - # 显式旧测试上下文仍走原始发送；正式上下文由 outbox writer 发布。 - self.eventmanager.send_event( - EventType.TransferComplete, - self._transfer_result_payload( - task, - transferinfo, - history.id if history else None, - ), - ) - elif self._is_subtitle_file(task.fileitem): - # 字幕整理完成事件 - self.eventmanager.send_event( - EventType.SubtitleTransferComplete, - { - "fileitem": task.fileitem, - "meta": task.meta, - "mediainfo": task.mediainfo, - "transferinfo": transferinfo, - "downloader": task.downloader, - "download_hash": task.download_hash, - "transfer_history_id": history.id if history else None, - }, - ) - elif self._is_audio_file(task.fileitem): - # 音频文件整理完成事件 - self.eventmanager.send_event( - EventType.AudioTransferComplete, - { - "fileitem": task.fileitem, - "meta": task.meta, - "mediainfo": task.mediainfo, - "transferinfo": transferinfo, - "downloader": task.downloader, - "download_hash": task.download_hash, - "transfer_history_id": history.id if history else None, - }, + if durable_event and not durable_transfer_complete: + # 显式旧测试上下文仍走原始发送；正式上下文由 outbox writer 发布。 + self._publish_transfer_result( + durable_event[1], + self._transfer_result_payload( + task, + transferinfo, + history.id if history else None, + ), ) # task登记转移成功文件清单 diff --git a/app/db/adapters/outbox.py b/app/db/adapters/outbox.py index 0aa372915..740a2005e 100644 --- a/app/db/adapters/outbox.py +++ b/app/db/adapters/outbox.py @@ -176,7 +176,6 @@ class SqlAlchemyOutboxRepository: ) self._session.commit() - class SqlAlchemyAsyncOutboxStager: """只负责把 outbox 意图加入调用方异步事务。""" diff --git a/app/db/maintenance.py b/app/db/maintenance.py index 248caac4a..7ac847a9f 100644 --- a/app/db/maintenance.py +++ b/app/db/maintenance.py @@ -2,10 +2,18 @@ from typing import Any, Callable, ContextManager +from sqlalchemy import delete, exists, select + +from app.db.base import execute_dml +from app.db.models.agentchat import AgentChat +from app.db.models.agenttask import AgentTask +from app.db.models.agenttaskrun import AgentTaskRun from app.db.models.downloadfailure import DownloadFailure from app.db.models.downloadhistory import DownloadFiles, DownloadHistory from app.db.models.message import Message +from app.db.models.outbox import OutboxMessage from app.db.models.siteuserdata import SiteUserData +from app.db.models.subscribehistory import SubscribeHistory from app.db.models.transferhistory import TransferHistory from app.db.uow import SqlAlchemyUnitOfWork @@ -67,3 +75,116 @@ class DatabaseCleanupRepository: before_time=cutoff, limit=limit, ) + + @staticmethod + def delete_subscribe_history(db: Any, cutoff: str, limit: int) -> int: + """分批删除超过用户保留期的已完成订阅快照。""" + return DatabaseCleanupRepository._delete_selected_ids( + db=db, + model=SubscribeHistory, + condition=SubscribeHistory.date < cutoff, + limit=limit, + ) + + @staticmethod + def delete_agent_chats(db: Any, cutoff: str, limit: int) -> int: + """清理旧会话，但保留仍被 Agent 定时任务引用的上下文。""" + task_reference = exists( + select(AgentTask.id).where(AgentTask.session_id == AgentChat.session_id) + ) + return DatabaseCleanupRepository._delete_selected_ids( + db=db, + model=AgentChat, + condition=(AgentChat.updated_at < cutoff) & ~task_reference, + limit=limit, + ) + + @staticmethod + def delete_agent_task_runs(db: Any, cutoff: str, limit: int) -> int: + """清理旧终态运行，但保留运行中和父任务最后一次运行。""" + latest_run_reference = exists( + select(AgentTask.id).where(AgentTask.last_run_id == AgentTaskRun.run_id) + ) + return DatabaseCleanupRepository._delete_selected_ids( + db=db, + model=AgentTaskRun, + condition=( + (AgentTaskRun.started_at < cutoff) + & (AgentTaskRun.status != "running") + & ~latest_run_reference + ), + limit=limit, + ) + + @staticmethod + def delete_outbox_completed(db: Any, cutoff: str, limit: int) -> int: + """删除过期 completed 记录，事务提交由维护用例统一控制。""" + return DatabaseCleanupRepository._delete_outbox_status( + db=db, + status="completed", + timestamp_column=OutboxMessage.completed_at, + cutoff=cutoff, + limit=limit, + ) + + @staticmethod + def delete_outbox_dead(db: Any, cutoff: str, limit: int) -> int: + """删除过期 dead-letter 记录，保留 pending/processing 恢复语义。""" + return DatabaseCleanupRepository._delete_outbox_status( + db=db, + status="dead", + timestamp_column=OutboxMessage.next_retry_at, + cutoff=cutoff, + limit=limit, + ) + + @staticmethod + def _delete_outbox_status( + *, + db: Any, + status: str, + timestamp_column: Any, + cutoff: str, + limit: int, + ) -> int: + """先锁定有限 ID 再删除，避免一次维护事务无界膨胀。""" + message_ids = db.execute( + select(OutboxMessage.id) + .where( + OutboxMessage.status == status, + timestamp_column.is_not(None), + timestamp_column < cutoff, + ) + .order_by(OutboxMessage.id) + .limit(limit) + ).scalars().all() + if not message_ids: + return 0 + return execute_dml( + db, + delete(OutboxMessage).where(OutboxMessage.id.in_(message_ids)), + execution_options={"synchronize_session": False}, + ) + + @staticmethod + def _delete_selected_ids( + *, + db: Any, + model: Any, + condition: Any, + limit: int, + ) -> int: + """按安全谓词锁定有限主键并暂存删除。""" + record_ids = db.execute( + select(model.id) + .where(condition) + .order_by(model.id) + .limit(limit) + ).scalars().all() + if not record_ids: + return 0 + return execute_dml( + db, + delete(model).where(model.id.in_(record_ids)), + execution_options={"synchronize_session": False}, + ) diff --git a/app/db/models/agentchat.py b/app/db/models/agentchat.py index 3ecdc7071..2057651da 100644 --- a/app/db/models/agentchat.py +++ b/app/db/models/agentchat.py @@ -46,6 +46,7 @@ class AgentChat(Base): Index("ix_agentchat_session_user", "session_id", "user_id"), Index("ix_agentchat_user_updated", "user_id", "updated_at", "id"), Index("ix_agentchat_channel_updated", "channel", "updated_at", "id"), + Index("ix_agentchat_updated_id", "updated_at", "id"), ) @classmethod diff --git a/app/db/models/agenttaskrun.py b/app/db/models/agenttaskrun.py index 5cd2803f0..4939d81fb 100644 --- a/app/db/models/agenttaskrun.py +++ b/app/db/models/agenttaskrun.py @@ -37,6 +37,7 @@ class AgentTaskRun(Base): __table_args__ = ( Index("ix_agenttaskrun_run_id", "run_id", unique=True), Index("ix_agenttaskrun_task_started", "task_id", "started_at", "id"), + Index("ix_agenttaskrun_status_started_id", "status", "started_at", "id"), ) @classmethod diff --git a/app/db/models/subscribehistory.py b/app/db/models/subscribehistory.py index 664789565..f83d75cda 100644 --- a/app/db/models/subscribehistory.py +++ b/app/db/models/subscribehistory.py @@ -102,6 +102,7 @@ class SubscribeHistory(Base): __table_args__ = ( media_identity_constraint("subscribehistory"), Index('ix_subscribehistory_type_date', 'type', 'date'), + Index('ix_subscribehistory_date_id', 'date', 'id'), Index('ix_subscribehistory_media_identity', 'media_source', 'media_id'), ) diff --git a/app/runtime/config.py b/app/runtime/config.py index 9ba21cb7f..1e390efd7 100644 --- a/app/runtime/config.py +++ b/app/runtime/config.py @@ -218,6 +218,16 @@ class ConfigModel(BaseModel): DATA_CLEANUP_TRANSFER_HISTORY_DAYS: int = 365 * 3 # 下载失败冷却记录保留天数，0为不清理 DATA_CLEANUP_DOWNLOAD_FAILURE_DAYS: int = 7 + # 订阅完成历史保留天数，0为不清理 + DATA_CLEANUP_SUBSCRIBE_HISTORY_DAYS: int = 365 * 3 + # Agent 会话历史保留天数，0为不清理 + DATA_CLEANUP_AGENT_CHAT_DAYS: int = 180 + # Agent 定时任务运行历史保留天数，0为不清理 + DATA_CLEANUP_AGENT_TASK_RUN_DAYS: int = 180 + # Outbox 已完成记录保留天数，0为不清理 + DATA_CLEANUP_OUTBOX_COMPLETED_DAYS: int = 30 + # Outbox 死信记录保留天数，0为不清理 + DATA_CLEANUP_OUTBOX_DEAD_DAYS: int = 90 # ==================== 缓存配置 ==================== # 缓存类型，支持 cachetools 和 redis，默认使用 cachetools diff --git a/app/runtime/event/contracts.py b/app/runtime/event/contracts.py index bab81eabe..bc188a5c6 100644 --- a/app/runtime/event/contracts.py +++ b/app/runtime/event/contracts.py @@ -159,9 +159,14 @@ _DURABLE_REQUIRED = { EventType.SubscribeAdded, EventType.SubscribeModified, EventType.SubscribeDeleted, + EventType.SubscribeComplete, EventType.DownloadAdded, EventType.TransferComplete, EventType.TransferFailed, + EventType.SubtitleTransferComplete, + EventType.SubtitleTransferFailed, + EventType.AudioTransferComplete, + EventType.AudioTransferFailed, } _TARGET_PLUGIN = {EventType.PluginAction, EventType.PluginTriggered} _HOST_ONLY = {EventType.SystemError, EventType.ConfigChanged, EventType.ModuleReload} diff --git a/app/scheduler.py b/app/scheduler.py index 05aefe3a4..77c67b945 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -126,6 +126,12 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): "DATA_CLEANUP_DOWNLOAD_HISTORY_DAYS", "DATA_CLEANUP_SITE_USERDATA_DAYS", "DATA_CLEANUP_TRANSFER_HISTORY_DAYS", + "DATA_CLEANUP_DOWNLOAD_FAILURE_DAYS", + "DATA_CLEANUP_SUBSCRIBE_HISTORY_DAYS", + "DATA_CLEANUP_AGENT_CHAT_DAYS", + "DATA_CLEANUP_AGENT_TASK_RUN_DAYS", + "DATA_CLEANUP_OUTBOX_COMPLETED_DAYS", + "DATA_CLEANUP_OUTBOX_DEAD_DAYS", "DB_BACKUP_ENABLE", "DB_BACKUP_CRON", "USAGE_STATISTIC_SHARE", @@ -591,7 +597,6 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): kwargs={"job_id": "outbox_dispatch"}, replace_existing=True, ) - # CookieCloud定时同步 if ( config.cookiecloud_interval diff --git a/app/startup/composition/configuration.py b/app/startup/composition/configuration.py index f4dd1adae..5bb6c08b6 100644 --- a/app/startup/composition/configuration.py +++ b/app/startup/composition/configuration.py @@ -120,6 +120,11 @@ def build_chain_runtime_config(settings: Settings) -> ChainRuntimeConfig: data_cleanup_site_userdata_days=settings.DATA_CLEANUP_SITE_USERDATA_DAYS, data_cleanup_transfer_history_days=settings.DATA_CLEANUP_TRANSFER_HISTORY_DAYS, data_cleanup_download_failure_days=settings.DATA_CLEANUP_DOWNLOAD_FAILURE_DAYS, + data_cleanup_subscribe_history_days=settings.DATA_CLEANUP_SUBSCRIBE_HISTORY_DAYS, + data_cleanup_agent_chat_days=settings.DATA_CLEANUP_AGENT_CHAT_DAYS, + data_cleanup_agent_task_run_days=settings.DATA_CLEANUP_AGENT_TASK_RUN_DAYS, + data_cleanup_outbox_completed_days=settings.DATA_CLEANUP_OUTBOX_COMPLETED_DAYS, + data_cleanup_outbox_dead_days=settings.DATA_CLEANUP_OUTBOX_DEAD_DAYS, download_subtitle=settings.DOWNLOAD_SUBTITLE, music_metadata_to_simplified=settings.MUSIC_METADATA_TO_SIMPLIFIED, recognize_plugin_first=settings.RECOGNIZE_PLUGIN_FIRST, diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index 95486f1fc..0e45ef3c0 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -374,7 +374,9 @@ def _build_outbox_dispatcher() -> OutboxDispatcher: message.payload, ), "subscribe.deleted.report": dispatch_subscribe_deleted_report, - "subscribe.complete": lambda message: EventManager().send_event( + durable_event_topic( + EventType.SubscribeComplete + ): lambda message: EventManager().send_event( EventType.SubscribeComplete, message.payload, ), @@ -398,6 +400,30 @@ def _build_outbox_dispatcher() -> OutboxDispatcher: EventType.TransferFailed, restore_transfer_result(message.payload), ), + durable_event_topic( + EventType.SubtitleTransferComplete + ): lambda message: EventManager().send_event( + EventType.SubtitleTransferComplete, + restore_transfer_result(message.payload), + ), + durable_event_topic( + EventType.SubtitleTransferFailed + ): lambda message: EventManager().send_event( + EventType.SubtitleTransferFailed, + restore_transfer_result(message.payload), + ), + durable_event_topic( + EventType.AudioTransferComplete + ): lambda message: EventManager().send_event( + EventType.AudioTransferComplete, + restore_transfer_result(message.payload), + ), + durable_event_topic( + EventType.AudioTransferFailed + ): lambda message: EventManager().send_event( + EventType.AudioTransferFailed, + restore_transfer_result(message.payload), + ), } validate_durable_event_handlers(handlers) session = SessionFactory() diff --git a/database/versions/a6c8e2f4b1d3_3_0_11.py b/database/versions/a6c8e2f4b1d3_3_0_11.py new file mode 100644 index 000000000..5955e1ad6 --- /dev/null +++ b/database/versions/a6c8e2f4b1d3_3_0_11.py @@ -0,0 +1,56 @@ +"""3.0.11 add retention cleanup indexes. + +Revision ID: a6c8e2f4b1d3 +Revises: e4f7a1b2c3d5 +Create Date: 2026-08-26 +""" + +import sqlalchemy as sa +from alembic import op + + +revision = "a6c8e2f4b1d3" +down_revision = "e4f7a1b2c3d5" +branch_labels = None +depends_on = None + + +_CLEANUP_INDEXES = { + "subscribehistory": ( + "ix_subscribehistory_date_id", + ("date", "id"), + ), + "agentchat": ( + "ix_agentchat_updated_id", + ("updated_at", "id"), + ), + "agenttaskrun": ( + "ix_agenttaskrun_status_started_id", + ("status", "started_at", "id"), + ), +} + + +def _index_names(table_name: str) -> set[str]: + """读取实时索引名，兼容 fresh metadata 已创建当前索引的路径。""" + return { + index["name"] + for index in sa.inspect(op.get_bind()).get_indexes(table_name) + } + + +def upgrade() -> None: + """为三张无界增长历史表补充保留期扫描索引。""" + table_names = set(sa.inspect(op.get_bind()).get_table_names()) + for table_name, (index_name, columns) in _CLEANUP_INDEXES.items(): + if table_name not in table_names or index_name in _index_names(table_name): + continue + op.create_index(index_name, table_name, list(columns), unique=False) + + +def downgrade() -> None: + """删除保留期扫描索引，不改动任何历史数据。""" + table_names = set(sa.inspect(op.get_bind()).get_table_names()) + for table_name, (index_name, _columns) in _CLEANUP_INDEXES.items(): + if table_name in table_names and index_name in _index_names(table_name): + op.drop_index(index_name, table_name=table_name) diff --git a/docs/adr/0007-background-action-reliability.md b/docs/adr/0007-background-action-reliability.md index df8794f34..e234557b9 100644 --- a/docs/adr/0007-background-action-reliability.md +++ b/docs/adr/0007-background-action-reliability.md @@ -19,8 +19,9 @@ MoviePilot 保持模块化单体，不把所有后台动作迁到分布式队列 `durable-required` 是目标语义，不代表当前实现已经 durable。ARCH-251 前，Event Registry 中标记该值的 事件仍应在风险报告中说明崩溃窗口。 -截至 2026-08-23，宿主正式装配的 `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted`、 -`SubscribeComplete`、`DownloadAdded`、`TransferComplete`、`TransferFailed` 广播已由业务事务内的 outbox +截至 2026-08-26，宿主正式装配的 `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted`、 +`SubscribeComplete`、`DownloadAdded` 以及媒体、字幕、音频的 `TransferComplete` / `TransferFailed` +广播已由业务事务内的 outbox intent 提供 at-least-once 恢复；订阅完成的历史新增、订阅删除、完成事件和完成统计 intent 同事务提交， 提交后通知/事件/统计仍按原顺序执行，事件与统计失败保持独立 pending。payload 保持插件 dict/对象 ABI， 并增加可选幂等键。下载和整理的 outbox 只保存 @@ -38,8 +39,7 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同 `DownloadFileDeleted`、`DownloadDeleted`。 - 消息/UI：`UserMessage`、`WebhookMessage`、`NoticeMessage`、`MessageAction`。 - 生命周期/诊断：`SystemError`、`ModuleReload`、`ConfigChanged`、`WorkflowExecute`、 - `AgentTokensUsage`、`MetadataScrape`、`SubscribeComplete`、`SubtitleTransferComplete`、 - `SubtitleTransferFailed`、`AudioTransferComplete`、`AudioTransferFailed`。 + `AgentTokensUsage`、`MetadataScrape`。 - 链式扩展：全部 22 个 `ChainEventType`（`PluginDataReset`、`NameRecognize`、 `MusicNameRecognize`、`MediaRecognize`、`MusicMediaRecognize`、`AuthVerification`、 `AuthIntercept`、`CommandRegister`、`TransferRename`、`TransferRenameBuild`、 @@ -50,11 +50,12 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同 ### E2：业务提交后的用户副作用 -- `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted`：订阅业务行 commit 是业务完成点；事件、 +- `SubscribeAdded`、`SubscribeModified`、`SubscribeDeleted`、`SubscribeComplete`：订阅业务行 commit 是业务完成点；事件、 通知和服务端上报必须由同事务 durable intent 驱动。ARCH-251 首选 `SubscribeAdded` pilot。 - `DownloadAdded`：下载器确认接收后，下载历史与事件 intent 已在返回前原子提交；通知和模块后处理只在 commit 后启动，事件由 Outbox 恢复投递。 -- `TransferComplete`、`TransferFailed`：整理步骤本身属于 E3，但向事件消费者发布结果属于 E2。 +- `TransferComplete`、`TransferFailed` 以及对应的字幕、音频完成/失败事件：整理步骤本身属于 E3， + 但历史行提交后向事件消费者发布结果属于 E2，使用同一 JSON 快照与恢复 handler。 ## 非 Event 后台机制映射 @@ -129,6 +130,11 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同 - E0：不重试或仅当前调用内有限重试；队列关停可丢弃，必须记录。 - E1：固定上限或指数退避，下一周期可重建；同一 job key 不并发重叠。 - E2：稳定 idempotency key；原子 claim；指数退避有上限；超过上限进入 dead letter，不无限刷日志。 +- Outbox 终态历史并入统一数据维护任务，受 `DATA_CLEANUP_ENABLE` 总开关控制；`completed` + 与 dead letter 默认分别保留 30/90 天，并可在高级设置中独立调整或设为 `0` 禁用。 + `pending` / `processing` 不参与保留期删除，恰好位于截止边界的记录继续保留。 +- 统一数据维护还覆盖全部具备安全时间边界的宿主追加表；Agent 会话保护任务引用，Agent 运行历史 + 保护运行中与最后一次运行。`transferpending` 与 `plugininstallation` 是恢复状态，不按年龄删除。 - E3：步骤级幂等、lease/heartbeat、重启恢复和人工决策入口；外部不可逆步骤必须记录补偿边界。 关停顺序为停止接收新任务、停止 claim、等待有界 drain、释放资源。超过预算的 E2/E3 任务保持持久 @@ -143,6 +149,6 @@ pending 状态交由下次启动，不以取消异常写成成功。 ## 验证与演进 - Event Registry 的 `delivery` 字段与本 ADR 同步进入 runtime baseline。 -- ARCH-251 已覆盖 Registry 中六种 `durable_required` 事件，并通过 commit 后崩溃、重复 claim、并发 +- ARCH-251 已覆盖 Registry 中十一种 `durable_required` 事件，并通过 commit 后崩溃、重复 claim、并发 claim、JSON 快照恢复和 dead-letter 测试；后续新增 E2 事件必须同时提供业务事务边界和恢复测试。 - ARCH-252 将 Scheduler 的定义、触发和执行状态拆分，但不提升不需要 durable 的 E0 信号。 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 90eaff6ce..ae4bca8b0 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -419,7 +419,13 @@ flowchart LR 和插件到 Model 的依赖，保证提交权不会被底层抢走。 - **Outbox 可靠副作用**：业务行与 durable intent 在同一 Session/UoW 中提交；提交后由 Outbox dispatcher 依据 topic、claim/lease、有限重试和 dead-letter 执行。完成通知、事件和统计 - 的 post-commit 逻辑必须保持幂等，不能用普通线程或 TaskRegistry 代替持久 intent。 + 的 post-commit 逻辑必须保持幂等，不能用普通线程或 TaskRegistry 代替持久 intent。终态历史随统一 + 数据维护任务分批清理，默认成功记录保留 30 天、dead letter 保留 90 天；总开关和两项保留期由 + 高级设置维护，待投递和 lease 中记录不参与清理。 +- **统一历史保留期**：所有可安全按时间回收的追加型数据均受 `DATA_CLEANUP_ENABLE` 控制，包括消息、 + 下载及孤儿文件、站点快照、整理历史、下载失败冷却、订阅历史、Agent 会话、Agent 任务运行和 Outbox + 终态。Agent 会话会保护任务引用，Agent 运行会保护运行中与最后一次运行；`transferpending` 和 + `plugininstallation` 承担恢复语义，禁止按年龄删除。 - 站点、历史、工作流、Agent 会话删除和插件数据重置已经形成同构事务切片；对应 Application Command/Service 持有 UoW，Oper 的 `stage_*` 方法只修改当前会话。插件数据重置从 `startup/initializers/plugins.py` 注入事务能力，插件直接使用 `PluginDataOper` 的旧 ABI 仅作兼容。 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 23bdaa3e5..c8d2bf2b4 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -453,6 +453,11 @@ Durable post-commit side effects have a separate boundary: - The dispatcher claims an intent with a lease, executes the topic handler, and records retry/dead-letter state. Handlers must be idempotent and must not rely on a live request object. +- Terminal history is part of the shared data-maintenance policy and is cleaned + in bounded daily batches only when that policy is enabled. Completed intents + default to 30-day retention and dead letters to 90 days; both values are + user-configurable and `0` disables that status cleanup. Pending or processing + intents must never be removed by retention cleanup. - `app/runtime/tasks.py` is only the in-process TaskRegistry boundary. It owns cancellation and bounded shutdown waiting, but it is not a durable queue and must not replace an Outbox or persistent task table. diff --git a/docs/rules/10-data-and-persistent.md b/docs/rules/10-data-and-persistent.md index d10c3d54f..7f006cc4c 100644 --- a/docs/rules/10-data-and-persistent.md +++ b/docs/rules/10-data-and-persistent.md @@ -140,10 +140,31 @@ startup composition supplies the repository, transaction scope and topic handlers. The dispatcher claims an intent with a lease, executes an idempotent handler, -and records bounded retries or dead-letter state. The `app/runtime/tasks.py` +and records bounded retries or dead-letter state. The shared data-maintenance +policy controls bounded terminal-history cleanup, with user-configurable 30-day +completed and 90-day dead-letter defaults; `0` disables either cleanup. It must +not delete pending or leased processing rows. The `app/runtime/tasks.py` TaskRegistry is only the owner for in-process work and bounded shutdown waiting; it is not a durable queue or a replacement for an Outbox/persistent task table. +All append-only or snapshot history owned by the host must participate in the +shared `DATA_CLEANUP_ENABLE` policy when it has a safe time boundary: + +- `message`, `downloadhistory` and orphaned `downloadfiles`, `siteuserdata`, + `transferhistory`, `downloadfailure`, and `subscribehistory` use their own + user-configurable retention periods. +- `agentchat` removes only expired sessions not referenced by an `agenttask`; + `agenttaskrun` removes only expired terminal runs that are neither running nor + the task's current `last_run_id`. +- `outboxmessage` has separate completed and dead-letter retention periods; + pending and processing intents are recovery state and are never age-deleted. + +`transferpending` and `plugininstallation` are recovery queues/journals rather +than history. Their age is not proof that they are disposable, so generic +retention cleanup must not delete them. Current-state tables keyed by a user, +site, plugin, workflow, passkey, or media-library item are likewise outside +time-based cleanup; their owning mutation lifecycle must replace or delete them. + Run `./.venv/bin/python scripts/architecture/baseline.py --check-host` after persistence changes. A deliberate debt reduction may refresh the low-water mark with `--write-host`; never refresh it to accept newly introduced debt. diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index fbad8b3ec..9b3c32bfb 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6767, - "edge_sha256": "659804b4d1c0f3ff4d96e8c9a059df0afb74c91f368a73188c740558114efa61", + "edge_count": 6773, + "edge_sha256": "3ba25180753ef7e15d9d08b7af25b66191306e8c0b465a27bed5613ed8d30c5d", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -3724,11 +3724,17 @@ "app.db.health -> app.db", "app.db.health -> app.db.session", "app.db.maintenance -> app.db", + "app.db.maintenance -> app.db.base", "app.db.maintenance -> app.db.models", + "app.db.maintenance -> app.db.models.agentchat", + "app.db.maintenance -> app.db.models.agenttask", + "app.db.maintenance -> app.db.models.agenttaskrun", "app.db.maintenance -> app.db.models.downloadfailure", "app.db.maintenance -> app.db.models.downloadhistory", "app.db.maintenance -> app.db.models.message", + "app.db.maintenance -> app.db.models.outbox", "app.db.maintenance -> app.db.models.siteuserdata", + "app.db.maintenance -> app.db.models.subscribehistory", "app.db.maintenance -> app.db.models.transferhistory", "app.db.maintenance -> app.db.uow", "app.db.models -> app.db", diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 4726b03a7..542613f29 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -1514,7 +1514,7 @@ "visibility": "plugin_public" }, "EventType.AudioTransferComplete": { - "delivery": "ephemeral", + "delivery": "durable_required", "error_behavior": "notify", "input_contract": "TransferResultContractData", "legacy_reason": null, @@ -1529,7 +1529,7 @@ "visibility": "plugin_public" }, "EventType.AudioTransferFailed": { - "delivery": "ephemeral", + "delivery": "durable_required", "error_behavior": "notify", "input_contract": "TransferResultContractData", "legacy_reason": null, @@ -1799,7 +1799,7 @@ "visibility": "plugin_public" }, "EventType.SubscribeComplete": { - "delivery": "ephemeral", + "delivery": "durable_required", "error_behavior": "notify", "input_contract": "SubscribeCompleteEventData", "legacy_reason": null, @@ -1844,7 +1844,7 @@ "visibility": "plugin_public" }, "EventType.SubtitleTransferComplete": { - "delivery": "ephemeral", + "delivery": "durable_required", "error_behavior": "notify", "input_contract": "TransferResultContractData", "legacy_reason": null, @@ -1859,7 +1859,7 @@ "visibility": "plugin_public" }, "EventType.SubtitleTransferFailed": { - "delivery": "ephemeral", + "delivery": "durable_required", "error_behavior": "notify", "input_contract": "TransferResultContractData", "legacy_reason": null, @@ -2253,6 +2253,10 @@ { "caller": "app.chain.transfer", "count": 1 + }, + { + "caller": "app.startup.initializers.modules", + "count": 1 } ] }, @@ -2262,6 +2266,10 @@ { "caller": "app.chain.transfer", "count": 1 + }, + { + "caller": "app.startup.initializers.modules", + "count": 1 } ] }, @@ -2521,6 +2529,10 @@ { "caller": "app.chain.transfer", "count": 1 + }, + { + "caller": "app.startup.initializers.modules", + "count": 1 } ] }, @@ -2530,6 +2542,10 @@ { "caller": "app.chain.transfer", "count": 1 + }, + { + "caller": "app.startup.initializers.modules", + "count": 1 } ] }, @@ -2555,7 +2571,7 @@ "producers": [ { "caller": "app.chain.transfer", - "count": 2 + "count": 1 }, { "caller": "app.startup.initializers.modules", @@ -2568,7 +2584,7 @@ "producers": [ { "caller": "app.chain.transfer", - "count": 2 + "count": 1 }, { "caller": "app.startup.initializers.modules", @@ -2609,7 +2625,7 @@ ] } }, - "producer_count": 78 + "producer_count": 80 }, "module_method_specs": { "anilist_credits": { diff --git a/tests/test_agent_task_run_migration.py b/tests/test_agent_task_run_migration.py index 7b1046051..434dc97e0 100644 --- a/tests/test_agent_task_run_migration.py +++ b/tests/test_agent_task_run_migration.py @@ -9,7 +9,6 @@ from sqlalchemy.schema import CreateTable from app.db.models.agenttask import AgentTask from app.db.models.agenttaskrun import AgentTaskRun - MIGRATION = "database.versions.f4c8d2a7b1e6_3_0_6" @@ -84,7 +83,18 @@ def test_agent_task_run_migration_accepts_fresh_current_schema(monkeypatch) -> N assert { column["name"] for column in inspector.get_columns("agenttaskrun") } == {column.name for column in AgentTaskRun.__table__.columns} - assert len(inspector.get_indexes("agenttaskrun")) == 2 + actual_indexes = { + index["name"]: (tuple(index["column_names"]), index["unique"]) + for index in inspector.get_indexes("agenttaskrun") + } + expected_indexes = { + index.name: ( + tuple(column.name for column in index.columns), + int(index.unique), + ) + for index in AgentTaskRun.__table__.indexes + } + assert actual_indexes == expected_indexes def test_agent_task_run_migration_matches_postgresql_identity() -> None: diff --git a/tests/test_cleanup_index_migration.py b/tests/test_cleanup_index_migration.py new file mode 100644 index 000000000..b95892653 --- /dev/null +++ b/tests/test_cleanup_index_migration.py @@ -0,0 +1,75 @@ +"""历史表清理索引迁移测试。""" + +import importlib + +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + + +MIGRATION_MODULE = "database.versions.a6c8e2f4b1d3_3_0_11" +INDEXES = { + "subscribehistory": ( + "ix_subscribehistory_date_id", + ("date", "id"), + ), + "agentchat": ( + "ix_agentchat_updated_id", + ("updated_at", "id"), + ), + "agenttaskrun": ( + "ix_agenttaskrun_status_started_id", + ("status", "started_at", "id"), + ), +} + + +def _index_columns(connection, table_name: str) -> dict[str, tuple[str, ...]]: + """返回测试表的索引字段签名。""" + return { + index["name"]: tuple(index.get("column_names") or ()) + for index in sa.inspect(connection).get_indexes(table_name) + } + + +def test_cleanup_index_migration_is_idempotent_and_reversible(monkeypatch) -> None: + """升级可重复执行并创建准确索引，降级只移除新增索引。""" + engine = sa.create_engine("sqlite://") + metadata = sa.MetaData() + sa.Table( + "subscribehistory", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("date", sa.String()), + ) + sa.Table( + "agentchat", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("updated_at", sa.String()), + ) + sa.Table( + "agenttaskrun", + metadata, + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("status", sa.String()), + sa.Column("started_at", sa.String()), + ) + + with engine.begin() as connection: + metadata.create_all(connection) + migration = importlib.import_module(MIGRATION_MODULE) + monkeypatch.setattr( + migration, + "op", + Operations(MigrationContext.configure(connection)), + ) + + migration.upgrade() + migration.upgrade() + for table_name, (index_name, columns) in INDEXES.items(): + assert _index_columns(connection, table_name)[index_name] == columns + + migration.downgrade() + for table_name, (index_name, _columns) in INDEXES.items(): + assert index_name not in _index_columns(connection, table_name) diff --git a/tests/test_data_cleanup_growth_tables.py b/tests/test_data_cleanup_growth_tables.py new file mode 100644 index 000000000..112e13a3e --- /dev/null +++ b/tests/test_data_cleanup_growth_tables.py @@ -0,0 +1,156 @@ +"""统一数据维护对追加型历史表的安全清理测试。""" + +from datetime import datetime + +from sqlalchemy import create_engine, select +from sqlalchemy.orm import sessionmaker + +from app.application.maintenance import CleanupPolicy, DataCleanupService +from app.db.base import Base +from app.db.maintenance import DatabaseCleanupRepository +from app.db.models.agentchat import AgentChat +from app.db.models.agenttask import AgentTask +from app.db.models.agenttaskrun import AgentTaskRun +from app.db.models.subscribehistory import SubscribeHistory + + +def _cleanup_policy() -> CleanupPolicy: + """只启用本组新增历史表的 30 天保留期。""" + return CleanupPolicy( + enabled=True, + message_days=0, + download_history_days=0, + site_userdata_days=0, + transfer_history_days=0, + download_failure_days=0, + subscribe_history_days=30, + agent_chat_days=30, + agent_task_run_days=30, + outbox_completed_days=0, + outbox_dead_days=0, + ) + + +def test_growth_table_cleanup_preserves_live_agent_recovery_state() -> None: + """旧历史可回收，但任务引用会话、最后运行和运行中记录必须保留。""" + engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(engine) + factory = sessionmaker(bind=engine) + old_time = "2026-06-01 12:00:00" + recent_time = "2026-08-20 12:00:00" + + with factory() as session: + session.add_all([ + SubscribeHistory(name="old-subscribe", date=old_time), + SubscribeHistory(name="recent-subscribe", date=recent_time), + AgentChat( + session_id="old-unreferenced", + title="old-unreferenced", + created_at=old_time, + updated_at=old_time, + ), + AgentChat( + session_id="task-context", + title="task-context", + created_at=old_time, + updated_at=old_time, + ), + AgentChat( + session_id="recent-chat", + title="recent-chat", + created_at=recent_time, + updated_at=recent_time, + ), + ]) + task = AgentTask( + name="cleanup-protected-task", + content="test", + trigger_type="cron", + enabled=True, + user_id="1", + session_id="task-context", + last_status="success", + last_run_id="latest-run", + run_count=2, + created_at=old_time, + updated_at=old_time, + ) + session.add(task) + session.flush() + session.add_all([ + AgentTaskRun( + run_id="expired-run", + task_id=task.id, + trigger_source="scheduled", + name=task.name, + content=task.content, + trigger_type=task.trigger_type, + user_id=task.user_id, + session_id=task.session_id, + status="success", + started_at=old_time, + finished_at=old_time, + ), + AgentTaskRun( + run_id="latest-run", + task_id=task.id, + trigger_source="scheduled", + name=task.name, + content=task.content, + trigger_type=task.trigger_type, + user_id=task.user_id, + session_id=task.session_id, + status="success", + started_at=old_time, + finished_at=old_time, + ), + AgentTaskRun( + run_id="running-run", + task_id=task.id, + trigger_source="manual", + name=task.name, + content=task.content, + trigger_type=task.trigger_type, + user_id=task.user_id, + session_id=task.session_id, + status="running", + started_at=old_time, + ), + AgentTaskRun( + run_id="recent-run", + task_id=task.id, + trigger_source="manual", + name=task.name, + content=task.content, + trigger_type=task.trigger_type, + user_id=task.user_id, + session_id=task.session_id, + status="failed", + started_at=recent_time, + finished_at=recent_time, + ), + ]) + session.commit() + + report = DataCleanupService( + repository=DatabaseCleanupRepository(session_factory=factory), + policy_reader=_cleanup_policy, + clock=lambda: datetime(2026, 8, 26, 12, 0, 0), + ).execute(batch_size=1) + + assert report["tables"]["subscribehistory"]["deleted"] == 1 + assert report["tables"]["agentchat"]["deleted"] == 1 + assert report["tables"]["agenttaskrun"]["deleted"] == 1 + with factory() as session: + assert set(session.execute(select(SubscribeHistory.name)).scalars()) == { + "recent-subscribe" + } + assert set(session.execute(select(AgentChat.session_id)).scalars()) == { + "task-context", + "recent-chat", + } + assert set(session.execute(select(AgentTaskRun.run_id)).scalars()) == { + "latest-run", + "running-run", + "recent-run", + } diff --git a/tests/test_data_cleanup_service.py b/tests/test_data_cleanup_service.py index adfda6f9f..a505ec16a 100644 --- a/tests/test_data_cleanup_service.py +++ b/tests/test_data_cleanup_service.py @@ -70,6 +70,26 @@ class FakeCleanupRepository: """模拟下载失败记录删除。""" return self._delete("downloadfailure") + def delete_subscribe_history(self, db, cutoff: str, limit: int) -> int: + """模拟订阅历史删除。""" + return self._delete("subscribehistory") + + def delete_agent_chats(self, db, cutoff: str, limit: int) -> int: + """模拟 Agent 会话删除。""" + return self._delete("agentchat") + + def delete_agent_task_runs(self, db, cutoff: str, limit: int) -> int: + """模拟 Agent 运行历史删除。""" + return self._delete("agenttaskrun") + + def delete_outbox_completed(self, db, cutoff: str, limit: int) -> int: + """模拟 Outbox 已完成记录删除。""" + return self._delete("outbox_completed") + + def delete_outbox_dead(self, db, cutoff: str, limit: int) -> int: + """模拟 Outbox 死信记录删除。""" + return self._delete("outbox_dead") + def _policy(**overrides) -> CleanupPolicy: """构造所有表默认启用的测试策略。""" @@ -80,6 +100,11 @@ def _policy(**overrides) -> CleanupPolicy: "site_userdata_days": 1, "transfer_history_days": 1, "download_failure_days": 1, + "subscribe_history_days": 1, + "agent_chat_days": 1, + "agent_task_run_days": 1, + "outbox_completed_days": 1, + "outbox_dead_days": 1, } values.update(overrides) return CleanupPolicy(**values) @@ -112,6 +137,11 @@ def test_cleanup_service_owns_batching_report_and_progress() -> None: "siteuserdata", "transferhistory", "downloadfailure", + "subscribehistory", + "agentchat", + "agenttaskrun", + "outbox_completed", + "outbox_dead", ] assert progress.call_args.kwargs["value"] == 100 @@ -128,7 +158,7 @@ def test_cleanup_service_finishes_other_tables_before_raising_partial_failure() with pytest.raises(RuntimeError, match="downloadhistory: boom"): service.execute(batch_size=2) - assert repository.calls[-1] == "downloadfailure" + assert repository.calls[-1] == "outbox_dead" assert repository.rollbacks == 1 diff --git a/tests/test_event_contracts.py b/tests/test_event_contracts.py index 34ea1c659..a5d9b08b6 100644 --- a/tests/test_event_contracts.py +++ b/tests/test_event_contracts.py @@ -89,6 +89,7 @@ def test_selected_user_side_effects_are_marked_durable_required() -> None: } assert set(DURABLE_EVENT_TOPICS) == durable_events + assert len(durable_events) == 11 assert len(set(DURABLE_EVENT_TOPICS.values())) == len(durable_events) diff --git a/tests/test_outbox.py b/tests/test_outbox.py index 12f2770e7..9db9965fa 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -7,10 +7,12 @@ import pytest from sqlalchemy import create_engine, select from sqlalchemy.orm import sessionmaker +from app.application.maintenance import CleanupPolicy, DataCleanupService from app.application.outbox import ClaimedOutboxMessage, OutboxDispatcher, OutboxIntent from app.application.subscription.write import CreateSubscriptionCommand from app.db.adapters.outbox import SqlAlchemyOutboxRepository from app.db.base import Base +from app.db.maintenance import DatabaseCleanupRepository from app.db.models.outbox import OutboxMessage @@ -171,3 +173,97 @@ def test_sync_outbox_claim_is_exclusive_for_event_key() -> None: assert message.status == "processing" assert message.attempt == 1 assert message.lease_until == lease_until.isoformat() + + +def test_outbox_cleanup_removes_only_expired_terminal_history_in_batches() -> None: + """清理只删除超过各自保留期的终态记录，并按批次持续收口。""" + engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(engine) + factory = sessionmaker(bind=engine) + now = datetime(2026, 8, 26, tzinfo=timezone.utc) + + def message( + event_key: str, + status: str, + *, + completed_at: datetime | None = None, + next_retry_at: datetime | None = None, + ) -> OutboxMessage: + """构造指定终态时间的最小 Outbox 测试记录。""" + return OutboxMessage( + event_key=event_key, + topic="test", + payload_version=1, + payload={}, + status=status, + attempt=1, + next_retry_at=(next_retry_at or now).isoformat(), + created_at=(now - timedelta(days=120)).isoformat(), + completed_at=completed_at.isoformat() if completed_at else None, + ) + + with factory() as session: + session.add_all([ + message( + "completed-expired-1", + "completed", + completed_at=now - timedelta(days=31), + ), + message( + "completed-expired-2", + "completed", + completed_at=now - timedelta(days=40), + ), + message( + "completed-boundary", + "completed", + completed_at=now - timedelta(days=30), + ), + message( + "dead-expired", + "dead", + next_retry_at=now - timedelta(days=91), + ), + message( + "dead-recent", + "dead", + next_retry_at=now - timedelta(days=20), + ), + message("pending-old", "pending"), + message("processing-old", "processing"), + ]) + session.commit() + + cleanup = DataCleanupService( + repository=DatabaseCleanupRepository(session_factory=factory), + policy_reader=lambda: CleanupPolicy( + enabled=True, + message_days=0, + download_history_days=0, + site_userdata_days=0, + transfer_history_days=0, + download_failure_days=0, + subscribe_history_days=0, + agent_chat_days=0, + agent_task_run_days=0, + outbox_completed_days=30, + outbox_dead_days=90, + ), + clock=lambda: now, + ) + report = cleanup.execute(batch_size=2) + + assert report["tables"]["outbox_completed"]["deleted"] == 2 + assert report["tables"]["outbox_dead"]["deleted"] == 1 + assert report["total_deleted"] == 3 + + with factory() as session: + remaining = set( + session.execute(select(OutboxMessage.event_key)).scalars().all() + ) + assert remaining == { + "completed-boundary", + "dead-recent", + "pending-old", + "processing-old", + } diff --git a/tests/test_transfer_overwrite_declined.py b/tests/test_transfer_overwrite_declined.py index 553e6c5c2..519ff0f79 100644 --- a/tests/test_transfer_overwrite_declined.py +++ b/tests/test_transfer_overwrite_declined.py @@ -9,10 +9,17 @@ __is_overwrite_declined 用于识别这一场景，__default_callback 失败分 from types import SimpleNamespace from unittest.mock import MagicMock, patch +import pytest + from app.chain.transfer import TransferChain from app.schemas.transfer import TransferInfo from app.schemas.types import EventType -from tests.test_transfer_job_manager import FakeMedia, make_task, make_transfer_chain +from tests.test_transfer_job_manager import ( + FakeMedia, + make_fileitem, + make_task, + make_transfer_chain, +) def make_history_oper(history=None, success_history=None, raise_on_query: bool = False): @@ -271,3 +278,49 @@ def test_default_callback_delegates_primary_failure_to_durable_writer(): event_type, event_payload = chain.eventmanager.send_event.call_args.args assert event_type == EventType.TransferFailed assert event_payload["idempotency_key"] == "transfer.failed:1:v1" + + +@pytest.mark.parametrize( + ("path", "success", "expected_topic", "expected_event"), + [ + ( + "/downloads/demo.srt", + True, + "transfer.subtitle.completed", + EventType.SubtitleTransferComplete, + ), + ( + "/downloads/demo.srt", + False, + "transfer.subtitle.failed", + EventType.SubtitleTransferFailed, + ), + ( + "/downloads/demo.flac", + True, + "transfer.audio.completed", + EventType.AudioTransferComplete, + ), + ( + "/downloads/demo.flac", + False, + "transfer.audio.failed", + EventType.AudioTransferFailed, + ), + ], +) +def test_subtitle_and_audio_results_use_durable_topics( + path: str, + success: bool, + expected_topic: str, + expected_event: EventType, +) -> None: + """字幕与音频结果必须和主要媒体结果共用 durable writer 分类。""" + chain = make_transfer_chain() + task = make_task(1) + task.fileitem = make_fileitem(path) + + assert chain._durable_transfer_event(task, success=success) == ( + expected_topic, + expected_event, + )