mirror of
https://github.com/euzu/tuliprox.git
synced 2026-10-20 06:42:20 +02:00
Fix/scheduled updates logger initialization zombie connections (#664)
* fix: Scheduler preserve forced_targets across config reloads. hot reload could cause loosing scheduler task targets. warn logs for: - process_source: warns when all inputs are disabled for given targets - process_sources: warns when no sources were processed at all * Fix: Logger initialization * Fix: TCP USER TIMEOUT for linux. If a client disrupts the connection, the tcp layer will hold it approx for 5-15 minutes. With this setting we can close in 30 seconds under linux
This commit is contained in:
@@ -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<Arc<str>> = 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::<Vec<&str>>().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<Sou
|
||||
|
||||
let errors = Arc::new(Mutex::<Vec<TuliproxError>>::new(vec![]));
|
||||
let stats = Arc::new(Mutex::<Vec<SourceStats>>::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<Sou
|
||||
let shared_stats = stats.clone();
|
||||
let ctx = processing_ctx.clone();
|
||||
|
||||
processed_any = true;
|
||||
if process_parallel {
|
||||
async_tasks.spawn(async move {
|
||||
// Hold the per-source lock for the full duration of this update.
|
||||
@@ -853,6 +864,14 @@ async fn process_sources(processing_ctx: &PlaylistProcessingContext) -> (Vec<Sou
|
||||
drop(update_lock);
|
||||
}
|
||||
}
|
||||
if !processed_any {
|
||||
warn!(
|
||||
"No sources were processed for the given targets. Check that:\n\
|
||||
- Sources have enabled targets matching your target selection\n\
|
||||
- CLI -t filter or schedule.targets are correct\n\
|
||||
- No playlist lock is blocking updates"
|
||||
);
|
||||
}
|
||||
while let Some(result) = async_tasks.join_next().await {
|
||||
if let Err(err) = result {
|
||||
error!("Playlist processing task failed: {err:?}");
|
||||
|
||||
Reference in New Issue
Block a user