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

refactor(m3u, channels): streamline stream assignment and server group management

Browse source 
- Renamed `_clear_stream_assignment_keys` to `_release_stale_stream_assignment` for clarity and updated its logic to release pool counters.
- Introduced new functions for managing credential slots during profile switches, enhancing the handling of shared connection limits across server groups.
- Removed the `max_streams` field from the `ServerGroup` model and updated related components to reflect this change, simplifying the server group management.
- Updated frontend components to integrate server group management, allowing for dynamic creation and editing of server groups.
- Enhanced error handling in stream URL generation to provide more informative feedback on connection issues.
- Added tests for stale assignment release and credential management during profile switches.
This commit is contained in:
SergeantPanda 2026-06-09 13:40:16 -05:00
parent 0f0e855507
commit bc046b3c92
17 changed files with 710 additions and 229 deletions
Download patch file Download diff file Expand all files Collapse all files

47
apps/channels/models.py
View file

@ -607,11 +607,22 @@ class Channel(models.Model):
ChannelState.CONNECTING,
)
def _clear_stream_assignment_keys(self, redis_client, stream_id=None) -> None:
"""Remove channel/stream profile assignment keys from Redis."""
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}")
if profile_id_bytes:
try:
profile_id = int(profile_id_bytes)
release_profile_slot(profile_id, redis_client)
except (ValueError, TypeError):
logger.debug(
"Invalid profile ID for stale assignment on stream %s: %s",
stream_id,
profile_id_bytes,
)
redis_client.delete(f"channel_stream:{self.id}")
if stream_id is not None:
redis_client.delete(f"stream_profile:{stream_id}")
redis_client.delete(f"stream_profile:{stream_id}")
def get_stream(self, requester=None):
"""
@ -659,10 +670,10 @@ class Channel(models.Model):
)
else:
logger.info(
f"Channel {self.uuid}: clearing stale stream assignment "
f"Channel {self.uuid}: releasing stale stream assignment "
f"(stream={stream_id}, proxy not active)"
)
self._clear_stream_assignment_keys(redis_client, stream_id)
self._release_stale_stream_assignment(redis_client, stream_id)
# No existing active stream, attempt to assign a new one
has_streams_but_maxed_out = False
@ -886,9 +897,29 @@ class Channel(models.Model):
if current_profile_id == new_profile_id:
return True
from apps.m3u.connection_pool import profile_connections_key
from apps.m3u.connection_pool import (
move_credential_slot_on_profile_switch,
profile_connections_key,
)
from apps.m3u.models import M3UAccountProfile
# Profile counters always move on switch; group totals stay unchanged (one stream).
old_profile = M3UAccountProfile.objects.select_related(
"m3u_account__server_group"
).get(id=current_profile_id)
new_profile = M3UAccountProfile.objects.select_related(
"m3u_account__server_group"
).get(id=new_profile_id)
if not move_credential_slot_on_profile_switch(
old_profile, new_profile, redis_client
):
logger.warning(
"Shared login pool full for profile %s during stream profile switch",
new_profile_id,
)
return False
# Profile counters always move on switch; credential totals move only when login changes.
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)

61
apps/m3u/connection_pool.py
View file

@ -2,8 +2,7 @@
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, a credential-scoped counter is checked on reserve/release
Redis counter (the pre-pool behavior). When an account belongs to a ServerGroup, 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. Account profiles with max_streams=0 skip
credential enforcement for that profile.
@ -129,9 +128,9 @@ def get_profile_credential_fingerprint(profile) -> Optional[str]:
def get_enforced_server_group_for_profile(profile):
"""Return the shared ServerGroup limit for this profile's account, if configured."""
"""Return the ServerGroup for credential pooling when the account is assigned to one."""
group = profile.m3u_account.server_group
if group and group.max_streams > 0:
if group:
return group
return None
@ -157,11 +156,6 @@ def get_credential_connection_count(profile, redis_client) -> int:
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:
"""Per-profile capacity check used when rotating across profiles on one account."""
if profile.max_streams == 0:
@ -188,6 +182,55 @@ def pool_has_capacity_for_profile(profile, redis_client) -> bool:
)
def profile_available_for_channel_switch(
profile, redis_client, *, channel_already_on_profile: bool
) -> bool:
"""
Non-mutating capacity check when selecting a profile for an in-flight channel.
If the channel already holds this profile's slots, skip re-checking capacity.
"""
if channel_already_on_profile:
return True
return pool_has_capacity_for_profile(profile, redis_client)
def move_credential_slot_on_profile_switch(
old_profile, new_profile, redis_client
) -> bool:
"""
Move the shared credential counter when switching to a different provider login.
Profile counters are managed separately by Channel.update_stream_profile().
Returns False when the new profile's credential pool is full.
"""
old_fp = get_profile_credential_fingerprint(old_profile)
new_fp = get_profile_credential_fingerprint(new_profile)
if old_fp == new_fp:
return True
released = _release_credential_slot_by_profile_id(old_profile.id, redis_client)
if not released:
_release_server_group_slot_for_profile(old_profile, redis_client)
cred_reserved, cred_key = _reserve_server_group_slot_for_profile(
new_profile, redis_client
)
if not cred_reserved:
restore_reserved, restore_key = _reserve_server_group_slot_for_profile(
old_profile, redis_client
)
if restore_reserved and restore_key:
_remember_credential_release_key(
old_profile.id, restore_key, redis_client
)
return False
if cred_key:
_remember_credential_release_key(new_profile.id, cred_key, redis_client)
return True
def _safe_decr(redis_client, key: str) -> None:
current = int(redis_client.get(key) or 0)
if current <= 0:

