fix(subscribe): tolerate partial site search failures

This commit is contained in:
jxxghp
2026-09-05 20:03:09 +08:00
parent 1e7fb6397b
commit b4be185e44
10 changed files with 220 additions and 21 deletions
+6 -2
View File
@@ -66,6 +66,10 @@ class SubscriptionExecutionAdmission:
return self._clock() >= lease.expires_at return self._clock() >= lease.expires_at
class SubscriptionSiteSearchFailed(RuntimeError):
"""表示本轮没有搜索源成功完成,无需输出内部异常堆栈。"""
@dataclass(slots=True) @dataclass(slots=True)
class SubscriptionExecutionContext: class SubscriptionExecutionContext:
"""一次订阅执行的显式取消、阶段和副作用边界。""" """一次订阅执行的显式取消、阶段和副作用边界。"""
@@ -101,9 +105,9 @@ class SubscriptionExecutionContext:
def raise_subscription_site_budget_failures(failures: tuple[str, ...]) -> None: def raise_subscription_site_budget_failures(failures: tuple[str, ...]) -> None:
"""成功站点结果完成处理后暴露其余站点聚合失败。""" """所有实际搜索源均未成功时暴露站点聚合失败。"""
if failures: if failures:
raise RuntimeError("".join(failures)) raise SubscriptionSiteSearchFailed("".join(failures))
def raise_subscription_site_budget_deferral( def raise_subscription_site_budget_deferral(
+1
View File
@@ -97,6 +97,7 @@ if TYPE_CHECKING:
matches_music_resource: Callable[..., Any] matches_music_resource: Callable[..., Any]
music_site_keywords: Callable[..., Any] music_site_keywords: Callable[..., Any]
process: Callable[..., Any] process: Callable[..., Any]
record_subscription_site_budget_success: Callable[..., Any]
record_subscription_site_budget_failure: Callable[..., Any] record_subscription_site_budget_failure: Callable[..., Any]
consume_subscription_site_budget_failures: Callable[..., Any] consume_subscription_site_budget_failures: Callable[..., Any]
record_subscription_site_budget_deferred: Callable[..., Any] record_subscription_site_budget_deferred: Callable[..., Any]
+19 -4
View File
@@ -52,26 +52,41 @@ class SearchChain(ChainBase):
"""仅为订阅搜索启用或清除站点预算,不影响其它搜索入口。""" """仅为订阅搜索启用或清除站点预算,不影响其它搜索入口。"""
self._subscription_site_budget = budget self._subscription_site_budget = budget
self._subscription_site_budget_failures: list[str] = [] self._subscription_site_budget_failures: list[str] = []
self._subscription_site_budget_successes: set[int] = set()
self._subscription_site_budget_deferrals: list[SubscriptionSiteBudgetDeferral] = [] self._subscription_site_budget_deferrals: list[SubscriptionSiteBudgetDeferral] = []
self._subscription_site_budget_failure_lock = threading.Lock() self._subscription_site_budget_failure_lock = threading.Lock()
def record_subscription_site_budget_success(self, site_id: int) -> None:
"""线程安全地记录一个已正常完成真实请求的站点。"""
lock = getattr(self, "_subscription_site_budget_failure_lock", None)
if lock is None:
return
with lock:
self._subscription_site_budget_successes.add(site_id)
def record_subscription_site_budget_failure(self, error: str) -> None: def record_subscription_site_budget_failure(self, error: str) -> None:
"""线程安全地记录一个站点执行失败,供订阅任务暴露聚合失败""" """线程安全地记录一个站点执行失败,供订阅任务判断轮次终态"""
lock = getattr(self, "_subscription_site_budget_failure_lock", None) lock = getattr(self, "_subscription_site_budget_failure_lock", None)
if lock is None: if lock is None:
return return
with lock: with lock:
self._subscription_site_budget_failures.append(error) self._subscription_site_budget_failures.append(error)
def consume_subscription_site_budget_failures(self) -> tuple[str, ...]: def consume_subscription_site_budget_failures(
"""读取并清空当前订阅搜索积累的站点预算失败。""" self,
*,
has_results: bool = False,
) -> tuple[str, ...]:
"""仅在没有成功搜索源时返回并清空本轮站点聚合失败。"""
lock = getattr(self, "_subscription_site_budget_failure_lock", None) lock = getattr(self, "_subscription_site_budget_failure_lock", None)
if lock is None: if lock is None:
return () return ()
with lock: with lock:
failures = tuple(self._subscription_site_budget_failures) failures = tuple(self._subscription_site_budget_failures)
has_successful_site = bool(self._subscription_site_budget_successes)
self._subscription_site_budget_failures.clear() self._subscription_site_budget_failures.clear()
return failures self._subscription_site_budget_successes.clear()
return () if has_results or has_successful_site else failures
def record_subscription_site_budget_deferred( def record_subscription_site_budget_deferred(
self, self,
+8 -5
View File
@@ -243,11 +243,14 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
raise raise
else: else:
budget.record_request(site_id, len(result or [])) budget.record_request(site_id, len(result or []))
if observation.attempted and observation.outcome not in {"success", "skipped"}: if observation.attempted:
failure = observation.error or observation.outcome if observation.outcome == "success":
self.record_subscription_site_budget_failure( self.record_subscription_site_budget_success(site_id)
f"站点 {site.get('name') or site_id} 搜索失败:{failure}" elif observation.outcome != "skipped":
) failure = observation.error or observation.outcome
message = f"站点 {site.get('name') or site_id} 搜索失败:{failure}"
self.record_subscription_site_budget_failure(message)
logger.warning(message)
return result return result
finally: finally:
try: try:
+8 -2
View File
@@ -19,6 +19,7 @@ from app.application.subscription.execution import (
SearchTaskSnapshot, SearchTaskSnapshot,
SubscriptionExecutionContext, SubscriptionExecutionContext,
SubscriptionSearchRepository, SubscriptionSearchRepository,
SubscriptionSiteSearchFailed,
raise_subscription_site_budget_deferral, raise_subscription_site_budget_deferral,
raise_subscription_site_budget_failures, raise_subscription_site_budget_failures,
) )
@@ -300,7 +301,10 @@ class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
except Exception as err: except Exception as err:
outcome = "failed" outcome = "failed"
reason = "error" reason = "error"
logger.error(f"订阅 {subscribe.name} 搜索失败:{str(err)}", exc_info=True) logger.error(
f"订阅 {subscribe.name} 搜索失败:{str(err)}",
exc_info=not isinstance(err, SubscriptionSiteSearchFailed),
)
finally: finally:
try: try:
if current and current.state == "N": if current and current.state == "N":
@@ -730,7 +734,9 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
custom_words=subscribe.custom_words.split("\n") if subscribe.custom_words else None, custom_words=subscribe.custom_words.split("\n") if subscribe.custom_words else None,
filter_params=self.get_params(subscribe), filter_params=self.get_params(subscribe),
) )
site_budget_failures = searchchain.consume_subscription_site_budget_failures() site_budget_failures = searchchain.consume_subscription_site_budget_failures(
has_results=bool(contexts),
)
site_budget_deferrals = searchchain.consume_subscription_site_budget_deferrals() site_budget_deferrals = searchchain.consume_subscription_site_budget_deferrals()
_ensure_execution_active(execution_context) _ensure_execution_active(execution_context)
if not contexts: if not contexts:
+5 -1
View File
@@ -15,6 +15,7 @@ from app.application.subscription.execution import (
SubscriptionExecutionContext, SubscriptionExecutionContext,
SubscriptionExecutionLease, SubscriptionExecutionLease,
SubscriptionSearchRepository, SubscriptionSearchRepository,
SubscriptionSiteSearchFailed,
handle_subscription_search_deferred, handle_subscription_search_deferred,
) )
from app.application.subscription.observability import ( from app.application.subscription.observability import (
@@ -239,7 +240,10 @@ class SubscriptionSearchTaskRunner:
self.summary.record, self.summary.record,
) )
except Exception as err: except Exception as err:
logger.error(f"订阅《{subscribe.name}》搜索失败:{str(err)}", exc_info=True) logger.error(
f"订阅《{subscribe.name}》搜索失败:{str(err)}",
exc_info=not isinstance(err, SubscriptionSiteSearchFailed),
)
self.queue.finish_task( self.queue.finish_task(
task_id=task_id, task_id=task_id,
lease_token=lease_token, lease_token=lease_token,
+13 -2
View File
@@ -49,6 +49,8 @@ _SPECIALIZED_SEARCH_ARGUMENTS = {
"RousiPro": ("keyword", "mtype", "cat", "page"), "RousiPro": ("keyword", "mtype", "cat", "page"),
} }
_UNKNOWN_SEARCH_FAILURE_MESSAGE = "站点请求或页面解析失败"
@dataclass(frozen=True) @dataclass(frozen=True)
class _IndexerSearchRequest: class _IndexerSearchRequest:
@@ -85,6 +87,15 @@ def _classify_search_failure(error: Optional[Exception]) -> str:
return "error" return "error"
def _search_failure_message(outcome: _IndexerSearchOutcome) -> Optional[str]:
"""返回供站点预算持久化和任务聚合展示的失败原因。"""
if outcome.error is not None:
return str(outcome.error)
if outcome.error_flag:
return _UNKNOWN_SEARCH_FAILURE_MESSAGE
return None
class IndexerModule(_ModuleBase): class IndexerModule(_ModuleBase):
""" """
索引模块 索引模块
@@ -450,7 +461,7 @@ class IndexerModule(_ModuleBase):
if outcome.error_flag or outcome.error is not None if outcome.error_flag or outcome.error is not None
else "success" else "success"
), ),
error=str(outcome.error) if outcome.error is not None else None, error=_search_failure_message(outcome),
) )
# 返回结果 # 返回结果
@@ -557,7 +568,7 @@ class IndexerModule(_ModuleBase):
if outcome.error_flag or outcome.error is not None if outcome.error_flag or outcome.error is not None
else "success" else "success"
), ),
error=str(outcome.error) if outcome.error is not None else None, error=_search_failure_message(outcome),
) )
# 返回结果 # 返回结果
@@ -329,8 +329,8 @@ RSS/Spider 保持现有逐站点串行刷新,不通过并发化换取几秒收
但进入 Search provider 后仍使用完全相同的站点租约、间隔和冷却,不能旁路; 但进入 Search provider 后仍使用完全相同的站点租约、间隔和冷却,不能旁路;
- 站点预算只包装订阅队列的同步搜索实例,不改变 RSS/Spider 刷新入口和普通手工资源搜索;索引器列表返回 - 站点预算只包装订阅队列的同步搜索实例,不改变 RSS/Spider 刷新入口和普通手工资源搜索;索引器列表返回
合同、同步/异步结果及既有站点健康统计语义保持不变; 合同、同步/异步结果及既有站点健康统计语义保持不变;
- 等待每秒检查批次取消,超过 5 秒短窗口的冷却或在途站点不阻塞整批:该站点本轮返回空页,其他站点继续 - 等待每秒检查批次取消,超过 5 秒短窗口的冷却或在途站点不阻塞整批:该站点本轮返回空页,其他站点继续
订阅在处理成功站点结果后以聚合失败收口,避免慢站点拖死后续订阅 部分站点失败只记录告警与冷却,仅在没有站点或其他搜索源成功完成时以聚合失败收口
- 验证:站点预算/队列/编排/provider 专项 `38 passed`,搜索和索引器兼容 `91 passed`,迁移与声明式模型 - 验证:站点预算/队列/编排/provider 专项 `38 passed`,搜索和索引器兼容 `91 passed`,迁移与声明式模型
`37 passed, 1 skipped`,架构合同组合 `149 passed`,错误级 Pylint 为 0Alembic 唯一 head 为 `37 passed, 1 skipped`,架构合同组合 `149 passed`,错误级 Pylint 为 0Alembic 唯一 head 为
`d2a7c5e9f1b4`Host 架构基线与 `git diff --check` 通过。 `d2a7c5e9f1b4`Host 架构基线与 `git diff --check` 通过。
@@ -339,8 +339,8 @@ RSS/Spider 保持现有逐站点串行刷新,不通过并发化换取几秒收
- 队列认领增加 15 分钟持久老化窗口:新手工任务继续优先,但等待超过窗口的 fallback 任务先于后续高优先级 - 队列认领增加 15 分钟持久老化窗口:新手工任务继续优先,但等待超过窗口的 fallback 任务先于后续高优先级
新任务执行,防止持续手工流量造成后台饥饿; 新任务执行,防止持续手工流量造成后台饥饿;
- 正式搜索队列继续完全不访问 Match 类级锁;联合测试在同一 provider fan-out 中证明冷却站点返回可聚合失败 - 正式搜索队列继续完全不访问 Match 类级锁;联合测试在同一 provider fan-out 中证明冷却或失败站点独立记录
健康站点仍独立完成并释放预算,慢站点不拖住整轮 健康站点仍独立完成并释放预算,部分站点失败不改变订阅任务终态
- 同站点第二个 worker 无法认领,另一站点可同时认领;任务租约与站点租约均验证过期后以新 token 恢复, - 同站点第二个 worker 无法认领,另一站点可同时认领;任务租约与站点租约均验证过期后以新 token 恢复,
单订阅失败继续后续任务且批次保持聚合 failed; 单订阅失败继续后续任务且批次保持聚合 failed;
- 手工入口仅改变队列优先级和启动时机,不改变站点预算、失败冷却或 single-flightRSS/Spider 仍未接入该预算, - 手工入口仅改变队列优先级和启动时机,不改变站点预算、失败冷却或 single-flightRSS/Spider 仍未接入该预算,
+71 -1
View File
@@ -12,7 +12,10 @@ from sqlalchemy.orm import sessionmaker
from app.application.site.observation import report_site_search_outcome from app.application.site.observation import report_site_search_outcome
from app.application.subscription.contract import SubscriptionSnapshot from app.application.subscription.contract import SubscriptionSnapshot
from app.application.subscription.execution import SubscriptionExecutionAdmission from app.application.subscription.execution import (
SubscriptionExecutionAdmission,
raise_subscription_site_budget_failures,
)
from app.application.subscription.sitebudget import ( from app.application.subscription.sitebudget import (
SubscriptionSearchCancelled, SubscriptionSearchCancelled,
SubscriptionSearchDeferred, SubscriptionSearchDeferred,
@@ -252,6 +255,73 @@ def test_fallback_queue_continues_after_one_subscription_failure(tmp_path, monke
assert chain._subscription_execution_admission.release(lease) is True assert chain._subscription_execution_admission.release(lease) is True
def test_site_search_failure_logs_without_internal_traceback(tmp_path, monkeypatch):
"""可预期的站点聚合失败仍标记任务失败,但不输出误导性的内部堆栈。"""
chain = _chain(tmp_path, [_subscribe(15)])
_make_tasks_ready(monkeypatch)
def process(*_args, **_kwargs):
"""通过正式聚合助手制造站点失败。"""
raise_subscription_site_budget_failures(
("站点 Generic 搜索失败:站点请求或页面解析失败",)
)
monkeypatch.setattr(
chain,
"_process_search_subscription",
process,
)
error_logs = []
monkeypatch.setattr(
"app.chain.subscribe.searchtask.logger.error",
lambda message, **kwargs: error_logs.append((message, kwargs)),
)
with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()):
batch_id = chain.search(state="R")
batch = chain.get_search_batch(batch_id)
assert batch.state == "failed"
assert batch.failed_count == 1
assert error_logs == [
(
"订阅《治理电影 15》搜索失败:站点 Generic 搜索失败:站点请求或页面解析失败",
{"exc_info": False},
)
]
def test_inline_site_search_failure_logs_without_internal_traceback(tmp_path, monkeypatch):
"""兼容内联搜索遇到站点聚合失败时也不得输出内部堆栈。"""
chain = _chain(tmp_path, [_subscribe(16)])
del chain.subscription_search_repository
monkeypatch.setattr(chain, "_wait_before_scheduled_search", lambda *_args: None)
def process(*_args, **_kwargs):
"""通过正式聚合助手制造内联站点失败。"""
raise_subscription_site_budget_failures(
("站点 Generic 搜索失败:站点请求或页面解析失败",)
)
monkeypatch.setattr(chain, "_process_search_subscription", process)
error_logs = []
monkeypatch.setattr(
subscribe_search.logger,
"error",
lambda message, **kwargs: error_logs.append((message, kwargs)),
)
with patch("app.chain.subscribe.search.SearchChain", return_value=Mock()):
chain.search(state="R")
assert error_logs == [
(
"订阅 治理电影 16 搜索失败:站点 Generic 搜索失败:站点请求或页面解析失败",
{"exc_info": False},
)
]
def test_site_budget_conflict_requeues_task_without_batch_failure(tmp_path, monkeypatch): def test_site_budget_conflict_requeues_task_without_batch_failure(tmp_path, monkeypatch):
"""站点预算冲突应自动排队重试,不能把订阅批次置为失败。""" """站点预算冲突应自动排队重试,不能把订阅批次置为失败。"""
subscribe = _subscribe(5) subscribe = _subscribe(5)
+85
View File
@@ -337,6 +337,17 @@ def test_search_provider_releases_successful_site_budget():
assert snapshot.candidate_count == 1 assert snapshot.candidate_count == 1
assert snapshot.failure_count == 0 assert snapshot.failure_count == 0
assert snapshot.release_failure_count == 0 assert snapshot.release_failure_count == 0
chain.record_subscription_site_budget_failure("站点 Flaky 搜索失败:HTTP 429")
assert not chain.consume_subscription_site_budget_failures()
def test_other_search_results_keep_site_failures_non_terminal():
"""插件等其他搜索源已有候选时,站点失败不得覆盖可用结果。"""
chain = object.__new__(SearchChain)
chain.configure_subscription_site_budget(None)
chain.record_subscription_site_budget_failure("站点 Flaky 搜索失败:HTTP 429")
assert not chain.consume_subscription_site_budget_failures(has_results=True)
def test_search_provider_aggregates_swallowed_indexer_failure(monkeypatch): def test_search_provider_aggregates_swallowed_indexer_failure(monkeypatch):
@@ -420,6 +431,80 @@ def test_search_provider_aggregates_swallowed_indexer_failure(monkeypatch):
assert snapshot.cooldown_seconds == 900.0 assert snapshot.cooldown_seconds == 900.0
def test_search_provider_explains_error_flag_without_exception(monkeypatch):
"""索引器仅返回错误标志时也必须持久化可读原因,不能向任务暴露裸 `error`。"""
captured = {}
class _Repository(_WaitingRepository):
"""提供立即可用租约并记录通用失败收口。"""
def claim_site(self, *, site_id: int, owner: str, lease_seconds: int) -> SiteBudgetClaim:
"""返回当前调用独占的站点租约。"""
del owner, lease_seconds
return SiteBudgetClaim(
site_id=site_id,
acquired=True,
retry_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
consecutive_failures=0,
lease_token="lease-token",
)
def finish_site(self, **kwargs) -> bool:
"""记录预算收口参数供断言。"""
captured.update(kwargs)
return True
monkeypatch.setattr(
IndexerModule,
"_IndexerModule__search_check",
staticmethod(lambda _site, _keyword=None: True),
)
monkeypatch.setattr(
IndexerModule,
"_IndexerModule__execute_search",
staticmethod(lambda _site, _request: (True, [])),
)
monkeypatch.setattr(
IndexerModule,
"_IndexerModule__indexer_statistic",
staticmethod(lambda **_kwargs: None),
)
monkeypatch.setattr(
IndexerModule,
"_IndexerModule__parse_result",
staticmethod(lambda **_kwargs: []),
)
budget = SubscriptionSiteBudget(
repository=_Repository(),
owner="fallback-worker",
cancelled=lambda: False,
stop_state=ProcessStopState(),
)
chain = object.__new__(SearchChain)
chain.configure_subscription_site_budget(budget)
chain.search_site_torrents = object.__new__(IndexerModule).search_torrents
warnings = []
monkeypatch.setattr(
"app.chain.search.provider.logger.warning",
warnings.append,
)
result = chain._search_site_torrents_with_budget( # pylint: disable=protected-access
site={"id": 15, "name": "Generic"},
keyword="movie",
mtype=None,
page=0,
)
assert result == []
assert captured["outcome"] == "error"
assert captured["error"] == "站点请求或页面解析失败"
assert chain.consume_subscription_site_budget_failures() == (
"站点 Generic 搜索失败:站点请求或页面解析失败",
)
assert warnings == ["站点 Generic 搜索失败:站点请求或页面解析失败"]
def test_search_provider_logs_site_budget_release_failure(monkeypatch): def test_search_provider_logs_site_budget_release_failure(monkeypatch):
"""站点租约收口返回失败时必须进入摘要并即时记录错误。""" """站点租约收口返回失败时必须进入摘要并即时记录错误。"""
class _Repository(_WaitingRepository): class _Repository(_WaitingRepository):