refactor(m3u, channels, proxy): enhance stream assignment logic and error handling

- Improved logging in the Stream model for better debugging of profile evaluations.
- Introduced a new method `_stream_assignment_is_reusable` in the Channel model to determine if existing stream assignments can be reused, enhancing efficiency.
- Updated the release logic in `release_profile_slot` to utilize stored credential keys, reducing unnecessary database lookups.
- Simplified error handling in the `get_stream_info_for_switch` function to ensure proper stream release on exceptions.
- Enhanced tests for connection pool management and error handling in the ServerGroupsTable component to improve reliability and user feedback.

apps/channels/models.py

@@ -228,7 +228,7 @@ class Stream(models.Model):
         ]
 
         for profile in profiles:
-            logger.info(profile)
+            logger.debug("Evaluating profile %s for stream %s", profile.id, self.id)
             # Skip inactive profiles
             if profile.is_active == False:
                 continue
@@ -607,6 +607,23 @@ class Channel(models.Model):
             ChannelState.CONNECTING,
         )
 
+    def _stream_assignment_is_reusable(self, redis_client, stream_id: int) -> bool:
+        """
+        Return True when an existing channel_stream assignment should be reused.
+
+        Reuse when the proxy is active, or when metadata is not written yet
+        (between get_stream() reserving slots and initialize_channel() starting).
+        When metadata exists but the proxy is inactive, the assignment is stale.
+        """
+        if self._channel_proxy_is_active(redis_client):
+            return True
+
+        metadata_key = RedisKeys.channel_metadata(str(self.uuid))
+        if not redis_client.exists(metadata_key):
+            return redis_client.get(f"stream_profile:{stream_id}") is not None
+
+        return False
+
     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}")
@@ -654,13 +671,13 @@ class Channel(models.Model):
                 stream_id = None
 
             if stream_id is not None:
-                if self._channel_proxy_is_active(redis_client):
+                if self._stream_assignment_is_reusable(redis_client, stream_id):
                     profile_id_bytes = redis_client.get(f"stream_profile:{stream_id}")
                     if profile_id_bytes:
                         try:
                             profile_id = int(profile_id_bytes)
                             logger.debug(
-                                f"Channel {self.uuid}: reusing active assignment "
+                                f"Channel {self.uuid}: reusing stream assignment "
                                 f"stream={stream_id} profile={profile_id}"
                             )
                             return stream_id, profile_id, None, False

apps/channels/tests/test_get_stream_assignment.py

