diff --git a/CHANGELOG.md b/CHANGELOG.md index b150bd9c5..0a649caa1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -166,9 +166,25 @@ active URL of the specified provider. item now stores retry/cooldown state for `resolve`, `probe`, and `tmdb`. - **TMDB No-Match Cooldown**: Added explicit TMDB cooldown handling. When TMDB resolve completes successfully but returns no match, TMDB reasons are suppressed for that item during cooldown to prevent endless requeue loops. +- **HLS/Catchup Provider Reservations**: Added short-lived provider-account reservations for HLS and catchup playback so follow-up requests can stay + on the same provider account without holding a real provider slot open between requests. New config fields: + `reverse_proxy.stream.hls_session_ttl_secs` (default `15`) and `reverse_proxy.stream.catchup_session_ttl_secs` (default `45`). +- **Channel Switch Friendly Reservations**: HLS/catchup reservations can now be taken over immediately by a new stream from the same client identity, + so channel switching does not have to wait for the reservation TTL to expire. - **Custom Stream Response Timeout**: Added support to limit how long custom fallback stream responses are served. - Set `config.custom_stream_response_timeout_secs` to a value `> 0` to auto-stop these streams after N seconds. - If unset or `0`, custom responses are streamed without timeout. + Set `config.custom_stream_response_timeout_secs` to a value `> 0` to auto-stop these streams after N seconds. If unset or `0`, custom responses + are streamed without timeout. + +## 🐛 Fixes + +- **Resolve Task Cooldown Persistence**: Resolve retry exhaustion is now persisted and consulted before enqueueing, so unresolved VOD/Series entries + are no longer recreated on every playlist refresh only to be skipped later in the worker. +- **Probe Handle Capacity Leak**: Fixed provider-slot leaks when internal probe tasks timed out, were dropped, or were preempted. Capacity is now + released reliably even when the underlying probe task does not complete normally. +- **Immediate Probe Preemption**: Higher-priority stream requests now cancel lower-priority probe tasks immediately instead of leaving a grace window + where the probe could continue holding upstream resources. +- **Anonymous Socket Cleanup**: Tracked anonymous incoming sockets are now pruned automatically after a TTL so stale UI/API keepalive registrations do + not remain visible forever in active-socket statistics. ## ⚙️ New Settings @@ -191,6 +207,9 @@ active URL of the specified provider. - FFprobe settings are configured under `metadata_update.ffprobe` (not under `video`). - Added `metadata_update.probe_fairness_resolve_burst` (default `200`) to control fairness between resolve and probe tasks. After N consecutive resolve-domain tasks, one pending probe-domain task is prioritized to avoid probe starvation. + - Added `reverse_proxy.stream.hls_session_ttl_secs` (`u64`, default `15`): keeps a short-lived provider-account reservation for HLS sessions. + - Added `reverse_proxy.stream.catchup_session_ttl_secs` (`u64`, default `45`): keeps a short-lived provider-account reservation for catchup + sessions and seek/reconnect flows. - Added `template_path` (optional): path to a template file (`template.yml`) or directory (`template.d` style). - **source.yml (input options)**: - Added `resolve_tmdb`: Triggers TMDB lookup if ID is missing. diff --git a/README.md b/README.md index a5f85e953..209e83959 100644 --- a/README.md +++ b/README.md @@ -521,7 +521,8 @@ metadata_update: - `probe.backoff_jitter_percent` (default `20`): Random jitter percentage applied to resolve/probe retry backoff to avoid synchronized retries. - `probe.user_priority` (default `127`): Connection priority assigned to probe tasks. Uses the same nice-style scale as user priorities (lower value = higher priority). At `127` (lowest priority), probe connections are always the first to be evicted when a regular user connects. - Reduce this value (e.g. `64`) to give probes more access to provider slots. + Reduce this value (e.g. `64`) to give probes more access to provider slots. When a higher-priority request preempts a probe, the probe is + cancelled immediately and its provider capacity is released right away. - `tmdb.cooldown` (default `7d`): Cooldown duration after a TMDB lookup completed successfully but returned no match. - `tmdb.enabled` / `tmdb.api_key` / `tmdb.rate_limit_ms` / `tmdb.cache_duration_days` / `tmdb.language`: TMDB resolver settings. - `tmdb.match_threshold` (default `86`): TMDB match threshold for search results for TMDB ID resolution. @@ -718,6 +719,8 @@ Attributes: - `grace_period_timeout_secs` default set to 4 seconds. - `grace_period_hold_stream` if set to `true`, the stream will only start after the grace period check has completed. Default is `true`. - `shared_burst_buffer_mb` optional (default `12`). Minimum burst buffer size (in MB) used for shared streams. +- `hls_session_ttl_secs` optional (default `15`). Keeps a short-lived provider-account reservation for HLS between playlist and segment requests. +- `catchup_session_ttl_secs` optional (default `45`). Keeps a short-lived provider-account reservation for catchup playback and seek/reconnect flows. ##### 1.7.1.1 `retry` @@ -779,6 +782,30 @@ client. This is useful for players that might time out or error if they receive data and then a "connections exhausted" stream switch occurs. Default is `true`. +##### 1.7.1.6 `hls_session_ttl_secs` + +HLS clients repeatedly connect, fetch a playlist or segment, disconnect, play for a while, and reconnect for the next request. Tuliprox therefore +cannot know the exact moment when playback has ended. + +To keep the upstream account stable without holding a real provider slot open all the time, Tuliprox uses a short-lived reservation: + +- The real provider slot is held only during the active playlist or segment request. +- Between requests, only the provider-account reservation is kept alive for the configured TTL. +- Requests from the same client/session try to reuse the same provider account. +- Channel switches from the same client can immediately take over the reservation and do not have to wait for the TTL to expire. + +This is useful for providers that abort HLS playback if the account changes in the middle of a session. + +##### 1.7.1.7 `catchup_session_ttl_secs` + +Catchup playback has the same account-affinity problem as HLS, especially during seeking, reconnects, and short gaps between requests. + +Tuliprox therefore keeps a short-lived reservation for catchup sessions: + +- The provider slot is released after the active request finishes. +- The reservation remains alive for the configured TTL so follow-up requests can stay on the same provider account. +- If the user switches to another stream, the new session from the same client can replace the old reservation immediately. + #### 1.7.2 `cache` LRU-Cache is for resources. If it is `enabled`, the resources/images are persisted in the given `directory`. If the cache size exceeds `size`, @@ -796,6 +823,8 @@ reverse_proxy: stream: throttle_kbps: 12500 retry: true + hls_session_ttl_secs: 15 + catchup_session_ttl_secs: 45 buffer: enabled: true size: 1024 diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 8dc5323d6..0eea90f7b 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -265,6 +265,20 @@ pub struct StreamOptions { pub pipe_provider_stream: bool, } +struct StreamingAcquireOptions<'a> { + force_provider: Option<&'a Arc>, + allow_provider_grace: bool, + user_priority: i8, + session_owner: Option<&'a str>, +} + +pub struct ForceStreamRequestContext<'a> { + pub req_headers: &'a HeaderMap, + pub input: &'a Arc, + pub user: &'a ProxyUserCredentials, + pub session_reservation_ttl_secs: u64, +} + /// Constructs a `StreamOptions` object based on the application's reverse proxy configuration. /// /// This function retrieves streaming-related settings from the `AppState`: @@ -357,19 +371,23 @@ async fn resolve_streaming_strategy( stream_url: &str, fingerprint: &Fingerprint, input: &ConfigInput, - force_provider: Option<&Arc>, - allow_provider_grace: bool, - user_priority: i8, + options: StreamingAcquireOptions<'_>, ) -> StreamingStrategy { // allocate a provider connection let mut forced_provider_allocated = false; - let provider_connection_handle = match force_provider { + 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(provider, &fingerprint.addr, allow_provider_grace, user_priority) + .acquire_exact_connection_with_grace_for_session( + provider, + &fingerprint.addr, + options.allow_provider_grace, + options.user_priority, + options.session_owner, + ) .await { forced_provider_allocated = true; @@ -382,14 +400,26 @@ async fn resolve_streaming_strategy( ); app_state .active_provider - .acquire_connection_with_grace(&input.name, &fingerprint.addr, allow_provider_grace, user_priority) + .acquire_connection_with_grace_for_session( + &input.name, + &fingerprint.addr, + options.allow_provider_grace, + options.user_priority, + options.session_owner, + ) .await } } None => { app_state .active_provider - .acquire_connection_with_grace(&input.name, &fingerprint.addr, allow_provider_grace, user_priority) + .acquire_connection_with_grace_for_session( + &input.name, + &fingerprint.addr, + options.allow_provider_grace, + options.user_priority, + options.session_owner, + ) .await } }; @@ -477,10 +507,21 @@ async fn create_stream_response_details( allow_provider_grace: bool, virtual_id: VirtualId, user_priority: i8, + session_owner: Option<&str>, ) -> Result { - let mut streaming_strategy = - resolve_streaming_strategy(app_state, stream_url, fingerprint, input, force_provider, allow_provider_grace, user_priority) - .await; + let mut streaming_strategy = resolve_streaming_strategy( + app_state, + stream_url, + fingerprint, + input, + StreamingAcquireOptions { + force_provider, + allow_provider_grace, + user_priority, + session_owner, + }, + ) + .await; let mut grace_period_options = app_state.get_grace_options(); grace_period_options.period_millis = get_grace_period_millis( connection_permission, @@ -790,9 +831,7 @@ pub async fn force_provider_stream_response( app_state: &Arc, user_session: &UserSession, mut stream_channel: StreamChannel, - req_headers: &HeaderMap, - input: &Arc, - user: &ProxyUserCredentials, + ctx: ForceStreamRequestContext<'_>, ) -> impl IntoResponse + Send { let stream_options = get_stream_options(app_state); let share_stream = false; @@ -815,15 +854,16 @@ pub async fn force_provider_stream_response( &stream_options, &user_session.stream_url, fingerprint, - req_headers, - input, + ctx.req_headers, + ctx.input, item_type, share_stream, connection_permission, preferred_provider, allow_provider_grace, stream_channel.virtual_id, - user.priority, + ctx.user.priority, + Some(user_session.token.as_str()), ) .await { @@ -840,17 +880,28 @@ pub async fn force_provider_stream_response( if stream_details.has_stream() || deferred_grace_hold_stream { let provider_response = stream_details.stream_info.as_ref().map(|(h, sc, url, cvt)| (h.clone(), *sc, url.clone(), *cvt)); - app_state.active_users.update_session_addr(&user.username, &user_session.token, &fingerprint.addr).await; + if ctx.session_reservation_ttl_secs > 0 { + if let Some(provider_name) = stream_details.provider_name.as_ref() { + app_state + .active_provider + .refresh_provider_reservation(provider_name, &user_session.token, ctx.session_reservation_ttl_secs) + .await; + } + } + app_state + .active_users + .update_session_addr(&ctx.user.username, &user_session.token, &fingerprint.addr) + .await; stream_channel.shared = share_stream; let stream = create_active_client_stream( stream_details, app_state, - user, + ctx.user, connection_permission, fingerprint, stream_channel, Some(&user_session.token), - req_headers, + ctx.req_headers, ) .await; @@ -863,7 +914,7 @@ pub async fn force_provider_stream_response( let body_stream = prepare_body_stream(app_state, item_type, stream); debug_if_enabled!( "Streaming provider forced stream request from {}", - sanitize_sensitive_info(resolve_request_url_for_logging(input, user_session.stream_url.as_ref()).as_ref()) + sanitize_sensitive_info(resolve_request_url_for_logging(ctx.input, user_session.stream_url.as_ref()).as_ref()) ); return try_unwrap_body!(response.body(body_stream)); } @@ -916,12 +967,7 @@ pub async fn stream_response( let virtual_id = stream_channel.virtual_id; let item_type = stream_channel.item_type; - // Grace candidates are intentionally not shared: - // they are transient and may be terminated after the grace check. - // Keeping them out of the shared-stream lifecycle avoids cross-effects on - // concurrently running streams of the same user. - let share_stream = - is_stream_share_enabled(item_type, target) && connection_permission != UserConnectionPermission::GracePeriod; + let share_stream = is_stream_share_enabled(item_type, target); let _shared_lock = if share_stream { let write_lock = app_state.app_config.file_locks.write_lock_str(stream_url).await; @@ -981,6 +1027,7 @@ pub async fn stream_response( true, stream_channel.virtual_id, user.priority, + Some(session_token), ) .await { @@ -1022,9 +1069,6 @@ pub async fn stream_response( } let mut is_stream_shared = share_stream; - if stream_details.provider_grace_active { - is_stream_shared = false; - } if let Some((_header, _status_code, _url, Some(_custom_video))) = stream_details.stream_info.as_ref() { if stream_details.stream.is_some() { is_stream_shared = false; @@ -1143,6 +1187,13 @@ pub async fn stream_response( connection_permission, ) .await; + let reservation_ttl_secs = get_session_reservation_ttl_secs(app_state, item_type); + if reservation_ttl_secs > 0 { + app_state + .active_provider + .refresh_provider_reservation(&provider, session_token, reservation_ttl_secs) + .await; + } } } @@ -1168,6 +1219,33 @@ fn get_stream_throttle(app_state: &Arc) -> u64 { .unwrap_or_default() } +fn get_stream_config_u64(app_state: &Arc, selector: impl FnOnce(&crate::model::StreamConfig) -> u64) -> u64 { + app_state + .app_config + .config + .load() + .reverse_proxy + .as_ref() + .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) + .map_or(0, selector) +} + +pub(crate) fn get_hls_session_ttl_secs(app_state: &Arc) -> u64 { + get_stream_config_u64(app_state, |stream| stream.hls_session_ttl_secs) +} + +pub(crate) fn get_catchup_session_ttl_secs(app_state: &Arc) -> u64 { + get_stream_config_u64(app_state, |stream| stream.catchup_session_ttl_secs) +} + +pub(crate) fn get_session_reservation_ttl_secs(app_state: &Arc, item_type: PlaylistItemType) -> u64 { + match item_type { + PlaylistItemType::LiveHls => get_hls_session_ttl_secs(app_state), + PlaylistItemType::Catchup => get_catchup_session_ttl_secs(app_state), + _ => 0, + } +} + #[allow(clippy::too_many_arguments)] async fn try_shared_stream_response_if_any( app_state: &Arc, @@ -1658,6 +1736,10 @@ pub fn create_session_fingerprint(fingerprint: &Fingerprint, username: &str, vir concat_string!(&fingerprint.key, "|", username, "|", &virtual_id.to_string()) } +pub fn create_catchup_session_key(fingerprint: &Fingerprint, username: &str, virtual_id: u32) -> String { + concat_string!("catchup|", &fingerprint.key, "|", username, "|", &virtual_id.to_string(), "|session") +} + pub fn stream_json_array

