Forbid output update after EOS (#833)

compositor_pipeline/src/pipeline.rs

@@ -378,6 +378,15 @@ impl Pipeline {
             .outputs
             .get(&output_id)
             .ok_or_else(|| UpdateSceneError::OutputNotRegistered(output_id.clone()))?;
+
+        if let Some(cond) = &output.video_end_condition {
+            if cond.did_output_end() {
+                // Ignore updates after EOS
+                warn!("Received output update on a finished output");
+                return Ok(());
+            }
+        }
+
         let (Some(resolution), Some(frame_format)) = (
             output.output.resolution(),
             output.output.output_frame_format(),
@@ -396,6 +405,19 @@ impl Pipeline {
         output_id: &OutputId,
         audio: AudioMixingParams,
     ) -> Result<(), UpdateSceneError> {
+        let output = self
+            .outputs
+            .get(output_id)
+            .ok_or_else(|| UpdateSceneError::OutputNotRegistered(output_id.clone()))?;
+
+        if let Some(cond) = &output.audio_end_condition {
+            if cond.did_output_end() {
+                // Ignore updates after EOS
+                warn!("Received output update on a finished output");
+                return Ok(());
+            }
+        }
+
         info!(?output_id, "Update audio mixer {:#?}", audio);
         self.audio_mixer.update_output(output_id, audio)
     }

compositor_pipeline/src/pipeline/encoder/opus.rs

@@ -38,7 +38,7 @@ impl OpusEncoder {
         std::thread::Builder::new()
             .name("Opus encoder thread".to_string())
             .spawn(move || {
-                let _span = span!(Level::INFO, "Opus encoder thread",).entered();
+                let _span = span!(Level::INFO, "Opus encoder thread").entered();
                 run_encoder_thread(encoder, samples_batch_receiver, packets_sender)
             })
             .unwrap();

compositor_pipeline/src/pipeline/pipeline_output.rs

@@ -234,6 +234,10 @@ impl PipelineOutputEndConditionState {
         EosStatus::None
     }
 
+    pub(super) fn did_output_end(&self) -> bool {
+        self.did_end
+    }
+
     pub(super) fn on_input_registered(&mut self, input_id: &InputId) {
         self.on_event(StateChange::AddInput(input_id))
     }

compositor_pipeline/src/queue/queue_thread.rs

@@ -236,7 +236,9 @@ impl VideoQueueProcessor {
         let pts = frames_batch.pts;
         debug!(?pts, "Pushing video frames.");
         if is_required {
-            self.sender.send(frames_batch).unwrap()
+            if self.sender.send(frames_batch).is_err() {
+                warn!(?pts, "Dropping video frame on queue output.");
+            }
         } else {
             let send_deadline = self.queue_start_time.add(frames_batch.pts);
             if self
@@ -341,7 +343,9 @@ impl AudioQueueProcessor {
         let pts_range = (samples.start_pts, samples.end_pts);
         debug!(?pts_range, "Pushing audio samples.");
         if is_required {
-            self.sender.send(samples).unwrap()
+            if self.sender.send(samples).is_err() {
+                warn!(?pts_range, "Dropping audio batch on queue output.");
+            }
         } else if self.sender.try_send(samples).is_err() {
             warn!(?pts_range, "Dropping audio batch on queue output.")
         }

integration_tests/src/tests/offline_processing.rs

@@ -123,7 +123,7 @@ pub fn offline_processing() -> Result<()> {
     if !(1.9..=2.1).contains(&duration) {
         return Err(anyhow!("Invalid duration: {}", duration));
     }
-    if !(930_000..=1_000_000).contains(&bit_rate) {
+    if !(930_000..=1_008_200).contains(&bit_rate) {
         return Err(anyhow!("Invalid bit rate: {}", bit_rate));
     }