fix(architecture): move site outcome state to budget

This commit is contained in:
jxxghp
2026-09-05 20:14:05 +08:00
parent b4be185e44
commit 1fb4a45064
4 changed files with 16 additions and 18 deletions
@@ -186,6 +186,8 @@ class SubscriptionSiteBudget:
self._clock = clock
self._phase_changed = phase_changed
self._metrics = metrics
self._outcome_lock = threading.Lock()
self._successful_site_ids: set[int] = set()
def acquire(self, site_id: int) -> SiteBudgetClaim:
"""只认领一次指定站点,未就绪时留待下一次正常调度。"""
@@ -217,6 +219,16 @@ class SubscriptionSiteBudget:
if self._metrics:
self._metrics.record_request(site_id, candidate_count)
def record_success(self, site_id: int) -> None:
"""记录一个已正常完成真实请求的站点。"""
with self._outcome_lock:
self._successful_site_ids.add(site_id)
def has_successful_site(self) -> bool:
"""判断本轮是否至少有一个站点正常完成真实请求。"""
with self._outcome_lock:
return bool(self._successful_site_ids)
def record_release_failure(self) -> None:
"""把站点租约释放异常写入轮次聚合器。"""
if self._metrics:
-1
View File
@@ -97,7 +97,6 @@ if TYPE_CHECKING:
matches_music_resource: Callable[..., Any]
music_site_keywords: Callable[..., Any]
process: Callable[..., Any]
record_subscription_site_budget_success: Callable[..., Any]
record_subscription_site_budget_failure: Callable[..., Any]
consume_subscription_site_budget_failures: Callable[..., Any]
record_subscription_site_budget_deferred: Callable[..., Any]
+3 -16
View File
@@ -52,18 +52,9 @@ class SearchChain(ChainBase):
"""仅为订阅搜索启用或清除站点预算,不影响其它搜索入口。"""
self._subscription_site_budget = budget
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_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:
"""线程安全地记录一个站点执行失败,供订阅任务判断轮次终态。"""
lock = getattr(self, "_subscription_site_budget_failure_lock", None)
@@ -72,20 +63,16 @@ class SearchChain(ChainBase):
with lock:
self._subscription_site_budget_failures.append(error)
def consume_subscription_site_budget_failures(
self,
*,
has_results: bool = False,
) -> tuple[str, ...]:
def consume_subscription_site_budget_failures(self, *, has_results: bool = False) -> tuple[str, ...]:
"""仅在没有成功搜索源时返回并清空本轮站点聚合失败。"""
lock = getattr(self, "_subscription_site_budget_failure_lock", None)
if lock is None:
return ()
with lock:
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_successes.clear()
budget = getattr(self, "_subscription_site_budget", None)
has_successful_site = isinstance(budget, SubscriptionSiteBudget) and budget.has_successful_site()
return () if has_results or has_successful_site else failures
def record_subscription_site_budget_deferred(
+1 -1
View File
@@ -245,7 +245,7 @@ class _SearchProviderSyncOwner(_SearchOwnerBase):
budget.record_request(site_id, len(result or []))
if observation.attempted:
if observation.outcome == "success":
self.record_subscription_site_budget_success(site_id)
budget.record_success(site_id)
elif observation.outcome != "skipped":
failure = observation.error or observation.outcome
message = f"站点 {site.get('name') or site_id} 搜索失败:{failure}"