From 74f6bb89b5902b8b8d2b6856982a18cae201a09f Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 1 Sep 2026 13:15:42 +0800 Subject: [PATCH] perf(subscribe): bound candidate routing work --- app/application/subscription/candidates.py | 8 ++- ...iepilot-subscription-governance-roadmap.md | 17 ++++-- tests/test_subscription_candidate_batch.py | 58 +++++++++++++++++++ 3 files changed, 76 insertions(+), 7 deletions(-) diff --git a/app/application/subscription/candidates.py b/app/application/subscription/candidates.py index 8774f5afe..814cfd014 100644 --- a/app/application/subscription/candidates.py +++ b/app/application/subscription/candidates.py @@ -77,6 +77,7 @@ class CandidateIndex: self._unknown: set[int] = set() self._reconcilable: set[int] = set() self._explicit_identity: dict[int, tuple[str, str]] = {} + self.last_examined_count = 0 for domain, contexts in candidates.items(): for context in contexts: position = len(self._ordered) @@ -136,9 +137,10 @@ class CandidateIndex: ) routed: CandidateGroups = {} - for position, (domain, context) in enumerate(self._ordered): - if position not in positions: - continue + self.last_examined_count = 0 + for position in sorted(positions): + domain, context = self._ordered[position] + self.last_examined_count += 1 if domains and domain not in domains: continue torrent_info = getattr(context, "torrent_info", None) diff --git a/docs/refactor/moviepilot-subscription-governance-roadmap.md b/docs/refactor/moviepilot-subscription-governance-roadmap.md index aa9dc78d5..9a0c26095 100644 --- a/docs/refactor/moviepilot-subscription-governance-roadmap.md +++ b/docs/refactor/moviepilot-subscription-governance-roadmap.md @@ -1,7 +1,7 @@ # MoviePilot 订阅执行治理 > 状态：`active（2026-09-01 已由 MoviePilot v3 接管）` -> 当前叶：`SUB-GOV-001E（条件评估）` +> 当前叶：`SUB-GOV-002A` > 迁移来源：`V3-RDY-009A`、`V3-RDY-009A1`、`V3-RDY-009A1B`、 > `V3-RDY-009A1C`、`V3-RDY-009A1D` > 适用范围：V3 订阅刷新、匹配、兜底搜索、下载提交与用户状态闭环 @@ -231,15 +231,15 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 | `SUB-GOV-001B` | 区分完整缓存与本轮 delta，建立无损候选索引；先证明命中集合与基线完全一致 | 001A | `completed（2026-09-01）` | | `SUB-GOV-001C` | 将资源匹配与完成对账拆开；只有受候选影响的订阅进入匹配，完成对账独立保持新鲜语义 | 001B | `completed（2026-09-01）` | | `SUB-GOV-001D` | 引入单轮新鲜事实租约，评估稳定元数据复用、轻量季集查询和单媒体服务器事实合并 | 001C | `completed（2026-09-01）` | -| `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `in_progress（条件评估）` | -| `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `pending` | +| `SUB-GOV-001E` | 若 001B–D 后仍有可重复等待热点，再通过隔离压测评估 2/4 worker 的有界准备；提交保持串行 | 001D | `not_activated（2026-09-01）` | +| `SUB-GOV-002A` | 将 24 小时兜底搜索移出 Match/提交长锁，建立批次、订阅 single-flight、可取消等待和恢复游标 | 001C | `in_progress` | | `SUB-GOV-002B` | 为兜底搜索建立每站点并发、间隔和错误冷却预算；RSS/Spider 保持基线压力 | 002A | `pending` | | `SUB-GOV-002C` | 联合运行日常 Match、手工搜索和兜底搜索，验证公平性、锁等待、站点压力和失败恢复 | 001D, 002B | `pending` | | `SUB-GOV-003A` | 建立跨入口的订阅级下载幂等、下载器不确定终态和取消补偿 | 001C, 002A | `pending` | | `SUB-GOV-003B` | 后端提供订阅及批次业务状态，前端展示排队、匹配、搜索、提交、完成、失败和取消 | 002A, 003A | `pending` | | `SUB-GOV-004` | 多条订阅记录指向同一媒体时的跨记录季集去重和产品规则 | 003A | `pending（最低优先级）` | -当前只激活 `SUB-GOV-001E` 的条件评估。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B +当前只激活 `SUB-GOV-002A`。001B–D 是日常 Match 主线；002A–C 是 24 小时兜底搜索主线；003A/B 只有在前两条链路的身份和终态稳定后实施。每个叶子完成验收并更新本表后，才激活下一个满足依赖的叶子。 ### 6.1 SUB-GOV-001A 验收证据 @@ -290,6 +290,15 @@ RSS/Spider 保持现有逐站点串行刷新，不通过并发化换取几秒收 - 验证：订阅与候选回放 `118 passed`，事实租约/治理回放/架构组合 `108 passed`，错误级源代码 Pylint 为 0， Host 架构基线与 `git diff --check` 通过。 +### 6.5 SUB-GOV-001E 条件评估 + +- 修正 `CandidateIndex` 的末端遍历：路由直接按选中位置读取，不再先枚举完整候选后判断位置集合； +- 受控样本固定为 `200` 条订阅、`20` 个站点、`1000` 条明确身份候选、`100` 个唯一媒体；基线组合量为 + `200000`，新索引实际检查并路由 `2000` 条，单轮事实租约执行 `100` 次加载并命中 `100` 次； +- 该样本证明本地 `S * C` 热点和同媒体重复识别已消除，但没有提供 001B–D 后相互独立的 provider/媒体服务器 + 等待仍主导批次的同 revision 证据；因此不激活 2/4 worker，不引入共享客户端线程安全与提交排序风险； +- 后续若受控生产形态证据再次显示准备等待占主导，可从本条件叶重新开启，但不得绕过 SiteBudget 和串行提交。 + ## 7. 上线前验证与验收 ### 7.1 场景 diff --git a/tests/test_subscription_candidate_batch.py b/tests/test_subscription_candidate_batch.py index 3b3dee143..dee071249 100644 --- a/tests/test_subscription_candidate_batch.py +++ b/tests/test_subscription_candidate_batch.py @@ -5,6 +5,7 @@ from unittest.mock import Mock, patch from app.application.subscription.candidates import CandidateBatch, CandidateIndex from app.application.subscription.contract import SubscriptionSnapshot +from app.application.subscription.facts import FreshFactLease from app.chain.torrents import TorrentsChain from app.domain.context import Context, MediaInfo, TorrentInfo from app.domain.metainfo import MetaInfo @@ -179,3 +180,60 @@ def test_candidate_index_custom_words_preserve_complete_candidate_set(): ) assert routed == candidates + + +def test_candidate_index_target_scale_avoids_subscription_candidate_product(): + """200 条订阅与 1000 候选只检查命中身份位置，并按唯一媒体合并事实读取。""" + media_count = 100 + subscription_count = 200 + site_count = 20 + candidates = {f"site-{site_index}.example": [] for site_index in range(site_count)} + for candidate_index in range(1000): + media_id = str(1000 + candidate_index % media_count) + domain = f"site-{candidate_index % site_count}.example" + candidates[domain].append( + _context( + f"目标剧集 {media_id} S01E{candidate_index + 1:04d}", + media_id=media_id, + meta_media_id=media_id, + enclosure=f"https://{domain}/{candidate_index}", + ) + ) + subscribes = [ + _subscribe( + id=index + 1, + name=f"目标剧集 {index % media_count}", + media_id=str(1000 + index % media_count), + ) + for index in range(subscription_count) + ] + + candidate_index = CandidateIndex(candidates) + examined = 0 + routed = 0 + for subscribe in subscribes: + groups = candidate_index.route_for_match(subscribe) + examined += candidate_index.last_examined_count + routed += CandidateBatch.count(groups) + + lease = FreshFactLease() + fact_loads = [] + for subscribe in subscribes: + lease.get_or_load( + subscribe, + lambda subscribe=subscribe: fact_loads.append(subscribe.media_id) + or MediaInfo( + media_source=MediaSource.TMDB, + media_id=subscribe.media_id, + type=MediaType.TV, + season=1, + ), + ) + + assert len(candidates) == site_count + assert CandidateBatch.count(candidates) == 1000 + assert examined == routed == 2000 + assert examined < subscription_count * CandidateBatch.count(candidates) // 50 + assert len(fact_loads) == media_count + assert lease.loads == media_count + assert lease.hits == subscription_count - media_count