fix: ci

video/transcoder/src/tests/transcoder/mod.rs

@@ -2,6 +2,7 @@
 #![allow(unused_imports)]
 #![allow(dead_code)]
 
+use core::panic;
 use std::{
     collections::HashMap, io::Cursor, net::SocketAddr, path::PathBuf, pin::Pin, sync::Arc,
     time::Duration,
@@ -391,17 +392,18 @@ async fn test_transcode() {
 
     tracing::debug!("finished sending frames");
 
-    tokio::time::sleep(Duration::from_millis(250)).await;
-
     watch_stream_req
         .send(Ok(WatchStreamResponse {
             data: Some(watch_stream_response::Data::ShuttingDown(true)),
         }))
         .await
         .unwrap();
 
-    let redis = global.redis.clone();
     drop(watch_stream_req);
+
+    tokio::time::sleep(Duration::from_millis(250)).await;
+
+    let redis = global.redis.clone();
     drop(global);
     handler
         .cancel()

video/transcoder/src/transcoder/job/mod.rs

@@ -275,6 +275,7 @@ fn report_to_ingest(
 ) -> impl Stream<Item = Result<()>> + Send + 'static {
     stream!({
         while let Some(msg) = channel.recv().await {
+            tracing::debug!("sending message: {:?}", msg);
             match client
                 .transcoder_event(msg)
                 .timeout(Duration::from_secs(5))
@@ -708,8 +709,12 @@ impl Job {
 
         drop(report);
 
+        tracing::debug!("waiting for report to ingest to exit");
+
         // Finish all the report futures
-        while report_fut.next().await.is_some() {}
+        while report_fut.next().await.is_some() {
+            tracing::debug!("report to ingest exited");
+        }
 
         tracing::info!("waiting for playlist update to exit");
 
@@ -804,6 +809,8 @@ impl Job {
             self.report_error("ffmpeg exited with non-zero status", false)
                 .await;
         }
+
+        tracing::debug!("ffmpeg exited");
     }
 
     async fn handle_msg(

video/transcoder/src/transcoder/mod.rs

@@ -40,6 +40,7 @@ pub async fn run(global: Arc<GlobalState>) -> Result<()> {
                 tokio::spawn(handle_message(global.clone(), m, child_token.clone()));
             },
             _ = global.ctx.done() => {
+                tracing::debug!("context done");
                 return Ok(());
             }
         }