From 079611acfccad652e4b15ca67051b1bbf835e976 Mon Sep 17 00:00:00 2001 From: dailz Date: Sat, 20 Jun 2026 21:53:27 +0800 Subject: [PATCH] fix(webrtc): propagate real capture PTS through WebRTC channel (closes #24) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause (3-stage bug found via Oracle round 1+2 review): Browser WebRTC clients accumulated 2-3 seconds jitter buffer under damage-driven variable frame rate (KWin Portal/PipeWire). User moved mouse, saw action 2-3 seconds later on client. Three coordinated bugs formed a chain that defeated any single-point fix: 1. state_portal.rs:455 used sequential frame counter as PTS instead of real capture time. (Portal path only — wlr-screencopy already correct.) 2. avhw.rs Channel output sent only Vec, DISCARDING AVPacket PTS. Even with correct encoder PTS, timing metadata was thrown away. 3. webrtc.rs:695 computed RTP timestamp as 'frame_number * 90000 / fps' from a counter, ignoring any real PTS. Browser saw uniform 33ms RTP spacing regardless of actual 1.6-57fps variable delivery, growing jitter buffer to compensate for perceived 'network jitter'. Fix (7-step ordered implementation per Oracle round 2): 1. EncodedH264Frame struct in avhw.rs carries data + pts_ticks 2. FrameOutput::Channel type changed from Sender> to Sender; both new_webrtc signatures updated 3. Channel drain in avhw.rs extracts pkt.pts(), normalizes via 'p - start_ts' (mirrors existing Muxer branch logic). Drops packets with missing PTS instead of silently emitting zero. 4. write_h264_frame signature: frame_number:u64 -> pts_ticks:i64 (both WebRtcState and WebRtcInner layers). Extracted pure function rtp_timestamp_from_pts_ticks(pts_ticks, fps) with 5 unit tests covering zero, one-frame, one-second, negative clamp, fps=0. 5. state_portal.rs receiver loop consumes EncodedH264Frame, passes .data and .pts_ticks to write_h264_frame. 6. state.rs (wlr-screencopy) receiver loop updated for compile compatibility — its existing real-time PTS computation at state.rs:606 was already correct, now properly propagates through new channel type. 7. state_portal.rs:467 PTS computation GATED on output mode: - WebRTC branch: compute_capture_pts() uses PipeWire's ns timestamp (or Instant fallback), normalizes to first-frame-origin, converts to encoder time_base units with i128 intermediate math, enforces monotonicity via safe checked_add pattern. - MP4 branch: KEEPS self.frames_encoded as i64 (sequential counter). File output does not need real-time PTS; changing it would alter file playback speed during static periods. Oracle round 2 critical revisions incorporated: - Single-point PTS normalization (only in avhw Channel drain), NOT at source. Avoids double-subtraction with existing Muxer logic. - MP4 path explicitly preserved — real PTS only applied to WebRTC branch. - Safe Rust monotonicity guard (no unsafe pointer tricks). - i64::try_from(ticks_i128) instead of broken i128::try_from(...).unwrap_or(i64::MAX). - pkt.pts() missing -> log + drop, not silent unwrap_or(0). - Updates span 4 files (avhw.rs, webrtc.rs, state_portal.rs, state.rs) because channel type change ripples through both Portal and wlr paths. Verification expectations: - Browser jitter buffer should stabilize at 100-500ms (typical) instead of growing to 2-3 seconds under damage-driven delivery. - chrome://webrtc-internals: jitterBufferDelay / jitterBufferEmittedCount ratio should drop significantly. - Server-side metrics (output_bps, frame rate, IDR size) unchanged. - MP4 file output (--output mode) behavior unchanged. Out of scope (deferred): - frame_age metric fix (Oracle: separate commit to isolate behavioral change from observability change) - VFR encoder redesign (1/fps time_base sufficient for this fix) - MP4 VFR recording (semantic change, separate decision) - Existing client jitter buffers may not auto-shrink; reconnect may be required for users with already-accumulated latency Tests: - cargo build --release: clean, 0 new warnings (19 pre-existing) - cargo test: 96 lib + 96 bin + 3 integration, 0 failed - 5 new RTP unit tests covering edge cases - SAFETY comments preserved verbatim - 4 files changed, +157/-27 lines --- src/avhw.rs | 46 +++++++++++++++++++++++++---- src/state.rs | 11 +++---- src/state_portal.rs | 70 +++++++++++++++++++++++++++++++++++++++------ src/webrtc.rs | 57 ++++++++++++++++++++++++++++++------ 4 files changed, 157 insertions(+), 27 deletions(-) diff --git a/src/avhw.rs b/src/avhw.rs index 8e3d888..1c5b3d8 100644 --- a/src/avhw.rs +++ b/src/avhw.rs @@ -711,9 +711,25 @@ impl EncState { // SwEncState - VAAPI GPU downscale + software H.264 encode // --------------------------------------------------------------------------- +/// Encoded H.264 frame with timing metadata for WebRTC output. +/// +/// MP4 file output (FrameOutput::Muxer) does NOT use this - it writes via +/// avformat which preserves PTS internally. WebRTC output (FrameOutput::Channel) +/// requires explicit PTS propagation so RTP timestamps reflect real capture time. +/// Without this, WebRTC clients' jitter buffers grow to seconds under +/// damage-driven variable frame rate. See issue #24. +#[derive(Debug)] +pub struct EncodedH264Frame { + /// H.264 NAL byte stream (Annex B or AVCC depending on encoder configuration) + pub data: Vec, + /// PTS in encoder time_base units (1/fps seconds), normalized so first frame = 0. + /// Derived from real capture time, NOT frame counter. + pub pts_ticks: i64, +} + pub enum FrameOutput { Muxer(ff::format::context::Output), - Channel(crossbeam_channel::Sender>), + Channel(crossbeam_channel::Sender), } /// Owned CPU NV12 frame data for cross-thread transfer. @@ -1078,7 +1094,7 @@ impl SwEncEncode { fps: u32, bitrate: u64, gop_size: u32, - tx: crossbeam_channel::Sender>, + tx: crossbeam_channel::Sender, webrtc_paused: Arc, bitrate_rx: crossbeam_channel::Receiver, resolution_rx: crossbeam_channel::Receiver, @@ -1377,18 +1393,36 @@ impl SwEncEncode { // slice is copied into a Vec before the packet is unreffed. let data: &[u8] = unsafe { std::slice::from_raw_parts(raw.data, raw.size as usize) }; - match tx.try_send(data.to_vec()) { + // Normalize PTS: subtract starting_timestamp so first frame = 0. + // Mirrors the Muxer branch normalization above. `start_ts` is + // self.starting_timestamp.unwrap_or(0) (passed by caller), so + // when no origin is recorded yet the subtraction is a no-op. + let pts_ticks = match pkt.pts() { + Some(p) => p - start_ts, + None => { + // libx264 should always set PTS; emitting RTP ts=0 + // here would recreate issue #24. Drop the packet. + tracing::warn!( + "encoder produced packet without PTS, dropping" + ); + continue; + } + }; + match tx.try_send(EncodedH264Frame { + data: data.to_vec(), + pts_ticks, + }) { Ok(()) => {} Err(crossbeam_channel::TrySendError::Full(frame)) => { tracing::warn!( "WebRTC channel full, dropping frame: {} bytes lost", - frame.len() + frame.data.len() ); } Err(crossbeam_channel::TrySendError::Disconnected(frame)) => { tracing::warn!( "WebRTC channel disconnected: {} bytes lost", - frame.len() + frame.data.len() ); self.webrtc_disconnected = true; break; @@ -1458,7 +1492,7 @@ impl SwEncState { fps: u32, bitrate: u64, gop_size: u32, - tx: crossbeam_channel::Sender>, + tx: crossbeam_channel::Sender, webrtc_paused: Arc, ) -> Result { tracing::info!( diff --git a/src/state.rs b/src/state.rs index 4513c71..352f4f7 100644 --- a/src/state.rs +++ b/src/state.rs @@ -43,7 +43,7 @@ use ffmpeg_next as ff; use ffmpeg_next::ffi; use crate::args::Args; -use crate::avhw::{AvHwDevCtx, EncState, SwEncState}; +use crate::avhw::{AvHwDevCtx, EncState, EncodedH264Frame, SwEncState}; use crate::cap_wlr_screencopy::CapWlrScreencopy; use crate::fps_limit::FpsLimit; use crate::stats::{FrameTimings, PipelineStats}; @@ -226,8 +226,8 @@ pub struct State { pub drm_device: Option, pub drm_device_from_compositor: Option, pub webrtc: Option, - pub webrtc_tx: Option>>, - webrtc_rx: Option>>, + pub webrtc_tx: Option>, + webrtc_rx: Option>, webrtc_frames_sent: u64, webrtc_paused: Option>, stats: PipelineStats, @@ -701,12 +701,13 @@ impl State { if let Some(ref rx) = self.webrtc_rx { let mut count = 0u32; - while let Ok(data) = rx.try_recv() { + while let Ok(enc_frame) = rx.try_recv() { if !connected { continue; } count += 1; - if let Err(e) = wrtc.write_h264_frame(&data, self.webrtc_frames_sent, self.args.fps) + if let Err(e) = wrtc + .write_h264_frame(&enc_frame.data, enc_frame.pts_ticks, self.args.fps) { tracing::debug!("WebRTC write frame error: {e}"); } diff --git a/src/state_portal.rs b/src/state_portal.rs index c2f5e3d..47f80c8 100644 --- a/src/state_portal.rs +++ b/src/state_portal.rs @@ -9,8 +9,8 @@ use anyhow::{bail, Result}; // 错误处理工具 use crate::args::Args; // 命令行参数 use crate::avhw::{ - self, BitrateCommand, CpuNv12Frame, EncodeOutcome, ResolutionChange, SwEncEncode, SwEncImport, - SwEncState, + self, BitrateCommand, CpuNv12Frame, EncodeOutcome, EncodedH264Frame, ResolutionChange, + SwEncEncode, SwEncImport, SwEncState, }; // 软件编码器状态（VAAPI 导入 + H.264 编码） use crate::cap_portal::{CapPortal, PwCtrlEvent, PwDmaBufFrame}; // PipeWire 屏幕采集端点 use crate::stats::{FrameTimings, PipelineStats}; // 管道统计（帧计时、每秒快照） @@ -64,6 +64,10 @@ pub struct StatePortal { last_capture_arrival: Option, // timestamp of last real frame arrival idle_log_start: Option, // when current idle period began (one-shot DEBUG log guard) shutdown_started: bool, // idempotency guard; plain bool because &mut self is exclusive (not AtomicBool) + // Issue #24: real-capture PTS origin/tracking for WebRTC RTP timestamps. + first_pts_ns: Option, + capture_start: Option, + last_pts_emitted: Option, } impl StatePortal { @@ -107,6 +111,9 @@ impl StatePortal { last_capture_arrival: None, idle_log_start: None, shutdown_started: false, + first_pts_ns: None, + capture_start: None, + last_pts_emitted: None, }) } @@ -452,7 +459,16 @@ impl StatePortal { } } let t_import_start = Instant::now(); - let pts = self.frames_encoded as i64; + // WebRTC: use real PipeWire capture time so RTP timestamps reflect reality + // (sequential counter caused client jitter buffers to grow to 2-3s under + // damage-driven variable fps — issue #24). MP4: keep sequential counter; + // file output doesn't need real-time PTS and changing it would alter + // playback speed during static periods. + let pts = if self.webrtc.is_some() { + self.compute_capture_pts(frame.pts) + } else { + self.frames_encoded as i64 + }; if let Some(enc) = self.enc.as_mut() { // 将 DMA-BUF 帧零拷贝导入 VAAPI 硬件帧池 @@ -537,6 +553,43 @@ impl StatePortal { Ok(()) } + /// Compute PTS in encoder time_base units (1/fps) from PipeWire's nanosecond + /// capture timestamp. Falls back to `Instant`-based elapsed time when PipeWire + /// does not provide PTS. Maintains strict monotonicity (encoder requirement). + fn compute_capture_pts(&mut self, pw_pts_ns: i64) -> i64 { + const NS_PER_SEC: i128 = 1_000_000_000; + + let raw_ns: i128 = if pw_pts_ns > 0 { + i128::from(pw_pts_ns) + } else { + let start = self.capture_start.get_or_insert_with(Instant::now); + i128::try_from(start.elapsed().as_nanos()).unwrap_or(0) + }; + + if self.first_pts_ns.is_none() && raw_ns > 0 { + self.first_pts_ns = Some(raw_ns); + } + + let origin = self.first_pts_ns.unwrap_or(0); + let relative_ns = if raw_ns >= origin { + raw_ns - origin + } else { + // PipeWire PTS went backwards (stream restart) — reset origin. + self.first_pts_ns = Some(raw_ns); + 0 + }; + let ticks_i128 = (relative_ns.saturating_mul(i128::from(self.args.fps))) / NS_PER_SEC; + let mut pts = i64::try_from(ticks_i128).unwrap_or(i64::MAX); + + if let Some(last) = self.last_pts_emitted { + if pts <= last { + pts = last.checked_add(1).unwrap_or(last); + } + } + self.last_pts_emitted = Some(pts); + pts + } + /// 关闭状态：刷新编码器并清理资源（幂等）。 /// /// `shutdown_started` 守卫在清理之前置位——防止 panic 时 `Drop` 重入 unwinding。 @@ -633,7 +686,7 @@ fn encode_thread_loop( fn webrtc_thread_loop( mut wrtc: WebRtcState, - webrtc_rx: crossbeam_channel::Receiver>, + webrtc_rx: crossbeam_channel::Receiver, fps: u32, enc_width: u32, enc_height: u32, @@ -736,8 +789,8 @@ fn webrtc_thread_loop( } if connected { - while let Ok(data) = webrtc_rx.try_recv() { - if let Err(e) = wrtc.write_h264_frame(&data, frames_sent, fps) { + while let Ok(enc_frame) = webrtc_rx.try_recv() { + if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks, fps) { tracing::debug!("WebRTC write frame error: {e}"); } frames_sent = frames_sent.saturating_add(1); @@ -752,9 +805,10 @@ fn webrtc_thread_loop( } match webrtc_rx.recv_timeout(timeout) { - Ok(data) => { + Ok(enc_frame) => { if wrtc.is_connected() { - if let Err(e) = wrtc.write_h264_frame(&data, frames_sent, fps) { + if let Err(e) = wrtc.write_h264_frame(&enc_frame.data, enc_frame.pts_ticks, fps) + { tracing::debug!("WebRTC write frame error: {e}"); } frames_sent = frames_sent.saturating_add(1); diff --git a/src/webrtc.rs b/src/webrtc.rs index 3aff975..36ee2e9 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -334,9 +334,9 @@ impl WebRtcState { self.poll_rtc() } - pub fn write_h264_frame(&mut self, data: &[u8], frame_number: u64, fps: u32) -> Result<()> { + pub fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64, fps: u32) -> Result<()> { let should_destroy = if let Some(inner) = self.inner.as_mut() { - inner.write_h264_frame(data, frame_number, fps)? + inner.write_h264_frame(data, pts_ticks, fps)? } else { false }; @@ -655,7 +655,7 @@ impl WebRtcInner { Ok(()) } - fn write_h264_frame(&mut self, data: &[u8], frame_number: u64, fps: u32) -> Result { + fn write_h264_frame(&mut self, data: &[u8], pts_ticks: i64, fps: u32) -> Result { if !self.connected { return Ok(false); } @@ -690,11 +690,9 @@ impl WebRtcInner { self.need_keyframe = false; } - let ticks_per_second = 90_000u64; - let fps = fps.max(1) as u64; - let rtp_timestamp = frame_number.saturating_mul(ticks_per_second) / fps; - self.rtp_clock = rtp_timestamp as u32; - let rtp_time = MediaTime::new(rtp_timestamp, Frequency::NINETY_KHZ); + let rtp_timestamp = rtp_timestamp_from_pts_ticks(pts_ticks, fps); + self.rtp_clock = rtp_timestamp; + let rtp_time = MediaTime::new(rtp_timestamp as u64, Frequency::NINETY_KHZ); let writer = match self.rtc.writer(mid) { Some(w) => w, @@ -724,6 +722,19 @@ impl WebRtcInner { } } +/// Convert PTS in encoder time_base units (1/fps) to RTP timestamp (90kHz clock). +/// +/// Extracted as a pure function for unit testing. Clamps negative pts_ticks to 0 +/// (encoder should never emit negative PTS, but defensive). Saturating multiply +/// to avoid overflow on long sessions. +pub fn rtp_timestamp_from_pts_ticks(pts_ticks: i64, fps: u32) -> u32 { + const TICKS_PER_SECOND: u64 = 90_000; + let fps_safe = (fps.max(1) as u64).max(1); + let pts_u64 = (pts_ticks.max(0) as u64).min(u64::MAX / TICKS_PER_SECOND); + let rtp_ts = pts_u64.saturating_mul(TICKS_PER_SECOND) / fps_safe; + rtp_ts as u32 +} + // ── 工具函数 ────────────────────────────────────────────────────────────── /// 从 HTTP 请求中提取 body（在 \r\n\r\n 之后） @@ -866,4 +877,34 @@ mod tests { let bps = default.as_u64(); assert_eq!(bps, 5_000_000); } + + // ── RTP timestamp conversion (issue #24) ── + + #[test] + fn rtp_timestamp_zero_pts() { + assert_eq!(rtp_timestamp_from_pts_ticks(0, 30), 0); + } + + #[test] + fn rtp_timestamp_one_frame() { + // 1 frame at 30fps = 33ms = 3000 RTP ticks (90kHz / 30) + assert_eq!(rtp_timestamp_from_pts_ticks(1, 30), 3000); + } + + #[test] + fn rtp_timestamp_one_second() { + // 30 frames at 30fps = 1 second = 90000 RTP ticks + assert_eq!(rtp_timestamp_from_pts_ticks(30, 30), 90000); + } + + #[test] + fn rtp_timestamp_negative_clamps_to_zero() { + assert_eq!(rtp_timestamp_from_pts_ticks(-5, 30), 0); + } + + #[test] + fn rtp_timestamp_zero_fps_does_not_panic() { + // fps=0 should clamp to 1 internally, not divide by zero + let _ = rtp_timestamp_from_pts_ticks(100, 0); + } }