From 78871db80549dfa7e8e6a19843b2b2e2da290b5f Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sat, 22 Aug 2026 21:59:44 +0800 Subject: [PATCH] refactor(route): unify endpoint media copy accounting (#200) --- .env.example | 7 +- docs/adr/0002-persistent-protected-rooms.md | 8 +- .../0006-fixed-high-native-sender-canary.md | 6 +- docs/deployment.md | 35 +- docs/project-memory.md | 1 + docs/research/native-shared-encode-sender.md | 6 +- docs/research/peer-assisted-media.md | 2 +- docs/status.md | 5 +- docs/方案设计.md | 10 +- native/sender/internal/app/server_test.go | 4 +- native/sender/internal/remote/wire.go | 2 +- native/sender/internal/remote/wire_test.go | 6 +- scripts/peer-assisted-benchmark.ts | 31 +- src/client/media/host-provisional-child.ts | 23 +- src/client/media/relay-capability.ts | 8 +- src/client/pages/HostPage.tsx | 8 +- src/client/pages/ViewerPage.tsx | 8 +- src/client/webrtc/media-assignment.ts | 9 +- src/client/webrtc/viewer-relay.ts | 12 +- src/server/app.ts | 2 +- src/server/config.ts | 25 +- src/server/hybrid-media-router.ts | 518 ++++++-------- src/server/media-route-controller.ts | 95 ++- src/server/peer-relay-topology.ts | 59 +- src/server/signaling.ts | 4 +- src/shared/media-copy-accounting.ts | 74 ++ src/shared/protocol.ts | 15 +- tests/client-state.test.ts | 10 +- tests/host-peer-tracks.test.ts | 48 +- tests/media-assignment.test.ts | 20 +- tests/media-copy-accounting.test.ts | 73 ++ tests/media-route-controller.test.ts | 39 +- tests/peer-assisted-benchmark.test.ts | 11 +- tests/peer-relay-topology.test.ts | 74 +- tests/protocol.test.ts | 12 + tests/relay-capability.test.ts | 4 +- tests/server-config.test.ts | 32 +- tests/server-http.test.ts | 2 +- tests/server-signal.test.ts | 663 ++++++++++-------- 39 files changed, 1000 insertions(+), 971 deletions(-) create mode 100644 src/shared/media-copy-accounting.ts create mode 100644 tests/media-copy-accounting.test.ts diff --git a/.env.example b/.env.example index c49dd05..370e82a 100644 --- a/.env.example +++ b/.env.example @@ -19,8 +19,9 @@ MAX_VIEWERS_PER_ROOM=8 # Bounded standard-WebRTC relay/SFU controller for every room when true. # Keep false outside measured tests. PEER_ASSISTED_MEDIA=false -# Release upload ceiling: Host 2, ordinary Browser Viewer 1. Range 1-2 only tightens it. -MAX_PEER_RELAY_DOWNSTREAM_EDGES=2 +# One steady outbound media-copy capacity for every non-server endpoint. +# A peer child or Host SFU publication consumes one copy. Valid values: 1, 2, 3. +ENDPOINT_MEDIA_COPY_CAPACITY=2 # Optional automatic SFU fallback capacity. Set PEER_ASSISTED_MEDIA=true and # supply all three values together; leaving them empty keeps LiveKit completely @@ -51,4 +52,4 @@ STUN_URLS= # LIVEKIT_API_KEY=replace-with-a-livekit-api-key # LIVEKIT_API_SECRET=replace-with-at-least-32-independent-random-bytes # MAX_SFU_ROOTS_PER_ROOM=2 -# MAX_PEER_RELAY_DOWNSTREAM_EDGES=2 +# ENDPOINT_MEDIA_COPY_CAPACITY=2 diff --git a/docs/adr/0002-persistent-protected-rooms.md b/docs/adr/0002-persistent-protected-rooms.md index 9d5790a..db6cbc6 100644 --- a/docs/adr/0002-persistent-protected-rooms.md +++ b/docs/adr/0002-persistent-protected-rooms.md @@ -57,7 +57,7 @@ Keep site access and room authorization as two ordered checks: `screener-v1` to `screener-v2`. That migration had no negotiation, v1 parser, dual write, or translator. A v1 first message received the universal fatal refresh outcome before room lookup and did not reconnect. Later atomic - migrations advanced the current wire to `screener-v5`. + migrations advanced the source wire to `screener-v6`. Every room has one of two Viewer policies: @@ -98,9 +98,9 @@ material and socket/Host session so stale asynchronous work cannot authorize or commit. At this ADR's implementation checkpoint, the wire remained `screener-v2`. -Current source and production have since migrated atomically to `screener-v5`; -the paragraph below records the v2 compatibility boundary rather than the -current wire. Only a Web Host that advertises +Current source has since migrated atomically to `screener-v6`, while production +remains on `screener-v5`; the paragraph below records the v2 compatibility +boundary rather than the current source wire. Only a Web Host that advertises `viewerPasswordSettings: true` may set/remove the password and receive `viewer-password-updated`; Native v2 does not advertise the capability and sees no new message type. Viewer `authenticate` may carry the current password diff --git a/docs/adr/0006-fixed-high-native-sender-canary.md b/docs/adr/0006-fixed-high-native-sender-canary.md index 6b0a35d..7c55b07 100644 --- a/docs/adr/0006-fixed-high-native-sender-canary.md +++ b/docs/adr/0006-fixed-high-native-sender-canary.md @@ -39,9 +39,9 @@ merged as product code from this state. ## 2026-08-20 Historical Wire Checkpoint This checkpoint targeted the then-current `screener-v2` wire and SQLite schema -v2. Current Web and source-only Native use `screener-v5`, and current room -persistence is SQLite schema v3. The dated behavior below remains experiment -provenance, not a current integration contract. +v2. Current Web and source-only Native use `screener-v6`, production remains on +`screener-v5`, and current room persistence is SQLite schema v3. The dated +behavior below remains experiment provenance, not a current integration contract. A clean local delivery candidate now targets only `screener-v2`: it exchanges `SITE_ACCESS_PASSWORD` for the bounded HttpOnly site-access cookie, uses diff --git a/docs/deployment.md b/docs/deployment.md index d835048..88e107f 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -2,9 +2,9 @@ Last verified against upstream documentation: 2026-08-21. -This page records exact release `9461e20` deployment behavior and syntax. Current -product direction and pending migrations are owned by [project memory](./project-memory.md) -and [the TODO ledger](./todo.md). +This page records exact release `9461e20` production facts and the current source +deployment contract. Product direction and pending migrations are owned by +[project memory](./project-memory.md) and [the TODO ledger](./todo.md). This section documents the repository's UDP-only deployment candidate: one Node.js process provides the built Web client, room API, and WebSocket signaling @@ -194,7 +194,7 @@ manager or a cross-platform Node command such as: node -e "console.log(require('node:crypto').randomBytes(32).toString('base64url'))" ``` -Use this production baseline: +Use this next-release baseline for the current source candidate: ```dotenv NODE_ENV=production @@ -207,6 +207,7 @@ ROOM_DATABASE_PATH=/var/lib/screener/rooms.sqlite ROOM_TTL_SECONDS=14400 MAX_ROOMS=1000 MAX_VIEWERS_PER_ROOM=8 +ENDPOINT_MEDIA_COPY_CAPACITY=2 STUN_URLS=stun:stun.example.com:3478 ``` @@ -215,7 +216,6 @@ To make automatic SFU fallback capacity available, add the complete tuple: ```dotenv PEER_ASSISTED_MEDIA=true -MAX_PEER_RELAY_DOWNSTREAM_EDGES=2 LIVEKIT_URL=wss://share.example.com LIVEKIT_API_KEY= LIVEKIT_API_SECRET= @@ -264,15 +264,22 @@ Omit the database path to keep random temporary rooms; `ROOM_TTL_SECONDS` applies only to those rooms. `MAX_VIEWERS_PER_ROOM` defaults to 8 and accepts 1 through 16. It is an admission limit, not evidence that the publisher can sustain that many streams. -In exact release `9461e20`, `MAX_PEER_RELAY_DOWNSTREAM_EDGES` defaults to 2 and accepts only 1 or 2. It can -tighten but cannot lift release policy: effective Host capacity is -`min(value, 2)` and effective ordinary Browser Viewer capacity is -`min(value, 1)`. Active/provisional/selected physical downstream overlays count, -and an active Host SFU publication uses one Host edge; ordinary upstream receive -does not. The current Browser advertises one. The unchanged protocol accepts -`0/1/2/3` only as a future capability envelope, so an old or malicious Viewer -advertising two or three remains effectively one. This setting does not change -the separate SFU root limit, which remains at most two. +`ENDPOINT_MEDIA_COPY_CAPACITY` defaults to 2 and accepts only 1, 2, or 3. It is +the single server-authoritative steady outbound media-copy cap for Host and +Viewer endpoints; role, browser, UA, and visibility do not create another tier. +A peer child or the Host's single SFU publication consumes one copy, upstream +receive is free, a committed selected TURN transport for the same child does not +consume a second copy, and an uncommitted selected carry does. Transition work +may reach only `min(C + 1, 3)`. A Host already sending three copies at `C=3` +must fail or wait before a fourth publication token or TURN grant is issued. +Supplying the removed `MAX_PEER_RELAY_DOWNSTREAM_EDGES`, even blank, fails +startup. The fixed per-room SFU-root and selected-lease guards remain separate +temporary server safety boundaries. + +Production release `9461e20` still runs the legacy Host-2/Browser-1 policy on +wire `screener-v5`. Deploy the current source server, Web assets, and Native +sender atomically on `screener-v6`; restore the exact prior environment and +release together when rolling back. The three `LIVEKIT_*` values must either all be absent or all be present, and a complete tuple requires `PEER_ASSISTED_MEDIA=true`. An empty tuple keeps the diff --git a/docs/project-memory.md b/docs/project-memory.md index 4837d6d..aa3227b 100644 --- a/docs/project-memory.md +++ b/docs/project-memory.md @@ -21,6 +21,7 @@ Last updated: 2026-08-22 ## Current Source And Production - Canonical source is the clean `main` branch. New branches and worktrees start from its exact current commit after accepted truth is merged. +- The Route wave A merge candidate implements the uniform `1/2/3` endpoint cap through one shared accounting guard, atomically advances Web/server/Native to `screener-v6`, rejects v5 before room authority, treats SFU-fed Viewers as leaves without outbound proof, and retains the fixed SFU-root and room-wide selected-lease guards pending later server admission waves. - Production runs exact `9461e207af62b4f38f6b7a8aa16f8beb49e0e4ee`, release `9461e20`, wire `screener-v5`; `21d5cd9f7139` is the rollback release. - Production still enforces Host downstream `2`, ordinary Browser downstream `1`, deployment values `1/2`, a fixed SFU-root limit of `2`, and legacy publication/selected accounting. This is a dated implementation divergence, not current product policy. - Production has automatic routing, bad-relay corroboration, peer-quality make-before-break with Viewer and Host provisional parents, selected-edge transport, stable route/connection details, 64/128/256 audio choices, access controls, self-check, diagnostic export, and signaling watchdog behavior. diff --git a/docs/research/native-shared-encode-sender.md b/docs/research/native-shared-encode-sender.md index 89c8f58..a1a0bc7 100644 --- a/docs/research/native-shared-encode-sender.md +++ b/docs/research/native-shared-encode-sender.md @@ -166,9 +166,9 @@ transient room TTL still caps the room, and restart drops it. A pre-auth connection or authentication failure gets one same-room/token/client-generation retry without a second room POST. SQLite remained v2 and ordinary Web creation was unchanged. Current Web and source-only Native have since moved to -`screener-v5`, and current room persistence is SQLite schema v3. Retaining this -paragraph records the dated candidate boundary; it does not authorize new work -against v2. +`screener-v6`, production remains on `screener-v5`, and current room persistence +is SQLite schema v3. Retaining this paragraph records the dated candidate +boundary; it does not authorize new work against v2. Android stop/capture-failure cleanup keeps the established signaling socket until its single-thread teardown sends `abandon-room` and then performs a normal close; pending-work cancellation only blocks callbacks and cancels active HTTP. No ACK diff --git a/docs/research/peer-assisted-media.md b/docs/research/peer-assisted-media.md index 00c6f07..f1b17d6 100644 --- a/docs/research/peer-assisted-media.md +++ b/docs/research/peer-assisted-media.md @@ -525,7 +525,7 @@ event arrives later. The application therefore needs a small text-message challenge rather than waiting for either protocol Ping/Pong or the old socket's close event. -The retained compatibility seam is response-only opt-in under `screener-v5`: +The retained compatibility seam is response-only opt-in under `screener-v6`: the Web client sends `signaling-challenge { sequence }`, and the server sends the exact `signaling-challenge-response { sequence }` only to that requesting socket. The Native sender's strict decoder rejects unknown server message diff --git a/docs/status.md b/docs/status.md index 944b816..60f77ae 100644 --- a/docs/status.md +++ b/docs/status.md @@ -14,9 +14,10 @@ This is the current execution index. Git history owns completed timelines; [veri ## Current Source - Canonical source is the clean `main` branch; auxiliary branches and worktrees do not own current truth. -- The accepted endpoint direction is one steady outbound media-copy cap for every non-server endpoint: default `2`, static `1/2/3`, a peer child or Host publication consumes one slot, upstream receive is free, selected TURN replaces the same copy's transport, and no Browser/UA/visibility tier exists. +- The Route wave A merge candidate implements one steady outbound media-copy cap for every non-server endpoint: default `2`, static `1/2/3`, a peer child or Host publication consumes one slot, upstream receive is free, a committed selected TURN transport replaces the same copy while hidden carry consumes another, and no Browser/UA/visibility tier exists. One shared guard owns the accounting and transition work is bounded by `min(C + 1, 3)`. +- The same candidate atomically advances Web/server/Native to `screener-v6`, rejects v5 before room authority, leaves an SFU-fed Viewer childless without independent outbound proof, and keeps quality events child-scoped without a fixed two-child parent drain. Fixed SFU-root and room-wide selected-lease guards remain until their independent server admission owners land. - Automatic P2P-first routing, one active upstream, an acyclic graph, exact identity/generation authority, media-proven make-before-break, and bounded failure remain accepted. -- The holistic assisted-route model is accepted: one controller performs allocation, child reparenting, and relay abdication/drain. One Host publication serves SFU subscriptions; TURN is an exact edge transport, not a topology node. Endpoint cap, transition overlap, media-binding generations, and independent SFU/TURN admission are governed by one bounded transaction and remain runtime validation work. +- The holistic assisted-route model is accepted: one controller performs allocation, child reparenting, and relay abdication/drain. One Host publication serves SFU subscriptions; TURN is an exact edge transport, not a topology node. Wave A supplies endpoint accounting and the wire boundary without pretending that later controller convergence, media-binding generations, or independent SFU/TURN admission already exist. - Parent selection is deterministic rather than random or score-based: hard eligibility filters, then shallowest result, remaining sender capacity, stable join order, and peer identity; one provisional edge uses standard ICE and media proof at a time. Viewer connection feedback and the strict-NAT emulator gate are accepted follow-up work, not deployed behavior. - Live 64/128/256 audio-ceiling mutation, a source-checkout LAN one-command launcher, and paused-share video codec switching are accepted but not yet implemented on canonical `main`. H.264 startup blur/low-FPS reports remain an evidence-led diagnosis, not a presumed codec defect. diff --git a/docs/方案设计.md b/docs/方案设计.md index 7f5e31a..282b568 100644 --- a/docs/方案设计.md +++ b/docs/方案设计.md @@ -106,7 +106,7 @@ Web Host 在 `authenticate` 中显式发送 `viewerPresence: true` capability， ## 信令协议 -WebSocket 使用无 query/fragment 的同源 `/signal`。upgrade 只记录 site-access cookie 是否有效；连接后五秒内第一条消息必须是 `screener-v5` 的房间级 `authenticate`。Host 消息携带 room、role、内部 Host token 与 clientId，并由 socket state 复核 site access；Web Host 可选发 `viewerPasswordSettings: true` capability，source-only Native 使用同一 wire 版本但不发送该 capability，也不接收密码设置或 presence 新消息。Viewer 消息携带 room、role、clientId，以及 private room 的 Viewer grant 或当前尝试的房间密码。服务端设置 64 KiB `maxPayload`、关闭 `perMessageDeflate`，并以 ping/pong 清理失效连接。单进程固定限制 2048 条总连接和 256 条未鉴权连接，容量在 upgrade 前预留，超限直接返回 HTTP 503 并销毁 socket。 +WebSocket 使用无 query/fragment 的同源 `/signal`。upgrade 只记录 site-access cookie 是否有效；连接后五秒内第一条消息必须是 `screener-v6` 的房间级 `authenticate`。Host 消息携带 room、role、内部 Host token 与 clientId，并由 socket state 复核 site access；Web Host 可选发 `viewerPasswordSettings: true` capability，source-only Native 使用同一 wire 版本但不发送该 capability，也不接收密码设置或 presence 新消息。Viewer 消息携带 room、role、clientId，以及 private room 的 Viewer grant 或当前尝试的房间密码。服务端设置 64 KiB `maxPayload`、关闭 `perMessageDeflate`，并以 ping/pong 清理失效连接。单进程固定限制 2048 条总连接和 256 条未鉴权连接，容量在 upgrade 前预留，超限直接返回 HTTP 503 并销毁 socket。 主要消息： @@ -126,7 +126,7 @@ WebSocket 使用无 query/fragment 的同源 `/signal`。upgrade 只记录 site- Zod 在服务边界拒绝未知类型、超长字符串、非法 SDP/candidate 结构和越权目标。服务端根据已认证会话决定路由：观看者不能给另一名观看者发信令，分享者也不能向本房间以外的 peerId 发送。 -历史上的授权迁移曾把唯一 literal 从 `screener-v1` 原子改为 `screener-v2`，不做版本协商；后续原子迁移已把当前 wire 推进到 `screener-v5`，旧 v1/v2 parser、双写和 translator 都不存在。不兼容旧页面的首条消息在任何角色进入 room lookup 前得到同一个 fatal refresh 终态并停止自动重连。缺失、过期、跨房、已 rotate 或 revoked 的当前 Viewer grant 使用同一个通用“邀请无效或已失效”结果，不区分房间不存在与凭据错误。 +历史上的授权迁移曾把唯一 literal 从 `screener-v1` 原子改为 `screener-v2`，不做版本协商；后续不兼容迁移已把当前 source wire 原子推进到 `screener-v6`，旧 parser、双写和 translator 都不存在。包括上一版 `screener-v5` 在内的不兼容旧页面，其首条消息在任何角色进入 room lookup 前得到同一个 fatal refresh 终态并停止自动重连。缺失、过期、跨房、已 rotate 或 revoked 的当前 Viewer grant 使用同一个通用“邀请无效或已失效”结果，不区分房间不存在与凭据错误。 ## WebRTC 协商 @@ -186,7 +186,7 @@ Viewer 本地连接详情只从当前 route 的唯一 inbound screen video/audio 质量自适应采用 ADR-0007 的 built-in-first 路径隔离双表示，而不是全房间按最差链路降档。Direct/peer 保留每条 PeerConnection 的 stock WebRTC GCC。每条 SFU path 从 `HIGH` ceiling 开始；优先发布共享 `HIGH+LOW` 并让 LiveKit BWE 独立降层和恢复。若该 gate 通过，Screener 不实现媒体层 selector。应用证据只决定下述 parent/leaf 拓扑资格与诊断。`LOW` 可以常驻；关闭空闲层只是资源优化。active representation/layer 数量硬限制为二，不能按 viewer 增长。标准浏览器不保证跨 PeerConnection 共用一个物理编码器，inactive 层也不保证释放 GPU；实际层数、实例数、CPU/GPU、游戏 FPS/p1 low、区间 encode cost、上传与释放行为必须实测。若没有合格 hardware/power-efficient 路径或额外表示破坏游戏/上传预算，则 fail closed，保留健康路径并明确显示弱路径降级或失败。 -质量证据保持 edge-local，只能影响证据覆盖的 parent/child authority，不能形成全房综合分数。现有实现包含 Viewer 与 Host provisional-parent、媒体证明后 promotion、失败保留旧边、坏 relay corroboration 和 bounded quarantine 等可复用机制；整体模型统一 candidate eligibility、质量 owner 的独立关联证据、sibling migration、migration mutex、capacity reservation 与 fallback 交互。无论策略取值如何，prepare/commit/rollback 都保留 exact session/share/revision/connection/media-binding identity，ICE connected 不能替代 live track、正 RTP 和 decoded-frame progress。 +质量证据保持 edge-local，只能影响证据覆盖的 parent/child authority，不能形成全房综合分数。现有实现包含 Viewer 与 Host provisional-parent、媒体证明后 promotion、失败保留旧边和 child-scoped reparent；relay-parent suspect/repair/drain 等待独立 evidence owner，不能从固定数量的 child 事件推断 parent failure。整体模型统一 candidate eligibility、质量 owner 的独立关联证据、sibling migration、migration mutex、capacity reservation 与 fallback 交互。无论策略取值如何，prepare/commit/rollback 都保留 exact session/share/revision/connection/media-binding identity，ICE connected 不能替代 live track、正 RTP 和 decoded-frame progress。 状态决策必须关联同一时间区间和媒体代次的三段证据： @@ -329,7 +329,7 @@ Token-free SFU standby prewarm 保持可选且不创建 participant、publicatio - RoomStore 的随机临时房间、TTL、Host token/Viewer grant 摘要验证、private/public policy、grant 到期/跨房拒绝、rotate/revoke 强撤销、停止分享等待、Viewer 上限和超额拒绝； - SQLite 房间库从 `1` 自增、重启恢复、v1/v2 -> v3 单事务迁移、nullable 32-byte grant 与 48-byte password material CHECK、畸形值 fail-closed、rotate/revoke/password 写失败原子性和数据最小化；数据库路径无 site access 时启动失败； - STUN/TURN UDP URL 严格校验、生产 STUN 必需、server-assisted tuple 完整性/TTL/secret 独立性、issuer HMAC/expiry/opaque tag、当前 route/session/revision/connection generation guard，以及 ordinary peer、未授权房间和 Native 的 STUN-only isolation；LiveKit `sfu-config` 保持独立边界； -- 单一 `screener-v5` 信令 literal、v3 旧页面 mismatch 的一次性刷新终态，以及协议非法 JSON、未知消息、错误方向、跨房间目标和 payload 限制； +- 单一 `screener-v6` 信令 literal、v5 旧页面 mismatch 的一次性刷新终态，以及协议非法 JSON、未知消息、错误方向、跨房间目标和 payload 限制； - 质量协议的 host 授权、严格字段与数值范围、last-wins、当前/未来 relay、已配置 SFU 和普通 P2P wire 兼容；sender 的四字段 requested/applied 读回、拒绝与持续受限提示；screen audio SDP helper 覆盖 CRLF、existing/missing fmtp、unknown key preservation、幂等、payload number、rejected/non-audio，以及 empty/no-Opus/ambiguous/parser 异常原对象 fallback，`ViewerPeer` 覆盖 initial/restart/rebuild 时同一结果的 local/signaling 一致性，SFU publisher 精确断言 high-quality stereo/forceStereo/DTX 且不覆盖 RED； - HTTP/WS 集成：nginx per-source site-access `limit_req` 与统一错误不泄露 secret，应用不建立 IP limiter 状态；Viewer 可在 Origin/容量门下 upgrade；Host 首条鉴权同时要求 upgrade-captured site access 与 Host token，Viewer 要么持 exact-room grant，要么先有 site access 再按 room policy；site-access cookie 不绕过 private grant/password，grant/password 不能建房、取得 Host 或跨房。覆盖 fragment/sessionStorage/replaceState、无 site access 的 public/password code-only 统一拒绝、通过 site access 后的 public/password 进入、不含明文于 URL/浏览器持久存储/SQLite/log、Native capability isolation，以及 rotate/revoke 后 grace/route 清理、旧 Viewer/上游媒体边关闭和 reconnect 拒绝； - 客户端策略与 stats 解析：会话替换不重连，ordinary ICE、LiveKit 与 selected-edge TURN protocol 分离，远端 TURN transport 保持未知； @@ -361,6 +361,8 @@ Token-free SFU standby prewarm 保持可选且不创建 participant、publicatio 7. 加入 first-frame-driven Viewer connection stages，再整合 source-checkout LAN launcher。Site access、room-scoped Viewer grant 与 SQLite v3 的已部署合同保持不变。 8. Windows Native fixed-`HIGH` 保持可选且不改变 Web 默认；shared encode、移动 lifecycle、可分发 sender 和平台输出按 TODO roadmap 后续推进。 +生产仍运行 `screener-v5`；当前 source 的 Wave A capacity slice 使用 `screener-v6`。不兼容 wire/schema 迁移须原子切换并保留迁移前备份。 + 任何一步只实现当前消费者需要的接口。不得为未来账号、第三种媒体拓扑或原生 sender 预建抽象层。 ## 自动路由设计 diff --git a/native/sender/internal/app/server_test.go b/native/sender/internal/app/server_test.go index 21e2f06..2504389 100644 --- a/native/sender/internal/app/server_test.go +++ b/native/sender/internal/app/server_test.go @@ -632,7 +632,7 @@ func newAppRemoteServer(t *testing.T, terminal chan<- []byte) *httptest.Server { if _, _, err = connection.Read(request.Context()); err != nil { return } - authenticated := []byte(`{"type":"authenticated","protocol":"screener-v5","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`) + authenticated := []byte(`{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`) if err = connection.Write(request.Context(), websocket.MessageText, authenticated); err != nil { return } @@ -688,7 +688,7 @@ func newDelayedAppRemoteServer( <-releaseAuthentication writeContext, cancelWrite := context.WithTimeout(context.Background(), time.Second) defer cancelWrite() - _ = connection.Write(writeContext, websocket.MessageText, []byte(`{"type":"authenticated","protocol":"screener-v5","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`)) + _ = connection.Write(writeContext, websocket.MessageText, []byte(`{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`)) }) return server } diff --git a/native/sender/internal/remote/wire.go b/native/sender/internal/remote/wire.go index febc425..5492911 100644 --- a/native/sender/internal/remote/wire.go +++ b/native/sender/internal/remote/wire.go @@ -16,7 +16,7 @@ import ( ) const ( - signalingProtocol = "screener-v5" + signalingProtocol = "screener-v6" maxProtocolViewers = 16 maxRouteRevision = int64(1<<53 - 1) maxSignalSDPBytes = 48 << 10 diff --git a/native/sender/internal/remote/wire_test.go b/native/sender/internal/remote/wire_test.go index 602201b..87efc06 100644 --- a/native/sender/internal/remote/wire_test.go +++ b/native/sender/internal/remote/wire_test.go @@ -6,9 +6,9 @@ import ( "testing" ) -const validHostAuthenticated = `{"type":"authenticated","protocol":"screener-v5","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}` +const validHostAuthenticated = `{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}` -const validPeerAssistedHostAuthenticated = `{"type":"authenticated","protocol":"screener-v5","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":["viewer-1","viewer-2","viewer-3"],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678","mediaMode":"peer-assisted","mediaAssignment":{"parentPeerId":null,"childPeerIds":["viewer-1","viewer-2","viewer-3"]},"routeRevision":7,"routeAssignment":{"upstream":{"kind":"none"},"childPeerIds":["viewer-1","viewer-2","viewer-3"],"sfuPublicationGeneration":null},"qualitySettings":{"resolution":"1080p","maxFramerate":60,"maxBitrate":8000000,"degradationPreference":"maintain-resolution","videoCodec":"automatic"}}` +const validPeerAssistedHostAuthenticated = `{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":["viewer-1","viewer-2","viewer-3"],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678","mediaMode":"peer-assisted","mediaAssignment":{"parentPeerId":null,"childPeerIds":["viewer-1","viewer-2","viewer-3"]},"routeRevision":7,"routeAssignment":{"upstream":{"kind":"none"},"childPeerIds":["viewer-1","viewer-2","viewer-3"],"sfuPublicationGeneration":null},"qualitySettings":{"resolution":"1080p","maxFramerate":60,"maxBitrate":8000000,"degradationPreference":"maintain-resolution","videoCodec":"automatic"}}` func TestMarshalHostSignalShapes(t *testing.T) { mid := "0" @@ -139,7 +139,7 @@ func TestDecodePeerAssistedRouteMessages(t *testing.T) { func TestDecodeOrdinaryHostAuthenticationRejectsInvalidVariants(t *testing.T) { tests := map[string]func(map[string]any){ "missing-protocol": func(message map[string]any) { delete(message, "protocol") }, - "wrong-protocol": func(message map[string]any) { message["protocol"] = "legacy-protocol" }, + "wrong-protocol": func(message map[string]any) { message["protocol"] = "screener-v5" }, "missing-max": func(message map[string]any) { delete(message, "maxViewers") }, "zero-max": func(message map[string]any) { message["maxViewers"] = 0 }, "too-large-max": func(message map[string]any) { message["maxViewers"] = 17 }, diff --git a/scripts/peer-assisted-benchmark.ts b/scripts/peer-assisted-benchmark.ts index ef28fff..3eb6054 100644 --- a/scripts/peer-assisted-benchmark.ts +++ b/scripts/peer-assisted-benchmark.ts @@ -7,12 +7,13 @@ import { fileURLToPath, pathToFileURL } from "node:url"; import WebSocket from "ws"; import { - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - CURRENT_HOST_MEDIA_EDGE_LIMIT, - DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES, MAX_VIEWERS_PER_ROOM_LIMIT, type ParticipantRouteAssignment, } from "../src/shared/protocol"; +import { + DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, + MAX_ENDPOINT_MEDIA_COPY_CAPACITY, +} from "../src/shared/media-copy-accounting"; import { createScreenerServer, type ScreenerServer, @@ -387,14 +388,14 @@ export function parseBenchmarkCanaryMode(value: string | undefined): BenchmarkCa export function parseExpectedEndpointCap(value: string | undefined): number { const parsed = value?.trim() ? Number(value) - : DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES; + : DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY; if ( !Number.isSafeInteger(parsed) || parsed < 1 || - parsed > CURRENT_HOST_MEDIA_EDGE_LIMIT + parsed > MAX_ENDPOINT_MEDIA_COPY_CAPACITY ) { throw new Error( - `BENCHMARK_EXPECTED_ENDPOINT_CAP must be an integer from 1 to ${CURRENT_HOST_MEDIA_EDGE_LIMIT}`, + `BENCHMARK_EXPECTED_ENDPOINT_CAP must be an integer from 1 to ${MAX_ENDPOINT_MEDIA_COPY_CAPACITY}`, ); } return parsed; @@ -863,12 +864,8 @@ export function buildRunChecks( summary: NonNullable, viewerCount: number, profileId: ProfileId, - expectedEndpointCap = DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES, + expectedEndpointCap = DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, ): RunCheck[] { - const expectedViewerCap = Math.min( - expectedEndpointCap, - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - ); const sfuConsistencyCheck: RunCheck = summary.sfuPublicationObserved ? { name: "sfu-route-consistency", @@ -898,9 +895,9 @@ export function buildRunChecks( }, { name: "relay-active-media-edges", - passed: summary.maxRelayActiveMediaEdges <= expectedViewerCap, + passed: summary.maxRelayActiveMediaEdges <= expectedEndpointCap, actual: summary.maxRelayActiveMediaEdges, - expected: `<= ${expectedViewerCap}`, + expected: `<= ${expectedEndpointCap}`, }, { name: "all-viewers-decoded", @@ -928,7 +925,7 @@ export function buildRunChecks( export function buildRecoveryCheck( recovery: RecoveryResult, - expectedEndpointCap = DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES, + expectedEndpointCap = DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, ): RunCheck { const passed = recovery.triggered && @@ -960,9 +957,7 @@ export function buildBenchmarkInitScript(options: { return `(() => { if (globalThis.__SCREENER_BENCHMARK__) return; const options = ${serialized}; - const expectedRoleCap = options.role === "host" - ? options.expectedEndpointCap - : Math.min(options.expectedEndpointCap, ${CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT}); + const expectedRoleCap = options.expectedEndpointCap; if (options.clearHostRoom) { try { localStorage.removeItem("screener:host-room:v1"); } catch {} } @@ -2664,7 +2659,7 @@ export async function main(): Promise { ALLOWED_ORIGINS: baseUrl, ROOM_DATABASE_PATH: "", PEER_ASSISTED_MEDIA: "true", - MAX_PEER_RELAY_DOWNSTREAM_EDGES: String(config.expectedEndpointCap), + ENDPOINT_MEDIA_COPY_CAPACITY: String(config.expectedEndpointCap), MAX_VIEWERS_PER_ROOM: String(Math.max( ...config.viewerCounts, config.canaryMode === "viewer-mbb" ? 3 : 1, diff --git a/src/client/media/host-provisional-child.ts b/src/client/media/host-provisional-child.ts index 6e6496e..c799cc1 100644 --- a/src/client/media/host-provisional-child.ts +++ b/src/client/media/host-provisional-child.ts @@ -3,6 +3,7 @@ import type { ParticipantRouteAssignment, SignalPayload, } from "../../shared/protocol"; +import { countEndpointMediaCopies } from "../../shared/media-copy-accounting"; import { HostPeer } from "../webrtc/host-peer"; import type { PeerSnapshot } from "../types"; import type { QualityProfile } from "./quality"; @@ -220,8 +221,13 @@ export function plannedHostProvisionalChild( input.activeChildPeerIds.some( (candidate) => !childPeerIds.includes(candidate), ) || - hostPhysicalMediaEdges(input.assignment, input.selectedPeerId) > - input.maxMediaEdges + countEndpointMediaCopies({ + childPeerIds: input.assignment.childPeerIds, + publicationGeneration: input.assignment.sfuPublicationGeneration, + selectedChildPeerIds: input.selectedPeerId + ? [input.selectedPeerId] + : [], + }) > input.maxMediaEdges ) { return null; } @@ -249,16 +255,3 @@ export function resolveHostPreparedChildActivation( } : { kind: "promote", peerId }; } - -function hostPhysicalMediaEdges( - assignment: ParticipantRouteAssignment, - selectedPeerId: string | null, -): number { - const selectedOverlayEdges = - selectedPeerId && !assignment.childPeerIds.includes(selectedPeerId) ? 1 : 0; - return ( - assignment.childPeerIds.length + - (assignment.sfuPublicationGeneration ? 1 : 0) + - selectedOverlayEdges - ); -} diff --git a/src/client/media/relay-capability.ts b/src/client/media/relay-capability.ts index d826e4a..3901259 100644 --- a/src/client/media/relay-capability.ts +++ b/src/client/media/relay-capability.ts @@ -1,7 +1,5 @@ -import { - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - type ClientMessage, -} from "../../shared/protocol"; +import { MAX_ENDPOINT_MEDIA_COPY_CAPACITY } from "../../shared/media-copy-accounting"; +import type { ClientMessage } from "../../shared/protocol"; type RelayCapacityMessage = Extract< ClientMessage, @@ -16,6 +14,6 @@ export function relayCapacityMessageForBrowser( } return { type: "relay-capacity", - downstreamEdges: CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, + downstreamEdges: MAX_ENDPOINT_MEDIA_COPY_CAPACITY, }; } diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 93b18dc..00b6a73 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -114,7 +114,7 @@ import type { } from "../types"; import { HostPeer } from "../webrtc/host-peer"; import { - MAX_HOST_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, reconcileBoundedMediaChildren, } from "../webrtc/media-assignment"; import { @@ -970,7 +970,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { assignment, activeChildPeerIds: activeHostChildPeerIdsRef.current, selectedPeerId: selectedHostChildRef.current?.peerId ?? null, - maxMediaEdges: MAX_HOST_MEDIA_CHILDREN, + maxMediaEdges: MAX_ENDPOINT_MEDIA_CHILDREN, iceConfig, stream, profile: qualitySettingsRef.current, @@ -989,7 +989,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { assignment, activeChildPeerIds, selectedPeerId: selectedHostChildRef.current?.peerId ?? null, - maxMediaEdges: MAX_HOST_MEDIA_CHILDREN, + maxMediaEdges: MAX_ENDPOINT_MEDIA_CHILDREN, }) ?? { kind: "ordinary" as const }; if (activation.kind === "promote") { hostProvisionalChildRef.current = null; @@ -1180,7 +1180,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { reconcileBoundedMediaChildren( peersRef.current.keys(), effectiveChildPeerIds, - MAX_HOST_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, removePeer, (peerId) => { void startPeer(peerId, generation).catch((error: unknown) => { diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index fb4627e..5d104db 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -74,7 +74,7 @@ import type { } from "../types"; import { limitMediaAssignment, - MAX_VIEWER_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, retainSelectedMediaParent, viewerRestartMessage, viewerSignalMessage, @@ -648,7 +648,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { parentPeerId: currentAssignment.parentPeerId, childPeerIds: [...childPeerIds], }, - MAX_VIEWER_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, ); reconcileRelayChildren(previousChildPeerIds, revision); }, @@ -685,7 +685,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { const previousChildPeerIds = currentAssignment.childPeerIds; currentAssignment = limitMediaAssignment( { parentPeerId: null, childPeerIds: assignment.childPeerIds }, - MAX_VIEWER_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, ); reconcileRelayChildren(previousChildPeerIds); const relay = ensureViewerRelay(); @@ -768,7 +768,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { function applyMediaAssignment(assignment: MediaAssignment, preserveUpstream = false): void { const nextAssignment = limitMediaAssignment( assignment, - MAX_VIEWER_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, ); const previousParentId = currentAssignment.parentPeerId; const previousChildPeerIds = currentAssignment.childPeerIds; diff --git a/src/client/webrtc/media-assignment.ts b/src/client/webrtc/media-assignment.ts index 96d1553..d92dde8 100644 --- a/src/client/webrtc/media-assignment.ts +++ b/src/client/webrtc/media-assignment.ts @@ -3,14 +3,9 @@ import type { MediaAssignment, SignalPayload, } from "../../shared/protocol"; -import { - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - CURRENT_HOST_MEDIA_EDGE_LIMIT, -} from "../../shared/protocol"; +import { MAX_ENDPOINT_MEDIA_COPY_CAPACITY } from "../../shared/media-copy-accounting"; -export const MAX_HOST_MEDIA_CHILDREN = CURRENT_HOST_MEDIA_EDGE_LIMIT; -export const MAX_VIEWER_MEDIA_CHILDREN = - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT; +export const MAX_ENDPOINT_MEDIA_CHILDREN = MAX_ENDPOINT_MEDIA_COPY_CAPACITY; export function limitMediaAssignment( assignment: MediaAssignment, diff --git a/src/client/webrtc/viewer-relay.ts b/src/client/webrtc/viewer-relay.ts index 202c643..0fd4cc5 100644 --- a/src/client/webrtc/viewer-relay.ts +++ b/src/client/webrtc/viewer-relay.ts @@ -9,7 +9,7 @@ import { } from "../media/quality"; import type { PeerSnapshot } from "../types"; import { HostPeer } from "./host-peer"; -import { MAX_VIEWER_MEDIA_CHILDREN } from "./media-assignment"; +import { MAX_ENDPOINT_MEDIA_CHILDREN } from "./media-assignment"; interface ViewerRelayEvents { sendSignal: (peerId: string, payload: SignalPayload) => boolean; @@ -99,7 +99,7 @@ export class ViewerRelay { Date.parse(message.expiresAt) <= now || connectionId !== message.oldConnectionId || (!retainedChildPeerIds.includes(message.viewerPeerId) && - retainedChildPeerIds.length >= MAX_VIEWER_MEDIA_CHILDREN) + retainedChildPeerIds.length >= MAX_ENDPOINT_MEDIA_CHILDREN) ) { return false; } @@ -158,7 +158,7 @@ export class ViewerRelay { !stream || planned.length !== plannedChildPeerIds.length || planned.length !== this.childPeerIds.length + 1 || - planned.length > MAX_VIEWER_MEDIA_CHILDREN || + planned.length > MAX_ENDPOINT_MEDIA_CHILDREN || this.childPeerIds.some((peerId) => !planned.includes(peerId)) || !childPeerId || this.peers.has(childPeerId) @@ -247,11 +247,11 @@ export class ViewerRelay { } const nextChildPeerIds = [...new Set(childPeerIds)].slice( 0, - MAX_VIEWER_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, ); const selectedPeerId = this.selectedChildConnection?.peerId ?? null; if (selectedPeerId && !nextChildPeerIds.includes(selectedPeerId)) { - if (nextChildPeerIds.length >= MAX_VIEWER_MEDIA_CHILDREN) { + if (nextChildPeerIds.length >= MAX_ENDPOINT_MEDIA_CHILDREN) { nextChildPeerIds.pop(); } nextChildPeerIds.push(selectedPeerId); @@ -631,7 +631,7 @@ export class ViewerRelay { if (peer) { this.retiredConnections.delete(childPeerId); this.retiredConnections.set(childPeerId, peer.connectionId); - while (this.retiredConnections.size > MAX_VIEWER_MEDIA_CHILDREN) { + while (this.retiredConnections.size > MAX_ENDPOINT_MEDIA_CHILDREN) { const oldestPeerId = this.retiredConnections.keys().next().value; if (typeof oldestPeerId === "string") { this.retiredConnections.delete(oldestPeerId); diff --git a/src/server/app.ts b/src/server/app.ts index f06639a..36c93f2 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -112,7 +112,7 @@ export async function createScreenerServer( server: httpServer, roomStore, peerAssistedMedia: config.peerAssistedMedia, - maxPeerRelayDownstreamEdges: config.maxPeerRelayDownstreamEdges, + endpointMediaCopyCapacity: config.endpointMediaCopyCapacity, ...(sfuFallback ? { sfuFallback } : {}), ...(config.selectedEdgeTurn ? { selectedEdgeTurn: config.selectedEdgeTurn } : {}), ice: iceOptions, diff --git a/src/server/config.ts b/src/server/config.ts index 635f0e5..4c7c056 100644 --- a/src/server/config.ts +++ b/src/server/config.ts @@ -1,12 +1,14 @@ import { - CURRENT_HOST_MEDIA_EDGE_LIMIT, CURRENT_SFU_ROOT_LIMIT, - DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES, MAX_ICE_SERVER_URLS, MAX_VIEWERS_PER_ROOM_LIMIT, stunUrlSchema, turnUrlSchema, } from "../shared/protocol.js"; +import { + DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, + MAX_ENDPOINT_MEDIA_COPY_CAPACITY, +} from "../shared/media-copy-accounting.js"; export type RuntimeEnvironment = "development" | "test" | "production"; @@ -29,6 +31,7 @@ const REMOVED_ENVIRONMENT_VARIABLES = [ "PEER_ICE_TURN_CREDENTIAL_TTL_SECONDS", "PEER_ASSISTED_ROOM_IDS", "HOST_ADMISSION_PASSWORD", + "MAX_PEER_RELAY_DOWNSTREAM_EDGES", ] as const; export interface LiveKitFallbackConfig { @@ -56,7 +59,7 @@ export interface ServerConfig { maxRooms: number; maxViewersPerRoom: number; peerAssistedMedia: boolean; - maxPeerRelayDownstreamEdges: number; + endpointMediaCopyCapacity: number; livekitFallback?: LiveKitFallbackConfig; selectedEdgeTurn?: SelectedEdgeTurnConfig; stunUrls: readonly string[]; @@ -272,7 +275,9 @@ export function loadConfig( for (const name of REMOVED_ENVIRONMENT_VARIABLES) { if (Object.prototype.hasOwnProperty.call(environment, name)) { throw new Error( - name === "HOST_ADMISSION_PASSWORD" + name === "MAX_PEER_RELAY_DOWNSTREAM_EDGES" + ? `${name} is no longer supported; use ENDPOINT_MEDIA_COPY_CAPACITY` + : name === "HOST_ADMISSION_PASSWORD" ? `${name} is no longer supported; use SITE_ACCESS_PASSWORD` : name === "PEER_ASSISTED_ROOM_IDS" ? `${name} is no longer supported; peer-assisted media applies to every room when enabled` @@ -336,12 +341,12 @@ export function loadConfig( false, "PEER_ASSISTED_MEDIA", ); - const maxPeerRelayDownstreamEdges = parseBoundedInteger( - environment.MAX_PEER_RELAY_DOWNSTREAM_EDGES, - DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES, - "MAX_PEER_RELAY_DOWNSTREAM_EDGES", + const endpointMediaCopyCapacity = parseBoundedInteger( + environment.ENDPOINT_MEDIA_COPY_CAPACITY, + DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, + "ENDPOINT_MEDIA_COPY_CAPACITY", 1, - CURRENT_HOST_MEDIA_EDGE_LIMIT, + MAX_ENDPOINT_MEDIA_COPY_CAPACITY, ); const livekitFallback = parseLiveKitFallback(environment, nodeEnv); const selectedEdgeTurn = parseSelectedEdgeTurn(environment); @@ -406,7 +411,7 @@ export function loadConfig( maxRooms: parsePositiveInteger(environment.MAX_ROOMS, 1_000, "MAX_ROOMS"), maxViewersPerRoom, peerAssistedMedia, - maxPeerRelayDownstreamEdges, + endpointMediaCopyCapacity, livekitFallback, selectedEdgeTurn, stunUrls, diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index f40705a..717d6ba 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -1,8 +1,6 @@ import { randomBytes } from "node:crypto"; import { - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - CURRENT_HOST_MEDIA_EDGE_LIMIT, CURRENT_SFU_ROOT_LIMIT, MAX_MEDIA_ROUTE_REVISION, type ClientMessage, @@ -14,6 +12,11 @@ import { type Role, type ServerMessage, } from "../shared/protocol.js"; +import { + assertEndpointMediaCopyCapacity, + countEndpointMediaCopies, + endpointMediaCopyCountFits, +} from "../shared/media-copy-accounting.js"; import type { SfuTokenIssuer } from "./livekit-token.js"; import type { SelectedEdgeTurnConfig } from "./config.js"; import { @@ -160,14 +163,6 @@ interface ViewerQualityEvidenceState { } | null; } -interface RelayParentQualityState { - parentSessionId: string; - shareGeneration: string; - lastChildPeerId: string; - lastConfirmedAtMs: number; - quarantinedUntilMs: number | null; -} - interface ForwardedViewerQualityEvidence { roomId: string; viewerSessionId: string; @@ -196,7 +191,7 @@ export interface SfuFallbackOptions { export interface HybridMediaRouterOptions { roomStore: RoomStore; - maxPeerRelayDownstreamEdges: RelayDownstreamEdges; + endpointMediaCopyCapacity: number; sfuFallback?: SfuFallbackOptions; selectedEdgeTurn?: SelectedEdgeTurnConfig; sendToSession: (sessionId: string, message: ServerMessage) => void; @@ -233,7 +228,7 @@ export interface AuthenticatedRouteParticipant { export class HybridMediaRouter { private readonly peerRelayTopology: PeerRelayTopology; - private readonly maxHostMediaEdges: number; + private readonly endpointMediaCopyCapacity: number; private readonly mediaRouteControllers = new Map< string, MediaRouteController @@ -251,7 +246,7 @@ export class HybridMediaRouter { Map >(); private readonly consumedSfuRefreshesByRoom = new Map>(); - private readonly selectedPeerEdgeLeases = new Map< + private readonly selectedPeerEdgeLeasesByRoom = new Map< string, SelectedPeerEdgeLease >(); @@ -284,16 +279,12 @@ export class HybridMediaRouter { string, DeferredHealthySfuReselection >(); - private readonly relayParentQualityStates = new Map(); - constructor(private readonly options: HybridMediaRouterOptions) { + assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity); this.peerRelayTopology = new PeerRelayTopology( - options.maxPeerRelayDownstreamEdges, - ); - this.maxHostMediaEdges = Math.min( - options.maxPeerRelayDownstreamEdges, - CURRENT_HOST_MEDIA_EDGE_LIMIT, + options.endpointMediaCopyCapacity, ); + this.endpointMediaCopyCapacity = options.endpointMediaCopyCapacity; const fallback = options.sfuFallback; if (options.selectedEdgeTurn && !fallback) { throw new Error("Selected-edge TURN requires SFU fallback"); @@ -320,10 +311,10 @@ export class HybridMediaRouter { clearTimeout(pending.timer); } this.pendingRoutePreparations.clear(); - for (const lease of this.selectedPeerEdgeLeases.values()) { + for (const lease of this.selectedPeerEdgeLeasesByRoom.values()) { clearTimeout(lease.timer); } - this.selectedPeerEdgeLeases.clear(); + this.selectedPeerEdgeLeasesByRoom.clear(); for (const attempt of this.selectedSfuIngressAttempts.values()) { clearTimeout(attempt.timer); } @@ -344,7 +335,6 @@ export class HybridMediaRouter { this.viewerQualityEvidenceStates.clear(); this.heldQualityParentExclusionsByViewer.clear(); this.roomQualityMigrationCooldownUntilMs.clear(); - this.relayParentQualityStates.clear(); } connectParticipant(input: AuthenticatedRouteParticipant): HybridAuthenticationState { @@ -355,7 +345,6 @@ export class HybridMediaRouter { input.peerId, ); this.clearRoomViewerQualityEvidenceStates(input.roomId); - this.refreshRelayParentQualityStates(input.roomId); this.clearHeldQualityParentExclusion(input.roomId, input.peerId); this.clearSfuRefreshesForPeer(input.roomId, input.peerId); if (input.role === "viewer") { @@ -457,16 +446,15 @@ export class HybridMediaRouter { signalKind: "candidate" | "description"; descriptionType?: "offer" | "answer"; }): boolean | "probe" | undefined { - const sourceKey = viewerConnectionKey(input.roomId, input.sourcePeerId); - const targetKey = viewerConnectionKey(input.roomId, input.targetPeerId); + const roomLease = this.selectedPeerEdgeLeasesByRoom.get(input.roomId); const lease = - this.selectedPeerEdgeLeases.get(sourceKey)?.parentPeerId === - input.targetPeerId - ? this.selectedPeerEdgeLeases.get(sourceKey) - : this.selectedPeerEdgeLeases.get(targetKey)?.parentPeerId === - input.sourcePeerId - ? this.selectedPeerEdgeLeases.get(targetKey) - : undefined; + roomLease && + ((roomLease.parentPeerId === input.sourcePeerId && + roomLease.viewerPeerId === input.targetPeerId) || + (roomLease.viewerPeerId === input.sourcePeerId && + roomLease.parentPeerId === input.targetPeerId)) + ? roomLease + : undefined; if (!lease) { const pending = this.pendingRoutePreparations.get(input.roomId); const qualityProbe = pending?.peerQualityReselection; @@ -540,10 +528,11 @@ export class HybridMediaRouter { viewerPeerId: string, connectionId: string, ): void { - const lease = this.selectedPeerEdgeLeases.get( - viewerConnectionKey(roomId, viewerPeerId), - ); - if (lease?.newConnectionId === connectionId) { + const lease = this.selectedPeerEdgeLeasesByRoom.get(roomId); + if ( + lease?.viewerPeerId === viewerPeerId && + lease.newConnectionId === connectionId + ) { lease.answered = true; clearTimeout(lease.timer); this.drainViewerRouteIntents(roomId); @@ -726,7 +715,6 @@ export class HybridMediaRouter { correlatedAtMs: now, framesPerSecond: pending.evidence.metrics.framesPerSecond, }; - this.refreshRelayParentQualityStates(roomId, now); const existingIntent = this.viewerRouteIntentsByRoom .get(roomId) ?.get(viewerPeerId); @@ -766,25 +754,11 @@ export class HybridMediaRouter { return; } - const relayParentBadEvent = this.handleRelayParentBadEvent( - roomId, - edge.parentPeerId, - parentSessionId, - viewerPeerId, - badWindowKind!, - now, - ); - if (relayParentBadEvent === "trigger-moved") { - this.viewerQualityEvidenceStates.delete(connectionKey); - return; - } const cooldownUntil = this.roomQualityMigrationCooldownUntilMs.get(roomId); if (cooldownUntil !== undefined && now < cooldownUntil) { - if (relayParentBadEvent === null) { - state.badWindowCount = 0; - state.badWindowKind = null; - return; - } + state.badWindowCount = 0; + state.badWindowKind = null; + return; } else if (cooldownUntil !== undefined) { this.roomQualityMigrationCooldownUntilMs.delete(roomId); } @@ -868,7 +842,6 @@ export class HybridMediaRouter { if (participant.role !== "viewer") { return; } - this.refreshRelayParentQualityStates(participant.roomId); let advertised = this.relayCapacitySessionsByRoom.get(participant.roomId); if (!advertised) { advertised = new Map(); @@ -988,20 +961,24 @@ export class HybridMediaRouter { const host = this.connectedPeer(roomId, hostPeerId); const parent = this.connectedPeer(roomId, parentPeerId); const activeParent = active.assignments.get(parentPeerId); - const capacity = Math.min( - this.peerRelayTopology.getDownstreamCapacity(roomId, parentPeerId), + const parentSupportsOutboundRelay = parentPeerId === hostPeerId - ? this.maxHostMediaEdges - : CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, + ? activeParent?.upstream.kind === "none" + : activeParent?.upstream.kind === "peer"; + const capacity = this.peerRelayTopology.getDownstreamCapacity( + roomId, + parentPeerId, ); if ( !host || !parent || !activeParent || - activeParent.childPeerIds.length + - (activeParent.sfuPublicationGeneration ? 1 : 0) + - 1 > - capacity || + !parentSupportsOutboundRelay || + !endpointMediaCopyCountFits( + this.activeParentMediaEdges(roomId, parentPeerId, activeParent) + 1, + capacity, + "transition", + ) || (parentPeerId !== hostPeerId && this.relayCapacitySessionsByRoom.get(roomId)?.get(parentPeerId) !== parent.sessionId) @@ -1013,7 +990,7 @@ export class HybridMediaRouter { hostPeerId, null, [], - this.maxHostMediaEdges, + this.endpointMediaCopyCapacity, ); const rootAssignment = assignments.get(participant.peerId)!; const parentAssignment = assignments.get(parentPeerId)!; @@ -1643,11 +1620,8 @@ export class HybridMediaRouter { if (pending?.peerQualityReselection) { return "quality"; } - return [...this.selectedPeerEdgeLeases.values()].some( - (lease) => - lease.roomId === roomId && - (!lease.answered || lease.viewerPeerId === viewerPeerId), - ) + const lease = this.selectedPeerEdgeLeasesByRoom.get(roomId); + return lease && (!lease.answered || lease.viewerPeerId === viewerPeerId) ? "selected" : null; } @@ -1668,21 +1642,16 @@ export class HybridMediaRouter { const controller = this.mediaRouteControllers.get(roomId); const active = controller?.getActiveRoute(); const intent = this.viewerRouteIntentsByRoom.get(roomId)?.get(viewerPeerId); - const key = viewerConnectionKey(roomId, viewerPeerId); const roomAttempt = this.roomPeerMigrationAttempt(roomId); if (roomAttempt === "healthy" || roomAttempt === "quality") { // The room probe already owns recovery, so suppress concurrent failback. return true; } - if (this.selectedPeerEdgeLeases.has(key)) { + const roomLease = this.selectedPeerEdgeLeasesByRoom.get(roomId); + if (roomLease?.viewerPeerId === viewerPeerId) { return true; } - if ( - roomAttempt === "selected" || - [...this.selectedPeerEdgeLeases.values()].some( - (lease) => lease.roomId === roomId, - ) - ) { + if (roomAttempt === "selected" || roomLease) { return false; } if ( @@ -1703,27 +1672,31 @@ export class HybridMediaRouter { const parent = this.connectedPeer(roomId, intent.failedParentPeerId); const shareGeneration = this.options.getShareGeneration(roomId); const parentAssignment = active.assignments.get(intent.failedParentPeerId); + const hostPeerId = this.peerRelayTopology.getHostPeerId(roomId); + const parentSupportsOutboundRelay = + intent.failedParentPeerId === hostPeerId + ? parentAssignment?.upstream.kind === "none" + : parentAssignment?.upstream.kind === "peer"; const parentChildLimit = this.peerRelayTopology.getAdvertisedDownstreamCapacity( roomId, intent.failedParentPeerId, ); - const selectedOverlayEdges = parentAssignment?.childPeerIds.includes( - viewerPeerId, - ) - ? 0 - : 1; const parentActiveEdges = parentAssignment - ? parentAssignment.childPeerIds.length + - (parentAssignment.sfuPublicationGeneration ? 1 : 0) + - selectedOverlayEdges + ? countEndpointMediaCopies({ + childPeerIds: parentAssignment.childPeerIds, + publicationGeneration: + parentAssignment.sfuPublicationGeneration, + selectedChildPeerIds: [viewerPeerId], + }) : Number.POSITIVE_INFINITY; if ( viewer?.sessionId !== intent.sessionId || parent?.sessionId !== intent.failedParentSessionId || !parentAssignment || + !parentSupportsOutboundRelay || parentChildLimit === 0 || - parentActiveEdges > parentChildLimit || + !endpointMediaCopyCountFits(parentActiveEdges, parentChildLimit) || !shareGeneration ) { return false; @@ -1773,7 +1746,7 @@ export class HybridMediaRouter { } const timer = setTimeout(() => { - const current = this.selectedPeerEdgeLeases.get(key); + const current = this.selectedPeerEdgeLeasesByRoom.get(roomId); if ( !current || current.newConnectionId !== newConnectionId || @@ -1781,7 +1754,7 @@ export class HybridMediaRouter { ) { return; } - const settled = this.settleSelectedPeerEdgeLease(key); + const settled = this.settleSelectedPeerEdgeLease(roomId); if (!settled) { return; } @@ -1817,7 +1790,7 @@ export class HybridMediaRouter { answered: false, timer, }; - this.selectedPeerEdgeLeases.set(key, lease); + this.selectedPeerEdgeLeasesByRoom.set(roomId, lease); // 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); @@ -2037,23 +2010,7 @@ export class HybridMediaRouter { if (message.connectionId === null) { return false; } - let key = viewerConnectionKey(participant.roomId, participant.peerId); - let lease = this.selectedPeerEdgeLeases.get(key); - if ( - !lease || - lease.viewerPeerId !== participant.peerId || - lease.viewerSessionId !== participant.sessionId - ) { - const parentEntry = [...this.selectedPeerEdgeLeases].find( - ([, candidate]) => - candidate.roomId === participant.roomId && - candidate.parentPeerId === participant.peerId && - candidate.parentSessionId === participant.sessionId, - ); - if (parentEntry) { - [key, lease] = parentEntry; - } - } + const lease = this.selectedPeerEdgeLeasesByRoom.get(participant.roomId); const active = this.mediaRouteControllers .get(participant.roomId) ?.getActiveRoute(); @@ -2079,7 +2036,7 @@ export class HybridMediaRouter { participant.roomId, lease.viewerPeerId, ); - this.settleSelectedPeerEdgeLease(key); + this.settleSelectedPeerEdgeLease(participant.roomId); this.sendError( participant.sessionId, "PEER_NOT_FOUND", @@ -2108,7 +2065,7 @@ export class HybridMediaRouter { hostPeerId, null, [], - this.maxHostMediaEdges, + this.endpointMediaCopyCapacity, ); const revision = controller.reconcileBaseline({ assignments, @@ -2143,7 +2100,7 @@ export class HybridMediaRouter { hostPeerId, null, [], - this.maxHostMediaEdges, + this.endpointMediaCopyCapacity, ); const revision = controller.reconcileBaseline({ assignments, @@ -2192,7 +2149,6 @@ export class HybridMediaRouter { } private drainViewerRouteIntents(roomId: string): void { - this.refreshRelayParentQualityStates(roomId); this.enqueueUnassignedViewerIntents(roomId); const pending = this.pendingRoutePreparations.get(roomId); if (pending) { @@ -2206,15 +2162,13 @@ export class HybridMediaRouter { } const intents = this.viewerRouteIntentsByRoom.get(roomId); if (!intents) { - this.evacuateQuarantinedRelayParents(roomId); return; } for (const [viewerPeerId, intent] of [...intents]) { if ( - this.selectedPeerEdgeLeases.has( - viewerConnectionKey(roomId, viewerPeerId), - ) + this.selectedPeerEdgeLeasesByRoom.get(roomId)?.viewerPeerId === + viewerPeerId ) { continue; } @@ -2412,9 +2366,6 @@ export class HybridMediaRouter { if (intents.size === 0) { this.viewerRouteIntentsByRoom.delete(roomId); } - if (this.roomPeerMigrationAttempt(roomId) !== "selected") { - this.evacuateQuarantinedRelayParents(roomId); - } } private startPeerQualityReselection( @@ -2454,13 +2405,15 @@ export class HybridMediaRouter { : this.options.roomStore.getConnectedViewer(roomId, peerId); const supportedParent = isHost ? assignment.upstream.kind === "none" - : assignment.upstream.kind === "peer" || - assignment.upstream.kind === "sfu"; + : assignment.upstream.kind === "peer"; if ( !participant || !supportedParent || - this.activeParentMediaEdges(roomId, peerId, assignment) + 1 > - this.peerRelayTopology.getDownstreamCapacity(roomId, peerId) + !endpointMediaCopyCountFits( + this.activeParentMediaEdges(roomId, peerId, assignment) + 1, + this.peerRelayTopology.getDownstreamCapacity(roomId, peerId), + "transition", + ) ) { candidateExclusions.add(peerId); } @@ -2755,7 +2708,15 @@ export class HybridMediaRouter { return "unavailable"; } - if (host.childPeerIds.length > this.maxHostMediaEdges - 1) { + if ( + !endpointMediaCopyCountFits( + countEndpointMediaCopies({ + childPeerIds: host.childPeerIds, + publicationGeneration: "candidate", + }), + this.endpointMediaCopyCapacity, + ) + ) { const budgetRootPeerId = requestedAssignment.upstream.kind === "peer" ? findHostBranchRoot(active.assignments, failedPeerId, hostPeerId) @@ -2790,8 +2751,13 @@ export class HybridMediaRouter { if ( rootPeerIds.length > fallback.maxRoots || !host || - host.childPeerIds.length + 1 > - this.maxHostMediaEdges + !endpointMediaCopyCountFits( + countEndpointMediaCopies({ + childPeerIds: host.childPeerIds, + publicationGeneration: "candidate", + }), + this.endpointMediaCopyCapacity, + ) ) { this.sendError( sourceSessionId, @@ -2801,6 +2767,23 @@ export class HybridMediaRouter { return "unavailable"; } + const activeHost = active.assignments.get(hostPeerId); + if ( + !activeHost || + !endpointMediaCopyCountFits( + this.activeParentMediaEdges(roomId, hostPeerId, activeHost) + 1, + this.endpointMediaCopyCapacity, + "transition", + ) + ) { + this.sendError( + sourceSessionId, + "PEER_NOT_FOUND", + "Endpoint media copy transition capacity is exhausted", + ); + return "unavailable"; + } + const generation = randomBytes(16).toString("base64url"); assignments.set(hostPeerId, { ...host, @@ -3056,8 +3039,7 @@ export class HybridMediaRouter { const supportedParent = probe?.newParentPeerId === hostPeerId ? parentAssignment?.upstream.kind === "none" - : parentAssignment?.upstream.kind === "peer" || - parentAssignment?.upstream.kind === "sfu"; + : parentAssignment?.upstream.kind === "peer"; return Boolean( probe && this.pendingRoutePreparations.get(roomId) === pending && @@ -3065,15 +3047,18 @@ export class HybridMediaRouter { this.options.getShareGeneration(roomId) === probe.shareGeneration && supportedParent && parentAssignment && - this.activeParentMediaEdges( - roomId, - probe.newParentPeerId, - parentAssignment, - ) + 1 <= + endpointMediaCopyCountFits( + this.activeParentMediaEdges( + roomId, + probe.newParentPeerId, + parentAssignment, + ) + 1, this.peerRelayTopology.getDownstreamCapacity( roomId, probe.newParentPeerId, - ) && + ), + "transition", + ) && planned?.revision === pending.revision && plannedUpstream?.kind === "peer" && plannedUpstream.peerId === probe.newParentPeerId && @@ -3087,19 +3072,15 @@ export class HybridMediaRouter { parentPeerId: string, assignment: ParticipantRouteAssignment, ): number { - const selectedOverlayEdges = [...this.selectedPeerEdgeLeases.values()].some( - (lease) => - lease.roomId === roomId && - lease.parentPeerId === parentPeerId && - !assignment.childPeerIds.includes(lease.viewerPeerId), - ) - ? 1 - : 0; - return ( - assignment.childPeerIds.length + - (assignment.sfuPublicationGeneration ? 1 : 0) + - selectedOverlayEdges - ); + const selectedLease = this.selectedPeerEdgeLeasesByRoom.get(roomId); + return countEndpointMediaCopies({ + childPeerIds: assignment.childPeerIds, + publicationGeneration: assignment.sfuPublicationGeneration, + selectedChildPeerIds: + selectedLease?.parentPeerId === parentPeerId + ? [selectedLease.viewerPeerId] + : [], + }); } private ownsActiveSfuForHealthyReselection( @@ -3156,9 +3137,8 @@ export class HybridMediaRouter { } if (participant.role === "viewer") { return ( - !this.selectedPeerEdgeLeases.has( - viewerConnectionKey(participant.roomId, participant.peerId), - ) && + this.selectedPeerEdgeLeasesByRoom.get(participant.roomId) + ?.viewerPeerId !== participant.peerId && assignment.upstream.kind === "sfu" && active.sfu.rootPeerIds.includes(participant.peerId) ); @@ -3462,7 +3442,7 @@ export class HybridMediaRouter { hostPeerId, previousSfu?.publicationGeneration ?? null, rootPeerIds, - this.maxHostMediaEdges, + this.endpointMediaCopyCapacity, ); if (!controller && initialRevision === 0 && assignments.size > 1) { initialRevision = 1; @@ -3471,7 +3451,7 @@ export class HybridMediaRouter { if (!controller) { controller = new MediaRouteController({ hostPeerId, - maxEndpointMediaEdges: this.maxHostMediaEdges, + endpointMediaCopyCapacity: this.endpointMediaCopyCapacity, assignments, revision: initialRevision, sfuPublicationGeneration: @@ -3563,14 +3543,18 @@ export class HybridMediaRouter { lease.parentPeerId, ); const activeParentAssignment = route.assignments.get(lease.parentPeerId); - const selectedOverlayEdges = - activeParentAssignment?.childPeerIds.includes(lease.viewerPeerId) === true - ? 0 - : 1; + const hostPeerId = this.peerRelayTopology.getHostPeerId(roomId); + const parentSupportsOutboundRelay = + lease.parentPeerId === hostPeerId + ? activeParentAssignment?.upstream.kind === "none" + : activeParentAssignment?.upstream.kind === "peer"; const activeParentEdges = activeParentAssignment - ? activeParentAssignment.childPeerIds.length + - (activeParentAssignment.sfuPublicationGeneration ? 1 : 0) + - selectedOverlayEdges + ? countEndpointMediaCopies({ + childPeerIds: activeParentAssignment.childPeerIds, + publicationGeneration: + activeParentAssignment.sfuPublicationGeneration, + selectedChildPeerIds: [lease.viewerPeerId], + }) : Number.POSITIVE_INFINITY; return ( route.revision === lease.currentRouteRevision && @@ -3579,13 +3563,16 @@ export class HybridMediaRouter { lease.parentSessionId && this.options.roomStore.getConnectedViewer(roomId, lease.viewerPeerId) ?.sessionId === lease.viewerSessionId && + parentSupportsOutboundRelay && viewerAssignment?.parentPeerId === lease.parentPeerId && parentAssignment?.childPeerIds.includes(lease.viewerPeerId) === true && - activeParentEdges <= + endpointMediaCopyCountFits( + activeParentEdges, this.peerRelayTopology.getAdvertisedDownstreamCapacity( roomId, lease.parentPeerId, - ) + ), + ) ); } @@ -3593,28 +3580,26 @@ export class HybridMediaRouter { roomId: string, route: RoomMediaRoute, ): void { - for (const [key, lease] of this.selectedPeerEdgeLeases) { - if (lease.roomId !== roomId) { - continue; - } - lease.currentRouteRevision = route.revision; - if (this.selectedPeerEdgeLeaseIsCurrent(roomId, lease, route)) { - continue; - } - this.settleSelectedPeerEdgeLease(key); + const lease = this.selectedPeerEdgeLeasesByRoom.get(roomId); + if (!lease) { + return; + } + lease.currentRouteRevision = route.revision; + if (!this.selectedPeerEdgeLeaseIsCurrent(roomId, lease, route)) { + this.settleSelectedPeerEdgeLease(roomId); } } private settleSelectedPeerEdgeLease( - key: string, + roomId: string, excludedPeerId?: string, ): SelectedPeerEdgeLease | undefined { - const lease = this.selectedPeerEdgeLeases.get(key); + const lease = this.selectedPeerEdgeLeasesByRoom.get(roomId); if (!lease) { return undefined; } clearTimeout(lease.timer); - this.selectedPeerEdgeLeases.delete(key); + this.selectedPeerEdgeLeasesByRoom.delete(roomId); const active = this.mediaRouteControllers.get(lease.roomId)?.getActiveRoute(); if (!active) { @@ -3653,17 +3638,16 @@ export class HybridMediaRouter { if (!peer || !assignment) { return; } - for (const lease of this.selectedPeerEdgeLeases.values()) { - if ( - lease.roomId === roomId && - (lease.parentPeerId === peerId || lease.viewerPeerId === peerId) && - this.selectedPeerEdgeLeaseIsCurrent(roomId, lease, route) - ) { - this.options.sendToSession(peer.sessionId, { - ...lease.grant, - revision: route.revision, - }); - } + const lease = this.selectedPeerEdgeLeasesByRoom.get(roomId); + if ( + lease && + (lease.parentPeerId === peerId || lease.viewerPeerId === peerId) && + this.selectedPeerEdgeLeaseIsCurrent(roomId, lease, route) + ) { + this.options.sendToSession(peer.sessionId, { + ...lease.grant, + revision: route.revision, + }); } this.options.sendToSession(peer.sessionId, { type: "route-update", @@ -3795,11 +3779,10 @@ export class HybridMediaRouter { this.mediaRouteControllers.delete(roomId); this.clearRoomFailedParents(roomId); this.viewerRouteIntentsByRoom.delete(roomId); - for (const [key, lease] of this.selectedPeerEdgeLeases) { - if (lease.roomId === roomId) { - clearTimeout(lease.timer); - this.selectedPeerEdgeLeases.delete(key); - } + const selectedLease = this.selectedPeerEdgeLeasesByRoom.get(roomId); + if (selectedLease) { + clearTimeout(selectedLease.timer); + this.selectedPeerEdgeLeasesByRoom.delete(roomId); } const ingress = this.selectedSfuIngressAttempts.get(roomId); if (ingress) { @@ -3812,20 +3795,18 @@ export class HybridMediaRouter { this.sfuDisabledRoomIds.delete(roomId); this.clearRoomViewerQualityEvidenceStates(roomId); this.roomQualityMigrationCooldownUntilMs.delete(roomId); - this.refreshRelayParentQualityStates(roomId, Number.MAX_SAFE_INTEGER); } private clearSelectedEdgeTurnsForParticipant( roomId: string, peerId: string, ): void { - for (const [key, lease] of [...this.selectedPeerEdgeLeases]) { - if ( - lease.roomId === roomId && - (lease.parentPeerId === peerId || lease.viewerPeerId === peerId) - ) { - this.settleSelectedPeerEdgeLease(key, peerId); - } + const lease = this.selectedPeerEdgeLeasesByRoom.get(roomId); + if ( + lease && + (lease.parentPeerId === peerId || lease.viewerPeerId === peerId) + ) { + this.settleSelectedPeerEdgeLease(roomId, peerId); } const ingress = this.selectedSfuIngressAttempts.get(roomId); if (ingress?.hostPeerId === peerId) { @@ -3965,132 +3946,10 @@ export class HybridMediaRouter { } } - private handleRelayParentBadEvent( - roomId: string, - parentPeerId: string, - parentSessionId: string, - childPeerId: string, - badWindowKind: ViewerQualityBadWindowKind, - now: number, - ): ViewerQualityBadWindowKind | "trigger-moved" | null { - const shareGeneration = this.options.getShareGeneration(roomId); - if ( - !shareGeneration || - this.options.roomStore.getConnectedViewer(roomId, parentPeerId) - ?.sessionId !== parentSessionId - ) { - return null; - } - const key = viewerConnectionKey(roomId, parentPeerId); - const previous = this.relayParentQualityStates.get(key); - if (previous?.quarantinedUntilMs == null) { - if ( - !previous || - previous.lastChildPeerId === childPeerId || - now - previous.lastConfirmedAtMs > VIEWER_QUALITY_EVIDENCE_GAP_MS - ) { - this.relayParentQualityStates.set(key, { - parentSessionId, - shareGeneration, - lastChildPeerId: childPeerId, - lastConfirmedAtMs: now, - quarantinedUntilMs: null, - }); - return null; - } - previous.lastChildPeerId = childPeerId; - previous.lastConfirmedAtMs = now; - previous.quarantinedUntilMs = - now + VIEWER_QUALITY_REASSIGN_COOLDOWN_MS; - this.peerRelayTopology.setViewerRelayEligible(roomId, parentPeerId, false); - } - return this.resolveActivePeerEdge(roomId, childPeerId)?.parentPeerId === - parentPeerId - ? badWindowKind - : "trigger-moved"; - } - - private evacuateRelayParentChildren( - roomId: string, - parentPeerId: string, - ): void { - const childPeerIds = - this.peerRelayTopology.getAssignment(roomId, parentPeerId)?.childPeerIds ?? []; - for (const childPeerId of childPeerIds) { - if ( - this.heldQualityParentExclusionsByViewer.get( - viewerConnectionKey(roomId, childPeerId), - )?.parentPeerId === parentPeerId - ) { - continue; - } - const failedParents = this.failedParentPeerIdsByViewer.get( - viewerConnectionKey(roomId, childPeerId), - ); - const excludedParents = new Set(failedParents); - excludedParents.add(parentPeerId); - const changes = this.peerRelayTopology.reassignViewer( - roomId, childPeerId, this.connectedPeerIds(roomId), excludedParents, - MAX_PEER_RELAY_DEPTH, - ); - if (!changes) continue; - this.clearChangedConnectionIds(roomId, changes); - this.reconcileMediaRoute(roomId); - this.sendMediaAssignmentChanges(roomId, changes); - } - } - - private evacuateQuarantinedRelayParents(roomId: string): void { - const prefix = `${roomId}:`; - for (const [key, state] of this.relayParentQualityStates) { - if (key.startsWith(prefix) && state.quarantinedUntilMs !== null) { - this.evacuateRelayParentChildren(roomId, key.slice(prefix.length)); - } - } - } - - private refreshRelayParentQualityStates( - roomId: string, - currentTime = this.options.now?.() ?? Date.now(), - ): void { - if (!Number.isFinite(currentTime)) return; - const prefix = `${roomId}:`; - const shareGeneration = this.options.getShareGeneration(roomId); - for (const [key, state] of this.relayParentQualityStates) { - if (!key.startsWith(prefix)) continue; - const parentPeerId = key.slice(prefix.length); - const identityCurrent = - state.parentSessionId === - this.options.roomStore.getConnectedViewer(roomId, parentPeerId) - ?.sessionId && state.shareGeneration === shareGeneration; - const active = - state.quarantinedUntilMs === null - ? currentTime - state.lastConfirmedAtMs <= - VIEWER_QUALITY_EVIDENCE_GAP_MS - : currentTime < state.quarantinedUntilMs; - if ( - identityCurrent && - currentTime >= state.lastConfirmedAtMs && - active - ) { - continue; - } - if (state.quarantinedUntilMs !== null) { - this.peerRelayTopology.setViewerRelayEligible(roomId, parentPeerId, true); - } - this.relayParentQualityStates.delete(key); - } - } - private clearViewerQualityStateForParticipant( roomId: string, peerId: string, ): void { - const relayParentKey = viewerConnectionKey(roomId, peerId); - if (this.relayParentQualityStates.get(relayParentKey)?.quarantinedUntilMs != null) { - this.peerRelayTopology.setViewerRelayEligible(roomId, peerId, true); - } - this.relayParentQualityStates.delete(relayParentKey); this.clearHeldQualityParentExclusion(roomId, peerId); const intents = this.viewerRouteIntentsByRoom.get(roomId); for (const [viewerPeerId, intent] of [...(intents ?? [])]) { @@ -4217,10 +4076,10 @@ function participantRoutesFromTopology( hostPeerId: string, publicationGeneration: string | null, requestedRootPeerIds: readonly string[], - maxHostMediaEdges: number, + endpointMediaCopyCapacity: number, ): Map { - // Keep the peer tree intact underneath the active SFU overlay so prepare - // aborts and one-shot failback never need to reconstruct a lost baseline. + // The topology remains the failback baseline. The active SFU overlay detaches + // children because an SFU-fed Viewer has no independent outbound proof yet. const assignments = new Map(); for (const [peerId, assignment] of topologyAssignments) { assignments.set(peerId, { @@ -4254,11 +4113,24 @@ function participantRoutesFromTopology( assignments.set(rootPeerId, { ...root, upstream: { kind: "sfu" }, + childPeerIds: [], }); + for (const childPeerId of root.childPeerIds) { + const child = assignments.get(childPeerId); + if ( + child?.upstream.kind === "peer" && + child.upstream.peerId === rootPeerId + ) { + assignments.set(childPeerId, { + ...child, + upstream: { kind: "none" }, + }); + } + } } const host = assignments.get(hostPeerId); if (host && rootPeerIds.length > 0) { - const directChildLimit = Math.max(0, maxHostMediaEdges - 1); + const directChildLimit = Math.max(0, endpointMediaCopyCapacity - 1); const directChildPeerIds = host.childPeerIds.slice(0, directChildLimit); for (const suppressedPeerId of host.childPeerIds.slice(directChildLimit)) { const suppressed = assignments.get(suppressedPeerId); @@ -4302,7 +4174,7 @@ function promoteSfuRoots( ): boolean { for (const rootPeerId of rootPeerIds) { const root = assignments.get(rootPeerId); - if (!root) { + if (!root || root.childPeerIds.length > 0) { return false; } if (root.upstream.kind === "peer") { diff --git a/src/server/media-route-controller.ts b/src/server/media-route-controller.ts index 91967f5..88ceecb 100644 --- a/src/server/media-route-controller.ts +++ b/src/server/media-route-controller.ts @@ -1,16 +1,18 @@ import { - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - CURRENT_HOST_MEDIA_EDGE_LIMIT, CURRENT_SFU_ROOT_LIMIT, - DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES, MAX_MEDIA_ROUTE_REVISION, - MAX_PEER_RELAY_DOWNSTREAM_EDGES, MAX_VIEWERS_PER_ROOM_LIMIT, participantRouteAssignmentSchema, sfuPublicationGenerationSchema, type MediaRoutePhase, type ParticipantRouteAssignment, } from "../shared/protocol.js"; +import { + assertEndpointMediaCopyCapacity, + assertEndpointMediaCopyCount, + countEndpointMediaCopies, + DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, +} from "../shared/media-copy-accounting.js"; const MAX_PARTICIPANTS_PER_ROOM = MAX_VIEWERS_PER_ROOM_LIMIT + 1; @@ -37,7 +39,7 @@ export interface MediaRouteControllerOptions { revision?: number; sfuPublicationGeneration?: string | null; sfuRootPeerIds?: readonly string[]; - maxEndpointMediaEdges?: number; + endpointMediaCopyCapacity?: number; } export interface PrepareMediaRouteInput { @@ -64,28 +66,14 @@ export class MediaRouteController { private pendingRoute: StoredPendingRoute | undefined; private latestRevision: number; private readonly hostPeerId: string; - private readonly maxHostMediaEdges: number; - private readonly maxViewerMediaEdges: number; + private readonly endpointMediaCopyCapacity: number; constructor(options: MediaRouteControllerOptions) { this.hostPeerId = options.hostPeerId; - const deploymentMediaEdgeLimit = - options.maxEndpointMediaEdges ?? DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES; - if ( - !Number.isSafeInteger(deploymentMediaEdgeLimit) || - deploymentMediaEdgeLimit < 1 || - deploymentMediaEdgeLimit > MAX_PEER_RELAY_DOWNSTREAM_EDGES - ) { - throw new Error("Endpoint media edge budget is invalid"); - } - this.maxHostMediaEdges = Math.min( - deploymentMediaEdgeLimit, - CURRENT_HOST_MEDIA_EDGE_LIMIT, - ); - this.maxViewerMediaEdges = Math.min( - deploymentMediaEdgeLimit, - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - ); + this.endpointMediaCopyCapacity = + options.endpointMediaCopyCapacity ?? + DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY; + assertEndpointMediaCopyCapacity(this.endpointMediaCopyCapacity); const revision = options.revision ?? 0; assertRevision(revision); this.activeRoute = createRoute( @@ -94,8 +82,7 @@ export class MediaRouteController { options.sfuPublicationGeneration ?? null, options.sfuRootPeerIds ?? [], this.hostPeerId, - this.maxHostMediaEdges, - this.maxViewerMediaEdges, + this.endpointMediaCopyCapacity, ); this.latestRevision = revision; } @@ -129,8 +116,7 @@ export class MediaRouteController { : input.sfuPublicationGeneration, input.sfuRootPeerIds ?? this.activeRoute.sfu.rootPeerIds, this.hostPeerId, - this.maxHostMediaEdges, - this.maxViewerMediaEdges, + this.endpointMediaCopyCapacity, ); if (hasSameTopology(route, this.activeRoute)) { return undefined; @@ -153,8 +139,7 @@ export class MediaRouteController { input.sfuPublicationGeneration ?? null, input.sfuRootPeerIds ?? [], this.hostPeerId, - this.maxHostMediaEdges, - this.maxViewerMediaEdges, + this.endpointMediaCopyCapacity, ); const expectedParticipantIds = new Set(input.expectedParticipantIds); if (expectedParticipantIds.size > MAX_PARTICIPANTS_PER_ROOM) { @@ -222,9 +207,10 @@ export class MediaRouteController { hostActiveMediaEdges(): number { const edgeCount = hostMediaEdges(this.activeRoute, this.hostPeerId); - if (edgeCount > this.maxHostMediaEdges) { - throw new Error("Host active media edge budget exceeded"); - } + assertEndpointMediaCopyCount( + edgeCount, + this.endpointMediaCopyCapacity, + ); return edgeCount; } @@ -242,8 +228,7 @@ function createRoute( publicationGeneration: string | null, rootPeerIds: readonly string[], hostPeerId: string, - maxHostMediaEdges: number, - maxViewerMediaEdges: number, + endpointMediaCopyCapacity: number, ): RoomMediaRoute { assertRevision(revision); if ( @@ -272,8 +257,7 @@ function createRoute( assertRouteInvariants( route, hostPeerId, - maxHostMediaEdges, - maxViewerMediaEdges, + endpointMediaCopyCapacity, ); return route; } @@ -281,8 +265,7 @@ function createRoute( function assertRouteInvariants( route: RoomMediaRoute, hostPeerId: string, - maxHostMediaEdges: number, - maxViewerMediaEdges: number, + endpointMediaCopyCapacity: number, ): void { const hostAssignment = route.assignments.get(hostPeerId); if (!hostAssignment) { @@ -312,11 +295,13 @@ function assertRouteInvariants( } for (const [peerId, assignment] of route.assignments) { - const downstreamEdgeLimit = - peerId === hostPeerId ? maxHostMediaEdges : maxViewerMediaEdges; - if (assignment.childPeerIds.length > downstreamEdgeLimit) { - throw new Error("Participant active media edge budget exceeded"); - } + assertEndpointMediaCopyCount( + countEndpointMediaCopies({ + childPeerIds: assignment.childPeerIds, + publicationGeneration: assignment.sfuPublicationGeneration, + }), + endpointMediaCopyCapacity, + ); if ( peerId !== hostPeerId && assignment.sfuPublicationGeneration !== null @@ -351,6 +336,12 @@ function assertRouteInvariants( if (assignment.upstream.kind === "sfu" && !rootPeerIds.has(peerId)) { throw new Error("Every SFU upstream must belong to the root allowlist"); } + if ( + assignment.upstream.kind === "sfu" && + assignment.childPeerIds.length > 0 + ) { + throw new Error("SFU-fed viewers require outbound relay proof"); + } } for (const rootPeerId of rootPeerIds) { @@ -361,9 +352,10 @@ function assertRouteInvariants( } assertAcyclicPeerEdges(route.assignments); - if (hostMediaEdges(route, hostPeerId) > maxHostMediaEdges) { - throw new Error("Host active media edge budget exceeded"); - } + assertEndpointMediaCopyCount( + hostMediaEdges(route, hostPeerId), + endpointMediaCopyCapacity, + ); } function assertAcyclicPeerEdges( @@ -384,11 +376,14 @@ function assertAcyclicPeerEdges( } function hostMediaEdges(route: RoomMediaRoute, hostPeerId: string): number { - const directChildren = route.assignments.get(hostPeerId)?.childPeerIds.length; - if (directChildren === undefined) { + const hostAssignment = route.assignments.get(hostPeerId); + if (!hostAssignment) { throw new Error("Media route is missing its host assignment"); } - return directChildren + (route.sfu.publicationGeneration === null ? 0 : 1); + return countEndpointMediaCopies({ + childPeerIds: hostAssignment.childPeerIds, + publicationGeneration: route.sfu.publicationGeneration, + }); } function hasSameTopology(left: RoomMediaRoute, right: RoomMediaRoute): boolean { diff --git a/src/server/peer-relay-topology.ts b/src/server/peer-relay-topology.ts index 86bcf82..33b1c7d 100644 --- a/src/server/peer-relay-topology.ts +++ b/src/server/peer-relay-topology.ts @@ -3,11 +3,9 @@ import type { RelayDownstreamEdges, } from "../shared/protocol.js"; import { - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - CURRENT_HOST_MEDIA_EDGE_LIMIT, - DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES, - MAX_PEER_RELAY_DOWNSTREAM_EDGES, -} from "../shared/protocol.js"; + assertEndpointMediaCopyCapacity, + DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, +} from "../shared/media-copy-accounting.js"; export interface MediaAssignmentChange { peerId: string; @@ -26,42 +24,26 @@ interface RelayViewer { parentPeerId: string | null; childPeerIds: string[]; downstreamEdges: RelayDownstreamEdges; - relayEligible: boolean; } interface RelayRoom { hostPeerId?: string; hostChildPeerIds: string[]; - maxHostDownstreamEdges: RelayDownstreamEdges; - maxViewerDownstreamEdges: RelayDownstreamEdges; + endpointMediaCopyCapacity: RelayDownstreamEdges; viewers: Map; nextOrder: number; } export class PeerRelayTopology { private readonly rooms = new Map(); - private readonly maxHostDownstreamEdges: RelayDownstreamEdges; - private readonly maxViewerDownstreamEdges: RelayDownstreamEdges; + private readonly endpointMediaCopyCapacity: RelayDownstreamEdges; constructor( - deploymentDownstreamEdgeLimit: RelayDownstreamEdges = - DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES, + endpointMediaCopyCapacity: RelayDownstreamEdges = + DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY, ) { - if ( - !Number.isSafeInteger(deploymentDownstreamEdgeLimit) || - deploymentDownstreamEdgeLimit < 1 || - deploymentDownstreamEdgeLimit > MAX_PEER_RELAY_DOWNSTREAM_EDGES - ) { - throw new Error("Peer relay downstream limit is invalid"); - } - this.maxHostDownstreamEdges = Math.min( - deploymentDownstreamEdgeLimit, - CURRENT_HOST_MEDIA_EDGE_LIMIT, - ); - this.maxViewerDownstreamEdges = Math.min( - deploymentDownstreamEdgeLimit, - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, - ); + assertEndpointMediaCopyCapacity(endpointMediaCopyCapacity); + this.endpointMediaCopyCapacity = endpointMediaCopyCapacity; } setHost( @@ -105,7 +87,6 @@ export class PeerRelayTopology { parentPeerId: null, childPeerIds: [], downstreamEdges: 0, - relayEligible: true, }); room.nextOrder += 1; } else { @@ -135,7 +116,7 @@ export class PeerRelayTopology { const before = snapshot(room); const boundedDownstreamEdges = Math.min( downstreamEdges, - room.maxViewerDownstreamEdges, + room.endpointMediaCopyCapacity, ); const previousDownstreamEdges = viewer.downstreamEdges; viewer.downstreamEdges = boundedDownstreamEdges; @@ -236,17 +217,6 @@ export class PeerRelayTopology { return room ? advertisedDownstreamCapacity(room, peerId) : 0; } - setViewerRelayEligible( - roomId: string, - peerId: string, - eligible: boolean, - ): void { - const viewer = this.rooms.get(roomId)?.viewers.get(peerId); - if (viewer) { - viewer.relayEligible = eligible; - } - } - getHostPeerId(roomId: string): string | undefined { return this.rooms.get(roomId)?.hostPeerId; } @@ -338,8 +308,7 @@ export class PeerRelayTopology { if (!room) { room = { hostChildPeerIds: [], - maxHostDownstreamEdges: this.maxHostDownstreamEdges, - maxViewerDownstreamEdges: this.maxViewerDownstreamEdges, + endpointMediaCopyCapacity: this.endpointMediaCopyCapacity, viewers: new Map(), nextOrder: 0, }; @@ -509,9 +478,7 @@ function downstreamCapacity( room: RelayRoom, peerId: string, ): RelayDownstreamEdges { - return room.hostPeerId === peerId || room.viewers.get(peerId)?.relayEligible - ? advertisedDownstreamCapacity(room, peerId) - : 0; + return advertisedDownstreamCapacity(room, peerId); } function advertisedDownstreamCapacity( @@ -519,7 +486,7 @@ function advertisedDownstreamCapacity( peerId: string, ): RelayDownstreamEdges { return room.hostPeerId === peerId - ? room.maxHostDownstreamEdges + ? room.endpointMediaCopyCapacity : (room.viewers.get(peerId)?.downstreamEdges ?? 0); } diff --git a/src/server/signaling.ts b/src/server/signaling.ts index a25e1eb..a3295e2 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -84,7 +84,7 @@ export interface SignalingOptions { server: HttpServer; roomStore: RoomStore; peerAssistedMedia: boolean; - maxPeerRelayDownstreamEdges: number; + endpointMediaCopyCapacity: number; sfuFallback?: SfuFallbackOptions; selectedEdgeTurn?: SelectedEdgeTurnConfig; ice: IceConfigOptions; @@ -158,7 +158,7 @@ export class SignalingServer { if (options.peerAssistedMedia) { this.hybridMediaRouter = new HybridMediaRouter({ roomStore: options.roomStore, - maxPeerRelayDownstreamEdges: options.maxPeerRelayDownstreamEdges, + endpointMediaCopyCapacity: options.endpointMediaCopyCapacity, sfuFallback: options.sfuFallback, selectedEdgeTurn: options.selectedEdgeTurn, sendToSession: (sessionId, message) => diff --git a/src/shared/media-copy-accounting.ts b/src/shared/media-copy-accounting.ts new file mode 100644 index 0000000..42b72c6 --- /dev/null +++ b/src/shared/media-copy-accounting.ts @@ -0,0 +1,74 @@ +export const DEFAULT_ENDPOINT_MEDIA_COPY_CAPACITY = 2; +export const MAX_ENDPOINT_MEDIA_COPY_CAPACITY = 3; + +export type EndpointMediaCopyPhase = "steady" | "transition"; + +export interface EndpointMediaCopySnapshot { + childPeerIds: readonly string[]; + publicationGeneration?: string | null; + selectedChildPeerIds?: Iterable; +} + +export function countEndpointMediaCopies( + snapshot: EndpointMediaCopySnapshot, +): number { + const childPeerIds = new Set(snapshot.childPeerIds); + const hiddenSelectedChildPeerIds = new Set(); + for (const childPeerId of snapshot.selectedChildPeerIds ?? []) { + if (!childPeerIds.has(childPeerId)) { + hiddenSelectedChildPeerIds.add(childPeerId); + } + } + return ( + snapshot.childPeerIds.length + + (snapshot.publicationGeneration ? 1 : 0) + + hiddenSelectedChildPeerIds.size + ); +} + +export function endpointMediaCopyLimit( + capacity: number, + phase: EndpointMediaCopyPhase, +): number { + assertEndpointMediaCopyCapacity(capacity); + return phase === "steady" + ? capacity + : Math.min(capacity + 1, MAX_ENDPOINT_MEDIA_COPY_CAPACITY); +} + +export function endpointMediaCopyCountFits( + count: number, + capacity: number, + phase: EndpointMediaCopyPhase = "steady", +): boolean { + return ( + Number.isSafeInteger(count) && + count >= 0 && + isEndpointMediaCopyCapacity(capacity) && + count <= endpointMediaCopyLimit(capacity, phase) + ); +} + +export function isEndpointMediaCopyCapacity(capacity: number): boolean { + return ( + Number.isSafeInteger(capacity) && + capacity >= 1 && + capacity <= MAX_ENDPOINT_MEDIA_COPY_CAPACITY + ); +} + +export function assertEndpointMediaCopyCapacity(capacity: number): void { + if (!isEndpointMediaCopyCapacity(capacity)) { + throw new Error("Endpoint media copy capacity must be 1, 2, or 3"); + } +} + +export function assertEndpointMediaCopyCount( + count: number, + capacity: number, + phase: EndpointMediaCopyPhase = "steady", +): void { + if (!endpointMediaCopyCountFits(count, capacity, phase)) { + throw new Error(`Endpoint ${phase} media copy capacity exceeded`); + } +} diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index 9448343..332a78b 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -1,11 +1,12 @@ import { z } from "zod"; +import { MAX_ENDPOINT_MEDIA_COPY_CAPACITY } from "./media-copy-accounting.js"; import { isCanonicalVideoCodecEvidence } from "./video-codec-evidence.js"; export const MAX_VIEWERS_PER_ROOM_LIMIT = 16; export const MAX_PARTICIPANTS_PER_ROOM_LIMIT = MAX_VIEWERS_PER_ROOM_LIMIT + 1; export const MAX_SIGNAL_BYTES = 64 * 1024; -export const SIGNALING_PROTOCOL = "screener-v5"; +export const SIGNALING_PROTOCOL = "screener-v6"; export const ROOM_CODE_LENGTH = 12; export const MAX_MEDIA_ROUTE_REVISION = Number.MAX_SAFE_INTEGER; export const MAX_SFU_TOKEN_LENGTH = 8 * 1024; @@ -204,18 +205,12 @@ export const DEFAULT_QUALITY_SETTINGS = { screenAudioQuality: "music", } as const satisfies QualitySettings; -export const DEFAULT_PEER_RELAY_DOWNSTREAM_EDGES = 2; -export const CURRENT_HOST_MEDIA_EDGE_LIMIT = 2; -export const CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT = 1; export const CURRENT_SFU_ROOT_LIMIT = 2; -// The wire retains the accepted future 0/1/2/3 capability envelope. Current -// release policy is enforced server-side by role and is intentionally lower. -export const MAX_PEER_RELAY_DOWNSTREAM_EDGES = 3; export const relayDownstreamEdgesSchema = z .number() .int() .min(0) - .max(MAX_PEER_RELAY_DOWNSTREAM_EDGES); + .max(MAX_ENDPOINT_MEDIA_COPY_CAPACITY); export type RelayDownstreamEdges = z.infer; function isValidStunUrl(value: string): boolean { @@ -354,7 +349,7 @@ export const mediaAssignmentSchema = z parentPeerId: opaqueIdSchema.nullable(), childPeerIds: z .array(opaqueIdSchema) - .max(MAX_PEER_RELAY_DOWNSTREAM_EDGES), + .max(MAX_ENDPOINT_MEDIA_COPY_CAPACITY), }) .strict(); export type MediaAssignment = z.infer; @@ -375,7 +370,7 @@ export const participantRouteAssignmentSchema = z upstream: mediaRouteUpstreamSchema, childPeerIds: z .array(opaqueIdSchema) - .max(MAX_PEER_RELAY_DOWNSTREAM_EDGES) + .max(MAX_ENDPOINT_MEDIA_COPY_CAPACITY) .refine((peerIds) => new Set(peerIds).size === peerIds.length), sfuPublicationGeneration: sfuPublicationGenerationSchema.nullable(), }) diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index 7a42f88..8a7c891 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -640,7 +640,7 @@ describe("client signaling recovery policy", () => { Object.defineProperty(authenticated, "data", { value: JSON.stringify({ type: "authenticated", - protocol: "screener-v5", + protocol: "screener-v6", role: "viewer", peerId: "viewer_12345678", roomExpiresAt: null, @@ -713,7 +713,7 @@ describe("client signaling recovery policy", () => { socket.dispatchEvent(new Event("open")); receive(socket, { type: "authenticated", - protocol: "screener-v5", + protocol: "screener-v6", role: "host", peerId: "host_12345678", roomExpiresAt: null, @@ -837,7 +837,7 @@ describe("client signaling recovery policy", () => { sockets[0]!.dispatchEvent(new Event("open")); receive({ type: "authenticated", - protocol: "screener-v5", + protocol: "screener-v6", role: "viewer", peerId: "viewer_12345678", roomExpiresAt: null, @@ -923,7 +923,7 @@ describe("client signaling recovery policy", () => { iceConfig: { iceServers: [] }, }), ])( - "terminates once when a server payload is incompatible with v2", + "terminates once when a server payload is incompatible with the current wire", (payload) => { const sockets: FakeWebSocket[] = []; class FakeWebSocket extends EventTarget { @@ -964,7 +964,7 @@ describe("client signaling recovery policy", () => { JSON.parse(String(sockets[0]!.send.mock.calls[0]![0])), ).toMatchObject({ type: "authenticate", - protocol: "screener-v5", + protocol: "screener-v6", }); const message = new Event("message"); Object.defineProperty(message, "data", { value: payload }); diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index 48e5d17..ccf03cf 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -1443,7 +1443,7 @@ describe("ViewerRelay downstream ownership", () => { relay.dispose(); }); - it("admits one provisional child only while the current Viewer cap has room", async () => { + it("admits provisional children until endpoint cap three is full", async () => { const relay = new ViewerRelay( { iceServers: [] }, QUALITY_PROFILES["720p30"], @@ -1455,14 +1455,22 @@ describe("ViewerRelay downstream ownership", () => { preparedConnection.connectionState = "connected"; relay.activateChildren(1, ["child-0"]); expect(FakePeerConnection.activeCount).toBe(1); + expect(relay.prepareChild(2, ["child-0", "child-1"])).toBe(true); + FakePeerConnection.latest!.connectionState = "connected"; + relay.activateChildren(2, ["child-0", "child-1"]); + expect(relay.prepareChild(3, ["child-0", "child-1", "child-2"])).toBe( + true, + ); + FakePeerConnection.latest!.connectionState = "connected"; + relay.activateChildren(3, ["child-0", "child-1", "child-2"]); expect( - relay.prepareChild(2, ["child-0", "child-1"]), + relay.prepareChild(4, ["child-0", "child-1", "child-2", "child-3"]), ).toBe(false); - expect(FakePeerConnection.activeCount).toBe(1); + expect(FakePeerConnection.activeCount).toBe(3); relay.dispose(); }); - it("clamps a malicious multi-child assignment to one physical Viewer edge", async () => { + it("clamps a malicious fourth child to three physical endpoint copies", async () => { const targets: string[] = []; const relay = new ViewerRelay( { iceServers: [] }, @@ -1479,10 +1487,17 @@ describe("ViewerRelay downstream ownership", () => { relay.setChildren(["child-a", "child-b", "child-c", "ignored-fourth"]); relay.setStream(createStream(firstVideo, firstAudio)); - await vi.waitFor(() => expect(targets).toEqual(["child-a"])); - const [childAConnection] = FakePeerConnection.instances; + await vi.waitFor(() => + expect(targets).toEqual(["child-a", "child-b", "child-c"]), + ); + const [childAConnection, childBConnection, childCConnection] = + FakePeerConnection.instances; expect(childAConnection).toBeDefined(); + expect(childBConnection).toBeDefined(); + expect(childCConnection).toBeDefined(); childAConnection!.connectionState = "connected"; + childBConnection!.connectionState = "connected"; + childCConnection!.connectionState = "connected"; const retiredConnectionId = relay.getSnapshot("child-a")!.connectionId; const nextVideo = createTrack("video", "next-video"); @@ -1494,13 +1509,12 @@ describe("ViewerRelay downstream ownership", () => { }); relay.setChildren(["child-b", "child-c", "child-d"]); - await vi.waitFor(() => expect(targets).toContain("child-b")); + await vi.waitFor(() => expect(targets).toContain("child-d")); expect(childAConnection!.connectionState).toBe("closed"); - const childBConnection = FakePeerConnection.latest!; - childBConnection.connectionState = "connected"; - expect(targets).toEqual(["child-a", "child-b"]); - expect(FakePeerConnection.activeCount).toBe(1); - expect(FakePeerConnection.peakActiveCount).toBe(1); + FakePeerConnection.latest!.connectionState = "connected"; + expect(targets).toEqual(["child-a", "child-b", "child-c", "child-d"]); + expect(FakePeerConnection.activeCount).toBe(3); + expect(FakePeerConnection.peakActiveCount).toBe(3); expect(targets).not.toContain("ignored-fourth"); await relay.acceptSignal("stale-child", { @@ -1510,14 +1524,14 @@ describe("ViewerRelay downstream ownership", () => { }, 7); await relay.recover("stale-child", "stale-connection", true); await relay.recover("child-b", "wrong-connection", true); - expect(FakePeerConnection.activeCount).toBe(1); + expect(FakePeerConnection.activeCount).toBe(3); const childBConnectionId = relay.getSnapshot("child-b")!.connectionId; await relay.recover("child-b", childBConnectionId, true); await vi.waitFor(() => expect(FakePeerConnection.latest).not.toBe(childBConnection), ); - expect(FakePeerConnection.activeCount).toBe(1); - expect(FakePeerConnection.peakActiveCount).toBe(1); + expect(FakePeerConnection.activeCount).toBe(3); + expect(FakePeerConnection.peakActiveCount).toBe(3); expect( relay.startSelectedEdgeTurn( @@ -1530,8 +1544,8 @@ describe("ViewerRelay downstream ownership", () => { 7, ), ).toBe(false); - expect(FakePeerConnection.activeCount).toBe(1); - expect(FakePeerConnection.peakActiveCount).toBe(1); + expect(FakePeerConnection.activeCount).toBe(3); + expect(FakePeerConnection.peakActiveCount).toBe(3); relay.dispose(); }); diff --git a/tests/media-assignment.test.ts b/tests/media-assignment.test.ts index 3dfba72..139bf63 100644 --- a/tests/media-assignment.test.ts +++ b/tests/media-assignment.test.ts @@ -3,8 +3,7 @@ import { describe, expect, it } from "vitest"; import type { SignalPayload } from "../src/shared/protocol.ts"; import { limitMediaAssignment, - MAX_HOST_MEDIA_CHILDREN, - MAX_VIEWER_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, reconcileBoundedMediaChildren, retainSelectedMediaParent, viewerRestartMessage, @@ -18,9 +17,8 @@ const offer: SignalPayload = { }; describe("peer-assisted client assignment", () => { - it("keeps host and viewer child counts within their client limits", () => { - expect(MAX_HOST_MEDIA_CHILDREN).toBe(2); - expect(MAX_VIEWER_MEDIA_CHILDREN).toBe(1); + it("keeps endpoint child counts within the shared client limit", () => { + expect(MAX_ENDPOINT_MEDIA_CHILDREN).toBe(3); const assignment = { parentPeerId: "parent_12345678", @@ -34,20 +32,15 @@ describe("peer-assisted client assignment", () => { }; expect( - limitMediaAssignment(assignment, MAX_HOST_MEDIA_CHILDREN), + limitMediaAssignment(assignment, MAX_ENDPOINT_MEDIA_CHILDREN), ).toEqual({ parentPeerId: "parent_12345678", childPeerIds: [ "child_12345678", "child_abcdefgh", + "child_qwertyui", ], }); - expect( - limitMediaAssignment(assignment, MAX_VIEWER_MEDIA_CHILDREN), - ).toEqual({ - parentPeerId: "parent_12345678", - childPeerIds: ["child_12345678"], - }); }); it("removes stale host edges before starting replacement children", () => { @@ -56,7 +49,7 @@ describe("peer-assisted client assignment", () => { reconcileBoundedMediaChildren( new Set(["old-child", "kept-child"]), ["kept-child", "new-child", "overflow-child"], - MAX_HOST_MEDIA_CHILDREN, + MAX_ENDPOINT_MEDIA_CHILDREN, (peerId) => operations.push(`remove:${peerId}`), (peerId) => operations.push(`start:${peerId}`), ); @@ -65,6 +58,7 @@ describe("peer-assisted client assignment", () => { "remove:old-child", "start:kept-child", "start:new-child", + "start:overflow-child", ]); }); diff --git a/tests/media-copy-accounting.test.ts b/tests/media-copy-accounting.test.ts new file mode 100644 index 0000000..e0eab78 --- /dev/null +++ b/tests/media-copy-accounting.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from "vitest"; + +import { + assertEndpointMediaCopyCapacity, + countEndpointMediaCopies, + endpointMediaCopyCountFits, + endpointMediaCopyLimit, +} from "../src/shared/media-copy-accounting.ts"; + +describe("endpoint media copy accounting", () => { + it("counts children and one Host publication while upstream remains free", () => { + expect( + countEndpointMediaCopies({ + childPeerIds: ["child-a", "child-b"], + }), + ).toBe(2); + expect( + countEndpointMediaCopies({ + childPeerIds: ["child-a", "child-b"], + publicationGeneration: "publication-a", + }), + ).toBe(3); + }); + + it("does not double-count a committed selected transport for the same edge", () => { + expect( + countEndpointMediaCopies({ + childPeerIds: ["child-a"], + selectedChildPeerIds: ["child-a"], + }), + ).toBe(1); + }); + + it("charges each hidden carried selected edge as another copy", () => { + expect( + countEndpointMediaCopies({ + childPeerIds: ["child-a"], + selectedChildPeerIds: ["child-b", "child-b", "child-c"], + }), + ).toBe(3); + }); + + it.each([ + [1, 1, 2], + [2, 2, 3], + [3, 3, 3], + ] as const)( + "bounds capacity %i at %i steady and %i during transition", + (capacity, steady, transition) => { + expect(endpointMediaCopyLimit(capacity, "steady")).toBe(steady); + expect(endpointMediaCopyLimit(capacity, "transition")).toBe(transition); + expect( + endpointMediaCopyCountFits(transition, capacity, "transition"), + ).toBe(true); + expect( + endpointMediaCopyCountFits(transition + 1, capacity, "transition"), + ).toBe(false); + }, + ); + + it.each([0, 4, 1.5, Number.NaN])( + "rejects invalid capacity %s", + (capacity) => { + expect(() => assertEndpointMediaCopyCapacity(capacity)).toThrow( + "must be 1, 2, or 3", + ); + }, + ); + + it("returns false instead of throwing for an unavailable runtime capacity", () => { + expect(endpointMediaCopyCountFits(0, 0)).toBe(false); + }); +}); diff --git a/tests/media-route-controller.test.ts b/tests/media-route-controller.test.ts index 9451855..971b233 100644 --- a/tests/media-route-controller.test.ts +++ b/tests/media-route-controller.test.ts @@ -129,7 +129,7 @@ describe("MediaRouteController", () => { sfuPublicationGeneration: GENERATION_A, sfuRootPeerIds: [SFU_ROOT_A, SFU_ROOT_B], }), - ).toThrow("Host active media edge budget exceeded"); + ).toThrow("Endpoint steady media copy capacity exceeded"); }); it("atomically commits only after every expected participant is ready", () => { @@ -299,7 +299,7 @@ describe("MediaRouteController", () => { sfuPublicationGeneration: GENERATION_A, sfuRootPeerIds: [SFU_ROOT_A], }), - ).toThrow("Host active media edge budget exceeded"); + ).toThrow("Endpoint steady media copy capacity exceeded"); const nonHostPublisher = directAssignments(); nonHostPublisher.set( @@ -338,7 +338,7 @@ describe("MediaRouteController", () => { ).toThrow("SFU roots and publication generation must be active together"); }); - it("does not let an explicit three-edge setting lift release role limits", () => { + it("applies capacity three uniformly to Host and Viewer endpoints", () => { const threeHostChildren = new Map([ [ HOST, @@ -351,13 +351,12 @@ describe("MediaRouteController", () => { [SFU_ROOT_A, assignment({ kind: "peer", peerId: HOST })], [THIRD_CHILD, assignment({ kind: "peer", peerId: HOST })], ]); - expect(() => - new MediaRouteController({ - hostPeerId: HOST, - assignments: threeHostChildren, - maxEndpointMediaEdges: 3, - }), - ).toThrow("Participant active media edge budget exceeded"); + const hostAtCapacity = new MediaRouteController({ + hostPeerId: HOST, + assignments: threeHostChildren, + endpointMediaCopyCapacity: 3, + }); + expect(hostAtCapacity.hostActiveMediaEdges()).toBe(3); const viewerWithTwoChildren = new Map([ [HOST, assignment({ kind: "none" }, [DIRECT_ROOT])], @@ -371,13 +370,13 @@ describe("MediaRouteController", () => { [SFU_ROOT_A, assignment({ kind: "peer", peerId: DIRECT_ROOT })], [SFU_ROOT_B, assignment({ kind: "peer", peerId: DIRECT_ROOT })], ]); - expect(() => + expect( new MediaRouteController({ hostPeerId: HOST, assignments: viewerWithTwoChildren, - maxEndpointMediaEdges: 3, - }), - ).toThrow("Participant active media edge budget exceeded"); + endpointMediaCopyCapacity: 3, + }).getActiveRoute().assignments.get(DIRECT_ROOT)?.childPeerIds, + ).toHaveLength(2); const publicationAtHostLimit = new Map([ [ @@ -390,7 +389,7 @@ describe("MediaRouteController", () => { const allowed = new MediaRouteController({ hostPeerId: HOST, assignments: publicationAtHostLimit, - maxEndpointMediaEdges: 3, + endpointMediaCopyCapacity: 3, sfuPublicationGeneration: GENERATION_A, sfuRootPeerIds: [SFU_ROOT_A], }); @@ -401,7 +400,7 @@ describe("MediaRouteController", () => { HOST, assignment( { kind: "none" }, - [DIRECT_ROOT, THIRD_CHILD], + [DIRECT_ROOT, THIRD_CHILD, SFU_ROOT_B], GENERATION_A, ), ); @@ -409,15 +408,19 @@ describe("MediaRouteController", () => { THIRD_CHILD, assignment({ kind: "peer", peerId: HOST }), ); + publicationOverHostLimit.set( + SFU_ROOT_B, + assignment({ kind: "peer", peerId: HOST }), + ); expect(() => new MediaRouteController({ hostPeerId: HOST, assignments: publicationOverHostLimit, - maxEndpointMediaEdges: 3, + endpointMediaCopyCapacity: 3, sfuPublicationGeneration: GENERATION_A, sfuRootPeerIds: [SFU_ROOT_A], }), - ).toThrow("Host active media edge budget exceeded"); + ).toThrow("Endpoint steady media copy capacity exceeded"); }); it("rejects reciprocal peer cycles", () => { diff --git a/tests/peer-assisted-benchmark.test.ts b/tests/peer-assisted-benchmark.test.ts index cd273f2..4807582 100644 --- a/tests/peer-assisted-benchmark.test.ts +++ b/tests/peer-assisted-benchmark.test.ts @@ -248,11 +248,12 @@ describe("peer topology loopback configuration", () => { ).toThrow(/selected viewer count/); }); - it("keeps Host cap2 by default and permits only deployment tightening", () => { + it("defaults to cap two and accepts the deployment range", () => { expect(parseExpectedEndpointCap(undefined)).toBe(2); expect(parseExpectedEndpointCap("1")).toBe(1); - for (const value of ["0", "2.5", "3", "4"]) { - expect(() => parseExpectedEndpointCap(value)).toThrow(/integer from 1 to 2/); + expect(parseExpectedEndpointCap("3")).toBe(3); + for (const value of ["0", "2.5", "4"]) { + expect(() => parseExpectedEndpointCap(value)).toThrow(/integer from 1 to 3/); } }); @@ -389,7 +390,7 @@ describe("peer topology loopback observations", () => { expect(summary.browserProcessResources.averageCpuUtilizationPercent).toBeCloseTo(50); }); - it("checks Host cap2 and Browser Viewer cap1 independently", () => { + it("checks one endpoint cap for Host and Viewer senders", () => { const initial = [ page("host", "host", 2, 0), ...Array.from({ length: 3 }, (_, index) => @@ -412,7 +413,7 @@ describe("peer topology loopback observations", () => { expect(checks.find((check) => check.name === "relay-active-media-edges")?.passed).toBe(true); const overused = { ...summary, - maxRelayActiveMediaEdges: 2, + maxRelayActiveMediaEdges: 3, }; expect( buildRunChecks(overused, 3, "720p30", 2).find( diff --git a/tests/peer-relay-topology.test.ts b/tests/peer-relay-topology.test.ts index d35c3f9..6e3a70c 100644 --- a/tests/peer-relay-topology.test.ts +++ b/tests/peer-relay-topology.test.ts @@ -30,18 +30,20 @@ function expectValidTree( roomId: string, hostPeerId: string, viewerPeerIds: readonly string[], - maxViewerChildren = 1, + endpointMediaCopyCapacity = 2, ): number { const hostAssignment = topology.getAssignment(roomId, hostPeerId)!; expect(hostAssignment.parentPeerId).toBeNull(); - expect(hostAssignment.childPeerIds.length).toBeLessThanOrEqual(2); + expect(hostAssignment.childPeerIds.length).toBeLessThanOrEqual( + endpointMediaCopyCapacity, + ); const allChildren = new Set(hostAssignment.childPeerIds); let maximumDepth = 0; for (const peerId of viewerPeerIds) { const assignment = topology.getAssignment(roomId, peerId)!; expect(assignment.childPeerIds.length).toBeLessThanOrEqual( - maxViewerChildren, + endpointMediaCopyCapacity, ); for (const childPeerId of assignment.childPeerIds) { expect(allChildren.has(childPeerId)).toBe(false); @@ -197,42 +199,7 @@ describe("PeerRelayTopology", () => { expect(topology.getAssignment("room", "viewer-d")?.parentPeerId).toBeNull(); }); - it("keeps server relay ineligibility separate from advertised capacity", () => { - const topology = new PeerRelayTopology(); - const peers = connected("host", "relay", "sibling", "child", "pending"); - topology.setHost("room", "host", peers); - topology.addViewer("room", "relay", peers); - topology.addViewer("room", "sibling", peers); - topology.setViewerRelayCapacity("room", "relay", 1, peers); - topology.addViewer("room", "child", peers); - - topology.setViewerRelayEligible("room", "relay", false); - expect(topology.getDownstreamCapacity("room", "relay")).toBe(0); - expect(topology.getAdvertisedDownstreamCapacity("room", "relay")).toBe(1); - expect(topology.getAssignment("room", "child")?.parentPeerId).toBe("relay"); - expect( - topology.setViewerRelayCapacity("room", "relay", 2, peers), - ).toEqual([]); - expect(topology.getAdvertisedDownstreamCapacity("room", "relay")).toBe(1); - topology.addViewer("room", "pending", peers); - expect(topology.getAssignment("room", "pending")?.parentPeerId).toBeNull(); - - topology.setViewerRelayEligible("room", "relay", true); - topology.setViewerRelayCapacity("room", "relay", 2, peers); - expect(topology.getAssignment("room", "pending")?.parentPeerId).toBeNull(); - topology.removeViewer( - "room", - "child", - connected("host", "relay", "sibling", "pending"), - ); - expect(topology.getAssignment("room", "pending")?.parentPeerId).toBe( - "relay", - ); - topology.setViewerRelayEligible("room", "host", false); - expect(topology.getDownstreamCapacity("room", "host")).toBe(2); - }); - - it("applies role-aware release limits even when code supplies three", () => { + it("applies one deployment capacity to Host and Viewer endpoints", () => { const peers = connected("host", "relay"); const defaultTopology = new PeerRelayTopology(); defaultTopology.setHost("default", "host", peers); @@ -243,7 +210,7 @@ describe("PeerRelayTopology", () => { ).toBe(2); expect( defaultTopology.getAdvertisedDownstreamCapacity("default", "relay"), - ).toBe(1); + ).toBe(2); const expandedTopology = new PeerRelayTopology(3); expandedTopology.setHost("expanded", "host", peers); @@ -251,10 +218,10 @@ describe("PeerRelayTopology", () => { expandedTopology.setViewerRelayCapacity("expanded", "relay", 3, peers); expect( expandedTopology.getAdvertisedDownstreamCapacity("expanded", "host"), - ).toBe(2); + ).toBe(3); expect( expandedTopology.getAdvertisedDownstreamCapacity("expanded", "relay"), - ).toBe(1); + ).toBe(3); const tightenedTopology = new PeerRelayTopology(1); tightenedTopology.setHost("tightened", "host", peers); @@ -358,7 +325,7 @@ describe("PeerRelayTopology", () => { expect(expectValidTree(topology, roomId, hostPeerId, viewerPeerIds)).toBe(4); }); - it("clamps old capacity-three viewers to one child in an eight-viewer tree", () => { + it("uses capacity three uniformly in an eight-viewer tree", () => { const topology = new PeerRelayTopology(3); const roomId = "room"; const hostPeerId = "host"; @@ -382,23 +349,25 @@ describe("PeerRelayTopology", () => { roomId, hostPeerId, viewerPeerIds.slice(0, index + 1), - 1, + 3, ); } expect(topology.getAssignment(roomId, "viewer-1")?.childPeerIds).toEqual([ - "viewer-3", + "viewer-4", + "viewer-5", + "viewer-6", ]); - expect(topology.getDownstreamCapacity(roomId, "viewer-1")).toBe(1); + expect(topology.getDownstreamCapacity(roomId, "viewer-1")).toBe(3); expect(topology.getAssignment(roomId, "viewer-5")?.parentPeerId).toBe( - "viewer-3", + "viewer-1", ); expect( - expectValidTree(topology, roomId, hostPeerId, viewerPeerIds, 1), - ).toBe(4); + expectValidTree(topology, roomId, hostPeerId, viewerPeerIds, 3), + ).toBe(2); }); - it("reattaches a relay child without creating a second viewer child", () => { + it("reattaches relay children within the shared endpoint capacity", () => { const topology = new PeerRelayTopology(); const peers = connected("host", "relay", "sibling", "child-a", "child-b"); topology.setHost("room", "host", peers); @@ -408,10 +377,9 @@ describe("PeerRelayTopology", () => { addRelayViewer(topology, "room", "child-b", peers, 2); expect(topology.getAssignment("room", "relay")?.childPeerIds).toEqual([ "child-a", - ]); - expect(topology.getAssignment("room", "sibling")?.childPeerIds).toEqual([ "child-b", ]); + expect(topology.getAssignment("room", "sibling")?.childPeerIds).toEqual([]); topology.removeViewer( "room", @@ -431,7 +399,7 @@ describe("PeerRelayTopology", () => { "room", "host", ["sibling", "child-a", "child-b"], - 1, + 2, ), ).toBe(2); }); diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index faa9276..c3a36fc 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -343,6 +343,18 @@ describe("client signaling protocol", () => { expect(result.success).toBe(false); }); + it("rejects the previous signaling wire", () => { + expect( + clientMessageSchema.safeParse({ + type: "authenticate", + protocol: "screener-v5", + roomId, + role: "viewer", + clientId: "client_12345678", + }).success, + ).toBe(false); + }); + it("requires a host token and rejects viewer tokens or malformed room codes", () => { expect( clientMessageSchema.safeParse({ diff --git a/tests/relay-capability.test.ts b/tests/relay-capability.test.ts index e38cacd..b12ce2d 100644 --- a/tests/relay-capability.test.ts +++ b/tests/relay-capability.test.ts @@ -5,10 +5,10 @@ import { } from "../src/client/media/relay-capability.ts"; describe("browser relay capability", () => { - it("advertises the current one-child browser release capacity", () => { + it("advertises the endpoint capability envelope without a Browser tier", () => { expect(relayCapacityMessageForBrowser(true)).toEqual({ type: "relay-capacity", - downstreamEdges: 1, + downstreamEdges: 3, }); }); diff --git a/tests/server-config.test.ts b/tests/server-config.test.ts index 3e242f7..f6d2b38 100644 --- a/tests/server-config.test.ts +++ b/tests/server-config.test.ts @@ -12,7 +12,7 @@ describe("server configuration", () => { expect(config.stunUrls).toEqual([]); expect(config.maxViewersPerRoom).toBe(8); expect(config.peerAssistedMedia).toBe(false); - expect(config.maxPeerRelayDownstreamEdges).toBe(2); + expect(config.endpointMediaCopyCapacity).toBe(2); expect(config.livekitFallback).toBeUndefined(); expect(config.selectedEdgeTurn).toBeUndefined(); }); @@ -227,28 +227,36 @@ describe("server configuration", () => { ).toThrow("MAX_VIEWERS_PER_ROOM"); }); - it.each(["1", "2"])( - "accepts a bounded peer relay downstream limit of %s", - (maxPeerRelayDownstreamEdges) => { + it.each(["1", "2", "3"])( + "accepts an endpoint media copy capacity of %s", + (endpointMediaCopyCapacity) => { expect( loadConfig({ - MAX_PEER_RELAY_DOWNSTREAM_EDGES: maxPeerRelayDownstreamEdges, - }).maxPeerRelayDownstreamEdges, - ).toBe(Number(maxPeerRelayDownstreamEdges)); + ENDPOINT_MEDIA_COPY_CAPACITY: endpointMediaCopyCapacity, + }).endpointMediaCopyCapacity, + ).toBe(Number(endpointMediaCopyCapacity)); }, ); - it.each(["0", "3", "4", "1.5"])( - "rejects an invalid peer relay downstream limit of %s", - (maxPeerRelayDownstreamEdges) => { + it.each(["0", "4", "1.5"])( + "rejects an invalid endpoint media copy capacity of %s", + (endpointMediaCopyCapacity) => { expect(() => loadConfig({ - MAX_PEER_RELAY_DOWNSTREAM_EDGES: maxPeerRelayDownstreamEdges, + ENDPOINT_MEDIA_COPY_CAPACITY: endpointMediaCopyCapacity, }), - ).toThrow("MAX_PEER_RELAY_DOWNSTREAM_EDGES"); + ).toThrow("ENDPOINT_MEDIA_COPY_CAPACITY"); }, ); + it("fails closed on the removed relay downstream setting", () => { + expect(() => + loadConfig({ MAX_PEER_RELAY_DOWNSTREAM_EDGES: "2" }), + ).toThrow( + "MAX_PEER_RELAY_DOWNSTREAM_EDGES is no longer supported; use ENDPOINT_MEDIA_COPY_CAPACITY", + ); + }); + it("enables the hybrid controller for every room when selected", () => { expect(loadConfig({ PEER_ASSISTED_MEDIA: "true" }).peerAssistedMedia).toBe( true, diff --git a/tests/server-http.test.ts b/tests/server-http.test.ts index 025f941..6ca6aa5 100644 --- a/tests/server-http.test.ts +++ b/tests/server-http.test.ts @@ -43,7 +43,7 @@ function testConfig(overrides: Partial = {}): ServerConfig { maxRooms: 10, maxViewersPerRoom: 8, peerAssistedMedia: false, - maxPeerRelayDownstreamEdges: 2, + endpointMediaCopyCapacity: 2, stunUrls: [], ...overrides, }; diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index a425cd8..4bd4f16 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -5,7 +5,6 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import WebSocket from "ws"; import { - CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT, MAX_PARENT_EDGE_QUALITY_EVIDENCE_BYTES, MAX_SIGNAL_BYTES, MAX_VIEWER_QUALITY_EVIDENCE_BYTES, @@ -156,7 +155,7 @@ function testConfig(): ServerConfig { maxRooms: 10, maxViewersPerRoom: 8, peerAssistedMedia: false, - maxPeerRelayDownstreamEdges: 2, + endpointMediaCopyCapacity: 2, stunUrls: [], }; } @@ -172,7 +171,7 @@ async function startHarness( persistent?: boolean; provisionalHostClaimSeconds?: number; peerAssistedMedia?: boolean; - maxPeerRelayDownstreamEdges?: number; + endpointMediaCopyCapacity?: number; stunUrls?: readonly string[]; now?: () => number; } = {}, @@ -180,8 +179,8 @@ async function startHarness( const config = testConfig(); config.siteAccessPassword = overrides.siteAccessPassword; config.peerAssistedMedia = overrides.peerAssistedMedia ?? false; - config.maxPeerRelayDownstreamEdges = - overrides.maxPeerRelayDownstreamEdges ?? 2; + config.endpointMediaCopyCapacity = + overrides.endpointMediaCopyCapacity ?? 2; config.stunUrls = overrides.stunUrls ?? []; const maxViewersPerRoom = overrides.maxViewersPerRoom ?? 8; config.maxViewersPerRoom = maxViewersPerRoom; @@ -642,7 +641,7 @@ async function startSfuHarness(options: { persistent?: boolean; prepareTimeoutMs?: number; maxRoots?: number; - maxPeerRelayDownstreamEdges?: number; + endpointMediaCopyCapacity?: number; viewerDisconnectGraceMs?: number; stunUrls?: readonly string[]; now?: () => number; @@ -665,8 +664,8 @@ async function startSfuHarness(options: { server: httpServer, roomStore, peerAssistedMedia: true, - maxPeerRelayDownstreamEdges: - options.maxPeerRelayDownstreamEdges ?? 2, + endpointMediaCopyCapacity: + options.endpointMediaCopyCapacity ?? 2, sfuFallback: { url: "wss://sfu.example.test", tokenIssuer: options.tokenIssuer, @@ -1171,16 +1170,8 @@ async function prepareSfuHostParentQualityProbe( ) { active.viewer.socket.send(JSON.stringify({ type: "relay-capacity", - downstreamEdges: 1, + downstreamEdges: 0, })); - const hostFiller = await openClient(harness.webSocketUrl); - await authenticate( - hostFiller, - harness.room, - "viewer", - `${prefix}-filler`, - 0, - ); const parent = await openClient(harness.webSocketUrl); const parentAuth = peerAssisted(await authenticate( parent, @@ -1191,8 +1182,16 @@ async function prepareSfuHostParentQualityProbe( )); expect(parentAuth.routeAssignment.upstream).toEqual({ kind: "peer", - peerId: active.viewerAuth.peerId, + peerId: active.hostAuth.peerId, }); + const hostFiller = await openClient(harness.webSocketUrl); + await authenticate( + hostFiller, + harness.room, + "viewer", + `${prefix}-filler`, + 0, + ); const child = await openClient(harness.webSocketUrl); const childAuth = peerAssisted(await authenticate( child, @@ -1214,7 +1213,7 @@ async function prepareSfuHostParentQualityProbe( ]); await sendTestOffer( - active.viewer, + active.host, parent, parentAuth.peerId, `${prefix}_parent_connection`, @@ -1249,108 +1248,130 @@ async function prepareSfuHostParentQualityProbe( }; } -async function activateTwoFailedViewerSfuRoots( +async function prepareTwoSelectedPeerEdgeCandidates( webSocketUrl: string, room: CreatedRoom, prefix: string, ) { - const first = await activateSingleViewerSfu( - webSocketUrl, + const host = await openClient(webSocketUrl); + const hostAuth = peerAssisted(await authenticate( + host, room, - `${prefix}-first`, - ); + "host", + `${prefix}-host`, + 1, + `${prefix}-share`, + )); + const firstParent = await openClient(webSocketUrl); + const firstParentAuth = peerAssisted(await authenticate( + firstParent, + room, + "viewer", + `${prefix}-first-parent`, + 1, + )); + const secondParent = await openClient(webSocketUrl); + const secondParentAuth = peerAssisted(await authenticate( + secondParent, + room, + "viewer", + `${prefix}-second-parent`, + 1, + )); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const firstViewer = await openClient(webSocketUrl); + const firstAuth = peerAssisted(await authenticate( + firstViewer, + room, + "viewer", + `${prefix}-first-viewer`, + 0, + )); const secondViewer = await openClient(webSocketUrl); - const secondAuth = peerAssisted( - await authenticate( - secondViewer, - room, - "viewer", - `${prefix}-second-viewer`, - 0, - ), - ); + const secondAuth = peerAssisted(await authenticate( + secondViewer, + room, + "viewer", + `${prefix}-second-viewer`, + 0, + )); + expect(firstAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: firstParentAuth.peerId, + }); expect(secondAuth.routeAssignment.upstream).toEqual({ kind: "peer", - peerId: first.hostAuth.peerId, + peerId: secondParentAuth.peerId, }); - - const hostConnectionId = `${prefix}-host-edge`; - first.host.socket.send(JSON.stringify({ - type: "signal", - targetPeerId: secondAuth.peerId, - payload: { - kind: "description", - connectionId: hostConnectionId, - description: { type: "offer", sdp: "v=0\r\n" }, - }, - })); - await secondViewer.inbox.next("signal"); - secondViewer.socket.send(JSON.stringify({ - type: "route-failed", - revision: secondAuth.routeRevision, - phase: "active", - connectionId: hostConnectionId, - })); - const reparented = await nextActiveRouteAfter( - secondViewer, - secondAuth.routeRevision, + const firstCurrent = await nextActiveRouteAfter( + firstViewer, + firstAuth.routeRevision, ); - expect(reparented.assignment.upstream).toEqual({ - kind: "peer", - peerId: first.viewerAuth.peerId, - }); + expect(firstCurrent.revision).toBe(secondAuth.routeRevision); - const peerConnectionId = `${prefix}-peer-edge`; - first.viewer.socket.send(JSON.stringify({ - type: "signal", - targetPeerId: secondAuth.peerId, - payload: { - kind: "description", - connectionId: peerConnectionId, - description: { type: "offer", sdp: "v=0\r\n" }, - }, - })); - await secondViewer.inbox.next("signal"); - secondViewer.socket.send(JSON.stringify({ - type: "route-failed", - revision: reparented.revision, - phase: "active", - connectionId: peerConnectionId, - })); - - const hostPrepare = await nextPreparedRoute(first.host); - const firstRootPrepare = await nextPreparedRoute(first.viewer); - const secondRootPrepare = await nextPreparedRoute(secondViewer); - await Promise.all([ - first.host.inbox.next("sfu-config"), - first.viewer.inbox.next("sfu-config"), - secondViewer.inbox.next("sfu-config"), - ]); - for (const client of [first.host, first.viewer, secondViewer]) { - client.socket.send(JSON.stringify({ - type: "route-ready", - revision: hostPrepare.revision, - phase: "prepare", - })); - } - const [, firstRootActive, secondRootActive] = await Promise.all([ - nextActiveRouteRevision(first.host, hostPrepare.revision), - nextActiveRouteRevision(first.viewer, firstRootPrepare.revision), - nextActiveRouteRevision(secondViewer, secondRootPrepare.revision), - ]); - expect(firstRootActive.assignment.upstream).toEqual({ kind: "sfu" }); - expect(secondRootActive.assignment.upstream).toEqual({ kind: "sfu" }); + const firstConnectionId = `${prefix}-first-edge`; + const secondConnectionId = `${prefix}-second-edge`; + await sendTestOffer( + firstParent, + firstViewer, + firstAuth.peerId, + firstConnectionId, + ); + await sendTestOffer( + secondParent, + secondViewer, + secondAuth.peerId, + secondConnectionId, + ); return { - host: first.host, - hostAuth: first.hostAuth, - failedViewer: secondViewer, - failedAuth: secondAuth, - secondRoot: first.viewer, - secondRootAuth: first.viewerAuth, - revision: hostPrepare.revision, + host, + hostAuth, + firstParent, + firstParentAuth, + firstViewer, + firstAuth, + firstConnectionId, + secondParent, + secondParentAuth, + secondViewer, + secondAuth, + secondConnectionId, + revision: firstCurrent.revision, }; } +function failPeerEdgeCandidate( + viewer: TestClient, + revision: number, + connectionId: string, +): void { + viewer.socket.send(JSON.stringify({ + type: "route-failed", + revision, + phase: "active", + connectionId, + })); +} + +async function startPeerSelectedCandidate( + viewer: TestClient, + parent: TestClient, + revision: number, + connectionId: string, +) { + failPeerEdgeCandidate(viewer, revision, connectionId); + const [viewerGrant, parentGrant] = await Promise.all([ + viewer.inbox.next("selected-edge-turn"), + parent.inbox.next("selected-edge-turn"), + ]); + expect(parentGrant).toEqual(viewerGrant); + if (viewerGrant.edgeKind !== "peer-selected") { + throw new Error("expected a peer-selected grant"); + } + return viewerGrant; +} + async function failActiveSfuRoot(client: TestClient, revision: number) { client.socket.send(JSON.stringify({ type: "refresh-sfu", @@ -2722,10 +2743,10 @@ describe("WebSocket signaling", () => { expect((await host.inbox.next("error")).code).toBe("FORBIDDEN"); }); - it("clamps legacy Viewer capacity across authentication and reconciliation", async () => { + it("applies capacity three to Viewer authentication and reconciliation", async () => { const harness = await startHarness({ peerAssistedMedia: true, - maxPeerRelayDownstreamEdges: 3, + endpointMediaCopyCapacity: 3, }); const host = await openClient(harness.webSocketUrl); const hostAuth = peerAssisted( @@ -2792,13 +2813,16 @@ describe("WebSocket signaling", () => { }); expect(childAuth.routeAssignment.upstream).toEqual({ kind: "peer", - peerId: relayAuth.peerId, + peerId: hostAuth.peerId, }); expect(siblingChildAuth.routeAssignment.upstream).toEqual({ kind: "peer", - peerId: siblingAuth.peerId, + peerId: relayAuth.peerId, + }); + expect(pendingAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: relayAuth.peerId, }); - expect(pendingAuth.routeAssignment.upstream).toEqual({ kind: "none" }); const replacementRelay = await openClient(harness.webSocketUrl); const replacementAuth = peerAssisted( @@ -2813,18 +2837,13 @@ describe("WebSocket signaling", () => { expect(replacementAuth.peerId).toBe(relayAuth.peerId); expect(replacementAuth.routeAssignment).toMatchObject({ upstream: { kind: "peer", peerId: hostAuth.peerId }, - childPeerIds: [childAuth.peerId], + childPeerIds: [siblingChildAuth.peerId, pendingAuth.peerId], }); replacementRelay.socket.send( JSON.stringify({ type: "relay-capacity", downstreamEdges: 3 }), ); - expect(await pending.inbox.next("error")).toMatchObject({ - code: "PEER_NOT_FOUND", - }); - await expect(pending.inbox.next("route-update", 40)).rejects.toThrow( - "Timed out", - ); + await pending.inbox.expectNone(40); await replacementRelay.inbox.expectNone(40); }); @@ -3912,6 +3931,7 @@ describe("WebSocket signaling", () => { let now = 128_000; const harness = await startSfuHarness({ now: () => now, + endpointMediaCopyCapacity: 3, tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; }, }, @@ -3924,8 +3944,20 @@ describe("WebSocket signaling", () => { ); active.viewer.socket.send(JSON.stringify({ type: "relay-capacity", - downstreamEdges: 1, + downstreamEdges: 0, })); + const parent = await openClient(harness.webSocketUrl); + const parentAuth = peerAssisted(await authenticate( + parent, + harness.room, + "viewer", + "quality-host-selected-budget-parent", + 1, + )); + expect(parentAuth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: active.hostAuth.peerId, + }); const hostFiller = await openClient(harness.webSocketUrl); const hostFillerAuth = peerAssisted(await authenticate( hostFiller, @@ -3938,18 +3970,6 @@ describe("WebSocket signaling", () => { kind: "peer", peerId: active.hostAuth.peerId, }); - const parent = await openClient(harness.webSocketUrl); - const parentAuth = peerAssisted(await authenticate( - parent, - harness.room, - "viewer", - "quality-host-selected-budget-parent", - 1, - )); - expect(parentAuth.routeAssignment.upstream).toEqual({ - kind: "peer", - peerId: active.viewerAuth.peerId, - }); const child = await openClient(harness.webSocketUrl); const childAuth = peerAssisted(await authenticate( child, @@ -3997,7 +4017,7 @@ describe("WebSocket signaling", () => { await active.host.inbox.next("signal"); await sendTestOffer( - active.viewer, + active.host, parent, parentAuth.peerId, "quality_host_selected_budget_parent_connection", @@ -4033,7 +4053,7 @@ describe("WebSocket signaling", () => { }); }); - it("uses a capable SFU root as the provisional peer parent", async () => { + it("keeps an SFU-fed Viewer leaf without outbound relay proof", async () => { let now = 128_000; const harness = await startSfuHarness({ now: () => now, @@ -4092,18 +4112,12 @@ describe("WebSocket signaling", () => { await correlateParentEdgeQualityEvidence(parent); now += 2_000; } - const moved = await completePreparedPeerMigration( - child, - active.viewer, - childAuth.peerId, - active.viewerAuth.peerId, - routeRevision, - "quality_sfu_root_probe", + await expect(child.inbox.next("route-update", 40)).rejects.toThrow( + "Timed out", + ); + await expect(active.viewer.inbox.next("signal", 40)).rejects.toThrow( + "Timed out", ); - expect(moved.assignment.upstream).toEqual({ - kind: "peer", - peerId: active.viewerAuth.peerId, - }); }); it("keeps the old relay edge when relative FPS has no alternate peer", async () => { @@ -4176,8 +4190,8 @@ describe("WebSocket signaling", () => { expect(issuedSfuTokens).toBe(0); }); - describe.skipIf(CURRENT_BROWSER_RELAY_DOWNSTREAM_EDGE_LIMIT < 2)( - "future capacity-two browser relay quality coordination", + describe( + "capacity-two relay quality coordination", () => { it("keeps a severe child on its old edge when no peer alternate exists", async () => { let now = 142_000; @@ -4293,11 +4307,11 @@ describe("WebSocket signaling", () => { expect(issuedSfuTokens).toBe(0); }); - it("quarantines a relay only after two children and restores it lazily", async () => { + it("does not drain a relay after two child-scoped quality events", async () => { let now = 145_000; let issuedSfuTokens = 0; const fixture = await startRelayRelativeFpsFixture( - "relay_parent_quarantine", + "relay_parent_child_scoped_quality", () => now, { alternateRelayCapacity: 0, @@ -4321,7 +4335,7 @@ describe("WebSocket signaling", () => { fixture.parent, secondChild, fixture.secondChildAuth!.peerId, - "relay_parent_quarantine_still_active", + "relay_parent_child_scoped_still_active", )).toMatchObject({ fromPeerId: fixture.parentAuth.peerId, }); @@ -4338,27 +4352,17 @@ describe("WebSocket signaling", () => { pending, fixture.harness.room, "viewer", - "relay-parent-quarantine-pending", + "relay-parent-child-scoped-pending", 0, ), ); - expect(pendingAuth.routeAssignment.upstream).toEqual({ kind: "none" }); - - now += 30_001; - fixture.parent.socket.send( - JSON.stringify({ type: "relay-capacity", downstreamEdges: 2 }), - ); - const restored = await nextActiveRouteAfter( - pending, - pendingAuth.routeRevision, - ); - expect(restored.assignment.upstream).toEqual({ + expect(pendingAuth.routeAssignment.upstream).toEqual({ kind: "peer", peerId: fixture.parentAuth.peerId, }); }); - it("keeps room cooldown when quarantine has no alternate", async () => { + it("serializes child-scoped migrations through the room cooldown", async () => { let now = 180_000; let issuedSfuTokens = 0; const fixture = await startRelayRelativeFpsFixture( @@ -5987,7 +5991,7 @@ describe("WebSocket signaling", () => { return `token-${peerId}`; }, }, - maxPeerRelayDownstreamEdges: 1, + endpointMediaCopyCapacity: 1, }); const host = await openClient(harness.webSocketUrl); const hostAuth = peerAssisted( @@ -6054,6 +6058,71 @@ describe("WebSocket signaling", () => { expect(failedPrepare.assignment.upstream).toEqual({ kind: "sfu" }); }); + it("rejects a fourth Host copy before issuing an SFU token at capacity three", async () => { + let issuedSfuTokens = 0; + const harness = await startSfuHarness({ + tokenIssuer: { + async issueToken({ peerId }) { + issuedSfuTokens += 1; + return `token-${peerId}`; + }, + }, + endpointMediaCopyCapacity: 3, + }); + const host = await openClient(harness.webSocketUrl); + const hostAuth = peerAssisted( + await authenticate(host, harness.room, "host", "capacity-three-host"), + ); + const viewers: Array<{ + client: TestClient; + auth: ReturnType; + }> = []; + for (let index = 0; index < 3; index += 1) { + const client = await openClient(harness.webSocketUrl); + const auth = peerAssisted( + await authenticate( + client, + harness.room, + "viewer", + `capacity-three-viewer-${index}`, + 0, + ), + ); + expect(auth.routeAssignment.upstream).toEqual({ + kind: "peer", + peerId: hostAuth.peerId, + }); + viewers.push({ client, auth }); + } + + const failed = viewers[2]!; + const connectionId = "capacity_three_third_edge"; + await sendTestOffer( + host, + failed.client, + failed.auth.peerId, + connectionId, + ); + failed.client.socket.send(JSON.stringify({ + type: "route-failed", + revision: failed.auth.routeRevision, + phase: "active", + connectionId, + })); + + expect(await failed.client.inbox.next("error")).toMatchObject({ + code: "PEER_NOT_FOUND", + message: "Endpoint media copy transition capacity is exhausted", + }); + expect(issuedSfuTokens).toBe(0); + await expect(failed.client.inbox.next("sfu-config", 40)).rejects.toThrow( + "Timed out", + ); + await expect(failed.client.inbox.next("route-update", 40)).rejects.toThrow( + "Timed out", + ); + }); + it("SFU root invariant gate: retains a zero-child root across commit and reauthentication", async () => { const issued: Array< Parameters[0] @@ -6808,6 +6877,7 @@ describe("WebSocket signaling", () => { let now = 205_000; const harness = await startSfuHarness({ now: () => now, + endpointMediaCopyCapacity: 3, tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; }, }, @@ -6860,6 +6930,7 @@ describe("WebSocket signaling", () => { let now = 208_000; const harness = await startSfuHarness({ now: () => now, + endpointMediaCopyCapacity: 3, tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; }, }, @@ -6921,6 +6992,7 @@ describe("WebSocket signaling", () => { let now = 210_000; const harness = await startSfuHarness({ now: () => now, + endpointMediaCopyCapacity: 3, tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; }, }, @@ -7015,7 +7087,7 @@ describe("WebSocket signaling", () => { await expect(active.host.inbox.next("route-update", 40)).rejects.toThrow("Timed out"); }); - it("keeps a browser parent to one downstream edge during overlap", async () => { + it("keeps an SFU-fed Viewer leaf despite advertised relay capacity", async () => { const harness = await startSfuHarness({ tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, }); @@ -7049,19 +7121,12 @@ describe("WebSocket signaling", () => { "healthy-relay-budget-root-child", 0, )); - expect(rootChildAuth.routeAssignment.upstream).toEqual({ - kind: "peer", - peerId: active.viewerAuth.peerId, - }); + expect(rootChildAuth.routeAssignment.upstream).toEqual({ kind: "none" }); const rootActive = await nextActiveRouteRevision( active.viewer, rootChildAuth.routeRevision, ); - expect(rootActive.assignment.childPeerIds).toEqual([rootChildAuth.peerId]); - requestHealthyReselection(active.viewer, rootActive.revision); - await expect(active.viewer.inbox.next("route-update", 40)).rejects.toThrow( - "Timed out", - ); + expect(rootActive.assignment.childPeerIds).toEqual([]); }); it("cleans a healthy probe across Viewer session replacement and stop", async () => { @@ -7249,95 +7314,93 @@ describe("WebSocket signaling", () => { it("admits only one pending peer-selected TURN attempt per room", async () => { const harness = await startSfuHarness({ - tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, + tokenIssuer: { async issueToken() { throw new Error("offline"); } }, selectedEdgeTurn: true, }); - const peers = await activateTwoFailedViewerSfuRoots( + const peers = await prepareTwoSelectedPeerEdgeCandidates( harness.webSocketUrl, harness.room, "selected-room-pending-cap", ); - const firstGrant = await startPeerSelectedTurn( - peers.failedViewer, - peers.secondRoot, + const firstGrant = await startPeerSelectedCandidate( + peers.firstViewer, + peers.firstParent, peers.revision, + peers.firstConnectionId, ); - await failActiveSfuRoot(peers.secondRoot, peers.revision); + failPeerEdgeCandidate( + peers.secondViewer, + firstGrant.revision, + peers.secondConnectionId, + ); await expect( - nextActiveRouteAfter(peers.secondRoot, peers.revision), - ).resolves.toMatchObject({ - assignment: { upstream: { kind: "peer" } }, - }); - const carry = await peers.secondRoot.inbox.next("selected-edge-turn"); - expect(carry).toMatchObject({ - parentPeerId: firstGrant.parentPeerId, - viewerPeerId: firstGrant.viewerPeerId, - oldConnectionId: firstGrant.oldConnectionId, - newConnectionId: firstGrant.newConnectionId, - }); + peers.secondViewer.inbox.next("selected-edge-turn", 40), + ).rejects.toThrow("Timed out"); await expect( - peers.secondRoot.inbox.next("selected-edge-turn", 40), + peers.secondParent.inbox.next("selected-edge-turn", 40), ).rejects.toThrow("Timed out"); }); it("keeps an answered peer-selected TURN edge inside the room cap", async () => { const harness = await startSfuHarness({ - tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, + tokenIssuer: { async issueToken() { throw new Error("offline"); } }, selectedEdgeTurn: true, }); - const peers = await activateTwoFailedViewerSfuRoots( + const peers = await prepareTwoSelectedPeerEdgeCandidates( harness.webSocketUrl, harness.room, "selected-room-answered-cap", ); - const grant = await startPeerSelectedTurn( - peers.failedViewer, - peers.secondRoot, + const grant = await startPeerSelectedCandidate( + peers.firstViewer, + peers.firstParent, peers.revision, + peers.firstConnectionId, ); - peers.secondRoot.socket.send(JSON.stringify({ + peers.firstParent.socket.send(JSON.stringify({ type: "signal", - targetPeerId: peers.failedAuth.peerId, + targetPeerId: peers.firstAuth.peerId, payload: { kind: "description", connectionId: grant.newConnectionId, description: { type: "offer", sdp: "v=0\r\n" }, }, })); - await peers.failedViewer.inbox.next("signal"); - peers.failedViewer.socket.send(JSON.stringify({ + await peers.firstViewer.inbox.next("signal"); + peers.firstViewer.socket.send(JSON.stringify({ type: "signal", - targetPeerId: peers.secondRootAuth.peerId, + targetPeerId: peers.firstParentAuth.peerId, payload: { kind: "description", connectionId: grant.newConnectionId, description: { type: "answer", sdp: "v=0\r\n" }, }, })); - await peers.secondRoot.inbox.next("signal"); + await peers.firstParent.inbox.next("signal"); - await failActiveSfuRoot(peers.secondRoot, peers.revision); - const routeAfterFailure = await nextActiveRouteAfter( - peers.secondRoot, - peers.revision, + failPeerEdgeCandidate( + peers.secondViewer, + grant.revision, + peers.secondConnectionId, ); - expect(routeAfterFailure).toMatchObject({ - assignment: { upstream: { kind: "peer" } }, - }); - const carry = await peers.secondRoot.inbox.next("selected-edge-turn"); - expect(carry).toMatchObject({ - revision: routeAfterFailure.revision, - parentPeerId: grant.parentPeerId, - viewerPeerId: grant.viewerPeerId, - oldConnectionId: grant.oldConnectionId, - newConnectionId: grant.newConnectionId, - }); await expect( - peers.secondRoot.inbox.next("selected-edge-turn", 40), + peers.secondViewer.inbox.next("selected-edge-turn", 40), ).rejects.toThrow("Timed out"); + peers.firstParent.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: peers.firstAuth.peerId, + payload: { + kind: "candidate", + connectionId: grant.newConnectionId, + candidate: null, + }, + })); + await expect(peers.firstViewer.inbox.next("signal")).resolves.toMatchObject({ + payload: { connectionId: grant.newConnectionId }, + }); }); it("revokes a Host selected overlay before another child exceeds cap two", async () => { @@ -7420,141 +7483,142 @@ describe("WebSocket signaling", () => { it("revokes a selected lease when its parent withdraws relay capacity", async () => { const harness = await startSfuHarness({ - tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, + tokenIssuer: { async issueToken() { throw new Error("offline"); } }, selectedEdgeTurn: true, }); - const peers = await activateTwoFailedViewerSfuRoots( + const peers = await prepareTwoSelectedPeerEdgeCandidates( harness.webSocketUrl, harness.room, "selected-parent-capacity-withdrawal", ); - const firstGrant = await startPeerSelectedTurn( - peers.failedViewer, - peers.secondRoot, + const firstGrant = await startPeerSelectedCandidate( + peers.firstViewer, + peers.firstParent, peers.revision, + peers.firstConnectionId, ); - peers.secondRoot.socket.send(JSON.stringify({ + peers.firstParent.socket.send(JSON.stringify({ type: "relay-capacity", downstreamEdges: 0, })); const [viewerAuthority, parentAuthority] = await Promise.all([ - peers.failedViewer.inbox.next("route-update"), - peers.secondRoot.inbox.next("route-update"), + nextActiveRouteRevision(peers.firstViewer, firstGrant.revision), + nextActiveRouteRevision(peers.firstParent, firstGrant.revision), ]); expect(viewerAuthority).toMatchObject({ - revision: peers.revision, + revision: firstGrant.revision, phase: "active", }); expect(parentAuthority).toMatchObject({ - revision: peers.revision, + revision: firstGrant.revision, phase: "active", }); - peers.secondRoot.socket.send(JSON.stringify({ - type: "signal", - targetPeerId: peers.failedAuth.peerId, - payload: { - kind: "candidate", - connectionId: firstGrant.newConnectionId, - candidate: null, - }, - })); - expect((await peers.secondRoot.inbox.next("error")).code).toBe("FORBIDDEN"); - - await failActiveSfuRoot(peers.secondRoot, peers.revision); - const [secondGrant, hostGrant] = await Promise.all([ - peers.secondRoot.inbox.next("selected-edge-turn"), - peers.host.inbox.next("selected-edge-turn"), - ]); - expect(hostGrant).toEqual(secondGrant); + const secondGrant = await startPeerSelectedCandidate( + peers.secondViewer, + peers.secondParent, + firstGrant.revision, + peers.secondConnectionId, + ); expect(secondGrant).toMatchObject({ edgeKind: "peer-selected", - parentPeerId: peers.hostAuth.peerId, - viewerPeerId: peers.secondRootAuth.peerId, + parentPeerId: peers.secondParentAuth.peerId, + viewerPeerId: peers.secondAuth.peerId, }); }); it("releases the room cap after the selected edge fails", async () => { const harness = await startSfuHarness({ - tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, + tokenIssuer: { async issueToken() { throw new Error("offline"); } }, selectedEdgeTurn: true, }); - const peers = await activateTwoFailedViewerSfuRoots( + const peers = await prepareTwoSelectedPeerEdgeCandidates( harness.webSocketUrl, harness.room, "selected-room-failure-release", ); - const firstGrant = await startPeerSelectedTurn( - peers.failedViewer, - peers.secondRoot, + const firstGrant = await startPeerSelectedCandidate( + peers.firstViewer, + peers.firstParent, peers.revision, + peers.firstConnectionId, ); - peers.failedViewer.socket.send(JSON.stringify({ + peers.firstViewer.socket.send(JSON.stringify({ type: "route-failed", - revision: peers.revision, + revision: firstGrant.revision, phase: "active", connectionId: firstGrant.newConnectionId, })); - await peers.failedViewer.inbox.next("error"); + await peers.firstViewer.inbox.next("error"); - await failActiveSfuRoot(peers.secondRoot, peers.revision); - const [secondGrant, hostGrant] = await Promise.all([ - peers.secondRoot.inbox.next("selected-edge-turn"), - peers.host.inbox.next("selected-edge-turn"), - ]); - expect(hostGrant).toEqual(secondGrant); + const secondGrant = await startPeerSelectedCandidate( + peers.secondViewer, + peers.secondParent, + firstGrant.revision, + peers.secondConnectionId, + ); expect(secondGrant).toMatchObject({ edgeKind: "peer-selected", - viewerPeerId: peers.secondRootAuth.peerId, + viewerPeerId: peers.secondAuth.peerId, }); }); it("releases the room cap when the selected Viewer disconnects", async () => { const harness = await startSfuHarness({ - tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, + tokenIssuer: { async issueToken() { throw new Error("offline"); } }, selectedEdgeTurn: true, viewerDisconnectGraceMs: 500, }); - const peers = await activateTwoFailedViewerSfuRoots( + const peers = await prepareTwoSelectedPeerEdgeCandidates( harness.webSocketUrl, harness.room, "selected-room-disconnect-release", ); - await startPeerSelectedTurn( - peers.failedViewer, - peers.secondRoot, + const firstGrant = await startPeerSelectedCandidate( + peers.firstViewer, + peers.firstParent, peers.revision, + peers.firstConnectionId, ); - await closeClient(peers.failedViewer); + await closeClient(peers.firstViewer); + peers.firstParent.socket.send(JSON.stringify({ + type: "relay-capacity", + downstreamEdges: 0, + })); - await failActiveSfuRoot(peers.secondRoot, peers.revision); - const [secondGrant, hostGrant] = await Promise.all([ - peers.secondRoot.inbox.next("selected-edge-turn"), - peers.host.inbox.next("selected-edge-turn"), - ]); - expect(hostGrant).toEqual(secondGrant); + const secondCurrent = await nextActiveRouteAfter( + peers.secondViewer, + firstGrant.revision, + ); + const secondGrant = await startPeerSelectedCandidate( + peers.secondViewer, + peers.secondParent, + secondCurrent.revision, + peers.secondConnectionId, + ); expect(secondGrant).toMatchObject({ edgeKind: "peer-selected", - viewerPeerId: peers.secondRootAuth.peerId, + viewerPeerId: peers.secondAuth.peerId, }); }); it("releases a selected edge when its Viewer session is replaced", async () => { const harness = await startSfuHarness({ - tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, + tokenIssuer: { async issueToken() { throw new Error("offline"); } }, selectedEdgeTurn: true, }); - const peers = await activateTwoFailedViewerSfuRoots( + const peers = await prepareTwoSelectedPeerEdgeCandidates( harness.webSocketUrl, harness.room, "selected-viewer-session-replacement", ); - const firstGrant = await startPeerSelectedTurn( - peers.failedViewer, - peers.secondRoot, + const firstGrant = await startPeerSelectedCandidate( + peers.firstViewer, + peers.firstParent, peers.revision, + peers.firstConnectionId, ); const replacement = await openClient(harness.webSocketUrl); @@ -7563,35 +7627,26 @@ describe("WebSocket signaling", () => { replacement, harness.room, "viewer", - "selected-viewer-session-replacement-second-viewer", + "selected-viewer-session-replacement-first-viewer", + 0, ), ); - expect(replacementAuth.peerId).toBe(peers.failedAuth.peerId); - await expect(peers.secondRoot.inbox.next("route-update")).resolves.toMatchObject({ - revision: replacementAuth.routeRevision, - phase: "active", - }); - peers.secondRoot.socket.send(JSON.stringify({ - type: "signal", - targetPeerId: replacementAuth.peerId, - payload: { - kind: "candidate", - connectionId: firstGrant.newConnectionId, - candidate: null, - }, - })); - expect((await peers.secondRoot.inbox.next("error")).code).toBe("FORBIDDEN"); + expect(replacementAuth.peerId).toBe(peers.firstAuth.peerId); + await nextActiveRouteRevision( + peers.firstParent, + replacementAuth.routeRevision, + ); - await failActiveSfuRoot(peers.secondRoot, replacementAuth.routeRevision); - const [secondGrant, hostGrant] = await Promise.all([ - peers.secondRoot.inbox.next("selected-edge-turn"), - peers.host.inbox.next("selected-edge-turn"), - ]); - expect(hostGrant).toEqual(secondGrant); + const secondGrant = await startPeerSelectedCandidate( + peers.secondViewer, + peers.secondParent, + replacementAuth.routeRevision, + peers.secondConnectionId, + ); if (secondGrant.edgeKind !== "peer-selected") { throw new Error("Expected a selected peer edge grant"); } - expect(secondGrant.viewerPeerId).toBe(peers.secondRootAuth.peerId); + expect(secondGrant.viewerPeerId).toBe(peers.secondAuth.peerId); expect(secondGrant.newConnectionId).not.toBe(firstGrant.newConnectionId); }); @@ -9677,7 +9732,7 @@ describe("WebSocket signaling", () => { oldClient.socket.send( JSON.stringify({ type: "authenticate", - protocol: "screener-v3", + protocol: "screener-v5", roomId: "999999999999", role: "viewer", clientId: "old-client",