diff --git a/.gitignore b/.gitignore index 44518a69b..dba79bc49 100644 --- a/.gitignore +++ b/.gitignore @@ -48,14 +48,13 @@ artifacts .envrc backend/Cargo.lock +#ignore d files +*.md +!docs/src/**/*.md +!README.md +!CHANGELOG.md +!CONTRIBUTING.md + # Ignore ai documentation files -AGENTS.md -PLANS.md -STEPS.md -CLAUDE.md -CODEX.md -.claude -.crush -.codex -.dual-graph/ -.dual-graph-context/ +.c[lor][adu]* +.dual* diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index fc607b603..f2d729c0f 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -463,6 +463,7 @@ pub struct StreamOptions { struct StreamingAcquireOptions<'a> { force_provider: Option<&'a Arc>, + allow_forced_provider_fallback: bool, allow_provider_grace: bool, user_priority: i8, connection_kind: crate::api::model::ConnectionKind, @@ -540,7 +541,9 @@ pub(in crate::api) async fn resolve_admission_with_strategies( soft_connections: u16, client_ip: &str, request_addr: &std::net::SocketAddr, - is_session_request: bool, + // This controls whether an existing logical playback session may reopen while the user is already at limit. + // It is intentionally independent from whether the session is socket-bound. + use_session_admission: bool, session_token: Option<&str>, activate_unbound_session: bool, eviction_reentry_guard: EvictionReentryGuard<'_>, @@ -553,7 +556,7 @@ pub(in crate::api) async fn resolve_admission_with_strategies( username, max_connections, soft_connections, - is_session_request, + use_session_admission, session_token, activate_unbound_session, ) @@ -613,7 +616,7 @@ pub(in crate::api) async fn resolve_admission_with_strategies( username, max_connections, soft_connections, - is_session_request, + use_session_admission, session_token, activate_unbound_session, ) @@ -644,6 +647,7 @@ struct SessionActivationRequest<'a> { stream_url: &'a str, connection_permission: UserConnectionPermission, connection_kind: crate::api::model::ConnectionKind, + socket_bound: bool, } async fn activate_session_before_stream_open( @@ -659,6 +663,7 @@ async fn activate_session_before_stream_open( stream_url, connection_permission, connection_kind, + socket_bound, } = request; let limits_enabled = app_state.app_config.config.load().user_access_control && (user.max_connections > 0 || user.soft_connections > 0); @@ -684,6 +689,7 @@ async fn activate_session_before_stream_open( addr: &fingerprint.addr, connection_permission, connection_kind: Some(connection_kind), + socket_bound, }) .await; @@ -697,7 +703,11 @@ async fn activate_session_before_stream_open( true, Some(session_token), true, - EvictionReentryGuard::SocketPlayback { virtual_id }, + if socket_bound { + EvictionReentryGuard::SocketPlayback { virtual_id } + } else { + EvictionReentryGuard::Session(session_token) + }, ) .await; @@ -770,7 +780,6 @@ async fn resolve_streaming_strategy( let provider_connection_handle = match options.force_provider { Some(provider) => { // First try to stay on the exact pinned provider account without over-allocating. - // If that account is no longer available, fall back to any available account in the same lineup. if let Some(handle) = app_state .active_provider .acquire_exact_connection_with_grace_for_session( @@ -785,7 +794,7 @@ async fn resolve_streaming_strategy( { forced_provider_allocated = true; Some(handle) - } else { + } else if options.allow_forced_provider_fallback { debug_if_enabled!( "Pinned provider {} unavailable for {}; falling back to lineup allocation", sanitize_sensitive_info(provider), @@ -802,6 +811,13 @@ async fn resolve_streaming_strategy( options.session_owner, ) .await + } else { + debug_if_enabled!( + "Pinned provider {} unavailable for {}; strict provider affinity prevents fallback", + sanitize_sensitive_info(provider), + sanitize_sensitive_info(&fingerprint.addr.to_string()) + ); + None } } None => { @@ -901,6 +917,7 @@ async fn create_stream_response_details( share_stream: bool, connection_permission: UserConnectionPermission, force_provider: Option<&Arc>, + allow_forced_provider_fallback: bool, allow_provider_grace: bool, virtual_id: VirtualId, user_priority: i8, @@ -915,6 +932,7 @@ async fn create_stream_response_details( input, StreamingAcquireOptions { force_provider, + allow_forced_provider_fallback, allow_provider_grace, user_priority, connection_kind, @@ -1238,9 +1256,10 @@ pub async fn force_provider_stream_response( // This is critical for users with a connection limit of 1 to avoid "Provider exhausted" or provider-side 502/509 errors during seeking. app_state.connection_manager.release_provider_connection(&user_session.addr).await; - // Keep seek/range reconnects on the same provider account whenever that exact account is still available. - // If it is not available anymore, resolve_streaming_strategy will fall back to another free account. + // Provider-affine playback must stay on the same provider account across seeks/range reconnects. + // Only non-affine sessions may fall back to a different account in the same lineup. let preferred_provider = Some(&user_session.provider); + let allow_forced_provider_fallback = !item_type.requires_provider_affinity(); // Never allow provider-side grace for forced seek/session reacquire. // Over-allocation here would break provider-side one-connection limits. let allow_provider_grace = false; @@ -1261,6 +1280,7 @@ pub async fn force_provider_stream_response( share_stream, connection_permission, preferred_provider, + allow_forced_provider_fallback, allow_provider_grace, stream_channel.virtual_id, connection_priority_for_kind(ctx.user, connection_kind), @@ -1400,6 +1420,7 @@ pub async fn stream_response( stream_url, connection_permission, connection_kind, + socket_bound: item_type.uses_socket_bound_session(), }, ) .await; @@ -1496,6 +1517,7 @@ pub async fn stream_response( connection_permission, None, true, + true, stream_channel.virtual_id, connection_priority_for_kind(user, connection_kind), connection_kind, @@ -1620,18 +1642,14 @@ pub async fn stream_response( StatusCode::BAD_REQUEST.into_response() } } else { - // Previously, we would always check if the provider redirected the request. - // If the provider redirected a movie from /movie/... to a temporary /live/... URL, - // We would save that redirected URL in your session. - // When we tried to seek or pause/resume, we would use that saved /live/ URL. - // However, providers often make these redirect links ephemeral or restricted—they - // might not support seeking, or they might trigger a 509 error if accessed again. - // For Movies/Series: We now ignore the redirect and always save the original, - // canonical URL (the one starting with /movie/) in your session. - // This ensures that every time you seek, we start "fresh" with the correct provider handshake, - // preventing the session from being "poisoned" by a temporary redirect. - // For everything else (Live): It continues to work as before, using the redirected URL if available, - // which is often desirable for live streams to stay on the same edge server. + // Previously, we always persisted the provider's final request URL into the session. + // For VOD-like playback that can be the wrong thing to reuse later: a seek or reopen + // should start from the canonical playback entrypoint, not from a provider-specific + // redirected target that happened to be used for an earlier request. + // For Movies/Series/Catchup we therefore keep the canonical request URL in the session. + // That avoids "session poisoning" where later seeks/resumes inherit a non-canonical URL. + // For live playback we still keep the redirected URL when available, because staying on + // the chosen upstream edge/server is often desirable there. let session_url: Cow<'_, str> = if matches!( item_type, PlaylistItemType::Catchup @@ -1639,6 +1657,8 @@ pub async fn stream_response( | PlaylistItemType::LocalVideo | PlaylistItemType::Series | PlaylistItemType::LocalSeries + | PlaylistItemType::SeriesInfo + | PlaylistItemType::LocalSeriesInfo ) { Cow::Owned(actual_request_url.to_string()) } else { @@ -1673,7 +1693,9 @@ pub async fn stream_response( | PlaylistItemType::LiveDash | PlaylistItemType::Video | PlaylistItemType::Series + | PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeries + | PlaylistItemType::LocalSeriesInfo | PlaylistItemType::Catchup ) { let _ = app_state @@ -1687,6 +1709,7 @@ pub async fn stream_response( addr: &fingerprint.addr, connection_permission, connection_kind: Some(connection_kind), + socket_bound: item_type.uses_socket_bound_session(), }) .await; let reservation_ttl_secs = get_session_reservation_ttl_secs(app_state, item_type); @@ -1847,6 +1870,7 @@ async fn try_shared_stream_response_if_any( addr: &fingerprint.addr, connection_permission: connect_permission, connection_kind: Some(connection_kind), + socket_bound: stream_channel.item_type.uses_socket_bound_session(), }) .await; } @@ -2037,6 +2061,7 @@ pub async fn local_stream_response( stream_url: &pli.url, connection_permission, connection_kind, + socket_bound: pli.item_type.uses_socket_bound_session(), }, ) .await; @@ -2086,6 +2111,7 @@ pub async fn local_stream_response( addr: &fingerprint.addr, connection_permission, connection_kind: Some(resolved_connection_kind), + socket_bound: pli.item_type.uses_socket_bound_session(), }) .await; } @@ -2341,12 +2367,12 @@ pub async fn resource_response( StatusCode::BAD_REQUEST.into_response() } -pub fn separate_number_and_remainder(input: &str) -> (String, Option) { +pub fn separate_number_and_remainder(input: &str) -> (&str, Option<&str>) { input.rfind('.').map_or_else( - || (input.to_string(), None), + || (input, None), |dot_index| { - let number_part = input[..dot_index].to_string(); - let rest = input[dot_index..].to_string(); + let number_part = &input[..dot_index]; + let rest = &input[dot_index..]; (number_part, if rest.len() < 2 { None } else { Some(rest) }) }, ) @@ -2640,7 +2666,10 @@ mod tests { PlaylistStorageState, SharedStreamManager, }, auth::Fingerprint, - model::{AppConfig, Config, ConfigInput, ConfigTarget, MediaToolCapabilities, ProcessTargets, ProxyUserCredentials, SourcesConfig}, + model::{ + AppConfig, Config, ConfigInput, ConfigInputAlias, ConfigTarget, MediaToolCapabilities, ProcessTargets, + ProxyUserCredentials, SourcesConfig, + }, utils::{GeoIp, FileLockManager}, }; use arc_swap::{ArcSwap, ArcSwapOption}; @@ -2651,7 +2680,7 @@ mod tests { model::{ConfigPaths, ConfigTargetOptions, InputFetchMethod, InputType, PlaylistItemType, ProcessingOrder, StreamChannel, XtreamCluster}, utils::{default_catchup_session_ttl_secs, default_hls_session_ttl_secs, Internable}, }; - use std::{borrow::Cow, collections::HashMap, sync::Arc}; + use std::{borrow::Cow, collections::HashMap, net::SocketAddr, sync::Arc}; use tokio::sync::mpsc; use crate::model::StreamHistoryConfig; @@ -2723,6 +2752,94 @@ mod tests { ); } + #[test] + fn provider_affinity_policy_matches_stream_types() { + assert!(!PlaylistItemType::Live.requires_provider_affinity()); + assert!(!PlaylistItemType::LiveUnknown.requires_provider_affinity()); + assert!(PlaylistItemType::LiveHls.requires_provider_affinity()); + assert!(PlaylistItemType::LiveDash.requires_provider_affinity()); + assert!(PlaylistItemType::Video.requires_provider_affinity()); + assert!(PlaylistItemType::Series.requires_provider_affinity()); + assert!(PlaylistItemType::Catchup.requires_provider_affinity()); + } + + #[tokio::test] + async fn resolve_streaming_strategy_honors_forced_provider_fallback_policy() { + let app_state = create_test_dual_provider_app_state(); + let input_name = "provider_1".intern(); + let input = app_state + .app_config + .sources + .load() + .get_input_by_name(&input_name) + .cloned() + .unwrap_or_else(|| unreachable!()); + let pinned_provider = "provider_1".intern(); + let busy_addr: SocketAddr = "127.0.0.1:55301".parse().unwrap_or_else(|_| unreachable!()); + let strict_addr: SocketAddr = "127.0.0.1:55302".parse().unwrap_or_else(|_| unreachable!()); + let fallback_addr: SocketAddr = "127.0.0.1:55303".parse().unwrap_or_else(|_| unreachable!()); + let stream_url = "http://provider-1.example/movie/1.mkv"; + + let busy = app_state + .active_provider + .acquire_exact_connection_with_grace( + &pinned_provider, + &busy_addr, + false, + 0, + crate::api::model::ConnectionKind::Normal, + ) + .await; + assert!(busy.is_some(), "setup should occupy the pinned provider"); + + let strict = resolve_streaming_strategy( + &app_state, + stream_url, + &create_test_fingerprint(strict_addr), + &input, + StreamingAcquireOptions { + force_provider: Some(&pinned_provider), + allow_forced_provider_fallback: false, + allow_provider_grace: false, + user_priority: 0, + connection_kind: crate::api::model::ConnectionKind::Normal, + session_owner: Some("vod-session"), + }, + ) + .await; + assert!(strict.provider_handle.is_none(), "strict provider affinity should not allocate a different provider"); + assert!( + matches!(strict.provider_stream_state, ProviderStreamState::Custom(_)), + "strict provider affinity should fail closed when the pinned provider is unavailable" + ); + + let fallback = resolve_streaming_strategy( + &app_state, + stream_url, + &create_test_fingerprint(fallback_addr), + &input, + StreamingAcquireOptions { + force_provider: Some(&pinned_provider), + allow_forced_provider_fallback: true, + allow_provider_grace: false, + user_priority: 0, + connection_kind: crate::api::model::ConnectionKind::Normal, + session_owner: Some("live-session"), + }, + ) + .await; + let fallback_provider = match fallback.provider_stream_state { + ProviderStreamState::Available(Some(provider_name), _) + | ProviderStreamState::GracePeriod(Some(provider_name), _) => provider_name, + _ => panic!("fallback-enabled request should allocate a provider"), + }; + assert_eq!(fallback_provider.as_ref(), "provider_2"); + + app_state.active_provider.release_connection(&busy_addr).await; + app_state.active_provider.release_connection(&strict_addr).await; + app_state.active_provider.release_connection(&fallback_addr).await; + } + #[test] fn test_should_allow_exhausted_shared_reconnect_only_for_matching_shared_session() { let session = UserSession { @@ -2732,7 +2849,10 @@ mod tests { provider: Arc::::from("provider"), stream_url: Arc::::from("http://provider/live/449924.ts"), addr: "127.0.0.1:1234".parse().unwrap_or_else(|_| unreachable!()), + socket_bound: false, + active_addrs: vec!["127.0.0.1:1234".parse().unwrap_or_else(|_| unreachable!())], ts: 1, + started_at:1, permission: UserConnectionPermission::Allowed, counted: false, }; @@ -2849,6 +2969,60 @@ mod tests { } } + fn create_test_dual_provider_app_config() -> AppConfig { + let input = Arc::new(ConfigInput { + id: 1, + name: "provider_1".intern(), + input_type: InputType::Xtream, + headers: HashMap::default(), + url: "http://provider-1.example".to_string(), + username: Some("user1".to_string()), + password: Some("pass1".to_string()), + enabled: true, + priority: 0, + max_connections: 1, + method: InputFetchMethod::default(), + aliases: Some(vec![ConfigInputAlias { + id: 2, + name: "provider_2".intern(), + url: "http://provider-2.example".to_string(), + username: Some("user2".to_string()), + password: Some("pass2".to_string()), + priority: 1, + max_connections: 1, + exp_date: None, + enabled: true, + }]), + ..ConfigInput::default() + }); + let sources = SourcesConfig { inputs: vec![input], ..SourcesConfig::default() }; + + AppConfig { + config: Arc::new(ArcSwap::from_pointee(Config::default())), + sources: Arc::new(ArcSwap::from_pointee(sources)), + hdhomerun: Arc::new(ArcSwapOption::default()), + api_proxy: Arc::new(ArcSwapOption::default()), + file_locks: Arc::new(FileLockManager::default()), + paths: Arc::new(ArcSwap::from_pointee(ConfigPaths { + home_path: String::new(), + config_path: String::new(), + storage_path: String::new(), + config_file_path: String::new(), + sources_file_path: String::new(), + mapping_file_path: None, + mapping_files_used: None, + template_file_path: None, + template_files_used: None, + api_proxy_file_path: String::new(), + custom_stream_response_path: None, + })), + custom_stream_response: Arc::new(ArcSwapOption::default()), + access_token_secret: [0; 32], + encrypt_secret: [0; 16], + media_tools: Arc::new(MediaToolCapabilities::new()), + } + } + fn create_test_app_state() -> Arc { create_test_app_state_for_config(Arc::new(create_test_app_config())) } @@ -2857,6 +3031,10 @@ mod tests { create_test_app_state_for_config(Arc::new(create_test_provider_app_config())) } + fn create_test_dual_provider_app_state() -> Arc { + create_test_app_state_for_config(Arc::new(create_test_dual_provider_app_config())) + } + fn create_test_app_state_for_config(app_cfg: Arc) -> Arc { let event_manager = Arc::new(EventManager::new()); let active_provider = Arc::new(ActiveProviderManager::new(&app_cfg, &event_manager)); @@ -3131,6 +3309,79 @@ mod tests { assert_eq!(grace_mode, None); } + #[tokio::test] + async fn resolve_admission_with_strategies_allows_existing_session_even_when_user_is_at_limit() { + let app_state = create_test_app_state(); + let addr = "127.0.0.1:55154".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let mut user = ProxyUserCredentials::default(); + user.username = "session-admission".to_string(); + user.max_connections = 1; + + app_state.connection_manager.add_connection(&addr).await; + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "vod-session", + virtual_id: 1, + provider: "provider-a", + stream_url: "http://provider-1.example/movie/1.mkv", + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + app_state + .connection_manager + .update_connection(crate::api::model::ConnectionParams { + meter_uid: 1, + username: &user.username, + max_connections: user.max_connections, + soft_connections: user.soft_connections, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 10, + fingerprint: &fingerprint, + provider: "provider-a", + stream_channel: &create_test_live_channel("http://provider-1.example/movie/1.mkv"), + user_agent: std::borrow::Cow::Borrowed("ua"), + session_token: Some("vod-session"), + }) + .await; + + let session_based = resolve_admission_with_strategies( + &app_state, + &user.username, + user.max_connections, + user.soft_connections, + &fingerprint.client_ip, + &fingerprint.addr, + true, + Some("vod-session"), + false, + EvictionReentryGuard::Session("vod-session"), + ) + .await; + assert_eq!(session_based.0.permission, UserConnectionPermission::Allowed); + + let connection_based = resolve_admission_with_strategies( + &app_state, + &user.username, + user.max_connections, + user.soft_connections, + &fingerprint.client_ip, + &fingerprint.addr, + false, + Some("vod-session"), + false, + EvictionReentryGuard::Session("vod-session"), + ) + .await; + assert_eq!(connection_based.0.permission, UserConnectionPermission::Exhausted); + } + #[tokio::test] async fn resolve_admission_with_strategies_prevents_recently_evicted_playback_ping_pong() { let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { @@ -4018,6 +4269,7 @@ mod tests { addr: &first_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, }) .await; app_state @@ -4031,6 +4283,7 @@ mod tests { addr: &second_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, }) .await; @@ -4095,6 +4348,7 @@ mod tests { addr: &first_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: true, }) .await; @@ -4158,6 +4412,7 @@ mod tests { addr: &first_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: true, }) .await; app_state @@ -4197,6 +4452,7 @@ mod tests { addr: &second_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(crate::api::model::ConnectionKind::Soft), + socket_bound: true, }) .await; app_state diff --git a/backend/src/api/config_file.rs b/backend/src/api/config_file.rs index a1f7ec001..e0df3025d 100644 --- a/backend/src/api/config_file.rs +++ b/backend/src/api/config_file.rs @@ -3,7 +3,7 @@ use crate::{ model::{Config, Mappings, ProcessTargets, SourcesConfig}, utils, utils::{ - prepare_sources_batch, read_config_file, read_mappings_file_unprepared, read_mappings_file_with_templates, + prepare_sources_batch, read_mappings_file_unprepared, read_mappings_file_with_templates, read_sources_file, read_sources_file_from_path_with_templates, read_templates, }, }; @@ -263,7 +263,7 @@ impl ConfigFile { let config_file = paths.config_file_path.clone(); // ── Parse ──────────────────────────────────────────────────── - let config_dto = read_config_file(config_file.as_str(), true, true)?; + let config_dto = crate::utils::read_config_file_with_options(config_file.as_str(), crate::utils::ReadConfigOptions::resolve_and_compute())?; let current_mapping_path = paths.mapping_file_path.clone().unwrap_or_else(|| { utils::resolve_mapping_file_path(paths.config_path.as_str(), None) diff --git a/backend/src/api/endpoints/hdhomerun_api.rs b/backend/src/api/endpoints/hdhomerun_api.rs index 3d3a96671..63fc05561 100644 --- a/backend/src/api/endpoints/hdhomerun_api.rs +++ b/backend/src/api/endpoints/hdhomerun_api.rs @@ -181,7 +181,7 @@ where fn create_device(app_state: &Arc) -> Option { if let Some(credentials) = app_state.app_state.app_config.get_user_credentials(&app_state.device.t_username) { - let server_info = app_state.app_state.app_config.get_user_server_info(&credentials); + let server_info = app_state.app_state.app_config.get_user_server_info(&credentials)?; let device = &app_state.device; let device_url = format!("{}://{}:{}", server_info.protocol, server_info.host, device.port); @@ -344,7 +344,9 @@ async fn lineup( .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) .body(axum::body::Body::from_stream(body_stream))); } else if (use_all || use_xtream) && target.has_output(TargetType::Xtream) { - let server_info = app_state.app_state.app_config.get_user_server_info(credentials); + let Some(server_info) = app_state.app_state.app_config.get_user_server_info(credentials) else { + return axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(); + }; let base_url = server_info.get_base_url(); let base_url_live = if credentials.proxy.is_redirect(PlaylistItemType::Live) diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index 48b494ffa..64d51a51a 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -104,6 +104,9 @@ pub(in crate::api) async fn handle_hls_stream_request( } let url = replace_url_extension(&normalized_hls_url, HLS_EXT); let server_info = app_state.app_config.get_user_server_info(user); + let Some(server_info) = server_info else { + return axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(); + }; let hls_session_ttl_secs = get_hls_session_ttl_secs(app_state); let (request_url, session_token, provider_handle) = if let Some(session) = user_session { @@ -129,29 +132,17 @@ pub(in crate::api) async fn handle_hls_stream_request( Some(handle) } else { debug_if_enabled!( - "HLS pinned provider {} unavailable for {}; falling back to lineup allocation", + "HLS pinned provider {} unavailable for {}; aborting allocation to prevent mid-session migration", sanitize_sensitive_info(&session.provider), sanitize_sensitive_info(&fingerprint.addr.to_string()) ); - app_state - .active_provider - .acquire_connection_with_grace( - &input.name, - &fingerprint.addr, - false, - connection_priority_for_kind( - user, - session - .connection_kind - .unwrap_or(connection_kind), - ), - session - .connection_kind - .unwrap_or(connection_kind), - ) - .await + None }; + if provider_handle.is_none() { + return axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + match provider_handle.as_ref().map(|handle| &handle.allocation) { Some(ProviderAllocation::Exhausted) => (url, None, provider_handle), Some(ProviderAllocation::Available(cfg) | ProviderAllocation::GracePeriod(cfg)) => { @@ -167,6 +158,7 @@ pub(in crate::api) async fn handle_hls_stream_request( addr: &fingerprint.addr, connection_permission, connection_kind: session.connection_kind, + socket_bound: PlaylistItemType::LiveHls.uses_socket_bound_session(), }) .await; app_state @@ -211,6 +203,7 @@ pub(in crate::api) async fn handle_hls_stream_request( addr: &fingerprint.addr, connection_permission, connection_kind: Some(connection_kind), + socket_bound: PlaylistItemType::LiveHls.uses_socket_bound_session(), }) .await; app_state diff --git a/backend/src/api/endpoints/m3u_api.rs b/backend/src/api/endpoints/m3u_api.rs index 266780071..3fe083fb0 100644 --- a/backend/src/api/endpoints/m3u_api.rs +++ b/backend/src/api/endpoints/m3u_api.rs @@ -182,7 +182,8 @@ async fn m3u_api_stream( debug_if_enabled!( "ID chain for m3u endpoint: request_stream_id={} -> action_stream_id={action_stream_id} -> req_virtual_id={req_virtual_id} -> virtual_id={virtual_id}", stream_req.stream_id); - let extension = stream_ext.clone().unwrap_or_else(|| extract_extension_from_url(&pli.url).unwrap_or_default()); + let extracted_ext = extract_extension_from_url(&pli.url).unwrap_or_default(); + let extension = stream_ext.unwrap_or(extracted_ext.as_str()); let session_key = if pli.item_type == PlaylistItemType::Catchup { create_catchup_session_key(fingerprint, &user.username, virtual_id) } else { @@ -190,11 +191,11 @@ async fn m3u_api_stream( fingerprint, &user.username, virtual_id, - !is_session_based_playback(pli.item_type, Some(extension.as_str())), + !is_session_based_playback(pli.item_type, Some(extension)), ) }; let eviction_reentry_guard = if pli.item_type == PlaylistItemType::Catchup - || is_session_based_playback(pli.item_type, Some(extension.as_str())) + || is_session_based_playback(pli.item_type, Some(extension)) { crate::api::api_utils::EvictionReentryGuard::Session(&session_key) } else { @@ -307,7 +308,7 @@ async fn m3u_api_stream( target: &target, input: &input, user: &user, - stream_ext: stream_ext.as_deref(), + stream_ext, req_context: context, action_path: "", // TODO is there timeshift or something like that ? }; @@ -316,7 +317,7 @@ async fn m3u_api_stream( return response.into_response(); } - let is_session_request = is_session_based_playback(pli.item_type, Some(extension.as_str())); + let is_session_request = is_session_based_playback(pli.item_type, Some(extension)); // Reverse proxy mode — only route genuine HLS into the HLS handler, not DASH if is_session_request && extension == shared::utils::HLS_EXT { return handle_hls_stream_request( diff --git a/backend/src/api/endpoints/v1_api_playlist.rs b/backend/src/api/endpoints/v1_api_playlist.rs index bcc1ce560..449869bfb 100644 --- a/backend/src/api/endpoints/v1_api_playlist.rs +++ b/backend/src/api/endpoints/v1_api_playlist.rs @@ -196,25 +196,25 @@ async fn load_epg_channels_for_input( }; let storage_dir = app_state.app_config.config.load().storage_dir.clone(); - let mut channels_by_source = Vec::new(); - let mut failed_sources = 0usize; + let mut channels_by_source = Vec::new(); + let mut failed_sources = 0usize; - for epg_source in &epg_config.sources { - let resolved_url = resolve_provider_url_with_input(input, &epg_source.url); - let raw_epg_path = match get_input_raw_epg_file_path(&resolved_url, input, &storage_dir).await { - Ok(path) => path, - Err(err) => { - debug!( - "Skipping EPG source {}: {}", - sanitize_sensitive_info(resolved_url.as_str()), - sanitize_sensitive_info(&err.to_string()) - ); - failed_sources += 1; - continue; - } - }; + for epg_source in &epg_config.sources { + let resolved_url = resolve_provider_url_with_input(input, &epg_source.url); + let raw_epg_path = match get_input_raw_epg_file_path(&resolved_url, input, &storage_dir).await { + Ok(path) => path, + Err(err) => { + debug!( + "Skipping EPG source {}: {}", + sanitize_sensitive_info(resolved_url.as_str()), + sanitize_sensitive_info(&err.to_string()) + ); + failed_sources += 1; + continue; + } + }; - let source_channels = if file_exists_async(&raw_epg_path).await { + let source_channels = if file_exists_async(&raw_epg_path).await { match parse_xmltv_for_web_ui_from_file(&raw_epg_path).await { Ok(ch) => ch, Err(file_err) => { @@ -430,6 +430,9 @@ fn playlist_webplayer( .and_then(|web_ui| web_ui.player_server.as_ref()) .map_or("default", |server_name| server_name.as_str()); let server_info = app_state.app_config.get_server_info(server_name); + let Some(server_info) = server_info else { + return axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(); + }; let base_url = server_info.get_base_url(); build_playlist_webplayer_url(&base_url, &access_token, target_id, virtual_id, cluster).into_response() } @@ -604,11 +607,19 @@ mod tests { utils::Internable, }; use std::sync::Arc; + use chrono::Utc; use tempfile::tempdir; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use tower::ServiceExt; + /// Generate an XMLTV datetime string in the format `YYYYMMDDHHmmss +0000` + /// offset by `hours_from_now` hours from the current time. + fn epg_dt(hours_from_now: i64) -> String { + let dt = Utc::now() + chrono::Duration::hours(hours_from_now); + dt.format("%Y%m%d%H%M%S %z").to_string() + } + fn test_app_config(input: Arc, source: ConfigSource) -> AppConfig { let sources = SourcesConfig { batch_files: vec![], @@ -1022,17 +1033,21 @@ mod tests { if let Some(parent) = raw_epg_path.parent() { tokio::fs::create_dir_all(parent).await.expect("epg dir"); } + let prog_start = epg_dt(0); + let prog_stop = epg_dt(1); tokio::fs::write( &raw_epg_path, - r#" + format!( + r#" Demo Channel - + Morning Show -"#, +"# + ), ) .await .expect("write epg"); @@ -1099,17 +1114,21 @@ mod tests { if let Some(parent) = raw_epg_path.parent() { tokio::fs::create_dir_all(parent).await.expect("epg dir"); } + let prog_start = epg_dt(0); + let prog_stop = epg_dt(1); tokio::fs::write( &raw_epg_path, - r#" + format!( + r#" Demo Channel - + Morning Show -"#, +"# + ), ) .await .expect("write epg"); @@ -1256,51 +1275,62 @@ mod tests { } } + let p0 = epg_dt(0); + let p1 = epg_dt(1); + let p2 = epg_dt(2); + let p3 = epg_dt(3); + tokio::fs::write( &secondary_path, - r#" + format!( + r#" Secondary Channel - + Secondary Show -"#, +"# + ), ) .await .expect("write secondary epg"); tokio::fs::write( &primary_path, - r#" + format!( + r#" Primary Channel - + Primary Show -"#, +"# + ), ) .await .expect("write primary epg"); tokio::fs::write( &same_priority_path, - r#" + format!( + r#" Same Priority Channel - + Duplicate Show - + Second Show -"#, +"# + ), ) .await .expect("write same priority epg"); diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index d606f7de3..3c42f9019 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -33,6 +33,7 @@ use std::{ use tokio::{io::AsyncWriteExt, sync::mpsc, task}; use tokio_stream::StreamExt; use tokio_util::io::ReaderStream; +use crate::model::ApiProxyServerInfo; pub fn get_empty_epg_response() -> axum::response::Response { try_unwrap_body!(axum::response::Response::builder() @@ -176,19 +177,21 @@ async fn serve_epg_with_rewrites( let server_info = app_state.app_config.get_user_server_info(user); let base_url = if !matches!(epg_processing_options.time_shift, EpgTimeShift::None) || epg_processing_options.rewrite_urls { - Some(concat_string!( - &server_info.get_base_url(), - "/", - storage_const::EPG_RESOURCE_PATH, - "/", - &user.username, - "/", - &user.password - )) + server_info.as_ref().map(|si| concat_string!( + &si.get_base_url(), + "/", + storage_const::EPG_RESOURCE_PATH, + "/", + &user.username, + "/", + &user.password + )) } else { None }; + let generator_info = server_info.as_ref().map(ApiProxyServerInfo::get_base_url).unwrap_or_default(); + let limit = limit.unwrap_or_default(); let bg_lock = app_state.app_config.file_locks.read_lock(epg_path).await; @@ -215,9 +218,6 @@ async fn serve_epg_with_rewrites( } }); - - let generator_info = server_info.get_base_url(); - let (mut tx, rx) = tokio::io::duplex(8192); tokio::spawn(async move { // Work-Around BytesText DocType escape, see below diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index 007d1902e..eb689ce60 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -188,16 +188,16 @@ pub(in crate::api) fn get_xtream_player_api_stream_url( } } -async fn get_user_info(user: &ProxyUserCredentials, app_state: &AppState) -> XtreamAuthorizationResponse { - let server_info = app_state.app_config.get_user_server_info(user); +async fn get_user_info(user: &ProxyUserCredentials, app_state: &AppState) -> Option { + let server_info = app_state.app_config.get_user_server_info(user)?; let active_connections = app_state.get_active_connections_for_user(&user.username).await; - XtreamAuthorizationResponse::new( + Some(XtreamAuthorizationResponse::new( &server_info, user, active_connections, app_state.app_config.config.load().user_access_control, - ) + )) } #[allow(clippy::too_many_lines)] @@ -259,8 +259,8 @@ async fn xtream_player_api_stream( return axum::http::StatusCode::BAD_REQUEST.into_response(); } - let stream_ext = stream_ext.filter(|s| !s.is_empty()) - .or_else(|| pli.get_container_extension().map(|e| concat_string!(".", e.as_ref()))); + let container_ext = pli.get_container_extension().map(|e| concat_string!(".", e.as_ref())); + let stream_ext = stream_ext.filter(|s| !s.is_empty()).or(container_ext.as_deref()); let input = try_option_bad_request!( app_state.app_config.get_input_by_name(&pli.input_name), @@ -338,11 +338,24 @@ async fn xtream_player_api_stream( debug_if_enabled!( "ID chain for xtream endpoint: request_stream_id={} -> action_stream_id={action_stream_id} -> req_virtual_id={req_virtual_id} -> virtual_id={virtual_id}", stream_req.stream_id); + let container_ext = pli.get_container_extension().map(|ext| concat_string!(".", ext.as_ref())); let requested_extension = stream_ext - .clone() .filter(|ext| !ext.is_empty()) - .or_else(|| pli.get_container_extension().map(|ext| concat_string!(".", ext.as_ref()))) + .or(container_ext.as_deref()) .unwrap_or_default(); + + // Derive the playback extension from get_query_path so session semantics match + // the actual path it will route. Falls back to requested_extension when empty. + #[allow(clippy::needless_borrow, clippy::borrow_deref_ref)] + let (_, playback_ext) = + get_query_path(stream_req.action_path, Some(&requested_extension), &pli, app_state); + #[allow(clippy::needless_borrow, clippy::borrow_deref_ref)] + let playback_ext: &str = if playback_ext.is_empty() { + &requested_extension + } else { + &playback_ext + }; + let session_key = if item_type == PlaylistItemType::Catchup { create_catchup_session_key(fingerprint, &user.username, virtual_id) } else { @@ -350,11 +363,11 @@ async fn xtream_player_api_stream( fingerprint, &user.username, virtual_id, - !is_session_based_playback(item_type, Some(requested_extension.as_str())), + !is_session_based_playback(item_type, Some(playback_ext)), ) }; let eviction_reentry_guard = if item_type == PlaylistItemType::Catchup - || is_session_based_playback(item_type, Some(requested_extension.as_str())) + || is_session_based_playback(item_type, Some(playback_ext)) { crate::api::api_utils::EvictionReentryGuard::Session(&session_key) } else { @@ -471,7 +484,7 @@ async fn xtream_player_api_stream( target: &target, input: &input, user: &user, - stream_ext: stream_ext.as_deref(), + stream_ext, req_context: context, action_path: stream_req.action_path, }; @@ -479,7 +492,8 @@ async fn xtream_player_api_stream( return response.into_response(); } - let (query_path, _extension) = get_query_path(stream_req.action_path, stream_ext.as_ref(), &pli, app_state); + #[allow(clippy::needless_borrow)] + let (query_path, _extension) = get_query_path(stream_req.action_path, Some(&requested_extension), &pli, app_state); let stream_url = try_option_bad_request!( get_xtream_player_api_stream_url(&input, stream_req.context, &query_path, &session_url), @@ -490,9 +504,9 @@ async fn xtream_player_api_stream( ) ); - let is_session_request = is_session_based_playback(item_type, Some(requested_extension.as_str())); + let is_session_request = is_session_based_playback(item_type, Some(playback_ext)); // Reverse proxy mode — only route genuine HLS into the HLS handler, not DASH - if is_session_request && requested_extension == shared::utils::HLS_EXT { + if is_session_request && playback_ext == shared::utils::HLS_EXT { return handle_hls_stream_request( fingerprint, app_state, @@ -532,7 +546,7 @@ async fn xtream_player_api_stream( pub(crate) fn get_query_path( action_path: &str, - stream_ext: Option<&String>, + stream_ext: Option<&str>, pli: &XtreamPlaylistItem, app_state: &Arc, ) -> (String, String) { @@ -620,14 +634,20 @@ async fn xtream_player_api_stream_with_token( .into_response(); } + let container_ext = pli.get_container_extension().map(|e| concat_string!(".", e.as_ref())); let requested_extension = stream_ext - .as_ref().filter(|s| !s.is_empty()).cloned() - .or_else(|| pli.get_container_extension().map(|e| concat_string!(".", e.as_ref()))) - .unwrap_or_default(); + .filter(|s| !s.is_empty()) + .or(container_ext.as_deref()); - let (query_path, _extension) = get_query_path(stream_req.action_path, stream_ext.as_ref(), &pli, app_state); + let (query_path, playback_ext) = + get_query_path(stream_req.action_path, requested_extension, &pli, app_state); + let playback_ext: Option<&str> = if playback_ext.is_empty() { + requested_extension + } else { + Some(&*playback_ext) + }; - let is_session_request = is_session_based_playback(pli.item_type, Some(requested_extension.as_str())); + let is_session_request = is_session_based_playback(pli.item_type, playback_ext); let session_key = create_session_fingerprint( fingerprint, "webui", @@ -638,7 +658,7 @@ async fn xtream_player_api_stream_with_token( // TODO how should we use fixed provider for hls in multi provider config? // Reverse proxy mode — only route genuine HLS into the HLS handler, not DASH - if is_session_request && requested_extension == shared::utils::HLS_EXT { + if is_session_request && playback_ext == Some(shared::utils::HLS_EXT) { return handle_hls_stream_request( fingerprint, app_state, @@ -1251,12 +1271,16 @@ async fn xtream_player_api(api_req: UserApiRequest, app_state: &Arc) - format!("Could not find any user for xc player api {}", api_req.username) ); if !target.has_output(TargetType::Xtream) { - return axum::response::Json(get_user_info(&user, app_state).await).into_response(); + return get_user_info(&user, app_state) + .await + .map_or_else(|| axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(), |info| axum::response::Json(info).into_response()); } let action = api_req.action.trim(); if action.is_empty() { - return axum::response::Json(get_user_info(&user, app_state).await).into_response(); + return get_user_info(&user, app_state) + .await + .map_or_else(|| axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(), |info| axum::response::Json(info).into_response()); } if user.permission_denied(app_state) { @@ -1280,7 +1304,9 @@ async fn xtream_player_api(api_req: UserApiRequest, app_state: &Arc) - match action { crate::model::XC_ACTION_GET_ACCOUNT_INFO => { - return axum::response::Json(get_user_info(&user, app_state).await).into_response(); + return get_user_info(&user, app_state) + .await + .map_or_else(|| axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(), |info| axum::response::Json(info).into_response()); } crate::model::XC_ACTION_GET_SERIES_INFO => { skip_json_response_if_flag_set!( @@ -1379,7 +1405,9 @@ async fn xtream_player_api(api_req: UserApiRequest, app_state: &Arc) - } Err(err) => { error!("Failed response for xtream target: {} action: {} error: {}", &target.name, action, err); - axum::response::Json(get_user_info(&user, app_state).await).into_response() + get_user_info(&user, app_state) + .await + .map_or_else(|| axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response(), |info| axum::response::Json(info).into_response()) } } } diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index eb7d45737..f85d729b0 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -365,7 +365,12 @@ async fn cancel_all_service_tokens(app_state: &Arc) { // Use the manager's shutdown() rather than cancelling the token directly so // the is_shutdown flag is set and workers do not attempt to restart after cancellation. app_state.metadata_manager.shutdown(); - app_state.connection_manager.shutdown().await; + if tokio::time::timeout(std::time::Duration::from_secs(30), app_state.connection_manager.shutdown()) + .await + .is_err() + { + warn!("Connection manager shutdown timed out after 30s, forcing exit"); + } } #[cfg(unix)] diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index ab6078b57..400d7f7eb 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -81,7 +81,10 @@ pub struct UserSession { pub provider: Arc, pub stream_url: Arc, pub addr: SocketAddr, + pub socket_bound: bool, + pub active_addrs: Vec, pub ts: u64, + pub started_at: u64, pub permission: UserConnectionPermission, pub connection_kind: Option, pub counted: bool, @@ -149,6 +152,22 @@ impl UserConnectionData { } } + fn has_session_addr(&self, addr: &SocketAddr) -> bool { + self.sessions + .iter() + .any(|session| session.addr == *addr || session.active_addrs.contains(addr)) + } + + fn release_addr_from_sessions(&mut self, addr: &SocketAddr) -> HashMap> { + let mut migrated_addrs = HashMap::new(); + for session in &mut self.sessions { + if session.addr == *addr || session.active_addrs.contains(addr) { + migrated_addrs.insert(session.token.clone(), release_session_addr(session, addr)); + } + } + migrated_addrs + } + fn increment_kind(&mut self, kind: ConnectionKind) { self.connections = self.connections.saturating_add(1); match kind { @@ -316,6 +335,34 @@ pub struct CreateUserSessionParams<'a> { pub addr: &'a SocketAddr, pub connection_permission: UserConnectionPermission, pub connection_kind: Option, + pub socket_bound: bool, +} + +fn remember_session_addr(session: &mut UserSession, addr: SocketAddr) { + if session.socket_bound { + session.active_addrs.clear(); + } else if let Some(position) = session.active_addrs.iter().position(|active_addr| *active_addr == addr) { + session.active_addrs.remove(position); + } + session.active_addrs.push(addr); + session.addr = addr; +} + +fn release_session_addr(session: &mut UserSession, addr: &SocketAddr) -> Option { + if let Some(position) = session.active_addrs.iter().position(|active_addr| active_addr == addr) { + session.active_addrs.remove(position); + } else if session.addr != *addr { + return None; + } + + if session.addr == *addr { + if let Some(next_addr) = session.active_addrs.last().copied() { + session.addr = next_addr; + return Some(next_addr); + } + } + + None } impl SocketRegistration { @@ -345,6 +392,7 @@ pub struct ActiveUserManager { last_logged_user_connection_count: AtomicUsize, cleanup_tx: tokio::sync::OnceCell>, provider_manager: tokio::sync::OnceCell>, + pub(crate) dropped_cleanup_events: AtomicU64, } impl ActiveUserManager { @@ -407,6 +455,7 @@ impl ActiveUserManager { last_logged_user_connection_count: AtomicUsize::new(0), cleanup_tx: tokio::sync::OnceCell::new(), provider_manager: tokio::sync::OnceCell::new(), + dropped_cleanup_events: AtomicU64::new(0), } } @@ -477,12 +526,22 @@ impl ActiveUserManager { let mut connection_changed = false; let mut promotion = None; if let Some(connection_data) = user_connections.by_key.get_mut(&username) { + let migrated_session_addrs = connection_data.release_addr_from_sessions(addr); if let Some(stream_idx) = connection_data .streams .iter() .position(|stream| stream.addr == *addr && !stream.preserved) { - if Self::should_preserve_session_stream(&connection_data.streams[stream_idx]) { + let migrated_addr = connection_data.streams[stream_idx] + .session_token + .as_deref() + .and_then(|token| migrated_session_addrs.get(token)) + .copied() + .flatten(); + if let Some(next_addr) = migrated_addr { + connection_data.streams[stream_idx].addr = next_addr; + connection_data.streams[stream_idx].ts = current_time_secs(); + } else if Self::should_preserve_session_stream(&connection_data.streams[stream_idx]) { if let Some(entry) = self.build_preserved_stream_expiry( &username, &connection_data.streams[stream_idx], @@ -550,6 +609,7 @@ impl ActiveUserManager { removed_stream } + #[allow(clippy::too_many_lines)] async fn release_connection_inner(&self, addr: &SocketAddr, preserve_session_streams: bool) -> ReleasedConnection { let (addr_removed, disconnected_user, removed_streams, expiry_entries, promotions) = { let mut user_connections = self.connections.write().await; @@ -560,12 +620,24 @@ impl ActiveUserManager { let mut expiry_entries = Vec::new(); let mut promotions = Vec::new(); if let Some(connection_data) = user_connections.by_key.get_mut(&username) { + let migrated_session_addrs = connection_data.release_addr_from_sessions(addr); let mut remaining_streams = Vec::with_capacity(connection_data.streams.len()); let mut released_kinds = Vec::new(); let mut removed_session_tokens = HashSet::new(); + let now = current_time_secs(); for mut stream_info in connection_data.streams.drain(..) { if stream_info.addr == *addr { - if preserve_session_streams && Self::should_preserve_session_stream(&stream_info) { + let migrated_addr = stream_info + .session_token + .as_deref() + .and_then(|token| migrated_session_addrs.get(token)) + .copied() + .flatten(); + if let Some(next_addr) = migrated_addr { + stream_info.addr = next_addr; + stream_info.ts = now; + remaining_streams.push(stream_info); + } else if preserve_session_streams && Self::should_preserve_session_stream(&stream_info) { if let Some(entry) = self.build_preserved_stream_expiry(&username, &stream_info, &connection_data.sessions) { @@ -971,6 +1043,13 @@ impl ActiveUserManager { connection_data.max_connections = max_connections; connection_data.soft_connections = soft_connections; + if let Some(token) = session_token { + if let Some(session) = connection_data.sessions.iter_mut().find(|session| session.token == token) { + session.ts = now; + remember_session_addr(session, fingerprint.addr); + } + } + let user_agent_string = user_agent.to_string(); let reserved_session_kind = session_token.and_then(|token| { connection_data @@ -988,6 +1067,12 @@ impl ActiveUserManager { None => stream_info.addr == fingerprint.addr && stream_info.session_token.is_none(), }) .map(|stream_idx| { + let session_started_at = session_token.and_then(|token| { + connection_data.sessions.iter() + .find(|s| s.token == token) + .map(|s| s.started_at) + }); + let stream_info = &mut connection_data.streams[stream_idx]; let client_ip = fingerprint.client_ip.clone(); let preserve_started_at = stream_info.session_token.is_some() @@ -1001,6 +1086,11 @@ impl ActiveUserManager { stream_info.channel = stream_channel.clone(); stream_info.provider = provider.to_string(); stream_info.user_agent.clone_from(&user_agent_string); + + if let Some(started_at) = session_started_at { + stream_info.started_at = started_at; + } + if preserve_started_at { let now = current_time_secs(); if utc_day_from_secs(stream_info.ts) != utc_day_from_secs(now) { @@ -1022,7 +1112,6 @@ impl ActiveUserManager { stream_info.previous_session_id = None; result }); - if let Some(stream_info) = existing_stream_info { if let Some(token) = session_token { if let Some(session) = connection_data.sessions.iter_mut().find(|session| session.token == token) { @@ -1037,7 +1126,7 @@ impl ActiveUserManager { let effective_connection_kind = reserved_session_kind.unwrap_or(connection_kind); let country_code = self.lookup_country(&fingerprint.client_ip); - let stream_info = StreamInfo::new( + let mut stream_info = StreamInfo::new( uid, meter_uid, username, @@ -1050,6 +1139,12 @@ impl ActiveUserManager { session_token, ); + if let Some(token) = session_token { + if let Some(session) = connection_data.sessions.iter().find(|s| s.token == token) { + stream_info.started_at = session.started_at; + } + } + if reserved_session_kind.is_none() { connection_data.increment_kind(effective_connection_kind); } @@ -1113,6 +1208,8 @@ impl ActiveUserManager { } } + + #[allow(clippy::too_many_arguments)] fn new_user_session( session_token: &str, virtual_id: u32, @@ -1121,14 +1218,19 @@ impl ActiveUserManager { addr: &SocketAddr, connection_permission: UserConnectionPermission, connection_kind: Option, + socket_bound: bool, ) -> UserSession { + let now = current_time_secs(); UserSession { token: session_token.to_string(), virtual_id, provider: provider.intern(), stream_url: stream_url.intern(), addr: *addr, - ts: current_time_secs(), + socket_bound, + active_addrs: vec![*addr], + ts: now, + started_at: now, permission: connection_permission, connection_kind, counted: false, @@ -1239,6 +1341,7 @@ impl ActiveUserManager { addr, connection_permission, connection_kind, + socket_bound, } = request; self.gc(); @@ -1251,7 +1354,8 @@ impl ActiveUserManager { if let Some(session) = connection_data.sessions.iter_mut().find(|session| session.token == session_token) { session.ts = current_time_secs(); - session.addr = *addr; + session.socket_bound = socket_bound; + remember_session_addr(session, *addr); if session.connection_kind.is_none() { session.connection_kind = connection_kind; } @@ -1269,6 +1373,7 @@ impl ActiveUserManager { addr, connection_permission, connection_kind, + socket_bound, )); true } @@ -1350,6 +1455,7 @@ impl ActiveUserManager { addr, connection_permission, connection_kind, + socket_bound, } = request; self.gc(); @@ -1367,6 +1473,7 @@ impl ActiveUserManager { addr, connection_permission, connection_kind, + socket_bound, ); data.add_session(session); data @@ -1376,7 +1483,8 @@ impl ActiveUserManager { for session in &mut connection_data.sessions { if session.token == session_token { session.ts = current_time_secs(); - session.addr = *addr; + session.socket_bound = socket_bound; + remember_session_addr(session, *addr); if &*session.stream_url != stream_url { session.stream_url = stream_url.intern(); } @@ -1409,6 +1517,7 @@ impl ActiveUserManager { addr, connection_permission, connection_kind, + socket_bound, ); let token = session.token.clone(); connection_data.add_session(session); @@ -1421,7 +1530,7 @@ impl ActiveUserManager { if let Some(connection_data) = user_connections.by_key.get_mut(username) { let update_result = if let Some(session) = connection_data.sessions.iter_mut().find(|s| s.token == token) { let previous_addr = session.addr; - session.addr = *addr; + remember_session_addr(session, *addr); session.ts = now; for stream in &mut connection_data.streams { if stream.addr == previous_addr { @@ -1430,7 +1539,7 @@ impl ActiveUserManager { } } let prune_previous_registration = previous_addr != *addr - && !connection_data.sessions.iter().any(|active_session| active_session.addr == previous_addr) + && !connection_data.has_session_addr(&previous_addr) && !connection_data.streams.iter().any(|stream| stream.addr == previous_addr); Some((previous_addr, prune_previous_registration)) } else { @@ -1505,7 +1614,7 @@ impl ActiveUserManager { for session in &mut connection_data.sessions { if session.token == token { session.ts = now; - session.addr = *addr; + remember_session_addr(session, *addr); touched_session = true; break; } @@ -1892,6 +2001,7 @@ impl ActiveUserManager { if let Some(tx) = self.cleanup_tx.get() { for (addr, stream_info) in cleanup_events { if tx.try_send(CleanupEvent::AdaptiveSessionExpired { stream_info }).is_err() { + self.dropped_cleanup_events.fetch_add(1, Ordering::Relaxed); debug!("Cleanup channel unavailable, dropping adaptive session expiry"); removed_addrs.push(addr); } @@ -2265,6 +2375,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -2319,6 +2430,7 @@ mod tests { addr: &kicked_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: true, }) .await; manager @@ -2331,6 +2443,7 @@ mod tests { addr: &survivor_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: true, }) .await; @@ -2451,6 +2564,7 @@ mod tests { addr: &first_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: true, }) .await; manager @@ -2527,6 +2641,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Soft), + socket_bound: true, }) .await; manager @@ -2657,6 +2772,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; let first = manager @@ -2750,6 +2866,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -2803,6 +2920,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -2878,6 +2996,7 @@ mod tests { addr: &normal_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -2968,6 +3087,7 @@ mod tests { addr: &addr_a, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -3051,6 +3171,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -3143,6 +3264,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -3225,6 +3347,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -3296,6 +3419,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -3380,6 +3504,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; let first = manager @@ -3571,6 +3696,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; @@ -3619,6 +3745,7 @@ mod tests { addr: &addr1, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; @@ -3669,7 +3796,7 @@ mod tests { } #[tokio::test] - async fn update_session_addr_prunes_previous_registration_when_session_moves_to_new_socket() { + async fn update_session_addr_prunes_previous_registration_for_socket_bound_session() { let config = Config::default(); let geoip = Arc::new(ArcSwapOption::::default()); let event_manager = Arc::new(EventManager::new()); @@ -3690,10 +3817,11 @@ mod tests { session_token: "tok-move", virtual_id: 9101, provider: "provider-a", - stream_url: "http://localhost/movie.mkv", + stream_url: "http://localhost/live.ts", addr: &old_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: true, }) .await; manager @@ -3709,14 +3837,14 @@ mod tests { fingerprint: &old_fingerprint, provider: "provider-a", stream_channel: &StreamChannel { - item_type: PlaylistItemType::Video, + item_type: PlaylistItemType::Live, ..test_channel(9101) }, user_agent: Cow::Borrowed("ua"), session_token: Some("tok-move"), }) .await - .expect("initial movie stream should register"); + .expect("initial live stream should register"); manager.update_session_addr("user1", "tok-move", &new_addr).await; @@ -3734,6 +3862,122 @@ mod tests { assert_eq!(connection_data.streams[0].addr, new_addr); } + #[tokio::test] + async fn vod_session_survives_overlapping_and_seek_sockets() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let base_addr: SocketAddr = "127.0.0.1:55131".parse().unwrap(); + let range_addr: SocketAddr = "127.0.0.1:55132".parse().unwrap(); + let seek_addr: SocketAddr = "127.0.0.1:55133".parse().unwrap(); + let base_fingerprint = Fingerprint::new("fp-vod-base".to_string(), "127.0.0.1".to_string(), base_addr); + let range_fingerprint = Fingerprint::new("fp-vod-range".to_string(), "127.0.0.1".to_string(), range_addr); + let mut user = ProxyUserCredentials::default(); + user.username = String::from("user1"); + user.max_connections = 1; + + manager.add_connection(&base_addr).await; + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: "tok-vod", + virtual_id: 9102, + provider: "provider-a", + stream_url: "http://localhost/movie.mkv", + addr: &base_addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, + }) + .await; + manager + .update_connection(ActiveUserConnectionParams { + uid: 302, + meter_uid: 402, + username: "user1", + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &base_fingerprint, + provider: "provider-a", + stream_channel: &StreamChannel { + item_type: PlaylistItemType::Video, + ..test_channel(9102) + }, + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-vod"), + }) + .await + .expect("initial vod stream should register"); + + manager.add_connection(&range_addr).await; + manager + .update_connection(ActiveUserConnectionParams { + uid: 303, + meter_uid: 403, + username: "user1", + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &range_fingerprint, + provider: "provider-a", + stream_channel: &StreamChannel { + item_type: PlaylistItemType::Video, + ..test_channel(9102) + }, + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-vod"), + }) + .await + .expect("overlapping range request should reuse the same vod session"); + + assert_eq!(manager.user_connections("user1").await, 1); + assert!(manager.release_stream(&range_addr).await.is_none()); + let released = manager.release_connection(&range_addr).await; + assert!(released.addr_removed); + assert!(released.removed_streams.is_empty()); + + { + let connections = manager.connections.read().await; + assert!(connections.key_by_addr.contains_key(&base_addr)); + let connection_data = connections.by_key.get("user1").expect("user connection data"); + assert_eq!(connection_data.sessions[0].addr, base_addr); + assert_eq!(connection_data.streams[0].addr, base_addr); + } + + manager.add_connection(&seek_addr).await; + manager.update_session_addr("user1", "tok-vod", &seek_addr).await; + + { + let connections = manager.connections.read().await; + assert!( + connections.key_by_addr.contains_key(&base_addr), + "existing vod socket must remain registered while the session spans multiple requests" + ); + assert!(connections.key_by_addr.contains_key(&seek_addr)); + + let connection_data = connections.by_key.get("user1").expect("user connection data"); + assert_eq!(connection_data.sessions[0].addr, seek_addr); + assert_eq!(connection_data.streams[0].addr, seek_addr); + } + + assert!(manager.release_stream(&seek_addr).await.is_none()); + let released = manager.release_connection(&seek_addr).await; + assert!(released.addr_removed); + assert!(released.removed_streams.is_empty()); + + let connections = manager.connections.read().await; + let connection_data = connections.by_key.get("user1").expect("user connection data"); + assert_eq!(connection_data.sessions[0].addr, base_addr); + assert_eq!(connection_data.streams[0].addr, base_addr); + } + #[tokio::test] async fn gc_keeps_active_ts_streams_even_when_user_timestamp_is_stale() { let config = Config::default(); @@ -3806,6 +4050,7 @@ mod tests { addr: &first_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -3818,6 +4063,7 @@ mod tests { addr: &second_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; @@ -3883,6 +4129,7 @@ mod tests { addr: &first_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; manager @@ -3895,6 +4142,7 @@ mod tests { addr: &second_addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; @@ -3988,6 +4236,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(ConnectionKind::Normal), + socket_bound: false, }) .await; diff --git a/backend/src/api/model/connection_manager.rs b/backend/src/api/model/connection_manager.rs index 5ca236e35..23f8d0aa0 100644 --- a/backend/src/api/model/connection_manager.rs +++ b/backend/src/api/model/connection_manager.rs @@ -21,7 +21,7 @@ use std::{ net::SocketAddr, str::FromStr, sync::{ - atomic::{AtomicU32, Ordering}, + atomic::{AtomicU32, AtomicU64, Ordering}, Arc, Mutex, MutexGuard, }, thread, @@ -51,6 +51,7 @@ struct BackpressureSender { state: Arc>>, queue_name: &'static str, overflow_capacity: usize, + dropped_events: AtomicU64, } impl BackpressureSender @@ -66,6 +67,7 @@ where })), queue_name, overflow_capacity, + dropped_events: AtomicU64::new(0), } } @@ -75,10 +77,13 @@ where let mut state = lock_backpressure_state(self.state.as_ref()); if state.draining { if state.overflow.len() >= self.overflow_capacity { - warn!( - "{} overflow buffer full (capacity={}), dropping event", - self.queue_name, self.overflow_capacity - ); + let count = self.dropped_events.fetch_add(1, Ordering::Relaxed) + 1; + if count == 1 || count.is_multiple_of(1024) { + warn!( + "{} overflow buffer full (capacity={}), {} events dropped total", + self.queue_name, self.overflow_capacity, count + ); + } return; } state.overflow.push_back(event); @@ -90,10 +95,13 @@ where Err(tokio::sync::mpsc::error::TrySendError::Full(event)) => { state.draining = true; if state.overflow.len() >= self.overflow_capacity { - warn!( - "{} overflow buffer full (capacity={}), dropping event", - self.queue_name, self.overflow_capacity - ); + let count = self.dropped_events.fetch_add(1, Ordering::Relaxed) + 1; + if count == 1 || count.is_multiple_of(1024) { + warn!( + "{} overflow buffer full (capacity={}), {} events dropped total", + self.queue_name, self.overflow_capacity, count + ); + } state.draining = false; return; } @@ -164,6 +172,10 @@ where state.overflow.clear(); state.draining = false; } + + fn dropped_count(&self) -> u64 { + self.dropped_events.load(Ordering::Relaxed) + } } fn lock_backpressure_state(state: &Mutex>) -> MutexGuard<'_, BackpressureState> { @@ -803,6 +815,8 @@ impl ConnectionManager { pub(crate) fn send_cleanup(&self, event: CleanupEvent) { self.cleanup_sender.enqueue(event); } + pub fn dropped_cleanup_events(&self) -> u64 { self.cleanup_sender.dropped_count() + self.user_manager.dropped_cleanup_events.load(Ordering::Relaxed) } + pub fn get_close_connection_channel(&self) -> tokio::sync::broadcast::Receiver { self.close_socket_signal_tx.subscribe() } pub async fn kick_connection(&self, addr: &SocketAddr, virtual_id: VirtualId, block_secs: u64) -> bool { @@ -1294,6 +1308,7 @@ mod tests { addr: &addr, connection_permission: UserConnectionPermission::Allowed, connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: true, }) .await; diff --git a/backend/src/api/model/qos_aggregation_manager.rs b/backend/src/api/model/qos_aggregation_manager.rs index 614f170ae..ea644820a 100644 --- a/backend/src/api/model/qos_aggregation_manager.rs +++ b/backend/src/api/model/qos_aggregation_manager.rs @@ -67,7 +67,7 @@ pub(in crate::api) fn exec_qos_aggregation(app_state: &Arc, cancel_tok tokio::spawn(async move { let repo = match QosSnapshotRepository::open(&storage_dir) { - Ok(repo) => repo, + Ok(repo) => std::sync::Arc::new(repo), Err(err) => { error!("Failed to open QoS snapshot repository: {err}"); return; @@ -76,7 +76,11 @@ pub(in crate::api) fn exec_qos_aggregation(app_state: &Arc, cancel_tok loop { let today = current_utc_day(); - if let Err(err) = run_aggregation_once(&repo, &history_dir, &today) { + let repo_clone = repo.clone(); + let history_dir_clone = history_dir.clone(); + if let Err(err) = tokio::task::spawn_blocking(move || { + run_aggregation_once(&repo_clone, &history_dir_clone, &today) + }).await.unwrap_or_else(|e| Err(std::io::Error::other(e.to_string()))) { warn!("QoS aggregation run failed: {err}"); } @@ -326,10 +330,11 @@ fn discover_history_days(history_dir: &Path) -> io::Result> { if !history_dir.exists() { return Ok(Vec::new()); } + let entries = std::fs::read_dir(history_dir) + .and_then(std::iter::Iterator::collect::, _>>)?; let mut days = BTreeSet::new(); - for entry in std::fs::read_dir(history_dir)? { - let entry = entry?; + for entry in entries { let name = entry.file_name(); let name = name.to_string_lossy(); if let Some(day) = extract_day_from_filename(&name) { @@ -406,11 +411,12 @@ fn discover_day_files(history_dir: &Path, day_utc: &str) -> io::Result, _>>)?; let mut archives = Vec::new(); let mut pending = Vec::new(); - for entry in std::fs::read_dir(history_dir)? { - let entry = entry?; + for entry in entries { let path = entry.path(); let name = path.file_name().unwrap_or_default().to_string_lossy().to_string(); if extract_day_from_filename(&name) != Some(day_utc) { @@ -678,14 +684,16 @@ mod tests { assert_eq!(bucket.total_provider_reconnect_count, 2); } - #[test] - fn history_day_revision_tracks_seconds_and_file_length() { + #[tokio::test] + async fn history_day_revision_tracks_seconds_and_file_length() { let temp = tempdir().expect("tempdir should succeed"); let history_dir = temp.path(); let record = base_record(EventType::Connect); write_pending_history_records(history_dir, vec![record.clone(), record]); - let revision = history_day_revision(history_dir, "2026-04-02") + let history_dir_copy = history_dir.to_path_buf(); + let revision = tokio::task::spawn_blocking(move || history_day_revision(&history_dir_copy, "2026-04-02")) + .await.unwrap() .expect("revision should load") .expect("revision should exist"); let pending_path = history_dir.join("stream-history-2026-04-02.pending"); @@ -796,8 +804,13 @@ mod tests { connect.shared_joined_existing = Some(false); write_pending_history_records(&history_dir, vec![connect]); - let repo = QosSnapshotRepository::open(temp.path()).expect("repo should open"); - run_aggregation_once(&repo, &history_dir, &today).expect("aggregation should succeed"); + let repo = std::sync::Arc::new(QosSnapshotRepository::open(temp.path()).expect("repo should open")); + + let repo_c = repo.clone(); + let h_dir = history_dir.clone(); + let t_day = today.clone(); + tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c, &h_dir, &t_day)) + .await.unwrap().expect("aggregation should succeed"); let snapshot = repo .get_snapshot("stream-a") @@ -825,10 +838,19 @@ mod tests { connect.shared_joined_existing = Some(false); write_pending_history_records(&history_dir, vec![connect]); - let repo = QosSnapshotRepository::open(temp.path()).expect("repo should open"); + let repo = std::sync::Arc::new(QosSnapshotRepository::open(temp.path()).expect("repo should open")); - run_aggregation_once(&repo, &history_dir, &today).expect("first aggregation should succeed"); - run_aggregation_once(&repo, &history_dir, &today).expect("second aggregation should succeed"); + let repo_c1 = repo.clone(); + let h_dir1 = history_dir.clone(); + let t_day1 = today.clone(); + tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c1, &h_dir1, &t_day1)) + .await.unwrap().expect("first aggregation should succeed"); + + let repo_c2 = repo.clone(); + let h_dir2 = history_dir.clone(); + let t_day2 = today.clone(); + tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c2, &h_dir2, &t_day2)) + .await.unwrap().expect("second aggregation should succeed"); let snapshot = repo .get_snapshot("stream-a") @@ -864,9 +886,13 @@ mod tests { write_pending_history_records(&history_dir, vec![connect]); } - let repo = QosSnapshotRepository::open(temp.path()).expect("repo should open"); + let repo = std::sync::Arc::new(QosSnapshotRepository::open(temp.path()).expect("repo should open")); - run_aggregation_once(&repo, &history_dir, "2036-04-05").expect("first aggregation should succeed"); + let repo_c1 = repo.clone(); + let h_dir1 = history_dir.clone(); + tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c1, &h_dir1, "2036-04-05")) + .await.unwrap().expect("first aggregation should succeed"); + let checkpoint = repo.load_checkpoint().expect("checkpoint should load"); assert_eq!(checkpoint.last_completed_day_utc.as_deref(), Some("2036-04-03")); @@ -877,7 +903,11 @@ mod tests { assert_eq!(snapshot.daily_buckets.len(), 3); assert_eq!(snapshot.window_7d.connect_count, 3); - run_aggregation_once(&repo, &history_dir, "2036-04-05").expect("second aggregation should succeed"); + let repo_c2 = repo.clone(); + let h_dir2 = history_dir.clone(); + tokio::task::spawn_blocking(move || run_aggregation_once(&*repo_c2, &h_dir2, "2036-04-05")) + .await.unwrap().expect("second aggregation should succeed"); + let checkpoint = repo.load_checkpoint().expect("checkpoint should load"); assert_eq!(checkpoint.last_completed_day_utc.as_deref(), Some("2036-04-04")); diff --git a/backend/src/api/model/streams/persist_pipe_stream.rs b/backend/src/api/model/streams/persist_pipe_stream.rs index 44e0a48eb..6842e91fc 100644 --- a/backend/src/api/model/streams/persist_pipe_stream.rs +++ b/backend/src/api/model/streams/persist_pipe_stream.rs @@ -123,29 +123,46 @@ pub async fn tee_dyn_reader( let mut buf = [0u8; 8192]; let mut reader = reader_arc; + let mut read_failed = false; loop { let n = match reader.read(&mut buf).await { - Ok(0) | Err(_) => break, + Ok(0) => break, + Err(err) => { + debug!("tee_dyn_reader: read error, terminating stream: {err}"); + read_failed = true; + break; + } Ok(n) => n, }; total_bytes += n; - if tx.write_all(&buf[..n]).await.is_err() { + if let Err(err) = tx.write_all(&buf[..n]).await { + debug!("tee_dyn_reader: downstream write error, terminating: {err}"); + read_failed = true; break; } - if writer.write_all(&buf[..n]).await.is_err() { + if let Err(err) = writer.write_all(&buf[..n]).await { + debug!("tee_dyn_reader: file write error, terminating: {err}"); + read_failed = true; break; } } - let _ = writer.flush().await; - let _ = tx.shutdown().await; + if let Err(err) = writer.flush().await { + debug!("tee_dyn_reader: flush error: {err}"); + read_failed = true; + } + if let Err(err) = tx.shutdown().await { + debug!("tee_dyn_reader: shutdown error: {err}"); + } - if let Some(cb) = callback { - cb(total_bytes); + if !read_failed { + if let Some(cb) = callback { + cb(total_bytes); + } } }); diff --git a/backend/src/api/model/streams/provider_stream_factory.rs b/backend/src/api/model/streams/provider_stream_factory.rs index 385682139..069077f78 100644 --- a/backend/src/api/model/streams/provider_stream_factory.rs +++ b/backend/src/api/model/streams/provider_stream_factory.rs @@ -890,7 +890,6 @@ pub async fn create_provider_stream( Ok(None) => None, Err(failure) => { let status = failure.status(); - app_state.connection_manager.release_provider_connection(&stream_options.addr).await; record_provider_open_failure( app_state, &stream_options, diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs index 87fb7cc24..4f703f0f7 100644 --- a/backend/src/api/model/streams/shared_stream_manager.rs +++ b/backend/src/api/model/streams/shared_stream_manager.rs @@ -20,7 +20,7 @@ use std::{ task::{Context, Poll}, }; use tokio::{ - sync::{mpsc, mpsc::Sender, RwLock}, + sync::{mpsc, mpsc::Sender, Mutex, RwLock}, time::{sleep, Duration, Instant}, }; use tokio_stream::wrappers::ReceiverStream; @@ -130,7 +130,7 @@ pub struct SharedStreamState { subscribers: RwLock>, broadcaster: tokio::sync::broadcast::Sender, stop_token: CancellationToken, - burst_buffer: Arc>, + burst_buffer: Arc>, task_handles: RwLock>>, } @@ -153,7 +153,7 @@ impl SharedStreamState { subscribers: RwLock::new(HashMap::new()), broadcaster, stop_token: CancellationToken::new(), - burst_buffer: Arc::new(RwLock::new(BurstBuffer::new(burst_buffer_size_in_bytes))), + burst_buffer: Arc::new(Mutex::new(BurstBuffer::new(burst_buffer_size_in_bytes))), task_handles: RwLock::new(Vec::new()), } } @@ -199,7 +199,7 @@ impl SharedStreamState { let handle = tokio::spawn(async move { let snapshot = { - let buffer = burst_buffer.read().await; + let buffer = burst_buffer.lock().await; buffer.snapshot() }; let sent_burst_chunks = send_burst_buffer(&snapshot, &client_tx_clone, &cancel_token).await; @@ -283,7 +283,7 @@ impl SharedStreamState { consecutive_lag_count = consecutive_lag_count.saturating_add(1); if last_lag_log.elapsed() > Duration::from_secs(5) { let buffered_bytes = { - let buffer = burst_buffer_for_log.read().await; + let buffer = burst_buffer_for_log.lock().await; buffer.current_bytes }; warn!( @@ -408,7 +408,7 @@ impl SharedStreamState { } } { - let mut buffer = burst_buffer.write().await; + let mut buffer = burst_buffer.lock().await; buffer.push(data.clone()); } diff --git a/backend/src/api/setup_api.rs b/backend/src/api/setup_api.rs index f9d79480c..626a628cb 100644 --- a/backend/src/api/setup_api.rs +++ b/backend/src/api/setup_api.rs @@ -3,7 +3,7 @@ use crate::{ auth::generate_password_from_input, model::validate_library_paths_from_dto, utils::{ - file_exists, get_default_web_root_path_for_home, read_api_proxy_file, read_config_file, read_sources_file, + file_exists, get_default_web_root_path_for_home, read_api_proxy_file, read_config_file_with_options, read_sources_file, read_templates_file, resolve_template_persist_file_path, sanitize_sources_for_persist, }, }; @@ -204,7 +204,7 @@ async fn build_initial_draft(paths: &ConfigPaths) -> AppConfigDto { let mut draft = create_default_draft(paths.home_path.as_str()); if file_exists(&paths.config_file_path) { - match read_config_file(paths.config_file_path.as_str(), true, false) { + match read_config_file_with_options(paths.config_file_path.as_str(), crate::utils::ReadConfigOptions { resolve_env: true, include_computed: false }) { Ok(cfg) => draft.config = cfg, Err(err) => warn!("Setup mode: failed to load existing config.yml: {err}"), } diff --git a/backend/src/main.rs b/backend/src/main.rs index 08d5da4f9..0b0bb4200 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -231,17 +231,14 @@ async fn print_info(app_config: &AppConfig) { let paths = > as Access>::load(&app_config.paths); info!("Current time: {}", chrono::offset::Local::now().format("%Y-%m-%d %H:%M:%S")); info!("Temp dir: {}", tempfile::env::temp_dir().display()); - info!("Storage dir: {:?}", &config.storage_dir); - info!("Config dir: {:?}", &paths.config_path); - info!("Config file: {:?}", &paths.config_file_path); - info!("Source file: {:?}", &paths.sources_file_path); - info!("Api Proxy File: {:?}", &paths.api_proxy_file_path); - info!("Mapping path: {:?}", &paths.mapping_file_path.as_ref().map_or_else(|| "not used", |v| v.as_str())); - info!("Template path: {:?}", &paths.template_file_path.as_ref().map_or_else(|| "not used", |v| v.as_str())); - info!( - "Resources path: {:?}", - &paths.custom_stream_response_path.as_ref().map_or_else(|| "not used", |v| v.as_str()) - ); + info!("Storage dir: {}", &config.storage_dir); + info!("Config dir: {}", &paths.config_path); + info!("Config file: {}", &paths.config_file_path); + info!("Source file: {}", &paths.sources_file_path); + info!("Api Proxy File: {}", &paths.api_proxy_file_path); + info!("Mapping path: {}", &paths.mapping_file_path.as_ref().map_or_else(|| "not used", |v| v.as_str())); + info!("Template path: {}", &paths.template_file_path.as_ref().map_or_else(|| "not used", |v| v.as_str())); + info!("Resources path: {}", &paths.custom_stream_response_path.as_ref().map_or_else(|| "not used", |v| v.as_str())); if let Some(mapping_paths) = paths.mapping_files_used.as_ref() { for mapping_path in mapping_paths { @@ -320,7 +317,10 @@ fn get_file_paths(args: &Args) -> ConfigPaths { utils::resolve_template_file_path(config_path.as_str(), Some(path.as_str())) }); let storage_path = if Path::new(&config_file).exists() { - match utils::read_config_file(&config_file, true, false) { + match utils::read_config_file_with_options( + &config_file, + crate::utils::ReadConfigOptions { resolve_env: true, include_computed: false }, + ) { Ok(cfg) => resolve_storage_path(&home_path, cfg.storage_dir.as_deref()), Err(err) => { eprintln!("Can't read config file {config_file} while resolving storage path: {err}"); diff --git a/backend/src/model/config/app.rs b/backend/src/model/config/app.rs index 756aa8b01..227a72821 100644 --- a/backend/src/model/config/app.rs +++ b/backend/src/model/config/app.rs @@ -439,21 +439,20 @@ impl AppConfig { } - /// # Panics - /// - /// Will panic if default server invalid - pub fn get_server_info(&self, server_info_name: &str) -> ApiProxyServerInfo { + pub fn get_server_info(&self, server_info_name: &str) -> Option { let guard = self.api_proxy.load(); if let Some(api_proxy) = guard.as_ref() { - let server_info_list = api_proxy.server.clone(); - server_info_list.iter().find(|c| c.name.eq(server_info_name)) - .map_or_else(|| server_info_list.first().unwrap().clone(), Clone::clone) + let server_info_list = &api_proxy.server; + server_info_list + .iter() + .find(|c| c.name.eq(server_info_name)) + .cloned() } else { - panic!("ApiProxyServer info not found"); + None } } - pub fn get_user_server_info(&self, user: &ProxyUserCredentials) -> ApiProxyServerInfo { + pub fn get_user_server_info(&self, user: &ProxyUserCredentials) -> Option { let server_info_name = user.server.as_ref().map_or("default", |server_name| server_name.as_str()); self.get_server_info(server_info_name) } diff --git a/backend/src/model/config/target.rs b/backend/src/model/config/target.rs index 3945b8c40..7d6ff42c1 100644 --- a/backend/src/model/config/target.rs +++ b/backend/src/model/config/target.rs @@ -181,6 +181,17 @@ pub enum TargetOutput { HdHomeRun(HdHomeRunTargetOutput), } +impl From<&TargetOutput> for TargetType { + fn from(output: &TargetOutput) -> Self { + match output { + TargetOutput::Xtream(_) => TargetType::Xtream, + TargetOutput::M3u(_) => TargetType::M3u, + TargetOutput::Strm(_) => TargetType::Strm, + TargetOutput::HdHomeRun(_) => TargetType::HdHomeRun, + } + } +} + macros::from_impl!(TargetOutput); impl From<&TargetOutputDto> for TargetOutput { fn from(dto: &TargetOutputDto) -> Self { @@ -228,57 +239,28 @@ impl ConfigTarget { } pub(crate) fn get_xtream_output(&self) -> Option<&XtreamTargetOutput> { - if let Some(TargetOutput::Xtream(output)) = self.output.iter().find(|o| matches!(o, TargetOutput::Xtream(_))) { - Some(output) - } else { - None - } + self.output.iter().find_map(|o| match o { + TargetOutput::Xtream(output) => Some(output), + _ => None, + }) } pub(crate) fn get_m3u_output(&self) -> Option<&M3uTargetOutput> { - if let Some(TargetOutput::M3u(output)) = self.output.iter().find(|o| matches!(o, TargetOutput::M3u(_))) { - Some(output) - } else { - None - } + self.output.iter().find_map(|o| match o { + TargetOutput::M3u(output) => Some(output), + _ => None, + }) } pub(crate) fn get_hdhomerun_output(&self) -> Option<&HdHomeRunTargetOutput> { - if let Some(TargetOutput::HdHomeRun(output)) = - self.output.iter().find(|o| matches!(o, TargetOutput::HdHomeRun(_))) - { - Some(output) - } else { - None - } + self.output.iter().find_map(|o| match o { + TargetOutput::HdHomeRun(output) => Some(output), + _ => None, + }) } pub fn has_output(&self, tt: TargetType) -> bool { - for target_output in &self.output { - match target_output { - TargetOutput::Xtream(_) => { - if tt == TargetType::Xtream { - return true; - } - } - TargetOutput::M3u(_) => { - if tt == TargetType::M3u { - return true; - } - } - TargetOutput::Strm(_) => { - if tt == TargetType::Strm { - return true; - } - } - TargetOutput::HdHomeRun(_) => { - if tt == TargetType::HdHomeRun { - return true; - } - } - } - } - false + self.output.iter().any(|o| TargetType::from(o) == tt) } pub fn is_force_redirect(&self, item_type: PlaylistItemType) -> bool { diff --git a/backend/src/model/xtream.rs b/backend/src/model/xtream.rs index ad64938e2..f32e36147 100644 --- a/backend/src/model/xtream.rs +++ b/backend/src/model/xtream.rs @@ -75,7 +75,7 @@ pub fn xtream_mapping_option_from_target_options(target: &ConfigTarget, target_o .map_or("", String::as_str); concat_path_leading_slash(web_ui_path, "api/v1/playlist/resource") } else { - app_config.get_user_server_info(user).get_base_url() + app_config.get_user_server_info(user).map(|si| si.get_base_url()).unwrap_or_default() }; diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index ec6497501..27b5d75be 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -389,8 +389,16 @@ async fn playlist_download_from_input( let cache_duration = input.cache_duration_seconds; // Ensure data directory exists - if !storage_path.exists() { - let _ = std::fs::create_dir_all(&storage_path); + match tokio::fs::try_exists(&storage_path).await { + Ok(false) => { + if let Err(err) = tokio::fs::create_dir_all(&storage_path).await { + warn!("Failed to create input storage directory '{}': {err}", storage_path.display()); + } + } + Err(err) => { + warn!("Failed to check existence of input storage directory '{}': {err}", storage_path.display()); + } + Ok(true) => {} } let hybrid = is_hybrid_m3u_xtream(input); diff --git a/backend/src/repository/bplustree.rs b/backend/src/repository/bplustree.rs index 9e1694785..54bb04b88 100644 --- a/backend/src/repository/bplustree.rs +++ b/backend/src/repository/bplustree.rs @@ -3954,7 +3954,7 @@ where /// Use `tokio::task::spawn_blocking(...)` when calling from async code. pub fn try_new_with_backoff_stats(filepath: &Path) -> io::Result<(Self, u64)> { let mut attempts = 0u64; - let mut backoff = Duration::from_micros(10); + let mut backoff = Duration::from_millis(1); let max_backoff = Duration::from_millis(10); let started_at = Instant::now(); let timeout = Duration::from_secs(30); diff --git a/backend/src/repository/m3u_playlist_iterator.rs b/backend/src/repository/m3u_playlist_iterator.rs index e7629542a..99ca682d5 100644 --- a/backend/src/repository/m3u_playlist_iterator.rs +++ b/backend/src/repository/m3u_playlist_iterator.rs @@ -175,8 +175,7 @@ impl M3uPlaylistIterator { flags.set(M3uPlaylistIteratorFlags::RewriteResource); } - let server_info = cfg.get_user_server_info(user); - let base_url = server_info.get_base_url(); + let base_url = cfg.get_user_server_info(user).map(|si| si.get_base_url()).unwrap_or_default(); let username = user.username.clone(); let password = user.password.clone(); let proxy_type = user.proxy; diff --git a/backend/src/repository/playlist_source.rs b/backend/src/repository/playlist_source.rs index 34521dd53..bf8273345 100644 --- a/backend/src/repository/playlist_source.rs +++ b/backend/src/repository/playlist_source.rs @@ -107,10 +107,14 @@ impl PlaylistSource for XtreamDiskPlaylistSource { } fn get_group_count(&mut self) -> usize { + fn collect_groups(query: &mut Option<(BPlusTreeQuery, Q)>, groups: &mut HashSet>) { + if let Some((query, _)) = query { for (_, item) in query.iter() { groups.insert(item.group.clone()); } } + } let mut groups = HashSet::new(); - if let Some((query, _)) = self.live.as_mut() { for (_, item) in query.iter() { groups.insert(item.group.clone()); } } - if let Some((query, _)) = self.vod.as_mut() { for (_, item) in query.iter() { groups.insert(item.group.clone()); } } - if let Some((query, _)) = self.series.as_mut() { for (_, item) in query.iter() { groups.insert(item.group.clone()); } } + + collect_groups(&mut self.live, &mut groups); + collect_groups(&mut self.vod, &mut groups); + collect_groups(&mut self.series, &mut groups); groups.len() } @@ -161,27 +165,19 @@ impl PlaylistSource for XtreamDiskPlaylistSource { } fn get_missing_vod_info_count(&mut self) -> usize { - let mut count = 0; - if let Some((query, _)) = self.vod.as_mut() { - for (_, item) in query.iter() { - if item.item_type == PlaylistItemType::Video && item.provider_id > 0 && !item.has_details() { - count += 1; - } - } - } - count + self.vod.as_mut().map_or(0, |(query, _)| { + query.iter().filter(|(_, item)| { + item.item_type == PlaylistItemType::Video && item.provider_id > 0 && !item.has_details() + }).count() + }) } fn get_missing_series_info_count(&mut self) -> usize { - let mut count = 0; - if let Some((query, _)) = self.series.as_mut() { - for (_, item) in query.iter() { - if item.item_type == PlaylistItemType::SeriesInfo && item.provider_id > 0 && !item.has_details() { - count += 1; - } - } - } - count + self.series.as_mut().map_or(0, |(query, _)| { + query.iter().filter(|(_, item)| { + item.item_type == PlaylistItemType::SeriesInfo && item.provider_id > 0 && !item.has_details() + }).count() + }) } fn deduplicate(&mut self, _duplicates: &mut HashSet) { diff --git a/backend/src/repository/strm_repository.rs b/backend/src/repository/strm_repository.rs index 7895ca531..ad962c0ab 100644 --- a/backend/src/repository/strm_repository.rs +++ b/backend/src/repository/strm_repository.rs @@ -1069,7 +1069,7 @@ fn get_credentials_and_server_info( ) -> Option<(ProxyUserCredentials, ApiProxyServerInfo)> { let username = username?; let credentials = cfg.get_user_credentials(username)?; - let server_info = cfg.get_user_server_info(&credentials); + let server_info = cfg.get_user_server_info(&credentials)?; Some((credentials, server_info)) } diff --git a/backend/src/utils/file/config_reader.rs b/backend/src/utils/file/config_reader.rs index 545348361..607b3c654 100644 --- a/backend/src/utils/file/config_reader.rs +++ b/backend/src/utils/file/config_reader.rs @@ -33,6 +33,45 @@ use shared::concat_string; use crate::utils::request::{is_uri}; use url::Url; +/// Options controlling how configuration files are parsed. +/// +/// Prefer constructing this struct over passing individual `bool` literals to +/// `read_config_file`, `read_sources_file_from_path`, etc. — bare `true`/`false` +/// arguments at call sites are hard to read without an IDE hover. +/// +/// # Example +/// ```ignore +/// // Before (opaque): +/// read_sources_file_from_path(&path, true, true, hdhr).await?; +/// +/// // After (self-documenting): +/// let opts = ReadConfigOptions::resolve_and_compute(); +/// read_sources_file_from_path(&path, opts.resolve_env, opts.include_computed, hdhr).await?; +/// ``` +#[derive(Debug, Clone, Copy)] +pub struct ReadConfigOptions { + // When `true`, `${ENV_VAR}` placeholders in config files are expanded. + pub resolve_env: bool, + // When `true`, computed / derived fields (e.g. auto EPG URLs) are filled in. + pub include_computed: bool, +} + +impl ReadConfigOptions { + // Expand environment variables and include all computed fields — the typical + // production default. + #[must_use] + pub const fn resolve_and_compute() -> Self { + Self { resolve_env: true, include_computed: true } + } + + // Parse the raw YAML without any variable expansion or computed fields — + // useful when you need the literal, unexpanded content (e.g. before saving). + #[must_use] + pub const fn raw() -> Self { + Self { resolve_env: false, include_computed: false } + } +} + pub(crate) struct PreparedTemplateBundle { pub(crate) definition: Option, pub(crate) prepared: Option>, @@ -145,13 +184,36 @@ pub async fn read_sources_file_from_path_with_templates( Ok(sources) } +pub async fn read_sources_file_from_path_with_options( + sources_file: &Path, + opts: ReadConfigOptions, + hdhr_config: Option<&HdHomeRunDeviceOverview>, + prepared_templates: Option<&[shared::model::PatternTemplate]>, +) -> Result { + read_sources_file_from_path_with_templates(sources_file, opts.resolve_env, opts.include_computed, hdhr_config, prepared_templates).await +} + pub async fn read_sources_file_from_path( sources_file: &Path, resolve_env: bool, include_computed: bool, hdhr_config: Option<&HdHomeRunDeviceOverview>, ) -> Result { - read_sources_file_from_path_with_templates(sources_file, resolve_env, include_computed, hdhr_config, None).await + read_sources_file_from_path_with_options( + sources_file, + ReadConfigOptions { resolve_env, include_computed }, + hdhr_config, + None + ).await +} + +pub async fn read_sources_file_with_options( + sources_file: &str, + opts: ReadConfigOptions, + hdhr_config: Option<&HdHomeRunDeviceOverview>, + prepared_templates: Option<&[shared::model::PatternTemplate]>, +) -> Result { + read_sources_file_from_path_with_options(&PathBuf::from(sources_file), opts, hdhr_config, prepared_templates).await } pub async fn read_sources_file( @@ -161,22 +223,21 @@ pub async fn read_sources_file( hdhr_config: Option<&HdHomeRunDeviceOverview>, prepared_templates: Option<&[shared::model::PatternTemplate]>, ) -> Result { - read_sources_file_from_path_with_templates(&PathBuf::from(sources_file), resolve_env, include_computed, hdhr_config, prepared_templates).await + read_sources_file_with_options(sources_file, ReadConfigOptions { resolve_env, include_computed }, hdhr_config, prepared_templates).await } -pub fn read_config_file( +pub fn read_config_file_with_options( config_file: &str, - resolve_env: bool, - include_computed: bool, + opts: ReadConfigOptions, ) -> Result { match open_file(&std::path::PathBuf::from(config_file)) { Ok(file) => { let maybe_config: Result = - serde_saphyr::from_reader(config_file_reader(file, resolve_env)); + serde_saphyr::from_reader(config_file_reader(file, opts.resolve_env)); match maybe_config { Ok(mut config) => { - if resolve_env { - config.prepare(include_computed)?; + if opts.resolve_env { + config.prepare(opts.include_computed)?; } Ok(config) } @@ -187,6 +248,14 @@ pub fn read_config_file( } } +pub fn read_config_file( + config_file: &str, + resolve_env: bool, + include_computed: bool, +) -> Result { + read_config_file_with_options(config_file, ReadConfigOptions { resolve_env, include_computed }) +} + #[allow(clippy::too_many_lines)] pub(crate) fn read_templates( template_file_path: Option<&str>, diff --git a/backend/src/utils/lru_cache.rs b/backend/src/utils/lru_cache.rs index c52dcf59f..0971d75bf 100644 --- a/backend/src/utils/lru_cache.rs +++ b/backend/src/utils/lru_cache.rs @@ -19,18 +19,34 @@ fn encode_cache_key(key: &str) -> String { /// automatically managing their lifecycle based on a specified maximum cache size. The cache evicts /// the least recently used files when the size limit is exceeded. /// +/// # Implementation note — O(1) LRU with lazy-deletion +/// +/// Instead of an O(n) `VecDeque::position()` scan on every cache hit, we use a **lazy-deletion** +/// strategy: +/// - Each entry in `cache` stores a monotonic `generation` counter alongside its data. +/// - `usage_order` is a `VecDeque<(key, generation)>`. Promoting an entry to MRU costs O(1): +/// bump the generation in `cache` and push a new `(key, new_gen)` tuple to the back. +/// - The old tuple(s) in `usage_order` that carry a stale generation are left in place and +/// discarded during the next eviction sweep — hence "lazy deletion". +/// - Amortised cost per operation is O(1); worst-case eviction is proportional to the number +/// of stale entries that accumulated, which is bounded by total access count. +/// /// # Fields -/// - `capacity`: The maximum cache size in bytes. Once the cache size exceeds this value, files are evicted. +/// - `capacity`: The maximum cache size in bytes. /// - `cache_dir`: The directory where cached files are stored. /// - `current_size`: The current total size of all files in the cache, in bytes. -/// - `cache`: A `HashMap` that maps a unique key to a tuple containing the file path and its size. -/// - `usage_order`: A `VecDeque` that tracks the access order of keys, with the oldest at the front. +/// - `cache`: Maps a unique key to `(path, mime_type, size_bytes, generation)`. +/// - `usage_order`: LRU deque of `(key, generation)` pairs; stale entries are skipped on eviction. +/// - `next_generation`: Monotonic counter for assigning unique generations. pub struct LRUResourceCache { capacity: usize, // Maximum size in bytes cache_dir: PathBuf, current_size: usize, // Current size in bytes - cache: HashMap, usize)>, - usage_order: VecDeque, + // (`path`, `mime_type`, `size_bytes`, `generation`) + cache: HashMap, usize, u64)>, + // Lazy-deletion LRU queue: (`cache_key`, `generation_at_enqueue`) + usage_order: VecDeque<(String, u64)>, + next_generation: u64, } impl LRUResourceCache { @@ -59,8 +75,9 @@ impl LRUResourceCache { capacity, cache_dir, current_size: 0, - cache: HashMap::, usize)>::with_capacity(estimated_entries), + cache: HashMap::, usize, u64)>::with_capacity(estimated_entries), usage_order: VecDeque::new(), + next_generation: 0, } } @@ -89,12 +106,13 @@ impl LRUResourceCache { let file_size = usize::try_from(metadata.len()).unwrap_or(0); // we need to duplicate because of closure we can't call insert_to_cache { // insert_to_cache - let mut path = self.cache_dir.clone(); path.push(&file_name); trace!("Added file to cache: {}", &path.to_string_lossy()); - self.cache.insert(key.clone(), (path.clone(), mime_type, file_size)); - self.usage_order.push_back(key); + let gen = self.next_generation; + self.next_generation = self.next_generation.wrapping_add(1); + self.cache.insert(key.clone(), (path.clone(), mime_type, file_size, gen)); + self.usage_order.push_back((key, gen)); self.current_size += file_size; } } @@ -127,8 +145,10 @@ impl LRUResourceCache { fn insert_to_cache(&mut self, key: String, mime_type: Option, file_size: usize) -> PathBuf { let path = self.get_store_path(&key, mime_type.as_deref()); debug!("Added file to cache: {}", &path.to_string_lossy()); - self.cache.insert(key.clone(), (path.clone(), mime_type, file_size)); - self.usage_order.push_back(key); + let gen = self.next_generation; + self.next_generation = self.next_generation.wrapping_add(1); + self.cache.insert(key.clone(), (path.clone(), mime_type, file_size, gen)); + self.usage_order.push_back((key, gen)); self.current_size += file_size; path } @@ -148,54 +168,67 @@ impl LRUResourceCache { path } - /// - Retrieves a file from the cache if it exists. - /// - Moves the file's key to the end of the usage queue to mark it as recently used. + /// - Moves the file's key to the end of the usage queue to mark it as recently used (O(1)). /// - Arguments: /// - `url`: The unique identifier for the file. /// - Returns: /// - The `PathBuf` of the file if it exists; `None` otherwise. pub fn get_content(&mut self, url: &str) -> Option<(PathBuf, Option)> { let key = encode_cache_key(url); - { - if let Some((path, mime_type, size)) = self.cache.get(&key) { - if path.exists() { - trace_if_enabled!("Responding resource from cache with key: {key} for url: {}", sanitize_sensitive_info(url)); - // Move to the end of the queue - if let Some(pos) = self.usage_order.iter().position(|k| k == &key) { - self.usage_order.remove(pos); // remove from queue - } - self.usage_order.push_back(key); // add to the to end - return Some((path.clone(), mime_type.clone())); - } - { - trace_if_enabled!("Cache inconsistency: file missing for key: {key}, url: {}", sanitize_sensitive_info(url)); - // this should not happen, someone deleted the file manually and the cache is not in sync - self.current_size -= size; - self.cache.remove(&key); - if let Some(pos) = self.usage_order.iter().position(|k| k == &key) { - self.usage_order.remove(pos); - } - } + if let Some((path, mime_type, size, gen)) = self.cache.get_mut(&key) { + if path.exists() { + trace_if_enabled!("Responding resource from cache with key: {key} for url: {}", sanitize_sensitive_info(url)); + // O(1) MRU promotion: bump the generation and push a new entry to the back. + // The old tuple in `usage_order` with the previous generation becomes stale + // and will be lazily discarded during the next eviction sweep. + let new_gen = self.next_generation; + self.next_generation = self.next_generation.wrapping_add(1); + *gen = new_gen; + self.usage_order.push_back((key, new_gen)); + return Some((path.clone(), mime_type.clone())); } + trace_if_enabled!("Cache inconsistency: file missing for key: {key}, url: {}", sanitize_sensitive_info(url)); + // this should not happen, someone deleted the file manually and the cache is not in sync + self.current_size -= *size; + self.cache.remove(&key); } None } fn evict_if_needed(&mut self) { - // if the cache size is to small and one element exceeds the size than the cache won't work, we ignore this + // Compact the usage_order deque when it has grown excessively relative to the number + // of live cache entries (each get_content() call appends a new entry, leaving stale + // entries from previous promotions). Rebuild from live cache data sorted by generation + // so the approximate MRU order is preserved without allocating a HashMap. + if self.usage_order.len() > self.cache.len().saturating_mul(4).max(64) { + let mut live: Vec<(String, u64)> = self.cache.iter().map(|(k, e)| (k.clone(), e.3)).collect(); + live.sort_unstable_by_key(|(_, gen)| *gen); + self.usage_order = live.into_iter().collect(); + } + + // if the cache size is too small and one element exceeds the size then the cache won't work, we ignore this while self.current_size > self.capacity { - if let Some(oldest_file) = self.usage_order.pop_front() { - if let Some((file, _mime_type, size)) = self.cache.remove(&oldest_file) { - self.current_size -= size; - if let Err(err) = fs::remove_file(&file) { - error!("Failed to delete cached file {} {err}", file.to_string_lossy()); - } else { - debug!("Removed file from cache: {}", file.to_string_lossy()); - } + // Pop from the front; skip stale entries (lazy-deletion). + let Some((oldest_key, queued_gen)) = self.usage_order.pop_front() else { + break; + }; + + // Check whether this queue entry is still the live generation. + let live_gen = self.cache.get(&oldest_key).map(|e| e.3); + if live_gen != Some(queued_gen) { + // Stale entry — the key was promoted to a newer generation; skip it. + continue; + } + + if let Some((file, _mime_type, size, _gen)) = self.cache.remove(&oldest_key) { + self.current_size -= size; + if let Err(err) = fs::remove_file(&file) { + error!("Failed to delete cached file {} {err}", file.to_string_lossy()); + } else { + debug!("Removed file from cache: {}", file.to_string_lossy()); } } } } } - diff --git a/docs/src/SUMMARY.md b/docs/src/SUMMARY.md index 4a81c534a..4a483efcd 100644 --- a/docs/src/SUMMARY.md +++ b/docs/src/SUMMARY.md @@ -11,6 +11,7 @@ - [Connection Handling](./configuration/connection-handling.md) - [Priorities, Soft Connections and Preemption](./configuration/connection-handling-priorities-and-preemption.md) - [Sessions, HLS, Catchup and Reconnects](./configuration/connection-handling-sessions-and-reconnects.md) + - [Session Handling Internals](./configuration/connection-handling-session-implementation-notes.md) - [Failures and User-Visible Behavior](./configuration/connection-handling-failures-and-user-visible-behavior.md) - [Metadata Updater & FFprobe](./configuration/metadata-update.md) - [Local Media Library](./configuration/local-library.md) diff --git a/docs/src/configuration/connection-handling-session-implementation-notes.md b/docs/src/configuration/connection-handling-session-implementation-notes.md new file mode 100644 index 000000000..d0e4a79bd --- /dev/null +++ b/docs/src/configuration/connection-handling-session-implementation-notes.md @@ -0,0 +1,350 @@ +# Session Handling Internals + +This page is for maintainers working on Tuliprox session code. + +If you only want the operator-facing behavior, read: + +- [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md) + +## Why this page exists + +Tuliprox uses the word "session" for several related but different mechanisms: + +- user admission reuse +- logical playback identity +- socket/address tracking +- provider/account affinity +- adaptive reconnect preservation +- provider-account affinity and reservation + +These mechanisms interact, but they are not the same thing. + +The most important rule for future changes is: + +- do not derive session-based admission from socket-binding rules +- do not derive provider/account affinity from socket-binding rules either + +That exact confusion breaks VOD, series, catchup, and local reopen/seek flows. + +## Core terms + +### Session token + +The session token is the stable logical playback identity. + +It is used to answer questions like: + +- is this the same playback as before? +- may this request reopen an existing playback while the user is already at the hard limit? +- should Tuliprox keep using the same provider context if possible? + +### Session admission + +Session admission means user-limit checks are performed with `connection_admission_for_session(...)` instead of plain `connection_admission(...)`. + +This happens in `resolve_admission_with_strategies(..., use_session_admission, session_token)`. + +If `use_session_admission` is `true` and the session token already exists, the request may continue even when the user is +already at the nominal connection limit. + +This is required for normal player behavior such as: + +- VOD size checks +- VOD seeks +- pause/resume reopen +- HLS segment fetches +- local media reopen + +### Socket-bound session + +Socket binding only controls how Tuliprox tracks request addresses for the same session. + +It does not decide: + +- whether the request should use session admission +- whether the logical playback must stay on the same provider account + +Current socket-binding policy: + +- `PlaylistItemType::Live` +- `PlaylistItemType::LiveUnknown` + +are socket-bound. + +Everything else is not socket-bound: + +- `LiveHls` +- `LiveDash` +- `Video` +- `Series` +- `Catchup` +- local playback types + +### Provider/account affinity + +Provider/account affinity answers a different question: + +- when several requests belong to the same logical playback, must they keep using the same provider account? + +The intended behavior is: + +- TS-style live playback is socket-bound, but not provider-account-bound across requests +- VOD/movie/series playback is not socket-bound, but is provider/account-bound for the logical stream +- HLS/DASH playback is not socket-bound, but is provider/account-bound for the logical stream + +Why: + +- TS-style live playback has no seek/range workflow and each new request can be treated independently +- VOD uses range and seek requests that still belong to one playback +- HLS fetches many chunk requests that still belong to one playback + +This means the same logical HLS or VOD stream must keep the same provider account for its follow-up requests, even though +those requests may arrive on different sockets. + +### Adaptive preserved session + +Adaptive live playback (`LiveHls`, `LiveDash`) can stay logically alive for a short TTL even after the current socket disconnects. + +That logic lives in `ActiveUserManager` and uses: + +- `should_preserve_session_stream(...)` +- `build_preserved_stream_expiry(...)` +- `process_due_adaptive_expiry_entries(...)` + +### Provider reservation + +Provider reservation is a provider-slot affinity mechanism, not a user admission mechanism. + +Current TTL mapping: + +- HLS/DASH use `hls_session_ttl_secs` +- Catchup uses `catchup_session_ttl_secs` +- other item types use `0` + +This affects provider reuse between short request gaps, not whether the session should use session admission. + +Important: + +- VOD/movie/series still have strict provider affinity on follow-up requests +- they just do not currently get an extra post-request reservation TTL like HLS/Catchup do + +## Where session tokens come from + +### TS live, VOD, series + +The regular M3U and Xtream playback endpoints build a stable playback token with: + +- `create_session_fingerprint(fingerprint, username, virtual_id)` + +### Catchup + +Catchup uses a different token namespace: + +- `create_catchup_session_key(fingerprint, username, virtual_id)` + +### HLS segment requests + +Rewritten HLS URLs can carry the session token inside the encrypted segment token. + +On segment fetch: + +- Tuliprox decodes the HLS token +- extracts the session token if present +- falls back to `create_session_fingerprint(...)` if needed + +### Local playback + +Local playback uses a stable `playback_session_token` passed into `local_stream_response(...)`. + +## The three decisions that must stay separate + +### 1. Should this request use session admission? + +For playback endpoints, the answer is yes. + +That means these paths should call `resolve_admission_with_strategies(..., true, Some(session_token))`: + +- M3U playback +- Xtream playback +- HLS session/segment handling +- local playback reopen logic + +Reason: + +- a second request for the same logical playback must be admitted as the same playback +- not as a brand-new connection + +### 2. Should this session be bound to exactly one current socket? + +This is controlled by `PlaylistItemType::uses_socket_bound_session()`. + +Reason: + +- plain TS-style live playback behaves like one active transport socket +- VOD, HLS, DASH, catchup, and local playback often use multiple short-lived sockets for one logical playback + +Do not replace decision 1 with decision 2. + +If you do that: + +- existing VOD sessions stop reopening correctly at `max_connections: 1` +- seek/size-check flows start failing with `UserConnectionsExhausted` + +### 3. Should this logical playback stay on the same provider account? + +This is separate from both admission and socket tracking. + +This policy is controlled by `PlaylistItemType::requires_provider_affinity()`. + +Intended model: + +- TS live: no provider/account pinning across requests +- VOD/movie/series: provider/account pinned across the logical stream +- HLS/DASH: provider/account pinned across the logical stream +- Catchup: provider/account pinned across the logical stream + +If you collapse this into socket-binding, you get the wrong matrix: + +- TS would be over-pinned +- VOD/HLS would be under-pinned + +## Current code flow + +### 1. Admission + +Main entry point: + +- `api_utils::resolve_admission_with_strategies(...)` + +If `use_session_admission` is `true`, Tuliprox checks: + +- `AppState::get_connection_admission_for_session(...)` +- `ActiveUserManager::connection_admission_for_session(...)` + +If `false`, it falls back to: + +- `AppState::get_connection_admission(...)` +- `ActiveUserManager::connection_admission(...)` + +### 2. Session creation or refresh + +Main entry point: + +- `ActiveUserManager::create_user_session(...)` + +This stores: + +- session token +- virtual id +- provider +- stream url +- current address +- `socket_bound` +- `active_addrs` +- connection permission and kind + +### 3. Stream registration + +Main entry point: + +- `ActiveUserManager::update_connection(...)` + +This creates or reuses the tracked logical stream entry. + +Important behavior: + +- same `session_token` reuses the logical stream +- stream metrics and duration stay tied to the logical session +- for non-socket-bound sessions, the session remembers multiple active addresses + +### 4. Lightweight HTTP activity + +Main entry point: + +- `ActiveUserManager::touch_http_activity(...)` + +This is used for session continuity without creating a new logical stream, especially for HLS follow-up requests. + +### 5. Seek / forced provider reopen + +Main entry point: + +- `api_utils::force_provider_stream_response(...)` + +This path: + +- releases the current provider-side connection for the session +- reacquires a provider connection +- moves the session to the request address with `update_session_addr(...)` + +For non-socket-bound sessions, `update_session_addr(...)` does not forget the older active session addresses. + +That is what allows a VOD session to survive: + +- overlapping size-check requests +- seek requests +- overlapping teardown/start windows + +The address move itself is not the provider-affinity rule. + +Provider-affinity is the separate invariant that follow-up requests for the same VOD/HLS playback should continue on the same provider account. + +### 6. Disconnect handling + +Main entry points: + +- `ActiveUserManager::release_stream(...)` +- `ActiveUserManager::release_connection(...)` + +For socket-bound sessions, closing the current address removes the old address from the session immediately. + +For non-socket-bound sessions: + +- the closed address is removed from `active_addrs` +- if another active address still belongs to that session, the session and logical stream fall back to that address + +This is the key behavior that keeps VOD and local playback stable across multiple sockets. + +## URL persistence rules + +Session URL handling is intentionally different by stream type. + +For: + +- `Video` +- `Series` +- `Catchup` +- local VOD/series playback + +Tuliprox stores the canonical request URL in the session, not an ephemeral redirect target. + +Reason: + +- provider-side redirects can resolve a request to a non-canonical final URL +- reusing that final URL later can change or break VOD seek/reopen behavior + +For live playback, following the redirected provider URL is still acceptable and often desirable. + +## File map for future changes + +Read these files together before changing session logic: + +- `backend/src/api/api_utils.rs` +- `backend/src/api/endpoints/m3u_api.rs` +- `backend/src/api/endpoints/xtream_api.rs` +- `backend/src/api/endpoints/hls_api.rs` +- `backend/src/api/model/active_user_manager.rs` +- `backend/src/api/model/active_provider_manager.rs` +- `shared/src/model/playlist.rs` + +## Tests that should stay green + +These tests cover the most fragile parts of the current logic: + +- `resolve_admission_with_strategies_allows_existing_session_even_when_user_is_at_limit` +- `local_stream_response_reuses_stable_playback_session_token_across_reopens` +- `vod_session_survives_overlapping_and_seek_sockets` +- `update_session_addr_prunes_previous_registration_for_socket_bound_session` +- `test_adaptive_session_release_connection_preserves_logical_stream_and_start_time` + +If you change session code and one of these assumptions no longer holds, update the documentation and the tests in the same change. diff --git a/docs/src/configuration/connection-handling-sessions-and-reconnects.md b/docs/src/configuration/connection-handling-sessions-and-reconnects.md index 54f44b945..5113e66db 100644 --- a/docs/src/configuration/connection-handling-sessions-and-reconnects.md +++ b/docs/src/configuration/connection-handling-sessions-and-reconnects.md @@ -4,6 +4,10 @@ This page explains why Tuliprox does not treat every request as a completely new This matters for operators because many seemingly strange connection patterns are actually normal player behavior. +If you are changing the code rather than the configuration, also read: + +- [Session Handling Internals](./connection-handling-session-implementation-notes.md) + ## Why sessions are needed at all Many operators initially assume this model: diff --git a/frontend/src/app/components/dashboard/stream_display/helpers.rs b/frontend/src/app/components/dashboard/stream_display/helpers.rs index fed0cdcf2..988133fd7 100644 --- a/frontend/src/app/components/dashboard/stream_display/helpers.rs +++ b/frontend/src/app/components/dashboard/stream_display/helpers.rs @@ -231,6 +231,7 @@ mod tests { client_ip: "127.0.0.1:1234".to_string(), user_agent: String::new(), ts: 0, + started_at: 0, country_code: None, session_token: has_session.then(|| "session".to_string()), preserved, diff --git a/frontend/src/app/components/dashboard/stream_display/item.rs b/frontend/src/app/components/dashboard/stream_display/item.rs index ad4f36f10..024ffe830 100644 --- a/frontend/src/app/components/dashboard/stream_display/item.rs +++ b/frontend/src/app/components/dashboard/stream_display/item.rs @@ -114,8 +114,8 @@ pub fn StreamDisplayItem(props: &StreamDisplayItemProps) -> Html {
{translate.t("LABEL.DURATION")} - - {format_duration(current_time_secs().saturating_sub(stream.ts))} + + {format_duration(current_time_secs().saturating_sub(if stream.started_at == 0 { stream.ts } else { stream.started_at }))}
if props.metrics_enabled && stream.meter_uid != 0 { diff --git a/frontend/src/hooks/use_server_status.rs b/frontend/src/hooks/use_server_status.rs index 76529fbca..4f3fe22af 100644 --- a/frontend/src/hooks/use_server_status.rs +++ b/frontend/src/hooks/use_server_status.rs @@ -95,6 +95,7 @@ fn download_task_to_stream_with_ts(download: &FileDownloadDto, ts: u64) -> Strea client_ip: BACKGROUND_TRANSFER_CLIENT_IP.to_string(), user_agent: "Tuliprox download worker".to_string(), ts, + started_at: ts, country_code: None, session_token: None, preserved: false, @@ -351,6 +352,7 @@ mod tests { client_ip: "127.0.0.1".to_string(), user_agent: "ua".to_string(), ts: 1, + started_at: 1, country_code: None, session_token: session_token.map(ToOwned::to_owned), preserved: false, diff --git a/shared/src/model/playlist.rs b/shared/src/model/playlist.rs index 69f72d204..4ff58557c 100644 --- a/shared/src/model/playlist.rs +++ b/shared/src/model/playlist.rs @@ -152,6 +152,31 @@ impl PlaylistItemType { pub fn is_live_adaptive(&self) -> bool { matches!(self, PlaylistItemType::LiveHls | PlaylistItemType::LiveDash) } + /// Controls address tracking only. + /// Do not use this to decide whether a playback request should use session-based admission + /// or whether a logical playback must stay on the same provider account. + pub fn uses_socket_bound_session(&self) -> bool { + matches!(self, PlaylistItemType::Live | PlaylistItemType::LiveUnknown) + } + + /// Controls whether follow-up requests for the same logical playback must stay on the + /// same provider account. + /// This is separate from both session admission and socket binding. + pub fn requires_provider_affinity(&self) -> bool { + matches!( + self, + PlaylistItemType::LiveHls + | PlaylistItemType::LiveDash + | PlaylistItemType::Video + | PlaylistItemType::LocalVideo + | PlaylistItemType::Series + | PlaylistItemType::SeriesInfo + | PlaylistItemType::LocalSeries + | PlaylistItemType::LocalSeriesInfo + | PlaylistItemType::Catchup + ) + } + pub fn as_u8(self) -> u8 { self as u8 } pub fn as_str(&self) -> &'static str { diff --git a/shared/src/model/stream_info.rs b/shared/src/model/stream_info.rs index 4962b8cb2..9cddecf1d 100644 --- a/shared/src/model/stream_info.rs +++ b/shared/src/model/stream_info.rs @@ -282,6 +282,8 @@ pub struct StreamInfo { pub user_agent: String, #[serde(default)] pub ts: u64, + #[serde(default)] + pub started_at: u64, #[serde(default, skip_serializing_if = "is_blank_optional_string")] pub country_code: Option, #[serde(default, skip_serializing_if = "is_blank_optional_string")] @@ -306,6 +308,7 @@ impl StreamInfo { country_code: Option, session_token: Option<&str>, ) -> Self { + let now = current_time_secs(); Self { uid, meter_uid, @@ -315,7 +318,8 @@ impl StreamInfo { addr: *addr, client_ip: client_ip.to_string(), user_agent, - ts: current_time_secs(), + ts: now, + started_at: now, country_code, session_token: session_token.map(|token| token.to_string()), preserved: false,