更新

backend/rpa_engine/batch_start.py

@@ -18,6 +18,10 @@ StartHandler = Callable[[int], Awaitable[dict[str, Any]]]
 JobToken = tuple[str, int]
 
 
+class _StartPreparationTimeout(Exception):
+    """Internal marker for the queue's own per-account deadline."""
+
+
 def _configured_concurrency() -> int:
     try:
         return max(1, min(8, int(os.getenv("KEFU_BATCH_START_CONCURRENCY", "2"))))
@@ -25,6 +29,16 @@ def _configured_concurrency() -> int:
         return 2
 
 
+def _configured_timeout_seconds() -> float:
+    try:
+        value = float(os.getenv("KEFU_BATCH_START_TIMEOUT_SECONDS", "90"))
+    except (TypeError, ValueError):
+        return 90.0
+    if value <= 0:
+        return 0.0
+    return max(5.0, min(600.0, value))
+
+
 def _utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
@@ -55,10 +69,16 @@ class BatchStartQueue:
         handler: StartHandler,
         concurrency: int | None = None,
         max_batches: int = 100,
+        timeout_seconds: float | None = None,
     ) -> None:
         self._handler = handler
         self.concurrency = max(1, int(concurrency or _configured_concurrency()))
         self.max_batches = max(10, int(max_batches or 100))
+        self.timeout_seconds = (
+            _configured_timeout_seconds()
+            if timeout_seconds is None
+            else max(0.0, float(timeout_seconds or 0.0))
+        )
         self._queue: asyncio.Queue[JobToken] = asyncio.Queue()
         self._pending_jobs: dict[int, JobToken] = {}
         self._active_tasks: dict[int, tuple[JobToken, asyncio.Task]] = {}
@@ -157,7 +177,21 @@ class BatchStartQueue:
                     )
                     self._active_tasks[account_id] = (job_token, handler_task)
 
-                result = await handler_task
+                if self.timeout_seconds > 0:
+                    try:
+                        result = await asyncio.wait_for(
+                            handler_task,
+                            timeout=self.timeout_seconds,
+                        )
+                    except asyncio.TimeoutError as exc:
+                        # wait_for cancels its task only when this queue's
+                        # deadline expires.  Preserve a TimeoutError raised by
+                        # the handler itself as its real account failure.
+                        if handler_task.cancelled():
+                            raise _StartPreparationTimeout from exc
+                        raise
+                else:
+                    result = await handler_task
                 async with self._lock:
                     record = self._batches.get(batch_id)
                     if record:
@@ -171,6 +205,28 @@ class BatchStartQueue:
                                 elapsed_seconds=round(time.monotonic() - started_at, 3),
                             )
                         record.updated_at = _utc_now()
+            except _StartPreparationTimeout:
+                elapsed = round(time.monotonic() - started_at, 3)
+                logger.warning(
+                    "Batch start timed out account=%s worker=%s after %.1fs",
+                    account_id,
+                    worker_number,
+                    self.timeout_seconds,
+                )
+                async with self._lock:
+                    record = self._batches.get(batch_id)
+                    if record:
+                        item = record.items[account_id]
+                        if item.get("status") != "cancelled":
+                            item.update(
+                                status="failed",
+                                message=(
+                                    f"启动准备超过 {self.timeout_seconds:g} 秒，"
+                                    "已跳过并继续处理后续账号"
+                                ),
+                                elapsed_seconds=elapsed,
+                            )
+                        record.updated_at = _utc_now()
             except asyncio.CancelledError:
                 async with self._lock:
                     record = self._batches.get(batch_id)

backend/rpa_engine/credential.py

@@ -6,7 +6,6 @@ from typing import Optional
 
 from rpa_engine.douyin_im.auth import DouyinAuth
 from rpa_engine.douyin_im.frontier import ensure_frontier_ws
-from rpa_engine.douyin_im.http_client import DouyinImHttpClient
 from rpa_engine.douyin_im.session import DouyinImSession
 from utils.cookie_store import analyze_cookie
 
