feat: support compensated output delay probe

Cargo.lock

@@ -1652,7 +1652,7 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
 
 [[package]]
 name = "lesavka_client"
-version = "0.19.3"
+version = "0.19.4"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -1686,7 +1686,7 @@ dependencies = [
 
 [[package]]
 name = "lesavka_common"
-version = "0.19.3"
+version = "0.19.4"
 dependencies = [
  "anyhow",
  "base64",
@@ -1698,7 +1698,7 @@ dependencies = [
 
 [[package]]
 name = "lesavka_server"
-version = "0.19.3"
+version = "0.19.4"
 dependencies = [
  "anyhow",
  "base64",

client/Cargo.toml

@@ -4,7 +4,7 @@ path                    = "src/main.rs"
 
 [package]
 name                    = "lesavka_client"
-version                 = "0.19.3"
+version                 = "0.19.4"
 edition                 = "2024"
 
 [dependencies]

client/src/bin/lesavka-relayctl.rs

@@ -72,6 +72,8 @@ struct Config {
     probe_pulse_period_ms: u32,
     probe_pulse_width_ms: u32,
     probe_event_width_codes: String,
+    probe_audio_delay_us: i64,
+    probe_video_delay_us: i64,
 }
 
 #[derive(Debug, Default, Eq, PartialEq)]
@@ -84,6 +86,8 @@ struct ParsedCommandArgs {
     probe_pulse_period_ms: u32,
     probe_pulse_width_ms: u32,
     probe_event_width_codes: String,
+    probe_audio_delay_us: i64,
+    probe_video_delay_us: i64,
 }
 
 #[derive(Debug, Eq, PartialEq)]
@@ -93,7 +97,7 @@ enum ParseOutcome {
 }
 
 fn usage() -> &'static str {
-    "Usage: lesavka-relayctl [--server http://HOST:50051] <status|version|upstream-sync|output-delay-probe [duration_s warmup_s period_ms width_ms codes]|calibration|calibrate <audio_delta_us> <video_delta_us> [note]|calibration-save-default|calibration-restore-default|calibration-restore-factory|auto|on|off|recover-usb|recover-uac|recover-uvc|reset-usb>"
+    "Usage: lesavka-relayctl [--server http://HOST:50051] <status|version|upstream-sync|output-delay-probe [duration_s warmup_s period_ms width_ms codes audio_delay_us video_delay_us]|calibration|calibrate <audio_delta_us> <video_delta_us> [note]|calibration-save-default|calibration-restore-default|calibration-restore-factory|auto|on|off|recover-usb|recover-uac|recover-uvc|reset-usb>"
 }
 
 fn parse_args_outcome_from<I, S>(args: I) -> Result<ParseOutcome>
@@ -141,6 +145,8 @@ where
         probe_pulse_period_ms: parsed.probe_pulse_period_ms,
         probe_pulse_width_ms: parsed.probe_pulse_width_ms,
         probe_event_width_codes: parsed.probe_event_width_codes,
+        probe_audio_delay_us: parsed.probe_audio_delay_us,
+        probe_video_delay_us: parsed.probe_video_delay_us,
     }))
 }
 
@@ -179,9 +185,9 @@ fn parse_command_args(command: CommandKind, args: Vec<String>) -> Result<ParsedC
 }
 
 fn parse_output_delay_probe_args(args: Vec<String>) -> Result<ParsedCommandArgs> {
-    if args.len() > 5 {
+    if args.len() > 7 {
         bail!(
-            "output-delay-probe accepts at most five arguments\n{}",
+            "output-delay-probe accepts at most seven arguments\n{}",
             usage()
         );
     }
@@ -195,12 +201,24 @@ fn parse_output_delay_probe_args(args: Vec<String>) -> Result<ParsedCommandArgs>
             .transpose()
             .map(|value| value.unwrap_or(0))
     };
+    let parse_i64 = |index: usize, name: &str| -> Result<i64> {
+        args.get(index)
+            .map(|value| {
+                value
+                    .parse::<i64>()
+                    .with_context(|| format!("parsing {name} `{value}`"))
+            })
+            .transpose()
+            .map(|value| value.unwrap_or(0))
+    };
     Ok(ParsedCommandArgs {
         probe_duration_seconds: parse_u32(0, "duration_s")?,
         probe_warmup_seconds: parse_u32(1, "warmup_s")?,
         probe_pulse_period_ms: parse_u32(2, "period_ms")?,
         probe_pulse_width_ms: parse_u32(3, "width_ms")?,
         probe_event_width_codes: args.get(4).cloned().unwrap_or_default(),
+        probe_audio_delay_us: parse_i64(5, "audio_delay_us")?,
+        probe_video_delay_us: parse_i64(6, "video_delay_us")?,
         ..ParsedCommandArgs::default()
     })
 }
