feat(portal): independent WebRTC thread + channel tuning for 60fps mouse latency

- Move WebRTC send to dedicated wl-webrtc-webrtc thread (was inline in main loop)
- Reduce frame_rx 16→1, input_tx 2→1 (drop-on-full), webrtc_tx 32→2
- Recv_timeout 10ms→2ms to reduce pipeline latency
- Fix sent_gap_p95 stats bug: compute gap at actual send time in WebRTC
  thread instead of batch-draining at snapshot time (was always 0.0ms)
- High profile via AVCodecContext.profile, veryfast preset, 5x bitrate
- Stats drain via sent_gap channel with record_send_from_thread()
- Shutdown: drop input_tx → join encode → drop webrtc_tx → join webrtc

.gitignore

@@ -17,3 +17,7 @@ Thumbs.db
 
 # Sisyphus orchestration artifacts
 .sisyphus/
+.omo/
+.playwright-mcp/
+wl-webrtc.log
+webrtc-p0-success.png

Cargo.lock

@@ -1126,6 +1126,15 @@ version = "0.4.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
 
+[[package]]
+name = "matchers"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"
+dependencies = [
+ "regex-automata",
+]
+
 [[package]]
 name = "memchr"
 version = "2.8.0"
@@ -2025,10 +2034,14 @@ version = "0.3.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319"
 dependencies = [
+ "matchers",
  "nu-ansi-term",
+ "once_cell",
+ "regex-automata",
  "sharded-slab",
  "smallvec",
  "thread_local",
+ "tracing",
  "tracing-core",
  "tracing-log",
 ]

Cargo.toml

@@ -14,7 +14,7 @@ signal-hook = "0.3"
 signal-hook-mio = { version = "0.2", features = ["support-v1_0"] }
 clap = { version = "4", features = ["derive"] }
 tracing = "0.1"
-tracing-subscriber = "0.3"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 anyhow = "1"
 drm = "0.12"
 drm-fourcc = "2"

src/args.rs

@@ -50,4 +50,8 @@ pub struct Args {
     /// Force re-authorization dialog (ignore saved portal restore token)
     #[arg(long)]
     pub no_persist: bool,
+
+    /// Enable per-second pipeline statistics output for stutter diagnosis
+    #[arg(long)]
+    pub stats: bool,
 }

src/avhw.rs

@@ -3,6 +3,7 @@ use std::mem;
 use std::os::fd::{AsRawFd, RawFd};
 use std::os::raw::c_void;
 use std::path::Path;
+use std::slice;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::ptr;
@@ -45,8 +46,9 @@ impl AvHwDevCtx {
         };
         if ret < 0 {
             bail!(
-                "Failed to create VAAPI device context from {}: error {ret}",
-                drm_device.display()
+                "Failed to create VAAPI device context from {}: {}",
+                drm_device.display(),
+                ff_err(ret)
             );
         }
         Ok(Self { ptr: p })
@@ -108,7 +110,7 @@ impl AvHwFrameCtx {
         if ret < 0 {
             // SAFETY: p is valid but init failed; clean up.
             unsafe { ffi::av_buffer_unref(&mut p) };
-            bail!("av_hwframe_ctx_init failed: error {ret}");
+            bail!("av_hwframe_ctx_init failed: {}", ff_err(ret));
         }
         Ok(Self { ptr: p })
     }
@@ -244,8 +246,7 @@ pub unsafe fn import_dma_buf_to_vaapi(
         )
     };
     if ret < 0 {
-        let err_str = av_err_to_string(ret);
-        bail!("av_hwframe_map failed: error {ret} ({err_str})");
+        bail!("av_hwframe_map failed: {}", ff_err(ret));
     }
 
     Ok(dst)
@@ -259,7 +260,8 @@ unsafe extern "C" fn cleanup_drm_descriptor(_opaque: *mut c_void, data: *mut u8)
     let _ = Box::from_raw(data as *mut ffi::AVDRMFrameDescriptor);
 }
 
