diff --git a/.gitignore b/.gitignore index dba79bc49..f2f6f98ed 100644 --- a/.gitignore +++ b/.gitignore @@ -57,4 +57,5 @@ backend/Cargo.lock # Ignore ai documentation files .c[lor][adu]* -.dual* +.dual-graph/ +.dual-graph-context/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ce806c89..32018559e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -106,6 +106,9 @@ - `grace_instant_stream` and `grace_hold_stream` are mutually exclusive. - Grace strategies require `grace_period_millis > 0`. - Added comprehensive connection handling documentation covering failures, user-visible behavior, priorities, sessions, and reconnects. + - Added two new runtime-flow handbook pages: + - operator-facing current runtime flow + - developer-facing runtime internals and activity flow - **Session Handling Boundary**: HLS and catchup remain session-based for continuity and provider affinity, but regular TS/VOD/local playback is now enforced as socket-bound admission. - A second non-HLS socket now counts as a second user connection even for the same user, IP, and stream. diff --git a/Cargo.lock b/Cargo.lock index 9e9823a9e..68f5bb694 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1796,6 +1796,8 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ + "allocator-api2", + "equivalent", "foldhash 0.1.5", ] @@ -2413,6 +2415,15 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "lru" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" +dependencies = [ + "hashbrown 0.15.5", +] + [[package]] name = "lru-slab" version = "0.1.2" @@ -4643,6 +4654,7 @@ dependencies = [ "libc", "log", "lol_html", + "lru", "lz4_flex", "memchr", "memmap2", diff --git a/README.md b/README.md index 3e0541eda..a69669170 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,14 @@ Jellyfin, Emby, Kodi and similar clients. ![tuliprox logo](https://github.com/user-attachments/assets/8ef9ea79-62ff-4298-978f-22326c5c3d02) +## Want to join the community + +[Join us on Discord](https://discord.gg/gkzCmWw9Tf) + +## License + +See [`LICENSE`](https://github.com/euzu/tuliprox/blob/develop/LICENSE). + ## πŸ† Key Features ### 1. Written in Rust β€” Maximum Performance, Minimal Footprint @@ -284,14 +292,6 @@ Generate all four formats simultaneously from the same source β€” one setup, eve `> ./docker/container-templates` -## Want to join the community - -[Join us on Discord](https://discord.gg/gkzCmWw9Tf) - -## License - -See [`LICENSE`](https://github.com/euzu/tuliprox/blob/develop/LICENSE). - ## Quick start Install with docker the latest image. diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 05af90b67..c41682b03 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -79,6 +79,7 @@ mime_guess = "2.0.5" crc32fast = "1.5.0" memchr = "2.8.0" http-body-util = "0.1.3" +lru = "0.12" [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index e8a666a44..e23b73eb8 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -5,8 +5,8 @@ use crate::{ create_active_client_stream, create_channel_unavailable_stream, create_custom_video_stream_response, create_provider_connections_exhausted_stream, create_provider_stream, get_stream_response_with_headers, tee_stream, AppState, CustomVideoStreamType, ProviderAllocation, ProviderConfig, ProviderStreamFactoryOptions, - ProviderStreamState, SharedStreamManager, StreamDetails, StreamError, StreamingStrategy, ThrottledStream, - UserApiRequest, UserSession, + PendingProviderReason, ProviderStreamState, SharedStreamManager, StreamDetails, StreamError, StreamingStrategy, + ThrottledStream, UserApiRequest, UserSession, }, }, auth::Fingerprint, @@ -38,7 +38,7 @@ use shared::{ }, utils::{ bin_serialize, extract_extension_from_url, human_readable_kbps, is_sanitize_sensitive_info_enabled, - replace_url_extension, sanitize_sensitive_info, + replace_url_extension, sanitize_sensitive_info, current_time_secs, trim_slash, Internable, CONTENT_TYPE_CBOR, CONTENT_TYPE_JSON, DASH_EXT, HLS_EXT, }, }; @@ -66,6 +66,130 @@ pub(crate) enum EvictionReentryGuard<'a> { SocketPlayback { virtual_id: VirtualId }, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum PlaybackRequestClass { + Prepare, + Activate, + FollowUp, + Terminate, +} + +#[derive(Clone, Copy)] +pub(crate) struct PlaybackRequestFacts<'a> { + pub(crate) item_type: PlaylistItemType, + pub(crate) existing_session: Option<&'a UserSession>, + pub(crate) prepare_only: bool, + pub(crate) terminate: bool, +} + +pub(crate) fn classify_playback_request(facts: PlaybackRequestFacts<'_>) -> PlaybackRequestClass { + if facts.terminate { + return PlaybackRequestClass::Terminate; + } + if facts.prepare_only { + return PlaybackRequestClass::Prepare; + } + if let Some(session) = facts.existing_session { + // FollowUp only for sessions that are actively counted. + // PendingProvider has no counted lease yet - activation is still pending. + // Prepared/Preserved/Expired sessions are not FollowUp. + if session.lifecycle.is_counted() { + return PlaybackRequestClass::FollowUp; + } + } + let _ = facts.item_type; + PlaybackRequestClass::Activate +} + +#[allow(clippy::too_many_arguments)] +pub(crate) async fn resolve_playback_request_admission( + app_state: &Arc, + user: &ProxyUserCredentials, + fingerprint: &Fingerprint, + item_type: PlaylistItemType, + user_session: Option<&UserSession>, + session_token: &str, + activate_unbound_session: bool, + eviction_reentry_guard: EvictionReentryGuard<'_>, + prepare_only: bool, + terminate: bool, +) -> (crate::api::model::ConnectionAdmission, Option, PlaybackRequestClass) { + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type, + existing_session: user_session, + prepare_only, + terminate, + }); + let limits_enabled = + (user.max_connections > 0 || user.soft_connections > 0) && app_state.app_config.config.load().user_access_control; + + // Handle explicit Terminate: run termination and return exhausted permission. + // No admission strategies are evaluated β€” termination immediately expires the playback. + if request_class == PlaybackRequestClass::Terminate { + if let Some(session) = user_session { + app_state + .active_users + .terminate_session(&user.username, session.token.as_str()) + .await; + } + return ( + crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Exhausted, + kind: user_session + .and_then(|session| session.connection_kind) + .or(Some(crate::api::model::ConnectionKind::Normal)), + }, + None, + request_class, + ); + } + + // Handle Prepare: no admission cost, just prepare state. Return Allowed without + // running strategies or modifying counted state. Caller handles the actual activation. + if request_class == PlaybackRequestClass::Prepare { + return ( + crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Allowed, + kind: user_session + .and_then(|session| session.connection_kind) + .or(Some(crate::api::model::ConnectionKind::Normal)), + }, + None, + request_class, + ); + } + + if request_class == PlaybackRequestClass::FollowUp || !limits_enabled { + return ( + crate::api::model::ConnectionAdmission { + permission: user_session + .map_or(UserConnectionPermission::Allowed, |session| session.permission), + kind: user_session + .and_then(|session| session.connection_kind) + .or(Some(crate::api::model::ConnectionKind::Normal)), + }, + None, + request_class, + ); + } + + let result = resolve_admission_with_strategies( + app_state, + &user.username, + user.max_connections, + user.soft_connections, + &fingerprint.client_ip, + &fingerprint.addr, + true, + Some(session_token), + activate_unbound_session, + eviction_reentry_guard, + ) + .await; + + (result.admission, result.grace_mode, request_class) +} + async fn should_suppress_eviction_for_recent_request( app_state: &Arc, username: &str, @@ -531,6 +655,31 @@ pub(in crate::api) fn get_stream_options(app_state: &Arc) -> StreamOpt StreamOptions { stream_retry, buffer_enabled, buffer_size, pipe_provider_stream } } +/// Metadata capturing which grace strategy was chosen and the original connection kind, +/// used to reconstruct the remaining-strategies slice on user-grace failure. +#[derive(Debug, Clone)] +#[allow(dead_code)] +pub(crate) struct GraceResolutionContext { + /// Index of the grace strategy that was actually used. + pub(crate) strategy_index: usize, + /// Full effective strategy list for stable reconstruction of the remaining slice. + pub(crate) strategies: Vec, + /// The original `ConnectionKind` from the admission decision that led to this grace. + /// Preserved so that the remaining-strategy fallback can return the correct kind + /// (e.g., `Soft`) even when the grace itself hardcoded `Normal`. + pub(crate) kind: Option, +} + +/// Structured result of evaluating admission strategies. +#[derive(Debug)] +#[allow(dead_code)] +pub(crate) struct AdmissionStrategyResolution { + pub(crate) admission: crate::api::model::ConnectionAdmission, + pub(crate) grace_mode: Option, + /// Present only when the request was admitted via a user-grace strategy. + pub(crate) grace_context: Option, +} + pub(in crate::api) fn get_effective_admission_strategies(app_state: &Arc) -> Vec { let config = app_state.app_config.config.load(); let stream_config = config.reverse_proxy.as_ref().and_then(|rp| rp.stream.as_ref()); @@ -547,59 +696,58 @@ pub(in crate::api) fn get_effective_admission_strategies(app_state: &Arc, - username: &str, +async fn evaluate_admission_strategy_loop<'a, F>( + app_state: &'a Arc, + username: &'a str, max_connections: u32, soft_connections: u16, - client_ip: &str, - request_addr: &std::net::SocketAddr, - // This controls whether an existing logical playback session may reopen while the user is already at limit. - // It is intentionally independent from whether the session is socket-bound. + client_ip: &'a str, + request_addr: &'a std::net::SocketAddr, use_session_admission: bool, - session_token: Option<&str>, + session_token: Option<&'a str>, activate_unbound_session: bool, - eviction_reentry_guard: EvictionReentryGuard<'_>, -) -> (crate::api::model::ConnectionAdmission, Option) { - use crate::api::model::{evaluate_strategy, AdmissionDecision, ConnectionAdmission, ConnectionKind, StrategyContext}; + eviction_reentry_guard: EvictionReentryGuard<'a>, + strategies: &'a [shared::model::AdmissionStrategy], + base_idx: usize, + admission: crate::api::model::ConnectionAdmission, + _kind_for_exhausted: Option, + build_grace_ctx: F, +) -> Option +where + F: Fn(usize) -> GraceResolutionContext, +{ + use crate::api::model::{evaluate_strategy, AdmissionDecision, StrategyContext}; use shared::model::UserConnectionPermission; - let admission = get_admission_for_request( - app_state, - username, - max_connections, - soft_connections, - use_session_admission, - session_token, - activate_unbound_session, - ) - .await; - - if admission.permission != UserConnectionPermission::Exhausted { - return (admission, None); - } - - let strategies = get_effective_admission_strategies(app_state); - if strategies.is_empty() { - debug!("No admission strategies configured, denying request for user {username}"); - return (admission, None); - } - let mut candidates = app_state.active_users.get_eviction_candidates(username, client_ip).await; - let ctx = StrategyContext { username, client_ip, strategies: &strategies }; - for strategy in &strategies { + let ctx = StrategyContext { username, client_ip, strategies }; + let mut idx = 0usize; + + for strategy in strategies { match evaluate_strategy(*strategy, &ctx, &candidates) { AdmissionDecision::NoMatch => {} AdmissionDecision::Grace(mode) => { if app_state.active_users.grant_grace(username).await { - return ( - ConnectionAdmission { + // Return a FRESH admission with GracePeriod permission (not the admission + // parameter, which may have Exhausted permission). The kind is preserved from + // the original admission. + return Some(AdmissionStrategyResolution { + admission: crate::api::model::ConnectionAdmission { permission: UserConnectionPermission::GracePeriod, - kind: Some(ConnectionKind::Normal), + kind: admission.kind, }, - Some(mode), - ); + grace_mode: Some(mode), + grace_context: Some(build_grace_ctx(base_idx + idx)), + }); } debug!("Grace grant rejected for user {username}, continuing with later strategies"); } @@ -636,20 +784,183 @@ pub(in crate::api) async fn resolve_admission_with_strategies( ) .await; if retry_admission.permission == UserConnectionPermission::Allowed { - return (retry_admission, None); + return Some(AdmissionStrategyResolution { + admission: retry_admission, + grace_mode: None, + grace_context: None, + }); } debug!("Admission still denied after eviction for user {username}, continuing with later strategies"); candidates = app_state.active_users.get_eviction_candidates(username, client_ip).await; } AdmissionDecision::Deny => { - debug!("Admission strategy denied for user {username}"); - return (admission, None); + // Caller constructs the final exhausted result. + return None; } } + idx += 1; + } + + // All strategies returned NoMatch β€” caller constructs the final exhausted result. + None +} + +#[allow(clippy::too_many_arguments)] +pub(in crate::api) async fn resolve_admission_with_strategies( + app_state: &Arc, + username: &str, + max_connections: u32, + soft_connections: u16, + client_ip: &str, + request_addr: &std::net::SocketAddr, + // This controls whether an existing logical playback session may reopen while the user is already at limit. + // It is intentionally independent from whether the session is socket-bound. + use_session_admission: bool, + session_token: Option<&str>, + activate_unbound_session: bool, + eviction_reentry_guard: EvictionReentryGuard<'_>, +) -> AdmissionStrategyResolution { + use shared::model::UserConnectionPermission; + + let admission = get_admission_for_request( + app_state, + username, + max_connections, + soft_connections, + use_session_admission, + session_token, + activate_unbound_session, + ) + .await; + + if admission.permission != UserConnectionPermission::Exhausted { + return AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None }; + } + + let strategies = get_effective_admission_strategies(app_state); + if strategies.is_empty() { + debug!("No admission strategies configured, denying request for user {username}"); + return AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None }; + } + + let build_grace_ctx = |global_idx: usize| GraceResolutionContext { + strategy_index: global_idx, + strategies: strategies.clone(), + kind: admission.kind, + }; + + if let Some(resolution) = evaluate_admission_strategy_loop( + app_state, + username, + max_connections, + soft_connections, + client_ip, + request_addr, + use_session_admission, + session_token, + activate_unbound_session, + eviction_reentry_guard, + &strategies, + 0, + admission, + admission.kind, + build_grace_ctx, + ) + .await + { + return resolution; } debug!("No admission strategy could admit user {username}"); - (admission, None) + AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None } +} + +/// Evaluates only the strategies that come AFTER the already-used grace strategy. +/// This is called when a user-grace has failed and the system needs to determine +/// whether a remaining eviction strategy can free a slot. +/// +/// Rules: +/// - Only `grace_context.strategies[(strategy_index + 1)..]` are evaluated +/// - `NoMatch` β†’ continue to next strategy +/// - `Evict` β†’ kick target, retry admission +/// - `Grace` β†’ technically possible under current config (only one grace allowed), but handled +/// - `Deny` β†’ final exhausted +/// - Empty remaining slice β†’ final exhausted +#[allow(clippy::too_many_arguments, clippy::too_many_lines)] +pub(in crate::api) async fn evaluate_remaining_strategies_after_grace( + app_state: &Arc, + username: &str, + max_connections: u32, + soft_connections: u16, + client_ip: &str, + request_addr: &std::net::SocketAddr, + use_session_admission: bool, + session_token: Option<&str>, + activate_unbound_session: bool, + eviction_reentry_guard: EvictionReentryGuard<'_>, + grace_context: &GraceResolutionContext, + original_kind: Option, +) -> AdmissionStrategyResolution { + use shared::model::UserConnectionPermission; + + let remaining = grace_context.strategy_index + 1; + let strategies = &grace_context.strategies; + if remaining >= strategies.len() { + debug!("No remaining strategies after grace for user {username}"); + return AdmissionStrategyResolution { + admission: crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Exhausted, + kind: original_kind, + }, + grace_mode: None, + grace_context: None, + }; + } + + // admission.kind is used only inside build_grace_ctx for the Grace case's + // GraceResolutionContext.kind. Both paths (helper early-return and caller + // exhausted construction) use original_kind, so this is safe. + let admission = crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Exhausted, + kind: original_kind, + }; + let build_grace_ctx = |global_idx: usize| GraceResolutionContext { + strategy_index: global_idx, + strategies: strategies.clone(), + kind: original_kind, + }; + + if let Some(resolution) = evaluate_admission_strategy_loop( + app_state, + username, + max_connections, + soft_connections, + client_ip, + request_addr, + use_session_admission, + session_token, + activate_unbound_session, + eviction_reentry_guard, + &strategies[remaining..], + remaining, + admission, + original_kind, + build_grace_ctx, + ) + .await + { + return resolution; + } + + debug!("No remaining strategy could admit user {username}"); + AdmissionStrategyResolution { + admission: crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Exhausted, + kind: original_kind, + }, + grace_mode: None, + grace_context: None, + } } struct SessionActivationRequest<'a> { @@ -657,42 +968,184 @@ struct SessionActivationRequest<'a> { input: &'a ConfigInput, user: &'a ProxyUserCredentials, session_token: &'a str, + request_class: Option, virtual_id: VirtualId, + item_type: PlaylistItemType, stream_url: &'a str, connection_permission: UserConnectionPermission, connection_kind: crate::api::model::ConnectionKind, socket_bound: bool, } +struct PlaybackActivationResult { + admission: crate::api::model::ConnectionAdmission, + grace_mode: Option, + grace_context: Option, + placeholder_transition_version: Option, +} + +/// # Panics +#[allow(clippy::too_many_lines)] async fn activate_session_before_stream_open( app_state: &Arc, request: SessionActivationRequest<'_>, -) -> (crate::api::model::ConnectionAdmission, Option, bool) { +) -> PlaybackActivationResult { let SessionActivationRequest { fingerprint, input, user, session_token, + request_class, virtual_id, + item_type, stream_url, connection_permission, connection_kind, socket_bound, } = request; + // Classify based on current session state, not the pre-computed value. + // If caller passes FollowUp, verify the session is still counted under the guard. + // A stale FollowUp would bypass admission β€” reclassify to catch this. + let effective_request_class = if let Some(request_class) = request_class { + if request_class == PlaybackRequestClass::FollowUp { + // Re-read session under the guard to ensure the counted lease is still held. + // Only reclassify if the session was in a counted state that has since been + // released. Pending sessions were never counted (admission pending provider + // acquisition) β€” keep FollowUp so no spurious placeholder is created. + let current_session = app_state + .active_users + .get_and_update_user_session(&user.username, session_token) + .await; + match current_session.as_ref().map(|s| &s.lifecycle) { + // Session is gone (expired/removed) or was never in a counted state β€” reclassify + // so admission runs and creates a fresh placeholder. + None => classify_playback_request(PlaybackRequestFacts { + item_type, + existing_session: None, + prepare_only: false, + terminate: false, + }), + // Had a counted lease but lost it β€” need to reclassify to Activate. + Some(crate::api::model::PlaybackLifecycle::Active | crate::api::model::PlaybackLifecycle::GraceActive) => { + classify_playback_request(PlaybackRequestFacts { + item_type, + existing_session: current_session.as_ref(), + prepare_only: false, + terminate: false, + }) + } + // Pending was never counted (waiting for provider slot); Prepared/Preserved hold + // no counted slot β€” caller-specified FollowUp is still valid, no reclassification. + _ => request_class, + } + } else { + request_class + } + } else { + let existing_session = app_state + .active_users + .get_and_update_user_session(&user.username, session_token) + .await; + classify_playback_request(PlaybackRequestFacts { + item_type, + existing_session: existing_session.as_ref(), + prepare_only: false, + terminate: false, + }) + }; let limits_enabled = app_state.app_config.config.load().user_access_control && (user.max_connections > 0 || user.soft_connections > 0); - if !limits_enabled || connection_permission == UserConnectionPermission::GracePeriod { - return ( - crate::api::model::ConnectionAdmission { + // Prepare: session setup without admission cost. The caller handles the actual activation. + // FollowUp: already counted, no re-admission needed. + // GracePeriod: grace already granted, no re-evaluation needed. + // No limits: skip admission entirely. + // GracePeriod permission is already resolved β€” skip admission strategies (re-run + // would evict the same session again). But we must still materialize the grace + // lifecycle (PendingProvider / GraceActive) so the session state is consistent. + if connection_permission == UserConnectionPermission::GracePeriod { + // Materialize grace lifecycle under the guard so the session state is consistent. + // Determine which grace mode applies by checking the current session state. + let current_session = app_state + .active_users + .get_and_update_user_session(&user.username, session_token) + .await; + let (_, resolved_grace) = match current_session.as_ref().map(|s| &s.lifecycle) { + Some(crate::api::model::PlaybackLifecycle::PendingProvider { .. }) => { + // Session already in PendingProvider β€” refresh deadline. + let deadline = current_time_secs().saturating_add(app_state.get_grace_options().timeout_secs); + let _ = app_state + .active_users + .mark_pending_provider(&user.username, session_token, PendingProviderReason::GraceHold, deadline) + .await; + (crate::api::model::PlaybackLifecycle::PendingProvider { + data: crate::api::model::PendingProviderState { + reason_code: PendingProviderReason::GraceHold, + created_at: current_time_secs(), + deadline, + version: current_session.as_ref().map_or(0, |s| { + if let crate::api::model::PlaybackLifecycle::PendingProvider { data } = &s.lifecycle { + data.version + } else { 0 } + }), + wake_source: None, + }, + }, Some(crate::api::model::GraceMode::Hold)) + } + Some(crate::api::model::PlaybackLifecycle::GraceActive) => { + // Already in GraceActive β€” nothing to refresh. + (crate::api::model::PlaybackLifecycle::GraceActive, Some(crate::api::model::GraceMode::Instant)) + } + _ => { + // Session not yet in grace state β€” infer from item_type defaults. + // Live/LiveHls/LiveDash default to Hold; VOD/Catchup to Instant. + if item_type.is_live() || item_type.is_live_adaptive() { + let deadline = current_time_secs().saturating_add(app_state.get_grace_options().timeout_secs); + let _ = app_state + .active_users + .mark_pending_provider(&user.username, session_token, PendingProviderReason::GraceHold, deadline) + .await; + (crate::api::model::PlaybackLifecycle::PendingProvider { + data: crate::api::model::PendingProviderState { + reason_code: PendingProviderReason::GraceHold, + created_at: current_time_secs(), + deadline, + version: 1, + wake_source: None, + }, + }, Some(crate::api::model::GraceMode::Hold)) + } else { + app_state.active_users.mark_grace_active(&user.username, session_token).await; + (crate::api::model::PlaybackLifecycle::GraceActive, Some(crate::api::model::GraceMode::Instant)) + } + } + }; + return PlaybackActivationResult { + admission: crate::api::model::ConnectionAdmission { permission: connection_permission, kind: Some(connection_kind), }, - None, - false, - ); + grace_mode: resolved_grace, + grace_context: None, + placeholder_transition_version: None, + }; + } + // No limits: skip admission entirely. FollowUp / Prepare: no re-admission needed. + if !limits_enabled + || effective_request_class == PlaybackRequestClass::FollowUp + || effective_request_class == PlaybackRequestClass::Prepare + { + return PlaybackActivationResult { + admission: crate::api::model::ConnectionAdmission { + permission: connection_permission, + kind: Some(connection_kind), + }, + grace_mode: None, + grace_context: None, + placeholder_transition_version: None, + }; } - let created_placeholder = app_state + let placeholder_transition_version = Some(app_state .active_users .ensure_user_session_placeholder(crate::api::model::CreateUserSessionParams { user, @@ -705,9 +1158,9 @@ async fn activate_session_before_stream_open( connection_kind: Some(connection_kind), socket_bound, }) - .await; + .await); - let (admission, grace_mode) = resolve_admission_with_strategies( + let result = resolve_admission_with_strategies( app_state, &user.username, user.max_connections, @@ -723,9 +1176,35 @@ async fn activate_session_before_stream_open( EvictionReentryGuard::Session(session_token) }, ) - .await; + .await; + let admission = result.admission; + let grace_mode = result.grace_mode; + let grace_context = result.grace_context; - (admission, grace_mode, created_placeholder) + if admission.permission == UserConnectionPermission::GracePeriod { + if matches!(grace_mode, Some(crate::api::model::GraceMode::Hold)) { + // Hold: session waits for provider slot. Does not count until provider is acquired. + let deadline = current_time_secs().saturating_add(app_state.get_grace_options().timeout_secs); + let _ = app_state + .active_users + .mark_pending_provider(&user.username, session_token, PendingProviderReason::GraceHold, deadline) + .await; + } else if matches!(grace_mode, Some(crate::api::model::GraceMode::Instant)) { + // Instant: session is provisionally active immediately. Counts against admission limits + // until the grace window resolves (success β†’ Active, failure β†’ Expired). + app_state + .active_users + .mark_grace_active(&user.username, session_token) + .await; + } + } + + PlaybackActivationResult { + admission, + grace_mode, + grace_context, + placeholder_transition_version, + } } pub fn get_stream_alternative_url(stream_url: &str, input: &ConfigInput, alias_input: &Arc) -> String { @@ -957,6 +1436,7 @@ async fn create_stream_response_details( is_reopen: bool, session_owner: Option<&str>, grace_hold_override: Option, + grace_resolution_context: Option, ) -> Result { let mut streaming_strategy = resolve_streaming_strategy( app_state, @@ -1027,6 +1507,7 @@ async fn create_stream_response_details( disable_provider_grace: false, reconnect_flag: None, provider_handle: streaming_strategy.provider_handle.clone(), + grace_resolution_context, }) } ProviderStreamState::Available(_provider_name, request_url) @@ -1132,6 +1613,7 @@ async fn create_stream_response_details( disable_provider_grace: false, reconnect_flag, provider_handle, + grace_resolution_context, }) } } @@ -1288,17 +1770,41 @@ pub async fn force_provider_stream_response( ctx: ForceStreamRequestContext<'_>, grace_mode: Option, ) -> impl IntoResponse + Send { + let _transition_guard = app_state + .active_users + .acquire_playback_transition(&ctx.user.username, &user_session.token) + .await; let stream_options = get_stream_options(app_state); let share_stream = false; let connection_permission = UserConnectionPermission::Allowed; let item_type = stream_channel.item_type; - // Release every stale provider slot and actively close every stale session socket before - // reacquiring. Otherwise the previous stream can linger long enough that repeated hold - // requests are abandoned by the client before provider capacity becomes visible again. - for addr in session_reacquire_cleanup_addrs(user_session, &fingerprint.addr) { - app_state.connection_manager.release_provider_connection(&addr).await; - let _ = app_state.connection_manager.close_connection_signal(&addr); + // Forced reopens must clear stale provider slots before reacquiring. For adaptive HLS/DASH + // sessions we only target old active stream sockets of the same session, never manifest-only + // session addresses, otherwise the controlling playlist request gets torn down. + let cleanup_addrs = if item_type.is_live_adaptive() { + app_state + .active_users + .adaptive_session_stream_cleanup_addrs(&ctx.user.username, &user_session.token, &fingerprint.addr) + .await + } else { + session_reacquire_cleanup_addrs(user_session, &fingerprint.addr) + }; + + if cleanup_addrs.is_empty() { + debug_if_enabled!( + "Forced reopen cleanup had no stale targets for item_type={item_type:?} session={} current_addr={}", + sanitize_sensitive_info(&user_session.token), + sanitize_sensitive_info(&fingerprint.addr.to_string()) + ); + } else { + debug_if_enabled!( + "Forced reopen cleanup releasing {} stale target(s) for item_type={item_type:?} session={} current_addr={}", + cleanup_addrs.len(), + sanitize_sensitive_info(&user_session.token), + sanitize_sensitive_info(&fingerprint.addr.to_string()) + ); + cleanup_forced_reopen_addrs(app_state, item_type, &cleanup_addrs).await; } // Provider-affine playback must stay on the same provider account across seeks/range reconnects. @@ -1333,6 +1839,7 @@ pub async fn force_provider_stream_response( true, Some(user_session.token.as_str()), grace_mode.map(|mode| matches!(mode, crate::api::model::GraceMode::Hold)), + None, ) .await { @@ -1340,7 +1847,7 @@ pub async fn force_provider_stream_response( Err(err) => { app_state .active_users - .release_unbound_session_reservation(&ctx.user.username, &user_session.token, false) + .release_unbound_session_reservation(&ctx.user.username, &user_session.token, None, false) .await; error!("Failed to stream: {err}"); return StatusCode::INTERNAL_SERVER_ERROR.into_response(); @@ -1409,7 +1916,7 @@ pub async fn force_provider_stream_response( app_state.connection_manager.release_provider_handle(stream_details.provider_handle).await; app_state .active_users - .release_unbound_session_reservation(&ctx.user.username, &user_session.token, false) + .release_unbound_session_reservation(&ctx.user.username, &user_session.token, None, false) .await; if let (Some(stream), _stream_info) = create_channel_unavailable_stream(&app_state.app_config, &[], StatusCode::SERVICE_UNAVAILABLE) @@ -1431,10 +1938,11 @@ pub async fn force_provider_stream_response( /// # Panics #[allow(clippy::too_many_arguments, clippy::too_many_lines)] -pub async fn stream_response( +pub(crate) async fn stream_response( fingerprint: &Fingerprint, app_state: &Arc, session_token: &str, + request_class: Option, mut stream_channel: StreamChannel, stream_url: &str, req_headers: &HeaderMap, @@ -1446,6 +1954,10 @@ pub async fn stream_response( allow_exhausted_shared_reconnect: bool, grace_mode: Option, ) -> impl IntoResponse + Send { + let _transition_guard = app_state + .active_users + .acquire_playback_transition(&user.username, session_token) + .await; let request_log_stream_url = resolve_request_url_for_logging(input, stream_url); if log_enabled!(log::Level::Trace) { trace!("Try to open stream {}", sanitize_sensitive_info(request_log_stream_url.as_ref())); @@ -1455,14 +1967,16 @@ pub async fn stream_response( let item_type = stream_channel.item_type; let mut connection_permission = connection_permission; let mut connection_kind = connection_kind; - let (final_admission, resolved_grace_mode, created_placeholder_session) = activate_session_before_stream_open( + let activation = activate_session_before_stream_open( app_state, SessionActivationRequest { fingerprint, input, user, session_token, + request_class, virtual_id, + item_type, stream_url, connection_permission, connection_kind, @@ -1470,9 +1984,9 @@ pub async fn stream_response( }, ) .await; - let grace_mode = resolved_grace_mode.or(grace_mode); - connection_permission = final_admission.permission; - connection_kind = final_admission.kind.unwrap_or(connection_kind); + let grace_mode = activation.grace_mode.or(grace_mode); + connection_permission = activation.admission.permission; + connection_kind = activation.admission.kind.unwrap_or(connection_kind); let allow_shared_reuse = connection_permission != UserConnectionPermission::Exhausted || allow_exhausted_shared_reconnect; @@ -1528,7 +2042,12 @@ pub async fn stream_response( if connection_permission == UserConnectionPermission::Exhausted { app_state .active_users - .release_unbound_session_reservation(&user.username, session_token, created_placeholder_session) + .release_unbound_session_reservation( + &user.username, + session_token, + activation.placeholder_transition_version, + activation.placeholder_transition_version.is_some(), + ) .await; record_connect_failed_attempt(ConnectFailedAttempt { app_state, @@ -1570,6 +2089,7 @@ pub async fn stream_response( false, Some(session_token), grace_mode.map(|m| matches!(m, crate::api::model::GraceMode::Hold)), + activation.grace_context.clone(), ) .await { @@ -1577,7 +2097,12 @@ pub async fn stream_response( Err(err) => { app_state .active_users - .release_unbound_session_reservation(&user.username, session_token, created_placeholder_session) + .release_unbound_session_reservation( + &user.username, + session_token, + activation.placeholder_transition_version, + activation.placeholder_transition_version.is_some(), + ) .await; error!("Failed to stream: {err}"); return StatusCode::INTERNAL_SERVER_ERROR.into_response(); @@ -1780,7 +2305,12 @@ pub async fn stream_response( app_state.connection_manager.release_provider_handle(stream_details.provider_handle).await; app_state .active_users - .release_unbound_session_reservation(&user.username, session_token, created_placeholder_session) + .release_unbound_session_reservation( + &user.username, + session_token, + activation.placeholder_transition_version, + activation.placeholder_transition_version.is_some(), + ) .await; StatusCode::BAD_REQUEST.into_response() } @@ -1858,6 +2388,20 @@ pub(crate) fn get_hls_session_ttl_secs(app_state: &Arc) -> u64 { get_stream_config_u64(app_state, |stream| stream.hls_session_ttl_secs, default_hls_session_ttl_secs()) } +async fn cleanup_forced_reopen_addrs( + app_state: &Arc, + item_type: PlaylistItemType, + cleanup_addrs: &[SocketAddr], +) { + let close_client_socket = !item_type.is_live_adaptive(); + for addr in cleanup_addrs { + app_state.connection_manager.release_provider_connection(addr).await; + if close_client_socket { + let _ = app_state.connection_manager.close_connection_signal(addr); + } + } +} + pub(crate) fn get_catchup_session_ttl_secs(app_state: &Arc) -> u64 { get_stream_config_u64(app_state, |stream| stream.catchup_session_ttl_secs, default_catchup_session_ttl_secs()) } @@ -1956,7 +2500,7 @@ async fn try_shared_stream_response_if_any( } #[allow(clippy::too_many_arguments, clippy::too_many_lines)] -pub async fn local_stream_response( +pub(crate) async fn local_stream_response( fingerprint: &Fingerprint, app_state: &Arc, pli: StreamChannel, @@ -1967,8 +2511,19 @@ pub async fn local_stream_response( connection_permission: UserConnectionPermission, connection_kind: crate::api::model::ConnectionKind, playback_session_token: Option<&str>, + request_class: Option, check_path: bool, ) -> impl IntoResponse + Send { + let _transition_guard = if let Some(session_token) = playback_session_token { + Some( + app_state + .active_users + .acquire_playback_transition(&user.username, session_token) + .await, + ) + } else { + None + }; if log_enabled!(log::Level::Trace) { trace!("Try to open stream {}", sanitize_sensitive_info(&pli.url)); } @@ -2097,14 +2652,16 @@ pub async fn local_stream_response( }; let mut connection_kind = connection_kind; if let Some(session_token) = playback_session_token { - let (final_admission, resolved_grace_mode, created_placeholder) = activate_session_before_stream_open( + let activation = activate_session_before_stream_open( app_state, SessionActivationRequest { fingerprint, input, user, session_token, + request_class, virtual_id: pli.virtual_id, + item_type: pli.item_type, stream_url: &pli.url, connection_permission, connection_kind, @@ -2112,14 +2669,19 @@ pub async fn local_stream_response( }, ) .await; - grace_mode = resolved_grace_mode; - connection_permission = final_admission.permission; - connection_kind = final_admission.kind.unwrap_or(connection_kind); + grace_mode = activation.grace_mode; + connection_permission = activation.admission.permission; + connection_kind = activation.admission.kind.unwrap_or(connection_kind); if connection_permission == UserConnectionPermission::Exhausted { app_state .active_users - .release_unbound_session_reservation(&user.username, session_token, created_placeholder) + .release_unbound_session_reservation( + &user.username, + session_token, + activation.placeholder_transition_version, + activation.placeholder_transition_version.is_some(), + ) .await; return create_custom_video_stream_response( app_state, @@ -2364,7 +2926,7 @@ async fn fetch_resource_with_retry( return Some(build_resource_stream_response(app_state, resource_url, response).await); } - // Non-retriable Status β†’ Upstream Response incl. Body + // Non-retriable Status -> Upstream Response incl. Body debug_if_enabled!("Failed to open resource got status {status} for {}", sanitize_sensitive_info(resource_url)); let mut response_builder = axum::response::Response::builder().status(status); @@ -2904,6 +3466,7 @@ mod tests { #[test] fn test_should_allow_exhausted_shared_reconnect_only_for_matching_shared_session() { let session = UserSession { + transition_version: 1, connection_kind: Some(crate::api::model::ConnectionKind::Normal), token: "tok".to_string(), virtual_id: 282, @@ -2915,7 +3478,7 @@ mod tests { ts: 1, started_at: 1, permission: UserConnectionPermission::Allowed, - counted: false, + lifecycle: crate::api::model::PlaybackLifecycle::Active, }; assert!(should_allow_exhausted_shared_reconnect( @@ -3161,6 +3724,7 @@ mod tests { stream_history: None, qos_aggregation: None, }), + user_access_control: true, ..Config::default() }; @@ -3205,6 +3769,864 @@ mod tests { } } + fn create_test_session( + token: &str, + item_type: PlaylistItemType, + lifecycle: crate::api::model::PlaybackLifecycle, + ) -> UserSession { + UserSession { + token: token.to_string(), + transition_version: 1, + virtual_id: 42, + provider: Arc::::from("provider-a"), + stream_url: Arc::::from(match item_type { + PlaylistItemType::LiveHls => "http://provider-1.example/live/42.m3u8", + _ => "http://provider-1.example/live/42.ts", + }), + addr: "127.0.0.1:55555".parse().unwrap_or_else(|_| unreachable!()), + socket_bound: item_type.uses_socket_bound_session(), + active_addrs: Vec::new(), + ts: 1, + started_at: 1, + permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + lifecycle, + } + } + + #[test] + fn classify_playback_request_marks_adaptive_playlist_request_as_prepare() { + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type: PlaylistItemType::LiveHls, + existing_session: None, + prepare_only: true, + terminate: false, + }); + + assert_eq!(request_class, PlaybackRequestClass::Prepare); + } + + #[test] + fn classify_playback_request_marks_preserved_session_as_activate() { + let session = create_test_session( + "tok-preserved", + PlaylistItemType::LiveHls, + crate::api::model::PlaybackLifecycle::Preserved, + ); + + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type: PlaylistItemType::LiveHls, + existing_session: Some(&session), + prepare_only: false, + terminate: false, + }); + + assert_eq!(request_class, PlaybackRequestClass::Activate); + } + + #[test] + fn classify_playback_request_marks_counted_session_as_follow_up() { + let session = create_test_session( + "tok-active", + PlaylistItemType::LiveHls, + crate::api::model::PlaybackLifecycle::Active, + ); + + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type: PlaylistItemType::LiveHls, + existing_session: Some(&session), + prepare_only: false, + terminate: false, + }); + + assert_eq!(request_class, PlaybackRequestClass::FollowUp); + } + + /// `PendingProvider` must NOT be classified as `FollowUp`. + /// `PendingProvider` has no counted lease yet β€” the session is still waiting + /// for a provider slot. A new request on a `PendingProvider` session should + /// be `Activate` so that full admission evaluation happens, not a cheap + /// `FollowUp` skip. + #[test] + fn classify_playback_request_marks_pending_provider_as_activate_not_follow_up() { + let session = create_test_session( + "tok-pending", + PlaylistItemType::LiveHls, + crate::api::model::PlaybackLifecycle::PendingProvider { + data: crate::api::model::PendingProviderState { + reason_code: crate::api::model::PendingProviderReason::GraceHold, + created_at: 1, + deadline: 30, + version: 1, + wake_source: None, + } + }, + ); + + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type: PlaylistItemType::LiveHls, + existing_session: Some(&session), + prepare_only: false, + terminate: false, + }); + + assert_eq!( + request_class, + PlaybackRequestClass::Activate, + "PendingProvider should not be FollowUp - it has no counted lease yet" + ); + } + + /// `Active` without a counted lease must NOT be classified as `FollowUp`. + /// `FollowUp` should only be returned when the session actually owns a + /// counted admission lease. A session with `Active` lifecycle but no counted + /// lease should go through `Activate` so that the counted lease is reacquired. + #[test] + fn classify_playback_request_marks_active_without_counted_as_activate_not_follow_up() { + let mut session = create_test_session( + "tok-active-uncounted", + PlaylistItemType::LiveHls, + crate::api::model::PlaybackLifecycle::Active, // counted=false via is_counted() + ); + // Manually force counted=false by setting to Prepared lifecycle, then restoring + // Note: is_counted() returns false for Prepared, true for Active + // For this test we need a session that is Active lifecycle but not counted + // The new model derives counted from lifecycle, so we must use a different lifecycle + // to represent "not counted". Use Prepared instead. + session.lifecycle = crate::api::model::PlaybackLifecycle::Prepared; + + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type: PlaylistItemType::LiveHls, + existing_session: Some(&session), + prepare_only: false, + terminate: false, + }); + + assert_eq!( + request_class, + PlaybackRequestClass::Activate, + "Active session with counted=false should not be FollowUp" + ); + } + + /// Prepared sessions must be classified as Activate. + #[test] + fn classify_playback_request_marks_prepared_session_as_activate() { + let session = create_test_session( + "tok-prepared", + PlaylistItemType::LiveHls, + crate::api::model::PlaybackLifecycle::Prepared, + ); + + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type: PlaylistItemType::LiveHls, + existing_session: Some(&session), + prepare_only: false, + terminate: false, + }); + + assert_eq!(request_class, PlaybackRequestClass::Activate); + } + + /// `GraceActive` without counted lease must NOT be classified as `FollowUp`. + #[test] + fn classify_playback_request_marks_grace_active_without_counted_as_activate() { + let mut session = create_test_session( + "tok-grace-uncounted", + PlaylistItemType::LiveHls, + crate::api::model::PlaybackLifecycle::Active, // is_counted() = true for GraceActive + ); + // Test scenario: session has GraceActive lifecycle but we need it NOT counted + // This represents the edge case before grace task resolves. Use Prepared lifecycle + // to model "not counted" since is_counted() returns false for Prepared. + session.lifecycle = crate::api::model::PlaybackLifecycle::Prepared; + + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type: PlaylistItemType::LiveHls, + existing_session: Some(&session), + prepare_only: false, + terminate: false, + }); + + assert_eq!( + request_class, + PlaybackRequestClass::Activate, + "GraceActive session with counted=false should not be FollowUp" + ); + } + + #[tokio::test] + async fn activate_session_before_stream_open_skips_placeholder_for_follow_up_session() { + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::EvictUserSameIpOldest]), + }); + let addr: SocketAddr = "127.0.0.1:55220".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let input = app_state.app_config.sources.load().inputs[0].clone(); + let mut user = ProxyUserCredentials::default(); + user.username = "follow-up-user".to_string(); + user.max_connections = 1; + let mut channel = create_test_live_channel("http://provider-1.example/live/55220.m3u8"); + channel.item_type = PlaylistItemType::LiveHls; + channel.virtual_id = 55220; + + app_state.connection_manager.add_connection(&addr).await; + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-follow-up", + virtual_id: channel.virtual_id, + provider: input.name.as_ref(), + stream_url: channel.url.as_ref(), + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: true, + }) + .await; + app_state + .connection_manager + .update_connection(crate::api::model::ConnectionParams { + meter_uid: 1, + username: &user.username, + max_connections: user.max_connections, + soft_connections: user.soft_connections, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 10, + fingerprint: &fingerprint, + provider: input.name.clone(), + stream_channel: &channel, + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-follow-up"), + }) + .await; + + let activation = activate_session_before_stream_open( + &app_state, + SessionActivationRequest { + fingerprint: &fingerprint, + input: input.as_ref(), + user: &user, + session_token: "tok-follow-up", + request_class: None, + virtual_id: channel.virtual_id, + item_type: PlaylistItemType::LiveHls, + stream_url: channel.url.as_ref(), + connection_permission: UserConnectionPermission::Allowed, + connection_kind: crate::api::model::ConnectionKind::Normal, + socket_bound: true, + }, + ) + .await; + + assert_eq!(activation.admission.permission, UserConnectionPermission::Allowed); + assert_eq!(activation.admission.kind, Some(crate::api::model::ConnectionKind::Normal)); + assert_eq!(activation.grace_mode, None); + assert!( + activation.placeholder_transition_version.is_none(), + "follow-up activation must not create a placeholder session" + ); + } + + #[tokio::test] + async fn activate_session_before_stream_open_uses_precomputed_follow_up_request_class() { + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::EvictUserSameIpOldest]), + }); + let addr: SocketAddr = "127.0.0.1:55221".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let input = app_state.app_config.sources.load().inputs[0].clone(); + let mut user = ProxyUserCredentials::default(); + user.username = "precomputed-follow-up-user".to_string(); + user.max_connections = 1; + let mut channel = create_test_live_channel("http://provider-1.example/live/55221.m3u8"); + channel.item_type = PlaylistItemType::LiveHls; + channel.virtual_id = 55221; + + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-precomputed-follow-up", + virtual_id: channel.virtual_id, + provider: input.name.as_ref(), + stream_url: channel.url.as_ref(), + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: true, + }) + .await; + + let activation = activate_session_before_stream_open( + &app_state, + SessionActivationRequest { + fingerprint: &fingerprint, + input: input.as_ref(), + user: &user, + session_token: "tok-precomputed-follow-up", + request_class: Some(PlaybackRequestClass::FollowUp), + virtual_id: channel.virtual_id, + item_type: PlaylistItemType::LiveHls, + stream_url: channel.url.as_ref(), + connection_permission: UserConnectionPermission::Allowed, + connection_kind: crate::api::model::ConnectionKind::Normal, + socket_bound: true, + }, + ) + .await; + + assert_eq!(activation.admission.permission, UserConnectionPermission::Allowed); + assert_eq!(activation.admission.kind, Some(crate::api::model::ConnectionKind::Normal)); + assert_eq!(activation.grace_mode, None); + assert!( + activation.placeholder_transition_version.is_none(), + "precomputed follow-up activation must not create a placeholder session" + ); + } + + // stale FollowUp revalidation + #[tokio::test] + async fn activate_session_before_stream_open_stale_follow_up_reclassified_on_counted_lease_release() { + // Scenario: pre-computed FollowUp, but session's counted lease was released before + // the guard was acquired. Must reclassify to Activate so admission runs. + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::EvictUserSameIpOldest]), + }); + let addr: SocketAddr = "127.0.0.1:55230".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let input = app_state.app_config.sources.load().inputs[0].clone(); + let mut user = ProxyUserCredentials::default(); + user.username = "stale-followup-user".to_string(); + user.max_connections = 1; + let mut channel = create_test_live_channel("http://provider-1.example/live/55230.m3u8"); + channel.item_type = PlaylistItemType::LiveHls; + channel.virtual_id = 55230; + + // Session created in Active (counted) state. + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-stale-followup", + virtual_id: channel.virtual_id, + provider: input.name.as_ref(), + stream_url: channel.url.as_ref(), + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: true, + }) + .await; + + // Simulate the counted lease being released before activation: + // expire the session so it no longer has a counted lease. + app_state + .active_users + .terminate_session(&user.username, "tok-stale-followup") + .await; + + // Call activate with stale FollowUp. Must NOT skip admission β€” reclassification + // to Activate must run so the placeholder is created. + let activation = activate_session_before_stream_open( + &app_state, + SessionActivationRequest { + fingerprint: &fingerprint, + input: input.as_ref(), + user: &user, + session_token: "tok-stale-followup", + request_class: Some(PlaybackRequestClass::FollowUp), + virtual_id: channel.virtual_id, + item_type: PlaylistItemType::LiveHls, + stream_url: channel.url.as_ref(), + connection_permission: UserConnectionPermission::Allowed, + connection_kind: crate::api::model::ConnectionKind::Normal, + socket_bound: true, + }, + ) + .await; + + // Must NOT skip β€” placeholder must be created since session is expired. + assert!( + activation.placeholder_transition_version.is_some(), + "stale FollowUp with expired session must run admission and create placeholder" + ); + } + + // pre-resolved Grace materialization + #[tokio::test] + async fn activate_session_before_stream_open_pre_resolved_grace_period_materializes_pending_provider() { + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]), + }); + let addr: SocketAddr = "127.0.0.1:55231".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let input = app_state.app_config.sources.load().inputs[0].clone(); + let mut user = ProxyUserCredentials::default(); + user.username = "pre-resolved-grace-user".to_string(); + user.max_connections = 1; + let mut channel = create_test_live_channel("http://provider-1.example/live/55231.m3u8"); + channel.item_type = PlaylistItemType::LiveHls; + channel.virtual_id = 55231; + + // Session in Prepared state (no grace lifecycle yet). + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-pre-resolved-grace", + virtual_id: channel.virtual_id, + provider: input.name.as_ref(), + stream_url: channel.url.as_ref(), + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: true, + }) + .await; + + // Call activation with pre-resolved GracePeriod permission. + let activation = activate_session_before_stream_open( + &app_state, + SessionActivationRequest { + fingerprint: &fingerprint, + input: input.as_ref(), + user: &user, + session_token: "tok-pre-resolved-grace", + request_class: None, + virtual_id: channel.virtual_id, + item_type: PlaylistItemType::LiveHls, + stream_url: channel.url.as_ref(), + connection_permission: UserConnectionPermission::GracePeriod, + connection_kind: crate::api::model::ConnectionKind::Normal, + socket_bound: true, + }, + ) + .await; + + assert_eq!(activation.admission.permission, UserConnectionPermission::GracePeriod); + assert_eq!(activation.grace_mode, Some(crate::api::model::GraceMode::Hold)); + + let session = app_state + .active_users + .get_and_update_user_session(&user.username, "tok-pre-resolved-grace") + .await; + assert!( + session.is_some_and(|s| matches!(s.lifecycle, crate::api::model::PlaybackLifecycle::PendingProvider { .. })), + "pre-resolved GracePeriod must materialize as PendingProvider lifecycle" + ); + } + + /// `activate_session_before_stream_open` skips placeholder for Prepare class. + #[tokio::test] + async fn activate_session_before_stream_open_skips_placeholder_for_prepare() { + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::EvictUserSameIpOldest]), + }); + let addr: SocketAddr = "127.0.0.1:55222".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let input = app_state.app_config.sources.load().inputs[0].clone(); + let mut user = ProxyUserCredentials::default(); + user.username = "prepare-user".to_string(); + user.max_connections = 1; + + let activation = activate_session_before_stream_open( + &app_state, + SessionActivationRequest { + fingerprint: &fingerprint, + input: input.as_ref(), + user: &user, + session_token: "tok-prepare", + // Explicitly pass Prepare class β€” placeholder and admission should be skipped. + request_class: Some(PlaybackRequestClass::Prepare), + virtual_id: 55222, + item_type: PlaylistItemType::LiveHls, + stream_url: "http://provider.example/live/test.ts", + connection_permission: UserConnectionPermission::Allowed, + connection_kind: crate::api::model::ConnectionKind::Normal, + socket_bound: true, + }, + ) + .await; + + // Prepare returns Allowed without running admission strategies. + assert_eq!(activation.admission.permission, UserConnectionPermission::Allowed); + assert_eq!(activation.grace_mode, None); + assert!( + activation.placeholder_transition_version.is_none(), + "Prepare activation must not create a placeholder session" + ); + } + + /// `resolve_playback_request_admission` with `prepare_only = true` returns `Prepare` class. + #[tokio::test] + async fn resolve_playback_request_admission_prepare_only_returns_prepare_class() { + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]), + }); + let addr: SocketAddr = "127.0.0.1:55223".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let mut user = ProxyUserCredentials::default(); + user.username = "prepare-only-user".to_string(); + user.max_connections = 1; + + let (admission, grace_mode, request_class) = resolve_playback_request_admission( + &app_state, + &user, + &fingerprint, + PlaylistItemType::LiveHls, + None, + "tok-prepare-only", + false, + EvictionReentryGuard::Session("tok-prepare-only"), + true, // prepare_only + false, // terminate + ) + .await; + + assert_eq!(request_class, PlaybackRequestClass::Prepare); + // Prepare returns Allowed without running strategies. + assert_eq!(admission.permission, UserConnectionPermission::Allowed); + assert_eq!(grace_mode, None); + } + + /// `resolve_playback_request_admission` with `terminate = true` returns `Terminate` class + /// and calls `terminate_session` on the existing session. + #[tokio::test] + async fn resolve_playback_request_admission_terminate_returns_terminate_class() { + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]), + }); + let addr: SocketAddr = "127.0.0.1:55224".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let mut user = ProxyUserCredentials::default(); + user.username = "terminate-user".to_string(); + user.max_connections = 2; + + // First create a session. + let session_token = "tok-terminate"; + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token, + virtual_id: 55224, + provider: "test-provider", + stream_url: "http://provider.example/test.ts", + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + + // Verify session exists. + let before = app_state + .active_users + .get_and_update_user_session(&user.username, session_token) + .await; + assert!(before.is_some(), "session should exist before terminate"); + + let (admission, grace_mode, request_class) = resolve_playback_request_admission( + &app_state, + &user, + &fingerprint, + PlaylistItemType::LiveHls, + before.as_ref(), + session_token, + false, + EvictionReentryGuard::Session(session_token), + false, // prepare_only + true, // terminate + ) + .await; + + assert_eq!(request_class, PlaybackRequestClass::Terminate); + assert_eq!(admission.permission, UserConnectionPermission::Exhausted); + assert_eq!(grace_mode, None); + + // Session should be expired after terminate. + let after = app_state + .active_users + .get_and_update_user_session(&user.username, session_token) + .await; + assert!(after.is_none(), "session should be removed after terminate"); + } + + /// `classify_playback_request` returns `Terminate` when `terminate = true`. + #[test] + fn classify_playback_request_returns_terminate_when_flag_set() { + let request_class = classify_playback_request(PlaybackRequestFacts { + item_type: PlaylistItemType::Live, + existing_session: None, + prepare_only: false, + terminate: true, + }); + assert_eq!(request_class, PlaybackRequestClass::Terminate); + } + + #[tokio::test] + async fn activate_session_before_stream_open_marks_pending_provider_for_grace_hold() { + let stream_cfg = crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]), + }; + let mut app_cfg = create_test_app_config(); + app_cfg.config = Arc::new(ArcSwap::from_pointee(Config { + user_access_control: true, + reverse_proxy: Some(crate::model::ReverseProxyConfig { + resource_rewrite_disabled: false, + rewrite_secret: [0; 16], + resource_retry: crate::model::ResourceRetryConfig::default(), + disabled_header: None, + stream: Some(stream_cfg), + cache: None, + rate_limit: None, + geoip: None, + stream_history: None, + qos_aggregation: None, + }), + ..Config::default() + })); + let app_state = create_test_app_state_for_config(Arc::new(app_cfg)); + let first_addr: SocketAddr = "127.0.0.1:55230".parse().unwrap_or_else(|_| unreachable!()); + let second_addr: SocketAddr = "127.0.0.1:55231".parse().unwrap_or_else(|_| unreachable!()); + let first_fingerprint = create_test_fingerprint(first_addr); + let second_fingerprint = create_test_fingerprint(second_addr); + let input = app_state.app_config.sources.load().inputs[0].clone(); + let mut user = ProxyUserCredentials::default(); + user.username = "grace-hold-user".to_string(); + user.max_connections = 1; + let first_channel = create_test_live_channel("http://provider-1.example/live/1.ts"); + let mut second_channel = create_test_live_channel("http://provider-1.example/live/2.m3u8"); + second_channel.item_type = PlaylistItemType::LiveHls; + second_channel.virtual_id = 55231; + + app_state.connection_manager.add_connection(&first_addr).await; + app_state + .connection_manager + .update_connection(crate::api::model::ConnectionParams { + meter_uid: 1, + username: &user.username, + max_connections: user.max_connections, + soft_connections: user.soft_connections, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 10, + fingerprint: &first_fingerprint, + provider: input.name.clone(), + stream_channel: &first_channel, + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-first"), + }) + .await; + + let activation = activate_session_before_stream_open( + &app_state, + SessionActivationRequest { + fingerprint: &second_fingerprint, + input: input.as_ref(), + user: &user, + session_token: "tok-grace-hold", + request_class: None, + virtual_id: second_channel.virtual_id, + item_type: PlaylistItemType::LiveHls, + stream_url: second_channel.url.as_ref(), + connection_permission: UserConnectionPermission::Allowed, + connection_kind: crate::api::model::ConnectionKind::Normal, + socket_bound: false, + }, + ) + .await; + + assert_eq!(activation.admission.permission, UserConnectionPermission::GracePeriod); + assert_eq!(activation.grace_mode, Some(crate::api::model::GraceMode::Hold)); + + let session = app_state + .active_users + .get_and_update_user_session(&user.username, "tok-grace-hold") + .await + .expect("placeholder session should exist"); + let crate::api::model::PlaybackLifecycle::PendingProvider { data: pending } = + &session.lifecycle + else { + panic!("grace hold should mark pending provider state") + }; + assert!(matches!(pending.reason_code, crate::api::model::PendingProviderReason::GraceHold)); + assert!(pending.deadline >= pending.created_at); + assert_eq!(app_state.active_users.user_connections(&user.username).await, 1); + assert!(!session.lifecycle.is_counted(), "pending provider placeholder must not consume an active user lease before commit"); + } + + #[tokio::test] + async fn activate_session_before_stream_open_does_not_commit_user_lease_before_provider_success() { + let stream_cfg = crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 0, + grace_period_timeout_secs: 8, + grace_period_hold_stream: false, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: None, + }; + let mut app_cfg = create_test_app_config(); + app_cfg.config = Arc::new(ArcSwap::from_pointee(Config { + user_access_control: true, + reverse_proxy: Some(crate::model::ReverseProxyConfig { + resource_rewrite_disabled: false, + rewrite_secret: [0; 16], + resource_retry: crate::model::ResourceRetryConfig::default(), + disabled_header: None, + stream: Some(stream_cfg), + cache: None, + rate_limit: None, + geoip: None, + stream_history: None, + qos_aggregation: None, + }), + ..Config::default() + })); + let app_state = create_test_app_state_for_config(Arc::new(app_cfg)); + let addr: SocketAddr = "127.0.0.1:55232".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let input = app_state.app_config.sources.load().inputs[0].clone(); + let mut user = ProxyUserCredentials::default(); + user.username = "atomic-commit-user".to_string(); + user.max_connections = 1; + let channel = create_test_live_channel("http://provider-1.example/live/3.ts"); + + let activation = activate_session_before_stream_open( + &app_state, + SessionActivationRequest { + fingerprint: &fingerprint, + input: input.as_ref(), + user: &user, + session_token: "tok-atomic-commit", + request_class: None, + virtual_id: channel.virtual_id, + item_type: channel.item_type, + stream_url: channel.url.as_ref(), + connection_permission: UserConnectionPermission::Allowed, + connection_kind: crate::api::model::ConnectionKind::Normal, + socket_bound: false, + }, + ) + .await; + + assert_eq!(activation.admission.permission, UserConnectionPermission::Allowed); + assert_eq!(activation.grace_mode, None); + + let session = app_state + .active_users + .get_and_update_user_session(&user.username, "tok-atomic-commit") + .await + .expect("placeholder session should exist"); + assert_eq!( + app_state.active_users.user_connections(&user.username).await, + 0, + "allowed activation should stay provisional until provider acquisition and stream commit succeed" + ); + assert!( + !session.lifecycle.is_counted(), + "placeholder session must stay uncounted until the provider side has been committed" + ); + assert!(!matches!(session.lifecycle, crate::api::model::PlaybackLifecycle::PendingProvider { .. })); + } + fn create_test_shared_target() -> ConfigTarget { ConfigTarget { id: 1, @@ -3288,7 +4710,215 @@ mod tests { } #[tokio::test] - async fn resolve_admission_with_strategies_falls_through_after_failed_grace_grant() { + async fn grace_context_is_populated_when_grace_strategy_is_actually_granted() { + // Use a DIFFERENT session token than the pre-existing counted session. + // Otherwise session-admission may treat it as a valid reopen and skip the exhausted path. + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::EvictUserOldest, + ]), + }); + + let addr1: SocketAddr = "127.0.0.1:55401".parse().unwrap_or_else(|_| unreachable!()); + let addr2: SocketAddr = "10.0.0.5:55402".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint1 = create_test_fingerprint(addr1); + let fingerprint2 = create_test_fingerprint(addr2); + // addr1 and addr2 have DIFFERENT IPs. + // EvictUserSameIpOldest will NOT match (different IP), so GraceHoldStream is evaluated. + let mut user = ProxyUserCredentials::default(); + user.username = "user-grace-ctx".to_string(); + user.max_connections = 1; + + // Register the connection first so update_connection succeeds + app_state.connection_manager.add_connection(&addr1).await; + + // Create the session β€” lifecycle starts as Prepared (uncounted) + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-existing-counted", + virtual_id: 55401, + provider: "provider_1", + stream_url: "http://provider-1.example/live/55401.m3u8", + addr: &addr1, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: true, + }) + .await; + + // update_connection promotes the session to Active (counted) and creates a stream. + // This exhausts the user's single slot (max_connections = 1). + app_state + .active_users + .update_connection(crate::api::model::ActiveUserConnectionParams { + uid: 55401, + meter_uid: 55401, + username: "user-grace-ctx", + max_connections: 1, + soft_connections: 0, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &fingerprint1, + provider: "provider_1".intern(), + stream_channel: &create_test_live_channel("http://provider-1.example/live/55401.m3u8"), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-existing-counted"), + }) + .await + .expect("stream should be created"); + + // Now the new request finds the slot exhausted and the grace strategy kicks in. + let result = resolve_admission_with_strategies( + &app_state, + &user.username, + user.max_connections, + user.soft_connections, + &fingerprint2.client_ip, + &fingerprint2.addr, + true, + Some("tok-new-request"), + true, + EvictionReentryGuard::Session("tok-new-request"), + ) + .await; + + assert_eq!(result.admission.permission, UserConnectionPermission::GracePeriod, "grace should be granted"); + assert!(matches!(result.grace_mode, Some(crate::api::model::GraceMode::Hold))); + let ctx = result.grace_context.expect("grace_context must be present when grace is granted"); + assert_eq!(ctx.strategy_index, 1, "GraceHoldStream is at index 1"); + assert_eq!(ctx.strategies.len(), 3); + assert!(matches!(ctx.strategies[ctx.strategy_index], AdmissionStrategy::GraceHoldStream)); + } + + #[tokio::test] + async fn evaluate_remaining_strategies_evicts_after_used_grace() { + // Strategies: [GraceHoldStream, EvictUserOldest] + // Grace was used at index 0, so only EvictUserOldest (index 1) is evaluated. + // Eviction frees the slot β†’ Allowed. + let strategies = vec![ + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::EvictUserOldest, + ]; + let grace_context = GraceResolutionContext { strategy_index: 0, strategies, kind: None }; + + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![ + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::EvictUserOldest, + ]), + }); + + let addr1: SocketAddr = "127.0.0.1:55701".parse().unwrap_or_else(|_| unreachable!()); + let addr2: SocketAddr = "10.0.0.5:55702".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint1 = create_test_fingerprint(addr1); + let fingerprint2 = create_test_fingerprint(addr2); + + app_state.connection_manager.add_connection(&addr1).await; + app_state.connection_manager.add_connection(&addr2).await; + + let mut user = ProxyUserCredentials::default(); + user.username = "remaining-evict".to_string(); + user.max_connections = 1; + + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-counted", + virtual_id: 55701, + provider: "provider-evict", + stream_url: "http://provider.example/live/1.ts", + addr: &addr1, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + + app_state + .active_users + .update_connection(crate::api::model::ActiveUserConnectionParams { + uid: 55701, + meter_uid: 55701, + username: "remaining-evict", + max_connections: 1, + soft_connections: 0, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &fingerprint1, + provider: "provider-evict".intern(), + stream_channel: &create_test_live_channel("http://provider.example/live/1.ts"), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-counted"), + }) + .await + .expect("stream should be created"); + + let result = evaluate_remaining_strategies_after_grace( + &app_state, + "remaining-evict", + 1, + 0, + &fingerprint2.client_ip, + &fingerprint2.addr, + true, + Some("tok-new"), + true, + EvictionReentryGuard::Session("tok-new"), + &grace_context, + Some(crate::api::model::ConnectionKind::Normal), + ) + .await; + + assert_eq!( + result.admission.permission, + UserConnectionPermission::Allowed, + "EvictUserOldest should free the slot" + ); + assert!(result.grace_context.is_none(), "no grace context on eviction success"); + } + + #[tokio::test] + async fn evaluate_remaining_strategies_skips_no_match_and_uses_later_eviction() { + // Strategies: [GraceHoldStream, EvictUserSameIpOldest, EvictUserOldest] + // Grace was at index 0, remaining are EvictUserSameIpOldest (index 1) and EvictUserOldest (index 2). + // The existing counted session is at a DIFFERENT IP, so EvictUserSameIpOldest β†’ NoMatch. + // EvictUserOldest succeeds β†’ Allowed. + let strategies = vec![ + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::EvictUserOldest, + ]; + let grace_context = GraceResolutionContext { strategy_index: 0, strategies, kind: None }; + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { retry: true, metrics_enabled: true, @@ -3304,6 +4934,463 @@ mod tests { admission_strategies: Some(vec![ AdmissionStrategy::GraceHoldStream, AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::EvictUserOldest, + ]), + }); + + let addr1: SocketAddr = "127.0.0.1:55801".parse().unwrap_or_else(|_| unreachable!()); + let addr2: SocketAddr = "10.0.0.5:55802".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint1 = create_test_fingerprint(addr1); + let fingerprint2 = create_test_fingerprint(addr2); + + app_state.connection_manager.add_connection(&addr1).await; + app_state.connection_manager.add_connection(&addr2).await; + + let mut user = ProxyUserCredentials::default(); + user.username = "remaining-skip-no-match".to_string(); + user.max_connections = 1; + + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-counted", + virtual_id: 55801, + provider: "provider-skip", + stream_url: "http://provider.example/live/1.ts", + addr: &addr1, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + + app_state + .active_users + .update_connection(crate::api::model::ActiveUserConnectionParams { + uid: 55801, + meter_uid: 55801, + username: "remaining-skip-no-match", + max_connections: 1, + soft_connections: 0, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &fingerprint1, + provider: "provider-skip".intern(), + stream_channel: &create_test_live_channel("http://provider.example/live/1.ts"), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-counted"), + }) + .await + .expect("stream should be created"); + + let result = evaluate_remaining_strategies_after_grace( + &app_state, + "remaining-skip-no-match", + 1, + 0, + &fingerprint2.client_ip, + &fingerprint2.addr, + true, + Some("tok-new"), + true, + EvictionReentryGuard::Session("tok-new"), + &grace_context, + Some(crate::api::model::ConnectionKind::Normal), + ) + .await; + + assert_eq!( + result.admission.permission, + UserConnectionPermission::Allowed, + "EvictUserSameIpOldest should NoMatch, EvictUserOldest should succeed" + ); + } + + #[tokio::test] + async fn evaluate_remaining_strategies_empty_slice_denies() { + // Strategies: [GraceHoldStream] + // Grace was at index 0, remaining slice is empty β†’ exhausted. + let strategies = vec![AdmissionStrategy::GraceHoldStream]; + let grace_context = GraceResolutionContext { strategy_index: 0, strategies, kind: None }; + + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]), + }); + + let addr: SocketAddr = "10.0.0.5:55901".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + + let result = evaluate_remaining_strategies_after_grace( + &app_state, + "no-remaining-strategies", + 1, + 0, + &fingerprint.client_ip, + &fingerprint.addr, + true, + Some("tok-new"), + true, + EvictionReentryGuard::Session("tok-new"), + &grace_context, + None, + ) + .await; + + assert_eq!( + result.admission.permission, + UserConnectionPermission::Exhausted, + "empty remaining slice should deny" + ); + } + + #[tokio::test] + async fn evaluate_remaining_strategies_preserves_soft_kind_on_exhausted() { + // Strategies: [GraceHoldStream] + // Grace was at index 0, remaining slice is empty β†’ exhausted. + // grace_context.kind is Soft β€” must be preserved in the exhausted result. + let strategies = vec![AdmissionStrategy::GraceHoldStream]; + let grace_context = GraceResolutionContext { strategy_index: 0, strategies, kind: Some(crate::api::model::ConnectionKind::Soft) }; + + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]), + }); + + let addr: SocketAddr = "10.0.0.6:55902".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + + let result = evaluate_remaining_strategies_after_grace( + &app_state, + "soft-kind-user", + 1, + 0, + &fingerprint.client_ip, + &fingerprint.addr, + true, + Some("tok-soft"), + true, + EvictionReentryGuard::Session("tok-soft"), + &grace_context, + Some(crate::api::model::ConnectionKind::Soft), + ) + .await; + + assert_eq!( + result.admission.permission, + UserConnectionPermission::Exhausted, + "empty remaining slice should deny" + ); + assert_eq!( + result.admission.kind, + Some(crate::api::model::ConnectionKind::Soft), + "exhausted result must preserve the original Soft connection kind" + ); + } + + #[tokio::test] + async fn evaluate_remaining_strategies_does_not_retry_used_prefix() { + // Strategies: [GraceHoldStream, GraceInstantStream, EvictUserOldest] + // Grace was at index 1 (GraceInstantStream). + // Remaining slice: [EvictUserOldest] (index 2). + // GraceHoldStream (index 0) must NOT be re-evaluated. + let strategies = vec![ + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::GraceInstantStream, + AdmissionStrategy::EvictUserOldest, + ]; + let strategies_for_config = strategies.clone(); + let grace_context = GraceResolutionContext { strategy_index: 1, strategies, kind: None }; + + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(strategies_for_config), + }); + + let addr1: SocketAddr = "127.0.0.1:56001".parse().unwrap_or_else(|_| unreachable!()); + let addr2: SocketAddr = "10.0.0.5:56002".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint1 = create_test_fingerprint(addr1); + let fingerprint2 = create_test_fingerprint(addr2); + + app_state.connection_manager.add_connection(&addr1).await; + app_state.connection_manager.add_connection(&addr2).await; + + let mut user = ProxyUserCredentials::default(); + user.username = "remaining-no-retry".to_string(); + user.max_connections = 1; + + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-counted", + virtual_id: 56001, + provider: "provider-no-retry", + stream_url: "http://provider.example/live/1.ts", + addr: &addr1, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + + app_state + .active_users + .update_connection(crate::api::model::ActiveUserConnectionParams { + uid: 56001, + meter_uid: 56001, + username: "remaining-no-retry", + max_connections: 1, + soft_connections: 0, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &fingerprint1, + provider: "provider-no-retry".intern(), + stream_channel: &create_test_live_channel("http://provider.example/live/1.ts"), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-counted"), + }) + .await + .expect("stream should be created"); + + let result = evaluate_remaining_strategies_after_grace( + &app_state, + "remaining-no-retry", + 1, + 0, + &fingerprint2.client_ip, + &fingerprint2.addr, + true, + Some("tok-new"), + true, + EvictionReentryGuard::Session("tok-new"), + &grace_context, + Some(crate::api::model::ConnectionKind::Normal), + ) + .await; + + assert_eq!( + result.admission.permission, + UserConnectionPermission::Allowed, + "only EvictUserOldest should be evaluated, not GraceHoldStream" + ); + } + + #[tokio::test] + async fn evaluate_remaining_strategies_empty_slice_uses_original_kind_not_context_kind() { + // grace_context.kind = Normal, original_kind = Soft + // remaining slice is empty β†’ exhausted result must use original_kind. + // This proves the empty-slice branch uses original_kind, not grace_context.kind. + let strategies = vec![AdmissionStrategy::GraceHoldStream]; + let grace_context = GraceResolutionContext { strategy_index: 0, strategies, kind: Some(crate::api::model::ConnectionKind::Normal) }; + let original_kind = Some(crate::api::model::ConnectionKind::Soft); + + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]), + }); + + let addr: SocketAddr = "10.0.0.7:55903".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + + let result = evaluate_remaining_strategies_after_grace( + &app_state, + "kind-mismatch-empty", + 1, + 0, + &fingerprint.client_ip, + &fingerprint.addr, + true, + Some("tok-empty"), + true, + EvictionReentryGuard::Session("tok-empty"), + &grace_context, + original_kind, + ) + .await; + + assert_eq!(result.admission.permission, UserConnectionPermission::Exhausted); + assert_eq!( + result.admission.kind, + original_kind, + "exhausted result must use original_kind (Soft), not grace_context.kind (Normal)" + ); + } + + #[tokio::test] + async fn evaluate_remaining_strategies_later_grace_uses_original_kind_not_context_kind() { + // grace_context.kind = Normal, original_kind = Soft + // Strategies: [GraceHoldStream, GraceInstantStream] + // Grace was used at index 0 (GraceHoldStream). + // Remaining slice contains GraceInstantStream (index 1). + // When the helper returns Grace for the remaining strategy, the new + // GraceResolutionContext.kind must be original_kind (Soft), not grace_context.kind (Normal). + // This proves build_grace_ctx uses original_kind as source of truth. + let strategies = vec![ + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::GraceInstantStream, + ]; + let grace_context = GraceResolutionContext { strategy_index: 0, strategies, kind: Some(crate::api::model::ConnectionKind::Normal) }; + let original_kind = Some(crate::api::model::ConnectionKind::Soft); + + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![ + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::GraceInstantStream, + ]), + }); + + let addr1: SocketAddr = "127.0.0.1:55710".parse().unwrap_or_else(|_| unreachable!()); + let addr2: SocketAddr = "10.0.0.8:55711".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint1 = create_test_fingerprint(addr1); + let fingerprint2 = create_test_fingerprint(addr2); + + app_state.connection_manager.add_connection(&addr1).await; + app_state.connection_manager.add_connection(&addr2).await; + + let mut user = ProxyUserCredentials::default(); + user.username = "kind-mismatch-grace".to_string(); + user.max_connections = 1; + + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-counted-grace", + virtual_id: 55710, + provider: "provider-grace-kind", + stream_url: "http://provider.example/live/1.ts", + addr: &addr1, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + + app_state + .active_users + .update_connection(crate::api::model::ActiveUserConnectionParams { + uid: 55710, + meter_uid: 55710, + username: "kind-mismatch-grace", + max_connections: 1, + soft_connections: 0, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &fingerprint1, + provider: "provider-grace-kind".intern(), + stream_channel: &create_test_live_channel("http://provider.example/live/1.ts"), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-counted-grace"), + }) + .await + .expect("stream should be created"); + + let result = evaluate_remaining_strategies_after_grace( + &app_state, + "kind-mismatch-grace", + 1, + 0, + &fingerprint2.client_ip, + &fingerprint2.addr, + true, + Some("tok-new-grace"), + true, + EvictionReentryGuard::Session("tok-new-grace"), + &grace_context, + original_kind, + ) + .await; + + assert_eq!( + result.admission.permission, + UserConnectionPermission::GracePeriod, + "remaining GraceInstantStream should grant GracePeriod" + ); + assert!( + result.grace_context.is_some(), + "grace_context must be present when grace is granted" + ); + assert_eq!( + result.grace_context.as_ref().unwrap().kind, + original_kind, + "GraceResolutionContext.kind in the result must be original_kind (Soft), not grace_context.kind (Normal)" + ); + } + + #[tokio::test] + #[allow(clippy::too_many_lines)] + async fn resolve_admission_with_strategies_falls_through_after_failed_grace_grant() { + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![ + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::EvictUserOldest, ]), }); @@ -3315,6 +5402,26 @@ mod tests { app_state.connection_manager.add_connection(&first_addr).await; app_state.connection_manager.add_connection(&second_addr).await; + let mut session_user = ProxyUserCredentials::default(); + session_user.username = "fallthrough".to_string(); + session_user.max_connections = 1; + session_user.soft_connections = 1; + + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &session_user, + session_token: "tok-first", + virtual_id: 1, + provider: "provider-a", + stream_url: "http://provider-1.example/live/1.ts", + addr: &first_addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + app_state .connection_manager .update_connection(crate::api::model::ConnectionParams { @@ -3335,6 +5442,21 @@ mod tests { assert!(app_state.active_users.grant_grace("fallthrough").await); + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &session_user, + session_token: "tok-second", + virtual_id: 2, + provider: "provider-a", + stream_url: "http://provider-1.example/live/2.ts", + addr: &second_addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Soft), + socket_bound: false, + }) + .await; + app_state .connection_manager .update_connection(crate::api::model::ConnectionParams { @@ -3353,7 +5475,7 @@ mod tests { }) .await; - let (admission, grace_mode) = resolve_admission_with_strategies( + let result = resolve_admission_with_strategies( &app_state, "fallthrough", 1, @@ -3366,9 +5488,11 @@ mod tests { EvictionReentryGuard::Session("tok-third"), ) .await; + let admission = result.admission; + let grace_mode = result.grace_mode; assert_eq!(admission.permission, UserConnectionPermission::Allowed); - assert_eq!(admission.kind, Some(crate::api::model::ConnectionKind::Soft)); + assert_eq!(admission.kind, Some(crate::api::model::ConnectionKind::Normal)); assert_eq!(grace_mode, None); } @@ -3427,7 +5551,7 @@ mod tests { EvictionReentryGuard::Session("vod-session"), ) .await; - assert_eq!(session_based.0.permission, UserConnectionPermission::Allowed); + assert_eq!(session_based.admission.permission, UserConnectionPermission::Allowed); let connection_based = resolve_admission_with_strategies( &app_state, @@ -3442,7 +5566,7 @@ mod tests { EvictionReentryGuard::Session("vod-session"), ) .await; - assert_eq!(connection_based.0.permission, UserConnectionPermission::Exhausted); + assert_eq!(connection_based.admission.permission, UserConnectionPermission::Exhausted); } #[tokio::test] @@ -3517,7 +5641,7 @@ mod tests { .await; app_state.connection_manager.release_connection_as_kicked(&victim_addr).await; - let (admission, grace_mode) = resolve_admission_with_strategies( + let result = resolve_admission_with_strategies( &app_state, "loop-user", 1, @@ -3530,6 +5654,8 @@ mod tests { EvictionReentryGuard::SocketPlayback { virtual_id: 9001 }, ) .await; + let admission = result.admission; + let grace_mode = result.grace_mode; assert_eq!(admission.permission, UserConnectionPermission::Exhausted); assert_eq!(grace_mode, None); @@ -3539,6 +5665,7 @@ mod tests { } #[tokio::test] + #[allow(clippy::too_many_lines)] async fn resolve_admission_with_strategies_allows_other_channel_after_recent_eviction() { let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { retry: true, @@ -3565,10 +5692,42 @@ mod tests { victim_channel.virtual_id = 9101; let mut winner_channel = create_test_live_channel("http://provider-1.example/live/9102.ts"); winner_channel.virtual_id = 9102; + let mut session_user = ProxyUserCredentials::default(); + session_user.username = "loop-user-2".to_string(); app_state.connection_manager.add_connection(&victim_addr).await; app_state.connection_manager.add_connection(&winner_addr).await; + // Create sessions before update_connection so streams are linked to counted sessions + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &session_user, + session_token: "session-victim", + virtual_id: 9101, + provider: "provider-a", + stream_url: "http://provider-1.example/live/9101.ts", + addr: &victim_addr, + connection_permission: shared::model::UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &session_user, + session_token: "session-winner", + virtual_id: 9102, + provider: "provider-a", + stream_url: "http://provider-1.example/live/9102.ts", + addr: &winner_addr, + connection_permission: shared::model::UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + app_state .connection_manager .update_connection(crate::api::model::ConnectionParams { @@ -3610,7 +5769,7 @@ mod tests { .await; app_state.connection_manager.release_connection_as_kicked(&victim_addr).await; - let (admission, _grace_mode) = resolve_admission_with_strategies( + let result = resolve_admission_with_strategies( &app_state, "loop-user-2", 1, @@ -3623,12 +5782,14 @@ mod tests { EvictionReentryGuard::SocketPlayback { virtual_id: 9103 }, ) .await; + let admission = result.admission; assert_eq!(admission.permission, UserConnectionPermission::Allowed); assert!(app_state.active_users.active_streams().await.is_empty()); } #[tokio::test] + #[allow(clippy::too_many_lines)] async fn resolve_admission_with_strategies_does_not_suppress_different_session_on_same_channel() { let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { retry: true, @@ -3654,10 +5815,42 @@ mod tests { let mut channel = create_test_live_channel("http://provider-1.example/live/9301.m3u8"); channel.virtual_id = 9301; channel.item_type = PlaylistItemType::LiveHls; + let mut session_user = ProxyUserCredentials::default(); + session_user.username = "loop-user-4".to_string(); app_state.connection_manager.add_connection(&victim_addr).await; app_state.connection_manager.add_connection(&winner_addr).await; + // Create sessions before update_connection so streams are linked to counted sessions + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &session_user, + session_token: "session-victim", + virtual_id: 9301, + provider: "provider-a", + stream_url: "http://provider-1.example/live/9301.m3u8", + addr: &victim_addr, + connection_permission: shared::model::UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &session_user, + session_token: "session-winner", + virtual_id: 9301, + provider: "provider-a", + stream_url: "http://provider-1.example/live/9301.m3u8", + addr: &winner_addr, + connection_permission: shared::model::UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + app_state .connection_manager .update_connection(crate::api::model::ConnectionParams { @@ -3699,7 +5892,7 @@ mod tests { .await; app_state.connection_manager.release_connection_as_kicked(&victim_addr).await; - let (admission, grace_mode) = resolve_admission_with_strategies( + let result = resolve_admission_with_strategies( &app_state, "loop-user-4", 1, @@ -3712,6 +5905,8 @@ mod tests { EvictionReentryGuard::Session("session-other"), ) .await; + let admission = result.admission; + let grace_mode = result.grace_mode; assert_eq!(admission.permission, UserConnectionPermission::Allowed); assert_eq!(grace_mode, None); @@ -3790,7 +5985,7 @@ mod tests { .await; app_state.connection_manager.release_connection_as_kicked(&victim_addr).await; - let (admission, grace_mode) = resolve_admission_with_strategies( + let result = resolve_admission_with_strategies( &app_state, "loop-user-3", 1, @@ -3803,6 +5998,8 @@ mod tests { EvictionReentryGuard::SocketPlayback { virtual_id: 9201 }, ) .await; + let admission = result.admission; + let grace_mode = result.grace_mode; assert_eq!(admission.permission, UserConnectionPermission::Allowed); assert_eq!(admission.kind, Some(crate::api::model::ConnectionKind::Soft)); @@ -3853,6 +6050,7 @@ mod tests { UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Normal, None, + None, false, ) .await @@ -3912,6 +6110,7 @@ mod tests { UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Normal, Some(&first_token), + None, false, ) .await @@ -3928,6 +6127,7 @@ mod tests { UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Normal, Some(&second_token), + None, false, ) .await @@ -4021,6 +6221,7 @@ mod tests { &soft_fingerprint, &app_state, "soft-session", + None, create_test_live_channel(stream_url), stream_url, &HeaderMap::default(), @@ -4043,6 +6244,78 @@ mod tests { assert_eq!(session_admission.kind, Some(crate::api::model::ConnectionKind::Soft)); } + #[tokio::test] + async fn stream_response_rolls_back_provisional_user_activation_when_provider_open_fails() { + let mut app_cfg = create_test_provider_app_config(); + app_cfg.config = Arc::new(ArcSwap::from_pointee(Config { + user_access_control: true, + ..Config::default() + })); + let app_state = create_test_app_state_for_config(Arc::new(app_cfg)); + let addr = "127.0.0.1:55143".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let input_name = "provider_1".intern(); + let input = app_state + .app_config + .get_input_by_name(&input_name) + .expect("provider input should exist"); + let target = Arc::new(ConfigTarget { + id: 1, + enabled: true, + name: "test".to_string(), + options: None, + sort: None, + filter: Filter::default(), + output: Vec::new(), + rename: None, + mapping_ids: None, + mapping: Arc::new(ArcSwapOption::default()), + favourites: None, + processing_order: ProcessingOrder::default(), + watch: None, + use_memory_cache: false, + }); + let mut user = ProxyUserCredentials::default(); + user.username = "rollback-user".to_string(); + user.max_connections = 1; + let stream_url = "provider://bad-url"; + let channel = create_test_live_channel(stream_url); + + let response = stream_response( + &fingerprint, + &app_state, + "rollback-session", + None, + channel, + stream_url, + &HeaderMap::default(), + &input, + &target, + &user, + UserConnectionPermission::Allowed, + crate::api::model::ConnectionKind::Normal, + false, + None, + ) + .await + .into_response(); + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!( + app_state.active_users.user_connections(&user.username).await, + 0, + "failed provider open must rollback provisional user activation" + ); + assert!( + app_state + .active_users + .get_and_update_user_session(&user.username, "rollback-session") + .await + .is_none(), + "failed provider open must remove the provisional placeholder session" + ); + } + #[tokio::test] async fn local_stream_response_disables_response_compression() { let app_state = create_test_app_state(); @@ -4083,6 +6356,7 @@ mod tests { UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Normal, None, + None, false, ) .await @@ -4133,6 +6407,7 @@ mod tests { UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Normal, Some(playback_session_token), + None, false, ) .await @@ -4149,6 +6424,7 @@ mod tests { UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Normal, Some(playback_session_token), + None, false, ) .await @@ -4204,6 +6480,7 @@ mod tests { UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Normal, Some(playback_session_token), + None, false, ) .await @@ -4220,6 +6497,7 @@ mod tests { UserConnectionPermission::Exhausted, crate::api::model::ConnectionKind::Normal, Some(playback_session_token), + None, false, ) .await @@ -4278,6 +6556,7 @@ mod tests { UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Soft, Some(playback_session_token), + None, false, ) .await @@ -4294,6 +6573,7 @@ mod tests { UserConnectionPermission::Exhausted, crate::api::model::ConnectionKind::Normal, Some(playback_session_token), + None, false, ) .await @@ -4309,7 +6589,7 @@ mod tests { } #[tokio::test] - async fn activated_session_admission_reserves_hls_slots_via_api_utils() { + async fn activated_session_admission_keeps_hls_placeholders_uncounted_via_api_utils() { let app_state = create_test_app_state(); let mut user = ProxyUserCredentials::default(); user.username = "hls-user".to_string(); @@ -4361,8 +6641,7 @@ mod tests { true, EvictionReentryGuard::Session("tok-hls-first"), ) - .await - .0; + .await; let second_admission = resolve_admission_with_strategies( &app_state, &user.username, @@ -4375,11 +6654,11 @@ mod tests { true, EvictionReentryGuard::Session("tok-hls-second"), ) - .await - .0; + .await; - assert_eq!(first_admission.permission, UserConnectionPermission::Allowed); - assert_eq!(second_admission.permission, UserConnectionPermission::Exhausted); + assert_eq!(first_admission.admission.permission, UserConnectionPermission::Allowed); + assert_eq!(second_admission.admission.permission, UserConnectionPermission::Allowed); + assert_eq!(app_state.active_users.user_connections(&user.username).await, 0); } #[tokio::test] @@ -4440,6 +6719,119 @@ mod tests { assert_eq!(admission.permission, UserConnectionPermission::Exhausted); } + #[tokio::test] + #[allow(clippy::too_many_lines)] + async fn resolve_admission_with_strategies_evicts_preserved_hls_session_for_same_user_ts_request() { + let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::EvictUserSameIpLatest, + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::EvictUserOldest, + AdmissionStrategy::EvictUserLatest, + ]), + }); + + let hls_addr: std::net::SocketAddr = "127.0.0.1:55176".parse().unwrap_or_else(|_| unreachable!()); + let ts_addr: std::net::SocketAddr = "127.0.0.1:55177".parse().unwrap_or_else(|_| unreachable!()); + let hls_fingerprint = create_test_fingerprint_with_user_agent(hls_addr, "player/1.0"); + let ts_fingerprint = create_test_fingerprint_with_user_agent(ts_addr, "player/1.0"); + let mut user = ProxyUserCredentials::default(); + user.username = "same-user".to_string(); + user.max_connections = 1; + + app_state.connection_manager.add_connection(&hls_addr).await; + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: "tok-hls-preserved", + virtual_id: 5001, + provider: "provider-a", + stream_url: "http://provider-1.example/live/5001.m3u8", + addr: &hls_addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + app_state + .connection_manager + .update_connection(crate::api::model::ConnectionParams { + meter_uid: 1, + username: &user.username, + max_connections: user.max_connections, + soft_connections: user.soft_connections, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &hls_fingerprint, + provider: "provider-a".intern(), + stream_channel: &StreamChannel { + item_type: PlaylistItemType::LiveHls, + virtual_id: 5001, + ..create_test_live_channel("http://provider-1.example/live/5001.m3u8") + }, + user_agent: std::borrow::Cow::Borrowed("player/1.0"), + session_token: Some("tok-hls-preserved"), + }) + .await; + + app_state.connection_manager.release_connection(&hls_addr).await; + assert_eq!(app_state.active_users.user_connections(&user.username).await, 0); + assert!(app_state.active_users.active_streams().await.is_empty()); + + let mut close_rx = app_state.connection_manager.get_close_connection_channel(); + let result = resolve_admission_with_strategies( + &app_state, + &user.username, + user.max_connections, + user.soft_connections, + &ts_fingerprint.client_ip, + &ts_fingerprint.addr, + false, + None, + false, + EvictionReentryGuard::SocketPlayback { virtual_id: 5001 }, + ) + .await; + let admission = result.admission; + let grace_mode = result.grace_mode; + + assert_eq!(admission.permission, UserConnectionPermission::Allowed); + assert_eq!(grace_mode, None); + assert!(app_state.active_users.active_streams().await.is_empty()); + assert_eq!( + tokio::time::timeout(std::time::Duration::from_millis(100), close_rx.recv()) + .await + .ok() + .and_then(Result::ok), + Some(crate::api::model::CloseConnectionSignal::WithReason( + hls_addr, + shared::model::DisconnectReason::ClientKicked, + )) + ); + assert!( + app_state + .active_users + .get_and_update_user_session(&user.username, "tok-hls-preserved") + .await + .is_none(), + "preserved session should be removed once the TS request evicts it" + ); + } + #[tokio::test] async fn socket_bound_playback_tokens_still_allow_soft_slots() { let app_state = create_test_app_state(); @@ -4563,6 +6955,203 @@ mod tests { assert!(socket_bound.contains(&fingerprint.addr.to_string())); } + #[tokio::test] + #[allow(clippy::too_many_lines)] + async fn xtream_hls_then_ts_uses_distinct_tokens_and_evicts_old_hls_session() { + let mut app_cfg = create_test_app_config(); + let config = Config { + user_access_control: true, + reverse_proxy: Some(crate::model::ReverseProxyConfig { + resource_rewrite_disabled: false, + rewrite_secret: [0; 16], + resource_retry: crate::model::ResourceRetryConfig::default(), + disabled_header: None, + stream: Some(crate::model::StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 8, + grace_period_hold_stream: true, + hls_session_ttl_secs: 10, + catchup_session_ttl_secs: 10, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::EvictUserSameIpLatest, + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::EvictUserOldest, + AdmissionStrategy::EvictUserLatest, + ]), + }), + cache: None, + rate_limit: None, + geoip: None, + stream_history: None, + qos_aggregation: None, + }), + ..Config::default() + }; + app_cfg.config = Arc::new(ArcSwap::from_pointee(config)); + let app_state = create_test_app_state_for_config(Arc::new(app_cfg)); + let hls_addr: SocketAddr = "127.0.0.1:55186".parse().unwrap_or_else(|_| unreachable!()); + let ts_addr: SocketAddr = "127.0.0.1:55187".parse().unwrap_or_else(|_| unreachable!()); + let hls_fingerprint = create_test_fingerprint_with_user_agent(hls_addr, "libmpv"); + let ts_fingerprint = create_test_fingerprint_with_user_agent(ts_addr, "libmpv"); + let mut user = ProxyUserCredentials::default(); + user.username = "xtream-hls-ts".to_string(); + user.max_connections = 1; + + let virtual_id = 7811; + let hls_token = create_session_fingerprint(&hls_fingerprint, &user.username, virtual_id, false); + let ts_token = create_session_fingerprint(&ts_fingerprint, &user.username, virtual_id, true); + assert_ne!(hls_token, ts_token, "Xtream .m3u8 and .ts must not share the same playback token"); + + let mut hls_channel = create_test_live_channel("http://provider-1.example/live/7811.m3u8"); + hls_channel.virtual_id = virtual_id; + hls_channel.item_type = PlaylistItemType::LiveHls; + let mut ts_channel = create_test_live_channel("http://provider-1.example/live/7811.ts"); + ts_channel.virtual_id = virtual_id; + + app_state.connection_manager.add_connection(&hls_addr).await; + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: &hls_token, + virtual_id, + provider: "provider_1", + stream_url: hls_channel.url.as_ref(), + addr: &hls_addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(crate::api::model::ConnectionKind::Normal), + socket_bound: false, + }) + .await; + app_state + .connection_manager + .update_connection(crate::api::model::ConnectionParams { + meter_uid: 1, + username: &user.username, + max_connections: user.max_connections, + soft_connections: user.soft_connections, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &hls_fingerprint, + provider: "provider_1".intern(), + stream_channel: &hls_channel, + user_agent: Cow::Borrowed("libmpv"), + session_token: Some(&hls_token), + }) + .await; + + app_state.connection_manager.release_connection(&hls_addr).await; + assert!( + app_state + .active_users + .get_and_update_user_session(&user.username, &hls_token) + .await + .is_some(), + "preserved HLS session should still exist before the competing TS request" + ); + assert_eq!( + app_state + .active_users + .connection_admission(&user.username, user.max_connections, user.soft_connections) + .await + .permission, + UserConnectionPermission::Exhausted, + "the preserved HLS playback must still reserve the user's only slot before the TS request is evaluated" + ); + assert_eq!( + app_state + .active_users + .get_eviction_candidates(&user.username, &ts_fingerprint.client_ip) + .await + .len(), + 1, + "the preserved HLS playback should be the single eviction candidate for the competing TS request" + ); + + let (ts_admission, ts_grace_mode, request_class) = resolve_playback_request_admission( + &app_state, + &user, + &ts_fingerprint, + PlaylistItemType::Live, + None, + &ts_token, + false, + EvictionReentryGuard::SocketPlayback { virtual_id }, + false, + false, + ) + .await; + assert_eq!(request_class, PlaybackRequestClass::Activate); + assert_eq!(ts_admission.permission, UserConnectionPermission::Allowed); + assert_eq!(ts_grace_mode, None); + assert!( + app_state + .active_users + .get_and_update_user_session(&user.username, &hls_token) + .await + .is_none(), + "the competing TS activation must remove the old preserved HLS session even though there is no live socket left to kick" + ); + + app_state.connection_manager.add_connection(&ts_addr).await; + app_state + .active_users + .create_user_session(crate::api::model::CreateUserSessionParams { + user: &user, + session_token: &ts_token, + virtual_id, + provider: "provider_1", + stream_url: ts_channel.url.as_ref(), + addr: &ts_addr, + connection_permission: ts_admission.permission, + connection_kind: ts_admission.kind, + socket_bound: true, + }) + .await; + app_state + .connection_manager + .update_connection(crate::api::model::ConnectionParams { + meter_uid: 2, + username: &user.username, + max_connections: user.max_connections, + soft_connections: user.soft_connections, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &ts_fingerprint, + provider: "provider_1".intern(), + stream_channel: &ts_channel, + user_agent: Cow::Borrowed("libmpv"), + session_token: Some(&ts_token), + }) + .await; + + assert!( + app_state + .active_users + .get_and_update_user_session(&user.username, &hls_token) + .await + .is_none(), + "after the competing TS request, the old Xtream HLS session must be gone so later /hls segment fetches cannot revive it" + ); + assert!( + app_state + .active_users + .get_and_update_user_session(&user.username, &ts_token) + .await + .is_some(), + "the winning TS playback should remain tracked under its socket-bound Xtream token" + ); + } + #[test] fn socket_bound_playback_session_matches_only_plain_live_playback() { assert!(is_socket_bound_playback_session(PlaylistItemType::Live, None)); @@ -4581,6 +7170,7 @@ mod tests { let seek: SocketAddr = "127.0.0.1:55193".parse().unwrap_or_else(|_| unreachable!()); let session = UserSession { token: "tok-vod".to_string(), + transition_version: 1, virtual_id: 9001, provider: "provider-a".intern(), stream_url: "http://localhost/movie.mkv".intern(), @@ -4591,7 +7181,7 @@ mod tests { started_at: 1, permission: UserConnectionPermission::Allowed, connection_kind: Some(crate::api::model::ConnectionKind::Normal), - counted: true, + lifecycle: crate::api::model::PlaybackLifecycle::Active, }; assert_eq!( @@ -4634,6 +7224,39 @@ mod tests { )); } + #[tokio::test] + async fn forced_reopen_cleanup_for_adaptive_streams_does_not_close_client_socket() { + let app_state = create_test_app_state(); + let addr: SocketAddr = "127.0.0.1:55220".parse().unwrap_or_else(|_| unreachable!()); + let mut close_rx = app_state.connection_manager.get_close_connection_channel(); + + cleanup_forced_reopen_addrs(&app_state, PlaylistItemType::LiveHls, &[addr]).await; + + let signal = tokio::time::timeout(std::time::Duration::from_millis(50), close_rx.recv()) + .await + .ok() + .and_then(Result::ok); + assert!(signal.is_none(), "adaptive cleanup should not hard-close the previous client socket"); + } + + #[tokio::test] + async fn forced_reopen_cleanup_for_non_adaptive_streams_closes_client_socket() { + let app_state = create_test_app_state(); + let addr: SocketAddr = "127.0.0.1:55221".parse().unwrap_or_else(|_| unreachable!()); + let mut close_rx = app_state.connection_manager.get_close_connection_channel(); + + cleanup_forced_reopen_addrs(&app_state, PlaylistItemType::Live, &[addr]).await; + + let signal = tokio::time::timeout(std::time::Duration::from_millis(50), close_rx.recv()) + .await + .ok() + .and_then(Result::ok); + assert!(matches!( + signal, + Some(crate::api::model::CloseConnectionSignal::WithReason(signal_addr, _)) if signal_addr == addr + )); + } + #[tokio::test] async fn get_query_path_strips_extension_for_live_with_flag() { use crate::model::ConfigInputFlags; diff --git a/backend/src/api/config_file.rs b/backend/src/api/config_file.rs index e0df3025d..1d039ace2 100644 --- a/backend/src/api/config_file.rs +++ b/backend/src/api/config_file.rs @@ -295,11 +295,11 @@ impl ConfigFile { // but without touching any live state. If anything fails, we return an error // and the currently-running state remains completely unchanged. let follow_up: PreparedFollowUp = if template_changed { - // Template path changed β†’ sources depend on new templates, reload everything. + // Template path changed -> sources depend on new templates, reload everything. let prepared = Self::prepare_sources_reload_with_config(&config, &effective_paths).await?; PreparedFollowUp::Sources(prepared) } else if mapping_changed { - // Only mapping path changed; templates are the same β†’ load templates once. + // Only mapping path changed; templates are the same -> load templates once. let prepared_templates = Self::load_prepared_global_templates_with_config(&effective_paths, &config).await?; let prepared = Self::prepare_mapping_reload( effective_paths.mapping_file_path.as_deref(), diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index b3c030b82..b349c340f 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -269,14 +269,36 @@ pub(in crate::api) async fn handle_hls_stream_request( user_token: session_token.as_deref(), }; let hls_content = rewrite_hls(user, &rewrite_hls_props); + if let Some(session_token) = session_token.as_deref() { + let _transition_guard = app_state + .active_users + .acquire_playback_transition(&user.username, session_token) + .await; + app_state + .active_users + .release_unbound_session_reservation(&user.username, session_token, None, false) + .await; + app_state + .active_users + .clear_unbound_session_addr(&user.username, session_token, &fingerprint.addr) + .await; + } hls_response(hls_content).into_response() } Err(err) => { error!("Failed to download m3u8 {}", sanitize_sensitive_info(err.to_string().as_str())); if let Some(session_token) = session_token.as_deref() { + let _transition_guard = app_state + .active_users + .acquire_playback_transition(&user.username, session_token) + .await; app_state .active_users - .release_unbound_session_reservation(&user.username, session_token, false) + .release_unbound_session_reservation(&user.username, session_token, None, false) + .await; + app_state + .active_users + .clear_unbound_session_addr(&user.username, session_token, &fingerprint.addr) .await; } @@ -463,31 +485,22 @@ async fn hls_api_stream( return axum::http::StatusCode::BAD_REQUEST.into_response(); } - let (connection_admission, grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0) - && app_state.app_config.config.load().user_access_control - { - crate::api::api_utils::resolve_admission_with_strategies( + let (connection_admission, grace_mode, request_class) = + crate::api::api_utils::resolve_playback_request_admission( &app_state, - &user.username, - user.max_connections, - user.soft_connections, - &fingerprint.client_ip, - &fingerprint.addr, - true, - Some(&session.token), + &user, + &fingerprint, + PlaylistItemType::LiveHls, + Some(session), + &session.token, true, crate::api::api_utils::EvictionReentryGuard::Session(&session.token), + // HLS playlist requests (.m3u8) are explicit Prepare: they set up session metadata + // but do not consume an admission slot. Segment and other media requests use Activate. + is_hls_url(&hls_url), + false, ) - .await - } else { - ( - crate::api::model::ConnectionAdmission { - permission: UserConnectionPermission::Allowed, - kind: session.connection_kind, - }, - None, - ) - }; + .await; let connection_permission = connection_admission.permission; let connection_kind = connection_admission .kind @@ -543,6 +556,7 @@ async fn hls_api_stream( connection_permission, connection_kind, Some(&session.token), + Some(request_class), false, ) .await diff --git a/backend/src/api/endpoints/m3u_api.rs b/backend/src/api/endpoints/m3u_api.rs index aed26b585..b1cd93f64 100644 --- a/backend/src/api/endpoints/m3u_api.rs +++ b/backend/src/api/endpoints/m3u_api.rs @@ -134,33 +134,25 @@ async fn m3u_api_stream( if pli.item_type.is_local() { let playback_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id, true); - let (admission, _grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0) - && app_state.app_config.config.load().user_access_control - { - crate::api::api_utils::resolve_admission_with_strategies( - app_state, - &user.username, - user.max_connections, - user.soft_connections, - &fingerprint.client_ip, - &fingerprint.addr, - true, - Some(playback_session_token.as_str()), - false, - crate::api::api_utils::EvictionReentryGuard::SocketPlayback { - virtual_id: pli.virtual_id, - }, - ) - .await - } else { - ( - crate::api::model::ConnectionAdmission { - permission: UserConnectionPermission::Allowed, - kind: Some(crate::api::model::ConnectionKind::Normal), - }, - None, - ) - }; + let user_session = app_state + .active_users + .get_and_update_user_session(&user.username, &playback_session_token) + .await; + let (admission, _grace_mode, request_class) = crate::api::api_utils::resolve_playback_request_admission( + app_state, + &user, + fingerprint, + pli.item_type, + user_session.as_ref(), + playback_session_token.as_str(), + false, + crate::api::api_utils::EvictionReentryGuard::SocketPlayback { + virtual_id: pli.virtual_id, + }, + false, + false, + ) + .await; return local_stream_response( fingerprint, app_state, @@ -172,6 +164,7 @@ async fn m3u_api_stream( admission.permission, admission.kind.unwrap_or(crate::api::model::ConnectionKind::Normal), Some(playback_session_token.as_str()), + Some(request_class), true, ) .await @@ -251,31 +244,19 @@ async fn m3u_api_stream( pli.url.clone() }; - let (connection_admission, grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0) - && app_state.app_config.config.load().user_access_control - { - crate::api::api_utils::resolve_admission_with_strategies( - app_state, - &user.username, - user.max_connections, - user.soft_connections, - &fingerprint.client_ip, - &fingerprint.addr, - true, - Some(&session_key), - false, - eviction_reentry_guard, - ) - .await - } else { - ( - crate::api::model::ConnectionAdmission { - permission: UserConnectionPermission::Allowed, - kind: user_session.as_ref().and_then(|session| session.connection_kind), - }, - None, - ) - }; + let (connection_admission, grace_mode, request_class) = crate::api::api_utils::resolve_playback_request_admission( + app_state, + &user, + fingerprint, + pli.item_type, + user_session.as_ref(), + &session_key, + false, + eviction_reentry_guard, + false, + false, + ) + .await; let connection_permission = connection_admission.permission; let connection_kind = connection_admission .kind @@ -341,6 +322,7 @@ async fn m3u_api_stream( fingerprint, app_state, &session_key, + Some(request_class), pli.to_stream_channel(target.id), &session_url, req_headers, diff --git a/backend/src/api/endpoints/v1_api_user.rs b/backend/src/api/endpoints/v1_api_user.rs index 20cca3833..d6dd457b5 100644 --- a/backend/src/api/endpoints/v1_api_user.rs +++ b/backend/src/api/endpoints/v1_api_user.rs @@ -37,7 +37,7 @@ async fn save_config_api_proxy_user( }; // ---------- Search for existing Target and existing User ---------- - let mut existing_target_index: Option = None; // index of target (target_name), falls vorhanden + let mut existing_target_index: Option = None; // index of target (target_name), if present let mut existing_user_target_index: Option = None; // index of existing users target let mut existing_user_index: Option = None; // index of the user in the targets credentials list @@ -206,11 +206,27 @@ async fn delete_config_api_proxy_user( axum::http::StatusCode::OK.into_response() } +/// Terminates a specific playback session for a user. +/// +/// This is the explicit `Terminate` path from the playback state machine: +/// - Removes the session and all associated streams +/// - Releases the counted lease if held +/// - Sets lifecycle to `Expired` +async fn terminate_user_session( + axum::extract::State(app_state): axum::extract::State>, + axum::extract::Path((username, session_token)): axum::extract::Path<(String, String)>, +) -> impl axum::response::IntoResponse { + app_state.active_users.terminate_session(&username, &session_token).await; + app_state.active_provider.clear_provider_reservation(&session_token).await; + (axum::http::StatusCode::NO_CONTENT).into_response() +} + pub fn v1_api_user_register(router: Router>) -> axum::Router> { router .route("/user/{target}", axum::routing::post(save_config_api_proxy_user)) .route("/user/{target}", axum::routing::put(save_config_api_proxy_user)) .route("/user/{target}/{username}", axum::routing::delete(delete_config_api_proxy_user)) + .route("/user/{username}/session/{session_token}", axum::routing::delete(terminate_user_session)) } pub fn v1_api_user_register_with_permissions( @@ -227,6 +243,10 @@ pub fn v1_api_user_register_with_permissions( "/{target}/{username}", axum::routing::delete(delete_config_api_proxy_user) ) + .route( + "/{username}/session/{session_token}", + axum::routing::delete(terminate_user_session) + ) .layer(permission_layer!(app_state, Permission::UserWrite)); router.nest("/user", user_write_routes) diff --git a/backend/src/api/endpoints/web_index.rs b/backend/src/api/endpoints/web_index.rs index 36c0f214f..2fd499c90 100644 --- a/backend/src/api/endpoints/web_index.rs +++ b/backend/src/api/endpoints/web_index.rs @@ -170,12 +170,12 @@ async fn token_refresh( fn inject_nonce_with_parser(html: String, nonce_b64: &str) -> String { let settings = RewriteStrSettings { element_content_handlers: vec![ - // 1) All