From 84dd05a8a6f9497690d8abaa11db4973302b95be Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 12 Jul 2026 13:59:54 +0200 Subject: [PATCH] multi-channel playout: concurrent tick, status endpoint, ffmpeg restart (#11) --- Cargo.lock | 1 + crates/playout/Cargo.toml | 1 + crates/playout/src/engine.rs | 64 ++++++++++++++++++++++++++++++------ crates/playout/src/http.rs | 13 +++++--- 4 files changed, 64 insertions(+), 15 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index df1a1c1..0b60a19 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1777,6 +1777,7 @@ dependencies = [ "chrono", "domain", "dotenvy", + "futures", "serde", "serde_json", "thiserror", diff --git a/crates/playout/Cargo.toml b/crates/playout/Cargo.toml index 0c901c0..b9ca7c2 100644 --- a/crates/playout/Cargo.toml +++ b/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"] } diff --git a/crates/playout/src/engine.rs b/crates/playout/src/engine.rs index 15aa244..1f59971 100644 --- a/crates/playout/src/engine.rs +++ b/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, @@ -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 { + 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 = self.channels.read().await.keys().copied().collect(); - for channel_id in channel_ids { - if let Err(e) = self.tick_channel(channel_id, now).await { - warn!(%channel_id, %e, "tick failed"); + let futures: Vec<_> = channel_ids + .into_iter() + .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) }; - if needs_transition { - info!(%channel_id, %current_slot_id, "slot transition"); + 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( - config, - self.store.clone(), - channel_id.to_string(), - ); + let handle = FfmpegHandle::spawn(config, self.store.clone(), channel_id.to_string()); Ok(handle) } diff --git a/crates/playout/src/http.rs b/crates/playout/src/http.rs index 61c3c62..91f9301 100644 --- a/crates/playout/src/http.rs +++ b/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( - "/playout/{channel_id}/{segment}", - get(get_segment), - ) + .route("/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) -> Response { axum::Json(ids).into_response() } +async fn channel_statuses(State(state): State) -> Response { + let statuses = state.engine.channel_statuses().await; + axum::Json(statuses).into_response() +} + async fn add_channel( State(state): State, Path(channel_id): Path,