From 2930e4c788b7502d2ecc125a2168e778498eb690 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Thu, 20 Aug 2026 19:53:58 +0800 Subject: [PATCH] feat: add bounded selected-edge TURN recovery (#76) Add one default-off, generation-bound relay-only rebuild for the original failed peer edge after Viewer-root SFU exhaustion while keeping ordinary Web and Native ICE STUN-only. --- .env.example | 6 + .../0005-automatic-hybrid-media-routing.md | 41 +-- docs/deployment.md | 32 +- docs/project-memory.md | 6 +- docs/research/advanced-peer-distribution.md | 2 +- docs/research/built-in-peer-ice-turn.md | 14 +- docs/research/low-server-media-routes.md | 6 +- docs/status.md | 9 +- docs/方案设计.md | 29 +- src/client/pages/HostPage.tsx | 46 ++- src/client/pages/ViewerPage.tsx | 69 +++- src/client/webrtc/host-peer.ts | 11 +- src/client/webrtc/viewer-peer.ts | 26 +- src/client/webrtc/viewer-relay.ts | 48 ++- src/server/app.ts | 1 + src/server/config.ts | 71 +++- src/server/hybrid-media-router.ts | 314 ++++++++++++++++++ src/server/selected-edge-turn.ts | 59 ++++ src/server/signaling.ts | 25 +- src/shared/protocol.ts | 53 +++ tests/host-peer-tracks.test.ts | 44 +++ tests/selected-edge-turn.test.ts | 42 +++ tests/server-config.test.ts | 49 ++- tests/server-signal.test.ts | 245 ++++++++++++++ 24 files changed, 1164 insertions(+), 84 deletions(-) create mode 100644 src/server/selected-edge-turn.ts create mode 100644 tests/selected-edge-turn.test.ts diff --git a/.env.example b/.env.example index 5e10814..79038de 100644 --- a/.env.example +++ b/.env.example @@ -31,6 +31,12 @@ LIVEKIT_API_SECRET= # Effective only when the complete LiveKit tuple is configured. Valid range: 1-2. MAX_SFU_ROOTS_PER_ROOM=2 +# Optional one-shot TURN/UDP transport for one controller-selected edge after +# peer and SFU recovery are exhausted. Set all three values together. +# SELECTED_EDGE_TURN_URLS=turn:turn.example.com:3478?transport=udp +# SELECTED_EDGE_TURN_SHARED_SECRET=replace-with-32-or-more-independent-random-bytes +# SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS=120 + # Ordinary peer ICE discovery. Production requires at least one STUN URL. # Separate multiple URLs with commas. STUN_URLS= diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index eeb8561..8d842e6 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -80,8 +80,9 @@ adds the bounded standby prewarm below. Production enables them only for persistent room `1`; other rooms stay ordinary P2P, and production remains STUN-only. The current source no longer contains the rejected participant-wide TURN config, issuer, capability, refresh wire, or client propagation. Stale -`PEER_ICE_TURN_*` keys fail startup even when blank. Selected-edge TURN remains -unimplemented and undeployed. +`PEER_ICE_TURN_*` keys fail startup even when blank. The source candidate now +implements a complete default-off selected-edge tuple and one post-SFU +relay-only rebuild; it remains undeployed and lacks real TURN/media evidence. The current runtime removes the old all-room coturn contract. Production requires STUN and authenticated ICE snapshots contain only STUN servers in @@ -96,7 +97,7 @@ permanent compatibility branch or advertised current transport. The repository also requires a strict, non-empty `PEER_ASSISTED_ROOM_IDS` deployment allowlist whenever `PEER_ASSISTED_MEDIA=true`. Only exact listed room IDs enter the controller or receive optional LiveKit standby, grants, or a -future selected-edge TURN attempt. All other rooms retain the current ordinary P2P +selected-edge TURN attempt. All other rooms retain the current ordinary P2P authentication shape, STUN-only ICE, and signaling/quality/lifecycle behavior. Missing, blank, malformed, or duplicate entries fail startup; there is no configuration state that enables all rooms. The boundary is a temporary deployment-only validation gate with no browser selector, percentage @@ -217,7 +218,7 @@ Minimal protocol additions: - `route-ready { revision, phase: "prepare" | "active" }`; - `route-failed { revision, phase, connectionId }`; and - `refresh-sfu { revision }`; -- a future controller-selected TURN grant for exactly one current edge; and +- a controller-selected TURN grant for exactly one current edge; and - a matching parent/child rebuild bound to the old and new connection identity. Native v2 retains its strict STUN-only authenticated shape. Pinned LiveKit @@ -227,8 +228,8 @@ publisher or subscriber and is not the selected peer-edge grant. The ordinary authenticated `iceConfig` contains only `iceServers` populated from `STUN_URLS`. The rejected participant-wide tuple, issuer, capability, refresh messages, and client propagation are removed, and stale -`PEER_ICE_TURN_*` keys fail startup. A future selected-edge slice must add its -controller consumer and one-use rebuild atomically; only that selected rebuild +`PEER_ICE_TURN_*` keys fail startup. The selected-edge source slice atomically +adds its controller consumer and one-use rebuild; only that selected rebuild may receive a short-lived relay-only configuration. No credential may enter a URL, log, browser persistence, room data, or SQLite. @@ -287,11 +288,12 @@ complete LiveKit tuple. route. A failure after commit creates a new rollback/failure revision; stale acknowledgements cannot revive the abandoned plan. -8. An active SFU transport may request one fresh short-lived grant. Current code - creates a newer peer baseline and disables SFU if that retry or token issuance - fails. The accepted replacement inserts one selected-edge TURN attempt before - terminal failure; it is not implemented yet. Stop or a new sharing generation - clears the one-shot circuit breaker. +8. An active Viewer SFU root may request one fresh short-lived grant. If its next + failure, or the initial SFU prepare, is exhausted, the router grants at most + one relay-only rebuild of that Viewer's original failed peer edge. Failure is + terminal for that attempt; stop, session replacement, topology revision or a + new sharing generation clears it. Host publisher/global SFU failure retains + the existing bounded peer failback. Production logs on 2026-08-20 observed two root participants for about 4.6 seconds and two short Host participants (about 0.46 and 0.27 seconds), all ending @@ -407,10 +409,11 @@ Healthy direct/peer UDP stays distributed. When no such path can satisfy admission or recovery, one SFU publication feeds one or two roots, which keep their bounded peer descendants. Only an edge that also cannot use SFU/UDP may receive a controller-selected authenticated TURN attempt. Production currently -ends in bounded failure and has no application TURN config. Source also has no +ends in bounded failure and has no application TURN config. Source has no participant-wide TURN path: the rejected candidate is removed and its stale -environment keys fail startup. Selected-edge config, wire, and rebuild remain -unimplemented and undeployed. +environment keys fail startup. The default-off selected-edge source candidate +uses a short coturn REST bearer and one server-generated connection identity for +the exact failed parent/Viewer pair. It is undeployed and has no real relay-media evidence. LiveKit participant-wide embedded/external TURN remains a separate ICE domain. The bounded cost model, privacy-safe ICE fields, and exact-room A/B sequence @@ -462,10 +465,10 @@ record which boundary is actually configured and the UI must not claim E2EE. 3. Add versioned route state and pure invariant/property tests. 4. Implement server prepare/commit/abort and strict authorization. 5. Integrate host/viewer first-frame switching while retaining relay children. -6. Extend the tracked `1/3/5/8` benchmark with automatic fallback, peer/SFU UDP, - bounded UDP-blocked failure, source/profile/pause/stop, and server-egress - measurements. After the SFU path passes, add one selected-edge TURN attempt - and measure allocation/RSS/ports and relay bandwidth. Only after those pass, +6. Land the bounded selected-edge function before treating performance evidence + as a prerequisite. Retain the tracked `1/3/5/8` benchmark and then extend it + with peer/SFU UDP, one selected relay edge, bounded UDP-blocked failure, + allocation/RSS/ports, relay bandwidth and server-egress measurements. Only after those pass, add the 20-viewer gate before changing the room default. @@ -505,7 +508,7 @@ Positive: - Host fanout remains bounded while server media egress is paid only for fallback roots and separately admitted exceptional viewers. - The ordinary peer path remains direct-first and STUN-only; current production - has no TURN media cost, and a future relay attempt is confined to one + has no TURN media cost, and a configured relay attempt is confined to one controller-selected exceptional edge after SFU/UDP. - Revisioned prepare/commit isolates stale asynchronous results without a continuous optimizer. diff --git a/docs/deployment.md b/docs/deployment.md index 4873a65..12698fd 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -13,8 +13,8 @@ current controller's automatic final media fallback. This capacity is dormant un `LIVEKIT_URL`/key/secret tuple is configured. It does not replace the P2P path, the peer-assisted experiment, or required STUN discovery. LiveKit remains ICE/UDP only. Source no longer contains the rejected participant-wide Peer ICE TURN -candidate; stale `PEER_ICE_TURN_*` keys fail startup even when blank, and -selected-edge TURN is not implemented or deployed. Production advertises +candidate; stale `PEER_ICE_TURN_*` keys fail startup even when blank. A +default-off selected-edge source candidate exists but is not deployed. Production advertises STUN-only ICE and the tracked coturn example remains `stun-only`. The shared host still retains its older authenticated-relay daemon configuration and firewall range, but the application advertises no credential. @@ -107,11 +107,11 @@ ICE. The accepted ladder is direct/peer UDP, then the revision-bound SFU/UDP virtual parent, then optional authenticated TURN for one controller-selected exceptional edge. LiveKit participants receive only revision-bound `sfu-config` URL/token messages and negotiate within LiveKit's separate ICE domain. The -selected-edge TURN config/wire is not yet implemented or deployed. +selected-edge TURN config/wire is source-only and not yet deployed. `PEER_ASSISTED_ROOM_IDS` remains the topology/SFU canary boundary. Listed and -unlisted ordinary peer connections both get STUN-only ICE; only a future -selected-edge implementation may grant TURN to the controller's then-current +unlisted ordinary peer connections both get STUN-only ICE; only the selected-edge +source path may grant TURN to the controller's then-current edge. Run candidate rooms on an isolated instance/hostname and keep the old release unchanged for rollback. If the candidate fails, roll back the release or instance; do not add a permanent dual-transport branch. @@ -190,12 +190,12 @@ MAX_SFU_ROOTS_PER_ROOM=2 ``` The rejected `PEER_ICE_TURN_*` participant-wide tuple is removed; supplying any -stale key, even blank, fails startup. A future selected-edge release must -introduce a distinct complete default-off tuple in the same coherent change as -its controller consumer, strict schema and rollback tests; deployment -documentation will name it only after that change lands. It -must accept one explicit TURN/UDP URI, use an independent secret and bounded TTL, -and issue credentials only for a current one-use route attempt. Credentials +stale key, even blank, fails startup. The source candidate uses the complete +`SELECTED_EDGE_TURN_URLS`, `SELECTED_EDGE_TURN_SHARED_SECRET`, and +`SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS` default-off tuple with one explicit +TURN/UDP URI, an independent secret and bounded TTL. Enable it only after the +exact-room peer/SFU tuple and bounded coturn service are ready; roll back by +removing the application tuple before changing coturn or firewall state. Credentials never enter URLs, logs, browser persistence, room rows, or SQLite. `PEER_ASSISTED_ROOM_IDS` is the required deployment canary boundary whenever @@ -206,7 +206,7 @@ leading zeroes, and malformed IDs fail startup. A non-empty allowlist requires `PEER_ASSISTED_MEDIA=true`. Only listed rooms receive peer-assisted authentication, routing, room quality state, or optional LiveKit fallback; every other room keeps ordinary P2P route fields and signaling behavior. The -future selected-edge TURN attempt is confined to a current controller edge in a +selected-edge TURN attempt is confined to a current controller edge in a listed room; every ordinary peer connection remains STUN-only. Omitting or blanking the variable fails startup rather than enabling every room. There is no browser control, percentage rollout, or all-room fail-open. @@ -240,7 +240,7 @@ that its authenticated message contains `mediaMode: "peer-assisted"` while a second non-allowlisted room contains none of `mediaMode`, `qualitySettings`, `routeRevision`, `routeAssignment`, or `sfuStandbyUrl`. Exercise join, offer and answer, stop, reconnect, and room deletion in both rooms. Ordinary Web and -Native-shaped clients remain STUN-only. With a future selected-edge tuple, only +Native-shaped clients remain STUN-only. With the selected-edge tuple, only the current controller-selected edge may receive a one-use grant; all other sessions and connections remain STUN-only. Roll back by disabling application issuance before restoring the exact recorded pre-canary coturn/firewall baseline, or direct @@ -418,8 +418,8 @@ compatibility. The rejected candidate names `PEER_ICE_TURN_URLS`, `PEER_ICE_TURN_SHARED_SECRET`, and `PEER_ICE_TURN_CREDENTIAL_TTL_SECONDS` must remain absent from production and -now fail startup even when blank. They are not aliases for the future -selected-edge tuple, which is not implemented or deployed. +now fail startup even when blank. They are not aliases for the source-only +selected-edge tuple, which is not deployed. ## HTTPS and WSS ingress @@ -516,7 +516,7 @@ not the optional selected-edge target or the current shared-host coturn state. T host retains an older authenticated-relay configuration and TCP/UDP 3478 plus UDP 49152-49251 firewall range, while Screener advertises no TURN credential and the final canary audit found zero allocations. Authenticated selected-edge -fallback remains unimplemented and default-off; its rollout stays blocked on +fallback is source-complete and default-off; its rollout stays blocked on retained SFU/UDP media and one forced-relay edge acceptance. Copy [`deploy/coturn/turnserver.conf.example`](../deploy/coturn/turnserver.conf.example) diff --git a/docs/project-memory.md b/docs/project-memory.md index 6093c01..31e7d7d 100644 --- a/docs/project-memory.md +++ b/docs/project-memory.md @@ -25,7 +25,7 @@ Last updated: 2026-08-20 - Keep experiments bounded: the standard representation sequence is below; native RTP relay, encoded-object striping, and FEC remain separate. - ADR-0006 stays no-go: one evidence-clarification run retained VP8 output and one first-generation binary `send()` return, then no post-send diagnostics and at least one sanitized fatal marker. The media-counter zeros are initial values, not Go reports; no Viewer was created, cleanup passed 5/5, and no product root cause is proven. A separate MF fixture proved RTX H.264 hardware over 360/360 frames, but `42c01f` misses Pion's default fmtp; Viewer interop remains unproven. Native v2 stays memory-only and leaves SQLite/Web unchanged. - Mobile Web Host is unsupported; feature-detect and fail clearly. Mobile Viewer stays leaf-only. After Windows native, gate Android 14+; iOS waits for stable iOS 27 ScreenCaptureKit. -- ADR-0005 accepts direct/peer UDP, bounded SFU roots, then optional selected-edge TURN. The built-in participant-wide candidate failed its real direct acceptance before forced relay and rolled back completely; it is rejected as the product path. Production advertises STUN-only ICE. +- ADR-0005 accepts direct/peer UDP, bounded SFU roots, then optional selected-edge TURN. Source has a default-off, undeployed one-shot rebuild after SFU; participant-wide TURN remains rejected and production stays STUN-only. - Keep the controller exact-room only: room `1` is the STUN/SFU smoke. Preserve sticky progressing P2P, mobile leaves and break-before-make. Recovery spends one attempt per layer (ICE restart, same-parent rebuild, alternate peer, then SFU), never three identical retries; active SFU gets one fresh grant before Peer failback. - C+B quality reparenting is edge-local and cooldown-bound; no score, timer, or global parent penalty. - Flagship media is UDP; HTTPS/WSS stays TLS/TCP; the old release remains rollback-only. @@ -40,7 +40,7 @@ Last updated: 2026-08-20 - The repository is one npm package using Node.js 24, React, TypeScript, Vite, native WebRTC, `ws`, Zod, Vitest, and separate coturn. - Production `31bee238bc1e901e823f34e50737e202dc4b04bf` deployed at 2026-08-20 15:33:21 +08 to `https://share.bonfire.icu`. Room `1` alone enables peer/SFU routing, C+B local reparenting, and admission rescue; unlisted rooms remain ordinary P2P. Immediate rollback is `7fea60ef6f2a` with the current v2 DB/env. -- Production ICE is STUN-only. Source no longer contains the rejected participant-wide TURN config, issuer, capability, refresh wire, or client propagation; stale `PEER_ICE_TURN_*` keys fail startup even when blank. Room `1` may feed two LiveKit roots with peer descendants; selected-edge TURN is not implemented or deployed. Real rescue/relay/SFU media and game quality remain unverified. +- Production ICE is STUN-only; stale `PEER_ICE_TURN_*` keys fail startup even blank. Source has a default-off coturn REST rebuild for the original failed Host/ViewerRelay edge after Viewer-root SFU exhaustion; it is undeployed/unverified and production advertises no TURN credential. - Production requires an independent `HOST_ADMISSION_PASSWORD` only for creation/Host role. Default private fragment grants and explicit public-watch authorize Viewers; four anonymous Chrome routes stayed neutral until authorization. There are no accounts/JWT/session rows; SQLite v2 still contains all four rooms, including room `1`. - Production uses nginx, Node.js 24.19.0, coturn 4.17.2, and pinned LiveKit 1.13.5 on UDP 7882 with TCP fallback disabled. The application TURN tuple is absent; coturn retains the old authenticated-relay config and TCP/UDP 3478 plus UDP 49152-49251 rules but receives no advertised credential. All services are active with zero automatic restarts; Screener used 38,031,360 bytes with zero cgroup events at 15:39:07. - Old all-room application TURN wire/config remains absent. Canary release `a11a73dfa79d-r4` is inactive after rollback to `7fea60e`; final health/config/firewall/SQLite passed with zero allocations. @@ -66,7 +66,7 @@ These are measurement gates, not performance claims. - Sustainable count by hardware, quality, network, and route: finish instrumented `1/3/5/8`, then pass a 20-viewer matrix before changing the accepted target default to 20. - Whether ADR-0004 passes fanout, re-encoding, depth-four latency, reparenting, silent-partition, and mobile-leaf gates. -- Whether one controller-selected edge can pass generation-bound parent/child rebuild, forced relay, expiry/failure, mobile, 1-GiB resource and relay-bandwidth gates after SFU/UDP fails. The rejected built-in canary left no application tuple, advertised credential, transient rule, or allocation; the old coturn relay config and ports remain. Media TCP is out of scope. +- Whether source-complete selected-edge passes forced relay, expiry/failure, mobile, 1-GiB resource and relay-bandwidth gates. Coturn's bearer is non-revocable until expiry; application guards and TTL only bound reuse. Performance comparison and the retained matrix follow functional landing; Media TCP is out of scope. - Exact mobile lifecycle behavior and whether Windows per-application audio is required for the first release. - Project license and distribution model, which determines whether GPL/AGPL sources can move beyond study-only use. - Initial deployment regions and expected mainland China, Hong Kong, and overseas network mix. diff --git a/docs/research/advanced-peer-distribution.md b/docs/research/advanced-peer-distribution.md index 9bbfd3d..2a81ebd 100644 --- a/docs/research/advanced-peer-distribution.md +++ b/docs/research/advanced-peer-distribution.md @@ -312,7 +312,7 @@ optimizer, temporary fanout above budget, self-reported geography or device quality, or cannot beat the unchanged route. The earlier admission-rescue case remains valid, while this quality slice moves only the affected Viewer-rooted subtree and excludes only its current failed parent through the existing -peer-first, then SFU path. A future selected-edge TURN attempt follows SFU +peer-first, then SFU path. The default-off source candidate's selected-edge TURN attempt follows SFU failure; healthy paths never enter relay optimization. ## Staged Connection Recovery Evidence diff --git a/docs/research/built-in-peer-ice-turn.md b/docs/research/built-in-peer-ice-turn.md index bca9a03..2184ea8 100644 --- a/docs/research/built-in-peer-ice-turn.md +++ b/docs/research/built-in-peer-ice-turn.md @@ -25,8 +25,9 @@ the durable findings that still apply are summarized below. The current source no longer contains the rejected participant-wide config, issuer, authentication capability, refresh wire, or client propagation. Supplying any -stale `PEER_ICE_TURN_*` key, even blank, fails startup. Selected-edge TURN config, -wire, and rebuild remain unimplemented and undeployed. +stale `PEER_ICE_TURN_*` key, even blank, fails startup. The replacement +selected-edge config, wire, and rebuild now exist as a default-off source +candidate; they remain undeployed and lack real relay-media evidence. ## Production Canary @@ -98,11 +99,10 @@ Coturn still cannot enforce the application edge identity. The application guards provide selected-edge authorization; short expiry and coturn quotas bound bearer reuse and resource cost. Do not claim cryptographic edge binding. -The exact protocol and environment names belong in the implementation PR, after -all parent roles (`HostPeer` and relay parents), the Viewer child, and -`HybridMediaRouter` share one reviewed generation contract. A config-only or -issuer-only partial slice is forbidden because it would create credentials with -no safe current consumer. +The source candidate uses the complete `SELECTED_EDGE_TURN_*` tuple only with +exact-room peer assistance and SFU configuration. `HostPeer`, relay parents, the +Viewer child, and `HybridMediaRouter` share one generation contract; a config-only +or issuer-only partial deployment remains forbidden. ## Acceptance Boundary diff --git a/docs/research/low-server-media-routes.md b/docs/research/low-server-media-routes.md index 78b9cd2..8a006cf 100644 --- a/docs/research/low-server-media-routes.md +++ b/docs/research/low-server-media-routes.md @@ -190,10 +190,10 @@ SFU, not ordinary peer PCs. Selected-edge coturn uses independent configuration and credentials. LiveKit TURN cannot rescue an unavailable SFU; the application controller selects whether independent coturn may rebuild one failed peer edge. -Production and current source remain STUN-only. The participant-wide TURN +Production and ordinary peer ICE remain STUN-only. The participant-wide TURN config, capability and refresh wire are removed; any stale `PEER_ICE_TURN_*` -key, including an empty value, fails startup. Selected-edge TURN remains -unimplemented and undeployed. Every ordinary room, Web peer and Native-shaped +key, including an empty value, fails startup. Selected-edge TURN is source-complete, +default-off and undeployed. Every ordinary room, Web peer and Native-shaped client remains STUN-only. The tracked coturn example is UDP `stun-only`; the LiveKit example exposes only ICE/UDP mux 7882, explicitly sets `tcp_port: 0` and `allow_tcp_fallback: false`, supplies the self-hosted STUN diff --git a/docs/status.md b/docs/status.md index 6290963..6e0d88b 100644 --- a/docs/status.md +++ b/docs/status.md @@ -37,6 +37,7 @@ and is not part of production. - The `a11a73d` built-in TURN canary is rejected: local allocation passed, but direct Host/Pion Viewer failed before forced relay. Full rollback restored STUN-only client ICE and zero allocations. - Native remains no-go: fixed VP8, one encoder and a generation-1 binary send returned; no later diagnostics arrived, a sanitized fatal marker was retained, and the media zeros are initial values rather than Go reports. No Viewer was created, cleanup passed 5/5, and no product root cause is proven. - Access protocol/config/HTTP/storage/SQLite/signaling focused tests pass, including commit-first teardown and v1 rollback. +- Selected-edge typecheck and 131 focused control tests pass; no TURN allocation/media evidence exists. ## Unverified Boundaries @@ -64,11 +65,9 @@ evacuation. Native stopped after one binary send return: no later diagnostics, a retained fatal marker, and uninterpretable media zeros. Classify this boundary before any new run. Audio A/B/C may proceed without displacing P0. -Ordinary Peer ICE stays STUN-only. Participant-wide TURN source and wire are -removed; stale `PEER_ICE_TURN_*` keys fail startup even blank. Selected-edge TURN -is unimplemented and undeployed. After SFU/UDP, bind one selected parent/child -rebuild to current generations before one forced-relay canary; do not repeat the -built-in direct canary. +Ordinary Peer ICE stays STUN-only; participant-wide TURN is removed. Selected-edge +is source-complete/default-off/undeployed. Next run one isolated forced-relay +canary; retained performance/resource benchmarks follow functional landing. ADR-0004 still requires a full-resolution 30-minute `1/3/5/8` network, resource, quality, latency, recovery, and browser/mobile-leaf matrix. A separate diff --git a/docs/方案设计.md b/docs/方案设计.md index aa2d8fb..e6013c1 100644 --- a/docs/方案设计.md +++ b/docs/方案设计.md @@ -232,9 +232,10 @@ fixed-`HIGH` -> 现有 WebRTC/一至两名 Web Viewer 的 P2 窄实验。硬件 当前仓库已经删除旧全房 TURN contract 和被拒绝的 participant-wide built-in config、issuer、capability、`refresh-ice`/`ice-config` wire 与客户端传播。 Production `ServerConfig` 必须配置 `STUN_URLS`；旧 `PEER_ICE_TURN_*` 即使为空也 -启动失败，且绝不是后续 selected-edge tuple 的 alias。Selected-edge TURN 仍未实现或 -部署；后续必须在同一 coherent change 中加入新的完整 default-off tuple 和 controller -consumer，部分配置、非法 UDP TURN URI、弱或复用 secret、越界 TTL 都必须启动失败。 +启动失败，且绝不是 selected-edge tuple 的 alias。Source candidate 已在同一 coherent +change 中加入完整 default-off `SELECTED_EDGE_TURN_*` tuple、controller consumer 与 +Host/ViewerRelay/Viewer rebuild；部分配置、非法 UDP TURN URI、弱或复用 secret、越界 +TTL 都启动失败。该 candidate 尚未部署或通过真实 relay-media 验收。 旧 `TURN_URLS`、`TURN_SHARED_SECRET`、`TURN_CREDENTIAL_TTL_SECONDS` 继续 fail-fast，绝不作为 alias。 @@ -244,10 +245,16 @@ Controller 先完成一次 ICE restart、一次同 parent rebuild、一个 alter selected-edge TURN attempt。该 attempt 必须同时绑定 room/share generation、viewer 与 parent session、route revision、replaced connection 和 server-generated next connection identity；parent 与 child 只为该新 PC 协同重建，Viewer 使用 relay-only -配置。任何代次变化、超时、成功或失败都消费内存态 grant；普通 PC、其他 edge、 +配置。pending 超时、失败或代次变化清理 attempt；answer 后 attempt 变为已使用并仅为 +该 edge 保留信令 owner，直到失败或 generation/session/topology/room cleanup。普通 PC、其他 edge、 Native v2 与 authenticated snapshot 继续 STUN-only。凭据不进入 URL、日志、browser storage、room row 或 SQLite。 +该临时 edge 不改写 active SFU assignment 或另建 route owner；attempt 存续期间 +presence 仍报告 `sfu`，本地 ICE path 才报告 TURN transport。签发前按当前 assignment +重新计算 parent 下游预算，Host 的 SFU publication 也占一边；Host 两边或 ViewerRelay +一边已满且 target 不是现有 child 时 fail closed，不替换无关 child。 + 普通 peer、LiveKit participant 与 selected-edge coturn 是三个明确 ICE 域。SFU publisher/subscriber 只消费 revision-bound `sfu-config {url, token}`，再由 LiveKit SDK 与服务端协商 candidates。tracked LiveKit 配置只开放 UDP mux，显式使用 @@ -256,8 +263,8 @@ external TURN 或 embedded TURN。Node 应用不能替代部署验证来证明 从公网可达；自托管 `stun_servers` 也避免 pinned LiveKit 1.13.5 继承默认公共 Google STUN。 -当前可执行梯级是 `direct/peer UDP -> SFU ICE/UDP -> failure`；accepted target 在 -SFU/UDP 后增加一次 selected-edge TURN，再失败则明确终止。Viewer 的 +Source candidate 的可执行梯级是 `direct/peer UDP -> SFU ICE/UDP -> selected-edge +TURN -> failure`，production 仍停在 SFU 后的 bounded failback。Viewer 的 现有两次有界 ICE restart/rebuild 后报告 route failure；router 先尝试局部 reparent，再在 root budget 内准备 SFU，最终没有可行 route 时返回明确的 `PEER_NOT_FOUND`，不长时间维持伪连接。HTTPS/WSS 始终使用 TLS/TCP，不 @@ -271,9 +278,11 @@ production 不根据 room allowlist 保留旧 release TURN 兼容分支。 Selected-edge TURN 复用 `HybridMediaRouter` 的 assignment/revision owner，不新增 第二个路由器或周期优化器。Coturn 只能验证 HMAC 与 expiry，不能自行验证 room、 -edge 或 generation；应用层必须在签发、parent/child rebuild、ready/failure 与清理 +edge 或 generation；应用层必须在签发、parent/child rebuild、signaling/failure 与清理 边界逐次重验，并用短 TTL、one-use attempt、endpoint fanout 与 allocation/bandwidth quota 限界。LiveKit participant-wide TURN 是独立 ICE 域，不能冒充 selected edge。 +Coturn bearer 在 TTL 内不可撤销，HMAC 只证明 expiry 与 secret；应用 guard 限制谁能 +收到和使用 rebuild，但不能宣称 coturn 对 room/edge 做了密码学绑定。 SFU 正常只作为 virtual parent 向一至两个必要 roots 发布；可靠 root 收到 媒体后继续服务 sticky descendants。host/native root 最多两条下游媒体边， @@ -363,7 +372,7 @@ hop 按[低服务器成本媒体路由](./research/low-server-media-routes.md) 3. 完成屏幕捕获、逐观看者 PeerConnection、观看播放和手动质量档。 4. 完成连接恢复和 stats 面板；随后以 STUN-only snapshot 替换旧全房 TURN credential/refresh wire。 5. 完成 automatic peer/SFU controller 与 exact-room canary 边界。 -6. 在 SFU/UDP retained-media gate 后增加默认关闭的 selected-edge TURN tuple、revision-bound one-use grant 与 parent/child rebuild；用有界网络矩阵验收 ordinary STUN-only、SFU/UDP、单条 forced relay、资源成本和明确失败，coturn/production rollout 保持独立验收边界。 +6. 默认关闭的 selected-edge TURN tuple、revision-bound one-use grant 与 parent/child rebuild 已在 source candidate 闭环；其后再用有界网络矩阵验收 ordinary STUN-only、SFU/UDP、单条 forced relay、资源成本和明确失败，coturn/production rollout 保持独立验收边界。保留现有 benchmark，但统一性能对比不是功能落地前置。 7. 在媒体主线阶段验收后，用一个原子 PR 迁移 Host admission、`screener-v2`、room-scoped Viewer grant 和 SQLite v2；同一部署切换客户端/服务端/环境和数据库备份，随后删除 v1 配置、wire 与测试。用户名/roster 保持独立。 8. P1 继续完成 Windows native fixed-`HIGH` staged gate 与 deployment；移动 Host 不阻断或先于它。 9. P2 只在上述门槛后运行 Android 14/API 34+ fixed-`HIGH` 原生 sender spike；失败即停，不扩展为跨平台 framework。 @@ -386,9 +395,9 @@ fail-open。未命中房间不得进入 不接收 TURN config。命中房间的 ordinary peer PC 与 Native v2 同样保持 STUN-only； 只有 controller-selected edge 可接收一次短期 grant。该边界不增加 UI、用户选择或百分比框架。 -产品硬目标是任一非服务器节点任何时刻最多两条下游媒体边。系统按一个或两个观看者直接 P2P、更多观看者 peer-assisted、部署 SFU 中央兜底、最后 selected-edge TURN 的顺序自动分配和恢复，分享者与观看者不选择拓扑。默认媒体 transport 只使用 UDP；SFU 作为 virtual parent 正常只送一至两个 roots，roots 之后仍优先沿受限 peer descendants 扩散；多个无法挂到健康 root 的极端 viewers 只可使用另设上限的中央 edge。Production 已把普通 ICE 收敛为 STUN-only、删除旧全房 TURN wire并只为 room `1` 开启 failure-only controller；source 也已删除失败的 participant-wide built-in code/wire，旧 `PEER_ICE_TURN_*` 配置启动失败，replacement selected-edge 尚未实现或部署。exact-room 公网 SFU media、selected relay 与资源仍未验收，旧 release/coturn 仅作回滚。ADR-0004/0005 的 broad-rollout 验收尚未完成。 +产品硬目标是任一非服务器节点任何时刻最多两条下游媒体边。系统按一个或两个观看者直接 P2P、更多观看者 peer-assisted、部署 SFU 中央兜底、最后 selected-edge TURN 的顺序自动分配和恢复，分享者与观看者不选择拓扑。默认媒体 transport 只使用 UDP；SFU 作为 virtual parent 正常只送一至两个 roots，roots 之后仍优先沿受限 peer descendants 扩散；多个无法挂到健康 root 的极端 viewers 只可使用另设上限的中央 edge。Production 已把普通 ICE 收敛为 STUN-only、删除旧全房 TURN wire并只为 room `1` 开启 failure-only controller；source 也已删除失败的 participant-wide built-in code/wire，旧 `PEER_ICE_TURN_*` 配置启动失败，replacement selected-edge 已完成 source candidate 但尚未部署。exact-room 公网 SFU media、selected relay 与资源仍未验收，旧 release/coturn 仅作回滚。ADR-0004/0005 的 broad-rollout 验收尚未完成。 -恢复目标按证据和成本逐级升级：短暂 `disconnected` 先观察同代 bytes/stats，`failed` 才执行一次 ICE restart；仍失败时重建一次相同 parent，再尝试不同的合格 peer parent，peer 候选耗尽后才申请 SFU。每个动作最多一次且成功即停，不实现“三次相同重试”。初始 `connecting`/`checking` 的软提示和硬窗口尚未实现；硬窗口必须先用移动网络实测确定，不能直接复用现有 3 秒恢复常量。已实现的 SFU active 失败只刷新一次短期 grant，第二次失败回到新 peer baseline。Host 只保留当前 route revision 的 `connect | source | video-publish | sender-config | audio-publish | transport` 枚举阶段作为本地事件提示，不上传原始错误、URL、token、candidate 或地址。 +恢复目标按证据和成本逐级升级：短暂 `disconnected` 先观察同代 bytes/stats，`failed` 才执行一次 ICE restart；仍失败时重建一次相同 parent，再尝试不同的合格 peer parent，peer 候选耗尽后才申请 SFU。每个动作最多一次且成功即停，不实现“三次相同重试”。初始 `connecting`/`checking` 的软提示和硬窗口尚未实现；硬窗口必须先用移动网络实测确定，不能直接复用现有 3 秒恢复常量。Viewer SFU root 只刷新一次短期 grant，随后至多重建一次原 failed parent selected edge；Host publisher/global SFU failure 维持 bounded peer failback。Host 只保留当前 route revision 的 `connect | source | video-publish | sender-config | audio-publish | transport` 枚举阶段作为本地事件提示，不上传原始错误、URL、token、candidate 或地址。 实验只实现一个由服务端确定的广度优先拓扑：host 最多两个 child，每名 viewer 最多一个 child，因此形成两条按加入顺序平衡的 sticky 链。父节点只按深度和 `serverJoinSequence` 选择第一个空 slot。新 viewer 加入通常不搬动已有节点；唯一例外是纯 peer active route 没有 pending/SFU 时，无上游、无 child、无失败 parent 历史且刚声明一槽容量的 viewer，可在 Host 两槽已满时按加入序替换一个 childless、零容量 Host leaf，并把该 leaf 作为自己的唯一 child。Host 客户端先关闭旧 leaf edge 再启动新 relay edge，同一同步拓扑变更只推进一个 route revision。Parent 离开时仍只把失去 parent 的 subtree root 重新分配到第一个空 slot，其 descendants 保持连接。该规则不采集或组合 RTT、带宽、CPU、地理位置与质量评分，也不主动重平衡。没有 parent 时明确等待或拒绝，不能创建分享端第三条边。 diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 7b49bd0..35a1ed7 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -86,6 +86,7 @@ type ViewerQualityEvidence = Extract< ServerMessage, { type: "viewer-quality-evidence" } >; +type SelectedEdgeTurn = Extract; interface CaptureDetails { resolution: string; @@ -206,6 +207,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { const signalRef = useRef(null); const iceConfigRef = useRef(null); const peersRef = useRef(new Map()); + const retiredConnectionsRef = useRef(new Map()); + const hostPeerIdRef = useRef(null); const viewerQualityEvidenceRef = useRef( new Map(), ); @@ -266,6 +269,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { signalRef.current?.stop(); peersRef.current.forEach((peer) => peer.dispose()); peersRef.current.clear(); + retiredConnectionsRef.current.clear(); + hostPeerIdRef.current = null; viewerQualityEvidenceTimersRef.current.forEach((timer) => window.clearTimeout(timer), ); @@ -367,6 +372,8 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { shareGenerationRef.current = null; peersRef.current.forEach((peer) => peer.dispose()); peersRef.current.clear(); + retiredConnectionsRef.current.clear(); + hostPeerIdRef.current = null; void hostSfuRouteRef.current?.disconnect(); hostSfuRouteRef.current = null; sfuStandbyPrewarmerRef.current?.setUrl(null); @@ -628,7 +635,11 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { function removePeer(peerId: string): void { clearViewerQualityEvidence(peerId); parentEdgeQualityEvidenceReporterRef.current.forget(peerId); - peersRef.current.get(peerId)?.dispose(); + const peer = peersRef.current.get(peerId); + if (peer) { + retiredConnectionsRef.current.set(peerId, peer.connectionId); + peer.dispose(); + } peersRef.current.delete(peerId); setPeerSnapshots((current) => { const next = new Map(current); @@ -641,6 +652,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { peerId: string, generation: number, attempt = 0, + selectedTurn?: SelectedEdgeTurn, ): Promise { if (!isCurrentGeneration(generation)) { return; @@ -649,7 +661,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { // negotiation whose offer or answer may have been lost with the WebSocket. const existing = peersRef.current.get(peerId); if (existing) { - if (existing.isConnected()) { + if (!selectedTurn && existing.isConnected()) { return; } removePeer(peerId); @@ -664,7 +676,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { let peer: HostPeer; peer = new HostPeer( peerId, - iceConfig, + selectedTurn ? { iceServers: [selectedTurn.iceServer] } : iceConfig, activeStream, qualitySettingsRef.current, { @@ -677,10 +689,16 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { isCurrentGeneration(generation) && peersRef.current.get(peerId) === peer ) { + if (selectedTurn && snapshot.connectionState === "failed") { + removePeer(peerId); + return; + } updatePeerSnapshot(snapshot); } }, }, + selectedTurn !== undefined, + selectedTurn?.newConnectionId, ); peersRef.current.set(peerId, peer); let started: boolean; @@ -697,7 +715,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } removePeer(peerId); - if (!isCurrentGeneration(generation) || attempt >= 1) { + if (selectedTurn || !isCurrentGeneration(generation) || attempt >= 1) { return; } window.setTimeout(() => { @@ -781,6 +799,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { return; } if (message.type === "authenticated" && message.role === "host") { + hostPeerIdRef.current = message.peerId; setViewerAccessUpdating(false); clearAllViewerQualityEvidence(); setSfuStandbyUrl( @@ -829,6 +848,25 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { } return; } + if (message.type === "selected-edge-turn") { + const current = peersRef.current.get(message.viewerPeerId); + const oldConnectionId = + current?.connectionId ?? + retiredConnectionsRef.current.get(message.viewerPeerId); + if ( + peerAssistedRef.current && + message.parentPeerId === hostPeerIdRef.current && + message.revision === activeRouteRevisionRef.current && + Date.parse(message.expiresAt) > Date.now() && + oldConnectionId === message.oldConnectionId + ) { + if (current) { + removePeer(message.viewerPeerId); + } + void startPeer(message.viewerPeerId, generation, 0, message); + } + return; + } if (message.type === "viewer-access-updated") { setViewerAccessUpdating(false); replaceViewerInvite(activeRoomId, message.inviteUrl); diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index 8bb9ed0..bc4da12 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -71,6 +71,7 @@ type ViewerQualityEvidence = Extract< ServerMessage, { type: "viewer-quality-evidence" } >; +type SelectedEdgeTurn = Extract; export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { const [accessState, setAccessState] = useState< @@ -130,6 +131,11 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { let preparedParentSignals: Array< Extract > = []; + let retiredUpstream: { + parentPeerId: string; + connectionId: string; + } | null = null; + let selectedEdgeTurn: SelectedEdgeTurn | null = null; const messageAuthority = new ViewerMessageAuthority(); let sfuStandbyPrewarmer: SfuStandbyPrewarmer | null = null; let relayChildEvidenceCurrent: ViewerQualityEvidence | null = null; @@ -245,9 +251,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { currentQualitySettings, { sendSignal: (targetPeerId, payload) => - active && - peerAssisted && - currentAssignment.childPeerIds[0] === targetPeerId + active && peerAssisted ? signal.send({ type: "signal", targetPeerId, @@ -407,7 +411,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { function clearUpstreamState(): void { qualityEvidenceReporter.reset(); - peerRef.current?.dispose(); + const peer = peerRef.current; + retiredUpstream = peer?.getConnectionIdentity() ?? retiredUpstream; + peer?.dispose(); peerRef.current = null; setRemoteStream(null); setPeerSnapshot(null); @@ -449,7 +455,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { return null; } const peer = new ViewerPeer( - currentIceConfig, + selectedEdgeTurn + ? { iceServers: [selectedEdgeTurn.iceServer] } + : currentIceConfig, { sendSignal: (targetPeerId, payload) => signal.send( @@ -496,6 +504,21 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { } }, onRecoveryExhausted: (parentPeerId, connectionId) => { + if ( + selectedEdgeTurn?.parentPeerId === parentPeerId && + selectedEdgeTurn.newConnectionId === connectionId + ) { + signal.send({ + type: "route-failed", + revision: selectedEdgeTurn.revision, + phase: "active", + connectionId, + }); + selectedEdgeTurn = null; + clearUpstreamState(); + setStatusText("无法建立媒体连接"); + return true; + } if (viewerSfuRoute) { return viewerSfuRoute.reportPeerFailure(parentPeerId, connectionId); } @@ -503,6 +526,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { return true; }, }, + selectedEdgeTurn !== null, ); peerRef.current = peer; return peer; @@ -513,6 +537,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { authorityToken: number, ): Promise { if (message.type === "authenticated") { + selectedEdgeTurn = null; viewerAuthenticated = true; setAccessState("ready"); clearRelayChildEvidence(); @@ -596,9 +621,43 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { } return; } + if (message.type === "selected-edge-turn") { + if ( + !peerAssisted || + message.revision !== currentRouteRevision || + Date.parse(message.expiresAt) <= Date.now() + ) { + return; + } + if (message.parentPeerId === currentPeerId) { + ensureViewerRelay()?.startSelectedEdgeTurn(message); + return; + } + const currentIdentity = + peerRef.current?.getConnectionIdentity() ?? retiredUpstream; + if ( + message.viewerPeerId !== currentPeerId || + currentIdentity?.parentPeerId !== message.parentPeerId || + currentIdentity.connectionId !== message.oldConnectionId + ) { + return; + } + selectedEdgeTurn = message; + clearViewerSfuRoute(); + clearUpstreamState(); + currentAssignment = { + parentPeerId: message.parentPeerId, + childPeerIds: currentAssignment.childPeerIds, + }; + setStatusText("正在恢复连接"); + return; + } if (message.type === "route-update") { if (peerAssisted) { if (message.phase === "active") { + if (selectedEdgeTurn?.revision !== message.revision) { + selectedEdgeTurn = null; + } if (message.revision !== currentRouteRevision) { clearRelayChildEvidence(); } diff --git a/src/client/webrtc/host-peer.ts b/src/client/webrtc/host-peer.ts index 9c04ac3..3afc743 100644 --- a/src/client/webrtc/host-peer.ts +++ b/src/client/webrtc/host-peer.ts @@ -16,6 +16,7 @@ import { } from "./stats"; const MAX_PENDING_CANDIDATES = 64; +type PeerIceConfig = Pick; type SignalCandidate = Extract< SignalPayload, { kind: "candidate" } @@ -59,7 +60,7 @@ function captureMetrics(track: MediaStreamTrack): CaptureMetrics { } export class HostPeer { - readonly connectionId = createOpaqueId(); + readonly connectionId: string; private readonly connection: RTCPeerConnection; private readonly pendingCandidates: SignalCandidate[] = []; @@ -79,13 +80,17 @@ export class HostPeer { constructor( readonly peerId: string, - iceConfig: IceConfig, + iceConfig: PeerIceConfig, private stream: MediaStream, private desiredProfile: QualityProfile, private readonly events: HostPeerEvents, + private readonly relayOnly = false, + connectionId = createOpaqueId(), ) { + this.connectionId = connectionId; this.connection = new RTCPeerConnection({ iceServers: iceConfig.iceServers, + ...(relayOnly ? { iceTransportPolicy: "relay" } : {}), }); this.snapshot = { peerId, @@ -250,7 +255,7 @@ export class HostPeer { } updateIceConfig(iceConfig: IceConfig): void { - if (this.disposed) { + if (this.disposed || this.relayOnly) { return; } try { diff --git a/src/client/webrtc/viewer-peer.ts b/src/client/webrtc/viewer-peer.ts index a4d4d7c..b387d2a 100644 --- a/src/client/webrtc/viewer-peer.ts +++ b/src/client/webrtc/viewer-peer.ts @@ -13,6 +13,7 @@ const MAX_PENDING_CANDIDATES = 64; const MAX_AUTOMATIC_RECOVERY_REQUESTS = 2; const INITIAL_CONNECTION_TIMEOUT_MS = 15_000; const AUTOMATIC_RECOVERY_TIMEOUT_MS = 3_000; +type PeerIceConfig = Pick; type SignalCandidate = Extract< SignalPayload, { kind: "candidate" } @@ -52,12 +53,13 @@ export class ViewerPeer { private automaticRecoveryRequests = 0; private recoveryExhaustedReported = false; private disposed = false; - private currentIceConfig: IceConfig; + private currentIceConfig: PeerIceConfig; private snapshot: PeerSnapshot | null = null; constructor( - iceConfig: IceConfig, + iceConfig: PeerIceConfig, private readonly events: ViewerPeerEvents, + private readonly relayOnly = false, ) { this.currentIceConfig = iceConfig; } @@ -150,6 +152,7 @@ export class ViewerPeer { this.setError(error, "处理分享端信令失败"); if ( payload.kind === "description" && + !this.relayOnly && this.offerRecoveryAttempts < 1 && this.events.sendRestartRequest( parentPeerId, @@ -167,6 +170,9 @@ export class ViewerPeer { } updateIceConfig(iceConfig: IceConfig): void { + if (this.relayOnly) { + return; + } this.currentIceConfig = iceConfig; if (!this.connection) { return; @@ -181,7 +187,7 @@ export class ViewerPeer { } requestRecovery(): boolean { - if (!this.connectionId || !this.parentPeerId) { + if (this.relayOnly || !this.connectionId || !this.parentPeerId) { return false; } const sent = this.events.sendRestartRequest( @@ -208,6 +214,15 @@ export class ViewerPeer { return this.connection !== null && this.connectionId === connectionId; } + getConnectionIdentity(): { + parentPeerId: string; + connectionId: string; + } | null { + return this.parentPeerId && this.connectionId + ? { parentPeerId: this.parentPeerId, connectionId: this.connectionId } + : null; + } + isConnected(): boolean { return this.connection?.connectionState === "connected"; } @@ -237,6 +252,7 @@ export class ViewerPeer { const connection = new RTCPeerConnection({ iceServers: this.currentIceConfig.iceServers, + ...(this.relayOnly ? { iceTransportPolicy: "relay" } : {}), }); this.connection = connection; this.snapshot = { @@ -330,6 +346,10 @@ export class ViewerPeer { ) { return; } + if (this.relayOnly) { + this.reportRecoveryExhausted(); + return; + } if (this.automaticRecoveryRequests >= MAX_AUTOMATIC_RECOVERY_REQUESTS) { this.reportRecoveryExhausted(); return; diff --git a/src/client/webrtc/viewer-relay.ts b/src/client/webrtc/viewer-relay.ts index 9fbfa81..1f17d0a 100644 --- a/src/client/webrtc/viewer-relay.ts +++ b/src/client/webrtc/viewer-relay.ts @@ -1,4 +1,8 @@ -import type { IceConfig, SignalPayload } from "../../shared/protocol"; +import type { + IceConfig, + ServerMessage, + SignalPayload, +} from "../../shared/protocol"; import type { QualityProfile } from "../media/quality"; import type { PeerSnapshot } from "../types"; import { HostPeer } from "./host-peer"; @@ -7,12 +11,15 @@ interface ViewerRelayEvents { sendSignal: (peerId: string, payload: SignalPayload) => boolean; onUpdate?: (snapshot: PeerSnapshot | null) => void; } +type SelectedEdgeTurn = Extract; export class ViewerRelay { private childPeerId: string | null = null; private stream: MediaStream | null = null; private peer: HostPeer | null = null; private snapshot: PeerSnapshot | null = null; + private retiredConnection: { peerId: string; connectionId: string } | null = + null; private syncQueue = Promise.resolve(); private disposed = false; @@ -28,6 +35,27 @@ export class ViewerRelay { : null; } + startSelectedEdgeTurn( + message: SelectedEdgeTurn, + ): boolean { + const stream = this.stream; + const connection = this.peer + ? { peerId: this.peer.peerId, connectionId: this.peer.connectionId } + : this.retiredConnection; + if ( + this.disposed || + !stream || + connection?.peerId !== message.viewerPeerId || + connection.connectionId !== message.oldConnectionId + ) { + return false; + } + this.disposePeer(); + this.childPeerId = message.viewerPeerId; + this.startPeer(message.viewerPeerId, stream, 0, message); + return true; + } + setChild(childPeerId: string | null): void { if (this.disposed) { return; @@ -178,6 +206,7 @@ export class ViewerRelay { childPeerId: string, stream: MediaStream, attempt = 0, + selectedTurn?: SelectedEdgeTurn, ): void { if ( this.disposed || @@ -191,7 +220,9 @@ export class ViewerRelay { let peer: HostPeer; peer = new HostPeer( childPeerId, - this.iceConfig, + selectedTurn + ? { iceServers: [selectedTurn.iceServer] } + : this.iceConfig, stream, this.desiredProfile, { @@ -209,6 +240,10 @@ export class ViewerRelay { snapshot.peerId === childPeerId && snapshot.connectionId === peer.connectionId ) { + if (selectedTurn && snapshot.connectionState === "failed") { + this.disposePeer(); + return; + } this.snapshot = { ...snapshot, metrics: { ...snapshot.metrics }, @@ -217,6 +252,8 @@ export class ViewerRelay { } }, }, + selectedTurn !== undefined, + selectedTurn?.newConnectionId, ); this.peer = peer; void peer @@ -228,6 +265,7 @@ export class ViewerRelay { } this.disposePeer(); if ( + !selectedTurn && attempt < 1 && !this.disposed && this.stream === stream && @@ -242,6 +280,12 @@ export class ViewerRelay { private disposePeer(): void { const peer = this.peer; + if (peer) { + this.retiredConnection = { + peerId: peer.peerId, + connectionId: peer.connectionId, + }; + } this.peer = null; this.snapshot = null; this.events.onUpdate?.(null); diff --git a/src/server/app.ts b/src/server/app.ts index 2680cc3..c448364 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -113,6 +113,7 @@ export async function createScreenerServer( peerAssistedMedia: config.peerAssistedMedia, peerAssistedRoomIds: config.peerAssistedRoomIds, ...(sfuFallback ? { sfuFallback } : {}), + ...(config.selectedEdgeTurn ? { selectedEdgeTurn: config.selectedEdgeTurn } : {}), ice: iceOptions, allowedOrigins: config.allowedOrigins, hostAdmissionAtUpgrade: (request) => diff --git a/src/server/config.ts b/src/server/config.ts index d78fbb5..78db164 100644 --- a/src/server/config.ts +++ b/src/server/config.ts @@ -3,6 +3,7 @@ import { MAX_VIEWERS_PER_ROOM_LIMIT, roomCodeSchema, stunUrlSchema, + turnUrlSchema, } from "../shared/protocol.js"; export type RuntimeEnvironment = "development" | "test" | "production"; @@ -10,6 +11,10 @@ export type RuntimeEnvironment = "development" | "test" | "production"; const MIN_HOST_ADMISSION_PASSWORD_BYTES = 8; const MAX_HOST_ADMISSION_PASSWORD_BYTES = 128; const MIN_LIVEKIT_API_SECRET_BYTES = 32; +const MIN_SELECTED_EDGE_TURN_SECRET_BYTES = 32; +const MAX_SELECTED_EDGE_TURN_SECRET_BYTES = 128; +const MIN_SELECTED_EDGE_TURN_TTL_SECONDS = 60; +const MAX_SELECTED_EDGE_TURN_TTL_SECONDS = 10 * 60; const MAX_PEER_ASSISTED_VIEWERS = 8; const DEFAULT_MAX_VIEWERS_PER_ROOM = 8; const DEFAULT_MAX_SFU_ROOTS_PER_ROOM = 2; @@ -31,6 +36,12 @@ export interface LiveKitFallbackConfig { maxSfuRootsPerRoom: number; } +export interface SelectedEdgeTurnConfig { + urls: readonly [string]; + sharedSecret: string; + credentialTtlSeconds: number; +} + export interface ServerConfig { nodeEnv: RuntimeEnvironment; port: number; @@ -45,6 +56,7 @@ export interface ServerConfig { peerAssistedMedia: boolean; peerAssistedRoomIds?: ReadonlySet; livekitFallback?: LiveKitFallbackConfig; + selectedEdgeTurn?: SelectedEdgeTurnConfig; stunUrls: readonly string[]; } @@ -190,6 +202,55 @@ function parseStunUrlList(value: string | undefined): string[] { }); } +function parseSelectedEdgeTurn( + environment: NodeJS.ProcessEnv, +): SelectedEdgeTurnConfig | undefined { + const names = [ + "SELECTED_EDGE_TURN_URLS", "SELECTED_EDGE_TURN_SHARED_SECRET", + "SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS", + ] as const; + const configuredNames = names.filter((name) => environment[name]?.trim()); + if (configuredNames.length === 0) { + return undefined; + } + if (configuredNames.length !== names.length) { + throw new Error(`${names.join(", ")} must be configured together`); + } + + const urls = parseUrlList( + environment.SELECTED_EDGE_TURN_URLS, + "SELECTED_EDGE_TURN_URLS", + ); + if (urls.length !== 1 || !turnUrlSchema.safeParse(urls[0]).success) { + throw new Error( + "SELECTED_EDGE_TURN_URLS must contain one UDP TURN URL with transport=udp", + ); + } + const sharedSecret = environment.SELECTED_EDGE_TURN_SHARED_SECRET!; + const secretBytes = Buffer.byteLength(sharedSecret); + if ( + !VISIBLE_ASCII_PATTERN.test(sharedSecret) || + secretBytes < MIN_SELECTED_EDGE_TURN_SECRET_BYTES || + secretBytes > MAX_SELECTED_EDGE_TURN_SECRET_BYTES + ) { + throw new Error( + "SELECTED_EDGE_TURN_SHARED_SECRET must contain 32 to 128 visible ASCII bytes", + ); + } + + return { + urls: [urls[0]!], + sharedSecret, + credentialTtlSeconds: parseBoundedInteger( + environment.SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS, + MIN_SELECTED_EDGE_TURN_TTL_SECONDS, + "SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS", + MIN_SELECTED_EDGE_TURN_TTL_SECONDS, + MAX_SELECTED_EDGE_TURN_TTL_SECONDS, + ), + }; +} + function parseOrigins(value: string | undefined, fallback: string): Set { const origins = parseUrlList(value, "ALLOWED_ORIGINS"); return new Set((origins.length > 0 ? origins : [fallback]).map(toOrigin)); @@ -299,6 +360,7 @@ export function loadConfig( environment.PEER_ASSISTED_ROOM_IDS, ); const livekitFallback = parseLiveKitFallback(environment, nodeEnv); + const selectedEdgeTurn = parseSelectedEdgeTurn(environment); if (peerAssistedRoomIds && !peerAssistedMedia) { throw new Error( @@ -313,14 +375,20 @@ export function loadConfig( if (livekitFallback && !peerAssistedMedia) { throw new Error("LiveKit fallback requires PEER_ASSISTED_MEDIA=true"); } + if (selectedEdgeTurn && (!peerAssistedMedia || !livekitFallback)) { + throw new Error( + "Selected-edge TURN requires peer-assisted media and LiveKit fallback", + ); + } const configuredSecrets = [ hostAdmissionPassword, livekitFallback?.apiKey, livekitFallback?.apiSecret, + selectedEdgeTurn?.sharedSecret, ].filter((secret): secret is string => secret !== undefined); if (new Set(configuredSecrets).size !== configuredSecrets.length) { throw new Error( - "HOST_ADMISSION_PASSWORD, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must use independent values", + "HOST_ADMISSION_PASSWORD, LIVEKIT_API_KEY, LIVEKIT_API_SECRET, and SELECTED_EDGE_TURN_SHARED_SECRET must use independent values", ); } if ( @@ -372,6 +440,7 @@ export function loadConfig( peerAssistedMedia, peerAssistedRoomIds, livekitFallback, + selectedEdgeTurn, stunUrls, }; } diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index 69eca4a..9b856d9 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -12,6 +12,7 @@ import { type ViewerMediaTopology, } from "../shared/protocol.js"; import type { SfuTokenIssuer } from "./livekit-token.js"; +import type { SelectedEdgeTurnConfig } from "./config.js"; import { MediaRouteController, type RoomMediaRoute, @@ -22,10 +23,12 @@ import { type MediaAssignmentChange, } from "./peer-relay-topology.js"; import type { RoomStore } from "./room-store.js"; +import { issueSelectedEdgeTurnCredential } from "./selected-edge-turn.js"; type ErrorCode = Extract["code"]; const DEFAULT_SFU_PREPARE_TIMEOUT_MS = 5_000; +const SELECTED_EDGE_TURN_ATTEMPT_TIMEOUT_MS = 30_000; const VIEWER_QUALITY_BAD_WINDOWS_TO_REASSIGN = 3; const VIEWER_QUALITY_EVIDENCE_GAP_MS = 5_000; const VIEWER_QUALITY_REASSIGN_COOLDOWN_MS = 30_000; @@ -47,12 +50,28 @@ interface PendingRoutePreparation { interface ViewerRouteIntent { failedParentPeerId: string | null; + failedParentSessionId?: string; + failedConnectionId?: string; sessionId: string; sfuAttempts: number; + selectedTurnAttempted?: boolean; unavailableReported: boolean; qualityGuard?: QualityRouteIntentGuard; } +interface SelectedEdgeTurnAttempt { + roomId: string; + revision: number; + shareGeneration: string; + parentPeerId: string; + parentSessionId: string; + viewerPeerId: string; + viewerSessionId: string; + newConnectionId: string; + answered: boolean; + timer: NodeJS.Timeout; +} + interface QualityRouteIntentGuard { viewerSessionId: string; connectionId: string; @@ -105,9 +124,11 @@ export interface SfuFallbackOptions { export interface HybridMediaRouterOptions { roomStore: RoomStore; sfuFallback?: SfuFallbackOptions; + selectedEdgeTurn?: SelectedEdgeTurnConfig; sendToSession: (sessionId: string, message: ServerMessage) => void; getConnectionId: (roomId: string, viewerPeerId: string) => string | undefined; deleteConnectionId: (roomId: string, viewerPeerId: string) => void; + getShareGeneration: (roomId: string) => string | undefined; onActiveRouteChanged?: (roomId: string) => void; now?: () => number; } @@ -146,6 +167,10 @@ export class HybridMediaRouter { Map >(); private readonly consumedSfuRefreshesByRoom = new Map>(); + private readonly selectedEdgeTurnAttempts = new Map< + string, + SelectedEdgeTurnAttempt + >(); private readonly relayCapacitySessionsByRoom = new Map< string, Map @@ -162,6 +187,9 @@ export class HybridMediaRouter { constructor(private readonly options: HybridMediaRouterOptions) { const fallback = options.sfuFallback; + if (options.selectedEdgeTurn && !fallback) { + throw new Error("Selected-edge TURN requires SFU fallback"); + } if (!fallback) { return; } @@ -183,6 +211,10 @@ export class HybridMediaRouter { clearTimeout(pending.timer); } this.pendingRoutePreparations.clear(); + for (const attempt of this.selectedEdgeTurnAttempts.values()) { + clearTimeout(attempt.timer); + } + this.selectedEdgeTurnAttempts.clear(); this.mediaRouteControllers.clear(); this.failedParentPeerIdsByViewer.clear(); this.viewerRouteIntentsByRoom.clear(); @@ -281,6 +313,63 @@ export class HybridMediaRouter { ); } + selectedEdgeTurnSignalAuthorization(input: { + roomId: string; + sourcePeerId: string; + sourceSessionId: string; + targetPeerId: string; + targetSessionId: string; + connectionId: string; + signalKind: "candidate" | "description"; + descriptionType?: "offer" | "answer"; + }): boolean | undefined { + const sourceKey = viewerConnectionKey(input.roomId, input.sourcePeerId); + const targetKey = viewerConnectionKey(input.roomId, input.targetPeerId); + const attempt = + this.selectedEdgeTurnAttempts.get(sourceKey)?.parentPeerId === + input.targetPeerId + ? this.selectedEdgeTurnAttempts.get(sourceKey) + : this.selectedEdgeTurnAttempts.get(targetKey)?.parentPeerId === + input.sourcePeerId + ? this.selectedEdgeTurnAttempts.get(targetKey) + : undefined; + if (!attempt) { + return undefined; + } + const parentToViewer = input.sourcePeerId === attempt.parentPeerId; + const currentRoute = this.mediaRouteControllers.get(input.roomId)?.getActiveRoute(); + return Boolean( + input.connectionId === attempt.newConnectionId && + currentRoute?.revision === attempt.revision && + this.options.getShareGeneration(input.roomId) === + attempt.shareGeneration && + ((parentToViewer && + input.sourceSessionId === attempt.parentSessionId && + input.targetSessionId === attempt.viewerSessionId) || + (!parentToViewer && + input.sourceSessionId === attempt.viewerSessionId && + input.targetSessionId === attempt.parentSessionId)) && + (input.signalKind === "candidate" || + (!attempt.answered && + ((parentToViewer && input.descriptionType === "offer") || + (!parentToViewer && input.descriptionType === "answer")))) + ); + } + + markSelectedEdgeTurnAnswered( + roomId: string, + viewerPeerId: string, + connectionId: string, + ): void { + const attempt = this.selectedEdgeTurnAttempts.get( + viewerConnectionKey(roomId, viewerPeerId), + ); + if (attempt?.newConnectionId === connectionId) { + attempt.answered = true; + clearTimeout(attempt.timer); + } + } + resolveActivePeerEdge( roomId: string, childPeerId: string, @@ -647,9 +736,19 @@ export class HybridMediaRouter { ); return; } + if ( + isSfuRoot && + participant.role === "viewer" && + this.startSelectedEdgeTurn(participant.roomId, participant.peerId, true) + ) { + return; + } this.failBackToPeerBaseline(participant.roomId); return; } + if (this.finishSelectedEdgeTurnFailure(participant, message)) { + return; + } if (participant.role !== "viewer") { this.sendError( participant.sessionId, @@ -715,6 +814,11 @@ export class HybridMediaRouter { } roomIntents.set(participant.peerId, { failedParentPeerId: assignment.upstream.peerId, + failedParentSessionId: this.connectedPeer( + participant.roomId, + assignment.upstream.peerId, + )?.sessionId, + failedConnectionId: message.connectionId, sessionId: participant.sessionId, sfuAttempts: 0, unavailableReported: false, @@ -737,6 +841,7 @@ export class HybridMediaRouter { } disconnectParticipant(roomId: string, peerId: string): void { + this.clearSelectedEdgeTurnsForParticipant(roomId, peerId); this.clearViewerQualityStateForParticipant(roomId, peerId); const pending = this.pendingRoutePreparations.get(roomId); if (!pending || this.pendingSessionsAreCurrent(roomId, pending)) { @@ -746,6 +851,7 @@ export class HybridMediaRouter { } removeViewer(roomId: string, peerId: string): void { + this.clearSelectedEdgeTurnsForParticipant(roomId, peerId); this.clearViewerQualityStateForParticipant(roomId, peerId); this.abortPendingRouteForTopologyChange(roomId); const changes = this.peerRelayTopology.removeViewer( @@ -813,6 +919,171 @@ export class HybridMediaRouter { this.abortPendingRoute(participant.roomId); } + private startSelectedEdgeTurn( + roomId: string, + viewerPeerId: string, + requireConsumedSfuRefresh: boolean, + ): boolean { + const config = this.options.selectedEdgeTurn; + const controller = this.mediaRouteControllers.get(roomId); + const active = controller?.getActiveRoute(); + const intent = this.viewerRouteIntentsByRoom.get(roomId)?.get(viewerPeerId); + const key = viewerConnectionKey(roomId, viewerPeerId); + if (this.selectedEdgeTurnAttempts.has(key)) { + return true; + } + if ( + !config || + !active || + !intent || + intent.selectedTurnAttempted || + !intent.failedParentPeerId || + !intent.failedParentSessionId || + !intent.failedConnectionId + ) { + return false; + } + const viewer = this.options.roomStore.getConnectedViewer( + roomId, + viewerPeerId, + ); + const parent = this.connectedPeer(roomId, intent.failedParentPeerId); + const shareGeneration = this.options.getShareGeneration(roomId); + const parentAssignment = active.assignments.get(intent.failedParentPeerId); + const parentChildLimit = + intent.failedParentPeerId === + this.peerRelayTopology.getHostPeerId(roomId) + ? 2 + : 1; + if ( + viewer?.sessionId !== intent.sessionId || + parent?.sessionId !== intent.failedParentSessionId || + !parentAssignment || + (parentAssignment.childPeerIds.length + + (parentAssignment.sfuPublicationGeneration ? 1 : 0) >= + parentChildLimit && + !parentAssignment.childPeerIds.includes(viewerPeerId)) || + !shareGeneration + ) { + return false; + } + if (requireConsumedSfuRefresh) { + const publicationGeneration = active.sfu.publicationGeneration; + if ( + !publicationGeneration || + !active.sfu.rootPeerIds.includes(viewerPeerId) || + !this.consumedSfuRefreshesByRoom + .get(roomId) + ?.has( + sfuRefreshKey( + publicationGeneration, + viewerPeerId, + viewer.sessionId, + ), + ) + ) { + return false; + } + } + + intent.selectedTurnAttempted = true; + const newConnectionId = randomBytes(16).toString("base64url"); + let credential; + try { + credential = issueSelectedEdgeTurnCredential( + config, + { + roomId, + shareGeneration, + revision: active.revision, + parentPeerId: parent.peerId, + parentSessionId: parent.sessionId, + viewerPeerId, + viewerSessionId: viewer.sessionId, + oldConnectionId: intent.failedConnectionId, + newConnectionId, + }, + this.options.now?.() ?? Date.now(), + ); + } catch { + console.error("Selected-edge TURN credential issuance failed"); + return false; + } + + const timer = setTimeout(() => { + const current = this.selectedEdgeTurnAttempts.get(key); + if ( + !current || + current.newConnectionId !== newConnectionId || + current.answered + ) { + return; + } + this.selectedEdgeTurnAttempts.delete(key); + this.sendError( + current.viewerSessionId, + "PEER_NOT_FOUND", + "Selected relay edge timed out", + ); + }, SELECTED_EDGE_TURN_ATTEMPT_TIMEOUT_MS); + timer.unref(); + const attempt: SelectedEdgeTurnAttempt = { + roomId, + revision: active.revision, + shareGeneration, + parentPeerId: parent.peerId, + parentSessionId: parent.sessionId, + viewerPeerId, + viewerSessionId: viewer.sessionId, + newConnectionId, + answered: false, + timer, + }; + this.selectedEdgeTurnAttempts.set(key, attempt); + const message: Extract = { + type: "selected-edge-turn", + revision: attempt.revision, + parentPeerId: attempt.parentPeerId, + viewerPeerId, + oldConnectionId: intent.failedConnectionId, + newConnectionId, + expiresAt: credential.expiresAt, + iceServer: credential.iceServer, + }; + // Queue the Viewer grant first. Any parent offer is handled on a later + // event-loop turn, so the Viewer's socket observes this grant first. + this.options.sendToSession(viewer.sessionId, message); + this.options.sendToSession(parent.sessionId, message); + return true; + } + + private finishSelectedEdgeTurnFailure( + participant: AuthenticatedRouteParticipant, + message: Extract, + ): boolean { + if (participant.role !== "viewer" || message.connectionId === null) { + return false; + } + const key = viewerConnectionKey(participant.roomId, participant.peerId); + const attempt = this.selectedEdgeTurnAttempts.get(key); + if ( + !attempt || + attempt.viewerSessionId !== participant.sessionId || + attempt.revision !== message.revision || + attempt.newConnectionId !== message.connectionId + ) { + return false; + } + clearTimeout(attempt.timer); + this.selectedEdgeTurnAttempts.delete(key); + this.sendError( + participant.sessionId, + "PEER_NOT_FOUND", + "Selected relay edge failed", + ); + return true; + } + private failBackToPeerBaseline(roomId: string): void { this.abortPendingRoute(roomId, undefined, false); const controller = this.mediaRouteControllers.get(roomId); @@ -1007,6 +1278,9 @@ export class HybridMediaRouter { continue; } } else { + if (assignment.upstream.kind === "sfu") { + continue; + } if ( assignment.upstream.kind !== "peer" || assignment.upstream.peerId !== intent.failedParentPeerId @@ -1088,6 +1362,13 @@ export class HybridMediaRouter { intent.unavailableReported = true; } + if ( + (intent.sfuAttempts > 0 || intent.unavailableReported) && + this.startSelectedEdgeTurn(roomId, viewerPeerId, false) + ) { + return; + } + if (!intent.unavailableReported) { this.sendError( intent.sessionId, @@ -1662,6 +1943,12 @@ export class HybridMediaRouter { current.sfu.publicationGeneration === active.sfu.publicationGeneration ) { + if ( + active.sfu.rootPeerIds.includes(peerId) && + this.startSelectedEdgeTurn(roomId, peerId, true) + ) { + return; + } this.failBackToPeerBaseline(roomId); } } @@ -1758,6 +2045,12 @@ export class HybridMediaRouter { route: RoomMediaRoute, excludedPeerId?: string, ): void { + for (const [key, attempt] of this.selectedEdgeTurnAttempts) { + if (attempt.roomId === roomId && attempt.revision !== route.revision) { + clearTimeout(attempt.timer); + this.selectedEdgeTurnAttempts.delete(key); + } + } this.clearRoomViewerQualityEvidenceStates(roomId); for (const [peerId, assignment] of route.assignments) { if (peerId === excludedPeerId) { @@ -1893,12 +2186,33 @@ export class HybridMediaRouter { this.mediaRouteControllers.delete(roomId); this.clearRoomFailedParents(roomId); this.viewerRouteIntentsByRoom.delete(roomId); + for (const [key, attempt] of this.selectedEdgeTurnAttempts) { + if (attempt.roomId === roomId) { + clearTimeout(attempt.timer); + this.selectedEdgeTurnAttempts.delete(key); + } + } this.consumedSfuRefreshesByRoom.delete(roomId); this.sfuDisabledRoomIds.delete(roomId); this.clearRoomViewerQualityEvidenceStates(roomId); this.roomQualityMigrationCooldownUntilMs.delete(roomId); } + private clearSelectedEdgeTurnsForParticipant( + roomId: string, + peerId: string, + ): void { + for (const [key, attempt] of this.selectedEdgeTurnAttempts) { + if ( + attempt.roomId === roomId && + (attempt.parentPeerId === peerId || attempt.viewerPeerId === peerId) + ) { + clearTimeout(attempt.timer); + this.selectedEdgeTurnAttempts.delete(key); + } + } + } + private clearSfuRefreshesForPeer(roomId: string, peerId: string): void { const consumed = this.consumedSfuRefreshesByRoom.get(roomId); if (!consumed) { diff --git a/src/server/selected-edge-turn.ts b/src/server/selected-edge-turn.ts new file mode 100644 index 0000000..1fa1836 --- /dev/null +++ b/src/server/selected-edge-turn.ts @@ -0,0 +1,59 @@ +import { createHmac } from "node:crypto"; + +import type { SelectedEdgeTurnConfig } from "./config.js"; + +export interface SelectedEdgeTurnIdentity { + roomId: string; + shareGeneration: string; + revision: number; + parentPeerId: string; + parentSessionId: string; + viewerPeerId: string; + viewerSessionId: string; + oldConnectionId: string; + newConnectionId: string; +} + +export function issueSelectedEdgeTurnCredential( + config: SelectedEdgeTurnConfig, + identity: SelectedEdgeTurnIdentity, + nowMs: number, +) { + if (!Number.isFinite(nowMs) || nowMs < 0) { + throw new Error("Selected-edge TURN grant time is invalid"); + } + const expiresAtSeconds = + Math.floor(nowMs / 1_000) + config.credentialTtlSeconds; + const opaqueIdentity = createHmac("sha256", config.sharedSecret) + .update( + JSON.stringify([ + "screener-selected-edge-turn-v1", + identity.roomId, + identity.shareGeneration, + identity.revision, + identity.parentPeerId, + identity.parentSessionId, + identity.viewerPeerId, + identity.viewerSessionId, + identity.oldConnectionId, + identity.newConnectionId, + expiresAtSeconds, + ]), + ) + .digest("base64url") + .slice(0, 32); + const username = `${expiresAtSeconds}:${opaqueIdentity}`; + const credential = createHmac("sha1", config.sharedSecret) + .update(username) + .digest("base64"); + const expiresAtMs = expiresAtSeconds * 1_000; + + return { + iceServer: { + urls: [...config.urls] as [string], + username, + credential, + }, + expiresAt: new Date(expiresAtMs).toISOString(), + }; +} diff --git a/src/server/signaling.ts b/src/server/signaling.ts index 5b090a5..94601f7 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -25,6 +25,7 @@ import { RoomStoreError, type ConnectedPeer, } from "./room-store.js"; +import type { SelectedEdgeTurnConfig } from "./config.js"; import { HybridMediaRouter, type SfuFallbackOptions, @@ -73,6 +74,7 @@ export interface SignalingOptions { peerAssistedMedia: boolean; peerAssistedRoomIds?: ReadonlySet; sfuFallback?: SfuFallbackOptions; + selectedEdgeTurn?: SelectedEdgeTurnConfig; ice: IceConfigOptions; allowedOrigins: ReadonlySet; hostAdmissionAtUpgrade: (request: IncomingMessage) => boolean; @@ -149,6 +151,7 @@ export class SignalingServer { this.hybridMediaRouter = new HybridMediaRouter({ roomStore: options.roomStore, sfuFallback: options.sfuFallback, + selectedEdgeTurn: options.selectedEdgeTurn, sendToSession: (sessionId, message) => this.sendToSession(sessionId, message), getConnectionId: (roomId, viewerPeerId) => @@ -157,6 +160,7 @@ export class SignalingServer { ), deleteConnectionId: (roomId, viewerPeerId) => this.deleteViewerConnectionId(roomId, viewerPeerId), + getShareGeneration: (roomId) => this.shareGenerationsByRoom.get(roomId), onActiveRouteChanged: (roomId) => this.sendViewerPresence(roomId), now: this.now, }); @@ -1028,11 +1032,23 @@ export class SignalingServer { message.payload.kind === "description" ? message.payload.description : undefined; - const authorized = description + const selectedEdgeAuthorized = + this.hybridMediaRouter!.selectedEdgeTurnSignalAuthorization({ + roomId: source.roomId, + sourcePeerId: source.peerId, + sourceSessionId: this.socketStates.get(sourceSocket)?.sessionId ?? "", + targetPeerId: target.peerId, + targetSessionId: target.sessionId, + connectionId: message.payload.connectionId, + signalKind: message.payload.kind, + ...(description ? { descriptionType: description.type } : {}), + }); + const assignedEdgeAuthorized = description ? description.type === "offer" ? parentToChild : childToParent : parentToChild || childToParent; + const authorized = selectedEdgeAuthorized ?? assignedEdgeAuthorized; if (!authorized) { this.sendError( sourceSocket, @@ -1054,6 +1070,13 @@ export class SignalingServer { fromPeerId: source.peerId, payload: message.payload, }); + if (selectedEdgeAuthorized && description?.type === "answer") { + this.hybridMediaRouter!.markSelectedEdgeTurnAnswered( + source.roomId, + source.peerId, + message.payload.connectionId, + ); + } } private routePeerAssistedRestart( diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index eda76f1..ab0c57a 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -9,6 +9,7 @@ export const ROOM_CODE_LENGTH = 12; export const MAX_MEDIA_ROUTE_REVISION = Number.MAX_SAFE_INTEGER; export const MAX_SFU_TOKEN_LENGTH = 8 * 1024; export const MAX_ICE_SERVER_URLS = 8; +export const MAX_TURN_CREDENTIAL_LENGTH = 256; export const MAX_VIEWER_QUALITY_EVIDENCE_BYTES = 2 * 1024; export const MAX_PARENT_EDGE_QUALITY_EVIDENCE_BYTES = 2 * 1024; export const VIEWER_QUALITY_EVIDENCE_INTERVAL_MS = 2_000; @@ -192,6 +193,46 @@ export const stunUrlSchema = z .max(512) .refine(isValidStunUrl, { message: "Invalid STUN URL" }); +function isValidTurnUrl(value: string): boolean { + const match = /^turn:([^/?#\s]+)\?transport=udp$/i.exec(value); + if (!match) { + return false; + } + let authority: URL; + try { + authority = new URL(`http://${match[1]}`); + } catch { + return false; + } + return Boolean( + authority.hostname && + !authority.username && + !authority.password && + authority.pathname === "/" && + !authority.search && + !authority.hash && + (!authority.port || Number(authority.port) > 0) + ); +} + +export const turnUrlSchema = z + .string() + .min(1) + .max(512) + .refine(isValidTurnUrl, { message: "Invalid TURN URL" }); + +const selectedEdgeTurnIceServerSchema = z + .object({ + urls: z.array(turnUrlSchema).length(1), + username: z + .string() + .min(1) + .max(128) + .regex(/^[1-9]\d{0,12}:[A-Za-z0-9_-]{16,64}$/), + credential: z.string().min(1).max(MAX_TURN_CREDENTIAL_LENGTH), + }) + .strict(); + const iceServerSchema = z .object({ urls: z.union([ @@ -607,6 +648,18 @@ export const serverMessageSchema = z.union([ token: z.string().min(1).max(MAX_SFU_TOKEN_LENGTH), }) .strict(), + z + .object({ + type: z.literal("selected-edge-turn"), + revision: mediaRouteRevisionSchema, + parentPeerId: opaqueIdSchema, + viewerPeerId: opaqueIdSchema, + oldConnectionId: opaqueIdSchema, + newConnectionId: opaqueIdSchema, + expiresAt: z.string().datetime(), + iceServer: selectedEdgeTurnIceServerSchema, + }) + .strict(), z .object({ type: z.literal("quality-settings"), diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index 3ef997d..f23d99b 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -819,6 +819,50 @@ describe("HostPeer source replacement", () => { }); describe("ViewerRelay downstream ownership", () => { + it("rebuilds a retired child as one relay-only selected edge", async () => { + const sendSignal = vi.fn(() => true); + const relay = new ViewerRelay( + { iceServers: [] }, + QUALITY_PROFILES["720p30"], + { sendSignal }, + ); + relay.setChild("selected-child"); + relay.setStream(createStream(createTrack("video", "selected-video"), null)); + await vi.waitFor(() => expect(sendSignal).toHaveBeenCalledOnce()); + const oldConnectionId = relay.getSnapshot()!.connectionId; + relay.setChild(null); + sendSignal.mockClear(); + const iceServer = { + urls: ["turn:turn.example.test:3478?transport=udp"], + username: "1787230000:opaque_identity_12345678", + credential: "short-lived-credential", + }; + + expect(relay.startSelectedEdgeTurn({ + type: "selected-edge-turn", + revision: 7, + parentPeerId: "selected-parent", + viewerPeerId: "selected-child", + oldConnectionId, + newConnectionId: "selected-connection-new", + expiresAt: "2026-08-20T12:00:00.000Z", + iceServer, + })).toBe(true); + await vi.waitFor(() => expect(sendSignal).toHaveBeenCalledOnce()); + expect(sendSignal).toHaveBeenCalledWith( + "selected-child", + expect.objectContaining({ + kind: "description", + connectionId: "selected-connection-new", + description: expect.objectContaining({ type: "offer" }), + }), + ); + expect(FakePeerConnection.latest?.configurations).toEqual([{ + iceServers: [iceServer], + iceTransportPolicy: "relay", + }]); + }); + it("exposes a defensive snapshot of current downstream send metrics", async () => { const relay = new ViewerRelay( { iceServers: [] }, diff --git a/tests/selected-edge-turn.test.ts b/tests/selected-edge-turn.test.ts new file mode 100644 index 0000000..73e4084 --- /dev/null +++ b/tests/selected-edge-turn.test.ts @@ -0,0 +1,42 @@ +import { createHmac } from "node:crypto"; +import { describe, expect, it } from "vitest"; +import { issueSelectedEdgeTurnCredential } from "../src/server/selected-edge-turn.ts"; +const config = { + urls: ["turn:turn.example.test:3478?transport=udp"] as [string], + sharedSecret: "t".repeat(32), + credentialTtlSeconds: 120, +}; +const identity = { + roomId: "123456789012", + shareGeneration: "share_generation_12345678", + revision: 7, + parentPeerId: "parent_peer_12345678", + parentSessionId: "parent_session_12345678", + viewerPeerId: "viewer_peer_12345678", + viewerSessionId: "viewer_session_12345678", + oldConnectionId: "old_connection_12345678", + newConnectionId: "new_connection_12345678", +}; +describe("selected-edge TURN", () => { + it("issues an opaque, expiring coturn REST credential bound to the edge", () => { + const grant = issueSelectedEdgeTurnCredential(config, identity, 1_700_000_000_000); + expect(grant.expiresAt).toBe("2023-11-14T22:15:20.000Z"); + expect(grant.iceServer.username).toMatch(/^1700000120:[A-Za-z0-9_-]{32}$/); + expect(grant.iceServer.credential).toBe( + createHmac("sha1", config.sharedSecret) + .update(grant.iceServer.username) + .digest("base64"), + ); + const serialized = JSON.stringify(grant); + for (const value of Object.values(identity).filter((value) => typeof value === "string")) { + expect(serialized).not.toContain(String(value)); + } + expect( + issueSelectedEdgeTurnCredential( + config, + { ...identity, shareGeneration: "share_generation_87654321" }, + 1_700_000_000_000, + ).iceServer.username, + ).not.toBe(grant.iceServer.username); + }); +}); diff --git a/tests/server-config.test.ts b/tests/server-config.test.ts index 8232141..dc012f7 100644 --- a/tests/server-config.test.ts +++ b/tests/server-config.test.ts @@ -14,6 +14,53 @@ describe("server configuration", () => { expect(config.peerAssistedMedia).toBe(false); expect(config.peerAssistedRoomIds).toBeUndefined(); expect(config.livekitFallback).toBeUndefined(); + expect(config.selectedEdgeTurn).toBeUndefined(); + }); + + it("enables selected-edge TURN only from its complete post-SFU tuple", () => { + const base = { + PEER_ASSISTED_MEDIA: "true", + PEER_ASSISTED_ROOM_IDS: "1", + LIVEKIT_URL: "wss://livekit.test", + LIVEKIT_API_KEY: "test-key", + LIVEKIT_API_SECRET: "s".repeat(32), + }; + const turn = { + SELECTED_EDGE_TURN_URLS: "turn:turn.test:3478?transport=udp", + SELECTED_EDGE_TURN_SHARED_SECRET: "t".repeat(32), + SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS: "120", + }; + expect(loadConfig({ ...base, ...turn }).selectedEdgeTurn).toEqual({ + urls: [turn.SELECTED_EDGE_TURN_URLS], + sharedSecret: turn.SELECTED_EDGE_TURN_SHARED_SECRET, + credentialTtlSeconds: 120, + }); + expect(loadConfig({ ...base, ...turn }).stunUrls).toEqual([]); + for (const name of Object.keys(turn)) { + const partial = { ...base, ...turn }; + delete partial[name as keyof typeof partial]; + expect(() => loadConfig(partial)).toThrow("must be configured together"); + } + }); + + it.each([ + "turn:turn.test:3478", + "turns:turn.test:5349?transport=udp", + "turn:user@turn.test:3478?transport=udp", + "turn:turn.test:3478?transport=tcp", + ])("rejects a non-canonical selected-edge TURN URL: %s", (url) => { + expect(() => + loadConfig({ + PEER_ASSISTED_MEDIA: "true", + PEER_ASSISTED_ROOM_IDS: "1", + LIVEKIT_URL: "wss://livekit.test", + LIVEKIT_API_KEY: "test-key", + LIVEKIT_API_SECRET: "s".repeat(32), + SELECTED_EDGE_TURN_URLS: url, + SELECTED_EDGE_TURN_SHARED_SECRET: "t".repeat(32), + SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS: "120", + }), + ).toThrow("one UDP TURN URL"); }); it("enables LiveKit fallback only for a complete credential tuple", () => { @@ -151,7 +198,7 @@ describe("server configuration", () => { }, ])("rejects reused infrastructure secrets", (environment) => { expect(() => loadConfig(environment)).toThrow( - "HOST_ADMISSION_PASSWORD, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must use independent values", + "must use independent values", ); }); diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index a8fc480..8bd122d 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -289,6 +289,7 @@ async function correlateParentEdgeQualityEvidence( async function startSfuHarness(options: { tokenIssuer: SfuTokenIssuer; + selectedEdgeTurn?: boolean; prepareTimeoutMs?: number; maxRoots?: number; viewerDisconnectGraceMs?: number; @@ -317,6 +318,15 @@ async function startSfuHarness(options: { maxRoots: options.maxRoots ?? 2, prepareTimeoutMs: options.prepareTimeoutMs, }, + ...(options.selectedEdgeTurn + ? { + selectedEdgeTurn: { + urls: ["turn:turn.example.test:3478?transport=udp"], + sharedSecret: "t".repeat(32), + credentialTtlSeconds: 120, + }, + } + : {}), ice: { stunUrls: options.stunUrls ?? [], }, @@ -630,6 +640,63 @@ async function prepareFallbackForTwoViewers( }; } +async function failSingleViewerDirect( + webSocketUrl: string, + room: CreatedRoom, + prefix: string, +) { + const host = await openClient(webSocketUrl); + const hostAuth = peerAssisted( + await authenticate(host, room, "host", `${prefix}-host`, 1, `${prefix}-share`), + ); + const viewer = await openClient(webSocketUrl); + const viewerAuth = peerAssisted( + await authenticate(viewer, room, "viewer", `${prefix}-viewer`), + ); + const oldConnectionId = `${prefix}-direct-connection`; + host.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: viewerAuth.peerId, + payload: { + kind: "description", + connectionId: oldConnectionId, + description: { type: "offer", sdp: "v=0\r\n" }, + }, + })); + await viewer.inbox.next("signal"); + viewer.socket.send(JSON.stringify({ + type: "route-failed", + revision: viewerAuth.routeRevision, + phase: "active", + connectionId: oldConnectionId, + })); + return { host, hostAuth, viewer, viewerAuth, oldConnectionId }; +} + +async function activateSingleViewerSfu( + webSocketUrl: string, + room: CreatedRoom, + prefix: string, +) { + const direct = await failSingleViewerDirect(webSocketUrl, room, prefix); + const { host, viewer } = direct; + const hostPrepare = await nextPreparedRoute(host); + const viewerPrepare = await nextPreparedRoute(viewer); + await Promise.all([host.inbox.next("sfu-config"), viewer.inbox.next("sfu-config")]); + for (const client of [host, viewer]) { + client.socket.send(JSON.stringify({ + type: "route-ready", + revision: viewerPrepare.revision, + phase: "prepare", + })); + } + await Promise.all([ + nextActiveRouteRevision(host, hostPrepare.revision), + nextActiveRouteRevision(viewer, viewerPrepare.revision), + ]); + return { ...direct, revision: viewerPrepare.revision }; +} + async function exhaustDeepViewerPeerRoutes( webSocketUrl: string, room: CreatedRoom, @@ -3599,6 +3666,169 @@ describe("WebSocket signaling", () => { expect(issued.length).toBeGreaterThan(issuedBeforeFailback); }); + it("falls through to selected TURN when SFU prepare is unavailable", async () => { + const harness = await startSfuHarness({ + tokenIssuer: { async issueToken() { throw new Error("offline"); } }, + selectedEdgeTurn: true, + }); + const direct = await failSingleViewerDirect( + harness.webSocketUrl, + harness.room, + "selected-unavailable", + ); + const [viewerGrant] = await Promise.all([ + direct.viewer.inbox.next("selected-edge-turn"), + direct.host.inbox.next("selected-edge-turn"), + ]); + expect(viewerGrant.oldConnectionId).toBe(direct.oldConnectionId); + }); + + it("uses one selected TURN edge after a Viewer SFU refresh is exhausted", async () => { + const harness = await startSfuHarness({ + tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, + selectedEdgeTurn: true, + }); + const direct = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "selected-host-parent", + ); + expect(direct.hostAuth.iceConfig.iceServers).toEqual([]); + direct.viewer.socket.send(JSON.stringify({ + type: "refresh-sfu", + revision: direct.revision, + })); + await direct.viewer.inbox.next("sfu-config"); + direct.viewer.socket.send(JSON.stringify({ + type: "route-failed", + revision: direct.revision, + phase: "active", + connectionId: null, + })); + const [viewerGrant, hostGrant] = await Promise.all([ + direct.viewer.inbox.next("selected-edge-turn"), + direct.host.inbox.next("selected-edge-turn"), + ]); + expect(hostGrant).toEqual(viewerGrant); + expect(viewerGrant).toMatchObject({ + revision: direct.revision, + parentPeerId: direct.hostAuth.peerId, + viewerPeerId: direct.viewerAuth.peerId, + oldConnectionId: direct.oldConnectionId, + iceServer: { urls: ["turn:turn.example.test:3478?transport=udp"] }, + }); + expect(viewerGrant.newConnectionId).not.toBe(direct.oldConnectionId); + direct.host.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: direct.viewerAuth.peerId, + payload: { + kind: "description", + connectionId: viewerGrant.newConnectionId, + description: { type: "offer", sdp: "v=0\r\n" }, + }, + })); + expect(await direct.viewer.inbox.next("signal")).toMatchObject({ + fromPeerId: direct.hostAuth.peerId, + payload: { connectionId: viewerGrant.newConnectionId }, + }); + direct.viewer.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: direct.hostAuth.peerId, + payload: { + kind: "description", + connectionId: viewerGrant.newConnectionId, + description: { type: "answer", sdp: "v=0\r\n" }, + }, + })); + await direct.host.inbox.next("signal"); + direct.viewer.socket.send(JSON.stringify({ + type: "route-failed", + revision: direct.revision, + phase: "active", + connectionId: viewerGrant.newConnectionId, + })); + expect((await direct.viewer.inbox.next("error")).code).toBe("PEER_NOT_FOUND"); + await expect( + direct.viewer.inbox.next("selected-edge-turn", 40), + ).rejects.toThrow("Timed out"); + }); + + it("authorizes a ViewerRelay parent and clears its grant on share stop", async () => { + const harness = await startSfuHarness({ + tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, + selectedEdgeTurn: true, + }); + const prepared = await prepareFallbackForTwoViewers( + harness.webSocketUrl, + harness.room, + "selected-viewer-parent", + "selected-viewer-parent-share", + ); + for (const client of [prepared.host, prepared.failedViewer]) { + client.socket.send(JSON.stringify({ + type: "route-ready", + revision: prepared.rootPrepare.revision, + phase: "prepare", + })); + } + await Promise.all([ + nextActiveRouteRevision(prepared.host, prepared.hostPrepare.revision), + nextActiveRouteRevision(prepared.failedViewer, prepared.rootPrepare.revision), + ]); + prepared.failedViewer.socket.send(JSON.stringify({ + type: "refresh-sfu", + revision: prepared.rootPrepare.revision, + })); + await prepared.failedViewer.inbox.next("sfu-config"); + prepared.failedViewer.socket.send(JSON.stringify({ + type: "route-failed", + revision: prepared.rootPrepare.revision, + phase: "active", + connectionId: null, + })); + const [childGrant, parentGrant] = await Promise.all([ + prepared.failedViewer.inbox.next("selected-edge-turn"), + prepared.rootViewer.inbox.next("selected-edge-turn"), + ]); + expect(parentGrant).toEqual(childGrant); + expect(childGrant.parentPeerId).toBe(prepared.rootAuth.peerId); + prepared.rootViewer.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: prepared.failedAuth.peerId, + payload: { + kind: "description", + connectionId: childGrant.newConnectionId, + description: { type: "offer", sdp: "v=0\r\n" }, + }, + })); + await prepared.failedViewer.inbox.next("signal"); + prepared.failedViewer.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: prepared.rootAuth.peerId, + payload: { + kind: "description", + connectionId: childGrant.newConnectionId, + description: { type: "answer", sdp: "v=0\r\n" }, + }, + })); + await prepared.rootViewer.inbox.next("signal"); + prepared.host.socket.send(JSON.stringify({ + type: "stop-sharing", + shareGeneration: "selected-viewer-parent-share", + })); + await prepared.failedViewer.inbox.next("sharing-stopped"); + prepared.rootViewer.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: prepared.failedAuth.peerId, + payload: { + kind: "candidate", + connectionId: childGrant.newConnectionId, + candidate: null, + }, + })); + expect((await prepared.rootViewer.inbox.next("error")).code).toBe("FORBIDDEN"); + }); + it("moves both a failed deep viewer and its host branch within two SFU roots", async () => { const issued: Array[0]> = []; const harness = await startSfuHarness({ @@ -3721,6 +3951,7 @@ describe("WebSocket signaling", () => { return `token-${request.peerId}`; }, }, + selectedEdgeTurn: true, }); const host = await openClient(harness.webSocketUrl); const hostAuth = peerAssisted( @@ -3785,6 +4016,20 @@ describe("WebSocket signaling", () => { peerId: hostAuth.peerId, }); expect(directAuth.routeAssignment.sfuPublicationGeneration).toBeNull(); + root.socket.send(JSON.stringify({ + type: "refresh-sfu", + revision: directAuth.routeRevision, + })); + await root.inbox.next("sfu-config"); + root.socket.send(JSON.stringify({ + type: "route-failed", + revision: directAuth.routeRevision, + phase: "active", + connectionId: null, + })); + await expect(root.inbox.next("selected-edge-turn", 40)).rejects.toThrow( + "Timed out", + ); }); it("retires an SFU publication when its root allowlist shrinks", async () => {