diff --git a/apps/channels/models.py b/apps/channels/models.py index 154f6fa2..9742626b 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -234,7 +234,9 @@ class Stream(models.Model): continue # Atomic slot reservation via shared connection pool helper - reserved, _count = reserve_profile_slot(profile, redis_client) + reserved, _count, _failure_reason = reserve_profile_slot( + profile, redis_client + ) if reserved: redis_client.set(f"channel_stream:{self.id}", self.id) @@ -694,7 +696,9 @@ class Channel(models.Model): has_active_profiles = True # Atomically check and reserve a slot (INCR-first pattern) - reserved, current_count = reserve_profile_slot(profile, redis_client) + reserved, current_count, failure_reason = reserve_profile_slot( + profile, redis_client + ) if reserved: # Slot reserved — assign stream to this channel @@ -712,11 +716,6 @@ 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, @@ -729,14 +728,14 @@ class Channel(models.Model): # return self.id, profile.id, victim_channel_id has_streams_but_maxed_out = True - if not profile_has_capacity_for_selection(profile, redis_client): + if failure_reason == "profile_full": 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): + elif failure_reason == "credential_full": logger.info( - f"Profile {profile.id} login or server group pool full, trying next profile" + f"Profile {profile.id} shared login pool full, trying next profile" ) else: logger.debug( diff --git a/apps/m3u/connection_pool.py b/apps/m3u/connection_pool.py index 35c35b0e..36a74d17 100644 --- a/apps/m3u/connection_pool.py +++ b/apps/m3u/connection_pool.py @@ -5,7 +5,8 @@ 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 so accounts sharing the same provider login share one limit without blocking -unrelated logins on the same group. +unrelated logins on the same group. Account profiles with max_streams=0 skip +credential enforcement for that profile. """ from __future__ import annotations @@ -13,11 +14,14 @@ from __future__ import annotations import hashlib import logging import re -from typing import Optional, Tuple +from typing import Literal, Optional, Tuple logger = logging.getLogger(__name__) +ReserveFailureReason = Literal["profile_full", "credential_full"] + PROFILE_CONNECTIONS_KEY = "profile_connections:{profile_id}" +PROFILE_CREDENTIAL_RELEASE_KEY = "profile_credential_release:{profile_id}" SERVER_GROUP_CONNECTIONS_KEY = "server_group_connections:{group_id}:{fingerprint}" _XC_URL_CREDENTIALS_RE = re.compile( @@ -30,6 +34,11 @@ def profile_connections_key(profile_id: int) -> str: return PROFILE_CONNECTIONS_KEY.format(profile_id=profile_id) +def profile_credential_release_key(profile_id: int) -> str: + """Redis key storing the credential counter to release when the profile row is gone.""" + return PROFILE_CREDENTIAL_RELEASE_KEY.format(profile_id=profile_id) + + 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( @@ -161,6 +170,8 @@ def profile_has_capacity_for_selection(profile, redis_client) -> bool: def group_has_capacity_for_profile(profile, redis_client) -> bool: + # Profiles with max_streams=0 skip credential enforcement entirely. An unlimited + # profile in a pooled group can still stream while other accounts share the login. group = get_enforced_server_group_for_profile(profile) if not group or profile.max_streams == 0: return True @@ -186,21 +197,43 @@ def _safe_decr(redis_client, key: str) -> None: redis_client.set(key, 0) -def _reserve_server_group_slot_for_profile(profile, redis_client) -> bool: +def _remember_credential_release_key( + profile_id: int, cred_key: str, redis_client +) -> None: + redis_client.set(profile_credential_release_key(profile_id), cred_key) + + +def _release_credential_slot_by_profile_id(profile_id: int, redis_client) -> bool: + """Release a reserved credential counter using the key stored at reserve time.""" + release_key = profile_credential_release_key(profile_id) + cred_key = redis_client.get(release_key) + if not cred_key: + return False + + if isinstance(cred_key, bytes): + cred_key = cred_key.decode() + _safe_decr(redis_client, cred_key) + redis_client.delete(release_key) + return True + + +def _reserve_server_group_slot_for_profile( + profile, redis_client +) -> Tuple[bool, Optional[str]]: group = get_enforced_server_group_for_profile(profile) if not group or profile.max_streams == 0: - return True + return True, None cred_key = _credential_counter_key(profile, group) if not cred_key: - return True + return True, None cred_count = redis_client.incr(cred_key) if cred_count <= profile.max_streams: - return True + return True, cred_key redis_client.decr(cred_key) - return False + return False, None def _release_server_group_slot_for_profile(profile, redis_client) -> None: @@ -212,11 +245,14 @@ def _release_server_group_slot_for_profile(profile, redis_client) -> None: _safe_decr(redis_client, cred_key) -def reserve_profile_slot(profile, redis_client) -> Tuple[bool, int]: +def reserve_profile_slot( + profile, redis_client +) -> Tuple[bool, int, Optional[ReserveFailureReason]]: """ Atomically reserve profile + optional credential slots (INCR-first). - Returns (reserved, profile_count_after_attempt). + Returns (reserved, profile_count_after_attempt, failure_reason). + failure_reason is set when reserved is False. """ profile_key = profile_connections_key(profile.id) profile_count = 0 @@ -225,20 +261,34 @@ def reserve_profile_slot(profile, redis_client) -> Tuple[bool, int]: profile_count = redis_client.incr(profile_key) if profile_count > profile.max_streams: redis_client.decr(profile_key) - return False, profile_count - 1 + return False, profile_count - 1, "profile_full" - if not _reserve_server_group_slot_for_profile(profile, redis_client): + cred_reserved, cred_key = _reserve_server_group_slot_for_profile( + profile, redis_client + ) + if not cred_reserved: if profile.max_streams > 0: redis_client.decr(profile_key) - return False, profile_count - 1 if profile.max_streams > 0 else 0 + return ( + False, + profile_count - 1 if profile.max_streams > 0 else 0, + "credential_full", + ) - return True, profile_count + if cred_key: + _remember_credential_release_key(profile.id, cred_key, redis_client) + + return True, profile_count, None def release_profile_slot(profile_id: int, redis_client) -> None: """Release profile and shared credential slots after a stream end.""" from apps.m3u.models import M3UAccountProfile + released_via_stored_key = _release_credential_slot_by_profile_id( + profile_id, redis_client + ) + try: profile = M3UAccountProfile.objects.get(id=profile_id) except M3UAccountProfile.DoesNotExist: @@ -250,5 +300,5 @@ def release_profile_slot(profile_id: int, redis_client) -> None: if current > 0: redis_client.decr(profile_key) - if profile: + if profile and not released_via_stored_key: _release_server_group_slot_for_profile(profile, redis_client) diff --git a/apps/m3u/models.py b/apps/m3u/models.py index e6ccaadb..46d1a068 100644 --- a/apps/m3u/models.py +++ b/apps/m3u/models.py @@ -201,7 +201,11 @@ class ServerGroup(models.Model): ) max_streams = models.PositiveIntegerField( default=0, - help_text="Maximum number of concurrent streams shared across all accounts in this group (0 for unlimited)", + 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): diff --git a/apps/m3u/tests/test_connection_pool.py b/apps/m3u/tests/test_connection_pool.py index 1dbfc2a9..e5de6c10 100644 --- a/apps/m3u/tests/test_connection_pool.py +++ b/apps/m3u/tests/test_connection_pool.py @@ -14,6 +14,7 @@ from apps.m3u.connection_pool import ( pool_has_capacity_for_profile, profile_has_capacity_for_selection, profile_connections_key, + profile_credential_release_key, release_profile_slot, reserve_profile_slot, server_group_connections_key, @@ -31,10 +32,15 @@ class FakeRedis: val = self._data.get(key) if val is None: return None + if isinstance(val, str): + return val.encode() return str(val).encode() def set(self, key, value, ex=None): - self._data[key] = int(value) + try: + self._data[key] = int(value) + except (ValueError, TypeError): + self._data[key] = value def incr(self, key): self._data[key] = self._data.get(key, 0) + 1 @@ -44,6 +50,9 @@ class FakeRedis: self._data[key] = self._data.get(key, 0) - 1 return self._data[key] + def delete(self, key): + self._data.pop(key, None) + def pipeline(self): return FakeRedisPipeline(self) @@ -125,10 +134,10 @@ class ManualServerGroupTests(TestCase): profile2.save() redis = FakeRedis() - reserved1, _ = reserve_profile_slot(profile1, redis) + reserved1, _, _ = reserve_profile_slot(profile1, redis) self.assertTrue(reserved1) - reserved2, _ = reserve_profile_slot(profile2, redis) + reserved2, _, _ = reserve_profile_slot(profile2, redis) self.assertFalse(reserved2) self.assertFalse(group_has_capacity_for_profile(profile2, redis)) @@ -153,7 +162,7 @@ class ManualServerGroupTests(TestCase): ) redis = FakeRedis() - reserved, _ = reserve_profile_slot(default, redis) + reserved, _, _ = reserve_profile_slot(default, redis) self.assertTrue(reserved) self.assertFalse(profile_has_capacity_for_selection(default, redis)) self.assertTrue(profile_has_capacity_for_selection(alt, redis)) @@ -182,7 +191,7 @@ class PoolEnforcementTests(TestCase): self.profile.save() def test_reserve_and_release_both_counters(self): - reserved, count = reserve_profile_slot(self.profile, self.redis) + reserved, count, _ = reserve_profile_slot(self.profile, self.redis) self.assertTrue(reserved) self.assertEqual(count, 1) @@ -286,13 +295,37 @@ class PoolEnforcementTests(TestCase): 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]) + reserved, _, failure_reason = reserve_profile_slot( + self.profile, self.redis + ) + self.assertTrue(reserved) + self.assertIsNone(failure_reason) 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) + self.assertEqual(self.redis._data[cred_key], 0) + self.assertNotIn(profile_credential_release_key(profile_id), self.redis._data) + + def test_reserve_returns_failure_reason_without_extra_checks(self): + account2 = M3UAccount.objects.create( + name="Reason Account", + account_type="XC", + username="user", + password="pass", + server_url="http://xc.example.com", + server_group=self.group, + max_streams=5, + ) + profile2 = M3UAccountProfile.objects.get(m3u_account=account2, is_default=True) + profile2.max_streams = 1 + profile2.save() + + self.assertTrue(reserve_profile_slot(self.profile, self.redis)[0]) + reserved, _count, reason = reserve_profile_slot(profile2, self.redis) + self.assertFalse(reserved) + self.assertEqual(reason, "credential_full") class UpdateStreamProfileTests(TestCase): diff --git a/apps/proxy/vod_proxy/multi_worker_connection_manager.py b/apps/proxy/vod_proxy/multi_worker_connection_manager.py index 77db3b1b..8d803dcf 100644 --- a/apps/proxy/vod_proxy/multi_worker_connection_manager.py +++ b/apps/proxy/vod_proxy/multi_worker_connection_manager.py @@ -745,7 +745,9 @@ class MultiWorkerVODConnectionManager: from apps.m3u.connection_pool import reserve_profile_slot try: - reserved, new_count = reserve_profile_slot(m3u_profile, self.redis_client) + reserved, new_count, _failure_reason = reserve_profile_slot( + m3u_profile, self.redis_client + ) if reserved: logger.info( f"[PROFILE-RESERVE] Profile {m3u_profile.id} slot reserved: " diff --git a/frontend/src/components/forms/M3U.jsx b/frontend/src/components/forms/M3U.jsx index 3a4161cb..6e19c93a 100644 --- a/frontend/src/components/forms/M3U.jsx +++ b/frontend/src/components/forms/M3U.jsx @@ -365,7 +365,7 @@ const M3U = ({ id="server_group" name="server_group" label="Server Group" - description="Share a connection limit across multiple accounts (e.g. XC + M3U on the same subscription)" + 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( diff --git a/frontend/src/components/forms/ServerGroup.jsx b/frontend/src/components/forms/ServerGroup.jsx index 229479ea..a01a0a76 100644 --- a/frontend/src/components/forms/ServerGroup.jsx +++ b/frontend/src/components/forms/ServerGroup.jsx @@ -70,7 +70,7 @@ const ServerGroupForm = ({ serverGroup = null, isOpen, onClose }) => { setValue('max_streams', value ?? 0)}