fix(server): align uvc sink to session clock

server/src/video_sinks/webcam_sink.rs

@@ -20,10 +20,24 @@ use crate::video_support::{contains_idr, dev_mode_enabled, pick_h264_decoder, re
 pub struct WebcamSink {
     appsrc: gst_app::AppSrc,
     pipe: gst::Pipeline,
+    clock_aligned: AtomicBool,
     next_pts_us: AtomicU64,
     frame_step_us: u64,
 }
 
+fn uvc_sink_session_clock_align_enabled() -> bool {
+    std::env::var("LESAVKA_UVC_SESSION_CLOCK_ALIGN")
+        .ok()
+        .map(|value| {
+            let trimmed = value.trim();
+            !(trimmed.eq_ignore_ascii_case("0")
+                || trimmed.eq_ignore_ascii_case("false")
+                || trimmed.eq_ignore_ascii_case("no")
+                || trimmed.eq_ignore_ascii_case("off"))
+        })
+        .unwrap_or(true)
+}
+
 impl WebcamSink {
     /// Build a new webcam sink pipeline.
     ///
@@ -36,6 +50,7 @@ impl WebcamSink {
         gst::init()?;
 
         let pipeline = gst::Pipeline::new();
+        let clock_align_enabled = uvc_sink_session_clock_align_enabled();
         let src = gst::ElementFactory::make("appsrc")
             .build()?
             .downcast::<gst_app::AppSrc>()
@@ -47,6 +62,10 @@ impl WebcamSink {
         let sink = gst::ElementFactory::make("fakesink")
             .build()
             .context("building fakesink")?;
+        if clock_align_enabled {
+            crate::media_timing::prepare_pipeline_clock_sync(&pipeline);
+            crate::media_timing::enable_sink_clock_sync(&sink);
+        }
         pipeline.add_many(&[src.upcast_ref(), &sink])?;
         gst::Element::link_many(&[src.upcast_ref(), &sink])?;
         pipeline.set_state(gst::State::Playing)?;
@@ -55,6 +74,7 @@ impl WebcamSink {
         Ok(Self {
             appsrc: src,
             pipe: pipeline,
+            clock_aligned: AtomicBool::new(!clock_align_enabled),
             next_pts_us: AtomicU64::new(0),
             frame_step_us,
         })
@@ -65,6 +85,7 @@ impl WebcamSink {
         gst::init()?;
 
         let pipeline = gst::Pipeline::new();
+        let clock_align_enabled = uvc_sink_session_clock_align_enabled();
 
         let width = cfg.width as i32;
         let height = cfg.height as i32;
@@ -83,6 +104,9 @@ impl WebcamSink {
             .map(|value| value != "0")
             .unwrap_or(false);
         src.set_property("block", block);
+        if clock_align_enabled {
+            crate::media_timing::prepare_pipeline_clock_sync(&pipeline);
+        }
 
         if use_mjpeg {
             let caps_mjpeg = gst::Caps::builder("image/jpeg")
@@ -101,8 +125,12 @@ impl WebcamSink {
                 .build()?;
             let sink = gst::ElementFactory::make("v4l2sink")
                 .property("device", uvc_dev)
-                .property("sync", false)
                 .build()?;
+            if clock_align_enabled {
+                crate::media_timing::enable_sink_clock_sync(&sink);
+            } else if sink.has_property("sync", None) {
+                sink.set_property("sync", false);
+            }
 
             pipeline.add_many([src.upcast_ref(), &queue, &capsfilter, &sink])?;
             gst::Element::link_many([src.upcast_ref(), &queue, &capsfilter, &sink])?;
@@ -131,8 +159,12 @@ impl WebcamSink {
                 .build()?;
             let sink = gst::ElementFactory::make("v4l2sink")
                 .property("device", uvc_dev)
-                .property("sync", false)
                 .build()?;
+            if clock_align_enabled {
+                crate::media_timing::enable_sink_clock_sync(&sink);
+            } else if sink.has_property("sync", None) {
+                sink.set_property("sync", false);
+            }
 
             pipeline.add_many([
                 src.upcast_ref(),
@@ -159,6 +191,7 @@ impl WebcamSink {
         Ok(Self {
             appsrc: src,
             pipe: pipeline,
+            clock_aligned: AtomicBool::new(!clock_align_enabled),
             next_pts_us: AtomicU64::new(0),
             frame_step_us,
         })
@@ -181,6 +214,12 @@ impl WebcamSink {
         let mut buf = gst::Buffer::from_slice(pkt.data);
         if let Some(meta) = buf.get_mut() {
             let pts_us = reserve_local_pts(&self.next_pts_us, pkt.pts, self.frame_step_us);
+            if !self
+                .clock_aligned
+                .swap(true, std::sync::atomic::Ordering::SeqCst)
+            {
+                crate::media_timing::align_pipeline_to_session_clock(&self.pipe, pts_us);
+            }
             let ts = gst::ClockTime::from_useconds(pts_us);
             meta.set_pts(Some(ts));
             meta.set_dts(Some(ts));
@@ -197,3 +236,23 @@ impl Drop for WebcamSink {
         let _ = self.pipe.set_state(gst::State::Null);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn uvc_session_clock_alignment_defaults_on_and_accepts_disable_overrides() {
+        temp_env::with_var_unset("LESAVKA_UVC_SESSION_CLOCK_ALIGN", || {
+            assert!(super::uvc_sink_session_clock_align_enabled());
+        });
+
+        for disabled in ["0", "false", "no", "off"] {
+            temp_env::with_var("LESAVKA_UVC_SESSION_CLOCK_ALIGN", Some(disabled), || {
+                assert!(!super::uvc_sink_session_clock_align_enabled());
+            });
+        }
+
+        temp_env::with_var("LESAVKA_UVC_SESSION_CLOCK_ALIGN", Some("1"), || {
+            assert!(super::uvc_sink_session_clock_align_enabled());
+        });
+    }
+}