|
|
|
|
|
|
| """匿名访客会话池。"""
|
|
|
| import asyncio
|
| import time
|
| from dataclasses import dataclass, field
|
| from threading import Lock
|
| from typing import Dict, Any, List, Optional, Set
|
|
|
| import httpx
|
| import jwt
|
|
|
| from app.core.config import settings
|
| from app.core.http_client import SharedHttpClients
|
| from app.core.headers import build_dynamic_headers as _build_dynamic_headers
|
| from app.core.httpx_errors import normalize_httpx_exception, normalize_httpx_response
|
| from app.core.resin_compat import apply_resin_account_header
|
| from app.core.upstream_urls import build_upstream_url
|
| from app.utils.fe_version import get_latest_fe_version
|
| from app.utils.logger import logger, log_exception
|
| from app.utils.utlis import mask_token
|
|
|
| AUTH_URL = "/api/v1/auths/"
|
| CHATS_URL = "/api/v1/chats/"
|
|
|
|
|
| def _decode_token_payload(token: str) -> Dict[str, Any]:
|
| """无需校验签名，仅解密 JWT Payload 以获取元数据。"""
|
| try:
|
| return jwt.decode(token, options={"verify_signature": False})
|
| except Exception:
|
| return {}
|
|
|
|
|
| @dataclass
|
| class GuestSession:
|
| """单个匿名访客会话。"""
|
|
|
| token: str
|
| user_id: str
|
| username: str
|
| created_at: float = field(default_factory=time.time)
|
| active_requests: int = 0
|
| valid: bool = True
|
| failure_count: int = 0
|
| last_failure_time: float = 0.0
|
|
|
| @property
|
| def age(self) -> float:
|
| """会话存活时间。"""
|
| return time.time() - self.created_at
|
|
|
| def snapshot(self) -> Dict[str, str]:
|
| """获取当前会话快照。"""
|
| return {
|
| "token": self.token,
|
| "user_id": self.user_id,
|
| "username": self.username,
|
| }
|
|
|
|
|
| class GuestSessionPool:
|
| """匿名访客会话池，支持最小负载获取与失败替换。"""
|
|
|
| def __init__(
|
| self,
|
| pool_size: int = 3,
|
| session_max_age: int = 480,
|
| maintenance_interval: int = 30,
|
| max_failures: int = 10,
|
| ):
|
| self.pool_size = max(1, pool_size)
|
| self.session_max_age = max(60, session_max_age)
|
| self.maintenance_interval = max(10, maintenance_interval)
|
| self.max_failures = max(1, max_failures)
|
| self._lock = Lock()
|
| self._sessions: Dict[str, GuestSession] = {}
|
| self._maintenance_task: Optional[asyncio.Task] = None
|
| self._capacity_lock = asyncio.Lock()
|
| self._create_rate_lock = asyncio.Lock()
|
| self._next_create_allowed_at = 0.0
|
| self._create_min_interval = 1.0
|
| self._http_clients = SharedHttpClients(follow_redirects=True)
|
| self._background_tasks: Set[asyncio.Task] = set()
|
| self._cleanup_parallelism = max(1, settings.GUEST_CLEANUP_PARALLELISM)
|
|
|
| @staticmethod
|
| def _is_retryable_create_error(exc: Exception) -> bool:
|
| """匿名会话创建阶段，哪些异常值得重试。"""
|
| return isinstance(exc, httpx.TransportError)
|
|
|
| def _track_background_task(self, coro) -> asyncio.Task:
|
| """跟踪后台任务，避免清理阻塞前台重试路径。"""
|
| task = asyncio.create_task(coro)
|
| self._background_tasks.add(task)
|
|
|
| def _on_done(done_task: asyncio.Task):
|
| self._background_tasks.discard(done_task)
|
| try:
|
| done_task.result()
|
| except asyncio.CancelledError:
|
| pass
|
| except Exception as exc:
|
| logger.warning(
|
| "⚠️ 匿名会话后台任务异常",
|
| exc_info=settings.DEBUG_LOGGING,
|
| )
|
|
|
| task.add_done_callback(_on_done)
|
| return task
|
|
|
| async def _wait_background_tasks(self, cancel: bool = False):
|
| """等待当前已注册的后台任务结束。"""
|
| pending = list(self._background_tasks)
|
| if pending:
|
| if cancel:
|
| for task in pending:
|
| task.cancel()
|
| await asyncio.gather(*pending, return_exceptions=True)
|
|
|
| async def _delete_sessions_concurrently(self, sessions: List[GuestSession]):
|
| """并发清理多枚匿名会话，加快池维护速度。"""
|
| if not sessions:
|
| return
|
|
|
| semaphore = asyncio.Semaphore(self._cleanup_parallelism)
|
|
|
| async def _cleanup(session: GuestSession):
|
| async with semaphore:
|
| await self._delete_all_chats(session)
|
|
|
| await asyncio.gather(*(_cleanup(session) for session in sessions))
|
|
|
| async def _wait_create_slot(self, reason: str) -> None:
|
| async with self._create_rate_lock:
|
| now = time.time()
|
| wait_seconds = max(0.0, self._next_create_allowed_at - now)
|
| if wait_seconds > 0:
|
| logger.debug(
|
| "[guest_session.create] throttled reason=%s wait_seconds=%.2f",
|
| reason,
|
| wait_seconds,
|
| )
|
| await asyncio.sleep(wait_seconds)
|
| self._next_create_allowed_at = time.time() + self._create_min_interval
|
|
|
| async def _create_session(self, reason: str = "unspecified") -> GuestSession:
|
| """创建一个新的匿名访客会话。
|
|
|
| 从上游获取真实 token，并解密其内部值作为 session 元数据。
|
| """
|
| max_retries = 3
|
|
|
| for attempt in range(1, max_retries + 1):
|
| try:
|
| await self._wait_create_slot(reason)
|
| fe_version = await get_latest_fe_version()
|
| headers = _build_dynamic_headers(fe_version)
|
|
|
| client = self._http_clients.get_client()
|
| response = await client.get(build_upstream_url(AUTH_URL), headers=headers)
|
|
|
| if response.status_code != 200:
|
| raise RuntimeError(
|
| f"匿名会话创建失败: HTTP {response.status_code} {response.text[:200]}"
|
| )
|
|
|
| data = response.json()
|
| token = str(data.get("token") or "").strip()
|
| if not token:
|
| raise RuntimeError(f"匿名会话创建失败: 未返回 token {data}")
|
|
|
| logger.debug(
|
| "[guest_session.create] reason=%s attempt=%s status=%s bytes=%s token=%s",
|
| reason,
|
| attempt,
|
| response.status_code,
|
| response.headers.get("content-length") or len(response.content or b""),
|
| mask_token(token),
|
| )
|
|
|
|
|
|
|
| payload = _decode_token_payload(token)
|
|
|
| user_id = str(
|
| payload.get("id")
|
| or data.get("id")
|
| or data.get("user_id")
|
| or f"guest-{token[:12]}"
|
| ).strip()
|
|
|
|
|
| raw_email = (
|
| payload.get("email") or data.get("name") or data.get("email") or ""
|
| )
|
| username = str(raw_email).split("@")[0] or f"Guest-{user_id[:8]}"
|
|
|
| logger.debug(
|
| "获取匿名会话成功: user_id=%s, username=%s, reason=%s", user_id, username, reason
|
| )
|
| return GuestSession(
|
| token=token,
|
| user_id=user_id,
|
| username=username,
|
| )
|
| except Exception as exc:
|
| if self._is_retryable_create_error(exc):
|
| logger.warning(
|
| "⚠️ %s",
|
| normalize_httpx_exception(
|
| exc,
|
| method="GET",
|
| url=build_upstream_url(AUTH_URL),
|
| context="guest_session.create",
|
| attempt=attempt,
|
| ),
|
| )
|
| if attempt < max_retries:
|
| await asyncio.sleep(min(2 ** (attempt - 1), 4))
|
| continue
|
| else:
|
| log_exception(logger, "❌ 匿名会话创建异常")
|
| raise
|
|
|
| async def _delete_all_chats(self, session: GuestSession) -> bool:
|
| """删除匿名会话的全部对话，尽量释放并发占用。"""
|
| fe_version = await get_latest_fe_version()
|
| headers = _build_dynamic_headers(fe_version)
|
| headers.update(
|
| {
|
| "Authorization": f"Bearer {session.token}",
|
| "Accept": "application/json",
|
| "Content-Type": "application/json",
|
| }
|
| )
|
| apply_resin_account_header(headers, session.token)
|
|
|
| try:
|
| client = self._http_clients.get_client()
|
| chats_url = build_upstream_url(CHATS_URL)
|
| response = await client.delete(chats_url, headers=headers)
|
|
|
| if response.status_code == 200:
|
| logger.debug("已清理匿名会话聊天记录: %s", session.user_id)
|
| return True
|
|
|
| logger.warning(
|
| "⚠️ 清理匿名会话聊天记录失败: %s",
|
| normalize_httpx_response(
|
| response.status_code,
|
| response.text,
|
| content_type=response.headers.get("content-type"),
|
| method="DELETE",
|
| url=chats_url,
|
| context=f"guest_session.cleanup:{session.user_id}",
|
| ),
|
| )
|
| except Exception as exc:
|
| logger.warning(
|
| "⚠️ 清理匿名会话聊天记录异常: %s",
|
| normalize_httpx_exception(
|
| exc,
|
| method="DELETE",
|
| url=build_upstream_url(CHATS_URL),
|
| context=f"guest_session.cleanup:{session.user_id}",
|
| ),
|
| exc_info=settings.DEBUG_LOGGING,
|
| )
|
|
|
| return False
|
|
|
| def _list_valid_sessions(
|
| self,
|
| exclude_user_ids: Optional[Set[str]] = None,
|
| ) -> List[GuestSession]:
|
| """获取有效匿名会话列表。"""
|
| excluded = exclude_user_ids or set()
|
| with self._lock:
|
| return [
|
| session
|
| for session in self._sessions.values()
|
| if session.valid
|
| and session.age < self.session_max_age
|
| and session.user_id not in excluded
|
| ]
|
|
|
| async def _ensure_capacity(self) -> bool:
|
| """补齐匿名会话池容量。返回 True 表示补齐成功或无需补齐，False 表示补齐失败（未建出任何会话）。"""
|
| async with self._capacity_lock:
|
| while True:
|
| valid_sessions = self._list_valid_sessions()
|
| need = self.pool_size - len(valid_sessions)
|
| if need <= 0:
|
| return True
|
|
|
| logger.debug(
|
| "[guest_session.ensure_capacity] valid_sessions=%s pool_size=%s need=%s",
|
| len(valid_sessions),
|
| self.pool_size,
|
| need,
|
| )
|
| results = await asyncio.gather(
|
| *[self._create_session(reason="capacity_refill") for _ in range(need)],
|
| return_exceptions=True,
|
| )
|
|
|
| created = 0
|
| duplicate_user_ids: set[str] = set()
|
| errors = set()
|
| with self._lock:
|
| for result in results:
|
| if isinstance(result, GuestSession):
|
| if result.user_id in self._sessions:
|
| duplicate_user_ids.add(result.user_id)
|
| continue
|
| self._sessions[result.user_id] = result
|
| created += 1
|
| elif isinstance(result, Exception):
|
| errors.add(
|
| normalize_httpx_exception(
|
| result,
|
| method="GET",
|
| url=build_upstream_url(AUTH_URL),
|
| context="guest_session.create",
|
| )
|
| )
|
|
|
| if errors:
|
| logger.warning(
|
| "⚠️ 匿名会话池补齐失败 (成功 %s/%s): %s",
|
| created,
|
| need,
|
| ", ".join(sorted(errors)),
|
| )
|
| if duplicate_user_ids:
|
| self._next_create_allowed_at = max(
|
| self._next_create_allowed_at,
|
| time.time() + self.maintenance_interval,
|
| )
|
| logger.warning(
|
| "⚠️ 匿名会话池补齐遇到重复 guest user_id，停止继续补池并冷却: duplicates=%s current_sessions=%s cooldown_seconds=%s",
|
| ", ".join(sorted(duplicate_user_ids)),
|
| len(self._sessions),
|
| self.maintenance_interval,
|
| )
|
|
|
| logger.debug(
|
| "[guest_session.ensure_capacity] refill_result created=%s requested=%s duplicates=%s errors=%s total_sessions=%s",
|
| created,
|
| need,
|
| len(duplicate_user_ids),
|
| len(errors),
|
| len(self._sessions),
|
| )
|
|
|
| if created == 0 or duplicate_user_ids:
|
| return False
|
|
|
| async def _maintenance_loop(self):
|
| """后台维护：回收过期/失效会话，并补齐池容量。"""
|
| consecutive_failures = 0
|
| recovery_wait_seconds = 300
|
| force_sleep_once = 0
|
|
|
| while True:
|
| try:
|
| if force_sleep_once > 0:
|
| sleep_time = force_sleep_once
|
| force_sleep_once = 0
|
| elif consecutive_failures > 0:
|
| sleep_time = self.maintenance_interval * (2 ** (consecutive_failures - 1))
|
| else:
|
| sleep_time = self.maintenance_interval
|
|
|
| await asyncio.sleep(sleep_time)
|
| stale_sessions: List[GuestSession] = []
|
| forced_removed_logs: List[str] = []
|
|
|
| with self._lock:
|
| for user_id, session in list(self._sessions.items()):
|
| force_remove = session.age > (2 * self.session_max_age)
|
| should_remove = (
|
| force_remove
|
| or (
|
| (not session.valid or session.age > self.session_max_age)
|
| and session.active_requests == 0
|
| )
|
| )
|
| if should_remove:
|
| removed = self._sessions.pop(user_id)
|
| stale_sessions.append(removed)
|
| if force_remove:
|
| forced_removed_logs.append(
|
| f"{removed.user_id}(age={removed.age:.1f}s, "
|
| f"active_requests={removed.active_requests}, valid={removed.valid})"
|
| )
|
|
|
| if forced_removed_logs:
|
| logger.warning(
|
| "⚠️ 匿名会话池强制回收超龄会话: " + ", ".join(forced_removed_logs)
|
| )
|
|
|
| await self._delete_sessions_concurrently(stale_sessions)
|
|
|
| success = await self._ensure_capacity()
|
| if success is False:
|
| consecutive_failures += 1
|
| logger.debug(
|
| "[guest_session.maintenance] ensure_capacity_failed consecutive_failures=%s",
|
| consecutive_failures,
|
| )
|
| if consecutive_failures >= self.max_failures:
|
| logger.warning(
|
| f"⚠️ 匿名会话池补齐连续失败 {consecutive_failures} 次达到限值，"
|
| f"进入恢复等待 {recovery_wait_seconds} 秒后继续"
|
| )
|
| consecutive_failures = 0
|
| force_sleep_once = recovery_wait_seconds
|
| else:
|
| next_sleep = self.maintenance_interval * (2 ** (consecutive_failures - 1))
|
| logger.warning(f"⚠️ 匿名会话池补齐连续失败 {consecutive_failures} 次, 下次重试将等待 {next_sleep} 秒")
|
| else:
|
| if consecutive_failures > 0:
|
| logger.debug("[pool] guest session pool maintenance success")
|
| consecutive_failures = 0
|
|
|
| except asyncio.CancelledError:
|
| return
|
| except Exception as exc:
|
| logger.warning(
|
| "⚠️ 匿名会话池后台维护异常",
|
| exc_info=settings.DEBUG_LOGGING,
|
| )
|
| consecutive_failures += 1
|
| if consecutive_failures >= self.max_failures:
|
| logger.warning(
|
| f"⚠️ 匿名会话池后台维护异常连续 {consecutive_failures} 次达到限值，"
|
| f"进入恢复等待 {recovery_wait_seconds} 秒后继续"
|
| )
|
| consecutive_failures = 0
|
| force_sleep_once = recovery_wait_seconds
|
|
|
| async def initialize(self):
|
| """初始化匿名会话池。"""
|
| if self._maintenance_task:
|
| return
|
|
|
| logger.debug(
|
| "[guest_session.initialize] start pool_size=%s maintenance_interval=%s session_max_age=%s",
|
| self.pool_size,
|
| self.maintenance_interval,
|
| self.session_max_age,
|
| )
|
| results = await asyncio.gather(
|
| *[self._create_session(reason="startup_initialize") for _ in range(self.pool_size)],
|
| return_exceptions=True,
|
| )
|
|
|
| created = 0
|
| errors = set()
|
| with self._lock:
|
| for result in results:
|
| if isinstance(result, GuestSession):
|
| self._sessions[result.user_id] = result
|
| created += 1
|
| elif isinstance(result, Exception):
|
| errors.add(
|
| normalize_httpx_exception(
|
| result,
|
| method="GET",
|
| url=build_upstream_url(AUTH_URL),
|
| context="guest_session.create",
|
| )
|
| )
|
|
|
| if errors:
|
| logger.warning(
|
| "⚠️ 匿名会话池初始化失败 (成功 %s/%s): %s",
|
| created,
|
| self.pool_size,
|
| ", ".join(sorted(errors)),
|
| )
|
|
|
| if created == 0:
|
| try:
|
| fallback = await self._create_session(reason="startup_fallback")
|
| with self._lock:
|
| self._sessions[fallback.user_id] = fallback
|
| created = 1
|
| logger.debug("[pool] guest session pool success initialize")
|
| except Exception as e:
|
| log_exception(
|
| logger,
|
| "[pool] guest session pool initialize failed，retrying in background"
|
| )
|
| else:
|
| logger.debug("[pool] guest session pool initialize success, current capacity: %s", created)
|
|
|
| self._maintenance_task = asyncio.create_task(self._maintenance_loop())
|
|
|
| async def close(self):
|
| """关闭匿名会话池。"""
|
| if self._maintenance_task:
|
| self._maintenance_task.cancel()
|
| try:
|
| await self._maintenance_task
|
| except asyncio.CancelledError:
|
| pass
|
| self._maintenance_task = None
|
|
|
| with self._lock:
|
| sessions = list(self._sessions.values())
|
| self._sessions.clear()
|
|
|
| await self._wait_background_tasks(cancel=True)
|
| idle_sessions = [session for session in sessions if session.active_requests == 0]
|
| try:
|
| await asyncio.wait_for(self._delete_sessions_concurrently(idle_sessions), timeout=3.0)
|
| except asyncio.TimeoutError:
|
| logger.warning("⚠️ 清理匿名会话记录超时，强制关闭")
|
| except Exception as e:
|
| logger.warning(
|
| "⚠️ 清理匿名会话记录异常",
|
| exc_info=settings.DEBUG_LOGGING,
|
| )
|
| finally:
|
| await self._http_clients.close()
|
|
|
| async def acquire(
|
| self,
|
| exclude_user_ids: Optional[Set[str]] = None,
|
| ) -> GuestSession:
|
| """按最小忙碌度获取一个可用匿名会话。"""
|
| excluded = exclude_user_ids or set()
|
| max_sessions = self.pool_size * 3
|
|
|
| while True:
|
| candidates = self._list_valid_sessions(exclude_user_ids=excluded)
|
| if candidates:
|
| session = min(
|
| candidates,
|
| key=lambda item: (item.active_requests, item.created_at),
|
| )
|
| with self._lock:
|
| current = self._sessions.get(session.user_id)
|
| if current and current.valid and current.user_id not in excluded:
|
| current.active_requests += 1
|
| logger.debug(
|
| "[guest_session.acquire] reuse user_id=%s active_requests=%s valid_sessions=%s",
|
| current.user_id,
|
| current.active_requests,
|
| len(candidates),
|
| )
|
| return current
|
|
|
| with self._lock:
|
| current_size = len(self._sessions)
|
| if current_size >= max_sessions:
|
| raise RuntimeError(
|
| f"匿名会话池容量超限: current={current_size}, max={max_sessions}"
|
| )
|
|
|
| logger.debug(
|
| "[guest_session.acquire] no reusable session, creating fallback current_size=%s excluded=%s",
|
| current_size,
|
| len(excluded),
|
| )
|
| new_session = await self._create_session(reason="acquire_fallback")
|
| if new_session.user_id in excluded:
|
| await self._delete_all_chats(new_session)
|
| continue
|
|
|
| with self._lock:
|
| current_size = len(self._sessions)
|
| if current_size >= max_sessions:
|
| overflow_message = (
|
| f"匿名会话池容量超限: current={current_size}, max={max_sessions}"
|
| )
|
| else:
|
| new_session.active_requests = 1
|
| self._sessions[new_session.user_id] = new_session
|
| return new_session
|
|
|
| await self._delete_all_chats(new_session)
|
| logger.warning(f"⚠️ {overflow_message}, 已放弃新建会话: {new_session.user_id}")
|
| raise RuntimeError(overflow_message)
|
|
|
| def release(self, user_id: str):
|
| """释放一个匿名会话占用。"""
|
| with self._lock:
|
| session = self._sessions.get(user_id)
|
| if session:
|
| session.active_requests = max(0, session.active_requests - 1)
|
| logger.debug(
|
| "[guest_session.release] user_id=%s active_requests=%s",
|
| user_id,
|
| session.active_requests,
|
| )
|
|
|
| async def report_failure(self, user_id: Optional[str] = None):
|
| """标记匿名会话失效，并尝试补一个新会话。"""
|
| session: Optional[GuestSession] = None
|
|
|
| if user_id:
|
| with self._lock:
|
| session = self._sessions.pop(user_id, None)
|
| if session:
|
| session.valid = False
|
| session.failure_count += 1
|
| session.last_failure_time = time.time()
|
| session.active_requests = 0
|
|
|
| if session:
|
| self._track_background_task(self._delete_all_chats(session))
|
| logger.debug("已淘汰匿名会话: %s", session.user_id)
|
|
|
| logger.debug(
|
| "[guest_session.report_failure] user_id=%s triggering_capacity_refill=%s",
|
| user_id,
|
| session is not None,
|
| )
|
| await self._ensure_capacity()
|
|
|
| async def refresh_auth(self, failed_user_id: Optional[str] = None):
|
| """兼容 glm-demo 命名：刷新匿名会话。"""
|
| await self.report_failure(failed_user_id)
|
|
|
| async def cleanup_idle_chats(self):
|
| """清理当前空闲匿名会话的聊天记录。"""
|
| with self._lock:
|
| idle_sessions = [
|
| session
|
| for session in self._sessions.values()
|
| if session.valid and session.active_requests == 0
|
| ]
|
|
|
| await self._delete_sessions_concurrently(idle_sessions)
|
|
|
| def get_pool_status(self) -> Dict[str, int]:
|
| """获取匿名会话池状态。"""
|
| with self._lock:
|
| sessions = list(self._sessions.values())
|
|
|
| valid_sessions = [
|
| session for session in sessions if session.valid and session.age < self.session_max_age
|
| ]
|
| busy_sessions = [session for session in valid_sessions if session.active_requests > 0]
|
|
|
| return {
|
| "total_sessions": len(sessions),
|
| "valid_sessions": len(valid_sessions),
|
| "available_sessions": len(valid_sessions),
|
| "busy_sessions": len(busy_sessions),
|
| "expired_sessions": len(
|
| [session for session in sessions if session.age >= self.session_max_age]
|
| ),
|
| }
|
|
|
|
|
| _guest_session_pool: Optional[GuestSessionPool] = None
|
| _guest_pool_lock = Lock()
|
|
|
|
|
| def get_guest_session_pool() -> Optional[GuestSessionPool]:
|
| """获取全局匿名会话池。"""
|
| return _guest_session_pool
|
|
|
|
|
| async def initialize_guest_session_pool(
|
| pool_size: int = 3,
|
| session_max_age: int = 480,
|
| maintenance_interval: int = 30,
|
| max_failures: int = 10,
|
| ) -> GuestSessionPool:
|
| """初始化全局匿名会话池。"""
|
| global _guest_session_pool
|
|
|
| if not settings.ANONYMOUS_MODE:
|
| raise RuntimeError("ANONYMOUS_MODE is disabled")
|
|
|
| with _guest_pool_lock:
|
| if _guest_session_pool is None:
|
| _guest_session_pool = GuestSessionPool(
|
| pool_size=pool_size,
|
| session_max_age=session_max_age,
|
| maintenance_interval=maintenance_interval,
|
| max_failures=max_failures,
|
| )
|
| pool = _guest_session_pool
|
|
|
| await pool.initialize()
|
| return pool
|
|
|
|
|
| async def close_guest_session_pool():
|
| """关闭全局匿名会话池。"""
|
| global _guest_session_pool
|
|
|
| with _guest_pool_lock:
|
| pool = _guest_session_pool
|
| _guest_session_pool = None
|
|
|
| if pool:
|
| await pool.close()
|
|
|
