fix(video-streamer): add codec-aware VP9 keyframe detection (#1702)

## Summary
- Add VP9 keyframe detection alongside existing VP8 support, based on
the VP9 bitstream specification (profiles 0-3)
- Thread the `VpxCodec` type through the iterator and block tag layers
so keyframe checks use the correct codec-specific logic
- Set `VpxEncoderPreset::BestPerformance` on the re-encoding encoder for
improved throughput during session shadowing

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: irvingoujAtDevolution

crates/video-streamer/src/streamer/block_tag.rs

@@ -1,6 +1,7 @@
 use std::fmt;
 
 use anyhow::Context;
+use cadeau::xmf::vpx::VpxCodec;
 use webm_iterable::matroska_spec::{Block, Master, MatroskaSpec, SimpleBlock};
 
 #[derive(Clone)]
@@ -35,7 +36,7 @@ impl fmt::Debug for VideoBlock {
 }
 
 impl VideoBlock {
-    pub(crate) fn new(tag: MatroskaSpec, cluster_timestamp: Option<u64>) -> anyhow::Result<Self> {
+    pub(crate) fn new(tag: MatroskaSpec, cluster_timestamp: Option<u64>, codec: VpxCodec) -> anyhow::Result<Self> {
         let result = match tag {
             MatroskaSpec::BlockGroup(Master::Full(children)) => {
                 let block = children
@@ -51,7 +52,10 @@ impl VideoBlock {
 
                 let block = Block::try_from(block)?;
                 let timestamp = block.timestamp;
-                let is_key_frame = block.read_frame_data()?.iter().any(|frame| is_key_frame(frame.data));
+                let is_key_frame = block
+                    .read_frame_data()?
+                    .iter()
+                    .any(|frame| is_vpx_key_frame(frame.data, codec));
 
                 Self {
                     cluster_timestamp,
@@ -121,9 +125,144 @@ impl VideoBlock {
     }
 }
 
-fn is_key_frame(buffer: &[u8]) -> bool {
+pub(crate) fn is_vpx_key_frame(buffer: &[u8], codec: VpxCodec) -> bool {
+    match codec {
+        VpxCodec::VP8 => is_vp8_key_frame(buffer),
+        VpxCodec::VP9 => is_vp9_key_frame(buffer),
+    }
+}
+
+/// VP8 keyframe detection.
+///
+/// RFC 6386 Section 9.1 "Uncompressed Data Chunk":
+/// https://datatracker.ietf.org/doc/html/rfc6386#section-9.1
+///
+/// First byte layout (LSB-first bitstream):
+///   bit 0: frame_type (0 = key frame, 1 = inter frame)
+///   bits 1-2: version
+///   bit 3: show_frame
+///   bits 4-7: first_part_size (bits 0-3)
+///
+/// We only need bit 0: `buffer[0] & 0x1 == 0` means keyframe.
+fn is_vp8_key_frame(buffer: &[u8]) -> bool {
     if buffer.is_empty() {
         return false;
     }
     buffer[0] & 0x1 == 0
 }
+
+/// VP9 keyframe detection.
+///
+/// VP9 Bitstream & Decoding Process Specification v0.6, Section 6.2 "Uncompressed header syntax":
+/// https://storage.googleapis.com/downloads.webmproject.org/docs/vp9/vp9-bitstream-specification-v0.6-20160331-draft.pdf
+///
+/// Unlike VP8 which uses a LSB-first bitstream, VP9 uses a MSB-first bitstream.
+/// The first byte layout depends on the profile:
+///
+/// For profiles 0-2 (first byte, MSB to LSB):
+///   bits 7-6: frame_marker (must be 0b10 to identify VP9)
+///   bit 5:    profile_low_bit
+///   bit 4:    profile_high_bit
+///   bit 3:    show_existing_frame (if 1, frame is a reference to an already-decoded frame, not a keyframe)
+///   bit 2:    frame_type (0 = KEY_FRAME, 1 = NON_KEY_FRAME)
+///   bits 1-0: (remaining header fields)
+///
+/// For profile 3 (first byte, MSB to LSB):
+///   bits 7-6: frame_marker (must be 0b10)
+///   bit 5:    profile_low_bit (1)
+///   bit 4:    profile_high_bit (1)
+///   bit 3:    reserved_zero
+///   bit 2:    show_existing_frame
+///   bit 1:    frame_type (0 = KEY_FRAME, 1 = NON_KEY_FRAME)
+///   bit 0:    (remaining header fields)
+///
+/// Profile 3 has an extra reserved_zero bit after the profile bits, which shifts
+/// show_existing_frame and frame_type one position to the right.
+///
+/// Note: the profile is encoded with swapped bit order: `profile = (high_bit << 1) | low_bit`,
+/// i.e. `profile = (bit4 << 1) | bit5`.
+///
+/// A frame is a keyframe when: show_existing_frame == 0 AND frame_type == 0.
+pub(crate) fn is_vp9_key_frame(buffer: &[u8]) -> bool {
+    if buffer.is_empty() {
+        return false;
+    }
+    let b0 = buffer[0];
+
+    // Validate frame_marker (bits 7-6) is 0b10
+    if (b0 >> 6) != 0b10 {
+        return false;
+    }
+
+    // profile = (high_bit << 1) | low_bit = (bit4 << 1) | bit5
+    let profile = (((b0 >> 4) & 1) << 1) | ((b0 >> 5) & 1);
+
+    if profile == 3 {
+        // Profile 3: show_existing_frame is bit 2, frame_type is bit 1
+        (b0 & 0x04) == 0 && (b0 & 0x02) == 0
+    } else {
+        // Profiles 0-2: show_existing_frame is bit 3, frame_type is bit 2
+        (b0 & 0x08) == 0 && (b0 & 0x04) == 0
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::is_vp9_key_frame;
+
+    #[test]
+    fn vp9_empty_buffer_is_not_keyframe() {
+        assert!(!is_vp9_key_frame(&[]));
+    }
+
+    #[test]
+    fn vp9_marker_mismatch_is_not_keyframe() {
+        // frame_marker (bits 7-6) != 0b10 → rejected even if other bits look like keyframe.
+        // 0x00 → frame_marker = 0b00
+        assert!(!is_vp9_key_frame(&[0x00]));
+    }
+
+    #[test]
+    fn vp9_profiles_0_to_2_keyframe_detected() {
+        // Profile 0: frame_marker=0b10, profile_low=0, profile_high=0,
+        // show_existing_frame(bit3)=0, frame_type(bit2)=0.
+        // 0b1000_0000 = 0x80
+        assert!(is_vp9_key_frame(&[0x80]));
+    }
+
+    #[test]
+    fn vp9_profiles_0_to_2_show_existing_frame_is_not_keyframe() {
+        // Profile 0: show_existing_frame(bit3)=1, frame_type(bit2)=0.
+        // 0b1000_1000 = 0x88
+        assert!(!is_vp9_key_frame(&[0x88]));
+    }
+
+    #[test]
+    fn vp9_profiles_0_to_2_inter_frame_is_not_keyframe() {
+        // Profile 0: show_existing_frame(bit3)=0, frame_type(bit2)=1.
+        // 0b1000_0100 = 0x84
+        assert!(!is_vp9_key_frame(&[0x84]));
+    }
+
+    #[test]
+    fn vp9_profile_3_keyframe_detected() {
+        // Profile 3: frame_marker=0b10, profile_low(bit5)=1, profile_high(bit4)=1,
+        // reserved_zero(bit3)=0, show_existing_frame(bit2)=0, frame_type(bit1)=0.
+        // 0b1011_0000 = 0xB0
+        assert!(is_vp9_key_frame(&[0xB0]));
+    }
+
+    #[test]
+    fn vp9_profile_3_show_existing_frame_is_not_keyframe() {
+        // Profile 3: show_existing_frame(bit2)=1, frame_type(bit1)=0.
+        // 0b1011_0100 = 0xB4
+        assert!(!is_vp9_key_frame(&[0xB4]));
+    }
+
+    #[test]
+    fn vp9_profile_3_inter_frame_is_not_keyframe() {
+        // Profile 3: show_existing_frame(bit2)=0, frame_type(bit1)=1.
+        // 0b1011_0010 = 0xB2
+        assert!(!is_vp9_key_frame(&[0xB2]));
+    }
+}

crates/video-streamer/src/streamer/iter.rs

@@ -1,12 +1,13 @@
 use std::io::Seek;
 
 use anyhow::Context;
-use cadeau::xmf::vpx::is_key_frame;
+use cadeau::xmf::vpx::VpxCodec;
 use thiserror::Error;
 use webm_iterable::WebmIterator;
 use webm_iterable::errors::TagIteratorError;
 use webm_iterable::matroska_spec::{Block, Master, MatroskaSpec, SimpleBlock};
 
+use super::block_tag::is_vpx_key_frame;
 use crate::reopenable::Reopenable;
 
 #[derive(Debug, Clone, Copy)]
@@ -40,6 +41,9 @@ pub(crate) struct WebmPositionedIterator<R: std::io::Read + Seek + Reopenable> {
     rolled_back_between_cluster: bool,
 
     should_emit_cache: Option<MatroskaSpec>,
+
+    // VPX codec type for codec-aware keyframe detection.
+    codec: VpxCodec,
 }
 
 #[derive(Debug, Error)]
@@ -60,19 +64,20 @@ impl<R> WebmPositionedIterator<R>
 where
     R: std::io::Read + Seek + Reopenable,
 {
-    pub(crate) fn new(mut inner: WebmIterator<R>) -> Self {
+    pub(crate) fn new(mut inner: WebmIterator<R>, codec: VpxCodec, cluster_start_position: usize) -> Self {
         inner.emit_master_end_when_eof(false);
         Self {
             inner: Some(inner),
-            previous_emitted_tag_postion: 0,
-            last_cluster_position: None,
+            previous_emitted_tag_postion: cluster_start_position,
+            last_cluster_position: Some(cluster_start_position),
             rollback_record: None,
             rolled_back_between_cluster: false,
             should_emit_cache: None,
             last_key_frame_info: LastKeyFrameInfo::NotMet {
                 cluster_timestamp: None,
-                cluster_start_position: None,
+                cluster_start_position: Some(cluster_start_position),
             },
+            codec,
         }
     }
 
@@ -132,7 +137,7 @@ where
                 return result.map(|result| result.map_err(|err| err.into()));
             }
 
-            match Self::is_key_frame(tag) {
+            match self.is_key_frame(tag) {
                 Err(e) => {
                     return Some(Err(e));
                 }
@@ -289,7 +294,7 @@ where
         Ok(())
     }
 
-    fn is_key_frame(tag: &MatroskaSpec) -> Result<bool, IteratorError> {
+    fn is_key_frame(&self, tag: &MatroskaSpec) -> Result<bool, IteratorError> {
         match tag {
             MatroskaSpec::BlockGroup(Master::Full(children)) => {
                 let block = children
@@ -308,7 +313,7 @@ where
                 let block = Block::try_from(block)?;
                 let frame = block.read_frame_data()?;
 
-                Ok(frame.into_iter().any(|frame| is_key_frame(frame.data)))
+                Ok(frame.into_iter().any(|frame| is_vpx_key_frame(frame.data, self.codec)))
             }
             MatroskaSpec::SimpleBlock(data) => {
                 let simple_block = SimpleBlock::try_from(data)?;

crates/video-streamer/src/streamer/mod.rs

@@ -35,14 +35,11 @@ pub fn webm_stream(
     config: StreamingConfig,
     when_new_chunk_appended: impl Fn() -> tokio::sync::oneshot::Receiver<()>,
 ) -> anyhow::Result<()> {
-    let mut webm_itr = WebmPositionedIterator::new(WebmIterator::new(
-        input_stream,
-        &[MatroskaSpec::BlockGroup(Master::Start)],
-    ));
+    let mut raw_itr = WebmIterator::new(input_stream, &[MatroskaSpec::BlockGroup(Master::Start)]);
     let mut headers = vec![];
 
     // we extract all the headers before the first cluster
-    while let Some(tag) = webm_itr.next() {
+    for tag in raw_itr.by_ref() {
         let tag = tag?;
         if matches!(tag, MatroskaSpec::Cluster(Master::Start)) {
             break;
@@ -51,6 +48,8 @@ pub fn webm_stream(
         headers.push(tag);
     }
     let encode_writer_config = EncodeWriterConfig::try_from((headers.as_slice(), &config))?;
+    let cluster_start_position = raw_itr.last_emitted_tag_offset();
+    let mut webm_itr = WebmPositionedIterator::new(raw_itr, encode_writer_config.codec, cluster_start_position);
 
     // we run to the last cluster, skipping everything that has been played
     while let Some(tag) = webm_itr.next() {

crates/video-streamer/src/streamer/tag_writers.rs

@@ -1,7 +1,7 @@
 use std::time::{Duration, Instant};
 
 use anyhow::Context;
-use cadeau::xmf::vpx::{VpxCodec, VpxDecoder, VpxEncoder};
+use cadeau::xmf::vpx::{VpxCodec, VpxDecoder, VpxEncoder, VpxEncoderPreset};
 use webm_iterable::errors::TagWriterError;
 use webm_iterable::matroska_spec::{Master, MatroskaSpec, SimpleBlock};
 use webm_iterable::{WebmWriter, WriteOptions};
@@ -108,6 +108,7 @@ where
     cluster_timestamp: Option<u64>,
     encoder: VpxEncoder,
     decoder: VpxDecoder,
+    codec: VpxCodec,
     cut_block_state: CutBlockState,
     last_encoded_abs_time: Option<u64>,
 
@@ -177,6 +178,7 @@ where
             .height(config.height)
             .threads(config.threads)
             .bitrate(256 * 1024)
+            .preset(VpxEncoderPreset::BestPerformance)
             .build()
             .inspect_err(|error| {
                 error!(
@@ -199,6 +201,7 @@ where
                 cluster_timestamp: None,
                 encoder,
                 decoder,
+                codec: config.codec,
                 cut_block_state: CutBlockState::HaventMet,
                 last_encoded_abs_time: None,
                 #[cfg(feature = "perf-diagnostics")]
@@ -267,7 +270,7 @@ where
             }
         }
 
-        let video_block = VideoBlock::new(tag, self.cluster_timestamp)?;
+        let video_block = VideoBlock::new(tag, self.cluster_timestamp, self.codec)?;
         perf_trace!(
             block_timestamp = video_block.timestamp,
             cluster_timestamp = ?self.cluster_timestamp,