Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-07-17 16:50:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Fix credential release on deleted profile and round 3 review items.

Browse source 
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 <cursoragent@cursor.com>
This commit is contained in:
Goldenfreddy0703 2026-06-04 19:51:20 -04:00
parent 6e07dce3f1
commit abdf2d7864
7 changed files with 123 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

19
apps/channels/models.py
View file

@ -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(

78
apps/m3u/connection_pool.py
View file

@ -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)

6
apps/m3u/models.py
View file

@ -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):

47
apps/m3u/tests/test_connection_pool.py
View file

@ -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):

4
apps/proxy/vod_proxy/multi_worker_connection_manager.py
View file

@ -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: "

2
frontend/src/components/forms/M3U.jsx
View file

@ -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(

2
frontend/src/components/forms/ServerGroup.jsx
View file

@ -70,7 +70,7 @@ const ServerGroupForm = ({ serverGroup = null, isOpen, onClose }) => {
<NumberInput
label="Max Streams"
description="Shared connection limit for all accounts in this group (0 = unlimited)"
description="Set above 0 to enable shared login pooling. Per-login limits use each account profile's max streams. Unlimited profiles (0) skip cross-account enforcement."
min={0}
value={maxStreams}
onChange={(value) => setValue('max_streams', value ?? 0)}