fix(dgw): fix streamer failure when frame rate is too low

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

@@ -27,7 +27,7 @@ pub(crate) enum LastKeyFrameInfo {
 pub(crate) struct WebmPositionedIterator<R: std::io::Read + Seek + Reopenable> {
     inner: Option<WebmIterator<R>>,
     // The absolute position of the last tag emitted
-    last_tag_position: usize,
+    previous_emitted_tag_postion: usize,
     // The absolute position of the last cluster start tag emitted
     last_cluster_position: Option<usize>,
 
@@ -66,7 +66,7 @@ where
         inner.emit_master_end_when_eof(false);
         Self {
             inner: Some(inner),
-            last_tag_position: 0,
+            previous_emitted_tag_postion: 0,
             last_cluster_position: None,
             rollback_record: None,
             rolled_back_between_cluster: false,
@@ -87,17 +87,17 @@ where
 
         if result.is_none() {
             let record = self.rollback_record.unwrap_or(0);
-            if record + inner.last_emitted_tag_offset() > self.last_tag_position {
-                self.last_tag_position = record + inner.last_emitted_tag_offset();
+            if record + inner.last_emitted_tag_offset() > self.previous_emitted_tag_postion {
+                self.previous_emitted_tag_postion = record + inner.last_emitted_tag_offset();
             }
             return None;
         }
 
         if let Some(Ok(tag)) = &result {
             let record = self.rollback_record.unwrap_or(0);
             // The last emitted tag is relative, i.e when rollback, the last_emitted_tag_offset() will be reset to 0
-            if record + inner.last_emitted_tag_offset() >= self.last_tag_position {
-                self.last_tag_position = record + inner.last_emitted_tag_offset();
+            if record + inner.last_emitted_tag_offset() >= self.previous_emitted_tag_postion {
+                self.previous_emitted_tag_postion = record + inner.last_emitted_tag_offset();
             }
 
             if matches!(tag, MatroskaSpec::BlockGroup(Master::Full(_))) {
@@ -108,7 +108,7 @@ where
                 if let Err(e) =
                     self.correct_for_blockgroup_header()
                         .map_err(|_| IteratorError::PositionCorrectionError {
-                            before_correct_position: self.last_tag_position as u64,
+                            before_correct_position: self.previous_emitted_tag_postion as u64,
                         })
                 {
                     return Some(Err(e));
@@ -140,7 +140,7 @@ where
                 }
                 Ok(false) => {}
                 Ok(true) => {
-                    trace!(last_tag_position = self.last_tag_position, last_key_frame_info =?self.last_key_frame_info, "Key Frame Found");
+                    trace!(last_tag_position = self.previous_emitted_tag_postion, last_key_frame_info =?self.last_key_frame_info, "Key Frame Found");
                     match self.last_key_frame_info {
                         LastKeyFrameInfo::NotMet {
                             cluster_timestamp,
@@ -155,33 +155,33 @@ where
                             };
 
                             self.last_key_frame_info = LastKeyFrameInfo::Met {
-                                position: self.last_tag_position,
+                                position: self.previous_emitted_tag_postion,
                                 cluster_timestamp,
                                 cluster_start_position,
                             }
                         }
                         LastKeyFrameInfo::Met { ref mut position, .. } => {
-                            *position = self.last_tag_position;
+                            *position = self.previous_emitted_tag_postion;
                         }
                     }
                 }
             };
 
             if let Some(Ok(MatroskaSpec::Cluster(Master::Start))) = &result {
-                self.last_cluster_position = Some(self.last_tag_position);
+                self.last_cluster_position = Some(self.previous_emitted_tag_postion);
 
                 match self.last_key_frame_info {
                     LastKeyFrameInfo::NotMet {
                         ref mut cluster_start_position,
                         ..
                     } => {
-                        cluster_start_position.replace(self.last_tag_position);
+                        cluster_start_position.replace(self.previous_emitted_tag_postion);
                     }
                     LastKeyFrameInfo::Met {
                         ref mut cluster_start_position,
                         ..
                     } => {
-                        *cluster_start_position = self.last_tag_position;
+                        *cluster_start_position = self.previous_emitted_tag_postion;
                     }
                 };
 
@@ -200,19 +200,19 @@ where
 
     pub(crate) fn rollback_to_last_successful_tag(&mut self) -> anyhow::Result<()> {
         debug!(
-            last_tag_position = self.last_tag_position,
+            last_tag_position = self.previous_emitted_tag_postion,
             "Rolling back to last successful tag"
         );
         let inner = self.inner.take().context("no inner iterator")?;
         let mut file = inner.into_inner();
         file.reopen()?;
-        file.seek(std::io::SeekFrom::Start(self.last_tag_position as u64))?;
-        self.inner = Some(WebmIterator::new(file, &[MatroskaSpec::BlockGroup(Master::Start)]));
-        self.rollback_record = Some(self.last_tag_position);
+        file.seek(std::io::SeekFrom::Start(self.previous_emitted_tag_postion as u64))?;
+        self.new_inner(file);
+        self.rollback_record = Some(self.previous_emitted_tag_postion);
 
         if self
             .last_cluster_position
-            .map(|last_cluster_position| last_cluster_position != self.last_tag_position)
+            .map(|last_cluster_position| last_cluster_position != self.previous_emitted_tag_postion)
             .unwrap_or(false)
         {
             self.rolled_back_between_cluster = true;
@@ -247,14 +247,14 @@ where
         file.reopen()?;
         file.seek(std::io::SeekFrom::Start(last_key_frame_position as u64))?;
         self.rollback_record = Some(last_key_frame_position);
-        self.last_tag_position = last_key_frame_position;
-        self.inner = Some(WebmIterator::new(file, &[MatroskaSpec::BlockGroup(Master::Start)]));
+        self.previous_emitted_tag_postion = last_key_frame_position;
+        self.new_inner(file);
         self.last_cluster_position = Some(cluster_start_position);
         Ok(self.last_key_frame_info)
     }
 
-    pub(crate) fn last_tag_position(&self) -> usize {
-        self.last_tag_position
+    pub(crate) fn previous_emitted_tag_postion(&self) -> usize {
+        self.previous_emitted_tag_postion
     }
 
     // The BlockGroup element binary layout is like this
@@ -263,7 +263,7 @@ where
     fn correct_for_blockgroup_header(&mut self) -> anyhow::Result<()> {
         let file = self.inner.as_mut().context("inner is none")?.get_mut();
         let current_position = file.stream_position()?;
-        file.seek(std::io::SeekFrom::Start(self.last_tag_position.try_into()?))?;
+        file.seek(std::io::SeekFrom::Start(self.previous_emitted_tag_postion.try_into()?))?;
         let mut lookback_range = [0u8; 16];
         file.seek_relative(-16)?;
         file.read_exact(&mut lookback_range)?;
@@ -273,7 +273,7 @@ where
             let slice = &lookback_range[i..];
             if slice[0] == 0xa0 && read_vint(&slice[1..]).is_ok_and(|opt| opt.is_some()) {
                 let trace_back_offset = 16 - i;
-                self.last_tag_position -= trace_back_offset;
+                self.previous_emitted_tag_postion -= trace_back_offset;
                 found = true;
                 break;
             }
@@ -315,6 +315,25 @@ where
             _ => Ok(false),
         }
     }
+
+    fn new_inner(&mut self, reader: R) {
+        let mut inner = WebmIterator::new(reader, &[MatroskaSpec::BlockGroup(Master::Start)]);
+        // Disable automatic Master::End or Master::Start tag emission at EOF.
+        //
+        // Scenario 1 - EOF within a Cluster:
+        // - When we hit EOF between Cluster(Master::Start) and expected Cluster(Master::End)
+        // - By default, iterator emits Cluster(Master::End) automatically
+        // - This causes last_emitted_tag_offset() to jump back to the Cluster(Master::Start) position
+        // - Our position tracking becomes incorrect as it's smaller than BlockGroup/SimpleBlock we read
+        //
+        // Scenario 2 - EOF when reading from middle of a Cluster:
+        // - When we start reading from middle of a Cluster (after rollback/seek)
+        // - At EOF, iterator assumes we need a matching Start tag.
+        // - It emits a Cluster(Master::Start) with offset 0
+        // - This resets last_emitted_tag_offset() to 0, breaking our position tracking
+        inner.emit_master_end_when_eof(false);
+        self.inner = Some(inner);
+    }
 }
 
 pub(crate) fn read_vint(buffer: &[u8]) -> anyhow::Result<Option<(u64, usize)>> {

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

@@ -56,7 +56,7 @@ pub fn webm_stream(
         }
     }
 
-    let cut_block_position = webm_itr.last_tag_position();
+    let cut_block_position = webm_itr.previous_emitted_tag_postion();
 
     let ws_frame = Framed::new(output_stream, ProtocolCodeC);
 
@@ -82,8 +82,8 @@ pub fn webm_stream(
         header_writer.write(header)?;
     }
 
-    let mut encode_writer = header_writer.into_encoded_writer(encode_writer_config)?;
-
+    let (mut encode_writer, cut_block_hit_marker) = header_writer.into_encoded_writer(encode_writer_config)?;
+    let mut cut_block_hit_marker = Some(cut_block_hit_marker);
     // Start muxing from the last key frame.
     // The WebM project requires the muxer to ensure the first Block/SimpleBlock is a keyframe.
     // However, the WebM file emitted by the CaptureStream API in Chrome does not adhere to this requirement.
@@ -118,8 +118,13 @@ pub fn webm_stream(
                 }
             }
             Some(Ok(tag)) => {
-                if webm_itr.last_tag_position() == cut_block_position {
-                    encode_writer.mark_cut_block_hit();
+                if webm_itr.previous_emitted_tag_postion() == cut_block_position {
+                    if let Some(cut_block_hit_marker) = cut_block_hit_marker.take() {
+                        encode_writer.mark_cut_block_hit(cut_block_hit_marker);
+                    } else {
+                        error_sender.blocking_send(UserFriendlyError::UnexpectedError)?;
+                        anyhow::bail!("cut block hit twice");
+                    }
                 }
 
                 match encode_writer.write(tag) {

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

@@ -47,8 +47,11 @@ where
         Ok(())
     }
 
-    pub(crate) fn into_encoded_writer(self, config: EncodeWriterConfig) -> anyhow::Result<CutCusterWriter<T>> {
-        let encoded_writer = CutCusterWriter::new(config, self)?;
+    pub(crate) fn into_encoded_writer(
+        self,
+        config: EncodeWriterConfig,
+    ) -> anyhow::Result<(CutClusterWriter<T>, CutBlockHitMarker)> {
+        let encoded_writer = CutClusterWriter::new(config, self)?;
         Ok(encoded_writer)
     }
 }
@@ -82,7 +85,7 @@ enum CutBlockState {
     },
 }
 
-pub(crate) struct CutCusterWriter<T>
+pub(crate) struct CutClusterWriter<T>
 where
     T: std::io::Write,
 {
@@ -94,11 +97,14 @@ where
     cut_block_state: CutBlockState,
 }
 
-impl<T> CutCusterWriter<T>
+/// A token type that enforces the one-time transition of cut block state.
+pub(crate) struct CutBlockHitMarker;
+
+impl<T> CutClusterWriter<T>
 where
     T: std::io::Write,
 {
-    fn new(config: EncodeWriterConfig, writer: HeaderWriter<T>) -> anyhow::Result<Self> {
+    fn new(config: EncodeWriterConfig, writer: HeaderWriter<T>) -> anyhow::Result<(Self, CutBlockHitMarker)> {
         let decoder = VpxDecoder::builder()
             .threads(config.threads)
             .width(config.width)
@@ -117,21 +123,24 @@ where
             .build()?;
 
         let HeaderWriter { writer } = writer;
-        Ok(Self {
-            writer,
-            cluster_timestamp: None,
-            encoder,
-            decoder,
-            cut_block_state: CutBlockState::HaventMet,
-        })
+        Ok((
+            Self {
+                writer,
+                cluster_timestamp: None,
+                encoder,
+                decoder,
+                cut_block_state: CutBlockState::HaventMet,
+            },
+            CutBlockHitMarker,
+        ))
     }
 }
 
 pub(crate) enum WriterResult {
     Continue,
 }
 
-impl<T> CutCusterWriter<T>
+impl<T> CutClusterWriter<T>
 where
     T: std::io::Write,
 {
@@ -296,7 +305,7 @@ where
         false
     }
 
-    pub(crate) fn mark_cut_block_hit(&mut self) {
+    pub(crate) fn mark_cut_block_hit(&mut self, _marker: CutBlockHitMarker) {
         self.cut_block_state = CutBlockState::AtCutBlock;
     }
 }