3 Commits
Author SHA1 Message Date
molluskandClaude Opus 4.8 0588d92537 release: 0.6.6
CI / check (push) Failing after 5m35s
The live-edge catch-up (8c4f4a0, b4a4c00) landed after the v0.6.5 tag, so
the 0.6.5 artifacts do not contain it — the same gap that left the fix out
of v0.6.4. Cut 0.6.6 so the published build actually carries it.

Local-only changes (no wire change; PROTO planes unchanged), so this is a
PATCH bump per VERSIONING.md.

601 lib tests green, clippy clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-19 15:57:05 -04:00
molluskandClaude Opus 4.8 b4a4c00711 fix(screenshare): make live-edge catch-up actually recover
CI / check (push) Failing after 2m37s
The first cut used a fixed 1.05x drain, which measurement showed was too
gentle to matter: clearing a 6 s backlog would take two minutes, which a
viewer experiences as still broken.

Two changes, both measured on the netem satellite rig (loopback
impairment, gst -> ffmpeg HTTP relay -> mpv, matching the http:// URL
production actually serves):

1. Proportional drain. Speed now scales with buffer depth,
   1 + 0.05*(cache - 0.5), clamped to 1.15x, keeping the hysteresis band
   so it cannot oscillate. Deep backlogs recover in tens of seconds;
   small excursions still get an inaudible nudge.

2. Bound the byte cache in Low latency. The demuxer cache is a *byte*
   budget, so at a given bitrate it sets the worst-case backlog: 2 MiB
   held ~6 s of a 2.5 Mbps share. Capping Low latency at 1 MiB halved the
   standing buffer, 6.0 s -> 2.8 s, on its own. Smooth keeps the user's
   value, since a deep buffer is that posture's whole point.

Measured effect with both: playback consumes 11.6% faster than realtime
while behind (ratio 1.1157 vs 0.9988 with catch-up off), i.e. ~9 s of
backlog cleared in 80 s where before it recovered nothing at all and the
viewer stayed behind for the rest of the call.

Rig caveat: its upstream queues hold an unbounded backlog, so the cache
never drops back through the low mark and the return-to-1x transition is
only covered by unit tests, not the rig.

