adding ability to control local video track encode parameters

.changeset/local-video-encoding-controls.md

@@ -0,0 +1,6 @@
+---
+libwebrtc: patch
+livekit: patch
+---
+
+Add simulcast-aware runtime video encoding limit controls for local video tracks, preserving the publish-time simulcast ladder while applying track-level caps through the `local_video` publisher/subscriber example RPC.

examples/local_video/Cargo.toml

@@ -49,6 +49,8 @@ parking_lot = { workspace = true, features = ["deadlock_detection"] }
 anyhow = { workspace = true }
 chrono = "0.4"
 bytemuck = { version = "1.16", features = ["derive"] }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
 
 nokhwa = { git = "https://github.com/l1npengtul/nokhwa", rev = "4923ecab7cf26f9dba83867a15a9d8662d021296", default-features = false, features = ["output-threaded"] }
 

examples/local_video/src/encoding_control.rs

@@ -0,0 +1,26 @@
+use serde::{Deserialize, Serialize};
+
+/// RPC method name used to negotiate video encoding limits between the
+/// subscriber (caller) and the publisher (responder).
+pub const SET_VIDEO_ENCODING_LIMITS_METHOD: &str = "set-video-encoding-limits";
+
+/// Payload sent by the subscriber to request new video encoding limits on a
+/// publisher's track.
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct SetEncodingLimitsRequest {
+    pub track_sid: String,
+    pub bitrate_bps: Option<u64>,
+    pub max_framerate: Option<f64>,
+    pub scale_resolution_down_by: Option<f64>,
+    pub reason: String,
+}
+
+/// Payload returned by the publisher describing the encoding limits it applied.
+#[derive(Debug, Deserialize, Serialize)]
+pub struct SetEncodingLimitsResponse {
+    pub applied_bitrate_bps: Option<u64>,
+    pub applied_max_framerate: Option<f64>,
+    pub applied_scale_resolution_down_by: Option<f64>,
+    pub track_sid: String,
+}

examples/local_video/src/publisher.rs

@@ -30,11 +30,15 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use yuv_sys;
 
 mod codec_display;
+mod encoding_control;
 mod test_pattern;
 mod timestamp_burn;
 mod video_display;
 mod viewport_aspect;
 
+use encoding_control::{
+    SetEncodingLimitsRequest, SetEncodingLimitsResponse, SET_VIDEO_ENCODING_LIMITS_METHOD,
+};
 use test_pattern::TestPattern;
 use timestamp_burn::TimestampOverlay;
 use video_display::{align_up, PublisherTimingSample, SharedYuv};
@@ -214,6 +218,7 @@ fn unix_time_us_now() -> u64 {
 }
 
 const MAX_BACKEND_CAPTURE_TIMESTAMP_AGE_US: u64 = 5_000_000;
