yongjiu shibai de fenxiang jilu zhijie tuo li duilie, bu zai rang shangyou chongshi

services/share_evidence_service.py

@@ -45,7 +45,11 @@ SHARE_EVIDENCE_STORE = DATA_DIR / "share_evidence.json"
 STATUS_PENDING = "pending"    # 已登记，等用户在保留期内按需取证
 STATUS_SHARED = "shared"      # 已生成分享链接，链接有效中
 STATUS_RELEASED = "released"  # 已回收（会话已隐藏、分享已撤）
-STATUS_ORPHAN = "orphan"      # 账号已不可用，回收失败待重试
+STATUS_ORPHAN = "orphan"      # 账号已不可用，回收失败待重试（仅限可恢复的失败）
+# 上游永久拒绝（token 吊销 / CF 拦截 / 会话已删）：重试不可能成功，直接移出队列。
+# ⚠️ 不能用「ORPHAN + 30 天退避」凑合 —— 30 天后又会全量重试一次，而记录只增不减，
+# 每次都把 cpu-basic 打满（2026-10-19 事故复发的直接原因）。
+STATUS_ABANDONED = "abandoned"
 
 _ACTIVE_STATUSES = (STATUS_PENDING, STATUS_SHARED, STATUS_ORPHAN)
 # 回收的兜底周期：正常路径由 `_next_due_ts()` 精确唤醒、到期即回收，这个周期只负责
@@ -77,6 +81,22 @@ def _is_permanent_failure(exc: BaseException) -> bool:
     return any(marker in text for marker in _PERMANENT_FAILURE_MARKERS)
 
 
+def _hopeless(record: dict[str, Any]) -> bool:
+    """该记录是否已经不可能成功、不该再碰上游。
+
+    两种：① 已被判为永久失败（历史遗留的 ``permanent`` / ``abandoned`` 标记）；
+    ② 尝试次数超过上限（老版本会无限重试，台账里积压了大量 attempts 200+ 的记录）。
+    """
+    if record.get("status") not in _ACTIVE_STATUSES:
+        return False
+    if record.get("permanent"):
+        return True
+    try:
+        return int(record.get("attempts") or 0) >= _MAX_RELEASE_ATTEMPTS
+    except (TypeError, ValueError):
+        return False
+
+
 def _now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
@@ -511,6 +531,7 @@ class ShareEvidenceService:
                 and float(record.get("next_attempt_ts") or 0) <= now
                 and effective_expires_ts(record) > now
                 and _clean(record.get("conversation_id")) not in self._inflight
+                and not _hopeless(record)
             ]
         due.sort(key=lambda item: float(item.get("created_ts") or 0))
 
@@ -537,13 +558,15 @@ class ShareEvidenceService:
                         attempts = int(stored.get("attempts") or 0) + 1
                         stored["attempts"] = attempts
                         permanent = _is_permanent_failure(exc) or attempts >= _MAX_RELEASE_ATTEMPTS
