diff --git a/apps/channels/models.py b/apps/channels/models.py index d09a50f5..e84f9768 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -607,11 +607,22 @@ class Channel(models.Model): ChannelState.CONNECTING, ) - def _clear_stream_assignment_keys(self, redis_client, stream_id=None) -> None: - """Remove channel/stream profile assignment keys from Redis.""" + def _release_stale_stream_assignment(self, redis_client, stream_id: int) -> None: + """Release pool counters and remove stale channel/stream assignment keys.""" + profile_id_bytes = redis_client.get(f"stream_profile:{stream_id}") + if profile_id_bytes: + try: + profile_id = int(profile_id_bytes) + release_profile_slot(profile_id, redis_client) + except (ValueError, TypeError): + logger.debug( + "Invalid profile ID for stale assignment on stream %s: %s", + stream_id, + profile_id_bytes, + ) + redis_client.delete(f"channel_stream:{self.id}") - if stream_id is not None: - redis_client.delete(f"stream_profile:{stream_id}") + redis_client.delete(f"stream_profile:{stream_id}") def get_stream(self, requester=None): """ @@ -659,10 +670,10 @@ class Channel(models.Model): ) else: logger.info( - f"Channel {self.uuid}: clearing stale stream assignment " + f"Channel {self.uuid}: releasing stale stream assignment " f"(stream={stream_id}, proxy not active)" ) - self._clear_stream_assignment_keys(redis_client, stream_id) + self._release_stale_stream_assignment(redis_client, stream_id) # No existing active stream, attempt to assign a new one has_streams_but_maxed_out = False @@ -886,9 +897,29 @@ class Channel(models.Model): if current_profile_id == new_profile_id: return True - from apps.m3u.connection_pool import profile_connections_key + from apps.m3u.connection_pool import ( + move_credential_slot_on_profile_switch, + profile_connections_key, + ) + from apps.m3u.models import M3UAccountProfile - # Profile counters always move on switch; group totals stay unchanged (one stream). + old_profile = M3UAccountProfile.objects.select_related( + "m3u_account__server_group" + ).get(id=current_profile_id) + new_profile = M3UAccountProfile.objects.select_related( + "m3u_account__server_group" + ).get(id=new_profile_id) + + if not move_credential_slot_on_profile_switch( + old_profile, new_profile, redis_client + ): + logger.warning( + "Shared login pool full for profile %s during stream profile switch", + new_profile_id, + ) + return False + + # Profile counters always move on switch; credential totals move only when login changes. old_profile_connections_key = profile_connections_key(current_profile_id) new_profile_connections_key = profile_connections_key(new_profile_id) old_count = int(redis_client.get(old_profile_connections_key) or 0) diff --git a/apps/m3u/connection_pool.py b/apps/m3u/connection_pool.py index 36a74d17..2d4c5830 100644 --- a/apps/m3u/connection_pool.py +++ b/apps/m3u/connection_pool.py @@ -2,8 +2,7 @@ Shared connection pool enforcement for M3U accounts in the same ServerGroup. Profile selection rotates across M3UAccountProfile rows using each profile's own -Redis counter (the pre-pool behavior). When an account belongs to a ServerGroup -with max_streams > 0, a credential-scoped counter is checked on reserve/release +Redis counter (the pre-pool behavior). When an account belongs to a ServerGroup, a credential-scoped counter is checked on reserve/release so accounts sharing the same provider login share one limit without blocking unrelated logins on the same group. Account profiles with max_streams=0 skip credential enforcement for that profile. @@ -129,9 +128,9 @@ def get_profile_credential_fingerprint(profile) -> Optional[str]: def get_enforced_server_group_for_profile(profile): - """Return the shared ServerGroup limit for this profile's account, if configured.""" + """Return the ServerGroup for credential pooling when the account is assigned to one.""" group = profile.m3u_account.server_group - if group and group.max_streams > 0: + if group: return group return None @@ -157,11 +156,6 @@ def get_credential_connection_count(profile, redis_client) -> int: return int(redis_client.get(cred_key) or 0) -def get_group_connection_count(profile, redis_client) -> int: - """Backwards-compatible alias for credential-scoped group usage.""" - return get_credential_connection_count(profile, redis_client) - - def profile_has_capacity_for_selection(profile, redis_client) -> bool: """Per-profile capacity check used when rotating across profiles on one account.""" if profile.max_streams == 0: @@ -188,6 +182,55 @@ def pool_has_capacity_for_profile(profile, redis_client) -> bool: ) +def profile_available_for_channel_switch( + profile, redis_client, *, channel_already_on_profile: bool +) -> bool: + """ + Non-mutating capacity check when selecting a profile for an in-flight channel. + + If the channel already holds this profile's slots, skip re-checking capacity. + """ + if channel_already_on_profile: + return True + return pool_has_capacity_for_profile(profile, redis_client) + + +def move_credential_slot_on_profile_switch( + old_profile, new_profile, redis_client +) -> bool: + """ + Move the shared credential counter when switching to a different provider login. + + Profile counters are managed separately by Channel.update_stream_profile(). + Returns False when the new profile's credential pool is full. + """ + old_fp = get_profile_credential_fingerprint(old_profile) + new_fp = get_profile_credential_fingerprint(new_profile) + if old_fp == new_fp: + return True + + released = _release_credential_slot_by_profile_id(old_profile.id, redis_client) + if not released: + _release_server_group_slot_for_profile(old_profile, redis_client) + + cred_reserved, cred_key = _reserve_server_group_slot_for_profile( + new_profile, redis_client + ) + if not cred_reserved: + restore_reserved, restore_key = _reserve_server_group_slot_for_profile( + old_profile, redis_client + ) + if restore_reserved and restore_key: + _remember_credential_release_key( + old_profile.id, restore_key, redis_client + ) + return False + + if cred_key: + _remember_credential_release_key(new_profile.id, cred_key, redis_client) + return True + + def _safe_decr(redis_client, key: str) -> None: current = int(redis_client.get(key) or 0) if current <= 0: diff --git a/apps/m3u/migrations/0020_servergroup_max_streams.py b/apps/m3u/migrations/0020_servergroup_max_streams.py deleted file mode 100644 index ce17d830..00000000 --- a/apps/m3u/migrations/0020_servergroup_max_streams.py +++ /dev/null @@ -1,21 +0,0 @@ -# Generated by Django 6.0.3 - -from django.db import migrations, models - - -class Migration(migrations.Migration): - - dependencies = [ - ("m3u", "0019_m3uaccountprofile_exp_date"), - ] - - operations = [ - migrations.AddField( - model_name="servergroup", - name="max_streams", - field=models.PositiveIntegerField( - default=0, - help_text="Maximum number of concurrent streams shared across all accounts in this group (0 for unlimited)", - ), - ), - ] diff --git a/apps/m3u/models.py b/apps/m3u/models.py index 46d1a068..53c18d38 100644 --- a/apps/m3u/models.py +++ b/apps/m3u/models.py @@ -194,19 +194,17 @@ class M3UFilter(models.Model): class ServerGroup(models.Model): - """Represents a logical grouping of servers or channels.""" + """ + Groups M3U accounts that share provider credentials. + + Accounts assigned to the same server group share credential-scoped connection + counters when their logins match. Limits come from each account profile's + max_streams, not from the group itself. + """ name = models.CharField( max_length=100, unique=True, help_text="Unique name for this server group." ) - max_streams = models.PositiveIntegerField( - default=0, - help_text=( - "Set above 0 to enable shared credential pooling for accounts in this group. " - "Per-login limits come from each account profile's max_streams. " - "Profiles with max_streams=0 (unlimited) skip cross-account credential enforcement." - ), - ) def __str__(self): return self.name diff --git a/apps/m3u/serializers.py b/apps/m3u/serializers.py index 29c4157f..485a3687 100644 --- a/apps/m3u/serializers.py +++ b/apps/m3u/serializers.py @@ -408,4 +408,4 @@ class ServerGroupSerializer(serializers.ModelSerializer): class Meta: model = ServerGroup - fields = ["id", "name", "max_streams"] + fields = ["id", "name"] diff --git a/apps/m3u/tests/test_connection_pool.py b/apps/m3u/tests/test_connection_pool.py index e5de6c10..5b05535a 100644 --- a/apps/m3u/tests/test_connection_pool.py +++ b/apps/m3u/tests/test_connection_pool.py @@ -7,7 +7,6 @@ from apps.m3u.connection_pool import ( extract_credentials_from_stream_url, get_credential_connection_count, get_enforced_server_group_for_profile, - get_group_connection_count, get_profile_connection_count, get_profile_credential_fingerprint, group_has_capacity_for_profile, @@ -94,8 +93,8 @@ class ExtractCredentialsTests(TestCase): class ManualServerGroupTests(TestCase): - def test_group_enforced_when_max_streams_set(self): - group = ServerGroup.objects.create(name="provider-a", max_streams=2) + def test_group_enforced_when_account_assigned(self): + group = ServerGroup.objects.create(name="provider-a") account = M3UAccount.objects.create( name="Account A", account_type="XC", @@ -108,7 +107,7 @@ class ManualServerGroupTests(TestCase): self.assertEqual(get_enforced_server_group_for_profile(profile), group) def test_accounts_in_same_group_share_credential_counter(self): - group = ServerGroup.objects.create(name="shared", max_streams=2) + group = ServerGroup.objects.create(name="shared") account1 = M3UAccount.objects.create( name="XC Account", account_type="XC", @@ -171,10 +170,7 @@ class ManualServerGroupTests(TestCase): class PoolEnforcementTests(TestCase): def setUp(self): self.redis = FakeRedis() - self.group = ServerGroup.objects.create( - name="test-pool", - max_streams=2, - ) + self.group = ServerGroup.objects.create(name="test-pool") self.account = M3UAccount.objects.create( name="Test Account", account_type="XC", @@ -207,8 +203,8 @@ class PoolEnforcementTests(TestCase): self.assertEqual(self.redis._data[cred_key], 0) self.assertEqual(self.redis._data[profile_key], 0) - def test_same_credential_capped_at_profile_max_not_group_max(self): - """Maintainer example: group max=2 but each login only allows 1.""" + def test_same_credential_capped_at_profile_max(self): + """Shared credential counter is capped by each profile's max_streams.""" account2 = M3UAccount.objects.create( name="Second Account", account_type="XC", @@ -229,11 +225,8 @@ class PoolEnforcementTests(TestCase): cred_key = server_group_connections_key(self.group.id, fp) self.assertEqual(self.redis._data[cred_key], 1) - def test_different_logins_both_stream_when_group_max_is_one(self): - """Regression: group max=1 must not block a second login on another profile.""" - self.group.max_streams = 1 - self.group.save() - + def test_different_logins_both_stream_in_same_group(self): + """Different provider logins keep separate credential counters in one group.""" account = M3UAccount.objects.create( name="Grouped multi-login", account_type="XC", @@ -288,7 +281,7 @@ class PoolEnforcementTests(TestCase): ): self.assertTrue(reserve_profile_slot(profile, self.redis)[0]) self.assertEqual(get_credential_connection_count(profile, self.redis), 0) - self.assertEqual(get_group_connection_count(profile, self.redis), 0) + self.assertEqual(get_credential_connection_count(profile, self.redis), 0) def test_release_when_profile_row_deleted(self): profile_id = self.profile.id @@ -328,12 +321,55 @@ class PoolEnforcementTests(TestCase): self.assertEqual(reason, "credential_full") +class StaleAssignmentTests(TestCase): + def test_stale_assignment_releases_counters(self): + from apps.channels.models import Channel, Stream + + redis = FakeRedis() + group = ServerGroup.objects.create(name="stale-group") + account = M3UAccount.objects.create( + name="Stale Account", + account_type="XC", + username="user", + password="pass", + server_url="http://xc.example.com", + server_group=group, + max_streams=5, + ) + profile = M3UAccountProfile.objects.get(m3u_account=account, is_default=True) + profile.max_streams = 1 + profile.save() + + stream = Stream.objects.create(name="Stale Stream", m3u_account=account) + channel = Channel.objects.create(channel_number=502, name="Stale Channel") + channel.streams.add(stream) + + reserve_profile_slot(profile, redis) + redis.set(f"channel_stream:{channel.id}", stream.id) + redis.set(f"stream_profile:{stream.id}", profile.id) + + profile_key = profile_connections_key(profile.id) + cred_key = server_group_connections_key( + group.id, get_profile_credential_fingerprint(profile) + ) + self.assertEqual(redis._data[profile_key], 1) + self.assertEqual(redis._data[cred_key], 1) + + with patch("core.utils.RedisClient.get_client", return_value=redis): + channel._release_stale_stream_assignment(redis, stream.id) + + self.assertNotIn(f"channel_stream:{channel.id}", redis._data) + self.assertNotIn(f"stream_profile:{stream.id}", redis._data) + self.assertEqual(redis._data[profile_key], 0) + self.assertEqual(redis._data[cred_key], 0) + + class UpdateStreamProfileTests(TestCase): def test_switch_updates_profile_counters_when_group_assigned(self): from apps.channels.models import Channel, Stream redis = FakeRedis() - group = ServerGroup.objects.create(name="switch-group", max_streams=2) + group = ServerGroup.objects.create(name="switch-group") account = M3UAccount.objects.create( name="Switch Account", account_type="XC", @@ -377,6 +413,58 @@ class UpdateStreamProfileTests(TestCase): self.assertEqual(redis._data[profile_connections_key(profile_b.id)], 1) self.assertEqual(redis._data[cred_key], cred_before) + def test_switch_moves_credential_counter_when_login_changes(self): + from apps.channels.models import Channel, Stream + + redis = FakeRedis() + group = ServerGroup.objects.create(name="cred-switch-group") + account = M3UAccount.objects.create( + name="Cred Switch Account", + account_type="XC", + username="login_a", + password="pass_a", + server_url="http://xc.example.com", + server_group=group, + max_streams=5, + ) + profile_a = M3UAccountProfile.objects.get(m3u_account=account, is_default=True) + profile_a.max_streams = 1 + profile_a.save() + profile_b = M3UAccountProfile.objects.create( + m3u_account=account, + name="alt_login", + is_default=False, + is_active=True, + max_streams=1, + search_pattern="", + replace_pattern="", + ) + + fp_a = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + fp_b = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + key_a = server_group_connections_key(group.id, fp_a) + key_b = server_group_connections_key(group.id, fp_b) + + stream = Stream.objects.create(name="Cred Switch Stream", m3u_account=account) + channel = Channel.objects.create(channel_number=503, name="Cred Switch Channel") + channel.streams.add(stream) + + with patch( + "apps.m3u.connection_pool.get_profile_credential_fingerprint", + side_effect=lambda profile: fp_a if profile.id == profile_a.id else fp_b, + ): + reserve_profile_slot(profile_a, redis) + redis.set(f"channel_stream:{channel.id}", stream.id) + redis.set(f"stream_profile:{stream.id}", profile_a.id) + self.assertEqual(redis._data[key_a], 1) + self.assertNotIn(key_b, redis._data) + + with patch("core.utils.RedisClient.get_client", return_value=redis): + self.assertTrue(channel.update_stream_profile(profile_b.id)) + + self.assertEqual(redis._data[key_a], 0) + self.assertEqual(redis._data[key_b], 1) + class VodProfileSelectionTests(TestCase): def test_get_m3u_profile_skips_default_when_profile_full(self): @@ -413,3 +501,66 @@ class VodProfileSelectionTests(TestCase): self.assertIsNotNone(result) selected, _connections = result self.assertEqual(selected.id, alt.id) + + def test_get_m3u_profile_skips_default_when_credential_pool_full(self): + from apps.proxy.vod_proxy.views import _get_m3u_profile + + group = ServerGroup.objects.create(name="vod-cred-pool") + account = M3UAccount.objects.create( + name="VOD pooled", + account_type="XC", + username="shared_user", + password="shared_pass", + server_url="http://xc.example.com", + server_group=group, + max_streams=5, + ) + default = M3UAccountProfile.objects.get(m3u_account=account, is_default=True) + default.max_streams = 1 + default.save() + + alt = M3UAccountProfile.objects.create( + m3u_account=account, + name="alt_login", + is_default=False, + is_active=True, + max_streams=1, + search_pattern="", + replace_pattern="", + ) + + other_account = M3UAccount.objects.create( + name="Other pooled account", + account_type="XC", + username="shared_user", + password="shared_pass", + server_url="http://xc.example.com", + server_group=group, + max_streams=5, + ) + other_profile = M3UAccountProfile.objects.get( + m3u_account=other_account, is_default=True + ) + other_profile.max_streams = 1 + other_profile.save() + + fp_shared = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc" + fp_alt = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd" + + redis = FakeRedis() + with patch( + "apps.m3u.connection_pool.get_profile_credential_fingerprint", + side_effect=lambda profile: ( + fp_shared + if profile.id in (default.id, other_profile.id) + else fp_alt + ), + ): + self.assertTrue(reserve_profile_slot(other_profile, redis)[0]) + + with patch("core.utils.RedisClient.get_client", return_value=redis): + result = _get_m3u_profile(account, None, None) + + self.assertIsNotNone(result) + selected, _connections = result + self.assertEqual(selected.id, alt.id) diff --git a/apps/proxy/live_proxy/url_utils.py b/apps/proxy/live_proxy/url_utils.py index 60d0e3d0..899ffc9a 100644 --- a/apps/proxy/live_proxy/url_utils.py +++ b/apps/proxy/live_proxy/url_utils.py @@ -8,7 +8,10 @@ from typing import Optional, Tuple, List from django.shortcuts import get_object_or_404 from apps.channels.models import Channel, Stream from apps.m3u.models import M3UAccount, M3UAccountProfile -from apps.m3u.connection_pool import profile_connections_key +from apps.m3u.connection_pool import ( + get_profile_connection_count, + profile_available_for_channel_switch, +) from core.models import UserAgent, CoreSettings, StreamProfile from .utils import get_logger from uuid import UUID @@ -54,12 +57,14 @@ def get_stream_object(id: str): logger.info(f"Fetching stream hash {id}") return get_object_or_404(Stream, stream_hash=id) -def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int], bool]: +def generate_stream_url( + channel_id: str, +) -> Tuple[str, str, bool, Optional[int], bool, Optional[str]]: """ Generate the appropriate stream URL for a channel or stream based on its profile settings. Returns: - Tuple: (stream_url, user_agent, transcode_flag, profile_id, slot_reserved) + Tuple: (stream_url, user_agent, transcode_flag, profile_id, slot_reserved, error_reason) """ try: channel_or_stream = get_stream_object(channel_id) @@ -71,12 +76,12 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int], if not stream.m3u_account: logger.error(f"Stream {stream.id} has no M3U account") - return None, None, False, None, False + return None, None, False, None, False, "Stream has no M3U account" stream_id, profile_id, error_reason, slot_reserved = stream.get_stream() if not stream_id or not profile_id: logger.error(f"No profile available for stream {stream.id}: {error_reason}") - return None, None, False, None, False + return None, None, False, None, False, error_reason try: profile = M3UAccountProfile.objects.get(id=profile_id) @@ -95,24 +100,23 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int], transcode = not stream_profile.is_proxy() stream_profile_id = stream_profile.id - return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved + return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved, None except Exception as e: logger.error(f"Error generating stream URL for stream {stream.id}: {e}") if slot_reserved: stream.release_stream() - return None, None, False, None, False + return None, None, False, None, False, str(e) # Handle channel preview (existing logic) channel = channel_or_stream # Get stream and profile for this channel - # Note: get_stream now returns 3 values (stream_id, profile_id, error_reason) stream_id, profile_id, error_reason, slot_reserved = channel.get_stream() if not stream_id or not profile_id: logger.error(f"No stream available for channel {channel_id}: {error_reason}") - return None, None, False, None, False + return None, None, False, None, False, error_reason # get_stream() allocated a connection slot - ensure it's released on any error try: @@ -142,16 +146,16 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int], stream_profile_id = stream_profile.id - return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved + return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved, None except Exception as e: logger.error(f"Error generating stream URL for channel {channel_id}: {e}") if slot_reserved: if not channel.release_stream(): logger.warning(f"Failed to release stream for channel {channel_id} after URL generation error") - return None, None, False, None, False + return None, None, False, None, False, str(e) except Exception as e: logger.error(f"Error generating stream URL: {e}") - return None, None, False, None, False + return None, None, False, None, False, str(e) def transform_url(input_url: str, search_pattern: str, replace_pattern: str) -> str: """ @@ -229,10 +233,6 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] selected_profile = None for profile in profiles: if redis_client: - current_connections = int( - redis_client.get(profile_connections_key(profile.id)) or 0 - ) - channel_using_profile = False existing_stream_id = redis_client.get(f"channel_stream:{channel.id}") if existing_stream_id: @@ -242,20 +242,22 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] if existing_profile_id and int(existing_profile_id) == profile.id: channel_using_profile = True - effective_connections = current_connections - ( - 1 if channel_using_profile else 0 - ) - - if profile.max_streams == 0 or effective_connections < profile.max_streams: + if profile_available_for_channel_switch( + profile, + redis_client, + channel_already_on_profile=channel_using_profile, + ): + current_connections = get_profile_connection_count( + profile, redis_client + ) selected_profile = profile logger.debug( f"Selected profile {profile.id} with " - f"{effective_connections}/{profile.max_streams} effective connections" + f"{current_connections}/{profile.max_streams} connections" ) break logger.debug( - f"Profile {profile.id} at max connections: " - f"{effective_connections}/{profile.max_streams}" + f"Profile {profile.id} unavailable for channel switch" ) else: selected_profile = profile @@ -368,38 +370,38 @@ def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = No selected_profile = None for profile in profiles: - # Check connection availability if redis_client: - profile_connections_key = f"profile_connections:{profile.id}" - current_connections = int(redis_client.get(profile_connections_key) or 0) - - # Check if this channel is already using this profile channel_using_profile = False existing_stream_id = redis_client.get(f"channel_stream:{channel.id}") if existing_stream_id: - # Decode bytes to string/int for proper Redis key lookup - existing_stream_id = existing_stream_id - existing_profile_id = redis_client.get(f"stream_profile:{existing_stream_id}") + existing_profile_id = redis_client.get( + f"stream_profile:{existing_stream_id}" + ) if existing_profile_id and int(existing_profile_id) == profile.id: channel_using_profile = True - logger.debug(f"Channel {channel.id} already using profile {profile.id}") + logger.debug( + f"Channel {channel.id} already using profile {profile.id}" + ) - # Calculate effective connections (subtract 1 if channel already using this profile) - effective_connections = current_connections - (1 if channel_using_profile else 0) - - # Check if profile has available slots - if profile.max_streams == 0 or effective_connections < profile.max_streams: + if profile_available_for_channel_switch( + profile, + redis_client, + channel_already_on_profile=channel_using_profile, + ): + current_connections = get_profile_connection_count( + profile, redis_client + ) selected_profile = profile logger.debug( f"Found available profile {profile.id} for stream {stream.id}: " - f"{effective_connections}/{profile.max_streams} effective " - f"(current: {current_connections}, already using: {channel_using_profile})" + f"{current_connections}/{profile.max_streams} " + f"(already using: {channel_using_profile})" ) break - else: - logger.debug(f"Profile {profile.id} at max connections: {effective_connections}/{profile.max_streams} (current: {current_connections}, already using: {channel_using_profile})") + logger.debug( + f"Profile {profile.id} unavailable for alternate stream {stream.id}" + ) else: - # No Redis available, assume first active profile is okay selected_profile = profile break diff --git a/apps/proxy/live_proxy/views.py b/apps/proxy/live_proxy/views.py index eb845743..d60d9703 100644 --- a/apps/proxy/live_proxy/views.py +++ b/apps/proxy/live_proxy/views.py @@ -221,9 +221,14 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): # Try to get a stream with fixed interval retries while should_retry and time.time() - wait_start_time < retry_timeout: attempt += 1 - stream_url, stream_user_agent, transcode, profile_value, slot_reserved = ( - generate_stream_url(channel_id) - ) + ( + stream_url, + stream_user_agent, + transcode, + profile_value, + slot_reserved, + error_reason, + ) = generate_stream_url(channel_id) if stream_url is not None: logger.info( @@ -233,7 +238,6 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): # On first failure, check if the error is retryable if attempt == 1: - _, _, error_reason, _ = channel.get_stream() if error_reason and "maximum connection limits" not in error_reason: logger.warning( f"[{client_id}] Can't retry - error not related to connection limits: {error_reason}" @@ -266,9 +270,14 @@ def stream_ts(request, channel_id, user=None, force_output_format=None): logger.info( f"[{client_id}] Making final attempt {attempt} at timeout boundary" ) - stream_url, stream_user_agent, transcode, profile_value, slot_reserved = ( - generate_stream_url(channel_id) - ) + ( + stream_url, + stream_user_agent, + transcode, + profile_value, + slot_reserved, + error_reason, + ) = generate_stream_url(channel_id) if stream_url is not None: logger.info( f"[{client_id}] Successfully obtained stream on final attempt for channel {channel_id}" diff --git a/apps/proxy/vod_proxy/views.py b/apps/proxy/vod_proxy/views.py index c4e84986..fc825cf4 100644 --- a/apps/proxy/vod_proxy/views.py +++ b/apps/proxy/vod_proxy/views.py @@ -262,7 +262,7 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None): from core.utils import RedisClient from apps.m3u.connection_pool import ( get_profile_connection_count, - profile_has_capacity_for_selection, + pool_has_capacity_for_profile, ) redis_client = RedisClient.get_client() @@ -316,7 +316,7 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None): profile_connections_key = f"profile_connections:{profile.id}" current_connections = get_profile_connection_count(profile, redis_client) - if profile_has_capacity_for_selection(profile, redis_client): + if pool_has_capacity_for_profile(profile, redis_client): logger.info(f"[PROFILE-SELECTION] Using requested profile {profile.id}: {current_connections}/{profile.max_streams} connections") return (profile, current_connections) logger.warning(f"[PROFILE-SELECTION] Requested profile {profile.id} is at capacity: {current_connections}/{profile.max_streams}") @@ -340,11 +340,14 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None): for profile in profiles: current_connections = get_profile_connection_count(profile, redis_client) - if profile_has_capacity_for_selection(profile, redis_client): + if pool_has_capacity_for_profile(profile, redis_client): logger.info(f"[PROFILE-SELECTION] Selected profile {profile.id} ({profile.name}): {current_connections}/{profile.max_streams} connections") return (profile, current_connections) else: - logger.debug(f"[PROFILE-SELECTION] Profile {profile.id} at capacity: {current_connections}/{profile.max_streams}") + logger.debug( + f"[PROFILE-SELECTION] Profile {profile.id} unavailable " + f"(profile={current_connections}/{profile.max_streams})" + ) # All profiles are at capacity - return None to trigger error response logger.error(f"[PROFILE-SELECTION] All profiles at capacity for M3U account {m3u_account.id}, rejecting request") diff --git a/frontend/src/components/ServerGroupsManagerModal.jsx b/frontend/src/components/ServerGroupsManagerModal.jsx new file mode 100644 index 00000000..5e3005ec --- /dev/null +++ b/frontend/src/components/ServerGroupsManagerModal.jsx @@ -0,0 +1,28 @@ +import { Modal } from '@mantine/core'; +import ServerGroupsTable from './tables/ServerGroupsTable'; + +const ServerGroupsManagerModal = ({ + isOpen, + onClose, + onGroupCreated, + openCreateOnMount = false, +}) => { + return ( + + {isOpen ? ( + + ) : null} + + ); +}; + +export default ServerGroupsManagerModal; diff --git a/frontend/src/components/__tests__/ServerGroupsManagerModal.test.jsx b/frontend/src/components/__tests__/ServerGroupsManagerModal.test.jsx new file mode 100644 index 00000000..a60ae4db --- /dev/null +++ b/frontend/src/components/__tests__/ServerGroupsManagerModal.test.jsx @@ -0,0 +1,59 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { render, screen, waitFor } from '@testing-library/react'; +import { MantineProvider } from '@mantine/core'; +import ServerGroupsManagerModal from '../ServerGroupsManagerModal'; + +vi.mock('../../api', () => ({ + default: { + deleteServerGroup: vi.fn(), + }, +})); + +vi.mock('../../store/serverGroups', () => ({ + default: (selector) => + selector({ + serverGroups: [ + { id: 1, name: 'Pool A' }, + { id: 2, name: 'Pool B' }, + ], + fetchServerGroups: vi.fn().mockResolvedValue(undefined), + }), +})); + +vi.mock('../../store/playlists', () => ({ + default: (selector) => + selector({ + playlists: [{ id: 10, server_group: 2 }], + }), +})); + +vi.mock('../forms/ServerGroup', () => ({ + default: () => null, +})); + +vi.mock('../../hooks/useLocalStorage', () => ({ + default: () => ['default', vi.fn()], +})); + +const renderModal = (props = {}) => + render( + + + + ); + +describe('ServerGroupsManagerModal', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('opens without Mantine Tooltip errors', async () => { + expect(() => renderModal()).not.toThrow(); + + await waitFor(() => { + expect(screen.getByText('Pool A')).toBeInTheDocument(); + expect(screen.getByText('Pool B')).toBeInTheDocument(); + expect(screen.getByText('Accounts')).toBeInTheDocument(); + }); + }); +}); diff --git a/frontend/src/components/forms/M3U.jsx b/frontend/src/components/forms/M3U.jsx index 6e19c93a..bfc015de 100644 --- a/frontend/src/components/forms/M3U.jsx +++ b/frontend/src/components/forms/M3U.jsx @@ -36,6 +36,7 @@ import { prepareSubmitValues, updatePlaylist, } from '../../utils/forms/M3uUtils.js'; +import ServerGroupsManagerModal from '../ServerGroupsManagerModal'; const M3U = ({ m3uAccount = null, @@ -56,6 +57,8 @@ const M3U = ({ const [groupFilterModalOpen, setGroupFilterModalOpen] = useState(false); const [filterModalOpen, setFilterModalOpen] = useState(false); const [scheduleType, setScheduleType] = useState('interval'); + const [serverGroupsManagerOpen, setServerGroupsManagerOpen] = useState(false); + const [serverGroupsCreateOnOpen, setServerGroupsCreateOnOpen] = useState(false); const form = useForm({ mode: 'uncontrolled', @@ -366,18 +369,36 @@ const M3U = ({ name="server_group" label="Server Group" description="Share login limits across accounts in a server group. Set max streams on each profile (unlimited profiles skip group enforcement)." - {...form.getInputProps('server_group')} key={form.key('server_group')} - data={[{ value: '0', label: '(None)' }].concat( - serverGroups.map((group) => ({ - label: - group.max_streams > 0 - ? `${group.name} (max: ${group.max_streams})` - : `${group.name} (unlimited)`, + value={form.getValues().server_group} + onChange={(value) => { + if (value === '__new__') { + setServerGroupsCreateOnOpen(true); + setServerGroupsManagerOpen(true); + return; + } + form.setFieldValue('server_group', value); + }} + data={[ + { value: '0', label: '(None)' }, + ...serverGroups.map((group) => ({ + label: group.name, value: `${group.id}`, - })) - )} + })), + { value: '__new__', label: '+ Add server group...' }, + ]} /> +