@@ -131,13 +130,26 @@ async def build_cookie_credential_detail(
 async def validate_im_session(
     session: DouyinImSession,
     _bypass_global_limit: bool = False,
+    *,
+    startup_priority: bool = False,
 ) -> tuple[bool, str]:
     if not _bypass_global_limit:
         from rpa_engine.douyin_im.traffic_control import get_traffic_controller
 
         controller = get_traffic_controller()
-        async with controller.background_slot(0, "credential validation"):
-            return await validate_im_session(session, _bypass_global_limit=True)
+        # Startup validation must not sit behind hundreds of recurring
+        # conversation polls.  It still shares the same global concurrency
+        # cap, so this changes ordering without increasing bandwidth usage.
+        async with controller.background_slot(
+            0,
+            "credential validation",
+            startup=startup_priority,
+        ):
+            return await validate_im_session(
+                session,
+                _bypass_global_limit=True,
+                startup_priority=startup_priority,
+            )
 
     if not session.can_direct_im():
         if not has_im_session_token(session):
@@ -155,17 +167,12 @@ async def validate_im_session(
         if not auth.is_sign_ready():
             return False, "缺少 IM 签名密钥（web_protect/keys），请用浏览器登录补全"
         session.my_uid = int(uid)
-        async with DouyinImHttpClient(session) as http:
-            await http.get_unread_count()
-            # 若已缓存到会话票据，优先校验其是否仍新鲜（最理想）。
-            if session.conv_meta:
-                ok, reason = await http.verify_messaging_capability(auth, session.my_uid)
-                if ok:
-                    return True, reason
-            # 没有缓存会话票据是首次登录的正常情况：会话 ticket 会在发送时即时
-            # 创建/获取（resolve_conversation_meta），因此只要 Cookie + sessionid +
-            # 签名密钥(web_protect/keys) + UID 齐全，就视为可 IM 直连托管，不必再开浏览器。
-            return True, "IM 凭证就绪（Cookie 与签名密钥齐全，可直连托管）"
+        # unread_count and ticket probes were previously issued here, but
+        # neither result changed the final decision: unread failures become
+        # zero and a stale/missing ticket is resolved lazily at send time.
+        # Keeping those probes doubled large-batch startup traffic without
+        # adding an authoritative validation signal.
+        return True, "IM 凭证就绪（Cookie 与签名密钥齐全，可直连托管）"
     except Exception as e:
         logger.warning(f"IM session validation failed: {e}")
         return False, f"IM 运行时验证失败: {e}"
@@ -174,6 +181,8 @@ async def validate_im_session(
 async def assess_account_credential(
     cookie_data: Optional[str],
     im_session_data: Optional[str] = None,
+    *,
+    startup_priority: bool = False,
 ) -> dict:
     cookie_info = analyze_cookie(cookie_data)
     result = {
@@ -212,7 +221,10 @@ async def assess_account_credential(
         result["should_reset"] = _should_reset_credentials(result)
         return result
 
-    im_ok, im_reason = await validate_im_session(session)
+    im_ok, im_reason = await validate_im_session(
+        session,
+        startup_priority=startup_priority,
+    )
     result["im_ready"] = im_ok
     if im_ok:
         result["can_skip_browser"] = True

backend/rpa_engine/douyin_im/http_client.py

@@ -737,7 +737,7 @@ class DouyinImHttpClient:
                     pass
         return total
 
-    async def get_conversations(self) -> list[dict]:
+    async def get_conversations(self, *, enrich_profiles: bool = True) -> list[dict]:
         """拉取会话列表，返回标准化会话"""
         payloads = [
             {"cursor": 0, "count": 50, "inbox_type": 0},
@@ -754,7 +754,12 @@ class DouyinImHttpClient:
             if data is None:
                 data = await self._request("GET", "/v1/conversation/list", body)
             if data is None:
-                continue
+                # Payload variants only help with schema compatibility.  They
+                # cannot repair a network outage, so stop after POST + GET
+                # both fail instead of occupying a scarce global slot for up
+                # to four more full request timeouts.
+                logger.warning("Conversation poll transport failed; skipping payload fallbacks")
+                break
 
             status_code = data.get("status_code") if isinstance(data, dict) else None
             error_text = ""
@@ -814,6 +819,9 @@ class DouyinImHttpClient:
         enriched: list[dict] = []
         for item in conversations:
             conv = enrich_conversation_item(item, my_uid)
+            if not enrich_profiles:
+                enriched.append(conv)
+                continue
             peer_uid = str(conv.get("peer_uid") or "")
             name = (conv.get("sender_name") or "").strip()
             avatar = str(conv.get("sender_avatar") or "").strip()

backend/rpa_engine/douyin_im/service.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import os
 import time
 from typing import Awaitable, Callable, Optional
 
@@ -26,6 +27,30 @@ LogFn = Callable[..., Awaitable[None]]
 ReceivedLogFn = Callable[..., Awaitable[None]]
 
 
+def _env_poll_seconds(name: str, default: float, minimum: float = 5.0) -> float:
+    try:
+        return max(minimum, float(os.getenv(name, str(default))))
+    except (TypeError, ValueError):
+        return default
+
+
+def _conversation_poll_timing(account_id: int, has_ws: bool) -> tuple[float, float]:
+    """Return the reconciliation interval and a stable per-account stagger.
+
+    WebSocket is the real-time receive path.  HTTP polling is only a safety
+    reconciliation when that path exists, so running it every 15 seconds for
+    hundreds of accounts wastes bandwidth and eventually starves new starts.
+    Accounts without WebSocket keep the original fast polling cadence.
+    """
+    interval = _env_poll_seconds(
+        "KEFU_WS_RECONCILE_INTERVAL_SECONDS" if has_ws else "KEFU_HTTP_POLL_INTERVAL_SECONDS",
+        120.0 if has_ws else 15.0,
+    )
+    spread_ms = max(1, int(interval * 1000))
+    stagger = ((int(account_id or 0) * 2654435761) % spread_ms) / 1000.0
+    return interval, stagger
+
+
 class DouyinImService:
     """抖音 IM 直连服务：WebSocket 实时监听 + HTTP 轮询 + 自动回复"""
 
@@ -734,11 +759,18 @@ class DouyinImService:
         controller = get_traffic_controller()
         async with controller.background_slot(self.account_id, "conversation poll"):
             async with DouyinImHttpClient(self.session, account_id=self.account_id) as http:
-                unread_total = await http.get_unread_count()
-                if unread_total:
-                    logger.info(f"IM unread total: {unread_total}")
-                conversations = await http.get_conversations()
-                await self._index_conversations(conversations)
+                conversations = await http.get_conversations(enrich_profiles=False)
+        # Profile enrichment may involve several slow third-party requests.
+        # Run it after releasing the conversation-list slot; each individual
+        # lookup re-enters the shared controller and yields fairly to startup
+        # validation and other accounts between profiles.
+        await self._index_conversations(conversations)
+        unread_total = sum(
+            max(0, int(item.get("unread_count") or 0))
+            for item in conversations
+        )
+        if unread_total:
+            logger.info(f"IM unread total: {unread_total}")
         # Message handling may wait in the global send lane.  Do not keep one
         # of the scarce background HTTP slots occupied while that happens.
         for conv in conversations:
@@ -811,8 +843,10 @@ class DouyinImService:
         )
         await self._ws_client.start()
 
+        initial_poll_succeeded = False
         try:
             await self._poll_conversations()
+            initial_poll_succeeded = True
         except Exception as e:
             logger.warning(f"Initial conversation poll failed: {e}")
             system_logger.record(
@@ -823,6 +857,34 @@ class DouyinImService:
                 account_id=self.account_id,
             )
 
+        ws_connected = bool(
+            self._ws_client and getattr(self._ws_client, "connected", False)
+        )
+        poll_interval, poll_stagger = _conversation_poll_timing(
+            self.account_id,
+            ws_connected,
+        )
+        loop = asyncio.get_running_loop()
+        if initial_poll_succeeded:
+            initial_retry_interval = poll_interval
+            initial_retry_stagger = poll_stagger
+        else:
+            # If the authoritative first poll failed, retry on the fast HTTP
+            # cadence even when WebSocket connected in the meantime.
+            initial_retry_interval, initial_retry_stagger = (
+                _conversation_poll_timing(self.account_id, False)
+            )
+        next_conversation_poll_at = (
+            loop.time() + initial_retry_interval + initial_retry_stagger
+        )
+        logger.info(
+            "Conversation reconciliation account=%s interval=%.1fs stagger=%.1fs ws=%s",
+            self.account_id,
+            poll_interval,
+            poll_stagger,
+            "yes" if ws_connected else "no",
+        )
+
         loop_count = 0
         while self._running:
             # The initial poll above is authoritative.  Sleep before the next
@@ -832,8 +894,39 @@ class DouyinImService:
                 break
             loop_count += 1
             try:
-                if loop_count % 3 == 0:
-                    await self._poll_conversations()
+                current_ws_connected = bool(
+                    self._ws_client
+                    and getattr(self._ws_client, "connected", False)
+                )
+                if current_ws_connected != ws_connected:
+                    ws_connected = current_ws_connected
+                    poll_interval, poll_stagger = _conversation_poll_timing(
+                        self.account_id,
+                        ws_connected,
+                    )
+                    candidate_poll_at = loop.time() + poll_interval + poll_stagger
+                    # Never postpone an already scheduled reconciliation.
+                    # In particular, reconnecting must preserve the earlier
+                    # fallback poll that covers messages missed while offline.
+                    next_conversation_poll_at = min(
+                        next_conversation_poll_at,
+                        candidate_poll_at,
+                    )
+                    logger.info(
+                        "Conversation reconciliation rescheduled account=%s "
+                        "interval=%.1fs ws=%s",
+                        self.account_id,
+                        poll_interval,
+                        "yes" if ws_connected else "no",
+                    )
+                if loop.time() >= next_conversation_poll_at:
+                    try:
+                        await self._poll_conversations()
+                    finally:
+                        # Advance on both success and failure.  Otherwise a
+                        # past deadline retries every five-second loop tick
+                        # during an outage and amplifies traffic.
+                        next_conversation_poll_at = loop.time() + poll_interval
                 if loop_count % 6 == 0:
                     logger.info(f"IM direct tick #{loop_count} account={self.account_id}")
                 # 关注欢迎语：约每 60s 检测一次新粉丝（独立于私信轮询，失败不影响主循环）

backend/rpa_engine/douyin_im/traffic_control.py

@@ -396,6 +396,15 @@ class TrafficController:
         self._background = asyncio.Semaphore(
             _env_int("KEFU_BACKGROUND_NETWORK_CONCURRENCY", 2)
         )
+        # Recurring polls can create hundreds of waiters when many accounts
+        # are online.  Admit at most one normal waiter to the semaphore at a
+        # time so startup validation can join near the front instead of being
+        # buried behind the entire polling backlog.  The shared semaphore is
+        # still the single bandwidth cap; startup work does not add extra
+        # network concurrency.
+        self._background_normal_admission = asyncio.Lock()
+        self._background_startup_clear = asyncio.Event()
+        self._background_startup_clear.set()
         self._browser = asyncio.Semaphore(
             _env_int("KEFU_BROWSER_START_CONCURRENCY", 1)
         )
@@ -410,12 +419,20 @@ class TrafficController:
         )
         self.background_waiting = 0
         self.background_active = 0
+        self.background_startup_waiting = 0
+        self.background_startup_active = 0
         self.browser_waiting = 0
         self.browser_active = 0
         self.media_proxy_active = 0
 
     @asynccontextmanager
-    async def background_slot(self, account_id: int = 0, description: str = "request"):
+    async def background_slot(
+        self,
+        account_id: int = 0,
+        description: str = "request",
+        *,
+        startup: bool = False,
+    ):
         current_task = asyncio.current_task()
         owner_task, depth = self._background_owner.get()
         if owner_task is current_task and depth > 0:
@@ -429,12 +446,29 @@ class TrafficController:
         started = asyncio.get_running_loop().time()
         self.background_waiting += 1
         try:
-            await self._background.acquire()
+            if startup:
+                self.background_startup_waiting += 1
+                self._background_startup_clear.clear()
+                try:
+                    await self._background.acquire()
+                finally:
+                    self.background_startup_waiting -= 1
+                    if self.background_startup_waiting == 0:
+                        self._background_startup_clear.set()
+            else:
+                # Only one recurring/background request may wait directly on
+                # the shared semaphore.  A later startup request therefore
+                # has at most one normal request ahead of it, not hundreds.
+                async with self._background_normal_admission:
+                    await self._background_startup_clear.wait()
+                    await self._background.acquire()
         except BaseException:
             self.background_waiting -= 1
             raise
         self.background_waiting -= 1
         self.background_active += 1
+        if startup:
+            self.background_startup_active += 1
         token = self._background_owner.set((current_task, 1))
         waited = asyncio.get_running_loop().time() - started
         if waited >= 1.0:
@@ -448,6 +482,8 @@ class TrafficController:
             yield
         finally:
             self._background_owner.reset(token)
+            if startup:
+                self.background_startup_active -= 1
             self.background_active -= 1
             self._background.release()
 
@@ -492,6 +528,8 @@ class TrafficController:
             "background": {
                 "active": self.background_active,
                 "waiting": self.background_waiting,
+                "startup_active": self.background_startup_active,
+                "startup_waiting": self.background_startup_waiting,
             },
             "browser": {
                 "active": self.browser_active,

backend/rpa_engine/douyin_im/ws_client.py

@@ -27,6 +27,7 @@ class DouyinImWsClient:
         self.on_message = on_message
         self.account_id = account_id
         self._running = False
+        self.connected = False
         self._task: Optional[asyncio.Task] = None
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._ws_app: Optional[WebSocketApp] = None
@@ -50,6 +51,7 @@ class DouyinImWsClient:
 
     async def stop(self):
         self._running = False
+        self.connected = False
         with self._ws_lock:
             if self._ws_app:
                 try:
@@ -150,6 +152,7 @@ class DouyinImWsClient:
             return
 
         def on_open(_ws):
+            self.connected = True
             logger.info("IM WebSocket connected")
             system_logger.record(
                 "实时接收通道已连接",
@@ -174,6 +177,7 @@ class DouyinImWsClient:
                 )
 
         def on_close(_ws, code, msg):
+            self.connected = False
             logger.info(f"IM WebSocket closed: code={code}, msg={msg}")
             if self._running:
                 system_logger.record(
@@ -206,6 +210,7 @@ class DouyinImWsClient:
         try:
             ws_app.run_forever(origin="https://www.douyin.com", ping_interval=20, ping_timeout=10)
         finally:
+            self.connected = False
             with self._ws_lock:
                 if self._ws_app is ws_app:
                     self._ws_app = None