From 120371bf05f0d9ab846b0d884fda8cca06826799 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sat, 22 Aug 2026 23:12:40 +0800 Subject: [PATCH 1/2] docs(route): bind native sender to authenticated endpoint cap --- docs/adr/0005-automatic-hybrid-media-routing.md | 14 ++++++++++++++ docs/project-memory.md | 2 +- docs/status.md | 4 ++-- docs/todo.md | 2 +- docs/方案设计.md | 4 ++-- docs/需求理解.md | 4 ++-- 6 files changed, 22 insertions(+), 8 deletions(-) diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index bcb8a0d..f92a19d 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -35,6 +35,20 @@ capacity: - one named configuration value and one shared implementation boundary own the policy. Route code must not repeat literal policy numbers. +The server returns the final deployment value as +`authenticated.endpointMediaCopyCapacity` on every successful signaling +authentication. Web and Native clients validate `1..3`; a Native Host uses that +same value for ordinary Viewer admission, authoritative peer-assisted child +assignments, and physical sender slots. Room `maxViewers` is a separate +participant-admission limit and never supplies a sender budget. A missing or +invalid capacity fails before route authority is accepted. + +Without the peer-assisted controller, signaling still admits at most that many +active Host children per room. Excess admitted Viewers wait without an +authorized media edge and are promoted in stable join order when a slot is +released; offer, answer, candidate, and restart routing is limited to the active +set. + Capacity counts active outbound media copies produced by a non-server endpoint: an ordinary peer child consumes one slot and the Host's single SFU publication consumes one Host slot. An upstream receive edge is free. A committed selected diff --git a/docs/project-memory.md b/docs/project-memory.md index a0f9854..e4468ee 100644 --- a/docs/project-memory.md +++ b/docs/project-memory.md @@ -21,7 +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. +- Current source implements the uniform `1/2/3` endpoint cap through one shared accounting guard, 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. Release readiness requires the v6 authenticated snapshot to carry the exact deployment cap so Native admission, peer-assisted assignments, and sender slots use the same authority. - 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/status.md b/docs/status.md index 871e7a2..0b9088f 100644 --- a/docs/status.md +++ b/docs/status.md @@ -14,8 +14,8 @@ 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 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. +- Current source 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 v6 release boundary rejects v5 before room authority and must carry the exact deployment cap in every authenticated snapshot so Native ordinary admission, peer-assisted child assignments, and sender slots share the server authority. It 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. 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. diff --git a/docs/todo.md b/docs/todo.md index 4d1cafb..1896ec6 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -6,7 +6,7 @@ Only items in **Now** are executable after their stated decision gate. A branch ## Now -1. **Route wave A: endpoint capacity and accounting.** From the latest canonical `main`, atomically bump the incompatible signaling wire and replace the Browser/role tiers with `ENDPOINT_MEDIA_COPY_CAPACITY=1|2|3`, default `2`. A peer child or Host publication consumes one steady sender slot, upstream receive is free, committed selected TURN replaces the same logical edge transport, and a media-producing candidate needs the one bounded overlap reservation. Reject an old client before room authority, reject a fourth Host copy when `C=3`, and keep parent-wide quality drain fail-closed until its accepted evidence owner exists. Do not merge the old configurable-cap worktree. +1. **Complete route wave A's v6 endpoint-cap boundary.** Add the exact deployment `endpointMediaCopyCapacity` to every authenticated snapshot; bound ordinary server-authorized Host children and their signaling to that value with stable waiting promotion; and make Native ordinary admission, peer-assisted child assignment, and sender-slot allocation consume it independently of room `maxViewers`. Strictly reject missing or out-of-range values, retain the shared accounting and transition guard, and validate `C=1/2/3` before release. Do not merge the old configurable-cap worktree. 2. **Route waves B-D: bounded fallback, selection, and controller convergence.** Replace the temporary fixed SFU subscriber guard with explicit deployment-wide SFU ingress/egress admission; then replace the temporary room-wide selected-lease guard with per-edge TURN allocation admission and exact accounting. Converge allocation, child reparent, and relay drain on one committed graph/controller. Parent candidates use hard eligibility filters followed by deterministic shallowest, remaining-capacity, join-order, and peer-identity ordering; standard ICE plus exact media proof tests one candidate at a time. Do not add raw-IP/NAT ranking, all-pairs probes, a weighted score, or periodic rebalancing. The temporary guards are safety boundaries, not accepted product limits, and must not be removed early. 3. **Switch screen-audio ceilings live.** Keep Opus fixed and serialize 64/128/256 kbps desired profile updates across every current Host, relay, and SFU audio sender. Each endpoint uses fresh `getParameters()`/`setParameters()` readback, ignores stale completions, retains media plus its prior applied value on failure, and applies the latest desired value to future edges. Do not add a remote applied-ack protocol or claim room-wide atomic convergence in this slice. 4. **Switch video codec while paused.** After the route/media-generation boundary is available, serialize the change with route mutation, freeze exact current route/media bindings, hold new joins, renegotiate P2P/relay and republish SFU while paused, then verify actual codec plus decoded frames on resume. A binding invalidated by leave or hard route failure exits this proof set and waits for routing to rebuild from the final committed preference after the bounded codec transaction settles. On failure pause again and perform bounded rollback; rollback proves the prior preference was reapplied plus a newly negotiated actual codec and decoded progress, not literal equality with the old actual codec. Zero targets update only the future-edge preference. Reset `automatic` with empty codec preferences. This is a paused bounded-gap transaction; unpaused make-before-break is a later extension. diff --git a/docs/方案设计.md b/docs/方案设计.md index a7e7f15..44113e3 100644 --- a/docs/方案设计.md +++ b/docs/方案设计.md @@ -108,7 +108,7 @@ Web Host 在 `authenticate` 中显式发送 `viewerPresence: true` capability， ## 信令协议 -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。 +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 或当前尝试的房间密码。成功的 `authenticated` 快照始终携带部署最终值 `endpointMediaCopyCapacity: 1|2|3`；普通模式的服务端 admission 只向 Host 激活稳定加入顺序中的至多该数量 Viewer，并在离开后提升等待者，双方信令只允许 active child。所有客户端严格校验该值，Native Host 也用它约束普通 admission、peer-assisted child assignment 与实际发送槽位，房间 `maxViewers` 只约束观看者总 admission。服务端设置 64 KiB `maxPayload`、关闭 `perMessageDeflate`，并以 ping/pong 清理失效连接。单进程固定限制 2048 条总连接和 256 条未鉴权连接，容量在 upgrade 前预留，超限直接返回 HTTP 503 并销毁 socket。 主要消息： @@ -374,7 +374,7 @@ Token-free SFU standby prewarm 保持可选且不创建 participant、publicatio Parent selection 使用硬过滤加字典序，不建立综合质量评分器。先拒绝 stale session、不可 source-reachable、成环、超深、无 sender/overlap slot、relay-ineligible、位于 failed-edge cooldown 或 server resource 未准入的候选；对剩余候选依次比较结果深度、分配后剩余稳态 sender slots、稳定加入顺序与 peer identity。controller 每次只 reserve/prepare 一个候选，使用该 `RTCPeerConnection` 自带的 ICE checklist/connectivity checks，再以 current-generation RTP、live track 与 decoded frame 验证；初次加入或 hard failure 采用达到 media-usable floor 的首个候选，soft-quality reparent 还需 quality owner 证明候选达到 recovery threshold 且优于仍健康旧 edge。失败释放 reservation 后才取下一个。应用不做 all-pairs ping、raw-IP/NAT/地理排名、周期探测或全房重平衡。一个 exact edge 的近期失败只进入该 edge 的有界 exclusion/cooldown，不污染其他路径。 -所有非服务器端点共享一个 server-authoritative steady outbound media-copy capacity：唯一配置 `ENDPOINT_MEDIA_COPY_CAPACITY` 默认 `2`，只接受 `1`、`2` 或 `3`，不兼容旧键。peer child 或 Host publication 各占一份，upstream receive 免费，committed selected TURN 只替换同一份 edge transport；SFU subscriber egress 与 TURN allocation 使用独立 server admission。客户端 capability 只能收紧该值，不能按角色、浏览器、UA、设备或页面可见性建立另一套 policy。代码通过一个 shared accounting guard 执行，不在 Host、Viewer、active、recovery 或 migration 分支重复 magic number；client/server/native wire 随该能力原子升级。首个迁移波次在独立 server admission 就绪前保留现有 SFU subscriber 与 room-wide selected-lease 安全护栏，但不把其固定值当作产品 policy；`C=3` 满载时必须在发出会产生第四份 Host media copy 的 token/grant 前明确等待或失败。 +所有非服务器端点共享一个 server-authoritative steady outbound media-copy capacity：唯一配置 `ENDPOINT_MEDIA_COPY_CAPACITY` 默认 `2`，只接受 `1`、`2` 或 `3`，不兼容旧键。peer child 或 Host publication 各占一份，upstream receive 免费，committed selected TURN 只替换同一份 edge transport；SFU subscriber egress 与 TURN allocation 使用独立 server admission。客户端 capability 只能收紧该值，不能按角色、浏览器、UA、设备或页面可见性建立另一套 policy。服务端通过 v6 `authenticated.endpointMediaCopyCapacity` 把最终值交给 Web 与 Native；Native 不从 `maxViewers` 推导发送预算，普通 admission、peer-assisted child assignment 和槽位分配共用该值。代码通过一个 shared accounting guard 执行，不在 Host、Viewer、active、recovery 或 migration 分支重复 magic number；client/server/native wire 随该能力原子升级。首个迁移波次在独立 server admission 就绪前保留现有 SFU subscriber 与 room-wide selected-lease 安全护栏，但不把其固定值当作产品 policy；`C=3` 满载时必须在发出会产生第四份 Host media copy 的 token/grant 前明确等待或失败。 普通 Web relay 把收到的 remote track 加入独立 downstream `RTCPeerConnection`，因此可能逐跳解码和重编码；浏览器不保证跨连接 shared encode。每条 ordinary child 是独立网络副本，实际 encode time、CPU/GPU、bitrate、loss、frame rate、depth 和 recovery 必须测量。 diff --git a/docs/需求理解.md b/docs/需求理解.md index e22c443..423d6a0 100644 --- a/docs/需求理解.md +++ b/docs/需求理解.md @@ -24,7 +24,7 @@ SFU/TURN 的 publication、subscription、transport、server-resource accounting - 观看端本地音量：播放控制栏提供 `0%` 至 `100%` 音量条和静音按钮，只改变当前浏览器的媒体元素，不修改 Host 音轨、编码、房间状态或其他观看者。切流、换父和 ICE 恢复保留当前页面内的值；刷新后不持久化。静音保留上次非零音量，拖动到非零值自动取消静音。 - 本地昵称与房间成员：用户可设置一个只保存在当前浏览器同源 `localStorage` 的显示名，界面统一称为“昵称”；默认以只读文本显示，铅笔编辑按钮紧贴昵称，点击后才进入编辑。Viewer 顶部标题直接显示“{Host 名称} 的屏幕”，不得在下方重复一行 Host 名称。信令只在当前在线 socket 会话内保存和广播 NFC 规范化、最多 24 个 Unicode code points、拒绝控制/换行/双向覆盖符与孤立 surrogate 的纯文本标签，不写 SQLite 或日志。空值使用通用角色 fallback。Web Host 与 Viewer 都必须看到全部当前在线 Viewer 的权威总数与 roster，不能拿最多两条 Host 媒体直连数冒充在线人数；presence 更新也不得创建媒体连接。展示身份默认只显示名称；仅当房内出现同名时追加当前房间 `peerId` 的至少六字符短后缀，并只在同名项后缀碰撞时有界延长。显示名、后缀和 `clientId` 都不是账号、授权身份或全局 UUID，重复名称允许且不得参与授权、路由或质量决策。Web Host 与 Viewer 通过同一 opt-in participant roster 看到 Host 名称；Native 使用同一版本化鉴权 wire，但不发送 presence capability，也不接收 Web roster。 - 一对多观看：房间 admission 与端点下游容量分别配置，超额加入必须明确拒绝。Production 当前默认 8、显式接受 1 至 16；未来默认值和上限是独立产品决策，必须经过相应资源、质量、深度和恢复验收，不能从端点 cap 或一次短跑推导。 -- 端点发送容量：所有非服务器端点使用同一个服务端权威稳态 outbound media-copy capacity，由唯一部署项 `ENDPOINT_MEDIA_COPY_CAPACITY` 配置，默认 `2` 且只接受 `1`、`2` 或 `3`；不读取旧键或保留第二套 tier。peer child 或 Host 的唯一 SFU publication 各消耗一份，上游接收不消耗该预算，committed selected TURN 只替换同一份 logical edge 的 transport。客户端只能声明不高于部署值的可用容量，角色、浏览器、UA、设备类型和页面可见性不产生不同上限。并行产生媒体的 candidate 必须持有唯一的 `TransitionOverlapSlot`，并在 commit 后回到稳态上限；SFU subscriber egress 与 TURN allocation 由独立 server admission 记账。该不兼容能力变化与单一 signaling wire 原子升级；旧客户端在取得房间或 route authority 前 fail closed。 +- 端点发送容量：所有非服务器端点使用同一个服务端权威稳态 outbound media-copy capacity，由唯一部署项 `ENDPOINT_MEDIA_COPY_CAPACITY` 配置，默认 `2` 且只接受 `1`、`2` 或 `3`；不读取旧键或保留第二套 tier。peer child 或 Host 的唯一 SFU publication 各消耗一份，上游接收不消耗该预算，committed selected TURN 只替换同一份 logical edge 的 transport。客户端只能声明不高于部署值的可用容量，角色、浏览器、UA、设备类型和页面可见性不产生不同上限。服务端在每次 `authenticated` 权威快照中返回最终 `endpointMediaCopyCapacity`；普通非 peer-assisted 模式也只授权稳定加入顺序中的至多 `C` 个 Host child，超额 Viewer 保持入房等待并在槽位释放后提升，信令不能绕过该集合。Web 与 Native 严格校验该值，Native Host 的普通 Viewer admission、peer-assisted child assignment 和实际发送槽位都使用同一值，且不得从独立的房间 `maxViewers` 推导。并行产生媒体的 candidate 必须持有唯一的 `TransitionOverlapSlot`，并在 commit 后回到稳态上限；SFU subscriber egress 与 TURN allocation 由独立 server admission 记账。该不兼容能力变化与单一 signaling wire 原子升级；缺失或非法 cap 与旧客户端都在取得房间或 route authority 前 fail closed。 - P2P 优先：每条普通逻辑媒体边先尝试 direct/peer UDP，服务器不得默认中转可直连媒体。更多观看者优先沿满足统一端点容量的无环 peer 图分布；每名 Viewer 同时至多一个 active upstream，失败时只影响需要重选的分支。 - NAT 与传输边界：STUN 必须可用，所有 ordinary peer `RTCPeerConnection` 只接收 STUN ICE servers，普通边不得预发 TURN candidate。所有恢复和候选授权必须绑定 exact room/share generation、endpoint session、route/assignment generation 与 connection identity；成功即停止，失败、超时或绑定失效即撤销，全部可用路径耗尽后明确等待或失败。已接受的顺序是保持既有 logical edge，必要时 direct/STUN 后使用 exact selected TURN；只有 logical ingress 不可用时才使用当前 Host publication 的 SFU subscription。Host 无 direct first-level path 时优先一条可经 TURN 的 Host-SFU publication，而不是逐 Viewer 建 Host-TURN peer 边。凭据不得进入 URL、日志、浏览器持久化或 SQLite；HTTPS/WSS 继续独立使用 TLS/TCP。 - 质量控制：保留 1080p60（8 Mbps）、1080p30（5 Mbps）、720p30（3 Mbps）三档推荐上限，三档与高级设置初始值统一采用 `balanced`（平衡）与 `automatic`（自动编码），`maintain-resolution`（清晰优先）、`maintain-framerate`（流畅优先）和 H.264/VP8 保留为显式选择。分享期间可直接换视频档位和 64/128/256 kbps 音频 sender ceiling，不重新选择来源或重建健康连接；视频 codec 首个可交付切换边界是在分享已显式暂停时执行有界重协商，未暂停热切换是后续复用同一事务的扩展。Peer-assisted relay 与已配置的 SFU 必须接收同一个严格校验的房间级权威设置，覆盖当前和以后建立的 sender；普通 P2P 信令结构保持不变。任何设置都只是浏览器上限和偏好，具体降级由浏览器决定；必须读回实际 sender 参数并通过 stats 观测结果、显示参数拒绝、改写和持续受限原因。不做自定义动态 FPS、综合评分或周期性全房间调档。 @@ -130,7 +130,7 @@ SFU/TURN 的 publication、subscription、transport、server-resource accounting ### 一对多与带宽保护 -- Given `ENDPOINT_MEDIA_COPY_CAPACITY=C`，其中 `C` 只能是 `1`、`2` 或 `3` 且默认 `2`，When Host 或任一 Viewer 建立、恢复、重鉴权或迁移媒体路径，Then peer child 与 Host publication 的稳态实际发送副本总数不超过 `C`，upstream receive 不计数，committed selected TURN 不重复计数，角色、浏览器和 UA 不改变 `C`；并行候选只有持有 `TransitionOverlapSlot` 时才可达到 `min(C + 1, 3)`，并在 commit/abort 后释放。When 此能力合同上线，Then client/server/native wire 同一 release 原子升级，旧 wire 在房间授权前得到明确 mismatch，且 `C=3` 满载 Host 在取得第四份 publication/token/grant 前明确等待或失败。 +- Given `ENDPOINT_MEDIA_COPY_CAPACITY=C`，其中 `C` 只能是 `1`、`2` 或 `3` 且默认 `2`，When Host 或任一 Viewer 建立、恢复、重鉴权或迁移媒体路径，Then peer child 与 Host publication 的稳态实际发送副本总数不超过 `C`，upstream receive 不计数，committed selected TURN 不重复计数，角色、浏览器和 UA 不改变 `C`；每个 `authenticated` 快照都携带同一个 `C`，Native Host 不用 `maxViewers` 或本地常量代替它，并行候选只有持有 `TransitionOverlapSlot` 时才可达到 `min(C + 1, 3)`，并在 commit/abort 后释放。When 此能力合同上线，Then client/server/native wire 同一 release 原子升级，缺失或越界的 `C` 与旧 wire 都在房间授权前得到明确 mismatch，且 `C=3` 满载 Host 在取得第四份 publication/token/grant 前明确等待或失败。 - Given peer-assisted 路径分别运行 1、3、5、8 名观看者，When 用户加入、容量释放、节点离开、重鉴权或当前 edge 失效，Then 服务端只为新 Viewer、等待 Viewer 或失去 parent 的 subtree root 选择一个 active parent；每名 Viewer 同时至多一个 active upstream，活动图无环，其余健康边保持 sticky。没有可用 parent 时明确等待或拒绝，不主动重平衡或引入常驻多源媒体。 - Given 一名分享者和八名观看者，When 在受控网络中运行已记录的质量档位，Then 记录每条边的拓扑深度、候选类型、码率、编码/解码负载、分享端与 relay 上行、首帧和玻璃到玻璃延迟；该条在真实矩阵完成前保持未验收。 - Given candidate 改变房间 admission 默认值或上限，When 在目标数量与代表性路径上运行规定时长并注入 relay departure、重连和一个弱路径，Then 每名观看者持续解码、每个非服务器端点不超过配置容量、深度与故障半径有界、未受影响 subtree 不迁移，并记录端点与中央服务的 CPU、GPU、内存、上行、出口、帧率、丢包率、冻结和恢复；任一门槛失败时不得切换默认值。 From ca1d715319815162bf8b167b7f5b788f582ea549 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sun, 23 Aug 2026 00:01:38 +0800 Subject: [PATCH 2/2] fix(route): enforce authenticated endpoint capacity --- native/sender/internal/app/server_test.go | 4 +- native/sender/internal/remote/admission.go | 55 ++-- .../sender/internal/remote/admission_test.go | 93 ++++--- native/sender/internal/remote/session.go | 127 ++++++--- native/sender/internal/remote/session_test.go | 14 +- native/sender/internal/remote/wire.go | 105 +++++--- native/sender/internal/remote/wire_test.go | 23 +- src/client/pages/HostPage.tsx | 6 + src/server/room-store.ts | 7 + src/server/signaling.ts | 120 ++++++++- src/shared/protocol.ts | 11 + tests/client-state.test.ts | 3 + tests/protocol.test.ts | 45 +++- tests/server-signal.test.ts | 254 +++++++++++++++++- 14 files changed, 702 insertions(+), 165 deletions(-) diff --git a/native/sender/internal/app/server_test.go b/native/sender/internal/app/server_test.go index 2504389..e016275 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-v6","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,"endpointMediaCopyCapacity":2,"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-v6","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,"endpointMediaCopyCapacity":2,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`)) }) return server } diff --git a/native/sender/internal/remote/admission.go b/native/sender/internal/remote/admission.go index beb6644..90684ee 100644 --- a/native/sender/internal/remote/admission.go +++ b/native/sender/internal/remote/admission.go @@ -17,8 +17,8 @@ type viewerAdmission struct { waitSet map[string]struct{} } -func newViewerAdmission(maxViewers int) *viewerAdmission { - edgeLimit := maxHostEdges +func newViewerAdmission(maxViewers, endpointMediaCopyCapacity int) *viewerAdmission { + edgeLimit := endpointMediaCopyCapacity if maxViewers < edgeLimit { edgeLimit = maxViewers } @@ -30,33 +30,53 @@ func newViewerAdmission(maxViewers int) *viewerAdmission { } } -func (admission *viewerAdmission) Join(peerID string) admissionAction { +func (admission *viewerAdmission) Activate(peerID string) admissionAction { if _, exists := admission.active[peerID]; exists { return admissionDuplicate } + _, wasWaiting := admission.waitSet[peerID] + if !wasWaiting && len(admission.active)+len(admission.waiting) >= admission.roomLimit { + return admissionFull + } + if len(admission.active) >= admission.edgeLimit { + return admissionFull + } + if wasWaiting { + admission.removeWaiting(peerID) + } + admission.active[peerID] = struct{}{} + return admissionActivate +} + +func (admission *viewerAdmission) Wait(peerID string) admissionAction { if _, exists := admission.waitSet[peerID]; exists { return admissionDuplicate } - if len(admission.active) < admission.edgeLimit { - admission.active[peerID] = struct{}{} - return admissionActivate - } - if len(admission.active)+len(admission.waiting) >= admission.roomLimit { + _, wasActive := admission.active[peerID] + if !wasActive && len(admission.active)+len(admission.waiting) >= admission.roomLimit { return admissionFull } + if wasActive { + delete(admission.active, peerID) + } admission.waiting = append(admission.waiting, peerID) admission.waitSet[peerID] = struct{}{} return admissionWait } -func (admission *viewerAdmission) Leave(peerID string) (wasActive bool, promoted string) { +func (admission *viewerAdmission) Leave(peerID string) (wasActive bool) { if _, exists := admission.active[peerID]; exists { delete(admission.active, peerID) - return true, admission.promoteNext() + return true } if _, exists := admission.waitSet[peerID]; !exists { - return false, "" + return false } + admission.removeWaiting(peerID) + return false +} + +func (admission *viewerAdmission) removeWaiting(peerID string) { delete(admission.waitSet, peerID) for index, waitingID := range admission.waiting { if waitingID != peerID { @@ -67,7 +87,6 @@ func (admission *viewerAdmission) Leave(peerID string) (wasActive bool, promoted admission.waiting = admission.waiting[:len(admission.waiting)-1] break } - return false, "" } func (admission *viewerAdmission) AbortActivation(peerID string) { @@ -82,15 +101,3 @@ func (admission *viewerAdmission) IsActive(peerID string) bool { _, exists := admission.active[peerID] return exists } - -func (admission *viewerAdmission) promoteNext() string { - if len(admission.waiting) == 0 { - return "" - } - promoted := admission.waiting[0] - admission.waiting[0] = "" - admission.waiting = admission.waiting[1:] - delete(admission.waitSet, promoted) - admission.active[promoted] = struct{}{} - return promoted -} diff --git a/native/sender/internal/remote/admission_test.go b/native/sender/internal/remote/admission_test.go index 30c8763..e21d604 100644 --- a/native/sender/internal/remote/admission_test.go +++ b/native/sender/internal/remote/admission_test.go @@ -2,41 +2,44 @@ package remote import "testing" -func TestViewerAdmissionNeverExceedsTwoEdgesAndPromotesInJoinOrder(t *testing.T) { - admission := newViewerAdmission(4) - assertAdmission(t, admission, "viewer-a", admissionActivate) - assertAdmission(t, admission, "viewer-b", admissionActivate) - assertAdmission(t, admission, "viewer-c", admissionWait) - assertAdmission(t, admission, "viewer-d", admissionWait) - assertAdmission(t, admission, "viewer-c", admissionDuplicate) - assertAdmission(t, admission, "viewer-e", admissionFull) +func TestViewerAdmissionMirrorsAuthoritativeActiveAndWaitingSets(t *testing.T) { + admission := newViewerAdmission(4, 2) + assertActiveAdmission(t, admission, "viewer-a", admissionActivate) + assertActiveAdmission(t, admission, "viewer-b", admissionActivate) + assertWaitingAdmission(t, admission, "viewer-c", admissionWait) + assertWaitingAdmission(t, admission, "viewer-d", admissionWait) + assertWaitingAdmission(t, admission, "viewer-c", admissionDuplicate) + assertWaitingAdmission(t, admission, "viewer-e", admissionFull) active, waiting := admission.Counts() if active != 2 || waiting != 2 { t.Fatalf("counts = %d active, %d waiting", active, waiting) } - wasActive, promoted := admission.Leave("viewer-b") - if !wasActive || promoted != "viewer-c" { - t.Fatalf("leave promoted %q from active=%t", promoted, wasActive) + if !admission.Leave("viewer-b") { + t.Fatal("active leave was not recorded") } - if admission.IsActive("viewer-b") || !admission.IsActive("viewer-c") { - t.Fatal("promotion did not atomically replace the freed edge") + if admission.IsActive("viewer-b") || admission.IsActive("viewer-c") { + t.Fatal("leave promoted a Viewer without server authority") } active, waiting = admission.Counts() + if active != 1 || waiting != 2 { + t.Fatalf("post-leave counts = %d active, %d waiting", active, waiting) + } + assertActiveAdmission(t, admission, "viewer-c", admissionActivate) + active, waiting = admission.Counts() if active != 2 || waiting != 1 { - t.Fatalf("post-promotion counts = %d active, %d waiting", active, waiting) + t.Fatalf("server promotion counts = %d active, %d waiting", active, waiting) } } func TestViewerAdmissionRemovesAWaitingViewerWithoutPromotion(t *testing.T) { - admission := newViewerAdmission(3) - assertAdmission(t, admission, "viewer-a", admissionActivate) - assertAdmission(t, admission, "viewer-b", admissionActivate) - assertAdmission(t, admission, "viewer-c", admissionWait) + admission := newViewerAdmission(3, 2) + assertActiveAdmission(t, admission, "viewer-a", admissionActivate) + assertActiveAdmission(t, admission, "viewer-b", admissionActivate) + assertWaitingAdmission(t, admission, "viewer-c", admissionWait) - wasActive, promoted := admission.Leave("viewer-c") - if wasActive || promoted != "" { - t.Fatalf("waiting leave = active %t, promotion %q", wasActive, promoted) + if admission.Leave("viewer-c") { + t.Fatal("waiting leave was reported as active") } active, waiting := admission.Counts() if active != 2 || waiting != 0 { @@ -45,9 +48,9 @@ func TestViewerAdmissionRemovesAWaitingViewerWithoutPromotion(t *testing.T) { } func TestViewerAdmissionUsesTheSmallerRoomLimit(t *testing.T) { - admission := newViewerAdmission(1) - assertAdmission(t, admission, "viewer-a", admissionActivate) - assertAdmission(t, admission, "viewer-b", admissionFull) + admission := newViewerAdmission(1, 3) + assertActiveAdmission(t, admission, "viewer-a", admissionActivate) + assertWaitingAdmission(t, admission, "viewer-b", admissionFull) active, waiting := admission.Counts() if active != 1 || waiting != 0 { t.Fatalf("counts = %d active, %d waiting", active, waiting) @@ -55,17 +58,15 @@ func TestViewerAdmissionUsesTheSmallerRoomLimit(t *testing.T) { } func TestViewerAdmissionClearsFailedActivation(t *testing.T) { - admission := newViewerAdmission(4) - assertAdmission(t, admission, "viewer-a", admissionActivate) - assertAdmission(t, admission, "viewer-b", admissionActivate) - assertAdmission(t, admission, "viewer-c", admissionWait) - assertAdmission(t, admission, "viewer-d", admissionWait) + admission := newViewerAdmission(4, 2) + assertActiveAdmission(t, admission, "viewer-a", admissionActivate) + assertActiveAdmission(t, admission, "viewer-b", admissionActivate) + assertWaitingAdmission(t, admission, "viewer-c", admissionWait) + assertWaitingAdmission(t, admission, "viewer-d", admissionWait) - _, promoted := admission.Leave("viewer-a") - if promoted != "viewer-c" { - t.Fatalf("first promotion = %q", promoted) - } - admission.AbortActivation(promoted) + admission.Leave("viewer-a") + assertActiveAdmission(t, admission, "viewer-c", admissionActivate) + admission.AbortActivation("viewer-c") if admission.IsActive("viewer-c") || admission.IsActive("viewer-d") { t.Fatal("failed activation left a viewer in the active set") } @@ -75,9 +76,25 @@ func TestViewerAdmissionClearsFailedActivation(t *testing.T) { } } -func assertAdmission(t *testing.T, admission *viewerAdmission, peerID string, want admissionAction) { - t.Helper() - if got := admission.Join(peerID); got != want { - t.Fatalf("join %q = %d, want %d", peerID, got, want) +func TestViewerAdmissionNeverActivatesAWaitingViewerIntoAFreeSlot(t *testing.T) { + admission := newViewerAdmission(4, 2) + assertWaitingAdmission(t, admission, "viewer-a", admissionWait) + active, waiting := admission.Counts() + if active != 0 || waiting != 1 || admission.IsActive("viewer-a") { + t.Fatalf("waiting authority = %d active, %d waiting", active, waiting) + } +} + +func assertActiveAdmission(t *testing.T, admission *viewerAdmission, peerID string, want admissionAction) { + t.Helper() + if got := admission.Activate(peerID); got != want { + t.Fatalf("activate %q = %d, want %d", peerID, got, want) + } +} + +func assertWaitingAdmission(t *testing.T, admission *viewerAdmission, peerID string, want admissionAction) { + t.Helper() + if got := admission.Wait(peerID); got != want { + t.Fatalf("wait %q = %d, want %d", peerID, got, want) } } diff --git a/native/sender/internal/remote/session.go b/native/sender/internal/remote/session.go index 03946e3..6ca6d35 100644 --- a/native/sender/internal/remote/session.go +++ b/native/sender/internal/remote/session.go @@ -24,7 +24,6 @@ import ( ) const ( - maxHostEdges = 3 maxRemoteResponse = 64 << 10 remoteRequestTimeout = 10 * time.Second signalWriteTimeout = 5 * time.Second @@ -77,28 +76,29 @@ type Session struct { shareGeneration string onEvent func(Event) - mu sync.Mutex - writeMu sync.Mutex - conn *websocket.Conn - ice iceConfig - peers map[string]*peer - admission *viewerAdmission - peerAssisted bool - routeRevision int64 - routePhase string - routeFailures [2]string - fanout *media.Fanout - audioFanout *media.AudioFanout - codec media.Codec - peerAPI *webrtc.API - codecPreferences []webrtc.RTPCodecParameters - closeOnce sync.Once - fatalOnce sync.Once - terminalOnce sync.Once - readDone chan struct{} - terminalResult chan error - readerStarted atomic.Bool - closing atomic.Bool + mu sync.Mutex + writeMu sync.Mutex + conn *websocket.Conn + ice iceConfig + peers map[string]*peer + admission *viewerAdmission + endpointMediaCopyCapacity int + peerAssisted bool + routeRevision int64 + routePhase string + routeFailures [2]string + fanout *media.Fanout + audioFanout *media.AudioFanout + codec media.Codec + peerAPI *webrtc.API + codecPreferences []webrtc.RTPCodecParameters + closeOnce sync.Once + fatalOnce sync.Once + terminalOnce sync.Once + readDone chan struct{} + terminalResult chan error + readerStarted atomic.Bool + closing atomic.Bool } type createRoomResponse struct { @@ -338,12 +338,13 @@ func (session *Session) connect(ctx context.Context) error { } session.mu.Lock() session.ice = message.IceConfig + session.endpointMediaCopyCapacity = message.EndpointMediaCopyCapacity session.peerAssisted = message.PeerAssisted if message.PeerAssisted { session.routeRevision = message.RouteRevision session.routePhase = "active" } else { - session.admission = newViewerAdmission(message.MaxViewers) + session.admission = newViewerAdmission(message.MaxViewers, message.EndpointMediaCopyCapacity) } session.mu.Unlock() if message.PeerAssisted { @@ -415,6 +416,11 @@ func (session *Session) handle(message serverMessage) error { return nil } return session.addViewer(message.PeerID) + case "peer-waiting": + if session.peerAssisted { + return nil + } + return session.waitViewer(message.PeerID) case "peer-left": if session.peerAssisted { return nil @@ -474,16 +480,15 @@ func (session *Session) addViewer(peerID string) error { session.mu.Unlock() return errors.New("viewer joined before host authentication completed") } - action := session.admission.Join(peerID) + action := session.admission.Activate(peerID) session.mu.Unlock() switch action { case admissionDuplicate: return nil case admissionWait: - session.emitViewerCounts() - return nil + return errors.New("active Viewer was assigned to the waiting set") case admissionFull: - return errors.New("viewer presence exceeded the authenticated room capacity") + return errors.New("active Viewer exceeds authenticated endpoint or room capacity") case admissionActivate: if err := session.activatePeer(peerID); err != nil { session.mu.Lock() @@ -498,6 +503,29 @@ func (session *Session) addViewer(peerID string) error { } } +func (session *Session) waitViewer(peerID string) error { + session.mu.Lock() + if session.admission == nil { + session.mu.Unlock() + return errors.New("waiting Viewer arrived before host authentication completed") + } + wasActive := session.admission.IsActive(peerID) + action := session.admission.Wait(peerID) + peer := session.peers[peerID] + if wasActive { + delete(session.peers, peerID) + } + session.mu.Unlock() + if peer != nil && wasActive { + peer.close() + } + if action == admissionFull { + return errors.New("waiting Viewer exceeds authenticated room capacity") + } + session.emitViewerCounts() + return nil +} + func (session *Session) activatePeer(peerID string) error { session.mu.Lock() configuration, err := pionConfiguration(session.ice) @@ -537,19 +565,11 @@ func (session *Session) removeViewer(peerID string) error { } return nil } - _, promoted := session.admission.Leave(peerID) + session.admission.Leave(peerID) session.mu.Unlock() if peer != nil { peer.close() } - if promoted != "" { - if err := session.activatePeer(promoted); err != nil { - session.mu.Lock() - session.admission.AbortActivation(promoted) - session.mu.Unlock() - return fmt.Errorf("promote waiting viewer: %w", err) - } - } session.emitViewerCounts() return nil } @@ -583,6 +603,9 @@ func (session *Session) rebuildPeer(peerID string) error { } func (session *Session) reconcileAuthoritativeChildren(peerIDs []string) error { + if len(peerIDs) > session.endpointMediaCopyCapacity { + return errors.New("authoritative children exceed the authenticated endpoint capacity") + } desired := make(map[string]struct{}, len(peerIDs)) for _, peerID := range peerIDs { desired[peerID] = struct{}{} @@ -622,12 +645,22 @@ func (session *Session) handleRouteUpdate(message serverMessage) error { if message.RouteRevision < session.routeRevision { return nil } + if !hostRouteAssignmentFitsCapacity( + message.RouteAssignment, + session.endpointMediaCopyCapacity, + message.RoutePhase == "prepare", + ) { + return errors.New("route update exceeds the authenticated endpoint capacity") + } session.routeRevision = message.RouteRevision session.routePhase = message.RoutePhase publicationGeneration, _ := decodeNullableOpaqueID(message.RouteAssignment.SFUPublicationGeneration) if publicationGeneration != "" { return session.failUnsupportedRoute(message.RouteRevision, message.RoutePhase, nil) } + if message.RoutePhase == "prepare" && !session.hasExactChildren(message.RouteAssignment.ChildPeerIDs) { + return session.failUnsupportedRoute(message.RouteRevision, message.RoutePhase, nil) + } if message.RoutePhase == "active" { if err := session.reconcileAuthoritativeChildren(message.RouteAssignment.ChildPeerIDs); err != nil { return err @@ -636,6 +669,20 @@ func (session *Session) handleRouteUpdate(message serverMessage) error { return session.send(map[string]any{"type": "route-ready", "revision": message.RouteRevision, "phase": message.RoutePhase}) } +func (session *Session) hasExactChildren(peerIDs []string) bool { + session.mu.Lock() + defer session.mu.Unlock() + if len(session.peers) != len(peerIDs) { + return false + } + for _, peerID := range peerIDs { + if session.peers[peerID] == nil { + return false + } + } + return true +} + func (session *Session) handleUnsupportedRouteMessage(message serverMessage, connectionID *string) error { if !session.peerAssisted || message.RouteRevision != session.routeRevision { return nil @@ -754,13 +801,13 @@ func (session *Session) diagnosticsLoop() { } func (session *Session) availableSlotLocked() int { - used := [maxHostEdges + 1]bool{} + used := [maxEndpointMediaCopyCapacity + 1]bool{} for _, activePeer := range session.peers { - if activePeer.slot > 0 && activePeer.slot <= maxHostEdges { + if activePeer.slot > 0 && activePeer.slot <= session.endpointMediaCopyCapacity { used[activePeer.slot] = true } } - for slot := 1; slot <= maxHostEdges; slot++ { + for slot := 1; slot <= session.endpointMediaCopyCapacity; slot++ { if !used[slot] { return slot } diff --git a/native/sender/internal/remote/session_test.go b/native/sender/internal/remote/session_test.go index 34cb3d2..4f5cc32 100644 --- a/native/sender/internal/remote/session_test.go +++ b/native/sender/internal/remote/session_test.go @@ -92,7 +92,7 @@ func TestStartRetriesTheSameProvisionalRoomOnce(t *testing.T) { expiresAt := time.Now().UTC().Add(24 * time.Hour).Truncate(time.Second) authenticated, _ := json.Marshal(map[string]any{ "type": "authenticated", "protocol": signalingProtocol, "role": "host", "peerId": "host-peer", - "roomExpiresAt": expiresAt.Format(time.RFC3339), "maxViewers": 3, "hostOnline": true, + "roomExpiresAt": expiresAt.Format(time.RFC3339), "maxViewers": 3, "endpointMediaCopyCapacity": 2, "hostOnline": true, "connectionId": nil, "viewerPeerIds": []string{}, "iceConfig": map[string]any{"iceServers": []any{}}, "viewerPolicy": privateViewerPolicy, "viewerAuthorizationGeneration": "viewer_generation_12345678", @@ -387,7 +387,8 @@ func TestPeerAssistedHostUsesOnlyAuthoritativeRoutes(t *testing.T) { defer cancel() session := &Session{ signalCtx: ctx, conn: connection, peerAssisted: true, - peers: make(map[string]*peer), routeRevision: 7, routePhase: "active", + peers: make(map[string]*peer), endpointMediaCopyCapacity: 2, + routeRevision: 7, routePhase: "active", } if err = session.handle(serverMessage{Type: "peer-joined", PeerID: "viewer-new"}); err != nil || len(session.peers) != 0 { @@ -407,6 +408,13 @@ func TestPeerAssistedHostUsesOnlyAuthoritativeRoutes(t *testing.T) { case <-time.After(50 * time.Millisecond): } + unsupportedPrepare := direct + unsupportedPrepare.ChildPeerIDs = []string{"viewer-new"} + if err = session.handle(serverMessage{Type: "route-update", RouteRevision: 8, RoutePhase: "prepare", RouteAssignment: unsupportedPrepare}); err != nil { + t.Fatal(err) + } + assertRouteStatus(t, receiveRouteStatus(t, received), "route-failed", 8, "prepare", nil) + if err = session.handle(serverMessage{Type: "route-update", RouteRevision: 8, RoutePhase: "active", RouteAssignment: direct}); err != nil { t.Fatal(err) } @@ -745,7 +753,7 @@ func newRemoteHarnessWithPassword(t *testing.T, roster []string, password string harness.authentication <- append([]byte(nil), payload...) authenticated, _ := json.Marshal(map[string]any{ "type": "authenticated", "protocol": signalingProtocol, "role": "host", "peerId": "host-peer", - "roomExpiresAt": nil, "maxViewers": 3, "hostOnline": true, + "roomExpiresAt": nil, "maxViewers": 3, "endpointMediaCopyCapacity": 2, "hostOnline": true, "connectionId": nil, "viewerPeerIds": roster, "iceConfig": map[string]any{"iceServers": []any{}}, "viewerPolicy": privateViewerPolicy, "viewerAuthorizationGeneration": "viewer_generation_12345678", diff --git a/native/sender/internal/remote/wire.go b/native/sender/internal/remote/wire.go index 5492911..ef6242b 100644 --- a/native/sender/internal/remote/wire.go +++ b/native/sender/internal/remote/wire.go @@ -16,11 +16,12 @@ import ( ) const ( - signalingProtocol = "screener-v6" - maxProtocolViewers = 16 - maxRouteRevision = int64(1<<53 - 1) - maxSignalSDPBytes = 48 << 10 - maxQualityEvidence = 2 << 10 + signalingProtocol = "screener-v6" + maxProtocolViewers = 16 + maxEndpointMediaCopyCapacity = 3 + maxRouteRevision = int64(1<<53 - 1) + maxSignalSDPBytes = 48 << 10 + maxQualityEvidence = 2 << 10 ) var opaqueIDPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{8,128}$`) @@ -91,27 +92,28 @@ type participantRouteAssignment struct { } type serverMessage struct { - Type string - Role string - PeerID string - FromPeerID string - ViewerPeerIDs []string - ConnectionID string - Rebuild bool - MaxViewers int - IceConfig iceConfig - ViewerPolicy string - Payload inboundSignalPayload - Code string - Message string - Reason string - PeerAssisted bool - MediaAssignment mediaAssignment - RouteAssignment participantRouteAssignment - RouteRevision int64 - RoutePhase string - EdgeKind string - NewConnectionID string + Type string + Role string + PeerID string + FromPeerID string + ViewerPeerIDs []string + ConnectionID string + Rebuild bool + MaxViewers int + EndpointMediaCopyCapacity int + IceConfig iceConfig + ViewerPolicy string + Payload inboundSignalPayload + Code string + Message string + Reason string + PeerAssisted bool + MediaAssignment mediaAssignment + RouteAssignment participantRouteAssignment + RouteRevision int64 + RoutePhase string + EdgeKind string + NewConnectionID string } func readServerMessage(ctx context.Context, conn *websocket.Conn) (serverMessage, error) { @@ -135,7 +137,7 @@ func decodeServerMessage(payload []byte) (serverMessage, error) { switch discriminator.Type { case "authenticated": return decodeAuthenticatedMessage(payload) - case "peer-joined", "peer-left": + case "peer-joined", "peer-waiting", "peer-left": var wire struct { Type string `json:"type"` PeerID string `json:"peerId"` @@ -327,6 +329,7 @@ func decodeAuthenticatedMessage(payload []byte) (serverMessage, error) { PeerID string `json:"peerId"` RoomExpiresAt json.RawMessage `json:"roomExpiresAt"` MaxViewers *int `json:"maxViewers"` + EndpointMediaCopyCapacity *int `json:"endpointMediaCopyCapacity"` HostOnline *bool `json:"hostOnline"` ConnectionID json.RawMessage `json:"connectionId"` ViewerPeerIDs []string `json:"viewerPeerIds"` @@ -343,7 +346,9 @@ func decodeAuthenticatedMessage(payload []byte) (serverMessage, error) { if err := decodeStrict(payload, &wire); err != nil || wire.Type != "authenticated" || wire.Protocol != signalingProtocol || wire.Role != "host" || !validOpaqueID(wire.PeerID) || wire.MaxViewers == nil || - *wire.MaxViewers < 1 || *wire.MaxViewers > maxProtocolViewers || wire.HostOnline == nil || + *wire.MaxViewers < 1 || *wire.MaxViewers > maxProtocolViewers || + wire.EndpointMediaCopyCapacity == nil || *wire.EndpointMediaCopyCapacity < 1 || + *wire.EndpointMediaCopyCapacity > maxEndpointMediaCopyCapacity || wire.HostOnline == nil || !*wire.HostOnline || !bytes.Equal(bytes.TrimSpace(wire.ConnectionID), []byte("null")) || wire.ViewerPeerIDs == nil || len(wire.ViewerPeerIDs) > *wire.MaxViewers || (wire.ViewerPolicy != "private-link" && wire.ViewerPolicy != "public-watch") || @@ -382,23 +387,27 @@ func decodeAuthenticatedMessage(payload []byte) (serverMessage, error) { if err != nil || strings.Join(assignment.ChildPeerIDs, "\x00") != strings.Join(routeAssignment.ChildPeerIDs, "\x00") { return serverMessage{}, errors.New("host authentication response is invalid") } + if !hostRouteAssignmentFitsCapacity(routeAssignment, *wire.EndpointMediaCopyCapacity, false) { + return serverMessage{}, errors.New("host authentication response is invalid") + } if wire.RouteRevision == nil || *wire.RouteRevision < 0 || *wire.RouteRevision > maxRouteRevision || len(wire.QualitySettings) == 0 { return serverMessage{}, errors.New("host authentication response is invalid") } routeRevision = *wire.RouteRevision } return serverMessage{ - Type: wire.Type, - Role: wire.Role, - PeerID: wire.PeerID, - MaxViewers: *wire.MaxViewers, - ViewerPeerIDs: append([]string(nil), wire.ViewerPeerIDs...), - IceConfig: config, - ViewerPolicy: wire.ViewerPolicy, - PeerAssisted: peerAssisted, - MediaAssignment: assignment, - RouteAssignment: routeAssignment, - RouteRevision: routeRevision, + Type: wire.Type, + Role: wire.Role, + PeerID: wire.PeerID, + MaxViewers: *wire.MaxViewers, + EndpointMediaCopyCapacity: *wire.EndpointMediaCopyCapacity, + ViewerPeerIDs: append([]string(nil), wire.ViewerPeerIDs...), + IceConfig: config, + ViewerPolicy: wire.ViewerPolicy, + PeerAssisted: peerAssisted, + MediaAssignment: assignment, + RouteAssignment: routeAssignment, + RouteRevision: routeRevision, }, nil } @@ -433,7 +442,7 @@ func decodeNullableOpaqueID(payload []byte) (string, error) { } func validChildPeerIDs(peerIDs []string) bool { - if peerIDs == nil || len(peerIDs) > maxHostEdges { + if peerIDs == nil || len(peerIDs) > maxEndpointMediaCopyCapacity { return false } seen := make(map[string]struct{}, len(peerIDs)) @@ -449,6 +458,22 @@ func validChildPeerIDs(peerIDs []string) bool { return true } +func hostRouteAssignmentFitsCapacity(assignment participantRouteAssignment, capacity int, transition bool) bool { + limit := capacity + if transition && limit < maxEndpointMediaCopyCapacity { + limit++ + } + publicationGeneration, err := decodeNullableOpaqueID(assignment.SFUPublicationGeneration) + if err != nil { + return false + } + count := len(assignment.ChildPeerIDs) + if publicationGeneration != "" { + count++ + } + return count <= limit +} + func decodeInboundSignalPayload(payload []byte) (inboundSignalPayload, error) { var discriminator struct { Kind string `json:"kind"` diff --git a/native/sender/internal/remote/wire_test.go b/native/sender/internal/remote/wire_test.go index 87efc06..a0c3d2f 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-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 validHostAuthenticated = `{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"endpointMediaCopyCapacity":2,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}` -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"}}` +const validPeerAssistedHostAuthenticated = `{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"endpointMediaCopyCapacity":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" @@ -58,7 +58,7 @@ func TestDecodeOrdinaryHostAuthentication(t *testing.T) { if err != nil { t.Fatal(err) } - if message.MaxViewers != 3 || message.Type != "authenticated" || message.Role != "host" || + if message.MaxViewers != 3 || message.EndpointMediaCopyCapacity != 2 || message.Type != "authenticated" || message.Role != "host" || message.ViewerPolicy != "private-link" { t.Fatalf("authentication = %+v", message) } @@ -138,13 +138,16 @@ 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"] = "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 }, - "peer-mode-extra": func(message map[string]any) { message["mediaMode"] = "peer-assisted" }, - "wrong-role": func(message map[string]any) { message["role"] = "viewer" }, + "missing-protocol": func(message map[string]any) { delete(message, "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 }, + "missing-endpoint-cap": func(message map[string]any) { delete(message, "endpointMediaCopyCapacity") }, + "zero-endpoint-cap": func(message map[string]any) { message["endpointMediaCopyCapacity"] = 0 }, + "too-large-endpoint-cap": func(message map[string]any) { message["endpointMediaCopyCapacity"] = 4 }, + "peer-mode-extra": func(message map[string]any) { message["mediaMode"] = "peer-assisted" }, + "wrong-role": func(message map[string]any) { message["role"] = "viewer" }, "legacy-ice-shape": func(message map[string]any) { message["iceConfig"].(map[string]any)["expiresAt"] = nil }, diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 00b6a73..97c10ad 100644 --- a/src/client/pages/HostPage.tsx +++ b/src/client/pages/HostPage.tsx @@ -1414,6 +1414,12 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) { }); return; } + if (message.type === "peer-waiting") { + if (!peerAssistedRef.current) { + removePeer(message.peerId); + } + return; + } if (message.type === "peer-left") { if (peerAssistedRef.current) { return; diff --git a/src/server/room-store.ts b/src/server/room-store.ts index a71bed4..26ca58d 100644 --- a/src/server/room-store.ts +++ b/src/server/room-store.ts @@ -563,6 +563,13 @@ export class RoomStore { return viewer?.sessionId ? { peerId, sessionId: viewer.sessionId } : undefined; } + getViewerPeerIds(roomId: string): string[] { + const room = this.rooms.get(roomId); + return room + ? [...room.viewers.values()].map((viewer) => viewer.peerId) + : []; + } + getConnectedViewers(roomId: string): ConnectedPeer[] { const room = this.rooms.get(roomId); if (!room) { diff --git a/src/server/signaling.ts b/src/server/signaling.ts index a3295e2..77c85c3 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -21,6 +21,7 @@ import { type ServerMessage, type ParticipantPresenceEntry, } from "../shared/protocol.js"; +import { assertEndpointMediaCopyCapacity } from "../shared/media-copy-accounting.js"; import { RoomStore, RoomStoreError, @@ -123,6 +124,10 @@ export class SignalingServer { private readonly shareGenerationsByRoom = new Map(); private readonly pausedShareGenerationsByRoom = new Map(); private readonly deferredViewerPresenceRooms = new Set(); + private readonly ordinaryActiveHostChildrenByRoom = new Map< + string, + Set + >(); private readonly hybridMediaRouter?: HybridMediaRouter; private readonly now: () => number; private readonly authenticationTimeoutMs: number; @@ -136,6 +141,7 @@ export class SignalingServer { private closing = false; constructor(private readonly options: SignalingOptions) { + assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity); this.now = options.now ?? Date.now; this.authenticationTimeoutMs = options.authenticationTimeoutMs ?? 5_000; this.viewerDisconnectGraceMs = options.viewerDisconnectGraceMs ?? 5_000; @@ -255,6 +261,7 @@ export class SignalingServer { this.viewerMediaReadyByRoom.clear(); this.shareGenerationsByRoom.clear(); this.pausedShareGenerationsByRoom.clear(); + this.ordinaryActiveHostChildrenByRoom.clear(); this.options.server.off("upgrade", this.upgradeHandler); for (const socket of this.webSocketServer.clients) { @@ -531,6 +538,9 @@ export class SignalingServer { participant.role === "host" ? this.options.roomStore.getConnectedViewers(participant.roomId) : []; + if (!this.isHybridMediaEnabled()) { + this.fillOrdinaryHostChildren(participant.roomId); + } const authenticatedMessage = { type: "authenticated" as const, @@ -539,6 +549,7 @@ export class SignalingServer { peerId: participant.peerId, roomExpiresAt: participant.expiresAt, maxViewers: this.options.roomStore.maxViewersPerRoom, + endpointMediaCopyCapacity: this.options.endpointMediaCopyCapacity, hostOnline: participant.hostOnline, ...(participant.role === "viewer" ? { @@ -549,7 +560,10 @@ export class SignalingServer { } : {}), connectionId, - viewerPeerIds: [...participant.viewerPeerIds], + viewerPeerIds: + participant.role === "host" && !this.isHybridMediaEnabled() + ? [...this.ordinaryHostChildPeerIds(participant.roomId)] + : [...participant.viewerPeerIds], iceConfig: this.iceConfig(), viewerPolicy: participant.viewerPolicy, viewerAuthorizationGeneration: @@ -606,8 +620,18 @@ export class SignalingServer { this.pausedShareGenerationsByRoom.get(participant.roomId) === shareGeneration, }); - if (!this.isHybridMediaEnabled()) { - this.send(socket, { type: "peer-joined", peerId: viewer.peerId }); + if ( + !this.isHybridMediaEnabled() && + !this.isOrdinaryHostChild(participant.roomId, viewer.peerId) + ) { + this.send(socket, { type: "peer-waiting", peerId: viewer.peerId }); + } + } + if (!this.isHybridMediaEnabled()) { + for (const viewer of connectedViewers) { + if (this.isOrdinaryHostChild(participant.roomId, viewer.peerId)) { + this.send(socket, { type: "peer-joined", peerId: viewer.peerId }); + } } } return; @@ -625,7 +649,9 @@ export class SignalingServer { // event idempotently, which closes races where its local peer was rebuilt // while the viewer retained the same server-side connection ID. this.sendToSession(host.sessionId, { - type: "peer-joined", + type: this.isOrdinaryHostChild(participant.roomId, participant.peerId) + ? "peer-joined" + : "peer-waiting", peerId: participant.peerId, }); } @@ -675,6 +701,12 @@ export class SignalingServer { this.sendError(socket, "FORBIDDEN", "P2P restart requests target the host implicitly"); return; } + if ( + !this.isOrdinaryHostChild(authenticated.roomId, authenticated.peerId) + ) { + this.sendError(socket, "FORBIDDEN", "Viewer has no active Host edge"); + return; + } this.routeToHost(socket, authenticated, { type: "restart-request", fromPeerId: authenticated.peerId, @@ -954,6 +986,7 @@ export class SignalingServer { if (update.revokedViewers.length > 0) { this.clearRoomGraceTimers(authenticated.roomId); this.clearRoomConnectionIds(authenticated.roomId); + this.ordinaryActiveHostChildrenByRoom.delete(authenticated.roomId); this.deferredViewerPresenceRooms.add(authenticated.roomId); try { @@ -1257,6 +1290,10 @@ export class SignalingServer { this.sendError(sourceSocket, "PEER_NOT_FOUND", "Viewer is not connected"); return; } + if (!this.isOrdinaryHostChild(source.roomId, viewer.peerId)) { + this.sendError(sourceSocket, "FORBIDDEN", "Viewer has no active Host edge"); + return; + } if (description?.type === "offer") { this.setViewerConnectionId( source.roomId, @@ -1276,6 +1313,10 @@ export class SignalingServer { this.sendError(sourceSocket, "FORBIDDEN", "Viewer signals may only reply to the host"); return; } + if (!this.isOrdinaryHostChild(source.roomId, source.peerId)) { + this.sendError(sourceSocket, "FORBIDDEN", "Viewer has no active Host edge"); + return; + } this.routeToHost(sourceSocket, source, { type: "signal", fromPeerId: source.peerId, @@ -1488,12 +1529,25 @@ export class SignalingServer { ); return; } + this.releaseOrdinaryHostChild(disconnected.roomId, disconnected.peerId); + const promotedPeerIds = this.fillOrdinaryHostChildren( + disconnected.roomId, + ); const host = this.options.roomStore.getConnectedHost(disconnected.roomId); if (host) { this.sendToSession(host.sessionId, { type: "peer-left", peerId: disconnected.peerId, }); + for (const peerId of promotedPeerIds) { + this.sendToSession(host.sessionId, { + type: "peer-joined", + peerId, + }); + } + } + if (promotedPeerIds.length > 0) { + this.sendViewerPresence(disconnected.roomId); } } }, this.viewerDisconnectGraceMs); @@ -1524,6 +1578,7 @@ export class SignalingServer { } this.clearRoomGraceTimers(roomId); this.clearRoomConnectionIds(roomId); + this.ordinaryActiveHostChildrenByRoom.delete(roomId); if (this.isHybridMediaEnabled()) { this.hybridMediaRouter!.deleteRoom(roomId); } @@ -1544,6 +1599,7 @@ export class SignalingServer { for (const expired of this.options.roomStore.expireRooms(this.now())) { this.clearRoomGraceTimers(expired.roomId); this.clearRoomConnectionIds(expired.roomId); + this.ordinaryActiveHostChildrenByRoom.delete(expired.roomId); if (this.isHybridMediaEnabled()) { this.hybridMediaRouter!.deleteRoom(expired.roomId); } @@ -1603,6 +1659,60 @@ export class SignalingServer { return this.hybridMediaRouter !== undefined; } + private ordinaryHostChildPeerIds(roomId: string): readonly string[] { + const active = this.ordinaryActiveHostChildren(roomId); + return this.options.roomStore + .getViewerPeerIds(roomId) + .filter((peerId) => active.has(peerId)); + } + + private isOrdinaryHostChild(roomId: string, peerId: string): boolean { + return this.ordinaryActiveHostChildren(roomId).has(peerId); + } + + private ordinaryActiveHostChildren(roomId: string): Set { + let active = this.ordinaryActiveHostChildrenByRoom.get(roomId); + if (!active) { + active = new Set(); + this.ordinaryActiveHostChildrenByRoom.set(roomId, active); + return active; + } + const members = new Set(this.options.roomStore.getViewerPeerIds(roomId)); + for (const peerId of active) { + if (!members.has(peerId)) { + active.delete(peerId); + } + } + return active; + } + + private fillOrdinaryHostChildren(roomId: string): string[] { + const active = this.ordinaryActiveHostChildren(roomId); + const promotedPeerIds: string[] = []; + for (const viewer of this.options.roomStore.getConnectedViewers(roomId)) { + if (active.size >= this.options.endpointMediaCopyCapacity) { + break; + } + if (active.has(viewer.peerId)) { + continue; + } + active.add(viewer.peerId); + promotedPeerIds.push(viewer.peerId); + } + return promotedPeerIds; + } + + private releaseOrdinaryHostChild(roomId: string, peerId: string): void { + const active = this.ordinaryActiveHostChildrenByRoom.get(roomId); + if (!active) { + return; + } + active.delete(peerId); + if (active.size === 0) { + this.ordinaryActiveHostChildrenByRoom.delete(roomId); + } + } + private hasConnectionCapacity(): boolean { return ( !this.closing && @@ -1690,7 +1800,7 @@ export class SignalingServer { } const upstream = this.isHybridMediaEnabled() ? this.hybridMediaRouter!.getViewerRouteUpstream(roomId, viewer.peerId) - : host + : host && this.isOrdinaryHostChild(roomId, viewer.peerId) ? { kind: "peer" as const, peerId: host.peerId } : { kind: "none" as const }; const ready = readyByViewer?.get(viewer.peerId); diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index 332a78b..a67e46c 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -661,6 +661,11 @@ const authenticatedMessageShape = { peerId: opaqueIdSchema, roomExpiresAt: z.string().datetime().nullable(), maxViewers: z.number().int().min(1).max(MAX_VIEWERS_PER_ROOM_LIMIT), + endpointMediaCopyCapacity: z + .number() + .int() + .min(1) + .max(MAX_ENDPOINT_MEDIA_COPY_CAPACITY), hostOnline: z.boolean(), hostPaused: z.boolean().optional(), connectionId: opaqueIdSchema.nullable(), @@ -699,6 +704,12 @@ export const serverMessageSchema = z.union([ peerId: opaqueIdSchema, }) .strict(), + z + .object({ + type: z.literal("peer-waiting"), + peerId: opaqueIdSchema, + }) + .strict(), z .object({ type: z.literal("peer-left"), diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index 8a7c891..553ecc8 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -645,6 +645,7 @@ describe("client signaling recovery policy", () => { peerId: "viewer_12345678", roomExpiresAt: null, maxViewers: 8, + endpointMediaCopyCapacity: 2, hostOnline: true, connectionId: null, viewerPeerIds: [], @@ -718,6 +719,7 @@ describe("client signaling recovery policy", () => { peerId: "host_12345678", roomExpiresAt: null, maxViewers: 8, + endpointMediaCopyCapacity: 2, hostOnline: true, connectionId: null, viewerPeerIds: [], @@ -842,6 +844,7 @@ describe("client signaling recovery policy", () => { peerId: "viewer_12345678", roomExpiresAt: null, maxViewers: 8, + endpointMediaCopyCapacity: 2, hostOnline: true, connectionId: "connection_12345678", viewerPeerIds: [], diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index c3a36fc..16fa78d 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -85,8 +85,8 @@ describe("client signaling protocol", () => { "utf8", ); - expect(nativeWire).toContain( - `signalingProtocol = "${SIGNALING_PROTOCOL}"`, + expect(nativeWire).toMatch( + new RegExp(`signalingProtocol\\s*=\\s*"${SIGNALING_PROTOCOL}"`), ); }); @@ -848,6 +848,7 @@ describe("server signaling protocol", () => { peerId: "host_12345678", roomExpiresAt: "2026-08-18T18:00:00.000Z", maxViewers, + endpointMediaCopyCapacity: 2, hostOnline: true, connectionId: null, viewerPeerIds, @@ -873,6 +874,46 @@ describe("server signaling protocol", () => { ).toBe(true); }); + it("requires the authenticated endpoint capacity to be 1, 2, or 3", () => { + for (const endpointMediaCopyCapacity of [1, 2, 3]) { + expect( + serverMessageSchema.safeParse({ + ...authenticatedMessage(8), + endpointMediaCopyCapacity, + }).success, + ).toBe(true); + } + for (const endpointMediaCopyCapacity of [undefined, 0, 4, 1.5]) { + const message = authenticatedMessage(8) as Record; + if (endpointMediaCopyCapacity === undefined) { + delete message.endpointMediaCopyCapacity; + } else { + message.endpointMediaCopyCapacity = endpointMediaCopyCapacity; + } + expect(serverMessageSchema.safeParse(message).success).toBe(false); + } + }); + + it("accepts only a strict peer-waiting notification", () => { + expect( + serverMessageSchema.safeParse({ + type: "peer-waiting", + peerId: "viewer_12345678", + }).success, + ).toBe(true); + expect( + serverMessageSchema.safeParse({ + type: "peer-waiting", + peerId: "viewer_12345678", + active: false, + }).success, + ).toBe(false); + expect( + serverMessageSchema.safeParse({ type: "peer-waiting", peerId: "short" }) + .success, + ).toBe(false); + }); + it("accepts a strict, unique and bounded Viewer presence snapshot", () => { const viewer = { role: "viewer", diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 4bd4f16..8159a57 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -9231,6 +9231,258 @@ describe("WebSocket signaling", () => { expect((await host.inbox.next("peer-left", 500)).peerId).toBe(firstAuth.peerId); }); + it.each([1, 2, 3] as const)( + "authorizes exactly %i ordinary Host children independently of room admission", + async (endpointMediaCopyCapacity) => { + const maxViewersPerRoom = endpointMediaCopyCapacity + 1; + const harness = await startHarness({ + endpointMediaCopyCapacity, + maxViewersPerRoom, + }); + const viewers: TestClient[] = []; + const viewerPeerIds: string[] = []; + for (let index = 0; index < maxViewersPerRoom; index += 1) { + const viewer = await openClient(harness.webSocketUrl); + const authenticated = await authenticate( + viewer, + harness.room, + "viewer", + `ordinary-cap-viewer-${index}`, + 1, + undefined, + { viewerPresence: true }, + ); + expect(authenticated.endpointMediaCopyCapacity).toBe( + endpointMediaCopyCapacity, + ); + viewers.push(viewer); + viewerPeerIds.push(authenticated.peerId); + } + + const host = await openClient(harness.webSocketUrl); + const hostAuth = await authenticate( + host, + harness.room, + "host", + "ordinary-cap-host", + 1, + undefined, + { viewerPresence: true }, + ); + expect(hostAuth).toMatchObject({ + maxViewers: maxViewersPerRoom, + endpointMediaCopyCapacity, + viewerPeerIds: viewerPeerIds.slice(0, endpointMediaCopyCapacity), + }); + + const presence = viewerPresenceEntries( + await host.inbox.next("viewer-presence"), + ); + for (let index = 0; index < presence.length; index += 1) { + expect(presence[index]).toMatchObject({ + peerId: viewerPeerIds[index], + upstream: + index < endpointMediaCopyCapacity + ? { kind: "peer", peerId: hostAuth.peerId } + : { kind: "none" }, + }); + } + + const waitingPeerId = viewerPeerIds.at(-1)!; + expect((await host.inbox.next("peer-waiting")).peerId).toBe( + waitingPeerId, + ); + for (const activePeerId of viewerPeerIds.slice( + 0, + endpointMediaCopyCapacity, + )) { + expect((await host.inbox.next("peer-joined")).peerId).toBe( + activePeerId, + ); + } + + host.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: waitingPeerId, + payload: { + kind: "description", + connectionId: "waiting-host-offer", + description: { type: "offer", sdp: "v=0\r\n" }, + }, + }), + ); + expect((await host.inbox.next("error")).code).toBe("FORBIDDEN"); + + host.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: waitingPeerId, + payload: { + kind: "candidate", + connectionId: "waiting-host-candidate", + candidate: null, + }, + }), + ); + expect((await host.inbox.next("error")).code).toBe("FORBIDDEN"); + + const waitingViewer = viewers.at(-1)!; + waitingViewer.socket.send( + JSON.stringify({ + type: "signal", + payload: { + kind: "description", + connectionId: "waiting-viewer-answer", + description: { type: "answer", sdp: "v=0\r\n" }, + }, + }), + ); + expect((await waitingViewer.inbox.next("error")).code).toBe("FORBIDDEN"); + + waitingViewer.socket.send( + JSON.stringify({ + type: "signal", + payload: { + kind: "candidate", + connectionId: "waiting-viewer-candidate", + candidate: null, + }, + }), + ); + expect((await waitingViewer.inbox.next("error")).code).toBe("FORBIDDEN"); + + waitingViewer.socket.send( + JSON.stringify({ + type: "restart-request", + connectionId: "waiting-viewer-restart", + rebuild: true, + }), + ); + expect((await waitingViewer.inbox.next("error")).code).toBe("FORBIDDEN"); + + host.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: viewerPeerIds[0], + payload: { + kind: "description", + connectionId: "active-host-offer", + description: { type: "offer", sdp: "v=0\r\n" }, + }, + }), + ); + expect(await viewers[0]!.inbox.next("signal")).toMatchObject({ + fromPeerId: hostAuth.peerId, + payload: { connectionId: "active-host-offer" }, + }); + }, + ); + + it("holds an active ordinary slot through grace and skips offline waiters", async () => { + const harness = await startHarness({ + endpointMediaCopyCapacity: 1, + maxViewersPerRoom: 3, + viewerDisconnectGraceMs: 100, + }); + const host = await openClient(harness.webSocketUrl); + const hostAuth = await authenticate( + host, + harness.room, + "host", + "ordinary-grace-host", + 1, + undefined, + { viewerPresence: true }, + ); + expect( + viewerPresenceEntries(await host.inbox.next("viewer-presence")), + ).toEqual([]); + + const activeViewer = await openClient(harness.webSocketUrl); + const activeAuth = await authenticate( + activeViewer, + harness.room, + "viewer", + "ordinary-grace-active", + ); + expect((await host.inbox.next("peer-joined")).peerId).toBe( + activeAuth.peerId, + ); + expect( + viewerPresenceEntries(await host.inbox.next("viewer-presence")), + ).toEqual([ + expect.objectContaining({ + peerId: activeAuth.peerId, + upstream: { kind: "peer", peerId: hostAuth.peerId }, + }), + ]); + + const offlineWaitingViewer = await openClient(harness.webSocketUrl); + const offlineWaitingAuth = await authenticate( + offlineWaitingViewer, + harness.room, + "viewer", + "ordinary-grace-offline-waiting", + ); + expect((await host.inbox.next("peer-waiting")).peerId).toBe( + offlineWaitingAuth.peerId, + ); + await host.inbox.next("viewer-presence"); + + const connectedWaitingViewer = await openClient(harness.webSocketUrl); + const connectedWaitingAuth = await authenticate( + connectedWaitingViewer, + harness.room, + "viewer", + "ordinary-grace-connected-waiting", + ); + expect((await host.inbox.next("peer-waiting")).peerId).toBe( + connectedWaitingAuth.peerId, + ); + await host.inbox.next("viewer-presence"); + + await closeClient(activeViewer); + await host.inbox.next("viewer-presence"); + await new Promise((resolve) => setTimeout(resolve, 20)); + await closeClient(offlineWaitingViewer); + await host.inbox.next("viewer-presence"); + await host.inbox.expectNone(40); + + expect((await host.inbox.next("peer-left", 300)).peerId).toBe( + activeAuth.peerId, + ); + expect((await host.inbox.next("peer-joined", 300)).peerId).toBe( + connectedWaitingAuth.peerId, + ); + expect( + viewerPresenceEntries(await host.inbox.next("viewer-presence")), + ).toEqual([ + expect.objectContaining({ + peerId: connectedWaitingAuth.peerId, + upstream: { kind: "peer", peerId: hostAuth.peerId }, + }), + ]); + + host.socket.send( + JSON.stringify({ + type: "signal", + targetPeerId: connectedWaitingAuth.peerId, + payload: { + kind: "description", + connectionId: "promoted-host-offer", + description: { type: "offer", sdp: "v=0\r\n" }, + }, + }), + ); + expect(await connectedWaitingViewer.inbox.next("signal")).toMatchObject({ + payload: { connectionId: "promoted-host-offer" }, + }); + expect((await host.inbox.next("peer-left", 300)).peerId).toBe( + offlineWaitingAuth.peerId, + ); + }); + it("clears the previous media generation when sharing stops", async () => { const harness = await startHarness({ viewerDisconnectGraceMs: 300 }); const host = await openClient(harness.webSocketUrl); @@ -9636,7 +9888,7 @@ describe("WebSocket signaling", () => { `viewer-client-${index}`, ); expect(viewerAuth.maxViewers).toBe(maxViewersPerRoom); - await host.inbox.next("peer-joined"); + await host.inbox.next(index <= 2 ? "peer-joined" : "peer-waiting"); viewers.push(viewer); }