multi-channel playout: concurrent tick, status endpoint, ffmpeg restart (#11)

Cargo.lock

@@ -1777,6 +1777,7 @@ dependencies = [
  "chrono",
  "domain",
  "dotenvy",
+ "futures",
  "serde",
  "serde_json",
  "thiserror",

crates/playout/Cargo.toml

@@ -27,6 +27,7 @@ thiserror = { workspace = true }
 anyhow = "1"
 dotenvy = "0.15"
 bytes = "1"
+futures = "0.3"
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }

crates/playout/src/engine.rs

@@ -7,6 +7,7 @@ use domain::models::{CurrentBroadcast, GeneratedSchedule};
 use domain::ports::ScheduleQuery;
 use domain::value_objects::{ChannelId, SourceUri};
 use domain::{ScheduleEngineService, SlotId};
+use serde::Serialize;
 use tokio::sync::RwLock;
 use tracing::{error, info, warn};
 
@@ -24,6 +25,14 @@ struct ChannelState {
     schedule_id: domain::value_objects::ScheduleId,
 }
 
+#[derive(Debug, Clone, Serialize)]
+pub struct ChannelStatus {
+    pub channel_id: String,
+    pub current_slot_id: String,
+    pub schedule_id: String,
+    pub ffmpeg_running: bool,
+}
+
 pub struct PlayoutEngine {
     config: PlayoutConfig,
     schedule_query: Arc<dyn ScheduleQuery>,
@@ -49,6 +58,13 @@ impl PlayoutEngine {
     }
 
     pub async fn add_channel(&self, channel_id: ChannelId) -> anyhow::Result<()> {
+        {
+            let channels = self.channels.read().await;
+            if channels.contains_key(&channel_id) {
+                return Err(anyhow::anyhow!("channel {channel_id} already active"));
+            }
+        }
+
         let now = Utc::now();
         let schedule = self
             .schedule_query
@@ -84,13 +100,32 @@ impl PlayoutEngine {
         self.channels.read().await.keys().copied().collect()
     }
 
+    pub async fn channel_statuses(&self) -> Vec<ChannelStatus> {
+        let mut channels = self.channels.write().await;
+        channels
+            .iter_mut()
+            .map(|(id, state)| ChannelStatus {
+                channel_id: id.to_string(),
+                current_slot_id: state.current_slot_id.to_string(),
+                schedule_id: state.schedule_id.to_string(),
+                ffmpeg_running: state.ffmpeg.is_running(),
+            })
+            .collect()
+    }
+
     pub async fn tick(&self) {
         let now = Utc::now();
         let channel_ids: Vec<ChannelId> = self.channels.read().await.keys().copied().collect();
 
-        for channel_id in channel_ids {
+        let futures: Vec<_> = channel_ids
-            if let Err(e) = self.tick_channel(channel_id, now).await {
+            .into_iter()
-                warn!(%channel_id, %e, "tick failed");
+            .map(|id| self.tick_channel(id, now))
+            .collect();
+
+        let results = futures::future::join_all(futures).await;
+        for result in results {
+            if let Err(e) = result {
+                warn!(%e, "channel tick failed");
             }
         }
     }
@@ -125,8 +160,21 @@ impl PlayoutEngine {
                 .unwrap_or(false)
         };
 
+        let ffmpeg_dead = {
+            let mut channels = self.channels.write().await;
+            channels
+                .get_mut(&channel_id)
+                .map(|s| !s.ffmpeg.is_running())
+                .unwrap_or(false)
+        };
+
+        if needs_transition || ffmpeg_dead {
             if needs_transition {
                 info!(%channel_id, %current_slot_id, "slot transition");
+            } else {
+                warn!(%channel_id, "ffmpeg process died, restarting");
+            }
+
             let mut channels = self.channels.write().await;
             if let Some(state) = channels.get_mut(&channel_id) {
                 state.ffmpeg.stop().await;
@@ -176,11 +224,7 @@ impl PlayoutEngine {
             segment_prefix: "seg".into(),
         };
 
-        let handle = FfmpegHandle::spawn(
+        let handle = FfmpegHandle::spawn(config, self.store.clone(), channel_id.to_string());
-            config,
-            self.store.clone(),
-            channel_id.to_string(),
-        );
 
         Ok(handle)
     }

crates/playout/src/http.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use axum::extract::{Path, State};
 use axum::http::{HeaderValue, StatusCode, header};
 use axum::response::{IntoResponse, Response};
-use axum::routing::{get, post, delete};
+use axum::routing::{delete, get, post};
 use axum::Router;
 use domain::value_objects::ChannelId;
 
@@ -25,11 +25,9 @@ pub fn router(state: AppState) -> Router {
             "/playout/{channel_id}/playlist.m3u8",
             get(get_playlist),
         )
-        .route(
+        .route("/playout/{channel_id}/{segment}", get(get_segment))
-            "/playout/{channel_id}/{segment}",
-            get(get_segment),
-        )
         .route("/playout/channels", get(list_channels))
+        .route("/playout/channels/status", get(channel_statuses))
         .route("/playout/channels/{channel_id}", post(add_channel))
         .route(
             "/playout/channels/{channel_id}",
@@ -85,6 +83,11 @@ async fn list_channels(State(state): State<AppState>) -> Response {
     axum::Json(ids).into_response()
 }
 
+async fn channel_statuses(State(state): State<AppState>) -> Response {
+    let statuses = state.engine.channel_statuses().await;
+    axum::Json(statuses).into_response()
+}
+
 async fn add_channel(
     State(state): State<AppState>,
     Path(channel_id): Path<String>,