p2p: on-demand pair connections via lobby call; hub_p2p single Group + /p2p/open + idle reclaim

frontends/hub_p2p.py

@@ -2,6 +2,7 @@
 import asyncio
 import os
 import socket
+import time
 from pathlib import Path
 
 from fastapi import Response
@@ -16,11 +17,10 @@ def install(app, *, web_port, token, here):
     signal = os.environ.get("GA_P2P_SIGNAL", "ws://47.101.182.29:49157/ws")
     name = os.environ.get("GA_P2P_NAME", "ga-hub-phone")
     rooms = Path(os.environ.get("GA_P2P_ROOMS", "~/.p2p_ws/rooms.json")).expanduser()
-    # 多人组成员名带角色前缀：hub: 导出本机 API，cust: 代理对方 API。一条 P2PSocket
-    # 只能有一个 HTTP 消费者，故同一进程以两个成员身份入组，每对连接角色唯一。
+    # 组内每个 hub 一个成员名；连接按需建立：主叫代理对方 API，被叫导出本机 API。
     member = os.environ.get("GA_P2P_MEMBER", f"{socket.gethostname()}-{web_port}")
     state = {"task": None, "invite": None, "status": "idle", "error": None,
-             "started": False, "group": None, "hubs": {}}
+             "started": False, "group": None, "proxies": {}}
     app.state.p2p_pair_open = True
 
     async def export(ws):
@@ -60,24 +60,29 @@ def install(app, *, web_port, token, here):
 
     async def use_hub(ws, peer):
         proxy = await HTTPLocalProxy(ws).start()
-        state["hubs"][peer[4:]] = proxy.url
+        state["proxies"][peer] = proxy
         try:
             await proxy.wait_closed()
         finally:
-            if state["hubs"].get(peer[4:]) == proxy.url:
-                del state["hubs"][peer[4:]]
+            if state["proxies"].get(peer) is proxy:
+                del state["proxies"][peer]
             await proxy.close()
 
+    async def reap_idle(group):
+        while True:  # 10 分钟没请求的代理断开，下次 /p2p/open 再按需重连
+            await asyncio.sleep(60)
+            for peer, proxy in list(state["proxies"].items()):
+                if time.monotonic() - proxy.last_used > 600:
+                    group.close(peer)
+
     def start_group():
         if state["group"]:
             return
         room = load_room(name, path=rooms)
-        hub, cust = f"hub:{member}", f"cust:{member}"
-        roles = ((hub, serve_customer, "cust:"), (cust, use_hub, "hub:"))
-        state["group"] = [asyncio.create_task(Group(
-            room["signal_url"], room["room"], me, serve,
-            want=lambda p, other=other: p.startswith(other) and p[len(other):] != member,
-        ).run()) for me, serve, other in roles]
+        group = state["group"] = Group(
+            room["signal_url"], room["room"], member, serve_customer, use_hub,
+        )
+        state["group_tasks"] = [asyncio.create_task(group.run()), asyncio.create_task(reap_idle(group))]
 
     async def new_pair():
         try:
