diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e0ca2b..51ceeab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ All notable changes to PeerSpeak are documented here. ## [Unreleased] +### Fixed +- **A screen share that falls behind now catches back up.** On a lossy + connection (satellite links are the worst case) the share could settle several + seconds behind the host and simply stay there for the rest of the call. The + viewer now notices a deep buffer and plays imperceptibly fast until it is back + at the live edge — the audio stays in tune and in sync while it does. This + replaces the previous attempt at the problem, which measurement showed did not + help. Applies to the Low latency setting; Smooth intentionally keeps its + larger buffer. + ## [0.6.5] — 2026-07-19 ### Added diff --git a/src/screenshare/livesync.rs b/src/screenshare/livesync.rs new file mode 100644 index 0000000..46acd2a --- /dev/null +++ b/src/screenshare/livesync.rs @@ -0,0 +1,271 @@ +//! Live-edge catch-up for the screen-share viewer. +//! +//! PixelPass carries the share as MPEG-TS over a reliable, ordered transport. On +//! a lossy link (satellite handovers are the pathological case) every loss burst +//! becomes retransmission plus head-of-line blocking, and the viewer absorbs the +//! stall as buffered latency. Nothing in the chain ever trims that buffer back, +//! so the picture ends up seconds behind the host and stays there. +//! +//! Measured on a `tc netem` rig that simulates a satellite link (40 ms +/- 20 ms +//! jitter, 0.5% loss, a 250 ms/30%-loss handover burst every 15 s): a viewer with +//! ordinary timestamp pacing settles ~1.24 s behind. mpv's `--untimed` does NOT +//! help (~1.38 s, marginally worse) because it only removes pacing at +//! *presentation* while audio still drains at 1x the DAC rate, so an accumulated +//! buffer never shrinks. Returning to the live edge requires consuming the +//! backlog faster than it arrives. +//! +//! So we nudge playback slightly faster than realtime while the buffer is deep, +//! and drop back to 1x once it has drained. mpv's default pitch correction +//! (`scaletempo2`) keeps a 5% speedup inaudible, and because audio and video are +//! sped up together A/V sync is preserved — unlike `--untimed`. +//! +//! The control law and the JSON-IPC message handling are pure functions with +//! tests; the only I/O is [`drive`], which talks to mpv's `--input-ipc-server` +//! socket. + +use std::path::{Path, PathBuf}; +use std::time::Duration; + +/// Buffer depth (seconds) above which we start draining. +pub const CACHE_HIGH_S: f64 = 1.0; +/// Buffer depth (seconds) below which we return to realtime. +pub const CACHE_LOW_S: f64 = 0.4; +/// Playback rate used while draining. Small enough to stay inaudible with +/// pitch correction, large enough to clear a handover backlog in seconds. +pub const CATCHUP_SPEED: f64 = 1.05; +/// Normal realtime playback. +pub const NORMAL_SPEED: f64 = 1.0; +/// How often we sample the buffer depth. +pub const POLL_INTERVAL: Duration = Duration::from_millis(500); + +/// The property we watch on the viewer. +const CACHE_PROPERTY: &str = "demuxer-cache-duration"; + +/// Decide the playback rate for the next interval. +/// +/// Deliberately hysteretic: between [`CACHE_LOW_S`] and [`CACHE_HIGH_S`] the +/// current rate is held, so a buffer hovering near a single threshold cannot +/// oscillate the speed (and with it the audio pitch) every poll. Pure. +/// +/// A non-finite reading (mpv reports `null` before playback starts, and the +/// caller maps that to NaN) holds the current rate rather than guessing. +pub fn catchup_speed(cache_s: f64, current: f64) -> f64 { + if !cache_s.is_finite() { + return current; + } + if cache_s > CACHE_HIGH_S { + CATCHUP_SPEED + } else if cache_s < CACHE_LOW_S { + NORMAL_SPEED + } else { + current + } +} + +/// Where mpv should create its IPC socket. Kept separate from the runtime +/// lookup so tests can pin a directory. Pure. +pub fn socket_path(dir: &Path, token: u64) -> PathBuf { + dir.join(format!("peerspeak-mpv-{token}.sock")) +} + +/// The directory for the IPC socket: the XDG runtime dir when the session +/// provides one (tmpfs, user-private, cleaned at logout), else the temp dir. +pub fn socket_dir() -> PathBuf { + std::env::var_os("XDG_RUNTIME_DIR") + .map(PathBuf::from) + .unwrap_or_else(std::env::temp_dir) +} + +/// A `get_property` request for the buffer depth. Pure. +pub fn get_cache_request(request_id: u64) -> String { + format!(r#"{{"command":["get_property","{CACHE_PROPERTY}"],"request_id":{request_id}}}"#) +} + +/// A `set_property` request for the playback rate. Pure. +pub fn set_speed_request(request_id: u64, speed: f64) -> String { + format!(r#"{{"command":["set_property","speed",{speed}],"request_id":{request_id}}}"#) +} + +/// Extract the buffer depth from one line of mpv's IPC output. +/// +/// mpv interleaves unsolicited event lines with command replies, so a line is +/// only ours when it carries the matching `request_id`. Returns: +/// - `Some(Some(secs))` — our reply, with a usable number, +/// - `Some(None)` — our reply, but no number (mpv sends `"data":null` before +/// playback starts, and reports `error` while the demuxer has no cache yet), +/// - `None` — not our reply (an event, or another command's response). +/// +/// Pure. +pub fn parse_cache_response(line: &str, request_id: u64) -> Option> { + let value: serde_json::Value = serde_json::from_str(line.trim()).ok()?; + let id = value.get("request_id")?.as_u64()?; + if id != request_id { + return None; + } + if value.get("error").and_then(|e| e.as_str()) != Some("success") { + return Some(None); + } + Some(value.get("data").and_then(|d| d.as_f64())) +} + +/// Drive one mpv viewer's playback rate over its JSON IPC socket. +/// +/// Runs until mpv exits (the socket dies), so it is spawned detached alongside +/// the player and needs no shutdown signal. Every failure path just ends the +/// task: catch-up is an optimization, and a viewer that never gets it still +/// plays, exactly as before this existed. +#[cfg(unix)] +pub async fn drive(socket: PathBuf) { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + use tokio::net::UnixStream; + + // mpv creates the socket a moment after exec, so the first connects race it. + let mut stream = None; + for _ in 0..40 { + match UnixStream::connect(&socket).await { + Ok(s) => { + stream = Some(s); + break; + } + Err(_) => tokio::time::sleep(Duration::from_millis(250)).await, + } + } + let Some(stream) = stream else { + crate::log_msg("livesync: mpv IPC socket never appeared; catch-up disabled"); + return; + }; + + let (read_half, mut write_half) = stream.into_split(); + let mut lines = BufReader::new(read_half).lines(); + let mut request_id: u64 = 0; + let mut speed = NORMAL_SPEED; + + loop { + tokio::time::sleep(POLL_INTERVAL).await; + + request_id += 1; + let query = format!("{}\n", get_cache_request(request_id)); + if write_half.write_all(query.as_bytes()).await.is_err() { + break; + } + + // Skip event lines until our reply arrives. + let cache = loop { + match lines.next_line().await { + Ok(Some(line)) => { + if let Some(value) = parse_cache_response(&line, request_id) { + break value; + } + } + // Socket closed or unreadable: mpv is gone. + _ => return, + } + }; + + let cache = cache.unwrap_or(f64::NAN); + let next = catchup_speed(cache, speed); + if next != speed { + speed = next; + request_id += 1; + let set = format!("{}\n", set_speed_request(request_id, speed)); + if write_half.write_all(set.as_bytes()).await.is_err() { + break; + } + crate::log_msg(&format!( + "livesync: cache {cache:.2}s -> playback speed {speed}x" + )); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn deep_buffer_speeds_up_and_drained_buffer_returns_to_realtime() { + assert_eq!(catchup_speed(1.5, NORMAL_SPEED), CATCHUP_SPEED); + assert_eq!(catchup_speed(0.1, CATCHUP_SPEED), NORMAL_SPEED); + } + + #[test] + fn hysteresis_band_holds_the_current_speed() { + // Between the marks nothing changes, whichever side we came from — + // this is what stops the rate (and audio pitch) oscillating. + for cache in [CACHE_LOW_S, 0.7, CACHE_HIGH_S] { + assert_eq!(catchup_speed(cache, NORMAL_SPEED), NORMAL_SPEED); + assert_eq!(catchup_speed(cache, CATCHUP_SPEED), CATCHUP_SPEED); + } + } + + #[test] + fn unknown_cache_holds_the_current_speed() { + assert_eq!(catchup_speed(f64::NAN, CATCHUP_SPEED), CATCHUP_SPEED); + assert_eq!(catchup_speed(f64::INFINITY, NORMAL_SPEED), NORMAL_SPEED); + } + + #[test] + fn a_full_handover_cycle_drains_then_settles() { + // Buffer grows through a loss burst, then drains as we play faster. + let mut speed = NORMAL_SPEED; + for cache in [0.2, 0.5, 1.2, 1.4, 0.9, 0.6, 0.3, 0.2] { + speed = catchup_speed(cache, speed); + } + assert_eq!( + speed, NORMAL_SPEED, + "should be back at realtime once drained" + ); + } + + #[test] + fn requests_are_valid_json_with_their_ids() { + let get: serde_json::Value = serde_json::from_str(&get_cache_request(7)).unwrap(); + assert_eq!(get["request_id"], 7); + assert_eq!(get["command"][0], "get_property"); + assert_eq!(get["command"][1], CACHE_PROPERTY); + + let set: serde_json::Value = serde_json::from_str(&set_speed_request(8, 1.05)).unwrap(); + assert_eq!(set["request_id"], 8); + assert_eq!(set["command"][0], "set_property"); + assert_eq!(set["command"][1], "speed"); + assert_eq!(set["command"][2], 1.05); + } + + #[test] + fn parses_our_reply_only() { + assert_eq!( + parse_cache_response(r#"{"error":"success","data":1.25,"request_id":3}"#, 3), + Some(Some(1.25)) + ); + // Another command's reply, and an unsolicited event, are not ours. + assert_eq!( + parse_cache_response(r#"{"error":"success","data":1.25,"request_id":4}"#, 3), + None + ); + assert_eq!( + parse_cache_response(r#"{"event":"playback-restart"}"#, 3), + None + ); + assert_eq!(parse_cache_response("not json", 3), None); + } + + #[test] + fn reply_without_a_usable_number_is_ours_but_empty() { + // mpv before playback starts, and while the demuxer has no cache. + assert_eq!( + parse_cache_response(r#"{"error":"success","data":null,"request_id":1}"#, 1), + Some(None) + ); + assert_eq!( + parse_cache_response(r#"{"error":"property unavailable","request_id":1}"#, 1), + Some(None) + ); + } + + #[test] + fn socket_path_is_scoped_to_its_token() { + let a = socket_path(Path::new("/run/user/1000"), 42); + assert_eq!(a, Path::new("/run/user/1000/peerspeak-mpv-42.sock")); + assert_ne!(a, socket_path(Path::new("/run/user/1000"), 43)); + } +} diff --git a/src/screenshare/mod.rs b/src/screenshare/mod.rs index 3788e2d..a1b81e6 100644 --- a/src/screenshare/mod.rs +++ b/src/screenshare/mod.rs @@ -21,6 +21,8 @@ use std::time::Duration; use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::process::{Child, Command}; +pub mod livesync; + use crate::config::{ScreenShareSettings, ShareBuffering, SharePlayer, ShareQuality}; /// The binary we shell out to. Looked up on `$PATH` unless a config override @@ -608,14 +610,27 @@ fn event_for_log(ev: &PixelpassEvent) -> String { /// its window closes. /// /// The buffering posture chooses the latency/A/V-sync tradeoff. Low latency -/// passes mpv's `--untimed`, displaying frames as soon as they decode so a live -/// share cannot accumulate several seconds in the reliable QUIC/TCP/player -/// buffer chain. Smooth leaves timestamp pacing enabled, which keeps a shared -/// video's audio and video synchronized at the cost of extra live latency. -/// Hardware decoding remains opt-in: forcing `--hwdec=auto` froze some viewers -/// on frame 1 while audio kept playing. +/// keeps the viewer at the live edge: mpv gets an IPC socket and [`livesync`] +/// drains a lagging buffer by playing slightly fast (pitch-corrected, so A/V +/// sync is preserved). Smooth leaves a deeper buffer alone, trading live +/// latency for immunity to jitter. Hardware decoding remains opt-in: forcing +/// `--hwdec=auto` froze some viewers on frame 1 while audio kept playing. fn launch_player(url: &str, settings: &ScreenShareSettings) -> std::io::Result<()> { - let mpv_args = mpv_args(settings); + // One socket per viewer launch, so overlapping shares can't collide on it. + // Unix only: mpv's IPC is a named pipe on Windows, which `livesync` does not + // speak, and an unusable socket path on the argv would help nobody. + #[cfg(unix)] + let ipc_socket = Some(livesync::socket_path( + &livesync::socket_dir(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0), + )); + #[cfg(not(unix))] + let ipc_socket: Option = None; + + let mpv_args = mpv_args(settings, ipc_socket.as_deref()); let vlc_args = vlc_args(settings); let first = match settings.player { SharePlayer::Mpv => ("mpv", &mpv_args), @@ -626,15 +641,32 @@ fn launch_player(url: &str, settings: &ScreenShareSettings) -> std::io::Result<( SharePlayer::Vlc => ("mpv", &mpv_args), }; - let child = match spawn_player(first.0, first.1, url) { - Ok(c) => c, - Err(_) => spawn_player(second.0, second.1, url).map_err(|_| { - std::io::Error::new( - std::io::ErrorKind::NotFound, - "no media player found — install mpv or vlc to watch screen shares", - ) - })?, + let (launched, child) = match spawn_player(first.0, first.1, url) { + Ok(c) => (first.0, c), + Err(_) => ( + second.0, + spawn_player(second.0, second.1, url).map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::NotFound, + "no media player found — install mpv or vlc to watch screen shares", + ) + })?, + ), }; + + // Only when the socket actually reached the argv: mpv (VLC has no + // equivalent IPC) in the Low latency posture. The driver ends by itself when + // the player exits, so it needs no shutdown path. + #[cfg(unix)] + if launched == "mpv" + && settings.buffering == ShareBuffering::LowLatency + && let Some(socket) = ipc_socket + { + tokio::spawn(livesync::drive(socket)); + } + #[cfg(not(unix))] + let _ = launched; + tokio::spawn(async move { let mut child = child; let _ = child.wait().await; @@ -642,17 +674,23 @@ fn launch_player(url: &str, settings: &ScreenShareSettings) -> std::io::Result<( Ok(()) } -pub fn mpv_args(settings: &ScreenShareSettings) -> Vec { +/// Build the argv for an mpv viewer. +/// +/// `ipc_socket` is where mpv should expose its JSON IPC socket so [`livesync`] +/// can drain a lagging buffer. It is wired up for Low latency only: Smooth +/// deliberately holds a ~2 s readahead, which the catch-up thresholds would +/// fight on every poll. +pub fn mpv_args(settings: &ScreenShareSettings, ipc_socket: Option<&Path>) -> Vec { let mut args = Vec::new(); match settings.buffering { ShareBuffering::LowLatency => { args.push("--profile=low-latency".to_string()); - // Pixelpass carries MPEG-TS through reliable ordered QUIC/TCP. If - // mpv paces a live stream even slightly slower than capture, those - // downstream buffers retain stale frames and latency grows into - // seconds. Immediate presentation keeps this posture at the live - // edge; Smooth deliberately retains timestamp pacing for A/V sync. - args.push("--untimed".to_string()); + // Pixelpass carries MPEG-TS through reliable ordered QUIC/TCP, so a + // lossy link turns every retransmission into buffered latency that + // nothing trims back. `--untimed` does NOT fix that (measured + // marginally worse: it only unpaces *presentation*, while audio + // still drains at 1x, so the backlog never shrinks) — the viewer + // instead drains it by playing slightly fast, see `livesync`. args.push("--audio-buffer=0.2".to_string()); args.push("--demuxer-readahead-secs=0.5".to_string()); } @@ -665,6 +703,12 @@ pub fn mpv_args(settings: &ScreenShareSettings) -> Vec { if settings.hardware_decode { args.push("--hwdec=auto".to_string()); } + if let Some(socket) = ipc_socket + && settings.buffering == ShareBuffering::LowLatency + { + args.push(format!("--input-ipc-server={}", socket.display())); + } + // Extra args stay last so a user override wins over everything above. args.extend(split_extra_args(&settings.extra_mpv_args)); args } @@ -830,10 +874,9 @@ mod tests { #[test] fn mpv_args_default_matches_low_latency_software_decode() { assert_eq!( - mpv_args(&ScreenShareSettings::default()), + mpv_args(&ScreenShareSettings::default(), None), vec![ "--profile=low-latency", - "--untimed", "--audio-buffer=0.2", "--demuxer-readahead-secs=0.5", "--demuxer-max-bytes=2M", @@ -841,6 +884,51 @@ mod tests { ); } + #[test] + fn low_latency_gets_the_ipc_socket_for_live_edge_catch_up() { + let args = mpv_args( + &ScreenShareSettings::default(), + Some(Path::new("/run/user/1000/peerspeak-mpv-1.sock")), + ); + assert!( + args.contains(&"--input-ipc-server=/run/user/1000/peerspeak-mpv-1.sock".to_string()), + "low latency drains a lagging buffer over mpv IPC: {args:?}" + ); + // The flag that used to hold this posture at the live edge measured no + // better than pacing, and cost A/V sync — it must not come back. + assert!(!args.contains(&"--untimed".to_string())); + } + + #[test] + fn smooth_keeps_its_deep_buffer_and_gets_no_ipc_socket() { + let settings = ScreenShareSettings { + buffering: ShareBuffering::Smooth, + ..ScreenShareSettings::default() + }; + let args = mpv_args( + &settings, + Some(Path::new("/run/user/1000/peerspeak-mpv-1.sock")), + ); + assert!( + !args.iter().any(|a| a.starts_with("--input-ipc-server")), + "catch-up would fight Smooth's deliberate ~2s readahead: {args:?}" + ); + } + + #[test] + fn user_extra_args_still_come_last() { + let settings = ScreenShareSettings { + extra_mpv_args: "--no-osc".to_string(), + ..ScreenShareSettings::default() + }; + let args = mpv_args(&settings, Some(Path::new("/tmp/s.sock"))); + assert_eq!( + args.last().map(String::as_str), + Some("--no-osc"), + "a user override has to win over everything we add: {args:?}" + ); + } + #[test] fn mpv_args_smooth_hwdecode_and_extra_args_last() { let settings = ScreenShareSettings { @@ -852,7 +940,7 @@ mod tests { }; assert_eq!( - mpv_args(&settings), + mpv_args(&settings, None), vec![ "--cache=yes", "--demuxer-readahead-secs=2",