21
apps/m3u/migrations/0020_servergroup_max_streams.py
View file

@ -1,21 +0,0 @@
# Generated by Django 6.0.3
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("m3u", "0019_m3uaccountprofile_exp_date"),
]
operations = [
migrations.AddField(
model_name="servergroup",
name="max_streams",
field=models.PositiveIntegerField(
default=0,
help_text="Maximum number of concurrent streams shared across all accounts in this group (0 for unlimited)",
),
),
]

16
apps/m3u/models.py
View file

@ -194,19 +194,17 @@ class M3UFilter(models.Model):
class ServerGroup(models.Model):
"""Represents a logical grouping of servers or channels."""
"""
Groups M3U accounts that share provider credentials.
Accounts assigned to the same server group share credential-scoped connection
counters when their logins match. Limits come from each account profile's
max_streams, not from the group itself.
"""
name = models.CharField(
max_length=100, unique=True, help_text="Unique name for this server group."
)
max_streams = models.PositiveIntegerField(
default=0,
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):
return self.name

2
apps/m3u/serializers.py
View file

@ -408,4 +408,4 @@ class ServerGroupSerializer(serializers.ModelSerializer):
class Meta:
model = ServerGroup
fields = ["id", "name", "max_streams"]
fields = ["id", "name"]

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

