From e681da07337462ac7ffd1673b73aa12582cece8c Mon Sep 17 00:00:00 2001 From: DarkBreakpoint <243206744+DarkBreakpoint@users.noreply.github.com> Date: Fri, 14 Nov 2025 10:11:08 -0600 Subject: [PATCH 01/10] Persist cache pipe via tokio::fs to avoid blocking the runtime --- CHANGELOG.md | 1 + backend/src/api/api_utils.rs | 4 +- .../api/model/streams/persist_pipe_stream.rs | 117 ++++++++++++------ backend/src/utils/file/file_utils.rs | 10 +- 4 files changed, 92 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b4f3bf658..aac993d65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ - 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 +- Persist cache write pipe using `tokio::fs` so response caching no longer blocks the async runtime # 3.2.0 (2025-11-14) - Added `name` attribute to Staged Input. diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index fbdd7f9a0..a086fb0b3 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -32,7 +32,7 @@ use shared::utils::{ }; use std::borrow::Cow; use std::collections::HashMap; -use std::io::BufWriter; +use tokio::io::BufWriter; use std::path::Path; use std::sync::Arc; use std::time::Duration; @@ -1076,7 +1076,7 @@ async fn build_stream_response( None }; if let Some(resource_path) = cache_resource_path { - if let Ok(file) = create_new_file_for_write(&resource_path) { + if let Ok(file) = create_new_file_for_write(&resource_path).await { let writer = BufWriter::new(file); let add_cache_content = get_add_cache_content(resource_url, &app_state.cache); let stream = PersistPipeStream::new(byte_stream, writer, add_cache_content); diff --git a/backend/src/api/model/streams/persist_pipe_stream.rs b/backend/src/api/model/streams/persist_pipe_stream.rs index 5864958e9..33fa1ec22 100644 --- a/backend/src/api/model/streams/persist_pipe_stream.rs +++ b/backend/src/api/model/streams/persist_pipe_stream.rs @@ -1,45 +1,34 @@ -use std::io::Write; +use std::collections::VecDeque; use std::pin::Pin; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::task::{Context, Poll}; use bytes::Bytes; use log::error; +use tokio::io::{AsyncWrite, AsyncWriteExt}; use tokio_stream::Stream; use crate::api::model::StreamError; /// `PersistPipeStream` /// -/// A stream wrapper that pipes data from an input stream to a writer while tracking -/// the total number of bytes processed. Upon completion, it triggers a user-provided -/// callback with the total size of the data written. -/// - `callback`: A user-provided function that is called with the total size of the processed data once the stream is completed. -/// -/// # Stream Implementation -/// - Implements the `Stream` trait to poll the underlying stream for data. -/// - For each chunk of data: -/// - Writes it to the writer. -/// - Updates the size tracker. -/// - When the stream is exhausted: -/// - Calls `on_complete()` to finalize the operation and trigger the callback. +/// Pipes bytes from an upstream stream to an async writer while tracking total size. +/// Once the stream completes and the writer is flushed, the provided callback is invoked +/// with the total number of bytes written. pub struct PersistPipeStream { inner: S, completed: bool, writer: W, size: AtomicUsize, callback: Arc, + pending_writes: VecDeque, + current_offset: usize, } impl PersistPipeStream where S: Stream + Unpin, - W: Write + Unpin + 'static, + W: AsyncWrite + Unpin + 'static, { - /// - Creates a new `PersistPipeStream` instance. - /// - Arguments: - /// - `inner`: The input stream providing the data. - /// - `writer`: The writer to which the data is written. - /// - `callback`: A callback function to be called with the total size upon stream completion. pub fn new(inner: S, writer: W, callback: Arc) -> Self { Self { inner, @@ -47,26 +36,63 @@ where writer, size: AtomicUsize::new(0), callback, + pending_writes: VecDeque::new(), + current_offset: 0, } } - fn on_complete(&mut self) { + fn enqueue_chunk(&mut self, bytes: Bytes) { + self.pending_writes.push_back(bytes); + } + + fn poll_pending_writes(&mut self, cx: &mut Context<'_>) -> Poll<()> { + while let Some(chunk) = self.pending_writes.front() { + if self.current_offset >= chunk.len() { + if let Some(finished) = self.pending_writes.pop_front() { + self.size.fetch_add(finished.len(), Ordering::SeqCst); + } + self.current_offset = 0; + continue; + } + + let remaining = &chunk[self.current_offset..]; + match Pin::new(&mut self.writer).poll_write(cx, remaining) { + Poll::Pending => return Poll::Pending, + Poll::Ready(Ok(written)) => { + if written == 0 { + // Avoid tight loop if writer makes no progress. + return Poll::Pending; + } + self.current_offset += written; + } + Poll::Ready(Err(err)) => { + error!("Error writing to resource file: {err}"); + self.pending_writes.pop_front(); + self.current_offset = 0; + } + } + } + + self.current_offset = 0; + Poll::Ready(()) + } + + fn poll_flush(&mut self, cx: &mut Context<'_>) -> Poll<()> { + match Pin::new(&mut self.writer).poll_flush(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(Ok(())) => Poll::Ready(()), + Poll::Ready(Err(err)) => { + error!("Error flushing resource file: {err}"); + Poll::Ready(()) + } + } + } + + fn finalize(&mut self) { if !self.completed { self.completed = true; let size = self.size.load(Ordering::SeqCst); - if self.writer.flush().is_ok() { - (self.callback)(size); - } - } - } - - fn on_data(&mut self, data: &Result) { - if let Ok(bytes) = data { - self.size.fetch_add(bytes.len(), Ordering::SeqCst); - let bytes_to_write = bytes.clone(); - if let Err(e) = self.writer.write_all(&bytes_to_write) { - error!("Error writing to resource file: {e}"); - } + (self.callback)(size); } } } @@ -74,20 +100,39 @@ where impl Stream for PersistPipeStream where S: Stream> + Unpin, - W: Write + Unpin + 'static, + W: AsyncWrite + Unpin + 'static, { type Item = Result; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let this = self.get_mut(); + + if !this.pending_writes.is_empty() { + if let Poll::Pending = this.poll_pending_writes(cx) { + return Poll::Pending; + } + } + match Pin::new(&mut this.inner).poll_next(cx) { Poll::Pending => Poll::Pending, Poll::Ready(None) => { - this.on_complete(); + if let Poll::Pending = this.poll_pending_writes(cx) { + return Poll::Pending; + } + if let Poll::Pending = this.poll_flush(cx) { + return Poll::Pending; + } + this.finalize(); Poll::Ready(None) } Poll::Ready(Some(item)) => { - this.on_data(&item); + if let Ok(bytes) = &item { + this.enqueue_chunk(bytes.clone()); + } + // Try to drain pending bytes after queuing new data. + if let Poll::Pending = this.poll_pending_writes(cx) { + // fall through: we still return the chunk to the caller even if persistence is pending + } Poll::Ready(Some(item)) } } diff --git a/backend/src/utils/file/file_utils.rs b/backend/src/utils/file/file_utils.rs index a52092805..d59fc75c0 100644 --- a/backend/src/utils/file/file_utils.rs +++ b/backend/src/utils/file/file_utils.rs @@ -10,6 +10,7 @@ use crate::utils::debug_if_enabled; use shared::utils::{API_PROXY_FILE, CONFIG_FILE, CONFIG_PATH, MAPPING_FILE, SOURCE_FILE, USER_FILE}; use log::{debug, error}; use path_clean::PathClean; +use tokio::fs as tokio_fs; pub fn file_writer(w: W) -> BufWriter where @@ -198,8 +199,13 @@ pub fn append_or_crate_file(path: &Path) -> std::io::Result { } #[inline] -pub fn create_new_file_for_write(path: &Path) -> std::io::Result { - OpenOptions::new().write(true).create(true).truncate(true).open(path) +pub async fn create_new_file_for_write(path: &Path) -> tokio::io::Result { + tokio_fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .open(path) + .await } #[inline] From 2b8c9ffc3ac37ea4421c7bd15ae57a9099920bd1 Mon Sep 17 00:00:00 2001 From: DarkBreakpoint <243206744+DarkBreakpoint@users.noreply.github.com> Date: Fri, 14 Nov 2025 10:56:03 -0600 Subject: [PATCH 02/10] Use Arc for shared-stream burst buffer snapshots --- CHANGELOG.md | 1 + .../model/streams/shared_stream_manager.rs | 19 ++++++++++--------- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b4f3bf658..7584a730f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ - 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 +- Shared stream burst buffer now reuses `Arc` so new subscribers no longer clone the entire burst data set. # 3.2.0 (2025-11-14) - Added `name` attribute to Staged Input. diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs index 752610d6f..6c55132c3 100644 --- a/backend/src/api/model/streams/shared_stream_manager.rs +++ b/backend/src/api/model/streams/shared_stream_manager.rs @@ -67,7 +67,7 @@ type SubscriberId = SocketAddr; struct BurstBuffer { - buffer: VecDeque, + buffer: VecDeque>, buffer_size: usize, current_bytes: usize, } @@ -91,11 +91,11 @@ impl BurstBuffer { } } - pub fn snapshot(&self) -> VecDeque { - self.buffer.iter().cloned().collect::>() + pub fn snapshot(&self) -> VecDeque> { + self.buffer.iter().cloned().collect::>>() } - pub fn push(&mut self, packet: &Bytes) { + pub fn push(&mut self, packet: Arc) { while self.current_bytes > self.buffer_size { if let Some(popped) = self.buffer.pop_front() { self.current_bytes -= popped.len(); @@ -105,18 +105,18 @@ impl BurstBuffer { } } self.current_bytes += packet.len(); - self.buffer.push_back(packet.clone()); + self.buffer.push_back(packet); } } async fn send_burst_buffer( - start_buffer: &VecDeque, + start_buffer: &VecDeque>, client_tx: &Sender, cancellation_token: &CancellationToken) { for buf in start_buffer { if cancellation_token.is_cancelled() { return; } - if let Err(err) = client_tx.send(buf.clone()).await { + if let Err(err) = client_tx.send(buf.as_ref().clone()).await { debug!("Error sending current chunk: {err}"); return; // stop on send error } @@ -244,12 +244,13 @@ impl SharedStreamState { item = source_stream.next() => { match item { Some(Ok(data)) => { + let arc_data = Arc::new(data); { let mut buffer = burst_buffer.lock().await; - buffer.push(&data); + buffer.push(arc_data.clone()); } - match sender.send(data) { + match sender.send(arc_data.as_ref().clone()) { Ok(clients) => { if clients == 0 { debug_if_enabled!("No shared stream subscribers closing {}", sanitize_sensitive_info(&streaming_url)); From 19de84a762237ff895d5d5f087a48c6399bda55b Mon Sep 17 00:00:00 2001 From: DarkBreakpoint <243206744+DarkBreakpoint@users.noreply.github.com> Date: Fri, 14 Nov 2025 12:18:48 -0600 Subject: [PATCH 03/10] Make M3U export use tokio::fs --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b4f3bf658..6d4be57d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ - 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 +- M3U playlist exports now stream via `tokio::fs` to keep the runtime responsive # 3.2.0 (2025-11-14) - Added `name` attribute to Staged Input. From 1c77d52cc227a1f3b7fb9e91af768d4a6b11c929 Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 14 Nov 2025 22:26:19 +0100 Subject: [PATCH 04/10] build tag for dev image --- bin/build_docker.sh | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/bin/build_docker.sh b/bin/build_docker.sh index 67c8ad936..c37f12a33 100755 --- a/bin/build_docker.sh +++ b/bin/build_docker.sh @@ -162,16 +162,17 @@ for IMAGE_NAME in "${!MULTI_PLATFORM_IMAGES[@]}"; do # Prepare tags based on branch DOCKER_TAGS="" - if [ "$BRANCH" = "master" ]; then + #if [ "$BRANCH" = "master" ]; then # For master branch: create both version and latest tags DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${VERSION} -t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}" BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${VERSION}") BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}") - elif [ "$BRANCH" = "develop" ]; then - # For develop branch: create only dev tag (no version tag) - DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}" - BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}") - fi + #elif [ "$BRANCH" = "develop" ]; then + # # For develop branch: create only dev tag (no version tag) + # DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}" + # BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${VERSION}") + # BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}") + #fi # Build and push multi-platform image directly with cache BUILDX_CACHE_ARGS=() From 9c43c389edabd3f4aabcc17a015c7009cd57f8f5 Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 14 Nov 2025 22:35:18 +0100 Subject: [PATCH 05/10] Persist pipe stream Drop implementation to ensure the callback is invoked --- backend/src/api/model/streams/persist_pipe_stream.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/backend/src/api/model/streams/persist_pipe_stream.rs b/backend/src/api/model/streams/persist_pipe_stream.rs index 33fa1ec22..dd36d8ecf 100644 --- a/backend/src/api/model/streams/persist_pipe_stream.rs +++ b/backend/src/api/model/streams/persist_pipe_stream.rs @@ -138,3 +138,9 @@ where } } } + +impl Drop for PersistPipeStream { + fn drop(&mut self) { + self.finalize(); + } +} \ No newline at end of file From 6f2be9693c0a8488ee3e82daeb04ecdced5a408b Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 14 Nov 2025 22:45:08 +0100 Subject: [PATCH 06/10] Using async file operations --- .../api/model/streams/persist_pipe_stream.rs | 34 +++++++------------ backend/src/repository/m3u_repository.rs | 6 ++-- backend/src/utils/network/request.rs | 4 +-- 3 files changed, 18 insertions(+), 26 deletions(-) diff --git a/backend/src/api/model/streams/persist_pipe_stream.rs b/backend/src/api/model/streams/persist_pipe_stream.rs index dd36d8ecf..550384350 100644 --- a/backend/src/api/model/streams/persist_pipe_stream.rs +++ b/backend/src/api/model/streams/persist_pipe_stream.rs @@ -1,13 +1,13 @@ -use std::collections::VecDeque; -use std::pin::Pin; -use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; -use std::task::{Context, Poll}; +use crate::api::model::StreamError; use bytes::Bytes; use log::error; -use tokio::io::{AsyncWrite, AsyncWriteExt}; +use std::collections::VecDeque; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::task::{Context, Poll}; +use tokio::io::AsyncWrite; use tokio_stream::Stream; -use crate::api::model::StreamError; /// `PersistPipeStream` /// @@ -99,7 +99,7 @@ where impl Stream for PersistPipeStream where - S: Stream> + Unpin, + S: Stream> + Unpin, W: AsyncWrite + Unpin + 'static, { type Item = Result; @@ -107,19 +107,17 @@ where fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let this = self.get_mut(); - if !this.pending_writes.is_empty() { - if let Poll::Pending = this.poll_pending_writes(cx) { - return Poll::Pending; - } + if !this.pending_writes.is_empty() && this.poll_pending_writes(cx).is_pending() { + return Poll::Pending; } match Pin::new(&mut this.inner).poll_next(cx) { Poll::Pending => Poll::Pending, Poll::Ready(None) => { - if let Poll::Pending = this.poll_pending_writes(cx) { + if this.poll_pending_writes(cx).is_pending() { return Poll::Pending; } - if let Poll::Pending = this.poll_flush(cx) { + if this.poll_flush(cx).is_pending() { return Poll::Pending; } this.finalize(); @@ -130,7 +128,7 @@ where this.enqueue_chunk(bytes.clone()); } // Try to drain pending bytes after queuing new data. - if let Poll::Pending = this.poll_pending_writes(cx) { + if this.poll_pending_writes(cx).is_pending() { // fall through: we still return the chunk to the caller even if persistence is pending } Poll::Ready(Some(item)) @@ -138,9 +136,3 @@ where } } } - -impl Drop for PersistPipeStream { - fn drop(&mut self) { - self.finalize(); - } -} \ No newline at end of file diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index 72430e6c3..552160385 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -108,8 +108,8 @@ pub async fn m3u_write_playlist( Err(err) => Err(cant_write_result!(&m3u_path_clone, err)), } }) - .await - .map_err(|err| create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {err}", m3u_path.display()))??; + .await + .map_err(|err| create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {err}", m3u_path.display()))??; Ok(()) } @@ -156,4 +156,4 @@ pub async fn iter_raw_m3u_playlist(config: &AppConfig, target: &ConfigTarget) -> Ok(reader) => Some((file_lock, reader)), Err(_) => None } -} +} \ No newline at end of file diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs index e9652ff87..46d613bf5 100644 --- a/backend/src/utils/network/request.rs +++ b/backend/src/utils/network/request.rs @@ -217,12 +217,12 @@ async fn decode_local_file_bytes(content: Vec) -> Result { } } -fn local_file_not_found(path: &PathBuf) -> Error { +fn local_file_not_found(path: &Path) -> 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 { +pub async fn get_local_file_content(file_path: &Path) -> Result { match tokio::fs::read(file_path).await { Ok(content) => decode_local_file_bytes(content).await, Err(_) => Err(local_file_not_found(file_path)), From 86f798f5bd6c4d9f5baf2fba00f0d2ff24ce0eca Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 14 Nov 2025 22:55:59 +0100 Subject: [PATCH 07/10] M3u repository playlist async persists --- backend/src/repository/m3u_repository.rs | 50 ++++++++++++------------ 1 file changed, 26 insertions(+), 24 deletions(-) diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index 552160385..e7f8df048 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -1,4 +1,4 @@ -use crate::api::model::{AppState}; +use crate::api::model::AppState; use crate::model::{AppConfig, ProxyUserCredentials}; use crate::model::{Config, ConfigTarget, M3uTargetOutput}; use crate::repository::indexed_document::{IndexedDocumentDirectAccess, IndexedDocumentIterator, IndexedDocumentWriter}; @@ -6,7 +6,6 @@ use crate::repository::m3u_playlist_iterator::M3uPlaylistM3uTextIterator; use crate::repository::storage::get_target_storage_path; use crate::repository::storage_const; use crate::utils; -use log::error; use shared::error::{create_tuliprox_error, info_err}; use shared::error::{str_to_io_error, TuliproxError, TuliproxErrorKind}; use shared::model::PlaylistItemType; @@ -35,33 +34,36 @@ pub fn m3u_get_epg_file_path(target_path: &Path) -> PathBuf { utils::add_prefix_to_filename(&path, "epg_", Some("xml")) } +macro_rules! await_playlist_write { + ($expr:expr, $fmt:literal $(, $args:expr)* ) => {{ + $expr.await.map_err(|err| { + create_tuliprox_error!(TuliproxErrorKind::Notify, $fmt $(, $args)*, err) + })? + }}; +} + async fn persist_m3u_playlist_as_text( cfg: &Config, target: &ConfigTarget, target_output: &M3uTargetOutput, m3u_playlist: Arc>, -) { - let Some(filename) = target_output.filename.as_ref() else { return }; - let Some(m3u_filename) = utils::get_file_path(&cfg.working_dir, Some(PathBuf::from(filename))) else { return }; +) -> Result<(), TuliproxError> { + let Some(filename) = target_output.filename.as_ref() else { return Ok(()); }; + let Some(m3u_filename) = utils::get_file_path(&cfg.working_dir, Some(PathBuf::from(filename))) else { return Ok(()); }; - match fs::File::create(&m3u_filename).await { - Ok(file) => { - let mut writer = AsyncBufWriter::new(file); - if writer.write_all(b"#EXTM3U\n").await.is_ok() { - for m3u in m3u_playlist.iter() { - let line = m3u.to_m3u(target.options.as_ref(), false); - if writer.write_all(line.as_bytes()).await.is_err() - || writer.write_all(b"\n").await.is_err() - { - error!("Failed to write entry to {}", m3u_filename.display()); - break; - } - } - let _ = writer.flush().await; - } - } - Err(_) => error!("Can't write m3u plain playlist {}", &m3u_filename.display()), + let file = await_playlist_write!(fs::File::create(&m3u_filename), "Can't write m3u plain playlist {} - {}", m3u_filename.display()); + let mut writer = AsyncBufWriter::new(file); + await_playlist_write!(writer.write_all(b"#EXTM3U\n"), "Failed to write header to {} - {}", m3u_filename.display()); + + for m3u in m3u_playlist.iter() { + let line = m3u.to_m3u(target.options.as_ref(), false); + await_playlist_write!(writer.write_all(line.as_bytes()), "Failed to write entry to {} - {}", m3u_filename.display()); + await_playlist_write!(writer.write_all(b"\n"), "Failed to write newline to {} - {}", m3u_filename.display()); } + + await_playlist_write!(writer.flush(), "Failed to flush {} - {}", m3u_filename.display()); + + Ok(()) } pub async fn m3u_write_playlist( @@ -85,7 +87,7 @@ pub async fn m3u_write_playlist( .collect::>(), ); - persist_m3u_playlist_as_text(&cfg.config.load(), target, target_output, Arc::clone(&m3u_playlist)).await; + let persist_result = persist_m3u_playlist_as_text(&cfg.config.load(), target, target_output, Arc::clone(&m3u_playlist)).await; let file_lock = cfg.file_locks.write_lock(&m3u_path).await; let playlist = Arc::clone(&m3u_playlist); @@ -111,7 +113,7 @@ pub async fn m3u_write_playlist( .await .map_err(|err| create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {err}", m3u_path.display()))??; - Ok(()) + persist_result } pub async fn m3u_load_rewrite_playlist( From cbb0da27cbdf8b9f976c28e869e24a9b06df18e0 Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 14 Nov 2025 23:02:56 +0100 Subject: [PATCH 08/10] version 3.2.3 --- Cargo.lock | 123 +++++++++++++++----------------------------- backend/Cargo.toml | 4 +- frontend/Cargo.toml | 4 +- shared/Cargo.toml | 2 +- 4 files changed, 46 insertions(+), 87 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b817c70be..b39068668 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -79,22 +79,22 @@ dependencies = [ [[package]] name = "anstyle-query" -version = "1.1.4" +version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e231f6134f61b71076a3eab506c379d4f36122f2af15a9ff04415ea4c3339e2" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] name = "anstyle-wincon" -version = "3.0.10" +version = "3.0.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3e0633414522a32ffaac8ac6cc8f748e090c5717661fddeea04219e2344f5f2a" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -165,9 +165,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" [[package]] name = "axum" -version = "0.8.6" +version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a18ed336352031311f4e0b4dd2ff392d4fbb370777c9d18d7fc9d7359f73871" +checksum = "5b098575ebe77cb6d14fc7f32749631a6e44edbef6b796f89b020e99ba20d425" dependencies = [ "axum-core", "axum-macros", @@ -357,15 +357,15 @@ checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43" [[package]] name = "bytes" -version = "1.10.1" +version = "1.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a" +checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3" [[package]] name = "cc" -version = "1.2.45" +version = "1.2.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "35900b6c8d709fb1d854671ae27aeaa9eec2f8b01b364e1619a40da3e6fe2afe" +checksum = "b97463e1064cb1b1c1384ad0a0b9c8abd0988e2a91f52606c80ef14aadb63e36" dependencies = [ "find-msvc-tools", "jobserver", @@ -395,7 +395,7 @@ dependencies = [ "js-sys", "num-traits", "wasm-bindgen", - "windows-link 0.2.1", + "windows-link", ] [[package]] @@ -1034,9 +1034,9 @@ dependencies = [ [[package]] name = "find-msvc-tools" -version = "0.1.4" +version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52051878f80a721bb68ebfbc930e07b65ba72f2da88968ea5c06fd6ca3d3a127" +checksum = "3a3076410a55c90011c298b04d0cfa770b00fa04e1e3c97d3f6c9de105a03844" [[package]] name = "flate2" @@ -1054,12 +1054,6 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" -[[package]] -name = "foldhash" -version = "0.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" - [[package]] name = "foldhash" version = "0.2.0" @@ -1102,7 +1096,7 @@ dependencies = [ [[package]] name = "frontend" -version = "3.2.2" +version = "3.2.3" dependencies = [ "anyhow", "base64", @@ -1740,9 +1734,9 @@ dependencies = [ [[package]] name = "governor" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "444405bbb1a762387aa22dd569429533b54a1d8759d35d3b64cb39b0293eaa19" +checksum = "6e23d5986fd4364c2fb7498523540618b4b8d92eec6c36a02e565f66748e2f79" dependencies = [ "cfg-if", "dashmap", @@ -1750,7 +1744,7 @@ dependencies = [ "futures-timer", "futures-util", "getrandom 0.3.4", - "hashbrown 0.15.5", + "hashbrown 0.16.0", "nonzero_ext", "parking_lot", "portable-atomic", @@ -1808,17 +1802,6 @@ version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" -[[package]] -name = "hashbrown" -version = "0.15.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" -dependencies = [ - "allocator-api2", - "equivalent", - "foldhash 0.1.5", -] - [[package]] name = "hashbrown" version = "0.16.0" @@ -1827,7 +1810,7 @@ checksum = "5419bdc4f6a9207fbeba6d11b604d481addf78ecd10c11ad51e76c2f6482748d" dependencies = [ "allocator-api2", "equivalent", - "foldhash 0.2.0", + "foldhash", ] [[package]] @@ -1925,9 +1908,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hyper" -version = "1.8.0" +version = "1.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1744436df46f0bde35af3eda22aeaba453aada65d8f1c171cd8a5f59030bd69f" +checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11" dependencies = [ "atomic-waker", "bytes", @@ -1994,9 +1977,9 @@ dependencies = [ [[package]] name = "hyper-util" -version = "0.1.17" +version = "0.1.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c6995591a8f1380fcb4ba966a252a4b29188d51d2b89e3a252f5305be65aea8" +checksum = "52e9a2a24dc5c6821e71a7030e1e14b7b632acac55c40e9d2e082c621261bb56" dependencies = [ "base64", "bytes", @@ -2732,7 +2715,7 @@ dependencies = [ "libc", "redox_syscall", "smallvec", - "windows-link 0.2.1", + "windows-link", ] [[package]] @@ -3438,9 +3421,9 @@ dependencies = [ [[package]] name = "rsa" -version = "0.9.8" +version = "0.9.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78928ac1ed176a5ca1d17e578a1825f3d81ca54cf41053a592584b020cfd691b" +checksum = "40a0376c50d0358279d9d643e4bf7b7be212f1f4ff1da9070a7b54d22ef75c88" dependencies = [ "const-oid", "digest", @@ -3782,7 +3765,7 @@ dependencies = [ [[package]] name = "shared" -version = "3.2.2" +version = "3.2.3" dependencies = [ "base64", "bitflags 2.10.0", @@ -4331,7 +4314,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "tuliprox" -version = "3.2.2" +version = "3.2.3" dependencies = [ "arc-swap", "async-compression", @@ -4758,9 +4741,9 @@ checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" dependencies = [ "windows-implement", "windows-interface", - "windows-link 0.2.1", - "windows-result 0.4.1", - "windows-strings 0.5.1", + "windows-link", + "windows-result", + "windows-strings", ] [[package]] @@ -4785,12 +4768,6 @@ dependencies = [ "syn 2.0.110", ] -[[package]] -name = "windows-link" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" - [[package]] name = "windows-link" version = "0.2.1" @@ -4799,22 +4776,13 @@ checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" [[package]] name = "windows-registry" -version = "0.5.3" +version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b8a9ed28765efc97bbc954883f4e6796c33a06546ebafacbabee9696967499e" +checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720" dependencies = [ - "windows-link 0.1.3", - "windows-result 0.3.4", - "windows-strings 0.4.2", -] - -[[package]] -name = "windows-result" -version = "0.3.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" -dependencies = [ - "windows-link 0.1.3", + "windows-link", + "windows-result", + "windows-strings", ] [[package]] @@ -4823,16 +4791,7 @@ version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" dependencies = [ - "windows-link 0.2.1", -] - -[[package]] -name = "windows-strings" -version = "0.4.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" -dependencies = [ - "windows-link 0.1.3", + "windows-link", ] [[package]] @@ -4841,7 +4800,7 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" dependencies = [ - "windows-link 0.2.1", + "windows-link", ] [[package]] @@ -4877,7 +4836,7 @@ version = "0.61.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" dependencies = [ - "windows-link 0.2.1", + "windows-link", ] [[package]] @@ -4902,7 +4861,7 @@ version = "0.53.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" dependencies = [ - "windows-link 0.2.1", + "windows-link", "windows_aarch64_gnullvm 0.53.1", "windows_aarch64_msvc 0.53.1", "windows_i686_gnu 0.53.1", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 77ab5377e..2a13255e6 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tuliprox" -version = "3.2.2" +version = "3.2.3" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -40,7 +40,7 @@ fastrand = "2" rpassword = "7.4" flate2 = "1.1" blake3 = "1.8" -bytes = "1.10" +bytes = "1.11" tokio-stream = { version = "0.1", features = ["sync"] } tokio = { version = "1.48", features = ["rt-multi-thread", "parking_lot", "fs"] } #tokio = { version = "1.46", features = ["rt-multi-thread", "parking_lot", "fs", "tracing"] } diff --git a/frontend/Cargo.toml b/frontend/Cargo.toml index 12a475134..123f5a7a3 100644 --- a/frontend/Cargo.toml +++ b/frontend/Cargo.toml @@ -1,10 +1,10 @@ [package] name = "frontend" -version = "3.2.2" +version = "3.2.3" edition = "2021" [dependencies] -shared = { version = "3.2.2", path = "../shared" } +shared = { version = "3.2.3", path = "../shared" } chrono = "0" yew = "0.21" yew-router = "0.18" diff --git a/shared/Cargo.toml b/shared/Cargo.toml index 0faab76aa..3e30a1de4 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "shared" -version = "3.2.2" +version = "3.2.3" edition = "2021" [dependencies] From f6adf3c498b433d6f49211aa9a432717be1398ac Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 14 Nov 2025 23:28:10 +0100 Subject: [PATCH 09/10] m3u repository save aync playlist file --- backend/src/repository/m3u_repository.rs | 3 ++- bin/build_docker.sh | 17 ++++------------- 2 files changed, 6 insertions(+), 14 deletions(-) diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index e7f8df048..8a66eec95 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -87,9 +87,10 @@ pub async fn m3u_write_playlist( .collect::>(), ); + let file_lock = cfg.file_locks.write_lock(&m3u_path).await; + let persist_result = persist_m3u_playlist_as_text(&cfg.config.load(), target, target_output, Arc::clone(&m3u_playlist)).await; - let file_lock = cfg.file_locks.write_lock(&m3u_path).await; let playlist = Arc::clone(&m3u_playlist); let m3u_path_clone = m3u_path.clone(); let idx_path_clone = idx_path.clone(); diff --git a/bin/build_docker.sh b/bin/build_docker.sh index c37f12a33..069d5b75b 100755 --- a/bin/build_docker.sh +++ b/bin/build_docker.sh @@ -161,19 +161,10 @@ for IMAGE_NAME in "${!MULTI_PLATFORM_IMAGES[@]}"; do echo "🎯 Building multi-platform image: ${IMAGE_NAME} with target ${BUILD_TARGET}" # Prepare tags based on branch - DOCKER_TAGS="" - #if [ "$BRANCH" = "master" ]; then - # For master branch: create both version and latest tags - DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${VERSION} -t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}" - BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${VERSION}") - BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}") - #elif [ "$BRANCH" = "develop" ]; then - # # For develop branch: create only dev tag (no version tag) - # DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}" - # BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${VERSION}") - # BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}") - #fi - + DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${VERSION} -t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}" + BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${VERSION}") + BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}") + # Build and push multi-platform image directly with cache BUILDX_CACHE_ARGS=() [ -n "${BUILDX_CACHE_FROM:-}" ] && BUILDX_CACHE_ARGS+=(--cache-from "${BUILDX_CACHE_FROM}") From 9825cd24cf17a40ae6aeb51e321995277d48725b Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 14 Nov 2025 23:39:26 +0100 Subject: [PATCH 10/10] m3u repository save aync playlist file --- backend/src/repository/m3u_repository.rs | 7 +++++-- bin/build_docker.sh | 4 ++-- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index 8a66eec95..7d97e0f3c 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -13,6 +13,7 @@ use shared::model::{M3uPlaylistItem, PlaylistGroup, PlaylistItem}; use std::io::Error; use std::path::{Path, PathBuf}; use std::sync::Arc; +use log::error; use tokio::fs; use tokio::io::{AsyncWriteExt, BufWriter as AsyncBufWriter}; use tokio::task; @@ -89,7 +90,9 @@ pub async fn m3u_write_playlist( let file_lock = cfg.file_locks.write_lock(&m3u_path).await; - let persist_result = persist_m3u_playlist_as_text(&cfg.config.load(), target, target_output, Arc::clone(&m3u_playlist)).await; + if let Err(err) = persist_m3u_playlist_as_text(&cfg.config.load(), target, target_output, Arc::clone(&m3u_playlist)).await { + error!("Persisting m3u playlist failed: {err}"); + } let playlist = Arc::clone(&m3u_playlist); let m3u_path_clone = m3u_path.clone(); @@ -114,7 +117,7 @@ pub async fn m3u_write_playlist( .await .map_err(|err| create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {err}", m3u_path.display()))??; - persist_result + Ok(()) } pub async fn m3u_load_rewrite_playlist( diff --git a/bin/build_docker.sh b/bin/build_docker.sh index 069d5b75b..e35eb0915 100755 --- a/bin/build_docker.sh +++ b/bin/build_docker.sh @@ -16,8 +16,8 @@ print_usage() { echo " branch Git branch name (only 'master' and 'develop' are supported)" echo echo "Examples:" - echo " $(basename "$0") master # Builds and pushes with :latest tag" - echo " $(basename "$0") develop # Builds and pushes with :dev tag" + echo " $(basename "$0") master # Builds and pushes with :VERSION and :latest tags" + echo " $(basename "$0") develop # Builds and pushes with :VERSION and :dev tags" exit 1 }