From 28827ec6a2f7e5d86aefe715716530f1e5710460 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sun, 27 Sep 2026 17:14:28 +0800 Subject: [PATCH] fix(media): improve sharing continuity and pause recovery (#431) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem and outcome / 问题与结果 Sharing could stall at Browser encoder handoff or retain a pause after a video binding was replaced. Source/audio changes could also overlap offer/answer negotiation, while Native recovery could request redundant keyframes. This phase repairs those existing owners, drains capture output during bounded shutdown, and normalizes configured Site origins consistently with Browser serialization. ## Release notes ### 中文 这是一次稳定性更新，重点改善分享中的画面连续性，以及暂停、换源和恢复播放。 - 修复浏览器编码复用中的重复帧率限制和交接等待，减少由这些问题造成的卡顿。 - 修复换源和音频变更时的协商重叠，并防止已关闭连接的异步操作干扰恢复。 - 修复房主暂停期间重连后无法恢复播放，以及编码复用回退后仍停在暂停状态的问题；正常取消播放不再被当作媒体故障。 - 减少原生采集的重复关键帧请求，并修复停止采集时输出未排空导致的退出等待问题。 - 修复部分国际化域名和 IPv6 站点地址的识别，并改善窄屏下分享设置项的文字排版。 升级 Piik App 或自部署 Server 后刷新页面即可，无需修改配置。部分设备和网络环境下报告的开播失败、偶发断流仍需现场日志确认，本次更新不代表所有此类反馈均已解决。 ### English This stability update improves picture continuity and recovery when pausing, changing sources, or resuming playback. - Fix duplicate frame-rate limiting and unnecessary handoff waits in Browser encoding reuse, reducing stalls caused by these issues. - Serialize negotiation across source and audio changes, and prevent asynchronous work on closed connections from interfering with recovery. - Fix playback remaining paused after reconnecting while the Host is paused, or after falling back from shared encoding. Normal playback cancellation is no longer treated as a media failure. - Avoid redundant Native capture keyframe requests and drain pending capture output during bounded shutdown. - Fix recognition of some internationalized domain names and IPv6 Site addresses, and improve sharing-setting labels on narrow screens. Update Piik App or your self-hosted Server, then refresh the page. No configuration changes are required. Some device- and network-specific reports of sharing startup failure or intermittent stream loss still need matching diagnostics; this update does not claim to resolve every reported case. ## Verification / 验证 - `npm run check`: 76 test files / 1,152 tests, TypeScript and production Web build passed. - Real Chrome VP8/H264 checks with two viewers passed live/paused profile changes, quiet-source recovery, source replacement and ordinary-sender fallback after producer failure. - New pause-owner regressions fail against the previous implementation; actual Viewer effects preserve same-binding local pause and resume replacement bindings. - Independent cumulative Browser and Native/config review found no confirmed release blocker. - `npm run check:go`: formatting, Go tests/vet, Windows/Linux builds and Windows native capture compile/probes passed. Darwin packaging and Go race detection run on the native CI runners. - Windows App H264 and VP8 Host gates passed decoded playback, live/paused quality changes, quiet/background-source recovery, source replacement, failure cleanup and subsequent sharing; experiment-owned processes, ports and profiles were cleaned up. - Documentation, repository hygiene and release-note extraction passed. ## Impact / 影响 Patch release v1.6.7. Browser/server v23, Native control v9, capture v7 and SQLite schema 2 remain unchanged. Existing WebRTC adaptation and route policy remain unchanged; no quality floor is introduced. Interaction and Windows x86 prototypes are excluded. Managed Server restart interrupts active media and preserves room authority. Physical macOS/Linux capture, broad public-network endurance and unmatched reporter environments remain outside local acceptance. --- docs/research/browser-local-encoding-pool.md | 93 ++++++++++++++- docs/research/native-client-lifecycle.md | 34 +++++- docs/standards/engineering.md | 3 + docs/standards/media-quality.md | 17 ++- docs/standards/presentation-lifecycle.md | 6 +- docs/todo.md | 34 +++--- go.mod | 2 +- internal/app/config/config.go | 6 +- internal/app/config/config_test.go | 21 ++-- internal/app/mediaedge/source.go | 7 +- internal/app/mediaedge/source_test.go | 32 ++++- internal/app/nativecapture/protocol_test.go | 2 +- internal/app/nativecapture/stream.go | 25 +++- internal/app/nativecapture/stream_test.go | 65 ++++++++++- internal/server/app/app_test.go | 47 ++++++++ internal/server/config/config.go | 66 +++++++++-- internal/server/config/config_test.go | 28 +++++ internal/server/config/local.go | 6 +- src/client/media/browser-encoding-pool.ts | 15 ++- src/client/media/quality.ts | 13 ++- src/client/media/video-frame-proof.ts | 9 -- src/client/media/viewer-playback.ts | 11 ++ src/client/pages/ViewerPage.tsx | 29 ++--- src/client/styles.css | 2 +- src/client/webrtc/host-peer.ts | 28 ++++- src/client/webrtc/viewer-peer.ts | 29 +++-- tests/browser-encoding-pool.test.ts | 84 +++++++++++++ tests/host-peer-tracks.test.ts | 117 ++++++++++++++++++- tests/quality.test.ts | 16 +++ tests/video-frame-proof.test.ts | 11 +- tests/viewer-peer-generation.test.ts | 57 ++++++++- tests/viewer-playback.test.ts | 33 +++++- 32 files changed, 831 insertions(+), 117 deletions(-) diff --git a/docs/research/browser-local-encoding-pool.md b/docs/research/browser-local-encoding-pool.md index 693741e..5e79bff 100644 --- a/docs/research/browser-local-encoding-pool.md +++ b/docs/research/browser-local-encoding-pool.md @@ -162,16 +162,107 @@ the same probe bundle; these are single-run observations, not averages. B retained 1080p throughout all four runs, with no reported connection failure. The pooled traces created one separate producer for constrained A, then rejoined the retained healthy producer; they did not show repeated producer churn. -The deeper VP8 dip remains a quality cost to investigate. These results do not +The deeper VP8 dip is a measured adaptation tradeoff, not by itself a confirmed +defect or a requirement to impose a resolution floor. These results do not establish the cause of an unknown reporter's blur or stream loss, nor test Native capture's separate keyframe-request path. Any further change must preserve healthy-child isolation and framework-owned adaptation. +A further control retained two separate producers from startup, disallowing +cross-member group compatibility only in an ignored copy of the same probe. +With the same VP8 source, audio and pulse schedule, A reached 720p at 4.7 seconds +and regained 1080p at 24.8 seconds; B stayed at 1080p and both producers retained +their identity. No connection error was recorded. This supports a cold-producer +cost alongside ordinary adaptation, without establishing a population average. +Keeping every producer separate sacrifices sharing; reversing the handoff would +instead move a healthy child onto a cold encoder. Neither is an accepted fix. + +Further serial checks on 2026-09-27 found avoidable handoff latency: successful +preparation and budget updates waited for another 500 ms sample before +re-evaluating membership. A falling budget could repeatedly fail the comparison +against the previous applied limit while the old high-rate output kept sending. +Both completions now invoke the existing coordinator; a successfully written +recovery frame still commits selection. Sampling cadence, budget source and +healthy-child isolation are unchanged. + +With duplicate carrier filtering removed in both runs, the VP8/audio pulse's +cold-producer preparation-to-selection interval was 1,712 ms before this repair +and 67 ms afterward. A regained 1080p at 12.7 s and 9.2 s after pulse start +respectively; B stayed at 1080p. The final run had no queue overflow, backward +source IDs or missed played-audio pulses. These are single-run observations, +not an average recovery guarantee. Both reached 270p during adaptation. A final H264 +pulse reached 720p and regained 1080p at 6.7 s; B stayed at 1080p and neither +output overflowed its queue. + +The remaining dip is consistent with a new encoder adapting at an already-low +allocation. Pinned WebRTC's [initial size check](https://webrtc.googlesource.com/src/+/6f37672d358475cd17544121a12494da454d85fb/video/video_stream_encoder.cc#2535) +uses the native startup allocation separately from encoder overshoot correction; restoring balanced +restarts that initial check as described above. Local traces also show the +producer's corrected encoder target below its configured budget while local BWE +remains higher. The upstream [encoder bitrate adjuster](https://webrtc.googlesource.com/src/+/6f37672d358475cd17544121a12494da454d85fb/video/encoder_bitrate_adjuster.h) +uses conservative startup +utilization until it has enough frames. These are distinct mechanisms; the +current evidence does not isolate each one's contribution to the lowest size. +There is no verified API for transferring a mature encoder's adaptation history +to the new producer. A bitrate multiplier, longer protection interval or delayed +demand would change adaptation policy, rather than repair these completion bugs. + Budget attribution remains unchanged: targetBitrate is the encoder's allocated target, not raw link bandwidth ([upstream stats correction](https://webrtc.googlesource.com/src/+/fe25b0e928ea4e64aa134f5dc8012343320deec5%5E%21/)). Replacing it with availableOutgoingBitrate would bypass native allocation and protection. Keep producer, carrier, egress and decoded observations separate. +### Carrier Capture Constraints + +Serial Chrome 152 pulse checks on 2026-09-27 found that `HostPeer` applied real +capture constraints to the producer-driven canvas. Filtering those clock frames +can leave encoded data queued; reaching the four-frame bound then discards the +dependency chain and requests recovery. A second FPS limit on its synthetic +sender also filtered unevenly delivered clock frames. Track identity excludes +only the canvas from capture constraints, and carrier configuration removes the +duplicate sender FPS limit. The real producer retains the Host's picture limits; +outgoing bitrate remains bounded and ordinary fallback restores sender FPS. + +With the same two-child VP8/audio fixture and 5 Mbps ceiling, queue-overflow +recovery counts across each complete run were: + +| Source ceiling | Before, A / B | Without canvas capture constraints, A / B | +| --- | --- | --- | +| 30 fps | 36 / 31 | 6 / 1 | +| 60 fps | 84 / 75 | 52 / 38 | + +Removing only the sender FPS ceiling left 29 / 27 at 30 fps. Removing both +ceilings yielded 1 / 0. A separate 30-second healthy check at the 60 fps ceiling +isolated the second filter: with capture constraints already removed, retaining +the sender FPS limit produced 23 / 15 overflows; removing it produced 0 / 0. +Decoded cadence rose from about 41 / 42 to 45 / 45 fps in that fixture. A +three-second quiet interval followed by resumed production also had no overflow; +retained carrier age stayed below 49 ms. Supplemental clock draws were tested +but added no benefit over removing the duplicate limit, so no pump or timer was +added. The four-frame bound and dependency-chain recovery remain intact. + +These checks establish a redundant filtering cost, not lossless delivery under +all load. A 60 fps short-pulse run still recorded four overflows on constrained A +and none on B; both recovered, and B retained 1080p. They do not establish +sustained 60 fps or game quality. Cold-producer adaptation remains separate from +carrier recovery. Raw traces and ablation fixtures remain ignored artifacts. + +VP8 and H264 checks on the final implementation cover live 720p60 settings, +quiet-source 480p30 changes, pause/resume, source replacement and injected +producer failure. Both retained fresh output within the selected ceilings and +returned to ordinary sending on the existing connections. H264 source replacement +had an approximately 0.5 s picture transition: the matching audio pulse arrived, +but its brief picture marker did not. Ordinary frames resumed before the new +pool selected output, so that gap is not a producer-readiness wait. Earlier +fixtures also have source-transition gaps; these observations do not establish +gap-free replacement or a new regression. They cover bounded synthetic media +lifecycle, not real-game endurance or every physical device. + +The [canvas capture draft](https://w3c.github.io/mediacapture-fromelement/#html-canvas-element-media-capture-extensions) +models `requestFrame()` as a pending request, not a counted queue of clock ticks. +Do not infer one delivered carrier frame per call or enlarge the encoded queue +to hide lost timing. + ## Cost And Accounting Before the CPU-carrier refinement, a matched actual-product VP8 1080p30 comparison's ordinary pair made diff --git a/docs/research/native-client-lifecycle.md b/docs/research/native-client-lifecycle.md index f09fc6e..08303c3 100644 --- a/docs/research/native-client-lifecycle.md +++ b/docs/research/native-client-lifecycle.md @@ -272,6 +272,29 @@ all game/driver recovery or the separate immediate share-start failure report. Some focus changes did not remain foreground throughout a sampling window; only recorded window states and receiver progress establish those transitions. +## Keyframe Request Ownership + +A direct Native source delivers a requested physical output through its +coalescing callback, which can also wake a quiet capture. Repeating that PLI in +the next input's output plan duplicates delivery. A relay differs: its callback +requests upstream input, while its local derived encoder still needs the frame +plan. Callback-free producers also retain that plan; group activation and +handoff keep their separate recovery requests. + +A bounded Windows VP8 control-delivery comparison used the actual capture +helper's decoder, output worker and WebRTC encoder with 640x360 encoded input. +Three requests produced three recovery frames. Sending each request again +immediately after reading the next `FrameBegin`, without an inserted delay, +produced three pairs of consecutive recovery frames in both repetitions. +The direct-source duplicate is removed at request ownership; relay and manual +frame-plan delivery remain. This demonstrates duplicate encoded output, not +its network cost or the cause of reported blur/disconnection. Full App/relay +behavior was checked separately: the Windows App VP8 gate passed playback, +quiet-source restoration, profile changes and source replacement. The real +Native relay fixture retained shared/split/rejoined lower outputs and independent +retirement with one decoder. These bounded checks do not establish public-network +or other hardware acceptance. + ## Windows Capture Borders The accepted scope keeps WGC for Windows capture and previews. Piik does not @@ -297,9 +320,14 @@ The stream owner previously connected parent cancellation directly to `exec.CommandContext`, killing the child before its existing bounded stop sequence could release the platform session. Parent cancellation now enters that same `Stream.Close` sequence; an unresponsive process still has the existing -one-second kill deadline. Subprocess regression checks cover explicit stop, -parent cancellation, concurrent stop and an unresponsive child. A real WGC -capture cancelled through this Go owner exited normally and cleared the border. +one-second kill deadline. Retirement also drains stdout after any in-flight +frame finishes: a full output pipe must not prevent the producer from reading +its stop command. A controlled pipe check reproduced a responsive child being +killed without this drain. The stream joins the drain before closing completes; +[`exec.Cmd.Wait`](https://pkg.go.dev/os/exec#Cmd.StdoutPipe) closes the output +pipe when the process ends. Subprocess regression checks cover explicit stop, +parent cancellation, concurrent stop, blocked output and an unresponsive child. +The earlier real WGC cancellation check exited normally and cleared the border. This establishes a retirement defect and its local repair, not the cause of every reported Windows 11 border: denied consent and other active captures remain distinct Windows-owned limits. diff --git a/docs/standards/engineering.md b/docs/standards/engineering.md index c6b0619..6f671dd 100644 --- a/docs/standards/engineering.md +++ b/docs/standards/engineering.md @@ -145,6 +145,9 @@ above rather than searching the tree. Queued events retain physical-owner validity until the final serialized write; reused logical IDs do not identify a replacement instance. New subscriptions follow the old owner's retirement, including repeated cleanup callers. + Closing a Browser connection may leave its API promises pending. Its owner + cancels the corresponding waits so a serialized successor never depends on + completion from a retired connection; late results still need identity checks. Later cancellation must not erase a failure already observed before cleanup. A caller deadline does not cancel a dependency that ignores it. Keep late work bounded and serialize its cleanup; verify the dependency's actual lifetime. diff --git a/docs/standards/media-quality.md b/docs/standards/media-quality.md index 657bf53..91141e6 100644 --- a/docs/standards/media-quality.md +++ b/docs/standards/media-quality.md @@ -31,6 +31,9 @@ production behavior and remaining acceptance. - An audio-only update with the same capture video preserves the Browser video sender, encoding state and startup evidence on both direct and SFU paths. A new video source owns a new startup baseline; audio changes do not. + On Browser direct edges, source/audio renegotiation waits for the current + answer and coalesces newer intent into the next offer. A local operation queue + alone does not serialize the remote offer/answer exchange. - Windows native capture follows an explicitly stretched active display path for an entire display or a window covering that display, when the captured frame matches its desktop source dimensions. Other frames retain their own @@ -48,7 +51,9 @@ production behavior and remaining acceptance. sharing. Native screen capture can include system playback audio and native window capture can include selected-process audio when the platform exposes it. - Authoritative Browser pause disables the current source and sender-owned tracks - while retaining the room and established routes. Native pause keeps capture + while retaining the room and established routes. The connection owns its + encoded-output pause even when shared encoding is unavailable; the pool pauses + a shared producer only when all its members pause. Native pause keeps capture alive but stops session output through the same owner. Black frames, track mute, or network failure are not interpreted as a user pause. - The Host preview displays the capture stream directly and creates no Viewer or @@ -95,6 +100,11 @@ The three recommended profiles are ceilings, not delivery guarantees: maintain-framerate` remain independent controls. Display video uses the standard `contentHint = "motion"` for game motion. +Framework-driven downscaling under bandwidth or device pressure is valid within +the selected degradation preference. A low decoded resolution alone is not a +defect and does not justify a new quality floor. Review avoidable disruption, +recovery and healthy-child isolation instead of requiring a minimum observed size. + Current acceptance targets are: - on a controlled P2P path with RTT at most 40 ms and loss at most 1%, glass- @@ -167,6 +177,11 @@ and producer observations remain distinct from the tiny carrier's statistics. Producer startup protection begins at actual outgoing publication, excluding local warmup and paused frames. The synthetic carrier uses screen-content transport probing; the real producer keeps the Host's picture adaptation. +Capture constraints and the picture's sender FPS ceiling belong to the real +source and producer, not the producer-driven carrier clock. Outgoing transport +bitrate still follows the Host ceiling; ordinary fallback restores its picture +sender limits. Preparation and applied-budget completion re-evaluate pending +membership immediately; accepting a recovery frame still commits the handoff. The first eligible consumer follows the same path; unsupported APIs or failed pooling use ordinary senders. Prepared ordinary quality candidates, Native ingress and Browser SFU remain independent compositions. diff --git a/docs/standards/presentation-lifecycle.md b/docs/standards/presentation-lifecycle.md index 5c35540..4fe86bc 100644 --- a/docs/standards/presentation-lifecycle.md +++ b/docs/standards/presentation-lifecycle.md @@ -113,9 +113,13 @@ Identity, connection details and topology stay in the deck. Replacement video bindings do not consult Viewer-local pause: when the Host is playing, they attempt playback subject to browser autoplay permission. There is -no separate remembered playback intent for source changes or reconnects. +no separate remembered playback intent for source changes or reconnects. A new +binding received while the Host is paused waits for the Host to resume; it does +not inherit the previous binding's Viewer-local pause. Adding, replacing or removing audio on the same video binding updates only its audio consumer; it preserves current-frame proof and Viewer-local playback. +A pending `play()` canceled by pause or rebinding is not a media failure; +cancellation alone must not invalidate frame proof or disable local playback. The bar spans the screen's lower edge, preferring one row. Narrow screens shorten the volume slider and compact horizontal spacing and button widths, retaining diff --git a/docs/todo.md b/docs/todo.md index e63135a..8d2e1a0 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -7,25 +7,27 @@ history. A parked idea is not implementation authority. ## Now -- [ ] **Investigate Browser short-pulse quality cost.** Matched ordinary/pool - checks with two 1080p30 children, audio and a one-second bandwidth constraint - reproduce a deeper VP8 dip in the pool (270p versus 720p), with faster recovery. - H264 reaches 720p on both paths; unaffected children stay at 1080p. The - [pool evidence](./research/browser-local-encoding-pool.md#balanced-startup-and-recovery) - distinguishes one cold producer handoff from repeated churn. Assess a smaller - adaptation cost without a custom bitrate floor, delay policy or weakening - healthy-child isolation. These bounded Browser results do not identify the - unknown reporters' capture paths or establish their causes. -- [ ] **Passive App attachment: design hold.** Site mode authorizes one selected - origin and supplies native media without starting a local room server. A - passive replacement needs an accepted site-consent/discovery flow; it must not - admit arbitrary sites or add another runtime owner. Keep Site mode until that - decision; removal of the Demo prefill does not authorize changing site trust. +- [ ] **Published-product bug and experience acceptance.** Prioritize share + startup, audio/source changes, sustained viewing, recovery and retirement, + admission, and narrow/touch UI. Verify confirmed fixes through the existing + owners before resuming feature polish. Keep field reports without matched + evidence separate from locally reproduced defects. - [ ] **Complete manual accessible-name review.** Verify the empty video's screen-reader output: Chromium exposes an unavailable-media description despite the literal shared-picture label and no media error. Also review accessible naming on disabled tooltip wrappers and the UI catalogue's paired-character example; automated checks leave those for manual review. + +## Deferred Feature Work + +Feature design and interaction polish are paused while the bug/experience phase +is active. Retain these decisions for resumption, not as parallel work. + +- [ ] **Passive App attachment: design hold.** Site mode authorizes one selected + origin and supplies native media without starting a local room server. A + passive replacement needs an accepted site-consent/discovery flow; it must not + admit arbitrary sites or add another runtime owner. Keep Site mode until that + decision; removal of the Demo prefill does not authorize changing site trust. - [ ] **Refine the isolated room-interaction prototype.** Review chat, optional danmaku and participant-targeted reactions for usability, placement and intended cross-view synchronization. Preserve room authorization and media-route owners. @@ -188,6 +190,10 @@ not establish better connection success or speed; this note adds no retry policy evidence and capacity assertions with the current sender-owned quality contract; its old Viewer-only trigger is not a valid quality acceptance gate. 2. **Broader quality work.** Reopen from measured benefit at acceptable complexity. + The [Browser pool comparison](./research/browser-local-encoding-pool.md#balanced-startup-and-recovery) + records a cold-encoder adaptation tradeoff; a low resolution by itself is not + a defect or active repair target. Prioritize avoidable interruptions, failed + recovery or degradation of healthy siblings when evidence establishes them. Preserve chosen profiles, bitrate ceilings, endpoint capacity and P2P-first routing unless a new accepted decision supports changing them. No weighted score, all-pairs probes, periodic rebalancing, parent-wide prediction or diff --git a/go.mod b/go.mod index 0fdb5d8..ea4a352 100644 --- a/go.mod +++ b/go.mod @@ -34,6 +34,7 @@ require ( go.uber.org/zap v1.28.0 golang.org/x/crypto v0.54.0 golang.org/x/mod v0.38.0 + golang.org/x/net v0.57.0 golang.org/x/text v0.41.0 google.golang.org/protobuf v1.36.11 modernc.org/sqlite v1.58.0 @@ -120,7 +121,6 @@ require ( go.uber.org/zap/exp v0.3.0 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect golang.org/x/exp v0.0.0-20260727155853-b88d891fe743 // indirect - golang.org/x/net v0.57.0 // indirect golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/time v0.15.0 // indirect diff --git a/internal/app/config/config.go b/internal/app/config/config.go index 5393993..b220f38 100644 --- a/internal/app/config/config.go +++ b/internal/app/config/config.go @@ -96,7 +96,11 @@ func NormalizeSite(value string) (string, error) { parsed.RawQuery != "" || parsed.Fragment != "" { return "", errors.New("Piik Site must be an HTTP or HTTPS origin") } - return serverconfig.Origin(parsed), nil + origin := serverconfig.Origin(parsed) + if origin == "" { + return "", errors.New("Piik Site must have a valid host and port") + } + return origin, nil } func load(path string) (Config, error) { diff --git a/internal/app/config/config_test.go b/internal/app/config/config_test.go index 4fc79e5..75257ed 100644 --- a/internal/app/config/config_test.go +++ b/internal/app/config/config_test.go @@ -58,13 +58,15 @@ func TestSavePersistsOneNormalizedSiteChoice(t *testing.T) { func TestNormalizeSiteAcceptsOnlyAPlainWebOrigin(t *testing.T) { for value, want := range map[string]string{ - "https://share.example": "https://share.example", - "https://Share.Example:443/": "https://share.example", - "http://Share.Example:80": "http://share.example", - "https://share.example:0443": "https://share.example", - "http://192.168.1.4:8787/": "http://192.168.1.4:8787", - "http://share.example:443": "http://share.example:443", - "https://[::1]:443": "https://[::1]", + "https://share.example": "https://share.example", + "https://Share.Example:443/": "https://share.example", + "http://Share.Example:80": "http://share.example", + "https://share.example:0443": "https://share.example", + "http://192.168.1.4:8787/": "http://192.168.1.4:8787", + "http://share.example:443": "http://share.example:443", + "https://[::1]:443": "https://[::1]", + "https://bücher.example:443": "https://xn--bcher-kva.example", + "http://[0:0:0:0:0:0:0:1]:8787": "http://[::1]:8787", } { if got, err := NormalizeSite(value); err != nil || got != want { t.Fatalf("NormalizeSite(%q) = %q, %v; want %q", value, got, err, want) @@ -75,6 +77,11 @@ func TestNormalizeSiteAcceptsOnlyAPlainWebOrigin(t *testing.T) { "https://user:pass@share.example", "https://share.example/room", "https://share.example?mode=host", + "https://example.com:65536", + "https://[fe80::1%25eth0]", + "https://example：443", + "https://example／evil", + "https://\u00ad", } { if _, err := NormalizeSite(value); err == nil { t.Fatalf("NormalizeSite accepted %q", value) diff --git a/internal/app/mediaedge/source.go b/internal/app/mediaedge/source.go index ff0dac7..de2ef3d 100644 --- a/internal/app/mediaedge/source.go +++ b/internal/app/mediaedge/source.go @@ -608,7 +608,12 @@ func (source *Source) detach(edge *Edge) { } func (source *Source) requestLayerKeyFrame(layer int) { - source.groupRecovery.Or(uint32(1) << layer) + // Direct capture receives this request through its callback, including while + // quiet. A relay's callback only wakes upstream; its local encoders still + // need recovery in the next frame's plan, as do callback-free producers. + if source.relay != nil || source.requestKeyFrame == nil { + source.groupRecovery.Or(uint32(1) << layer) + } source.requestRecovery(layer) } diff --git a/internal/app/mediaedge/source_test.go b/internal/app/mediaedge/source_test.go index ba2faf6..7d70a28 100644 --- a/internal/app/mediaedge/source_test.go +++ b/internal/app/mediaedge/source_test.go @@ -44,8 +44,8 @@ func TestSourceRecoveryCoalescesTargetsWithoutWakingUnrelatedOutputs(t *testing. case <-time.After(time.Second): t.Fatal("lost output recovery request") } - if source.groupRecovery.Load() != 5 { - t.Fatal("wake callback consumed the frame-boundary recovery plan") + if source.groupRecovery.Load() != 0 { + t.Fatal("direct capture queued a second recovery through the frame plan") } select { case layers := <-requested: @@ -56,6 +56,31 @@ func TestSourceRecoveryCoalescesTargetsWithoutWakingUnrelatedOutputs(t *testing. } } +func TestSourceRecoveryRetainsLocalPlanWithoutDirectCaptureCallback(t *testing.T) { + for _, relay := range []bool{false, true} { + t.Run(fmt.Sprint(relay), func(t *testing.T) { + source := &Source{} + if relay { + source.relay = &relayDerivation{source: source} + source.requestKeyFrame = func([]int) {} + source.recoveryRequests = make(chan struct{}, 1) + } + source.requestLayerKeyFrame(0) + source.requestLayerKeyFrame(2) + if source.groupRecovery.Load() != 5 { + t.Fatal("local recovery was lost when the callback cannot reach its encoders") + } + if relay && source.pendingRecovery.Load() != 5 { + t.Fatal("relay recovery did not also request its upstream input") + } + source.RequestRecoveryFrame() + if source.groupRecovery.Load() != 5 || relay && source.pendingRecovery.Load() != allRecoveryLayers { + t.Fatal("upstream wake widened or lost the local recovery targets") + } + }) + } +} + func TestSourcePLIUsesPhysicalOutputIdentity(t *testing.T) { requested := make(chan []int, 16) source, err := (&Engine{}).NewSource("vp8", 2, 2, func(layers []int) { requested <- layers }) @@ -99,6 +124,9 @@ func TestSourcePLIUsesPhysicalOutputIdentity(t *testing.T) { case <-time.After(time.Second): t.Fatal("PLI did not wake its output") } + if source.groupRecovery.Load() != 0 { + t.Fatal("forwarded PLI would request the direct capture output twice") + } } } diff --git a/internal/app/nativecapture/protocol_test.go b/internal/app/nativecapture/protocol_test.go index 361b7ef..d386284 100644 --- a/internal/app/nativecapture/protocol_test.go +++ b/internal/app/nativecapture/protocol_test.go @@ -152,7 +152,7 @@ func TestOutputProfilesAndControlFramesShareOneBoundedContract(t *testing.T) { reader, writer := io.Pipe() done := make(chan error) close(done) - stream := &Stream{key: writer, done: done, outputs: outputs, cancel: func() {}} + stream := &Stream{input: io.NopCloser(bytes.NewReader(nil)), key: writer, done: done, outputs: outputs, cancel: func() {}} control := make(chan []byte, 1) go func() { payload, _ := io.ReadAll(reader) diff --git a/internal/app/nativecapture/stream.go b/internal/app/nativecapture/stream.go index af49cee..b8caf53 100644 --- a/internal/app/nativecapture/stream.go +++ b/internal/app/nativecapture/stream.go @@ -128,6 +128,7 @@ type Stream struct { controlMu sync.Mutex closed bool closeOnce sync.Once + readMu sync.Mutex readEnd sync.Once logger *slog.Logger ctx context.Context @@ -504,7 +505,9 @@ func logCaptureFailure(ctx context.Context, err error, exitCode int, stderr []by } func (stream *Stream) Read() (Frame, error) { + stream.readMu.Lock() frame, err := readFrame(stream.input) + stream.readMu.Unlock() if err != nil && stream.logger != nil { stream.readEnd.Do(func() { stream.logger.DebugContext(stream.ctx, "piik-client", "event", "capture-stream-ended", @@ -599,17 +602,27 @@ func (stream *Stream) Done() <-chan error { } func (stream *Stream) Close() error { - // Start the existing stop budget before waiting for a possibly blocked control write. - timer := time.AfterFunc(captureStopTimeout, stream.cancel) - defer timer.Stop() stream.closeOnce.Do(func() { + // Start the stop budget before waiting for a possibly blocked control write. + timer := time.AfterFunc(captureStopTimeout, stream.cancel) + defer timer.Stop() + // A retired consumer must not leave the producer blocked on stdout before + // it can read Q. Finish any in-flight frame before discarding the rest. + drained := make(chan struct{}) + go func() { + stream.readMu.Lock() + _, _ = io.Copy(io.Discard, stream.input) + stream.readMu.Unlock() + close(drained) + }() stream.controlMu.Lock() - defer stream.controlMu.Unlock() stream.closed = true _ = writeFrame(stream.key, Frame{Kind: FrameControl, Data: []byte("Q")}) _ = stream.key.Close() + stream.controlMu.Unlock() + <-stream.done + stream.cancel() + <-drained }) - <-stream.done - stream.cancel() return nil } diff --git a/internal/app/nativecapture/stream_test.go b/internal/app/nativecapture/stream_test.go index a5f0661..e59f58f 100644 --- a/internal/app/nativecapture/stream_test.go +++ b/internal/app/nativecapture/stream_test.go @@ -10,6 +10,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" ) @@ -26,6 +27,13 @@ func TestCaptureStopHelper(t *testing.T) { time.Sleep(10 * time.Second) os.Exit(3) } + if strings.HasPrefix(mode, "blocked-output-") { + // A producer can fill stdout before returning to its control reader. + if err := writeFrame(os.Stdout, Frame{Kind: FrameVP8, Width: 1280, Height: 720, + Duration: time.Second / 30, Data: make([]byte, 1024*1024)}); err != nil { + os.Exit(6) + } + } frame, err := readFrame(os.Stdin) if err != nil || frame.Kind != FrameControl || string(frame.Data) != "Q" { os.Exit(4) @@ -43,7 +51,8 @@ func TestCaptureStopUsesOneBoundedRetirementPath(t *testing.T) { if err != nil { t.Fatal(err) } - for _, mode := range []string{"explicit", "parent", "both", "unresponsive"} { + for _, mode := range []string{"explicit", "parent", "both", "unresponsive", + "blocked-output-explicit", "blocked-output-parent", "blocked-output-both"} { t.Run(mode, func(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() @@ -58,10 +67,11 @@ func TestCaptureStopUsesOneBoundedRetirementPath(t *testing.T) { if _, err := stream.Read(); err != nil { t.Fatal(err) } - if mode != "explicit" { + stop := strings.TrimPrefix(mode, "blocked-output-") + if stop != "explicit" { cancel() } - if mode == "explicit" || mode == "both" { + if stop == "explicit" || stop == "both" { go stream.Close() } select { @@ -81,6 +91,55 @@ func TestCaptureStopUsesOneBoundedRetirementPath(t *testing.T) { } } +func TestCaptureStopPreservesAnInFlightFrame(t *testing.T) { + input, output := io.Pipe() + control, commands := io.Pipe() + done := make(chan error) + var finishOnce sync.Once + finish := func() { + finishOnce.Do(func() { + _ = output.Close() + _ = input.Close() + _ = control.Close() + close(done) + }) + } + stream := &Stream{input: input, key: commands, done: done, cancel: finish} + t.Cleanup(func() { finish(); _ = stream.Close() }) + expected := Frame{Kind: FrameVP8, Width: 1280, Height: 720, + Duration: time.Second / 30, Data: bytes.Repeat([]byte{0x12}, 1024*1024)} + var encoded bytes.Buffer + if err := writeFrame(&encoded, expected); err != nil { + t.Fatal(err) + } + read := make(chan error, 1) + go func() { + frame, err := stream.Read() + if err == nil && (frame.Kind != expected.Kind || !bytes.Equal(frame.Data, expected.Data)) { + err = errors.New("retirement discarded part of the in-flight frame") + } + read <- err + }() + // Pipe.Write returns after Read has entered the frame, before its header is complete. + if _, err := output.Write(encoded.Bytes()[:1]); err != nil { + t.Fatal(err) + } + closed := make(chan struct{}) + go func() { _ = stream.Close(); close(closed) }() + frame, err := readFrame(control) + if err != nil || frame.Kind != FrameControl || string(frame.Data) != "Q" { + t.Fatalf("stop command = %+v, %v", frame, err) + } + if _, err := output.Write(encoded.Bytes()[1:]); err != nil { + t.Fatal(err) + } + if err := <-read; err != nil { + t.Fatal(err) + } + finish() + <-closed +} + func TestDiagnosticStderrOverflowDoesNotStopTheChildReader(t *testing.T) { for _, discard := range []bool{false, true} { buffer := &boundedBuffer{limit: 4, discardOverflow: discard} diff --git a/internal/server/app/app_test.go b/internal/server/app/app_test.go index bc398e6..1665cfa 100644 --- a/internal/server/app/app_test.go +++ b/internal/server/app/app_test.go @@ -632,6 +632,53 @@ func TestLocalPasswordUsesConfiguredDestinationCookieAcrossLANAndPublicLink(t *t } } +func TestConfiguredOriginMatchesBrowserSerialization(t *testing.T) { + for _, pair := range [][2]string{ + {"https://bücher.example:443", "https://xn--bcher-kva.example"}, + {"http://[0:0:0:0:0:0:0:1]:8787", "http://[::1]:8787"}, + } { + for _, explicit := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/explicit=%t", pair[0], explicit), func(t *testing.T) { + env := map[string]string{"PUBLIC_BASE_URL": pair[0], "ROOM_DATABASE_PATH": ":memory:", "STUN_URLS": ""} + if explicit { + env["ALLOWED_ORIGINS"] = pair[0] + } + configuration, err := config.Load(env) + if err != nil { + t.Fatal(err) + } + configuration.Port = 0 + configuration.ListenHost = "127.0.0.1" + server := start(t, Options{Config: configuration}) + for _, origin := range []string{pair[1], "https://foreign.test", pair[1] + "/path"} { + for _, path := range []string{"/api/rooms", "/signal"} { + request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(`{"codeEntryPolicy":"open"}`)) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Origin", origin) + if path == "/signal" { + request.Method = http.MethodGet + request.Header.Set("Connection", "Upgrade") + request.Header.Set("Upgrade", "websocket") + } + recorder := httptest.NewRecorder() + server.ServeHTTP(recorder, request) + want := http.StatusForbidden + if origin == pair[1] { + want = http.StatusCreated + if path == "/signal" { + want = http.StatusBadRequest + } // Missing WebSocket version, past origin admission. + } + if recorder.Code != want { + t.Errorf("%s origin=%q: %d, want %d (%s)", path, origin, recorder.Code, want, recorder.Body.String()) + } + } + } + }) + } + } +} + func TestSiteAccessRequiresAnAllowedOriginAndAValidLoginBody(t *testing.T) { server := start(t, Options{Config: testConfig(t)}) forbidden := `{"error":"Origin not allowed"}` diff --git a/internal/server/config/config.go b/internal/server/config/config.go index 3ec7d38..481c0da 100644 --- a/internal/server/config/config.go +++ b/internal/server/config/config.go @@ -13,6 +13,10 @@ import ( "strconv" "strings" + "golang.org/x/net/idna" + "golang.org/x/text/cases" + "golang.org/x/text/language" + "github.com/TNTcraftHIM/Piik/internal/server/protocol" ) @@ -358,7 +362,11 @@ func toOrigin(value string) (string, error) { if parsed.Scheme != "http" && parsed.Scheme != "https" { return "", errors.New("Allowed origins must use http or https") } - return Origin(parsed), nil + origin := Origin(parsed) + if origin == "" { + return "", errors.New("Allowed origins must have a valid host and port") + } + return origin, nil } // parseOriginURL validates an origin and returns the WHATWG-normalised URL @@ -379,7 +387,11 @@ func parseOriginURL(value, name, scheme, alternative string) (*url.URL, error) { return nil, fmt.Errorf( "%s must be an origin without credentials, path, query, or fragment", name) } - return &url.URL{Scheme: parsed.Scheme, Host: normalizedHost(parsed), Path: "/"}, nil + host := normalizedHost(parsed) + if host == "" { + return nil, fmt.Errorf("%s must have a valid host and port", name) + } + return &url.URL{Scheme: parsed.Scheme, Host: host, Path: "/"}, nil } // hasUserinfo ports `url.username || url.password`: WHATWG reports empty @@ -396,21 +408,55 @@ func hasUserinfo(parsed *url.URL) bool { // Origin ports the WHATWG URL `origin` getter for the http/https/ws/wss schemes // this package accepts: new URL("https://Example.com:443/").origin is // "https://example.com". It is exported so app and cmd can compare request -// origins without a second copy of the rule. +// origins without a second copy of the rule. Invalid hosts/ports return empty. func Origin(parsed *url.URL) string { - return parsed.Scheme + "://" + normalizedHost(parsed) + host := normalizedHost(parsed) + if host == "" { + return "" + } + return parsed.Scheme + "://" + host } -// normalizedHost lowercases the host, keeps an IPv6 literal bracketed and drops -// a port equal to the scheme default. Non-ASCII hosts are not IDNA-encoded, so -// they keep the form they were configured with. +// Match browser domain-to-ASCII rules; default Lookup also rejects underscores +// and hyphen positions that web origins permit. See WHATWG URL host parsing. +var originIDNA = idna.New(idna.MapForLookup(), idna.StrictDomainName(false), + idna.Transitional(false), idna.CheckHyphens(false), idna.BidiRule()) + +// normalizedHost owns configured host spelling for HTTP and local App trust. +// Request consumers still require the header to equal its canonical origin. func normalizedHost(parsed *url.URL) string { - host := strings.ToLower(parsed.Hostname()) - if strings.Contains(host, ":") { + host := parsed.Hostname() + if strings.HasPrefix(parsed.Host, "[") { + address, err := netip.ParseAddr(host) + if err != nil || !address.Is6() || address.Zone() != "" { + return "" + } + host = address.String() + if address.Is4In6() { + // Go prints a dotted tail; browsers serialize all IPv6 pieces as hex. + bytes := address.As16() + host = fmt.Sprintf("::ffff:%x:%x", uint16(bytes[12])<<8|uint16(bytes[13]), + uint16(bytes[14])<<8|uint16(bytes[15])) + } host = "[" + host + "]" + } else { + var err error + // Use full IDNA casing: strings.ToLower loses the dot in İ; the + // installed IDNA mapping table still maps capital ẞ to the domain ss. + // Final sigma handling is disabled as required for IDNA, not prose. + host = cases.Lower(language.Und, cases.HandleFinalSigma(false)).String(host) + host, err = originIDNA.ToASCII(host) + if err != nil || host == "" || strings.ContainsAny(host, "#/:<>?@[\\]^|%*") || + strings.IndexFunc(host, func(r rune) bool { return r <= 0x20 || r == 0x7f }) >= 0 { + return "" + } } // WHATWG parses the port as a number, so "0443" and "443" are both default. - if number, err := strconv.Atoi(parsed.Port()); err == nil { + if parsed.Port() != "" { + number, err := strconv.Atoi(parsed.Port()) + if err != nil || number < 0 || number > 65535 { + return "" + } if port := strconv.Itoa(number); port != defaultSchemePort(parsed.Scheme) { host += ":" + port } diff --git a/internal/server/config/config_test.go b/internal/server/config/config_test.go index cf4cb1b..c6a5ed3 100644 --- a/internal/server/config/config_test.go +++ b/internal/server/config/config_test.go @@ -538,6 +538,15 @@ func TestOrigin(t *testing.T) { {"http://localhost:9123", "http", "https", "http://localhost:9123"}, {"https://[2001:DB8::1]:8443", "http", "https", "https://[2001:db8::1]:8443"}, {"http://[::1]:7880", "http", "https", "http://[::1]:7880"}, + {"https://bücher.example:443", "http", "https", "https://xn--bcher-kva.example"}, + {"https://faß.de", "http", "https", "https://xn--fa-hia.de"}, + {"https://İ.example", "http", "https", "https://xn--i-9bb.example"}, + {"https://ẞ.example", "http", "https", "https://xn--zca.example"}, + {"https://example.ΟΣ", "http", "https", "https://example.xn--0xai"}, + {"https://EXAMPLE。com./", "http", "https", "https://example.com."}, + {"https://foo_bar.ab--cd.test", "http", "https", "https://foo_bar.ab--cd.test"}, + {"http://[0:0:0:0:0:0:0:1]:8787", "http", "https", "http://[::1]:8787"}, + {"http://[::ffff:192.0.2.1]:80", "http", "https", "http://[::ffff:c000:201]"}, } for _, testCase := range cases { t.Run(testCase.raw, func(t *testing.T) { @@ -554,3 +563,22 @@ func TestOrigin(t *testing.T) { }) } } + +func TestOriginConfigurationRejectsInvalidHostAndPort(t *testing.T) { + for _, raw := range []string{ + "https://example.com:65536", "https://[fe80::1%25eth0]", "https://[not-an-ip]", + "https://example／evil", "https://example：443", "https://example＠evil", + "https://a％2eb", "https://＊.example", "https://\u00ad", "https://a\u200db.test", + } { + t.Run(raw, func(t *testing.T) { + for _, key := range []string{"PUBLIC_BASE_URL", "ALLOWED_ORIGINS"} { + if _, err := Load(map[string]string{key: raw}); err == nil { + t.Errorf("%s accepted %q", key, raw) + } + } + if _, err := publicHTTPSOrigin(raw); err == nil { + t.Errorf("Local public origin accepted %q", raw) + } + }) + } +} diff --git a/internal/server/config/local.go b/internal/server/config/local.go index 28c40b1..8bec5b9 100644 --- a/internal/server/config/local.go +++ b/internal/server/config/local.go @@ -125,7 +125,11 @@ func publicHTTPSOrigin(value string) (*url.URL, error) { parsed.RawQuery != "" || parsed.Fragment != "" { return nil, errors.New("Local public origin must be an HTTPS origin") } - return &url.URL{Scheme: "https", Host: normalizedHost(parsed), Path: "/"}, nil + host := normalizedHost(parsed) + if host == "" { + return nil, errors.New("Local public origin must be an HTTPS origin") + } + return &url.URL{Scheme: "https", Host: host, Path: "/"}, nil } // Node's isIP(value) === 4 accepts only a dotted diff --git a/src/client/media/browser-encoding-pool.ts b/src/client/media/browser-encoding-pool.ts index ed48803..7fa1c06 100644 --- a/src/client/media/browser-encoding-pool.ts +++ b/src/client/media/browser-encoding-pool.ts @@ -59,7 +59,6 @@ export class BrowserEncodingPool { // A fresh connection has no useful raw output to preserve. Start its // tiny clock immediately, rather than warming two full encoders first. if (!member.carrier) encoded.passthrough(requestKey); - encoded.setPaused(member.paused); if (!this.timer) this.timer = setInterval(() => void this.poll(), BUDGET_SAMPLE_MS); return { carrierScale: () => { @@ -77,7 +76,6 @@ export class BrowserEncodingPool { if (member.disposed || member.paused === paused) return; member.paused = paused; member.previousOutput = undefined; - member.encoded.setPaused(paused); this.updatePauses(); }, metrics: (transport) => this.metrics(member, transport), @@ -271,7 +269,12 @@ export class BrowserEncodingPool { group.updating = group.producer.update(group.profile, budget).catch(() => { if (this.groups.has(group)) this.failSource(group.source); }) - .finally(() => { group.updating = undefined; }); + .finally(() => { + group.updating = undefined; + // A ready downgrade can use this applied budget now. Waiting for + // another sample compares the previous limit to falling demand again. + if (!this.disposed && this.groups.has(group)) this.reconcile(); + }); } } } @@ -290,7 +293,11 @@ export class BrowserEncodingPool { const group: Group = { producer, source: member.source, profile, codecKey: member.codecKey!, budget, ready: false, failed: false }; this.groups.add(group); debugEvent("encoding-pool", "producer-preparing", { producerId: producer.id, trackId: member.source.id, codec: member.codecKey, budget, profile }); - void producer.start(budget).then(() => { if (this.groups.has(group)) group.ready = true; }, (error) => { + void producer.start(budget).then(() => { + if (!this.groups.has(group)) return; + group.ready = true; + this.reconcile(); + }, (error) => { debugError("encoding-pool", "producer-start-failed", error, { producerId: producer.id }); if (this.groups.has(group)) { group.failed = true; this.failSource(group.source); } }); diff --git a/src/client/media/quality.ts b/src/client/media/quality.ts index bdb860e..59098e2 100644 --- a/src/client/media/quality.ts +++ b/src/client/media/quality.ts @@ -373,7 +373,7 @@ function sameParameter( export async function configureVideoSender( sender: RTCRtpSender, profile: QualityProfile, - scaleResolutionDownBy?: number, + carrierScale?: number, ): Promise { const parameters = sender.getParameters(); if (parameters.encodings.length === 0) { @@ -381,11 +381,14 @@ export async function configureVideoSender( } // Pinned LiveKit orders simulcast encodings by increasing spatial resolution. const encodingIndex = parameters.encodings.length - 1; - parameters.encodings[encodingIndex]!.maxBitrate = profile.maxBitrate; - parameters.encodings[encodingIndex]!.maxFramerate = profile.maxFramerate; + const encoding = parameters.encodings[encodingIndex]!; + encoding.maxBitrate = profile.maxBitrate; + // The producer owns picture cadence. A second limit on its synthetic clock + // can discard ticks and strand already-encoded frames when delivery is uneven. + if (carrierScale === undefined) encoding.maxFramerate = profile.maxFramerate; + else delete encoding.maxFramerate; if (parameters.encodings.length === 1) { - parameters.encodings[encodingIndex]!.scaleResolutionDownBy = - scaleResolutionDownBy ?? requestedScaleResolutionDownBy(sender, profile); + encoding.scaleResolutionDownBy = carrierScale ?? requestedScaleResolutionDownBy(sender, profile); } parameters.degradationPreference = profile.degradationPreference; diff --git a/src/client/media/video-frame-proof.ts b/src/client/media/video-frame-proof.ts index 7c395ae..bfba29f 100644 --- a/src/client/media/video-frame-proof.ts +++ b/src/client/media/video-frame-proof.ts @@ -9,15 +9,6 @@ type VideoWithFrameCallback = HTMLVideoElement & { webkitDecodedFrameCount?: number; }; -export function isAutoplayPolicyRejection(error: unknown): boolean { - return ( - typeof error === "object" && - error !== null && - "name" in error && - error.name === "NotAllowedError" - ); -} - export function observeCompositedVideoFrame( video: HTMLVideoElement, expectedStream: MediaStream, diff --git a/src/client/media/viewer-playback.ts b/src/client/media/viewer-playback.ts index 16739b1..cbee9e1 100644 --- a/src/client/media/viewer-playback.ts +++ b/src/client/media/viewer-playback.ts @@ -1,5 +1,12 @@ type ViewerPlaybackElement = Pick; +export function playbackFailure(error: unknown): "autoplay-blocked" | "playback-failed" | null { + const name = typeof error === "object" && error !== null && "name" in error ? error.name : undefined; + // pause()/load() cancel pending play promises; cancellation is not failed media. + if (name === "AbortError") return null; + return name === "NotAllowedError" ? "autoplay-blocked" : "playback-failed"; +} + export interface RemoteMediaBinding { stream: MediaStream; generation: number; @@ -28,7 +35,11 @@ export function prepareViewerPlayback( video: ViewerPlaybackElement, stream: MediaStream, paused: boolean, + hostPause: { active: boolean; resume: boolean }, ): boolean { + // A new picture must resume with the Host, not inherit the old local pause. + hostPause.active = paused; + hostPause.resume = paused; if (video.srcObject !== stream) { video.srcObject = stream; } diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index deb7765..575f616 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -88,11 +88,8 @@ import { type ViewerPresentationAction, type ViewerRouteKind, } from "../media/viewer-presentation"; -import { nextViewerMediaBinding, prepareViewerPlayback, type RemoteMediaBinding } from "../media/viewer-playback"; -import { - isAutoplayPolicyRejection, - observeCompositedVideoFrame, -} from "../media/video-frame-proof"; +import { nextViewerMediaBinding, playbackFailure, prepareViewerPlayback, type RemoteMediaBinding } from "../media/viewer-playback"; +import { observeCompositedVideoFrame } from "../media/video-frame-proof"; import { exactPeerSignalOwner } from "../media/route-transition"; import { ViewerSfuRoute } from "../media/viewer-sfu-route"; import type { @@ -417,23 +414,16 @@ export function ViewerPage({ }); }, (error: unknown) => { - if (!mediaBindingIsCurrent(binding)) { + const failure = playbackFailure(error); + if (!mediaBindingIsCurrent(binding) || !failure) { return; } invalidatePresentedMedia(); - if (isAutoplayPolicyRejection(error)) { - dispatchPresentation({ - type: "autoplay-blocked", - generation: binding.generation, - revision: binding.boundAtRevision, - }); - } else { - dispatchPresentation({ - type: "playback-failed", - generation: binding.generation, - revision: binding.boundAtRevision, - }); - } + dispatchPresentation({ + type: failure, + generation: binding.generation, + revision: binding.boundAtRevision, + }); }, ); } @@ -1883,6 +1873,7 @@ export function ViewerPage({ video, remoteMedia.stream, presentationState.host === "paused", + hostPlaybackPauseRef.current, ) ) { attemptPlayback(video, remoteMedia); diff --git a/src/client/styles.css b/src/client/styles.css index a230d24..f0ef1a5 100644 --- a/src/client/styles.css +++ b/src/client/styles.css @@ -1128,7 +1128,7 @@ html[data-theme="dark"] .lr-invite-url { background: #1a2634; border-color: #334 .lr-tile:is(:hover, :focus-visible):not(:disabled):not([aria-disabled="true"]):not(.is-selected) { background: var(--wall-2); transform: translateY(-2px); } .lr-tile:active:not(:disabled):not([aria-disabled="true"]) { transform: translateY(2px) scale(.97); box-shadow: none; } .lr-tile.is-selected { background: var(--ink); color: var(--paper); animation: lr-pop .2s var(--spring); } -.lr-tile small { font-family: var(--mono); font-size: 11px; letter-spacing: .02em; color: #5d7086; white-space: nowrap; } +.lr-tile small { max-width: 100%; font-family: var(--mono); font-size: 11px; letter-spacing: .02em; color: #5d7086; text-wrap: balance; overflow-wrap: anywhere; } .lr-tile.is-selected small { color: var(--paper); opacity: .75; } /* text inputs */ diff --git a/src/client/webrtc/host-peer.ts b/src/client/webrtc/host-peer.ts index ffe4df6..1b17f26 100644 --- a/src/client/webrtc/host-peer.ts +++ b/src/client/webrtc/host-peer.ts @@ -97,6 +97,7 @@ export class HostPeer { private profileRevision = 0; private negotiationEpoch = 0; private ordinaryAnswerEpoch: number | null = null; + private pendingNegotiation = false; private senderMutationTail: Promise = Promise.resolve(); private negotiationTail: Promise = Promise.resolve(); private startupVideoProfilePending: boolean; @@ -184,6 +185,7 @@ export class HostPeer { this.encodedOutput = new BrowserEncodingOutput(this.videoSender, () => { this.fail(); }, { connectionId: this.connectionId, peerId: this.peerId }); + this.encodedOutput.setPaused(this.paused); } this.attachVideoPool(this.stream.getVideoTracks()[0]!); this.audioTransceiver = this.connection.addTransceiver(audioTrack ?? "audio", { @@ -331,6 +333,7 @@ export class HostPeer { setPaused(paused: boolean): void { this.paused = paused; + this.encodedOutput?.setPaused(paused); this.pooledVideo?.setPaused(paused); this.applyPausedState(this.senderVideoTrack, this.audioSender?.track ?? null); this.applyPausedState(this.replacementVideoTrack, this.replacementAudioTrack); @@ -376,7 +379,7 @@ export class HostPeer { ) { return false; } - if (updateCaptureConstraints) { + if (updateCaptureConstraints && videoTrack !== this.encodedOutput?.track) { try { await this.waitForOperation(() => applyVideoCaptureProfile(videoTrack, profile)); } catch (error) { @@ -460,6 +463,7 @@ export class HostPeer { return this.enqueueNegotiation(async () => { if ( this.disposed || + this.ordinaryAnswerEpoch !== null || this.connection.signalingState !== "stable" ) { return false; @@ -506,6 +510,7 @@ export class HostPeer { this.lifetime.abort(); this.nextNegotiationEpoch(); this.ordinaryAnswerEpoch = null; + this.pendingNegotiation = false; if (this.statsTimer !== null) { window.clearInterval(this.statsTimer); this.statsTimer = null; @@ -574,7 +579,11 @@ export class HostPeer { const next = binding?.carrierScale() === undefined ? cloneSenderVideoTrack(source) : this.encodedOutput!.track; try { this.applyPausedState(next, null); - await this.waitForOperation(() => applyVideoCaptureProfile(next, this.desiredProfile)); + // The producer already limits real frames. Filtering its requested + // canvas clock frames again leaves encoded frames waiting for a tick. + if (next !== this.encodedOutput?.track) { + await this.waitForOperation(() => applyVideoCaptureProfile(next, this.desiredProfile)); + } if (!owns()) { if (next !== this.encodedOutput?.track) next.stop(); return false; } if (next !== previous) await this.waitForOperation(() => sender.replaceTrack(next)); await this.waitForOperation(() => configureVideoSender(sender, @@ -637,6 +646,13 @@ export class HostPeer { if (this.disposed) { return false; } + // Source/microphone changes may arrive before the current answer. Keep + // one offer in flight: another offer would let its predecessor's answer + // consume the newer epoch. The next offer reads the latest transceivers. + if (this.ordinaryAnswerEpoch !== null) { + this.pendingNegotiation = true; + return true; + } return this.createOwnedOffer(restart, this.nextNegotiationEpoch()); }); } @@ -713,12 +729,20 @@ export class HostPeer { await this.flushCandidates(); if (this.ownsAnswer(epoch)) { this.ordinaryAnswerEpoch = null; + if (this.pendingNegotiation) { + this.pendingNegotiation = false; + // Already inside negotiationTail; enqueueing here would deadlock. + if (!(await this.createOwnedOffer(false, this.nextNegotiationEpoch()))) { + this.fail(); + } + } } } catch (error) { if (!this.ownsAnswer(epoch)) { return; } this.ordinaryAnswerEpoch = null; + this.pendingNegotiation = false; throw error; } } diff --git a/src/client/webrtc/viewer-peer.ts b/src/client/webrtc/viewer-peer.ts index 45fcdd7..0c1cf5c 100644 --- a/src/client/webrtc/viewer-peer.ts +++ b/src/client/webrtc/viewer-peer.ts @@ -2,6 +2,7 @@ import type { CopyKey } from "../ui/copy"; import type { IceConfig, SignalPayload } from "../../shared/protocol"; import { debugError } from "../lib/debug"; import { observeDebugConnection } from "../lib/debug-webrtc"; +import { waitForConnectionOperation } from "./connection-operation"; import { EMPTY_METRICS, type PeerSnapshot, @@ -73,6 +74,7 @@ export interface ViewerMediaPeer { export class ViewerPeer implements ViewerMediaPeer { private connection: RTCPeerConnection | null = null; + private connectionLifetime: AbortController | null = null; private connectionId: string | null = null; private parentPeerId: string | null = null; private remoteStream = new MediaStream(); @@ -281,6 +283,7 @@ export class ViewerPeer implements ViewerMediaPeer { ), ); this.connection = connection; + this.connectionLifetime = new AbortController(); this.localIceCandidates = localIceCandidates; this.snapshot = { peerId: parentPeerId, @@ -338,7 +341,7 @@ export class ViewerPeer implements ViewerMediaPeer { readFramesDecoded: async () => { const track = this.remoteStream.getVideoTracks()[0]; return decodedVideoFrames( - await connection.getStats(), + await this.waitForOperation(connection, () => connection.getStats()), track ? { trackIdentifier: track.id } : null, ); }, @@ -363,7 +366,7 @@ export class ViewerPeer implements ViewerMediaPeer { ): Promise { const connectionId = payload.connectionId; try { - await connection.setRemoteDescription(payload.description); + await this.waitForOperation(connection, () => connection.setRemoteDescription(payload.description)); if (!this.isCurrentConnection(connection, connectionId)) { return; } @@ -371,11 +374,11 @@ export class ViewerPeer implements ViewerMediaPeer { if (!this.isCurrentConnection(connection, connectionId)) { return; } - const answer = preferScreenAudioStereo(await connection.createAnswer()); + const answer = preferScreenAudioStereo(await this.waitForOperation(connection, () => connection.createAnswer())); if (!this.isCurrentConnection(connection, connectionId)) { return; } - await connection.setLocalDescription(answer); + await this.waitForOperation(connection, () => connection.setLocalDescription(answer)); if (!this.isCurrentConnection(connection, connectionId)) { return; } @@ -434,7 +437,7 @@ export class ViewerPeer implements ViewerMediaPeer { const connection = this.connection; const connectionId = payload.connectionId; try { - await addRemoteIceCandidate(connection, payload.candidate); + await this.waitForOperation(connection, () => addRemoteIceCandidate(connection, payload.candidate)); } catch (error) { if (this.isCurrentConnection(connection, connectionId)) { this.setError(error, "host.fail.connection"); @@ -601,7 +604,7 @@ export class ViewerPeer implements ViewerMediaPeer { const candidates = this.pendingByConnection.get(connectionId) ?? []; this.pendingByConnection.delete(connectionId); for (const candidate of candidates) { - await addRemoteIceCandidate(connection, candidate); + await this.waitForOperation(connection, () => addRemoteIceCandidate(connection, candidate)); if (!this.isCurrentConnection(connection, connectionId)) { return; } @@ -621,11 +624,11 @@ export class ViewerPeer implements ViewerMediaPeer { this.statsInFlightConnection = connection; const statsAccumulator = this.statsAccumulator; try { - const metrics = await collectConnectionMetrics( + const metrics = await this.waitForOperation(connection, () => collectConnectionMetrics( connection, "receive", statsAccumulator, - ); + )); if (!this.isCurrentConnection(connection, connectionId)) { return; } @@ -642,6 +645,14 @@ export class ViewerPeer implements ViewerMediaPeer { } } + private waitForOperation(connection: RTCPeerConnection, operation: () => Promise): Promise { + const lifetime = this.connectionLifetime; + if (this.connection !== connection || !lifetime) { + return Promise.reject(new DOMException("Retired Viewer connection", "AbortError")); + } + return waitForConnectionOperation(lifetime.signal, operation); + } + private isCurrentConnection( connection: RTCPeerConnection, connectionId: string, @@ -709,6 +720,8 @@ export class ViewerPeer implements ViewerMediaPeer { } this.localIceCandidates?.discard(); this.localIceCandidates = null; + this.connectionLifetime?.abort(); + this.connectionLifetime = null; this.connection?.close(); this.connection = null; this.connectionId = null; diff --git a/tests/browser-encoding-pool.test.ts b/tests/browser-encoding-pool.test.ts index afdca05..85b8234 100644 --- a/tests/browser-encoding-pool.test.ts +++ b/tests/browser-encoding-pool.test.ts @@ -39,6 +39,16 @@ async function sample(count = 1) { await vi.advanceTimersByTimeAsync(500 * count); } +function holdNextProducerStart(): () => void { + let ready!: () => void; + const start = vi.mocked(BrowserEncodingProducer.prototype.start).getMockImplementation()!; + vi.mocked(BrowserEncodingProducer.prototype.start).mockImplementationOnce(async function (this: BrowserEncodingProducer, budget) { + await start.call(this, budget); + await new Promise((resolve) => { ready = resolve; }); + }); + return () => ready(); +} + beforeEach(() => { vi.useFakeTimers(); vi.stubGlobal("RTCRtpSender", { getCapabilities: () => ({ codecs: [codec] }) }); @@ -69,6 +79,32 @@ afterEach(() => { }); describe("Browser encoding pool ownership", () => { + it("pauses a shared producer only when all of its members pause", async () => { + const first = member(), second = member(); + await sample(4); + expect(producers.size).toBe(1); + const producer = [...producers.keys()][0]!; + + first.handle.setPaused(true); + expect(producer.setPaused).toHaveBeenLastCalledWith(false); + second.handle.setPaused(true); + expect(producer.setPaused).toHaveBeenLastCalledWith(true); + first.handle.setPaused(false); + expect(producer.setPaused).toHaveBeenLastCalledWith(false); + expect(first.producerId()).toBe(second.producerId()); + expect(producer.dispose).not.toHaveBeenCalled(); + }); + + it("selects a prepared producer without waiting for another budget sample", async () => { + const ready = holdNextProducerStart(); + const first = member(); + await sample(); + expect(first.select).not.toHaveBeenCalled(); + ready(); + await vi.advanceTimersByTimeAsync(0); + expect(first.producerId()).toBe([...producers.keys()][0]!.id); + }); + it("keeps a shared encoder through a bitrate spike with unchanged native demand", async () => { const first = member(), second = member(); await sample(4); @@ -143,4 +179,52 @@ describe("Browser encoding pool ownership", () => { retired!(); expect(begin).toHaveBeenCalledOnce(); }); + + it("rechecks a ready handoff after applying a falling budget, without another stats tick", async () => { + const strong = member(), weak = member(); + await sample(4); + const original = [...producers.keys()][0]!; + const ready = holdNextProducerStart(); + weak.demand.budget = 500_000; + await sample(); + const candidate = [...producers.keys()][1]!; + let complete!: () => void; + vi.mocked(candidate.update).mockImplementationOnce(async (_profile, budget) => { + await new Promise((resolve) => { complete = resolve; }); + producers.get(candidate)!.budget = budget; + }); + weak.demand.budget = 100_000; + await sample(); + ready(); + await vi.advanceTimersByTimeAsync(0); + expect(candidate.update).toHaveBeenCalledWith(profile, 100_000); + expect(weak.producerId()).toBe(original.id); + complete(); + await vi.advanceTimersByTimeAsync(0); + expect(weak.producerId()).toBe(candidate.id); + expect(strong.producerId()).toBe(original.id); + expect(producers.get(original)!.budget).toBe(2_000_000); + }); + + it.each(["failed", "retired"])("does not install a %s budget update's candidate", async (outcome) => { + member(); + const weak = member(); + await sample(4); + const ready = holdNextProducerStart(); + weak.demand.budget = 500_000; + await sample(); + const candidate = [...producers.keys()][1]!; + let resolve!: () => void, reject!: (error: Error) => void; + vi.mocked(candidate.update).mockImplementationOnce(() => new Promise((yes, no) => { resolve = yes; reject = no; })); + weak.demand.budget = 100_000; + await sample(); + ready(); + await vi.advanceTimersByTimeAsync(0); + const selections = weak.select.mock.calls.length; + if (outcome === "retired") { weak.handle.dispose(); resolve(); } + else reject(new Error("encoder settings failed")); + await vi.advanceTimersByTimeAsync(0); + expect(weak.select).toHaveBeenCalledTimes(selections); + expect(weak.fatal).not.toHaveBeenCalled(); + }); }); diff --git a/tests/host-peer-tracks.test.ts b/tests/host-peer-tracks.test.ts index 5511b50..99a9761 100644 --- a/tests/host-peer-tracks.test.ts +++ b/tests/host-peer-tracks.test.ts @@ -212,6 +212,7 @@ class FakePeerConnection { return; } this.localDescription = description as RTCSessionDescription; + this.signalingState = "have-local-offer"; } async setRemoteDescription( @@ -225,6 +226,7 @@ class FakePeerConnection { }); } this.remoteDescription = description as RTCSessionDescription; + this.signalingState = "stable"; } releaseDeferredRemoteDescription(): void { @@ -550,7 +552,67 @@ function encodedPeerFixture() { failAudio: () => audio.error(new Error("audio transform failed")) }; } -describe("HostPeer terminal media failure", () => { +describe("HostPeer encoded output ownership", () => { + it.each(["initial", "replacement"])("owns output pause when the pool declines the %s source", async (phase) => { + const fixture = encodedPeerFixture(); + const source = createTrack("video", "source"); + const peer = new HostPeer("child", { iceServers: [] }, createStream(source, null), + QUALITY_PROFILES["720p30"], { sendSignal: () => true, onUpdate() {} }, + VP8_ONLY_VIDEO_CODEC, undefined, false, fixture.pool); + try { + if (phase === "initial") { + peer.setPaused(true); + fixture.create.mockReturnValueOnce(null); + } + await peer.start(); + if (phase === "replacement") { + peer.setPaused(true); + fixture.create.mockReturnValueOnce(null); + expect(await peer.replaceStream(createStream(createTrack("video", "declined-source"), null))).toBe(true); + } + const sender = FakePeerConnection.latest!.senders[0]!; + expect(sender.track!.enabled).toBe(false); + expect(fixture.output.passthrough).toHaveBeenCalled(); + expect.soft(fixture.output.setPaused).toHaveBeenLastCalledWith(true); + + peer.setPaused(false); + expect(sender.track!.enabled).toBe(true); + expect(fixture.output.setPaused).toHaveBeenLastCalledWith(false); + expect(source.stop).not.toHaveBeenCalled(); + } finally { + peer.dispose(); + } + }); + + it("keeps capture constraints on real tracks across pool attachment, profile changes and fallback", async () => { + const fixture = encodedPeerFixture(); + const source = createTrack("video", "source"); + const peer = new HostPeer("child", { iceServers: [] }, createStream(source, null), + QUALITY_PROFILES["720p30"], { sendSignal: () => true, onUpdate() {} }, + VP8_ONLY_VIDEO_CODEC, undefined, false, fixture.pool); + await peer.start(); + await acceptPeerAnswer(peer); + const sender = FakePeerConnection.latest!.senders[0]!; + await peer.updateCaptureProfile(QUALITY_PROFILES["720p30"]); + expect(sender.track!.applyConstraints).toHaveBeenCalled(); + fixture.binding.carrierScale.mockReturnValue(1); + expect(await fixture.create.mock.calls[0]![5]()).toBe(true); + expect(sender.track).toBe(fixture.output.track); + for (const id of ["1080p60", "720p30"] as const) { + expect(await peer.updateCaptureProfile(QUALITY_PROFILES[id])).toBe(true); + expect(sender.getParameters().encodings[0]).not.toHaveProperty("maxFramerate"); + } + expect(fixture.output.track.applyConstraints).not.toHaveBeenCalled(); + fixture.binding.carrierScale.mockReturnValue(undefined); + expect(await fixture.create.mock.calls[0]![5]()).toBe(true); + expect(sender.track).not.toBe(source); + expect(sender.track).not.toBe(fixture.output.track); + expect(sender.track!.applyConstraints).toHaveBeenCalledWith(expect.objectContaining({ frameRate: { ideal: 30, max: 30 } })); + expect(sender.getParameters().encodings[0]!.maxFramerate).toBe(30); + peer.dispose(); + expect(source.stop).not.toHaveBeenCalled(); + }); + it.each(["video", "audio", "fallback", "rollback"])("reports %s failure once and keeps borrowed source tracks alive", async (kind) => { const fixture = encodedPeerFixture(); const source = createTrack("video", "source"), audio = createTrack("audio", "audio"); @@ -1082,6 +1144,7 @@ describe("HostPeer source replacement", () => { ); await expect(peer.start()).resolves.toBe(true); + await acceptPeerAnswer(peer); const connection = FakePeerConnection.latest!; expect(connection.transceiverInputs[1]?.trackOrKind).toBe("audio"); expect(connection.senders[1]?.track).toBeNull(); @@ -1109,6 +1172,7 @@ describe("HostPeer source replacement", () => { ); await expect(peer.start()).resolves.toBe(true); + await acceptPeerAnswer(peer); const connection = FakePeerConnection.latest!; await expect( peer.replaceStream( @@ -1229,6 +1293,57 @@ describe("HostPeer source replacement", () => { ); }); + it.each([false, true])("defers audio renegotiation until the current answer (restart=%s)", async (restart) => { + const video = createTrack("video", "shared-video"); + const peer = createPeer(createStream(video, null)); + const connection = FakePeerConnection.latest!; + const answer = (sdp: string) => peer.acceptSignal({ kind: "description", connectionId: peer.connectionId, + description: { type: "answer", sdp } }); + try { + await expect(peer.start()).resolves.toBe(true); + if (restart) { + await answer("initial-answer"); + await expect(peer.restartIce()).resolves.toBe(true); + } + const offers = connection.createOfferCallCount; + // Rapid microphone toggles coalesce, rather than superseding the offer. + for (const enabled of [true, false, true]) { + await expect(peer.replaceStream(createStream(video, + enabled ? createTrack("audio", "microphone") : null))).resolves.toBe(true); + } + expect(connection.createOfferCallCount).toBe(offers); + await expect(peer.restartIce()).resolves.toBe(false); + await answer("previous-answer-audio-inactive"); + expect(connection.createOfferCallCount).toBe(offers + 1); + expect(connection.transceivers[1]?.direction).toBe("sendonly"); + await answer("latest-answer-audio-recvonly"); + expect(connection.remoteDescription?.sdp).toBe("latest-answer-audio-recvonly"); + expect(connection.signalingState).toBe("stable"); + expect(connection.createOfferCallCount).toBe(offers + 1); + } finally { peer.dispose(); } + }); + + it.each(["dispose", "offer-failure", "signal-failure"] as const)("retires deferred renegotiation on %s", async (ending) => { + const video = createTrack("video", "shared-video"); + const sendSignal = vi.fn(() => true); + const onUpdate = vi.fn(); + const peer = new HostPeer("viewer", { iceServers: [] }, createStream(video, null), + QUALITY_PROFILES["720p30"], { sendSignal, onUpdate }); + const connection = FakePeerConnection.latest!; + await peer.start(); + await peer.replaceStream(createStream(video, createTrack("audio", "microphone"))); + if (ending === "dispose") peer.dispose(); + if (ending === "offer-failure") FakePeerConnection.offersFailing = 1; + if (ending === "signal-failure") sendSignal.mockReturnValue(false); + await peer.acceptSignal({ kind: "description", connectionId: peer.connectionId, + description: { type: "answer", sdp: "previous-answer" } }); + const failures = onUpdate.mock.calls.filter(([snapshot]) => snapshot.connectionState === "failed"); + expect(failures).toHaveLength(ending === "dispose" ? 0 : 1); + expect(connection.connectionState).toBe("closed"); + expect(connection.createOfferCallCount).toBe(ending === "dispose" ? 1 : 2); + peer.dispose(); + }); + it("retains restart candidates that arrive before the new answer", async () => { const peer = createPeer(createStream(createTrack("video", "video"), null)); await peer.start(); diff --git a/tests/quality.test.ts b/tests/quality.test.ts index 51797b4..67104a2 100644 --- a/tests/quality.test.ts +++ b/tests/quality.test.ts @@ -253,6 +253,22 @@ describe("realtime quality controls", () => { ]); }); + it("removes only the carrier's duplicate FPS ceiling and restores ordinary sender limits", async () => { + let applied = { encodings: [{ maxFramerate: 30 }] } as RTCRtpSendParameters; + const sender = { + getParameters: () => applied, + setParameters: async (parameters: RTCRtpSendParameters) => { applied = parameters; }, + } as unknown as RTCRtpSender; + const carrier = await configureVideoSender(sender, QUALITY_PROFILES["1080p60"], 1); + expect(applied.encodings[0]).not.toHaveProperty("maxFramerate"); + expect(carrier).toMatchObject({ requested: { maxFramerate: null, maxBitrate: 8_000_000 }, mismatches: [] }); + for (const id of ["720p30", "1080p60"] as const) { + const readback = await configureVideoSender(sender, QUALITY_PROFILES[id]); + expect(applied.encodings[0]!.maxFramerate).toBe(QUALITY_PROFILES[id].maxFramerate); + expect(readback.mismatches).toEqual([]); + } + }); + it("does not scale a remote source that already owns the room profile", async () => { let applied = { encodings: [{}] } as unknown as RTCRtpSendParameters; const sender = { diff --git a/tests/video-frame-proof.test.ts b/tests/video-frame-proof.test.ts index bb5d050..755e432 100644 --- a/tests/video-frame-proof.test.ts +++ b/tests/video-frame-proof.test.ts @@ -1,9 +1,6 @@ import { describe, expect, it, vi } from "vitest"; -import { - isAutoplayPolicyRejection, - observeCompositedVideoFrame, -} from "../src/client/media/video-frame-proof.ts"; +import { observeCompositedVideoFrame } from "../src/client/media/video-frame-proof.ts"; interface FakeVideo extends EventTarget { srcObject: MediaStream | null; @@ -28,12 +25,6 @@ function fakeVideo(stream: MediaStream): FakeVideo { } describe("composited video frame proof", () => { - it("recognizes only browser autoplay policy rejection", () => { - expect(isAutoplayPolicyRejection({ name: "NotAllowedError" })).toBe(true); - expect(isAutoplayPolicyRejection({ name: "AbortError" })).toBe(false); - expect(isAutoplayPolicyRejection(new Error("not allowed"))).toBe(false); - }); - it("binds requestVideoFrameCallback proof to the expected srcObject", () => { const stream = {} as MediaStream; const replacement = {} as MediaStream; diff --git a/tests/viewer-peer-generation.test.ts b/tests/viewer-peer-generation.test.ts index f34236b..0a9d92c 100644 --- a/tests/viewer-peer-generation.test.ts +++ b/tests/viewer-peer-generation.test.ts @@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { IceConfig, SignalPayload } from "../src/shared/protocol.ts"; import type { PeerSnapshot } from "../src/client/types.ts"; import { ViewerPeer } from "../src/client/webrtc/viewer-peer.ts"; +import { ViewerSfuRoute } from "../src/client/media/viewer-sfu-route"; interface Deferred { promise: Promise; @@ -243,6 +244,59 @@ afterEach(() => { }); describe("ViewerPeer connection generations", () => { + it.each(["dispose", "replace"])("settles pending answer work on %s without a browser completion", async (retirement) => { + const gate = createDeferred(); + FakePeerConnection.plans.push({ localDescriptionGate: gate.promise }); + const signals: SignalPayload[] = []; + const peer = createPeer(signals, []); + let settled = false; + const old = peer.acceptSignal("parent", offer("old")); + void old.then(() => { settled = true; }); + await flushAsyncWork(); + expect(FakePeerConnection.instances[0]!.setLocalDescription).toHaveBeenCalledOnce(); + if (retirement === "dispose") peer.dispose(); + else await peer.acceptSignal("parent", offer("new")); + await flushAsyncWork(); + const settledByRetirement = settled; + gate.resolve(); + await old; + peer.dispose(); + expect(signals.filter(value => value.connectionId === "old")).toEqual([]); + expect(settledByRetirement).toBe(true); + }); + + it("releases authoritative route resync when its old browser operation never completes", async () => { + const gate = createDeferred(); + FakePeerConnection.plans.push({ localDescriptionGate: gate.promise }); + const signals: SignalPayload[] = []; + let peer: ViewerPeer | null = createPeer(signals, []); + let connection = 0; + const route = new ViewerSfuRoute("viewer-old", { + activatePeer: async () => { + peer ??= createPeer(signals, []); + await peer.acceptSignal("parent", offer(`connection-${++connection}`)); + }, + resetMedia: () => { peer?.dispose(); peer = null; }, + onSfuStream: () => undefined, + send: () => true, + }); + const assignment = { upstream: { kind: "peer" as const, peerId: "parent" }, childPeerIds: [], sfuPublicationGeneration: null }; + route.accept({ revision: 1, phase: "active", assignment }); + await flushAsyncWork(); + expect(FakePeerConnection.instances[0]!.setLocalDescription).toHaveBeenCalledOnce(); + let settled = false; + const resync = route.resyncAuthoritative({ revision: 2, phase: "active", assignment }, "viewer-new"); + void resync.then(() => { settled = true; }); + await flushAsyncWork(); + const resyncedWithoutRetiredWork = settled; + gate.resolve(); + await resync; + await route.disconnect(); + peer?.dispose(); + expect(signals.filter(value => value.connectionId === "connection-1")).toEqual([]); + expect(resyncedWithoutRetiredWork).toBe(true); + }); + it.each(["route", "viewer"] as const)("handles construction failure with the %s recovery owner and exact identity", async (recoveryOwner) => { const sendSignal = vi.fn(() => true); const sendRestartRequest = vi.fn(() => true); @@ -977,8 +1031,9 @@ describe("ViewerPeer connection generations", () => { await peer.acceptSignal("host", offer("connection-new")); const newConnection = FakePeerConnection.instances[1]!; - firstCandidate.resolve(); + // Retirement must settle the old flush without awaiting the browser gate. await oldAccept; + firstCandidate.resolve(); expect(oldConnection.addIceCandidate).toHaveBeenCalledTimes(1); expect(newConnection.addIceCandidate).not.toHaveBeenCalled(); diff --git a/tests/viewer-playback.test.ts b/tests/viewer-playback.test.ts index 370b54a..88caabb 100644 --- a/tests/viewer-playback.test.ts +++ b/tests/viewer-playback.test.ts @@ -1,8 +1,17 @@ import { describe, expect, it, vi } from "vitest"; -import { nextViewerMediaBinding, prepareViewerPlayback } from "../src/client/media/viewer-playback.ts"; +import { nextViewerMediaBinding, playbackFailure, prepareViewerPlayback } from "../src/client/media/viewer-playback.ts"; describe("Viewer playback binding", () => { + it("distinguishes canceled playback from autoplay denial and actual failures", () => { + expect(playbackFailure(new DOMException("Paused", "AbortError"))).toBeNull(); + expect(playbackFailure({ name: "AbortError" })).toBeNull(); + expect(playbackFailure(new DOMException("Blocked", "NotAllowedError"))).toBe("autoplay-blocked"); + expect(playbackFailure(new DOMException("Unsupported", "NotSupportedError"))).toBe("playback-failed"); + expect(playbackFailure(new Error("decode failed"))).toBe("playback-failed"); + expect(playbackFailure(null)).toBe("playback-failed"); + }); + it("retains video identity across audio arrival, replacement and removal", () => { let audio: { id: string }[] = []; const stream = { getVideoTracks: () => [{ id: "video" }], getAudioTracks: () => audio } as MediaStream; @@ -34,13 +43,29 @@ describe("Viewer playback binding", () => { srcObject: null as MediaProvider | null, pause: vi.fn(), }; + const pauseState = { active: false, resume: false }; - expect(prepareViewerPlayback(video, stream, false)).toBe(true); + expect(prepareViewerPlayback(video, stream, false, pauseState)).toBe(true); expect(video.srcObject).toBe(stream); - expect(prepareViewerPlayback(video, stream, false)).toBe(true); + expect(prepareViewerPlayback(video, stream, false, pauseState)).toBe(true); expect(video.pause).not.toHaveBeenCalled(); - expect(prepareViewerPlayback(video, stream, true)).toBe(false); + expect(prepareViewerPlayback(video, stream, true, pauseState)).toBe(false); expect(video.pause).toHaveBeenCalledOnce(); }); + + it.each([false, true])("arms Host resume on a new binding regardless of the old local pause (%s)", (resume) => { + const stream = {} as MediaStream; + const video = { srcObject: stream, pause: vi.fn() }; + const pauseState = { active: true, resume }; + + // A new video generation can reuse the same MediaStream object. + expect(prepareViewerPlayback(video, stream, true, pauseState)).toBe(false); + expect(video.pause).toHaveBeenCalledOnce(); + expect(pauseState).toEqual({ active: true, resume: true }); + + // If the Host has already resumed, the new binding starts immediately. + expect(prepareViewerPlayback(video, stream, false, pauseState)).toBe(true); + expect(pauseState).toEqual({ active: false, resume: false }); + }); });