601 lib tests green, clippy clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-19 15:47:32 -04:00
molluskandClaude Opus 4.8 8c4f4a0b8b feat(screenshare): drain a lagging viewer back to the live edge
CI / check (push) Failing after 2m12s
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>
2026-07-19 15:36:06 -04:00
6 changed files with 493 additions and 30 deletions
+21
View File
@@ -4,6 +4,27 @@ All notable changes to PeerSpeak are documented here.
## [Unreleased] ## [Unreleased]
## [0.6.6] — 2026-07-19
### 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.
### Changed
- **Low latency now keeps a tighter viewer buffer.** The screen-share cache
setting is a size in megabytes, which at a given bitrate quietly decides how
many *seconds* behind a viewer can drift — a 2 MB buffer turned out to hold
about six seconds of a typical share. Low latency now caps that buffer at 1 MB
regardless of the setting, which halved how far behind a share fell on a bad
connection before anything else kicked in. Smooth still honors the value you
choose, since a deep buffer is the point of that mode.
## [0.6.5] — 2026-07-19 ## [0.6.5] — 2026-07-19
### Added ### Added
Generated
+1 -1
View File
@@ -4871,7 +4871,7 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec"
[[package]] [[package]]
name = "peerspeak" name = "peerspeak"
version = "0.6.5" version = "0.6.6"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-trait", "async-trait",
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "peerspeak" name = "peerspeak"
version = "0.6.5" version = "0.6.6"
edition = "2024" edition = "2024"
description = "Decentralized peer-to-peer voice chat (Rust/iroh/PipeWire/Opus/iced)" description = "Decentralized peer-to-peer voice chat (Rust/iroh/PipeWire/Opus/iced)"
license = "MIT" license = "MIT"
+1 -1
View File
@@ -12,7 +12,7 @@
; (x86_64-pc-windows-gnu, statically linked -- no extra DLLs needed). ; (x86_64-pc-windows-gnu, statically linked -- no extra DLLs needed).
#define MyAppName "PeerSpeak" #define MyAppName "PeerSpeak"
#define MyAppVersion "0.6.5" #define MyAppVersion "0.6.6"
#define MyAppPublisher "mollusk" #define MyAppPublisher "mollusk"
#define MyAppExeName "peerspeak.exe" #define MyAppExeName "peerspeak.exe"
+314
View File
@@ -0,0 +1,314 @@
//! 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;
/// The buffer depth we aim to sit at; the drain rate is proportional to how far
/// above this the buffer actually is.
pub const CACHE_TARGET_S: f64 = 0.5;
/// Extra playback rate per second of excess buffer.
pub const CATCHUP_GAIN: f64 = 0.05;
/// Hard ceiling on the drain rate. Beyond this the speedup stops being
/// unnoticeable, and a share that far behind is better served by the operator
/// restarting it than by a chipmunk impression.
pub const MAX_CATCHUP_SPEED: f64 = 1.15;
/// 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);
/// Smallest rate change worth sending to the player.
pub const SPEED_EPSILON: f64 = 0.005;
/// The property we watch on the viewer.
const CACHE_PROPERTY: &str = "demuxer-cache-duration";
/// Decide the playback rate for the next interval.
///
/// Proportional, because a fixed small speedup cannot recover a large backlog in
/// any reasonable time: draining 6 s at 1.05x takes two minutes, which a viewer
/// experiences as "still broken". The drain rate instead scales with how deep
/// the buffer is, so a bad handover is cleared in tens of seconds while a small
/// excursion still gets only a gentle, inaudible nudge.
///
/// 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_LOW_S {
return NORMAL_SPEED;
}
if cache_s <= CACHE_HIGH_S {
return current;
}
let excess = cache_s - CACHE_TARGET_S;
(NORMAL_SPEED + CATCHUP_GAIN * excess).clamp(NORMAL_SPEED, MAX_CATCHUP_SPEED)
}
/// 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);
// A proportional law would otherwise re-send on every wobble of the
// reading; only a change worth hearing is worth a round trip.
if (next - speed).abs() > SPEED_EPSILON {
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!(catchup_speed(1.5, NORMAL_SPEED) > NORMAL_SPEED);
assert_eq!(catchup_speed(0.1, MAX_CATCHUP_SPEED), NORMAL_SPEED);
}
#[test]
fn drain_rate_scales_with_how_far_behind_we_are() {
// The point of the proportional law: a small excursion gets a gentle
// nudge, a deep backlog gets real recovery.
let small = catchup_speed(1.5, NORMAL_SPEED);
let large = catchup_speed(4.0, NORMAL_SPEED);
assert!(
large > small,
"deeper buffer must drain faster: {small} vs {large}"
);
assert!(
(small - 1.05).abs() < 1e-9,
"1.5s buffer -> 1.05x, got {small}"
);
}
#[test]
fn drain_rate_is_capped_so_it_never_sounds_absurd() {
// The ~6 s standing buffer measured on the netem rig, and far worse.
assert_eq!(catchup_speed(6.0, NORMAL_SPEED), MAX_CATCHUP_SPEED);
assert_eq!(catchup_speed(600.0, NORMAL_SPEED), MAX_CATCHUP_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, MAX_CATCHUP_SPEED), MAX_CATCHUP_SPEED);
}
}
#[test]
fn unknown_cache_holds_the_current_speed() {
assert_eq!(
catchup_speed(f64::NAN, MAX_CATCHUP_SPEED),
MAX_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, 3.4, 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));
}
}
+155 -27
View File
@@ -21,6 +21,8 @@ use std::time::Duration;
use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, Command}; use tokio::process::{Child, Command};
pub mod livesync;
use crate::config::{ScreenShareSettings, ShareBuffering, SharePlayer, ShareQuality}; use crate::config::{ScreenShareSettings, ShareBuffering, SharePlayer, ShareQuality};
/// The binary we shell out to. Looked up on `$PATH` unless a config override /// The binary we shell out to. Looked up on `$PATH` unless a config override
@@ -45,6 +47,12 @@ const MAX_TICKET_LEN: usize = 512;
/// are short ("Firefox", "mpv"); this only guards against a pathological value. /// are short ("Firefox", "mpv"); this only guards against a pathological value.
const MAX_APP_NAME_LEN: usize = 256; const MAX_APP_NAME_LEN: usize = 256;
/// Ceiling on the viewer's demuxer byte cache in the Low latency posture. The
/// cache is a *byte* budget, so at a given bitrate it sets the worst-case
/// backlog in seconds; keeping it tight is what stops a lossy link parking the
/// viewer seconds behind before [`livesync`] even gets a chance to drain it.
const LOW_LATENCY_CACHE_CAP_MB: u32 = 1;
/// How long to wait for the host to emit its ticket / the viewer to connect /// How long to wait for the host to emit its ticket / the viewer to connect
/// before giving up and killing the child. Startup is normally sub-second; this /// before giving up and killing the child. Startup is normally sub-second; this
/// is only a safety net so a hung pixelpass can't wedge the caller forever. /// is only a safety net so a hung pixelpass can't wedge the caller forever.
@@ -608,14 +616,27 @@ fn event_for_log(ev: &PixelpassEvent) -> String {
/// its window closes. /// its window closes.
/// ///
/// The buffering posture chooses the latency/A/V-sync tradeoff. Low latency /// 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<()> { 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 vlc_args = vlc_args(settings);
let first = match settings.player { let first = match settings.player {
SharePlayer::Mpv => ("mpv", &mpv_args), SharePlayer::Mpv => ("mpv", &mpv_args),
@@ -626,15 +647,32 @@ fn launch_player(url: &str, settings: &ScreenShareSettings) -> std::io::Result<(
SharePlayer::Vlc => ("mpv", &mpv_args), 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 { tokio::spawn(async move {
let mut child = child; let mut child = child;
let _ = child.wait().await; let _ = child.wait().await;
@@ -642,17 +680,23 @@ fn launch_player(url: &str, settings: &ScreenShareSettings) -> std::io::Result<(
Ok(()) 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(); let mut args = Vec::new();
match settings.buffering { match settings.buffering {
ShareBuffering::LowLatency => { ShareBuffering::LowLatency => {
args.push("--profile=low-latency".to_string()); 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("--audio-buffer=0.2".to_string());
args.push("--demuxer-readahead-secs=0.5".to_string()); args.push("--demuxer-readahead-secs=0.5".to_string());
} }
@@ -661,10 +705,25 @@ pub fn mpv_args(settings: &ScreenShareSettings) -> Vec<String> {
args.push("--demuxer-readahead-secs=2".to_string()); args.push("--demuxer-readahead-secs=2".to_string());
} }
} }
args.push(format!("--demuxer-max-bytes={}M", settings.cache_mb)); // The byte cap is what bounds how far behind a viewer can silently fall:
// a demuxer allowed 2 MiB will happily sit on ~6 s of a 2.5 Mbps share (as
// measured on the netem rig) and call it a buffer. Low latency therefore
// gets a tighter ceiling than the user's Smooth-oriented setting, so the
// catch-up has less to claw back after a bad patch of link.
let cache_mb = match settings.buffering {
ShareBuffering::LowLatency => settings.cache_mb.min(LOW_LATENCY_CACHE_CAP_MB),
ShareBuffering::Smooth => settings.cache_mb,
};
args.push(format!("--demuxer-max-bytes={cache_mb}M"));
if settings.hardware_decode { if settings.hardware_decode {
args.push("--hwdec=auto".to_string()); 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.extend(split_extra_args(&settings.extra_mpv_args));
args args
} }
@@ -830,17 +889,86 @@ mod tests {
#[test] #[test]
fn mpv_args_default_matches_low_latency_software_decode() { fn mpv_args_default_matches_low_latency_software_decode() {
assert_eq!( assert_eq!(
mpv_args(&ScreenShareSettings::default()), mpv_args(&ScreenShareSettings::default(), None),
vec![ vec![
"--profile=low-latency", "--profile=low-latency",
"--untimed",
"--audio-buffer=0.2", "--audio-buffer=0.2",
"--demuxer-readahead-secs=0.5", "--demuxer-readahead-secs=0.5",
"--demuxer-max-bytes=2M", "--demuxer-max-bytes=1M",
] ]
); );
} }
#[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 low_latency_caps_the_byte_cache_but_smooth_keeps_the_user_value() {
// The cache is a byte budget, so at a given bitrate it sets the
// worst-case backlog: 2 MiB held ~6 s of a 2.5 Mbps share on the rig.
let generous = ScreenShareSettings {
cache_mb: 32,
..ScreenShareSettings::default()
};
assert!(
mpv_args(&generous, None)
.contains(&format!("--demuxer-max-bytes={LOW_LATENCY_CACHE_CAP_MB}M")),
"low latency must bound how far behind the viewer can silently fall"
);
let smooth = ScreenShareSettings {
cache_mb: 32,
buffering: ShareBuffering::Smooth,
..ScreenShareSettings::default()
};
assert!(
mpv_args(&smooth, None).contains(&"--demuxer-max-bytes=32M".to_string()),
"smooth is the posture where the user asked for a deep buffer"
);
}
#[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] #[test]
fn mpv_args_smooth_hwdecode_and_extra_args_last() { fn mpv_args_smooth_hwdecode_and_extra_args_last() {
let settings = ScreenShareSettings { let settings = ScreenShareSettings {
@@ -852,7 +980,7 @@ mod tests {
}; };
assert_eq!( assert_eq!(
mpv_args(&settings), mpv_args(&settings, None),
vec![ vec![
"--cache=yes", "--cache=yes",
"--demuxer-readahead-secs=2", "--demuxer-readahead-secs=2",