diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 5000d3e59..362775a5c 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -452,6 +452,106 @@ pub(in crate::api) fn get_stream_options(app_state: &Arc) -> StreamOpt StreamOptions { stream_retry, buffer_enabled, buffer_size, pipe_provider_stream } } +pub(in crate::api) fn get_effective_admission_strategies(app_state: &Arc) -> Vec { + use crate::model::AdmissionStrategy; + let config = app_state.app_config.config.load(); + let stream_config = config.reverse_proxy.as_ref().and_then(|rp| rp.stream.as_ref()); + match stream_config { + Some(sc) if sc.admission_strategies.is_some() => sc.admission_strategies.clone().unwrap_or_default(), + Some(sc) if sc.grace_period_millis > 0 => { + vec![if sc.grace_period_hold_stream { + AdmissionStrategy::GraceHoldStream + } else { + AdmissionStrategy::GraceInstantStream + }] + } + _ => Vec::new(), + } +} + +#[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, + _fingerprint: &Fingerprint, + is_session_request: bool, + session_token: Option<&str>, +) -> (crate::api::model::ConnectionAdmission, Option) { + use crate::api::model::{AdmissionDecision, ConnectionAdmission, ConnectionKind, StrategyContext, evaluate_strategy}; + use shared::model::UserConnectionPermission; + + let admission = if is_session_request { + app_state.get_connection_admission_for_session( + username, max_connections, soft_connections, session_token.unwrap_or_default(), + ).await + } else { + app_state.get_connection_admission(username, max_connections, soft_connections).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 { + match evaluate_strategy(*strategy, &ctx, &candidates) { + AdmissionDecision::NoMatch => {} + AdmissionDecision::Grace(mode) => { + if app_state.active_users.grant_grace(username).await { + return ( + ConnectionAdmission { + permission: UserConnectionPermission::GracePeriod, + kind: Some(ConnectionKind::Normal), + }, + Some(mode), + ); + } + debug!("Grace grant rejected for user {username}, continuing with later strategies"); + } + AdmissionDecision::Evict(target) => { + debug!("Evicting connection {} for user {username}", target.addr); + app_state.connection_manager.release_connection_as_kicked(&target.addr).await; + let retry_admission = if is_session_request { + app_state + .get_connection_admission_for_session( + username, + max_connections, + soft_connections, + session_token.unwrap_or_default(), + ) + .await + } else { + app_state + .get_connection_admission(username, max_connections, soft_connections) + .await + }; + if retry_admission.permission == UserConnectionPermission::Allowed { + return (retry_admission, 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); + } + } + } + + debug!("No admission strategy could admit user {username}"); + (admission, None) +} + pub fn get_stream_alternative_url(stream_url: &str, input: &ConfigInput, alias_input: &Arc) -> String { let Some(input_user_info) = input.get_user_info() else { return stream_url.to_string(); @@ -654,6 +754,7 @@ async fn create_stream_response_details( user_priority: i8, connection_kind: crate::api::model::ConnectionKind, session_owner: Option<&str>, + grace_hold_override: Option, ) -> Result { let mut streaming_strategy = resolve_streaming_strategy( app_state, @@ -675,6 +776,9 @@ async fn create_stream_response_details( &streaming_strategy.provider_stream_state, grace_period_options.period_millis, ); + if let Some(hold) = grace_hold_override { + grace_period_options.hold_stream = hold; + } let provider_grace_active = matches!( streaming_strategy.provider_stream_state, ProviderStreamState::GracePeriod(_, _) @@ -971,6 +1075,7 @@ pub async fn force_provider_stream_response( user_session: &UserSession, mut stream_channel: StreamChannel, ctx: ForceStreamRequestContext<'_>, + grace_mode: Option, ) -> impl IntoResponse + Send { let stream_options = get_stream_options(app_state); let share_stream = false; @@ -1009,6 +1114,7 @@ pub async fn force_provider_stream_response( connection_priority_for_kind(ctx.user, connection_kind), connection_kind, Some(user_session.token.as_str()), + grace_mode.map(|mode| matches!(mode, crate::api::model::GraceMode::Hold)), ) .await { @@ -1112,6 +1218,7 @@ pub async fn stream_response( connection_permission: UserConnectionPermission, connection_kind: crate::api::model::ConnectionKind, allow_exhausted_shared_reconnect: bool, + grace_mode: Option, ) -> impl IntoResponse + Send { let request_log_stream_url = resolve_request_url_for_logging(input, stream_url); if log_enabled!(log::Level::Trace) { @@ -1209,6 +1316,7 @@ pub async fn stream_response( connection_priority_for_kind(user, connection_kind), connection_kind, Some(session_token), + grace_mode.map(|m| matches!(m, crate::api::model::GraceMode::Hold)), ) .await { @@ -1257,7 +1365,7 @@ pub async fn stream_response( ) .await; - let mut is_stream_shared = share_stream && !deferred_grace_hold_stream; + let mut is_stream_shared = share_stream && !stream_details.has_deferred_provider_open(); if let Some((_header, _status_code, _url, Some(_custom_video))) = stream_details.stream_info.as_ref() { if stream_details.stream.is_some() { is_stream_shared = false; @@ -2556,6 +2664,26 @@ mod tests { Fingerprint::new(format!("fp-{addr}"), addr.ip().to_string(), addr) } + fn create_test_app_state_with_stream_config(stream: crate::model::StreamConfig) -> Arc { + let mut config = Config::default(); + config.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), + cache: None, + rate_limit: None, + geoip: None, + stream_history: None, + qos_aggregation: None, + }); + + let mut app_cfg = create_test_app_config(); + app_cfg.config = Arc::new(ArcSwap::from_pointee(config)); + create_test_app_state_for_config(Arc::new(app_cfg)) + } + fn create_test_local_channel(url: &str) -> StreamChannel { StreamChannel { target_id: 1, @@ -2631,6 +2759,132 @@ mod tests { assert!(admission_failure_video_type(ConnectFailureReason::ProviderError).is_none()); } + #[tokio::test] + async fn effective_admission_strategies_use_legacy_grace_when_field_missing() { + 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: None, + }); + + assert_eq!( + get_effective_admission_strategies(&app_state), + vec![crate::model::AdmissionStrategy::GraceHoldStream] + ); + } + + #[tokio::test] + async fn effective_admission_strategies_respect_explicit_empty_list() { + 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![]), + }); + + assert!(get_effective_admission_strategies(&app_state).is_empty()); + } + + #[tokio::test] + 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![ + crate::model::AdmissionStrategy::GraceHoldStream, + crate::model::AdmissionStrategy::EvictUserSameIpOldest, + ]), + }); + + let first_addr: std::net::SocketAddr = "127.0.0.1:55151".parse().unwrap_or_else(|_| unreachable!()); + let second_addr: std::net::SocketAddr = "127.0.0.1:55152".parse().unwrap_or_else(|_| unreachable!()); + let first_fingerprint = create_test_fingerprint(first_addr); + let second_fingerprint = create_test_fingerprint(second_addr); + + app_state.connection_manager.add_connection(&first_addr).await; + app_state.connection_manager.add_connection(&second_addr).await; + + app_state + .connection_manager + .update_connection(crate::api::model::ConnectionParams { + meter_uid: 1, + username: "fallthrough", + max_connections: 1, + soft_connections: 1, + connection_kind: crate::api::model::ConnectionKind::Normal, + priority: 0, + soft_priority: 10, + fingerprint: &first_fingerprint, + provider: "provider-a", + stream_channel: &create_test_live_channel("http://provider-1.example/live/1.ts"), + user_agent: std::borrow::Cow::Borrowed("ua"), + session_token: Some("tok-first"), + }) + .await; + + assert!(app_state.active_users.grant_grace("fallthrough").await); + + app_state + .connection_manager + .update_connection(crate::api::model::ConnectionParams { + meter_uid: 2, + username: "fallthrough", + max_connections: 1, + soft_connections: 1, + connection_kind: crate::api::model::ConnectionKind::Soft, + priority: 0, + soft_priority: 10, + fingerprint: &second_fingerprint, + provider: "provider-a", + stream_channel: &create_test_live_channel("http://provider-1.example/live/2.ts"), + user_agent: std::borrow::Cow::Borrowed("ua"), + session_token: Some("tok-second"), + }) + .await; + + let (admission, grace_mode) = resolve_admission_with_strategies( + &app_state, + "fallthrough", + 1, + 1, + "127.0.0.1", + &create_test_fingerprint("127.0.0.1:55153".parse().unwrap_or_else(|_| unreachable!())), + true, + Some("tok-third"), + ) + .await; + + assert_eq!(admission.permission, UserConnectionPermission::Allowed); + assert_eq!(admission.kind, Some(crate::api::model::ConnectionKind::Soft)); + assert_eq!(grace_mode, None); + } + #[tokio::test] async fn local_stream_response_registers_active_local_stream() { let app_state = create_test_app_state(); @@ -2765,6 +3019,7 @@ mod tests { admission.permission, admission.kind.unwrap_or(crate::api::model::ConnectionKind::Normal), false, + None, ) .await .into_response(); diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index cd7f81cec..1b933cbb0 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -454,6 +454,7 @@ async fn hls_api_stream( user: &user, session_reservation_ttl_secs: get_hls_session_ttl_secs(&app_state), }, + None, ) .await .into_response(); @@ -462,22 +463,28 @@ async fn hls_api_stream( return axum::http::StatusCode::BAD_REQUEST.into_response(); } - let connection_admission = if (user.max_connections > 0 || user.soft_connections > 0) + let (connection_admission, grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0) && app_state.app_config.config.load().user_access_control { - app_state - .get_connection_admission_for_session( - &user.username, - user.max_connections, - user.soft_connections, - &session.token, - ) - .await + crate::api::api_utils::resolve_admission_with_strategies( + &app_state, + &user.username, + user.max_connections, + user.soft_connections, + &fingerprint.client_ip, + &fingerprint, + true, + Some(&session.token), + ) + .await } else { - crate::api::model::ConnectionAdmission { - permission: UserConnectionPermission::Allowed, - kind: session.connection_kind, - } + ( + crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Allowed, + kind: session.connection_kind, + }, + None, + ) }; let connection_permission = connection_admission.permission; let connection_kind = connection_admission @@ -545,6 +552,7 @@ async fn hls_api_stream( user: &user, session_reservation_ttl_secs: get_hls_session_ttl_secs(&app_state), }, + grace_mode, ) .await .into_response() diff --git a/backend/src/api/endpoints/m3u_api.rs b/backend/src/api/endpoints/m3u_api.rs index 07e3f3333..07f189b87 100644 --- a/backend/src/api/endpoints/m3u_api.rs +++ b/backend/src/api/endpoints/m3u_api.rs @@ -132,10 +132,30 @@ async fn m3u_api_stream( } if pli.item_type.is_local() { - let admission = app_state - .get_connection_admission(&user.username, user.max_connections, user.soft_connections) - .await; let playback_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id); + 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, + true, + Some(playback_session_token.as_str()), + ) + .await + } else { + ( + crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Allowed, + kind: Some(crate::api::model::ConnectionKind::Normal), + }, + None, + ) + }; return local_stream_response( fingerprint, app_state, @@ -202,6 +222,7 @@ async fn m3u_api_stream( user: &user, session_reservation_ttl_secs: get_session_reservation_ttl_secs(app_state, pli.item_type), }, + None, ) .await .into_response(); @@ -211,22 +232,28 @@ async fn m3u_api_stream( pli.url.clone() }; - let connection_admission = if (user.max_connections > 0 || user.soft_connections > 0) + let (connection_admission, grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0) && app_state.app_config.config.load().user_access_control { - app_state - .get_connection_admission_for_session( - &user.username, - user.max_connections, - user.soft_connections, - &session_key, - ) - .await + crate::api::api_utils::resolve_admission_with_strategies( + app_state, + &user.username, + user.max_connections, + user.soft_connections, + &fingerprint.client_ip, + fingerprint, + true, + Some(&session_key), + ) + .await } else { - crate::api::model::ConnectionAdmission { - permission: UserConnectionPermission::Allowed, - kind: user_session.as_ref().and_then(|session| session.connection_kind), - } + ( + crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Allowed, + kind: user_session.as_ref().and_then(|session| session.connection_kind), + }, + None, + ) }; let connection_permission = connection_admission.permission; let connection_kind = connection_admission @@ -306,6 +333,7 @@ async fn m3u_api_stream( connection_permission, connection_kind, allow_exhausted_shared_reconnect, + grace_mode, ) .await .into_response() diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index d37efe25f..cd4bdbf36 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -283,19 +283,30 @@ async fn xtream_player_api_stream( } if pli.item_type.is_local() { - let admission = if (user.max_connections > 0 || user.soft_connections > 0) + let playback_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id); + let (admission, _grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0) && app_state.app_config.config.load().user_access_control { - app_state - .get_connection_admission(&user.username, user.max_connections, user.soft_connections) - .await + crate::api::api_utils::resolve_admission_with_strategies( + app_state, + &user.username, + user.max_connections, + user.soft_connections, + &fingerprint.client_ip, + fingerprint, + true, + Some(playback_session_token.as_str()), + ) + .await } else { - crate::api::model::ConnectionAdmission { - permission: UserConnectionPermission::Allowed, - kind: Some(crate::api::model::ConnectionKind::Normal), - } + ( + crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Allowed, + kind: Some(crate::api::model::ConnectionKind::Normal), + }, + None, + ) }; - let playback_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id); return local_stream_response( fingerprint, app_state, @@ -369,6 +380,7 @@ async fn xtream_player_api_stream( user: &user, session_reservation_ttl_secs: get_session_reservation_ttl_secs(app_state, item_type), }, + None, ) .await .into_response(); @@ -379,22 +391,28 @@ async fn xtream_player_api_stream( pli.url.clone() }; - let connection_admission = if (user.max_connections > 0 || user.soft_connections > 0) + let (connection_admission, grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0) && app_state.app_config.config.load().user_access_control { - app_state - .get_connection_admission_for_session( - &user.username, - user.max_connections, - user.soft_connections, - &session_key, - ) - .await + crate::api::api_utils::resolve_admission_with_strategies( + app_state, + &user.username, + user.max_connections, + user.soft_connections, + &fingerprint.client_ip, + fingerprint, + true, + Some(&session_key), + ) + .await } else { - crate::api::model::ConnectionAdmission { - permission: UserConnectionPermission::Allowed, - kind: user_session.as_ref().and_then(|session| session.connection_kind), - } + ( + crate::api::model::ConnectionAdmission { + permission: UserConnectionPermission::Allowed, + kind: user_session.as_ref().and_then(|session| session.connection_kind), + }, + None, + ) }; let connection_permission = connection_admission.permission; let connection_kind = connection_admission @@ -483,6 +501,7 @@ async fn xtream_player_api_stream( connection_permission, connection_kind, allow_exhausted_shared_reconnect, + grace_mode, ) .await .into_response() @@ -630,6 +649,7 @@ async fn xtream_player_api_stream_with_token( UserConnectionPermission::Allowed, crate::api::model::ConnectionKind::Normal, false, + None, ) .await .into_response() diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 0fd19aa7c..a1c187a4a 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/backend/src/api/model/active_user_manager.rs @@ -489,7 +489,7 @@ impl ActiveUserManager { removed_stream } - pub async fn release_connection(&self, addr: &SocketAddr) -> ReleasedConnection { + async fn release_connection_inner(&self, addr: &SocketAddr, preserve_session_streams: bool) -> ReleasedConnection { let (addr_removed, disconnected_user, removed_streams, expiry_entries, promotions) = { let mut user_connections = self.connections.write().await; @@ -501,9 +501,10 @@ impl ActiveUserManager { if let Some(connection_data) = user_connections.by_key.get_mut(&username) { let mut remaining_streams = Vec::with_capacity(connection_data.streams.len()); let mut released_kinds = Vec::new(); + let mut removed_session_tokens = HashSet::new(); for mut stream_info in connection_data.streams.drain(..) { if stream_info.addr == *addr { - if Self::should_preserve_session_stream(&stream_info) { + if preserve_session_streams && Self::should_preserve_session_stream(&stream_info) { if let Some(entry) = self.build_preserved_stream_expiry(&username, &stream_info, &connection_data.sessions) { @@ -522,6 +523,9 @@ impl ActiveUserManager { released_kinds.push(kind); } connection_data.stream_normal_priorities.remove(&stream_info.uid); + if let Some(token) = stream_info.session_token.as_ref() { + removed_session_tokens.insert(token.clone()); + } removed_streams.push(stream_info); } } else { @@ -529,6 +533,11 @@ impl ActiveUserManager { } } connection_data.streams = remaining_streams; + if !preserve_session_streams && !removed_session_tokens.is_empty() { + connection_data + .sessions + .retain(|session| !removed_session_tokens.contains(&session.token)); + } for kind in released_kinds { connection_data.decrement_kind(kind); } @@ -574,6 +583,14 @@ impl ActiveUserManager { ReleasedConnection { addr_removed, removed_streams } } + pub async fn release_connection(&self, addr: &SocketAddr) -> ReleasedConnection { + self.release_connection_inner(addr, true).await + } + + pub async fn release_connection_as_kicked(&self, addr: &SocketAddr) -> ReleasedConnection { + self.release_connection_inner(addr, false).await + } + pub fn update_config(&self, config: &Config) { let log_active_user = config.log.as_ref().is_some_and(|l| l.log_active_user); let (grace_period_millis, grace_period_timeout_secs) = get_grace_options(config); @@ -596,7 +613,6 @@ impl ActiveUserManager { username: &str, connection_data: &mut UserConnectionData, ) -> ConnectionAdmission { - let current_connections = connection_data.connections; let selected_kind = decide_connection_kind( connection_data.counts, connection_data.max_connections, @@ -604,9 +620,11 @@ impl ActiveUserManager { ); if let Some(kind) = selected_kind { - // Reset grace period because the user is back under max_connections - connection_data.granted_grace = false; - connection_data.grace_ts = 0; + // Reset grace only once the user is back below the hard limit. + if connection_data.connections < connection_data.max_connections { + connection_data.granted_grace = false; + connection_data.grace_ts = 0; + } return ConnectionAdmission { permission: UserConnectionPermission::Allowed, kind: Some(kind), @@ -616,7 +634,7 @@ impl ActiveUserManager { let now = get_current_timestamp(); // Check if user already used a grace period if connection_data.granted_grace { - if current_connections > connection_data.max_connections + if connection_data.connections >= connection_data.max_connections && now - connection_data.grace_ts <= self.grace_period_timeout_secs.load(Ordering::Relaxed) { // Grace timeout, still active, deny connection @@ -627,26 +645,12 @@ impl ActiveUserManager { }; } // Grace timeout expired, reset grace counters - connection_data.granted_grace = false; - connection_data.grace_ts = 0; + if connection_data.connections < connection_data.max_connections { + connection_data.granted_grace = false; + connection_data.grace_ts = 0; + } } - if self.grace_period_millis.load(Ordering::Relaxed) > 0 && current_connections == connection_data.max_connections { - // Intentional asymmetry: grace is granted when current == max (AT limit), while - // Exhausted is returned when current > max (OVER limit after the grace window). - // This allows exactly one extra connection during the grace window — the new - // connection is accepted now, and a background check evicts it if the count is - // still over max after the grace period elapses. - connection_data.granted_grace = true; - connection_data.grace_ts = now; - debug!("Granted a grace period for user access: {username}"); - return ConnectionAdmission { - permission: UserConnectionPermission::GracePeriod, - kind: Some(ConnectionKind::Normal), - }; - } - - // Too many connections, no grace allowed debug!("User access denied, too many connections: {username}"); ConnectionAdmission { permission: UserConnectionPermission::Exhausted, @@ -745,6 +749,66 @@ impl ActiveUserManager { .permission } + pub(crate) async fn get_eviction_candidates( + &self, + username: &str, + client_ip: &str, + ) -> Vec { + let connections = self.connections.read().await; + let Some(connection_data) = connections.by_key.get(username) else { + return Vec::new(); + }; + let mut addr_counts = HashMap::new(); + for stream in connection_data.streams.iter().filter(|stream| !stream.preserved) { + addr_counts + .entry(stream.addr) + .and_modify(|count: &mut u8| *count = count.saturating_add(1)) + .or_insert(1_u8); + } + connection_data + .streams + .iter() + .filter(|stream| !stream.preserved && stream.client_ip == client_ip) + .filter(|stream| matches!(addr_counts.get(&stream.addr), Some(1))) + .map(|s| crate::api::model::EvictionCandidate { + addr: s.addr, + client_ip: s.client_ip.clone(), + ts: s.ts, + }) + .collect() + } + + pub(crate) async fn grant_grace(&self, username: &str) -> bool { + if self.grace_period_millis.load(Ordering::Relaxed) == 0 { + debug!("Grace grant denied, grace_period_millis is zero for {username}"); + return false; + } + let mut connections = self.connections.write().await; + if let Some(connection_data) = connections.by_key.get_mut(username) { + let now = get_current_timestamp(); + if connection_data.connections != connection_data.max_connections { + debug!( + "Grace grant denied for {username}, expected connections == max_connections but got {}/{}", + connection_data.connections, + connection_data.max_connections + ); + return false; + } + if connection_data.granted_grace + && connection_data.connections >= connection_data.max_connections + && now - connection_data.grace_ts <= self.grace_period_timeout_secs.load(Ordering::Relaxed) + { + debug!("Grace grant denied, still within active grace timeout for {username}"); + return false; + } + connection_data.granted_grace = true; + connection_data.grace_ts = now; + debug!("Granted a grace period for user access: {username}"); + return true; + } + false + } + pub async fn active_users_and_connections(&self) -> (usize, usize) { self.gc(); let user_connections = self.connections.read().await; @@ -1529,6 +1593,13 @@ mod tests { } } + fn test_adaptive_channel(virtual_id: u32) -> StreamChannel { + StreamChannel { + item_type: PlaylistItemType::LiveHls, + ..test_channel(virtual_id) + } + } + #[tokio::test] async fn test_multi_session_same_addr_counts_and_releases_individually() { let config = Config::default(); @@ -1563,7 +1634,7 @@ mod tests { assert_eq!(manager.user_connections(username).await, 1); assert_eq!( manager.connection_permission(username, 1, 0).await, - UserConnectionPermission::GracePeriod + UserConnectionPermission::Exhausted ); let second = manager @@ -1593,6 +1664,323 @@ mod tests { assert_eq!(manager.user_connections(username).await, 0); } + #[tokio::test] + async fn test_grant_grace_only_at_limit() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let at_limit_addr: SocketAddr = "127.0.0.1:55011".parse().unwrap(); + let at_limit_fingerprint = Fingerprint::new("fp-limit".to_string(), "127.0.0.1".to_string(), at_limit_addr); + let over_limit_addr: SocketAddr = "127.0.0.1:55012".parse().unwrap(); + let over_limit_fingerprint = Fingerprint::new("fp-over".to_string(), "127.0.0.1".to_string(), over_limit_addr); + + manager.add_connection(&at_limit_addr).await; + manager + .update_connection(ActiveUserConnectionParams { + uid: 10, + meter_uid: 0, + username: "at-limit", + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &at_limit_fingerprint, + provider: "provider-a", + stream_channel: &test_channel(1010), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-limit"), + }) + .await; + + assert!(manager.grant_grace("at-limit").await); + + manager.add_connection(&over_limit_addr).await; + manager + .update_connection(ActiveUserConnectionParams { + uid: 11, + meter_uid: 0, + username: "over-limit", + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &over_limit_fingerprint, + provider: "provider-a", + stream_channel: &test_channel(1011), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-over-1"), + }) + .await; + manager.add_connection(&"127.0.0.1:55013".parse().unwrap()).await; + let second_fingerprint = + Fingerprint::new("fp-over-2".to_string(), "127.0.0.1".to_string(), "127.0.0.1:55013".parse().unwrap()); + manager + .update_connection(ActiveUserConnectionParams { + uid: 12, + meter_uid: 0, + username: "over-limit", + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &second_fingerprint, + provider: "provider-a", + stream_channel: &test_channel(1012), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-over-2"), + }) + .await; + + assert!(!manager.grant_grace("over-limit").await); + } + + #[tokio::test] + async fn eviction_candidates_ignore_ambiguous_socket_addrs() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let shared_addr: SocketAddr = "127.0.0.1:55031".parse().unwrap(); + let unique_addr: SocketAddr = "127.0.0.1:55032".parse().unwrap(); + let shared_fp = Fingerprint::new("fp-shared".to_string(), "127.0.0.1".to_string(), shared_addr); + let unique_fp = Fingerprint::new("fp-unique".to_string(), "127.0.0.1".to_string(), unique_addr); + + manager.add_connection(&shared_addr).await; + manager.add_connection(&unique_addr).await; + + for (uid, token, channel_id) in [(31, "tok-31", 1031), (32, "tok-32", 1032)] { + manager + .update_connection(ActiveUserConnectionParams { + uid, + meter_uid: 0, + username: "same-user", + max_connections: 3, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &shared_fp, + provider: "provider-a", + stream_channel: &test_channel(channel_id), + user_agent: Cow::Borrowed("ua"), + session_token: Some(token), + }) + .await; + } + + manager + .update_connection(ActiveUserConnectionParams { + uid: 33, + meter_uid: 0, + username: "same-user", + max_connections: 3, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &unique_fp, + provider: "provider-a", + stream_channel: &test_channel(1033), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-33"), + }) + .await; + + let candidates = manager.get_eviction_candidates("same-user", "127.0.0.1").await; + assert_eq!(candidates.len(), 1); + assert_eq!(candidates[0].addr, unique_addr); + } + + #[tokio::test] + async fn test_kicked_release_does_not_preserve_adaptive_stream() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let addr: SocketAddr = "127.0.0.1:55014".parse().unwrap(); + let fingerprint = Fingerprint::new("fp-adaptive".to_string(), "127.0.0.1".to_string(), addr); + let mut user = ProxyUserCredentials::default(); + user.username = String::from("user-adaptive"); + user.max_connections = 1; + + manager.add_connection(&addr).await; + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: "tok-adaptive", + virtual_id: 2014, + provider: "provider-a", + stream_url: "http://localhost/live.m3u8", + addr: &addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(ConnectionKind::Normal), + }) + .await; + manager + .update_connection(ActiveUserConnectionParams { + uid: 14, + meter_uid: 0, + username: &user.username, + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &fingerprint, + provider: "provider-a", + stream_channel: &test_adaptive_channel(2014), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-adaptive"), + }) + .await; + + let removed = manager.release_connection_as_kicked(&addr).await; + assert!(removed.addr_removed); + assert_eq!(removed.removed_streams.len(), 1); + assert!(manager.active_streams().await.is_empty()); + } + + #[tokio::test] + async fn test_kicked_release_invalidates_removed_session_tokens() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let kicked_addr: SocketAddr = "127.0.0.1:55015".parse().unwrap(); + let survivor_addr: SocketAddr = "127.0.0.1:55016".parse().unwrap(); + let kicked_fingerprint = Fingerprint::new("fp-kicked".to_string(), "127.0.0.1".to_string(), kicked_addr); + let survivor_fingerprint = Fingerprint::new("fp-survivor".to_string(), "127.0.0.1".to_string(), survivor_addr); + let mut user = ProxyUserCredentials::default(); + user.username = String::from("kicked-user"); + user.max_connections = 1; + + manager.add_connection(&kicked_addr).await; + manager.add_connection(&survivor_addr).await; + + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: "tok-kicked", + virtual_id: 2015, + provider: "provider-a", + stream_url: "http://localhost/live-1.ts", + addr: &kicked_addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(ConnectionKind::Normal), + }) + .await; + manager + .create_user_session(CreateUserSessionParams { + user: &user, + session_token: "tok-survivor", + virtual_id: 2016, + provider: "provider-a", + stream_url: "http://localhost/live-2.ts", + addr: &survivor_addr, + connection_permission: UserConnectionPermission::Allowed, + connection_kind: Some(ConnectionKind::Normal), + }) + .await; + + manager + .update_connection(ActiveUserConnectionParams { + uid: 15, + meter_uid: 0, + username: &user.username, + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &kicked_fingerprint, + provider: "provider-a", + stream_channel: &test_channel(2015), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-kicked"), + }) + .await; + manager + .update_connection(ActiveUserConnectionParams { + uid: 16, + meter_uid: 0, + username: &user.username, + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &survivor_fingerprint, + provider: "provider-a", + stream_channel: &test_channel(2016), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-survivor"), + }) + .await; + + let removed = manager.release_connection_as_kicked(&kicked_addr).await; + assert!(removed.addr_removed); + assert_eq!(removed.removed_streams.len(), 1); + assert_eq!( + manager + .connection_admission_for_session(&user.username, 1, 0, "tok-kicked") + .await + .permission, + UserConnectionPermission::Exhausted + ); + assert_eq!( + manager + .connection_admission_for_session(&user.username, 1, 0, "tok-survivor") + .await + .permission, + UserConnectionPermission::Allowed + ); + } + + #[tokio::test] + async fn test_grace_at_limit_remains_active_until_connections_drop_below_limit() { + let config = Config::default(); + let geoip = Arc::new(ArcSwapOption::::default()); + let event_manager = Arc::new(EventManager::new()); + let manager = ActiveUserManager::new(&config, &geoip, &event_manager); + + let addr: SocketAddr = "127.0.0.1:55017".parse().unwrap(); + let fingerprint = Fingerprint::new("fp-grace".to_string(), "127.0.0.1".to_string(), addr); + + manager.add_connection(&addr).await; + manager + .update_connection(ActiveUserConnectionParams { + uid: 17, + meter_uid: 0, + username: "grace-at-limit", + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &fingerprint, + provider: "provider-a", + stream_channel: &test_channel(2017), + user_agent: Cow::Borrowed("ua"), + session_token: Some("tok-grace"), + }) + .await; + + assert!(manager.grant_grace("grace-at-limit").await); + assert_eq!( + manager.connection_admission("grace-at-limit", 1, 0).await.permission, + UserConnectionPermission::Exhausted + ); + assert!(!manager.grant_grace("grace-at-limit").await); + } + #[tokio::test] async fn test_same_session_token_on_new_addr_reuses_logical_connection() { let config = Config::default(); diff --git a/backend/src/api/model/admission_strategy.rs b/backend/src/api/model/admission_strategy.rs new file mode 100644 index 000000000..4e83a20e4 --- /dev/null +++ b/backend/src/api/model/admission_strategy.rs @@ -0,0 +1,290 @@ +use crate::model::AdmissionStrategy; +use log::debug; +use std::net::SocketAddr; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AdmissionDecision { + NoMatch, + Evict(EvictionTarget), + Grace(GraceMode), + #[cfg_attr(not(test), allow(dead_code))] + Deny, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum GraceMode { + Instant, + Hold, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EvictionTarget { + pub addr: SocketAddr, +} + +#[derive(Debug, Clone)] +pub struct StrategyContext<'a> { + pub username: &'a str, + pub client_ip: &'a str, + #[cfg_attr(not(test), allow(dead_code))] + pub strategies: &'a [AdmissionStrategy], +} + +#[cfg_attr(not(test), allow(dead_code))] +pub fn evaluate_admission_strategies( + ctx: &StrategyContext<'_>, + candidates: &[EvictionCandidate], +) -> AdmissionDecision { + if ctx.strategies.is_empty() { + debug!("No admission strategies configured, denying request for user {}", ctx.username); + return AdmissionDecision::Deny; + } + + debug!( + "Evaluating {} admission strategies for user {}", + ctx.strategies.len(), + ctx.username + ); + + for strategy in ctx.strategies { + let decision = evaluate_strategy(*strategy, ctx, candidates); + match decision { + AdmissionDecision::NoMatch => {} + AdmissionDecision::Evict(target) => { + debug!( + "Strategy {:?} selected eviction target {} for user {}", + strategy, target.addr, ctx.username + ); + return AdmissionDecision::Evict(target); + } + AdmissionDecision::Grace(mode) => { + debug!( + "Strategy {:?} selected grace mode {:?} for user {} (blocking later strategies)", + strategy, mode, ctx.username + ); + return AdmissionDecision::Grace(mode); + } + AdmissionDecision::Deny => { + debug!("Strategy {:?} denied request for user {}", strategy, ctx.username); + return AdmissionDecision::Deny; + } + } + } + + debug!( + "No admission strategy matched for user {}, denying", + ctx.username + ); + AdmissionDecision::Deny +} + +pub(in crate::api) fn evaluate_strategy( + strategy: AdmissionStrategy, + ctx: &StrategyContext<'_>, + candidates: &[EvictionCandidate], +) -> AdmissionDecision { + match strategy { + AdmissionStrategy::EvictUserSameIpOldest => { + evaluate_evict_same_ip(ctx, candidates, EvictionOrder::Oldest) + } + AdmissionStrategy::EvictUserSameIpLatest => { + evaluate_evict_same_ip(ctx, candidates, EvictionOrder::Latest) + } + AdmissionStrategy::GraceInstantStream => AdmissionDecision::Grace(GraceMode::Instant), + AdmissionStrategy::GraceHoldStream => AdmissionDecision::Grace(GraceMode::Hold), + } +} + +#[derive(Clone, Copy)] +enum EvictionOrder { + Oldest, + Latest, +} + +fn evaluate_evict_same_ip( + ctx: &StrategyContext<'_>, + candidates: &[EvictionCandidate], + order: EvictionOrder, +) -> AdmissionDecision { + let mut selected: Option<&EvictionCandidate> = None; + for candidate in candidates { + if candidate.client_ip != ctx.client_ip { + continue; + } + let should_replace = match selected { + None => true, + Some(current) => match order { + EvictionOrder::Oldest => candidate.ts < current.ts, + EvictionOrder::Latest => candidate.ts > current.ts, + }, + }; + if should_replace { + selected = Some(candidate); + } + } + + if selected.is_none() { + debug!( + "Strategy EvictUserSameIp{:?}: no same-IP candidates for user {}", + match order { + EvictionOrder::Oldest => "Oldest", + EvictionOrder::Latest => "Latest", + }, + ctx.username + ); + return AdmissionDecision::NoMatch; + } + + match selected { + Some(candidate) => AdmissionDecision::Evict(EvictionTarget { + addr: candidate.addr, + }), + None => AdmissionDecision::NoMatch, + } +} + +#[derive(Debug, Clone)] +pub struct EvictionCandidate { + pub addr: SocketAddr, + pub client_ip: String, + pub ts: u64, +} + + + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr}; + + fn addr(port: u16) -> SocketAddr { + SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), port) + } + + fn candidate(port: u16, ip: &str, ts: u64) -> EvictionCandidate { + EvictionCandidate { + addr: addr(port), + client_ip: ip.to_string(), + ts, + } + } + + #[test] + fn empty_strategy_list_returns_deny() { + let ctx = StrategyContext { + username: "user1", + client_ip: "1.1.1.1", + strategies: &[], + }; + assert_eq!(evaluate_admission_strategies(&ctx, &[]), AdmissionDecision::Deny); + } + + #[test] + fn evaluator_stops_at_first_matching_strategy() { + let ctx = StrategyContext { + username: "user1", + client_ip: "1.1.1.1", + strategies: &[ + AdmissionStrategy::GraceInstantStream, + AdmissionStrategy::EvictUserSameIpOldest, + ], + }; + let candidates = vec![candidate(1000, "1.1.1.1", 100)]; + let result = evaluate_admission_strategies(&ctx, &candidates); + assert_eq!(result, AdmissionDecision::Grace(GraceMode::Instant)); + } + + #[test] + fn grace_is_blocking() { + let ctx = StrategyContext { + username: "user1", + client_ip: "1.1.1.1", + strategies: &[ + AdmissionStrategy::GraceHoldStream, + AdmissionStrategy::EvictUserSameIpOldest, + ], + }; + let candidates = vec![candidate(1000, "1.1.1.1", 100)]; + let result = evaluate_admission_strategies(&ctx, &candidates); + assert_eq!(result, AdmissionDecision::Grace(GraceMode::Hold)); + } + + #[test] + fn evict_oldest_same_ip() { + let ctx = StrategyContext { + username: "user1", + client_ip: "1.1.1.1", + strategies: &[AdmissionStrategy::EvictUserSameIpOldest], + }; + let candidates = vec![ + candidate(1000, "1.1.1.1", 100), + candidate(1001, "1.1.1.1", 200), + candidate(1002, "2.2.2.2", 50), + ]; + let result = evaluate_admission_strategies(&ctx, &candidates); + match result { + AdmissionDecision::Evict(target) => assert_eq!(target.addr, addr(1000)), + other => panic!("Expected Evict, got {other:?}"), + } + } + + #[test] + fn evict_latest_same_ip() { + let ctx = StrategyContext { + username: "user1", + client_ip: "1.1.1.1", + strategies: &[AdmissionStrategy::EvictUserSameIpLatest], + }; + let candidates = vec![ + candidate(1000, "1.1.1.1", 100), + candidate(1001, "1.1.1.1", 200), + candidate(1002, "2.2.2.2", 50), + ]; + let result = evaluate_admission_strategies(&ctx, &candidates); + match result { + AdmissionDecision::Evict(target) => assert_eq!(target.addr, addr(1001)), + other => panic!("Expected Evict, got {other:?}"), + } + } + + #[test] + fn no_same_ip_candidate_returns_no_match_then_deny() { + let ctx = StrategyContext { + username: "user1", + client_ip: "1.1.1.1", + strategies: &[AdmissionStrategy::EvictUserSameIpOldest], + }; + let candidates = vec![candidate(1000, "2.2.2.2", 100)]; + let result = evaluate_admission_strategies(&ctx, &candidates); + assert_eq!(result, AdmissionDecision::Deny); + } + + #[test] + fn eviction_falls_through_to_grace() { + let ctx = StrategyContext { + username: "user1", + client_ip: "1.1.1.1", + strategies: &[ + AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::GraceHoldStream, + ], + }; + let candidates = vec![candidate(1000, "2.2.2.2", 100)]; + let result = evaluate_admission_strategies(&ctx, &candidates); + assert_eq!(result, AdmissionDecision::Grace(GraceMode::Hold)); + } + + #[test] + fn empty_candidates_with_eviction_then_grace() { + let ctx = StrategyContext { + username: "user1", + client_ip: "1.1.1.1", + strategies: &[ + AdmissionStrategy::EvictUserSameIpLatest, + AdmissionStrategy::GraceInstantStream, + ], + }; + let result = evaluate_admission_strategies(&ctx, &[]); + assert_eq!(result, AdmissionDecision::Grace(GraceMode::Instant)); + } +} diff --git a/backend/src/api/model/connection_manager.rs b/backend/src/api/model/connection_manager.rs index 8e7357774..5bafdcf71 100644 --- a/backend/src/api/model/connection_manager.rs +++ b/backend/src/api/model/connection_manager.rs @@ -139,7 +139,11 @@ async fn release_connection_parts( reason: &DisconnectReason, send_shared_stop_signal: bool, ) { - let removed = user_manager.release_connection(addr).await; + let removed = if matches!(reason, DisconnectReason::ClientKicked) { + user_manager.release_connection_as_kicked(addr).await + } else { + user_manager.release_connection(addr).await + }; for stream_info in &removed.removed_streams { let (bytes_sent, first_byte_latency_ms) = event_manager.read_meter_qos(stream_info.meter_uid).await; event_manager.unregister_meter_client(stream_info.uid).await; @@ -675,6 +679,7 @@ impl ConnectionManager { } pub async fn release_connection_as_kicked(&self, addr: &SocketAddr) { + let _ = self.close_connection_with_reason(addr, DisconnectReason::ClientKicked); release_connection_with_reason(self, addr, &DisconnectReason::ClientKicked, true).await; } @@ -1023,6 +1028,22 @@ mod tests { ); } + #[tokio::test] + async fn release_connection_as_kicked_sends_kick_close_signal() { + let manager = create_test_connection_manager(); + let mut rx = manager.get_close_connection_channel(); + let addr: SocketAddr = "127.0.0.1:2234".parse().unwrap_or_else(|_| unreachable!()); + + manager.release_connection_as_kicked(&addr).await; + assert_eq!( + tokio::time::timeout(Duration::from_millis(100), rx.recv()) + .await + .ok() + .and_then(Result::ok), + Some(CloseConnectionSignal::WithReason(addr, DisconnectReason::ClientKicked)) + ); + } + #[tokio::test] async fn close_connection_signal_sends_generic_close_signal() { let manager = create_test_connection_manager(); diff --git a/backend/src/api/model/mod.rs b/backend/src/api/model/mod.rs index 22c21dbb8..4f68ca73a 100644 --- a/backend/src/api/model/mod.rs +++ b/backend/src/api/model/mod.rs @@ -1,5 +1,6 @@ mod active_provider_manager; mod active_user_manager; +mod admission_strategy; mod app_state; mod connection_manager; mod download; @@ -28,5 +29,8 @@ pub(in crate::api) use self::{ active_user_manager::*, download::*, model_utils::*, provider_config::*, qos_aggregation_manager::*, recording_worker::*, request::*, stream_error::*, xtream::*, }; +pub(in crate::api) use self::admission_strategy::{ + AdmissionDecision, EvictionCandidate, GraceMode, StrategyContext, evaluate_strategy, +}; mod batch_result_collector; pub use self::batch_result_collector::*; diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index 59a6619ae..ad72acaf0 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/backend/src/api/model/streams/active_client_stream.rs @@ -16,7 +16,7 @@ use crate::{ use axum::http::{header::USER_AGENT, HeaderMap}; use bytes::Bytes; use futures::{task::AtomicWaker, Future, Stream, StreamExt}; -use log::{error, info, trace}; +use log::{error, info}; use shared::{ model::{StreamChannel, UserConnectionPermission, VirtualId}, utils::sanitize_sensitive_info, @@ -932,8 +932,6 @@ fn stream_grace_period(request: GracePeriodParams<'_>) -> (Option> None }; - trace!("grace hold stream {hold_stream}"); - if provider_grace_check.is_some() || user_grace_check.is_some() { let stream_strategy_flag = Arc::new(AtomicU8::new( if hold_stream { StreamMode::GracePending as u8 } else { StreamMode::Inner as u8 }, diff --git a/backend/src/model/config/stream.rs b/backend/src/model/config/stream.rs index 17d939c2e..59a342198 100644 --- a/backend/src/model/config/stream.rs +++ b/backend/src/model/config/stream.rs @@ -1,8 +1,48 @@ -use shared::model::{StreamBufferConfigDto, StreamConfigDto}; +use shared::model::{AdmissionStrategyDto, StreamBufferConfigDto, StreamConfigDto}; use shared::utils::parse_to_kbps; use crate::api::model::TransportStreamBuffer; use crate::model::macros; +#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)] +pub enum AdmissionStrategy { + EvictUserSameIpOldest, + EvictUserSameIpLatest, + GraceInstantStream, + GraceHoldStream, +} + +impl From<&AdmissionStrategyDto> for AdmissionStrategy { + fn from(dto: &AdmissionStrategyDto) -> Self { + match dto { + AdmissionStrategyDto::EvictUserSameIpOldest => Self::EvictUserSameIpOldest, + AdmissionStrategyDto::EvictUserSameIpLatest => Self::EvictUserSameIpLatest, + AdmissionStrategyDto::GraceInstantStream => Self::GraceInstantStream, + AdmissionStrategyDto::GraceHoldStream => Self::GraceHoldStream, + } + } +} + +impl From<&AdmissionStrategy> for AdmissionStrategyDto { + fn from(domain: &AdmissionStrategy) -> Self { + match domain { + AdmissionStrategy::EvictUserSameIpOldest => Self::EvictUserSameIpOldest, + AdmissionStrategy::EvictUserSameIpLatest => Self::EvictUserSameIpLatest, + AdmissionStrategy::GraceInstantStream => Self::GraceInstantStream, + AdmissionStrategy::GraceHoldStream => Self::GraceHoldStream, + } + } +} + +impl AdmissionStrategy { + pub fn is_grace(&self) -> bool { + matches!(self, Self::GraceInstantStream | Self::GraceHoldStream) + } + + pub fn is_grace_hold(&self) -> bool { + matches!(self, Self::GraceHoldStream) + } +} + #[derive(Debug, Clone)] pub struct StreamBufferConfig { pub enabled: bool, @@ -41,6 +81,7 @@ pub struct StreamConfig { pub throttle_str: Option, pub throttle_kbps: u64, pub shared_burst_buffer_mb: u64, + pub admission_strategies: Option>, } macros::from_impl!(StreamConfig); @@ -58,6 +99,10 @@ impl From<&StreamConfigDto> for StreamConfig { throttle_str: dto.throttle.clone(), throttle_kbps: dto.throttle.as_ref().map_or(0u64, |throttle| parse_to_kbps(throttle).unwrap_or(0u64)), shared_burst_buffer_mb: dto.shared_burst_buffer_mb, + admission_strategies: dto + .admission_strategies + .as_ref() + .map(|entries| entries.iter().map(AdmissionStrategy::from).collect()), } } } @@ -76,10 +121,65 @@ impl From<&StreamConfig> for StreamConfigDto { throttle: instance.throttle_str.clone(), throttle_kbps: instance.throttle_kbps, shared_burst_buffer_mb: instance.shared_burst_buffer_mb, + admission_strategies: instance + .admission_strategies + .as_ref() + .map(|entries| entries.iter().map(AdmissionStrategyDto::from).collect()), } } } +#[cfg(test)] +mod tests { + use super::{AdmissionStrategy, StreamConfig}; + use shared::model::{AdmissionStrategyDto, StreamConfigDto}; + + #[test] + fn stream_config_preserves_missing_admission_strategies() { + let domain = StreamConfig::from(&StreamConfigDto::default()); + assert_eq!(domain.admission_strategies, None); + } + + #[test] + fn stream_config_preserves_explicit_empty_admission_strategies() { + let domain = StreamConfig::from(&StreamConfigDto { + admission_strategies: Some(vec![]), + ..StreamConfigDto::default() + }); + assert_eq!(domain.admission_strategies, Some(vec![])); + } + + #[test] + fn stream_config_roundtrips_admission_strategies() { + let domain = StreamConfig { + retry: true, + metrics_enabled: true, + buffer: None, + grace_period_millis: 2_000, + grace_period_timeout_secs: 10, + grace_period_hold_stream: true, + hls_session_ttl_secs: 5, + catchup_session_ttl_secs: 5, + throttle_str: None, + throttle_kbps: 0, + shared_burst_buffer_mb: 1, + admission_strategies: Some(vec![ + AdmissionStrategy::EvictUserSameIpOldest, + AdmissionStrategy::GraceHoldStream, + ]), + }; + + let dto = StreamConfigDto::from(&domain); + assert_eq!( + dto.admission_strategies, + Some(vec![ + AdmissionStrategyDto::EvictUserSameIpOldest, + AdmissionStrategyDto::GraceHoldStream, + ]) + ); + } +} + #[derive(Debug, Clone)] pub struct CustomStreamResponse { pub channel_unavailable: Option, diff --git a/docker/Dockerfile b/docker/Dockerfile index 3e5059c96..9bf171489 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -13,12 +13,12 @@ FROM mwader/static-ffmpeg:7.1 AS ffmpeg-static # ----------------------------------------------------------------- # Stage 1: Build the Rust binary for production # ----------------------------------------------------------------- -FROM rust:bookworm AS rust-build +FROM rust@sha256:e8e2bb5ff27ad3b369a4f667392464e6ec399cfe81c1230ae78edb1036b9bd74 AS rust-build # Get target architecture from build argument ARG RUST_TARGET=x86_64-unknown-linux-musl -# Install build dependencies for Debian (bookworm) +# Install build dependencies for Debian (trixie) RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \ pkg-config libssl-dev musl-tools && \ apt-get clean && \ @@ -51,7 +51,7 @@ RUN cargo build -p tuliprox --target ${RUST_TARGET} --release # ----------------------------------------------------------------- # Stage 2: Build the rust frontend # ----------------------------------------------------------------- -FROM rust:bookworm AS trunk-build +FROM rust@sha256:e8e2bb5ff27ad3b369a4f667392464e6ec399cfe81c1230ae78edb1036b9bd74 AS trunk-build # Install dependencies for Trunk and WebAssembly RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \ diff --git a/docs/src/SUMMARY.md b/docs/src/SUMMARY.md index 0584acdaf..4a81c534a 100644 --- a/docs/src/SUMMARY.md +++ b/docs/src/SUMMARY.md @@ -8,6 +8,10 @@ - [Configuration & Setup (Overview)](./configuration/overview.md) - [config.yml (Core System)](./configuration/config.md) - [Streaming & Proxy Behavior](./configuration/reverse-proxy.md) + - [Connection Handling](./configuration/connection-handling.md) + - [Priorities, Soft Connections and Preemption](./configuration/connection-handling-priorities-and-preemption.md) + - [Sessions, HLS, Catchup and Reconnects](./configuration/connection-handling-sessions-and-reconnects.md) + - [Failures and User-Visible Behavior](./configuration/connection-handling-failures-and-user-visible-behavior.md) - [Metadata Updater & FFprobe](./configuration/metadata-update.md) - [Local Media Library](./configuration/local-library.md) - [source.yml (Inputs, Panel API & Targets)](./configuration/source.md) diff --git a/docs/src/configuration/connection-handling-failures-and-user-visible-behavior.md b/docs/src/configuration/connection-handling-failures-and-user-visible-behavior.md new file mode 100644 index 000000000..0c48a09eb --- /dev/null +++ b/docs/src/configuration/connection-handling-failures-and-user-visible-behavior.md @@ -0,0 +1,250 @@ +# Connection Handling: Failures and User-Visible Behavior + +This page explains what operators and end users will see when connection handling rules take effect. + +The most important point: + +Not every "problem" is a defect. Very often Tuliprox is reacting exactly as configured: + +- a stream is intentionally rejected +- a soft connection is intentionally preempted +- a shared stream is intentionally terminated as a whole +- a request intentionally waits in a short grace phase + +## The most common visible outcomes + +### 1. The user gets "too many connections" + +This usually means: + +- from the user-rule perspective, no further legal stream can be admitted + +Possible causes: + +- `max_connections` is reached +- `soft_connections` are not available or already consumed +- admission strategies could not help +- an earlier grace is still active +- a session is still counted + +What to check: + +- `max_connections` +- `soft_connections` +- `admission_strategies` +- session TTLs +- grace settings + +### 2. The user gets provider-exhausted behavior + +This means: + +- the user might still be allowed to stream on the user side +- but the provider currently has no free slot left + +Possible causes: + +- real provider exhaustion +- no reusable shared stream was available +- no valid preemption target existed +- equal priority prevented preemption + +What to check: + +- provider capacity +- priorities of the affected users +- whether shared streams are in use +- whether a higher-priority user currently has precedence + +### 3. The user sees `low_priority_preempted` + +This means: + +- the stream started legally +- but later a more important request arrived +- and the current stream became the valid victim + +Important: + +- this is not a random interruption +- it is an intentional result of the priority model + +If a shared stream was affected, several viewers may see the same result at once. + +### 4. A stream starts with a short delay + +This is often not a bug, but one of these behaviors: + +- `GraceHoldStream` +- provider-side grace +- intentional holding so Tuliprox does not make a wrong decision too early + +In plain language: + +- Tuliprox is waiting briefly to see whether the transition resolves cleanly + +### 5. A stream starts first and is terminated later + +This can happen when: + +- a grace phase allows the start temporarily +- but the situation does not become legal in time + +Then the behavior is: + +- the start was provisionally allowed +- the later final decision was negative + +### 6. A kick feels "harder" than a normal disconnect + +This is intentional. + +With a kick: + +- the stream ends +- the related session is also invalidated + +That prevents the same session from immediately undoing the kick. + +## Typical operator scenarios + +### Scenario A: "The user says they only switched channels" + +Likely causes: + +- the old and new stream briefly overlapped +- grace was too short or not well suited +- the provider still looked full during the switch moment + +Check: + +- `grace_period_millis` +- `grace_period_hold_stream` +- priority +- provider limits + +### Scenario B: "The user says they only seeked" + +Likely causes: + +- the seek created overlapping requests +- the session window was too short +- the user limit was very tight + +Check: + +- `hls_session_ttl_secs` +- `catchup_session_ttl_secs` +- grace configuration + +### Scenario C: "Two users watched the same channel and then both dropped at once" + +Likely cause: + +- they were attached to a shared stream +- that shared stream was preempted as a whole + +Check: + +- the priority of the later requester +- whether the shared stream only had low-priority viewers +- whether `low_priority_preempted` fallback content is configured + +### Scenario D: "The same group of users sometimes switches successfully and sometimes not" + +Likely causes: + +- sometimes a shared stream is reused and sometimes not +- sometimes a slot is freed in time and sometimes not +- equal priority blocks preemption +- the session is sometimes still valid and sometimes already expired + +Check: + +- whether it was truly the same channel or a different source +- session TTLs +- provider load +- shared-stream usage + +## How to classify failures correctly + +### First ask: user side or provider side? + +User side: + +- user limits +- soft connections +- admission strategies +- grace +- sessions + +Provider side: + +- no free slots +- priorities +- shared streams +- preemption + +Many incidents become easier to understand once you separate those two layers. + +### Then ask: immediate or delayed? + +Immediate: + +- admission or provider allocation was not possible right away + +Delayed: + +- grace expired +- a higher-priority request arrived later +- a shared stream was preempted later + +## What operators often misread + +### "If something ended later, it must have been a network issue" + +Not necessarily. It can just as easily have been a later priority or grace decision. + +### "If several users are affected at once, Tuliprox must be unstable" + +Not necessarily. During shared-stream preemption that is exactly the expected behavior. + +### "If a user was allowed to start, they should never be stopped later" + +Not necessarily. Soft connections and low-priority streams can be sacrificed later. + +## Operational recommendations + +### If you want as few complaints as possible + +- use priorities deliberately and sparingly +- use soft connections only where preemption is acceptable +- tune session TTLs for the actual player types in use +- do not set grace too aggressively low + +### If you want hard precedence rules + +- give critical users clearly stronger priority values +- do not rely on "it will probably distribute fairly" + +### If you rely heavily on shared streams + +- document internally that one shared upstream connection can affect all attached viewers together + +## Quick incident checklist + +When an operator wants to classify an incident, these questions are often enough: + +1. Was it the same session or a truly new start? +2. Was the user already at the normal limit? +3. Was any soft capacity still available? +4. Was the provider full? +5. Were multiple viewers attached to a shared stream? +6. Did someone with higher priority arrive later? +7. Was a grace phase active? + +## Further reading + +- [Connection Handling Overview](./connection-handling.md) +- [Priorities, Soft Connections and Preemption](./connection-handling-priorities-and-preemption.md) +- [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md) diff --git a/docs/src/configuration/connection-handling-priorities-and-preemption.md b/docs/src/configuration/connection-handling-priorities-and-preemption.md new file mode 100644 index 000000000..afcbca858 --- /dev/null +++ b/docs/src/configuration/connection-handling-priorities-and-preemption.md @@ -0,0 +1,275 @@ +# Connection Handling: Priorities, Soft Connections and Preemption + +This page explains how Tuliprox decides which stream may stay alive when no provider capacity is left. + +This is usually the hardest part for operators to reason about because several rules interact here: + +- normal connections +- soft connections +- priorities +- shared streams +- real provider limits + +## The most important rule first + +Tuliprox does not preempt at random. It follows fixed precedence rules. + +The general idea is: + +- important streams should remain stable +- less important streams may be sacrificed +- shared streams save capacity, but they are not automatically protected + +## How priority is read + +In Tuliprox: + +- lower numbers mean higher priority + +Examples: + +- `-10` is stronger than `0` +- `0` is stronger than `5` +- `5` is stronger than `20` + +Many operators intuitively expect the opposite. In Tuliprox, the smaller number is the stronger priority. + +## Two types of connections + +### Normal connection + +This is a stream within the user's normal limit. + +Normal connections are the stronger class. + +### Soft connection + +This is an extra stream above `max_connections` when `soft_connections` are enabled. + +Soft connections are the weaker class. They are intentionally easier to preempt. + +## Who may preempt whom + +### Normal against soft + +A normal request may preempt an existing soft connection. + +It does not need a better numeric priority value for that. The difference between `Normal` and `Soft` is already enough. + +In plain language: + +- a real normal slot is always stronger than a bonus slot + +### Soft against normal + +A soft request may not preempt a normal connection. + +In plain language: + +- a bonus slot may not push out a real standard slot + +### Normal against normal + +Here priority decides. + +Rule: + +- only if the new request truly has higher priority +- equal priority does not preempt + +### Soft against soft + +Here priority also decides. + +Rule: + +- only a soft request with higher priority may preempt another soft connection + +## What happens with equal priority + +Equal priority does not preempt. + +This is intentional so the behavior remains stable and predictable. + +Example: + +- User A has priority `0` +- User B has priority `0` +- The provider is full +- User B starts a new stream + +Result: + +- User B may not simply throw out A + +Instead, Tuliprox needs: + +- a free slot +- a shared-stream reuse case +- a grace-based solution +- or the request is rejected + +## What happens in an exact tie + +If several possible victims share the same weak priority, Tuliprox chooses the oldest victim first. + +In plain language: + +- the longest-running stream in that weakest group is removed first + +This avoids unpredictable switching between candidates. + +## How shared streams fit into these rules + +A shared stream is only one upstream connection from the provider's perspective. + +That means: + +- 2, 5, or 20 viewers may be attached to the same live stream +- on the provider side this still consumes just one shared slot + +## Which priority a shared stream has + +A shared stream does not simply keep "the priority of the first viewer". + +Instead, Tuliprox derives an effective shared priority from the active viewers. + +Simplified: + +- within the currently effective connection class, Tuliprox uses the strongest remaining subscriber priority +- if at least one subscriber is `Normal`, the shared stream is treated as `Normal` +- only if all subscribers are `Soft`, the shared stream is treated as `Soft` + +What this means for operators: + +- a later high-priority subscriber can make the shared stream harder to preempt +- if that important subscriber leaves, the shared stream can become weaker again + +## Can a shared stream be preempted completely? + +Yes. + +This is important: + +- if no provider capacity is free +- and a shared stream is the best valid victim +- Tuliprox does not remove only one viewer +- it terminates the shared provider connection as a whole + +Consequence: + +- all viewers on that shared stream lose their upstream at the same time + +If a `low_priority_preempted` fallback output is configured, those viewers may see the fallback content instead of a hard disconnect. + +## Example: several low-priority users share the same stream + +Situation: + +- several users with low priority share channel A +- an important user wants channel B +- no free provider connection is available + +Result: + +- if channel A and channel B compete for the same provider lineup +- and the new user really has higher priority +- the shared stream for A may be preempted as a whole + +Then the following happens: + +- the shared provider connection is released +- the new important stream receives the slot +- all viewers on the old shared stream are affected at once + +## Example: an important user joins the same already shared channel + +Situation: + +- low-priority users are already sharing channel A +- an important user also starts channel A + +Result: + +- in this case Tuliprox often does not need a new provider connection +- the user can join the existing shared stream +- therefore no preemption may be necessary at all + +## Provider grace during switching + +There is not only user grace, but also provider-side transition handling. + +This matters in situations such as: + +- a quick channel switch +- the old stream is just about to close +- the new stream has equal importance +- the provider appears full for a very short moment + +In those cases Tuliprox does not always reject immediately. It may wait briefly to see whether the old slot is released in time. + +This is especially important so equal-priority switch operations do not fail unnecessarily. + +## What soft connections are good for + +Soft connections are useful when you want to give operators or premium users more flexibility without weakening the overall protection model too much. + +They are useful for: + +- short-term overbooking +- comfort during quiet periods +- bonus slots for users who do not need a hard guarantee + +They are not suitable for: + +- hard guarantees +- critical primary streams +- user groups that must never be preempted + +## Practical operator rules + +### If important users should remain stable + +- give them a smaller priority number +- do not rely on soft connections for critical streams + +### If many equal users should be treated fairly + +- give them the same priority +- then they do not preempt each other + +### If shared streams are used heavily + +- remember that shared streams save provider slots +- but they can still be preempted as a whole in a conflict + +### If soft connections trigger complaints + +That is often not a bug, but the intended design: + +- soft is supposed to help +- but soft is not supposed to be as strong as normal + +## Common misunderstandings + +### "More viewers on a shared stream should make it safer" + +No. The decisive factors are not the number of viewers, but: + +- the connection class +- the effective priority +- provider scarcity + +### "Equal priority means the newer request wins" + +No. Equal priority does not preempt. + +### "Soft is just another fully normal slot" + +No. Soft is intentionally weak and remains preemptable. + +## Next step + +If you want to understand why many players create multiple requests for what users perceive as a single stream, read: + +- [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md) diff --git a/docs/src/configuration/connection-handling-sessions-and-reconnects.md b/docs/src/configuration/connection-handling-sessions-and-reconnects.md new file mode 100644 index 000000000..633a386b9 --- /dev/null +++ b/docs/src/configuration/connection-handling-sessions-and-reconnects.md @@ -0,0 +1,246 @@ +# Connection Handling: Sessions, HLS, Catchup and Reconnects + +This page explains why Tuliprox does not treat every request as a completely new stream and why sessions are so important for streaming stability. + +This matters for operators because many seemingly strange connection patterns are actually normal player behavior. + +## Why sessions are needed at all + +Many operators initially assume this model: + +- a user starts one stream +- then one connection stays open until the end + +In practice this is often not how players behave. + +Many players behave more like this: + +- several short follow-up requests +- segment changes +- reconnects +- brief overlap during seeks +- channel switching with an overlap window + +Without session logic, Tuliprox would often count these operations incorrectly as "yet another new stream". + +## What a session means in Tuliprox + +A session is the logical playback identity of a user for one stream activity. + +It helps Tuliprox decide: + +- is this the same playback? +- should Tuliprox try to stay on the same provider account? +- may this reconnect continue the same logical connection? +- or should this be treated as a completely new start? + +## HLS and why it is especially noticeable + +HLS is not "one file, one connection". + +Typical behavior: + +- the player downloads a playlist +- then continuously fetches many small segments +- there are short idle gaps in between +- channel switches or quality changes may trigger new requests + +For operators this can look like many connections in the logs. Tuliprox therefore tries to treat these follow-up requests as one logical session. + +## Catchup and why it can look even more unstable + +Catchup and archive playback often produce: + +- jumps +- backward and forward moves +- repeated requests after timeline changes + +That is why catchup usually needs a longer session TTL than HLS. + +## What the session TTL does + +Tuliprox keeps session information for a limited time so that a short interruption does not immediately look like a completely new stream. + +Important values in the reverse proxy block: + +- `hls_session_ttl_secs` +- `catchup_session_ttl_secs` + +In plain language: + +- HLS gets a short recognition window +- catchup usually gets a somewhat longer one + +## What the session TTL does not mean + +The TTL does not automatically mean: + +- that a real provider stream always stays open the entire time + +Instead, the TTL is about recognition, continuity, and provider affinity. + +## Provider affinity for sessions + +When a session continues, Tuliprox tries to stay on the same provider context whenever possible. + +This matters because some upstream services react badly if one logical playback keeps jumping between different provider accounts or slots. + +For operators this means: + +- session logic is not only a comfort feature +- it also protects against unstable provider-side behavior + +## What happens during a quick channel switch + +During a channel switch, two things can be true at the same time: + +- the old stream is not fully released yet +- the new stream is already being requested + +To prevent unnecessary failures, Tuliprox uses transition tools such as grace and session reuse. + +The goal is: + +- ordinary switching should work +- real parallel viewing should still remain controlled + +## What happens during a seek + +Seek is one of the most common reasons for apparently duplicated connections. + +Why? + +- the player requests the new position already +- the old connection is often not fully closed yet +- for a brief moment both sides exist at the same time + +Without grace and session logic, this would often end as: + +- "too many connections" +- or "provider is full" + +even though the user only performed a normal seek. + +## User grace and session behavior + +Tuliprox uses user grace as a short transition mechanism when a new request would otherwise be blocked by the user limit for a short moment. + +Important for operators: + +- grace is not a permanent extra entitlement +- grace is meant to bridge short transition moments +- grace is only evaluated after normal and soft admission are no longer sufficient + +There are two broad modes: + +- start immediately and verify later +- hold the start briefly and only continue once the decision is clear + +Which one is better depends on the player and the behavior you want. + +## Provider grace and session switches + +Besides the user side, there is also a provider-side grace view. + +This is especially important during: + +- equal-priority switching +- moments where provider slots are still briefly occupied + +Here Tuliprox tries not to turn the new request into an immediate failure if the old slot is likely to be released in time. + +## Why HLS and catchup are still not "free" + +Session reuse does not mean that HLS or catchup are ignored indefinitely. + +If Tuliprox recognizes that this is genuinely a new situation, the request is counted as a new activity again. + +Examples: + +- a different channel +- a different session +- the session has expired +- a kick or another hard termination happened + +## What happens on a kick + +When a connection is deliberately kicked, not only the current stream ends. + +Important for operators: + +- the related session is also invalidated +- the same session should not immediately be accepted as "still valid" + +This prevents a deliberately terminated stream from slipping right back through via the same session identity. + +## What happens on a normal disconnect + +A normal disconnect is not automatically a fault. + +Depending on timing, Tuliprox may: + +- keep the session briefly +- allow a resume +- or discard the session once the window expires + +## What matters when shared streams and sessions meet + +Shared streams and sessions solve different problems: + +- shared streams save provider slots +- sessions help with recognition and continuity + +So a user can simultaneously: + +- be part of a shared stream +- and still rely on session logic for HLS or catchup continuity + +## Common operator questions + +### "Why do I see many requests even though the user only watches one stream?" + +Usually because of: + +- HLS segments +- catchup jumps +- reconnects +- normal player behavior + +### "Why does a session remain for a short time after the stream is gone?" + +So that brief reconnects are not punished as a completely new playback. + +### "Why was the same session no longer accepted after a kick?" + +Because a kick is intentionally supposed to invalidate the session side as well. + +## Recommendations for operators + +### If HLS users often have trouble while switching + +Check: + +- `hls_session_ttl_secs` +- `grace_period_millis` +- `grace_period_hold_stream` + +### If catchup users often fail during timeline jumps + +Check: + +- `catchup_session_ttl_secs` +- whether the player performs very aggressive seeks + +### If users complain about "too many connections" even though they are only seeking + +Check: + +- admission strategies +- grace configuration +- session TTLs + +## Next step + +If you want to understand which user-visible symptoms result from these rules, continue with: + +- [Failures and User-Visible Behavior](./connection-handling-failures-and-user-visible-behavior.md) diff --git a/docs/src/configuration/connection-handling.md b/docs/src/configuration/connection-handling.md new file mode 100644 index 000000000..f9b27a5cf --- /dev/null +++ b/docs/src/configuration/connection-handling.md @@ -0,0 +1,266 @@ +# Connection Handling + +This handbook explains how Tuliprox decides whether a stream is started, reused, briefly tolerated, preempted, or rejected. + +It is written for operators rather than developers. The goal is to explain: + +- what Tuliprox does in everyday operation +- why a user can sometimes start streaming immediately and sometimes cannot +- which rules apply to priorities, sessions, shared streams, and provider limits +- which visible effects operators and end users will notice + +If you want to go deeper into individual topics, continue with: + +- [Priorities, Soft Connections and Preemption](./connection-handling-priorities-and-preemption.md) +- [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md) +- [Failures and User-Visible Behavior](./connection-handling-failures-and-user-visible-behavior.md) + +## Why this matters + +Every stream request in Tuliprox is evaluated against several limits at the same time: + +- user connection limits +- soft connections +- provider capacity +- session reuse +- shared streams +- priorities +- grace behavior during switching and seeking + +Because of that, a stream can fail or be terminated later for very different reasons. Without a clear model, +the behavior can look random. In reality it follows fixed rules. + +## The most important terms in plain language + +### User limit + +This is the normal number of simultaneous streams a user is allowed to have. + +Example: + +- `max_connections: 2` +- The user may normally watch 2 streams in parallel. + +### Soft connection + +A soft connection is an extra stream above the normal user limit. + +Important: + +- It is a bonus, not a guaranteed right. +- It can be preempted more easily than a normal connection. + +### Priority + +Priority decides which stream is protected when there is a conflict. + +Important: + +- lower numbers mean higher priority +- `0` is stronger than `5` +- `-10` is stronger than `0` + +### Provider capacity + +Tuliprox does not only enforce user limits. It also respects the actual free slots available from the upstream provider. + +That means: + +- a user may still be allowed to open another stream on the user side +- but the request can still be blocked +- if the provider currently has no capacity left + +### Shared stream + +Multiple viewers can reuse the same running live stream. + +In that case: + +- for the provider this is only one shared upstream connection +- for viewers, more than one client is attached to the same source + +### Session + +A session is Tuliprox's way of recognizing that several requests belong to the same logical playback. + +This matters especially for: + +- HLS +- catchup +- quick reconnects +- channel switches with a brief overlap + +### Grace + +Grace is a short, controlled transition window. + +It exists so a new request does not fail immediately just because: + +- the old stream is still shutting down +- a player sends overlapping requests during a seek +- a provider slot has not been released yet during a switch + +## The simple overall flow + +When a user starts a stream, Tuliprox roughly thinks in this order: + +1. Is this an existing session or a truly new start? +2. Does the user still have a free normal connection? +3. If not, is a soft connection still available? +4. If not, do configured admission strategies or grace rules apply? +5. If provider capacity is missing, can a shared stream be reused or can a lower-priority connection be preempted? +6. If no legal solution exists, the request is rejected. + +This is the short version. In practice, user rules and provider rules run in parallel: + +- user side: "Is this user allowed to have another stream?" +- provider side: "Is there actually a free upstream connection for it?" + +Both sides must succeed. + +## What is checked first + +### 1. Same session or new stream? + +Tuliprox first tries to recognize whether this is the same logical playback. + +This matters because many players do not use "one long connection", but many short follow-up requests. + +Examples: + +- HLS constantly fetches new segments +- catchup performs frequent jumps +- some players briefly overlap old and new requests during seeks or channel switches + +If Tuliprox recognizes the same playback, not every follow-up request is treated like a completely new stream. + +### 2. Normal connection + +If the user is still below the normal limit, the stream is admitted as a normal connection. + +This is the preferred case. + +### 3. Soft connection + +If the normal user limit is already reached, Tuliprox checks whether a soft connection is still allowed. + +Soft connections are intentionally weaker. They help with short-term overbooking in quiet periods, but they are the first candidates for later preemption. + +### 4. Admission strategies and user grace + +If neither a normal nor a soft connection is possible, Tuliprox can still evaluate configured admission strategies. + +Examples: + +- `GraceInstantStream` +- `GraceHoldStream` +- `EvictUserSameIpOldest` +- `EvictUserSameIpLatest` + +These rules are only evaluated after ordinary user admission is already exhausted. + +### 5. Provider side + +Even if the user is still allowed to stream, the upstream provider must also have room. + +Tuliprox then checks: + +- is there still a free provider slot? +- can an already running shared stream be reused? +- may a lower-priority connection be preempted? +- can a short provider grace transition help? + +## The two major protection layers + +### User rules + +These rules protect the operator-defined usage limits: + +- normal connections +- soft connections +- sessions +- admission strategies + +### Provider rules + +These rules protect the real upstream: + +- free provider slots +- shared streams +- priorities +- preemption +- provider grace during switch moments + +A request must pass both layers. + +## Typical everyday situations + +### Case 1: A user simply starts another channel + +- If there is still room in both the user limit and provider capacity, the stream starts normally. + +### Case 2: A user reaches the limit + +- If soft connections are allowed, a soft connection may still be created. +- If that is no longer possible either, only admission strategies or rejection remain. + +### Case 3: A user performs a seek + +- The player may briefly request the old and the new position in parallel. +- Tuliprox tries not to treat this as abuse immediately. +- This is where grace and session recognition matter. + +### Case 4: Many users share the same live stream + +- On the provider side this counts as one shared upstream connection. +- If preemption happens later, this shared connection can be affected as a whole. + +### Case 5: An important user needs a slot + +- If no provider capacity is free, Tuliprox may preempt lower-priority streams. +- This also applies to shared streams. + +## Important operator rules + +### Equal priority does not preempt + +If two requests are equally important, Tuliprox does not simply sacrifice the existing stream. + +### Higher priority wins + +If a genuinely more important request arrives while no provider capacity is free, a less important stream may be preempted. + +### Soft is weaker than normal + +Normal connections are protected more strongly than soft connections. + +### Shared streams are not a magic shield + +A shared stream saves provider slots, but it is not automatically untouchable. + +If it is the best valid victim, it can be terminated completely. + +### A kick ends both the stream and its session + +When a connection is deliberately kicked, Tuliprox should not immediately treat the same session as still valid. + +## Operator checklist + +If connection handling looks "strange" for your users, check these first: + +- How high is `max_connections`? +- Are `soft_connections` configured? +- What `priority` does the affected user have? +- Are `admission_strategies` configured? +- Is `grace_period_millis` set sensibly? +- Are `hls_session_ttl_secs` and `catchup_session_ttl_secs` appropriate for the player in use? +- Is `share_live_streams` enabled and are multiple users watching the same channel? +- Is the actual provider already at capacity? + +## Recommended reading order + +If you want to fully understand the system, read these next: + +1. [Priorities, Soft Connections and Preemption](./connection-handling-priorities-and-preemption.md) +2. [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md) +3. [Failures and User-Visible Behavior](./connection-handling-failures-and-user-visible-behavior.md) diff --git a/docs/src/configuration/reverse-proxy.md b/docs/src/configuration/reverse-proxy.md index 70b0eb39a..cc0311679 100644 --- a/docs/src/configuration/reverse-proxy.md +++ b/docs/src/configuration/reverse-proxy.md @@ -13,6 +13,26 @@ It also contains the optional telemetry pipeline for stream reliability analysis * `qos_aggregation` periodically condenses those raw events into compact QoS snapshots. * The Web UI can then inspect both the raw history view and the aggregated QoS summaries. +## Read This First: Connection Handling Handbook + +The `reverse_proxy.stream` block now interacts with a broader connection-handling model: + +* user limits +* soft connections +* priorities +* admission strategies +* grace periods +* HLS and catchup sessions +* shared streams +* provider-side preemption + +If you want to understand why a stream starts, waits, gets reused, gets preempted, or is rejected, read these pages first: + +* [Connection Handling](./connection-handling.md) +* [Connection Handling: Priorities, Soft Connections and Preemption](./connection-handling-priorities-and-preemption.md) +* [Connection Handling: Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md) +* [Connection Handling: Failures and User-Visible Behavior](./connection-handling-failures-and-user-visible-behavior.md) + ## Top-level entries ```yaml diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 0b8157f9e..74fdd0358 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -435,7 +435,7 @@ "INPUT_TYPE": "Defines the streaming protocol or source format (e.g. xtream, m3u).", "LIVE_SOURCE": "Source selector for the Live cluster when staged input is enabled (`default`, `staged`, `input`, `skip`).", "PROVIDERS": "Provider profiles define reusable upstream URLs and DNS behavior that this input can reference." - , + , "SERIES_SOURCE": "Source selector for the Series cluster when staged input is enabled (`default`, `staged`, `input`, `skip`).", "VOD_SOURCE": "Source selector for the VOD cluster when staged input is enabled (`default`, `staged`, `input`, `skip`)." }, @@ -611,6 +611,7 @@ "URL": "Target URL that receives the notification webhook." }, "REVERSE_PROXY_CONFIG": { + "ADMISSION_STRATEGIES": "Defines what Tuliprox may still try after the normal user admission is already exhausted.\n\nUse this to control short transition grace during switching or to evict an existing same-IP stream instead of rejecting the new request immediately.\n\nStrategies are evaluated from top to bottom.\n\nGrace strategies only work when `grace_period_millis` is greater than 0. Only one grace strategy can be active at a time.", "FAILOVER_REDIRECT_PATTERNS": "Regex patterns determining which HTTP error responses trigger an automatic failover rotation.", "RESOURCE_REWRITE_DISABLED": "If enabled, URL rewriting for logos and EPG is disabled (also disables caching).", "REWRITE_SECRET": "Persistent secret for stable resource URLs across restarts. \nExample:\n```\nreverse_proxy:\n rewrite_secret: A1B2C3D4E5F60718293A4B5C6D7E8F90\n```", @@ -805,6 +806,10 @@ "ADD_TRAKT_LIST": "Add a Trakt list", "ADD_URL": "Add Url", "ADD_WATCH": "Add Watch", + "ADMISSION_STRATEGY_EVICT_USER_SAME_IP_LATEST": "Evict same-IP latest stream", + "ADMISSION_STRATEGY_EVICT_USER_SAME_IP_OLDEST": "Evict same-IP oldest stream", + "ADMISSION_STRATEGY_GRACE_HOLD_STREAM": "Grace hold stream", + "ADMISSION_STRATEGY_GRACE_INSTANT_STREAM": "Grace instant stream", "ADVANCED": "Advanced", "ALIAS": "Alias", "ALIASES": "Aliases", @@ -1387,6 +1392,7 @@ "VIEW_TYPE_PLAYLIST_EXPLORER": "Playlist Explorer", "VIEW_TYPE_PLAYLIST_EPG": "Playlist EPG", "VIEW_TYPE_RBAC": "WebUI Users", + "ADMISSION_STRATEGIES": "Admission Strategies", "RUNTIME_CONFIG_REPORT": "Runtime Config Report", "RUNTIME_CONFIG_REPORT_FORMAT": "Runtime Config Report Format" }, diff --git a/frontend/scss/_default.scss b/frontend/scss/_default.scss index dfdea8e3d..6baf64051 100644 --- a/frontend/scss/_default.scss +++ b/frontend/scss/_default.scss @@ -148,6 +148,20 @@ li { } } +.tp__toolbar { + display: flex; + flex-flow: row wrap; + gap: var(--gap-small); + box-sizing: border-box; + + .tp__icon-button { + &:hover { + background-color: var(--icon-button-hover-background-color); + border-radius: 50%; + } + } +} + ::selection { background-color: #0078d4; color: #ffffff; diff --git a/frontend/scss/_theme.scss b/frontend/scss/_theme.scss index 4a653f55f..708e39163 100644 --- a/frontend/scss/_theme.scss +++ b/frontend/scss/_theme.scss @@ -35,6 +35,7 @@ body { --scrollbar-border-radius: 4px; --action-card-width: 380px; + --gap-micro: 0.3rem; --gap-mini: 0.3rem; --gap-small: 0.3rem; --gap-default: 0.5rem; diff --git a/frontend/scss/app/components/_table.scss b/frontend/scss/app/components/_table.scss index f328e76d5..ac7e1a187 100644 --- a/frontend/scss/app/components/_table.scss +++ b/frontend/scss/app/components/_table.scss @@ -40,10 +40,9 @@ background-color: var(--table-header-background-color); box-shadow: inset 0 -1px 0 var(--border-color); - // if you want cell border - //&:not(:last-child) { - // border-right: 1px solid var(--border-color); - //} + &:not(:last-child) { + border-right: 1px solid var(--border-color); + } svg, .svg-icon { @@ -84,9 +83,11 @@ vertical-align: top; padding: var(--padding-mini); + /* cell border &:not(:last-child) { border-right: 1px solid var(--border-color); } + */ svg, .svg-icon { diff --git a/frontend/scss/app/components/config/_config_view.scss b/frontend/scss/app/components/config/_config_view.scss index dc496d0b8..61d711e09 100644 --- a/frontend/scss/app/components/config/_config_view.scss +++ b/frontend/scss/app/components/config/_config_view.scss @@ -60,7 +60,6 @@ display: flex; flex-flow: row wrap; gap: var(--gap-default); - height: 2.5rem; align-items: center; li { diff --git a/frontend/src/app/components/config/reverse_proxy_config_view.rs b/frontend/src/app/components/config/reverse_proxy_config_view.rs index 0f550bb6c..167ce2971 100644 --- a/frontend/src/app/components/config/reverse_proxy_config_view.rs +++ b/frontend/src/app/components/config/reverse_proxy_config_view.rs @@ -8,20 +8,21 @@ use crate::{ config_view_context::ConfigViewContext, use_emit_mapped_option, }, - Card, Chip, + Card, Chip, IconButton, TextButton, }, context::ConfigContext, }, config_field, config_field_bool, config_field_child, config_field_custom, config_field_hide, config_field_optional, edit_field_bool, edit_field_list, edit_field_number, edit_field_number_f64, edit_field_number_u16, edit_field_number_u64, edit_field_number_usize, edit_field_text, edit_field_text_option, generate_form_reducer, - i18n::use_translation, + i18n::{use_translation, YewI18n}, + utils::t_safe, }; use shared::{ model::{ - CacheConfigDto, GeoIpConfigDto, QosAggregationConfigDto, RateLimitConfigDto, ResourceRetryConfigDto, - ReverseProxyConfigDto, ReverseProxyDisabledHeaderConfigDto, StreamBufferConfigDto, StreamConfigDto, - StreamHistoryConfigDto, + AdmissionStrategyDto, CacheConfigDto, GeoIpConfigDto, QosAggregationConfigDto, RateLimitConfigDto, + ResourceRetryConfigDto, ReverseProxyConfigDto, ReverseProxyDisabledHeaderConfigDto, StreamBufferConfigDto, + StreamConfigDto, StreamHistoryConfigDto, }, utils::{default_secret, format_float_localized}, }; @@ -31,7 +32,6 @@ const LABEL_CACHE: &str = "LABEL.CACHE"; const LABEL_ENABLED: &str = "LABEL.ENABLED"; const LABEL_SIZE: &str = "LABEL.SIZE"; const LABEL_DIRECTORY: &str = "LABEL.DIRECTORY"; - const LABEL_STREAM: &str = "LABEL.STREAM"; const LABEL_STREAM_GRACE: &str = "LABEL.STREAM_GRACE"; const LABEL_STREAM_SESSION: &str = "LABEL.STREAM_SESSION"; @@ -53,6 +53,8 @@ const LABEL_PERIOD_MILLIS: &str = "LABEL.PERIOD_MILLIS"; const LABEL_BURST_SIZE: &str = "LABEL.BURST_SIZE"; const LABEL_SHARED_BURST_BUFFER_MB: &str = "LABEL.SHARED_BURST_BUFFER_BYTES"; +const LABEL_ADMISSION_STRATEGIES: &str = "LABEL.ADMISSION_STRATEGIES"; + const LABEL_SETTINGS: &str = "LABEL.SETTINGS"; const LABEL_RESOURCE_REWRITE_DISABLED: &str = "LABEL.RESOURCE_REWRITE_DISABLED"; const LABEL_REWRITE_SECRET: &str = "LABEL.REWRITE_SECRET"; @@ -78,6 +80,10 @@ const LABEL_STREAM_HISTORY_RETENTION_DAYS: &str = "LABEL.STREAM_HISTORY_RETENTIO const LABEL_QOS_AGGREGATION: &str = "LABEL.QOS_AGGREGATION"; const LABEL_QOS_AGGREGATION_ENABLED: &str = "LABEL.QOS_AGGREGATION_ENABLED"; const LABEL_INTERVAL_SECS: &str = "LABEL.INTERVAL_SECS"; +const LABEL_ADMISSION_STRATEGY_EVICT_USER_SAME_IP_OLDEST: &str = "LABEL.ADMISSION_STRATEGY_EVICT_USER_SAME_IP_OLDEST"; +const LABEL_ADMISSION_STRATEGY_EVICT_USER_SAME_IP_LATEST: &str = "LABEL.ADMISSION_STRATEGY_EVICT_USER_SAME_IP_LATEST"; +const LABEL_ADMISSION_STRATEGY_GRACE_INSTANT_STREAM: &str = "LABEL.ADMISSION_STRATEGY_GRACE_INSTANT_STREAM"; +const LABEL_ADMISSION_STRATEGY_GRACE_HOLD_STREAM: &str = "LABEL.ADMISSION_STRATEGY_GRACE_HOLD_STREAM"; generate_form_reducer!( state: CacheConfigFormState { form: CacheConfigDto }, @@ -119,6 +125,171 @@ impl FailoverPatternsDto { pub fn is_empty(&self) -> bool { self.patterns.is_empty() } } +#[derive(Debug, Clone, Default, PartialEq)] +pub struct AdmissionStrategiesDto { + pub strategies: Option>, +} + +fn admission_strategy_tag(strategy: AdmissionStrategyDto) -> &'static str { + match strategy { + AdmissionStrategyDto::EvictUserSameIpOldest => "evict_user_same_ip_oldest", + AdmissionStrategyDto::EvictUserSameIpLatest => "evict_user_same_ip_latest", + AdmissionStrategyDto::GraceInstantStream => "grace_instant_stream", + AdmissionStrategyDto::GraceHoldStream => "grace_hold_stream", + } +} + +fn parse_admission_strategy_tag(tag: &str) -> Option { + match tag.trim() { + "evict_user_same_ip_oldest" => Some(AdmissionStrategyDto::EvictUserSameIpOldest), + "evict_user_same_ip_latest" => Some(AdmissionStrategyDto::EvictUserSameIpLatest), + "grace_instant_stream" => Some(AdmissionStrategyDto::GraceInstantStream), + "grace_hold_stream" => Some(AdmissionStrategyDto::GraceHoldStream), + _ => None, + } +} + +fn admission_strategy_label_key(strategy: AdmissionStrategyDto) -> &'static str { + match strategy { + AdmissionStrategyDto::EvictUserSameIpOldest => LABEL_ADMISSION_STRATEGY_EVICT_USER_SAME_IP_OLDEST, + AdmissionStrategyDto::EvictUserSameIpLatest => LABEL_ADMISSION_STRATEGY_EVICT_USER_SAME_IP_LATEST, + AdmissionStrategyDto::GraceInstantStream => LABEL_ADMISSION_STRATEGY_GRACE_INSTANT_STREAM, + AdmissionStrategyDto::GraceHoldStream => LABEL_ADMISSION_STRATEGY_GRACE_HOLD_STREAM, + } +} + +fn admission_strategy_label(translate: &YewI18n, strategy: AdmissionStrategyDto) -> String { + t_safe(translate, admission_strategy_label_key(strategy)).unwrap_or_else(|| match strategy { + AdmissionStrategyDto::EvictUserSameIpOldest => "Evict same-IP oldest stream".to_string(), + AdmissionStrategyDto::EvictUserSameIpLatest => "Evict same-IP latest stream".to_string(), + AdmissionStrategyDto::GraceInstantStream => "Grace instant stream".to_string(), + AdmissionStrategyDto::GraceHoldStream => "Grace hold stream".to_string(), + }) +} + +fn is_grace_strategy(strategy: AdmissionStrategyDto) -> bool { + matches!(strategy, AdmissionStrategyDto::GraceInstantStream | AdmissionStrategyDto::GraceHoldStream) +} + +fn is_grace_strategy_tag(tag: &str) -> bool { + let tag = tag.trim(); + tag.starts_with("grace_") || parse_admission_strategy_tag(tag).is_some_and(is_grace_strategy) +} + +fn admission_strategy_tags(strategies: Option<&Vec>) -> Option> { + strategies.map(|entries| entries.iter().map(|entry| admission_strategy_tag(*entry).to_string()).collect()) +} + +fn parse_admission_strategy_tags(tags: Option<&[String]>) -> Option> { + let tags = tags?; + let mut parsed = Vec::new(); + for tag in tags { + if let Some(strategy) = parse_admission_strategy_tag(tag) { + if !parsed.contains(&strategy) { + parsed.push(strategy); + } + } + } + Some(parsed) +} + +fn filter_disabled_grace_strategy_tags(tags: Vec, grace_period_millis: u64) -> Vec { + if grace_period_millis == 0 { + tags.into_iter().filter(|tag| !is_grace_strategy_tag(tag)).collect() + } else { + tags + } +} + +fn filter_disabled_grace_strategies( + strategies: Option>, + grace_period_millis: u64, +) -> Option> { + strategies.map(|entries| { + if grace_period_millis == 0 { + entries.into_iter().filter(|strategy| !is_grace_strategy(*strategy)).collect() + } else { + entries + } + }) +} + +fn admission_strategy_tag_label(translate: &YewI18n, tag: &str) -> String { + parse_admission_strategy_tag(tag) + .map(|strategy| admission_strategy_label(translate, strategy)) + .unwrap_or_else(|| tag.to_string()) +} + +fn legacy_admission_strategy_tags(stream: &StreamConfigDto) -> Vec { + if stream.grace_period_millis == 0 { + Vec::new() + } else { + vec![admission_strategy_tag(if stream.grace_period_hold_stream { + AdmissionStrategyDto::GraceHoldStream + } else { + AdmissionStrategyDto::GraceInstantStream + }) + .to_string()] + } +} + +fn displayed_admission_strategy_tags(state: &AdmissionStrategiesDto, stream: &StreamConfigDto) -> Vec { + filter_disabled_grace_strategy_tags( + state.strategies.clone().unwrap_or_else(|| { + admission_strategy_tags(stream.admission_strategies.as_ref()) + .unwrap_or_else(|| legacy_admission_strategy_tags(stream)) + }), + stream.grace_period_millis, + ) +} + +fn available_admission_strategies(selected_tags: &[String], grace_period_millis: u64) -> Vec { + let has_grace = selected_tags.iter().filter_map(|tag| parse_admission_strategy_tag(tag)).any(is_grace_strategy); + + [ + AdmissionStrategyDto::EvictUserSameIpOldest, + AdmissionStrategyDto::EvictUserSameIpLatest, + AdmissionStrategyDto::GraceInstantStream, + AdmissionStrategyDto::GraceHoldStream, + ] + .into_iter() + .filter(|strategy| { + let tag = admission_strategy_tag(*strategy); + let grace_available = grace_period_millis > 0 || !is_grace_strategy(*strategy); + !selected_tags.iter().any(|selected| selected == tag) + && grace_available + && (!has_grace || !is_grace_strategy(*strategy)) + }) + .collect() +} + +fn add_admission_strategy_tag(current: &[String], strategy: AdmissionStrategyDto) -> Vec { + let mut next = current.to_vec(); + let tag = admission_strategy_tag(strategy).to_string(); + if !next.iter().any(|selected| selected == &tag) { + next.push(tag); + } + next +} + +fn remove_admission_strategy_tag(current: &[String], index: usize) -> Vec { + let mut next = current.to_vec(); + if index < next.len() { + next.remove(index); + } + next +} + +fn move_admission_strategy_tag(current: &[String], index: usize, delta: isize) -> Vec { + let mut next = current.to_vec(); + if let Some(target_index) = index.checked_add_signed(delta) { + if index < next.len() && target_index < next.len() { + next.swap(index, target_index); + } + } + next +} + generate_form_reducer!( state: FailoverPatternsFormState { form: FailoverPatternsDto }, action_name: FailoverPatternsFormAction, @@ -127,6 +298,14 @@ generate_form_reducer!( } ); +generate_form_reducer!( + state: AdmissionStrategiesFormState { form: AdmissionStrategiesDto }, + action_name: AdmissionStrategiesFormAction, + fields { + Strategies => strategies: Option>, + } +); + generate_form_reducer!( state: StreamConfigFormState { form: StreamConfigDto }, action_name: StreamConfigFormAction, @@ -235,6 +414,8 @@ pub fn ReverseProxyConfigView() -> Html { let failover_patterns_state: UseReducerHandle = use_reducer(|| FailoverPatternsFormState { form: FailoverPatternsDto::default(), modified: false }); + let admission_strategies_state: UseReducerHandle = + use_reducer(|| AdmissionStrategiesFormState { form: AdmissionStrategiesDto::default(), modified: false }); let stream_history_state: UseReducerHandle = use_reducer(|| StreamHistoryConfigFormState { form: StreamHistoryConfigDto::default(), modified: false }); let qos_aggregation_state: UseReducerHandle = @@ -251,6 +432,7 @@ pub fn ReverseProxyConfigView() -> Html { let geoip_state = geoip_state.clone(); let stream_buffer_state = stream_buffer_state.clone(); let failover_patterns_state = failover_patterns_state.clone(); + let admission_strategies_state = admission_strategies_state.clone(); let stream_history_state = stream_history_state.clone(); let qos_aggregation_state = qos_aggregation_state.clone(); let last_emitted_form = last_emitted_form.clone(); @@ -267,6 +449,7 @@ pub fn ReverseProxyConfigView() -> Html { geoip_state.form.clone(), stream_buffer_state.form.clone(), failover_patterns_state.form.clone(), + admission_strategies_state.form.clone(), stream_history_state.form.clone(), qos_aggregation_state.form.clone(), ), @@ -280,6 +463,7 @@ pub fn ReverseProxyConfigView() -> Html { geoip_state.modified, stream_buffer_state.modified, failover_patterns_state.modified, + admission_strategies_state.modified, stream_history_state.modified, qos_aggregation_state.modified, ), @@ -296,6 +480,7 @@ pub fn ReverseProxyConfigView() -> Html { geoip, stream_buffer, failover_patterns, + admission_strategies, stream_history, qos_aggregation, ), @@ -309,6 +494,7 @@ pub fn ReverseProxyConfigView() -> Html { geoip_modified, stream_buffer_modified, failover_patterns_modified, + admission_strategies_modified, stream_history_modified, qos_aggregation_modified, ), @@ -316,6 +502,10 @@ pub fn ReverseProxyConfigView() -> Html { let mut form = rp.clone(); let mut stream_form = stream.clone(); stream_form.buffer = if stream_buffer.is_empty() { None } else { Some(stream_buffer.clone()) }; + stream_form.admission_strategies = filter_disabled_grace_strategies( + parse_admission_strategy_tags(admission_strategies.strategies.as_deref()), + stream_form.grace_period_millis, + ); form.cache = Some(cache.clone()); form.rate_limit = Some(rl.clone()); @@ -338,6 +528,7 @@ pub fn ReverseProxyConfigView() -> Html { || geoip_modified || stream_buffer_modified || failover_patterns_modified + || admission_strategies_modified || stream_history_modified || qos_aggregation_modified; let next_form = ConfigForm::ReverseProxy(modified, form); @@ -362,6 +553,7 @@ pub fn ReverseProxyConfigView() -> Html { let geoip_state = geoip_state.clone(); let stream_buffer_state = stream_buffer_state.clone(); let failover_patterns_state = failover_patterns_state.clone(); + let admission_strategies_state = admission_strategies_state.clone(); let stream_history_state = stream_history_state.clone(); let qos_aggregation_state = qos_aggregation_state.clone(); @@ -424,6 +616,16 @@ pub fn ReverseProxyConfigView() -> Html { failover_patterns_state.dispatch(FailoverPatternsFormAction::SetAll(target_failover_patterns)); } + let target_admission_strategies = AdmissionStrategiesDto { + strategies: admission_strategy_tags( + rp.stream.as_ref().and_then(|stream| stream.admission_strategies.as_ref()), + ), + }; + if admission_strategies_state.form != target_admission_strategies { + admission_strategies_state + .dispatch(AdmissionStrategiesFormAction::SetAll(target_admission_strategies)); + } + let target_stream_history = rp.stream_history.as_ref().map_or_else(StreamHistoryConfigDto::default, |s| s.clone()); if stream_history_state.form != target_stream_history { @@ -482,6 +684,12 @@ pub fn ReverseProxyConfigView() -> Html { failover_patterns_state.dispatch(FailoverPatternsFormAction::SetAll(target_failover_patterns)); } + let target_admission_strategies = AdmissionStrategiesDto::default(); + if admission_strategies_state.form != target_admission_strategies { + admission_strategies_state + .dispatch(AdmissionStrategiesFormAction::SetAll(target_admission_strategies)); + } + let target_stream_history = StreamHistoryConfigDto::default(); if stream_history_state.form != target_stream_history { stream_history_state.dispatch(StreamHistoryConfigFormAction::SetAll(target_stream_history)); @@ -507,6 +715,7 @@ pub fn ReverseProxyConfigView() -> Html { } }; let render_stream = || { + let strategy_tags = displayed_admission_strategy_tags(&admission_strategies_state.form, &stream_state.form); html! { <> @@ -528,6 +737,22 @@ pub fn ReverseProxyConfigView() -> Html { { config_field!(stream_state.form, translate.t(LABEL_HLS_SESSION_TTL_SECS), hls_session_ttl_secs) } { config_field!(stream_state.form, translate.t(LABEL_CATCHUP_SESSION_TTL_SECS), catchup_session_ttl_secs) } + +

