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 <cursoragent@cursor.com>

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)

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:

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)