From f87c2ba0ed09c15e11ec540712ced74ea4108685 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Wed, 26 Aug 2026 01:21:34 +0800 Subject: [PATCH] fix(route): preserve media while converging direct paths --- .env.example | 10 +- .../0002-memory-resident-protected-rooms.md | 10 +- .../0005-automatic-hybrid-media-routing.md | 51 +-- docs/deployment.md | 16 +- docs/research/advanced-peer-distribution.md | 40 ++- docs/todo.md | 26 +- docs/方案设计.md | 6 +- docs/需求理解.md | 6 +- scripts/access-privacy-gate.ts | 2 +- scripts/native-one-viewer-gate.ts | 3 +- scripts/peer-assisted-benchmark.ts | 3 +- src/client/media/viewer-presentation.ts | 24 +- src/client/pages/ViewerPage.tsx | 16 +- src/server/app.ts | 8 +- src/server/config.ts | 52 --- src/server/room-route-controller.ts | 335 ++++++++++++------ src/server/room-store.ts | 8 +- src/shared/protocol.ts | 1 + tests/hybrid-media-router.test.ts | 21 +- tests/room-route-controller.test.ts | 199 ++++++++--- tests/room-store.test.ts | 16 +- tests/server-config.test.ts | 69 +--- tests/server-http.test.ts | 29 +- tests/server-signal.test.ts | 9 +- tests/viewer-presentation.test.ts | 13 + 25 files changed, 606 insertions(+), 367 deletions(-) diff --git a/.env.example b/.env.example index d327c35..467ba0f 100644 --- a/.env.example +++ b/.env.example @@ -12,7 +12,6 @@ SITE_ACCESS_PASSWORD= # Dormant rooms remain resumable by the exact Host token for this lease. # Active sharing does not expire; process restart clears every room. ROOM_LEASE_SECONDS=86400 -MAX_ROOMS=1000 MAX_VIEWERS_PER_ROOM=8 # Bounded standard-WebRTC relay/SFU controller for every room when true. # Keep false outside measured tests. @@ -22,15 +21,14 @@ PEER_ASSISTED_MEDIA=false ENDPOINT_MEDIA_COPY_CAPACITY=2 # Optional automatic SFU fallback. Set PEER_ASSISTED_MEDIA=true and supply all -# six values together; leaving them empty keeps LiveKit completely dormant. -# Capacities have no defaults and must come from this deployment's measurements. +# four values together; leaving them empty keeps LiveKit completely dormant. +# Ingress uses the 9000-room code capacity; egress uses that capacity times the +# configured per-room Viewer admission. # Production requires wss and an independent API secret of 32+ bytes. LIVEKIT_URL= LIVEKIT_API_URL= LIVEKIT_API_KEY= LIVEKIT_API_SECRET= -SFU_INGRESS_CAPACITY= -SFU_EGRESS_CAPACITY= # Ordinary peer ICE discovery. Production requires at least one STUN URL. # Separate multiple URLs with commas. @@ -45,6 +43,4 @@ STUN_URLS= # LIVEKIT_API_URL=http://127.0.0.1:7880 # LIVEKIT_API_KEY=replace-with-a-livekit-api-key # LIVEKIT_API_SECRET=replace-with-at-least-32-independent-random-bytes -# SFU_INGRESS_CAPACITY=replace-with-measured-positive-integer -# SFU_EGRESS_CAPACITY=replace-with-measured-positive-integer # ENDPOINT_MEDIA_COPY_CAPACITY=2 diff --git a/docs/adr/0002-memory-resident-protected-rooms.md b/docs/adr/0002-memory-resident-protected-rooms.md index 66671e1..0535c7e 100644 --- a/docs/adr/0002-memory-resident-protected-rooms.md +++ b/docs/adr/0002-memory-resident-protected-rooms.md @@ -41,9 +41,10 @@ All room state lives in the authoritative application process: The free-code pool contains every currently unallocated four-digit code. A room creation request may name one locally preferred code: the store takes it only if it is still in the free pool, otherwise it selects uniformly from the current -pool. Room release returns the code to that pool. `MAX_ROOMS` must not exceed the -9,000-code space; its product default remains 1,000. A room code is a locator and -best-effort preference, not a secret, reservation, or permanent identity. +pool. Room release returns the code to that pool. The complete `1000..9999` +space fixes active room capacity at 9,000; there is no separate room-limit +configuration. A room code is a locator and best-effort preference, not a +secret, reservation, or permanent identity. `ROOM_LEASE_SECONDS` is the single room-lifetime setting and defaults to 86,400 seconds. An authenticated, actively sharing Host prevents expiry. When sharing @@ -154,7 +155,8 @@ the generic `SERVER_ERROR`. None of those paths is folded into ## Acceptance Gates - Allocation uses only `1000` through `9999`, never duplicates an active code, - respects `MAX_ROOMS`, and returns released codes to the free pool. + admits at most the fixed 9,000-code capacity, and returns released codes to + the free pool. - An active Host is not expired. Stop/disconnect starts the configured dormant lease; the exact Host token resumes before expiry; Viewer activity does not. - Expiry and process restart reject the old Host token, Viewer grant, room diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index 1de18c6..76639cf 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -116,11 +116,23 @@ remains an observed acceptance metric. A candidate identity is one logical upstream path. An operation opened by a failed edge seeds that exact current candidate as already tried, so the same -parent cannot immediately repeat. Other eligible direct/STUN parents use -deterministic order first, followed by SFU reuse or direct Host publication -ingress. A new external fact starts a new operation and may make the old -candidate eligible again. The operation records only exact candidates already -tried; failure does not globally exclude that parent from later room events. +parent cannot immediately repeat. A new external fact may make the old +candidate eligible again; failure does not globally exclude that parent. + +Foreground availability and direct convergence have separate priorities but +use the same controller and candidate transaction. Without a current Host +publication, route acquisition first tries the best eligible direct parent, +then SFU, then retains the remaining direct suffix. Once the room has a healthy Host publication, another +joining Viewer first commits an SFU subscription so an earlier slow direct +candidate cannot serialize every waiting Viewer. A Viewer that committed SFU +from foreground acquisition retains that healthy route while the controller later tries its +remaining direct parents, one exact candidate at a time. Newly unrouted or +repairing Viewers always preempt this background direct work. The first direct +candidate that proves a decoded frame replaces SFU make-before-break; exhausting +the finite parent list simply keeps SFU. A candidate that would require a +bounded-gap cutover is skipped because background convergence may not interrupt +current media. This is event-driven route convergence, not periodic rebalancing +or quality-based switching. If an SFU subscription is needed while the Host has no publication and all Host slots are occupied, reconciliation uses the same child operation to convert one @@ -141,16 +153,13 @@ PeerConnection, and publication generation remains owned by the SFU resource lifecycle. A single pending child-operation object aggregates these bindings, the deterministic candidate list and cursor, its current candidate and reservations, -and one total operation deadline. Candidate failure advances the cursor without -resetting that deadline. These fields do not become separate gates or state -machines, and this route wave adds no -assignment, media-binding, or proof generation to the wire. -The same one timer derives wake boundaries from the route classes actually -present in the deterministic list: direct peer and SFU. The total deadline is -divided equally between those semantic stages, without fixed per-candidate -milliseconds. Hard failures may advance through multiple candidates inside a -stage; a silent direct candidate at its boundary skips the remaining direct -candidates so it cannot consume the SFU suffix. +and one deadline. Candidate failure advances only that exact cursor entry. +Initial availability keeps one bounded direct window before SFU. Background +direct convergence uses the same bounded window for each remaining exact parent; +its finite tried set survives between attempts, and a higher-priority route +demand aborts the current attempt without marking the parent failed. These +fields do not create another graph, concurrent pending route revision, periodic +timer, assignment generation, or all-pairs probe. A `prepare` route update names that operation's exact child, route kind, and server-issued candidate connection identity. Parent and child therefore prepare the same connection; neither endpoint infers candidate authority from @@ -334,12 +343,12 @@ extra/private fields, deny the request to Viewers, and prove departure and room-deletion cleanup. For the single-process deployment, SFU admission is one injected authority with -deployment-wide ingress and egress counters. Enabling LiveKit requires explicit -positive safe-integer `SFU_INGRESS_CAPACITY` and `SFU_EGRESS_CAPACITY` values; -neither has a product default and neither is derived from endpoint capacity, -Viewer admission, or a fixed root count. One exact room/share/publication entry -owns the Host publication ingress, while an exact Viewer subscription handle -under that entry owns one egress unit. The current publication is reused as +deployment-wide ingress and egress counters. The fixed 9,000-room code space is +also the ingress ceiling because each room owns at most one Host publication; +egress is the same room capacity multiplied by per-room Viewer admission. +Neither has an independent setting. One exact room/share/publication entry owns +the Host publication ingress, while an exact Viewer subscription handle under +that entry owns one egress unit. The current publication is reused as Viewers enter or leave the SFU path; each child candidate reserves and commits only its own subscription handle. Reserved, committed, and draining handles, concurrent publication generations, and generations still draining from diff --git a/docs/deployment.md b/docs/deployment.md index e5ea5d4..faa7bf8 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -238,7 +238,6 @@ PUBLIC_BASE_URL=https://share.example.com ALLOWED_ORIGINS=https://share.example.com SITE_ACCESS_PASSWORD= ROOM_LEASE_SECONDS=86400 -MAX_ROOMS=1000 MAX_VIEWERS_PER_ROOM=20 ENDPOINT_MEDIA_COPY_CAPACITY=2 @@ -253,8 +252,6 @@ LIVEKIT_URL=wss://share.example.com LIVEKIT_API_URL=http://127.0.0.1:7880 LIVEKIT_API_KEY= LIVEKIT_API_SECRET= -SFU_INGRESS_CAPACITY= -SFU_EGRESS_CAPACITY= ``` Supplying any stale `PEER_ICE_TURN_*` or `SELECTED_EDGE_TURN_*` key, even blank, @@ -301,13 +298,10 @@ atomically on `screener-v12`; every stale Browser or executable-sender wire fail before room authority. Native senders and helpers are outside this release. The four `LIVEKIT_*` values must either all be absent or all be present, and a -complete tuple requires `PEER_ASSISTED_MEDIA=true` plus explicit positive -safe-integer `SFU_INGRESS_CAPACITY` and `SFU_EGRESS_CAPACITY` values. The two -capacities have no defaults and must be selected from the instance's measured -publisher/subscriber/bitrate and accepted concurrency matrix. Supplying either -capacity without the complete fallback configuration fails startup rather than -silently enabling or ignoring a partial policy. An empty tuple with neither -capacity keeps the optional SDK and server path dormant. `LIVEKIT_URL` must be a +complete tuple requires `PEER_ASSISTED_MEDIA=true`. The four-digit code space +fixes ingress at 9,000 Host publications; egress is 9,000 multiplied by +`MAX_VIEWERS_PER_ROOM`. There is no separate SFU capacity configuration. An +empty tuple keeps the optional SDK and server path dormant. `LIVEKIT_URL` must be a plain `ws:` or `wss:` origin with no `/rtc` suffix; production requires `wss:`. `LIVEKIT_API_URL` must be a plain `http:` or `https:` origin with no path; production permits plaintext only on loopback and otherwise requires `https:`. @@ -380,7 +374,7 @@ The Host may update or revoke that grant without changing code-entry policy. Room allocation and lifetime use one deliberately small model: - random unallocated codes come only from `1000` through `9999`, never duplicate - an active code, respect `MAX_ROOMS <= 9000`, and return on room release; + an active code, use the fixed 9,000-room capacity, and return on room release; - active Host sharing does not expire; stop, capture-track end, or Host disconnect starts `ROOM_LEASE_SECONDS`; - only the exact Host token resumes and renews before expiry, Viewer activity diff --git a/docs/research/advanced-peer-distribution.md b/docs/research/advanced-peer-distribution.md index 33539be..82a30d4 100644 --- a/docs/research/advanced-peer-distribution.md +++ b/docs/research/advanced-peer-distribution.md @@ -1,6 +1,6 @@ # Advanced Peer Distribution -- Research date: 2026-08-24 +- Research date: 2026-08-26 - Scope: current route admission up to twenty trusted viewers, sub-second interactive media, endpoint downstream cap `1..3`, and minimal central-server media egress; retained advanced-media measurements may cover smaller cohorts @@ -269,6 +269,33 @@ errors, URLs, tokens, candidates, and addresses out of wire and logs. ## Automatic Route Controller Boundary +The 2026-08-26 production trace established a controller defect rather than a +missing relay topology: one SFU-fed Viewer was repeatedly tried as the first +direct parent, and its silent 10-second attempt caused every later direct parent +to be skipped before SFU reuse. Offer, answer, trickled candidates, relay +capacity, and the untried P2P-fed parent were all present. A room-wide serial +candidate therefore amplified one slow parent into queue delay for every later +Viewer. + +Browser ICE already races and paces address, interface, and NAT candidate pairs +inside one exact `RTCPeerConnection`. There is no mature Web API that races +multiple remote peers as one connection; each additional peer requires another +offer/answer, ICE/DTLS/RTP state, reservation, and potentially another media +encode. Jitsi provides the closest deployed model: it retains its JVB session, +attempts one exact P2P session, and stops using JVB only after P2P ICE succeeds. +Chunk swarms such as CoolStreaming, WebTorrent, PeerTube, and commercial eCDN +systems can contact several suppliers because buffered segments are +interchangeable; that model does not apply to one live RTP `MediaStream`. + +The retained Browser design consequently keeps one current candidate and one +authoritative graph. Initial acquisition gives one best direct parent a bounded +head start before SFU; once SFU has a current decoded frame, the remaining exact +direct parents are attempted one at a time behind that media. A new unrouted or +repairing Viewer preempts this low-priority convergence. Direct success commits +make-before-break on its first decoded frame; exhaustion leaves healthy SFU +unchanged. Literal multi-parent racing or concurrent child operations require a +new wire/revision/reservation model and remain outside the current result. + The accepted controller preserves healthy peer edges and chooses one route per logical edge: direct/STUN first, then the dedicated SFU/UDP path when needed. A server-fed ingress uses the room's single Host publication plus exact SFU @@ -326,6 +353,14 @@ Sources checked on 2026-08-20 through 2026-08-24: source copied. - [WebRTC Recommendation](https://www.w3.org/TR/webrtc/) - `failed` ICE restart and `disconnected` bytes/stats guidance. +- [ICE, RFC 8445](https://www.rfc-editor.org/rfc/rfc8445.html), + [ICE Happy Eyeballs, RFC 8421](https://www.rfc-editor.org/rfc/rfc8421.html), + [Trickle ICE, RFC 8838](https://www.rfc-editor.org/rfc/rfc8838.html), and + [ICE PAC, RFC 8863](https://www.rfc-editor.org/rfc/rfc8863.html) - one exact + peer's native candidate checks and why application deadlines cannot infer an + untried parent. +- [JitsiConference P2P/JVB ownership](https://github.com/jitsi/lib-jitsi-meet/blob/master/JitsiConference.ts) + - mature server-media retention until one exact P2P session succeeds. - [LiveKit client 2.22 reconnect policy](https://github.com/livekit/client-sdk-js/blob/v2.22.0/src/room/DefaultReconnectPolicy.ts), [room defaults](https://github.com/livekit/client-sdk-js/blob/v2.22.0/src/room/defaults.ts), and [state reconciliation](https://github.com/livekit/client-sdk-js/blob/v2.22.0/src/room/Room.ts) @@ -353,8 +388,7 @@ Sources checked on 2026-08-20 through 2026-08-24: [WebTorrent](https://github.com/webtorrent/webtorrent), and [P2P Media Loader](https://github.com/Novage/p2p-media-loader) - mature chunk/segment swarms, not an RTP route implementation. -- [WebRTC statistics](https://www.w3.org/TR/webrtc-stats/), - [ICE, RFC 8445](https://www.rfc-editor.org/rfc/rfc8445.html), and +- [WebRTC statistics](https://www.w3.org/TR/webrtc-stats/) and [LiveKit SFU](https://docs.livekit.io/reference/internals/livekit-sfu/) - observable path boundaries and the centralized low-latency alternative. - [LiveKit client 2.22 room events](https://github.com/livekit/client-sdk-js/blob/v2.22.0/src/room/Room.ts) diff --git a/docs/todo.md b/docs/todo.md index 42bf0b4..82e76a4 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -1,26 +1,38 @@ # Current TODO Ledger -Last reviewed: 2026-08-25 +Last reviewed: 2026-08-26 Only **Now** is executable. A branch, old experiment, observation, or accepted later topic is not implementation authority by itself. ## Now -1. **Preferred room reacquisition.** While sharing, keep the last assigned code +1. **Restore bounded direct-parent selection.** Diagnose the `2244`/`7723` + production traces and replace the class-stage deadline that can let one + silent P2P candidate skip every remaining direct parent. Preserve exact + candidate failure, direct-before-SFU order, and bounded latency at 20 + Viewers without linear exhaustive waiting or all-pairs probing. +2. **Converge Viewer route presentation.** Current decoded media must clear a + stale no-route/reconnecting overlay. Manual media reconnect must refresh the + current route facts and presentation state without requiring a page reload. +3. **Verify multi-room SFU admission.** Replace production's single-publication + limit with the fixed 9,000-room ingress and + `9000 * MAX_VIEWERS_PER_ROOM` subscription egress derived from product + admission, then verify two rooms can use independent SFU publications. +4. **Preferred room reacquisition.** While sharing, keep the last assigned code as an unexpired local preference; normal stop starts the configured room lease countdown. A new room may request that code, but the server uses it only while free and always creates a new room incarnation and credentials. -2. **Browser VP8 hardware evidence.** Measure supported Chrome/Edge platforms +5. **Browser VP8 hardware evidence.** Measure supported Chrome/Edge platforms under real game load using actual codec, encoder implementation, encode time, frame rate, and CPU/GPU attribution. Do not add another codec, custom encoder, GPU selector, or Native helper without a new decision. -3. **Mobile Viewer lifecycle.** Run Android Chrome and iOS Safari matrices for +6. **Mobile Viewer lifecycle.** Run Android Chrome and iOS Safari matrices for autoplay gesture, foreground/background audio, foreground video recovery, lock/page reclamation, rotation, Wi-Fi/cellular migration, and assigned-relay survival or controller recovery. Web does not promise background video or relay execution after OS suspension. -4. **Finish real-network route and media acceptance.** Exercise direct peer, +7. **Finish real-network route and media acceptance.** Exercise direct peer, Browser relay, SFU, relay-ingress recovery with subtree retention, disconnect/capacity drain, Pause/Resume, source replacement, screen-audio continuity, and real-game A/V sync across representative IPv4/IPv6, @@ -33,7 +45,9 @@ later topic is not implementation authority by itself. 1. **Quality-based topology research.** After the real-network baseline, decide whether measured user-visible quality justifies any active parent-selection - mechanism. Current evidence cannot compare an active route with an + mechanism. A weak-Host production session showed a usable P2P Viewer with + worse playback than SFU Viewers, so reachability alone is not the eventual + optimization target. Current evidence still cannot compare an active route with an unconnected alternative, so no quality score, all-pairs probing, periodic rebalancing, relay abdication threshold, or active parent switch is accepted. Prefer mature algorithms and one general model if this boundary is reopened. diff --git a/docs/方案设计.md b/docs/方案设计.md index 00324a9..7d58716 100644 --- a/docs/方案设计.md +++ b/docs/方案设计.md @@ -84,7 +84,7 @@ Host 的同源 `localStorage` creation profile 保存显示名、code-only polic 邀请使用 `/r/{roomId}#v={grant}`。RFC 3986 要求 fragment 在 dereference 前由 user agent 分离，因此 HTTP request-target 与反向代理 access log 看不到它；WHATWG WebSocket 还明确拒绝含 fragment 的 URL。客户端只从页面 URL 严格解析一次，把 grant 写入 `screener:viewer-grant:{roomId}` 的 `sessionStorage`，随即以 `history.replaceState` 把地址栏替换为 canonical `/r/{roomId}`；刷新和页面内 WSS 重连继续使用该页面会话的值。Host 建房/rotate 收到的 raw grant 同样只在内存和 room-scoped `sessionStorage` 中保留，用它临时构造复制链接；`localStorage` 的 room ownership 只包含 canonical URL、room ID、Host token 和期限。grant 不进入 cookie、query、WebSocket URL、Referrer、日志或错误。生产继续发送 `Referrer-Policy: no-referrer`；fragment 与该 header 降低传输泄漏，但不能防止原邀请消息、浏览器扩展、截图或用户转发，UI 必须把链接称为访问凭据。 -房间只有一种易失模型。进程启动时建立 `1000..9999` 的空闲码池；创建房间可携带一个严格四位 `preferredRoomId`，若它仍在空闲池则以同一个 swap-remove 取走，否则用加密随机索引从当前空闲集合分配，释放时放回。偏好请求不预占、不替换现存房间，也不改变新房凭据生成。这样在 9000 个号码内没有碰撞重试、递增身份或数据库序列。`MAX_ROOMS` 默认 1000 且不得超过 9000。 +房间只有一种易失模型。进程启动时建立 `1000..9999` 的空闲码池；创建房间可携带一个严格四位 `preferredRoomId`，若它仍在空闲池则以同一个 swap-remove 取走，否则用加密随机索引从当前空闲集合分配，释放时放回。偏好请求不预占、不替换现存房间，也不改变新房凭据生成。这样在固定 9000 个活动房间容量内没有碰撞重试、递增身份、独立房间上限或数据库序列。 `ROOM_LEASE_SECONDS` 默认 86400。RoomStore 对连接中的 Host 不执行到期；Host 明确停止或连接消失时将 `expiresAt` 设为 `now + lease`，同一 Host token 在期限内重新鉴权并分享时续租，Viewer 活动不修改期限。清理循环删除 dormant expired room、关闭其 session/route/resource 并把号码归还空闲池。应用进程重启直接清空全部房间；配置不再包含 `ROOM_DATABASE_PATH`、`ROOM_TTL_SECONDS` 或任何 schema/migration。所有房间继续生成 256-bit Host token，Host 浏览器保存 raw token，服务器当前进程只保存摘要；不使用 IP、UA、设备信息或浏览器指纹认领。 @@ -336,7 +336,7 @@ Token-free SFU standby prewarm 保持可选且不创建 participant、publicatio - SiteAccess 的 production 必配、8 至 128 visible-ASCII bytes/独立 secret、local-dev 显式关闭、12 小时签名 cookie、过期、篡改、cookie 属性、已移除 access key fail-fast 和无服务端 session 状态； - RoomStore 的四位空闲码随机分配/释放、活动分享免过期、dormant lease/续租、Host token/Viewer grant 摘要验证、22 字符 128-bit grant 与 exact room incarnation 同寿、`open/private` 正交 policy、私密无密码仅邀请可入、跨房/同号重建拒绝、rotate/revoke 强撤销、Viewer 上限和超额拒绝； -- 进程重启后 room state 全空、旧 Host/grant/password fail closed、释放号码复用时旧 grant 仍失败；`ROOM_LEASE_SECONDS` 的默认/边界与 `MAX_ROOMS <= 9000` 校验，已删除的 `ROOM_DATABASE_PATH`/`ROOM_TTL_SECONDS` fail-fast； +- 进程重启后 room state 全空、旧 Host/grant/password fail closed、释放号码复用时旧 grant 仍失败；`ROOM_LEASE_SECONDS` 的默认/边界与固定 9000 房间码池，已删除的 `ROOM_DATABASE_PATH`/`ROOM_TTL_SECONDS` fail-fast； - STUN UDP URL 严格校验、生产 STUN 必需、LiveKit tuple 完整性/TTL/secret 独立性、当前 route/session/revision/connection generation guard，以及 ordinary peer、未授权房间和 Native 的 STUN-only isolation；LiveKit `sfu-config` 保持独立边界； - Accepted v12 contract 使用单一 `screener-v12` 信令 literal、严格 JSON schema、方向与房间授权校验、payload 上限和 client-mismatch 刷新终态；分享状态由 `shareGeneration`-fenced paused boolean 表达，missing room 使用 `ROOM_NOT_FOUND`，现存房间准入失败使用 `ROOM_ACCESS_DENIED`，Viewer route failure 为 typed terminal，Host-only bounded route snapshot 保留给自动验收但不进入产品 UI。`qualityProfileIdSchema` 保持三档，`qualityResolutionSchema` 包含 `480p`，`DEFAULT_QUALITY_SETTINGS` 为 1080p30，quality schema 严格拒绝 codec 字段； - 质量协议的 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-off； @@ -384,7 +384,7 @@ Pending candidate 的 receive-transport owner 只在当前 exact generation 下 所有非服务器端点共享一个 server-authoritative steady outbound media-copy cap `C`：唯一配置 `ENDPOINT_MEDIA_COPY_CAPACITY` 默认 `2`，只接受 `1`、`2` 或 `3`，不兼容旧键。服务端将 endpoint 当前广告的 `0..C` 可用值 clamp 为唯一 `effectiveDownstreamCapacity`；该值只表示当前可用 sender slots，不建立角色、浏览器、UA、设备或页面可见性 tier。peer child 或 Host publication 各占一份，upstream receive 免费；SFU subscriber egress 使用独立 server admission。Browser `screener-v12` 的 `authenticated.endpointMediaCopyCapacity` 把部署 `C` 交给 Host 与 Viewer；两端用该值校验稳态 assignment，并让 active+pending 的物理 copy 只在 controller 已授权 overlap 时达到 `min(C + 1, 3)`。代码通过一个 shared accounting guard 执行，不重复 magic number；其他 wire 与 Native executable 在 authority 前拒绝，不保留兼容路径。`C=3` 满载时先收到 active bounded-gap retirement再收到 prepare；任何会产生第四份 endpoint media copy 的 token都不得签发。 -SFU admission 在应用单进程中只有一个注入实例。LiveKit credential tuple 完整时，必须同时显式配置私有 `LIVEKIT_API_URL`、正 safe integer `SFU_INGRESS_CAPACITY` 与 `SFU_EGRESS_CAPACITY`；容量没有产品默认值，也不从 Viewer 数、endpoint cap 或历史 root 数推导。专用 LiveKit 实例设置 `room.auto_create: false`。账本只保存全局已用计数，以及 exact room/share/publication generation 下的一份 publication handle 和按 Viewer identity 索引的 subscription handle。Publication 的 `reserved | committed | draining` 占 ingress `1`，每份 subscription 的同类状态占 egress `1`；离开 route 的 subscription 在该 generation 删除并确认 absent 前保持 draining，同一 Viewer handle 再次启用不重复计数。首个 SFU child 先 reserve publication 与自己的 subscription，再通过 `RoomService.CreateRoom` 创建 exact-generation room，成功后才 prepare 和签发 token；后续 child 只在同 generation reserve subscription 并复用 Host publication。exact candidate child 的 first decoded frame 后 commit 自己的资源；publication replacement 将旧 generation 及其全部 subscription 一起转入 draining，失败、超时、失联、分享换代、停止或删房按资源所有权进入同一生命周期。只有 `RoomService.DeleteRoom` 成功且 follow-up lookup 证明 room 不存在，账本才扣除该 generation 的 ingress/egress；auto-create 关闭后，旧 token 不能重建已删除 room。进程启动在接受流量前拒绝任何非 Screener room，删除全部遗留 Screener room并确认 namespace 为空；关闭时先 drain 再遗忘计数。Host 信令离线期间若 exact room 中的 Host participant 仍存在就继续计费并保留健康媒体，participant 消失后在一个有界轮询周期内回收。稳态 reserve/commit/drain bookkeeping 仍是 Map lookup 与整数加减；只有 generation 创建、删除、启动清场和 Host 离线检查访问 LiveKit control plane。多应用进程部署必须先增加共享原子 admission/lifecycle owner。 +SFU admission 在应用单进程中只有一个注入实例。LiveKit credential tuple 完整时必须显式配置私有 `LIVEKIT_API_URL`；固定四位房号池给出 9000 个 publication ingress，上界 `9000 * MAX_VIEWERS_PER_ROOM` 给出 subscription egress，不存在独立 SFU 容量配置。专用 LiveKit 实例设置 `room.auto_create: false`。账本只保存全局已用计数，以及 exact room/share/publication generation 下的一份 publication handle 和按 Viewer identity 索引的 subscription handle。Publication 的 `reserved | committed | draining` 占 ingress `1`，每份 subscription 的同类状态占 egress `1`；离开 route 的 subscription 在该 generation 删除并确认 absent 前保持 draining，同一 Viewer handle 再次启用不重复计数。首个 SFU child 先 reserve publication 与自己的 subscription，再通过 `RoomService.CreateRoom` 创建 exact-generation room，成功后才 prepare 和签发 token；后续 child 只在同 generation reserve subscription 并复用 Host publication。exact candidate child 的 first decoded frame 后 commit 自己的资源；publication replacement 将旧 generation 及其全部 subscription 一起转入 draining，失败、超时、失联、分享换代、停止或删房按资源所有权进入同一生命周期。只有 `RoomService.DeleteRoom` 成功且 follow-up lookup 证明 room 不存在，账本才扣除该 generation 的 ingress/egress；auto-create 关闭后，旧 token 不能重建已删除 room。进程启动在接受流量前拒绝任何非 Screener room，删除全部遗留 Screener room并确认 namespace 为空；关闭时先 drain 再遗忘计数。Host 信令离线期间若 exact room 中的 Host participant 仍存在就继续计费并保留健康媒体，participant 消失后在一个有界轮询周期内回收。稳态 reserve/commit/drain bookkeeping 仍是 Map lookup 与整数加减；只有 generation 创建、删除、启动清场和 Host 离线检查访问 LiveKit control plane。多应用进程部署必须先增加共享原子 admission/lifecycle owner。 只有实际因 deployment-wide SFU admission 失败的房间进入一个 `resourceWaiterRoomIds` Set。LiveKit delete-plus-absent 使真实 usage 下降时，owner 一次 drain 该 Set，对每个仍存在的 controller 调用 `touchExternalFacts()` 并排入普通 reconcile。它不轮询、不按资源类型复制 route state，也不唤醒未等待资源的房间。 diff --git a/docs/需求理解.md b/docs/需求理解.md index 3e1a3ea..ce05b0c 100644 --- a/docs/需求理解.md +++ b/docs/需求理解.md @@ -54,7 +54,7 @@ SFU 的 publication、subscription、server-resource accounting、临时 overlap - 原生 shared encode relay：具备原生 relay 能力的非服务器端点同样受统一服务端权威下游容量约束，并应复用兼容编码输出而不再次编码。当前普通浏览器 relay 使用逐边重编码的标准 WebRTC 路径，不伪称 shared encode。 - SVC 重开条件：当前标准 Web/LiveKit SVC 已静态拒绝为 `no-go-web-svc-cross-path-hardware-contract`。sender `scalabilityMode` 没有 direct/peer receiver 对应的按层选择或跨 PeerConnection shared-encode 契约；requested/applied mode、Media Capabilities `powerEfficient` 和 RTCStats encoder 字段也不能证明没有静默软件 fallback。Pinned client 2.22.0 还会把 screen-share SVC 覆盖为三层 temporal-only `L1T3`，违反当前 Browser 无 SVC、单表示合同。只有未来 native sender 能明确选择硬件编码器、最多发布两层、将同一编码输出复用于 direct/peer/SFU、保持每条路径独立选择并通过真实游戏和亚秒延迟门槛时，才可用新的 capability decision 重开 SVC；不能把浏览器 API 支持或 UA 推断当成采用理由。 - 分包/分层多树实验：把编码包、对象或可独立解码层拆到多个 peer 路径以分散上传，只能在独立 spike 中验证。必须保持亚秒级，处理多父重组、排序、丢包恢复、关键帧、同步和节点退出；未通过门槛前不进入产品路由控制器。 -- SFU 中央兜底：产品部署提供有界 SFU 资源，但不要求用户选模式，也不得默认变成整房 always-SFU。只有 Host 拥有一个 authoritative publication；SFU-fed 与 peer-fed endpoint 使用同一个 provisional-child first-frame transaction，严格受限 endpoint 在没有成功 child candidate 时自然保持 leaf。启用 LiveKit fallback 时必须使用专用且关闭 `room.auto_create` 的实例，并显式提供私有 `LIVEKIT_API_URL`、无默认值的正整数 `SFU_INGRESS_CAPACITY` 与 `SFU_EGRESS_CAPACITY`。单进程全局账本按 exact room/share/publication 保存一份 Host publication ingress，并在该 generation 下按 exact Viewer 保存 subscription egress；当前 publication 被后续 SFU Viewer 复用，每个 child candidate 只 reserve/commit 自己的 subscription。一份 publication 占一个 ingress，每份 reserved、committed 或 draining subscription 占一个 egress；离开 route 的 subscription 在整个 publication room 删除并确认 absent 前仍作为 draining 计费，同一 Viewer handle 再次启用不重复计数。新 generation 在发 token 前 reserve publication 与首个 subscription，并先通过 RoomService 创建对应 room；first-frame ready同步commit admission后才promote graph。publication replacement、abort、timeout、disconnect、share rollover、room stop/delete按资源差集或controller `dispose()`进入draining，只有DeleteRoom成功和absent readback才释放该generation的ingress/egress。启动时必须先拒绝外来room、清除全部遗留Screener room并确认namespace为空；Host信令离线但LiveKit participant仍存在时继续保留和计费，participant消失后有界回收。实际因deployment-wide admission失败的房间只进入一个waiter Set；LiveKit drain完成时一次性touch这些controller的external fact并走普通reconcile，不轮询。固定root数和普通endpoint child数不能代替准入；资源耗尽进入明确waiting或failure。多应用进程部署必须先引入共享原子owner。 +- SFU 中央兜底：产品部署提供有界 SFU 资源，但不要求用户选模式，也不得默认变成整房 always-SFU。只有 Host 拥有一个 authoritative publication；SFU-fed 与 peer-fed endpoint 使用同一个 provisional-child first-frame transaction，严格受限 endpoint 在没有成功 child candidate 时自然保持 leaf。启用 LiveKit fallback 时必须使用专用且关闭 `room.auto_create` 的实例，并显式提供私有 `LIVEKIT_API_URL`。固定四位房号池同时给出 9000 个 publication ingress 上界，subscription egress 上界为 `9000 * MAX_VIEWERS_PER_ROOM`，不另设 SFU 容量参数。单进程全局账本按 exact room/share/publication 保存一份 Host publication ingress，并在该 generation 下按 exact Viewer 保存 subscription egress；当前 publication 被后续 SFU Viewer 复用，每个 child candidate 只 reserve/commit 自己的 subscription。一份 publication 占一个 ingress，每份 reserved、committed 或 draining subscription 占一个 egress；离开 route 的 subscription 在整个 publication room 删除并确认 absent 前仍作为 draining 计费，同一 Viewer handle 再次启用不重复计数。新 generation 在发 token 前 reserve publication 与首个 subscription，并先通过 RoomService 创建对应 room；first-frame ready同步commit admission后才promote graph。publication replacement、abort、timeout、disconnect、share rollover、room stop/delete按资源差集或controller `dispose()`进入draining，只有DeleteRoom成功和absent readback才释放该generation的ingress/egress。启动时必须先拒绝外来room、清除全部遗留Screener room并确认namespace为空；Host信令离线但LiveKit participant仍存在时继续保留和计费，participant消失后有界回收。实际因deployment-wide admission失败的房间只进入一个waiter Set；LiveKit drain完成时一次性touch这些controller的external fact并走普通reconcile，不轮询。固定root数和普通endpoint child数不能代替准入；资源耗尽进入明确waiting或failure。多应用进程部署必须先引入共享原子owner。 - 分享端桌面封装：用 Electron 固定 Chromium 版本、改善选源/热键/系统音频体验，同时保留 Web 观看端。 - Windows 应用音频：Windows 11 Native sender 使用本地 opaque window target 绑定 HWND、PID 与进程创建时间，只捕获所选游戏及其子进程的音频，避免把语音聊天或通知声一并分享；“全部系统声音”只能由分享者显式选择。普通 Windows 10 不具备当前 process-loopback API 门槛，必须明确标为未解决/不支持而不是伪兼容。浏览器无法证明或隔离任意进程音频时必须如实提示，原生捕获失败、无声或进程退出不得静默扩大为整机捕获。 - 受限网络边界：当前媒体只使用 ordinary direct/STUN UDP 与 LiveKit SFU/UDP；全 UDP 封锁、只允许标准 HTTPS 出站等网络在有界窗口后明确失败。不得用 TCP 探测、客户端声称的 NAT 类型或另一套应用 route candidate 猜测可达性。若真实矩阵证明该失败边界需要扩展，先比较 LiveKit 内建 ICE/TCP 或其内部安全 relay transport，并另行接受安全、端口、延迟和部署合同。 @@ -95,7 +95,7 @@ SFU 的 publication、subscription、server-resource accounting、临时 overlap ### 房间模式与生命周期 -- Given 当前空闲房间码池，When 连续创建房间，Then 每间房从 `1000..9999` 的未占用集合随机取得一个四位码，不重复活动号码，`MAX_ROOMS` 不超过 9000 且默认 1000；释放房间后号码返回同一空闲池。 +- Given 当前空闲房间码池，When 连续创建房间，Then 每间房从 `1000..9999` 的未占用集合随机取得一个四位码，不重复活动号码，活动房间容量固定为 9000；释放房间后号码返回同一空闲池。 - Given Host 正在分享，When `ROOM_LEASE_SECONDS` 到达，Then 活动房间不被回收。Given 分享者明确停止、捕获轨结束或 Host 断开，When 同一 Host token 在默认 86400 秒内重新分享，Then 同一进程内恢复原房间并重新续租；超过期限则关闭房间、释放号码并拒绝旧 Host token、grant 和房间密码。 - Given 分享期间只有 host 信令短暂断开，When 现有 P2P 媒体仍健康，Then 客户端继续播放媒体并恢复信令，而不是错误显示“已停止分享”。 - Given 应用以 production 模式启动但 `SITE_ACCESS_PASSWORD` 缺失，或配置仍提供已删除的 `ACCESS_PASSWORD`、`ROOM_DATABASE_PATH`、`ROOM_TTL_SECONDS`，When 应用启动，Then 配置校验明确失败，而不是兼容旧 gate、旧持久化模式或公开建房。 @@ -145,7 +145,7 @@ SFU 的 publication、subscription、server-resource accounting、临时 overlap - Given ADR-0004 的任一硬门槛只有在引入 Encoded Transform/DataChannel/WebCodecs 媒体、自定义拥塞控制、FEC/RTX、多树、综合评分、转码、codec ladder、浏览器 RTP 注入 hack 或放宽 endpoint capacity 后才能满足，When 结束 bounded spike，Then 放弃并删除该浏览器 relay 实验路径，保留标准 P2P 与整体路由模型已接受的 fallback；后续 native shared encode sender 仍按独立 ADR 推进，但不能据此把失败的 relay 拓扑判为通过。 - Given 新观看者会使预计总上行超过安全预算，When 其请求加入，Then 系统只选择整体路由模型已授权且满足配置容量与资源准入的路径；没有合格路线时明确等待或拒绝，而不是创建超过配置容量的 ordinary edge 或持续堆积发送队列。各 transport 的拥塞控制适应该路径，诊断证据不改变拓扑或合成全房降档。 - Given SFU 参与路线，When 计算准入、迁移 overlap 或失败恢复，Then 同时检查统一端点容量和独立的 server ingress/egress、SFU 资源预算；不得把普通端点 child 数当作 server cap，也不得借兜底形成无界整房 fanout。 -- Given 配置完整 LiveKit credential tuple，When `LIVEKIT_API_URL` 缺失或不是允许的 control-plane origin，或 `SFU_INGRESS_CAPACITY` / `SFU_EGRESS_CAPACITY` 缺失、非正整数或超出 JavaScript safe-integer 范围，Then 服务必须在启动时 fail closed。Given 配置有效，When startup listing 出现非 Screener room、遗留 room 无法删除、删除后仍存在，或 exact candidate room 无法显式创建，Then 应用不得签发 token 或接受可能重复计算的 SFU generation。Given 当前 reserved、committed 与 draining 的 ingress 或 egress 会越过任一部署容量，Then controller 必须在签发任何新 SFU token 前拒绝该 candidate，并保持旧 route 和计数不变。Given generation 被 commit、abort、timeout、stop、delete 或 Host participant 离线回收，When DeleteRoom 尚未成功或 absent readback 尚未完成，Then 它必须继续占用账本资源且旧 token 不能创建 room。 +- Given 配置完整 LiveKit credential tuple，When `LIVEKIT_API_URL` 缺失或不是允许的 control-plane origin，Then 服务必须在启动时 fail closed。Given 配置有效，When startup listing 出现非 Screener room、遗留 room 无法删除、删除后仍存在，或 exact candidate room 无法显式创建，Then 应用不得签发 token 或接受可能重复计算的 SFU generation。Given 当前 reserved、committed 与 draining 的 ingress 或 egress 会越过房号池及每房 Viewer 准入推导出的容量，Then controller 必须在签发任何新 SFU token 前拒绝该 candidate，并保持旧 route 和计数不变。Given generation 被 commit、abort、timeout、stop、delete 或 Host participant 离线回收，When DeleteRoom 尚未成功或 absent readback 尚未完成，Then 它必须继续占用账本资源且旧 token 不能创建 room。 - Given Host、relay parent 或 leaf 处于受限网络，When direct/peer 路径不足，Then 只执行 ADR-0005 授权的 SFU publisher/subscriber fallback；不得靠未授权的固定数量、静态 lease、TCP probe 或 claimed NAT type 猜测实现。 ### 质量和恢复 diff --git a/scripts/access-privacy-gate.ts b/scripts/access-privacy-gate.ts index ad4aa9c..e58d488 100644 --- a/scripts/access-privacy-gate.ts +++ b/scripts/access-privacy-gate.ts @@ -96,7 +96,7 @@ async function main(): Promise { nodeEnv: "production", port: appPort, listenHost: "127.0.0.1", publicBaseUrl: new URL(baseUrl), allowedOrigins: new Set([baseUrl]), siteAccessPassword: sitePassword, roomLeaseMs: 86_400_000, - maxRooms: 4, maxViewersPerRoom: 2, endpointMediaCopyCapacity: 2, + maxViewersPerRoom: 2, endpointMediaCopyCapacity: 2, peerAssistedMedia: false, stunUrls: [], }; server = await createScreenerServer({ config, staticDirectory: resolve(root, "dist/client") }); diff --git a/scripts/native-one-viewer-gate.ts b/scripts/native-one-viewer-gate.ts index a3ff772..3026a54 100644 --- a/scripts/native-one-viewer-gate.ts +++ b/scripts/native-one-viewer-gate.ts @@ -11,6 +11,7 @@ import WebSocket from "ws"; import { createScreenerServer, type ScreenerServer } from "../src/server/app"; import { loadConfig } from "../src/server/config"; import { + ROOM_CAPACITY, RoomStore, type ConnectedParticipant, type ConnectParticipantInput, @@ -381,7 +382,7 @@ async function main(): Promise { }); const roomStore = new ObservedRoomStore({ leaseMs: config.roomLeaseMs, - maxRooms: config.maxRooms, + maxRooms: ROOM_CAPACITY, maxViewersPerRoom: config.maxViewersPerRoom, }, () => senderLedger.record({ hostWssAuthenticated: true })); server = await createScreenerServer({ config, roomStore }); diff --git a/scripts/peer-assisted-benchmark.ts b/scripts/peer-assisted-benchmark.ts index c2c8655..58deb55 100644 --- a/scripts/peer-assisted-benchmark.ts +++ b/scripts/peer-assisted-benchmark.ts @@ -21,6 +21,7 @@ import { } from "../src/server/app"; import { loadConfig } from "../src/server/config"; import { + ROOM_CAPACITY, RoomStore, } from "../src/server/room-store"; import { @@ -3042,7 +3043,7 @@ export async function main(): Promise { config: serverConfig, roomStore: new RoomStore({ leaseMs: serverConfig.roomLeaseMs, - maxRooms: serverConfig.maxRooms, + maxRooms: ROOM_CAPACITY, maxViewersPerRoom: serverConfig.maxViewersPerRoom, }), }); diff --git a/src/client/media/viewer-presentation.ts b/src/client/media/viewer-presentation.ts index 9395120..96fc3b7 100644 --- a/src/client/media/viewer-presentation.ts +++ b/src/client/media/viewer-presentation.ts @@ -246,6 +246,12 @@ export function reduceViewerPresentation( const revisionChanged = state.revision !== action.revision; const terminal = action.state === "failed"; const currentFrame = hasCurrentFrame(state); + const currentMedia = + terminal && state.media?.revision === action.revision + ? { ...state.media, framePresented: false } + : terminal + ? null + : state.media; return { ...state, revision: action.revision, @@ -253,7 +259,7 @@ export function reduceViewerPresentation( revision: action.revision, state: action.state, }, - media: terminal ? null : state.media, + media: currentMedia, autoplayBlockedGeneration: revisionChanged || terminal ? null : state.autoplayBlockedGeneration, @@ -319,8 +325,12 @@ export function reduceViewerPresentation( connection: "connected", media: { ...state.media, framePresented: true }, retainedFrame: false, + routeStatus: null, failure: - state.failure === "PLAYBACK_FAILED" ? null : state.failure, + state.failure === "PLAYBACK_FAILED" || + state.failure === "ROUTE_EXHAUSTED" + ? null + : state.failure, }; case "autoplay-blocked": if ( @@ -395,13 +405,21 @@ export function reduceViewerPresentation( return state; } const terminalRoute = action.failure === "ROUTE_EXHAUSTED"; + const terminalMedia = + terminalRoute && + action.revision !== undefined && + state.media?.revision === action.revision + ? { ...state.media, framePresented: false } + : terminalRoute + ? null + : state.media; return { ...state, revision: action.revision === undefined ? state.revision : Math.max(state.revision ?? 0, action.revision), - media: terminalRoute ? null : state.media, + media: terminalMedia, retainedFrame: state.retainedFrame || (terminalRoute && hasCurrentFrame(state)), autoplayBlockedGeneration: terminalRoute diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index e1bdc2a..d316cdf 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -1065,6 +1065,7 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { failure: "ROUTE_EXHAUSTED", revision: currentRouteRevision, }); + setFrameProofEpoch((current) => current + 1); return true; }, }, @@ -1211,11 +1212,13 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { revision: message.revision, connectionId: message.candidate.connectionId, }; - acceptAssignedRoute( - message.revision, - message.assignment.upstream, - "prepare", - ); + if (remoteMediaRef.current === null) { + acceptAssignedRoute( + message.revision, + message.assignment.upstream, + "prepare", + ); + } } if (message.phase === "active") { const samePeerUpstream = @@ -1273,6 +1276,9 @@ export function ViewerPage({ roomId, viewerGrant }: ViewerPageProps) { revision: message.revision, state: message.state, }); + if (message.state === "failed") { + setFrameProofEpoch((current) => current + 1); + } return; } if (message.type === "viewer-quality-evidence") { diff --git a/src/server/app.ts b/src/server/app.ts index eb8b8c2..c7f1f36 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -15,7 +15,7 @@ import type { SfuFallbackOptions } from "./hybrid-media-router.js"; import type { SfuTokenIssuer } from "./livekit-token.js"; import type { SfuRoomControl } from "./sfu-room-control.js"; import { SfuResourceAdmission } from "./sfu-resource-admission.js"; -import { RoomStore, RoomStoreError } from "./room-store.js"; +import { ROOM_CAPACITY, RoomStore, RoomStoreError } from "./room-store.js"; import { SignalingServer, type SignalingOptions } from "./signaling.js"; export interface CreateServerOptions { @@ -71,8 +71,8 @@ export async function createScreenerServer( maxViewersPerRoom: config.maxViewersPerRoom, }), admission: new SfuResourceAdmission({ - ingressCapacity: livekitFallback.ingressCapacity, - egressCapacity: livekitFallback.egressCapacity, + ingressCapacity: ROOM_CAPACITY, + egressCapacity: ROOM_CAPACITY * config.maxViewersPerRoom, }), roomControl: sfuRoomControl, }; @@ -81,7 +81,7 @@ export async function createScreenerServer( options.roomStore ?? new RoomStore({ leaseMs: config.roomLeaseMs, - maxRooms: config.maxRooms, + maxRooms: ROOM_CAPACITY, maxViewersPerRoom: config.maxViewersPerRoom, now, }); diff --git a/src/server/config.ts b/src/server/config.ts index 5d2a753..d98eae9 100644 --- a/src/server/config.ts +++ b/src/server/config.ts @@ -15,7 +15,6 @@ const MAX_SITE_ACCESS_PASSWORD_BYTES = 128; const MIN_LIVEKIT_API_SECRET_BYTES = 32; const DEFAULT_MAX_VIEWERS_PER_ROOM = 8; const DEFAULT_ROOM_LEASE_SECONDS = 86_400; -const MAX_ROOMS = 9_000; const VISIBLE_ASCII_PATTERN = /^[\x21-\x7e]+$/; const REMOVED_ENVIRONMENT_VARIABLES = [ "TURN_URLS", @@ -31,7 +30,6 @@ const REMOVED_ENVIRONMENT_VARIABLES = [ "PEER_ASSISTED_ROOM_IDS", "HOST_ADMISSION_PASSWORD", "MAX_PEER_RELAY_DOWNSTREAM_EDGES", - "MAX_SFU_ROOTS_PER_ROOM", "ROOM_DATABASE_PATH", "ROOM_TTL_SECONDS", ] as const; @@ -41,8 +39,6 @@ export interface LiveKitFallbackConfig { apiUrl: string; apiKey: string; apiSecret: string; - ingressCapacity: number; - egressCapacity: number; } export interface ServerConfig { @@ -53,7 +49,6 @@ export interface ServerConfig { allowedOrigins: ReadonlySet; siteAccessPassword?: string; roomLeaseMs: number; - maxRooms: number; maxViewersPerRoom: number; peerAssistedMedia: boolean; endpointMediaCopyCapacity: number; @@ -90,20 +85,6 @@ function parsePositiveInteger( return parsed; } -function parseRequiredPositiveInteger( - value: string | undefined, - name: string, -): number { - if (value === undefined || value === "") { - throw new Error(`${name} must be a positive integer`); - } - const parsed = Number(value); - if (!Number.isSafeInteger(parsed) || parsed <= 0) { - throw new Error(`${name} must be a positive integer`); - } - return parsed; -} - function parseBoundedInteger( value: string | undefined, fallback: number, @@ -139,18 +120,7 @@ function parseLiveKitFallback( const apiKey = environment.LIVEKIT_API_KEY?.trim() || undefined; const apiSecret = environment.LIVEKIT_API_SECRET?.trim() || undefined; const configuredValues = [url, apiUrl, apiKey, apiSecret].filter(Boolean).length; - const ingressCapacity = environment.SFU_INGRESS_CAPACITY?.trim() || undefined; - const egressCapacity = environment.SFU_EGRESS_CAPACITY?.trim() || undefined; - const configuredCapacities = [ingressCapacity, egressCapacity].filter( - Boolean, - ).length; - if (configuredValues === 0) { - if (configuredCapacities > 0) { - throw new Error( - "SFU_INGRESS_CAPACITY and SFU_EGRESS_CAPACITY require LiveKit fallback", - ); - } return undefined; } if (configuredValues !== 4) { @@ -158,11 +128,6 @@ function parseLiveKitFallback( "LIVEKIT_URL, LIVEKIT_API_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must be configured together", ); } - if (configuredCapacities !== 2) { - throw new Error( - "SFU_INGRESS_CAPACITY and SFU_EGRESS_CAPACITY must be configured with LiveKit fallback", - ); - } if (Buffer.byteLength(apiSecret!) < MIN_LIVEKIT_API_SECRET_BYTES) { throw new Error("LIVEKIT_API_SECRET must contain at least 32 bytes"); } @@ -224,14 +189,6 @@ function parseLiveKitFallback( apiUrl: parsedApiUrl.origin, apiKey: apiKey!, apiSecret: apiSecret!, - ingressCapacity: parseRequiredPositiveInteger( - ingressCapacity, - "SFU_INGRESS_CAPACITY", - ), - egressCapacity: parseRequiredPositiveInteger( - egressCapacity, - "SFU_EGRESS_CAPACITY", - ), }; } @@ -297,8 +254,6 @@ export function loadConfig( ? `${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` - : name === "MAX_SFU_ROOTS_PER_ROOM" - ? `${name} is no longer supported; use SFU_INGRESS_CAPACITY and SFU_EGRESS_CAPACITY` : `${name} is no longer supported; ordinary ICE accepts STUN_URLS only`, ); } @@ -413,13 +368,6 @@ export function loadConfig( DEFAULT_ROOM_LEASE_SECONDS, "ROOM_LEASE_SECONDS", ) * 1_000, - maxRooms: parseBoundedInteger( - environment.MAX_ROOMS, - 1_000, - "MAX_ROOMS", - 1, - MAX_ROOMS, - ), maxViewersPerRoom, peerAssistedMedia, endpointMediaCopyCapacity, diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index b54d26c..b7897f0 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -129,6 +129,13 @@ interface ChildOperation { current?: Attempt; } +interface DirectContinuation { + childSessionId: string; + sfuConnectionId: string; + publicationGeneration: string; + parentPeerIds: string[]; +} + export interface OperationSnapshot { childPeerId: string; childSessionId: string; @@ -203,6 +210,7 @@ export class RoomRouteController { private nextJoinOrder = 0; private paused = false; private readonly routeTimings = new Map(); + private readonly directContinuations = new Map(); constructor(private readonly options: ControllerOptions) { assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity); @@ -310,14 +318,22 @@ export class RoomRouteController { } const capacity = this.effectiveCapacity(input.effectiveDownstreamCapacity); const current = this.participants.get(input.peerId); + const released: Resource[] = []; + if ( + !current && + input.role === "viewer" && + this.operation?.reason === "direct-convergence" + ) { + released.push(...this.abortOperation(nowMs, "aborted")); + } if (current && current.role !== input.role) throw new Error("Route role cannot change"); if (current) { let changed = false; - const released: Resource[] = []; const previousSessionId = current.sessionId; if (previousSessionId !== input.sessionId) { const revisionBefore = this.revision; released.push(...this.abortOperation(nowMs, "stale")); + this.directContinuations.delete(input.peerId); current.sessionId = input.sessionId; const rebound = this.rebindCommittedSession(input.peerId, input.sessionId); released.push(...rebound.released); @@ -370,13 +386,14 @@ export class RoomRouteController { this.recordDemand(input.peerId, nowMs, "join"); } this.touchFacts(); - return []; + return released; } disconnectSession(peerId: string, sessionId: string): boolean { const participant = this.participants.get(peerId); if (!participant || participant.sessionId !== sessionId) return false; participant.sessionId = null; + this.directContinuations.delete(peerId); this.touchFacts(); this.debug("participant-disconnected", { participant: this.debugPeer(peerId), @@ -393,6 +410,7 @@ export class RoomRouteController { participant.sessionId = null; participant.departureConfirmed = true; participant.effectiveDownstreamCapacity = 0; + this.directContinuations.delete(peerId); if (nowMs !== undefined) { for (const childPeerId of this.childrenOf(peerId)) { this.recordDemand(childPeerId, nowMs, "parent-departed"); @@ -482,6 +500,7 @@ export class RoomRouteController { } if (edge.usable) { edge.usable = false; + this.directContinuations.delete(guard.childPeerId); child.blockedAtFactVersion = undefined; this.touchFacts(); child.failedTuple = { @@ -508,6 +527,7 @@ export class RoomRouteController { publication.connectionId !== guard.connectionId) return false; if (publication.usable) { publication.usable = false; + this.directContinuations.clear(); if (nowMs !== undefined) { for (const [childPeerId, edge] of this.upstreamByViewer) { if (edge.kind === "sfu") { @@ -612,20 +632,36 @@ export class RoomRouteController { for (let remaining = this.participants.size; remaining > 0; remaining -= 1) { const bootstrap = this.bootstrapForBlockedDemand(); - const childPeerId = bootstrap ?? this.selectNextChild(); + const routeChildPeerId = bootstrap ?? this.selectNextChild(); + const continuation = routeChildPeerId + ? undefined + : this.selectDirectContinuation(); + const childPeerId = routeChildPeerId ?? continuation?.childPeerId; if (!childPeerId) return { removedPeerIds, failedPeerIds, released }; const child = this.participants.get(childPeerId)!; - const reason = bootstrap - ? "sfu-bootstrap" - : (this.routeTimings.get(childPeerId)?.reason ?? - this.routeDemandReason(childPeerId)); + const reason: RouteDemandReason = continuation + ? "direct-convergence" + : bootstrap + ? "sfu-bootstrap" + : (this.routeTimings.get(childPeerId)?.reason ?? + this.routeDemandReason(childPeerId)); this.ensureDemand(childPeerId, nowMs, reason); - const candidates = this.buildCandidates(childPeerId, bootstrap !== undefined); + const candidates = continuation + ? [continuation.plan] + : this.buildCandidates( + childPeerId, + bootstrap !== undefined, + reason, + ); if (candidates.length === 0) { this.debug("operation-unavailable", { child: this.debugPeer(childPeerId), reason, }); + if (reason === "direct-convergence") { + this.directContinuations.delete(childPeerId); + continue; + } if (this.retireInvalidOperationEdge(childPeerId, released)) { this.revision = this.allocateRevision(); this.touchFacts(); @@ -650,7 +686,11 @@ export class RoomRouteController { baseRevision: this.revision, candidates, cursor: 0, - deadlineAtMs: nowMs + this.options.operationTimeoutMs, + deadlineAtMs: + nowMs + + (reason === "direct-convergence" + ? this.directHeadStartMs() + : this.options.operationTimeoutMs), builtAtFactVersion: this.factVersion, }; this.debug("operation-started", { @@ -781,6 +821,7 @@ export class RoomRouteController { if (!operation || operation.current || !this.cursorGuardMatches(guard, operation)) { return { accepted: false, released: validation.released, exhausted: validation.exhausted }; } + this.consumeDirectContinuationCandidate(operation); this.noteRejection(operation.childPeerId, bucket); this.debug("candidate-rejected", { child: this.debugPeer(operation.childPeerId), @@ -927,6 +968,7 @@ export class RoomRouteController { candidate: this.debugTuple(operation.current.tuple), revision: operation.current.revision, }); + this.consumeDirectContinuationCandidate(operation); this.noteRejection(operation.childPeerId, "candidate-failed"); } const validation = this.validateOrAdvance(nowMs, guard); @@ -946,6 +988,7 @@ export class RoomRouteController { cursor: this.operation.cursor, candidateCount: this.operation.candidates.length, }); + this.consumeDirectContinuationCandidate(this.operation); } const validation = this.validateOrAdvance(nowMs); return { accepted: validation.expired, exhausted: validation.exhausted, activeRevision: this.revision, released: validation.released }; @@ -964,6 +1007,7 @@ export class RoomRouteController { this.hostPublication = null; this.participants.clear(); this.routeTimings.clear(); + this.directContinuations.clear(); this.paused = true; return [...resources]; } @@ -987,6 +1031,10 @@ export class RoomRouteController { let operation = this.operation; if (!operation) return result; if (nowMs >= operation.deadlineAtMs) { + const directConvergence = operation.reason === "direct-convergence"; + if (directConvergence) { + this.consumeDirectContinuationCandidate(operation); + } if (operation.current) released.push(...reservationResources(operation.current.reservation)); const revisionAdvanced = Boolean(operation.current); if (operation.current) this.advanceActiveRevision(operation); @@ -994,25 +1042,30 @@ export class RoomRouteController { this.finishTiming( operation.childPeerId, nowMs, - factsChanged ? "waiting" : "failed", + directConvergence + ? this.currentFinalRoute(operation.childPeerId) + : factsChanged + ? "waiting" + : "failed", factsChanged ? "stale" : "operation-deadline", ); this.blockAndClear( operation, - !factsChanged, + !directConvergence && !factsChanged, released, revisionAdvanced, ); return { ...result, - exhausted: !factsChanged, - exhaustedChildPeerId: factsChanged - ? undefined - : operation.childPeerId, + exhausted: directConvergence ? undefined : !factsChanged, + exhaustedChildPeerId: + directConvergence || factsChanged + ? undefined + : operation.childPeerId, expired: true, }; } - if (this.advanceExpiredCandidateStage(operation, nowMs, released)) { + if (this.advanceExpiredDirectHeadStart(operation, nowMs, released)) { result.expired = true; } if (this.participants.get(operation.childPeerId)?.sessionId !== operation.childSessionId) { @@ -1040,6 +1093,9 @@ export class RoomRouteController { if (!guardFailed) { this.noteRejection(operation.childPeerId, "stale"); } + if (operation.reason === "direct-convergence") { + this.consumeDirectContinuationCandidate(operation); + } released.push(...reservationResources(attempt.reservation)); this.advanceActiveRevision(operation); operation.current = undefined; @@ -1051,15 +1107,23 @@ export class RoomRouteController { while (operation.cursor < operation.candidates.length && !this.candidateValid(operation.childPeerId, operation.candidates[operation.cursor]!)) { this.noteRejection(operation.childPeerId, "stale"); + if (operation.reason === "direct-convergence") { + this.consumeDirectContinuationCandidate(operation); + } operation.cursor += 1; this.clearCandidateTiming(operation.childPeerId); } if (operation.cursor < operation.candidates.length) return result; + const directConvergence = operation.reason === "direct-convergence"; const factsChanged = operation.builtAtFactVersion !== this.factVersion; this.finishTiming( operation.childPeerId, nowMs, - factsChanged ? "waiting" : "failed", + directConvergence + ? this.currentFinalRoute(operation.childPeerId) + : factsChanged + ? "waiting" + : "failed", factsChanged ? "stale" : this.routeTimings.get(operation.childPeerId)?.rejectionBucket === @@ -1069,11 +1133,13 @@ export class RoomRouteController { ); this.blockAndClear( operation, - !factsChanged, + !directConvergence && !factsChanged, released, this.revision !== activeRevisionAtStart, ); - result.exhausted = !factsChanged && !this.bootstrapForBlockedDemand(); + result.exhausted = directConvergence + ? undefined + : !factsChanged && !this.bootstrapForBlockedDemand(); if (result.exhausted) { result.exhaustedChildPeerId = operation.childPeerId; } @@ -1097,6 +1163,7 @@ export class RoomRouteController { parentPeerId: attempt.tuple.parentPeerId, parentSessionId, transport: "direct", connectionId: attempt.connectionId, usable: true, physicalActive: true, }); + this.directContinuations.delete(operation.childPeerId); } else { const generation = attempt.publicationGeneration!; if (attempt.tuple.publication !== "reuse") { @@ -1123,6 +1190,27 @@ export class RoomRouteController { publicationGeneration: generation, transport: "sfu", connectionId: attempt.connectionId, usable: true, physicalActive: true, resource: edge, }); + const remainingParentPeerIds = + operation.reason !== "sfu-bootstrap" && + operation.reason !== "direct-convergence" + ? operation.candidates + .slice(operation.cursor + 1) + .flatMap((candidate) => + candidate.tuple.kind === "peer" + ? [candidate.tuple.parentPeerId] + : [], + ) + : []; + if (remainingParentPeerIds.length > 0) { + this.directContinuations.set(operation.childPeerId, { + childSessionId: operation.childSessionId, + sfuConnectionId: attempt.connectionId, + publicationGeneration: generation, + parentPeerIds: [...new Set(remainingParentPeerIds)], + }); + } else { + this.directContinuations.delete(operation.childPeerId); + } } this.revision = attempt.revision; this.operation = undefined; @@ -1149,7 +1237,11 @@ export class RoomRouteController { return released; } - private buildCandidates(childPeerId: string, sfuOnly: boolean): CandidatePlan[] { + private buildCandidates( + childPeerId: string, + sfuOnly: boolean, + reason: RouteDemandReason, + ): CandidatePlan[] { const child = this.participants.get(childPeerId)!; const failedKey = child.failedTuple?.factVersion === this.factVersion ? child.failedTuple.key @@ -1161,18 +1253,31 @@ export class RoomRouteController { .sort((left, right) => this.depth(left.peerId) - this.depth(right.peerId) || this.remaining(right.peerId, childPeerId) - this.remaining(left.peerId, childPeerId) || compareParticipant(left, right)); - const candidates: CandidateTuple[] = []; - if (!sfuOnly) { - for (const parent of parents) candidates.push({ kind: "peer", parentPeerId: parent.peerId, transport: "direct" }); - } + const directCandidates: CandidateTuple[] = parents.map((parent) => ({ + kind: "peer", + parentPeerId: parent.peerId, + transport: "direct", + })); + let sfuCandidate: Extract | undefined; if (this.options.sfuEnabled) { if (this.hostPublication?.usable && this.hostPublication.physicalActive) { - candidates.push({ kind: "sfu", publication: "reuse" }); + sfuCandidate = { kind: "sfu", publication: "reuse" }; } else if (this.hostPublication || this.publicationFits(childPeerId, false)) { const publication = this.hostPublication ? "replace" : "create"; - candidates.push({ kind: "sfu", publication }); + sfuCandidate = { kind: "sfu", publication }; } } + const candidates: CandidateTuple[] = sfuOnly + ? sfuCandidate + ? [sfuCandidate] + : [] + : sfuCandidate + ? reason === "join" && sfuCandidate.publication === "reuse" + ? [sfuCandidate, ...directCandidates] + : directCandidates.length > 0 + ? [directCandidates[0]!, sfuCandidate, ...directCandidates.slice(1)] + : [sfuCandidate] + : [...directCandidates, ...(sfuCandidate ? [sfuCandidate] : [])]; return candidates.filter((candidate, index, all) => tupleKey(candidate) !== failedKey && all.findIndex((other) => tupleKey(other) === tupleKey(candidate)) === index) .map((candidate) => this.planCandidate(childPeerId, candidate)) @@ -1198,6 +1303,80 @@ export class RoomRouteController { return unusable?.peerId ?? staleSfu?.peerId ?? viewers.find(({ peerId }) => !this.upstreamByViewer.has(peerId))?.peerId; } + private selectDirectContinuation(): { + childPeerId: string; + plan: CandidatePlan; + } | undefined { + for (const [childPeerId, continuation] of this.directContinuations) { + const child = this.participants.get(childPeerId); + const edge = this.upstreamByViewer.get(childPeerId); + if ( + !child?.sessionId || + child.departureConfirmed || + child.sessionId !== continuation.childSessionId || + edge?.kind !== "sfu" || + !edge.usable || + !edge.physicalActive || + edge.connectionId !== continuation.sfuConnectionId || + edge.publicationGeneration !== continuation.publicationGeneration || + !this.hostPublication?.usable || + !this.hostPublication.physicalActive || + this.hostPublication.generation !== continuation.publicationGeneration + ) { + this.directContinuations.delete(childPeerId); + continue; + } + while (continuation.parentPeerIds.length > 0) { + const parentPeerId = continuation.parentPeerIds[0]!; + const tuple: CandidateTuple = { + kind: "peer", + parentPeerId, + transport: "direct", + }; + const plan = this.planCandidate(childPeerId, tuple); + if ( + plan && + plan.endpointTransition.kind !== "bounded-gap" && + this.candidateValid(childPeerId, plan) + ) { + return { childPeerId, plan }; + } + continuation.parentPeerIds.shift(); + } + this.directContinuations.delete(childPeerId); + this.debug("direct-convergence-complete", { + child: this.debugPeer(childPeerId), + route: "sfu", + }); + } + return undefined; + } + + private directHeadStartMs(): number { + return Math.max(1, Math.floor(this.options.operationTimeoutMs / 2)); + } + + private consumeDirectContinuationCandidate( + operation: ChildOperation, + ): void { + const tuple = operation.current?.tuple ?? operation.candidates[operation.cursor]?.tuple; + if (operation.reason !== "direct-convergence" || tuple?.kind !== "peer") { + return; + } + const continuation = this.directContinuations.get(operation.childPeerId); + if (!continuation) return; + const index = continuation.parentPeerIds.indexOf(tuple.parentPeerId); + if (index >= 0) continuation.parentPeerIds.splice(index, 1); + this.debug("direct-convergence-advanced", { + child: this.debugPeer(operation.childPeerId), + parent: this.debugPeer(tuple.parentPeerId), + remaining: continuation.parentPeerIds.length, + }); + if (continuation.parentPeerIds.length === 0) { + this.directContinuations.delete(operation.childPeerId); + } + } + private bootstrapForBlockedDemand(): string | undefined { if (!this.options.sfuEnabled || this.hostPublication || this.hostHasPublicationSlot()) return undefined; const demand = this.availableViewers(true).find((viewer) => viewer.blockedAtFactVersion === this.factVersion); @@ -1353,6 +1532,7 @@ export class RoomRouteController { this.upstreamByViewer.delete(participant.peerId); this.participants.delete(participant.peerId); this.routeTimings.delete(participant.peerId); + this.directContinuations.delete(participant.peerId); removed.push(participant.peerId); changed = true; } @@ -1477,26 +1657,28 @@ export class RoomRouteController { connectionId: operation.current.connectionId } : undefined }; } - private advanceExpiredCandidateStage( + private advanceExpiredDirectHeadStart( operation: ChildOperation, nowMs: number, released: Resource[], ): boolean { - const plan = operation.current - ? { tuple: operation.current.tuple } - : operation.candidates[operation.cursor]; - if (!plan) { - return false; - } - const stage = candidateStage(plan.tuple); - const stageDeadlineAtMs = this.stageDeadlineAt(operation, stage); + const currentTuple = + operation.current?.tuple ?? operation.candidates[operation.cursor]?.tuple; + const fallbackTuple = operation.candidates[operation.cursor + 1]?.tuple; if ( - stageDeadlineAtMs >= operation.deadlineAtMs || - nowMs < stageDeadlineAtMs + operation.reason === "direct-convergence" || + operation.cursor !== 0 || + currentTuple?.kind !== "peer" || + (fallbackTuple?.kind !== "sfu" && !this.bootstrapCandidateAvailable()) ) { return false; } - const skippedFrom = operation.cursor; + const headStartDeadlineAtMs = + operation.deadlineAtMs - this.options.operationTimeoutMs + + this.directHeadStartMs(); + if (nowMs < headStartDeadlineAtMs) { + return false; + } const activeCandidate = operation.current ? this.debugTuple(operation.current.tuple) : null; @@ -1508,62 +1690,32 @@ export class RoomRouteController { operation.cursor += 1; this.clearCandidateTiming(operation.childPeerId); } - while ( - operation.cursor < operation.candidates.length && - candidateStage(operation.candidates[operation.cursor]!.tuple) === stage - ) { - operation.cursor += 1; - } - this.debug("candidate-stage-expired", { + this.debug("direct-head-start-expired", { child: this.debugPeer(operation.childPeerId), - stage, activeCandidate, - skippedFrom, - skippedTo: operation.cursor, - deadlineAtMs: stageDeadlineAtMs, + cursor: operation.cursor, + deadlineAtMs: headStartDeadlineAtMs, observedAtMs: nowMs, }); return true; } private operationWakeAt(operation: ChildOperation): number { - const tuple = + const currentTuple = operation.current?.tuple ?? operation.candidates[operation.cursor]?.tuple; - return tuple - ? this.stageDeadlineAt(operation, candidateStage(tuple)) - : operation.deadlineAtMs; - } - - private stageDeadlineAt( - operation: ChildOperation, - stage: CandidateStage, - ): number { - const stages = this.operationStages(operation); - const index = stages.indexOf(stage); - if (index === -1 || index === stages.length - 1) { - return operation.deadlineAtMs; - } - const startedAtMs = - operation.deadlineAtMs - this.options.operationTimeoutMs; - return ( - startedAtMs + - Math.floor( - (this.options.operationTimeoutMs * (index + 1)) / stages.length, - ) - ); - } - - private operationStages( - operation: ChildOperation, - ): CandidateStage[] { - const stages = candidateStages(operation.candidates); + const fallbackTuple = operation.candidates[operation.cursor + 1]?.tuple; if ( - !stages.includes("sfu") && - this.bootstrapCandidateAvailable() + operation.reason !== "direct-convergence" && + operation.cursor === 0 && + currentTuple?.kind === "peer" && + (fallbackTuple?.kind === "sfu" || this.bootstrapCandidateAvailable()) ) { - stages.push("sfu"); + return ( + operation.deadlineAtMs - this.options.operationTimeoutMs + + this.directHeadStartMs() + ); } - return stages; + return operation.deadlineAtMs; } private bootstrapCandidateAvailable(): boolean { @@ -1580,7 +1732,6 @@ export class RoomRouteController { return Boolean( participant?.sessionId && edge?.kind === "peer" && - edge.transport === "direct" && edge.usable && edge.physicalActive, ); @@ -1953,24 +2104,6 @@ function tupleKey(tuple: CandidateTuple): string { return tuple.kind === "peer" ? `peer:${tuple.parentPeerId}` : `sfu:${tuple.publication}`; } -type CandidateStage = "direct" | "sfu"; - -function candidateStage(tuple: CandidateTuple): CandidateStage { - if (tuple.kind === "sfu") { - return "sfu"; - } - return "direct"; -} - -function candidateStages(candidates: readonly CandidatePlan[]): CandidateStage[] { - const stages = new Set( - candidates.map((candidate) => candidateStage(candidate.tuple)), - ); - return (["direct", "sfu"] as const).filter((stage) => - stages.has(stage), - ); -} - function edgeTupleKey(edge: CommittedEdge): string { return edge.kind === "peer" ? `peer:${edge.parentPeerId}` : "sfu:reuse"; } diff --git a/src/server/room-store.ts b/src/server/room-store.ts index cd372db..812ded4 100644 --- a/src/server/room-store.ts +++ b/src/server/room-store.ts @@ -22,7 +22,7 @@ export type RoomStoreErrorCode = | "HOST_ALREADY_CONNECTED" | "ROOM_LIMIT"; -const ROOM_CODE_COUNT = 9_000; +export const ROOM_CAPACITY = 9_000; const ROOM_CODE_FIRST = 1_000; const VIEWER_PASSWORD_SALT_BYTES = 16; const VIEWER_PASSWORD_VERIFIER_BYTES = 32; @@ -155,7 +155,7 @@ export interface RoomStoreOptions { export class RoomStore { private readonly rooms = new Map(); private readonly freeRoomCodes = Array.from( - { length: ROOM_CODE_COUNT }, + { length: ROOM_CAPACITY }, (_, index) => (ROOM_CODE_FIRST + index).toString(), ); private readonly now: () => number; @@ -169,10 +169,10 @@ export class RoomStore { if ( !Number.isSafeInteger(options.maxRooms) || options.maxRooms <= 0 || - options.maxRooms > ROOM_CODE_COUNT + options.maxRooms > ROOM_CAPACITY ) { throw new Error( - `Room limit must be an integer between 1 and ${ROOM_CODE_COUNT}`, + `Room limit must be an integer between 1 and ${ROOM_CAPACITY}`, ); } if ( diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index abcfafd..bf45333 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -346,6 +346,7 @@ export const routeDemandReasonSchema = z.enum([ "parent-departed", "capacity-reduction", "sfu-bootstrap", + "direct-convergence", ]); export type RouteDemandReason = z.infer; diff --git a/tests/hybrid-media-router.test.ts b/tests/hybrid-media-router.test.ts index ee5171e..ea290a1 100644 --- a/tests/hybrid-media-router.test.ts +++ b/tests/hybrid-media-router.test.ts @@ -723,15 +723,32 @@ describe("HybridMediaRouter v9 runtime", () => { router.setViewerRelayCapacity(first, 1); const child = connectViewer(store, room, "sfu-peer-child"); complete(router, child); + await vi.waitFor(() => + expect(preparedFor(sent, child.sessionId)?.candidate.transport).toBe( + "sfu", + ), + ); + const sfuPrepare = preparedFor(sent, child.sessionId)!; + router.handleRouteReady(child, { + type: "route-ready", + revision: sfuPrepare.revision, + phase: "prepare", + }); + await vi.waitFor(() => + expect( + router.resolveActiveViewerMediaEdge(room.roomId, child.peerId) + ?.upstream, + ).toEqual({ kind: "sfu" }), + ); await vi.waitFor(() => expect(preparedFor(sent, child.sessionId)?.candidate.transport).toBe( "direct", ), ); - const childPrepare = preparedFor(sent, child.sessionId)!; + const directPrepare = preparedFor(sent, child.sessionId)!; router.handleRouteReady(child, { type: "route-ready", - revision: childPrepare.revision, + revision: directPrepare.revision, phase: "prepare", }); await vi.waitFor(() => diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts index 66b61d5..896ea4c 100644 --- a/tests/room-route-controller.test.ts +++ b/tests/room-route-controller.test.ts @@ -411,7 +411,7 @@ describe("RoomRouteController", () => { }, ); - it("keeps healthy branches while 18 silent Viewers fall back serially", () => { + it("admits waiting Viewers through existing SFU before direct convergence", () => { const routes = controller(2, { sfuEnabled: true }); addViewer(routes, B, 2); routes.hydrateHostPublication( @@ -437,22 +437,14 @@ describe("RoomRouteController", () => { let nowMs = 0; waiting.forEach((peerId, index) => { - const directOperation = routes.reconcile(nowMs).operation!; - expect(directOperation.childPeerId).toBe(peerId); - beginCandidate(routes, { - nowMs: nowMs + 1, - connectionId: `${peerId}_silent_direct`, - reservation: { kind: "direct" }, - }); - routes.operationExpired(directOperation.wakeAtMs); - - const sfuOperation = routes.snapshot().operation!; + const sfuOperation = routes.reconcile(nowMs).operation!; + expect(sfuOperation.childPeerId).toBe(peerId); expect(sfuOperation.candidates[sfuOperation.cursor]!.tuple).toEqual({ kind: "sfu", publication: "reuse", }); const prepared = beginCandidate(routes, { - nowMs: directOperation.wakeAtMs + 1, + nowMs: nowMs + 1, connectionId: `${peerId}_sfu`, reservation: { kind: "sfu-reuse", @@ -467,10 +459,10 @@ describe("RoomRouteController", () => { revision: prepared.current!.revision, connectionId: `${peerId}_sfu`, }, - directOperation.wakeAtMs + 2, + nowMs + 2, ).accepted, ).toBe(true); - nowMs = directOperation.wakeAtMs + 3; + nowMs += 3; expect(routes.snapshot().upstreamByViewer.get(B)).toMatchObject({ kind: "sfu", @@ -493,6 +485,9 @@ describe("RoomRouteController", () => { generation: "publication_generation", usable: true, }); + expect(routes.reconcile(nowMs).operation).toMatchObject({ + reason: "direct-convergence", + }); }); it.each([1, 2, 3] as const)( @@ -696,7 +691,7 @@ describe("RoomRouteController", () => { }); }); - it("reaches a healthy SFU reuse stage after silent peer transport", () => { + it("keeps SFU active while converging through an SFU-fed parent", () => { const routes = controller(2, { sfuEnabled: true }); addViewer(routes, B, 2); routes.hydrateHostPublication( @@ -716,31 +711,12 @@ describe("RoomRouteController", () => { routes.hydrateEdge(C, peerEdge(HOST, "c_from_host")); addViewer(routes, A, 0); - const operation = routes.reconcile(0).operation!; - expect(operation.candidates.map((candidate) => candidate.tuple)).toEqual( - expect.arrayContaining([ - { kind: "peer", parentPeerId: B, transport: "direct" }, - { kind: "sfu", publication: "reuse" }, - ]), - ); - - beginCandidate(routes, { - nowMs: 1, - connectionId: "silent_direct", - reservation: { kind: "direct" }, - }); - routes.operationExpired(operation.wakeAtMs); - - expect( - routes.snapshot().operation!.candidates[ - routes.snapshot().operation!.cursor - ]!.tuple, - ).toEqual({ + expect(routes.reconcile(0).operation?.candidates[0]?.tuple).toEqual({ kind: "sfu", publication: "reuse", }); const prepared = beginCandidate(routes, { - nowMs: 7_000, + nowMs: 1, connectionId: "a_sfu_reuse", reservation: { kind: "sfu-reuse", edge: "a_subscription" }, }).operation!; @@ -752,7 +728,7 @@ describe("RoomRouteController", () => { revision: prepared.current!.revision, connectionId: "a_sfu_reuse", }, - 8_000, + 2, ).accepted, ).toBe(true); expect(routes.snapshot().hostPublication).toMatchObject({ @@ -764,6 +740,144 @@ describe("RoomRouteController", () => { publicationGeneration: "publication_generation", usable: true, }); + const convergence = routes.reconcile(3).operation!; + expect(convergence).toMatchObject({ + childPeerId: A, + reason: "direct-convergence", + }); + expect(convergence.candidates[0]?.tuple).toEqual({ + kind: "peer", + parentPeerId: B, + transport: "direct", + }); + const direct = beginCandidate(routes, { + nowMs: 4, + connectionId: "a_from_b", + reservation: { kind: "direct" }, + }).operation!; + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + kind: "sfu", + connectionId: "a_sfu_reuse", + }); + expect(routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: direct.current!.revision, + connectionId: "a_from_b", + }, 5).accepted).toBe(true); + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + kind: "peer", + parentPeerId: B, + connectionId: "a_from_b", + }); + }); + + it("advances exact direct convergence failures without failing SFU", () => { + const routes = controller(2, { sfuEnabled: true }); + addViewer(routes, D, 0); + routes.hydrateEdge(D, peerEdge(HOST, "host_full")); + routes.hydrateHostPublication("publication_1", "publication_resource"); + for (const peerId of [B, C]) { + addViewer(routes, peerId, 1); + routes.hydrateEdge(peerId, { + kind: "sfu", + publicationGeneration: "publication_1", + transport: "sfu", + connectionId: `${peerId}_sfu`, + usable: true, + physicalActive: true, + resource: `${peerId}_subscription`, + }); + } + addViewer(routes, A, 0, 0); + routes.reconcile(0); + const sfu = beginCandidate(routes, { + nowMs: 1, + connectionId: "a_sfu", + reservation: { kind: "sfu-reuse", edge: "a_subscription" }, + }).operation!; + expect(routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: sfu.current!.revision, + connectionId: "a_sfu", + }, 2).accepted).toBe(true); + + const first = routes.reconcile(3).operation!; + expect(first.candidates[0]?.tuple).toMatchObject({ + kind: "peer", + parentPeerId: B, + }); + const failed = beginCandidate(routes, { + nowMs: 4, + connectionId: "a_from_b", + reservation: { kind: "direct" }, + }).operation!; + expect(routes.candidateFailed({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: failed.current!.revision, + connectionId: "a_from_b", + }, 5).exhausted).toBeUndefined(); + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + kind: "sfu", + connectionId: "a_sfu", + }); + expect(routes.reconcile(6).operation?.candidates[0]?.tuple).toMatchObject({ + kind: "peer", + parentPeerId: C, + }); + }); + + it("preempts direct convergence when another Viewer needs a route", () => { + const routes = controller(2, { sfuEnabled: true }); + addViewer(routes, B, 2); + routes.hydrateHostPublication("publication_1", "publication_resource"); + routes.hydrateEdge(B, { + kind: "sfu", + publicationGeneration: "publication_1", + transport: "sfu", + connectionId: "b_sfu", + usable: true, + physicalActive: true, + resource: "b_subscription", + }); + addViewer(routes, A, 0, 0); + routes.reconcile(0); + const sfu = beginCandidate(routes, { + nowMs: 1, + connectionId: "a_sfu", + reservation: { kind: "sfu-reuse", edge: "a_subscription" }, + }).operation!; + routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: sfu.current!.revision, + connectionId: "a_sfu", + }, 2); + expect(routes.reconcile(3).operation?.reason).toBe("direct-convergence"); + beginCandidate(routes, { + nowMs: 4, + connectionId: "a_direct_pending", + reservation: { kind: "direct", overlap: "a_overlap" }, + }); + + const released = routes.upsertParticipant({ + peerId: C, + role: "viewer", + sessionId: `${C}_session`, + effectiveDownstreamCapacity: 0, + }, 5); + expect(released).toEqual(["a_overlap"]); + expect(routes.snapshot().operation).toBeUndefined(); + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + kind: "sfu", + connectionId: "a_sfu", + }); + expect(routes.reconcile(6).operation).toMatchObject({ + childPeerId: C, + reason: "join", + }); }); it("reserves a stage for Host-full SFU bootstrap before it is a tuple", () => { @@ -949,7 +1063,7 @@ describe("RoomRouteController", () => { expect(blocked.reconcile(1).operation).toBeUndefined(); }); - it("uses an SFU-fed endpoint as an ordinary parent and aborts pause resources", () => { + it("aborts pending SFU resources on pause and repairs publication", () => { const routes = controller(1, { sfuEnabled: true }); addViewer(routes, A, 1); addViewer(routes, B, 0); @@ -965,16 +1079,15 @@ describe("RoomRouteController", () => { }); const operation = routes.reconcile(0).operation!; expect(operation.candidates[0]?.tuple).toEqual({ - kind: "peer", - parentPeerId: A, - transport: "direct", + kind: "sfu", + publication: "reuse", }); beginCandidate(routes, { nowMs: 1, connectionId: "b_candidate", - reservation: { kind: "direct" }, + reservation: { kind: "sfu-reuse", edge: "b_subscription" }, }); - expect(routes.setPaused(true)).toEqual([]); + expect(routes.setPaused(true)).toEqual(["b_subscription"]); expect(routes.snapshot().operation).toBeUndefined(); expect(routes.snapshot().upstreamByViewer.get(A)?.connectionId).toBe("a_sfu"); routes.setPaused(false); diff --git a/tests/room-store.test.ts b/tests/room-store.test.ts index f21bb55..05eec70 100644 --- a/tests/room-store.test.ts +++ b/tests/room-store.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from "vitest"; -import { RoomStore, RoomStoreError } from "../src/server/room-store.ts"; +import { + ROOM_CAPACITY, + RoomStore, + RoomStoreError, +} from "../src/server/room-store.ts"; interface MutableClock { nowMs: number; @@ -52,15 +56,15 @@ function expectRoomError(run: () => unknown, code: string) { describe("RoomStore", () => { it("allocates every free four-digit room code and recycles releases", async () => { const { store: roomStore } = store({ - maxRooms: 9_000, + maxRooms: ROOM_CAPACITY, random: (size) => Buffer.alloc(size), }); const rooms = []; - for (let index = 0; index < 9_000; index += 1) { + for (let index = 0; index < ROOM_CAPACITY; index += 1) { rooms.push(await roomStore.createRoom()); } - expect(new Set(rooms.map((room) => room.roomId)).size).toBe(9_000); + expect(new Set(rooms.map((room) => room.roomId)).size).toBe(ROOM_CAPACITY); expect(rooms.every((room) => /^[1-9]\d{3}$/.test(room.roomId))).toBe(true); await expect(roomStore.createRoom()).rejects.toEqual( new RoomStoreError("ROOM_LIMIT"), @@ -86,8 +90,8 @@ describe("RoomStore", () => { }); it("rejects room limits beyond the four-digit code space", () => { - expect(() => store({ maxRooms: 9_001 }).store).toThrow( - "Room limit must be an integer between 1 and 9000", + expect(() => store({ maxRooms: ROOM_CAPACITY + 1 }).store).toThrow( + `Room limit must be an integer between 1 and ${ROOM_CAPACITY}`, ); }); diff --git a/tests/server-config.test.ts b/tests/server-config.test.ts index 9698821..89c5d6c 100644 --- a/tests/server-config.test.ts +++ b/tests/server-config.test.ts @@ -5,8 +5,6 @@ import { loadConfig } from "../src/server/config.ts"; const liveKitAdmission = { LIVEKIT_API_URL: "https://livekit-api.test", - SFU_INGRESS_CAPACITY: "4", - SFU_EGRESS_CAPACITY: "16", } as const; describe("server configuration", () => { @@ -30,8 +28,6 @@ describe("server configuration", () => { LIVEKIT_API_URL: " http://livekit.test:7880 ", LIVEKIT_API_KEY: " test-key ", LIVEKIT_API_SECRET: ` ${"s".repeat(32)} `, - SFU_INGRESS_CAPACITY: " 4 ", - SFU_EGRESS_CAPACITY: " 16 ", }); expect(config.livekitFallback).toEqual({ @@ -39,35 +35,9 @@ describe("server configuration", () => { apiUrl: "http://livekit.test:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - ingressCapacity: 4, - egressCapacity: 16, }); }); - it.each(["1", "9007199254740991"])( - "accepts an explicit positive safe SFU capacity of %s", - (capacity) => { - const config = loadConfig({ - PEER_ASSISTED_MEDIA: "true", - LIVEKIT_URL: "wss://livekit.test", - LIVEKIT_API_URL: "https://livekit-api.test", - LIVEKIT_API_KEY: "test-key", - LIVEKIT_API_SECRET: "s".repeat(32), - SFU_INGRESS_CAPACITY: capacity, - SFU_EGRESS_CAPACITY: capacity, - }); - - expect(config.livekitFallback?.ingressCapacity).toBe(Number(capacity)); - expect(config.livekitFallback?.egressCapacity).toBe(Number(capacity)); - }, - ); - - it("rejects SFU capacity without the LiveKit credential tuple", () => { - expect(() => loadConfig({ SFU_INGRESS_CAPACITY: "2" })).toThrow( - "require LiveKit fallback", - ); - }); - it("requires peer-assisted media for LiveKit fallback", () => { expect(() => loadConfig({ @@ -142,8 +112,6 @@ describe("server configuration", () => { LIVEKIT_API_URL: apiUrl, LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), - SFU_INGRESS_CAPACITY: "4", - SFU_EGRESS_CAPACITY: "16", }), ).toThrow("LIVEKIT_API_URL"); }); @@ -158,8 +126,6 @@ describe("server configuration", () => { LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), - SFU_INGRESS_CAPACITY: "4", - SFU_EGRESS_CAPACITY: "16", }; expect(() => loadConfig({ @@ -175,7 +141,7 @@ describe("server configuration", () => { ).toBe("http://127.0.0.1:7880"); }); - it("rejects a short LiveKit API secret and invalid SFU capacities", () => { + it("rejects a short LiveKit API secret", () => { const fallback = { LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", @@ -186,32 +152,6 @@ describe("server configuration", () => { expect(() => loadConfig({ ...fallback, LIVEKIT_API_SECRET: "too-short" }), ).toThrow("LIVEKIT_API_SECRET must contain at least 32 bytes"); - for (const capacity of ["0", "1.5", "9007199254740992"]) { - expect(() => - loadConfig({ - ...fallback, - SFU_EGRESS_CAPACITY: capacity, - }), - ).toThrow("SFU_EGRESS_CAPACITY"); - } - }); - - it("requires both SFU capacities with LiveKit and rejects the removed root key", () => { - const fallback = { - LIVEKIT_URL: "wss://livekit.test", - LIVEKIT_API_URL: "https://livekit-api.test", - LIVEKIT_API_KEY: "test-key", - LIVEKIT_API_SECRET: "s".repeat(32), - }; - expect(() => loadConfig(fallback)).toThrow( - "must be configured with LiveKit fallback", - ); - expect(() => - loadConfig({ ...fallback, SFU_INGRESS_CAPACITY: "2" }), - ).toThrow("must be configured with LiveKit fallback"); - expect(() => loadConfig({ MAX_SFU_ROOTS_PER_ROOM: "" })).toThrow( - "MAX_SFU_ROOTS_PER_ROOM is no longer supported", - ); }); it.each([ @@ -461,13 +401,6 @@ describe("server configuration", () => { ); }); - it("bounds MAX_ROOMS to the four-digit code space", () => { - expect(loadConfig({ MAX_ROOMS: "9000" }).maxRooms).toBe(9_000); - expect(() => loadConfig({ MAX_ROOMS: "9001" })).toThrow( - "MAX_ROOMS must be between 1 and 9000", - ); - }); - it.each(["ROOM_DATABASE_PATH", "ROOM_TTL_SECONDS"] as const)( "rejects removed room persistence configuration even when %s is blank", (name) => { diff --git a/tests/server-http.test.ts b/tests/server-http.test.ts index dacc01a..42c2f79 100644 --- a/tests/server-http.test.ts +++ b/tests/server-http.test.ts @@ -11,6 +11,7 @@ import { type ScreenerServer, } from "../src/server/app.ts"; import type { ServerConfig } from "../src/server/config.ts"; +import { RoomStore } from "../src/server/room-store.ts"; import { FakeSfuRoomControl } from "./fake-sfu-room-control.ts"; const allowedOrigin = "http://allowed.test"; @@ -30,7 +31,6 @@ function testConfig(overrides: Partial = {}): ServerConfig { allowedOrigins: new Set([allowedOrigin]), siteAccessPassword, roomLeaseMs: 86_400_000, - maxRooms: 10, maxViewersPerRoom: 8, peerAssistedMedia: false, endpointMediaCopyCapacity: 2, @@ -43,7 +43,11 @@ async function start( config = testConfig(), options: Pick< CreateServerOptions, - "now" | "siteAccessTtlSeconds" | "sfuTokenIssuer" | "sfuRoomControl" + | "now" + | "roomStore" + | "siteAccessTtlSeconds" + | "sfuTokenIssuer" + | "sfuRoomControl" > = {}, ): Promise { const sfuRoomControl = @@ -558,7 +562,14 @@ describe("room HTTP API", () => { it("returns service unavailable at the global room bound", async () => { const baseUrl = await start( - testConfig({ siteAccessPassword: undefined, maxRooms: 1 }), + testConfig({ siteAccessPassword: undefined }), + { + roomStore: new RoomStore({ + leaseMs: 86_400_000, + maxRooms: 1, + maxViewersPerRoom: 8, + }), + }, ); const create = () => createRoom(baseUrl); @@ -577,8 +588,6 @@ describe("server HTTP listener and health", () => { apiUrl: "http://livekit.test:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - ingressCapacity: 4, - egressCapacity: 16, }, }), { @@ -608,8 +617,6 @@ describe("server HTTP listener and health", () => { apiUrl: "http://127.0.0.1:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - ingressCapacity: 4, - egressCapacity: 16, }, }), { @@ -640,8 +647,6 @@ describe("server HTTP listener and health", () => { apiUrl: "http://127.0.0.1:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - ingressCapacity: 4, - egressCapacity: 16, }, }), serveFrontend: false, @@ -688,8 +693,6 @@ describe("server HTTP listener and health", () => { apiUrl: "http://127.0.0.1:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - ingressCapacity: 4, - egressCapacity: 16, }, }), serveFrontend: false, @@ -724,8 +727,6 @@ describe("server HTTP listener and health", () => { apiUrl: "http://127.0.0.1:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - ingressCapacity: 4, - egressCapacity: 16, }, }), serveFrontend: false, @@ -782,8 +783,6 @@ describe("server HTTP listener and health", () => { apiUrl: "http://127.0.0.1:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - ingressCapacity: 4, - egressCapacity: 16, }, }), serveFrontend: false, diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index c8c76f4..12350f6 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -18,7 +18,11 @@ import { type ScreenerServer, } from "../src/server/app.ts"; import type { ServerConfig } from "../src/server/config.ts"; -import { RoomStore, type CreatedRoom } from "../src/server/room-store.ts"; +import { + ROOM_CAPACITY, + RoomStore, + type CreatedRoom, +} from "../src/server/room-store.ts"; const allowedOrigin = "http://allowed.test"; let runningServer: ScreenerServer | undefined; @@ -123,7 +127,6 @@ function testConfig(): ServerConfig { publicBaseUrl: new URL("https://share.example.test"), allowedOrigins: new Set([allowedOrigin]), roomLeaseMs: 86_400_000, - maxRooms: 10, maxViewersPerRoom: 8, peerAssistedMedia: false, endpointMediaCopyCapacity: 2, @@ -155,7 +158,7 @@ async function startHarness( config.maxViewersPerRoom = maxViewersPerRoom; const roomStore = new RoomStore({ leaseMs: config.roomLeaseMs, - maxRooms: config.maxRooms, + maxRooms: ROOM_CAPACITY, maxViewersPerRoom, now: overrides.now, }); diff --git a/tests/viewer-presentation.test.ts b/tests/viewer-presentation.test.ts index fced1e2..2b949c9 100644 --- a/tests/viewer-presentation.test.ts +++ b/tests/viewer-presentation.test.ts @@ -373,6 +373,19 @@ describe("Viewer presentation reducer", () => { hasRetainedFrame: true, failureCode: "ROUTE_EXHAUSTED", }); + + const stillPlaying = reduceViewerPresentation(failed, { + type: "frame-presented", + generation: 3, + revision: 4, + }); + expect(deriveViewerPresentation(stillPlaying)).toMatchObject({ + stage: "playing", + overlay: "none", + hasCurrentFrame: true, + hasRetainedFrame: false, + failureCode: null, + }); }); it("honors terminal access, Host pause, and typed route status priority", () => {