fix(route): preserve media while converging direct paths

This commit is contained in:
AAA committed 2026-08-26 01:21:34 +08:00
1 parent d55b089e7b
commit f87c2ba0ed
25 files changed
+606 -367

No files matched your search

+3 -7
View File
@@ -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
@@ -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
+30 -21
View File
@@ -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
+5 -11
View File
@@ -238,7 +238,6 @@ PUBLIC_BASE_URL=https://share.example.com
ALLOWED_ORIGINS=https://share.example.com
SITE_ACCESS_PASSWORD=<INDEPENDENT_8_TO_128_BYTE_ACCESS_KEY>
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=<GENERATED_LIVEKIT_API_KEY>
LIVEKIT_API_SECRET=<INDEPENDENT_SECRET_OF_AT_LEAST_32_BYTES>
SFU_INGRESS_CAPACITY=<MEASURED_DEPLOYMENT_INGRESS_COPIES>
SFU_EGRESS_CAPACITY=<MEASURED_DEPLOYMENT_EGRESS_COPIES>
```
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
+37 -3
View File
@@ -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)
+20 -6
View File
@@ -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.
+3 -3
View File
@@ -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,也不唤醒未等待资源的房间。
+3 -3
View File
@@ -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 猜测实现。
### 质量和恢复
+1 -1
View File
@@ -96,7 +96,7 @@ async function main(): Promise<void> {
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") });
+2 -1
View File
@@ -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<void> {
});
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 });
+2 -1
View File
@@ -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<number> {
config: serverConfig,
roomStore: new RoomStore({
leaseMs: serverConfig.roomLeaseMs,
maxRooms: serverConfig.maxRooms,
maxRooms: ROOM_CAPACITY,
maxViewersPerRoom: serverConfig.maxViewersPerRoom,
}),
});
+21 -3
View File
@@ -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
+11 -5
View File
@@ -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") {
+4 -4
View File
@@ -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,
});
-52
View File
@@ -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<string>;
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,
+234 -101
View File
@@ -129,6 +129,13 @@ interface ChildOperation<Resource> {
current?: Attempt<Resource>;
}
interface DirectContinuation {
childSessionId: string;
sfuConnectionId: string;
publicationGeneration: string;
parentPeerIds: string[];
}
export interface OperationSnapshot {
childPeerId: string;
childSessionId: string;
@@ -203,6 +210,7 @@ export class RoomRouteController<Resource = unknown> {
private nextJoinOrder = 0;
private paused = false;
private readonly routeTimings = new Map<string, RouteTimingRecord>();
private readonly directContinuations = new Map<string, DirectContinuation>();
constructor(private readonly options: ControllerOptions) {
assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity);
@@ -310,14 +318,22 @@ export class RoomRouteController<Resource = unknown> {
}
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
}
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
this.hostPublication = null;
this.participants.clear();
this.routeTimings.clear();
this.directContinuations.clear();
this.paused = true;
return [...resources];
}
@@ -987,6 +1031,10 @@ export class RoomRouteController<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
);
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
.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<CandidateTuple, { kind: "sfu" }> | 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<Resource = unknown> {
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<Resource>,
): 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<Resource = unknown> {
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<Resource = unknown> {
connectionId: operation.current.connectionId } : undefined };
}
private advanceExpiredCandidateStage(
private advanceExpiredDirectHeadStart(
operation: ChildOperation<Resource>,
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<Resource = unknown> {
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<Resource>): 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<Resource>,
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<Resource>,
): 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<Resource = unknown> {
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<Resource>(edge: CommittedEdge<Resource>): string {
return edge.kind === "peer" ? `peer:${edge.parentPeerId}` : "sfu:reuse";
}
+4 -4
View File
@@ -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<string, Room>();
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 (
+1
View File
@@ -346,6 +346,7 @@ export const routeDemandReasonSchema = z.enum([
"parent-departed",
"capacity-reduction",
"sfu-bootstrap",
"direct-convergence",
]);
export type RouteDemandReason = z.infer<typeof routeDemandReasonSchema>;
+19 -2
View File
@@ -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(() =>
+156 -43
View File
@@ -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);
+10 -6
View File
@@ -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}`,
);
});
+1 -68
View File
@@ -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) => {
+14 -15
View File
@@ -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> = {}): 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<string> {
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,
+6 -3
View File
@@ -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,
});
+13
View File
@@ -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", () => {