feat(screenshare): drain a lagging viewer back to the live edge

On a lossy link the reliable PixelPass transport turns every loss burst
into buffered latency that nothing trims back, so the viewer settles
seconds behind the host and stays there. Measured on a tc netem satellite
simulation: a viewer parks at a ~6 s standing buffer indefinitely.

--untimed (0.6.5) does NOT fix this and measured marginally worse (+1.38 s
vs +1.24 s): it only unpaces presentation, while audio still drains at 1x
the DAC rate, so an accumulated backlog never shrinks. Drop it.

Instead give mpv a JSON IPC socket in the Low latency posture and drive
playback slightly fast while the buffer is deep, returning to 1x once it
drains. Pitch correction keeps it inaudible and A/V sync is preserved,
because audio and video speed up together.

The control law and IPC message handling are pure functions with unit
tests; the only I/O is livesync::drive, which ends by itself when the
player exits. Smooth is deliberately excluded — its ~2 s readahead is the
point of that posture, and catch-up would fight it every poll.

Known limitation: 1.05x needs ~120 s to clear a 6 s backlog, so recovery
is slower than ideal. Tuning (a proportional law, or a seek-to-live for
large backlogs) is the follow-up.

598 lib tests green (+11), clippy clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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

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<Option<f64>> {
+    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));
+    }
+}

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
+/// keeps the viewer at the live edge: mpv gets an IPC socket and [`livesync`]
-/// share cannot accumulate several seconds in the reliable QUIC/TCP/player
+/// drains a lagging buffer by playing slightly fast (pitch-corrected, so A/V
-/// buffer chain. Smooth leaves timestamp pacing enabled, which keeps a shared
+/// sync is preserved). Smooth leaves a deeper buffer alone, trading live
-/// video's audio and video synchronized at the cost of extra live latency.
+/// latency for immunity to jitter. Hardware decoding remains opt-in: forcing
-/// Hardware decoding remains opt-in: forcing `--hwdec=auto` froze some viewers
+/// `--hwdec=auto` froze some viewers on frame 1 while audio kept playing.
-/// 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<PathBuf> = 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) {
+    let (launched, child) = match spawn_player(first.0, first.1, url) {
-        Ok(c) => c,
+        Ok(c) => (first.0, c),
-        Err(_) => spawn_player(second.0, second.1, url).map_err(|_| {
+        Err(_) => (
-            std::io::Error::new(
+            second.0,
-                std::io::ErrorKind::NotFound,
+            spawn_player(second.0, second.1, url).map_err(|_| {
-                "no media player found — install mpv or vlc to watch screen shares",
+                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<String> {
+/// 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<String> {
     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
+            // Pixelpass carries MPEG-TS through reliable ordered QUIC/TCP, so a
-            // mpv paces a live stream even slightly slower than capture, those
+            // lossy link turns every retransmission into buffered latency that
-            // downstream buffers retain stale frames and latency grows into
+            // nothing trims back. `--untimed` does NOT fix that (measured
-            // seconds. Immediate presentation keeps this posture at the live
+            // marginally worse: it only unpaces *presentation*, while audio
-            // edge; Smooth deliberately retains timestamp pacing for A/V sync.
+            // still drains at 1x, so the backlog never shrinks) — the viewer
-            args.push("--untimed".to_string());
+            // 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<String> {
     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",