+const HIGH_RID: &str = "f";
 
 #[derive(Default)]
 struct CaptureTimestampLogState {
@@ -329,24 +334,83 @@ struct PublisherTimingSummary {
     capture_to_webrtc_total_ms: RollingMs,
 }
 
-fn find_video_outbound_encoder(stats: &[livekit::webrtc::stats::RtcStats]) -> Option<&str> {
-    let mut fallback = None;
+#[derive(Clone, Default)]
+struct PublisherVideoOutboundStats {
+    rid: String,
+    active: bool,
+    encoder_implementation: Option<String>,
+    target_bitrate_bps: Option<f64>,
+    frame_width: u32,
+    frame_height: u32,
+    frames_per_second: f64,
+}
+
+fn collect_video_outbound_stats(
+    stats: &[livekit::webrtc::stats::RtcStats],
+) -> Vec<PublisherVideoOutboundStats> {
+    let mut outbounds = Vec::new();
     for stat in stats {
         let livekit::webrtc::stats::RtcStats::OutboundRtp(outbound) = stat else {
             continue;
         };
-        if outbound.stream.kind != "video" || outbound.outbound.encoder_implementation.is_empty() {
+        if outbound.stream.kind != "video" {
             continue;
         }
 
-        let implementation = outbound.outbound.encoder_implementation.as_str();
-        if outbound.outbound.active {
-            return Some(implementation);
-        }
-        fallback.get_or_insert(implementation);
+        outbounds.push(PublisherVideoOutboundStats {
+            rid: outbound.outbound.rid.clone(),
+            active: outbound.outbound.active,
+            encoder_implementation: (!outbound.outbound.encoder_implementation.is_empty())
+                .then(|| outbound.outbound.encoder_implementation.clone()),
+            target_bitrate_bps: (outbound.outbound.target_bitrate > 0.0)
+                .then_some(outbound.outbound.target_bitrate),
+            frame_width: outbound.outbound.frame_width,
+            frame_height: outbound.outbound.frame_height,
+            frames_per_second: outbound.outbound.frames_per_second,
+        });
     }
 
-    fallback
+    outbounds
+}
+
+fn find_video_outbound_stats(
+    outbounds: &[PublisherVideoOutboundStats],
+) -> PublisherVideoOutboundStats {
+    outbounds
+        .iter()
+        .find(|outbound| outbound.active && outbound.rid == HIGH_RID)
+        .or_else(|| outbounds.iter().find(|outbound| outbound.active))
+        .or_else(|| outbounds.iter().find(|outbound| outbound.rid == HIGH_RID))
+        .or_else(|| outbounds.first())
+        .cloned()
+        .unwrap_or_default()
+}
+
+fn format_video_outbound_layers(outbounds: &[PublisherVideoOutboundStats]) -> Option<String> {
+    if outbounds.len() <= 1 {
+        return None;
+    }
+
+    let layers = outbounds
+        .iter()
+        .map(|outbound| {
+            let rid = if outbound.rid.is_empty() { "-" } else { outbound.rid.as_str() };
+            let active = if outbound.active { "*" } else { "" };
+            let bitrate = outbound
+                .target_bitrate_bps
+                .map(|bps| format!("{:.2}mbps", bps / 1_000_000.0))
+                .unwrap_or_else(|| "?mbps".to_string());
+            format!(
+                "{rid}{active}: {}x{} {:.1}fps {bitrate}",
+                outbound.frame_width,
+                outbound.frame_height,
+                outbound.frames_per_second.max(0.0),
+            )
+        })
+        .collect::<Vec<_>>()
+        .join(", ");
+
+    Some(format!("Publisher RTP encodings: {layers}"))
 }
 
 async fn update_publisher_encoder_overlay(
@@ -356,6 +420,7 @@ async fn update_publisher_encoder_overlay(
 ) {
     let mut logged_initial = false;
     let mut last_implementation = String::new();
+    let mut last_layers_line = String::new();
     let mut interval = tokio::time::interval(Duration::from_secs(1));
     interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
@@ -366,14 +431,29 @@ async fn update_publisher_encoder_overlay(
 
         match track.get_stats().await {
             Ok(stats) => {
-                if let Some(implementation) = find_video_outbound_encoder(&stats) {
+                let outbounds = collect_video_outbound_stats(&stats);
+                if let Some(layers_line) = format_video_outbound_layers(&outbounds) {
+                    debug!("{layers_line}");
+                    if layers_line != last_layers_line {
+                        info!("{layers_line}");
+                        last_layers_line = layers_line;
+                    }
+                }
+
+                let outbound = find_video_outbound_stats(&outbounds);
+                if let Some(implementation) = outbound.encoder_implementation {
                     if implementation != last_implementation {
                         info!("Publisher video encoder implementation: {implementation}");
-                        last_implementation = implementation.to_string();
+                        last_implementation = implementation.clone();
                     }
 
                     let mut shared = shared.lock();
-                    shared.codec_implementation = implementation.to_string();
+                    shared.codec_implementation = implementation;
+                    if let Some(target_bitrate_bps) = outbound.target_bitrate_bps {
+                        shared.encode_bitrate_mbps = Some(target_bitrate_bps / 1_000_000.0);
+                    }
+                } else if let Some(target_bitrate_bps) = outbound.target_bitrate_bps {
+                    shared.lock().encode_bitrate_mbps = Some(target_bitrate_bps / 1_000_000.0);
                 }
                 logged_initial = true;
             }
@@ -541,6 +621,80 @@ fn update_shared_timing_sample(
     }
 }
 
+fn register_encoding_limits_rpc(room: &Arc<Room>, publication: LocalTrackPublication) {
+    room.local_participant().register_rpc_method(
+        SET_VIDEO_ENCODING_LIMITS_METHOD.to_string(),
+        move |data| {
+            let publication = publication.clone();
+            Box::pin(async move {
+                debug!(
+                    "Raw video encoding limits RPC from {}: {}",
+                    data.caller_identity, data.payload
+                );
+                let request: SetEncodingLimitsRequest = serde_json::from_str(&data.payload)
+                    .map_err(|err| {
+                        RpcError::new(
+                            400,
+                            "invalid encoding limits request".to_string(),
+                            Some(err.to_string()),
+                        )
+                    })?;
+
+                let publication_sid = publication.sid().to_string();
+                if request.track_sid != publication_sid {
+                    return Err(RpcError::new(
+                        404,
+                        "track not found".to_string(),
+                        Some(request.track_sid),
+                    ));
+                }
+
+                info!(
+                    "{} requested video encoding limits: track={}, {:?} bps, {:?} fps, {:?}x scale ({})",
+                    data.caller_identity,
+                    request.track_sid,
+                    request.bitrate_bps,
+                    request.max_framerate,
+                    request.scale_resolution_down_by,
+                    request.reason
+                );
+
+                let limits = VideoEncodingLimits {
+                    max_bitrate: request.bitrate_bps,
+                    max_framerate: request.max_framerate,
+                    scale_resolution_down_by: request.scale_resolution_down_by,
+                };
+                debug!("Applying track-level publisher encoding limits");
+                publication.set_video_encoding_limits(limits).map_err(|err| {
+                    RpcError::new(500, format!("set encoding limits failed: {err}"), None)
+                })?;
+
+                info!(
+                    "Applied video encoding limits: track={}, {:?} bps, {:?} fps, {:?}x scale",
+                    publication_sid,
+                    request.bitrate_bps,
+                    request.max_framerate,
+                    request.scale_resolution_down_by
+                );
+
+                serde_json::to_string(&SetEncodingLimitsResponse {
+                    applied_bitrate_bps: request.bitrate_bps,
+                    applied_max_framerate: request.max_framerate,
+                    applied_scale_resolution_down_by: request.scale_resolution_down_by,
+                    track_sid: publication_sid,
+                })
+                .map_err(|err| {
+                    RpcError::new(
+                        500,
+                        "failed to serialize encoding limits response".to_string(),
+                        Some(err.to_string()),
+                    )
+                })
+            })
+        },
+    );
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -955,21 +1109,23 @@ async fn run(args: Args, ctrl_c_received: Arc<AtomicBool>) -> Result<()> {
         .publish_track(LocalTrack::Video(track.clone()), publish_opts(requested_codec))
         .await;
 
-    let actual_codec = if let Err(e) = publish_result {
-        if matches!(requested_codec, VideoCodec::H265) {
+    let (publication, actual_codec) = match publish_result {
+        Ok(publication) => {
+            info!("Published camera track");
+            (publication, requested_codec)
+        }
+        Err(e) if matches!(requested_codec, VideoCodec::H265) => {
             log::warn!("H.265 publish failed ({}). Falling back to H.264...", e);
-            room.local_participant()
+            let publication = room
+                .local_participant()
                 .publish_track(LocalTrack::Video(track.clone()), publish_opts(VideoCodec::H264))
                 .await?;
             info!("Published camera track with H.264 fallback");
-            VideoCodec::H264
-        } else {
-            return Err(e.into());
+            (publication, VideoCodec::H264)
         }
-    } else {
-        info!("Published camera track");
-        requested_codec
+        Err(e) => return Err(e.into()),
     };
+    register_encoding_limits_rpc(&room, publication);
 
     let capture_config = CaptureConfig {
         fps: args.fps,