From 2c07d52b932fe16211e9a21373195c42f80780ea Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Fri, 25 Sep 2026 08:22:11 +0200 Subject: [PATCH] Fixed max user connection screen when grace granted (#880) * **Bug Fixes** * Improved live-stream session handling during instant grace periods, including when a session has not yet been created. * Corrected connection-limit and eviction tracking so active streams remain eligible for eviction even when their session record is unavailable. * Improved reliability when switching between live channels under connection limits. * **Improvements** * Updated staged M3U overlays for Xtream providers: staged groups determine selected categories and channels, while unmatched categories remain available. Provider categories with no usable staged channels are retained. * **Documentation** * Clarified how staged Xtream and M3U playlists affect provider categories and channels. --- Cargo.lock | 1 + backend/app/src/api/api_utils/mod.rs | 41 +++-- backend/app/src/api/api_utils/tests.rs | 61 +++++--- .../src/processor/playlist/ingest.rs | 64 ++++++-- .../src/processor/playlist/tests.rs | 52 ++++++- .../session/src/active_user_manager/mod.rs | 23 +-- .../session/src/active_user_manager/tests.rs | 41 +++++ docs/src/configuration/source.md | 18 ++- ...-switch-single-user-two-provider-slots.yml | 53 +++++++ tools/tuliprox-testkit/Cargo.toml | 1 + tools/tuliprox-testkit/src/config.rs | 64 ++++++++ tools/tuliprox-testkit/src/main.rs | 143 +++++++++++++++++- tools/tuliprox-testkit/src/observation.rs | 10 +- tools/tuliprox-testkit/src/protocol.rs | 4 + 14 files changed, 496 insertions(+), 80 deletions(-) create mode 100644 test/fixtures/testkit/scenarios/live-ts-channel-switch-single-user-two-provider-slots.yml diff --git a/Cargo.lock b/Cargo.lock index 68f0460c1..283e10560 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4914,6 +4914,7 @@ dependencies = [ "base64 0.23.1", "blake3", "bytes", + "chrono", "clap", "crc32fast", "futures", diff --git a/backend/app/src/api/api_utils/mod.rs b/backend/app/src/api/api_utils/mod.rs index 3e8e48a16..308baacdf 100644 --- a/backend/app/src/api/api_utils/mod.rs +++ b/backend/app/src/api/api_utils/mod.rs @@ -495,6 +495,7 @@ struct SessionActivationRequest<'a> { stream_url: &'a str, connection_permission: UserConnectionPermission, connection_kind: crate::api::model::ConnectionKind, + granted_grace_mode: Option, socket_bound: bool, } @@ -522,6 +523,7 @@ async fn activate_session_before_stream_open( stream_url, connection_permission, connection_kind, + granted_grace_mode, socket_bound, } = request; // Classify based on current session state, not the pre-computed value. @@ -563,11 +565,26 @@ async fn activate_session_before_stream_open( // would evict the same session again). But we must still materialize the grace // lifecycle (PendingProvider / GraceActive) so the session state is consistent. if connection_permission == UserConnectionPermission::GracePeriod { + if loaded_session.as_ref().is_none_or(Option::is_none) { + app_state + .active_users + .ensure_user_session_placeholder(crate::api::model::CreateUserSessionParams { + user, + session_token, + virtual_id: virtual_id.get(), + provider: input.name.as_ref(), + stream_url, + addr: &fingerprint.addr, + connection_permission, + connection_kind: Some(connection_kind), + socket_bound, + }) + .await; + } // Materialize grace lifecycle under the guard so the session state is consistent. - // Determine which grace mode applies by checking the current session state. let current_session = match loaded_session { - Some(session) => session, - None => app_state.active_users.get_and_update_user_session(&user.username, session_token).await, + Some(Some(session)) => Some(session), + _ => app_state.active_users.get_and_update_user_session(&user.username, session_token).await, }; let (_, resolved_grace) = match current_session.as_ref().map(|s| &s.lifecycle) { Some(crate::api::model::PlaybackLifecycle::PendingProvider { .. }) => { @@ -597,18 +614,14 @@ async fn activate_session_before_stream_open( ) } Some(crate::api::model::PlaybackLifecycle::GraceActive) => { - // Already in GraceActive — infer mode from item_type. - let mode = if item_type.is_live() || item_type.is_live_adaptive() { - crate::api::model::GraceMode::Hold - } else { - crate::api::model::GraceMode::Instant - }; - (crate::api::model::PlaybackLifecycle::GraceActive, Some(mode)) + (crate::api::model::PlaybackLifecycle::GraceActive, Some(crate::api::model::GraceMode::Instant)) } _ => { - // Session not yet in grace state — infer from item_type defaults. - // Live/LiveHls/LiveDash default to Hold; VOD/Catchup to Instant. - if item_type.is_live() || item_type.is_live_adaptive() { + let hold_stream = granted_grace_mode.map_or_else( + || item_type.is_live() || item_type.is_live_adaptive(), + |mode| matches!(mode, crate::api::model::GraceMode::Hold), + ); + if hold_stream { let deadline = current_time_secs().saturating_add(app_state.get_grace_options().timeout_secs); let _ = app_state .active_users @@ -2363,6 +2376,7 @@ pub(crate) async fn stream_response( stream_url, connection_permission, connection_kind, + granted_grace_mode: grace_mode, socket_bound, }, ) @@ -3470,6 +3484,7 @@ pub(crate) async fn local_stream_response( stream_url: &pli.url, connection_permission, connection_kind, + granted_grace_mode: None, socket_bound, }, ) diff --git a/backend/app/src/api/api_utils/tests.rs b/backend/app/src/api/api_utils/tests.rs index e1784971f..f07e6a303 100644 --- a/backend/app/src/api/api_utils/tests.rs +++ b/backend/app/src/api/api_utils/tests.rs @@ -5246,6 +5246,7 @@ async fn activate_session_before_stream_open_skips_placeholder_for_follow_up_ses stream_url: channel.url.as_ref(), connection_permission: UserConnectionPermission::Allowed, connection_kind: crate::api::model::ConnectionKind::Normal, + granted_grace_mode: None, socket_bound: true, }, ) @@ -5317,6 +5318,7 @@ async fn activate_session_before_stream_open_revalidates_precomputed_follow_up_r stream_url: channel.url.as_ref(), connection_permission: UserConnectionPermission::Allowed, connection_kind: crate::api::model::ConnectionKind::Normal, + granted_grace_mode: None, socket_bound: true, }, ) @@ -5398,6 +5400,7 @@ async fn activate_session_before_stream_open_stale_follow_up_reclassified_on_cou stream_url: channel.url.as_ref(), connection_permission: UserConnectionPermission::Allowed, connection_kind: crate::api::model::ConnectionKind::Normal, + granted_grace_mode: None, socket_bound: true, }, ) @@ -5440,23 +5443,7 @@ async fn activate_session_before_stream_open_pre_resolved_grace_period_materiali channel.item_type = PlaylistItemType::LiveHls; channel.virtual_id = 55231; - // Session in Prepared state (no grace lifecycle yet). - app_state - .active_users - .create_user_session(crate::api::model::CreateUserSessionParams { - user: &user, - session_token: "tok-pre-resolved-grace", - virtual_id: channel.virtual_id, - provider: input.name.as_ref(), - stream_url: channel.url.as_ref(), - addr: &addr, - connection_permission: UserConnectionPermission::Allowed, - connection_kind: Some(crate::api::model::ConnectionKind::Normal), - socket_bound: true, - }) - .await; - - // Call activation with pre-resolved GracePeriod permission. + // A pre-resolved grace grant reaches activation before a session exists. let activation = activate_session_before_stream_open( &app_state, SessionActivationRequest { @@ -5470,6 +5457,7 @@ async fn activate_session_before_stream_open_pre_resolved_grace_period_materiali stream_url: channel.url.as_ref(), connection_permission: UserConnectionPermission::GracePeriod, connection_kind: crate::api::model::ConnectionKind::Normal, + granted_grace_mode: Some(crate::api::model::GraceMode::Hold), socket_bound: true, }, ) @@ -5485,6 +5473,42 @@ async fn activate_session_before_stream_open_pre_resolved_grace_period_materiali ); } +#[tokio::test] +async fn pre_resolved_instant_grace_creates_counted_live_session() { + let app_state = create_test_app_state(); + let addr: SocketAddr = "127.0.0.1:55233".parse().unwrap_or_else(|_| unreachable!()); + let fingerprint = create_test_fingerprint(addr); + let input = app_state.app_config.sources.load().inputs[0].clone(); + let mut user = ProxyUserCredentials::default(); + user.username = "instant-grace-live-user".to_string(); + user.max_connections = 1; + let channel = create_test_live_channel("http://provider-1.example/live/55233.ts"); + + let activation = activate_session_before_stream_open( + &app_state, + SessionActivationRequest { + fingerprint: &fingerprint, + input: input.as_ref(), + user: &user, + session_token: "tok-instant-grace", + request_class: Some(PlaybackRequestClass::Activate), + virtual_id: VirtualId::new(channel.virtual_id), + item_type: PlaylistItemType::Live, + stream_url: channel.url.as_ref(), + connection_permission: UserConnectionPermission::GracePeriod, + connection_kind: crate::api::model::ConnectionKind::Normal, + granted_grace_mode: Some(crate::api::model::GraceMode::Instant), + socket_bound: true, + }, + ) + .await; + + assert_eq!(activation.grace_mode, Some(crate::api::model::GraceMode::Instant)); + assert_eq!(app_state.active_users.user_connections(&user.username).await, 1); + let session = app_state.active_users.get_and_update_user_session(&user.username, "tok-instant-grace").await; + assert!(session.is_some_and(|session| session.lifecycle == crate::api::model::PlaybackLifecycle::GraceActive)); +} + /// `activate_session_before_stream_open` skips placeholder for Prepare class. #[tokio::test] async fn activate_session_before_stream_open_skips_placeholder_for_prepare() { @@ -5526,6 +5550,7 @@ async fn activate_session_before_stream_open_skips_placeholder_for_prepare() { stream_url: "http://provider.example/live/test.ts", connection_permission: UserConnectionPermission::Allowed, connection_kind: crate::api::model::ConnectionKind::Normal, + granted_grace_mode: None, socket_bound: true, }, ) @@ -5749,6 +5774,7 @@ async fn activate_session_before_stream_open_marks_pending_provider_for_grace_ho stream_url: second_channel.url.as_ref(), connection_permission: UserConnectionPermission::Allowed, connection_kind: crate::api::model::ConnectionKind::Normal, + granted_grace_mode: None, socket_bound: false, }, ) @@ -5833,6 +5859,7 @@ async fn activate_session_before_stream_open_does_not_commit_user_lease_before_p stream_url: channel.url.as_ref(), connection_permission: UserConnectionPermission::Allowed, connection_kind: crate::api::model::ConnectionKind::Normal, + granted_grace_mode: None, socket_bound: false, }, ) diff --git a/backend/processing/src/processor/playlist/ingest.rs b/backend/processing/src/processor/playlist/ingest.rs index 2614eeb03..1e220dc11 100644 --- a/backend/processing/src/processor/playlist/ingest.rs +++ b/backend/processing/src/processor/playlist/ingest.rs @@ -542,8 +542,8 @@ fn prepare_staged_xtream_group( /// names may differ, the ids do not. Staged groups are therefore resolved against the provider groups /// by id, never by name ([`match_staged_groups`]). An overlaid group keeps the provider category id; /// a staged group that introduces a new category keeps its own id only when that id is still free. A -/// staged group whose channels carry no numeric stream id leaves the provider category untouched, so -/// unusable overlay rows cannot remove provider content. +/// staged group whose channels carry no numeric stream id leaves the provider category untouched for +/// Xtream staged input, while an M3U staged input remains authoritative for the selected clusters. pub(crate) fn apply_staged_overlay_groups( provider: &ConfigInput, staged_type: StagedInputType, @@ -619,6 +619,10 @@ pub(crate) fn apply_staged_overlay_groups( let mut used_ids = GroupCategoryIds::from_groups(&provider_groups); let assignment = match_staged_groups(&provider_groups, &staged_selected, staged_type, clusters); + if staged_type == StagedInputType::M3u { + return apply_staged_m3u_template(provider_groups, staged_selected, &assignment, clusters, used_ids); + } + let mut result: Vec = Vec::with_capacity(provider_groups.len() + staged_selected.len()); for (provider_idx, provider_group) in provider_groups.into_iter().enumerate() { @@ -629,8 +633,6 @@ pub(crate) fn apply_staged_overlay_groups( let matched = assignment.staged_of(provider_idx).and_then(|staged_idx| staged_selected[staged_idx].take()); let Some(mut staged) = matched.filter(|staged| !staged.channels.is_empty()) else { - // A staged group whose channels carried no numeric stream id keeps the provider category, - // so the overlaid cluster is never emptied by unusable overlay rows. result.push(provider_group); continue; }; @@ -645,17 +647,7 @@ pub(crate) fn apply_staged_overlay_groups( // Append newly introduced staged groups that had no corresponding provider group. for mut staged in staged_selected.into_iter().flatten().filter(|staged| !staged.channels.is_empty()) { - if staged.id == 0 { - // A staged group without a category id introduces a new category, so it only needs a free id. - staged.id = used_ids.allocate(staged.xtream_cluster); - } else if !used_ids.claim(staged.xtream_cluster, staged.id) { - let allocated = used_ids.allocate(staged.xtream_cluster); - warn!( - "Staged group '{}' uses category id {} which is already taken in {}; renumbered to {allocated}", - staged.title, staged.id, staged.xtream_cluster - ); - staged.id = allocated; - } + used_ids.assign_new_group(&mut staged); sync_channel_category_id(&mut staged); result.push(staged); } @@ -663,6 +655,35 @@ pub(crate) fn apply_staged_overlay_groups( result } +fn apply_staged_m3u_template( + provider_groups: Vec, + staged_groups: Vec>, + assignment: &StagedOverlayAssignment, + clusters: ClusterFlags, + mut used_ids: GroupCategoryIds, +) -> Vec { + let mut result = Vec::with_capacity(provider_groups.len() + staged_groups.len()); + for (staged_idx, staged) in staged_groups.into_iter().enumerate() { + let Some(mut staged) = staged.filter(|group| !group.channels.is_empty()) else { + continue; + }; + if let Some(provider_idx) = assignment.provider_by_staged[staged_idx] { + let provider_group = &provider_groups[provider_idx]; + staged.id = if provider_group.id == 0 { + used_ids.allocate(provider_group.xtream_cluster) + } else { + provider_group.id + }; + } else { + used_ids.assign_new_group(&mut staged); + } + sync_channel_category_id(&mut staged); + result.push(staged); + } + result.extend(provider_groups.into_iter().filter(|group| !cluster_selected(group.xtream_cluster, clusters))); + result +} + /// Keeps the channels of a group consistent with its final category id. /// /// Persistence re-syncs this through [`PlaylistGroup::on_load`], but the overlay result is consumed as @@ -898,6 +919,19 @@ impl GroupCategoryIds { id != 0 && self.used.entry(cluster).or_default().insert(id) } + fn assign_new_group(&mut self, group: &mut PlaylistGroup) { + if group.id == 0 { + group.id = self.allocate(group.xtream_cluster); + } else if !self.claim(group.xtream_cluster, group.id) { + let allocated = self.allocate(group.xtream_cluster); + warn!( + "Staged group '{}' uses category id {} which is already taken in {}; renumbered to {allocated}", + group.title, group.id, group.xtream_cluster + ); + group.id = allocated; + } + } + /// Reserves and returns the next free id of `cluster`. fn allocate(&mut self, cluster: XtreamCluster) -> u32 { let used = self.used.entry(cluster).or_default(); diff --git a/backend/processing/src/processor/playlist/tests.rs b/backend/processing/src/processor/playlist/tests.rs index fb1a2fc71..ea1f9bb0e 100644 --- a/backend/processing/src/processor/playlist/tests.rs +++ b/backend/processing/src/processor/playlist/tests.rs @@ -1443,14 +1443,56 @@ fn staged_m3u_unknown_streams_do_not_match_positional_category_id() { vec![staged_new], ); - assert_eq!(groups.len(), 2); - assert_eq!(groups[0].title.as_ref(), "Sports"); - assert_eq!(groups[0].channels[0].header.id.as_ref(), "100"); - assert_eq!(groups[1].title.as_ref(), "New"); - assert_eq!(groups[1].channels[0].header.id.as_ref(), "300"); + assert_eq!(groups.len(), 1); + assert_eq!(groups[0].title.as_ref(), "New"); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "300"); assert_unique_category_ids(&groups); } +#[test] +fn staged_m3u_replaces_selected_xtream_groups_and_keeps_other_clusters() { + let provider = ConfigInput { + name: "provider".intern(), + input_type: InputType::Xtream, + url: "http://provider.example".to_string(), + username: Some("real-user".to_string()), + password: Some("real-pass".to_string()), + ..Default::default() + }; + let mut provider_news = test_group(XtreamCluster::Live, "Provider News", "provider"); + provider_news.id = 5; + provider_news.channels[0].header.id = "100".intern(); + let mut provider_sports = test_group(XtreamCluster::Live, "Provider Sports", "provider"); + provider_sports.id = 7; + provider_sports.channels[0].header.id = "200".intern(); + provider_sports.channels[0].header.url = "http://provider.example/direct/200.ts".intern(); + let mut provider_kids = test_group(XtreamCluster::Live, "Provider Kids", "provider"); + provider_kids.id = 9; + provider_kids.channels[0].header.id = "300".intern(); + let provider_vod = test_group(XtreamCluster::Video, "Provider VOD", "provider"); + let mut staged_custom = test_group(XtreamCluster::Live, "My Channels", "staged"); + staged_custom.channels[0].header.id = "200".intern(); + let mut staged_news = test_group(XtreamCluster::Live, "My News", "staged"); + staged_news.channels[0].header.id = "100".intern(); + + let groups = super::apply_staged_overlay_groups( + &provider, + StagedInputType::M3u, + ClusterFlags::Live, + vec![provider_news, provider_sports, provider_kids, provider_vod], + vec![staged_custom, staged_news], + ); + + assert_eq!(groups.len(), 3); + assert_eq!(groups[0].title.as_ref(), "My Channels"); + assert_eq!(groups[0].channels[0].header.id.as_ref(), "200"); + assert_eq!(groups[0].channels[0].header.input_name.as_ref(), "provider"); + assert_eq!(groups[0].channels[0].header.url.as_ref(), "http://provider.example/direct/200.ts"); + assert_eq!(groups[1].title.as_ref(), "My News"); + assert_eq!(groups[1].channels[0].header.id.as_ref(), "100"); + assert_eq!(groups[2].title.as_ref(), "Provider VOD"); +} + #[test] fn staged_xtream_overlay_falls_back_to_the_secondary_stream_overlap() { let provider = ConfigInput { diff --git a/backend/session/src/active_user_manager/mod.rs b/backend/session/src/active_user_manager/mod.rs index 7e636a464..b8b8b2d10 100644 --- a/backend/session/src/active_user_manager/mod.rs +++ b/backend/session/src/active_user_manager/mod.rs @@ -1586,17 +1586,9 @@ impl ActiveUserManager { }; let mut addr_counts = HashMap::new(); for stream in &connection_data.streams { - // Preserved streams do not occupy a counted slot — exclude from addr counts. - // They are still valid eviction candidates (see filter below), but they don't - // consume connection capacity, so they don't contribute to the "singleton addr" logic. - let contributes_to_count = if stream.preserved { - false - } else if let Some(token) = stream.session_token.as_deref() { - connection_data.sessions.iter().any(|s| s.token == token && s.lifecycle.is_counted()) - } else { - true // orphan streams without a session are counted - }; - if contributes_to_count { + // The stream kind is the charged slot. Session lifecycle may lag behind + // the stream registration, so it cannot determine eviction eligibility. + if !stream.preserved && connection_data.stream_kinds.contains_key(&stream.uid) { addr_counts .entry(stream.addr) .and_modify(|count: &mut u8| *count = count.saturating_add(1)) @@ -1606,14 +1598,7 @@ impl ActiveUserManager { let candidates: Vec<_> = connection_data .streams .iter() - .filter(|stream| { - if let Some(token) = stream.session_token.as_deref() { - connection_data.sessions.iter().any(|s| s.token == token && s.lifecycle.is_counted()) - || stream.preserved - } else { - true - } - }) + .filter(|stream| stream.preserved || connection_data.stream_kinds.contains_key(&stream.uid)) .filter(|stream| { let addr_count = addr_counts.get(&stream.addr).copied().unwrap_or(0); if stream.preserved { diff --git a/backend/session/src/active_user_manager/tests.rs b/backend/session/src/active_user_manager/tests.rs index 455c2b86a..60185b7c1 100644 --- a/backend/session/src/active_user_manager/tests.rs +++ b/backend/session/src/active_user_manager/tests.rs @@ -1484,6 +1484,47 @@ async fn eviction_candidates_ignore_ambiguous_socket_addrs() { assert_eq!(candidates[0].addr, unique_addr); } +#[tokio::test] +async fn eviction_candidates_include_charged_stream_with_uncounted_session() { + 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:55034".parse().unwrap(); + let fingerprint = Fingerprint::new("fp-switch".to_string(), "10.41.41.170".to_string(), addr); + let username = "switch-user"; + manager.add_connection(&addr).await; + manager + .update_connection(ActiveUserConnectionParams { + uid: 34, + meter_uid: 0, + username, + max_connections: 1, + soft_connections: 0, + connection_kind: ConnectionKind::Normal, + priority: 0, + soft_priority: 0, + fingerprint: &fingerprint, + provider: "provider-a".intern(), + stream_channel: &test_channel(1034), + user_agent: Cow::Borrowed("ua"), + session_token: Some("missing-session"), + }) + .await; + + assert_eq!(manager.user_connections(username).await, 1); + assert_eq!(manager.connection_permission(username, 1, 0).await, UserConnectionPermission::Exhausted); + + let candidates = manager.get_eviction_candidates(username, &fingerprint.client_ip).await; + assert_eq!(candidates.len(), 1); + assert_eq!(candidates[0].addr, addr); + assert_eq!(candidates[0].client_ip, fingerprint.client_ip); + + manager.release_connection_as_kicked(&addr).await; + assert_eq!(manager.connection_permission(username, 1, 0).await, UserConnectionPermission::Allowed); +} + #[tokio::test] async fn eviction_candidates_include_other_ips_for_user_wide_rules() { let config = Config::default(); diff --git a/docs/src/configuration/source.md b/docs/src/configuration/source.md index ef4d0bc1c..6f1458d66 100644 --- a/docs/src/configuration/source.md +++ b/docs/src/configuration/source.md @@ -999,9 +999,9 @@ the actual streams are still opened against the Xtream or M3U provider. **Data flow:** * `staged input -> provider input`: the staged input is an overlay for that provider. Inside the clusters listed in - `staged.clusters`, each staged group of an `xtream` provider replaces the provider category it belongs to; provider - categories that the staged playlist does not represent stay as they are. For any other provider type the listed - clusters are replaced entirely by the staged groups. Clusters not listed are loaded from the provider input itself. + `staged.clusters`, an `m3u` staged playlist determines all groups and channels. An `xtream` staged playlist replaces + matching provider categories and keeps unmatched provider categories. Clusters not listed are loaded from the provider + input itself. The merged playlist is persisted under the provider input, and streaming/API requests still target the provider input. * `staged input -> target` is not supported. Use a normal `m3u` or `xtream` input if the source should be connected directly to a target. @@ -1014,12 +1014,14 @@ the actual streams are still opened against the Xtream or M3U provider. known, an `xtream` staged group is resolved by category ID; an `m3u` staged group skips that step because its category IDs are positional. The group title acts as a last resort. A group title therefore never overrides a stream ID match, so a renamed or split group cannot take over an unrelated category by accident. -* The overlaid category keeps its own category ID; the staged playlist supplies the group name and the channel order. +* The overlaid category keeps its own category ID; the staged playlist supplies the group name and channel order. + With `staged_type: m3u`, it also supplies the group order. * Known streams keep their provider playback URL, including a direct source URL. New streams use the provider address and credentials; for VOD, the staged stream's container metadata supplies the extension when available. * Every staged channel needs a numeric provider stream ID in its `header.id`, either from an `xui-id` / `cuid` attribute or from the numeric last URL segment. Rows without one are skipped, a single warning per group reports how - many rows were dropped, and the provider category stays in place instead of being emptied. + many rows were dropped. A group with no usable channels is omitted for `m3u` staged input; for `xtream` staged input, + a matching provider category stays in place. * A staged group that matches no provider category is added as a new category. Its category ID is reused when it is still free, otherwise Tuliprox assigns an unused ID, so two groups of one cluster never share an ID. @@ -1064,9 +1066,9 @@ inputs: `staged.clusters` names the clusters whose categories are overlaid from the staged input. * The referenced provider supplies all clusters not listed in `staged.clusters`. -* Inside a listed cluster of an `xtream` provider, only the categories represented by the staged playlist are overlaid; - provider categories without a staged counterpart stay available. For any other provider type the listed clusters are - replaced entirely. +* Inside a listed cluster of an `xtream` provider, an `m3u` staged playlist determines which groups and channels appear. + Provider categories without an `m3u` staged counterpart are omitted. With `staged_type: xtream`, unmatched provider + categories remain available. For any other provider type the listed clusters are replaced entirely. * `staged.for_input` must reference an existing non-staged `m3u` or `xtream` input. * Each provider input can have at most one staged overlay. * `staged.clusters` must not be empty. diff --git a/test/fixtures/testkit/scenarios/live-ts-channel-switch-single-user-two-provider-slots.yml b/test/fixtures/testkit/scenarios/live-ts-channel-switch-single-user-two-provider-slots.yml new file mode 100644 index 000000000..a8e5def9e --- /dev/null +++ b/test/fixtures/testkit/scenarios/live-ts-channel-switch-single-user-two-provider-slots.yml @@ -0,0 +1,53 @@ +schema_version: 1 +name: live-ts-channel-switch-single-user-two-provider-slots +tuliprox: + base_url: http://fixture.invalid + execution_mode: isolated_fixture + playback_endpoint: xtream + fixture_stream: + share_live_hls: true + share_live_mpeg_ts: false + bootstrap: + command: "${env:TULIPROX_TESTKIT_SUT_BINARY}" + arguments: ["-s", "-c", "{config_file}", "-i", "{source_file}", "-a", "{api_proxy_file}"] + readiness_url: "{base_url}/healthcheck" +origin: + run_id: live-ts-channel-switch-single-user-two-provider-slots + account_limit: 2 + limit_mode: reject_new +actors: + - id: tivimate + agent: local + username: admission-user + user_agent: TiviMate/5.1.6-testkit + client_ip: { mode: forwarded, value: 10.41.41.170 } +channels: + live-11: { origin_marker: 11, protocol: xtream_ts } + live-12: { origin_marker: 12, protocol: xtream_ts } + live-13: { origin_marker: 13, protocol: xtream_ts } +steps: + - command_id: start-11 + start: { actor: tivimate, playback_id: playback-11, session_group: playback-11, channel: live-11 } + await: { valid_frames: 3 } + - command_id: switch-12 + start: { actor: tivimate, playback_id: playback-12, session_group: playback-12, channel: live-12, expect_evicted: true } + await: { valid_frames: 3 } + assert_origin: { active_body: 2, max_active_body: 2, no_limit_rejections: true } + - command_id: release-11 + stop: { playback_id: playback-11 } + assert_origin: { active_body: 1, no_limit_rejections: true } + - command_id: switch-13 + start: { actor: tivimate, playback_id: playback-13, session_group: playback-13, channel: live-13 } + await: { valid_frames: 5 } + assert_origin: { active_body: 1, max_active_body: 2, no_limit_rejections: true } +policy_contract: + user_access_control: true + users: + admission-user: { max_connections: 1, soft_connections: 0 } + admission_strategies: + - grace_instant_stream + - evict_user_same_ip_latest + grace: { mode: instant, timeout_millis: 2000 } + provider_pool: + - { name: account-a, max_connections: 1 } + - { name: account-b, max_connections: 1 } diff --git a/tools/tuliprox-testkit/Cargo.toml b/tools/tuliprox-testkit/Cargo.toml index 8bfae1960..65ee0d2db 100644 --- a/tools/tuliprox-testkit/Cargo.toml +++ b/tools/tuliprox-testkit/Cargo.toml @@ -13,6 +13,7 @@ base64.workspace = true blake3.workspace = true bytes.workspace = true clap = { version = "4.6.6", features = ["derive"] } +chrono.workspace = true crc32fast = "1.5.1" futures.workspace = true reqwest = { version = "0.13.4", features = ["json", "stream", "rustls"] } diff --git a/tools/tuliprox-testkit/src/config.rs b/tools/tuliprox-testkit/src/config.rs index c100610cd..29bb299a5 100644 --- a/tools/tuliprox-testkit/src/config.rs +++ b/tools/tuliprox-testkit/src/config.rs @@ -340,6 +340,8 @@ pub struct Start { pub playback_id: String, pub session_group: String, #[serde(default)] + pub expect_evicted: bool, + #[serde(default)] pub url: Option, #[serde(default)] pub channel: Option, @@ -383,6 +385,7 @@ impl Scenario { Ok(scenario) } + #[allow(clippy::too_many_lines)] pub fn validate(&self) -> Result<(), TestkitError> { if self.schema_version != 1 { return Err(TestkitError::Configuration("unsupported scenario schema version".to_owned())); @@ -412,6 +415,7 @@ impl Scenario { let mut command_ids = HashSet::new(); let mut playback_ids = HashSet::new(); let mut seen_playback_ids = HashSet::new(); + let mut expected_evictions = HashSet::new(); for step in &steps { if !command_ids.insert(step.command_id.as_str()) || step.command_id.is_empty() { return Err(TestkitError::Configuration("command IDs must be unique and non-empty".to_owned())); @@ -429,12 +433,34 @@ impl Scenario { stop.playback_id ))); } + if expected_evictions.contains(stop.playback_id.as_str()) { + return Err(TestkitError::Configuration(format!( + "playback {} expects eviction and cannot be explicitly stopped", + stop.playback_id + ))); + } playback_ids.remove(stop.playback_id.as_str()); continue; }; if !actor_ids.contains(start.actor.as_str()) { return Err(TestkitError::Configuration(format!("unknown actor {}", start.actor))); } + if start.expect_evicted { + if step.expect.is_rejected() + || self.actors.iter().any(|actor| actor.id == start.actor && actor.agent != "local") + || ((start.vod_object.is_some() + || start.range.is_some() + || start.read_limit_bytes.is_some() + || start.method.is_some()) + && (start.post_read_action.is_some_and(|action| action != PostReadAction::KeepOpen) + || start.method.as_deref() == Some("HEAD"))) + { + return Err(TestkitError::Configuration( + "expect_evicted requires a held streaming playback on a local actor".to_owned(), + )); + } + expected_evictions.insert(start.playback_id.as_str()); + } if start.playback_id.is_empty() || start.session_group.is_empty() { return Err(TestkitError::Configuration("start requires playback_id and session_group".to_owned())); } @@ -572,6 +598,43 @@ pub fn lease_duration(millis: u64) -> Duration { Duration::from_millis(millis.ma mod tests { use super::*; + #[test] + fn expected_eviction_requires_held_vod_playback() -> Result<(), TestkitError> { + let mut scenario: Scenario = serde_saphyr::from_str( + r" +schema_version: 1 +name: vod-eviction-validation +tuliprox: { base_url: http://example.invalid, execution_mode: existing_instance } +actors: [{ id: a, agent: local }] +steps: + - command_id: start + start: { actor: a, playback_id: vod, session_group: vod, vod_object: movie.mkv, expect_evicted: true, post_read_action: close } +", + ) + .map_err(|error| TestkitError::Configuration(error.to_string()))?; + assert!(scenario.validate().is_err()); + + scenario.steps[0] + .start + .as_mut() + .ok_or_else(|| TestkitError::Configuration("missing start".to_owned()))? + .post_read_action = Some(PostReadAction::Pause); + assert!(scenario.validate().is_err()); + scenario.steps[0] + .start + .as_mut() + .ok_or_else(|| TestkitError::Configuration("missing start".to_owned()))? + .post_read_action = Some(PostReadAction::KeepOpen); + assert!(scenario.validate().is_ok()); + scenario.steps[0] + .start + .as_mut() + .ok_or_else(|| TestkitError::Configuration("missing start".to_owned()))? + .method = Some("HEAD".to_owned()); + assert!(scenario.validate().is_err()); + Ok(()) + } + #[test] fn duplicate_playbacks_are_rejected() { let scenario = Scenario { @@ -639,6 +702,7 @@ mod tests { "vod-range-reopen-strict-cap.yml", "vod-reopen-backpressured-body.yml", "live-ts-same-channel-retry-latest-wins.yml", + "live-ts-channel-switch-single-user-two-provider-slots.yml", "reentry-suppresses-evicted-retry.yml", "soft-slot-precedes-eviction.yml", "soft-slot-exhausts-without-upstream-leak.yml", diff --git a/tools/tuliprox-testkit/src/main.rs b/tools/tuliprox-testkit/src/main.rs index e11e39d2b..dd12cd401 100644 --- a/tools/tuliprox-testkit/src/main.rs +++ b/tools/tuliprox-testkit/src/main.rs @@ -1128,7 +1128,17 @@ async fn run_agent_until_released( chunk = tokio::time::timeout(PLAYBACK_IDLE_TIMEOUT, body.next()) => { let chunk = match chunk { Ok(Some(Ok(chunk))) => chunk, - Ok(Some(Err(err))) => return Ok(PlaybackOutcome::TransportError { message: err.to_string() }), + Ok(Some(Err(err))) => { + return Ok(if ready.is_none() { + PlaybackOutcome::StreamInterrupted { + frames: received_frames, + bytes: received_bytes, + message: err.to_string(), + } + } else { + PlaybackOutcome::TransportError { message: err.to_string() } + }); + } Ok(None) => return Ok(PlaybackOutcome::UnexpectedEof { frames: received_frames, bytes: received_bytes }), Err(_) => return Ok(PlaybackOutcome::IdleTimeout), }; @@ -1212,6 +1222,62 @@ struct HeldPlayback { task: tokio::task::JoinHandle>, } +fn matches_expected_eviction( + playback_id: &str, + expected_terminations: &HashSet, + outcome: &PlaybackOutcome, + kicked_session: bool, +) -> bool { + expected_terminations.contains(playback_id) + && (matches!(outcome, PlaybackOutcome::UnexpectedEof { .. }) + || (kicked_session && matches!(outcome, PlaybackOutcome::StreamInterrupted { .. }))) +} + +fn observe_new_stream_session( + streams: &serde_json::Value, + username: &str, + seen_uids: &mut HashSet, +) -> Option { + let mut session_id = None; + for stream in streams.as_array()? { + let (Some(uid), Some(ts)) = (stream["uid"].as_u64(), stream["ts"].as_u64()) else { + continue; + }; + let (Ok(uid), Ok(ts)) = (u32::try_from(uid), u32::try_from(ts)) else { + continue; + }; + if seen_uids.insert(uid) && stream["username"].as_str() == Some(username) { + if session_id.is_some() { + return None; + } + session_id = Some((u64::from(ts) << 32) | u64::from(uid)); + } + } + session_id +} + +fn history_confirms_kick(history: &serde_json::Value, session_id: u64) -> bool { + history["items"].as_array().is_some_and(|items| { + items.iter().any(|item| { + item["session_id"].as_u64() == Some(session_id) + && item["disconnect_reason"].as_str() == Some("client_kicked") + }) + }) +} + +async fn wait_for_kicked_session(observer: &TuliproxObserver, session_id: u64) -> bool { + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + loop { + if observer.stream_history(session_id).await.is_ok_and(|history| history_confirms_kick(&history, session_id)) { + return true; + } + if tokio::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } +} + #[derive(Clone)] pub struct OriginObserver { control_base_url: String, @@ -2181,6 +2247,8 @@ async fn execute_scenario_steps<'a>( let mut remote_playbacks: HashMap = HashMap::new(); let mut expected_terminations = HashSet::new(); let mut playback_users = HashMap::new(); + let mut seen_stream_uids = HashSet::new(); + let mut playback_session_ids = HashMap::new(); let mut admission_oracles = scenario.policy_contract.as_ref().map(|contract| { contract .users @@ -2473,6 +2541,9 @@ async fn execute_scenario_steps<'a>( start.playback_id.as_str(), if step.expect.is_streaming() { "received_valid_frames" } else { "rejected_as_expected" }, )); + if start.expect_evicted && actual_streaming { + expected_terminations.insert(start.playback_id.clone()); + } if remote_outcome.as_ref().is_some_and(PlaybackOutcome::is_streaming) { remote_playbacks.insert(start.playback_id.clone(), actor.agent.clone()); } @@ -2520,6 +2591,15 @@ async fn execute_scenario_steps<'a>( } if let Ok(snapshot) = observer.runtime_snapshot().await { events.push((start.playback_id.as_str(), "runtime_observed")); + if actual_streaming && actor.agent == "local" { + if let Some(username) = actor.username.as_deref() { + if let Some(session_id) = + observe_new_stream_session(&snapshot.streams, username, &mut seen_stream_uids) + { + playback_session_ids.insert(start.playback_id.clone(), session_id); + } + } + } if let Some(policy) = &scenario.policy_contract { if let Ok(slots) = provider_slot_count(&snapshot.status) { if policy.provider_capacity().is_some_and(|limit| slots > limit) { @@ -2568,7 +2648,23 @@ async fn execute_scenario_steps<'a>( events.push((event_playback_id, "unexpected_clean_terminal")); any_failed = true; } - Ok(Ok(Ok(_))) => events.push((event_playback_id, "evicted")), + Ok(Ok(Ok(outcome))) => { + let kicked_session = if matches!(outcome, PlaybackOutcome::StreamInterrupted { .. }) { + if let Some(session_id) = playback_session_ids.get(&held.playback_id) { + wait_for_kicked_session(observer, *session_id).await + } else { + false + } + } else { + false + }; + if matches_expected_eviction(&held.playback_id, &expected_terminations, &outcome, kicked_session) { + events.push((event_playback_id, "evicted")); + } else { + events.push((event_playback_id, "unexpected_eviction_terminal")); + any_failed = true; + } + } Ok(Ok(Err(_)) | Err(_)) => { events.push((event_playback_id, "unexpected_eviction_terminal")); any_failed = true; @@ -2924,6 +3020,49 @@ fn marker_from_url(url: &str) -> Result { mod tests { use super::*; + #[test] + fn expected_eviction_requires_matching_playback_stream_termination() { + let expected = HashSet::from(["playback-a".to_owned()]); + let eof = PlaybackOutcome::UnexpectedEof { frames: 3, bytes: 100 }; + assert!(matches_expected_eviction("playback-a", &expected, &eof, false)); + let interrupted = PlaybackOutcome::StreamInterrupted { + frames: 3, + bytes: 100, + message: "error decoding response body".to_owned(), + }; + assert!(!matches_expected_eviction("playback-a", &expected, &interrupted, false)); + assert!(matches_expected_eviction("playback-a", &expected, &interrupted, true)); + assert!(!matches_expected_eviction("playback-b", &expected, &interrupted, true)); + assert!(!matches_expected_eviction("playback-b", &expected, &eof, false)); + assert!(!matches_expected_eviction("playback-a", &expected, &PlaybackOutcome::IdleTimeout, true)); + assert!(!matches_expected_eviction( + "playback-a", + &expected, + &PlaybackOutcome::TransportError { message: "connection failed".to_owned() }, + true, + )); + } + + #[test] + fn interrupted_stream_needs_matching_kick_record() { + let streams = serde_json::json!([ + {"uid": 17, "ts": 123, "username": "alice"}, + {"uid": 18, "ts": 123, "username": "bob"} + ]); + let mut seen = HashSet::new(); + let session_id = observe_new_stream_session(&streams, "alice", &mut seen); + assert_eq!(session_id, Some((123_u64 << 32) | 17)); + let history = serde_json::json!({"items": [ + {"session_id": (123_u64 << 32) | 18, "disconnect_reason": "client_kicked"}, + {"session_id": (123_u64 << 32) | 17, "disconnect_reason": "client_closed"} + ]}); + assert!(!history_confirms_kick(&history, session_id.unwrap_or_default())); + let kicked = serde_json::json!({"items": [ + {"session_id": (123_u64 << 32) | 17, "disconnect_reason": "client_kicked"} + ]}); + assert!(history_confirms_kick(&kicked, session_id.unwrap_or_default())); + } + fn test_origin_state(run_id: &str) -> OriginState { OriginState { run_id: RunId::new(run_id), diff --git a/tools/tuliprox-testkit/src/observation.rs b/tools/tuliprox-testkit/src/observation.rs index bba8f843b..4b957a91f 100644 --- a/tools/tuliprox-testkit/src/observation.rs +++ b/tools/tuliprox-testkit/src/observation.rs @@ -27,7 +27,15 @@ impl TuliproxObserver { pub async fn streams(&self) -> Result { self.get_json("streams").await } - pub async fn stream_history(&self) -> Result { self.get_json("stream-history").await } + pub async fn stream_history(&self, session_id: u64) -> Result { + let timestamp = i64::try_from(session_id >> 32) + .map_err(|error| TestkitError::Protocol(format!("invalid stream history session timestamp: {error}")))?; + let date = chrono::DateTime::from_timestamp(timestamp, 0) + .ok_or_else(|| TestkitError::Protocol("invalid stream history session timestamp".to_owned()))? + .format("%Y-%m-%d"); + let today = chrono::Utc::now().format("%Y-%m-%d"); + self.get_json(&format!("stream-history?from={date}&to={today}")).await + } pub async fn runtime_snapshot(&self) -> Result { let (status, streams) = tokio::join!(self.status(), self.streams()); diff --git a/tools/tuliprox-testkit/src/protocol.rs b/tools/tuliprox-testkit/src/protocol.rs index e45d7747a..22f98258f 100644 --- a/tools/tuliprox-testkit/src/protocol.rs +++ b/tools/tuliprox-testkit/src/protocol.rs @@ -105,6 +105,7 @@ pub enum PlaybackOutcome { AdmissionRejected { reason: RejectionReason }, HttpError { status: u16 }, TransportError { message: String }, + StreamInterrupted { frames: u64, bytes: u64, message: String }, InvalidData { message: String }, UnexpectedEof { frames: u64, bytes: u64 }, IdleTimeout, @@ -127,6 +128,9 @@ impl std::fmt::Display for PlaybackOutcome { Self::AdmissionRejected { reason } => write!(f, "rejected({reason})"), Self::HttpError { status } => write!(f, "http_error({status})"), Self::TransportError { message } => write!(f, "transport_error({message})"), + Self::StreamInterrupted { frames, bytes, message } => { + write!(f, "stream_interrupted({frames} frames, {bytes} bytes, {message})") + } Self::InvalidData { message } => write!(f, "invalid_data({message})"), Self::UnexpectedEof { frames, bytes } => write!(f, "unexpected_eof({frames} frames, {bytes} bytes)"), Self::IdleTimeout => write!(f, "idle_timeout"),