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>
This commit is contained in:
2026-07-19 15:47:32 -04:00
co-authored by Claude Opus 4.8
parent 8c4f4a0b8b
commit b4a4c00711
2 changed files with 100 additions and 17 deletions
+58 -15
View File
@@ -30,19 +30,33 @@ use std::time::Duration;
pub const CACHE_HIGH_S: f64 = 1.0; pub const CACHE_HIGH_S: f64 = 1.0;
/// Buffer depth (seconds) below which we return to realtime. /// Buffer depth (seconds) below which we return to realtime.
pub const CACHE_LOW_S: f64 = 0.4; pub const CACHE_LOW_S: f64 = 0.4;
/// Playback rate used while draining. Small enough to stay inaudible with /// The buffer depth we aim to sit at; the drain rate is proportional to how far
/// pitch correction, large enough to clear a handover backlog in seconds. /// above this the buffer actually is.
pub const CATCHUP_SPEED: f64 = 1.05; 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. /// Normal realtime playback.
pub const NORMAL_SPEED: f64 = 1.0; pub const NORMAL_SPEED: f64 = 1.0;
/// How often we sample the buffer depth. /// How often we sample the buffer depth.
pub const POLL_INTERVAL: Duration = Duration::from_millis(500); 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. /// The property we watch on the viewer.
const CACHE_PROPERTY: &str = "demuxer-cache-duration"; const CACHE_PROPERTY: &str = "demuxer-cache-duration";
/// Decide the playback rate for the next interval. /// 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 /// Deliberately hysteretic: between [`CACHE_LOW_S`] and [`CACHE_HIGH_S`] the
/// current rate is held, so a buffer hovering near a single threshold cannot /// 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. /// oscillate the speed (and with it the audio pitch) every poll. Pure.
@@ -53,13 +67,14 @@ pub fn catchup_speed(cache_s: f64, current: f64) -> f64 {
if !cache_s.is_finite() { if !cache_s.is_finite() {
return current; return current;
} }
if cache_s > CACHE_HIGH_S { if cache_s < CACHE_LOW_S {
CATCHUP_SPEED return NORMAL_SPEED;
} else if cache_s < CACHE_LOW_S {
NORMAL_SPEED
} else {
current
} }
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 /// Where mpv should create its IPC socket. Kept separate from the runtime
@@ -164,7 +179,9 @@ pub async fn drive(socket: PathBuf) {
let cache = cache.unwrap_or(f64::NAN); let cache = cache.unwrap_or(f64::NAN);
let next = catchup_speed(cache, speed); let next = catchup_speed(cache, speed);
if next != 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; speed = next;
request_id += 1; request_id += 1;
let set = format!("{}\n", set_speed_request(request_id, speed)); let set = format!("{}\n", set_speed_request(request_id, speed));
@@ -184,8 +201,31 @@ mod tests {
#[test] #[test]
fn deep_buffer_speeds_up_and_drained_buffer_returns_to_realtime() { fn deep_buffer_speeds_up_and_drained_buffer_returns_to_realtime() {
assert_eq!(catchup_speed(1.5, NORMAL_SPEED), CATCHUP_SPEED); assert!(catchup_speed(1.5, NORMAL_SPEED) > NORMAL_SPEED);
assert_eq!(catchup_speed(0.1, CATCHUP_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] #[test]
@@ -194,13 +234,16 @@ mod tests {
// this is what stops the rate (and audio pitch) oscillating. // this is what stops the rate (and audio pitch) oscillating.
for cache in [CACHE_LOW_S, 0.7, CACHE_HIGH_S] { 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, NORMAL_SPEED), NORMAL_SPEED);
assert_eq!(catchup_speed(cache, CATCHUP_SPEED), CATCHUP_SPEED); assert_eq!(catchup_speed(cache, MAX_CATCHUP_SPEED), MAX_CATCHUP_SPEED);
} }
} }
#[test] #[test]
fn unknown_cache_holds_the_current_speed() { fn unknown_cache_holds_the_current_speed() {
assert_eq!(catchup_speed(f64::NAN, CATCHUP_SPEED), CATCHUP_SPEED); assert_eq!(
catchup_speed(f64::NAN, MAX_CATCHUP_SPEED),
MAX_CATCHUP_SPEED
);
assert_eq!(catchup_speed(f64::INFINITY, NORMAL_SPEED), NORMAL_SPEED); assert_eq!(catchup_speed(f64::INFINITY, NORMAL_SPEED), NORMAL_SPEED);
} }
@@ -208,7 +251,7 @@ mod tests {
fn a_full_handover_cycle_drains_then_settles() { fn a_full_handover_cycle_drains_then_settles() {
// Buffer grows through a loss burst, then drains as we play faster. // Buffer grows through a loss burst, then drains as we play faster.
let mut speed = NORMAL_SPEED; let mut speed = NORMAL_SPEED;
for cache in [0.2, 0.5, 1.2, 1.4, 0.9, 0.6, 0.3, 0.2] { 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); speed = catchup_speed(cache, speed);
} }
assert_eq!( assert_eq!(
+42 -2
View File
@@ -47,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.
@@ -699,7 +705,16 @@ pub fn mpv_args(settings: &ScreenShareSettings, ipc_socket: Option<&Path>) -> Ve
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());
} }
@@ -879,7 +894,7 @@ mod tests {
"--profile=low-latency", "--profile=low-latency",
"--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",
] ]
); );
} }
@@ -915,6 +930,31 @@ mod tests {
); );
} }
#[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] #[test]
fn user_extra_args_still_come_last() { fn user_extra_args_still_come_last() {
let settings = ScreenShareSettings { let settings = ScreenShareSettings {