From 6e07dce3f17532fc3519c3ef2b052467c06ead05 Mon Sep 17 00:00:00 2001 From: Goldenfreddy0703 <62456796+Goldenfreddy0703@users.noreply.github.com> Date: Fri, 29 May 2026 17:05:59 -0400 Subject: [PATCH] Fix maintainer review items without breaking multi-login rotation. Cap credential-scoped group counters at profile.max_streams (not group.max_streams), skip credential counters when fingerprint resolution fails, and always swap profile counters on update_stream_profile. Restore per-profile-only selection for VOD/live switches so a second stream can use a different login without invalid HTTP errors. Co-authored-by: Cursor --- apps/channels/models.py | 44 +++--- apps/m3u/connection_pool.py | 94 ++++++++----- apps/m3u/tests/test_connection_pool.py | 178 +++++++++++++++++++++---- 3 files changed, 233 insertions(+), 83 deletions(-) diff --git a/apps/channels/models.py b/apps/channels/models.py index 38ebbbc7..154f6fa2 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -712,6 +712,11 @@ class Channel(models.Model): True, ) # Return newly assigned stream and matched profile else: + from apps.m3u.connection_pool import ( + group_has_capacity_for_profile, + profile_has_capacity_for_selection, + ) + # At capacity: try to preempt a lower-impact channel on this profile victim_channel_id = self._pick_channel_to_preempt( profile_id=profile.id, @@ -723,12 +728,21 @@ class Channel(models.Model): logger.info(f"Preempting channel {victim_channel_id} for new stream on profile {profile.id}") # return self.id, profile.id, victim_channel_id - # This profile is at max connections has_streams_but_maxed_out = True - logger.debug( - f"Profile {profile.id} at max connections: " - f"{current_count}/{profile.max_streams}" - ) + if not profile_has_capacity_for_selection(profile, redis_client): + logger.info( + f"Profile {profile.id} at max connections: " + f"{current_count}/{profile.max_streams}, trying next profile" + ) + elif not group_has_capacity_for_profile(profile, redis_client): + logger.info( + f"Profile {profile.id} login or server group pool full, trying next profile" + ) + else: + logger.debug( + f"Profile {profile.id} reservation failed: " + f"{current_count}/{profile.max_streams}" + ) # No available streams - determine specific reason if has_streams_but_maxed_out: @@ -873,25 +887,9 @@ class Channel(models.Model): if current_profile_id == new_profile_id: return True - from apps.m3u.models import M3UAccountProfile - from apps.m3u.connection_pool import ( - get_enforced_server_group_for_profile, - profile_connections_key, - ) + from apps.m3u.connection_pool import profile_connections_key - new_profile = M3UAccountProfile.objects.get(id=new_profile_id) - if get_enforced_server_group_for_profile(new_profile): - # Shared group pool: one active stream uses one group slot regardless - # of which profile supplies the URL. - redis_client.set(f"stream_profile:{stream_id}", new_profile_id) - logger.info( - f"Updated stream {stream_id} profile from {current_profile_id} to " - f"{new_profile_id} (shared group pool unchanged)" - ) - return True - - # Use pipeline for atomic profile switch to prevent counter drift - # if an exception occurs between DECR and INCR + # Profile counters always move on switch; group totals stay unchanged (one stream). 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 9817d86c..35c35b0e 100644 --- a/apps/m3u/connection_pool.py +++ b/apps/m3u/connection_pool.py @@ -3,8 +3,9 @@ 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, the group counter is scoped by provider login fingerprint -so profiles that rewrite to different IPTV credentials keep independent limits. +with max_streams > 0, 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. """ from __future__ import annotations @@ -29,10 +30,12 @@ def profile_connections_key(profile_id: int) -> str: return PROFILE_CONNECTIONS_KEY.format(profile_id=profile_id) -def server_group_connections_key(group_id: int, fingerprint: Optional[str] = None) -> str: - """Redis key for a manual ServerGroup slot, scoped by provider login.""" - fp = (fingerprint or "unknown")[:16] - return SERVER_GROUP_CONNECTIONS_KEY.format(group_id=group_id, fingerprint=fp) +def server_group_connections_key(group_id: int, fingerprint: str) -> str: + """Redis key for per-credential usage within a ServerGroup.""" + return SERVER_GROUP_CONNECTIONS_KEY.format( + group_id=group_id, + fingerprint=fingerprint[:16], + ) def compute_credential_fingerprint(username: str, password: str) -> Optional[str]: @@ -124,22 +127,30 @@ def get_enforced_server_group_for_profile(profile): return None -def _group_counter_key(profile, group) -> str: - return server_group_connections_key( - group.id, - get_profile_credential_fingerprint(profile), - ) +def _credential_counter_key(profile, group) -> Optional[str]: + fingerprint = get_profile_credential_fingerprint(profile) + if not fingerprint: + return None + return server_group_connections_key(group.id, fingerprint) def get_profile_connection_count(profile, redis_client) -> int: return int(redis_client.get(profile_connections_key(profile.id)) or 0) -def get_group_connection_count(profile, redis_client) -> int: +def get_credential_connection_count(profile, redis_client) -> int: group = get_enforced_server_group_for_profile(profile) if not group: return 0 - return int(redis_client.get(_group_counter_key(profile, group)) or 0) + cred_key = _credential_counter_key(profile, group) + if not cred_key: + return 0 + 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: @@ -151,35 +162,22 @@ def profile_has_capacity_for_selection(profile, redis_client) -> bool: def group_has_capacity_for_profile(profile, redis_client) -> bool: group = get_enforced_server_group_for_profile(profile) - if not group: + if not group or profile.max_streams == 0: return True - return get_group_connection_count(profile, redis_client) < group.max_streams + cred_key = _credential_counter_key(profile, group) + if not cred_key: + return True + return get_credential_connection_count(profile, redis_client) < profile.max_streams def pool_has_capacity_for_profile(profile, redis_client) -> bool: - """Non-mutating check before reserve: profile slot and group slot if applicable.""" + """Non-mutating check before reserve: profile slot and credential slot if applicable.""" return profile_has_capacity_for_selection(profile, redis_client) and group_has_capacity_for_profile( profile, redis_client ) -def _reserve_server_group_slot_for_profile(profile, redis_client) -> bool: - group = get_enforced_server_group_for_profile(profile) - if not group: - return True - key = _group_counter_key(profile, group) - group_count = redis_client.incr(key) - if group_count <= group.max_streams: - return True - redis_client.decr(key) - return False - - -def _release_server_group_slot_for_profile(profile, redis_client) -> None: - group = get_enforced_server_group_for_profile(profile) - if not group: - return - key = _group_counter_key(profile, group) +def _safe_decr(redis_client, key: str) -> None: current = int(redis_client.get(key) or 0) if current <= 0: return @@ -188,9 +186,35 @@ def _release_server_group_slot_for_profile(profile, redis_client) -> None: redis_client.set(key, 0) +def _reserve_server_group_slot_for_profile(profile, redis_client) -> bool: + group = get_enforced_server_group_for_profile(profile) + if not group or profile.max_streams == 0: + return True + + cred_key = _credential_counter_key(profile, group) + if not cred_key: + return True + + cred_count = redis_client.incr(cred_key) + if cred_count <= profile.max_streams: + return True + + redis_client.decr(cred_key) + return False + + +def _release_server_group_slot_for_profile(profile, redis_client) -> None: + group = get_enforced_server_group_for_profile(profile) + if not group or profile.max_streams == 0: + return + cred_key = _credential_counter_key(profile, group) + if cred_key: + _safe_decr(redis_client, cred_key) + + def reserve_profile_slot(profile, redis_client) -> Tuple[bool, int]: """ - Atomically reserve profile + optional group slots (INCR-first). + Atomically reserve profile + optional credential slots (INCR-first). Returns (reserved, profile_count_after_attempt). """ @@ -212,7 +236,7 @@ def reserve_profile_slot(profile, redis_client) -> Tuple[bool, int]: def release_profile_slot(profile_id: int, redis_client) -> None: - """Release profile and shared group slots after a stream ends.""" + """Release profile and shared credential slots after a stream end.""" from apps.m3u.models import M3UAccountProfile try: diff --git a/apps/m3u/tests/test_connection_pool.py b/apps/m3u/tests/test_connection_pool.py index 44281211..1dbfc2a9 100644 --- a/apps/m3u/tests/test_connection_pool.py +++ b/apps/m3u/tests/test_connection_pool.py @@ -5,6 +5,7 @@ from unittest.mock import patch 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, @@ -43,6 +44,37 @@ class FakeRedis: self._data[key] = self._data.get(key, 0) - 1 return self._data[key] + def pipeline(self): + return FakeRedisPipeline(self) + + +class FakeRedisPipeline: + def __init__(self, redis): + self.redis = redis + self._ops = [] + + def decr(self, key): + self._ops.append(("decr", key)) + return self + + def incr(self, key): + self._ops.append(("incr", key)) + return self + + def set(self, key, value): + self._ops.append(("set", key, value)) + return self + + def execute(self): + for op in self._ops: + if op[0] == "decr": + self.redis.decr(op[1]) + elif op[0] == "incr": + self.redis.incr(op[1]) + elif op[0] == "set": + self.redis.set(op[1], op[2]) + self._ops = [] + class ExtractCredentialsTests(TestCase): def test_extract_credentials_from_xc_style_url(self): @@ -66,8 +98,8 @@ class ManualServerGroupTests(TestCase): self.assertEqual(get_enforced_server_group_for_profile(profile), group) - def test_accounts_in_same_group_share_counter(self): - group = ServerGroup.objects.create(name="shared", max_streams=1) + def test_accounts_in_same_group_share_credential_counter(self): + group = ServerGroup.objects.create(name="shared", max_streams=2) account1 = M3UAccount.objects.create( name="XC Account", account_type="XC", @@ -87,6 +119,10 @@ class ManualServerGroupTests(TestCase): ) profile1 = M3UAccountProfile.objects.get(m3u_account=account1, is_default=True) profile2 = M3UAccountProfile.objects.get(m3u_account=account2, is_default=True) + profile1.max_streams = 1 + profile1.save() + profile2.max_streams = 1 + profile2.save() redis = FakeRedis() reserved1, _ = reserve_profile_slot(profile1, redis) @@ -97,7 +133,6 @@ class ManualServerGroupTests(TestCase): self.assertFalse(group_has_capacity_for_profile(profile2, redis)) def test_profile_rotation_when_default_profile_full(self): - """Pre-pool behavior: try the next profile on the same account.""" account = M3UAccount.objects.create( name="Multi-profile", account_type="XC", @@ -129,20 +164,21 @@ class PoolEnforcementTests(TestCase): self.redis = FakeRedis() self.group = ServerGroup.objects.create( name="test-pool", - max_streams=1, + max_streams=2, ) self.account = M3UAccount.objects.create( name="Test Account", account_type="XC", username="user", password="pass", + server_url="http://xc.example.com", server_group=self.group, max_streams=5, ) self.profile = M3UAccountProfile.objects.get( m3u_account=self.account, is_default=True ) - self.profile.max_streams = 5 + self.profile.max_streams = 1 self.profile.save() def test_reserve_and_release_both_counters(self): @@ -150,31 +186,45 @@ class PoolEnforcementTests(TestCase): self.assertTrue(reserved) self.assertEqual(count, 1) - group_key = server_group_connections_key( + cred_key = server_group_connections_key( self.group.id, get_profile_credential_fingerprint(self.profile), ) profile_key = profile_connections_key(self.profile.id) - self.assertEqual(self.redis._data[group_key], 1) + self.assertEqual(self.redis._data[cred_key], 1) self.assertEqual(self.redis._data[profile_key], 1) release_profile_slot(self.profile.id, self.redis) - self.assertEqual(self.redis._data[group_key], 0) + self.assertEqual(self.redis._data[cred_key], 0) self.assertEqual(self.redis._data[profile_key], 0) - def test_reserve_fails_when_group_at_capacity(self): - group_key = server_group_connections_key( - self.group.id, - get_profile_credential_fingerprint(self.profile), + def test_same_credential_capped_at_profile_max_not_group_max(self): + """Maintainer example: group max=2 but each login only allows 1.""" + account2 = M3UAccount.objects.create( + name="Second Account", + account_type="XC", + username="user", + password="pass", + server_url="http://xc.example.com", + server_group=self.group, + max_streams=5, ) - self.redis.set(group_key, 1) + profile2 = M3UAccountProfile.objects.get(m3u_account=account2, is_default=True) + profile2.max_streams = 1 + profile2.save() - reserved, _count = reserve_profile_slot(self.profile, self.redis) - self.assertFalse(reserved) - self.assertFalse(pool_has_capacity_for_profile(self.profile, self.redis)) + self.assertTrue(reserve_profile_slot(self.profile, self.redis)[0]) + self.assertFalse(reserve_profile_slot(profile2, self.redis)[0]) + + fp = get_profile_credential_fingerprint(self.profile) + 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_in_group_do_not_block_each_other(self): - """Profiles with different provider logins keep separate group counters.""" account = M3UAccount.objects.create( name="Grouped multi-login", account_type="XC", @@ -204,17 +254,96 @@ class PoolEnforcementTests(TestCase): "apps.m3u.connection_pool.get_profile_credential_fingerprint", side_effect=lambda profile: fp_a if profile.id == default.id else fp_b, ): - reserved_default, _ = reserve_profile_slot(default, self.redis) - self.assertTrue(reserved_default) - - reserved_alt, _ = reserve_profile_slot(alt, self.redis) - self.assertTrue(reserved_alt) + self.assertTrue(reserve_profile_slot(default, self.redis)[0]) + self.assertTrue(reserve_profile_slot(alt, self.redis)[0]) key_a = server_group_connections_key(self.group.id, fp_a) key_b = server_group_connections_key(self.group.id, fp_b) self.assertEqual(self.redis._data[key_a], 1) self.assertEqual(self.redis._data[key_b], 1) + def test_no_fingerprint_skips_credential_counter(self): + account = M3UAccount.objects.create( + name="No creds", + account_type="STD", + server_group=self.group, + max_streams=5, + ) + profile = M3UAccountProfile.objects.get(m3u_account=account, is_default=True) + profile.max_streams = 1 + profile.save() + + with patch( + "apps.m3u.connection_pool.get_profile_credential_fingerprint", + return_value=None, + ): + 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) + + def test_release_when_profile_row_deleted(self): + profile_id = self.profile.id + fp = get_profile_credential_fingerprint(self.profile) + cred_key = server_group_connections_key(self.group.id, fp) + + self.assertTrue(reserve_profile_slot(self.profile, self.redis)[0]) + self.profile.delete() + + release_profile_slot(profile_id, self.redis) + + self.assertEqual(self.redis._data[profile_connections_key(profile_id)], 0) + self.assertEqual(self.redis._data[cred_key], 1) + + +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) + account = M3UAccount.objects.create( + name="Switch Account", + account_type="XC", + username="user", + password="pass", + 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", + is_default=False, + is_active=True, + max_streams=1, + search_pattern="", + replace_pattern="", + ) + + stream = Stream.objects.create(name="Test Stream", m3u_account=account) + channel = Channel.objects.create(channel_number=501, name="Switch Channel") + channel.streams.add(stream) + + 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) + + cred_key = server_group_connections_key( + group.id, get_profile_credential_fingerprint(profile_a) + ) + cred_before = redis._data[cred_key] + + with patch("core.utils.RedisClient.get_client", return_value=redis): + self.assertTrue(channel.update_stream_profile(profile_b.id)) + + self.assertEqual(int(redis.get(f"stream_profile:{stream.id}")), profile_b.id) + self.assertEqual(redis._data[profile_connections_key(profile_a.id)], 0) + self.assertEqual(redis._data[profile_connections_key(profile_b.id)], 1) + self.assertEqual(redis._data[cred_key], cred_before) + class VodProfileSelectionTests(TestCase): def test_get_m3u_profile_skips_default_when_profile_full(self): @@ -243,8 +372,7 @@ class VodProfileSelectionTests(TestCase): ) redis = FakeRedis() - reserved, _ = reserve_profile_slot(default, redis) - self.assertTrue(reserved) + self.assertTrue(reserve_profile_slot(default, redis)[0]) with patch("core.utils.RedisClient.get_client", return_value=redis): result = _get_m3u_profile(account, None, None)