diff --git a/README.md b/README.md
index e284f51ed..25843ac52 100644
--- a/README.md
+++ b/README.md
@@ -714,8 +714,8 @@ Attributes:
 - `retry`, default `true`, retries the stream if it fails.
 - `buffer`
 - `throttle` Allowed units are `KB/s`,`MB/s`,`KiB/s`,`MiB/s`,`kbps`,`mbps`,`Mibps`. Default unit is `kbps`
-- `grace_period_millis`  default set to 300 milliseconds.
-- `grace_period_timeout_secs` default set to 2 seconds.
+- `grace_period_millis`  default set to 2000 milliseconds.
+- `grace_period_timeout_secs` default set to 4 seconds.
 - `grace_period_hold_stream` if set to `true`, the stream will only start after the grace period check has completed. Default is `true`.
 - `shared_burst_buffer_mb` optional (default `12`). Minimum burst buffer size (in MB) used for shared streams.
 
@@ -765,12 +765,12 @@ Default unit is `kbps`.
 If you have a provider or a user where the max_connection attribute is greater than 0,
 a grace period can be given during the switchover.
 If this period is set too short, it may result in access being denied in some cases.
-The default is 300 milliseconds.
+The default is 2000 milliseconds.
 If the connection is not throttled, the player will play its buffered content longer than expected.
 
 ##### 1.7.1.4 `grace_period_timeout_secs`
 
-How long the grace grant will last, until another grace grant can made. Default is `2` seconds.
+How long the grace grant will last, until another grace grant can made. Default is `4` seconds.
 
 ##### 1.7.1.5 `grace_period_hold_stream`
 
@@ -1057,14 +1057,14 @@ If you want to send a picture instead of black screen when a channel is not avai
 Following attributes are available:
 
 - `channel_unavailable`: _optional_
-- `user_account_expired`: _optional_
 - `user_connections_exhausted`: _optional_
 - `provider_connections_exhausted`: _optional_
-- `panel_api_provisioning`: _optional_
 - `low_priority_preempted`: _optional_
+- `user_account_expired`: _optional_
+- `panel_api_provisioning`: _optional_
 
-Video files with name `channel_unavailable.ts`, `user_account_expired.ts`, `user_connections_exhausted.ts`,  
-`provider_connections_exhausted.ts`, `panel_api_provisioning.ts`, `low_priority_preempted.ts`  
+Video files with name `channel_unavailable.ts`, `user_connections_exhausted.ts`, `provider_connections_exhausted.ts`,  
+`low_priority_preempted.ts`, `user_account_expired.ts`, `panel_api_provisioning.ts`  
 are already available in the docker image.
 
 You can convert an image with `ffmpeg`.
@@ -1077,10 +1077,10 @@ and add it to the `config.yml`.
 
 - `user_account_expired.ts`
 - `provider_connections_exhausted.ts`
+- `low_priority_preempted.ts`
 - `user_connections_exhausted.ts`
 - `channel_unavailable.ts`
 - `panel_api_provisioning.ts`