@ -7,7 +7,6 @@ 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,
get_profile_credential_fingerprint,
group_has_capacity_for_profile,
@ -94,8 +93,8 @@ class ExtractCredentialsTests(TestCase):
class ManualServerGroupTests(TestCase):
def test_group_enforced_when_max_streams_set(self):
group = ServerGroup.objects.create(name="provider-a", max_streams=2)
def test_group_enforced_when_account_assigned(self):
group = ServerGroup.objects.create(name="provider-a")
account = M3UAccount.objects.create(
name="Account A",
account_type="XC",
@ -108,7 +107,7 @@ class ManualServerGroupTests(TestCase):
self.assertEqual(get_enforced_server_group_for_profile(profile), group)
def test_accounts_in_same_group_share_credential_counter(self):
group = ServerGroup.objects.create(name="shared", max_streams=2)
group = ServerGroup.objects.create(name="shared")
account1 = M3UAccount.objects.create(
name="XC Account",
account_type="XC",
@ -171,10 +170,7 @@ class ManualServerGroupTests(TestCase):
class PoolEnforcementTests(TestCase):
def setUp(self):
self.redis = FakeRedis()
self.group = ServerGroup.objects.create(
name="test-pool",
max_streams=2,
)
self.group = ServerGroup.objects.create(name="test-pool")
self.account = M3UAccount.objects.create(
name="Test Account",
account_type="XC",
@ -207,8 +203,8 @@ class PoolEnforcementTests(TestCase):
self.assertEqual(self.redis._data[cred_key], 0)
self.assertEqual(self.redis._data[profile_key], 0)
def test_same_credential_capped_at_profile_max_not_group_max(self):
"""Maintainer example: group max=2 but each login only allows 1."""
def test_same_credential_capped_at_profile_max(self):
"""Shared credential counter is capped by each profile's max_streams."""
account2 = M3UAccount.objects.create(
name="Second Account",
account_type="XC",
@ -229,11 +225,8 @@ class PoolEnforcementTests(TestCase):
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_both_stream_in_same_group(self):
"""Different provider logins keep separate credential counters in one group."""
account = M3UAccount.objects.create(
name="Grouped multi-login",
account_type="XC",
@ -288,7 +281,7 @@ class PoolEnforcementTests(TestCase):
):
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)
self.assertEqual(get_credential_connection_count(profile, self.redis), 0)
def test_release_when_profile_row_deleted(self):
profile_id = self.profile.id
@ -328,12 +321,55 @@ class PoolEnforcementTests(TestCase):
self.assertEqual(reason, "credential_full")
class StaleAssignmentTests(TestCase):
def test_stale_assignment_releases_counters(self):
from apps.channels.models import Channel, Stream
redis = FakeRedis()
group = ServerGroup.objects.create(name="stale-group")
account = M3UAccount.objects.create(
name="Stale Account",
account_type="XC",
username="user",
password="pass",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
profile = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
profile.max_streams = 1
profile.save()
stream = Stream.objects.create(name="Stale Stream", m3u_account=account)
channel = Channel.objects.create(channel_number=502, name="Stale Channel")
channel.streams.add(stream)
reserve_profile_slot(profile, redis)
redis.set(f"channel_stream:{channel.id}", stream.id)
redis.set(f"stream_profile:{stream.id}", profile.id)
profile_key = profile_connections_key(profile.id)
cred_key = server_group_connections_key(
group.id, get_profile_credential_fingerprint(profile)
)
self.assertEqual(redis._data[profile_key], 1)
self.assertEqual(redis._data[cred_key], 1)
with patch("core.utils.RedisClient.get_client", return_value=redis):
channel._release_stale_stream_assignment(redis, stream.id)
self.assertNotIn(f"channel_stream:{channel.id}", redis._data)
self.assertNotIn(f"stream_profile:{stream.id}", redis._data)
self.assertEqual(redis._data[profile_key], 0)
self.assertEqual(redis._data[cred_key], 0)
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)
group = ServerGroup.objects.create(name="switch-group")
account = M3UAccount.objects.create(
name="Switch Account",
account_type="XC",
@ -377,6 +413,58 @@ class UpdateStreamProfileTests(TestCase):
self.assertEqual(redis._data[profile_connections_key(profile_b.id)], 1)
self.assertEqual(redis._data[cred_key], cred_before)
def test_switch_moves_credential_counter_when_login_changes(self):
from apps.channels.models import Channel, Stream
redis = FakeRedis()
group = ServerGroup.objects.create(name="cred-switch-group")
account = M3UAccount.objects.create(
name="Cred Switch Account",
account_type="XC",
username="login_a",
password="pass_a",
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_login",
is_default=False,
is_active=True,
max_streams=1,
search_pattern="",
replace_pattern="",
)
fp_a = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
fp_b = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
key_a = server_group_connections_key(group.id, fp_a)
key_b = server_group_connections_key(group.id, fp_b)
stream = Stream.objects.create(name="Cred Switch Stream", m3u_account=account)
channel = Channel.objects.create(channel_number=503, name="Cred Switch Channel")
channel.streams.add(stream)
with patch(
"apps.m3u.connection_pool.get_profile_credential_fingerprint",
side_effect=lambda profile: fp_a if profile.id == profile_a.id else fp_b,
):
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)
self.assertEqual(redis._data[key_a], 1)
self.assertNotIn(key_b, redis._data)
with patch("core.utils.RedisClient.get_client", return_value=redis):
self.assertTrue(channel.update_stream_profile(profile_b.id))
self.assertEqual(redis._data[key_a], 0)
self.assertEqual(redis._data[key_b], 1)
class VodProfileSelectionTests(TestCase):
def test_get_m3u_profile_skips_default_when_profile_full(self):
@ -413,3 +501,66 @@ class VodProfileSelectionTests(TestCase):
self.assertIsNotNone(result)
selected, _connections = result
self.assertEqual(selected.id, alt.id)
def test_get_m3u_profile_skips_default_when_credential_pool_full(self):
from apps.proxy.vod_proxy.views import _get_m3u_profile
group = ServerGroup.objects.create(name="vod-cred-pool")
account = M3UAccount.objects.create(
name="VOD pooled",
account_type="XC",
username="shared_user",
password="shared_pass",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
default = M3UAccountProfile.objects.get(m3u_account=account, is_default=True)
default.max_streams = 1
default.save()
alt = M3UAccountProfile.objects.create(
m3u_account=account,
name="alt_login",
is_default=False,
is_active=True,
max_streams=1,
search_pattern="",
replace_pattern="",
)
other_account = M3UAccount.objects.create(
name="Other pooled account",
account_type="XC",
username="shared_user",
password="shared_pass",
server_url="http://xc.example.com",
server_group=group,
max_streams=5,
)
other_profile = M3UAccountProfile.objects.get(
m3u_account=other_account, is_default=True
)
other_profile.max_streams = 1
other_profile.save()
fp_shared = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
fp_alt = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
redis = FakeRedis()
with patch(
"apps.m3u.connection_pool.get_profile_credential_fingerprint",
side_effect=lambda profile: (
fp_shared
if profile.id in (default.id, other_profile.id)
else fp_alt
),
):
self.assertTrue(reserve_profile_slot(other_profile, redis)[0])
with patch("core.utils.RedisClient.get_client", return_value=redis):
result = _get_m3u_profile(account, None, None)
self.assertIsNotNone(result)
selected, _connections = result
self.assertEqual(selected.id, alt.id)

88
apps/proxy/live_proxy/url_utils.py
View file

@ -8,7 +8,10 @@ from typing import Optional, Tuple, List
from django.shortcuts import get_object_or_404
from apps.channels.models import Channel, Stream
from apps.m3u.models import M3UAccount, M3UAccountProfile
from apps.m3u.connection_pool import profile_connections_key
from apps.m3u.connection_pool import (
get_profile_connection_count,
profile_available_for_channel_switch,
)
from core.models import UserAgent, CoreSettings, StreamProfile
from .utils import get_logger
from uuid import UUID
@ -54,12 +57,14 @@ def get_stream_object(id: str):
logger.info(f"Fetching stream hash {id}")
return get_object_or_404(Stream, stream_hash=id)
def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int], bool]:
def generate_stream_url(
channel_id: str,
) -> Tuple[str, str, bool, Optional[int], bool, Optional[str]]:
"""
Generate the appropriate stream URL for a channel or stream based on its profile settings.
Returns:
Tuple: (stream_url, user_agent, transcode_flag, profile_id, slot_reserved)
Tuple: (stream_url, user_agent, transcode_flag, profile_id, slot_reserved, error_reason)
"""
try:
channel_or_stream = get_stream_object(channel_id)
@ -71,12 +76,12 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int],
if not stream.m3u_account:
logger.error(f"Stream {stream.id} has no M3U account")
return None, None, False, None, False
return None, None, False, None, False, "Stream has no M3U account"
stream_id, profile_id, error_reason, slot_reserved = stream.get_stream()
if not stream_id or not profile_id:
logger.error(f"No profile available for stream {stream.id}: {error_reason}")
return None, None, False, None, False
return None, None, False, None, False, error_reason
try:
profile = M3UAccountProfile.objects.get(id=profile_id)
@ -95,24 +100,23 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int],
transcode = not stream_profile.is_proxy()
stream_profile_id = stream_profile.id
return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved
return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved, None
except Exception as e:
logger.error(f"Error generating stream URL for stream {stream.id}: {e}")
if slot_reserved:
stream.release_stream()
return None, None, False, None, False
return None, None, False, None, False, str(e)
# Handle channel preview (existing logic)
channel = channel_or_stream
# Get stream and profile for this channel
# Note: get_stream now returns 3 values (stream_id, profile_id, error_reason)
stream_id, profile_id, error_reason, slot_reserved = channel.get_stream()
if not stream_id or not profile_id:
logger.error(f"No stream available for channel {channel_id}: {error_reason}")
return None, None, False, None, False
return None, None, False, None, False, error_reason
# get_stream() allocated a connection slot - ensure it's released on any error
try:
@ -142,16 +146,16 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int],
stream_profile_id = stream_profile.id
return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved
return stream_url, stream_user_agent, transcode, stream_profile_id, slot_reserved, None
except Exception as e:
logger.error(f"Error generating stream URL for channel {channel_id}: {e}")
if slot_reserved:
if not channel.release_stream():
logger.warning(f"Failed to release stream for channel {channel_id} after URL generation error")
return None, None, False, None, False
return None, None, False, None, False, str(e)
except Exception as e:
logger.error(f"Error generating stream URL: {e}")
return None, None, False, None, False
return None, None, False, None, False, str(e)
def transform_url(input_url: str, search_pattern: str, replace_pattern: str) -> str:
"""
@ -229,10 +233,6 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
selected_profile = None
for profile in profiles:
if redis_client:
current_connections = int(
redis_client.get(profile_connections_key(profile.id)) or 0
)
channel_using_profile = False
existing_stream_id = redis_client.get(f"channel_stream:{channel.id}")
if existing_stream_id:
@ -242,20 +242,22 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int]
if existing_profile_id and int(existing_profile_id) == profile.id:
channel_using_profile = True
effective_connections = current_connections - (
1 if channel_using_profile else 0
)
if profile.max_streams == 0 or effective_connections < profile.max_streams:
if profile_available_for_channel_switch(
profile,
redis_client,
channel_already_on_profile=channel_using_profile,
):
current_connections = get_profile_connection_count(
profile, redis_client
)
selected_profile = profile
logger.debug(
f"Selected profile {profile.id} with "
f"{effective_connections}/{profile.max_streams} effective connections"
f"{current_connections}/{profile.max_streams} connections"
)
break
logger.debug(
f"Profile {profile.id} at max connections: "
f"{effective_connections}/{profile.max_streams}"
f"Profile {profile.id} unavailable for channel switch"
)
else:
selected_profile = profile
@ -368,38 +370,38 @@ def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = No
selected_profile = None
for profile in profiles:
# Check connection availability
if redis_client:
profile_connections_key = f"profile_connections:{profile.id}"
current_connections = int(redis_client.get(profile_connections_key) or 0)
# Check if this channel is already using this profile
channel_using_profile = False
existing_stream_id = redis_client.get(f"channel_stream:{channel.id}")
if existing_stream_id:
# Decode bytes to string/int for proper Redis key lookup
existing_stream_id = existing_stream_id
existing_profile_id = redis_client.get(f"stream_profile:{existing_stream_id}")
existing_profile_id = redis_client.get(
f"stream_profile:{existing_stream_id}"
)
if existing_profile_id and int(existing_profile_id) == profile.id:
channel_using_profile = True
logger.debug(f"Channel {channel.id} already using profile {profile.id}")
logger.debug(
f"Channel {channel.id} already using profile {profile.id}"
)
# Calculate effective connections (subtract 1 if channel already using this profile)
effective_connections = current_connections - (1 if channel_using_profile else 0)
# Check if profile has available slots
if profile.max_streams == 0 or effective_connections < profile.max_streams:
if profile_available_for_channel_switch(
profile,
redis_client,
channel_already_on_profile=channel_using_profile,
):
current_connections = get_profile_connection_count(
profile, redis_client
)
selected_profile = profile
logger.debug(
f"Found available profile {profile.id} for stream {stream.id}: "
f"{effective_connections}/{profile.max_streams} effective "
f"(current: {current_connections}, already using: {channel_using_profile})"
f"{current_connections}/{profile.max_streams} "
f"(already using: {channel_using_profile})"
)
break
else:
logger.debug(f"Profile {profile.id} at max connections: {effective_connections}/{profile.max_streams} (current: {current_connections}, already using: {channel_using_profile})")
logger.debug(
f"Profile {profile.id} unavailable for alternate stream {stream.id}"
)
else:
# No Redis available, assume first active profile is okay
selected_profile = profile
break

23
apps/proxy/live_proxy/views.py
View file

@ -221,9 +221,14 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
# Try to get a stream with fixed interval retries
while should_retry and time.time() - wait_start_time < retry_timeout:
attempt += 1
stream_url, stream_user_agent, transcode, profile_value, slot_reserved = (
generate_stream_url(channel_id)
)
(
stream_url,
stream_user_agent,
transcode,
profile_value,
slot_reserved,
error_reason,
) = generate_stream_url(channel_id)
if stream_url is not None:
logger.info(
@ -233,7 +238,6 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
# On first failure, check if the error is retryable
if attempt == 1:
_, _, error_reason, _ = channel.get_stream()
if error_reason and "maximum connection limits" not in error_reason:
logger.warning(
f"[{client_id}] Can't retry - error not related to connection limits: {error_reason}"
@ -266,9 +270,14 @@ def stream_ts(request, channel_id, user=None, force_output_format=None):
logger.info(
f"[{client_id}] Making final attempt {attempt} at timeout boundary"
)
stream_url, stream_user_agent, transcode, profile_value, slot_reserved = (
generate_stream_url(channel_id)
)
(
stream_url,
stream_user_agent,
transcode,
profile_value,
slot_reserved,
error_reason,
) = generate_stream_url(channel_id)
if stream_url is not None:
logger.info(
f"[{client_id}] Successfully obtained stream on final attempt for channel {channel_id}"

11
apps/proxy/vod_proxy/views.py
View file

@ -262,7 +262,7 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None):
from core.utils import RedisClient
from apps.m3u.connection_pool import (
get_profile_connection_count,
profile_has_capacity_for_selection,
pool_has_capacity_for_profile,
)
redis_client = RedisClient.get_client()
@ -316,7 +316,7 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None):
profile_connections_key = f"profile_connections:{profile.id}"
current_connections = get_profile_connection_count(profile, redis_client)
if profile_has_capacity_for_selection(profile, redis_client):
if pool_has_capacity_for_profile(profile, redis_client):
logger.info(f"[PROFILE-SELECTION] Using requested profile {profile.id}: {current_connections}/{profile.max_streams} connections")
return (profile, current_connections)
logger.warning(f"[PROFILE-SELECTION] Requested profile {profile.id} is at capacity: {current_connections}/{profile.max_streams}")
@ -340,11 +340,14 @@ def _get_m3u_profile(m3u_account, profile_id, session_id=None):
for profile in profiles:
current_connections = get_profile_connection_count(profile, redis_client)
if profile_has_capacity_for_selection(profile, redis_client):
if pool_has_capacity_for_profile(profile, redis_client):
logger.info(f"[PROFILE-SELECTION] Selected profile {profile.id} ({profile.name}): {current_connections}/{profile.max_streams} connections")
return (profile, current_connections)
else:
logger.debug(f"[PROFILE-SELECTION] Profile {profile.id} at capacity: {current_connections}/{profile.max_streams}")
logger.debug(
f"[PROFILE-SELECTION] Profile {profile.id} unavailable "
f"(profile={current_connections}/{profile.max_streams})"
)
# All profiles are at capacity - return None to trigger error response
logger.error(f"[PROFILE-SELECTION] All profiles at capacity for M3U account {m3u_account.id}, rejecting request")

28
frontend/src/components/ServerGroupsManagerModal.jsx Normal file
View file

@ -0,0 +1,28 @@
import { Modal } from '@mantine/core';
import ServerGroupsTable from './tables/ServerGroupsTable';
const ServerGroupsManagerModal = ({
isOpen,
onClose,
onGroupCreated,
openCreateOnMount = false,
}) => {
return (
<Modal
opened={isOpen}
onClose={onClose}
title="Server Groups"
size="lg"
centered
>
{isOpen ? (
<ServerGroupsTable
onGroupCreated={onGroupCreated}
openCreateOnMount={openCreateOnMount}
/>
) : null}
</Modal>
);
};
export default ServerGroupsManagerModal;

59
frontend/src/components/__tests__/ServerGroupsManagerModal.test.jsx Normal file
View file

@ -0,0 +1,59 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { render, screen, waitFor } from '@testing-library/react';
import { MantineProvider } from '@mantine/core';
import ServerGroupsManagerModal from '../ServerGroupsManagerModal';
vi.mock('../../api', () => ({
default: {
deleteServerGroup: vi.fn(),
},
}));
vi.mock('../../store/serverGroups', () => ({
default: (selector) =>
selector({
serverGroups: [
{ id: 1, name: 'Pool A' },
{ id: 2, name: 'Pool B' },
],
fetchServerGroups: vi.fn().mockResolvedValue(undefined),
}),
}));
vi.mock('../../store/playlists', () => ({
default: (selector) =>
selector({
playlists: [{ id: 10, server_group: 2 }],
}),
}));
vi.mock('../forms/ServerGroup', () => ({
default: () => null,
}));
vi.mock('../../hooks/useLocalStorage', () => ({
default: () => ['default', vi.fn()],
}));
const renderModal = (props = {}) =>
render(
<MantineProvider>
<ServerGroupsManagerModal isOpen onClose={vi.fn()} {...props} />
</MantineProvider>
);
describe('ServerGroupsManagerModal', () => {
beforeEach(() => {
vi.clearAllMocks();
});
it('opens without Mantine Tooltip errors', async () => {
expect(() => renderModal()).not.toThrow();
await waitFor(() => {
expect(screen.getByText('Pool A')).toBeInTheDocument();
expect(screen.getByText('Pool B')).toBeInTheDocument();
expect(screen.getByText('Accounts')).toBeInTheDocument();
});
});
});

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

@ -36,6 +36,7 @@ import {
prepareSubmitValues,
updatePlaylist,
} from '../../utils/forms/M3uUtils.js';
import ServerGroupsManagerModal from '../ServerGroupsManagerModal';
const M3U = ({
m3uAccount = null,
@ -56,6 +57,8 @@ const M3U = ({
const [groupFilterModalOpen, setGroupFilterModalOpen] = useState(false);
const [filterModalOpen, setFilterModalOpen] = useState(false);
const [scheduleType, setScheduleType] = useState('interval');
const [serverGroupsManagerOpen, setServerGroupsManagerOpen] = useState(false);
const [serverGroupsCreateOnOpen, setServerGroupsCreateOnOpen] = useState(false);
const form = useForm({
mode: 'uncontrolled',
@ -366,18 +369,36 @@ const M3U = ({
name="server_group"
label="Server Group"
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(
serverGroups.map((group) => ({
label:
group.max_streams > 0
? `${group.name} (max: ${group.max_streams})`
: `${group.name} (unlimited)`,
value={form.getValues().server_group}
onChange={(value) => {
if (value === '__new__') {
setServerGroupsCreateOnOpen(true);
setServerGroupsManagerOpen(true);
return;
}
form.setFieldValue('server_group', value);
}}
data={[
{ value: '0', label: '(None)' },
...serverGroups.map((group) => ({
label: group.name,
value: `${group.id}`,
}))
)}
})),
{ value: '__new__', label: '+ Add server group...' },
]}
/>
<Button
variant="subtle"
size="compact-xs"
onClick={() => {
setServerGroupsCreateOnOpen(false);
setServerGroupsManagerOpen(true);
}}
style={{ alignSelf: 'flex-start' }}
>
Manage server groups
</Button>
<Select
id="user_agent"
@ -513,6 +534,20 @@ const M3U = ({
/>
</>
)}
<ServerGroupsManagerModal
isOpen={serverGroupsManagerOpen}
onClose={() => {
setServerGroupsManagerOpen(false);
setServerGroupsCreateOnOpen(false);
}}
openCreateOnMount={serverGroupsCreateOnOpen}
onGroupCreated={(group) => {
if (group?.id) {
form.setFieldValue('server_group', `${group.id}`);
}
}}
/>
</>
);
};

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

@ -3,20 +3,21 @@ import { useForm } from 'react-hook-form';
import { yupResolver } from '@hookform/resolvers/yup';
import * as Yup from 'yup';
import API from '../../api';
import { Button, Flex, Modal, NumberInput, TextInput } from '@mantine/core';
import { Button, Flex, Modal, TextInput } from '@mantine/core';
const schema = Yup.object({
name: Yup.string().required('Name is required'),
max_streams: Yup.number()
.min(0, 'Must be 0 or greater')
.required('Max streams is required'),
});
const ServerGroupForm = ({ serverGroup = null, isOpen, onClose }) => {
const ServerGroupForm = ({
serverGroup = null,
isOpen,
onClose,
onSaved,
}) => {
const defaultValues = useMemo(
() => ({
name: serverGroup?.name || '',
max_streams: serverGroup?.max_streams ?? 0,
}),
[serverGroup]
);
@ -26,23 +27,21 @@ const ServerGroupForm = ({ serverGroup = null, isOpen, onClose }) => {
handleSubmit,
formState: { errors, isSubmitting },
reset,
setValue,
watch,
} = useForm({
defaultValues,
resolver: yupResolver(schema),
});
const onSubmit = async (values) => {
const payload = {
...values,
max_streams: Number(values.max_streams),
};
let response;
if (serverGroup?.id) {
await API.updateServerGroup({ id: serverGroup.id, ...payload });
response = await API.updateServerGroup({ id: serverGroup.id, ...values });
} else {
await API.addServerGroup(payload);
response = await API.addServerGroup(values);
}
if (response) {
onSaved?.(response);
}
reset();
@ -57,26 +56,23 @@ const ServerGroupForm = ({ serverGroup = null, isOpen, onClose }) => {
return null;
}
const maxStreams = watch('max_streams');
return (
<Modal opened={isOpen} onClose={onClose} title="Server Group">
<Modal
opened={isOpen}
onClose={onClose}
title="Server Group"
centered
withinPortal
zIndex={400}
>
<form onSubmit={handleSubmit(onSubmit)}>
<TextInput
label="Name"
description="Accounts in this group share connection limits when they use the same provider login. Limits come from each account profile's max streams."
{...register('name')}
error={errors.name?.message}
/>
<NumberInput
label="Max Streams"
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)}
error={errors.max_streams?.message}
/>
<Flex mih={50} gap="xs" justify="flex-end" align="flex-end">
<Button size="small" type="submit" disabled={isSubmitting}>
Submit

47
frontend/src/components/tables/M3UsTable.jsx
View file

@ -8,6 +8,7 @@ import React, {
import API from '../../api';
import usePlaylistsStore from '../../store/playlists';
import M3UForm from '../forms/M3U';
import ServerGroupsManagerModal from '../ServerGroupsManagerModal';
import { TableHelper } from '../../helpers';
import {
useMantineTheme,
@ -144,6 +145,7 @@ const M3UTable = () => {
const [data, setData] = useState([]);
const [sorting, setSorting] = useState([{ id: 'name', desc: '' }]);
const [deleting, setDeleting] = useState(false);
const [serverGroupsManagerOpen, setServerGroupsManagerOpen] = useState(false);
const playlists = usePlaylistsStore((s) => s.playlists);
const refreshProgress = usePlaylistsStore((s) => s.refreshProgress);
@ -988,21 +990,31 @@ const M3UTable = () => {
>
M3U Accounts
</Text>
<Button
leftSection={<SquarePlus size={14} />}
variant="light"
size="xs"
onClick={() => editPlaylist()}
p={5}
color="green"
style={{
borderWidth: '1px',
borderColor: 'green',
color: 'white',
}}
>
Add M3U
</Button>
<Flex gap={6}>
<Button
variant="light"
size="xs"
onClick={() => setServerGroupsManagerOpen(true)}
p={5}
>
Server Groups
</Button>
<Button
leftSection={<SquarePlus size={14} />}
variant="light"
size="xs"
onClick={() => editPlaylist()}
p={5}
color="green"
style={{
borderWidth: '1px',
borderColor: 'green',
color: 'white',
}}
>
Add M3U
</Button>
</Flex>
</Flex>
<Paper
@ -1051,6 +1063,11 @@ const M3UTable = () => {
playlistCreated={playlistCreated}
/>
<ServerGroupsManagerModal
isOpen={serverGroupsManagerOpen}
onClose={() => setServerGroupsManagerOpen(false)}
/>
<ConfirmationDialog
opened={confirmDeleteOpen}
onClose={() => setConfirmDeleteOpen(false)}

170
frontend/src/components/tables/ServerGroupsTable.jsx
View file

@ -1,7 +1,10 @@
import { useEffect, useMemo, useState } from 'react';
import { useEffect, useMemo, useRef, useState } from 'react';
import API from '../../api';
import useServerGroupsStore from '../../store/serverGroups';
import usePlaylistsStore from '../../store/playlists';
import useWarningsStore from '../../store/warnings';
import ServerGroupForm from '../forms/ServerGroup';
import ConfirmationDialog from '../ConfirmationDialog';
import {
ActionIcon,
Box,
@ -11,15 +14,15 @@ import {
Paper,
Stack,
Text,
Tooltip,
} from '@mantine/core';
import { SquareMinus, SquarePen, SquarePlus } from 'lucide-react';
import { CustomTable, useTable } from './CustomTable';
import useLocalStorage from '../../hooks/useLocalStorage';
import './table.css';
const RowActions = ({ row, editServerGroup, deleteServerGroup }) => {
return (
<>
<Flex gap={4}>
<ActionIcon
variant="transparent"
size="sm"
@ -36,35 +39,56 @@ const RowActions = ({ row, editServerGroup, deleteServerGroup }) => {
>
<SquareMinus size="18" />
</ActionIcon>
</>
</Flex>
);
};
const ServerGroupsTable = () => {
const ServerGroupsTable = ({
onGroupCreated,
openCreateOnMount = false,
}) => {
const [serverGroup, setServerGroup] = useState(null);
const [serverGroupModalOpen, setServerGroupModalOpen] = useState(false);
const [confirmDeleteOpen, setConfirmDeleteOpen] = useState(false);
const [deleteTarget, setDeleteTarget] = useState(null);
const [groupToDelete, setGroupToDelete] = useState(null);
const [deleting, setDeleting] = useState(false);
const openedCreateOnMount = useRef(false);
const isWarningSuppressed = useWarningsStore((s) => s.isWarningSuppressed);
const suppressWarning = useWarningsStore((s) => s.suppressWarning);
const serverGroups = useServerGroupsStore((state) => state.serverGroups);
const fetchServerGroups = useServerGroupsStore(
(state) => state.fetchServerGroups
);
const playlists = usePlaylistsStore((state) => state.playlists);
const [tableSize] = useLocalStorage('table-size', 'default');
const tableData = useMemo(
() =>
serverGroups.map((group) => ({
...group,
accountCount: playlists.filter(
(playlist) => playlist.server_group === group.id
).length,
})),
[serverGroups, playlists]
);
const columns = useMemo(
() => [
{
header: 'Name',
accessorKey: 'name',
size: 175,
size: 150,
cell: ({ cell }) => <Text size="sm">{cell.getValue()}</Text>,
},
{
header: 'Max Streams',
accessorKey: 'max_streams',
size: 100,
cell: ({ cell }) => {
const value = cell.getValue();
return value === 0 ? 'Unlimited' : value;
},
header: 'Accounts',
accessorKey: 'accountCount',
size: 80,
cell: ({ cell }) => <Text size="sm">{cell.getValue() ?? 0}</Text>,
},
{
id: 'actions',
@ -82,8 +106,26 @@ const ServerGroupsTable = () => {
setServerGroupModalOpen(true);
};
const deleteServerGroup = async (id) => {
await API.deleteServerGroup(id);
const executeDeleteServerGroup = async (id) => {
setDeleting(true);
try {
await API.deleteServerGroup(id);
} finally {
setDeleting(false);
setConfirmDeleteOpen(false);
}
};
const deleteServerGroup = (id) => {
const group = tableData.find((item) => item.id === id);
setGroupToDelete(group);
setDeleteTarget(id);
if (isWarningSuppressed('delete-server-group')) {
return executeDeleteServerGroup(id);
}
setConfirmDeleteOpen(true);
};
const closeServerGroupForm = () => {
@ -91,17 +133,32 @@ const ServerGroupsTable = () => {
setServerGroupModalOpen(false);
};
const handleServerGroupSaved = (savedGroup) => {
if (!serverGroup?.id) {
onGroupCreated?.(savedGroup);
}
};
useEffect(() => {
fetchServerGroups().finally(() => setIsLoading(false));
}, [fetchServerGroups]);
const renderHeaderCell = (header) => {
return (
<Text size="sm" name={header.id}>
{header.column.columnDef.header}
</Text>
);
};
useEffect(() => {
if (!openCreateOnMount) {
openedCreateOnMount.current = false;
return;
}
if (!isLoading && !openedCreateOnMount.current) {
openedCreateOnMount.current = true;
editServerGroup();
}
}, [openCreateOnMount, isLoading]);
const renderHeaderCell = (header) => (
<Text size="sm" name={header.id}>
{header.column.columnDef.header}
</Text>
);
const renderBodyCell = ({ row }) => {
return (
@ -115,14 +172,14 @@ const ServerGroupsTable = () => {
const table = useTable({
columns,
data: serverGroups,
allRowIds: serverGroups.map((group) => group.id),
data: tableData,
allRowIds: tableData.map((group) => group.id),
bodyCellRenderFns: {
actions: renderBodyCell,
},
headerCellRenderFns: {
name: renderHeaderCell,
max_streams: renderHeaderCell,
accountCount: renderHeaderCell,
actions: renderHeaderCell,
},
});
@ -146,23 +203,22 @@ const ServerGroupsTable = () => {
}}
>
<Flex gap={6}>
<Tooltip label="Create a shared connection pool for multiple accounts">
<Button
leftSection={<SquarePlus size={18} />}
variant="light"
size="xs"
onClick={() => editServerGroup()}
p={5}
color="green"
style={{
borderWidth: '1px',
borderColor: 'green',
color: 'white',
}}
>
Add Server Group
</Button>
</Tooltip>
<Button
leftSection={<SquarePlus size={18} />}
variant="light"
size="xs"
onClick={() => editServerGroup()}
p={5}
color="green"
title="Create a shared connection pool for multiple accounts"
style={{
borderWidth: '1px',
borderColor: 'green',
color: 'white',
}}
>
Add Server Group
</Button>
</Flex>
</Box>
</Paper>
@ -185,7 +241,7 @@ const ServerGroupsTable = () => {
borderRadius: 'var(--mantine-radius-default)',
}}
>
<div style={{ minWidth: 400 }}>
<div style={{ minWidth: 320 }}>
<CustomTable table={table} />
</div>
</Box>
@ -195,6 +251,34 @@ const ServerGroupsTable = () => {
serverGroup={serverGroup}
isOpen={serverGroupModalOpen}
onClose={closeServerGroupForm}
onSaved={handleServerGroupSaved}
/>
<ConfirmationDialog
opened={confirmDeleteOpen}
onClose={() => setConfirmDeleteOpen(false)}
onConfirm={() => executeDeleteServerGroup(deleteTarget)}
loading={deleting}
title="Confirm Server Group Deletion"
message={
groupToDelete ? (
<div style={{ whiteSpace: 'pre-line' }}>
{`Are you sure you want to delete the following server group?
Name: ${groupToDelete.name}
Accounts: ${groupToDelete.accountCount ?? 0}
Accounts in this group will no longer share connection limits. This action cannot be undone.`}
</div>
) : (
'Are you sure you want to delete this server group? This action cannot be undone.'
)
}
confirmLabel="Delete"
cancelLabel="Cancel"
actionKey="delete-server-group"
onSuppressChange={suppressWarning}
zIndex={401}
/>
</Stack>
);

62
frontend/src/components/tables/__tests__/ServerGroupsTable.test.jsx Normal file
View file

@ -0,0 +1,62 @@
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { render, screen, waitFor } from '@testing-library/react';
import { MantineProvider } from '@mantine/core';
import ServerGroupsTable from '../ServerGroupsTable';
const renderTable = () =>
render(
<MantineProvider>
<ServerGroupsTable />
</MantineProvider>
);
vi.mock('../../../api', () => ({
default: {
deleteServerGroup: vi.fn(),
},
}));
vi.mock('../../../store/serverGroups', () => ({
default: (selector) =>
selector({
serverGroups: [
{ id: 1, name: 'Pool A' },
{ id: 2, name: 'Pool B' },
],
fetchServerGroups: vi.fn().mockResolvedValue(undefined),
}),
}));
vi.mock('../../../store/playlists', () => ({
default: (selector) =>
selector({
playlists: [
{ id: 10, server_group: 2 },
{ id: 11, server_group: 1 },
],
}),
}));
vi.mock('../../forms/ServerGroup', () => ({
default: () => null,
}));
vi.mock('../../../hooks/useLocalStorage', () => ({
default: () => ['default', vi.fn()],
}));
describe('ServerGroupsTable', () => {
beforeEach(() => {
vi.clearAllMocks();
});
it('renders server groups without tooltip errors', async () => {
expect(() => renderTable()).not.toThrow();
await waitFor(() => {
expect(screen.getByText('Pool A')).toBeInTheDocument();
expect(screen.getByText('Pool B')).toBeInTheDocument();
expect(screen.getByText('Accounts')).toBeInTheDocument();
});
});
});

16
frontend/src/pages/Settings.jsx
View file

@ -14,9 +14,6 @@ import {
const UserAgentsTable = React.lazy(
() => import('../components/tables/UserAgentsTable.jsx')
);
const ServerGroupsTable = React.lazy(
() => import('../components/tables/ServerGroupsTable.jsx')
);
const StreamProfilesTable = React.lazy(
() => import('../components/tables/StreamProfilesTable.jsx')
);
@ -138,19 +135,6 @@ const SettingsPage = () => {
</AccordionPanel>
</AccordionItem>
<AccordionItem value="server-groups">
<AccordionControl>Server Groups</AccordionControl>
<AccordionPanel>
<ErrorBoundary>
<Suspense fallback={<Loader />}>
<ServerGroupsTable
active={accordianValue === 'server-groups'}
/>
</Suspense>
</ErrorBoundary>
</AccordionPanel>
</AccordionItem>
<AccordionItem value="user-agents">
<AccordionControl>User-Agents</AccordionControl>
<AccordionPanel>