From 0fae9f3edd015923910601d17e3415df540d67ca Mon Sep 17 00:00:00 2001 From: euzu Date: Sun, 30 Nov 2025 13:12:59 +0100 Subject: [PATCH] Atomic ordering changed, Fixed race conditions on user access --- .../src/api/model/provider_lineup_manager.rs | 4 +-- .../src/api/model/streams/client_stream.rs | 2 +- .../api/model/streams/persist_pipe_stream.rs | 2 +- backend/src/api/scheduler.rs | 2 +- backend/src/processing/processor/playlist.rs | 2 +- backend/src/tools/atomic_once_flag.rs | 2 +- backend/src/utils/file/file_lock_manager.rs | 13 ------- frontend/src/services/event_service.rs | 6 ++-- frontend/src/services/toastr_service.rs | 2 +- frontend/src/services/websocket_service.rs | 34 +++++++++---------- 10 files changed, 28 insertions(+), 41 deletions(-) diff --git a/backend/src/api/model/provider_lineup_manager.rs b/backend/src/api/model/provider_lineup_manager.rs index c7aa77567..b25455cce 100644 --- a/backend/src/api/model/provider_lineup_manager.rs +++ b/backend/src/api/model/provider_lineup_manager.rs @@ -397,7 +397,7 @@ impl MultiProviderLineup { /// ``` async fn acquire(&self, with_grace: bool, grace_period_timeout_secs: u64) -> ProviderAllocation { let provider_count = self.providers.len(); - let start = self.index.fetch_add(1, Ordering::Acquire) % provider_count; + let start = self.index.fetch_add(1, Ordering::AcqRel) % provider_count; let mut idx = start; loop { @@ -435,7 +435,7 @@ impl MultiProviderLineup { async fn get_next(&self, grace_period_timeout_secs: u64) -> Option> { let provider_count = self.providers.len(); - let start = self.index.fetch_add(1, Ordering::Acquire) % provider_count; + let start = self.index.fetch_add(1, Ordering::AcqRel) % provider_count; let mut idx = start; loop { diff --git a/backend/src/api/model/streams/client_stream.rs b/backend/src/api/model/streams/client_stream.rs index 547d955f9..64f0c5fd7 100644 --- a/backend/src/api/model/streams/client_stream.rs +++ b/backend/src/api/model/streams/client_stream.rs @@ -41,7 +41,7 @@ impl Stream for ClientStream { // Empty payload signals upstream closure; notify and let consumer see final chunk self.close_signal.notify(); } else if let Some(counter) = self.total_bytes.as_ref() { - counter.fetch_add(bytes.len(), Ordering::Relaxed); + counter.fetch_add(bytes.len(), Ordering::AcqRel); } Poll::Ready(Some(Ok(bytes))) diff --git a/backend/src/api/model/streams/persist_pipe_stream.rs b/backend/src/api/model/streams/persist_pipe_stream.rs index 91b4bac9f..fd0dfb4c7 100644 --- a/backend/src/api/model/streams/persist_pipe_stream.rs +++ b/backend/src/api/model/streams/persist_pipe_stream.rs @@ -50,7 +50,7 @@ where let chunk_len = chunk.len(); if self.current_offset >= chunk_len { if let Some(finished) = self.pending_writes.pop_front() { - self.size.fetch_add(finished.len(), Ordering::Acquire); + self.size.fetch_add(finished.len(), Ordering::AcqRel); } self.current_offset = 0; continue; diff --git a/backend/src/api/scheduler.rs b/backend/src/api/scheduler.rs index 3cfbfddab..a29b85b27 100644 --- a/backend/src/api/scheduler.rs +++ b/backend/src/api/scheduler.rs @@ -119,7 +119,7 @@ mod tests { let expression = "0/1 * * * * * *"; // every second let runs = AtomicU8::new(0); - let run_me = || runs.fetch_add(1, Ordering::Acquire); + let run_me = || runs.fetch_add(1, Ordering::AcqRel); let start = std::time::Instant::now(); if let Ok(schedule) = Schedule::from_str(expression) { diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 30d1d16fc..ec2f5ee1e 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -241,7 +241,7 @@ fn map_playlist_counter(target: &ConfigTarget, playlist: &mut [PlaylistGroup]) { for channel in &mut plg.channels { let provider = ValueProvider { pli: channel }; if counter.filter.filter(&provider) { - let cntval = counter.value.fetch_add(1, core::sync::atomic::Ordering::Acquire); + let cntval = counter.value.fetch_add(1, core::sync::atomic::Ordering::AcqRel); let padded_cntval = if counter.padding > 0 { format!("{:0width$}", cntval, width = counter.padding as usize) } else { diff --git a/backend/src/tools/atomic_once_flag.rs b/backend/src/tools/atomic_once_flag.rs index cf549da9d..6b2903eba 100644 --- a/backend/src/tools/atomic_once_flag.rs +++ b/backend/src/tools/atomic_once_flag.rs @@ -1,7 +1,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; /// A flag that is initially active (`true`) and can only be disabled once. -/// Once the flag is disabled by calling `disable()`, it remains inactive (`false`) forever. +/// Once the flag is disabled by calling [`notify()`], it remains inactive (`false`) forever. /// /// ## Use Case /// This type is useful when you need a one-way toggle to mark a resource or state as "finalized", diff --git a/backend/src/utils/file/file_lock_manager.rs b/backend/src/utils/file/file_lock_manager.rs index 943107bc6..b2fd1c164 100644 --- a/backend/src/utils/file/file_lock_manager.rs +++ b/backend/src/utils/file/file_lock_manager.rs @@ -133,16 +133,3 @@ fn normalize_path(path: &Path) -> PathBuf { base.clean() } - - -#[cfg(test)] -mod test { - use super::*; - #[test] - fn test_normalize_path() { - let path = PathBuf::from("https://10.1.41.41"); - let normalized = normalize_path(&path); - assert_eq!(normalized.display().to_string(), "dsd sdf sf d"); - - } -} \ No newline at end of file diff --git a/frontend/src/services/event_service.rs b/frontend/src/services/event_service.rs index 459cff90b..f3f06eb26 100644 --- a/frontend/src/services/event_service.rs +++ b/frontend/src/services/event_service.rs @@ -37,8 +37,8 @@ impl EventService { pub fn set_config_change_message_blocked(&self, value: bool) { if value { // Re-block and bump epoch to invalidate any pending unblocks. - self.block_config_updated_message.store(true, Ordering::Relaxed); - self.block_epoch.fetch_add(1, Ordering::Relaxed); + self.block_config_updated_message.store(true, Ordering::Release); + self.block_epoch.fetch_add(1, Ordering::AcqRel); } else { let flag = Rc::clone(&self.block_config_updated_message); let epoch_now = self.block_epoch.load(Ordering::Relaxed); @@ -53,7 +53,7 @@ impl EventService { } pub fn subscribe(&self, callback: F) -> usize { - let sub_id = self.subscriber_id.fetch_add(1, Ordering::Acquire); + let sub_id = self.subscriber_id.fetch_add(1, Ordering::AcqRel); self.subscribers.borrow_mut().insert(sub_id, Box::new(callback)); sub_id } diff --git a/frontend/src/services/toastr_service.rs b/frontend/src/services/toastr_service.rs index 2fa1d7670..21c7a0736 100644 --- a/frontend/src/services/toastr_service.rs +++ b/frontend/src/services/toastr_service.rs @@ -93,7 +93,7 @@ impl ToastrService { ) { let mut state = self.state.borrow_mut(); let toast = Toast { - id: self.counter.fetch_add(1, Ordering::Acquire), + id: self.counter.fetch_add(1, Ordering::AcqRel), message: msg.into(), toast_type: toast_type.clone(), close_mode: options.close_mode, diff --git a/frontend/src/services/websocket_service.rs b/frontend/src/services/websocket_service.rs index d0cb15073..90a702d56 100644 --- a/frontend/src/services/websocket_service.rs +++ b/frontend/src/services/websocket_service.rs @@ -1,8 +1,7 @@ use wasm_bindgen::JsCast; use web_sys::{WebSocket, MessageEvent, Event, ErrorEvent, CloseEvent}; -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::rc::Rc; -use std::sync::atomic::{AtomicBool, AtomicU16, Ordering}; use web_sys::js_sys::{Uint8Array, ArrayBuffer}; use log::{error, trace, warn}; use shared::model::{ProtocolMessage, PROTOCOL_VERSION}; @@ -14,11 +13,11 @@ use wasm_bindgen::closure::Closure; const WS_RECONNECT_BASE_MS: u32 = 300; const WS_RECONNECT_MAX_MS: u32 = 2000; -const WS_RECONNECT_MAX_ATTEMPTS: u32 = 20; +const WS_RECONNECT_MAX_ATTEMPTS: u16 = 20; -fn reconnect_delay(attempt: u32) -> u32 { +fn reconnect_delay(attempt: u16) -> u32 { if attempt < 6 { - let d = WS_RECONNECT_BASE_MS * (attempt+1); + let d = WS_RECONNECT_BASE_MS * (attempt as u32 +1u32); d.min(WS_RECONNECT_MAX_MS) } else { WS_RECONNECT_MAX_MS @@ -31,8 +30,8 @@ type JsOnErrorCallback = Option>; type JsOnOpenCallback = Option>; pub struct WebSocketService { - connected: Rc, - attempt_counter: Rc, + connected: Rc>, + attempt_counter: Rc>, ws: Rc>>, status_service: Rc, event_service: Rc, @@ -49,8 +48,8 @@ impl WebSocketService { pub fn new(status_service: Rc, event_service: Rc) -> Self { let base_href = get_base_href(); Self { - connected: Rc::new(AtomicBool::new(false)), - attempt_counter: Rc::new(AtomicU16::new(0)), + connected: Rc::new(Cell::new(false)), + attempt_counter: Rc::new(Cell::new(0)), ws: Rc::new(RefCell::new(None)), status_service, event_service, @@ -79,7 +78,7 @@ impl WebSocketService { } pub fn connect_ws_with_backoff(&self) { - if self.connected.load(Ordering::Acquire) { + if self.connected.get() { return; } match WebSocket::new(&self.ws_path) { @@ -115,7 +114,7 @@ impl WebSocketService { // on open is called on a connect attempt, it does not mean it is connected! trace!("WebSocket connection opened."); if Self::try_send_message(ws_open_clone.borrow().as_ref(), ProtocolMessage::Version(PROTOCOL_VERSION)) { - connected_clone.store(true, Ordering::Release); + connected_clone.set(true); } })); socket.set_onopen(Some(onopen_callback.as_ref().unchecked_ref())); @@ -158,7 +157,7 @@ impl WebSocketService { // it will be dropped when the service or field is taken elsewhere if desired. *ws_close_rc.borrow_mut() = None; - connected_clone.store(false, Ordering::Release); + connected_clone.set(false); event_service_clone.broadcast(EventMessage::WebSocketStatus(false)); // schedule reconnect @@ -177,7 +176,7 @@ impl WebSocketService { let onerror_callback = Closure::::wrap(Box::new(move |e: ErrorEvent| { error!("WebSocket error: {:?}", e); - connected_clone.store(false, Ordering::Release); + connected_clone.set(false); event_service_clone.broadcast(EventMessage::WebSocketStatus(false)); // ws_service_reconnect_clone.schedule_reconnect(); })); @@ -190,7 +189,8 @@ impl WebSocketService { fn schedule_reconnect(&self) { // increment attempts atomically and get the previous value - let attempt = self.attempt_counter.fetch_add(1, Ordering::Acquire) as u32; + let attempt = self.attempt_counter.get() + 1; + self.attempt_counter.set(attempt); if attempt >= WS_RECONNECT_MAX_ATTEMPTS { warn!("WebSocket reconnect attempts exceeded ({attempt}). Giving up."); @@ -230,7 +230,7 @@ impl WebSocketService { } pub async fn get_server_status(&self) { - if self.connected.load(Ordering::Acquire) { + if self.connected.get() { if let Some(token) = get_token() { self.send_message(ProtocolMessage::StatusRequest(token)); } @@ -248,7 +248,7 @@ impl WebSocketService { } } -fn handle_socket_protocol_msg(event: MessageEvent, event_service: &Rc, attempt_counter: &Rc) -> Option{ +fn handle_socket_protocol_msg(event: MessageEvent, event_service: &Rc, attempt_counter: &Rc>) -> Option{ if let Ok(buf) = event.data().dyn_into::() { let array = Uint8Array::new(&buf); let bytes = bytes::Bytes::from(array.to_vec()); @@ -295,7 +295,7 @@ fn handle_socket_protocol_msg(event: MessageEvent, event_service: &Rc { - attempt_counter.store(0, Ordering::Release); + attempt_counter.set(0); event_service.broadcast(EventMessage::WebSocketStatus(true)); if let Some(token) = get_token() { return Some(ProtocolMessage::Auth(token));