{translate.t(LABEL_ADMISSION_STRATEGIES)}

+ { config_field_child!(translate.t(LABEL_ADMISSION_STRATEGIES), "REVERSE_PROXY_CONFIG.ADMISSION_STRATEGIES", { + html! { +
+ if strategy_tags.is_empty() { + + } else { + { for strategy_tags.iter().map(|strategy| { + html! { } + }) } + } +
+ } + })} + } }; @@ -684,6 +909,9 @@ pub fn ReverseProxyConfigView() -> Html { }; let render_stream_edit = || { + let strategy_tags = displayed_admission_strategy_tags(&admission_strategies_state.form, &stream_state.form); + let available_strategies = + available_admission_strategies(&strategy_tags, stream_state.form.grace_period_millis); html! { <> @@ -705,6 +933,86 @@ pub fn ReverseProxyConfigView() -> Html { { edit_field_number_u64!(stream_state, translate.t(LABEL_HLS_SESSION_TTL_SECS), hls_session_ttl_secs, StreamConfigFormAction::HlsSessionTtlSecs) } { edit_field_number_u64!(stream_state, translate.t(LABEL_CATCHUP_SESSION_TTL_SECS), catchup_session_ttl_secs, StreamConfigFormAction::CatchupSessionTtlSecs) } + +

{translate.t(LABEL_ADMISSION_STRATEGIES)}

+
+ if strategy_tags.is_empty() { + + } else { + { for strategy_tags.iter().enumerate().map(|(index, strategy)| { + let remove_state = admission_strategies_state.clone(); + let remove_tags = strategy_tags.clone(); + let move_up_state = admission_strategies_state.clone(); + let move_up_tags = strategy_tags.clone(); + let move_down_state = admission_strategies_state.clone(); + let move_down_tags = strategy_tags.clone(); + let label = admission_strategy_tag_label(&translate, strategy); + html! { +
+ + if index > 0 { + + } + if index + 1 < strategy_tags.len() { + + } + +
+ } + }) } + } +
+
+ { + for available_strategies.into_iter().map(|strategy| { + let add_state = admission_strategies_state.clone(); + let add_tags = strategy_tags.clone(); + let strategy_name = admission_strategy_label(&translate, strategy); + html! { + + } + }) + } +
+ } }; @@ -809,3 +1117,81 @@ pub fn ReverseProxyConfigView() -> Html { } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn admission_strategy_tags_roundtrip() { + let tags = admission_strategy_tags(Some(&vec![ + AdmissionStrategyDto::EvictUserSameIpOldest, + AdmissionStrategyDto::GraceHoldStream, + ])) + .unwrap_or_default(); + assert_eq!( + parse_admission_strategy_tags(Some(&tags)), + Some(vec![AdmissionStrategyDto::EvictUserSameIpOldest, AdmissionStrategyDto::GraceHoldStream,]) + ); + } + + #[test] + fn invalid_admission_strategy_tags_are_ignored() { + let tags = vec![ + "evict_user_same_ip_latest".to_string(), + "not-a-strategy".to_string(), + "evict_user_same_ip_latest".to_string(), + ]; + assert_eq!(parse_admission_strategy_tags(Some(&tags)), Some(vec![AdmissionStrategyDto::EvictUserSameIpLatest])); + } + + #[test] + fn displayed_admission_strategies_fall_back_to_legacy_grace() { + let state = AdmissionStrategiesDto::default(); + let stream = StreamConfigDto { + grace_period_millis: 2_000, + grace_period_hold_stream: true, + ..StreamConfigDto::default() + }; + + assert_eq!(displayed_admission_strategy_tags(&state, &stream), vec!["grace_hold_stream".to_string()]); + } + + #[test] + fn available_admission_strategies_hide_second_grace_option() { + let available = available_admission_strategies(&["grace_hold_stream".to_string()], 2_000); + assert!(!available.contains(&AdmissionStrategyDto::GraceInstantStream)); + assert!(!available.contains(&AdmissionStrategyDto::GraceHoldStream)); + assert!(available.contains(&AdmissionStrategyDto::EvictUserSameIpOldest)); + } + + #[test] + fn available_admission_strategies_hide_grace_when_disabled() { + let available = available_admission_strategies(&[], 0); + assert!(!available.contains(&AdmissionStrategyDto::GraceInstantStream)); + assert!(!available.contains(&AdmissionStrategyDto::GraceHoldStream)); + assert!(available.contains(&AdmissionStrategyDto::EvictUserSameIpOldest)); + assert!(available.contains(&AdmissionStrategyDto::EvictUserSameIpLatest)); + } + + #[test] + fn displayed_admission_strategies_hide_disabled_grace_tags() { + let state = AdmissionStrategiesDto { strategies: Some(vec!["grace_hold_stream".to_string()]) }; + let stream = StreamConfigDto { grace_period_millis: 0, ..StreamConfigDto::default() }; + + assert_eq!(displayed_admission_strategy_tags(&state, &stream), Vec::::new()); + } + + #[test] + fn filtered_admission_strategies_drop_grace_when_disabled() { + let parsed = parse_admission_strategy_tags(Some(&[ + "evict_user_same_ip_oldest".to_string(), + "grace_hold_stream".to_string(), + ])); + + assert_eq!( + filter_disabled_grace_strategies(parsed, 0), + Some(vec![AdmissionStrategyDto::EvictUserSameIpOldest]) + ); + } +} diff --git a/frontend/src/app/mod.rs b/frontend/src/app/mod.rs index a18f61a7c..70bf5a6b4 100644 --- a/frontend/src/app/mod.rs +++ b/frontend/src/app/mod.rs @@ -21,6 +21,8 @@ use yew::prelude::*; use yew_hooks::{use_async_with_options, UseAsyncOptions}; use yew_router::prelude::*; +const STATIC_ASSET_VERSION: &str = env!("CARGO_PKG_VERSION"); + /// App routes #[derive(Routable, Debug, Clone, PartialEq, Eq)] pub enum AppRoute { @@ -41,6 +43,13 @@ pub fn switch(route: AppRoute) -> Html { } } +fn versioned_static_asset_url(path: &str) -> String { + let separator = if path.contains('?') { '&' } else { '?' }; + format!("{path}{separator}v={STATIC_ASSET_VERSION}") +} + +fn versioned_config_url() -> String { versioned_static_asset_url("config.json") } + #[component] pub fn App() -> Html { let supported_languages = vec!["en"]; @@ -56,7 +65,7 @@ pub fn App() -> Html { let futures = languages .iter() .map(|lang| async move { - let url = format!("assets/i18n/{lang}.json"); + let url = versioned_static_asset_url(&format!("assets/i18n/{lang}.json")); let result: Result, Error> = request_get(&url, None, None).await; (lang.to_string(), result) }) @@ -79,7 +88,8 @@ pub fn App() -> Html { let config_state = configuration_state.clone(); use_async_with_options::<_, (), Error>( async move { - match request_get::("config.json", None, None).await { + let config_url = versioned_config_url(); + match request_get::(&config_url, None, None).await { Ok(Some(cfg)) => { if let Some(tab_title) = cfg.tab_title.as_deref() { if let Some(win) = window() { @@ -108,7 +118,8 @@ pub fn App() -> Html { let icon_state = icon_state.clone(); use_async_with_options::<_, (), Error>( async move { - match request_get("assets/icons.json", None, None).await { + let icons_url = versioned_static_asset_url("assets/icons.json"); + match request_get(&icons_url, None, None).await { Ok(Some(icons)) => icon_state.set(Some(icons)), Ok(None) => icon_state.set(Some(Vec::new())), Err(err) => { @@ -150,3 +161,29 @@ pub fn App() -> Html { pub(in crate::app) struct CardContext { pub custom_class: UseStateHandle, } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn versioned_static_asset_url_appends_release_version() { + assert_eq!( + versioned_static_asset_url("assets/i18n/en.json"), + format!("assets/i18n/en.json?v={}", env!("CARGO_PKG_VERSION")) + ); + } + + #[test] + fn versioned_static_asset_url_keeps_existing_query_params() { + assert_eq!( + versioned_static_asset_url("assets/i18n/en.json?lang=en"), + format!("assets/i18n/en.json?lang=en&v={}", env!("CARGO_PKG_VERSION")) + ); + } + + #[test] + fn versioned_config_url_uses_release_version() { + assert_eq!(versioned_config_url(), format!("config.json?v={}", env!("CARGO_PKG_VERSION"))); + } +} diff --git a/shared/src/model/config/stream.rs b/shared/src/model/config/stream.rs index 5a2bd0c36..c67b92338 100644 --- a/shared/src/model/config/stream.rs +++ b/shared/src/model/config/stream.rs @@ -12,6 +12,18 @@ use crate::{ const STREAM_QUEUE_SIZE: usize = 1024; // mpsc channel holding messages. with 8192byte chunks and 2Mbit/s approx 8MB const MIN_SHARED_BURST_BUFFER_MB: u64 = 1; +#[derive(Debug, Copy, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)] +pub enum AdmissionStrategyDto { + #[serde(rename = "evict_user_same_ip_oldest")] + EvictUserSameIpOldest, + #[serde(rename = "evict_user_same_ip_latest")] + EvictUserSameIpLatest, + #[serde(rename = "grace_instant_stream")] + GraceInstantStream, + #[serde(rename = "grace_hold_stream")] + GraceHoldStream, +} + #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)] #[serde(deny_unknown_fields)] pub struct StreamBufferConfigDto { @@ -59,6 +71,8 @@ pub struct StreamConfigDto { pub throttle_kbps: u64, #[serde(default = "default_shared_burst_buffer_mb", skip_serializing_if = "is_default_shared_burst_buffer_mb")] pub shared_burst_buffer_mb: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub admission_strategies: Option>, } impl Default for StreamConfigDto { @@ -75,6 +89,7 @@ impl Default for StreamConfigDto { grace_period_hold_stream: true, hls_session_ttl_secs: default_hls_session_ttl_secs(), catchup_session_ttl_secs: default_catchup_session_ttl_secs(), + admission_strategies: None, } } } @@ -92,6 +107,7 @@ impl StreamConfigDto { && self.grace_period_hold_stream && self.hls_session_ttl_secs == default_hls_session_ttl_secs() && self.catchup_session_ttl_secs == default_catchup_session_ttl_secs() + && self.admission_strategies.is_none() } pub(crate) fn prepare(&mut self) -> Result<(), TuliproxError> { @@ -122,6 +138,131 @@ impl StreamConfigDto { ))); } + if let Some(strategies) = &self.admission_strategies { + if let Err(err) = validate_admission_strategies(strategies, self.grace_period_millis) { + return Err(TuliproxError::ConfigStream(err)); + } + } + Ok(()) } } + +fn validate_admission_strategies(strategies: &[AdmissionStrategyDto], grace_period_millis: u64) -> Result<(), String> { + use std::collections::HashSet; + use AdmissionStrategyDto::*; + + let mut seen = HashSet::new(); + for s in strategies { + if !seen.insert(*s) { + return Err(format!( + "Duplicate admission strategy: {}", + serde_json::to_string(s).unwrap_or_default().trim_matches('"') + )); + } + } + + let has_instant = strategies.iter().any(|s| matches!(s, GraceInstantStream)); + let has_hold = strategies.iter().any(|s| matches!(s, GraceHoldStream)); + if has_instant && has_hold { + return Err("admission_strategies: grace_instant_stream and grace_hold_stream are mutually exclusive".into()); + } + + if grace_period_millis == 0 && (has_instant || has_hold) { + return Err("admission_strategies: grace strategies require grace_period_millis > 0".into()); + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_empty_list_accepted() { + let mut dto = StreamConfigDto::default(); + dto.admission_strategies = Some(vec![]); + assert!(dto.prepare().is_ok()); + } + + #[test] + fn test_none_strategies_accepted() { + let mut dto = StreamConfigDto::default(); + assert!(dto.prepare().is_ok()); + } + + #[test] + fn test_duplicate_rejected() { + let mut dto = StreamConfigDto::default(); + dto.admission_strategies = + Some(vec![AdmissionStrategyDto::EvictUserSameIpOldest, AdmissionStrategyDto::EvictUserSameIpOldest]); + let err = dto.prepare().unwrap_err(); + let msg = format!("{err}"); + assert!(msg.contains("Duplicate admission strategy"), "msg: {msg}"); + } + + #[test] + fn test_mutually_exclusive_grace_rejected() { + let mut dto = StreamConfigDto::default(); + dto.admission_strategies = + Some(vec![AdmissionStrategyDto::GraceInstantStream, AdmissionStrategyDto::GraceHoldStream]); + let err = dto.prepare().unwrap_err(); + let msg = format!("{err}"); + assert!(msg.contains("mutually exclusive"), "msg: {msg}"); + } + + #[test] + fn test_valid_ordered_subset_accepted() { + let mut dto = StreamConfigDto::default(); + dto.admission_strategies = + Some(vec![AdmissionStrategyDto::EvictUserSameIpOldest, AdmissionStrategyDto::GraceHoldStream]); + assert!(dto.prepare().is_ok()); + } + + #[test] + fn test_grace_strategy_requires_positive_grace_period() { + let mut dto = StreamConfigDto::default(); + dto.grace_period_millis = 0; + dto.admission_strategies = Some(vec![AdmissionStrategyDto::GraceHoldStream]); + let err = dto.prepare().unwrap_err(); + let msg = format!("{err}"); + assert!(msg.contains("grace_period_millis"), "msg: {msg}"); + } + + #[test] + fn test_single_strategy_accepted() { + for s in [ + AdmissionStrategyDto::EvictUserSameIpOldest, + AdmissionStrategyDto::EvictUserSameIpLatest, + AdmissionStrategyDto::GraceInstantStream, + AdmissionStrategyDto::GraceHoldStream, + ] { + let mut dto = StreamConfigDto::default(); + dto.admission_strategies = Some(vec![s]); + assert!(dto.prepare().is_ok(), "Failed for {s:?}"); + } + } + + #[test] + fn test_is_empty_with_strategies() { + let mut dto = StreamConfigDto::default(); + dto.admission_strategies = Some(vec![AdmissionStrategyDto::GraceInstantStream]); + assert!(!dto.is_empty()); + } + + #[test] + fn test_is_empty_with_explicit_empty_strategy_list() { + let mut dto = StreamConfigDto::default(); + dto.admission_strategies = Some(vec![]); + assert!(!dto.is_empty()); + } + + #[test] + fn test_explicit_empty_strategy_list_serializes() { + let dto = StreamConfigDto { admission_strategies: Some(vec![]), ..StreamConfigDto::default() }; + + let serialized = serde_json::to_string(&dto).unwrap_or_default(); + assert!(serialized.contains("admission_strategies"), "serialized: {serialized}"); + } +}