(iter: Box + Send>) -> axum::response::Response where P: serde::Serialize + Send + 'static, diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index 76ca9d16c..0bef4b36b 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -2,6 +2,7 @@ use crate::{ api::{ api_utils::{ create_session_fingerprint, force_provider_stream_response, get_headers_from_request, + get_hls_session_ttl_secs, get_stream_alternative_url, is_seek_request, local_stream_response, try_option_bad_request, try_unwrap_body, HeaderFilter, }, @@ -103,74 +104,104 @@ 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 (request_url, session_token) = match user_session { - Some(session) => { - let handle = if let Some(handle) = app_state - .active_provider - .acquire_exact_connection_with_grace(&session.provider, &fingerprint.addr, false, user.priority) - .await - { - Some(handle) - } else { - debug_if_enabled!( - "HLS pinned provider {} unavailable for {}; falling back to lineup allocation", - 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, user.priority).await - }; + 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 { + let provider_handle = if let Some(handle) = app_state + .active_provider + .acquire_connection_with_grace_for_session( + &input.name, + &fingerprint.addr, + false, + user.priority, + Some(session.token.as_str()), + ) + .await + { + Some(handle) + } else { + debug_if_enabled!( + "HLS pinned provider {} unavailable for {}; falling back to lineup allocation", + 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, user.priority).await + }; - match handle { - Some(provider_handle) => match provider_handle.allocation { - ProviderAllocation::Exhausted => (url, None), - ProviderAllocation::Available(cfg) | ProviderAllocation::GracePeriod(cfg) => { - let stream_url = get_stream_alternative_url(&url, input, &cfg); - let session_token = app_state - .active_users - .create_user_session( - user, - &session.token, - virtual_id, - &cfg.name, - &stream_url, - &fingerprint.addr, - connection_permission, - ) - .await; - (stream_url, Some(session_token)) - } - }, - None => (url, None), - } - } - None => match app_state.active_provider.get_next_provider(&input.name).await { - Some(provider_cfg) => { - let stream_url = get_stream_alternative_url(&url, input, &provider_cfg); - debug_if_enabled!( - "API endpoint [HLS] create_session_fingerprint user={} virtual_id={virtual_id} provider={} stream_url={}", - sanitize_sensitive_info(&user.username), - provider_cfg.name, - sanitize_sensitive_info(&stream_url) - ); - let user_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id); + match provider_handle.as_ref().map(|handle| &handle.allocation) { + Some(ProviderAllocation::Exhausted) => (url, None, provider_handle), + Some(ProviderAllocation::Available(cfg) | ProviderAllocation::GracePeriod(cfg)) => { + let stream_url = get_stream_alternative_url(&url, input, cfg); let session_token = app_state .active_users .create_user_session( user, - &user_session_token, + &session.token, virtual_id, - &provider_cfg.name, + &cfg.name, &stream_url, &fingerprint.addr, connection_permission, ) .await; - (stream_url, Some(session_token)) + app_state + .active_provider + .refresh_provider_reservation(&cfg.name, &session_token, hls_session_ttl_secs) + .await; + (stream_url, Some(session_token), provider_handle) } - None => (url, None), - }, + None => (url, None, None), + } + } else { + let user_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id); + match app_state + .active_provider + .acquire_connection_with_grace_for_session( + &input.name, + &fingerprint.addr, + false, + user.priority, + Some(&user_session_token), + ) + .await + { + Some(provider_handle) => match provider_handle.allocation.get_provider_config() { + Some(provider_cfg) => { + let stream_url = get_stream_alternative_url(&url, input, &provider_cfg); + debug_if_enabled!( + "API endpoint [HLS] create_session_fingerprint user={} virtual_id={virtual_id} provider={} stream_url={}", + sanitize_sensitive_info(&user.username), + provider_cfg.name, + sanitize_sensitive_info(&stream_url) + ); + let session_token = app_state + .active_users + .create_user_session( + user, + &user_session_token, + virtual_id, + &provider_cfg.name, + &stream_url, + &fingerprint.addr, + connection_permission, + ) + .await; + app_state + .active_provider + .refresh_provider_reservation(&provider_cfg.name, &session_token, hls_session_ttl_secs) + .await; + (stream_url, Some(session_token), Some(provider_handle)) + } + None => (url, None, Some(provider_handle)), + }, + None => (url, None, None), + } }; + // Playlist requests only need the chosen provider account to derive the URL and pin the session. + // Holding the provider slot until the first segment request causes stale active connections and + // breaks forced same-account reuse on the next HLS/Catchup stream request. + app_state.connection_manager.release_provider_handle(provider_handle).await; + // Don't forward Range on playlist fetch; segments use original headers in provider path let filter_header: HeaderFilter = Some(Box::new(|name: &str| !name.eq_ignore_ascii_case("range"))); let forwarded = get_headers_from_request(req_headers, &filter_header); @@ -314,9 +345,18 @@ async fn hls_api_stream( ); debug_if_enabled!("ID chain for hls endpoint: request_stream_id={} -> virtual_id={virtual_id}", params.stream_id); - let user_session_token = create_session_fingerprint(&fingerprint, &user.username, virtual_id); - let mut user_session = - app_state.active_users.get_and_update_user_session(&user.username, &user_session_token).await; + let encrypt_secret = app_state.get_encrypt_secret(); + let Some(decoded_hls_token) = get_hls_session_token_and_url_from_token(&encrypt_secret, ¶ms.token) else { + return axum::http::StatusCode::BAD_REQUEST.into_response(); + }; + let lookup_session_token = decoded_hls_token + .0 + .clone() + .unwrap_or_else(|| create_session_fingerprint(&fingerprint, &user.username, virtual_id)); + let mut user_session = app_state + .active_users + .get_and_update_user_session(&user.username, &lookup_session_token) + .await; if let Some(session) = &mut user_session { if session.permission == UserConnectionPermission::Exhausted { @@ -337,10 +377,9 @@ async fn hls_api_stream( .into_response(); } - let encrypt_secret = app_state.get_encrypt_secret(); - let hls_url = match get_hls_session_token_and_url_from_token(&encrypt_secret, ¶ms.token) { - Some((Some(session_token), hls_url)) if session.token.eq(&session_token) => hls_url, - Some((None, hls_url)) => hls_url, + let hls_url = match decoded_hls_token { + (Some(session_token), hls_url) if session.token.eq(&session_token) => hls_url, + (None, hls_url) => hls_url, _ => return axum::http::StatusCode::BAD_REQUEST.into_response(), }; let hls_url = hls_url.intern(); @@ -354,9 +393,12 @@ async fn hls_api_stream( &app_state, session, stream_channel, - &req_headers, - &input, - &user, + crate::api::api_utils::ForceStreamRequestContext { + req_headers: &req_headers, + input: &input, + user: &user, + session_reservation_ttl_secs: get_hls_session_ttl_secs(&app_state), + }, ) .await .into_response(); @@ -409,7 +451,18 @@ async fn hls_api_stream( } let stream_channel = resolve_stream_channel(&app_state, &target, virtual_id, &hls_url).await; - force_provider_stream_response(&fingerprint, &app_state, session, stream_channel, &req_headers, &input, &user) + force_provider_stream_response( + &fingerprint, + &app_state, + session, + stream_channel, + crate::api::api_utils::ForceStreamRequestContext { + req_headers: &req_headers, + input: &input, + user: &user, + session_reservation_ttl_secs: get_hls_session_ttl_secs(&app_state), + }, + ) .await .into_response() } else { diff --git a/backend/src/api/endpoints/m3u_api.rs b/backend/src/api/endpoints/m3u_api.rs index cf2030333..03a66856d 100644 --- a/backend/src/api/endpoints/m3u_api.rs +++ b/backend/src/api/endpoints/m3u_api.rs @@ -1,9 +1,10 @@ use crate::{ api::{ api_utils::{ - create_session_fingerprint, force_provider_stream_response, get_user_target, - get_user_target_by_credentials, is_seek_request, local_stream_response, redirect, redirect_response, - resource_response, separate_number_and_remainder, stream_response, try_option_bad_request, + create_catchup_session_key, create_session_fingerprint, force_provider_stream_response, + get_session_reservation_ttl_secs, get_user_target, get_user_target_by_credentials, is_seek_request, + local_stream_response, redirect, redirect_response, resource_response, separate_number_and_remainder, + stream_response, try_option_bad_request, try_result_bad_request, try_result_not_found, try_unwrap_body, RedirectParams, }, endpoints::{ @@ -142,7 +143,11 @@ 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 session_key = create_session_fingerprint(fingerprint, &user.username, virtual_id); + let session_key = if pli.item_type == PlaylistItemType::Catchup { + create_catchup_session_key(fingerprint, &user.username, virtual_id) + } else { + create_session_fingerprint(fingerprint, &user.username, virtual_id) + }; let user_session = app_state.active_users.get_and_update_user_session(&user.username, &session_key).await; let session_url = if let Some(session) = &user_session { @@ -170,9 +175,12 @@ async fn m3u_api_stream( app_state, session, pli.to_stream_channel(target.id), - req_headers, - &input, - &user, + crate::api::api_utils::ForceStreamRequestContext { + req_headers, + input: &input, + user: &user, + session_reservation_ttl_secs: get_session_reservation_ttl_secs(app_state, pli.item_type), + }, ) .await .into_response(); diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index 6ebf1f4d8..fd667f376 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -4,10 +4,10 @@ use crate::{ api::{ api_utils, api_utils::{ - create_api_proxy_user, create_session_fingerprint, empty_json_response_as_array, - empty_json_response_as_object, force_provider_stream_response, get_user_target, - get_user_target_by_credentials, internal_server_error, is_seek_request, local_stream_response, redirect, - redirect_response, resource_response, separate_number_and_remainder, stream_response, + create_api_proxy_user, create_catchup_session_key, create_session_fingerprint, empty_json_response_as_array, + empty_json_response_as_object, force_provider_stream_response, get_session_reservation_ttl_secs, + get_user_target, get_user_target_by_credentials, internal_server_error, is_seek_request, + local_stream_response, redirect, redirect_response, resource_response, separate_number_and_remainder, stream_response, try_option_bad_request, try_result_bad_request, try_result_not_found, try_unwrap_body, RedirectParams, }, endpoints::{ @@ -298,7 +298,11 @@ 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 session_key = create_session_fingerprint(fingerprint, &user.username, virtual_id); + let session_key = if item_type == PlaylistItemType::Catchup { + create_catchup_session_key(fingerprint, &user.username, virtual_id) + } else { + create_session_fingerprint(fingerprint, &user.username, virtual_id) + }; let user_session = app_state.active_users.get_and_update_user_session(&user.username, &session_key).await; let session_url = if let Some(session) = &user_session { @@ -329,9 +333,12 @@ async fn xtream_player_api_stream( app_state, session, stream_channel, - req_headers, - &input, - &user, + crate::api::api_utils::ForceStreamRequestContext { + req_headers, + input: &input, + user: &user, + session_reservation_ttl_secs: get_session_reservation_ttl_secs(app_state, item_type), + }, ) .await .into_response(); diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs index 90eb71fd5..3ae198b70 100644 --- a/backend/src/api/model/active_provider_manager.rs +++ b/backend/src/api/model/active_provider_manager.rs @@ -6,7 +6,7 @@ use crate::{ model::{AppConfig, ConfigInput, GracePeriodOptions}, utils::debug_if_enabled, }; -use log::{error, warn}; +use log::error; use shared::utils::sanitize_sensitive_info; use std::cmp::Reverse; use std::collections::{BTreeMap, HashMap, HashSet}; @@ -18,11 +18,10 @@ use std::{ }, time::{Duration, Instant}, }; -use tokio::sync::{RwLock, Semaphore}; +use tokio::sync::RwLock; +use tokio::time::Instant as TokioInstant; use tokio_util::sync::CancellationToken; -const PREEMPTED_PROBE_CANCEL_GRACE: Duration = Duration::from_secs(2); -const PREEMPTED_GRACE_MAX_PENDING: usize = 64; static DUMMY_ADDR: LazyLock = LazyLock::new(|| SocketAddr::from(([127, 0, 0, 1], 0))); pub type ClientConnectionId = SocketAddr; @@ -79,7 +78,12 @@ struct ActiveConnectionInfo { cancel_token: CancellationToken, created_at: Instant, priority: i8, - is_probe: bool, +} + +#[derive(Debug, Clone)] +struct ProviderReservation { + owner: Arc, + expires_at: TokioInstant, } #[derive(Debug, Clone, Default)] @@ -105,20 +109,21 @@ struct Connections { pub struct ActiveProviderManager { providers: ProviderLineupManager, connections: RwLock, + reservations: RwLock, ProviderReservation>>, next_allocation_id: AtomicU64, - preempted_grace_semaphore: Arc, shared_stream_manager: OnceLock>, } impl ActiveProviderManager { + pub fn new(cfg: &AppConfig, event_manager: &Arc) -> Self { let grace_period_options = Self::get_grace_options(cfg); let inputs = Self::get_config_inputs(cfg); Self { providers: ProviderLineupManager::new(inputs, grace_period_options, event_manager), connections: RwLock::new(Connections::default()), + reservations: RwLock::new(HashMap::new()), next_allocation_id: AtomicU64::new(1), - preempted_grace_semaphore: Arc::new(Semaphore::new(PREEMPTED_GRACE_MAX_PENDING)), shared_stream_manager: OnceLock::new(), } } @@ -165,6 +170,149 @@ impl ActiveProviderManager { self.providers.reconcile_connections(counts).await; } + fn prune_expired_reservations(reservations: &mut HashMap, ProviderReservation>) { + let now = TokioInstant::now(); + reservations.retain(|_, reservation| reservation.expires_at > now); + } + + // Reservation ownership is grouped by the `fingerprint|username` prefix and intentionally drops the + // trailing session/input suffix. That lets a client switch channels and immediately reuse the same + // reserved provider account without waiting for the old channel reservation to expire. + // Tradeoff: a new reservation for one channel clears any other reservation in the same family, so + // concurrent multi-channel playback for the same client family is not supported by this grouping. + // If concurrent streams must be supported, reservation matching must use the full owner identifier. + fn reservation_family(owner: &str) -> &str { + owner.rsplit_once('|').map_or(owner, |(family, _)| family) + } + + // Compare reservations at the family level rather than exact owner equality so channel-switching can + // take over an existing reservation across different `virtual_id`s for the same client family. + // This shares the same tradeoff as `reservation_family`: same-family reservations replace each other. + fn is_same_reservation_family(reservation_owner: &str, session_owner: &str) -> bool { + Self::reservation_family(reservation_owner) == Self::reservation_family(session_owner) + } + + async fn is_reserved_for_other(&self, provider_name: &Arc, session_owner: Option<&str>) -> bool { + let mut reservations = self.reservations.write().await; + Self::prune_expired_reservations(&mut reservations); + reservations.get(provider_name).is_some_and(|reservation| { + session_owner.is_none_or(|owner| !Self::is_same_reservation_family(reservation.owner.as_ref(), owner)) + }) + } + + async fn get_reserved_provider_for_owner(&self, input_name: &Arc, session_owner: &str) -> Option> { + let mut reservations = self.reservations.write().await; + Self::prune_expired_reservations(&mut reservations); + reservations.iter().find_map(|(provider_name, reservation)| { + if Self::is_same_reservation_family(reservation.owner.as_ref(), session_owner) + && self.providers.is_provider_for_input(provider_name, input_name) + { + Some(provider_name.clone()) + } else { + None + } + }) + } + + async fn reserved_provider_names_for_other( + &self, + input_name: &Arc, + session_owner: Option<&str>, + ) -> HashSet> { + let mut reserved = HashSet::new(); + for provider_name in self.providers.provider_names_for_input(input_name) { + if self.is_reserved_for_other(&provider_name, session_owner).await { + reserved.insert(provider_name); + } + } + reserved + } + + pub async fn refresh_provider_reservation(&self, provider_name: &Arc, session_owner: &str, ttl_secs: u64) { + let mut reservations = self.reservations.write().await; + Self::prune_expired_reservations(&mut reservations); + // Refresh operates on the reservation family, not the exact owner string. This is what allows a + // same-client channel switch to move the pinned provider account to the new stream immediately. + // The tradeoff is that other reservations from the same family are cleared here as well. + reservations.retain(|_, reservation| !Self::is_same_reservation_family(reservation.owner.as_ref(), session_owner)); + if ttl_secs == 0 { + return; + } + reservations.insert( + provider_name.clone(), + ProviderReservation { + owner: Arc::from(session_owner), + expires_at: TokioInstant::now() + Duration::from_secs(ttl_secs), + }, + ); + } + + async fn acquire_exact_connection_inner( + &self, + provider_name: &Arc, + addr: &SocketAddr, + allow_grace: bool, + priority: i8, + session_owner: Option<&str>, + ) -> Option { + if self.is_reserved_for_other(provider_name, session_owner).await { + return None; + } + let allocation = self.providers.acquire_exact_connection_with_grace_override(provider_name, allow_grace).await; + if matches!(allocation, ProviderAllocation::Exhausted) { + return None; + } + self.register_allocation(allocation, addr, priority).await + } + + async fn finalize_lineup_allocation( + &self, + input_name: &Arc, + addr: &SocketAddr, + allow_grace: bool, + priority: i8, + allocation: ProviderAllocation, + session_owner: Option<&str>, + ) -> Option { + if matches!(&allocation, ProviderAllocation::GracePeriod(_)) + && self.evict_lower_priority_on_input(input_name, priority, session_owner).await + { + let evicted_on_same = !self.providers.is_over_limit(&allocation.get_provider_name().unwrap_or_default()).await; + if !evicted_on_same { + let new_alloc = self.providers.acquire_connection_with_grace_override(input_name, allow_grace).await; + if !matches!(new_alloc, ProviderAllocation::Exhausted) { + if let Some(provider_name) = new_alloc.get_provider_name() { + if !self.is_reserved_for_other(&provider_name, session_owner).await { + allocation.release().await; + return self.register_allocation(new_alloc, addr, priority).await; + } + } + new_alloc.release().await; + } + } + } + + if matches!(&allocation, ProviderAllocation::GracePeriod(_)) + && self.has_non_preemptable_connection_on_input(input_name, priority).await + { + allocation.release().await; + return None; + } + + self.register_allocation(allocation, addr, priority).await + } + + async fn has_non_preemptable_connection_on_input(&self, input_name: &Arc, new_priority: i8) -> bool { + let connections = self.connections.read().await; + connections.priority_index.iter().any(|(prov_name, tree)| { + self.providers.is_provider_for_input(prov_name, input_name) + && tree + .iter() + .next() + .is_some_and(|((prio, _, _), _)| *prio <= new_priority) + }) + } + async fn acquire_connection_inner( &self, provider_or_input_name: &Arc, @@ -172,60 +320,59 @@ impl ActiveProviderManager { force: bool, allow_grace_override: Option, priority: i8, - is_probe: bool, + session_owner: Option<&str>, ) -> Option { - // 1. Try to acquire directly - let (allow_grace, allocation) = if force { - (true, self.providers.force_exact_acquire_connection(provider_or_input_name).await) - } else { - match allow_grace_override { - Some(allow_grace) => ( - allow_grace, - self.providers.acquire_connection_with_grace_override(provider_or_input_name, allow_grace).await, - ), - None => (true, self.providers.acquire_connection(provider_or_input_name).await), - } - }; + let allow_grace = if force { true } else { allow_grace_override.unwrap_or(true) }; - if !matches!(allocation, ProviderAllocation::Exhausted) { - if matches!(&allocation, ProviderAllocation::GracePeriod(_)) && !force { - // Grace allocation received — try to evict a lower-priority victim - // across the entire input lineup (all aliases) to free capacity. - if self.evict_lower_priority_on_input(provider_or_input_name, priority).await { - // A victim was evicted. If it was on the same alias as our grace - // allocation, the provider is back at max (not over limit). - // If it was on a different alias, release the grace allocation - // and re-acquire on the input to land on the freed alias. - let evicted_on_same = !self.providers.is_over_limit( - &allocation.get_provider_name().unwrap_or_default(), - ).await; - if !evicted_on_same { - // Victim was on a different alias — first try to re-acquire. - // Keep the existing grace allocation as fallback if re-acquire fails. - let new_alloc = self.providers.acquire_connection_with_grace_override( - provider_or_input_name, allow_grace, - ).await; - if !matches!(new_alloc, ProviderAllocation::Exhausted) { - allocation.release().await; - return self.register_allocation(new_alloc, addr, priority, is_probe).await; - } - return self.register_allocation(allocation, addr, priority, is_probe).await; - } - } - // Eviction on same alias succeeded, or no victim found → keep grace allocation + if force { + let allocation = self.providers.force_exact_acquire_connection(provider_or_input_name).await; + if matches!(allocation, ProviderAllocation::Exhausted) { + return None; } - return self.register_allocation(allocation, addr, priority, is_probe).await; + return self.register_allocation(allocation, addr, priority).await; } - // 2. If exhausted, try preemption (kick lower priority connection) - if !force { - if let Some(preempted_alloc) = - self.try_preempt_connection(provider_or_input_name, priority, allow_grace).await - { - return self.register_allocation(preempted_alloc, addr, priority, is_probe).await; + if let Some(owner) = session_owner { + if let Some(reserved_provider) = self.get_reserved_provider_for_owner(provider_or_input_name, owner).await { + return self.acquire_exact_connection_inner(&reserved_provider, addr, allow_grace, priority, Some(owner)).await; } } + let candidate_count = self.providers.provider_names_for_input(provider_or_input_name).len(); + let attempts = candidate_count.max(1); + let mut skipped_reserved = HashSet::new(); + for _ in 0..attempts { + let allocation = self.providers.acquire_connection_with_grace_override(provider_or_input_name, allow_grace).await; + if matches!(allocation, ProviderAllocation::Exhausted) { + break; + } + if let Some(provider_name) = allocation.get_provider_name() { + if self.is_reserved_for_other(&provider_name, session_owner).await { + debug_if_enabled!( + "Skipping reserved provider {} for {}", + sanitize_sensitive_info(&provider_name), + sanitize_sensitive_info(&addr.to_string()) + ); + skipped_reserved.insert(provider_name); + allocation.release().await; + if skipped_reserved.len() >= attempts { + break; + } + continue; + } + } + return self + .finalize_lineup_allocation(provider_or_input_name, addr, allow_grace, priority, allocation, session_owner) + .await; + } + + if let Some(preempted_alloc) = self + .try_preempt_connection(provider_or_input_name, priority, allow_grace, session_owner) + .await + { + return self.register_allocation(preempted_alloc, addr, priority).await; + } + None } @@ -234,7 +381,6 @@ impl ActiveProviderManager { allocation: ProviderAllocation, addr: &SocketAddr, priority: i8, - is_probe: bool, ) -> Option { let provider_name = allocation.get_provider_name().unwrap_or_default(); let allocation_id = self.next_allocation_id.fetch_add(1, Ordering::Relaxed); @@ -251,7 +397,6 @@ impl ActiveProviderManager { cancel_token: cancel_token.clone(), created_at: now, priority, - is_probe, }, ); @@ -275,7 +420,10 @@ impl ActiveProviderManager { &self, input_name: &Arc, new_priority: i8, + session_owner: Option<&str>, ) -> bool { + let reserved_providers = self.reserved_provider_names_for_other(input_name, session_owner).await; + // Find best victim across all aliases under read lock let victim: Option<(PriorityOwner, AllocationId, i8, Instant)> = { let connections = self.connections.read().await; @@ -284,20 +432,12 @@ impl ActiveProviderManager { if !self.providers.is_provider_for_input(prov_name, input_name) { continue; } - for ((prio, Reverse(created_at), alloc_id), owner) in tree.iter().rev() { + if reserved_providers.contains(prov_name) { + continue; + } + if let Some(((prio, Reverse(created_at), alloc_id), owner)) = tree.iter().next_back() { if *prio <= new_priority { - break; // No more victims on this alias - } - if let PriorityOwner::Shared(shared_id) = owner { - let evictable = connections - .shared - .shared_by_allocation_id - .get(shared_id) - .and_then(|key| connections.shared.by_key.get(key)) - .is_some_and(|s| s.connections.len() == 1); - if !evictable { - continue; - } + continue; } let is_better = match found { None => true, @@ -308,7 +448,6 @@ impl ActiveProviderManager { if is_better { found = Some((*owner, *alloc_id, *prio, *created_at)); } - break; // Only best candidate per alias } } found @@ -331,13 +470,12 @@ impl ActiveProviderManager { return false; }; - let still_single = connections.shared.by_key.get(&key).is_some_and(|shared| { + let still_match = connections.shared.by_key.get(&key).is_some_and(|shared| { shared.allocation_id == alloc_id - && shared.connections.len() == 1 && shared.priority == v_prio && shared.created_at == victim_created_at }); - if !still_single { + if !still_match { return false; } @@ -413,24 +551,8 @@ impl ActiveProviderManager { }; let Some(info) = removed_info else { return false; }; - let token = info.cancel_token; - if info.is_probe { - if let Ok(permit) = Arc::clone(&self.preempted_grace_semaphore).try_acquire_owned() { - tokio::spawn(async move { - let _permit = permit; - tokio::time::sleep(PREEMPTED_PROBE_CANCEL_GRACE).await; - token.cancel(); - }); - } else { - warn!( - "Preemption grace semaphore exhausted ({PREEMPTED_GRACE_MAX_PENDING} tasks pending); \ - falling back to immediate cancellation for preempted probe" - ); - token.cancel(); - } - } else { - token.cancel(); - } + // Preempted probes must stop immediately; they must not keep a custom stream alive. + info.cancel_token.cancel(); info.allocation.release().await; } } @@ -444,7 +566,9 @@ impl ActiveProviderManager { input_name: &Arc, new_priority: i8, allow_grace: bool, + session_owner: Option<&str>, ) -> Option { + let reserved_providers = self.reserved_provider_names_for_other(input_name, session_owner).await; // Victim: (owner, alloc_id, priority, created_at) let mut victim: Option<(PriorityOwner, AllocationId, i8, Instant)> = None; @@ -456,21 +580,13 @@ impl ActiveProviderManager { if !self.providers.is_provider_for_input(prov_name, input_name) { continue; } - // Iterate from highest priority value (lowest importance) = best victim - for ((prio, Reverse(created_at), alloc_id), owner) in tree.iter().rev() { + if reserved_providers.contains(prov_name) { + continue; + } + // Highest priority value (lowest importance) is the best victim candidate. + if let Some(((prio, Reverse(created_at), alloc_id), owner)) = tree.iter().next_back() { if *prio <= new_priority { - break; // No more victims on this alias - } - if let PriorityOwner::Shared(shared_id) = owner { - let evictable = connections - .shared - .shared_by_allocation_id - .get(shared_id) - .and_then(|key| connections.shared.by_key.get(key)) - .is_some_and(|s| s.connections.len() == 1); - if !evictable { - continue; - } + continue; } let is_better = match victim { None => true, @@ -481,7 +597,6 @@ impl ActiveProviderManager { if is_better { victim = Some((*owner, *alloc_id, *prio, *created_at)); } - break; // Only need the best candidate per alias } } } @@ -496,15 +611,13 @@ impl ActiveProviderManager { let mut connections = self.connections.write().await; let key = connections.shared.shared_by_allocation_id.get(&shared_id).cloned()?; - // Revalidate the selected shared victim under WRITE lock to avoid evicting - // a connection that gained additional listeners concurrently. - let still_single = connections.shared.by_key.get(&key).is_some_and(|shared| { + // Revalidate the selected shared victim under WRITE lock. + let still_match = connections.shared.by_key.get(&key).is_some_and(|shared| { shared.allocation_id == alloc_id - && shared.connections.len() == 1 && shared.priority == v_prio && shared.created_at == victim_created_at }); - if !still_single { + if !still_match { None } else if let Some(shared) = connections.shared.by_key.remove(&key) { connections.shared.shared_by_allocation_id.remove(&shared.allocation_id); @@ -589,34 +702,25 @@ impl ActiveProviderManager { return None; }; - let token = info.cancel_token; - if info.is_probe { - // Probe preemption gets a short grace window, but cap detached sleep/cancel - // tasks so bursts cannot spawn unbounded background work. - if let Ok(permit) = Arc::clone(&self.preempted_grace_semaphore).try_acquire_owned() { - tokio::spawn(async move { - let _permit = permit; - tokio::time::sleep(PREEMPTED_PROBE_CANCEL_GRACE).await; - token.cancel(); - }); - } else { - warn!( - "Preemption grace semaphore exhausted ({PREEMPTED_GRACE_MAX_PENDING} tasks pending); \ - falling back to immediate cancellation for preempted probe" - ); - token.cancel(); - } - } else { - token.cancel(); - } - + // Preempted probes must stop immediately; they must not keep a custom stream alive. + info.cancel_token.cancel(); info.allocation.release().await; } } // Now try acquire again preserving the original grace policy. - let allocation = self.providers.acquire_connection_with_grace_override(input_name, allow_grace).await; - if !matches!(allocation, ProviderAllocation::Exhausted) { + let attempts = self.providers.provider_names_for_input(input_name).len().max(1); + for _ in 0..attempts { + let allocation = self.providers.acquire_connection_with_grace_override(input_name, allow_grace).await; + if matches!(allocation, ProviderAllocation::Exhausted) { + break; + } + if let Some(provider_name) = allocation.get_provider_name() { + if reserved_providers.contains(&provider_name) { + allocation.release().await; + continue; + } + } return Some(allocation); } } @@ -631,11 +735,18 @@ impl ActiveProviderManager { allow_grace: bool, priority: i8, ) -> Option { - let allocation = self.providers.acquire_exact_connection_with_grace_override(provider_name, allow_grace).await; - if matches!(allocation, ProviderAllocation::Exhausted) { - return None; - } - self.register_allocation(allocation, addr, priority, false).await + self.acquire_exact_connection_with_grace_for_session(provider_name, addr, allow_grace, priority, None).await + } + + pub async fn acquire_exact_connection_with_grace_for_session( + &self, + provider_name: &Arc, + addr: &SocketAddr, + allow_grace: bool, + priority: i8, + session_owner: Option<&str>, + ) -> Option { + self.acquire_exact_connection_inner(provider_name, addr, allow_grace, priority, session_owner).await } pub async fn force_exact_acquire_connection( @@ -650,7 +761,7 @@ impl ActiveProviderManager { // Returns the next available provider connection pub async fn acquire_connection(&self, input_name: &Arc, addr: &SocketAddr, priority: i8) -> Option { - self.acquire_connection_inner(input_name, addr, false, None, priority, false).await + self.acquire_connection_inner(input_name, addr, false, None, priority, None).await } /// Acquire a provider connection while explicitly controlling provider-side grace allocations. @@ -661,13 +772,24 @@ impl ActiveProviderManager { allow_grace: bool, priority: i8, ) -> Option { - self.acquire_connection_inner(input_name, addr, false, Some(allow_grace), priority, false).await + self.acquire_connection_with_grace_for_session(input_name, addr, allow_grace, priority, None).await + } + + pub async fn acquire_connection_with_grace_for_session( + &self, + input_name: &Arc, + addr: &SocketAddr, + allow_grace: bool, + priority: i8, + session_owner: Option<&str>, + ) -> Option { + self.acquire_connection_inner(input_name, addr, false, Some(allow_grace), priority, session_owner).await } /// Acquire a provider connection for probe tasks with configurable priority. /// Probes never consume grace capacity. pub async fn acquire_connection_for_probe(&self, input_name: &Arc, priority: i8) -> Option { - self.acquire_connection_inner(input_name, &DUMMY_ADDR, false, Some(false), priority, true).await + self.acquire_connection_inner(input_name, &DUMMY_ADDR, false, Some(false), priority, None).await } // This method is used for redirects to cycle through the provider @@ -997,6 +1119,7 @@ impl ActiveProviderManager { } pub async fn get_provider_connections_count(&self) -> usize { self.providers.active_connection_count().await } + } #[cfg(test)] @@ -1180,8 +1303,8 @@ mod tests { manager.release_connection(&client_2_addr).await; } - #[tokio::test(start_paused = true)] - async fn test_probe_preemption_releases_capacity_immediately_and_cancels_after_grace() { + #[tokio::test] + async fn test_probe_preemption_releases_capacity_and_cancels_immediately() { let app_cfg = create_test_app_config_single_provider_pool(); let event_manager = Arc::new(EventManager::new()); let manager = ActiveProviderManager::new(&app_cfg, &event_manager); @@ -1200,27 +1323,80 @@ mod tests { .expect("user allocation should preempt probe"); assert_eq!(user_alloc.allocation.get_provider_name().as_deref(), Some(input_name.as_ref())); - // Let the detached cancellation task start and arm its sleep timer on the paused clock. + // Cancellation happens inline during preemption; yield once to observe any deferred work. tokio::task::yield_now().await; - - // Probe cancellation is intentionally delayed by a small grace window. - tokio::time::advance(Duration::from_millis(100)).await; - assert!(!probe_token.is_cancelled(), "probe token should not be cancelled immediately"); - - let cancel_wait_timeout = super::PREEMPTED_PROBE_CANCEL_GRACE + Duration::from_millis(500); - let wait_token = probe_token.clone(); - let cancel_wait = - tokio::spawn(async move { tokio::time::timeout(cancel_wait_timeout, wait_token.cancelled()).await }); - tokio::task::yield_now().await; - tokio::time::advance(cancel_wait_timeout).await; - assert!( - cancel_wait.await.expect("cancel wait task should join").is_ok(), - "probe token should be cancelled before timeout after grace" - ); + assert!(probe_token.is_cancelled(), "probe token should be cancelled immediately after preemption"); manager.release_connection(&user_addr).await; } + #[tokio::test(start_paused = true)] + async fn test_session_provider_reservation_blocks_other_sessions_until_ttl_expires() { + let app_cfg = create_test_app_config_single_provider_pool(); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveProviderManager::new(&app_cfg, &event_manager); + + let input_name = "provider_1".intern(); + let owner_1 = "session-owner-1"; + let owner_2 = "session-owner-2"; + let addr_1: SocketAddr = "127.0.0.1:43101".parse().unwrap(); + let addr_2: SocketAddr = "127.0.0.1:43102".parse().unwrap(); + + manager.refresh_provider_reservation(&input_name, owner_1, 15).await; + + let first = manager + .acquire_connection_with_grace_for_session(&input_name, &addr_1, false, default_user_priority(), Some(owner_1)) + .await + .expect("reserved owner should reacquire its provider"); + assert_eq!(first.allocation.get_provider_name().as_deref(), Some(input_name.as_ref())); + manager.release_connection(&addr_1).await; + + let blocked = manager + .acquire_connection_with_grace_for_session(&input_name, &addr_2, false, default_user_priority(), Some(owner_2)) + .await; + assert!(blocked.is_none(), "other sessions must not take a reserved provider before TTL expiry"); + + tokio::time::advance(Duration::from_secs(16)).await; + + let second = manager + .acquire_connection_with_grace_for_session(&input_name, &addr_2, false, default_user_priority(), Some(owner_2)) + .await + .expect("reservation should expire after TTL"); + assert_eq!(second.allocation.get_provider_name().as_deref(), Some(input_name.as_ref())); + manager.release_connection(&addr_2).await; + } + + #[tokio::test(start_paused = true)] + async fn test_same_client_family_can_reuse_reserved_provider_after_channel_switch() { + let app_cfg = create_test_app_config_single_provider_pool(); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveProviderManager::new(&app_cfg, &event_manager); + + let input_name = "provider_1".intern(); + let first_addr: SocketAddr = "127.0.0.1:43111".parse().unwrap(); + let second_addr: SocketAddr = "127.0.0.1:43112".parse().unwrap(); + let owner_channel_1 = "client|ua|user|100"; + let owner_channel_2 = "client|ua|user|200"; + + manager.refresh_provider_reservation(&input_name, owner_channel_1, 15).await; + + let reacquired = manager + .acquire_connection_with_grace_for_session( + &input_name, + &second_addr, + false, + default_user_priority(), + Some(owner_channel_2), + ) + .await + .expect("same client family should be able to reuse the reserved provider after a channel switch"); + + assert_eq!(reacquired.allocation.get_provider_name().as_deref(), Some(input_name.as_ref())); + + manager.release_connection(&first_addr).await; + manager.release_connection(&second_addr).await; + } + #[tokio::test] async fn test_higher_priority_user_preempts_lower_priority_user() { // User with priority 5 (low) is connected; user with priority -1 (high) arrives. @@ -1350,9 +1526,7 @@ mod tests { } #[tokio::test] - async fn test_grace_period_no_victim_keeps_grace_behavior() { - // Provider full, same-prio user arrives with grace allowed, - // no victim available → normal grace behavior (allocation still returned). + async fn test_equal_priority_user_does_not_preempt_and_gets_rejected() { let app_cfg = create_test_app_config_single_provider_pool(); let event_manager = Arc::new(EventManager::new()); let manager = ActiveProviderManager::new(&app_cfg, &event_manager); @@ -1372,16 +1546,13 @@ mod tests { // Provider is now exhausted assert!(manager.is_exhausted(&input_name).await); - // User 2 arrives with same priority and grace allowed - // Should get a grace allocation but NOT evict user 1 + // User 2 arrives with same priority. + // Equal priority must not preempt and must not receive a grace allocation. let alloc2 = manager .acquire_connection(&input_name, &user_2_addr, default_user_priority()) .await; - // Grace allocation should be returned (provider allows grace) - if let Some(alloc2) = &alloc2 { - assert_eq!(alloc2.allocation.get_provider_name().as_deref(), Some(input_name.as_ref())); - } + assert!(alloc2.is_none(), "same-priority user must be rejected when provider capacity is exhausted"); // User 1 should NOT be cancelled assert!(!token1.is_cancelled(), "same-prio user should not be evicted"); @@ -1436,6 +1607,50 @@ mod tests { manager.release_connection(&addr_b).await; } + #[tokio::test] + async fn test_higher_priority_user_preempts_shared_stream_with_multiple_subscribers() { + let app_cfg = create_test_app_config_single_provider_pool(); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveProviderManager::new(&app_cfg, &event_manager); + + let input_name = "provider_1".intern(); + let stream_key = "http://example.com/shared/live/multi"; + let addr_a: SocketAddr = "127.0.0.1:48601".parse().unwrap(); + let addr_b: SocketAddr = "127.0.0.1:48602".parse().unwrap(); + let addr_high: SocketAddr = "127.0.0.1:48603".parse().unwrap(); + + let shared_alloc = manager + .acquire_connection(&input_name, &addr_a, 5) + .await + .expect("low-priority shared stream should get initial connection"); + let shared_token = shared_alloc.cancel_token.clone().expect("shared allocation should have cancel token"); + manager.make_shared_connection(&addr_a, stream_key).await; + + manager + .add_shared_connection(&addr_b, stream_key, 6) + .await + .expect("second subscriber should join shared stream"); + + let high_alloc = manager + .acquire_connection(&input_name, &addr_high, 0) + .await + .expect("higher-priority user should preempt lower-priority shared stream"); + assert_eq!(high_alloc.allocation.get_provider_name().as_deref(), Some(input_name.as_ref())); + assert!(shared_token.is_cancelled(), "shared stream should be cancelled when preempted"); + + { + let connections = manager.connections.read().await; + assert!( + !connections.shared.by_key.contains_key(stream_key), + "preempted shared stream must be removed even with multiple subscribers" + ); + } + + manager.release_connection(&addr_high).await; + manager.release_connection(&addr_a).await; + manager.release_connection(&addr_b).await; + } + #[tokio::test] async fn test_btree_index_consistent_after_lifecycle() { // Verify that the priority_index stays consistent through add, evict, release. diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index ebd1a884e..3fc450407 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -28,6 +28,7 @@ use tokio::sync::RwLock; const USER_GC_TTL: u64 = 900; // 15 Min const USER_CON_TTL: u64 = 10_800; // 3 hours const USER_SESSION_LIMIT: usize = 50; +const ANON_SOCKET_TTL: u64 = 300; // 5 Min fn get_grace_options(config: &Config) -> (u64, u64) { let (grace_period_millis, grace_period_timeout_secs) = @@ -89,7 +90,22 @@ impl UserConnectionData { struct UserConnections { kicked: HashMap, by_key: HashMap, - key_by_addr: HashMap, + key_by_addr: HashMap, +} + +#[derive(Clone, Debug)] +struct SocketRegistration { + username: String, + ts: u64, +} + +impl SocketRegistration { + fn anonymous() -> Self { + Self { + username: String::new(), + ts: current_time_secs(), + } + } } pub struct ActiveUserManager { @@ -112,7 +128,7 @@ impl ActiveUserManager { fps: String::from("30"), video_codec: String::from("H.264"), audio_codec: String::from("AAC"), - audio_channels: String::from("2.0"), + audio_channels: String::from("Stereo"), } } @@ -158,7 +174,7 @@ impl ActiveUserManager { let (removed_stream, username) = { let mut user_connections = self.connections.write().await; - let Some(username) = user_connections.key_by_addr.get(addr).cloned() else { + let Some(username) = user_connections.key_by_addr.get(addr).map(|reg| reg.username.clone()) else { return false; }; @@ -196,7 +212,8 @@ impl ActiveUserManager { let (addr_removed, disconnected_user) = { let mut user_connections = self.connections.write().await; - if let Some(username) = user_connections.key_by_addr.remove(addr) { + if let Some(registration) = user_connections.key_by_addr.remove(addr) { + let username = registration.username; if let Some(connection_data) = user_connections.by_key.get_mut(&username) { let prev_len = connection_data.streams.len(); connection_data.streams.retain(|c| c.addr != *addr); @@ -322,7 +339,7 @@ impl ActiveUserManager { let mut user_connections = self.connections.write().await; let username = { match user_connections.key_by_addr.get(addr) { - Some(username) => username.clone(), + Some(registration) => registration.username.clone(), None => return None, } }; @@ -342,10 +359,13 @@ impl ActiveUserManager { #[allow(clippy::too_many_arguments)] pub async fn add_connection(&self, addr: &SocketAddr) { + self.gc(); let mut user_connections = self.connections.write().await; - if !user_connections.key_by_addr.contains_key(addr) { - user_connections.key_by_addr.insert(*addr, String::new()); - } + user_connections + .key_by_addr + .entry(*addr) + .and_modify(|registration| registration.ts = current_time_secs()) + .or_insert_with(SocketRegistration::anonymous); } #[allow(clippy::too_many_arguments)] @@ -362,13 +382,20 @@ impl ActiveUserManager { let stream_info = { let mut user_connections = self.connections.write().await; - // needs to be registered through socket connection to avoid race time conditions through short disconnect - if !user_connections.key_by_addr.contains_key(&fingerprint.addr) { - return None; + let now = current_time_secs(); + if let Some(registration) = user_connections.key_by_addr.get_mut(&fingerprint.addr) { + registration.username = username.to_string(); + registration.ts = now; + } else { + user_connections.key_by_addr.insert( + fingerprint.addr, + SocketRegistration { + username: username.to_string(), + ts: now, + }, + ); } - user_connections.key_by_addr.insert(fingerprint.addr, username.to_string()); - let connection_data = user_connections .by_key .entry(username.to_string()) @@ -422,12 +449,12 @@ impl ActiveUserManager { session_token, ); - let total_active = user_connections.key_by_addr.len(); + let tracked_socket_count = user_connections.key_by_addr.len(); if let Some(connection_data) = user_connections.by_key.get_mut(username) { connection_data.connections += 1; connection_data.streams.push(stream_info.clone()); - Self::log_connection_added(username, &fingerprint.addr, connection_data, total_active); + Self::log_connection_added(username, &fingerprint.addr, connection_data, tracked_socket_count); } stream_info @@ -488,6 +515,7 @@ impl ActiveUserManager { for session in &mut connection_data.sessions { if session.token == session_token { session.ts = current_time_secs(); + session.addr = *addr; if &*session.stream_url != stream_url { session.stream_url = stream_url.intern(); } @@ -579,7 +607,7 @@ impl ActiveUserManager { username: &str, addr: &SocketAddr, connection_data: &UserConnectionData, - total_active_sockets: usize, + tracked_socket_count: usize, ) { if log::log_enabled!(log::Level::Debug) { let active_for_user = connection_data.connections; @@ -604,7 +632,7 @@ impl ActiveUserManager { ); } else { debug_if_enabled!( - "Added new connection for {username} at {} (active user connections={active_for_user}, total connections={total_active_sockets})", + "Added new connection for {username} at {} (active user connections={active_for_user}, tracked sockets={tracked_socket_count})", sanitize_sensitive_info(&addr.to_string()) ); } @@ -623,7 +651,12 @@ impl ActiveUserManager { let mut connections = self.connections.write().await; let now = current_time_secs(); connections.kicked.retain(|_, (expires_at, _)| *expires_at > now); - if let Some(username) = connections.key_by_addr.get(addr).cloned() { + if let Some(username) = connections + .key_by_addr + .get(addr) + .map(|registration| registration.username.clone()) + .filter(|username| !username.is_empty()) + { let expires_at = now + block_for_secs; connections.kicked.insert(username, (expires_at, virtual_id)); } @@ -631,7 +664,12 @@ impl ActiveUserManager { } pub async fn get_username_for_addr(&self, addr: &SocketAddr) -> Option { - self.connections.read().await.key_by_addr.get(addr).cloned() + self.connections + .read() + .await + .key_by_addr + .get(addr) + .map(|registration| registration.username.clone()) } fn gc(&self) { @@ -650,6 +688,9 @@ impl ActiveUserManager { for connection_data in user_connections.by_key.values_mut() { connection_data.sessions.retain(|s| now.saturating_sub(s.ts) < USER_CON_TTL); } + user_connections.key_by_addr.retain(|_, registration| { + !(registration.username.is_empty() && now.saturating_sub(registration.ts) >= ANON_SOCKET_TTL) + }); } else { // Lock contention: release the GC claim so a subsequent caller can retry immediately. let _ = gc_ts.compare_exchange(now, ts, Ordering::AcqRel, Ordering::Relaxed); @@ -736,6 +777,34 @@ mod tests { assert!(manager.release_stream(&addr).await); assert_eq!(manager.user_connections(username).await, 0); } + + #[tokio::test] + async fn stale_anonymous_socket_registration_is_pruned_by_gc() { + 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 stale_addr: SocketAddr = "127.0.0.1:55011".parse().unwrap(); + let fresh_addr: SocketAddr = "127.0.0.1:55012".parse().unwrap(); + + manager.add_connection(&stale_addr).await; + { + let mut connections = manager.connections.write().await; + let registration = connections.key_by_addr.get_mut(&stale_addr).expect("socket registration should exist"); + registration.ts = registration.ts.saturating_sub(ANON_SOCKET_TTL + 1); + } + + if let Some(gc_ts) = &manager.gc_ts { + gc_ts.store(current_time_secs().saturating_sub(USER_GC_TTL + 1), Ordering::Release); + } + + manager.add_connection(&fresh_addr).await; + + let connections = manager.connections.read().await; + assert!(!connections.key_by_addr.contains_key(&stale_addr)); + assert!(connections.key_by_addr.contains_key(&fresh_addr)); + } } // diff --git a/backend/src/api/model/metadata_update_manager.rs b/backend/src/api/model/metadata_update_manager.rs index 038861a74..d3cb65149 100644 --- a/backend/src/api/model/metadata_update_manager.rs +++ b/backend/src/api/model/metadata_update_manager.rs @@ -32,7 +32,7 @@ use std::{ io, path::{Path, PathBuf}, sync::{ - atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, + atomic::{AtomicBool, AtomicI64, AtomicU64, AtomicUsize, Ordering}, Arc, OnceLock, Weak, }, time::{Duration, Instant}, @@ -274,6 +274,16 @@ impl TaskKey { } } +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct ScopedTaskKey { + input_name: Arc, + task_key: TaskKey, +} + +impl ScopedTaskKey { + fn new(input_name: Arc, task_key: TaskKey) -> Self { Self { input_name, task_key } } +} + #[derive(Debug, Clone)] struct RetryState { attempts: u8, @@ -553,6 +563,15 @@ pub struct MetadataUpdateManager { cancel_token: ArcSwap, /// Monotonic worker generation id used to avoid removing a newly spawned worker context. next_worker_id: AtomicU64, + /// Producer-visible view of active resolve cooldowns so repeated playlist refreshes + /// can skip creating tasks that are still suppressed anyway. + resolve_enqueue_suppressions: Arc>, + /// Tracks inputs for which producer-side retry suppression state was already loaded from disk. + enqueue_state_loaded_inputs: Arc, ()>>, + /// Backoff for failed producer-side retry-state loads to avoid repeated blocking disk reads. + enqueue_state_load_retry_at_ts: Arc, i64>>, + /// Periodic prune marker for stale producer-side resolve suppression entries. + last_resolve_enqueue_suppression_prune_at_ts: AtomicI64, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -573,6 +592,10 @@ impl MetadataUpdateManager { update_pause_gate: Arc::new(RwLock::new(())), cancel_token: ArcSwap::from_pointee(cancel_token), next_worker_id: AtomicU64::new(1), + resolve_enqueue_suppressions: Arc::new(DashMap::new()), + enqueue_state_loaded_inputs: Arc::new(DashMap::new()), + enqueue_state_load_retry_at_ts: Arc::new(DashMap::new()), + last_resolve_enqueue_suppression_prune_at_ts: AtomicI64::new(0), } } @@ -582,6 +605,9 @@ impl MetadataUpdateManager { let _guard = self.worker_lifecycle_lock.lock(); self.is_shutdown_flag.store(true, Ordering::Release); self.cancel_token.load_full().cancel(); + self.resolve_enqueue_suppressions.clear(); + self.enqueue_state_loaded_inputs.clear(); + self.enqueue_state_load_retry_at_ts.clear(); } /// Returns true once shutdown has been requested. @@ -600,6 +626,9 @@ impl MetadataUpdateManager { let old_token = self.cancel_token.swap(Arc::new(cancel_token)); self.workers.clear(); + self.resolve_enqueue_suppressions.clear(); + self.enqueue_state_loaded_inputs.clear(); + self.enqueue_state_load_retry_at_ts.clear(); old_token }; old_token.cancel(); @@ -616,6 +645,121 @@ impl MetadataUpdateManager { *guard = Some(app_state); } + fn scoped_task_key(input_name: &str, task: &UpdateTask) -> ScopedTaskKey { + ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(task)) + } + + fn has_pending_task(&self, input_name: &str, task: &UpdateTask) -> bool { + self.workers + .get(input_name) + .is_some_and(|ctx| !ctx.sender.is_closed() && ctx.pending_tasks.contains_key(&TaskKey::from_task(task))) + } + + fn prune_resolve_enqueue_suppressions(&self, now_ts: i64) { + let last_pruned_at = self.last_resolve_enqueue_suppression_prune_at_ts.load(Ordering::Relaxed); + if last_pruned_at != 0 && now_ts.saturating_sub(last_pruned_at) < RETRY_STATE_PRUNE_INTERVAL_SECS { + return; + } + self.last_resolve_enqueue_suppression_prune_at_ts.store(now_ts, Ordering::Relaxed); + self.resolve_enqueue_suppressions + .retain(|_, suppressed_until_ts| *suppressed_until_ts > now_ts); + } + + fn should_skip_enqueue_cached(&self, input_name: &str, task: &UpdateTask) -> bool { + let now_ts = chrono::Utc::now().timestamp(); + self.prune_resolve_enqueue_suppressions(now_ts); + + if self.is_redundant_with_pending_task(input_name, task) { + return true; + } + + if InputWorker::retry_domain_for_task(task) != RetryDomain::Resolve { + return false; + } + + let scoped_key = Self::scoped_task_key(input_name, task); + if let Some(entry) = self.resolve_enqueue_suppressions.get(&scoped_key) { + let suppressed_until_ts = *entry; + drop(entry); + if now_ts < suppressed_until_ts { + if self.has_pending_task(input_name, task) { + return false; + } + debug!( + "[Task] Skipping enqueue (resolve suppression) for input {}: {} (until_ts={}, remaining={}s)", + input_name, + task, + suppressed_until_ts, + suppressed_until_ts.saturating_sub(now_ts) + ); + return true; + } + self.resolve_enqueue_suppressions.remove(&scoped_key); + } + + false + } + + async fn ensure_enqueue_state_loaded_for_input(&self, input_name: &Arc) { + if self.enqueue_state_loaded_inputs.contains_key(input_name) { + return; + } + let now_ts = chrono::Utc::now().timestamp(); + if self + .enqueue_state_load_retry_at_ts + .get(input_name) + .is_some_and(|retry_at_ts| now_ts < *retry_at_ts) + { + return; + } + + let app_state_weak = { + let guard = self.app_state.lock().await; + guard.clone() + }; + let runtime_settings = MetadataUpdateRuntimeSettings::from_app_state(app_state_weak.as_ref()); + let Some(app_state) = app_state_weak.and_then(|weak| Weak::upgrade(&weak)) else { + return; + }; + let retry_at_ts = now_ts.saturating_add(runtime_settings.metadata_retry_load_retry_delay_secs); + + let storage_dir = app_state.app_config.config.load().storage_dir.clone(); + let Ok(storage_path) = get_input_storage_path(input_name, &storage_dir).await else { + self.enqueue_state_load_retry_at_ts.insert(input_name.clone(), retry_at_ts); + return; + }; + let retry_path = storage_path.join(METADATA_RETRY_STATE_FILE); + let input_name_cloned = input_name.clone(); + let loaded = spawn_blocking_limited(move || load_metadata_retry_states_from_disk(&retry_path)) + .await + .ok() + .and_then(Result::ok); + let Some(states) = loaded else { + self.enqueue_state_load_retry_at_ts.insert(input_name.clone(), retry_at_ts); + return; + }; + + for (task_key, state) in states { + if let Some(resolve_state) = state.resolve.as_ref() { + let suppressed_until_ts = resolve_state + .cooldown_until_ts + .unwrap_or(resolve_state.next_allowed_at_ts); + if suppressed_until_ts > now_ts { + let scoped_key = ScopedTaskKey::new(input_name_cloned.clone(), task_key); + self.resolve_enqueue_suppressions.insert(scoped_key, suppressed_until_ts); + } + } + } + + self.enqueue_state_load_retry_at_ts.remove(input_name); + self.enqueue_state_loaded_inputs.insert(input_name.clone(), ()); + } + + pub async fn should_skip_enqueue(&self, input_name: Arc, task: &UpdateTask) -> bool { + self.ensure_enqueue_state_loaded_for_input(&input_name).await; + self.should_skip_enqueue_cached(input_name.as_ref(), task) + } + /// Spawn a background task to queue the update. /// This is a fire-and-forget method that returns immediately. pub fn queue_task_background(self: &Arc, input_name: Arc, task: UpdateTask) { @@ -698,6 +842,7 @@ impl MetadataUpdateManager { last_retry_state_prune_at_ts: None, scheduled_requeues: Arc::new(DashMap::new()), recently_completed_no_change: HashMap::new(), + resolve_enqueue_suppressions: Arc::clone(&self.resolve_enqueue_suppressions), }, )); @@ -1000,12 +1145,14 @@ struct InputWorker { // Prevents repeated resolution of already-resolved items across playlist refreshes. // Stores the reason set so that tasks with new/different reasons are not wrongly skipped. recently_completed_no_change: HashMap, + resolve_enqueue_suppressions: Arc>, } #[derive(Debug, Clone, Copy)] struct ProcessTaskOutcome { task_changed: bool, tmdb_pending: bool, + probe_pending: bool, } impl InputWorker { @@ -1164,22 +1311,26 @@ impl InputWorker { if !skip_execution { if let Some(state_bundle) = self.retry_states.get(¤t_key) { if let Some(active_state) = state_bundle.get(active_retry_domain) { - if active_retry_domain == RetryDomain::Probe { - if let Some(cooldown_until_ts) = active_state.cooldown_until_ts { - if now_ts < cooldown_until_ts { - debug!( - "[Metadata-Task] Skipping task (probe cooldown) for input {}: {} (cooldown_until={}, remaining={}s)", - input_name, - task_for_execution, - cooldown_until_ts, - cooldown_until_ts.saturating_sub(now_ts) - ); - self.scheduled_requeues.remove(¤t_key); - remove_current_task = true; - skip_execution = true; + if let Some(cooldown_until_ts) = active_state.cooldown_until_ts { + if now_ts < cooldown_until_ts { + let cooldown_label = if active_retry_domain == RetryDomain::Probe { + "probe" } else { - clear_active_retry_state = true; - } + "resolve" + }; + debug!( + "[Metadata-Task] Skipping task ({} cooldown) for input {}: {} (cooldown_until={}, remaining={}s)", + cooldown_label, + input_name, + task_for_execution, + cooldown_until_ts, + cooldown_until_ts.saturating_sub(now_ts) + ); + self.scheduled_requeues.remove(¤t_key); + remove_current_task = true; + skip_execution = true; + } else { + clear_active_retry_state = true; } } @@ -1221,6 +1372,9 @@ impl InputWorker { if should_remove_retry_entry { self.retry_states.remove(¤t_key); } + if clear_active_retry_state && active_retry_domain == RetryDomain::Resolve { + self.clear_resolve_enqueue_suppression(¤t_key); + } metadata_persist_state = Some(state_after_clear); } } @@ -1258,10 +1412,11 @@ impl InputWorker { Self::should_trigger_playlist_update_for_task(&task_for_execution, task_outcome.task_changed); cycle_had_changes |= trigger_playlist_update; debug!( - "[Metadata-Task] Task succeeded for input {input_name}: {task_for_execution} (changed={}, trigger_playlist_update={}, tmdb_pending={})", + "[Metadata-Task] Task succeeded for input {input_name}: {task_for_execution} (changed={}, trigger_playlist_update={}, tmdb_pending={}, probe_pending={})", task_outcome.task_changed, trigger_playlist_update, - task_outcome.tmdb_pending + task_outcome.tmdb_pending, + task_outcome.probe_pending ); let active_retry_domain = Self::retry_domain_for_task(&task_for_execution); @@ -1330,12 +1485,16 @@ impl InputWorker { } self.resolve_exhausted.remove(¤t_key); + if active_retry_domain == RetryDomain::Resolve { + self.clear_resolve_enqueue_suppression(¤t_key); + } self.scheduled_requeues.remove(¤t_key); // Cache tasks that completed with no changes to skip redundant re-resolution. if !task_outcome.task_changed && Self::is_resolve_task(&task_for_execution) && !task_outcome.tmdb_pending + && !task_outcome.probe_pending { let reasons = Self::task_reason(&task_for_execution); debug!( @@ -1400,27 +1559,28 @@ impl InputWorker { runtime_settings.probe_cooldown_secs ); } else { + let cooldown_until_ts = + now_ts.saturating_add(runtime_settings.resolve_exhaustion_reset_gap_secs); + let state_bundle_after_update = { + let state_bundle = self.retry_states.entry(current_key.clone()).or_default(); + let state = state_bundle.get_mut_or_insert(RetryDomain::Resolve); + state.attempts = runtime_settings.max_attempts_resolve; + state.next_allowed_at_ts = cooldown_until_ts; + state.cooldown_until_ts = Some(cooldown_until_ts); + state.last_error = Some(e.message.clone()); + state_bundle.touch(now_ts); + state_bundle.clone() + }; + self.set_resolve_enqueue_suppression(¤t_key, cooldown_until_ts); self.resolve_exhausted.insert(current_key.clone(), now_ts); - let mut should_remove_retry_entry = false; - let mut state_after_clear: Option = None; - if let Some(state_bundle) = self.retry_states.get_mut(¤t_key) { - state_bundle.clear_domain(retry_domain); - if state_bundle.is_empty() { - should_remove_retry_entry = true; - } else { - state_bundle.touch(now_ts); - state_after_clear = Some(state_bundle.clone()); - } - } - if should_remove_retry_entry { - self.retry_states.remove(¤t_key); - } - metadata_persist_state = Some(state_after_clear); + metadata_persist_state = Some(Some(state_bundle_after_update)); remove_current_task = true; debug!( - "[Metadata-Task] Resolve task marked exhausted after permanent not-found for input {}: {} (error={})", + "[Metadata-Task] Resolve task entering cooldown after permanent not-found for input {}: {} (cooldown_until={}, cooldown_duration={}s, error={})", input_name, task_for_execution, + cooldown_until_ts, + runtime_settings.resolve_exhaustion_reset_gap_secs, e.message ); } @@ -1501,35 +1661,41 @@ impl InputWorker { e.message ); } else { + let cooldown_until_ts = + now_ts.saturating_add(runtime_settings.resolve_exhaustion_reset_gap_secs); + let state_bundle_after_update = { + let state_bundle = self.retry_states.entry(current_key.clone()).or_default(); + let state = state_bundle.get_mut_or_insert(RetryDomain::Resolve); + state.attempts = max_attempts; + state.next_allowed_at_ts = cooldown_until_ts; + state.cooldown_until_ts = Some(cooldown_until_ts); + state.last_error = Some(e.message.clone()); + state_bundle.touch(now_ts); + state_bundle.clone() + }; + self.set_resolve_enqueue_suppression(¤t_key, cooldown_until_ts); self.resolve_exhausted.insert(current_key.clone(), now_ts); - let mut should_remove_retry_entry = false; - let mut state_after_clear: Option = None; - if let Some(state_bundle) = self.retry_states.get_mut(¤t_key) { - state_bundle.clear_domain(retry_domain); - if state_bundle.is_empty() { - should_remove_retry_entry = true; - } else { - state_bundle.touch(now_ts); - state_after_clear = Some(state_bundle.clone()); - } - } - if should_remove_retry_entry { - self.retry_states.remove(¤t_key); - } - metadata_persist_state = Some(state_after_clear); + metadata_persist_state = Some(Some(state_bundle_after_update)); remove_current_task = true; debug!( - "[Metadata-Task] Resolve task exhausted (max attempts reached) for input {}: {} (attempts={}/{}, error={})", + "[Metadata-Task] Resolve task exhausted (max attempts reached) for input {}: {} (attempts={}/{}, cooldown_duration={}s, error={})", input_name, task_for_execution, attempts, max_attempts, + runtime_settings.resolve_exhaustion_reset_gap_secs, e.message ); } } else { schedule_requeue_at_ts = Some(state_after_update.next_allowed_at_ts); metadata_persist_state = Some(Some(state_bundle_after_update)); + if retry_domain == RetryDomain::Resolve { + self.set_resolve_enqueue_suppression( + ¤t_key, + state_after_update.next_allowed_at_ts, + ); + } debug!( "[Metadata-Task] Task failed, scheduling retry for input {}: {} (attempt={}/{}, next_allowed_at={}, backoff={}s, error={})", input_name, @@ -1708,6 +1874,7 @@ impl InputWorker { // (for example by the next playlist update), because state alone does not contain the // full `UpdateTask` payload for all variants. for (key, state) in loaded { + self.sync_resolve_enqueue_suppression_from_retry_state(&key, &state, now_ts); self.retry_states.insert(key, state); } self.prune_retry_tracking_maps(now_ts, runtime_settings).await; @@ -1977,6 +2144,7 @@ impl InputWorker { for key in stale_retry_keys { if self.retry_states.remove(&key).is_some() { + self.clear_resolve_enqueue_suppression(&key); debug_if_enabled!("Pruned stale metadata retry state for input {}: {:?}", self.input_name, key); self.persist_metadata_retry_state(&key, None).await; } @@ -1995,6 +2163,7 @@ impl InputWorker { .collect(); for key in stale_resolve_exhausted_keys { self.resolve_exhausted.remove(&key); + self.clear_resolve_enqueue_suppression(&key); } let no_change_ttl = Duration::from_secs(runtime_settings.no_change_cache_ttl_secs); @@ -2138,6 +2307,32 @@ impl InputWorker { false } + fn set_resolve_enqueue_suppression(&self, current_key: &TaskKey, until_ts: i64) { + let scoped_key = ScopedTaskKey::new(self.input_name.clone(), current_key.clone()); + self.resolve_enqueue_suppressions.insert(scoped_key, until_ts); + } + + fn clear_resolve_enqueue_suppression(&self, current_key: &TaskKey) { + let scoped_key = ScopedTaskKey::new(self.input_name.clone(), current_key.clone()); + self.resolve_enqueue_suppressions.remove(&scoped_key); + } + + fn sync_resolve_enqueue_suppression_from_retry_state(&self, current_key: &TaskKey, state: &TaskRetryState, now_ts: i64) { + let Some(resolve_state) = state.resolve.as_ref() else { + self.clear_resolve_enqueue_suppression(current_key); + return; + }; + + let suppressed_until_ts = resolve_state + .cooldown_until_ts + .unwrap_or(resolve_state.next_allowed_at_ts); + if suppressed_until_ts > now_ts { + self.set_resolve_enqueue_suppression(current_key, suppressed_until_ts); + } else { + self.clear_resolve_enqueue_suppression(current_key); + } + } + #[inline] fn should_trigger_playlist_update_for_task(task: &UpdateTask, task_changed: bool) -> bool { task_changed && !Self::is_probe_task(task) && !Self::is_probe_only_resolve_task(task) @@ -2965,7 +3160,7 @@ impl InputWorker { app_state.connection_manager.release_provider_handle(provider_handle).await; } match res { - Ok(tmdb_and_date_present) => { + Ok((tmdb_and_date_present, probe_pending)) => { let task_changed = match task { UpdateTask::ResolveVod { .. } => collector.vod.len() > pre_vod_updates, UpdateTask::ResolveSeries { .. } => collector.series.len() > pre_series_updates, @@ -2982,7 +3177,7 @@ impl InputWorker { } UpdateTask::ProbeLive { .. } | UpdateTask::ProbeStream { .. } => false, }; - Ok(ProcessTaskOutcome { task_changed, tmdb_pending }) + Ok(ProcessTaskOutcome { task_changed, tmdb_pending, probe_pending }) } Err(e) => Err(e), } @@ -3012,9 +3207,10 @@ impl InputWorker { collector: &mut BatchResultCollector, db_handles: &mut HashMap, failed_clusters: &mut HashSet, - ) -> Result { - // The returned bool indicates whether both TMDB id and release date are already present in the DB - // (used by the caller to avoid false-positive "no match" cooldowns). + ) -> Result<(bool, bool), TuliproxError> { + // The returned tuple is `(tmdb_and_date_present, probe_pending)`. + // `tmdb_and_date_present` avoids false-positive TMDB "no match" cooldowns. + // `probe_pending` prevents skipped/aborted probes from being cached as no-op. match task { UpdateTask::ResolveVod { id, reason, .. } => { let fetch_info = reason.contains(ResolveReason::Info); @@ -3037,6 +3233,7 @@ impl InputWorker { }; let tmdb_and_date_present = AtomicBool::new(false); + let probe_pending = AtomicBool::new(false); match update_vod_metadata( &app_state.app_config, client, @@ -3051,14 +3248,21 @@ impl InputWorker { will_probe, query_opt, Some(&tmdb_and_date_present), + Some(&probe_pending), ) .await { Ok(Some(props)) => { collector.add_vod(id.clone(), props); - Ok(tmdb_and_date_present.load(Ordering::Relaxed)) + Ok(( + tmdb_and_date_present.load(Ordering::Relaxed), + probe_pending.load(Ordering::Relaxed), + )) } - Ok(None) => Ok(tmdb_and_date_present.load(Ordering::Relaxed)), + Ok(None) => Ok(( + tmdb_and_date_present.load(Ordering::Relaxed), + probe_pending.load(Ordering::Relaxed), + )), Err(e) => Err(e), } } @@ -3084,6 +3288,7 @@ impl InputWorker { }; let tmdb_and_date_present = AtomicBool::new(false); + let probe_pending = AtomicBool::new(false); match update_series_metadata( &app_state.app_config, client, @@ -3099,14 +3304,21 @@ impl InputWorker { series_probe_settings, query_opt, Some(&tmdb_and_date_present), + Some(&probe_pending), ) .await { Ok(Some(props)) => { collector.add_series(id.clone(), props); - Ok(tmdb_and_date_present.load(Ordering::Relaxed)) + Ok(( + tmdb_and_date_present.load(Ordering::Relaxed), + probe_pending.load(Ordering::Relaxed), + )) } - Ok(None) => Ok(tmdb_and_date_present.load(Ordering::Relaxed)), + Ok(None) => Ok(( + tmdb_and_date_present.load(Ordering::Relaxed), + probe_pending.load(Ordering::Relaxed), + )), Err(e) => Err(e), } } @@ -3127,9 +3339,9 @@ impl InputWorker { { Ok(Some(props)) => { collector.add_live(id.clone(), props); - Ok(false) + Ok((false, false)) } - Ok(None) => Ok(false), + Ok(None) => Ok((false, false)), Err(e) => Err(e), } } @@ -3155,7 +3367,7 @@ impl InputWorker { .await?; match outcome { - GenericProbeOutcome::Updated | GenericProbeOutcome::Noop => Ok(false), + GenericProbeOutcome::Updated | GenericProbeOutcome::Noop => Ok((false, false)), GenericProbeOutcome::ProbeFailed => Err(shared::error::info_err!( "Probe stream task failed for key {:?} ({})", task_key, @@ -3201,6 +3413,7 @@ mod tests { last_retry_state_prune_at_ts: None, scheduled_requeues: Arc::new(DashMap::new()), recently_completed_no_change: HashMap::new(), + resolve_enqueue_suppressions: Arc::new(DashMap::new()), } } @@ -3239,6 +3452,30 @@ mod tests { manager.shutdown(); } + #[test] + fn should_skip_enqueue_respects_resolve_suppression() { + let manager = MetadataUpdateManager::new(CancellationToken::new()); + let input_name = "input_suppressed"; + let task = UpdateTask::ResolveVod { + id: ProviderIdType::Id(42), + reason: ResolveReasonSet::from_variants(&[ResolveReason::Info]), + delay: 1, + }; + let scoped_key = ScopedTaskKey::new(Arc::from(input_name), TaskKey::from_task(&task)); + manager + .resolve_enqueue_suppressions + .insert(scoped_key.clone(), chrono::Utc::now().timestamp().saturating_add(60)); + + assert!(manager.should_skip_enqueue_cached(input_name, &task)); + + manager + .resolve_enqueue_suppressions + .insert(scoped_key.clone(), chrono::Utc::now().timestamp().saturating_sub(1)); + + assert!(!manager.should_skip_enqueue_cached(input_name, &task)); + assert!(!manager.resolve_enqueue_suppressions.contains_key(&scoped_key)); + } + #[tokio::test] async fn submit_task_merges_existing_task_and_increments_generation() { let (tx, mut rx) = mpsc::channel::(8); @@ -3652,7 +3889,12 @@ mod tests { let path = dir.path().join("metadata_retry_state.db"); let key = TaskKey::Stream { scope: Arc::from("input_a"), id: Arc::from("stream_1") }; let state = TaskRetryState { - resolve: None, + resolve: Some(RetryState { + attempts: 4, + next_allowed_at_ts: 1_700_043_200, + cooldown_until_ts: Some(1_700_043_200), + last_error: Some("resolve exhausted".to_string()), + }), probe: Some(RetryState { attempts: 3, next_allowed_at_ts: 1_700_000_000, @@ -3672,6 +3914,10 @@ mod tests { let loaded = load_metadata_retry_states_from_disk(&path).expect("state load should succeed"); let loaded_state = loaded.get(&key).expect("probe key should be present"); + let loaded_resolve = loaded_state.resolve.as_ref().expect("resolve retry state should be present"); + assert_eq!(loaded_resolve.attempts, 4); + assert_eq!(loaded_resolve.cooldown_until_ts, Some(1_700_043_200)); + assert_eq!(loaded_resolve.last_error.as_deref(), Some("resolve exhausted")); let loaded_probe = loaded_state.probe.as_ref().expect("probe retry state should be present"); assert_eq!(loaded_probe.attempts, 3); assert_eq!(loaded_probe.cooldown_until_ts, Some(1_700_086_400)); diff --git a/backend/src/api/model/provider_lineup_manager.rs b/backend/src/api/model/provider_lineup_manager.rs index aac3419a3..5e561c0da 100644 --- a/backend/src/api/model/provider_lineup_manager.rs +++ b/backend/src/api/model/provider_lineup_manager.rs @@ -160,6 +160,13 @@ impl fmt::Display for ProviderLineup { } impl ProviderLineup { + fn provider_names(&self) -> Vec> { + match self { + ProviderLineup::Single(lineup) => vec![lineup.provider.name.clone()], + ProviderLineup::Multi(lineup) => lineup.provider_names(), + } + } + async fn get_next(&self, grace_period_timeout_secs: u64) -> Option> { match self { ProviderLineup::Single(lineup) => lineup.get_next(grace_period_timeout_secs).await, @@ -283,6 +290,19 @@ impl MultiProviderLineup { Self { name: cfg_input.name.clone(), providers } } + fn provider_names(&self) -> Vec> { + let mut names = Vec::new(); + for group in &self.providers { + match group { + ProviderPriorityGroup::SingleProviderGroup(provider) => names.push(provider.name.clone()), + ProviderPriorityGroup::MultiProviderGroup(_, providers) => { + names.extend(providers.iter().map(|provider| provider.name.clone())); + } + } + } + names + } + /// Attempts to acquire the next available provider from a specific priority group. /// /// # Parameters @@ -777,11 +797,6 @@ impl ProviderLineupManager { allocation } - // Returns the next available provider connection - pub(crate) async fn acquire_connection(&self, input_name: &Arc) -> ProviderAllocation { - self.acquire_connection_with_grace_override(input_name, true).await - } - /// Acquire a provider connection, optionally allowing provider-side grace allocations. /// /// When `allow_grace` is `false`, the lineup will not allocate providers in `GracePeriod`, @@ -867,6 +882,12 @@ impl ProviderLineupManager { } } + pub fn provider_names_for_input(&self, input_name: &Arc) -> Vec> { + let snapshot = self.snapshot.load_full(); + Self::get_provider_config_by_name(input_name, &snapshot.providers) + .map_or_else(Vec::new, |(lineup, _)| lineup.provider_names()) + } + pub async fn active_connections(&self) -> Option, usize>> { let mut result = HashMap::, usize>::new(); let mut add_provider = |name: &Arc, count: usize| { diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index eb10e10b2..1ec4f20ca 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/backend/src/api/model/streams/active_client_stream.rs @@ -13,7 +13,7 @@ use crate::{ use axum::http::{header::USER_AGENT, HeaderMap}; use bytes::Bytes; use futures::{task::AtomicWaker, Future, Stream, StreamExt}; -use log::{debug, error, info}; +use log::{error, info, trace}; use shared::{ model::{StreamChannel, UserConnectionPermission, VirtualId}, utils::sanitize_sensitive_info, @@ -661,7 +661,7 @@ fn stream_grace_period( None }; - debug!("grace hold stream {hold_stream}"); + trace!("grace hold stream {hold_stream}"); if provider_grace_check.is_some() || user_grace_check.is_some() { let stream_strategy_flag = Arc::new(AtomicU8::new( diff --git a/backend/src/model/config/stream.rs b/backend/src/model/config/stream.rs index af4a366d4..1a8afed4d 100644 --- a/backend/src/model/config/stream.rs +++ b/backend/src/model/config/stream.rs @@ -35,6 +35,8 @@ pub struct StreamConfig { pub grace_period_millis: u64, pub grace_period_timeout_secs: u64, pub grace_period_hold_stream: bool, + pub hls_session_ttl_secs: u64, + pub catchup_session_ttl_secs: u64, pub throttle_str: Option, pub throttle_kbps: u64, pub shared_burst_buffer_mb: u64, @@ -48,10 +50,12 @@ impl From<&StreamConfigDto> for StreamConfig { buffer: dto.buffer.as_ref().map(Into::into), grace_period_millis: dto.grace_period_millis, grace_period_timeout_secs: dto.grace_period_timeout_secs, + grace_period_hold_stream: dto.grace_period_hold_stream, + hls_session_ttl_secs: dto.hls_session_ttl_secs, + catchup_session_ttl_secs: dto.catchup_session_ttl_secs, throttle_str: dto.throttle.clone(), throttle_kbps: dto.throttle.as_ref().map_or(0u64, |throttle| parse_to_kbps(throttle).unwrap_or(0u64)), shared_burst_buffer_mb: dto.shared_burst_buffer_mb, - grace_period_hold_stream: dto.grace_period_hold_stream, } } } @@ -63,10 +67,12 @@ impl From<&StreamConfig> for StreamConfigDto { buffer: instance.buffer.as_ref().map(Into::into), grace_period_millis: instance.grace_period_millis, grace_period_timeout_secs: instance.grace_period_timeout_secs, + grace_period_hold_stream: instance.grace_period_hold_stream, + hls_session_ttl_secs: instance.hls_session_ttl_secs, + catchup_session_ttl_secs: instance.catchup_session_ttl_secs, throttle: instance.throttle_str.clone(), throttle_kbps: instance.throttle_kbps, shared_burst_buffer_mb: instance.shared_burst_buffer_mb, - grace_period_hold_stream: instance.grace_period_hold_stream, } } } diff --git a/backend/src/processing/processor/library.rs b/backend/src/processing/processor/library.rs index ac665b2c5..89ce7b903 100644 --- a/backend/src/processing/processor/library.rs +++ b/backend/src/processing/processor/library.rs @@ -3,7 +3,7 @@ use crate::model::{AppConfig, ConfigInput}; use shared::concat_string; use shared::error::TuliproxError; use shared::model::UUIDType; -use shared::model::{EpisodeStreamProperties, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, SeriesStreamDetailSeasonProperties, SeriesStreamProperties, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster}; +use shared::model::{EpisodeStreamProperties, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, SeriesStreamDetailSeasonProperties, SeriesStreamProperties, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster, normalize_episode_title}; use shared::utils::{generate_playlist_uuid, Internable}; use std::collections::HashMap; use std::path::Path; @@ -263,6 +263,8 @@ pub fn metadata_cache_entry_to_xtream_series_info( .unwrap_or_default(); let episode_release_date = episode.aired.as_ref().map(ToString::to_string).unwrap_or_default(); let tmdb_id = (episode.tmdb_id > 0).then_some(episode.tmdb_id); + let raw_episode_title: Arc = episode.title.clone().into(); + let series_title: Arc = series.title.clone().into(); let season_entry = season_data.entry(episode.season).or_insert_with(|| { SeriesStreamDetailSeasonProperties { @@ -283,7 +285,7 @@ pub fn metadata_cache_entry_to_xtream_series_info( id: tmdb_id.unwrap_or_default(), episode_num: episode.episode, season: episode.season, - title: episode.title.clone().into(), + title: normalize_episode_title(&raw_episode_title, &series_title, episode.season, episode.episode), container_extension: container_extension.into(), custom_sid: None, added: episode.file_modified.to_string().into(), @@ -344,4 +346,4 @@ pub fn metadata_cache_entry_to_xtream_series_info( }; Some(StreamProperties::Series(Box::new(properties))) -} \ No newline at end of file +} diff --git a/backend/src/processing/processor/mod.rs b/backend/src/processing/processor/mod.rs index c16739491..0c4c039ad 100644 --- a/backend/src/processing/processor/mod.rs +++ b/backend/src/processing/processor/mod.rs @@ -9,43 +9,19 @@ mod trakt; mod library; mod stream_probe; mod filtered_playlist_source; +mod probe_handle_guard; +mod resolve_options; pub use self::playlist::*; pub use self::epg::*; pub use self::xtream::*; pub use self::xtream_vod::*; pub use self::xtream_series::*; pub use self::stream_probe::*; +pub(crate) use self::probe_handle_guard::*; +pub use self::resolve_options::*; use crate::api::model::ProviderHandle; -use shared::create_bitset; use tokio_util::sync::CancellationToken; -create_bitset!(u8, ResolveOptionsFlags, Resolve, TmdbMissing, Probe, Background); - -pub struct ResolveOptions { - flags: ResolveOptionsFlagsSet, - pub resolve_delay: u16, -} - -impl ResolveOptions { - #[inline] - pub fn has_flag(&self, flag: ResolveOptionsFlags) -> bool { - self.flags.contains(flag) - } - #[inline] - pub fn unset_flag(&mut self, flag: ResolveOptionsFlags) { - self.flags.unset(flag); - } -} - -impl Default for ResolveOptions { - fn default() -> Self { - Self { - flags: ResolveOptionsFlagsSet::from_variants(&[ResolveOptionsFlags::Background]), - resolve_delay: default_resolve_delay_secs(), - } - } -} - pub(crate) const FOREGROUND_BATCH_SIZE: usize = 200; pub(crate) const FOREGROUND_RETRY_BATCH_MAX_SIZE: usize = FOREGROUND_BATCH_SIZE * 4; pub(crate) const FOREGROUND_MIN_RETRY_DELAY_SECS: u64 = 1; @@ -293,4 +269,3 @@ macro_rules! process_foreground_retry_once { }; } pub(crate) use process_foreground_retry_once; -use shared::utils::default_resolve_delay_secs; diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 1503483c4..ac6057210 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -1180,7 +1180,7 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, delay: probe_delay, interval: interval_secs, }; - if mgr.is_redundant_with_pending_task(input_name.as_ref(), &task) { + if mgr.should_skip_enqueue(input_name.clone(), &task).await { continue; } if log_enabled!(Level::Debug) { @@ -1249,7 +1249,7 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, reason: ResolveReasonSet::from_variants(&[ResolveReason::MissingDetails]), delay: opts.probe_delay, }; - if mgr.is_redundant_with_pending_task(input_name.as_ref(), &task) { + if mgr.should_skip_enqueue(input_name.clone(), &task).await { continue; } debug!( diff --git a/backend/src/processing/processor/probe_handle_guard.rs b/backend/src/processing/processor/probe_handle_guard.rs new file mode 100644 index 000000000..712675f27 --- /dev/null +++ b/backend/src/processing/processor/probe_handle_guard.rs @@ -0,0 +1,120 @@ +use crate::api::model::{ActiveProviderManager, ProviderHandle}; +use std::sync::Arc; + +pub(crate) struct ProbeHandleGuard { + manager: Arc, + handle: Option, +} + +impl ProbeHandleGuard { + pub(crate) fn new(manager: &Arc, handle: ProviderHandle) -> Self { + Self { + manager: Arc::clone(manager), + handle: Some(handle), + } + } + + #[inline] + pub(crate) fn handle(&self) -> Option<&ProviderHandle> { self.handle.as_ref() } + + pub(crate) async fn release(mut self) { + if let Some(handle) = self.handle.take() { + self.manager.release_handle(&handle).await; + } + } +} + +impl Drop for ProbeHandleGuard { + fn drop(&mut self) { + let Some(handle) = self.handle.take() else { + return; + }; + let manager = Arc::clone(&self.manager); + if let Ok(runtime_handle) = tokio::runtime::Handle::try_current() { + runtime_handle.spawn(async move { + manager.release_handle(&handle).await; + }); + } + } +} + +#[cfg(test)] +mod tests { + use super::ProbeHandleGuard; + use crate::{ + api::model::{ActiveProviderManager, EventManager}, + model::{AppConfig, Config, ConfigInput, SourcesConfig}, + utils::FileLockManager, + }; + use arc_swap::{ArcSwap, ArcSwapOption}; + use shared::{ + model::{ConfigPaths, InputFetchMethod, InputType}, + utils::{default_probe_user_priority, Internable}, + }; + use std::{collections::HashMap, sync::Arc}; + + fn create_test_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(), + ..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], + ffprobe_available: Arc::default(), + } + } + + #[tokio::test] + async fn probe_handle_guard_releases_provider_slot_on_drop() { + let app_cfg = create_test_app_config(); + let event_manager = Arc::new(EventManager::new()); + let manager = Arc::new(ActiveProviderManager::new(&app_cfg, &event_manager)); + let input_name = "provider_1".intern(); + + let handle = manager + .acquire_connection_for_probe(&input_name, default_probe_user_priority()) + .await + .expect("probe allocation should succeed"); + assert_eq!(manager.get_provider_connections_count().await, 1); + + let guard = ProbeHandleGuard::new(&manager, handle); + drop(guard); + tokio::task::yield_now().await; + tokio::task::yield_now().await; + + assert_eq!(manager.get_provider_connections_count().await, 0); + } +} diff --git a/backend/src/processing/processor/resolve_options.rs b/backend/src/processing/processor/resolve_options.rs new file mode 100644 index 000000000..72657e130 --- /dev/null +++ b/backend/src/processing/processor/resolve_options.rs @@ -0,0 +1,29 @@ +use shared::create_bitset; + +create_bitset!(u8, ResolveOptionsFlags, Resolve, TmdbMissing, Probe, Background); + +pub struct ResolveOptions { + pub(crate) flags: ResolveOptionsFlagsSet, + pub resolve_delay: u16, +} + +impl ResolveOptions { + #[inline] + pub fn has_flag(&self, flag: ResolveOptionsFlags) -> bool { + self.flags.contains(flag) + } + + #[inline] + pub fn unset_flag(&mut self, flag: ResolveOptionsFlags) { + self.flags.unset(flag); + } +} + +impl Default for ResolveOptions { + fn default() -> Self { + Self { + flags: ResolveOptionsFlagsSet::from_variants(&[ResolveOptionsFlags::Background]), + resolve_delay: shared::utils::default_resolve_delay_secs(), + } + } +} diff --git a/backend/src/processing/processor/stream_probe.rs b/backend/src/processing/processor/stream_probe.rs index 767a82fec..fe2fe3b15 100644 --- a/backend/src/processing/processor/stream_probe.rs +++ b/backend/src/processing/processor/stream_probe.rs @@ -1,7 +1,7 @@ use crate::api::model::ActiveProviderManager; use crate::model::ConfigInput; use crate::model::{AppConfig}; -use crate::processing::processor::select_cancel_token; +use crate::processing::processor::{select_cancel_token, ProbeHandleGuard}; use crate::repository::{get_input_m3u_playlist_file_path, get_input_storage_path, get_input_local_library_playlist_file_path, xtream_get_file_path, BPlusTreeUpdate}; use crate::utils::{debug_if_enabled, ffmpeg}; use crate::utils::ffmpeg::{ProbeFailureKind, ProbeUrlOutcome}; @@ -90,7 +90,10 @@ pub async fn update_generic_stream_metadata( let acquired_handle = if !needs_provider_connection || active_handle.is_some() { None } else { - active_provider.acquire_connection_for_probe(&input.name, probe_priority).await + active_provider + .acquire_connection_for_probe(&input.name, probe_priority) + .await + .map(|handle| ProbeHandleGuard::new(active_provider, handle)) }; if needs_provider_connection && active_handle.is_none() && acquired_handle.is_none() { @@ -117,7 +120,10 @@ pub async fn update_generic_stream_metadata( debug_if_enabled!("Probing Generic Stream '{unique_id}'"); - let cancel_token = select_cancel_token(acquired_handle.as_ref(), active_handle); + let cancel_token = select_cancel_token( + acquired_handle.as_ref().and_then(ProbeHandleGuard::handle), + active_handle, + ); let probe_data = ffmpeg::probe_url_with_cancel( &probe_url, user_agent.as_deref(), @@ -130,7 +136,7 @@ pub async fn update_generic_stream_metadata( .await; if let Some(handle) = acquired_handle { - active_provider.release_handle(&handle).await; + handle.release().await; } let (raw_video, raw_audio) = match probe_data { diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index 21f1bf8ad..9c6a42a76 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -9,6 +9,7 @@ use crate::processing::parser::xtream::parse_xtream_series_info; use crate::processing::processor::playlist::{PlaylistProcessingContext, ProcessingPipe}; use crate::processing::processor::{ create_resolve_options_function_for_xtream_target, process_foreground_retry_once, select_cancel_token, + ProbeHandleGuard, ResolveOptions, ResolveOptionsFlags, FOREGROUND_BATCH_SIZE as BATCH_SIZE, FOREGROUND_MIN_RETRY_DELAY_SECS, FOREGROUND_RETRY_BATCH_MAX_SIZE as RETRY_BATCH_MAX_SIZE, }; @@ -111,7 +112,7 @@ async fn playlist_resolve_series_info( let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb); let groups_to_add = if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() { - queue_background_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled) + queue_background_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled).await } else { process_immediate_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled).await }; @@ -177,7 +178,7 @@ fn sync_resolved_series_properties(provider_fpl: &mut FetchedPlaylist<'_>, proce } } -fn queue_background_series_info( +async fn queue_background_series_info( ctx: &PlaylistProcessingContext, fpl: &mut FetchedPlaylist<'_>, filter: impl Fn(&PlaylistItem) -> bool, @@ -213,7 +214,7 @@ fn queue_background_series_info( reason: reasons, delay: resolve_options.resolve_delay, }; - if !mgr.is_redundant_with_pending_task(input_name_arc.as_ref(), &task) { + if !mgr.should_skip_enqueue(input_name_arc.clone(), &task).await { if log_enabled!(Level::Debug) { let has_details = pli.has_details(); let (has_tmdb, has_date) = match pli.header.additional_properties.as_ref() { @@ -542,6 +543,7 @@ async fn update_series_info_immediate( probe_settings, db_query, None, + None, ) .await } @@ -639,6 +641,7 @@ pub async fn update_series_metadata( probe_settings: SeriesProbeSettings, db_query: Option>>>, tmdb_and_date_present_out: Option<&AtomicBool>, + probe_pending_out: Option<&AtomicBool>, ) -> Result, TuliproxError> { let storage_dir = &app_config.config.load().storage_dir; let storage_path = get_input_storage_path(&input.name, storage_dir) @@ -716,6 +719,7 @@ pub async fn update_series_metadata( let mut fetched_new = false; let mut properties_updated = false; let mut probe_failure: Option = None; + let mut probe_pending = false; let display_id = series_id_opt.map_or_else(|| "StringID".to_string(), |v| v.to_string()); @@ -870,7 +874,10 @@ pub async fn update_series_metadata( let temp_handle = if active_handle.is_some() { None } else { - active_provider.acquire_connection_for_probe(&input.name, probe_priority).await + active_provider + .acquire_connection_for_probe(&input.name, probe_priority) + .await + .map(|handle| ProbeHandleGuard::new(active_provider, handle)) }; if active_handle.is_some() || temp_handle.is_some() { @@ -890,7 +897,10 @@ pub async fn update_series_metadata( ep.title, ep.season, ep.episode_num, missing_reason ); - let cancel_token = select_cancel_token(temp_handle.as_ref(), active_handle); + let cancel_token = select_cancel_token( + temp_handle.as_ref().and_then(ProbeHandleGuard::handle), + active_handle, + ); match crate::utils::ffmpeg::probe_url_with_cancel( &episode_url, user_agent.as_deref(), @@ -917,23 +927,26 @@ pub async fn update_series_metadata( probe_failure = Some(ProbeFailureKind::NotFound); } ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => { + probe_pending = true; if probe_failure.is_none() { probe_failure = Some(ProbeFailureKind::Other); } } ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => { + probe_pending = true; if probe_failure.is_none() { probe_failure = Some(ProbeFailureKind::Cancelled); } } } - - if let Some(h) = temp_handle { - active_provider.release_handle(&h).await; - } } else { + probe_pending = true; warn!("Skipping probe for series episode {} due to connection limits", ep.title); } + + if let Some(guard) = temp_handle { + guard.release().await; + } } } if probed_count > 0 { @@ -949,6 +962,10 @@ pub async fn update_series_metadata( } } + if let Some(out) = probe_pending_out { + out.store(probe_pending, Ordering::Relaxed); + } + let probe_only_unresolved = do_probe && !fetch_info && !resolve_tmdb && !properties_updated && !fetched_new; if probe_only_unresolved { if let Some(kind) = probe_failure { diff --git a/backend/src/processing/processor/xtream_vod.rs b/backend/src/processing/processor/xtream_vod.rs index b3bc3fa88..d7306b4c0 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -9,6 +9,7 @@ use crate::processing::input_cache::resolve_input_storage_path; use crate::processing::processor::playlist::PlaylistProcessingContext; use crate::processing::processor::{ create_resolve_options_function_for_xtream_target, process_foreground_retry_once, select_cancel_token, + ProbeHandleGuard, ResolveOptions, ResolveOptionsFlags, FOREGROUND_BATCH_SIZE as BATCH_SIZE, FOREGROUND_MIN_RETRY_DELAY_SECS, FOREGROUND_RETRY_BATCH_MAX_SIZE as RETRY_BATCH_MAX_SIZE, }; @@ -133,7 +134,7 @@ async fn playlist_resolve_vod_info( let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb); if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() { - queue_background_vod_info(ctx, fpl, filter, &resolve_options, do_probe, resolve_tmdb_enabled); + queue_background_vod_info(ctx, fpl, filter, &resolve_options, do_probe, resolve_tmdb_enabled).await; } else { process_immediate_vod_info(ctx, fpl, filter, resolve_options, do_probe, resolve_tmdb_enabled).await; } @@ -415,7 +416,7 @@ fn check_resolve_tmdb( } } -fn queue_background_vod_info( +async fn queue_background_vod_info( ctx: &PlaylistProcessingContext, fpl: &mut FetchedPlaylist<'_>, filter: impl Fn(&PlaylistItem) -> bool, @@ -444,7 +445,7 @@ fn queue_background_vod_info( if !reasons.is_empty() { let task = UpdateTask::ResolveVod { id: provider_id.clone(), reason: reasons, delay: resolve_options.resolve_delay }; - if mgr.is_redundant_with_pending_task(input.name.as_ref(), &task) { + if mgr.should_skip_enqueue(input.name.clone(), &task).await { continue; } @@ -500,6 +501,7 @@ async fn update_vod_info_immediate( reasons.contains(ResolveReason::Probe), db_query, None, + None, ) .await } @@ -527,6 +529,7 @@ pub async fn update_vod_metadata( do_probe: bool, db_query: Option>>>, tmdb_resolved_out: Option<&AtomicBool>, + probe_pending_out: Option<&AtomicBool>, ) -> Result, TuliproxError> { let storage_dir = &app_config.config.load().storage_dir; let storage_path = resolve_input_storage_path(storage_dir, &input.name).await; @@ -599,6 +602,7 @@ pub async fn update_vod_metadata( let mut fetched_new = false; let mut properties_updated = false; let mut probe_failure: Option = None; + let mut probe_pending = false; // Determine the title to use for logging and fallback. // Cloning here to satisfy borrow checker when 'props' is mutated later. @@ -819,60 +823,74 @@ pub async fn update_vod_metadata( // Acquire Connection logic let temp_handle = if active_handle.is_some() { - None // No new handle needed + None } else { - active_provider.acquire_connection_for_probe(&input.name, probe_priority).await + active_provider + .acquire_connection_for_probe(&input.name, probe_priority) + .await + .map(|handle| ProbeHandleGuard::new(active_provider, handle)) }; - if active_handle.is_some() || temp_handle.is_some() { - debug_if_enabled!("Probing VOD '{}' (ID: {})", display_title, display_id); - let cancel_token = select_cancel_token(temp_handle.as_ref(), active_handle); - match crate::utils::ffmpeg::probe_url_with_cancel( - &stream_url, - user_agent.as_deref(), - analyze_duration, - probe_size, - ffprobe_timeout, - config.proxy.as_ref(), - cancel_token, - ) - .await - { - ProbeUrlOutcome::Success(_quality, raw_video, raw_audio) => { - if let Some(details) = properties.details.as_mut() { - if let Some(v) = raw_video { - details.video = Some(v.to_string().into()); - properties_updated = true; - } - if let Some(a) = raw_audio { - details.audio = Some(a.to_string().into()); - properties_updated = true; - } + if active_handle.is_some() || temp_handle.is_some() { + debug_if_enabled!("Probing VOD '{}' (ID: {})", display_title, display_id); + let cancel_token = select_cancel_token( + temp_handle.as_ref().and_then(ProbeHandleGuard::handle), + active_handle, + ); + match crate::utils::ffmpeg::probe_url_with_cancel( + &stream_url, + user_agent.as_deref(), + analyze_duration, + probe_size, + ffprobe_timeout, + config.proxy.as_ref(), + cancel_token, + ) + .await + { + ProbeUrlOutcome::Success(_quality, raw_video, raw_audio) => { + if let Some(details) = properties.details.as_mut() { + if let Some(v) = raw_video { + details.video = Some(v.to_string().into()); + properties_updated = true; } - } - ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => { - probe_failure = Some(ProbeFailureKind::NotFound); - } - ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => { - if probe_failure.is_none() { - probe_failure = Some(ProbeFailureKind::Other); - } - } - ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => { - if probe_failure.is_none() { - probe_failure = Some(ProbeFailureKind::Cancelled); + if let Some(a) = raw_audio { + details.audio = Some(a.to_string().into()); + properties_updated = true; } } } - if let Some(h) = temp_handle { - active_provider.release_handle(&h).await; + ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => { + probe_failure = Some(ProbeFailureKind::NotFound); + } + ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => { + probe_pending = true; + if probe_failure.is_none() { + probe_failure = Some(ProbeFailureKind::Other); + } + } + ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => { + probe_pending = true; + if probe_failure.is_none() { + probe_failure = Some(ProbeFailureKind::Cancelled); + } + } } } else { + probe_pending = true; warn!("Skipping probe for VOD {display_id} due to connection limits"); } + + if let Some(guard) = temp_handle { + guard.release().await; + } } } + if let Some(out) = probe_pending_out { + out.store(probe_pending, Ordering::Relaxed); + } + let probe_only_unresolved = do_probe && !fetch_info && !resolve_tmdb && !properties_updated && !fetched_new; if probe_only_unresolved { if let Some(kind) = probe_failure { diff --git a/backend/src/utils/network/xtream.rs b/backend/src/utils/network/xtream.rs index 83b373d0e..e1a60eb94 100644 --- a/backend/src/utils/network/xtream.rs +++ b/backend/src/utils/network/xtream.rs @@ -122,7 +122,7 @@ pub async fn get_xtream_stream_info(client: &reqwest::Client, return value; } } - Err(err) => error!("Failed to persist video info: {err}") + Err(err) => error!("Failed to persist video info for provider id {}: {err}", pli.provider_id) } } } diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 7a13c0540..11a71077b 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -562,9 +562,11 @@ "SIZE": "Size of the stream buffer in 8192-byte chunks. A value of 1024 means approximately 8MB." }, "STREAM_CONFIG": { + "CATCHUP_SESSION_TTL_SECS": "How long to keep a catchup provider-account reservation alive after the last request so resume/seek can stay on the same provider account.", "GRACE_PERIOD_HOLD_STREAM": "If enabled, keeps the provider stream open briefly after the last client disconnects to allow quick reconnects.", "GRACE_PERIOD_MILLIS": "Duration in milliseconds to allow exceeding max_connections during rapid channel switches (default 2000).", "GRACE_PERIOD_TIMEOUT_SECS": "How many seconds to wait before terminating excess connections after grace period ends (default 5).", + "HLS_SESSION_TTL_SECS": "How long to keep an HLS provider-account reservation alive between playlist or segment requests so the same provider account can be reused.", "RETRY": "Number of times to retry connecting to the provider on initial stream failure.", "SHARED_BURST_BUFFER_MB": "Initial buffer size in MB to provide immediate data to new clients joining a shared stream.", "THROTTLE": "Enables bandwidth limiting for outgoing streams.", @@ -734,6 +736,7 @@ "CHANNELS": "Channels", "CHAT_IDS": "Chat Ids", "CLEANUP": "Cleanup", + "CATCHUP_SESSION_TTL_SECS": "Catchup Session TTL (secs)", "CLEAR": "Clear", "CLIENT_IP": "Client IP", "CLOSE": "Close", @@ -835,6 +838,7 @@ "GRACE_PERIOD_MILLIS": "Grace Period (millis)", "GRACE_PERIOD_TIMEOUT": "Grace period timeout sec", "GRACE_PERIOD_TIMEOUT_SECS": "Grace Period Timeout (secs)", + "HLS_SESSION_TTL_SECS": "HLS Session TTL (secs)", "GROUP": "Group", "GROUPS": "Groups", "GROUP_PATTERN": "Group Pattern", diff --git a/frontend/scss/_theme.scss b/frontend/scss/_theme.scss index 1b37dd6ac..cd085b8d4 100644 --- a/frontend/scss/_theme.scss +++ b/frontend/scss/_theme.scss @@ -514,24 +514,6 @@ body[data-theme="bright"] { --table-card-background-color: #f7fbff; --table-alternate-background-color: #f0f6fc; --table-border-color: #cfdae6; - --stream-tech-chip-container-color: #0f3d2e; - --stream-tech-chip-container-background-color: #c4f1dd; - --stream-tech-chip-container-border-color: #65c79a; - --stream-tech-chip-video-codec-color: #123a66; - --stream-tech-chip-video-codec-background-color: #c9e4ff; - --stream-tech-chip-video-codec-border-color: #6ca9e9; - --stream-tech-chip-resolution-color: #4b2c0c; - --stream-tech-chip-resolution-background-color: #fde2b4; - --stream-tech-chip-resolution-border-color: #efb861; - --stream-tech-chip-fps-color: #5a1f4e; - --stream-tech-chip-fps-background-color: #f6d2ee; - --stream-tech-chip-fps-border-color: #d98cc8; - --stream-tech-chip-audio-codec-color: #15395a; - --stream-tech-chip-audio-codec-background-color: #c9eff9; - --stream-tech-chip-audio-codec-border-color: #71bdd0; - --stream-tech-chip-audio-channels-color: #2f3f10; - --stream-tech-chip-audio-channels-background-color: #deefc2; - --stream-tech-chip-audio-channels-border-color: #9ac56a; /* Tags & Statuses */ --tag-background-color: #f1f5f9; diff --git a/frontend/src/app/components/config/reverse_proxy_config_view.rs b/frontend/src/app/components/config/reverse_proxy_config_view.rs index 3122b51ab..67e35a5d3 100644 --- a/frontend/src/app/components/config/reverse_proxy_config_view.rs +++ b/frontend/src/app/components/config/reverse_proxy_config_view.rs @@ -36,6 +36,8 @@ const LABEL_THROTTLE: &str = "LABEL.THROTTLE"; const LABEL_GRACE_PERIOD_MILLIS: &str = "LABEL.GRACE_PERIOD_MILLIS"; const LABEL_GRACE_PERIOD_TIMEOUT_SECS: &str = "LABEL.GRACE_PERIOD_TIMEOUT_SECS"; const LABEL_GRACE_PERIOD_HOLD_STREAM: &str = "LABEL.GRACE_PERIOD_HOLD_STREAM"; +const LABEL_HLS_SESSION_TTL_SECS: &str = "LABEL.HLS_SESSION_TTL_SECS"; +const LABEL_CATCHUP_SESSION_TTL_SECS: &str = "LABEL.CATCHUP_SESSION_TTL_SECS"; const LABEL_THROTTLE_KBPS: &str = "LABEL.THROTTLE_KBPS"; const LABEL_STREAM_BUFFER: &str = "LABEL.STREAM_BUFFER"; const LABEL_BUFFER_ENABLED: &str = "LABEL.BUFFER_ENABLED"; @@ -123,6 +125,8 @@ generate_form_reducer!( ThrottleKbps => throttle_kbps: u64, SharedBurstBufferMb => shared_burst_buffer_mb: u64, GracePeriodHoldStream => grace_period_hold_stream: bool, + HlsSessionTtlSecs => hls_session_ttl_secs: u64, + CatchupSessionTtlSecs => catchup_session_ttl_secs: u64, } ); @@ -401,6 +405,8 @@ pub fn ReverseProxyConfigView() -> Html { { config_field!(stream_state.form, translate.t(LABEL_GRACE_PERIOD_MILLIS), grace_period_millis) } { config_field!(stream_state.form, translate.t(LABEL_GRACE_PERIOD_TIMEOUT_SECS), grace_period_timeout_secs) } { config_field_bool!(stream_state.form, translate.t(LABEL_GRACE_PERIOD_HOLD_STREAM), grace_period_hold_stream) } + { config_field!(stream_state.form, translate.t(LABEL_HLS_SESSION_TTL_SECS), hls_session_ttl_secs) } + { config_field!(stream_state.form, translate.t(LABEL_CATCHUP_SESSION_TTL_SECS), catchup_session_ttl_secs) } { config_field_optional!(stream_state.form, translate.t(LABEL_THROTTLE), throttle) } { config_field!(stream_state.form, translate.t(LABEL_THROTTLE_KBPS), throttle_kbps) } { config_field!(stream_state.form, translate.t(LABEL_SHARED_BURST_BUFFER_MB), shared_burst_buffer_mb) } @@ -567,6 +573,8 @@ pub fn ReverseProxyConfigView() -> Html { { edit_field_number_u64!(stream_state, translate.t(LABEL_GRACE_PERIOD_MILLIS), grace_period_millis, StreamConfigFormAction::GracePeriodMillis) } { edit_field_number_u64!(stream_state, translate.t(LABEL_GRACE_PERIOD_TIMEOUT_SECS), grace_period_timeout_secs, StreamConfigFormAction::GracePeriodTimeoutSecs) } { edit_field_bool!(stream_state, translate.t(LABEL_GRACE_PERIOD_HOLD_STREAM), grace_period_hold_stream, StreamConfigFormAction::GracePeriodHoldStream) } + { edit_field_number_u64!(stream_state, translate.t(LABEL_HLS_SESSION_TTL_SECS), hls_session_ttl_secs, StreamConfigFormAction::HlsSessionTtlSecs) } + { edit_field_number_u64!(stream_state, translate.t(LABEL_CATCHUP_SESSION_TTL_SECS), catchup_session_ttl_secs, StreamConfigFormAction::CatchupSessionTtlSecs) } { edit_field_text_option!(stream_state, translate.t(LABEL_THROTTLE), throttle, StreamConfigFormAction::Throttle) } { edit_field_number_u64!(stream_state, translate.t(LABEL_THROTTLE_KBPS), throttle_kbps, StreamConfigFormAction::ThrottleKbps) } { edit_field_number_u64!(stream_state, translate.t(LABEL_SHARED_BURST_BUFFER_MB), shared_burst_buffer_mb, StreamConfigFormAction::SharedBurstBufferMb) } diff --git a/shared/src/model/config/stream.rs b/shared/src/model/config/stream.rs index 5d13ffb3e..9f8e53da0 100644 --- a/shared/src/model/config/stream.rs +++ b/shared/src/model/config/stream.rs @@ -1,9 +1,11 @@ use crate::{ error::{info_err_res, TuliproxError, TuliproxErrorKind}, utils::{ - default_as_true, default_grace_period_millis, default_grace_period_timeout_secs, - default_shared_burst_buffer_mb, is_blank_optional_string, is_default_grace_period_millis, - is_default_grace_period_timeout_secs, is_default_shared_burst_buffer_mb, is_true, parse_to_kbps, + default_as_true, default_catchup_session_ttl_secs, default_grace_period_millis, + default_grace_period_timeout_secs, default_hls_session_ttl_secs, default_shared_burst_buffer_mb, + is_blank_optional_string, is_default_catchup_session_ttl_secs, is_default_grace_period_millis, + is_default_grace_period_timeout_secs, is_default_hls_session_ttl_secs, is_default_shared_burst_buffer_mb, + is_true, parse_to_kbps, }, }; @@ -47,6 +49,10 @@ pub struct StreamConfigDto { /// If true (default), wait for grace period check before streaming. #[serde(default = "default_as_true", skip_serializing_if = "is_true")] pub grace_period_hold_stream: bool, + #[serde(default = "default_hls_session_ttl_secs", skip_serializing_if = "is_default_hls_session_ttl_secs")] + pub hls_session_ttl_secs: u64, + #[serde(default = "default_catchup_session_ttl_secs", skip_serializing_if = "is_default_catchup_session_ttl_secs")] + pub catchup_session_ttl_secs: u64, #[serde(default, skip)] pub throttle_kbps: u64, #[serde(default = "default_shared_burst_buffer_mb", skip_serializing_if = "is_default_shared_burst_buffer_mb")] @@ -64,6 +70,8 @@ impl Default for StreamConfigDto { throttle_kbps: 0, shared_burst_buffer_mb: default_shared_burst_buffer_mb(), grace_period_hold_stream: true, + hls_session_ttl_secs: default_hls_session_ttl_secs(), + catchup_session_ttl_secs: default_catchup_session_ttl_secs(), } } } @@ -78,6 +86,8 @@ impl StreamConfigDto { && self.throttle_kbps == 0 && self.shared_burst_buffer_mb == default_shared_burst_buffer_mb() && self.grace_period_hold_stream + && self.hls_session_ttl_secs == default_hls_session_ttl_secs() + && self.catchup_session_ttl_secs == default_catchup_session_ttl_secs() } pub(crate) fn prepare(&mut self) -> Result<(), TuliproxError> { diff --git a/shared/src/model/stream_properties.rs b/shared/src/model/stream_properties.rs index 347e13c83..6cf08841d 100644 --- a/shared/src/model/stream_properties.rs +++ b/shared/src/model/stream_properties.rs @@ -4,13 +4,42 @@ use crate::{ arc_str_default_on_null, arc_str_none_default_on_null, arc_str_option_serde, deserialize_as_option_arc_str, deserialize_as_string_array, deserialize_json_as_opt_string, deserialize_number_from_string, deserialize_number_from_string_or_zero, serialize_json_as_opt_string, serialize_option_string_as_null_if_empty, - Internable, + Internable, CONSTANTS, }, }; use log::warn; use serde::{Deserialize, Serialize}; use std::sync::Arc; +fn format_episode_code(season: u32, episode: u32) -> Option { + match (season, episode) { + (0, 0) => None, + (season, 0) => Some(format!("S{season:02}")), + (0, episode) => Some(format!("E{episode:02}")), + (season, episode) => Some(format!("S{season:02}E{episode:02}")), + } +} + +fn title_contains_episode_code(title: &str) -> bool { CONSTANTS.re_episode_code.is_match(title) } + +pub fn normalize_episode_title(raw_title: &Arc, series_name: &Arc, season: u32, episode: u32) -> Arc { + let title = raw_title.trim(); + let series_name = series_name.trim(); + let Some(code) = format_episode_code(season, episode) else { + return if title.is_empty() { series_name.intern() } else { title.intern() }; + }; + + if title.is_empty() || (!series_name.is_empty() && title.eq_ignore_ascii_case(series_name)) { + return code.into(); + } + + if title_contains_episode_code(title) { + return title.intern(); + } + + format!("{code} - {title}").into() +} + #[derive(Default, Serialize, Deserialize, Debug, Clone, PartialEq)] pub struct LiveStreamProperties { #[serde(default, deserialize_with = "arc_str_none_default_on_null")] @@ -760,7 +789,7 @@ impl SeriesStreamProperties { id: e.id, episode_num: e.episode_num, season: e.season, - title: e.title.clone(), + title: normalize_episode_title(&e.title, &info.info.name, e.season, e.episode_num), container_extension: e.container_extension.clone(), custom_sid: e.custom_sid.clone(), added: e.added.clone(), @@ -850,7 +879,7 @@ impl SeriesStreamProperties { }), episode_num: e.episode_num, season: e.season, - title: e.title.clone(), + title: normalize_episode_title(&e.title, &info.info.name, e.season, e.episode_num), container_extension: e.container_extension.clone(), custom_sid: e.custom_sid.clone(), added: e.added.clone(), @@ -994,4 +1023,22 @@ mod tests { assert_eq!(props.tmdb, Some(7788)); assert_eq!(props.is_adult, 1); } + + #[test] + fn normalize_episode_title_injects_missing_episode_code() { + let normalized = normalize_episode_title(&"Pilot".into(), &"Example Show".into(), 1, 2); + assert_eq!(normalized.as_ref(), "S01E02 - Pilot"); + } + + #[test] + fn normalize_episode_title_replaces_series_name_only_with_episode_code() { + let normalized = normalize_episode_title(&"Example Show".into(), &"Example Show".into(), 1, 2); + assert_eq!(normalized.as_ref(), "S01E02"); + } + + #[test] + fn normalize_episode_title_keeps_existing_episode_code() { + let normalized = normalize_episode_title(&"S01E02".into(), &"Example Show".into(), 1, 2); + assert_eq!(normalized.as_ref(), "S01E02"); + } } diff --git a/shared/src/utils/constants.rs b/shared/src/utils/constants.rs index 5340e3ce9..79ad55859 100644 --- a/shared/src/utils/constants.rs +++ b/shared/src/utils/constants.rs @@ -117,6 +117,7 @@ pub struct Constants { pub re_quality: Regex, pub re_classifier_year: Regex, pub re_classifier_cleanup: Regex, + pub re_episode_code: Regex, pub re_classifier_episode: Regex, pub re_classifier_season: Regex, pub re_classifier_moviedb_id: Regex, @@ -177,6 +178,10 @@ pub static CONSTANTS: LazyLock = LazyLock::new(|| { re_quality: Regex::new(r"(?i)\b(x265|4K|UHD|8K|2160p?|1080p?|720p?|480p?|BLURAY|HDTV|DVDRIP|BRRIP|CAM|TS|HDR|DV|SDR)\b").unwrap(), re_classifier_year: Regex::new(r"[\(\[]?(\d{4})[\)\]]?").unwrap(), re_classifier_cleanup: Regex::new(r"(?i)[\s\._-]*(?:s\d+e\d+|\d+x\d+|season[\s\._-]*\d+|episode[\s\._-]*\d+).*$").unwrap(), + re_episode_code: Regex::new( + r"(?i)\b(?:s\d{1,2}e\d{1,3}|\d{1,2}x\d{1,3}|season[\s._-]*\d{1,2}(?:[\s._-]*episode[\s._-]*\d{1,3})?|episode[\s._-]*\d{1,3})\b" + ) + .unwrap(), re_classifier_episode: Regex::new(r"(?i)(?:e|episode|x)[\s\._-]*(\d+)").unwrap(), re_classifier_season: Regex::new(r"(?i)(?:s|season)[\s\._-]*(\d+)").unwrap(), re_classifier_moviedb_id: Regex::new(r"(?i)\b(tmdb|tvdb|imdb)[\s._=-]?(\d+)\b").unwrap(), diff --git a/shared/src/utils/default_utils.rs b/shared/src/utils/default_utils.rs index 2bc929cae..f0ea7590a 100644 --- a/shared/src/utils/default_utils.rs +++ b/shared/src/utils/default_utils.rs @@ -42,6 +42,10 @@ pub const fn default_shared_burst_buffer_mb() -> u64 { 12 } pub const fn is_default_shared_burst_buffer_mb(v: &u64) -> bool { *v == default_shared_burst_buffer_mb() } pub const fn default_grace_period_timeout_secs() -> u64 { 4 } pub const fn is_default_grace_period_timeout_secs(v: &u64) -> bool { *v == default_grace_period_timeout_secs() } +pub const fn default_hls_session_ttl_secs() -> u64 { 15 } +pub const fn is_default_hls_session_ttl_secs(v: &u64) -> bool { *v == default_hls_session_ttl_secs() } +pub const fn default_catchup_session_ttl_secs() -> u64 { 45 } +pub const fn is_default_catchup_session_ttl_secs(v: &u64) -> bool { *v == default_catchup_session_ttl_secs() } pub const fn default_panel_api_provision_timeout_secs() -> u64 { 65 } pub const fn default_panel_api_provision_probe_interval_secs() -> u64 { 15 } pub const fn default_panel_api_provision_cooldown_secs() -> u64 { 0 }