-- `low_priority_preempted.ts`
 
 ```yaml
 custom_stream_response_path: /home/tuliprox/resources
diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs
index a162ada74..19cc59637 100644
--- a/backend/src/api/main_api.rs
+++ b/backend/src/api/main_api.rs
@@ -37,6 +37,7 @@ use axum::{
     middleware::Next,
     Router,
 };
+use dashmap::DashSet;
 use log::{debug, error, info, warn};
 use shared::{
     error::TuliproxError,
@@ -58,16 +59,14 @@ const METADATA_TRIGGER_WAIT_CYCLE_LIMIT: u32 = 900;
 
 fn collect_rescheduled_targets(
     targets_set: &HashSet<String>,
-    running_trigger_targets: &Arc<std::sync::Mutex<HashSet<String>>>,
-    pending_trigger_targets: &Arc<std::sync::Mutex<HashSet<String>>>,
+    running_trigger_targets: &Arc<DashSet<String>>,
+    pending_trigger_targets: &Arc<DashSet<String>>,
 ) -> Vec<String> {
-    let mut running = running_trigger_targets.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
-    let mut pending = pending_trigger_targets.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
     let mut targets_to_respawn = Vec::new();
     for target in targets_set {
-        running.remove(target);
-        if pending.remove(target) {
-            running.insert(target.clone());
+        running_trigger_targets.remove(target);
+        if pending_trigger_targets.remove(target).is_some() {
+            running_trigger_targets.insert(target.clone());
             targets_to_respawn.push(target.clone());
         }
     }
@@ -77,8 +76,8 @@ fn collect_rescheduled_targets(
 fn spawn_metadata_trigger_update(
     app_state: &Arc<AppState>,
     targets_to_spawn: &[String],
-    running_trigger_targets: &Arc<std::sync::Mutex<HashSet<String>>>,
-    pending_trigger_targets: &Arc<std::sync::Mutex<HashSet<String>>>,
+    running_trigger_targets: &Arc<DashSet<String>>,
+    pending_trigger_targets: &Arc<DashSet<String>>,
 ) {
     if targets_to_spawn.is_empty() {
         return;
@@ -628,13 +627,13 @@ fn exec_input_update_listener(app_state: &Arc<AppState>, targets: &Arc<ProcessTa
     tokio::spawn(async move {
         let mut rx = app_state.event_manager.get_event_channel();
         // Map<TargetName, Set<InputName>>: Tracks which inputs are currently updating for a given target
-        let mut active_target_inputs: HashMap<String, HashSet<String>> = HashMap::new();
+        let mut active_target_inputs: HashMap<String, HashSet<Arc<str>>> = HashMap::new();
         // Set<TargetName>: Tracks which targets are pending an update once their inputs are done
         let mut pending_targets: HashSet<String> = HashSet::new();
         // Set<TargetName>: Tracks which targets have an active spawned playlist-update task (dedup guard)
-        let running_trigger_targets: Arc<std::sync::Mutex<HashSet<String>>> = Arc::new(std::sync::Mutex::new(HashSet::new()));
+        let running_trigger_targets: Arc<DashSet<String>> = Arc::new(DashSet::new());
         // Set<TargetName>: Tracks follow-up triggers that arrive while a target is already running.
-        let pending_trigger_targets: Arc<std::sync::Mutex<HashSet<String>>> = Arc::new(std::sync::Mutex::new(HashSet::new()));
+        let pending_trigger_targets: Arc<DashSet<String>> = Arc::new(DashSet::new());
 
         loop {
             match rx.recv().await {
@@ -651,7 +650,7 @@ fn exec_input_update_listener(app_state: &Arc<AppState>, targets: &Arc<ProcessTa
                                 active_target_inputs
                                     .entry(target.name.clone())
                                     .or_default()
-                                    .insert(input_name.to_string());
+                                    .insert(input_name.clone());
                                 // Mark target as potentially needing an update
                                 pending_targets.insert(target.name.clone());
                             }
@@ -659,37 +658,29 @@ fn exec_input_update_listener(app_state: &Arc<AppState>, targets: &Arc<ProcessTa
                     }
                 }
                 Ok(EventMessage::InputMetadataUpdatesCompleted(input_name)) => {
-                    // 1. Remove this input from all active sets
-                    let target_names: Vec<String> = active_target_inputs.keys().cloned().collect();
                     let mut targets_to_trigger = Vec::new();
-
-                    for target_name in target_names {
-                        if let std::collections::hash_map::Entry::Occupied(mut entry) =
-                            active_target_inputs.entry(target_name.clone())
-                        {
-                            let inputs = entry.get_mut();
-                            inputs.remove(input_name.as_ref());
-                            if inputs.is_empty() {
-                                entry.remove();
-                                // If this target was pending, it's now ready to trigger
-                                if pending_targets.remove(&target_name) {
-                                    targets_to_trigger.push(target_name);
-                                }
+                    // Remove this input from all active sets in-place and collect targets
+                    // that became ready without cloning all keys first.
+                    active_target_inputs.retain(|target_name, inputs| {
+                        inputs.remove(&input_name);
+                        if inputs.is_empty() {
+                            if pending_targets.remove(target_name) {
+                                targets_to_trigger.push(target_name.clone());
                             }
+                            false
+                        } else {
+                            true
                         }
-                    }
+                    });
 
                     // Deduplicate: only spawn for targets that don't already have a running trigger task
                     let targets_to_spawn: Vec<String> = {
-                        let mut running = running_trigger_targets.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
-                        let mut pending =
-                            pending_trigger_targets.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
                         let mut to_spawn = Vec::new();
                         for target in targets_to_trigger {
-                            if running.insert(target.clone()) {
+                            if running_trigger_targets.insert(target.clone()) {
                                 to_spawn.push(target);
                             } else {
-                                pending.insert(target);
+                                pending_trigger_targets.insert(target);
                             }
                         }
                         to_spawn
diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs
index 4ba14d1ec..90eb71fd5 100644
--- a/backend/src/api/model/active_provider_manager.rs
+++ b/backend/src/api/model/active_provider_manager.rs
@@ -9,7 +9,7 @@ use crate::{
 use log::{error, warn};
 use shared::utils::sanitize_sensitive_info;
 use std::cmp::Reverse;
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::{
     net::SocketAddr,
     sync::{
@@ -75,18 +75,18 @@ struct SharedAllocation {
 #[derive(Debug, Clone)]
 struct ActiveConnectionInfo {
     allocation: ProviderAllocation,
-    priority: i8,
-    is_probe: bool,
     // Used to signal preemption to the consumer of this connection
     cancel_token: CancellationToken,
     created_at: Instant,
+    priority: i8,
+    is_probe: bool,
 }
 
 #[derive(Debug, Clone, Default)]
 struct SharedConnections {
-    by_key: HashMap<String, SharedAllocation>,
-    key_by_addr: HashMap<ClientConnectionId, String>,
-    shared_by_allocation_id: HashMap<AllocationId, String>,
+    by_key: HashMap<Arc<str>, SharedAllocation>,
+    key_by_addr: HashMap<ClientConnectionId, Arc<str>>,
+    shared_by_allocation_id: HashMap<AllocationId, Arc<str>>,
 }
 
 #[derive(Debug, Clone, Default)]
@@ -95,8 +95,8 @@ struct Connections {
     single: HashMap<ClientConnectionId, HashMap<AllocationId, ActiveConnectionInfo>>,
     shared: SharedConnections,
     // Index to quickly find connections by provider name for preemption
-    // ProviderName -> Vec<(ClientConnectionId, AllocationId)>
-    by_provider: HashMap<Arc<str>, Vec<(ClientConnectionId, AllocationId)>>,
+    // ProviderName -> Set<(ClientConnectionId, AllocationId)>
+    by_provider: HashMap<Arc<str>, HashSet<(ClientConnectionId, AllocationId)>>,
     // Priority index per provider alias for O(log n) victim lookup
     // ProviderName -> BTreeMap<PriorityKey, PriorityOwner>
     priority_index: HashMap<Arc<str>, BTreeMap<PriorityKey, PriorityOwner>>,
@@ -248,14 +248,14 @@ impl ActiveProviderManager {
             allocation_id,
             ActiveConnectionInfo {
                 allocation: allocation.clone(),
-                priority,
-                is_probe,
                 cancel_token: cancel_token.clone(),
                 created_at: now,
+                priority,
+                is_probe,
             },
         );
 
-        connections.by_provider.entry(provider_name.clone()).or_default().push((*addr, allocation_id));
+        connections.by_provider.entry(provider_name.clone()).or_default().insert((*addr, allocation_id));
         connections.priority_index.entry(provider_name.clone()).or_default()
             .insert((priority, Reverse(now), allocation_id), PriorityOwner::Single(*addr));
 
@@ -403,9 +403,7 @@ impl ActiveProviderManager {
                     }
                     if let Some(ref name) = removed_provider_name {
                         if let Some(list) = connections.by_provider.get_mut(name) {
-                            if let Some(idx) = list.iter().position(|(a, i)| *a == addr && *i == alloc_id) {
-                                list.remove(idx);
-                            }
+                            list.remove(&(addr, alloc_id));
                         }
                         if let Some(tree) = connections.priority_index.get_mut(name) {
                             tree.remove(&(v_prio, Reverse(victim_created_at), alloc_id));
@@ -577,9 +575,7 @@ impl ActiveProviderManager {
                         }
                         if let Some(name) = removed_provider_name {
                             if let Some(list) = connections.by_provider.get_mut(&name) {
-                                if let Some(idx) = list.iter().position(|(a, i)| *a == addr && *i == alloc_id) {
-                                    list.remove(idx);
-                                }
+                                list.remove(&(addr, alloc_id));
                             }
                             if let Some(tree) = connections.priority_index.get_mut(&name) {
                                 tree.remove(&(v_prio, Reverse(victim_created_at), alloc_id));
@@ -700,9 +696,7 @@ impl ActiveProviderManager {
                 for (id, info) in &allocations {
                     if let Some(name) = info.allocation.get_provider_name() {
                         if let Some(list) = connections.by_provider.get_mut(&name) {
-                            if let Some(idx) = list.iter().position(|(a, i)| *a == *addr && *i == *id) {
-                                list.remove(idx);
-                            }
+                            list.remove(&(*addr, *id));
                         }
                         if let Some(tree) = connections.priority_index.get_mut(&name) {
                             tree.remove(&(info.priority, Reverse(info.created_at), *id));
@@ -812,11 +806,7 @@ impl ActiveProviderManager {
                     // Remove from by_provider index
                     if let Some(name) = released.as_ref().and_then(ProviderAllocation::get_provider_name) {
                         if let Some(list) = connections.by_provider.get_mut(&name) {
-                            if let Some(idx) =
-                                list.iter().position(|(a, i)| *a == handle.client_id && *i == handle.allocation_id)
-                            {
-                                list.remove(idx);
-                            }
+                            list.remove(&(handle.client_id, handle.allocation_id));
                         }
                         released_priority_key = Some((name, pkey));
                     }
@@ -859,6 +849,7 @@ impl ActiveProviderManager {
         let extras = {
             let mut connections = self.connections.write().await;
             let mut extras = Vec::new();
+            let shared_key: Arc<str> = Arc::from(key);
 
             // Find the allocation to promote (must be single)
             // Logic change: we must find the specific allocation if multiple exist, but usually per client only 1 active?
@@ -875,9 +866,7 @@ impl ActiveProviderManager {
                         // Cleanup indices
                         if let Some(name) = info.allocation.get_provider_name() {
                             if let Some(list) = connections.by_provider.get_mut(&name) {
-                                if let Some(idx) = list.iter().position(|(a, alloc_id)| *a == *addr && *alloc_id == id) {
-                                    list.remove(idx);
-                                }
+                                list.remove(&(*addr, id));
                             }
                             // Remove old single entry from priority_index
                             if let Some(tree) = connections.priority_index.get_mut(&name) {
@@ -888,12 +877,7 @@ impl ActiveProviderManager {
                         for (extra_id, extra_info) in &extra_entries {
                             if let Some(extra_provider_name) = extra_info.allocation.get_provider_name() {
                                 if let Some(list) = connections.by_provider.get_mut(&extra_provider_name) {
-                                    if let Some(idx) = list
-                                        .iter()
-                                        .position(|(extra_addr, alloc_id)| *extra_addr == *addr && *alloc_id == *extra_id)
-                                    {
-                                        list.remove(idx);
-                                    }
+                                    list.remove(&(*addr, *extra_id));
                                 }
                                 if let Some(tree) = connections.priority_index.get_mut(&extra_provider_name) {
                                     tree.remove(&(extra_info.priority, Reverse(extra_info.created_at), *extra_id));
@@ -902,6 +886,7 @@ impl ActiveProviderManager {
                         }
 
                         for (_, extra_info) in extra_entries {
+                            extra_info.cancel_token.cancel();
                             extras.push(extra_info.allocation);
                         }
 
@@ -929,7 +914,7 @@ impl ActiveProviderManager {
                 );
 
                 connections.shared.by_key.insert(
-                    key.to_string(),
+                    Arc::clone(&shared_key),
                     SharedAllocation {
                         allocation_id: handle.0.allocation_id,
                         allocation: handle.0.allocation.clone(),
@@ -939,8 +924,11 @@ impl ActiveProviderManager {
                         cancel_token: handle.0.cancel_token.clone(),
                     },
                 );
-                connections.shared.key_by_addr.insert(*addr, key.to_string());
-                connections.shared.shared_by_allocation_id.insert(handle.0.allocation_id, key.to_string());
+                connections.shared.key_by_addr.insert(*addr, Arc::clone(&shared_key));
+                connections
+                    .shared
+                    .shared_by_allocation_id
+                    .insert(handle.0.allocation_id, shared_key);
 
                 // Insert new shared entry into priority_index
                 connections.priority_index.entry(provider_name.clone()).or_default()
@@ -1004,7 +992,7 @@ impl ActiveProviderManager {
             }
         }
 
-        connections.shared.key_by_addr.insert(*addr, key.to_string());
+        connections.shared.key_by_addr.insert(*addr, Arc::from(key));
         Ok(())
     }
 
diff --git a/backend/src/api/model/connection_manager.rs b/backend/src/api/model/connection_manager.rs
index e022042ff..5ffef316f 100644
--- a/backend/src/api/model/connection_manager.rs
+++ b/backend/src/api/model/connection_manager.rs
@@ -14,6 +14,7 @@ use shared::{
 use std::{borrow::Cow, net::SocketAddr, sync::Arc};
 use tokio::sync::{mpsc, Notify};
 
+const CLEANUP_QUEUE_CAPACITY: usize = 4096;
 fn notify_capacity(capacity_notify: &Notify) { capacity_notify.notify_waiters(); }
 
 pub(crate) enum CleanupEvent {
@@ -37,7 +38,7 @@ pub struct ConnectionManager {
     pub shared_stream_manager: Arc<SharedStreamManager>,
     event_manager: Arc<EventManager>,
     close_socket_signal_tx: tokio::sync::broadcast::Sender<SocketAddr>,
-    cleanup_tx: mpsc::UnboundedSender<CleanupEvent>,
+    cleanup_tx: mpsc::Sender<CleanupEvent>,
     capacity_notify: Arc<Notify>,
 }
 
@@ -49,7 +50,7 @@ impl ConnectionManager {
         event_manager: &Arc<EventManager>,
     ) -> Self {
         let (close_socket_signal_tx, _) = tokio::sync::broadcast::channel(256);
-        let (cleanup_tx, cleanup_rx) = mpsc::unbounded_channel();
+        let (cleanup_tx, cleanup_rx) = mpsc::channel(CLEANUP_QUEUE_CAPACITY);
         let capacity_notify = Arc::new(Notify::new());
 
         let mgr = Self {
@@ -75,7 +76,7 @@ impl ConnectionManager {
     }
 
     fn spawn_cleanup_worker(
-        mut rx: mpsc::UnboundedReceiver<CleanupEvent>,
+        mut rx: mpsc::Receiver<CleanupEvent>,
         user_manager: Arc<ActiveUserManager>,
         provider_manager: Arc<ActiveProviderManager>,
         shared_stream_manager: Arc<SharedStreamManager>,
@@ -103,15 +104,21 @@ impl ConnectionManager {
                         // Release provider handle first to avoid a race window where the user
                         // connection count drops (making capacity appear available) before the
                         // provider slot is actually freed.
-                        if let Some(h) = handle {
+                        let provider_released = if let Some(h) = handle {
                             provider_manager.release_handle(&h).await;
-                        }
-                        if user_manager.release_stream(&addr).await {
+                            true
+                        } else {
+                            false
+                        };
+                        let stream_released = user_manager.release_stream(&addr).await;
+                        if stream_released {
                             event_manager.send_event(EventMessage::ActiveUser(
                                 ActiveUserConnectionChange::Disconnected(addr),
                             ));
                         }
-                        notify_capacity(capacity_notify.as_ref());
+                        if provider_released || stream_released {
+                            notify_capacity(capacity_notify.as_ref());
+                        }
                     }
                     CleanupEvent::UpdateDetailAndReleaseProvider { addr, video_type, handle } => {
                         if let Some(stream_info) = user_manager.update_stream_detail(&addr, video_type).await {
@@ -141,8 +148,14 @@ impl ConnectionManager {
     }
 
     pub(crate) fn send_cleanup(&self, event: CleanupEvent) {
-        if let Err(e) = self.cleanup_tx.send(event) {
-            debug!("Cleanup channel closed, dropping event: {e}");
+        match self.cleanup_tx.try_send(event) {
+            Ok(()) => {}
+            Err(tokio::sync::mpsc::error::TrySendError::Full(_event)) => {
+                warn!("Cleanup queue full, dropping event");
+            }
+            Err(tokio::sync::mpsc::error::TrySendError::Closed(_event)) => {
+                debug!("Cleanup channel closed, dropping cleanup event");
+            }
         }
     }
 
@@ -176,26 +189,26 @@ impl ConnectionManager {
         if removed {
             self.event_manager.send_event(EventMessage::ActiveUser(ActiveUserConnectionChange::Disconnected(*addr)));
         }
-        self.capacity_notify.notify_waiters();
+        notify_capacity(self.capacity_notify.as_ref());
     }
 
     pub async fn release_provider_connection(&self, addr: &SocketAddr) {
         self.provider_manager.release_connection(addr).await;
         self.shared_stream_manager.release_connection(addr, false).await;
-        self.capacity_notify.notify_waiters();
+        notify_capacity(self.capacity_notify.as_ref());
     }
 
     pub async fn release_stream(&self, addr: &SocketAddr) {
         if self.user_manager.release_stream(addr).await {
             self.event_manager.send_event(EventMessage::ActiveUser(ActiveUserConnectionChange::Disconnected(*addr)));
-            self.capacity_notify.notify_waiters();
+            notify_capacity(self.capacity_notify.as_ref());
         }
     }
 
     pub async fn release_provider_handle(&self, provider_handle: Option<ProviderHandle>) {
         if let Some(handle) = provider_handle {
             self.provider_manager.release_handle(&handle).await;
-            self.capacity_notify.notify_waiters();
+            notify_capacity(self.capacity_notify.as_ref());
         }
     }
 
diff --git a/backend/src/api/model/stream_error.rs b/backend/src/api/model/stream_error.rs
index 32d81c0d3..ae3222c8f 100644
--- a/backend/src/api/model/stream_error.rs
+++ b/backend/src/api/model/stream_error.rs
@@ -29,7 +29,7 @@ impl std::fmt::Display for StreamError {
             StreamError::StdIo(e) => write!(f, "IO error: {e}"),
             // StreamError::ReceiverClosed =>  write!(f, "Receiver closed"),
             StreamError::ReceiverError(e) => write!(f, "Receiver error {e}"),
-            StreamError::Stream(e) => write!(f, "Stream error: {e}"),
+            StreamError::Stream(e) => write!(f, "Stream: {e}"),
             StreamError::LockError(e) => write!(f, "LockError: {e}"),
             StreamError::MalformedPacket(e) => write!(f, "MalformedPacket: {e}"),
             StreamError::InvalidTimestamp(e) => write!(f, "InvalidTimestamp: {e}"),
diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs
index 3349e6182..0a1ca645f 100644
--- a/backend/src/api/model/streams/active_client_stream.rs
+++ b/backend/src/api/model/streams/active_client_stream.rs
@@ -157,7 +157,6 @@ impl ActiveClientStreamState {
         self.provider_stopped = true;
         self.preempt_cancelled = None;
         self.stop_grace_task();
-        self.release_user_stream();
 
         let mut serve_preempted_custom = false;
         if self.provider_handle.is_some() {
@@ -188,11 +187,16 @@ impl ActiveClientStreamState {
                 "Provider stream preempted for {}; stopping client stream",
                 sanitize_sensitive_info(&addr.to_string())
             );
-            self.connection_manager.send_cleanup(CleanupEvent::UpdateDetailAndReleaseProvider {
-                addr,
-                video_type: CustomVideoStreamType::LowPriorityPreempted,
-                handle,
-            });
+            if serve_preempted_custom {
+                self.connection_manager.send_cleanup(CleanupEvent::UpdateDetailAndReleaseProvider {
+                    addr,
+                    video_type: CustomVideoStreamType::LowPriorityPreempted,
+                    handle,
+                });
+            } else {
+                self.release_user_stream();
+                self.connection_manager.send_cleanup(CleanupEvent::ReleaseProviderHandle { handle });
+            }
         }
         serve_preempted_custom
     }
@@ -577,12 +581,20 @@ pub(crate) async fn create_active_client_stream(
         .and_then(|h| h.cancel_token.as_ref())
         .map(|token| Box::pin(token.clone().cancelled_owned()));
 
+    let mut send_custom_stream_flag = grace_stop_flag;
+    if send_custom_stream_flag.is_none()
+        && preempt_cancelled.is_some()
+        && custom_video.low_priority_preempted.is_some()
+    {
+        send_custom_stream_flag = Some(Arc::new(AtomicU8::new(StreamMode::Inner as u8)));
+    }
+
     let state = ActiveClientStreamState {
         inner: stream,
         preempt_cancelled,
         grace_task_handle,
         provider_handle: stream_details.provider_handle,
-        send_custom_stream_flag: grace_stop_flag,
+        send_custom_stream_flag,
         provisionable: has_provisioning,
         custom_video,
         waker: Some(waker),
diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs
index afad5fb0f..5837f6522 100644
--- a/backend/src/api/model/streams/shared_stream_manager.rs
+++ b/backend/src/api/model/streams/shared_stream_manager.rs
@@ -69,7 +69,7 @@ fn convert_stream(stream: BoxStream<Bytes>) -> BoxStream<Result<Bytes, StreamErr
 type SubscriberId = SocketAddr;
 
 struct BurstBuffer {
-    buffer: VecDeque<Arc<Bytes>>,
+    buffer: VecDeque<Bytes>,
     buffer_size: usize,
     current_bytes: usize,
 }
@@ -89,9 +89,9 @@ impl BurstBuffer {
         Self { buffer: VecDeque::with_capacity(buf_size), buffer_size: buf_size, current_bytes: 0 }
     }
 
-    pub fn snapshot(&self) -> VecDeque<Arc<Bytes>> { self.buffer.iter().cloned().collect::<VecDeque<Arc<Bytes>>>() }
+    pub fn snapshot(&self) -> VecDeque<Bytes> { self.buffer.iter().cloned().collect::<VecDeque<Bytes>>() }
 
-    pub fn push(&mut self, packet: Arc<Bytes>) {
+    pub fn push(&mut self, packet: Bytes) {
         while self.current_bytes + packet.len() > self.buffer_size {
             if let Some(popped) = self.buffer.pop_front() {
                 self.current_bytes -= popped.len();
@@ -106,7 +106,7 @@ impl BurstBuffer {
 }
 
 async fn send_burst_buffer(
-    start_buffer: &VecDeque<Arc<Bytes>>,
+    start_buffer: &VecDeque<Bytes>,
     client_tx: &Sender<Bytes>,
     cancellation_token: &CancellationToken,
 ) -> usize {
@@ -115,7 +115,7 @@ async fn send_burst_buffer(
         if cancellation_token.is_cancelled() {
             return sent;
         }
-        if let Err(err) = client_tx.send(buf.as_ref().clone()).await {
+        if let Err(err) = client_tx.send(buf.clone()).await {
             debug!("Failed sending burst-buffer chunk to client: {err}");
             return sent; // stop on send error
         }
@@ -132,6 +132,8 @@ pub struct SharedStreamState {
     headers: Vec<(String, String)>,
     buf_size: usize,
     provider_guard: Option<ProviderHandle>,
+    low_priority_preempted: Option<crate::api::model::TransportStreamBuffer>,
+    preempted_token: CancellationToken,
     subscribers: RwLock<HashMap<SubscriberId, CancellationToken>>,
     broadcaster: tokio::sync::broadcast::Sender<Bytes>,
     stop_token: CancellationToken,
@@ -145,6 +147,7 @@ impl SharedStreamState {
         buf_size: usize,
         provider_guard: Option<ProviderHandle>,
         min_burst_buffer_size: usize,
+        low_priority_preempted: Option<crate::api::model::TransportStreamBuffer>,
     ) -> Self {
         let (broadcaster, _) = tokio::sync::broadcast::channel(buf_size);
         // TODO channel size versus byte size,  channels are chunk sized, burst_buffer byte sized
@@ -153,6 +156,8 @@ impl SharedStreamState {
             headers,
             buf_size,
             provider_guard,
+            low_priority_preempted,
+            preempted_token: CancellationToken::new(),
             subscribers: RwLock::new(HashMap::new()),
             broadcaster,
             stop_token: CancellationToken::new(),
@@ -197,6 +202,8 @@ impl SharedStreamState {
         let mut last_lag_log = Instant::now().checked_sub(Duration::from_secs(10)).unwrap_or_else(Instant::now);
         let mut consecutive_lag_count: u32 = 0;
         let subscriber_buf_size = self.buf_size;
+        let preempted_token = self.preempted_token.clone();
+        let low_priority_preempted = self.low_priority_preempted.clone();
 
         let address = *addr;
         let subscriber_started_at = Instant::now();
@@ -310,7 +317,35 @@ impl SharedStreamState {
                                     .min(1_600);
                                 sleep(Duration::from_millis(backoff_ms)).await;
                             }
-                            Err(_) => break,
+                            Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                                if preempted_token.is_cancelled() {
+                                    if let Some(mut fallback) = low_priority_preempted {
+                                        debug_if_enabled!(
+                                            "Shared stream subscriber {} switching to low_priority_preempted fallback",
+                                            sanitize_sensitive_info(&address.to_string())
+                                        );
+                                        loop {
+                                            if cancel_token.is_cancelled() || client_tx_clone.is_closed() {
+                                                break;
+                                            }
+
+                                            if let Some(chunk) = fallback.next_chunk() {
+                                                if let Err(err) = client_tx.send(chunk).await {
+                                                    debug!(
+                                                        "Shared stream fallback send error for {}: {err}",
+                                                        sanitize_sensitive_info(&address.to_string())
+                                                    );
+                                                    break;
+                                                }
+                                            } else {
+                                                // End of fallback stream; close subscriber task.
+                                                break;
+                                            }
+                                        }
+                                    }
+                                }
+                                break;
+                            }
                         }
                     }
                 }
@@ -392,13 +427,12 @@ impl SharedStreamState {
                                    startup_stats_logged = true;
                                }
                            }
-                           let arc_data = Arc::new(data);
                            {
                              let mut buffer = burst_buffer.write().await;
-                             buffer.push(arc_data.clone());
+                             buffer.push(data.clone());
                            }
 
-                             match sender.send(arc_data.as_ref().clone()) {
+                             match sender.send(data) {
                              Ok(clients) =>  {
                                  if clients == 0 {
                                     debug_if_enabled!("No shared stream subscribers closing {}", sanitize_sensitive_info(&streaming_url));
@@ -446,8 +480,8 @@ impl SharedStreamState {
 
 #[derive(Debug, Clone, Default)]
 struct SharedStreamsRegister {
-    by_key: HashMap<String, Arc<SharedStreamState>>,
-    key_by_addr: HashMap<SubscriberId, String>,
+    by_key: HashMap<Arc<str>, Arc<SharedStreamState>>,
+    key_by_addr: HashMap<SubscriberId, Arc<str>>,
 }
 
 pub struct SharedStreamManager {
@@ -475,7 +509,7 @@ impl SharedStreamManager {
             let remove_keys: Vec<SocketAddr> = shared_streams
                 .key_by_addr
                 .iter()
-                .filter_map(|(addr, url)| if url == stream_url { Some(*addr) } else { None })
+                .filter_map(|(addr, url)| if url.as_ref() == stream_url { Some(*addr) } else { None })
                 .collect();
             for k in remove_keys {
                 shared_streams.key_by_addr.remove(&k);
@@ -524,7 +558,7 @@ impl SharedStreamManager {
             let remove_keys: Vec<SocketAddr> = shared_streams
                 .key_by_addr
                 .iter()
-                .filter_map(|(addr, url)| if url == stream_url { Some(*addr) } else { None })
+                .filter_map(|(addr, url)| if url.as_ref() == stream_url { Some(*addr) } else { None })
                 .collect();
             for k in remove_keys {
                 shared_streams.key_by_addr.remove(&k);
@@ -539,9 +573,7 @@ impl SharedStreamManager {
                 sanitize_sensitive_info(stream_url)
             );
 
-            for handle in shared_state.task_handles.write().await.drain(..) {
-                handle.abort();
-            }
+            shared_state.preempted_token.cancel();
 
             // Cancel stop_token to terminate the broadcast task.
             // Do NOT release provider_guard — the preemption caller already released the allocation.
@@ -610,8 +642,12 @@ impl SharedStreamManager {
     ) -> Option<(BoxedProviderStream, Option<Arc<str>>)> {
         let shared_state_opt = {
             let mut shared_streams = self.shared_streams.write().await;
-            if let Some(shared_state) = shared_streams.by_key.get(stream_url).cloned() {
-                shared_streams.key_by_addr.insert(*addr, stream_url.to_owned());
+            if let Some((stream_key, shared_state)) = shared_streams
+                .by_key
+                .get_key_value(stream_url)
+                .map(|(stream_key, shared_state)| (Arc::clone(stream_key), Arc::clone(shared_state)))
+            {
+                shared_streams.key_by_addr.insert(*addr, stream_key);
                 Some(shared_state)
             } else {
                 None
@@ -632,8 +668,9 @@ impl SharedStreamManager {
 
     async fn register(&self, addr: &SocketAddr, stream_url: &str, shared_state: Arc<SharedStreamState>) {
         let mut shared_streams = self.shared_streams.write().await;
-        shared_streams.by_key.insert(stream_url.to_string(), shared_state);
-        shared_streams.key_by_addr.insert(*addr, stream_url.to_string());
+        let stream_key: Arc<str> = Arc::from(stream_url);
+        shared_streams.by_key.insert(Arc::clone(&stream_key), shared_state);
+        shared_streams.key_by_addr.insert(*addr, stream_key);
         debug_if_enabled!(
             "Registered shared stream {} for initial subscriber {}",
             sanitize_sensitive_info(stream_url),
@@ -660,7 +697,19 @@ impl SharedStreamManager {
         let buf_size = CHANNEL_SIZE.max(buffer_size);
         let config = app_state.app_config.config.load();
         let min_buffer_bytes = resolve_min_burst_buffer_bytes(&config);
-        let shared_state = Arc::new(SharedStreamState::new(headers, buf_size, provider_handle, min_buffer_bytes));
+        let low_priority_preempted = app_state
+            .app_config
+            .custom_stream_response
+            .load()
+            .as_ref()
+            .and_then(|c| c.low_priority_preempted.clone());
+        let shared_state = Arc::new(SharedStreamState::new(
+            headers,
+            buf_size,
+            provider_handle,
+            min_buffer_bytes,
+            low_priority_preempted,
+        ));
         app_state.shared_stream_manager.register(addr, stream_url, Arc::clone(&shared_state)).await;
         app_state.active_provider.make_shared_connection(addr, stream_url).await;
         let subscribed_stream = Self::subscribe_shared_stream(app_state, stream_url, addr, user_priority).await;
@@ -675,6 +724,11 @@ impl SharedStreamManager {
                 "Created shared provider stream {} (channel_capacity={buf_size}, burst_buffer_min={min_buffer_bytes} bytes)",
                 sanitize_sensitive_info(stream_url)
             );
+        } else {
+            warn!(
+                "Shared stream subscribe failed for {}; broadcaster will not start",
+                sanitize_sensitive_info(stream_url)
+            );
         }
         subscribed_stream
     }
@@ -716,11 +770,13 @@ mod tests {
         utils::FileLockManager,
     };
     use arc_swap::{ArcSwap, ArcSwapOption};
+    use futures::StreamExt;
     use shared::{
         model::{ConfigPaths, InputFetchMethod, InputType},
         utils::Internable,
     };
     use std::{collections::HashMap, net::SocketAddr, sync::Arc};
+    use tokio::time::{timeout, Duration};
     use tokio_util::sync::CancellationToken;
 
     fn create_test_app_config() -> AppConfig {
@@ -779,13 +835,19 @@ mod tests {
         let addr_1: SocketAddr = "127.0.0.1:41001".parse().unwrap_or_else(|_| unreachable!());
         let addr_2: SocketAddr = "127.0.0.1:41002".parse().unwrap_or_else(|_| unreachable!());
 
-        let state = Arc::new(SharedStreamState::new(Vec::new(), CHANNEL_SIZE.max(8), None, 1024));
+        let state = Arc::new(SharedStreamState::new(
+            Vec::new(),
+            CHANNEL_SIZE.max(8),
+            None,
+            1024,
+            None,
+        ));
 
         {
             let mut reg = shared_manager.shared_streams.write().await;
-            reg.by_key.insert(stream_url.to_string(), Arc::clone(&state));
-            reg.key_by_addr.insert(addr_1, stream_url.to_string());
-            reg.key_by_addr.insert(addr_2, stream_url.to_string());
+            reg.by_key.insert(Arc::from(stream_url), Arc::clone(&state));
+            reg.key_by_addr.insert(addr_1, Arc::from(stream_url));
+            reg.key_by_addr.insert(addr_2, Arc::from(stream_url));
         }
 
         {
@@ -822,12 +884,18 @@ mod tests {
 
         let stream_url = "https://example.invalid/live/single.ts";
         let addr_1: SocketAddr = "127.0.0.1:42001".parse().unwrap_or_else(|_| unreachable!());
-        let state = Arc::new(SharedStreamState::new(Vec::new(), CHANNEL_SIZE.max(8), None, 1024));
+        let state = Arc::new(SharedStreamState::new(
+            Vec::new(),
+            CHANNEL_SIZE.max(8),
+            None,
+            1024,
+            None,
+        ));
 
         {
             let mut reg = shared_manager.shared_streams.write().await;
-            reg.by_key.insert(stream_url.to_string(), Arc::clone(&state));
-            reg.key_by_addr.insert(addr_1, stream_url.to_string());
+            reg.by_key.insert(Arc::from(stream_url), Arc::clone(&state));
+            reg.key_by_addr.insert(addr_1, Arc::from(stream_url));
         }
 
         {
@@ -857,4 +925,46 @@ mod tests {
             assert!(subs.is_empty(), "duplicate release must keep subscribers empty");
         }
     }
+
+    #[tokio::test]
+    async fn test_preempted_shared_subscriber_switches_to_low_priority_fallback() {
+        let app_cfg = create_test_app_config();
+        let event_manager = Arc::new(EventManager::new());
+        let provider_manager = Arc::new(ActiveProviderManager::new(&app_cfg, &event_manager));
+        let shared_manager = Arc::new(SharedStreamManager::new(provider_manager));
+
+        let addr: SocketAddr = "127.0.0.1:43001".parse().unwrap_or_else(|_| unreachable!());
+
+        // Minimal valid TS packet (sync byte + padding) to feed TransportStreamBuffer.
+        let mut ts_packet = vec![0_u8; 188];
+        ts_packet[0] = 0x47;
+
+        let low_priority_fallback = crate::api::model::TransportStreamBuffer::new(ts_packet);
+        let state = Arc::new(SharedStreamState::new(
+            Vec::new(),
+            CHANNEL_SIZE.max(8),
+            None,
+            1024,
+            Some(low_priority_fallback),
+        ));
+
+        let (mut stream, _provider) = state.subscribe(&addr, Arc::clone(&shared_manager)).await;
+
+        // Simulate preemption teardown semantics:
+        // 1) mark as preempted, 2) drop state so broadcaster sender closes.
+        state.preempted_token.cancel();
+        drop(state);
+
+        let first = timeout(Duration::from_secs(2), stream.next()).await;
+        let maybe_chunk = match first {
+            Ok(v) => v,
+            Err(_) => panic!("timed out waiting for fallback chunk"),
+        };
+        let chunk = match maybe_chunk {
+            Some(Ok(bytes)) => bytes,
+            Some(Err(err)) => panic!("fallback stream returned error: {err}"),
+            None => panic!("fallback stream ended unexpectedly"),
+        };
+        assert!(!chunk.is_empty(), "fallback chunk must contain MPEG-TS bytes");
+    }
 }
diff --git a/backend/src/api/model/streams/transport_stream_buffer.rs b/backend/src/api/model/streams/transport_stream_buffer.rs
index 32daa66f4..262d0c647 100644
--- a/backend/src/api/model/streams/transport_stream_buffer.rs
+++ b/backend/src/api/model/streams/transport_stream_buffer.rs
@@ -288,6 +288,46 @@ pub fn calculate_duration_ticks(buffer: &[u8], packet_indices: &PacketIndices) -
     }
 }
 
+fn calculate_pcr_duration_ticks(buffer: &[u8]) -> Option<u64> {
+    let mut first_pcr = None;
+    let mut last_pcr = None;
+    let mut i = 0;
+    while i + TS_PACKET_SIZE <= buffer.len() {
+        if buffer[i] != SYNC_BYTE {
+            i += 1;
+            continue;
+        }
+        let packet = &buffer[i..i + TS_PACKET_SIZE];
+        let afc = (packet[3] >> 4) & 0b11;
+        if afc == 2 || afc == 3 {
+            let adaptation_len = packet[4] as usize;
+            if adaptation_len >= 7 && (packet[5] & ADAPTATION_FIELD_FLAG_PCR) != 0 {
+                let pcr = decode_pcr(&packet[6..12]);
+                if first_pcr.is_none() {
+                    first_pcr = Some(pcr);
+                }
+                last_pcr = Some(pcr);
+            }
+        }
+        i += TS_PACKET_SIZE;
+    }
+
+    let (Some(first), Some(last)) = (first_pcr, last_pcr) else {
+        return None;
+    };
+    if first == last {
+        return None;
+    }
+
+    let diff_27mhz = if last >= first {
+        last - first
+    } else {
+        MAX_PCR - first + last
+    };
+    let ticks_90khz = diff_27mhz / 300;
+    (ticks_90khz > 0).then_some(ticks_90khz)
+}
+
 type PacketIndices = Vec<(usize, Option<(usize, Option<usize>, u16)>)>;
 
 pub struct TransportStreamBuffer {
@@ -303,6 +343,7 @@ pub struct TransportStreamBuffer {
     cc_entries: Box<[Option<(u8, bool)>; 8192]>,
     waker: Arc<AtomicWaker>,
     first_pcr: Option<u64>,
+    force_discontinuity_on_wrap: bool,
 }
 
 impl std::fmt::Debug for TransportStreamBuffer {
@@ -314,6 +355,7 @@ impl std::fmt::Debug for TransportStreamBuffer {
             .field("timestamp_offset", &self.timestamp_offset)
             .field("stream_duration_90khz", &self.stream_duration_90khz)
             .field("first_pcr", &self.first_pcr)
+            .field("force_discontinuity_on_wrap", &self.force_discontinuity_on_wrap)
             .finish_non_exhaustive()
     }
 }
@@ -333,6 +375,9 @@ impl Clone for TransportStreamBuffer {
             cc_entries: Box::new([None; 8192]),
             waker: Arc::clone(&self.waker),
             first_pcr: self.first_pcr,
+            // Start each clone with one initial discontinuity marker to keep
+            // continuity behavior consistent for fresh consumers.
+            force_discontinuity_on_wrap: true,
         }
     }
 }
@@ -349,7 +394,10 @@ impl TransportStreamBuffer {
         let (packet_indices, _) = extract_pts_dts_indices_with_continuity(&raw);
         let length = packet_indices.len();
 
-        let stream_duration_90khz = calculate_duration_ticks(&raw, &packet_indices);
+        let mut stream_duration_90khz = calculate_duration_ticks(&raw, &packet_indices);
+        if stream_duration_90khz == 0 {
+            stream_duration_90khz = calculate_pcr_duration_ticks(&raw).unwrap_or(0);
+        }
 
         // Scan for the first PCR in the buffer to use as a reference for discontinuity packets
         let mut first_pcr = None;
@@ -383,6 +431,9 @@ impl TransportStreamBuffer {
             cc_entries: Box::new([None; 8192]),
             waker: Arc::new(AtomicWaker::new()),
             first_pcr,
+            // Emit a discontinuity packet on startup. Subsequent injections are
+            // governed by wrap handling and duration/discontinuity logic.
+            force_discontinuity_on_wrap: true,
         }
     }
 
@@ -549,9 +600,19 @@ impl TransportStreamBuffer {
                 // increasing across loops. Resetting to 0 causes decoders (MPV, ffmpeg) to see
                 // a backward timestamp jump and treat the loop as end-of-stream or corrupt data.
                 self.current_pos = 0;
-                self.timestamp_offset =
-                    self.timestamp_offset.wrapping_add(self.stream_duration_90khz) % MAX_PTS_DTS;
-                self.current_dts = 0;
+                // Advance timestamps by one full source duration per loop so output time is
+                // monotonic for clients. Resetting to zero causes backward jumps that some
+                // players interpret as stream end/corruption after the first cycle.
+                if self.stream_duration_90khz > 0 {
+                    self.timestamp_offset =
+                        self.timestamp_offset.wrapping_add(self.stream_duration_90khz) % MAX_PTS_DTS;
+                    self.current_dts = self.current_dts.wrapping_add(self.stream_duration_90khz) % MAX_PTS_DTS;
+                } else {
+                    // PCR-only (or malformed) assets may not expose PTS/DTS-derived duration.
+                    // Force one discontinuity marker after wrap so decoders do not see identical
+                    // timestamp cycles as a continuous timeline.
+                    self.force_discontinuity_on_wrap = true;
+                }
 
                 // Reset only the discontinuity-sent flag so injection packets are emitted at the
                 // start of the next loop. Continuity counter values keep running so CC remains
@@ -577,24 +638,23 @@ impl TransportStreamBuffer {
             }
             let (counter, discontinuity_sent) = entry.as_mut().unwrap();
 
-            // Inject a discontinuity packet for every PID on first appearance and at loop
-            // boundaries. This covers both PES PIDs (audio/video) and PSI PIDs (PAT/PMT)
-            // so that decoders accept the CC reset during mid-stream switches.
-            let inject_discontinuity = !*discontinuity_sent;
-
-            let payload_packet_cc;
+            // Disable synthetic discontinuity packet insertion for looped custom streams.
+            // In practice this can produce demuxer corruption on some clients (PES mismatch).
+            // Monotonic timestamps + stable continuity counters are sufficient here.
+            let inject_discontinuity = self.force_discontinuity_on_wrap && !*discontinuity_sent;
             if inject_discontinuity {
-                // Discontinuity packet is adaptation-only and does not consume a CC step.
-                // Keep CC progression tied strictly to payload packets.
                 let extra_packet_cc = if packet_has_payload { *counter } else { packet[3] & 0x0F };
-                payload_packet_cc = if packet_has_payload { *counter } else { packet[3] & 0x0F };
-                *discontinuity_sent = true;
-
-                // Write discontinuity packet directly into the output buffer — no allocation.
-                Self::generate_discontinuity_packet(packet, extra_packet_cc, self.first_pcr, self.timestamp_offset, &mut bytes);
-            } else {
-                payload_packet_cc = if packet_has_payload { *counter } else { packet[3] & 0x0F };
+                Self::generate_discontinuity_packet(
+                    packet,
+                    extra_packet_cc,
+                    self.first_pcr,
+                    self.timestamp_offset,
+                    &mut bytes,
+                );
+                self.force_discontinuity_on_wrap = false;
             }
+            *discontinuity_sent = true;
+            let payload_packet_cc = if packet_has_payload { *counter } else { packet[3] & 0x0F };
 
             // TS continuity counter increments only when payload is present (AFC=01/11).
             if packet_has_payload {
diff --git a/backend/src/model/config/app.rs b/backend/src/model/config/app.rs
index 248a7b403..e9a9b83de 100644
--- a/backend/src/model/config/app.rs
+++ b/backend/src/model/config/app.rs
@@ -12,6 +12,10 @@ use rand::Rng;
 use shared::error::{TuliproxError, TuliproxErrorKind};
 use shared::info_err_res;
 use shared::model::ConfigPaths;
+use shared::utils::{
+    CHANNEL_UNAVAILABLE, LOW_PRIORITY_PREEMPTED, PANEL_API_PROVISIONING, PROVIDER_CONNECTIONS_EXHAUSTED,
+    USER_ACCOUNT_EXPIRED, USER_CONNECTIONS_EXHAUSTED,
+};
 use std::borrow::Cow;
 use std::collections::HashSet;
 use std::fs::File;
@@ -19,7 +23,6 @@ use std::io::Read;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use tokio::sync::OnceCell;
-use shared::utils::{CHANNEL_UNAVAILABLE, LOW_PRIORITY_PREEMPTED, PANEL_API_PROVISIONING, PROVIDER_CONNECTIONS_EXHAUSTED, USER_ACCOUNT_EXPIRED, USER_CONNECTIONS_EXHAUSTED};
 
 fn generate_secret() -> [u8; 32] {
     let mut rng = rand::rng();
diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs
index 66767f709..c3ed6e628 100644
--- a/backend/src/utils/network/request.rs
+++ b/backend/src/utils/network/request.rs
@@ -1611,32 +1611,6 @@ mod tests {
         Arc::new(ConfigProvider::from(&dto))
     }
 
-    async fn start_plain_http_server() -> (SocketAddr, Arc<AtomicUsize>, tokio::task::JoinHandle<()>) {
-        let listener = TcpListener::bind("127.0.0.1:0").await.expect("tcp bind should work");
-        let addr = listener.local_addr().expect("local addr should exist");
-        let accepted = Arc::new(AtomicUsize::new(0));
-        let accepted_clone = Arc::clone(&accepted);
-
-        let handle = tokio::spawn(async move {
-            loop {
-                let Ok((mut socket, _)) = listener.accept().await else {
-                    continue;
-                };
-                accepted_clone.fetch_add(1, Ordering::SeqCst);
-                tokio::spawn(async move {
-                    let mut buf = vec![0_u8; 2048];
-                    let _ = socket.read(&mut buf).await;
-                    let _ = socket
-                        .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
-                        .await;
-                    let _ = socket.shutdown().await;
-                });
-            }
-        });
-
-        (addr, accepted, handle)
-    }
-
     #[test]
     fn test_url_mask() {
         // Replace with "***"
@@ -1816,9 +1790,43 @@ mod tests {
         assert_eq!(https_second.connect_ip.map(|ip| ip.to_string()), Some("203.0.113.11".to_string()));
     }
 
+    async fn start_plain_http_server() -> std::io::Result<(SocketAddr, Arc<AtomicUsize>, tokio::task::JoinHandle<()>)> {
+        let listener = TcpListener::bind("127.0.0.1:0").await?;
+        let addr = listener.local_addr()?;
+        let accepted = Arc::new(AtomicUsize::new(0));
+        let accepted_clone = Arc::clone(&accepted);
+
+        let handle = tokio::spawn(async move {
+            loop {
+                let Ok((mut socket, _)) = listener.accept().await else {
+                    continue;
+                };
+                accepted_clone.fetch_add(1, Ordering::SeqCst);
+                tokio::spawn(async move {
+                    let mut buf = vec![0_u8; 2048];
+                    let _ = socket.read(&mut buf).await;
+                    let _ = socket
+                        .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
+                        .await;
+                    let _ = socket.shutdown().await;
+                });
+            }
+        });
+
+        Ok((addr, accepted, handle))
+    }
+
     #[tokio::test]
     async fn test_on_connect_error_try_next_ip_before_provider_rotation() {
-        let (addr, accepted, server_handle) = start_plain_http_server().await;
+
+        let (addr, accepted, server_handle) = match start_plain_http_server().await {
+            Ok(server) => server,
+            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => {
+                eprintln!("skipping test_on_connect_error_try_next_ip_before_provider_rotation: {err}");
+                return;
+            }
+            Err(err) => panic!("failed to start test http server: {err}"),
+        };
 
         let mut cfg = Config {
             connect_timeout_secs: 1,
diff --git a/bin/build_resources.sh b/bin/build_resources.sh
index 0bd796370..6c736758c 100755
--- a/bin/build_resources.sh
+++ b/bin/build_resources.sh
@@ -22,7 +22,7 @@ while getopts "fh" opt; do
   esac
 done
 
-declare -a resources=("channel_unavailable" "user_connections_exhausted" "provider_connections_exhausted" "user_account_expired" "panel_api_provisioning" "low_priority_preempted")
+declare -a resources=("channel_unavailable" "user_connections_exhausted" "provider_connections_exhausted" "low_priority_preempted" "user_account_expired" "panel_api_provisioning")
 
 for resource in "${resources[@]}"; do
   if [ "$flag_force" = false ]; then
@@ -46,7 +46,7 @@ for resource in "${resources[@]}"; do
       exit 1
     fi
   else
-    echo "ffmpeg not found"
+    echo "ffmpeg not found" >&2
     exit 1
   fi
 done
diff --git a/bin/inc_version.sh b/bin/inc_version.sh
index 426cc6ac5..89f13a15c 100755
--- a/bin/inc_version.sh
+++ b/bin/inc_version.sh
@@ -1,20 +1,25 @@
 #!/usr/bin/env bash
-set -eo pipefail
+set -euo pipefail
 
 WORKING_DIR=$(pwd)
-BACKEND_DIR="${WORKING_DIR}/backend"
 
-if ! command -v cargo-set-version &> /dev/null
-then
-    echo "🧨 cargo-set-version could not be found. Install it with 'cargo install cargo-edit'"
-    exit 1
-fi
+ROOT_TOML="${WORKING_DIR}/Cargo.toml"
 
-
-# Read current version from Cargo.toml
-OLD_VERSION="$(grep '^version' "${BACKEND_DIR}/Cargo.toml" | head -n1 | cut -d'"' -f2 || true)"
+# Read [workspace.package].version from Cargo.toml
+OLD_VERSION="$(
+  awk '
+    /^\[workspace\.package\][[:space:]]*$/ { in_ws_pkg=1; next }
+    /^\[[^]]+\][[:space:]]*$/ { if (in_ws_pkg) exit }
+    in_ws_pkg && /^[[:space:]]*version[[:space:]]*=/ {
+      if (match($0, /"[^"]+"/)) {
+        print substr($0, RSTART + 1, RLENGTH - 2)
+        exit
+      }
+    }
+  ' "${ROOT_TOML}" || true
+)"
 if [ -z "${OLD_VERSION}" ]; then
-    echo "🧨 Failed to read version from '${BACKEND_DIR}/Cargo.toml' (expected a line like: version = \"x.y.z\")."
+    echo "🧨 Failed to read [workspace.package].version from '${ROOT_TOML}'."
     exit 1
 fi
 
@@ -42,8 +47,35 @@ esac
 
 NEW_VERSION="${major}.${minor}.${patch}"
 
-
-cargo set-version "$NEW_VERSION"
+# Update only [workspace.package].version in Cargo.toml
+TMP_FILE="$(mktemp)"
+if awk -v new_version="${NEW_VERSION}" '
+    BEGIN { in_ws_pkg=0; version_updated=0 }
+    /^\[workspace\.package\][[:space:]]*$/ { in_ws_pkg=1; print; next }
+    /^\[[^]]+\][[:space:]]*$/ {
+      if (in_ws_pkg && !version_updated) {
+        in_ws_pkg=0
+      }
+    }
+    {
+      if (in_ws_pkg && !version_updated && /^[[:space:]]*version[[:space:]]*=/) {
+        sub(/[[:space:]]*=[[:space:]]*"[^"]*"/, " = \"" new_version "\"")
+        version_updated=1
+      }
+      print
+    }
+    END {
+      if (!version_updated) {
+        exit 2
+      }
+    }
+  ' "${ROOT_TOML}" > "${TMP_FILE}"; then
+  mv "${TMP_FILE}" "${ROOT_TOML}"
+else
+  rm -f "${TMP_FILE}"
+  echo "🧨 Failed to update [workspace.package].version in '${ROOT_TOML}'."
+  exit 1
+fi
 
 VERSION=v$NEW_VERSION
 echo "🛠️ Set version $VERSION"