@@ -599,6 +617,8 @@ async fn main() -> Result<()> {
             pulse_period_ms: config.probe_pulse_period_ms,
             pulse_width_ms: config.probe_pulse_width_ms,
             event_width_codes: config.probe_event_width_codes.clone(),
+            audio_delay_us: config.probe_audio_delay_us,
+            video_delay_us: config.probe_video_delay_us,
         };
         let mut stream = client
             .run_output_delay_probe(Request::new(request))
@@ -796,6 +816,8 @@ mod tests {
             "1000",
             "120",
             "1,2,3,4",
+            "0",
+            "157712",
         ])
         .expect("probe config");
 
@@ -805,6 +827,8 @@ mod tests {
         assert_eq!(config.probe_pulse_period_ms, 1000);
         assert_eq!(config.probe_pulse_width_ms, 120);
         assert_eq!(config.probe_event_width_codes, "1,2,3,4");
+        assert_eq!(config.probe_audio_delay_us, 0);
+        assert_eq!(config.probe_video_delay_us, 157_712);
     }
 
     #[test]
@@ -832,7 +856,20 @@ mod tests {
         assert!(parse_args_from(["status", "extra"]).is_err());
         assert!(parse_args_from(["calibrate"]).is_err());
         assert!(parse_args_from(["calibrate", "0", "not-int"]).is_err());
-        assert!(parse_args_from(["output-delay-probe", "1", "2", "3", "4", "1", "extra"]).is_err());
+        assert!(
+            parse_args_from([
+                "output-delay-probe",
+                "1",
+                "2",
+                "3",
+                "4",
+                "1",
+                "0",
+                "0",
+                "extra"
+            ])
+            .is_err()
+        );
     }
 
     #[test]
@@ -938,6 +975,8 @@ mod tests {
             probe_pulse_period_ms: 0,
             probe_pulse_width_ms: 0,
             probe_event_width_codes: String::new(),
+            probe_audio_delay_us: 0,
+            probe_video_delay_us: 0,
         };
         let request = calibration_request_for(&config).expect("request");
         assert_eq!(request.action, CalibrationAction::AdjustActive as i32);

common/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name                    = "lesavka_common"
-version                 = "0.19.3"
+version                 = "0.19.4"
 edition                 = "2024"
 build                   = "build.rs"
 

common/proto/lesavka.proto

