From 1fb4a450643b51c55837a13d36940dfd0346733a Mon Sep 17 00:00:00 2001
From: jxxghp <jxxghp@gmail.com>
Date: Sat, 5 Sep 2026 20:14:05 +0800
Subject: [PATCH] fix(architecture): move site outcome state to budget

---
 app/application/subscription/sitebudget.py | 12 ++++++++++++
 app/chain/search/contract.py               |  1 -
 app/chain/search/facade.py                 | 19 +++----------------
 app/chain/search/provider.py               |  2 +-
 4 files changed, 16 insertions(+), 18 deletions(-)

diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py
index ff3c38e14..b1ac30dad 100644
--- a/app/application/subscription/sitebudget.py
+++ b/app/application/subscription/sitebudget.py
@@ -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:
diff --git a/app/chain/search/contract.py b/app/chain/search/contract.py
index cf56ac060..ef3bd7600 100644
--- a/app/chain/search/contract.py
+++ b/app/chain/search/contract.py
@@ -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]
diff --git a/app/chain/search/facade.py b/app/chain/search/facade.py
index 21acb55cd..de58a1f71 100644
--- a/app/chain/search/facade.py
+++ b/app/chain/search/facade.py
@@ -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(
diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py
index 67d75fbf2..9d4bcc612 100644
--- a/app/chain/search/provider.py
+++ b/app/chain/search/provider.py
@@ -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}"