Refactored connection handling to avoid race conditions

CHANGELOG.md

@@ -23,11 +23,11 @@
 - Added `order: none` support for group/channel sorting so mappings can opt out of any reordering and keep the source order.
 - Session tracking now matches repeated HLS segment connections by session token so a single user keeps one active connection count even when new TCP sockets are opened.
 - EPG icon urls are now rewritten on reverse proxy mode.
-- Short EPG is now served from local disc, if available
+- Short EPG is now served from local disk, if available
 - WebUI Api-User Category selection implemented
 - Stream Table Copy-To-Clipboard functions added
 - Refactored provider connection handling to avoid possible race conditions
-- `exp_date` field added to inputs, aliases and csv batch files - a date as "YYYY-MM-DD HH:MM:SS" format like `2028-11-30 12:34:12` or Unix timestamp (seconds since epoch)
+- Added `exp_date` field to inputs, aliases, and CSV batch files; accepts date in "YYYY-MM-DD HH:MM:SS" format or Unix timestamp (seconds since epoch).
 
 # 3.2.0 (2025-11-14)
 - Added `name` attribute to Staged Input.

backend/src/api/api_utils.rs

@@ -391,7 +391,7 @@ async fn resolve_streaming_strategy(
                 }
             }
         } else {
-            debug!("Input {} is exhausted. No connections allowed.", input.name);
+            debug!("Provider {} is exhausted. No connections allowed.", input.name);
             let stream = create_provider_connections_exhausted_stream(&app_state.app_config, &[]);
             ProviderStreamState::Custom(stream)
         };

backend/src/api/model/active_provider_manager.rs

@@ -74,10 +74,6 @@ impl ActiveProviderManager {
         self.providers.update_config(inputs, grace_period_millis, grace_period_timeout_secs).await;
     }
 
