p2p: multi-hub/multi-customer group (lobby member list + per-pair P2PSocket rooms); hub exposes /p2p/hubs

frontends/hub_p2p.py

@@ -1,21 +1,26 @@
 """Optional P2P phone pairing plug-in for hub.py."""
 import asyncio
 import os
-import random
-from collections import deque
+import socket
 from pathlib import Path
 
 from fastapi import Response
 from fastapi.responses import JSONResponse
-from p2p_ws_client import HTTPExporter, connect_saved, create_code, load_room, save_room
+from p2p_ws_client import (
+    Group, HTTPExporter, HTTPLocalProxy, connect_saved, create_code, load_room,
+    retry_forever, save_room,
+)
 
 
 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 消费者，故同一进程以两个成员身份入组，每对连接角色唯一。
+    member = os.environ.get("GA_P2P_MEMBER", f"{socket.gethostname()}-{web_port}")
     state = {"task": None, "invite": None, "status": "idle", "error": None,
-             "started": False}
+             "started": False, "group": None, "hubs": {}}
     app.state.p2p_pair_open = True
 
     async def export(ws):
@@ -30,38 +35,49 @@ def install(app, *, web_port, token, here):
             await exporter.close()
 
     async def reconnect():
-        # 重连节奏三层:
-        # 1) 本轮尝试活过 STABLE 秒(正常会话被掉线/被顶号, 或已久候后信令断) -> 0.2s 秒级重连;
-        # 2) 秒断/连不上 -> 抖动指数退避 1,2,4..60s, 防"配对即断"空转;
-        # 3) 兜底硬上限: 任意 WINDOW 秒内最多 BUDGET 次建连, 远低于信令服务器
-        #    每 IP 2/s、突发 20 的限速, 免得把同出口 IP 的手机一起 429 掉。
-        STABLE, WINDOW, BUDGET = 30, 60, 8
-        loop = asyncio.get_running_loop()
-        delay, recent = 0, deque()
-        while True:
-            now = loop.time()
-            while recent and now - recent[0] >= WINDOW:
-                recent.popleft()
-            if len(recent) >= BUDGET:
-                await asyncio.sleep(recent[0] + WINDOW - now)
-                continue
-            recent.append(now)
-            t0 = now
+        async def once():
+            state["status"], state["error"] = "reconnecting", None
+            # 已配对候房不超时；信令断开仍立即失败并走退避。
+            ws = await connect_saved(name, path=rooms, direct_timeout=15, peer_timeout=None)
             try:
-                state["status"], state["error"] = "reconnecting", None
-                # 已配对候房不超时；信令断开仍立即失败并走退避。
-                ws = await connect_saved(name, path=rooms, direct_timeout=15, peer_timeout=None)
-                try:
-                    await export(ws)
-                finally:
-                    await ws.close()  # 不关会泄漏信令连接: 旧连接与新连接自配对占满房间
-            except asyncio.CancelledError:
-                raise
-            except Exception as exc:
-                state["status"], state["error"] = "error", str(exc)
-            # 抖动也避免服务端重启时上千个 hub 同一秒齐涌。
-            delay = 0 if loop.time() - t0 >= 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 export(ws)
+            finally:
+                await ws.close()  # 不关会泄漏信令连接: 旧连接与新连接自配对占满房间
+
+        def failed(exc):
+            state["status"], state["error"] = "error", str(exc)
+
+        await retry_forever(once, failed)
+
+    async def serve_customer(ws, peer):
+        exporter = await HTTPExporter(
+            ws, f"http://127.0.0.1:{web_port}", allow=("/api/",), query={"t": token},
+        ).start()
+        try:
+            await exporter.wait_closed()
+        finally:
+            await exporter.close()
+
+    async def use_hub(ws, peer):
+        proxy = await HTTPLocalProxy(ws).start()
+        state["hubs"][peer[4:]] = proxy.url
+        try:
+            await proxy.wait_closed()
+        finally:
+            if state["hubs"].get(peer[4:]) == proxy.url:
+                del state["hubs"][peer[4:]]
+            await proxy.close()
+
+    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]
 
     async def new_pair():
         try:
