diff --git a/backend/package/yuxi/channels/plugins/wechat_woc/woc_bridge_client.py b/backend/package/yuxi/channels/plugins/wechat_woc/woc_bridge_client.py index cb6d9acd..8e261666 100644 --- a/backend/package/yuxi/channels/plugins/wechat_woc/woc_bridge_client.py +++ b/backend/package/yuxi/channels/plugins/wechat_woc/woc_bridge_client.py @@ -433,6 +433,51 @@ class WocBridgeClient: # 通用 HTTP 执行 # ------------------------------------------------------------------ + async def _resolve_endpoint( + self, + account_id: str, + path: str, + override_bridge_url: str | None, + bridge_token: str | None, + ) -> tuple[str, str | None]: + """一次性解析 bridge URL 与鉴权 token，避免双重凭证缓存查询。 + + 非 wizard 模式（``override_bridge_url`` 为空）时从 CachePort 凭证缓存 + 读取一次凭证字典，同时提取 ``bridge_url`` 与 ``bridge_token``，避免 + ``_build_bridge_url`` 与 ``_get_bridge_token`` 各调一次 ``_get_credentials`` + 导致同一份凭证查 2 次。wizard 模式下直接使用调用方传入的 + ``override_bridge_url``，``bridge_token`` 透传。 + + @pre + - ``account_id`` 非空 + - ``path`` 以 ``/`` 开头 + - ``override_bridge_url`` 非空时必须以 ``http://`` 或 ``https://`` 开头 + + @post + - 返回 ``(url, bridge_token_or_none)`` 元组 + + @failure + - 透传 ``_get_credentials`` 的 ``DependencyError`` + - ``DependencyError``：bridge_url 为空字符串 + """ + if override_bridge_url is not None: + base = override_bridge_url.rstrip("/") + url = f"{base}{path}" if path.startswith("/") else f"{base}/{path}" + return url, bridge_token + # 非 wizard 模式：一次读取凭证字典，同时提取 bridge_url 与 bridge_token + credentials = await self._get_credentials(account_id) + bridge_url = credentials.get("bridge_url", "").rstrip("/") + if not bridge_url: + raise DependencyError( + dep=WOC_BRIDGE_DEP, + cause=Exception(f"bridge_url_empty: account_id={account_id}"), + ) + if not path.startswith("/"): + path = f"/{path}" + # 调用方未显式传入 bridge_token 时从凭证读取；传入时优先使用调用方值 + resolved_token = bridge_token if bridge_token is not None else credentials.get("bridge_token", "") + return f"{bridge_url}{path}", resolved_token + async def _execute_http( self, method: str, @@ -454,19 +499,15 @@ class WocBridgeClient: 在途请求计数：进入时 +1，退出时 -1，归零时设置 ``_drain_event`` 供 lifecycle 的 ``drain`` 等待。 - URL 构造：默认通过 ``_build_bridge_url(account_id, path)`` 从 CachePort - 凭证缓存读取 ``bridge_url`` 拼接；当 ``override_bridge_url`` 非空时直接与 - ``path`` 拼接，绕过缓存读取（用于 wizard 阶段账户未创建时探活，对齐 - wechat_ilink ``getconfig_with_token`` 模式）。 + URL 构造与鉴权注入：通过 ``_resolve_endpoint`` 一次性从凭证缓存读取 + ``bridge_url`` 与 ``bridge_token``（避免双重缓存查询），非空 token 注入 + ``Authorization: Bearer `` 头。``override_bridge_url`` 非空时 + 绕过缓存读取（wizard 阶段账户未创建时探活）。 - 鉴权注入：``bridge_token`` 非空时注入 ``Authorization: Bearer `` - 头，用于通过 WechatOnCloud 面板 ``/api/bridge/:id/*`` 反代路由的 M2M - 鉴权（与面板管理员会话二选一）。``bridge_token`` 为 ``None`` 或空字符串 - 时不注入该头，回退到面板会话鉴权（兼容未启用 M2M 鉴权的旧部署）。 - - 重试策略：``httpx.TimeoutException`` / ``httpx.NetworkError`` / 5xx - ``HTTPStatusError`` 最多重试 ``_HTTP_RETRY_ATTEMPTS`` 次（指数退避）， - 4xx（含 429）不重试，由调用方或 outbox 决策（P2-3 超时重试）。 + 重试策略：仅对 ``httpx.NetworkError``（连接失败/DNS 失败等瞬时网络抖动） + 做 1 次快速重试（不等待），超时与 5xx 不重试，由调用方或 outbox 机制 + 决策（P2-3 超时重试）。避免 bridge 客户端重试与 outbox 退避重试双重 + 放大延迟。 @pre - ``method`` 为合法 HTTP 方法（GET / POST） @@ -493,16 +534,10 @@ class WocBridgeClient: ) # 生成请求级 trace_id，注入 X-Trace-Id 头与 error_translator 供异常追踪 trace_id = uuid4().hex - if override_bridge_url is not None: - base = override_bridge_url.rstrip("/") - url = f"{base}{path}" if path.startswith("/") else f"{base}/{path}" - else: - url = await self._build_bridge_url(account_id, path) - # 鉴权 token 统一获取：非 wizard 模式（override_bridge_url 为空）且 - # 调用方未显式传入 bridge_token 时，从凭证缓存读取。wizard 模式下 - # bridge_token 由调用方显式传入（可为 None，表示未配置 M2M 鉴权）。 - if bridge_token is None and override_bridge_url is None: - bridge_token = await self._get_bridge_token(account_id) + # 一次性解析 URL 与鉴权 token，避免双重凭证缓存查询 + url, resolved_token = await self._resolve_endpoint( + account_id, path, override_bridge_url, bridge_token + ) # 合并请求头：调用方显式传入的 headers 优先（覆盖 User-Agent / X-Trace-Id） request_headers: dict[str, str] = { "User-Agent": self._user_agent, @@ -510,11 +545,11 @@ class WocBridgeClient: } if json_body is not None: request_headers["Content-Type"] = "application/json" - if bridge_token: + if resolved_token: # 注入 M2M 鉴权头，通过 WechatOnCloud 面板反代路由的双鉴权校验。 - # bridge_token 为 None 或空字符串时不注入，回退到面板会话鉴权 + # resolved_token 为 None 或空字符串时不注入，回退到面板会话鉴权 # （兼容未启用 M2M 鉴权的旧部署）。 - request_headers["Authorization"] = f"Bearer {bridge_token}" + request_headers["Authorization"] = f"Bearer {resolved_token}" if headers is not None: request_headers.update(headers) # 使用 path 作为 operation 上下文，供 translate_call_error 丰富超时错误信息 @@ -523,7 +558,7 @@ class WocBridgeClient: self._drain_event.clear() last_exc: httpx.HTTPError | None = None try: - for attempt in range(1, _HTTP_RETRY_ATTEMPTS + 1): + for attempt in range(1, _HTTP_NETWORK_ERROR_MAX_ATTEMPTS + 1): # 每次重试前重新检查活跃状态，stop/pause 后立即拒绝新请求 if not self._active: raise DependencyError( @@ -554,26 +589,12 @@ class WocBridgeClient: resp.raise_for_status() last_exc = None break - except httpx.TimeoutException as exc: - last_exc = exc - if attempt == _HTTP_RETRY_ATTEMPTS: - break - await self._logger.warn( - "woc bridge HTTP request timeout, retrying", - method=method, - path=path, - attempt=attempt, - trace_id=trace_id, - ) - await asyncio.sleep( - min( - _HTTP_RETRY_WAIT_MAX_SECONDS, - _HTTP_RETRY_WAIT_MIN_SECONDS * (2 ** (attempt - 1)), - ) - ) except httpx.NetworkError as exc: + # 仅对网络错误（连接失败/DNS 失败等瞬时抖动）做 1 次快速重试， + # 不等待。超时与 5xx 不重试，交给 outbox 退避重试机制处理， + # 避免双重重试放大延迟。 last_exc = exc - if attempt == _HTTP_RETRY_ATTEMPTS: + if attempt == _HTTP_NETWORK_ERROR_MAX_ATTEMPTS: break await self._logger.warn( "woc bridge HTTP request network error, retrying", @@ -582,33 +603,14 @@ class WocBridgeClient: attempt=attempt, trace_id=trace_id, ) - await asyncio.sleep( - min( - _HTTP_RETRY_WAIT_MAX_SECONDS, - _HTTP_RETRY_WAIT_MIN_SECONDS * (2 ** (attempt - 1)), - ) - ) except httpx.HTTPStatusError as exc: if exc.response.status_code < 500: # 4xx（含 429）为客户端/限流错误，不重试 raise + # 5xx 服务端错误不重试，交给 outbox 退避重试机制处理 last_exc = exc - if attempt == _HTTP_RETRY_ATTEMPTS: - break - await self._logger.warn( - "woc bridge HTTP request server error, retrying", - method=method, - path=path, - attempt=attempt, - status=exc.response.status_code, - trace_id=trace_id, - ) - await asyncio.sleep( - min( - _HTTP_RETRY_WAIT_MAX_SECONDS, - _HTTP_RETRY_WAIT_MIN_SECONDS * (2 ** (attempt - 1)), - ) - ) + break + # httpx.TimeoutException 不重试，交给 outbox 退避重试机制处理 if last_exc is not None: raise last_exc except httpx.HTTPError as exc: @@ -1030,8 +1032,10 @@ class WocBridgeClient: cause=Exception("client_inactive_after_stop"), ) trace_id = uuid4().hex - bridge_url = await self._build_bridge_url(account_id, "/api/messages/stream") - bridge_token = await self._get_bridge_token(account_id) + # 一次性解析 URL 与鉴权 token，避免双重凭证缓存查询 + bridge_url, bridge_token = await self._resolve_endpoint( + account_id, "/api/messages/stream", None, None + ) request_headers: dict[str, str] = { "User-Agent": self._user_agent, "X-Trace-Id": trace_id,