@@ -119,8 +124,24 @@ poll();</script>''', media_type="text/html")
 
     @app.get("/p2p/hubs")
     async def p2p_hubs():
-        """同组其他 hub 的本地代理地址 {成员名: http://127.0.0.1:端口}，按原 /api/ 访问。"""
-        return JSONResponse(state["hubs"])
+        """同组在线 hub {成员名: 本地代理地址，未连接为 null}。"""
+        group, proxies = state["group"], state["proxies"]
+        members = sorted(group.members) if group else []
+        return JSONResponse({m: getattr(proxies.get(m), "url", None) for m in members})
+
+    @app.get("/p2p/open")
+    async def p2p_open(hub: str):
+        """按需连接 hub，返回本地代理地址，之后按原 /api/ 访问。"""
+        group = state["group"]
+        if not group or hub not in group.members:
+            return JSONResponse({"error": "hub offline"}, status_code=404)
+        group.open(hub)
+        for _ in range(200):
+            if proxy := state["proxies"].get(hub):
+                proxy.last_used = time.monotonic()
+                return JSONResponse({"url": proxy.url})
+            await asyncio.sleep(0.1)
+        return JSONResponse({"error": "connect timeout"}, status_code=504)
 
     @app.get("/pair/status")
     async def pair_status():

frontends/p2p_ws_client.py

@@ -826,6 +826,7 @@ class HTTPLocalProxy:
         self._app = self._runner = self._site = self._receiver = None
         self._pending, self._closed = {}, asyncio.Event()
         self._counter = 0
+        self.last_used = time.monotonic()  # 供上层回收空闲代理
 
     async def start(self):
         if self._runner:
@@ -868,6 +869,7 @@ class HTTPLocalProxy:
         return {"Access-Control-Allow-Origin": origin, "Vary": "Origin"} if allowed else {}
 
     async def _handle(self, request):
+        self.last_used = time.monotonic()
         _, web, _ = _http_modules()
         cors_headers = self._cors_headers(request)
         if request.method == "OPTIONS":
@@ -1086,7 +1088,7 @@ async def connect_code(
 
 async def retry_forever(attempt, on_error=None, *, stable=30, window=60, budget=8):
     """永久重试 attempt()。节奏三层:
-    1) 本轮活过 stable 秒(正常会话被掉线/被顶号) -> 0.2s 秒级重连;
+    1) 本轮活过 stable 秒(正常会话被掉线/被顶号) -> 0.2~1s 秒级重连;
     2) 秒断/连不上 -> 抖动指数退避 1,2,4..60s, 防"配对即断"空转;
     3) 兜底: 任意 window 秒内最多 budget 次, 远低于信令服务器每 IP 限速。
     抖动也避免服务端重启时大量客户端同一秒齐涌。"""
@@ -1108,36 +1110,47 @@ async def retry_forever(attempt, on_error=None, *, stable=30, window=60, budget=
             if on_error:
                 on_error(exc)
         delay = 0 if loop.time() - now >= stable else min(max(delay * 2, 1), 60)
-        await asyncio.sleep(0.2 if delay == 0 else delay * random.uniform(0.5, 1.5))
+        await asyncio.sleep(random.uniform(0.2, 1) if delay == 0 else delay * random.uniform(0.5, 1.5))
 
 
-def pair_room(room: str, a: str, b: str) -> str:
-    """两成员的专属房间：双方独立算出同一值，且不同于原 1v1 房间。"""
-    x, y = sorted((a, b))
-    return hashlib.sha256(f"{room}\n{x}\n{y}".encode()).hexdigest()
+def pair_room(room: str, caller: str, callee: str) -> str:
+    """主叫到被叫的专属房间：双方独立算出同一值；有方向，A 呼 B 与 B 呼 A 互不干扰。"""
+    return hashlib.sha256(f"{room}\n{caller}\n{callee}".encode()).hexdigest()
 
 
 class Group:
     """以成员名 me 加入 UUID room 的多人组。
 
-    大厅连接只收在线名单(成员名列表，变化时推送，无轮询)；对每个在线且 want(peer)
-    的成员各维护一条普通 P2PSocket，并交给 serve(socket, peer) 处理直到断开。
-    成员离开才取消其连接；大厅自身断线不影响已建立的两两连接。
+    大厅连接收在线名单(变化时推送，无轮询)并转发呼叫。两两连接按需建立：
+    open(peer) 经大厅呼叫对方，双方进入 pair_room(主叫, 被叫) 的普通 P2PSocket，
+    主叫交给 use(socket, peer)，被叫交给 serve(socket, peer)。主叫断线自动重呼；
+    close(peer) 停止，对方离组时暂停、回来后自动恢复。不 open 就不占连接。
     同名重入会顶掉旧连接，因此每个进程须用唯一成员名。
     """
 
-    def __init__(self, signal_url, room, me, serve, *, want=lambda peer: True,
-                 access_key=DEFAULT_ACCESS_KEY):
+    def __init__(self, signal_url, room, me, serve, use, *, access_key=DEFAULT_ACCESS_KEY):
         self.signal_url, self.room, self.me = signal_url, room, me
-        self.serve, self.want, self.access_key = serve, want, access_key
+        self.serve, self.use, self.access_key = serve, use, access_key
         self.members: set[str] = set()
-        self._pairs: dict[str, asyncio.Task] = {}
+        self._ws = None
+        self._wanted: set[str] = set()
+        self._out: dict[str, asyncio.Task] = {}  # 本端主叫
+        self._in: dict[str, asyncio.Task] = {}   # 对方呼入
 
     async def run(self):
         try:
             await retry_forever(self._lobby)
         finally:
-            for task in self._pairs.values():
+            for task in [*self._out.values(), *self._in.values()]:
+                task.cancel()
+
+    def open(self, peer):
+        self._wanted.add(peer)
+        self._sync(self.members)
+
+    def close(self, peer):
+        self._wanted.discard(peer)
+        if task := self._out.pop(peer, None):
             task.cancel()
 
     async def _lobby(self):
@@ -1148,28 +1161,51 @@ class Group:
         async with websockets.connect(
             signed_url(url, "ws", room, self.access_key), ping_interval=15, ping_timeout=10,
         ) as ws:
+            self._ws = ws
+            try:
                 async for raw in ws:
                     msg = json.loads(raw)
                     if msg.get("type") == "members" and msg["seq"] > seq:
                         seq = msg["seq"]
                         self._sync(set(msg["members"]) - {self.me})
+                    elif msg.get("type") == "call":
+                        self._answer(msg["from"])
+            finally:
+                self._ws = None
 
     def _sync(self, members):
         self.members = members
-        for peer in set(self._pairs) - members:
-            self._pairs.pop(peer).cancel()
-        for peer in members - set(self._pairs):
-            if self.want(peer):
-                self._pairs[peer] = asyncio.create_task(
-                    retry_forever(lambda peer=peer: self._pair(peer)))
+        for tasks in (self._out, self._in):
+            for peer in set(tasks) - members:
+                tasks.pop(peer).cancel()
+        for peer in (self._wanted & members) - set(self._out):
+            self._out[peer] = asyncio.create_task(retry_forever(lambda peer=peer: self._call(peer)))
 
-    async def _pair(self, peer):
+    async def _call(self, peer):
+        if self._ws is None:
+            raise ConnectionError("lobby offline")
+        await self._ws.send(json.dumps({"type": "call", "to": peer}))
+        await self._link(pair_room(self.room, self.me, peer), self.use, peer)
+
+    def _answer(self, peer):
+        """每次呼叫只接一次；对方重呼时旧连接必已失效，直接替换。"""
+        if old := self._in.pop(peer, None):
+            old.cancel()
+
+        async def once():
+            with suppress(Exception):
+                await self._link(pair_room(self.room, peer, self.me), self.serve, peer)
+            if self._in.get(peer) is task:
+                del self._in[peer]
+        task = self._in[peer] = asyncio.create_task(once())
+
+    async def _link(self, room, handler, peer):
         socket = await P2PSocket.connect(
-            self.signal_url, pair_room(self.room, self.me, peer),
-            access_key=self.access_key, direct_timeout=15, peer_timeout=None,
+            self.signal_url, room, access_key=self.access_key,
+            direct_timeout=15, peer_timeout=15,
         )
         try:
-            await self.serve(socket, peer)
+            await handler(socket, peer)
         finally:
             await socket.close()