-fn av_err_to_string(err: i32) -> String {
+/// Convert an FFmpeg error code to a human-readable string.
+pub(crate) fn av_err_to_string(err: i32) -> String {
     let mut buf = vec![0u8; 128];
     // SAFETY: buf points to 128 writable bytes and lives for the duration of
     // av_strerror.
@@ -270,6 +272,12 @@ fn av_err_to_string(err: i32) -> String {
         .trim_end_matches('\0')
         .to_string()
 }
+
+/// Format an FFmpeg error code with both numeric value and description.
+/// Example output: "error -22 (Invalid argument)"
+pub(crate) fn ff_err(ret: i32) -> String {
+    format!("error {ret} ({})", av_err_to_string(ret))
+}
 // ---------------------------------------------------------------------------
 // EncState
 // ---------------------------------------------------------------------------
@@ -326,7 +334,7 @@ impl EncState {
             transform,
         )?;
 
-        let mut sink_ctx = video_filter.get("out").unwrap();
+        let mut sink_ctx = video_filter.get("out").ok_or_else(|| anyhow::anyhow!("filter 'out' not found"))?;
         // SAFETY: sink_ctx is a live buffersink; the returned hw_frames_ctx is
         // borrowed, so av_buffer_ref creates an owned reference.
         let sink_hw_frames = unsafe {
@@ -371,6 +379,15 @@ impl EncState {
     enc.set_time_base(ff::Rational::new(1, fps as i32));
     enc.set_max_b_frames(0);
 
+    // VBV rate limiting: caps IDR burst size for WebRTC.  Without this a 4K
+    // scene change can produce a 256KB keyframe that overflows the UDP send
+    // buffer.  bufsize=bitrate/4 ≈ 250ms of video at the target bitrate.
+    unsafe {
+        let ctx_ptr = enc.as_mut_ptr();
+        (*ctx_ptr).rc_max_rate = bitrate as i64;
+        (*ctx_ptr).rc_buffer_size = (bitrate / 4) as i32;
+    }
+
         // SAFETY: AV_CODEC_FLAG_GLOBAL_HEADER must be set BEFORE opening the encoder.
         // It triggers SPS/PPS extradata generation needed by the muxer for
         // Annex B to AVCC conversion.
@@ -395,7 +412,7 @@ impl EncState {
                 ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0)
             };
             if ret < 0 {
-                tracing::warn!("av_opt_set repeat_pps failed (error {ret}), likely FFmpeg < 7.0; continuing without per-frame PPS");
+                tracing::warn!("av_opt_set repeat_pps failed ({}), likely FFmpeg < 7.0; continuing without per-frame PPS", ff_err(ret));
             }
         }
 
@@ -420,7 +437,7 @@ impl EncState {
             )
         };
         if ret < 0 || fmt_ctx_ptr.is_null() {
-            bail!("Failed to allocate output format context: error {ret}");
+            bail!("Failed to allocate output format context: {}", ff_err(ret));
         }
 
         // SAFETY: avformat_query_codec checks codec+format compatibility.
@@ -444,7 +461,7 @@ impl EncState {
             ffi::avcodec_parameters_from_context((*stream_ptr).codecpar, enc_video.as_ptr())
         };
         if ret < 0 {
-            bail!("Failed to copy encoder parameters to stream: error {ret}");
+            bail!("Failed to copy encoder parameters to stream: {}", ff_err(ret));
         }
 
         // SAFETY: Copy encoder time_base to stream.
@@ -462,15 +479,16 @@ impl EncState {
         };
         if ret < 0 {
             bail!(
-                "Failed to open output file '{}': error {ret}",
-                output_path.display()
+                "Failed to open output file '{}': {}",
+                output_path.display(),
+                ff_err(ret)
             );
         }
 
         // SAFETY: avformat_write_header writes the container header.
         let ret = unsafe { ffi::avformat_write_header(fmt_ctx_ptr, ptr::null_mut()) };
         if ret < 0 {
-            bail!("Failed to write output header: error {ret}");
+            bail!("Failed to write output header: {}", ff_err(ret));
         }
 
         // SAFETY: We created fmt_ctx_ptr above and it's valid.
@@ -492,9 +510,9 @@ impl EncState {
     }
 
     pub fn encode_frame(&mut self, hw_frame: &ff::frame::Video) -> Result<()> {
-        let mut filter_src_ctx = self.video_filter.get("in").unwrap();
+        let mut filter_src_ctx = self.video_filter.get("in").ok_or_else(|| anyhow::anyhow!("filter 'in' not found"))?;
         let mut filter_src = filter_src_ctx.source();
-        let mut filter_sink_ctx = self.video_filter.get("out").unwrap();
+        let mut filter_sink_ctx = self.video_filter.get("out").ok_or_else(|| anyhow::anyhow!("filter 'out' not found"))?;
         let mut filter_sink = filter_sink_ctx.sink();
 
         // SAFETY: hw_frame is a valid VAAPI hardware frame from capture.
@@ -524,7 +542,7 @@ impl EncState {
             let ret =
                 unsafe { ffi::avcodec_send_frame(self.enc_video.as_mut_ptr(), filtered.as_ptr()) };
             if ret < 0 {
-                bail!("avcodec_send_frame failed: error {ret}");
+                bail!("avcodec_send_frame failed: {}", ff_err(ret));
             }
             self.drain_encoder(start_ts)?;
         }
@@ -534,12 +552,14 @@ impl EncState {
 
     pub fn flush(&mut self) -> Result<()> {
         // Flush filter graph
-        let mut filter_src_ctx = self.video_filter.get("in").unwrap();
+        let mut filter_src_ctx = self.video_filter.get("in").ok_or_else(|| anyhow::anyhow!("filter 'in' not found"))?;
         let mut filter_src = filter_src_ctx.source();
-        let _ = filter_src.flush();
+        if let Err(e) = filter_src.flush() {
+            tracing::debug!("filter source flush error: {e}");
+        }
 
         // Drain filter
-        let mut filter_sink_ctx = self.video_filter.get("out").unwrap();
+        let mut filter_sink_ctx = self.video_filter.get("out").ok_or_else(|| anyhow::anyhow!("filter 'out' not found"))?;
         let mut filter_sink = filter_sink_ctx.sink();
         loop {
             let mut filtered = ff::frame::Video::empty();
@@ -552,7 +572,7 @@ impl EncState {
                         ffi::avcodec_send_frame(self.enc_video.as_mut_ptr(), filtered.as_ptr())
                     };
                     if ret < 0 {
-                        bail!("avcodec_send_frame failed during flush: error {ret}");
+                        bail!("avcodec_send_frame failed during flush: {}", ff_err(ret));
                     }
                     self.drain_encoder(start_ts)?;
                 }
@@ -589,7 +609,7 @@ impl EncState {
                 if ret == ffi::AVERROR(ffi::EAGAIN) || ret == ffi::AVERROR_EOF {
                     break;
                 }
-                bail!("avcodec_receive_packet failed: error {ret}");
+                bail!("avcodec_receive_packet failed: {}", ff_err(ret));
             }
 
             // Rescale timestamps from encoder time_base to stream time_base
@@ -633,39 +653,34 @@ pub enum FrameOutput {
     Channel(crossbeam_channel::Sender<Vec<u8>>),
 }
 
-pub struct SwEncState {
+/// Owned CPU NV12 frame data for cross-thread transfer.
+/// Produced by main thread (VAAPI import + GPU scale + transfer), consumed by encode thread.
+pub struct CpuNv12Frame {
+    pub y_data: Vec<u8>,
+    pub uv_data: Vec<u8>,
+    pub y_stride: usize,
+    pub uv_stride: usize,
+    pub pts: i64,
+}
+
+pub struct SwEncImport {
     hw_dev: AvHwDevCtx,
     frames_rgb: AvHwFrameCtx,
     filter_graph: ff::filter::Graph,
-    sws_ctx: *mut ffi::SwsContext,
-    enc_video: ff::codec::encoder::video::Video,
-    output: Option<FrameOutput>,
-    yuv_frame: *mut ffi::AVFrame,
-    starting_timestamp: Option<i64>,
-    frames_written: bool,
-    webrtc_disconnected: bool,
-    webrtc_paused: Option<Arc<AtomicBool>>,
+    enc_width: u32,
+    enc_height: u32,
 }
 
-unsafe impl Send for SwEncState {}
-
-impl SwEncState {
+impl SwEncImport {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         drm_device: &Path,
-        output_path: &Path,
         width: u32,
         height: u32,
         enc_width: u32,
         enc_height: u32,
         fps: u32,
-        bitrate: u64,
-        gop_size: u32,
     ) -> Result<Self> {
-        tracing::info!(
-            "SwEncState::new: GPU downscale {width}x{height} BGRA -> {enc_width}x{enc_height} NV12, software H.264"
-        );
-
         let hw_dev = AvHwDevCtx::new_vaapi(drm_device)?;
         let frames_rgb =
             AvHwFrameCtx::for_capture(&hw_dev, width, height, ff::format::Pixel::BGRA)?;
@@ -679,88 +694,32 @@ impl SwEncState {
             fps,
         )?;
 
-        let sws_ctx = create_nv12_to_yuv420p_sws(enc_width, enc_height)?;
-        let (enc_video, octx) =
-            create_software_h264_muxer(output_path, enc_width, enc_height, fps, bitrate, gop_size)?;
-        let yuv_frame = alloc_yuv420p_frame(enc_width, enc_height)?;
-
         Ok(Self {
             hw_dev,
             frames_rgb,
             filter_graph,
-            sws_ctx,
-            enc_video,
-            output: Some(FrameOutput::Muxer(octx)),
-            yuv_frame,
-            starting_timestamp: None,
-            frames_written: false,
-            webrtc_disconnected: false,
-            webrtc_paused: None,
-        })
-    }
-
-    #[allow(clippy::too_many_arguments)]
-    pub fn new_webrtc(
-        drm_device: &Path,
-        width: u32,
-        height: u32,
-        enc_width: u32,
-        enc_height: u32,
-        fps: u32,
-        bitrate: u64,
-        gop_size: u32,
-        tx: crossbeam_channel::Sender<Vec<u8>>,
-        webrtc_paused: Arc<AtomicBool>,
-    ) -> Result<Self> {
-        tracing::info!(
-            "SwEncState::new_webrtc: GPU downscale {width}x{height} BGRA -> {enc_width}x{enc_height} NV12, software H.264 -> WebRTC"
-        );
-
-        let hw_dev = AvHwDevCtx::new_vaapi(drm_device)?;
-        let frames_rgb =
-            AvHwFrameCtx::for_capture(&hw_dev, width, height, ff::format::Pixel::BGRA)?;
-        let filter_graph = build_swenc_filter_graph(
-            &hw_dev,
-            &frames_rgb,
-            width,
-            height,
             enc_width,
             enc_height,
-            fps,
-        )?;
-        let sws_ctx = create_nv12_to_yuv420p_sws(enc_width, enc_height)?;
-        let enc_video = create_software_h264_encoder(enc_width, enc_height, fps, bitrate, gop_size)?;
-        let yuv_frame = alloc_yuv420p_frame(enc_width, enc_height)?;
-
-        Ok(Self {
-            hw_dev,
-            frames_rgb,
-            filter_graph,
-            sws_ctx,
-            enc_video,
-            output: Some(FrameOutput::Channel(tx)),
-            yuv_frame,
-            starting_timestamp: None,
-            frames_written: false,
-            webrtc_disconnected: false,
-            webrtc_paused: Some(webrtc_paused),
         })
     }
 
     pub fn frames_rgb(&self) -> &AvHwFrameCtx {
+        let _ = self.hw_dev.as_ptr();
         &self.frames_rgb
     }
 
-    pub fn encode_frame(&mut self, hw_frame: &ff::frame::Video) -> Result<()> {
-        let mut filter_src_ctx = self.filter_graph.get("in").unwrap();
+    pub fn import_and_scale(&mut self, hw_frame: &ff::frame::Video) -> Result<CpuNv12Frame> {
+        let mut filter_src_ctx = self.filter_graph.get("in").ok_or_else(|| anyhow::anyhow!("filter 'in' not found"))?;
         let mut filter_src = filter_src_ctx.source();
-        let mut filter_sink_ctx = self.filter_graph.get("out").unwrap();
+        let mut filter_sink_ctx = self.filter_graph.get("out").ok_or_else(|| anyhow::anyhow!("filter 'out' not found"))?;
         let mut filter_sink = filter_sink_ctx.sink();
 
         filter_src
             .add(hw_frame)
             .map_err(|e| anyhow::anyhow!("software pipeline filter source add failed: {e}"))?;
 
+        let mut first = None;
+        let mut extra_count = 0usize;
         loop {
             let mut filtered = ff::frame::Video::empty();
             match filter_sink.frame(&mut filtered) {
@@ -768,61 +727,47 @@ impl SwEncState {
                     if filtered.pts().is_none() {
                         filtered.set_pts(hw_frame.pts());
                     }
-                    self.encode_filtered_frame(&filtered)?;
+                    let cpu_frame = self.transfer_filtered_to_cpu(&filtered)?;
+                    if first.is_none() {
+                        first = Some(cpu_frame);
+                    } else {
+                        extra_count += 1;
+                    }
                 }
                 Err(ff::Error::Other { errno }) if errno == ffi::EAGAIN => break,
                 Err(e) => bail!("software pipeline filter sink get frame failed: {e}"),
             }
         }
 
-        Ok(())
+        if extra_count > 0 {
+            tracing::warn!("software import filter produced {extra_count} extra frame(s); dropping extras");
         }
 
-    pub fn flush(&mut self) -> Result<()> {
-        let mut filter_src_ctx = self.filter_graph.get("in").unwrap();
-        let mut filter_src = filter_src_ctx.source();
-        let _ = filter_src.flush();
+        first.ok_or_else(|| anyhow::anyhow!("software pipeline produced no scaled frame"))
+    }
 
-        let mut filter_sink_ctx = self.filter_graph.get("out").unwrap();
+    pub fn flush_import(&mut self) -> Result<Vec<CpuNv12Frame>> {
+        let mut filter_src_ctx = self.filter_graph.get("in").ok_or_else(|| anyhow::anyhow!("filter 'in' not found"))?;
+        let mut filter_src = filter_src_ctx.source();
+        if let Err(e) = filter_src.flush() {
+            tracing::debug!("filter source flush error: {e}");
+        }
+
+        let mut filter_sink_ctx = self.filter_graph.get("out").ok_or_else(|| anyhow::anyhow!("filter 'out' not found"))?;
         let mut filter_sink = filter_sink_ctx.sink();
+        let mut frames = Vec::new();
         loop {
             let mut filtered = ff::frame::Video::empty();
             match filter_sink.frame(&mut filtered) {
-                Ok(()) => self.encode_filtered_frame(&filtered)?,
+                Ok(()) => frames.push(self.transfer_filtered_to_cpu(&filtered)?),
                 Err(_) => break,
             }
         }
 
-        // SAFETY: Sending a null frame flushes the opened software encoder;
-        // no frame data is dereferenced. enc_video is exclusively borrowed via &mut self.
-        unsafe {
-            let ret = ffi::avcodec_send_frame(self.enc_video.as_mut_ptr(), ptr::null());
-            if ret < 0 && ret != ffi::AVERROR_EOF {
-                bail!("software encoder flush send failed: error {ret}");
-            }
-        }
-        let start_ts = self.starting_timestamp.unwrap_or(0);
-        self.drain_encoder(start_ts)?;
-
-        if self.frames_written {
-            if let Some(FrameOutput::Muxer(ref mut octx)) = self.output {
-                octx.write_trailer()
-                    .map_err(|e| anyhow::anyhow!("Failed to write trailer: {e}"))?;
-            }
+        Ok(frames)
     }
 
-        Ok(())
-    }
-
-    fn encode_filtered_frame(&mut self, filtered: &ff::frame::Video) -> Result<()> {
-        if self.webrtc_disconnected {
-            return Ok(());
-        }
-        if let Some(ref paused) = self.webrtc_paused {
-            if paused.load(Ordering::Relaxed) {
-                return Ok(());
-            }
-        }
+    fn transfer_filtered_to_cpu(&self, filtered: &ff::frame::Video) -> Result<CpuNv12Frame> {
         // SAFETY: av_frame_alloc returns a newly allocated AVFrame or null,
         // which is checked below.
         let mut sw_nv12 = unsafe { ffi::av_frame_alloc() };
@@ -837,32 +782,171 @@ impl SwEncState {
             // SAFETY: sw_nv12 was allocated above and has not been freed yet.
             unsafe { ffi::av_frame_free(&mut sw_nv12) };
             bail!(
-                "av_hwframe_transfer_data failed for GPU-downscaled frame: error {transfer_ret} ({})",
-                av_err_to_string(transfer_ret)
+                "av_hwframe_transfer_data failed for GPU-downscaled frame: {}",
+                ff_err(transfer_ret)
             );
         }
 
+        // SAFETY: sw_nv12 was filled by av_hwframe_transfer_data. NV12 planes 0 and 1 are
+        // initialized for enc_width x enc_height; linesize values define each row's byte span.
+        let frame = unsafe {
+            let y_ptr = (*sw_nv12).data[0];
+            let uv_ptr = (*sw_nv12).data[1];
+            if y_ptr.is_null() || uv_ptr.is_null() {
+                ffi::av_frame_free(&mut sw_nv12);
+                bail!("NV12 transfer frame missing Y/UV plane data");
+            }
+            let y_stride = (*sw_nv12).linesize[0] as usize;
+            let uv_stride = (*sw_nv12).linesize[1] as usize;
+            if (*sw_nv12).width != self.enc_width as i32 || (*sw_nv12).height != self.enc_height as i32 {
+                ffi::av_frame_free(&mut sw_nv12);
+                bail!("NV12 transfer frame has unexpected dimensions");
+            }
+            let y_len = y_stride * self.enc_height as usize;
+            let uv_len = uv_stride * (self.enc_height as usize / 2);
+            let y_data = slice::from_raw_parts(y_ptr, y_len).to_vec();
+            let uv_data = slice::from_raw_parts(uv_ptr, uv_len).to_vec();
+            let pts = filtered.pts().unwrap_or(0);
+            ffi::av_frame_free(&mut sw_nv12);
+            CpuNv12Frame {
+                y_data,
+                uv_data,
+                y_stride,
+                uv_stride,
+                pts,
+            }
+        };
+
+        Ok(frame)
+    }
+}
+
+pub struct SwEncEncode {
+    sws_ctx: *mut ffi::SwsContext,
+    enc_video: ff::codec::encoder::video::Video,
+    output: Option<FrameOutput>,
+    yuv_frame: *mut ffi::AVFrame,
+    starting_timestamp: Option<i64>,
+    frames_written: bool,
+    webrtc_disconnected: bool,
+    webrtc_paused: Option<Arc<AtomicBool>>,
+    enc_width: u32,
+    enc_height: u32,
+}
+
+// SAFETY: SwEncEncode owns sws_ctx/yuv_frame/enc_video exclusively after construction.
+// It is moved to a single encode thread and only accessed through &mut self there.
+unsafe impl Send for SwEncEncode {}
+
+impl SwEncEncode {
+    #[allow(clippy::too_many_arguments)]
+    fn new_muxer(
+        output_path: &Path,
+        enc_width: u32,
+        enc_height: u32,
+        fps: u32,
+        bitrate: u64,
+        gop_size: u32,
+    ) -> Result<Self> {
+        let sws_ctx = create_nv12_to_yuv420p_sws(enc_width, enc_height)?;
+        let (enc_video, octx) =
+            create_software_h264_muxer(output_path, enc_width, enc_height, fps, bitrate, gop_size)?;
+        let yuv_frame = alloc_yuv420p_frame(enc_width, enc_height)?;
+
+        Ok(Self {
+            sws_ctx,
+            enc_video,
+            output: Some(FrameOutput::Muxer(octx)),
+            yuv_frame,
+            starting_timestamp: None,
+            frames_written: false,
+            webrtc_disconnected: false,
+            webrtc_paused: None,
+            enc_width,
+            enc_height,
+        })
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub fn new_webrtc(
+        enc_width: u32,
+        enc_height: u32,
+        fps: u32,
+        bitrate: u64,
+        gop_size: u32,
+        tx: crossbeam_channel::Sender<Vec<u8>>,
+        webrtc_paused: Arc<AtomicBool>,
+    ) -> Result<Self> {
+        let sws_ctx = create_nv12_to_yuv420p_sws(enc_width, enc_height)?;
+        let enc_video = create_software_h264_encoder(enc_width, enc_height, fps, bitrate, gop_size)?;
+        let yuv_frame = alloc_yuv420p_frame(enc_width, enc_height)?;
+
+        Ok(Self {
+            sws_ctx,
+            enc_video,
+            output: Some(FrameOutput::Channel(tx)),
+            yuv_frame,
+            starting_timestamp: None,
+            frames_written: false,
+            webrtc_disconnected: false,
+            webrtc_paused: Some(webrtc_paused),
+            enc_width,
+            enc_height,
+        })
+    }
+
+    pub fn flush(&mut self) -> Result<()> {
+        // SAFETY: Sending a null frame flushes the opened software encoder;
+        // no frame data is dereferenced. enc_video is exclusively borrowed via &mut self.
+        unsafe {
+            let ret = ffi::avcodec_send_frame(self.enc_video.as_mut_ptr(), ptr::null());
+            if ret < 0 && ret != ffi::AVERROR_EOF {
+                bail!("software encoder flush send failed: {}", ff_err(ret));
+            }
+        }
+        let start_ts = self.starting_timestamp.unwrap_or(0);
+        self.drain_encoder(start_ts)?;
+
+        Ok(())
+    }
+
+    pub fn encode_cpu_frame(&mut self, frame: &CpuNv12Frame) -> Result<()> {
+        if self.webrtc_disconnected {
+            return Ok(());
+        }
+        if frame.y_stride < self.enc_width as usize || frame.uv_stride < self.enc_width as usize {
+            bail!("CPU NV12 frame stride is smaller than encoder width");
+        }
+        if let Some(ref paused) = self.webrtc_paused {
+            if paused.load(Ordering::Relaxed) {
+                return Ok(());
+            }
+        }
+
         // SAFETY: yuv_frame is an owned reusable YUV420P frame at the same dimensions as sw_nv12;
         // sws_ctx was created for NV12 -> YUV420P with no resize, so sws_scale only converts format.
         unsafe {
             let ret = ffi::av_frame_make_writable(self.yuv_frame);
             if ret < 0 {
-                ffi::av_frame_free(&mut sw_nv12);
-                bail!("av_frame_make_writable failed: error {ret}");
+                bail!("av_frame_make_writable failed: {}", ff_err(ret));
             }
-            ffi::sws_scale(
+            let src_slices = [frame.y_data.as_ptr(), frame.uv_data.as_ptr(), ptr::null(), ptr::null()];
+            let src_strides = [frame.y_stride as i32, frame.uv_stride as i32, 0, 0];
+            let scaled = ffi::sws_scale(
                 self.sws_ctx,
-                (*sw_nv12).data.as_ptr() as *const *const u8,
-                (*sw_nv12).linesize.as_ptr() as *const i32,
+                src_slices.as_ptr(),
+                src_strides.as_ptr(),
                 0,
-                (*sw_nv12).height,
+                self.enc_height as i32,
                 (*self.yuv_frame).data.as_ptr() as *mut *mut u8,
                 (*self.yuv_frame).linesize.as_ptr() as *const i32,
             );
-            ffi::av_frame_free(&mut sw_nv12);
+            if scaled < 0 {
+                bail!("sws_scale failed for software encoder: {scaled}");
+            }
         }
 
-        let pts = filtered.pts().unwrap_or(0);
+        let pts = frame.pts;
         if self.starting_timestamp.is_none() {
             self.starting_timestamp = Some(pts);
         }
@@ -873,13 +957,23 @@ impl SwEncState {
             (*self.yuv_frame).pts = pts;
             let ret = ffi::avcodec_send_frame(self.enc_video.as_mut_ptr(), self.yuv_frame);
             if ret < 0 {
-                bail!("avcodec_send_frame failed for software encoder: error {ret}");
+                bail!("avcodec_send_frame failed for software encoder: {}", ff_err(ret));
             }
         }
 
         self.drain_encoder(start_ts)
     }
 
+    fn write_trailer_if_needed(&mut self) -> Result<()> {
+        if self.frames_written {
+            if let Some(FrameOutput::Muxer(ref mut octx)) = self.output {
+                octx.write_trailer()
+                    .map_err(|e| anyhow::anyhow!("Failed to write trailer: {e}"))?;
+            }
+        }
+        Ok(())
+    }
+
     fn drain_encoder(&mut self, start_ts: i64) -> Result<()> {
         loop {
             let mut pkt = ff::Packet::empty();
@@ -891,7 +985,7 @@ impl SwEncState {
                 if ret == ffi::AVERROR(ffi::EAGAIN) || ret == ffi::AVERROR_EOF {
                     break;
                 }
-                bail!("avcodec_receive_packet failed: error {ret}");
+                bail!("avcodec_receive_packet failed: {}", ff_err(ret));
             }
 
             match self.output {
@@ -961,7 +1055,7 @@ impl SwEncState {
     }
 }
 
-impl Drop for SwEncState {
+impl Drop for SwEncEncode {
     fn drop(&mut self) {
         if !self.sws_ctx.is_null() {
             // SAFETY: sws_ctx is owned by this state and was returned by sws_getContext.
@@ -975,6 +1069,83 @@ impl Drop for SwEncState {
     }
 }
 
+pub struct SwEncState {
+    import: SwEncImport,
+    encode: SwEncEncode,
+}
+
+// SAFETY: SwEncState owns import and encode state exclusively and existing sync callers move it
+// between threads only with external serialization; all FFI handles are accessed through &mut self.
+unsafe impl Send for SwEncState {}
+
+impl SwEncState {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        drm_device: &Path,
+        output_path: &Path,
+        width: u32,
+        height: u32,
+        enc_width: u32,
+        enc_height: u32,
+        fps: u32,
+        bitrate: u64,
+        gop_size: u32,
+    ) -> Result<Self> {
+        tracing::info!(
+            "SwEncState::new: GPU downscale {width}x{height} BGRA -> {enc_width}x{enc_height} NV12, software H.264"
+        );
+        let import = SwEncImport::new(drm_device, width, height, enc_width, enc_height, fps)?;
+        let encode = SwEncEncode::new_muxer(output_path, enc_width, enc_height, fps, bitrate, gop_size)?;
+        Ok(Self { import, encode })
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    pub fn new_webrtc(
+        drm_device: &Path,
+        width: u32,
+        height: u32,
+        enc_width: u32,
+        enc_height: u32,
+        fps: u32,
+        bitrate: u64,
+        gop_size: u32,
+        tx: crossbeam_channel::Sender<Vec<u8>>,
+        webrtc_paused: Arc<AtomicBool>,
+    ) -> Result<Self> {
+        tracing::info!(
+            "SwEncState::new_webrtc: GPU downscale {width}x{height} BGRA -> {enc_width}x{enc_height} NV12, software H.264 -> WebRTC"
+        );
+        let import = SwEncImport::new(drm_device, width, height, enc_width, enc_height, fps)?;
+        let encode = SwEncEncode::new_webrtc(
+            enc_width,
+            enc_height,
+            fps,
+            bitrate,
+            gop_size,
+            tx,
+            webrtc_paused,
+        )?;
+        Ok(Self { import, encode })
+    }
+
+    pub fn frames_rgb(&self) -> &AvHwFrameCtx {
+        self.import.frames_rgb()
+    }
+
+    pub fn encode_frame(&mut self, hw_frame: &ff::frame::Video) -> Result<()> {
+        let cpu_frame = self.import.import_and_scale(hw_frame)?;
+        self.encode.encode_cpu_frame(&cpu_frame)
+    }
+
+    pub fn flush(&mut self) -> Result<()> {
+        for frame in self.import.flush_import()? {
+            self.encode.encode_cpu_frame(&frame)?;
+        }
+        self.encode.flush()?;
+        self.encode.write_trailer_if_needed()
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Shared encoder creation (used by both wlr-screencopy and portal paths)
 // ---------------------------------------------------------------------------
@@ -1065,7 +1236,7 @@ fn build_swenc_filter_graph(
         let ret = ffi::av_buffersrc_parameters_set(src_ctx.as_mut_ptr(), par);
         ffi::av_free(par as *mut _);
         if ret < 0 {
-            bail!("av_buffersrc_parameters_set failed: error {ret}");
+            bail!("av_buffersrc_parameters_set failed: {}", ff_err(ret));
         }
     }
 
@@ -1124,7 +1295,7 @@ fn alloc_yuv420p_frame(width: u32, height: u32) -> Result<*mut ffi::AVFrame> {
         let ret = ffi::av_frame_get_buffer(frame, 0);
         if ret < 0 {
             ffi::av_frame_free(&mut frame);
-            bail!("av_frame_get_buffer failed: error {ret}");
+            bail!("av_frame_get_buffer failed: {}", ff_err(ret));
         }
         Ok(frame)
     }
@@ -1159,7 +1330,7 @@ fn create_software_h264_muxer(
     enc.set_bit_rate(bitrate as usize);
     enc.set_gop(gop_size);
     enc.set_time_base(ff::Rational::new(1, fps as i32));
-    enc.set_max_b_frames(0);
+    enc.set_max_b_frames(3);
 
     // SAFETY: global headers are needed by MP4 and harmless for other common muxers.
     unsafe {
@@ -1167,17 +1338,16 @@ fn create_software_h264_muxer(
     }
 
     if codec_name == "libx264" {
-        // SAFETY: priv_data belongs to the unopened encoder; strings live for each call.
+        // SAFETY: priv_data and codec context belong to the unopened encoder;
+        // strings live for each av_opt_set call.
         unsafe {
             let key = CString::new("preset").unwrap();
-            let val = CString::new("ultrafast").unwrap();
-            ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0);
-            let key = CString::new("tune").unwrap();
-            let val = CString::new("zerolatency").unwrap();
+            let val = CString::new("fast").unwrap();
             ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0);
             let key = CString::new("threads").unwrap();
             let val = CString::new("6").unwrap();
             ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0);
+            (*enc.as_mut_ptr()).profile = ffi::AV_PROFILE_H264_HIGH as i32;
         }
     }
 
@@ -1212,7 +1382,7 @@ fn create_software_h264_muxer(
         )
     };
     if ret < 0 || fmt_ctx_ptr.is_null() {
-        bail!("Failed to allocate output format context: error {ret}");
+        bail!("Failed to allocate output format context: {}", ff_err(ret));
     }
 
     // SAFETY: fmt_ctx_ptr is valid; stream and codec parameters are owned by the format context.
@@ -1225,7 +1395,7 @@ fn create_software_h264_muxer(
     let ret =
         unsafe { ffi::avcodec_parameters_from_context((*stream_ptr).codecpar, enc_video.as_ptr()) };
     if ret < 0 {
-        bail!("Failed to copy codec parameters to stream: error {ret}");
+        bail!("Failed to copy codec parameters to stream: {}", ff_err(ret));
     }
     // SAFETY: stream_ptr is valid and writable during muxer setup.
     unsafe {
@@ -1242,8 +1412,9 @@ fn create_software_h264_muxer(
             );
             if ret < 0 {
                 bail!(
-                    "Failed to open output file '{}': error {ret}",
-                    output_path.display()
+                    "Failed to open output file '{}': {}",
+                    output_path.display(),
+                    ff_err(ret)
                 );
             }
         }
@@ -1252,7 +1423,7 @@ fn create_software_h264_muxer(
     // SAFETY: fmt_ctx_ptr is fully configured.
     let ret = unsafe { ffi::avformat_write_header(fmt_ctx_ptr, ptr::null_mut()) };
     if ret < 0 {
-        bail!("Failed to write output header: error {ret}");
+        bail!("Failed to write output header: {}", ff_err(ret));
     }
 
     // SAFETY: ownership of fmt_ctx_ptr transfers to ffmpeg-next Output wrapper.
@@ -1286,11 +1457,11 @@ fn create_software_h264_encoder(
     enc.set_max_b_frames(0);
 
     if codec_name == "libx264" {
-        // SAFETY: priv_data belongs to the unopened encoder context; each
-        // CString lives for the duration of its av_opt_set call.
+        // SAFETY: priv_data and codec context belong to the unopened encoder;
+        // each CString lives for the duration of its av_opt_set call.
         unsafe {
             let key = CString::new("preset").unwrap();
-            let val = CString::new("ultrafast").unwrap();
+            let val = CString::new("veryfast").unwrap();
             ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0);
             let key = CString::new("tune").unwrap();
             let val = CString::new("zerolatency").unwrap();
@@ -1298,6 +1469,9 @@ fn create_software_h264_encoder(
             let key = CString::new("threads").unwrap();
             let val = CString::new("6").unwrap();
             ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0);
+            // High profile via AVCodecContext.profile (not x264opts — x264 rejects it there).
+            // High enables CABAC + 8x8dct automatically.
+            (*enc.as_mut_ptr()).profile = ffi::AV_PROFILE_H264_HIGH as i32;
             let key = CString::new("x264opts").unwrap();
             let val = CString::new("repeat_headers=1").unwrap();
             ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0);
@@ -1307,7 +1481,7 @@ fn create_software_h264_encoder(
     let opened = enc
         .open()
         .map_err(|e| anyhow::anyhow!("Failed to open {codec_name} encoder: {e}"))?;
-    tracing::info!("WebRTC encoder: {codec_name} {width}x{height} @ {fps}fps {bitrate}bps");
+    tracing::info!("WebRTC encoder: {codec_name} {width}x{height} @ {fps}fps {bitrate}bps (profile High, preset veryfast)");
     Ok(opened.0)
 }
 
@@ -1361,7 +1535,7 @@ fn build_filter_graph(
         let ret = ffi::av_buffersrc_parameters_set(src_ctx.as_mut_ptr(), par);
         ffi::av_free(par as *mut _);
         if ret < 0 {
-            bail!("av_buffersrc_parameters_set failed: error {ret}");
+            bail!("av_buffersrc_parameters_set failed: {}", ff_err(ret));
         }
     }
 

src/backend_detect.rs

@@ -190,6 +190,7 @@ mod tests {
             backend: backend.map(String::from),
             port: 0,
             no_persist: false,
+            stats: false,
         }
     }
 

src/bin/sw_encode_bench.rs

@@ -66,6 +66,7 @@ fn receive_first_frame(cap: &CapPortal) -> Result<wl_webrtc::cap_portal::PwDmaBu
         if let Ok(ctrl) = cap.event_receiver().try_recv() {
             match ctrl {
                 PwCtrlEvent::StreamEnded => bail!("PipeWire stream ended before first frame"),
+                PwCtrlEvent::FormatChanged { .. } => {}
                 PwCtrlEvent::Error(e) => bail!("PipeWire error: {e}"),
             }
         }
@@ -114,6 +115,7 @@ fn main() -> Result<()> {
         backend: Some("portal".to_string()),
         port: 0,
         no_persist: false,
+        stats: false,
     };
 
     let cap = CapPortal::new(&portal_args)?;
@@ -327,6 +329,7 @@ fn main() -> Result<()> {
                     eprintln!("PipeWire error after {} frames: {}", frames_encoded, e);
                     break;
                 }
+                PwCtrlEvent::FormatChanged { .. } => {}
             }
         }
 

src/bin/vaapi_import_bench.rs

@@ -138,6 +138,7 @@ fn receive_first_frame(cap: &CapPortal) -> Result<wl_webrtc::cap_portal::PwDmaBu
         if let Ok(ctrl) = cap.event_receiver().try_recv() {
             match ctrl {
                 PwCtrlEvent::StreamEnded => bail!("PipeWire stream ended before first frame"),
+                PwCtrlEvent::FormatChanged { .. } => {}
                 PwCtrlEvent::Error(e) => bail!("PipeWire error: {e}"),
             }
         }
@@ -519,6 +520,7 @@ fn run_cpu_pipeline(
                     "PipeWire error after {} CPU frames: {e}",
                     stats.frames_encoded
                 ),
+                PwCtrlEvent::FormatChanged { .. } => {}
             }
         }
 
@@ -659,6 +661,7 @@ fn run_gpu_pipeline(
                     "PipeWire error after {} GPU frames: {e}",
                     stats.frames_encoded
                 ),
+                PwCtrlEvent::FormatChanged { .. } => {}
             }
         }
 
@@ -883,6 +886,7 @@ fn main() -> Result<()> {
         backend: Some("portal".to_string()),
         port: 0,
         no_persist: false,
+        stats: false,
     };
 
     let cap = CapPortal::new(&portal_args)?;

src/cap_portal.rs

@@ -54,6 +54,8 @@ pub struct PwDmaBufFrame {
 pub enum PwCtrlEvent {
     /// 流已结束（PipeWire 流断开连接或进入错误状态）
     StreamEnded,
+    /// Format/dimensions changed mid-stream
+    FormatChanged { width: u32, height: u32 },
     /// 发生错误，包含错误描述信息
     Error(String),
 }
@@ -73,6 +75,7 @@ pub struct CapPortal {
     event_rx: Receiver<PwCtrlEvent>,
     pw_thread: Option<JoinHandle<()>>,
     rt: Runtime,
+    pw_dropped: Arc<AtomicU64>,
 }
 
 /// PipeWire 捕获线程的上下文数据
@@ -95,7 +98,7 @@ impl CapPortal {
     /// 执行流程:
     ///   1. 创建 Tokio 运行时（用于异步 Portal 调用）
     ///   2. 通过 XDG Desktop Portal 请求屏幕录制权限，获取 PipeWire fd 和 node_id
-    ///   3. 创建有界通道（容量 16）用于帧传递
+    ///   3. 创建有界通道（容量 1）用于帧传递（最新帧优先，避免队列积压延迟）
     ///   4. 创建 eventfd 对，用于线程安全的关闭信号传递
     ///   5. 启动 PipeWire 捕获线程
     pub fn new(args: &Args) -> Result<Self> {
@@ -104,7 +107,7 @@ impl CapPortal {
         let no_persist = args.no_persist;
         let (pw_fd, node_id) = rt.block_on(async { Self::setup_portal(no_persist).await })?;
 
-        let (frame_tx, frame_rx) = bounded(16);
+        let (frame_tx, frame_rx) = bounded(1);
         let (event_tx, event_rx) = bounded(8);
 
         let efd = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) };
@@ -149,6 +152,7 @@ impl CapPortal {
             event_rx,
             pw_thread: Some(pw_thread),
             rt,
+            pw_dropped,
         })
     }
 
@@ -160,6 +164,16 @@ impl CapPortal {
         &self.event_rx
     }
 
+    /// Returns the total number of PipeWire frames dropped due to channel backlog.
+    pub fn dropped_count(&self) -> u64 {
+        self.pw_dropped.load(Ordering::Relaxed)
+    }
+
+    /// Returns the number of frames currently waiting in the capture channel.
+    pub fn capture_queue_depth(&self) -> usize {
+        self.frame_rx.len()
+    }
+
     /// 通过 XDG Desktop Portal 建立屏幕录制会话
     ///
     /// 与桌面环境的 D-Bus 服务交互，请求用户授权屏幕录制。
@@ -476,7 +490,9 @@ fn pipewire_thread(ctx: PwThreadCtx) {
     let mainloop = match pw::main_loop::MainLoopBox::new(None) {
         Ok(ml) => ml,
         Err(e) => {
-            let _ = event_tx.try_send(PwCtrlEvent::Error(format!("MainLoop::new failed: {e}")));
+            if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("MainLoop::new failed: {e}"))) {
+                tracing::error!("MainLoop::new failed and error channel also failed: {e}");
+            }
             return;
         }
     };
@@ -484,7 +500,9 @@ fn pipewire_thread(ctx: PwThreadCtx) {
     let context = match pw::context::ContextBox::new(mainloop.loop_(), None) {
         Ok(c) => c,
         Err(e) => {
-            let _ = event_tx.try_send(PwCtrlEvent::Error(format!("Context::new failed: {e}")));
+            if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("Context::new failed: {e}"))) {
+                tracing::error!("Context::new failed and error channel also failed: {e}");
+            }
             return;
         }
     };
@@ -492,7 +510,9 @@ fn pipewire_thread(ctx: PwThreadCtx) {
     let core = match context.connect_fd(pw_fd, None) {
         Ok(c) => c,
         Err(e) => {
-            let _ = event_tx.try_send(PwCtrlEvent::Error(format!("connect_fd failed: {e}")));
+            if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("connect_fd failed: {e}"))) {
+                tracing::error!("connect_fd failed and error channel also failed: {e}");
+            }
             return;
         }
     };
@@ -514,7 +534,9 @@ fn pipewire_thread(ctx: PwThreadCtx) {
     ) {
         Ok(s) => s,
         Err(e) => {
-            let _ = event_tx.try_send(PwCtrlEvent::Error(format!("Stream::new failed: {e}")));
+            if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("Stream::new failed: {e}"))) {
+                tracing::error!("Stream::new failed and error channel also failed: {e}");
+            }
             return;
         }
     };
@@ -534,7 +556,13 @@ fn pipewire_thread(ctx: PwThreadCtx) {
                 pw::stream::StreamState::Unconnected => {
                     let _ = event_tx_state.try_send(PwCtrlEvent::StreamEnded);
                 }
-                _ => {}
+                pw::stream::StreamState::Paused => {
+                    tracing::warn!("PipeWire stream paused (compositor may be switching content)");
+                }
+                pw::stream::StreamState::Streaming => {
+                    tracing::info!("PipeWire stream (re)started");
+                }
+                pw::stream::StreamState::Connecting => {}
             }
         })
         // 参数变化回调（格式协商）
@@ -542,6 +570,7 @@ fn pipewire_thread(ctx: PwThreadCtx) {
         // id 为参数类型，param 包含具体的格式参数（分辨率、像素格式等）
         .param_changed({
             let format_info = format_info.clone();
+            let event_tx = event_tx.clone();
             move |_, _, id, param| {
                 // 仅处理 Format 类型的参数变化
                 let Some(param) = param else { return };
@@ -563,7 +592,18 @@ fn pipewire_thread(ctx: PwThreadCtx) {
                 let framerate = info.framerate();
                 let max_framerate = info.max_framerate();
                 // 保存协商后的格式信息，供 process 回调读取
+                let previous_format = format_info.get();
                 format_info.set(Some((width, height, drm_format, modifier)));
+                if let Some((previous_width, previous_height, _, _)) = previous_format {
+                    if width != previous_width || height != previous_height {
+                        tracing::warn!(
+                            "PipeWire dimensions changed: {}x{} (format renegotiation)",
+                            width,
+                            height
+                        );
+                        let _ = event_tx.try_send(PwCtrlEvent::FormatChanged { width, height });
+                    }
+                }
                 tracing::info!(
                     "PipeWire format negotiated: {width}x{height}, \
                      drm_format={drm_format:#010x}, modifier={modifier:#x}, \
@@ -584,12 +624,14 @@ fn pipewire_thread(ctx: PwThreadCtx) {
 
                 let raw_buf = unsafe { stream.dequeue_raw_buffer() };
                 if raw_buf.is_null() {
+                    tracing::trace!("process: null raw_buf");
                     return;
                 }
 
                 // 获取 SPA buffer 结构体，包含数据数组、元数据等
                 let spa_buf = unsafe { (*raw_buf).buffer };
                 if spa_buf.is_null() {
+                    tracing::trace!("process: null spa_buf");
                     unsafe { stream.queue_raw_buffer(raw_buf) };
                     return;
                 }
@@ -599,6 +641,7 @@ fn pipewire_thread(ctx: PwThreadCtx) {
                 let n_datas = unsafe { (*spa_buf).n_datas };
                 let datas_ptr = unsafe { (*spa_buf).datas };
                 if n_datas == 0 || datas_ptr.is_null() {
+                    tracing::trace!("process: no data (n_datas={n_datas})");
                     unsafe { stream.queue_raw_buffer(raw_buf) };
                     return;
                 }
@@ -609,11 +652,13 @@ fn pipewire_thread(ctx: PwThreadCtx) {
                     unsafe { &*(datas_ptr as *const pw::spa::buffer::Data) };
                 let fd = data_ref.fd();
                 if fd < 0 {
+                    tracing::trace!("process: invalid fd={fd}");
                     unsafe { stream.queue_raw_buffer(raw_buf) };
                     return;
                 }
 
                 if data_ref.as_raw().chunk.is_null() {
+                    tracing::trace!("process: null chunk");
                     unsafe { stream.queue_raw_buffer(raw_buf) };
                     return;
                 }
@@ -651,6 +696,7 @@ fn pipewire_thread(ctx: PwThreadCtx) {
                     return;
                 };
                 if width == 0 || height == 0 || format == 0 {
+                    tracing::trace!("process: invalid dimensions {width}x{height} format={format}");
                     unsafe { stream.queue_raw_buffer(raw_buf) };
                     return;
                 }
@@ -676,11 +722,12 @@ fn pipewire_thread(ctx: PwThreadCtx) {
                     pts,
                 };
 
-                if let Err(crossbeam_channel::TrySendError::Full(_)) = frame_tx.try_send(frame) {
-                    let prev = dropped.fetch_add(1, Ordering::Relaxed);
-                    if prev > 0 && prev % 30 == 0 {
-                        tracing::warn!("dropped {prev} frames total: encoder backlog");
+                match frame_tx.try_send(frame) {
+                    Ok(()) => {}
+                    Err(crossbeam_channel::TrySendError::Full(_)) => {
+                        dropped.fetch_add(1, Ordering::Relaxed);
                     }
+                    Err(crossbeam_channel::TrySendError::Disconnected(_)) => {}
                 }
                 unsafe { stream.queue_raw_buffer(raw_buf) };
             }
@@ -695,7 +742,9 @@ fn pipewire_thread(ctx: PwThreadCtx) {
         StreamFlags::AUTOCONNECT | StreamFlags::MAP_BUFFERS,
         &mut params,
     ) {
-        let _ = event_tx.try_send(PwCtrlEvent::Error(format!("stream.connect failed: {e}")));
+        if let Err(e) = event_tx.try_send(PwCtrlEvent::Error(format!("stream.connect failed: {e}"))) {
+            tracing::error!("stream.connect failed and error channel also failed: {e}");
+        }
         return;
     }
 

src/lib.rs

@@ -4,6 +4,7 @@ pub mod backend_detect;
 pub mod cap_portal;
 pub mod cap_wlr_screencopy;
 pub mod fps_limit;
+pub mod stats;
 pub mod state;
 pub mod state_portal;
 pub mod transform;

src/main.rs

@@ -15,6 +15,7 @@ mod backend_detect; // 截屏后端自动检测（wlroots vs Portal/PipeWire）
 mod cap_portal; // XDG Portal 屏幕捕获
 mod cap_wlr_screencopy; // wlroots wlr-screencopy 截屏协议
 mod fps_limit; // 帧率限制器
+mod stats; // 管道性能统计（卡顿诊断）
 mod state; // wlr-screencopy 后端的主状态机
 mod state_portal; // Portal/PipeWire 后端的主状态机
 mod transform; // 图像变换（旋转/翻转）
@@ -43,18 +44,23 @@ fn main() -> Result<()> {
     // 解析命令行参数
     let args = Args::parse();
 
-    // 根据是否启用 verbose 模式设置日志级别
-    tracing_subscriber::fmt()
-        .with_max_level(if args.verbose {
-            tracing::Level::DEBUG
+    // 根据 verbose 模式或 RUST_LOG 环境变量设置日志级别
+    // 支持 RUST_LOG 粒度控制（如 RUST_LOG=wl_webrtc::webrtc=trace）
+    let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
+        .unwrap_or_else(|_| {
+            if args.verbose {
+                tracing_subscriber::EnvFilter::new("debug")
             } else {
-            tracing::Level::INFO
-        })
+                tracing_subscriber::EnvFilter::new("info")
+            }
+        });
+    tracing_subscriber::fmt()
+        .with_env_filter(env_filter)
         .with_writer(std::io::stderr)
         .init();
 
     tracing::info!("wl-webrtc starting");
-    tracing::debug!("Args: {:?}", args);
+    tracing::debug!("Args: output={:?} fps={} codec={} port={} verbose={}", args.output, args.fps, args.codec, args.port, args.verbose);
 
     // MVP 阶段仅支持 H.264 编码，不支持 HEVC
     if args.codec != "h264" {
@@ -250,7 +256,7 @@ fn run_wlr_screencopy(args: Args) -> Result<()> {
 
         // 状态机遇到致命错误时退出
         if state.errored {
-            tracing::error!("Fatal error in state machine, exiting");
+            tracing::error!("Fatal error in state machine (check preceding error logs), exiting");
             running = false;
         }
 
@@ -346,7 +352,7 @@ fn run_portal_pipewire(args: Args) -> Result<()> {
 
         // Portal 状态机遇到致命错误时退出
         if state.is_errored() {
-            tracing::error!("Fatal error in portal state machine, exiting");
+            tracing::error!("Fatal error in portal state machine (check preceding error logs), exiting");
             running = false;
         }
     }

src/state.rs

@@ -46,6 +46,7 @@ use crate::args::Args;
 use crate::avhw::{AvHwDevCtx, EncState, SwEncState};
 use crate::cap_wlr_screencopy::CapWlrScreencopy;
 use crate::fps_limit::FpsLimit;
+use crate::stats::{FrameTimings, PipelineStats};
 use crate::transform::{transpose_if_transform_transposed, Transform};
 use crate::webrtc::WebRtcState;
 
@@ -213,6 +214,9 @@ pub struct State<S: CaptureSource> {
     pub stage: EncConstructionStage<S>,
     pub in_flight_surface: InFlightSurface<S>,
     pub starting_timestamp: Option<i64>,
+    pub stats_start_time: Option<Instant>,
+    pub stats_last_time: Option<Instant>,
+    pub stats_frames: u64,
     pub first_frame: bool,
     pub args: Args,
     pub errored: bool,
@@ -226,6 +230,7 @@ pub struct State<S: CaptureSource> {
     webrtc_rx: Option<crossbeam_channel::Receiver<Vec<u8>>>,
     webrtc_frames_sent: u64,
     webrtc_paused: Option<Arc<AtomicBool>>,
+    stats: PipelineStats,
 }
 
 // ---------------------------------------------------------------------------
@@ -302,6 +307,9 @@ impl<S: CaptureSource> State<S> {
             },
             in_flight_surface: InFlightSurface::None,
             starting_timestamp: None,
+            stats_start_time: None,
+            stats_last_time: None,
+            stats_frames: 0,
             first_frame: true,
             fps_limit: FpsLimit::new(fps),
             args,
@@ -315,6 +323,7 @@ impl<S: CaptureSource> State<S> {
             webrtc_rx,
             webrtc_frames_sent: 0,
             webrtc_paused,
+            stats: PipelineStats::new(),
         };
 
         // registry_queue_init consumes registry events internally during its
@@ -492,7 +501,7 @@ impl<S: CaptureSource> State<S> {
         // is a freshly allocated empty Video frame.
         let ret = unsafe { ffi::av_hwframe_get_buffer(frames_rgb_ctx, surface.as_mut_ptr(), 0) };
         if ret < 0 {
-            tracing::error!("av_hwframe_get_buffer failed: error {}", ret);
+            tracing::error!("av_hwframe_get_buffer failed: {}", crate::avhw::ff_err(ret));
             self.errored = true;
             return;
         }
@@ -505,7 +514,7 @@ impl<S: CaptureSource> State<S> {
         }
         let ret = unsafe { ffi::av_hwframe_map(map_frame.as_mut_ptr(), surface.as_ptr(), 0) };
         if ret < 0 {
-            tracing::error!("av_hwframe_map failed: error {}", ret);
+            tracing::error!("av_hwframe_map failed: {}", crate::avhw::ff_err(ret));
             self.errored = true;
             return;
         }
@@ -530,7 +539,7 @@ impl<S: CaptureSource> State<S> {
                 // takes ownership of the fd, and the original fd is owned by map_frame.
                 let fd_dup = unsafe { libc::dup(obj.fd) };
                 if fd_dup < 0 {
-                    tracing::error!("failed to dup dma-buf fd");
+                    tracing::error!("failed to dup dma-buf fd: {}", std::io::Error::last_os_error());
                     // wayland-client does not auto-destroy params on Drop.
                     params.destroy();
                     self.errored = true;
@@ -574,6 +583,8 @@ impl<S: CaptureSource> State<S> {
     where
         S::Frame: Default,
     {
+        self.stats.record_capture();
+
         let (mut surface, _drm_map, frame, buffer) =
             match mem::replace(&mut self.in_flight_surface, InFlightSurface::None) {
                 InFlightSurface::CopyQueued {
@@ -614,10 +625,29 @@ impl<S: CaptureSource> State<S> {
                 .is_some()
         };
         if should_encode {
+            let encode_start = Instant::now();
             if let Err(e) = enc.encode_frame(&surface) {
                 tracing::error!("encode_frame failed: {}", e);
                 self.errored = true;
             }
+            let encode_elapsed = encode_start.elapsed().as_micros() as u64;
+            self.stats.record_encode(&FrameTimings {
+                total_us: encode_elapsed,
+                ..Default::default()
+            });
+        }
+        self.stats_frames += 1;
+        if let Some(last) = self.stats_last_time {
+            if last.elapsed() >= std::time::Duration::from_secs(10) {
+                let delta = self.stats_frames;
+                let fps = delta as f64 / last.elapsed().as_secs_f64();
+                tracing::info!(frames = self.stats_frames, fps = format!("{fps:.1}"), "encoding stats");
+                self.stats_last_time = Some(std::time::Instant::now());
+                self.stats_frames = 0;
+            }
+        } else {
+            self.stats_start_time = Some(std::time::Instant::now());
+            self.stats_last_time = Some(std::time::Instant::now());
         }
     }
 
@@ -670,13 +700,23 @@ impl<S: CaptureSource> State<S> {
                 if let Err(e) = wrtc.write_h264_frame(&data, self.webrtc_frames_sent, self.args.fps) {
                     tracing::debug!("WebRTC write frame error: {e}");
                 }
+                self.stats.record_send(0.0, None);
                 self.webrtc_frames_sent = self.webrtc_frames_sent.saturating_add(1);
             }
             if count > 0 {
-                tracing::info!("WebRTC forwarded {count} frames from channel");
+                tracing::debug!("WebRTC forwarded {count} frames from channel");
             }
         }
 
+        if self.args.stats && self.stats.should_snapshot() {
+            self.stats.set_queue_depths(
+                0,
+                self.webrtc_rx.as_ref().map(|r| r.len()).unwrap_or(0),
+            );
+            let snap = self.stats.snapshot_and_reset();
+            tracing::info!("stats: {snap}");
+        }
+
         Ok(())
     }
 
@@ -995,7 +1035,6 @@ impl<S: CaptureSource> Dispatch<WlRegistry, GlobalListContents> for State<S> {
         qhandle: &QueueHandle<State<S>>,
     ) {
         use wayland_client::protocol::wl_registry::Event as RegistryEvent;
-        tracing::debug!("Dispatch<WlRegistry>::event fired: {:?}", event);
         
         match event {
             RegistryEvent::Global {

src/state_portal.rs

@@ -8,8 +8,9 @@ use std::time::{Duration, Instant};
 use anyhow::{bail, Result};     // 错误处理工具
 
 use crate::args::Args;          // 命令行参数
-use crate::avhw::{self, SwEncState}; // 软件编码器状态（VAAPI 导入 + H.264 编码）
+use crate::avhw::{self, CpuNv12Frame, SwEncEncode, SwEncImport, SwEncState}; // 软件编码器状态（VAAPI 导入 + H.264 编码）
 use crate::cap_portal::{CapPortal, PwCtrlEvent, PwDmaBufFrame}; // PipeWire 屏幕采集端点
+use crate::stats::{FrameTimings, PipelineStats}; // 管道统计（帧计时、每秒快照）
 use crate::webrtc::WebRtcState; // WebRTC 信令与媒体传输
 
 /// 门户采集的阶段状态
@@ -20,6 +21,23 @@ enum PortalStage {
     Streaming,
 }
 
+struct EncodeThreadTiming {
+    sws_us: u64,
+    encode_us: u64,
+    output_bytes: usize,
+}
+
+struct EncodeThread {
+    handle: Option<std::thread::JoinHandle<()>>,
+    input_tx: crossbeam_channel::Sender<CpuNv12Frame>,
+    timing_rx: crossbeam_channel::Receiver<EncodeThreadTiming>,
+}
+
+struct WebrtcThread {
+    handle: Option<std::thread::JoinHandle<()>>,
+    sent_gap_rx: crossbeam_channel::Receiver<f64>,
+}
+
 /// 门户模式的主状态机
 ///
 /// 负责管理从 PipeWire 采集屏幕帧、通过 VAAPI 硬件编码的完整生命周期。
@@ -27,19 +45,25 @@ enum PortalStage {
 pub struct StatePortal {
     stage: PortalStage,                             // 当前采集阶段（等待首帧 / 流式编码中）
     enc: Option<SwEncState>,                        // 软件编码器，首帧到达后初始化
+    enc_import: Option<SwEncImport>,
+    enc_thread: Option<EncodeThread>,
     cap: CapPortal,                                 // PipeWire 屏幕采集端点
     args: Args,                                     // 用户命令行参数
     errored: bool,                                  // 是否遇到不可恢复的错误
     drm_device: Option<PathBuf>,                    // DRM 渲染设备路径（可自动检测）
-    frames_encoded: u64,                            // 已编码帧数
+    frames_encoded: u64,                            // 已编码帧数（用于 PTS 编号）
     start_time: Option<Instant>,                    // 编码开始时间
-    last_stats_time: Option<Instant>,               // 上一次统计日志时间
-    last_stats_frames: u64,                         // 上一次统计时的已编码帧数
-    webrtc: Option<WebRtcState>,                    // WebRTC 状态（仅 WebRTC 模式启用）
-    webrtc_tx: Option<crossbeam_channel::Sender<Vec<u8>>>,   // 编码帧发送通道
-    webrtc_rx: Option<crossbeam_channel::Receiver<Vec<u8>>>,
-    webrtc_frames_sent: u64,
+    stats: PipelineStats,                           // 管道统计（窗口化帧计时 + 每秒快照）
+    pw_dropped_prev: u64,                           // 上一窗口的 PipeWire 丢弃帧数（用于增量计算）
+    webrtc: Option<WebRtcState>,
+    webrtc_thread: Option<WebrtcThread>,
     webrtc_paused: Option<Arc<AtomicBool>>,
+    last_capture_arrival: Option<Instant>,  // timestamp of last real frame arrival
+    stall_start: Option<Instant>,           // when current stall began
+    last_stall_log: Option<Instant>,        // rate-limiting for stall warnings
+    last_fillable_frame: Option<CpuNv12Frame>,  // cached last frame for filler duplication
+    next_filler_at: Option<Instant>,            // when to send next filler frame
+    filler_frames_sent: u64,
 }
 
 impl StatePortal {
@@ -56,44 +80,45 @@ impl StatePortal {
 
         let cap = CapPortal::new(&args)?;
 
-        let (webrtc, webrtc_tx, webrtc_rx, webrtc_paused) = if args.port > 0 {
-            let (tx, rx) = crossbeam_channel::bounded(32);
+        let (webrtc, webrtc_paused) = if args.port > 0 {
             let wrtc = WebRtcState::new(args.port, args.fps)?;
             let paused = Arc::new(AtomicBool::new(true));
-            (Some(wrtc), Some(tx), Some(rx), Some(paused))
+            (Some(wrtc), Some(paused))
         } else {
-            (None, None, None, None)
+            (None, None)
         };
 
         Ok(Self {
             stage: PortalStage::WaitingForFormat,
             enc: None,
+            enc_import: None,
+            enc_thread: None,
             cap,
             args,
             errored: false,
             drm_device,
             frames_encoded: 0,
             start_time: None,
-            last_stats_time: None,
-            last_stats_frames: 0,
+            stats: PipelineStats::new(),
+            pw_dropped_prev: 0,
             webrtc,
-            webrtc_tx,
-            webrtc_rx,
-            webrtc_frames_sent: 0,
+            webrtc_thread: None,
             webrtc_paused,
+            last_capture_arrival: None,
+            stall_start: None,
+            last_stall_log: None,
+            last_fillable_frame: None,
+            next_filler_at: None,
+            filler_frames_sent: 0,
         })
     }
 
     /// 轮询 PipeWire 事件并编码帧
     ///
-    /// `block=true` 时使用 recv_timeout 阻塞等待帧（最多 10ms），
+    /// `block=true` 时使用 recv_timeout 阻塞等待帧（最多 2ms），
     /// `block=false` 时使用 try_recv 非阻塞检查。
     /// 返回 `Ok(true)` 表示已处理事件，`Ok(false)` 表示暂无数据。
     pub fn poll_and_encode(&mut self, block: bool) -> Result<bool> {
-        // 先处理 WebRTC 信令、网络轮询，并转发已编码帧
-        // WebRTC: process signaling, network, and forward encoded frames
-        self.poll_webrtc()?;
-
         // 检查 PipeWire 控制事件（流结束 / 错误）
         if let Ok(ctrl) = self.cap.event_receiver().try_recv() {
             match ctrl {
@@ -107,23 +132,39 @@ impl StatePortal {
                     self.errored = true;
                     return Ok(true);
                 }
+                PwCtrlEvent::FormatChanged { width, height } => {
+                    tracing::warn!(
+                        "PipeWire format renegotiation: new dimensions {}x{} — encoder output remains at original resolution",
+                        width,
+                        height
+                    );
+                    // No action yet — VAAPI import/scale handles the conversion.
+                    // Full encoder reinit is a future enhancement.
+                }
             }
         }
 
         // 根据阻塞模式选择不同的帧接收策略
         let frame = if block {
-            // 阻塞模式：最多等待 10ms 接收帧
-            match self.cap.frame_receiver().recv_timeout(std::time::Duration::from_millis(10)) {
+            // 阻塞模式：最多等待 2ms 接收帧
+            match self.cap.frame_receiver().recv_timeout(std::time::Duration::from_millis(2)) {
                 Ok(frame) => frame,
-                Err(_) => return Ok(false),
+                Err(_) => {
+                    self.record_capture_timeout();
+                    return Ok(false);
+                }
             }
         } else {
             // 非阻塞模式：立即尝试接收，无数据则返回
             match self.cap.frame_receiver().try_recv() {
                 Ok(frame) => frame,
-                Err(_) => return Ok(false),
+                Err(_) => {
+                    self.record_capture_timeout();
+                    return Ok(false);
+                }
             }
         };
+        self.record_frame_arrival();
 
         match self.stage {
             PortalStage::WaitingForFormat => {
@@ -151,11 +192,11 @@ impl StatePortal {
                 );
                 // 码率：未指定时按分辨率 × 帧率动态计算
                 let actual_bitrate = self.args.bitrate.unwrap_or_else(|| {
-                    2 * (enc_width as u64) * (enc_height as u64) * (self.args.fps as u64) / 100
+                    5 * (enc_width as u64) * (enc_height as u64) * (self.args.fps as u64) / 100
                 });
                 // GOP 大小：WebRTC 模式使用更小的 GOP（fps/2，最低10），MP4 模式使用 fps
                 let actual_gop_size = self.args.gop_size.unwrap_or_else(|| {
-                    if self.webrtc_tx.is_some() {
+                    if self.webrtc.is_some() {
                         (self.args.fps / 2).max(10)
                     } else {
                         self.args.fps
@@ -163,26 +204,51 @@ impl StatePortal {
                 });
 
                 // 根据是否启用 WebRTC 选择不同的编码器构造方式
-                let enc = if let Some(ref tx) = self.webrtc_tx {
+                if self.webrtc.is_some() {
                     let paused = self.webrtc_paused.as_ref()
                         .ok_or_else(|| anyhow::anyhow!("internal invariant broken: webrtc_paused missing while WebRTC mode is active"))?;
-                    avhw::SwEncState::new_webrtc(
+                    let import = SwEncImport::new(
                         &drm_path,
                         frame.width,
                         frame.height,
                         enc_width,
                         enc_height,
                         self.args.fps,
+                    )?;
+                    let (webrtc_tx, webrtc_rx) = crossbeam_channel::bounded(2);
+                    let (input_tx, input_rx) = crossbeam_channel::bounded::<CpuNv12Frame>(1);
+                    let (timing_tx, timing_rx) = crossbeam_channel::bounded::<EncodeThreadTiming>(32);
+                    let encode = SwEncEncode::new_webrtc(
+                        enc_width,
+                        enc_height,
+                        self.args.fps,
                         actual_bitrate,
                         actual_gop_size,
-                        tx.clone(),
+                        webrtc_tx,
                         paused.clone(),
-                    )?
+                    )?;
+                    let handle = std::thread::Builder::new()
+                        .name("wl-webrtc-encode".into())
+                        .spawn(move || encode_thread_loop(encode, input_rx, timing_tx))?;
+                    self.enc_import = Some(import);
+                    self.enc_thread = Some(EncodeThread { handle: Some(handle), input_tx, timing_rx });
+
+                    let wrtc = self.webrtc.take()
+                        .ok_or_else(|| anyhow::anyhow!("internal: WebRtcState missing during init"))?;
+                    let paused = self.webrtc_paused.as_ref()
+                        .ok_or_else(|| anyhow::anyhow!("internal: webrtc_paused missing"))?
+                        .clone();
+                    let fps = self.args.fps;
+                    let (sent_gap_tx, sent_gap_rx) = crossbeam_channel::bounded(64);
+                    let webrtc_handle = std::thread::Builder::new()
+                        .name("wl-webrtc-webrtc".into())
+                        .spawn(move || webrtc_thread_loop(wrtc, webrtc_rx, fps, paused, sent_gap_tx))?;
+                    self.webrtc_thread = Some(WebrtcThread { handle: Some(webrtc_handle), sent_gap_rx });
                 } else {
                     // MP4 模式：编码输出写入文件
                     let output_path = self.args.output.as_deref()
                         .ok_or_else(|| anyhow::anyhow!("--output is required in MP4 file output mode; use --port > 0 for WebRTC mode"))?;
-                    avhw::SwEncState::new(
+                    let enc = avhw::SwEncState::new(
                         &drm_path,
                         std::path::Path::new(output_path),
                         frame.width,
@@ -192,29 +258,143 @@ impl StatePortal {
                         self.args.fps,
                         actual_bitrate,
                         actual_gop_size,
-                    )?
-                };
-
+                    )?;
                     self.enc = Some(enc);
+                };
                 self.stage = PortalStage::Streaming; // 切换到流式编码阶段
                 self.start_time = Some(Instant::now());
-                self.last_stats_time = Some(Instant::now());
                 tracing::info!("First frame processed, encoder initialized, transitioning to Streaming");
                 drop(frame); // 首帧仅用于初始化，不参与编码
             }
             PortalStage::Streaming => {
+                // 记录采集帧到达（用于 capture gap 和 capture_fps 统计）
+                self.stats.record_capture();
+                self.last_capture_arrival = Some(Instant::now());
                 // 流式编码阶段：直接处理帧
                 self.handle_pw_frame(frame)?;
             }
         }
 
-        // 在返回前再次轮询 WebRTC，确保本帧编码后的数据及时转发
-        // WebRTC: drain encoded frames produced by this poll before returning.
-        self.poll_webrtc()?;
+        // 每秒输出一次结构化管道统计（仅 --stats 启用时记录日志）
+        if self.args.stats && self.stats.should_snapshot() {
+            self.stats.set_pipewire_dropped(0, 0);
+            self.stats.set_queue_depths(0, 0);
+            if let Some(ref enc_thread) = self.enc_thread {
+                while let Ok(timing) = enc_thread.timing_rx.try_recv() {
+                    self.stats.record_encode_thread(
+                        timing.sws_us,
+                        timing.encode_us,
+                        timing.output_bytes,
+                    );
+                }
+            }
+            if let Some(ref webrtc_thread) = self.webrtc_thread {
+                while let Ok(gap_ms) = webrtc_thread.sent_gap_rx.try_recv() {
+                    self.stats.record_send_from_thread(gap_ms);
+                }
+            }
+            let snap = self.stats.snapshot_and_reset();
+            if self.filler_frames_sent > 0 {
+                tracing::info!("stats: {snap} filler_frames_sent={}", self.filler_frames_sent);
+            } else {
+                tracing::info!("stats: {snap}");
+            }
+        }
 
         Ok(true)
     }
 
+    fn record_capture_timeout(&mut self) {
+        let Some(last_capture_arrival) = self.last_capture_arrival else {
+            return;
+        };
+
+        let now = Instant::now();
+        let frame_interval = Duration::from_secs_f64(1.0 / f64::from(self.args.fps.max(1)));
+        let stall_threshold = Duration::from_millis(100).max(frame_interval * 3);
+        if now.duration_since(last_capture_arrival) <= stall_threshold {
+            return;
+        }
+
+        if self.stall_start.is_none() {
+            self.stall_start = Some(now);
+            self.last_stall_log = Some(now);
+            tracing::warn!("compositor frame delivery stalled");
+        } else {
+            let should_log = self
+                .last_stall_log
+                .map_or(true, |last_log| now.duration_since(last_log) >= Duration::from_secs(1));
+            if should_log {
+                self.last_stall_log = Some(now);
+                tracing::warn!("compositor frame delivery stalled");
+            }
+        }
+
+        self.maybe_send_filler_frame();
+    }
+
+    fn maybe_send_filler_frame(&mut self) {
+        if self.webrtc_thread.is_none() || self.stall_start.is_none() {
+            return;
+        }
+        let Some(cached) = &self.last_fillable_frame else {
+            return;
+        };
+
+        const MAX_FILLER_DURATION: Duration = Duration::from_secs(2);
+        if let Some(stall_start) = self.stall_start {
+            if stall_start.elapsed() > MAX_FILLER_DURATION {
+                return;
+            }
+        }
+
+        let now = Instant::now();
+        let frame_interval = Duration::from_secs_f64(1.0 / f64::from(self.args.fps.max(1)));
+
+        let Some(next) = self.next_filler_at else {
+            self.next_filler_at = Some(now + frame_interval);
+            return;
+        };
+        if now < next {
+            return;
+        }
+
+        let filler = CpuNv12Frame {
+            y_data: cached.y_data.clone(),
+            uv_data: cached.uv_data.clone(),
+            y_stride: cached.y_stride,
+            uv_stride: cached.uv_stride,
+            pts: self.frames_encoded as i64,
+        };
+
+        if let Some(enc_thread) = &self.enc_thread {
+            match enc_thread.input_tx.try_send(filler) {
+                Ok(()) => {
+                    self.frames_encoded += 1;
+                    self.filler_frames_sent += 1;
+                    self.next_filler_at = Some(next + frame_interval);
+                }
+                Err(crossbeam_channel::TrySendError::Full(_)) => {}
+                Err(crossbeam_channel::TrySendError::Disconnected(_)) => {
+                    tracing::error!("Encode thread disconnected during filler");
+                    self.errored = true;
+                }
+            }
+        }
+    }
+
+    fn record_frame_arrival(&mut self) {
+        if let Some(stall_start) = self.stall_start.take() {
+            tracing::info!(
+                "compositor frame delivery resumed after {:.0}ms",
+                stall_start.elapsed().as_secs_f64() * 1000.0
+            );
+            self.last_stall_log = None;
+        }
+        self.last_capture_arrival = Some(Instant::now());
+        self.next_filler_at = None;
+    }
+
     /// 为当前帧解析可用的 DRM 渲染设备
     ///
     /// 如果用户已通过 `--drm-device` 指定设备，直接返回；
@@ -271,12 +451,10 @@ impl StatePortal {
     /// 通过 `av_hwframe_map` 零拷贝导入 VAAPI，然后交给 SwEncState 完成：
     /// scale_vaapi GPU 缩放、2K NV12 回读、YUV420P 格式转换、软件 H.264 编码。
     fn handle_pw_frame(&mut self, frame: PwDmaBufFrame) -> Result<()> {
-        // 获取已初始化的编码器引用
-        let enc = match self.enc.as_mut() {
-            Some(enc) => enc,
-            None => bail!("encoder not initialized"),
-        };
+        let t_import_start = Instant::now();
+        let pts = self.frames_encoded as i64;
 
+        if let Some(enc) = self.enc.as_mut() {
             // 将 DMA-BUF 帧零拷贝导入 VAAPI 硬件帧池
             let mut vaapi_frame = unsafe {
                 avhw::import_dma_buf_to_vaapi(
@@ -291,29 +469,74 @@ impl StatePortal {
                 )
             }?;
 
+            let import_us = t_import_start.elapsed().as_micros() as u64;
+            let t_encode_start = Instant::now();
+
             // 设置帧的显示时间戳（PTS），基于已编码帧序号
-        let pts = self.frames_encoded as i64;
             unsafe {
                 (*vaapi_frame.as_mut_ptr()).pts = pts;
             }
 
             // 送入编码器完成：缩放 → 回读 → 格式转换 → H.264 编码
             enc.encode_frame(&vaapi_frame)?;
+            let total_us = t_import_start.elapsed().as_micros() as u64;
+            let encode_us = t_encode_start.elapsed().as_micros() as u64;
+
             self.frames_encoded += 1;
 
-        // 每 10 秒输出一次编码统计（已编码帧数、实时帧率）
-        if let Some(last) = self.last_stats_time {
-            if last.elapsed() >= Duration::from_secs(10) {
-                let delta_frames = self.frames_encoded - self.last_stats_frames;
-                let delta_secs = last.elapsed().as_secs_f64();
-                let fps = delta_frames as f64 / delta_secs;
-                tracing::info!(
-                    "encoded={}, fps={fps:.1}",
-                    self.frames_encoded,
-                );
-                self.last_stats_time = Some(Instant::now());
-                self.last_stats_frames = self.frames_encoded;
+            // 记录帧计时到管道统计（import + encode 内部各阶段暂不可分离，用 total 覆盖）
+            let timings = FrameTimings {
+                import_us,
+                encode_us,
+                total_us,
+                ..Default::default()
+            };
+            self.stats.record_encode(&timings);
+        } else if let Some(import) = self.enc_import.as_mut() {
+            let mut vaapi_frame = unsafe {
+                avhw::import_dma_buf_to_vaapi(
+                    import.frames_rgb().as_ptr(),
+                    frame.fd.as_raw_fd(),
+                    frame.width,
+                    frame.height,
+                    frame.format,
+                    frame.modifier,
+                    frame.stride,
+                    frame.offset,
+                )
+            }?;
+            unsafe {
+                (*vaapi_frame.as_mut_ptr()).pts = pts;
             }
+
+            let cpu_nv12 = import.import_and_scale(&vaapi_frame)?;
+            let import_us = t_import_start.elapsed().as_micros() as u64;
+            self.stats.record_import(import_us);
+
+            let enc_thread = self.enc_thread.as_ref()
+                .ok_or_else(|| anyhow::anyhow!("internal invariant broken: encode thread missing while async import is active"))?;
+            let fillable_frame = CpuNv12Frame {
+                y_data: cpu_nv12.y_data.clone(),
+                uv_data: cpu_nv12.uv_data.clone(),
+                y_stride: cpu_nv12.y_stride,
+                uv_stride: cpu_nv12.uv_stride,
+                pts: 0,
+            };
+            match enc_thread.input_tx.try_send(cpu_nv12) {
+                Ok(()) => {
+                    self.frames_encoded += 1;
+                    self.last_fillable_frame = Some(fillable_frame);
+                }
+                Err(crossbeam_channel::TrySendError::Full(_)) => {
+                    tracing::debug!("Encode thread input full, dropping portal frame");
+                }
+                Err(crossbeam_channel::TrySendError::Disconnected(_frame)) => {
+                    tracing::error!("Encode thread input disconnected");
+                    self.errored = true;
+                }
+            }
+        } else {
+            bail!("encoder not initialized");
         }
 
         Ok(())
@@ -323,9 +546,26 @@ impl StatePortal {
     ///
     /// 使用 `enc.take()` 确保编码器只被 flush 一次，即使多次调用也安全（幂等）。
     pub fn shutdown(&mut self) {
-        // 先 drop receiver，使 flush() 中的 try_send() 立即返回 Disconnected
-        // 而非在满通道上阻塞（修复 issue #8 死锁）
-        self.webrtc_rx = None;
+        self.last_fillable_frame = None;
+        // 1. Stop encode thread (drops webrtc_tx → signals WebRTC thread to exit)
+        if let Some(mut enc_thread) = self.enc_thread.take() {
+            drop(enc_thread.input_tx);
+            if let Some(handle) = enc_thread.handle.take() {
+                if handle.join().is_err() {
+                    tracing::error!("Encode thread panicked during shutdown");
+                }
+            }
+        }
+        self.enc_import = None;
+        // 2. Wait for WebRTC thread (exits when webrtc_tx is dropped by encode thread)
+        if let Some(mut webrtc_thread) = self.webrtc_thread.take() {
+            if let Some(handle) = webrtc_thread.handle.take() {
+                if handle.join().is_err() {
+                    tracing::error!("WebRTC thread panicked during shutdown");
+                }
+            }
+        }
+        // 3. Flush MP4 encoder if present
         if let Some(mut enc) = self.enc.take() {
             if let Err(e) = enc.flush() {
                 tracing::error!("Flush error during shutdown: {e}");
@@ -350,20 +590,66 @@ impl StatePortal {
     pub fn is_errored(&self) -> bool {
         self.errored
     }
+}
 
-    /// 轮询 WebRTC 信令通道并转发编码帧
-    ///
-    /// 处理信令交换、网络轮询，以及从编码通道中取出已编码的 H.264 数据
-    /// 并通过 WebRTC 发送。
-    fn poll_webrtc(&mut self) -> Result<()> {
-        let Some(ref mut wrtc) = self.webrtc else { return Ok(()) };
+fn encode_thread_loop(
+    mut encode: SwEncEncode,
+    input_rx: crossbeam_channel::Receiver<CpuNv12Frame>,
+    timing_tx: crossbeam_channel::Sender<EncodeThreadTiming>,
+) {
+    loop {
+        match input_rx.recv() {
+            Ok(frame) => {
+                let t_start = Instant::now();
+                match encode.encode_cpu_frame(&frame) {
+                    Ok(()) => {
+                        let elapsed = t_start.elapsed().as_micros() as u64;
+                        let _ = timing_tx.try_send(EncodeThreadTiming {
+                            sws_us: 0,
+                            encode_us: elapsed,
+                            output_bytes: 0,
+                        });
+                    }
+                    Err(e) => {
+                        tracing::error!("Encode thread error: {e}");
+                        break;
+                    }
+                }
+            }
+            Err(_) => {
+                tracing::info!("Encode thread input closed, flushing encoder");
+                if let Err(e) = encode.flush() {
+                    tracing::error!("Encode thread flush error: {e}");
+                }
+                break;
+            }
+        }
+    }
+    tracing::info!("Encode thread exiting");
+}
 
-        wrtc.handle_signaling()?;
-        wrtc.poll_and_feed()?;
+fn webrtc_thread_loop(
+    mut wrtc: WebRtcState,
+    webrtc_rx: crossbeam_channel::Receiver<Vec<u8>>,
+    fps: u32,
+    paused: Arc<AtomicBool>,
+    sent_gap_tx: crossbeam_channel::Sender<f64>,
+) {
+    let mut frames_sent: u64 = 0;
+    let mut last_send: Option<std::time::Instant> = None;
+    let timeout = Duration::from_millis(1);
+
+    loop {
+        if let Err(e) = wrtc.handle_signaling() {
+            tracing::error!("WebRTC signaling error: {e}");
+            break;
+        }
+        if let Err(e) = wrtc.poll_and_feed() {
+            tracing::error!("WebRTC poll error: {e}");
+            break;
+        }
 
         let connected = wrtc.is_connected();
-
-        if let Some(ref paused) = self.webrtc_paused {
         let was_paused = paused.load(Ordering::Relaxed);
         let now_paused = !connected;
         if was_paused && !now_paused {
@@ -372,27 +658,46 @@ impl StatePortal {
             tracing::warn!("WebRTC client disconnected, pausing encoding");
         }
         paused.store(now_paused, Ordering::Relaxed);
-        }
 
-        if let Some(ref rx) = self.webrtc_rx {
-            let mut count = 0u32;
-            while let Ok(data) = 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 connected {
+            while let Ok(data) = webrtc_rx.try_recv() {
+                if let Err(e) = wrtc.write_h264_frame(&data, frames_sent, fps) {
                     tracing::debug!("WebRTC write frame error: {e}");
                 }
-                self.webrtc_frames_sent = self.webrtc_frames_sent.saturating_add(1);
+                frames_sent = frames_sent.saturating_add(1);
+                let gap_ms = last_send
+                    .map(|l| l.elapsed().as_secs_f64() * 1000.0)
+                    .unwrap_or(0.0);
+                last_send = Some(std::time::Instant::now());
+                let _ = sent_gap_tx.try_send(gap_ms);
+            }
+        } else {
+            while webrtc_rx.try_recv().is_ok() {}
+        }
+
+        match webrtc_rx.recv_timeout(timeout) {
+            Ok(data) => {
+                if wrtc.is_connected() {
+                    if let Err(e) = wrtc.write_h264_frame(&data, frames_sent, fps) {
+                        tracing::debug!("WebRTC write frame error: {e}");
+                    }
+                    frames_sent = frames_sent.saturating_add(1);
+                    let gap_ms = last_send
+                        .map(|l| l.elapsed().as_secs_f64() * 1000.0)
+                        .unwrap_or(0.0);
+                    last_send = Some(std::time::Instant::now());
+                    let _ = sent_gap_tx.try_send(gap_ms);
+                }
+            }
+            Err(crossbeam_channel::RecvTimeoutError::Timeout) => {}
+            Err(crossbeam_channel::RecvTimeoutError::Disconnected) => {
+                tracing::info!("WebRTC channel disconnected, exiting thread");
+                return;
             }
-            if count > 0 {
-                tracing::info!("WebRTC forwarded {count} frames from channel");
         }
     }
 
-        Ok(())
-    }
+    tracing::info!("WebRTC thread exiting");
 }
 
 impl Drop for StatePortal {
@@ -510,6 +815,7 @@ mod tests {
             backend: None,
             port: 0,
             no_persist: false,
+            stats: false,
         };
         let result = resolve_drm_device(&args).unwrap();
         assert_eq!(
@@ -533,6 +839,7 @@ mod tests {
             backend: None,
             port: 0,
             no_persist: false,
+            stats: false,
         };
         let result = resolve_drm_device(&args).unwrap();
         assert_eq!(result, None);

src/stats.rs

@@ -0,0 +1,500 @@
+// stats.rs — Lightweight windowed pipeline statistics for stutter diagnosis
+//
+// Tracks per-second snapshots of capture/encode/send pipeline metrics.
+// Designed for low overhead: only counters and timing samples are collected,
+// with one structured log line emitted per second when `--stats` is enabled.
+
+use std::time::Instant;
+
+/// Per-stage timing for a single encode pipeline frame.
+///
+/// All values are in microseconds. The caller records timestamps around
+/// each stage and passes the deltas to [`PipelineStats::record_frame`].
+#[derive(Debug, Default)]
+pub struct FrameTimings {
+    /// DMA-BUF import (av_hwframe_map)
+    pub import_us: u64,
+    /// GPU scale (scale_vaapi filter)
+    pub scale_us: u64,
+    /// GPU→CPU transfer (av_hwframe_transfer_data)
+    pub transfer_us: u64,
+    /// sws_scale NV12→YUV420P
+    pub sws_us: u64,
+    /// H.264 encode (avcodec_send_frame + receive_packet)
+    pub encode_us: u64,
+    /// Wall-clock total for this frame (import through encode output)
+    pub total_us: u64,
+    /// Encoded output size in bytes
+    pub output_bytes: usize,
+}
+
+/// Windowed statistics aggregator for the encode/send pipeline.
+///
+/// Collects counters and timing samples within a one-second window,
+/// then computes avg/p95/max when the snapshot is taken.
+pub struct PipelineStats {
+    // --- counters (reset each window) ---
+    capture_frames: u64,
+    encoded_frames: u64,
+    sent_frames: u64,
+    pipewire_dropped: u64,
+    over_budget_count: u64,
+
+    // --- queue depth at last observation ---
+    capture_queue_depth: usize,
+    encoded_queue_depth: usize,
+
+    // --- timing samples ---
+    capture_gaps_ms: Vec<f64>,
+    encoded_gaps_ms: Vec<f64>,
+    sent_gaps_ms: Vec<f64>,
+    frame_age_ms: Vec<f64>,
+    send_wait_ms: Vec<f64>,
+
+    // --- per-stage timing (microseconds) ---
+    import_us: Vec<u64>,
+    scale_us: Vec<u64>,
+    transfer_us: Vec<u64>,
+    sws_us: Vec<u64>,
+    encode_us: Vec<u64>,
+    total_us: Vec<u64>,
+    output_bytes: Vec<usize>,
+
+    // --- timing state ---
+    last_capture_time: Option<Instant>,
+    last_encode_time: Option<Instant>,
+    last_send_time: Option<Instant>,
+    window_start: Instant,
+}
+
+impl PipelineStats {
+    pub fn new() -> Self {
+        Self {
+            capture_frames: 0,
+            encoded_frames: 0,
+            sent_frames: 0,
+            pipewire_dropped: 0,
+            over_budget_count: 0,
+            capture_queue_depth: 0,
+            encoded_queue_depth: 0,
+            capture_gaps_ms: Vec::new(),
+            encoded_gaps_ms: Vec::new(),
+            sent_gaps_ms: Vec::new(),
+            frame_age_ms: Vec::new(),
+            send_wait_ms: Vec::new(),
+            import_us: Vec::new(),
+            scale_us: Vec::new(),
+            transfer_us: Vec::new(),
+            sws_us: Vec::new(),
+            encode_us: Vec::new(),
+            total_us: Vec::new(),
+            output_bytes: Vec::new(),
+            last_capture_time: None,
+            last_encode_time: None,
+            last_send_time: None,
+            window_start: Instant::now(),
+        }
+    }
+
+    /// Record that a capture frame was received from PipeWire.
+    pub fn record_capture(&mut self) {
+        let now = Instant::now();
+        if let Some(last) = self.last_capture_time {
+            let gap_ms = last.elapsed().as_secs_f64() * 1000.0;
+            self.capture_gaps_ms.push(gap_ms);
+        }
+        self.last_capture_time = Some(now);
+        self.capture_frames += 1;
+    }
+
+    /// Record that a frame completed encoding with the given timings.
+    pub fn record_encode(&mut self, timings: &FrameTimings) {
+        let now = Instant::now();
+        if let Some(last) = self.last_encode_time {
+            let gap_ms = last.elapsed().as_secs_f64() * 1000.0;
+            self.encoded_gaps_ms.push(gap_ms);
+        }
+        self.last_encode_time = Some(now);
+        self.encoded_frames += 1;
+
+        self.import_us.push(timings.import_us);
+        self.scale_us.push(timings.scale_us);
+        self.transfer_us.push(timings.transfer_us);
+        self.sws_us.push(timings.sws_us);
+        self.encode_us.push(timings.encode_us);
+        self.total_us.push(timings.total_us);
+        self.output_bytes.push(timings.output_bytes);
+    }
+
+    pub fn record_import(&mut self, import_us: u64) {
+        self.import_us.push(import_us);
+    }
+
+    pub fn record_encode_thread(&mut self, sws_us: u64, encode_us: u64, output_bytes: usize) {
+        let now = Instant::now();
+        if let Some(last) = self.last_encode_time {
+            let gap_ms = last.elapsed().as_secs_f64() * 1000.0;
+            self.encoded_gaps_ms.push(gap_ms);
+        }
+        self.last_encode_time = Some(now);
+        self.encoded_frames += 1;
+
+        self.sws_us.push(sws_us);
+        self.encode_us.push(encode_us);
+        self.total_us.push(sws_us.saturating_add(encode_us));
+        self.output_bytes.push(output_bytes);
+    }
+
+    /// Record that a frame was sent via WebRTC.
+    /// `wait_ms` is time spent blocked waiting to send into the channel.
+    /// `capture_time` is when the frame was originally captured (for frame age).
+    pub fn record_send(&mut self, wait_ms: f64, capture_time: Option<Instant>) {
+        let now = Instant::now();
+        if let Some(last) = self.last_send_time {
+            let gap_ms = last.elapsed().as_secs_f64() * 1000.0;
+            self.sent_gaps_ms.push(gap_ms);
+        }
+        self.last_send_time = Some(now);
+        self.sent_frames += 1;
+
+        if wait_ms > 0.0 {
+            self.send_wait_ms.push(wait_ms);
+        }
+        if let Some(ct) = capture_time {
+            let age_ms = ct.elapsed().as_secs_f64() * 1000.0;
+            self.frame_age_ms.push(age_ms);
+        }
+    }
+
+    /// Record a frame sent from a background WebRTC thread.
+    /// `gap_ms` is the pre-computed time since the previous send (0.0 = first frame).
+    /// Unlike `record_send`, this does not sample `Instant::now()`, so it remains
+    /// accurate even when batch-drained at stats snapshot time.
+    pub fn record_send_from_thread(&mut self, gap_ms: f64) {
+        if gap_ms > 0.0 {
+            self.sent_gaps_ms.push(gap_ms);
+        }
+        self.sent_frames += 1;
+    }
+
+    /// Update PipeWire dropped counter (absolute value from AtomicU64).
+    pub fn set_pipewire_dropped(&mut self, total_dropped: u64, prev_dropped: u64) {
+        self.pipewire_dropped = total_dropped.saturating_sub(prev_dropped);
+    }
+
+    /// Update queue depth observations.
+    pub fn set_queue_depths(&mut self, capture: usize, encoded: usize) {
+        self.capture_queue_depth = capture;
+        self.encoded_queue_depth = encoded;
+    }
+
+    /// Record that a frame exceeded its time budget.
+    pub fn record_over_budget(&mut self) {
+        self.over_budget_count += 1;
+    }
+
+    /// Returns true if at least 1 second has elapsed since the last snapshot
+    /// (or since creation). If true, call `snapshot_and_reset` to get the stats.
+    pub fn should_snapshot(&self) -> bool {
+        self.window_start.elapsed().as_secs() >= 1
+    }
+
+    /// Compute a snapshot of the current window and reset all counters.
+    pub fn snapshot_and_reset(&mut self) -> StatsSnapshot {
+        let elapsed = self.window_start.elapsed().as_secs_f64();
+        let snap = StatsSnapshot {
+            elapsed_secs: elapsed,
+            capture_fps: self.capture_frames as f64 / elapsed,
+            encoded_fps: self.encoded_frames as f64 / elapsed,
+            sent_fps: self.sent_frames as f64 / elapsed,
+            capture_frames: self.capture_frames,
+            encoded_frames: self.encoded_frames,
+            sent_frames: self.sent_frames,
+            pipewire_dropped: self.pipewire_dropped,
+            over_budget_count: self.over_budget_count,
+            capture_queue_depth: self.capture_queue_depth,
+            encoded_queue_depth: self.encoded_queue_depth,
+            capture_gap_avg_ms: avg_f64(&self.capture_gaps_ms),
+            capture_gap_p95_ms: p95_f64(&self.capture_gaps_ms),
+            capture_gap_max_ms: max_f64(&self.capture_gaps_ms),
+            encoded_gap_avg_ms: avg_f64(&self.encoded_gaps_ms),
+            encoded_gap_p95_ms: p95_f64(&self.encoded_gaps_ms),
+            encoded_gap_max_ms: max_f64(&self.encoded_gaps_ms),
+            sent_gap_avg_ms: avg_f64(&self.sent_gaps_ms),
+            sent_gap_p95_ms: p95_f64(&self.sent_gaps_ms),
+            sent_gap_max_ms: max_f64(&self.sent_gaps_ms),
+            frame_age_avg_ms: avg_f64(&self.frame_age_ms),
+            frame_age_p95_ms: p95_f64(&self.frame_age_ms),
+            frame_age_max_ms: max_f64(&self.frame_age_ms),
+            send_wait_p95_ms: p95_f64(&self.send_wait_ms),
+            import_avg_ms: avg_ms(&self.import_us),
+            import_p95_ms: p95_ms(&self.import_us),
+            scale_avg_ms: avg_ms(&self.scale_us),
+            scale_p95_ms: p95_ms(&self.scale_us),
+            transfer_avg_ms: avg_ms(&self.transfer_us),
+            transfer_p95_ms: p95_ms(&self.transfer_us),
+            sws_avg_ms: avg_ms(&self.sws_us),
+            sws_p95_ms: p95_ms(&self.sws_us),
+            encode_avg_ms: avg_ms(&self.encode_us),
+            encode_p95_ms: p95_ms(&self.encode_us),
+            total_avg_ms: avg_ms(&self.total_us),
+            total_p95_ms: p95_ms(&self.total_us),
+            output_bytes_per_sec: sum_usize(&self.output_bytes) as f64 / elapsed,
+            output_frame_bytes_p95: p95_usize(&self.output_bytes),
+            output_frame_bytes_max: max_usize(&self.output_bytes),
+        };
+
+        // Reset all counters and sample buffers
+        self.capture_frames = 0;
+        self.encoded_frames = 0;
+        self.sent_frames = 0;
+        self.pipewire_dropped = 0;
+        self.over_budget_count = 0;
+        self.capture_queue_depth = 0;
+        self.encoded_queue_depth = 0;
+        self.capture_gaps_ms.clear();
+        self.encoded_gaps_ms.clear();
+        self.sent_gaps_ms.clear();
+        self.frame_age_ms.clear();
+        self.send_wait_ms.clear();
+        self.import_us.clear();
+        self.scale_us.clear();
+        self.transfer_us.clear();
+        self.sws_us.clear();
+        self.encode_us.clear();
+        self.total_us.clear();
+        self.output_bytes.clear();
+        self.window_start = Instant::now();
+
+        snap
+    }
+}
+
+/// A one-second snapshot of pipeline statistics.
+#[derive(Debug)]
+pub struct StatsSnapshot {
+    pub elapsed_secs: f64,
+    // FPS
+    pub capture_fps: f64,
+    pub encoded_fps: f64,
+    pub sent_fps: f64,
+    // Counters
+    pub capture_frames: u64,
+    pub encoded_frames: u64,
+    pub sent_frames: u64,
+    pub pipewire_dropped: u64,
+    pub over_budget_count: u64,
+    // Queue depths
+    pub capture_queue_depth: usize,
+    pub encoded_queue_depth: usize,
+    // Gap timing (ms)
+    pub capture_gap_avg_ms: f64,
+    pub capture_gap_p95_ms: f64,
+    pub capture_gap_max_ms: f64,
+    pub encoded_gap_avg_ms: f64,
+    pub encoded_gap_p95_ms: f64,
+    pub encoded_gap_max_ms: f64,
+    pub sent_gap_avg_ms: f64,
+    pub sent_gap_p95_ms: f64,
+    pub sent_gap_max_ms: f64,
+    // Frame age (capture → send)
+    pub frame_age_avg_ms: f64,
+    pub frame_age_p95_ms: f64,
+    pub frame_age_max_ms: f64,
+    // Send wait
+    pub send_wait_p95_ms: f64,
+    // Per-stage encode timing (ms)
+    pub import_avg_ms: f64,
+    pub import_p95_ms: f64,
+    pub scale_avg_ms: f64,
+    pub scale_p95_ms: f64,
+    pub transfer_avg_ms: f64,
+    pub transfer_p95_ms: f64,
+    pub sws_avg_ms: f64,
+    pub sws_p95_ms: f64,
+    pub encode_avg_ms: f64,
+    pub encode_p95_ms: f64,
+    pub total_avg_ms: f64,
+    pub total_p95_ms: f64,
+    // Output size
+    pub output_bytes_per_sec: f64,
+    pub output_frame_bytes_p95: usize,
+    pub output_frame_bytes_max: usize,
+}
+
+impl std::fmt::Display for StatsSnapshot {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "capture_fps={:.1} encoded_fps={:.1} sent_fps={:.1} \
+             pw_dropped={} over_budget={} \
+             cap_q={} enc_q={} \
+             cap_gap_p95={:.1}ms cap_gap_max={:.1}ms \
+             enc_gap_p95={:.1}ms enc_gap_max={:.1}ms \
+             sent_gap_p95={:.1}ms sent_gap_max={:.1}ms \
+             frame_age_p95={:.1}ms frame_age_max={:.1}ms \
+             send_wait_p95={:.1}ms \
+             import_p95={:.1}ms scale_p95={:.1}ms transfer_p95={:.1}ms \
+             sws_p95={:.1}ms encode_p95={:.1}ms total_p95={:.1}ms \
+             output_bps={:.0} frame_bytes_max={}",
+            self.capture_fps,
+            self.encoded_fps,
+            self.sent_fps,
+            self.pipewire_dropped,
+            self.over_budget_count,
+            self.capture_queue_depth,
+            self.encoded_queue_depth,
+            self.capture_gap_p95_ms,
+            self.capture_gap_max_ms,
+            self.encoded_gap_p95_ms,
+            self.encoded_gap_max_ms,
+            self.sent_gap_p95_ms,
+            self.sent_gap_max_ms,
+            self.frame_age_p95_ms,
+            self.frame_age_max_ms,
+            self.send_wait_p95_ms,
+            self.import_p95_ms,
+            self.scale_p95_ms,
+            self.transfer_p95_ms,
+            self.sws_p95_ms,
+            self.encode_p95_ms,
+            self.total_p95_ms,
+            self.output_bytes_per_sec,
+            self.output_frame_bytes_max,
+        )
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Statistics helpers
+// ---------------------------------------------------------------------------
+
+fn avg_f64(data: &[f64]) -> f64 {
+    if data.is_empty() {
+        return 0.0;
+    }
+    data.iter().sum::<f64>() / data.len() as f64
+}
+
+fn p95_f64(data: &[f64]) -> f64 {
+    if data.is_empty() {
+        return 0.0;
+    }
+    let mut sorted: Vec<f64> = data.to_vec();
+    sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
+    let idx = ((sorted.len() as f64) * 0.95).floor() as usize;
+    sorted[idx.min(sorted.len() - 1)]
+}
+
+fn max_f64(data: &[f64]) -> f64 {
+    data.iter().copied().fold(0.0_f64, f64::max)
+}
+
+fn avg_ms(data: &[u64]) -> f64 {
+    if data.is_empty() {
+        return 0.0;
+    }
+    data.iter().sum::<u64>() as f64 / data.len() as f64 / 1000.0
+}
+
+fn p95_ms(data: &[u64]) -> f64 {
+    if data.is_empty() {
+        return 0.0;
+    }
+    let mut sorted = data.to_vec();
+    sorted.sort_unstable();
+    let idx = ((sorted.len() as f64) * 0.95).floor() as usize;
+    sorted[idx.min(sorted.len() - 1)] as f64 / 1000.0
+}
+
+fn sum_usize(data: &[usize]) -> usize {
+    data.iter().sum()
+}
+
+fn p95_usize(data: &[usize]) -> usize {
+    if data.is_empty() {
+        return 0;
+    }
+    let mut sorted = data.to_vec();
+    sorted.sort_unstable();
+    let idx = ((sorted.len() as f64) * 0.95).floor() as usize;
+    sorted[idx.min(sorted.len() - 1)]
+}
+
+fn max_usize(data: &[usize]) -> usize {
+    data.iter().copied().max().unwrap_or(0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn empty_stats_snapshot() {
+        let mut stats = PipelineStats::new();
+        let snap = stats.snapshot_and_reset();
+        assert_eq!(snap.capture_frames, 0);
+        assert_eq!(snap.encoded_frames, 0);
+        assert_eq!(snap.sent_frames, 0);
+    }
+
+    #[test]
+    fn record_and_snapshot_counts() {
+        let mut stats = PipelineStats::new();
+        stats.record_capture();
+        stats.record_capture();
+        stats.record_encode(&FrameTimings {
+            total_us: 5000,
+            output_bytes: 1000,
+            ..Default::default()
+        });
+        stats.record_send(0.1, None);
+
+        let snap = stats.snapshot_and_reset();
+        assert_eq!(snap.capture_frames, 2);
+        assert_eq!(snap.encoded_frames, 1);
+        assert_eq!(snap.sent_frames, 1);
+    }
+
+    #[test]
+    fn p95_computation() {
+        // 100 values: 0.0 through 99.0
+        let data: Vec<f64> = (0..100).map(|i| i as f64).collect();
+        let result = p95_f64(&data);
+        assert!((result - 95.0).abs() < 1.0, "p95 of 0..100 should be ~95, got {result}");
+    }
+
+    #[test]
+    fn p95_ms_microseconds() {
+        let data: Vec<u64> = (0..100).map(|i| i * 1000).collect(); // 0ms..99ms
+        let result = p95_ms(&data);
+        assert!((result - 95.0).abs() < 1.0, "p95_ms should be ~95ms, got {result}");
+    }
+
+    #[test]
+    fn snapshot_resets_counters() {
+        let mut stats = PipelineStats::new();
+        stats.record_capture();
+        let _ = stats.snapshot_and_reset();
+        let snap = stats.snapshot_and_reset();
+        assert_eq!(snap.capture_frames, 0);
+    }
+
+    #[test]
+    fn display_format_contains_key_fields() {
+        let mut stats = PipelineStats::new();
+        stats.record_capture();
+        stats.record_encode(&FrameTimings {
+            total_us: 10000,
+            output_bytes: 5000,
+            ..Default::default()
+        });
+        stats.record_send(0.5, None);
+        let snap = stats.snapshot_and_reset();
+        let text = format!("{snap}");
+        assert!(text.contains("capture_fps="));
+        assert!(text.contains("encoded_fps="));
+        assert!(text.contains("sent_fps="));
+        assert!(text.contains("total_p95="));
+    }
+}

src/webrtc.rs

@@ -19,11 +19,13 @@ const HTML_PAGE: &str = r#"<!DOCTYPE html>
 video{max-width:90vw;max-height:80vh;border:1px solid #333}
 #status{margin:12px;font-size:14px;color:#aaa}
 #debug{position:fixed;bottom:8px;left:8px;font-size:11px;color:#666;max-width:90vw;white-space:pre-wrap}
+#stats-panel{position:fixed;top:8px;right:8px;background:rgba(0,0,0,0.7);color:#0f0;font:11px monospace;padding:6px 10px;border-radius:4px;z-index:100;pointer-events:none;max-width:90vw;white-space:pre;line-height:1.5}
 </style></head>
 <body>
 <div id="status">Connecting...</div>
 <video id="video" autoplay playsinline muted></video>
 <pre id="debug"></pre>
+<div id="stats-panel"></div>
 <script>
 const status = document.getElementById('status');
 const video  = document.getElementById('video');
@@ -51,25 +53,92 @@ function preferH264(sdp) {
 }
 
 function installStatsLogger(peer) {
+  const panel = document.getElementById('stats-panel');
+  let prev = null;
+  const intervalSecs = 1;
+
   setInterval(() => {
     if (peer !== pc) return;
-    const v = video;
-    log(`video: readyState=${v.readyState} currentTime=${v.currentTime.toFixed(2)} ` +
-        `paused=${v.paused} width=${v.videoWidth} height=${v.videoHeight} ` +
-        `srcObject=${v.srcObject ? 'yes' : 'no'}`);
     peer.getStats().then(stats => {
+      let rtp = null, rtt = null, codecStr = '';
+      let freezeCount = null, totalFreezesDuration = null;
+
       stats.forEach(report => {
-        if (report.type === 'inbound-rtp' && report.kind === 'video') {
-          log(`RTP-in: packetsReceived=${report.packetsReceived} packetsLost=${report.packetsLost} ` +
-              `bytesReceived=${report.bytesReceived} framesDecoded=${report.framesDecoded} ` +
-              `framesDropped=${report.framesDropped} codecId=${report.codecId}`);
-        }
-        if (report.type === 'codec' && report.mimeType && report.mimeType.includes('H264')) {
-          log(`Codec: ${report.mimeType} ${report.payloadType} sdpFmtpLine=${report.sdpFmtpLine}`);
+        if (report.type === 'inbound-rtp' && report.kind === 'video') rtp = report;
+        if (report.type === 'codec' && report.mimeType && report.mimeType.includes('H264'))
+          codecStr = report.mimeType + ' ' + (report.payloadType || '');
+        // candidate-pair: feature-detect 'selected' property
+        if (report.type === 'candidate-pair') {
+          const isSel = ('selected' in report) ? report.selected : report.state === 'succeeded';
+          if (isSel && typeof report.currentRoundTripTime === 'number') rtt = report.currentRoundTripTime;
         }
       });
+
+      // Freeze stats (feature-detect)
+      if (rtp && typeof rtp.freezeCount !== 'undefined') {
+        freezeCount = rtp.freezeCount;
+        totalFreezesDuration = rtp.totalFreezesDuration;
+      }
+
+      if (!rtp) return;
+
+      const cur = {
+        framesDecoded: rtp.framesDecoded || 0,
+        framesDropped: rtp.framesDropped || 0,
+        framesPerSecond: rtp.framesPerSecond || 0,
+        packetsLost: rtp.packetsLost || 0,
+        jitter: rtp.jitter || 0,
+        bytesReceived: rtp.bytesReceived || 0,
+        totalDecodeTime: rtp.totalDecodeTime || 0,
+        jitterBufferDelay: rtp.jitterBufferDelay || 0,
+        jitterBufferEmittedCount: rtp.jitterBufferEmittedCount || 0,
+        freezeCount: freezeCount,
+        totalFreezesDuration: totalFreezesDuration,
+        rtt: rtt,
+      };
+
+      // Raw log to debug element (backward compat)
+      log('RTP-in: decoded=' + cur.framesDecoded + ' lost=' + cur.packetsLost +
+          ' bytes=' + cur.bytesReceived + ' fps=' + cur.framesPerSecond +
+          (codecStr ? ' codec=' + codecStr : ''));
+
+      if (!prev) { prev = cur; return; }
+
+      // Compute deltas
+      const dFrames = cur.framesDecoded - prev.framesDecoded;
+      const dDropped = cur.framesDropped - prev.framesDropped;
+      const dLost = cur.packetsLost - prev.packetsLost;
+      const dBytes = cur.bytesReceived - prev.bytesReceived;
+      const dDecodeTime = cur.totalDecodeTime - prev.totalDecodeTime;
+      const dJitterBufDelay = cur.jitterBufferDelay - prev.jitterBufferDelay;
+      const dJitterBufCount = cur.jitterBufferEmittedCount - prev.jitterBufferEmittedCount;
+      const kbps = Math.round(dBytes * 8 / intervalSecs / 1000);
+      const decodeMs = dFrames > 0 ? (dDecodeTime / dFrames * 1000).toFixed(1) : '—';
+      const jitterBufMs = dJitterBufCount > 0 ? (dJitterBufDelay / dJitterBufCount * 1000).toFixed(1) : '—';
+      const jitterMs = (cur.jitter * 1000).toFixed(1);
+      const rttMs = cur.rtt !== null ? (cur.rtt * 1000).toFixed(1) : null;
+
+      let line = 'FPS:' + cur.framesPerSecond +
+        ' Decoded:' + cur.framesDecoded + '(+' + dFrames + ')' +
+        ' Dropped:' + cur.framesDropped + (dDropped > 0 ? '(+' + dDropped + ')' : '') +
+        ' Lost:' + dLost +
+        ' Jitter:' + jitterMs + 'ms' +
+        (rttMs !== null ? ' RTT:' + rttMs + 'ms' : '') +
+        ' Decode:' + decodeMs + 'ms' +
+        ' JBuf:' + jitterBufMs + 'ms';
+
+      if (freezeCount !== null) {
+        const dFreeze = cur.freezeCount - (prev.freezeCount || 0);
+        if (cur.freezeCount > 0 || dFreeze > 0)
+          line += ' Freeze:' + cur.freezeCount + '(+' + dFreeze + ')';
+      }
+
+      line += ' ' + kbps + 'kbps';
+
+      panel.textContent = line;
+      prev = cur;
     }).catch(() => {});
-  }, 2000);
+  }, intervalSecs * 1000);
 }
 
 function connect() {
@@ -178,12 +247,16 @@ impl WebRtcState {
                     HTML_PAGE.len(),
                     HTML_PAGE
                 );
-                let _ = stream.write_all(resp.as_bytes());
+                if let Err(e) = stream.write_all(resp.as_bytes()) {
+                    tracing::debug!("HTTP write error: {e}");
+                }
             } else if req_str.starts_with("POST /sdp") {
                 let body = extract_body(&req_str);
                 if body.is_empty() {
                     let resp = "HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\nempty body";
-                    let _ = stream.write_all(resp.as_bytes());
+                    if let Err(e) = stream.write_all(resp.as_bytes()) {
+                        tracing::debug!("HTTP write error: {e}");
+                    }
                     continue;
                 }
 
@@ -206,17 +279,23 @@ impl WebRtcState {
                             answer_json.len(),
                             answer_json
                         );
-                        let _ = stream.write_all(resp.as_bytes());
+                        if let Err(e) = stream.write_all(resp.as_bytes()) {
+                            tracing::debug!("HTTP write error: {e}");
+                        }
                     }
                     Err(e) => {
                         tracing::error!("SDP offer handling failed: {e}");
-                        let resp = format!("HTTP/1.1 500 Error\r\nConnection: close\r\n\r\n{e}");
-                        let _ = stream.write_all(resp.as_bytes());
+                        let resp = "HTTP/1.1 500 Internal Server Error\r\nConnection: close\r\n\r\n";
+                        if let Err(e) = stream.write_all(resp.as_bytes()) {
+                            tracing::debug!("HTTP write error: {e}");
+                        }
                     }
                 }
             } else {
                 let resp = "HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\n";
-                let _ = stream.write_all(resp.as_bytes());
+                if let Err(e) = stream.write_all(resp.as_bytes()) {
+                    tracing::debug!("HTTP write error: {e}");
+                }
             }
         }
         Ok(handled)
@@ -225,7 +304,7 @@ impl WebRtcState {
     pub fn poll_rtc(&mut self) -> Result<()> {
         if let Some(inner) = self.inner.as_mut() {
             if inner.poll_rtc()? {
-                tracing::warn!("WebRTC connection closed/failed; clearing connection state");
+                tracing::info!("WebRTC connection closed; clearing connection state");
                 self.inner = None;
             }
         }
@@ -246,8 +325,14 @@ impl WebRtcState {
     }
 
     pub fn write_h264_frame(&mut self, data: &[u8], frame_number: u64, fps: u32) -> Result<()> {
-        if let Some(inner) = self.inner.as_mut() {
-            inner.write_h264_frame(data, frame_number, fps)?;
+        let should_destroy = if let Some(inner) = self.inner.as_mut() {
+            inner.write_h264_frame(data, frame_number, fps)?
+        } else {
+            false
+        };
+        if should_destroy {
+            tracing::info!("WebRTC connection failed during write; clearing connection state");
+            self.inner = None;
         }
         Ok(())
     }
@@ -264,10 +349,49 @@ impl WebRtcInner {
 
         let socket = UdpSocket::bind("0.0.0.0:0")?;
         socket.set_nonblocking(true)?;
+
+        // Increase UDP send buffer to absorb IDR frame bursts (256KB IDR → ~145 RTP
+        // packets in a single poll_rtc loop).  Default Linux wmem is ~208KB which
+        // causes EAGAIN on large keyframes.  2MB comfortably buffers several IDRs.
+        const SND_BUF_REQ: usize = 2 * 1024 * 1024;
+        // SAFETY: fd is a valid UDP socket; setsockopt/getsockopt with SOL_SOCKET +
+        // SO_SNDBUF are safe on Linux.  We check the return value and log the actual
+        // kernel-assigned buffer (Linux may cap at wmem_max and/or double the value).
+        unsafe {
+            let fd = std::os::unix::io::AsRawFd::as_raw_fd(&socket);
+            let val: libc::c_int = SND_BUF_REQ as libc::c_int;
+            let ret = libc::setsockopt(
+                fd,
+                libc::SOL_SOCKET,
+                libc::SO_SNDBUF,
+                &val as *const libc::c_int as *const libc::c_void,
+                std::mem::size_of::<libc::c_int>() as libc::socklen_t,
+            );
+            if ret < 0 {
+                tracing::warn!("setsockopt SO_SNDBUF failed (errno {})", std::io::Error::last_os_error());
+            }
+            let mut actual: libc::c_int = 0;
+            let mut actual_len: libc::socklen_t = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
+            let gret = libc::getsockopt(
+                fd,
+                libc::SOL_SOCKET,
+                libc::SO_SNDBUF,
+                &mut actual as *mut libc::c_int as *mut libc::c_void,
+                &mut actual_len,
+            );
+            if gret == 0 {
+                tracing::info!(
+                    "UDP send buffer: requested {}KB, actual {}KB",
+                    SND_BUF_REQ / 1024,
+                    actual / 1024,
+                );
+            }
+        }
+
         let local_addr = socket.local_addr()?;
 
         let lan_ip = local_ip().unwrap_or_else(|| {
-            tracing::warn!("Failed to detect LAN IP, falling back to 127.0.0.1");
+            tracing::debug!("Failed to detect LAN IP, falling back to 127.0.0.1");
             "127.0.0.1".to_string()
         });
         let candidate_addr: SocketAddr = format!("{lan_ip}:{}", local_addr.port()).parse()?;
@@ -311,18 +435,14 @@ impl WebRtcInner {
     }
 
     fn discover_video_params(&mut self) {
-        for s in ["0", "1", "2", "3"] {
-            let mid: Mid = s.into();
-            if let Some(media) = self.rtc.media(mid) {
-                if media.kind() == MediaKind::Video {
-                    tracing::info!("Found video media: mid={mid}");
-                    self.video_mid = Some(mid);
-                    break;
+        let mid = match self.video_mid {
+            Some(m) => m,
+            None => {
+                tracing::debug!("discover_video_params: no video_mid yet");
+                return;
             }
-            }
-        }
-
-        if let Some(mid) = self.video_mid {
+        };
+        self.video_pt = None;
         if let Some(writer) = self.rtc.writer(mid) {
             for pp in writer.payload_params() {
                 tracing::debug!("Codec: pt={:?} spec={:?}", pp.pt(), pp.spec());
@@ -333,6 +453,8 @@ impl WebRtcInner {
                 }
             }
         }
+        if self.video_pt.is_none() {
+            tracing::warn!("discover_video_params: no H.264 codec found for mid={mid}");
         }
     }
 
@@ -340,13 +462,20 @@ impl WebRtcInner {
         loop {
             match self.rtc.poll_output() {
                 Ok(Output::Transmit(t)) => {
-                    tracing::info!("TX {} bytes -> {}", t.contents.len(), t.destination);
+                    tracing::trace!("TX {} bytes -> {}", t.contents.len(), t.destination);
                     if let Err(e) = self.socket.send_to(&t.contents, t.destination) {
-                        tracing::warn!("UDP send error: {e}");
+                        if e.kind() == std::io::ErrorKind::WouldBlock {
+                            tracing::debug!(
+                                "UDP send WouldBlock ({} bytes) — send buffer full",
+                                t.contents.len(),
+                            );
+                        } else {
+                            tracing::warn!("UDP send error to {}: {e}", t.destination);
+                        }
                     }
                 }
                 Ok(Output::Event(e)) => {
-                    tracing::info!("RTC event: {e:?}");
+                    tracing::debug!("RTC event: {e:?}");
                     match &e {
                         Event::Connected => {
                             tracing::info!("WebRTC connected!");
@@ -357,9 +486,21 @@ impl WebRtcInner {
                         Event::IceConnectionStateChange(IceConnectionState::Disconnected) => {
                             tracing::warn!("WebRTC disconnected");
                             self.connected = false;
+                            return Ok(true);
                         }
                         Event::MediaAdded(ma) => {
-                            tracing::info!("Media added: mid={:?}", ma.mid);
+                            tracing::info!("Media added: mid={} kind={:?}", ma.mid, ma.kind);
+                            if ma.kind == MediaKind::Video {
+                                if let Some(media) = self.rtc.media(ma.mid) {
+                                    if media.direction().is_sending()
+                                        && self.video_mid.is_none()
+                                    {
+                                        self.video_mid = Some(ma.mid);
+                                        tracing::info!("Captured video mid: {}", ma.mid);
+                                        self.discover_video_params();
+                                    }
+                                }
+                            }
                         }
                         _ => {
                             tracing::debug!("WebRTC event: {:?}", e);
@@ -369,7 +510,8 @@ impl WebRtcInner {
                 Ok(Output::Timeout(_t)) => break,
                 Err(e) => {
                     tracing::error!("rtc.poll_output error: {e}");
-                    break;
+                    self.connected = false;
+                    return Ok(true);
                 }
             }
         }
@@ -383,7 +525,7 @@ impl WebRtcInner {
                 Ok((n, source)) => {
                     recv_count += 1;
                     if recv_count <= 5 {
-                        tracing::info!("UDP recv {} bytes from {}", n, source);
+                        tracing::trace!("UDP recv {} bytes from {}", n, source);
                     }
                     let input = Input::Receive(
                         Instant::now(),
@@ -413,23 +555,23 @@ 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], frame_number: u64, fps: u32) -> Result<bool> {
         if !self.connected {
-            return Ok(());
+            return Ok(false);
         }
 
         let mid = match self.video_mid {
             Some(m) => m,
             None => {
-                tracing::warn!("write_h264: no video_mid");
-                return Ok(());
+                tracing::debug!("write_h264: no video_mid");
+                return Ok(false);
             }
         };
         let pt = match self.video_pt {
             Some(p) => p,
             None => {
-                tracing::warn!("write_h264: no video_pt");
-                return Ok(());
+                tracing::debug!("write_h264: no video_pt");
+                return Ok(false);
             }
         };
 
@@ -439,7 +581,7 @@ impl WebRtcInner {
                     "write_h264: skipping non-IDR frame ({} bytes), waiting for keyframe",
                     data.len()
                 );
-                return Ok(());
+                return Ok(false);
             }
             tracing::info!(
                 "write_h264: got IDR keyframe ({} bytes), starting playback",
@@ -457,8 +599,8 @@ impl WebRtcInner {
         let writer = match self.rtc.writer(mid) {
             Some(w) => w,
             None => {
-                tracing::warn!("write_h264: no writer for mid={mid}");
-                return Ok(());
+                tracing::debug!("write_h264: no writer for mid={mid}");
+                return Ok(false);
             }
         };
 
@@ -472,9 +614,9 @@ impl WebRtcInner {
             .write(pt, Instant::now(), rtp_time, data)
             .map_err(|e| anyhow::anyhow!("writer.write: {e}"))?;
 
-        self.poll_rtc()?;
+        let should_destroy = self.poll_rtc()?;
 
-        Ok(())
+        Ok(should_destroy)
     }
 
     fn is_connected(&self) -> bool {
@@ -510,16 +652,17 @@ fn local_ip() -> Option<String> {
 
 fn is_idr_nalu(data: &[u8]) -> bool {
     let mut i = 0;
-    while i + 4 < data.len() {
-        if data[i..i + 4] == [0, 0, 0, 1] {
-            let nal_type = data[i + 4] & 0x1F;
-            if nal_type == 5 {
+    while i < data.len() {
+        let tail = &data[i..];
+        if tail.starts_with(&[0, 0, 0, 1]) {
+            let Some(&header) = tail.get(4) else { break };
+            if header & 0x1F == 5 {
                 return true;
             }
             i += 5;
-        } else if i + 3 < data.len() && data[i..i + 3] == [0, 0, 1] {
-            let nal_type = data[i + 3] & 0x1F;
-            if nal_type == 5 {
+        } else if tail.starts_with(&[0, 0, 1]) {
+            let Some(&header) = tail.get(3) else { break };
+            if header & 0x1F == 5 {
                 return true;
             }
             i += 4;
@@ -529,3 +672,78 @@ fn is_idr_nalu(data: &[u8]) -> bool {
     }
     false
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn empty_data() {
+        assert!(!is_idr_nalu(&[]));
+    }
+
+    #[test]
+    fn short_data_no_start_code() {
+        assert!(!is_idr_nalu(&[0]));
+        assert!(!is_idr_nalu(&[0, 0]));
+        assert!(!is_idr_nalu(&[1, 2, 3]));
+    }
+
+    #[test]
+    fn three_byte_start_code_no_nal_header() {
+        assert!(!is_idr_nalu(&[0, 0, 1]));
+    }
+
+    #[test]
+    fn four_byte_start_code_no_nal_header() {
+        assert!(!is_idr_nalu(&[0, 0, 0, 1]));
+    }
+
+    #[test]
+    fn three_byte_start_code_idr_at_tail() {
+        assert!(is_idr_nalu(&[0, 0, 1, 0x65]));
+        assert!(!is_idr_nalu(&[0, 0, 1, 0x01]));
+    }
+
+    #[test]
+    fn four_byte_start_code_idr_at_tail() {
+        assert!(is_idr_nalu(&[0, 0, 0, 1, 0x65]));
+        assert!(!is_idr_nalu(&[0, 0, 0, 1, 0x01]));
+    }
+
+    #[test]
+    fn idr_in_middle_of_frame() {
+        let data: Vec<u8> = [
+            &[0, 0, 0, 1, 0x67][..], // SPS
+            &[0, 0, 0, 1, 0x68][..], // PPS
+            &[0, 0, 0, 1, 0x65][..], // IDR
+        ]
+        .concat();
+        assert!(is_idr_nalu(&data));
+    }
+
+    #[test]
+    fn no_idr_in_frame() {
+        let data: Vec<u8> = [
+            &[0, 0, 0, 1, 0x67][..], // SPS
+            &[0, 0, 0, 1, 0x68][..], // PPS
+        ]
+        .concat();
+        assert!(!is_idr_nalu(&data));
+    }
+
+    #[test]
+    fn mixed_start_code_lengths() {
+        let data: Vec<u8> = [
+            &[0, 0, 0, 1, 0x67][..], // SPS (4-byte start code)
+            &[0, 0, 1, 0x65][..],    // IDR (3-byte start code)
+        ]
+        .concat();
+        assert!(is_idr_nalu(&data));
+    }
+
+    #[test]
+    fn all_zeros() {
+        assert!(!is_idr_nalu(&[0, 0, 0, 0, 0, 0, 0, 0]));
+    }
+}