fix(dgw): fix streamer failure for RDM

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

@@ -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.
@@ -119,7 +119,9 @@ pub fn webm_stream(
             }
             Some(Ok(tag)) => {
                 if webm_itr.last_tag_position() == cut_block_position {
-                    encode_writer.mark_cut_block_hit();
+                    if let Some(cut_block_hit_marker) = cut_block_hit_marker.take() {
+                        encode_writer.mark_cut_block_hit(cut_block_hit_marker);
+                    }
                 }
 
                 match encode_writer.write(tag) {

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

@@ -47,7 +47,10 @@ where
         Ok(())
     }
 
-    pub(crate) fn into_encoded_writer(self, config: EncodeWriterConfig) -> anyhow::Result<CutCusterWriter<T>> {
+    pub(crate) fn into_encoded_writer(
+        self,
+        config: EncodeWriterConfig,
+    ) -> anyhow::Result<(CutCusterWriter<T>, CutBlockHitMarker)> {
         let encoded_writer = CutCusterWriter::new(config, self)?;
         Ok(encoded_writer)
     }
@@ -94,11 +97,13 @@ where
     cut_block_state: CutBlockState,
 }
 
+pub(crate) struct CutBlockHitMarker;
+
 impl<T> CutCusterWriter<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,13 +122,16 @@ 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,
+        ))
     }
 }
 
@@ -203,7 +211,9 @@ where
                 let current_block_absolute_time = current_video_block.absolute_timestamp()?;
                 let cluster_relative_timestamp = current_block_absolute_time - cut_block_absolute_time;
                 if self.should_write_new_cluster(current_block_absolute_time) {
-                    self.start_new_cluster(cluster_relative_timestamp)?;
+                    self.start_new_cluster(cluster_relative_timestamp).inspect_err(|e| {
+                        error!("failed to start new cluster: {}", e);
+                    })?;
 
                     self.cut_block_state = CutBlockState::Met {
                         cut_block_absolute_time,
@@ -296,7 +306,8 @@ where
         false
     }
 
-    pub(crate) fn mark_cut_block_hit(&mut self) {
+    pub(crate) fn mark_cut_block_hit(&mut self, _marker: CutBlockHitMarker) {
+        debug!("marking cut block hit");
         self.cut_block_state = CutBlockState::AtCutBlock;
     }
 }