@@ -0,0 +1,173 @@
+"""Tests for Channel.get_stream() assignment reuse and stale cleanup."""
+
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from apps.channels.models import Channel, ChannelStream, Stream
+from apps.m3u.models import M3UAccount, M3UAccountProfile
+from apps.proxy.live_proxy.constants import ChannelMetadataField, ChannelState
+from apps.proxy.live_proxy.redis_keys import RedisKeys
+
+
+class FakeAssignmentRedis:
+    """In-memory Redis for channel_stream assignment tests."""
+
+    def __init__(self):
+        self._strings = {}
+        self._hashes = {}
+
+    def _decode(self, value):
+        if isinstance(value, bytes):
+            return value.decode()
+        return value
+
+    def get(self, key):
+        value = self._strings.get(key)
+        if value is None:
+            return None
+        if isinstance(value, int):
+            return str(value).encode()
+        return str(value).encode()
+
+    def set(self, key, value):
+        self._strings[key] = value
+
+    def delete(self, key):
+        self._strings.pop(key, None)
+        self._hashes.pop(key, None)
+
+    def exists(self, key):
+        return key in self._strings or key in self._hashes
+
+    def hget(self, key, field):
+        return self._hashes.get(key, {}).get(field)
+
+    def hset(self, key, mapping=None, **kwargs):
+        bucket = self._hashes.setdefault(key, {})
+        if mapping:
+            bucket.update(mapping)
+        bucket.update(kwargs)
+
+    def incr(self, key):
+        current = int(self._decode(self.get(key)) or 0)
+        current += 1
+        self._strings[key] = current
+        return current
+
+    def decr(self, key):
+        current = int(self._decode(self.get(key)) or 0)
+        current -= 1
+        self._strings[key] = current
+        return current
+
+
+class ChannelGetStreamAssignmentTests(TestCase):
+    def setUp(self):
+        self.redis = FakeAssignmentRedis()
+        self.account = M3UAccount.objects.create(
+            name="assignment-test",
+            account_type="XC",
+            username="user",
+            password="pass",
+            max_streams=5,
+        )
+        self.profile = M3UAccountProfile.objects.get(
+            m3u_account=self.account, is_default=True
+        )
+        self.profile.max_streams = 2
+        self.profile.save()
+
+        self.stream = Stream.objects.create(
+            name="Test Stream",
+            url="http://example.com/live/user/pass/1.ts",
+            m3u_account=self.account,
+        )
+        self.channel = Channel.objects.create(channel_number=501, name="Assignment Ch")
+        ChannelStream.objects.create(channel=self.channel, stream=self.stream, order=0)
+
+        self.metadata_key = RedisKeys.channel_metadata(str(self.channel.uuid))
+
+    def _seed_assignment(self):
+        self.redis.set(f"channel_stream:{self.channel.id}", self.stream.id)
+        self.redis.set(f"stream_profile:{self.stream.id}", self.profile.id)
+
+    @patch("apps.channels.models.RedisClient.get_client")
+    @patch("apps.channels.models.reserve_profile_slot")
+    def test_reuses_assignment_when_proxy_active(
+        self, mock_reserve, mock_get_client
+    ):
+        mock_get_client.return_value = self.redis
+        self._seed_assignment()
+        self.redis.hset(
+            self.metadata_key,
+            {ChannelMetadataField.STATE: ChannelState.ACTIVE},
+        )
+
+        stream_id, profile_id, error, slot_reserved = self.channel.get_stream()
+
+        self.assertEqual(stream_id, self.stream.id)
+        self.assertEqual(profile_id, self.profile.id)
+        self.assertIsNone(error)
+        self.assertFalse(slot_reserved)
+        mock_reserve.assert_not_called()
+
+    @patch("apps.channels.models.RedisClient.get_client")
+    @patch("apps.channels.models.reserve_profile_slot")
+    def test_reuses_assignment_during_init_before_metadata(
+        self, mock_reserve, mock_get_client
+    ):
+        mock_get_client.return_value = self.redis
+        self._seed_assignment()
+
+        stream_id, profile_id, error, slot_reserved = self.channel.get_stream()
+
+        self.assertEqual(stream_id, self.stream.id)
+        self.assertEqual(profile_id, self.profile.id)
+        self.assertIsNone(error)
+        self.assertFalse(slot_reserved)
+        mock_reserve.assert_not_called()
+
+    @patch("apps.channels.models.RedisClient.get_client")
+    @patch("apps.channels.models.release_profile_slot")
+    @patch("apps.channels.models.reserve_profile_slot")
+    def test_releases_stale_assignment_when_proxy_stopped(
+        self, mock_reserve, mock_release, mock_get_client
+    ):
+        mock_get_client.return_value = self.redis
+        mock_reserve.return_value = (True, 1, None)
+        self._seed_assignment()
+        self.redis.hset(
+            self.metadata_key,
+            {ChannelMetadataField.STATE: ChannelState.STOPPED},
+        )
+
+        stream_id, profile_id, error, slot_reserved = self.channel.get_stream()
+
+        mock_release.assert_called_once_with(self.profile.id, self.redis)
+        mock_reserve.assert_called_once()
+        self.assertEqual(stream_id, self.stream.id)
+        self.assertEqual(profile_id, self.profile.id)
+        self.assertTrue(slot_reserved)
+
+    @patch("apps.channels.models.RedisClient.get_client")
+    def test_stream_assignment_is_reusable_during_init_pending(self, mock_get_client):
+        mock_get_client.return_value = self.redis
+        self._seed_assignment()
+
+        self.assertTrue(
+            self.channel._stream_assignment_is_reusable(self.redis, self.stream.id)
+        )
+
+    @patch("apps.channels.models.RedisClient.get_client")
+    def test_stream_assignment_not_reusable_when_stopped(self, mock_get_client):
+        mock_get_client.return_value = self.redis
+        self._seed_assignment()
+        self.redis.hset(
+            self.metadata_key,
+            {ChannelMetadataField.STATE: ChannelState.STOPPED},
+        )
+
+        self.assertFalse(
+            self.channel._stream_assignment_is_reusable(self.redis, self.stream.id)
+        )

apps/m3u/connection_pool.py

@@ -172,7 +172,7 @@ def group_has_capacity_for_profile(profile, redis_client) -> bool:
     cred_key = _credential_counter_key(profile, group)
     if not cred_key:
         return True
-    return get_credential_connection_count(profile, redis_client) < profile.max_streams
+    return int(redis_client.get(cred_key) or 0) < profile.max_streams
 
 
 def pool_has_capacity_for_profile(profile, redis_client) -> bool:
