From a9ff7cb498b339bb036fd67941602bb9f0c65707 Mon Sep 17 00:00:00 2001 From: host Date: Tue, 27 Oct 2026 22:16:43 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E6=B3=A8=E5=86=8C=E5=BC=80=E9=80=9A?= =?UTF-8?q?=E5=A4=8D=E7=94=A8=E8=BF=9B=E7=A8=8B=E5=86=85=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E5=91=98=E4=BC=9A=E8=AF=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实测（真网关）：单次开通链路约 0.29s，其中管理员登录 111ms、 建号 80ms、用户登录 79ms、查组 5ms、发 key 10ms、授余额 9ms。 外部 OAuth 两步为必需串行调用；查组/授余额仅个位数毫秒， 不值得为其加缓存复杂度，故本批只做一项：同一进程内第二 次起的开通复用管理员 token 元组（实例不跨线程，401 时走 刷新->重登录自愈，密码轮换后自动用新密码），免一次登录。 真网关对照：冷开通 0.294s，热开通 0.179s（约 -39%）， 对照组每次新登录两次均为 0.255s 左右。 测试：新增共享会话复用/过期重取 2 个；全量 cloud 794 （4 跳过）、host 69、deploy 19 全绿。 --- cloud/sub2api.py | 62 +++++++++++++++++++++++++++++++++++-- cloud/tests/test_sub2api.py | 51 ++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 3 deletions(-) diff --git a/cloud/sub2api.py b/cloud/sub2api.py index 3fddbe7..4583de4 100644 --- a/cloud/sub2api.py +++ b/cloud/sub2api.py @@ -16,6 +16,7 @@ import json import os import secrets +import threading import time import urllib.error import urllib.request @@ -112,8 +113,57 @@ def get_user_gateway_key(conn, user_id): return None +# 进程内共享管理员会话：(origin, email) -> (token, refresh, expires_at)。 +# 只复用 token 元组，不共用实例（实例不跨线程）；注册与首任务开通 +# 每次新建实例都要登录一次（实测约 111ms，占开通链路约 1/3）， +# 共享后同一进程内第二次起免登录。401/刷新失败会回落到重登录， +# 密码轮换后下一次登录即用新密码刷新缓存，不会卡死。 +_SHARED_ADMIN = {} +_SHARED_ADMIN_LOCK = threading.Lock() + + +def _store_shared_session(admin): + key = getattr(admin, "_shared_key", None) + if key is None or not admin._token: + return + with _SHARED_ADMIN_LOCK: + _SHARED_ADMIN[key] = (admin._token, admin._refresh, + admin._expires_at) + + +def _evict_shared_session(admin): + key = getattr(admin, "_shared_key", None) + if key is None: + return + with _SHARED_ADMIN_LOCK: + _SHARED_ADMIN.pop(key, None) + + +def shared_admin_session(request=None): + """带进程内 token 复用的管理员会话（生产开通路径默认用它）。 + + 缓存命中且未过期时直接播种 token，免一次管理员登录； + 未命中/已过期时照常登录，成功后写回缓存。调用方显式传 + admin 的单测不受影响（不走这条工厂）。 + """ + admin = Sub2apiAdmin.from_env(request=request) + key = (admin.origin, admin.email) + admin._shared_key = key + with _SHARED_ADMIN_LOCK: + cached = _SHARED_ADMIN.get(key) + if cached and cached[2] > time.time(): + admin._token, admin._refresh, admin._expires_at = cached + return admin + + class Sub2apiAdmin: - """管理员会话：登录、建/删用户、查组。token 自动刷新一次。""" + """管理员会话：登录、建/删用户、查组。token 自动刷新一次。 + + 每次实例只在进程内复用 token 元组（见 shared_admin_session）： + 实例本身不跨线程共用，只复用 (token, refresh, expires_at)， + 401 时走已有刷新->重登录链路自愈，密码轮换后下一次登录 + 即用新密码并刷新缓存，不会卡在旧 token 上。 + """ def __init__(self, origin, email, password, request=None): if not origin: @@ -127,6 +177,7 @@ class Sub2apiAdmin: self._token = None self._refresh = None self._expires_at = 0 + self._shared_key = None @classmethod def from_env(cls, request=None): @@ -141,6 +192,7 @@ class Sub2apiAdmin: {"email": self.email, "password": self.password}) payload = (data or {}).get("data") or {} if st != 200 or not payload.get("access_token"): + _evict_shared_session(self) raise Sub2apiError( "sub2api 管理员登录失败: %s" % ((data or {}).get("message") or st), status=st) @@ -151,6 +203,7 @@ class Sub2apiAdmin: except (TypeError, ValueError): ttl = 3600 self._expires_at = time.time() + max(60, ttl - 60) + _store_shared_session(self) def _refresh_token(self): if not self._refresh: @@ -169,6 +222,7 @@ class Sub2apiAdmin: except (TypeError, ValueError): ttl = 3600 self._expires_at = time.time() + max(60, ttl - 60) + _store_shared_session(self) return self._login() @@ -310,7 +364,9 @@ def ensure_user_gateway_key(conn, user_id, admin=None, request=None): 入库失败时尽力清理刚建的 key 与账号（避免孤儿 key），再抛原错； 授余额失败同样走这条清理（余额为 0 的 key 在网关侧直接 403， 留着没有意义）。同一用户并发开通时建号撞 email 已存在， - 则轮询复用胜出者入库的行。调用方负责 conn 的提交语义：本函数在入库后 + 则轮询复用胜出者入库的行。管理员工单默认走进程内共享会话 + （同一进程内第二次起免管理员登录，见 shared_admin_session）。 + 调用方负责 conn 的提交语义：本函数在入库后 commit（开通是独立事务，不跟调用方的业务事务绑在一起）。 """ ensure_gateway_keys_table(conn) @@ -320,7 +376,7 @@ def ensure_user_gateway_key(conn, user_id, admin=None, request=None): origin = gateway_origin() if not origin: raise Sub2apiError("SUB2API_BASE_URL 未配置，无法开通网关账号") - admin = admin or Sub2apiAdmin.from_env(request=request) + admin = admin or shared_admin_session(request=request) req = request or _default_request email = "codex-%s@codex.local" % user_id username = "codex-" + str(user_id)[:12] diff --git a/cloud/tests/test_sub2api.py b/cloud/tests/test_sub2api.py index 7533c1b..e89c1d2 100644 --- a/cloud/tests/test_sub2api.py +++ b/cloud/tests/test_sub2api.py @@ -216,6 +216,57 @@ class TestEnsureGatewayKey(unittest.TestCase): " WHERE user_id = ?", ("user-race2",)).fetchone() self.assertEqual(row[0], 0) + def _admin_env(self): + old = {k: os.environ.get(k) for k in + ("SUB2API_ADMIN_EMAIL", "SUB2API_ADMIN_PASSWORD")} + os.environ["SUB2API_ADMIN_EMAIL"] = "adm@x" + os.environ["SUB2API_ADMIN_PASSWORD"] = "pw" + sub2api._SHARED_ADMIN.clear() + return old + + def _restore_admin_env(self, old): + for k, v in old.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + sub2api._SHARED_ADMIN.clear() + + def _admin_login_count(self): + return len([ + c for c in self.api.calls + if c[0] == "POST" and c[1] == "/api/v1/auth/login" + and (c[2] or {}).get("email") == "adm@x"]) + + def test_shared_admin_session_reuses_login(self): + old = self._admin_env() + try: + conn1 = _conn() + key1 = ensure_user_gateway_key(conn1, "user-s1", + request=self.api) + self.assertTrue(key1.startswith("sk-probe-")) + conn2 = _conn() + key2 = ensure_user_gateway_key(conn2, "user-s2", + request=self.api) + self.assertTrue(key2.startswith("sk-probe-")) + # 同一进程内第二次开通不再管理员登录（只剩首次那一次）。 + self.assertEqual(self._admin_login_count(), 1) + finally: + self._restore_admin_env(old) + + def test_shared_admin_session_expired_refetches(self): + old = self._admin_env() + sub2api._SHARED_ADMIN[("http://x:28084", "adm@x")] = ( + "old-tok", "old-ref", 0) + try: + conn = _conn() + key = ensure_user_gateway_key(conn, "user-s3", + request=self.api) + self.assertTrue(key.startswith("sk-probe-")) + self.assertEqual(self._admin_login_count(), 1) + finally: + self._restore_admin_env(old) + class TestProviderCommand(unittest.TestCase): def test_server_command_sub2api(self):