-                        # 30s → 60s → 120s → 240s → 480s → 600s 封顶；永久失败直接远期搁置
-                        backoff = (
-                            _LONG_SLEEP_SECONDS
-                            if permanent
-                            else min(30 * (2 ** min(attempts - 1, 5)), 600)
-                        )
-                        stored["next_attempt_ts"] = time.time() + backoff
+                        if permanent:
+                            # 上游已判死刑：直接移出队列，别再让它在下次重启后重来。
+                            stored["status"] = STATUS_ABANDONED
+                            stored["permanent"] = True
+                        else:
+                            # 30s → 60s → 120s → 240s → 480s → 600s 封顶
+                            stored["next_attempt_ts"] = time.time() + min(
+                                30 * (2 ** min(attempts - 1, 5)), 600
+                            )
                         stored["last_error"] = str(exc)[:300]
                         self._save_locked()
                 logger.warning({
@@ -557,6 +580,35 @@ class ShareEvidenceService:
 
     # ------------------------------------------------------------------ 回收
 
+    def _abandon_hopeless(self) -> int:
+        """把所有「重试不可能成功」的记录一次性移出队列，且**不碰上游**。
+
+        必须在做任何 due 计算之前跑：老版本在台账里留下了大量 attempts 200+ 的
+        401/403 记录，每次容器重启都会整批重新尝试，每一条都要等 Cloudflare JS
+        挑战超时（CPU 密集、持 GIL），2 核直接被吃满，整站连 /health 都不响应。
+        只有「启动即清理、且永不重试」才能彻底断掉这个循环。
+        """
+        with self._lock:
+            victims = [cid for cid, rec in self._records.items() if _hopeless(rec)]
+            if not victims:
+                return 0
+            now = time.time()
+            for conversation_id in victims:
+                record = self._records.get(conversation_id)
+                if record is None:
+                    continue
+                record["status"] = STATUS_ABANDONED
+                record["permanent"] = True
+                record["next_attempt_ts"] = now + _LONG_SLEEP_SECONDS
+                record.setdefault("last_error", "上游永久拒绝（401/403/404）或尝试次数超限，已停止重试")
+            self._save_locked()
+        logger.info({
+            "event": "share_evidence_abandoned_hopeless",
+            "count": len(victims),
+            "reason": "permanent_upstream_rejection_or_attempt_cap",
+        })
+        return len(victims)
+
     def reap_expired(self, *, limit: int = 20) -> dict[str, Any]:
         """回收所有已过保留期的会话：隐藏会话（顺带废掉链接）+ 删除分享。
 
@@ -564,6 +616,7 @@ class ShareEvidenceService:
         快照值——站长把保留时长调短后，旧记录必须立刻跟着到期，而不是继续按旧值熬。
         """
         now = time.time()
+        self._abandon_hopeless()
         with self._lock:
             due = [
                 dict(record)
@@ -574,6 +627,7 @@ class ShareEvidenceService:
                 # 「到期即唤醒」精确调度就会变成每 0.5 秒狂刷上游（旧实现靠 5 分钟大
                 # 周期兜着才没暴露）。
                 and float(record.get("next_attempt_ts") or 0) <= now
+                and not _hopeless(record)
             ]
         due.sort(key=effective_expires_ts)
 
@@ -591,17 +645,18 @@ class ShareEvidenceService:
                     stored = self._record_locked(conversation_id)
                     if stored is not None and stored.get("status") in _ACTIVE_STATUSES:
                         attempts = int(stored.get("attempts") or 0) + 1
-                        stored["status"] = STATUS_ORPHAN
                         stored["attempts"] = attempts
-                        permanent = _is_permanent_failure(exc) or attempts >= _MAX_RELEASE_ATTEMPTS
-                        # 30s → 60s → 120s → 240s → 480s → 600s 封顶；永久失败直接远期搁置
-                        backoff = (
-                            _LONG_SLEEP_SECONDS
-                            if permanent
-                            else min(30 * (2 ** min(attempts - 1, 5)), 600)
-                        )
-                        stored["next_attempt_ts"] = time.time() + backoff
                         stored["last_error"] = str(exc)[:300]
+                        if _is_permanent_failure(exc) or attempts >= _MAX_RELEASE_ATTEMPTS:
+                            # 永久失败：标记后永久移出队列。之前只做「30 天远期搁置」，
+                            # 重启后整批重新到期 → 又全体撞一次上游 403 → V8 求解打满 CPU。
+                            stored["status"] = STATUS_ABANDONED
+                            stored["permanent"] = True
+                        else:
+                            stored["status"] = STATUS_ORPHAN
+                            stored["next_attempt_ts"] = time.time() + min(
+                                30 * (2 ** min(attempts - 1, 5)), 600
+                            )
                         self._save_locked()
                 logger.warning({
                     "event": "share_evidence_release_failed",
@@ -733,6 +788,10 @@ class ShareEvidenceService:
             for record in self._records.values():
                 if record.get("status") not in _ACTIVE_STATUSES:
                     continue
+                if _hopeless(record):
+                    # 已判死刑但还没被 _abandon_hopeless 收走的记录，不能参与唤醒，
+                    # 否则 worker 会被它反复叫醒、空转抢 CPU。
+                    continue
                 expires_ts = effective_expires_ts(record)
                 if expires_ts <= 0:
                     continue
@@ -785,6 +844,12 @@ class ShareEvidenceService:
             self.reconcile_retention()
         except Exception as exc:
             logger.warning({"event": "share_evidence_reconcile_error", "error": str(exc)})
+        # 启动先清掉不可能成功的积压记录，**在**任何一轮 due 计算之前。否则重启后
+        # 整批 401/403 会重新撞上游（403 → V8 求解打满 2 核 → 整站无响应）。
+        try:
+            self._abandon_hopeless()
+        except Exception as exc:
+            logger.warning({"event": "share_evidence_abandon_error", "error": str(exc)})
 
         def run() -> None:
             last_reap = time.time()