@@ -326,22 +326,26 @@ def reserve_profile_slot(
 
 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:
-        profile = None
-
     profile_key = profile_connections_key(profile_id)
-    if profile is None or profile.max_streams > 0:
-        current = int(redis_client.get(profile_key) or 0)
-        if current > 0:
-            redis_client.decr(profile_key)
+    current = int(redis_client.get(profile_key) or 0)
+    if current > 0:
+        redis_client.decr(profile_key)
 
-    if profile and not released_via_stored_key:
-        _release_server_group_slot_for_profile(profile, redis_client)
+    if released_via_stored_key:
+        return
+
+    # Legacy fallback for reservations made before credential release keys were stored.
+    from apps.m3u.models import M3UAccountProfile
+
+    try:
+        profile = M3UAccountProfile.objects.select_related(
+            "m3u_account__server_group"
+        ).get(id=profile_id)
+    except M3UAccountProfile.DoesNotExist:
+        return
+
+    _release_server_group_slot_for_profile(profile, redis_client)

apps/m3u/tests/test_connection_pool.py

@@ -186,6 +186,18 @@ class PoolEnforcementTests(TestCase):
         self.profile.max_streams = 1
         self.profile.save()
 
+    def test_group_has_capacity_reads_credential_counter_directly(self):
+        cred_key = server_group_connections_key(
+            self.group.id,
+            get_profile_credential_fingerprint(self.profile),
+        )
+        self.redis.set(cred_key, 1)
+
+        self.assertFalse(group_has_capacity_for_profile(self.profile, self.redis))
+
+        self.redis.set(cred_key, 0)
+        self.assertTrue(group_has_capacity_for_profile(self.profile, self.redis))
+
     def test_reserve_and_release_both_counters(self):
         reserved, count, _ = reserve_profile_slot(self.profile, self.redis)
         self.assertTrue(reserved)
@@ -301,6 +313,22 @@ class PoolEnforcementTests(TestCase):
         self.assertEqual(self.redis._data[cred_key], 0)
         self.assertNotIn(profile_credential_release_key(profile_id), self.redis._data)
 
+    def test_release_uses_stored_credential_key_without_db_lookup(self):
+        profile_id = self.profile.id
+        cred_key = server_group_connections_key(
+            self.group.id,
+            get_profile_credential_fingerprint(self.profile),
+        )
+
+        self.assertTrue(reserve_profile_slot(self.profile, self.redis)[0])
+
+        with patch("apps.m3u.models.M3UAccountProfile.objects.get") as mock_get:
+            release_profile_slot(profile_id, self.redis)
+            mock_get.assert_not_called()
+
+        self.assertEqual(self.redis._data[profile_connections_key(profile_id)], 0)
+        self.assertEqual(self.redis._data[cred_key], 0)
+
     def test_reserve_returns_failure_reason_without_extra_checks(self):
         account2 = M3UAccount.objects.create(
             name="Reason Account",

apps/proxy/live_proxy/channel_status.py

@@ -501,15 +501,7 @@ class ChannelStatus:
             m3u_profile_id = metadata.get(ChannelMetadataField.M3U_PROFILE)
             if m3u_profile_id:
                 try:
-                    profile_id_int = int(m3u_profile_id)
-                    info['m3u_profile_id'] = profile_id_int
-                    try:
-                        from apps.m3u.models import M3UAccountProfile
-                        m3u_profile = M3UAccountProfile.objects.filter(id=profile_id_int).first()
-                        if m3u_profile:
-                            info['m3u_profile_name'] = m3u_profile.name
-                    except Exception as e:
-                        logger.warning(f"Failed to get M3U profile name for ID {profile_id_int}: {e}")
+                    info['m3u_profile_id'] = int(m3u_profile_id)
                 except ValueError:
                     logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id}")
 

apps/proxy/live_proxy/url_utils.py

@@ -203,6 +203,8 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
     Returns:
         dict: Stream information including URL, user agent and transcode flag
     """
+    slot_reserved = False
+    channel = None
     try:
         from core.utils import RedisClient
 
@@ -268,28 +270,18 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
 
             m3u_profile_id = selected_profile.id
         else:
-            stream_id, m3u_profile_id, error_reason, _slot_reserved = channel.get_stream()
+            stream_id, m3u_profile_id, error_reason, slot_reserved = channel.get_stream()
             if stream_id is None or m3u_profile_id is None:
                 return {'error': error_reason or 'No stream assigned to channel'}
 
-        # Get the stream and profile objects directly
         stream = get_object_or_404(Stream, pk=stream_id)
         profile = get_object_or_404(M3UAccountProfile, pk=m3u_profile_id)
 
-        # Check connections left
         m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id)
-        #connections_left = get_connections_left(m3u_profile_id)
-
-        #if connections_left <= 0:
-            #logger.warning(f"No connections left for M3U account {m3u_account.id}")
-            #return {'error': 'No connections left'}
-
-        # Get the user agent from the M3U account
         user_agent = m3u_account.get_user_agent().user_agent
 
         stream_url = _resolve_live_stream_url(stream, m3u_account, profile)
 
-        # Get transcode info from the channel's stream profile
         stream_profile = channel.get_stream_profile()
         transcode = not (stream_profile.is_proxy() or stream_profile is None)
         profile_value = stream_profile.id
@@ -304,6 +296,8 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
             'stream_name': stream.name,
         }
     except Exception as e:
+        if slot_reserved and channel is not None:
+            channel.release_stream()
         logger.error(f"Error getting stream info for switch: {e}", exc_info=True)
         return {'error': f'Error: {str(e)}'}