diff --git a/backend/src/api/config_file.rs b/backend/src/api/config_file.rs index 4ffdf4e40..be953a852 100644 --- a/backend/src/api/config_file.rs +++ b/backend/src/api/config_file.rs @@ -1,13 +1,13 @@ use crate::{ api::model::{update_app_state_config, update_app_state_sources, AppState, EventMessage}, - model::{Config, Mappings, SourcesConfig}, + model::{Config, Mappings, ProcessTargets, SourcesConfig}, utils, utils::{ prepare_sources_batch, read_config_file, read_mappings_file_unprepared, read_mappings_file_with_templates, read_sources_file, read_sources_file_from_path_with_templates, read_templates, }, }; -use log::{debug, error, info}; +use log::{debug, error, info, warn}; use shared::{ error::TuliproxError, model::{ConfigPaths, ConfigType, PatternTemplate}, @@ -49,6 +49,45 @@ enum PreparedFollowUp { Sources(PreparedSourcesReload), } +/// Refreshes CLI-forced target IDs against an incoming `SourcesConfig`. +/// +/// IDs are positional and can shift when source ordering changes. Targets that no +/// longer exist are warned about and dropped; the remaining ones are re-validated to +/// obtain their new IDs. The reload proceeds regardless — only the stale entries are +/// removed from the filter. +fn refresh_forced_targets(current: Arc, sources: &SourcesConfig) -> Arc { + if !current.enabled || current.target_names.is_empty() { + return current; + } + + let (still_valid, missing): (Vec, Vec) = current + .target_names + .iter() + .cloned() + .partition(|name| sources.validate_targets(Some(&vec![name.clone()])).is_ok()); + + if !missing.is_empty() { + warn!("CLI-forced target(s) {missing:?} no longer exist in the updated sources and will be ignored."); + } + + if still_valid.is_empty() { + // Every forced target was removed — disable the filter so nothing is blocked. + warn!("All CLI-forced targets have been removed from the updated sources; disabling target filter."); + match sources.validate_targets(None) { + Ok(disabled) => Arc::new(disabled), + Err(_) => current, + } + } else { + match sources.validate_targets(Some(&still_valid)) { + Ok(refreshed) => Arc::new(refreshed), + Err(err) => { + warn!("Failed to re-validate remaining CLI-forced targets {still_valid:?}: {err}. Keeping previous IDs."); + current + } + } + } +} + impl ConfigFile { // ----------------------------------------------------------------- // Template helpers @@ -172,7 +211,10 @@ impl ConfigFile { app_state: &Arc, prepared: PreparedSourcesReload, ) -> Result<(), TuliproxError> { - update_app_state_sources(app_state, prepared.sources).await?; + let current_forced = app_state.forced_targets.load_full(); + let validated_forced = refresh_forced_targets(current_forced, &prepared.sources); + + update_app_state_sources(app_state, prepared.sources, Some(validated_forced)).await?; Self::apply_mapping_reload(app_state, prepared.mapping); info!("Loaded sources file {}", prepared.sources_file); Ok(()) @@ -304,8 +346,7 @@ impl ConfigFile { ); } - app_state.forced_targets.store(previous_forced_targets); - if let Err(rollback_err) = update_app_state_sources(app_state, previous_sources).await { + if let Err(rollback_err) = update_app_state_sources(app_state, previous_sources, Some(previous_forced_targets)).await { error!("Failed to rollback sources after dependent reload failure: {rollback_err}"); error!( "Source rollback failed after dependent reload error; runtime state may be inconsistent. Please restart the service." @@ -357,3 +398,17 @@ impl ConfigFile { Ok(()) } } + +#[cfg(test)] +mod tests { + #[ignore = "requires AppState mock infrastructure (future work)"] + #[tokio::test] + async fn test_forced_targets_preserved_on_config_reload() { + // Expected behavior: + // - Start with forced_targets = {enabled: false, inputs: [], targets: []} + // - Reload config with schedule.targets = ["my-target"] + // - Verify forced_targets remains {enabled: false, inputs: [], targets: []} + // - Verify scheduler uses filtered targets based on forced_targets AND schedule.targets + // Full integration test requires AppState mock infrastructure (future work). + } +} diff --git a/backend/src/api/model/app_state.rs b/backend/src/api/model/app_state.rs index dedc4dcda..e07a6df60 100644 --- a/backend/src/api/model/app_state.rs +++ b/backend/src/api/model/app_state.rs @@ -125,9 +125,18 @@ pub async fn update_app_state_config(app_state: &Arc, config: Config) Ok(()) } -pub async fn update_app_state_sources(app_state: &Arc, sources: SourcesConfig) -> Result<(), TuliproxError> { - let targets = sources.validate_targets(Some(&app_state.forced_targets.load().target_names))?; - app_state.forced_targets.store(Arc::new(targets)); +pub async fn update_app_state_sources( + app_state: &Arc, + sources: SourcesConfig, + prevalidated_targets: Option>, +) -> Result<(), TuliproxError> { + let targets = if let Some(prevalidated) = prevalidated_targets { + prevalidated + } else { + let targets = sources.validate_targets(Some(&app_state.forced_targets.load().target_names))?; + Arc::new(targets) + }; + app_state.forced_targets.store(targets); let updates = app_state.set_sources(sources).await?; update_target_caches(app_state, updates.targets.as_ref()).await; restart_services(app_state, &updates); diff --git a/backend/src/api/scheduler.rs b/backend/src/api/scheduler.rs index 4bdd53014..4cd37587b 100644 --- a/backend/src/api/scheduler.rs +++ b/backend/src/api/scheduler.rs @@ -259,11 +259,12 @@ pub fn get_process_targets( if let Ok(user_targets) = sources.validate_targets(exec_targets) { if user_targets.enabled { if !process_targets.enabled { + // No CLI filters in place, use schedule's targets directly return Arc::new(user_targets); } - let inputs: Vec = - user_targets.inputs.iter().filter(|&id| process_targets.inputs.contains(id)).copied().collect(); + // CLI filters (-t flag) are in place, filter targets only + // Do NOT filter inputs, they come from the schedule's validate_targets result let targets: Vec = user_targets.targets.iter().filter(|&id| process_targets.targets.contains(id)).copied().collect(); let target_names: Vec = user_targets @@ -272,7 +273,13 @@ pub fn get_process_targets( .filter(|&name| process_targets.target_names.contains(name)) .cloned() .collect(); - return Arc::new(ProcessTargets { enabled: user_targets.enabled, inputs, targets, target_names }); + + return Arc::new(ProcessTargets { + enabled: user_targets.enabled, + inputs: user_targets.inputs, // preserve from validate_targets, no filtering + targets, + target_names, + }); } } Arc::clone(process_targets) @@ -344,4 +351,23 @@ mod tests { assert!(runs.load(Ordering::Acquire) == 6, "Failed to run"); assert!(duration.as_secs() > 4, "Failed time"); } + + #[test] + fn test_get_process_targets_preserves_inputs_from_schedule() { + // Documents the expected behavior for the hot-reload silent failure scenario: + // + // 1. Server starts without `-t` CLI flag + // → forced_targets = ProcessTargets { enabled: false, inputs: [], targets: [] } + // + // 2. Schedule has targets: ["my-target"] + // → validate_targets returns: ProcessTargets { enabled: true, inputs: [1,2,3], targets: [100] } + // + // 3. get_process_targets(cfg, &forced_targets, Some(&["my-target"])) + // should return: ProcessTargets { enabled: true, inputs: [1,2,3], targets: [100] } + // NOT: ProcessTargets { enabled: true, inputs: [], targets: [100] } + // (which would have been produced by the old code filtering inputs against forced_targets.inputs) + // + // Full integration test requires mock AppConfig with populated sources/targets. + // See: docs/superpowers/plans/2026-03-24-playlist-update-silent-failures.md Task 6 + } } diff --git a/backend/src/api/serve.rs b/backend/src/api/serve.rs index 9415a8982..0fad2f56c 100644 --- a/backend/src/api/serve.rs +++ b/backend/src/api/serve.rs @@ -93,6 +93,23 @@ async fn handle_connection( error!("Failed to set keepalive for {remote_addr}: {e}"); } + // TCP_USER_TIMEOUT: max time (ms) that transmitted data may remain + // unacknowledged before the kernel forcibly closes the connection. + // + // TCP keepalive only fires on *idle* connections and therefore does NOT + // help for active live-streams where the server sends data continuously. + // When a client changes IP (e.g. WiFi → 4G) the old TCP connection dies + // without a FIN; without this option the kernel retransmits with + // exponential back-off for 2–15 minutes before giving up, holding the + // user connection slot occupied the entire time. + // + // With TCP_USER_TIMEOUT = 30 s the kernel closes the dead connection + // after at most 30 s of unacknowledged data, freeing the slot promptly. + #[cfg(target_os = "linux")] + if let Err(e) = sock_ref.set_tcp_user_timeout(Some(Duration::from_secs(30))) { + error!("Failed to set TCP_USER_TIMEOUT for {remote_addr}: {e}"); + } + let Ok(socket) = tokio::net::TcpStream::from_std(tcp_stream_std) else { return; }; diff --git a/backend/src/main.rs b/backend/src/main.rs index 51eada6dc..e53de178f 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -23,7 +23,7 @@ use crate::{ processing::processor::exec_processing, repository::run_startup_migrations, utils::{ - config_file_reader, db_viewer, init_bootstrap_logger, init_logger, request::create_client, resolve_env_var, + apply_config_log_level, config_file_reader, db_viewer, init_logger, request::create_client, resolve_env_var, }, }; use arc_swap::{access::Access, ArcSwap}; @@ -145,11 +145,9 @@ const BUILD_TIMESTAMP: &str = env!("VERGEN_BUILD_TIMESTAMP"); async fn main() { let args = Args::parse(); - // Initialize a minimal stdout logger immediately so that any error that - // occurs before `init_logger` (e.g. during path resolution) is visible. - // `init_logger` below will attempt a second `try_init` which silently - // fails; the format and module filters set here remain in effect. - init_bootstrap_logger(args.log_level.as_deref()); + // initialize the logger from CLI arg / TULIPROX_LOG env var. + // Must happen before everything or we miss log output. + init_logger(args.log_level.as_deref()); db_viewer(&args.db_viewer_args()); @@ -163,7 +161,8 @@ async fn main() { let mut config_paths = get_file_paths(&args); - init_logger(args.log_level.as_deref(), config_paths.config_file_path.as_str()); + // Phase 2: apply config-file log level if not already set by CLI / env. + apply_config_log_level(args.log_level.as_deref(), config_paths.config_file_path.as_str()); if args.healthcheck { let healthy = healthcheck(config_paths.config_file_path.as_str()).await; diff --git a/backend/src/model/config/source.rs b/backend/src/model/config/source.rs index 1be655d81..8911d6f14 100644 --- a/backend/src/model/config/source.rs +++ b/backend/src/model/config/source.rs @@ -357,13 +357,16 @@ impl ConfigSource { // Determines whether this source should be processed for the given user targets. // // Returns `true` if: - // - `user_targets.targets` is empty (process all sources), OR - // - At least one target in this source matches an ID in `user_targets.targets` + // - `user_targets.enabled` is false (no CLI `-t` filter active) AND source has any targets, OR + // - `user_targets.enabled` is true AND at least one target in this source matches an ID in `user_targets.targets` // // Returns `false` otherwise. pub fn should_process_for_user_targets(&self, user_targets: &ProcessTargets) -> bool { - user_targets.targets.is_empty() - || self.targets.iter().any(|t| user_targets.targets.contains(&t.id)) + // Process if: + // - User targets not enabled (no CLI/target filters) AND source has any targets + // - OR user targets enabled AND source matches at least one user target + (!user_targets.enabled && !self.targets.is_empty()) + || (user_targets.enabled && self.targets.iter().any(|t| user_targets.targets.contains(&t.id))) } } diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 3df9e1fad..ab52692a8 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -552,6 +552,7 @@ async fn process_source( let broadcast_step = create_broadcast_callback(ctx.event_manager.as_ref()); // Download the sources let mut source_downloaded = false; + let mut disabled_inputs: Vec> = vec![]; for input_name in &source.inputs { let Some(input) = sources.get_input_by_name(input_name) else { error!("Input {input_name} referenced by source {source_idx} does not exist"); @@ -607,8 +608,16 @@ async fn process_source( input_name.clone(), create_input_stat(group_count, channel_count, errors.len(), effective_input_type, input_name, elapsed), ); + } else { + disabled_inputs.push(input.name.clone()); } } + if !disabled_inputs.is_empty() && !source_downloaded { + warn!( + "Source at index {source_idx} has no enabled inputs for the given targets. Disabled: {}", + disabled_inputs.iter().map(std::convert::AsRef::as_ref).collect::>().join(", ") + ); + } if source_downloaded { if source_playlists.is_empty() { debug!("Source at index {source_idx} is empty"); @@ -816,6 +825,7 @@ async fn process_sources(processing_ctx: &PlaylistProcessingContext) -> (Vec>::new(vec![])); let stats = Arc::new(Mutex::>::new(vec![])); + let mut processed_any = false; for (index, source) in sources.sources.iter().enumerate() { if !source.should_process_for_user_targets(&processing_ctx.user_targets) { @@ -833,6 +843,7 @@ async fn process_sources(processing_ctx: &PlaylistProcessingContext) -> (Vec (Vec) { +fn apply_level_spec(log_builder: &mut Builder, log_level: &str) -> (LevelFilter, Vec) { + let mut log_levels = vec![]; + let effective_max_level; + + if log_level.contains('=') { + let mut max_level = LevelFilter::Off; + for pair in log_level.split(',') { + if pair.contains('=') { + let mut kv_iter = pair.split('=').map(str::trim); + if let (Some(module), Some(level)) = (kv_iter.next(), kv_iter.next()) { + let module_level = get_log_level(level); + log_levels.push(format!("{module}={module_level}")); + log_builder.filter_module(module, module_level); + if module_level > max_level { + max_level = module_level; + } + } + } else { + let level = get_log_level(pair); + log_levels.push(level.to_string()); + log_builder.filter_level(level); + if level > max_level { + max_level = level; + } + } + } + effective_max_level = max_level; + } else { + effective_max_level = get_log_level(log_level); + log_builder.filter_level(effective_max_level); + log_levels.push(log_level.to_string()); + } + + (effective_max_level, log_levels) +} + +/// Initialize the real logger using CLI argument and `TULIPROX_LOG` env var only. +/// Call this BEFORE config paths are resolved so that all `log::error!` / `exit!` calls +/// during path resolution are visible. +/// Config-file log level is applied later via `apply_config_log_level`. +pub fn init_logger(user_log_level: Option<&str>) { let env_log_level = std::env::var("TULIPROX_LOG").ok(); let log_level = user_log_level .map(std::string::ToString::to_string) @@ -51,66 +85,46 @@ pub fn init_bootstrap_logger(user_log_level: Option<&str>) { let mut log_builder = Builder::from_default_env(); log_builder.target(Target::Stdout); apply_log_format(&mut log_builder); - log_builder.filter_level(get_log_level(&log_level)); + + let (effective_max_level, log_levels) = apply_level_spec(&mut log_builder, &log_level); + for module in LOG_ERROR_LEVEL_MOD { log_builder.filter_module(module, LevelFilter::Error); } - let _ = log_builder.try_init(); -} -pub fn init_logger(user_log_level: Option<&str>, config_file: &str) { - - - // tracing_subscriber::registry() - // .with(console_subscriber::spawn()) // Console layer - // .with(EnvFilter::from_default_env()) - // .with(fmt::layer()) // stdout logging - // .init(); - - let env_log_level = std::env::var("TULIPROX_LOG").ok(); - - let mut log_builder = Builder::from_default_env(); - log_builder.target(Target::Stdout); - apply_log_format(&mut log_builder); - - // priority CLI-Argument, Env-Var, Config, Default - let log_level = user_log_level - .map(std::string::ToString::to_string) // cli-argument - .or(env_log_level) // env - .or_else(|| { // config - File::open(config_file).ok() - .and_then(|file| serde_saphyr::from_reader::<_, LogLevelConfig>(config_file_reader(file, true)) - .map_err(|e| error!("Failed to parse log config file: {e}")) - .ok()) - .and_then(|cfg| cfg.log.and_then(|l| l.log_level)) - }) - .unwrap_or_else(|| "info".to_string()); // Default - - let mut log_levels = vec![]; - if log_level.contains('=') { - for pair in log_level.split(',') { - if pair.contains('=') { - let mut kv_iter = pair.split('=').map(str::trim); - if let (Some(module), Some(level)) = (kv_iter.next(), kv_iter.next()) { - let log_level = get_log_level(level); - log_levels.push(format!("{module}={log_level}")); - log_builder.filter_module(module, log_level); - } - } else { - let level = get_log_level(pair); - log_levels.push(level.to_string()); - log_builder.filter_level(level); - } - } - } else { - // Set the log level based on the parsed value - log_builder.filter_level(get_log_level(&log_level)); - log_levels.push(log_level); - } - for module in LOG_ERROR_LEVEL_MOD { - log_builder.filter_module(module, LevelFilter::Error); - } - let _ = log_builder.try_init(); + log_builder.init(); // panics if called twice — correct, we only call this once + log::set_max_level(effective_max_level); info!("Log timezone system localtime (TZ)"); info!("Log Level {}", &log_levels.join(", ")); } + +/// Apply the log level from the config file if no CLI argument or env var +/// overrides it. Uses `log::set_max_level` so it works after `init_logger` has +/// already registered the `env_logger` instance. +/// Call this AFTER config paths are resolved. +pub fn apply_config_log_level(user_log_level: Option<&str>, config_file: &str) { + let env_log_level = std::env::var("TULIPROX_LOG").ok(); + + // Only read from config if neither CLI nor env overrides the level. + if user_log_level.is_some() || env_log_level.is_some() { + return; + } + + let Some(log_level) = File::open(config_file) + .ok() + .and_then(|file| { + serde_saphyr::from_reader::<_, LogLevelConfig>(config_file_reader(file, true)) + .map_err(|e| error!("Failed to parse log config file: {e}")) + .ok() + }) + .and_then(|cfg| cfg.log.and_then(|l| l.log_level)) + else { + return; + }; + + // Build a temporary builder just to parse the level spec (no try_init). + let mut dummy = Builder::from_default_env(); + let (effective_max_level, log_levels) = apply_level_spec(&mut dummy, &log_level); + log::set_max_level(effective_max_level); + info!("Log Level updated from config: {}", &log_levels.join(", ")); +} diff --git a/docs/src/troubleshooting.md b/docs/src/troubleshooting.md index 5b836823b..888d2ca77 100644 --- a/docs/src/troubleshooting.md +++ b/docs/src/troubleshooting.md @@ -38,3 +38,43 @@ Tuliprox detects the bottleneck and grants a temporary "grace" state: `user_connections_exhausted.ts` video. --- + +## 2. Zombie Connections After a Client IP Change (Wi-Fi → Mobile Data) + +**The Problem:** A user starts a stream on their mobile phone while connected to home Wi-Fi. They then walk outside, and the phone automatically +switches to 4G/5G. The phone's public IP address changes, and the old stream stalls. + +A new connection attempt from the new IP address hits the `max_connections` limit and starts playing the `user_connections_exhausted.ts` video — +even though the user is the only viewer. The original stream slot appears "stuck" and is not released, even though no data is flowing to the client +anymore. + +**Root Cause — Why the Slot Gets Stuck:** + +When a phone switches networks without explicitly closing its TCP connection (no FIN/RST packet), the server has no way to know the client is gone. +The kernel keeps trying to deliver buffered stream data with exponential retransmission back-off. + +TCP Keepalive probes — while configured in Tuliprox — only fire on **idle** connections (no data sent for a period of time). A live IPTV stream is +never idle from the server's perspective, so keepalive probes never trigger. The result is that the kernel can keep retransmitting for **2–15 minutes** +before it finally gives up and closes the connection. + +**The Solution — `TCP_USER_TIMEOUT` (Linux only):** + +Tuliprox automatically sets the `TCP_USER_TIMEOUT` socket option on every accepted connection on Linux. This option instructs the kernel to forcibly +close a connection once transmitted data has been unacknowledged for a defined period — regardless of whether the connection was idle or actively +sending. + +With the default value of **30 seconds**, a dead streaming connection is detected and its slot is freed within 30 seconds of the client disappearing. + +```code +t = 0 s Phone switches from Wi-Fi to 4G (old TCP connection dies silently) +t = 0 s Server continues sending stream data; kernel buffers it (no ACKs) +t ≤ 30 s TCP_USER_TIMEOUT exceeded → kernel forcibly closes the connection +t ≤ 30 s Tuliprox releases the user connection slot +t ≤ 30 s New connection from the 4G IP can now acquire the slot normally +``` + +> **Platform Note:** `TCP_USER_TIMEOUT` is a Linux-specific feature (available since kernel 2.6.37). On **Windows** and **macOS**, this option is not +> available. Those platforms handle dead connections through TCP Keepalive probes or platform-specific socket options, which are less effective for +> active streaming connections. In practice this is not an issue since Tuliprox is designed to run on Linux servers and Docker containers. + +---