diff --git a/CHANGELOG.md b/CHANGELOG.md index da124823e..b4f3bf658 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ # 3.3.0 (2025-11-xx) - Avoid blocking the runtime when warming cache - Normalize FileLockManager paths so aliases share the same lock +- Use async file operations for playlist persistence to avoid blocking async paths # 3.2.0 (2025-11-14) - Added `name` attribute to Staged Input. diff --git a/Cargo.lock b/Cargo.lock index af3a6b4f1..b817c70be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1102,7 +1102,7 @@ dependencies = [ [[package]] name = "frontend" -version = "3.2.1" +version = "3.2.2" dependencies = [ "anyhow", "base64", @@ -3782,7 +3782,7 @@ dependencies = [ [[package]] name = "shared" -version = "3.2.1" +version = "3.2.2" dependencies = [ "base64", "bitflags 2.10.0", @@ -4331,7 +4331,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "tuliprox" -version = "3.2.1" +version = "3.2.2" dependencies = [ "arc-swap", "async-compression", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 8e8b975de..77ab5377e 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tuliprox" -version = "3.2.1" +version = "3.2.2" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html diff --git a/backend/src/api/config_watch.rs b/backend/src/api/config_watch.rs index 7d90a8c9d..31ba94b75 100644 --- a/backend/src/api/config_watch.rs +++ b/backend/src/api/config_watch.rs @@ -82,7 +82,7 @@ impl ConfigFile { let sources_file = paths.sources_file_path.as_str(); let config = > as Access>::load(&app_state.app_config.config); let mut sources_dto = read_sources_file(sources_file, true, true, config.get_hdhr_device_overview().as_ref())?; - prepare_sources_batch(&mut sources_dto, true)?; + prepare_sources_batch(&mut sources_dto, true).await?; let sources: SourcesConfig = SourcesConfig::try_from(sources_dto)?; info!("Loaded sources file {sources_file}"); update_app_state_sources(app_state, sources).await?; diff --git a/backend/src/api/endpoints/v1_api_config.rs b/backend/src/api/endpoints/v1_api_config.rs index e3e5a1f8a..5873383ff 100644 --- a/backend/src/api/endpoints/v1_api_config.rs +++ b/backend/src/api/endpoints/v1_api_config.rs @@ -93,7 +93,7 @@ async fn config( let paths = app_state.app_config.paths.load(); match utils::read_app_config_dto(&paths, true, false) { Ok(mut app_config) => { - if let Err(err) = prepare_sources_batch(&mut app_config.sources, false) { + if let Err(err) = prepare_sources_batch(&mut app_config.sources, false).await { error!("Failed to prepare sources batch: {err}"); axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response() } else if let Err(err) = prepare_users(&mut app_config, &app_state.app_config).await { diff --git a/backend/src/utils/file/config_reader.rs b/backend/src/utils/file/config_reader.rs index 4d39e30e2..e7dc8855f 100644 --- a/backend/src/utils/file/config_reader.rs +++ b/backend/src/utils/file/config_reader.rs @@ -125,7 +125,7 @@ pub fn read_app_config_dto(paths: &ConfigPaths, }) } -pub fn prepare_sources_batch(sources: &mut SourcesConfigDto, include_computed: bool) -> Result<(), TuliproxError> { +pub async fn prepare_sources_batch(sources: &mut SourcesConfigDto, include_computed: bool) -> Result<(), TuliproxError> { let mut current_index = 0; @@ -148,7 +148,7 @@ pub fn prepare_sources_batch(sources: &mut SourcesConfigDto, include_computed: b for source in &mut sources.sources { for input in &mut source.inputs { - match get_batch_aliases(input.input_type, input.url.as_str()) { + match get_batch_aliases(input.input_type, input.url.as_str()).await { Ok(Some((_, aliases))) => { if let Some(idx) = input.prepare_batch(aliases, current_index)? { current_index = idx; @@ -167,9 +167,9 @@ pub fn prepare_sources_batch(sources: &mut SourcesConfigDto, include_computed: b Ok(()) } -pub fn get_batch_aliases(input_type: InputType, url: &str) -> Result)>, TuliproxError> { +pub async fn get_batch_aliases(input_type: InputType, url: &str) -> Result)>, TuliproxError> { if input_type == InputType::M3uBatch || input_type == InputType::XtreamBatch { - return match utils::csv_read_inputs(input_type, url) { + return match utils::csv_read_inputs(input_type, url).await { Ok((file_path, batch_aliases)) => { Ok(Some((file_path, batch_aliases))) } @@ -213,7 +213,7 @@ pub async fn read_initial_app_config(paths: &mut ConfigPaths, let config_dto = read_config_file(config_file, resolve_env, include_computed)?; let mut sources_dto = read_sources_file(sources_file, resolve_env, include_computed, config_dto.get_hdhr_device_overview().as_ref())?; - prepare_sources_batch(&mut sources_dto, include_computed)?; + prepare_sources_batch(&mut sources_dto, include_computed).await?; let sources: SourcesConfig = SourcesConfig::try_from(sources_dto)?; let mut config: Config = Config::from(config_dto); config.prepare(config_path)?; diff --git a/backend/src/utils/file/csv_input_reader.rs b/backend/src/utils/file/csv_input_reader.rs index 622440654..776a41156 100644 --- a/backend/src/utils/file/csv_input_reader.rs +++ b/backend/src/utils/file/csv_input_reader.rs @@ -1,5 +1,4 @@ use shared::error::{str_to_io_error, to_io_error}; -use crate::utils::request::{get_local_file_content}; use crate::utils::EnvResolvingReader; use crate::utils::{file_reader, resolve_relative_path}; use log::error; @@ -9,6 +8,7 @@ use std::path::PathBuf; use url::Url; use shared::model::{ConfigInputAliasDto, InputType}; use shared::utils::{get_credentials_from_url, trim_last_slash}; +use crate::utils::request::get_local_file_content; const CSV_SEPARATOR: char = ';'; const HEADER_PREFIX: char = '#'; @@ -150,9 +150,9 @@ pub fn csv_read_inputs_from_reader(batch_input_type: InputType, reader: impl Buf } -pub fn csv_read_inputs(input_type: InputType, file_uri: &str) -> Result<(PathBuf, Vec), Error> { +pub async fn csv_read_inputs(input_type: InputType, file_uri: &str) -> Result<(PathBuf, Vec), Error> { let file_path = get_csv_file_path(file_uri)?; - match get_local_file_content(&file_path) { + match get_local_file_content(&file_path).await { Ok(content) => { Ok((file_path, csv_read_inputs_from_reader(input_type, EnvResolvingReader::new(file_reader(Cursor::new(content))))?)) } @@ -251,4 +251,4 @@ input_2;de566567;de2345f43g5;http://provider_2.tv:8080;1 let resolved = resolve_env_var("${env:HOME}"); assert_eq!(resolved, std::env::var("HOME").unwrap()); } -} \ No newline at end of file +} diff --git a/backend/src/utils/file/file_utils.rs b/backend/src/utils/file/file_utils.rs index fd1051215..a52092805 100644 --- a/backend/src/utils/file/file_utils.rs +++ b/backend/src/utils/file/file_utils.rs @@ -129,15 +129,12 @@ pub fn open_file(file_name: &Path) -> Result { File::open(file_name) } -pub fn persist_file(persist_file: Option, text: &str) { +pub async fn persist_file(persist_file: Option, text: &str) { if let Some(path_buf) = persist_file { let filename = &path_buf.to_str().unwrap_or("?"); - match File::create(&path_buf) { - Ok(mut file) => match file.write_all(text.as_bytes()) { - Ok(()) => debug!("persisted: {filename}"), - Err(e) => error!("failed to persist file {filename}, {e}") - }, - Err(e) => error!("failed to persist file {filename}, {e}") + match tokio::fs::write(&path_buf, text).await { + Ok(()) => debug!("persisted: {filename}"), + Err(e) => error!("failed to persist file {filename}, {e}"), } } } diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs index ab4024309..e9652ff87 100644 --- a/backend/src/utils/network/request.rs +++ b/backend/src/utils/network/request.rs @@ -1,7 +1,6 @@ use std::collections::{HashMap, HashSet}; -use std::fs; use std::fs::File; -use std::io::{BufWriter, Error, ErrorKind, Read, Write}; +use std::io::{BufWriter, Error, ErrorKind, Write}; use std::path::{Path, PathBuf}; use std::pin::Pin; use std::sync::Arc; @@ -70,12 +69,9 @@ pub async fn get_input_epg_content_as_file(client: Arc, input: if filepath.exists() { if let Some(persist_file_value) = persist_filepath { let to_file = &persist_file_value; - match fs::copy(&filepath, to_file) { - Ok(_) => {} - Err(e) => { - error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e); - return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e); - } + if let Err(e) = tokio::fs::copy(&filepath, to_file).await { + error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e); + return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e); } } @@ -117,16 +113,13 @@ pub async fn get_input_text_content(client: Arc, input: &InputS if filepath.exists() { if let Some(persist_file_value) = persist_filepath { let to_file = &persist_file_value; - match fs::copy(&filepath, to_file) { - Ok(_) => {} - Err(e) => { - error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e); - return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e); - } + if let Err(e) = tokio::fs::copy(&filepath, to_file).await { + error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e); + return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e); } } - match get_local_file_content(&filepath) { + match get_local_file_content(&filepath).await { Ok(content) => Some(content), Err(err) => { return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed : {}", err); @@ -211,25 +204,38 @@ pub fn get_request_headers(request_header headers } -pub fn get_local_file_content(file_path: &PathBuf) -> Result { - // Check if the file is accessible - if file_path.exists() && file_path.is_file() { - if let Ok(content) = fs::read(file_path) { - if content.len() >= 2 && is_gzip(&content[0..2]) { - let mut decoder = flate2::read::GzDecoder::new(&content[..]); - let mut decode_buffer = String::new(); - return match decoder.read_to_string(&mut decode_buffer) { - Ok(_) => Ok(decode_buffer), - Err(err) => Err(str_to_io_error(&format!("failed to decode gzip content {err}"))) - }; - } - return Ok(String::from_utf8_lossy(&content).parse().unwrap_or_default()); - } +async fn decode_local_file_bytes(content: Vec) -> Result { + if content.len() >= 2 && is_gzip(&content[0..2]) { + let mut decoder = async_compression::tokio::bufread::GzipDecoder::new(&content[..]); + let mut decode_buffer = String::new(); + decoder + .read_to_string(&mut decode_buffer).await + .map_err(|err| str_to_io_error(&format!("failed to decode gzip content {err}")))?; + Ok(decode_buffer) + } else { + Ok(String::from_utf8_lossy(&content).parse().unwrap_or_default()) } - let file_str = file_path.to_str().unwrap_or("?"); - Err(Error::new(ErrorKind::InvalidData, format!("Cant find file {file_str}"))) } +fn local_file_not_found(path: &PathBuf) -> Error { + let file_str = path.to_str().unwrap_or("?"); + Error::new(ErrorKind::InvalidData, format!("Cant find file {file_str}")) +} + +pub async fn get_local_file_content(file_path: &PathBuf) -> Result { + match tokio::fs::read(file_path).await { + Ok(content) => decode_local_file_bytes(content).await, + Err(_) => Err(local_file_not_found(file_path)), + } +} + +// pub fn get_local_file_content_blocking(file_path: &PathBuf) -> Result { +// match fs::read(file_path) { +// Ok(content) => decode_local_file_bytes(content).await, +// Err(_) => Err(local_file_not_found(file_path)), +// } +// } + async fn get_remote_content_as_file(client: Arc, input: &ConfigInput, url: &Url, file_path: &Path) -> Result { let start_time = Instant::now(); @@ -349,26 +355,39 @@ async fn download_epg_content_as_file(client: Arc, input: &Conf } } -pub async fn download_text_content(client: Arc, disabled_headers: Option<&ReverseProxyDisabledHeaderConfig>, input: &InputSource, headers: Option<&HeaderMap>, persist_filepath: Option) -> Result<(String, String), Error> { +pub async fn download_text_content( + client: Arc, + disabled_headers: Option<&ReverseProxyDisabledHeaderConfig>, + input: &InputSource, + headers: Option<&HeaderMap>, + persist_filepath: Option, +) -> Result<(String, String), Error> { if let Ok(url) = input.url.parse::() { let result = if url.scheme() == "file" { - url.to_file_path().map_or_else(|()| Err(str_to_io_error(&format!("Unknown file {}", sanitize_sensitive_info(&input.url)))), |file_path| - get_local_file_content(&file_path).map(|c| (c, url.to_string())), - ) + match url.to_file_path() { + Ok(file_path) => get_local_file_content(&file_path).await.map(|c| (c, url.to_string())), + Err(()) => Err(str_to_io_error(&format!( + "Unknown file {}", + sanitize_sensitive_info(&input.url) + ))), + } } else { get_remote_content(client, input, headers, &url, disabled_headers).await }; match result { Ok((content, response_url)) => { if persist_filepath.is_some() { - persist_file(persist_filepath, &content); + persist_file(persist_filepath, &content).await; } Ok((content, response_url)) } - Err(err) => Err(err) + Err(err) => Err(err), } } else { - Err(str_to_io_error(&format!("Malformed URL {}", sanitize_sensitive_info(&input.url)))) + Err(str_to_io_error(&format!( + "Malformed URL {}", + sanitize_sensitive_info(&input.url) + ))) } } diff --git a/frontend/Cargo.toml b/frontend/Cargo.toml index 18bf0722c..12a475134 100644 --- a/frontend/Cargo.toml +++ b/frontend/Cargo.toml @@ -1,10 +1,10 @@ [package] name = "frontend" -version = "3.2.1" +version = "3.2.2" edition = "2021" [dependencies] -shared = { version = "3.2.1", path = "../shared" } +shared = { version = "3.2.2", path = "../shared" } chrono = "0" yew = "0.21" yew-router = "0.18" diff --git a/shared/Cargo.toml b/shared/Cargo.toml index e94bc1039..0faab76aa 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "shared" -version = "3.2.1" +version = "3.2.2" edition = "2021" [dependencies]