From abdf2d7864d29a4e834c5c521522ad0144e129f1 Mon Sep 17 00:00:00 2001 From: Goldenfreddy0703 <62456796+Goldenfreddy0703@users.noreply.github.com> Date: Thu, 4 Jun 2026 19:51:20 -0400 Subject: [PATCH] Fix credential release on deleted profile and round 3 review items. Store credential Redis keys at reserve so release works when the profile row is deleted. Return reserve failure reasons to avoid fingerprint DB queries on logging paths. Document unlimited profile bypass in pool logic and Server Group UI. Co-authored-by: Cursor --- apps/channels/models.py | 19 +++-- apps/m3u/connection_pool.py | 78 +++++++++++++++---- apps/m3u/models.py | 6 +- apps/m3u/tests/test_connection_pool.py | 47 +++++++++-- .../multi_worker_connection_manager.py | 4 +- frontend/src/components/forms/M3U.jsx | 2 +- frontend/src/components/forms/ServerGroup.jsx | 2 +- 7 files changed, 123 insertions(+), 35 deletions(-) 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)}