@@ -71,6 +71,10 @@ message OutputDelayProbeRequest {
   uint32 pulse_period_ms = 3;
   uint32 pulse_width_ms = 4;
   string event_width_codes = 5;
+  // Probe-local output delays used to test whether server-side scheduling can
+  // compensate the raw UVC/UAC path. Positive values delay that stream.
+  int64 audio_delay_us = 6;
+  int64 video_delay_us = 7;
 }
 
 message OutputDelayProbeReply {

scripts/manual/run_upstream_av_sync.sh

@@ -23,6 +23,8 @@ PROBE_TIMEOUT_SECONDS=${PROBE_TIMEOUT_SECONDS:-$((PROBE_DURATION_SECONDS + PROBE
 PROBE_PULSE_PERIOD_MS=${PROBE_PULSE_PERIOD_MS:-1000}
 PROBE_PULSE_WIDTH_MS=${PROBE_PULSE_WIDTH_MS:-120}
 PROBE_EVENT_WIDTH_CODES=${PROBE_EVENT_WIDTH_CODES:-1,2,1,3,2,4,1,1,3,1,4,2,1,2,3,4,1,3,2,2,4,1,2,4,3,1,1,4,2,3,1,2}
+LESAVKA_OUTPUT_DELAY_PROBE_AUDIO_DELAY_US=${LESAVKA_OUTPUT_DELAY_PROBE_AUDIO_DELAY_US:-0}
+LESAVKA_OUTPUT_DELAY_PROBE_VIDEO_DELAY_US=${LESAVKA_OUTPUT_DELAY_PROBE_VIDEO_DELAY_US:-0}
 # Do not open the UVC host capture far ahead of the probe. The gadget side only
 # has frames once the sync probe is feeding the server, and some hosts time out
 # VIDIOC_STREAMON if the camera is starved during pre-roll.
@@ -1279,7 +1281,9 @@ set +e
       "${PROBE_WARMUP_SECONDS}" \
       "${PROBE_PULSE_PERIOD_MS}" \
       "${PROBE_PULSE_WIDTH_MS}" \
-      "${PROBE_EVENT_WIDTH_CODES}"
+      "${PROBE_EVENT_WIDTH_CODES}" \
+      "${LESAVKA_OUTPUT_DELAY_PROBE_AUDIO_DELAY_US}" \
+      "${LESAVKA_OUTPUT_DELAY_PROBE_VIDEO_DELAY_US}"
 ) 2>&1 | tee "${LOCAL_SERVER_PROBE_REPLY}"
 probe_status=${PIPESTATUS[0]}
 set -e

server/Cargo.toml

@@ -10,7 +10,7 @@ bench                   = false
 
 [package]
 name                    = "lesavka_server"
-version                 = "0.19.3"
+version                 = "0.19.4"
 edition                 = "2024"
 autobins                = false
 

server/src/output_delay_probe.rs

@@ -62,6 +62,8 @@ struct ProbeConfig {
     pulse_period: Duration,
     pulse_width: Duration,
     event_width_codes: Vec<u32>,
+    audio_delay: Duration,
+    video_delay: Duration,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -91,6 +93,8 @@ struct OutputDelayProbeTimeline {
     audio_sample_rate: u32,
     audio_channels: usize,
     audio_chunk_ms: u64,
+    audio_delay_us: u64,
+    video_delay_us: u64,
     pulse_period_ms: u64,
     pulse_width_ms: u64,
     warmup_us: u64,
@@ -140,6 +144,8 @@ impl OutputDelayProbeTimeline {
             audio_sample_rate: AUDIO_SAMPLE_RATE,
             audio_channels: AUDIO_CHANNELS,
             audio_chunk_ms: AUDIO_CHUNK_MS,
+            audio_delay_us: duration_us(config.audio_delay),
+            video_delay_us: duration_us(config.video_delay),
             pulse_period_ms: config.pulse_period.as_millis() as u64,
             pulse_width_ms: config.pulse_width.as_millis() as u64,
             warmup_us: duration_us(config.warmup),
@@ -215,6 +221,8 @@ impl ProbeConfig {
             pulse_period: Duration::from_millis(u64::from(pulse_period_ms)),
             pulse_width: Duration::from_millis(u64::from(pulse_width_ms)),
             event_width_codes,
+            audio_delay: positive_delay(request.audio_delay_us, "audio_delay_us")?,
+            video_delay: positive_delay(request.video_delay_us, "video_delay_us")?,
         })
     }
 
@@ -270,6 +278,13 @@ fn non_zero_or_default(value: u32, default: u32, name: &str) -> Result<u32> {
     Ok(value)
 }
 
+fn positive_delay(value_us: i64, name: &str) -> Result<Duration> {
+    if value_us < 0 {
+        bail!("{name} must be zero or positive for the direct output-delay probe");
+    }
+    Ok(Duration::from_micros(value_us as u64))
+}
+
 fn parse_event_width_codes(raw: &str) -> Result<Vec<u32>> {
     let trimmed = raw.trim();
     if trimmed.is_empty() {
@@ -330,13 +345,24 @@ pub async fn run_server_output_delay_probe(
     loop {
         let next_frame_pts = duration_mul(frame_step, frame_index);
         let next_audio_pts = duration_mul(audio_chunk, audio_index);
-        let next_pts = next_frame_pts.min(next_audio_pts);
+        let frame_active = next_frame_pts <= config.duration;
-        if next_pts > config.duration {
+        let audio_active = next_audio_pts <= config.duration;
+        if !frame_active && !audio_active {
             break;
         }
-        tokio::time::sleep_until(start + next_pts).await;
+        let next_frame_due = if frame_active {
+            next_frame_pts.saturating_add(config.video_delay)
+        } else {
+            Duration::MAX
+        };
+        let next_audio_due = if audio_active {
+            next_audio_pts.saturating_add(config.audio_delay)
+        } else {
+            Duration::MAX
+        };
+        tokio::time::sleep_until(start + next_frame_due.min(next_audio_due)).await;
 
-        if next_audio_pts <= next_frame_pts && next_audio_pts <= config.duration {
+        if audio_active && next_audio_due <= next_frame_due {
             let pts_us = duration_us(next_audio_pts);
             let event_slot = config.event_slot_at(next_audio_pts);
             let data = render_audio_chunk(&config, next_audio_pts, samples_per_chunk);
@@ -358,7 +384,7 @@ pub async fn run_server_output_delay_probe(
             audio_index = audio_index.saturating_add(1);
         }
 
-        if next_frame_pts <= next_audio_pts && next_frame_pts <= config.duration {
+        if frame_active && next_frame_due <= next_audio_due {
             let pts_us = duration_us(next_frame_pts);
             let event_slot = config.event_slot_at(next_frame_pts);
             let code = event_slot.map(|slot| slot.code);
@@ -658,6 +684,8 @@ mod tests {
             pulse_period_ms: 1_000,
             pulse_width_ms: 120,
             event_width_codes: "3".to_string(),
+            audio_delay_us: 0,
+            video_delay_us: 0,
         })
         .expect("config");
 

testing/tests/client_manual_sync_script_contract.rs

@@ -47,6 +47,8 @@ fn upstream_sync_script_tunnels_auto_server_addr_through_ssh() {
         "LESAVKA_OUTPUT_DELAY_MAX_ABS_SKEW_MS=${LESAVKA_OUTPUT_DELAY_MAX_ABS_SKEW_MS:-5000}",
         "LESAVKA_OUTPUT_DELAY_MAX_DRIFT_MS=${LESAVKA_OUTPUT_DELAY_MAX_DRIFT_MS:-80}",
         "LESAVKA_OUTPUT_DELAY_MAX_STEP_US=${LESAVKA_OUTPUT_DELAY_MAX_STEP_US:-1500000}",
+        "LESAVKA_OUTPUT_DELAY_PROBE_AUDIO_DELAY_US=${LESAVKA_OUTPUT_DELAY_PROBE_AUDIO_DELAY_US:-0}",
+        "LESAVKA_OUTPUT_DELAY_PROBE_VIDEO_DELAY_US=${LESAVKA_OUTPUT_DELAY_PROBE_VIDEO_DELAY_US:-0}",
         "write_output_delay_calibration",
         "extract_server_timeline",
         "write_output_delay_correlation",
@@ -64,6 +66,8 @@ fn upstream_sync_script_tunnels_auto_server_addr_through_ssh() {
         "probe_media_path\": \"server generated signatures -> UVC/UAC sinks -> lab host capture\"",
         "audio_after_video_positive",
         "PROBE_EVENT_WIDTH_CODES=${PROBE_EVENT_WIDTH_CODES:-1,2,1,3",
+        "\"${LESAVKA_OUTPUT_DELAY_PROBE_AUDIO_DELAY_US}\"",
+        "\"${LESAVKA_OUTPUT_DELAY_PROBE_VIDEO_DELAY_US}\"",
         "REMOTE_PULSE_CAPTURE_TOOL=${REMOTE_PULSE_CAPTURE_TOOL:-gst}",
         "output_delay_calibration_json",
         "direct UVC/UAC output-delay calibration",