@@ -71,6 +87,7 @@ def install(app, *, web_port, token, here):
             state["status"], state["error"] = "waiting", None
             ws = await invite.connect(direct_timeout=15)
             save_room(signal, invite.room, name=name, path=rooms)
+            start_group()
             try:
                 await export(ws)
             finally:
@@ -100,6 +117,11 @@ document.querySelector('#state').textContent=s.status+(s.error?' : '+s.error:'')
 if(s.status!=='connected')setTimeout(poll,1000)}catch(e){document.querySelector('#state').textContent=e}}
 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"])
+
     @app.get("/pair/status")
     async def pair_status():
         invite = state["invite"]
@@ -119,6 +141,7 @@ poll();</script>''', media_type="text/html")
         except KeyError:
             return
         state["task"] = asyncio.create_task(reconnect())
+        start_group()
 
     app.add_event_handler("startup", startup)
     return state

frontends/p2p_ws_client.py

@@ -8,11 +8,13 @@ import json
 import logging
 import math
 import os
+import random
 import re
 import time
 import urllib.error
 import urllib.request
 import zlib
+from collections import deque
 from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
@@ -1080,12 +1082,104 @@ async def connect_code(
     )
 
 
+# ---- 多人组：同一 UUID 即全部权限；名单走大厅，数据走两两派生的普通 1v1 房间 ----
+
+async def retry_forever(attempt, on_error=None, *, stable=30, window=60, budget=8):
+    """永久重试 attempt()。节奏三层:
+    1) 本轮活过 stable 秒(正常会话被掉线/被顶号) -> 0.2s 秒级重连;
+    2) 秒断/连不上 -> 抖动指数退避 1,2,4..60s, 防"配对即断"空转;
+    3) 兜底: 任意 window 秒内最多 budget 次, 远低于信令服务器每 IP 限速。
+    抖动也避免服务端重启时大量客户端同一秒齐涌。"""
+    loop = asyncio.get_running_loop()
+    delay, recent = 0, deque()
+    while True:
+        now = loop.time()
+        while recent and now - recent[0] >= window:
+            recent.popleft()
+        if len(recent) >= budget:
+            await asyncio.sleep(recent[0] + window - now)
+            continue
+        recent.append(now)
+        try:
+            await attempt()
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            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))
+
+
+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()
+
+
+class Group:
+    """以成员名 me 加入 UUID room 的多人组。
+
+    大厅连接只收在线名单(成员名列表，变化时推送，无轮询)；对每个在线且 want(peer)
+    的成员各维护一条普通 P2PSocket，并交给 serve(socket, peer) 处理直到断开。
+    成员离开才取消其连接；大厅自身断线不影响已建立的两两连接。
+    同名重入会顶掉旧连接，因此每个进程须用唯一成员名。
+    """
+
+    def __init__(self, signal_url, room, me, serve, *, want=lambda peer: True,
+                 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.members: set[str] = set()
+        self._pairs: dict[str, asyncio.Task] = {}
+
+    async def run(self):
+        try:
+            await retry_forever(self._lobby)
+        finally:
+            for task in self._pairs.values():
+                task.cancel()
+
+    async def _lobby(self):
+        room = hashlib.sha256(f"{self.room}:lobby".encode()).hexdigest()
+        sep = "&" if "?" in self.signal_url else "?"
+        url = f"{self.signal_url}{sep}{urlencode({'room': room, 'member': self.me})}"
+        seq = 0  # 服务端广播在锁外发送可能乱序，只认更大的 seq
+        async with websockets.connect(
+            signed_url(url, "ws", room, self.access_key), ping_interval=15, ping_timeout=10,
+        ) as ws:
+            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})
+
+    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)))
+
+    async def _pair(self, 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,
+        )
+        try:
+            await self.serve(socket, peer)
+        finally:
+            await socket.close()
+
+
 connect = P2PSocket.connect
 
 __all__ = [
     "DEFAULT_ACCESS_KEY", "ROOMS_FILE", "Invite", "P2PSocket", "Signal",
     "HTTPExporter", "HTTPLocalProxy", "export_http", "mount_http",
     "connect", "connect_code", "connect_saved", "create_code", "load_room",
-    "save_room", "signed_url",
+    "save_room", "signed_url", "Group", "pair_room", "retry_forever",
 ]