-    // fn get_next_handle_id(&self) -> usize {
-    //     self.next_handle_id.fetch_update(Ordering::AcqRel,Ordering::Acquire,|x| Some(x.wrapping_add(1))).unwrap_or_default()
-    // }
-
     async fn acquire_connection_inner(&self, provider_or_input_name: &str, addr: &SocketAddr, force: bool) -> Option<ProviderHandle> {
         // Call the specific acquisition function
         let allocation = if force {

backend/src/model/xmltv.rs

@@ -247,11 +247,11 @@ async fn parse_xmltv_for_web_ui<R: AsyncRead + Send + Unpin>(reader: R) -> Resul
 
     // only 1 day old epg
     let now = Utc::now();
-    let yesterday_start = Utc.with_ymd_and_hms(now.year(), now.month(), now.day(), 0, 0, 0).unwrap()
+    let yesterday_start = Utc.with_ymd_and_hms(now.year(), now.month(), now.day(), 0, 0, 0)
+        .single().expect("Current date at midnight should always be valid")
         - chrono::Duration::days(1);
     let threshold_ts = yesterday_start.timestamp();
 
-
     loop {
         match reader.read_event_into_async(&mut buf).await {
             Ok(Event::Empty(e) | Event::Start(e)) => {

backend/src/repository/epg_repository.rs

@@ -5,33 +5,32 @@ use crate::repository::xtream_repository::{xtream_get_epg_file_path, xtream_get_
 use crate::utils::debug_if_enabled;
 use shared::error::{notify_err, TuliproxError};
 use std::path::Path;
-use log::error;
 use tokio::io::AsyncWriteExt;
 
+
+// Due to an error in quick_xml we cant write doc type through event. The quotes are escaped and the xml file is invalid.
+//
+// // XML Header
+// writer.write_event_async(quick_xml::events::Event::Decl(quick_xml::events::BytesDecl::new("1.0", Some("utf-8"), None)))
+//     .await.map_err(|e| notify_err!(format!("failed to write XML header: {}", e)))?;
+//
+// // DOCTYPE
+// writer.write_event_async(quick_xml::events::Event::DocType(quick_xml::events::BytesText::new(r#"tv SYSTEM "xmltv.dtd""#)))
+//     .await.map_err(|e| notify_err!(format!("failed to write doctype: {}", e)))?;
 pub async fn epg_write_file(target: &ConfigTarget, epg: &Epg, path: &Path) -> Result<(), TuliproxError> {
     let file = tokio::fs::File::create(path).await
         .map_err(|e| notify_err!(format!("failed to create epg file: {}", e)))?;
     let mut buf_writer = tokio::io::BufWriter::new(file);
 
     // Work-Around BytesText DocType escape, see below
-    if let Err(err)  = buf_writer.write_all(b"<?xml version=\"1.0\" encoding=\"utf-8\"?>\n").await {
-        error!("EPG: Failed to write xml header {err}");
-    }
+    buf_writer.write_all(b"<?xml version=\"1.0\" encoding=\"utf-8\"?>\n").await
+        .map_err(|e| notify_err!(format!("failed to write XML header: {}", e)))?;
 
-    if let Err(err)  = buf_writer.write_all(b"<!DOCTYPE tv SYSTEM \"xmltv.dtd\">\n").await {
-        error!("EPG: Failed to write epg doc type {err}");
-    }
+    buf_writer.write_all(b"<!DOCTYPE tv SYSTEM \"xmltv.dtd\">\n").await
+        .map_err(|e| notify_err!(format!("failed to write doctype: {}", e)))?;
 
     let mut writer = quick_xml::writer::Writer::new(buf_writer);
 
-    // // XML Header
-    // writer.write_event_async(quick_xml::events::Event::Decl(quick_xml::events::BytesDecl::new("1.0", Some("utf-8"), None)))
-    //     .await.map_err(|e| notify_err!(format!("failed to write XML header: {}", e)))?;
-    //
-    // // DOCTYPE
-    // writer.write_event_async(quick_xml::events::Event::DocType(quick_xml::events::BytesText::new(r#"tv SYSTEM "xmltv.dtd""#)))
-    //     .await.map_err(|e| notify_err!(format!("failed to write doctype: {}", e)))?;
-
     // EPG Content
     epg.write_to_async(&mut writer).await.map_err(|e| notify_err!(format!("failed to write epg: {}", e)))?;
 

frontend/public/assets/i18n/en.json

@@ -342,6 +342,7 @@
     "SCHEDULE_EXISTS": "Schedule already exists",
     "CLIPBOARD_NOT_SUPPORTED": "Clipboard not supported.\nYour browser or current context does not allow clipboard access.\nPlease use HTTPS or localhost.",
     "FAILED_TO_KICK_USER_STREAM":  "Failed to kick user stream",
+    "FAILED_TO_RETRIEVE_WEBPLAYER_URL": "Failed to retrieve webplayer URL",
     "DOWNLOAD": {
       "SUCCESS": "Successfully downloaded",
       "FAIL": "Failed to download!",

frontend/scss/app/components/api_user/_api_user_view.scss

@@ -89,16 +89,10 @@
   }
 
   &__categories {
-    //display: flex;
-    //flex-flow: row wrap;
-    //width: 100%;
-    //gap: var(--gap-default);
-    padding: var(--padding-default) 0;
-    //box-sizing: border-box;
-
     display: grid;
     grid-template-columns: repeat(auto-fit, minmax(size.$playlist-categories-grid-cell-width, 1fr));
     gap: var(--gap-large);
+    padding: var(--padding-default) 0;
     box-sizing: border-box;
     width: 100%;
 

frontend/src/app/components/api_user/api_user_view.rs

@@ -50,7 +50,7 @@ pub fn ApiUserView() -> Html {
                         </div>
                     </div>
                     <div class="tp__app-main__body">
-                        <ApiUserPlaylist></ApiUserPlaylist>
+                        <ApiUserPlaylist />
                     </div>
               </div>
             </div>

frontend/src/app/components/api_user/playlist.rs

@@ -1,22 +1,22 @@
+use crate::app::components::api_user::target_playlist::{BouquetSelection, UserTargetPlaylist};
+use crate::app::components::{Panel, RadioButtonGroup, TextButton};
+use crate::hooks::use_service_context;
+use crate::model::{BusyStatus, EventMessage};
+use shared::error::TuliproxError;
+use shared::info_err;
+use shared::model::{PlaylistBouquetDto, PlaylistCategoriesDto, PlaylistClusterBouquetDto};
 use std::cell::RefCell;
-use yew::prelude::*;
 use std::collections::HashMap;
 use std::fmt;
 use std::rc::Rc;
 use std::str::FromStr;
+use yew::prelude::*;
 use yew_i18n::use_translation;
-use shared::error::TuliproxError;
-use shared::info_err;
-use shared::model::{PlaylistBouquetDto, PlaylistCategoriesDto, PlaylistClusterBouquetDto};
-use crate::app::components::{Panel, RadioButtonGroup, TextButton};
-use crate::app::components::api_user::target_playlist::{BouquetSelection, UserTargetPlaylist};
-use crate::hooks::use_service_context;
-use crate::model::{BusyStatus, EventMessage};
 
 #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 enum ApiUserPlaylistPage {
     Xtream,
-    M3u
+    M3u,
 }
 
 impl FromStr for ApiUserPlaylistPage {
@@ -54,7 +54,7 @@ fn to_playlist_cluster(count: (usize, usize, usize), bouquet: Option<&Rc<RefCell
         };
 
         let live = selected_vec(&selections.live).filter(|v| v.len() != count.0);
-        let vod  = selected_vec(&selections.vod).filter(|v| v.len() != count.1);
+        let vod = selected_vec(&selections.vod).filter(|v| v.len() != count.1);
         let series = selected_vec(&selections.series).filter(|v| v.len() != count.2);
 
         // if all three are None, return None
@@ -112,7 +112,12 @@ pub fn ApiUserPlaylist() -> Html {
                         bouquets.set(bouquet.clone());
                         categories.set(cats.clone());
                     }
-                    _ => {
+                    (Err(e1), Err(e2)) => {
+                        log::error!("Failed to load bouquet: {e1:?}, categories: {e2:?}");
+                        services.toastr.error(translate.t("MESSAGES.DOWNLOAD.USER_BOUQUET.FAIL"));
+                    }
+                    (Err(e), _) | (_, Err(e)) => {
+                        log::error!("Failed to load user data: {e:?}");
                         services.toastr.error(translate.t("MESSAGES.DOWNLOAD.USER_BOUQUET.FAIL"));
                     }
                 }
@@ -127,21 +132,23 @@ pub fn ApiUserPlaylist() -> Html {
     let on_save = {
         let selections = selections.clone();
         let services = service_ctx.clone();
+        let translate = translate.clone();
         let categories = categories.clone();
 
         Callback::from(move |_| {
             let selections = selections.clone();
             let services = services.clone();
+            let translate = translate.clone();
             let categories_xtream_count = categories.as_ref().and_then(|plc| plc.xtream.as_ref().map(|x|
                 (x.live.as_ref().map(|v| v.len()).unwrap_or(0),
                  x.vod.as_ref().map(|v| v.len()).unwrap_or(0),
                  x.series.as_ref().map(|v| v.len()).unwrap_or(0))
-            )).unwrap_or((0,0,0));
+            )).unwrap_or((0, 0, 0));
             let categories_m3u_count = categories.as_ref().and_then(|plc| plc.m3u.as_ref().map(|x|
                 (x.live.as_ref().map(|v| v.len()).unwrap_or(0),
                  x.vod.as_ref().map(|v| v.len()).unwrap_or(0),
                  x.series.as_ref().map(|v| v.len()).unwrap_or(0))
-            )).unwrap_or((0,0,0));
+            )).unwrap_or((0, 0, 0));
 
             wasm_bindgen_futures::spawn_local(async move {
                 services.event.broadcast(EventMessage::Busy(BusyStatus::Show));
@@ -149,13 +156,13 @@ pub fn ApiUserPlaylist() -> Html {
                     let selects = selections.borrow();
                     PlaylistBouquetDto {
                         xtream: to_playlist_cluster(categories_xtream_count, selects.get(&ApiUserPlaylistPage::Xtream)),
-                        m3u:    to_playlist_cluster(categories_m3u_count, selects.get(&ApiUserPlaylistPage::M3u)),
+                        m3u: to_playlist_cluster(categories_m3u_count, selects.get(&ApiUserPlaylistPage::M3u)),
                     }
                 };
 
                 match services.user_api.save_playlist_bouquet(&result).await {
-                     Ok(()) =>  services.toastr.success("Bouquet saved successfully"),
-                     Err(_) => services.toastr.error("Failed to save playlist"),
+                    Ok(()) =>  services.toastr.success(translate.t("MESSAGES.SAVE.BOUQUET.SUCCESS")),
+                    Err(_) => services.toastr.error(translate.t("MESSAGES.SAVE.BOUQUET.FAIL")),
                 }
 
                 services.event.broadcast(EventMessage::Busy(BusyStatus::Hide));

frontend/src/app/components/dashboard/streams_table.rs

@@ -267,10 +267,13 @@ pub fn StreamsTable(props: &StreamsTableProps) -> Html {
                             let virtual_id = dto.channel.virtual_id;
                             let cluster = dto.channel.cluster;
                             let services = services.clone();
+                            let translate = translate.clone();
                             let copy_to_clipboard = copy_to_clipboard.clone();
                             spawn_local(async move {
                                 if let Some(url) = services.playlist.get_playlist_webplayer_url(target_id, virtual_id, cluster).await {
                                     copy_to_clipboard.emit(url);
+                                } else {
+                                    services.toastr.error(translate.t("MESSAGES.FAILED_TO_RETRIEVE_WEBPLAYER_URL"));
                                 }
                             });
                         }

frontend/src/app/components/playlist/playlist_explorer.rs

@@ -152,6 +152,8 @@ pub fn PlaylistExplorer() -> Html {
                                             if let Some(url) = services.playlist.get_playlist_webplayer_url(target_id, virtual_id, cluster).await {
                                                 copy_to_clipboard.emit(url);
                                                 services.toastr.success(translate_clone.t("MESSAGES.PLAYLIST.WEBPLAYER_URL_COPY_TO_CLIPBOARD"));
+                                            } else {
+                                                services.toastr.error(translate_clone.t("MESSAGES.FAILED_TO_RETRIEVE_WEBPLAYER_URL"));
                                             }
                                         });
                                     }

frontend/src/services/user_api_service.rs

@@ -21,23 +21,15 @@ impl UserApiService {
     }
 
     pub async fn get_playlist_categories(&self) -> Result<Option<Rc<PlaylistCategoriesDto>>, Error> {
-        match request_get::<Rc<PlaylistCategoriesDto>>(&self.user_playlist_categories_path, None, None).await {
-            Ok(categories) => { Ok(categories) },
-            Err(err) => {
-                error!("{err}");
-                Err(err)
-            }
-        }
+        request_get::<Rc<PlaylistCategoriesDto>>(&self.user_playlist_categories_path, None, None)
+            .await
+            .inspect_err(|err| error!("{err}"))
     }
 
     pub async fn get_playlist_bouquet(&self) -> Result<Option<Rc<PlaylistBouquetDto>>, Error> {
-        match request_get::<Rc<PlaylistBouquetDto>>(&self.user_playlist_bouquet_path, None, None).await {
-            Ok(bouquet) => { Ok(bouquet) },
-            Err(err) => {
-                error!("{err}");
-                Err(err)
-            }
-        }
+        request_get::<Rc<PlaylistBouquetDto>>(&self.user_playlist_bouquet_path, None, None)
+            .await
+            .inspect_err(|err| error!("{err}"))
     }
 
     pub async fn save_playlist_bouquet(&self, bouquet: &PlaylistBouquetDto) -> Result<(), Error> {

shared/src/model/playlist.rs

@@ -41,10 +41,10 @@ impl FromStr for XtreamCluster {
     type Err = String;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        match s {
-            "live" | "Live" => Ok(XtreamCluster::Live),
-            "video" | "Video" | "vod" | "movie" | "Movie" => Ok(XtreamCluster::Video),
-            "series" | "Series" => Ok(XtreamCluster::Series),
+        match s.to_lowercase().as_str() {
+            "live" => Ok(XtreamCluster::Live),
+            "video" | "vod" | "movie" => Ok(XtreamCluster::Video),
+            "series" => Ok(XtreamCluster::Series),
             _ => Err(format!("Invalid XtreamCluster: {s}")),
         }
     }

shared/src/model/playlist_categories.rs

@@ -1,14 +1,3 @@
-
-// #[derive(Debug, serde::Serialize, serde::Deserialize, Default)]
-// pub struct PlaylistCategories {
-//     #[serde(skip_serializing_if = "Option::is_none")]
-//     pub live: Option<Vec<String>>,
-//     #[serde(skip_serializing_if = "Option::is_none")]
-//     pub vod: Option<Vec<String>>,
-//     #[serde(skip_serializing_if = "Option::is_none")]
-//     pub series: Option<Vec<String>>,
-// }
-
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)]
 pub struct PlaylistClusterCategoriesDto {
     #[serde(skip_serializing_if = "Option::is_none")]