Connection Admission (#710)

* Connection admission
* Fetch assets with version
* Dockerfile udate
This commit is contained in:
euzu
2026-04-10 14:56:57 +02:00
committed by GitHub
parent f6e189f89c
commit 5dd7f2dc91
25 changed files with 2859 additions and 101 deletions
+256 -1
View File
@@ -452,6 +452,106 @@ pub(in crate::api) fn get_stream_options(app_state: &Arc<AppState>) -> StreamOpt
StreamOptions { stream_retry, buffer_enabled, buffer_size, pipe_provider_stream }
}
pub(in crate::api) fn get_effective_admission_strategies(app_state: &Arc<AppState>) -> Vec<crate::model::AdmissionStrategy> {
use crate::model::AdmissionStrategy;
let config = app_state.app_config.config.load();
let stream_config = config.reverse_proxy.as_ref().and_then(|rp| rp.stream.as_ref());
match stream_config {
Some(sc) if sc.admission_strategies.is_some() => sc.admission_strategies.clone().unwrap_or_default(),
Some(sc) if sc.grace_period_millis > 0 => {
vec![if sc.grace_period_hold_stream {
AdmissionStrategy::GraceHoldStream
} else {
AdmissionStrategy::GraceInstantStream
}]
}
_ => Vec::new(),
}
}
#[allow(clippy::too_many_arguments)]
pub(in crate::api) async fn resolve_admission_with_strategies(
app_state: &Arc<AppState>,
username: &str,
max_connections: u32,
soft_connections: u16,
client_ip: &str,
_fingerprint: &Fingerprint,
is_session_request: bool,
session_token: Option<&str>,
) -> (crate::api::model::ConnectionAdmission, Option<crate::api::model::GraceMode>) {
use crate::api::model::{AdmissionDecision, ConnectionAdmission, ConnectionKind, StrategyContext, evaluate_strategy};
use shared::model::UserConnectionPermission;
let admission = if is_session_request {
app_state.get_connection_admission_for_session(
username, max_connections, soft_connections, session_token.unwrap_or_default(),
).await
} else {
app_state.get_connection_admission(username, max_connections, soft_connections).await
};
if admission.permission != UserConnectionPermission::Exhausted {
return (admission, None);
}
let strategies = get_effective_admission_strategies(app_state);
if strategies.is_empty() {
debug!("No admission strategies configured, denying request for user {username}");
return (admission, None);
}
let mut candidates = app_state.active_users.get_eviction_candidates(username, client_ip).await;
let ctx = StrategyContext { username, client_ip, strategies: &strategies };
for strategy in &strategies {
match evaluate_strategy(*strategy, &ctx, &candidates) {
AdmissionDecision::NoMatch => {}
AdmissionDecision::Grace(mode) => {
if app_state.active_users.grant_grace(username).await {
return (
ConnectionAdmission {
permission: UserConnectionPermission::GracePeriod,
kind: Some(ConnectionKind::Normal),
},
Some(mode),
);
}
debug!("Grace grant rejected for user {username}, continuing with later strategies");
}
AdmissionDecision::Evict(target) => {
debug!("Evicting connection {} for user {username}", target.addr);
app_state.connection_manager.release_connection_as_kicked(&target.addr).await;
let retry_admission = if is_session_request {
app_state
.get_connection_admission_for_session(
username,
max_connections,
soft_connections,
session_token.unwrap_or_default(),
)
.await
} else {
app_state
.get_connection_admission(username, max_connections, soft_connections)
.await
};
if retry_admission.permission == UserConnectionPermission::Allowed {
return (retry_admission, None);
}
debug!("Admission still denied after eviction for user {username}, continuing with later strategies");
candidates = app_state.active_users.get_eviction_candidates(username, client_ip).await;
}
AdmissionDecision::Deny => {
debug!("Admission strategy denied for user {username}");
return (admission, None);
}
}
}
debug!("No admission strategy could admit user {username}");
(admission, None)
}
pub fn get_stream_alternative_url(stream_url: &str, input: &ConfigInput, alias_input: &Arc<ProviderConfig>) -> String {
let Some(input_user_info) = input.get_user_info() else {
return stream_url.to_string();
@@ -654,6 +754,7 @@ async fn create_stream_response_details(
user_priority: i8,
connection_kind: crate::api::model::ConnectionKind,
session_owner: Option<&str>,
grace_hold_override: Option<bool>,
) -> Result<StreamDetails, TuliproxError> {
let mut streaming_strategy = resolve_streaming_strategy(
app_state,
@@ -675,6 +776,9 @@ async fn create_stream_response_details(
&streaming_strategy.provider_stream_state,
grace_period_options.period_millis,
);
if let Some(hold) = grace_hold_override {
grace_period_options.hold_stream = hold;
}
let provider_grace_active = matches!(
streaming_strategy.provider_stream_state,
ProviderStreamState::GracePeriod(_, _)
@@ -971,6 +1075,7 @@ pub async fn force_provider_stream_response(
user_session: &UserSession,
mut stream_channel: StreamChannel,
ctx: ForceStreamRequestContext<'_>,
grace_mode: Option<crate::api::model::GraceMode>,
) -> impl IntoResponse + Send {
let stream_options = get_stream_options(app_state);
let share_stream = false;
@@ -1009,6 +1114,7 @@ pub async fn force_provider_stream_response(
connection_priority_for_kind(ctx.user, connection_kind),
connection_kind,
Some(user_session.token.as_str()),
grace_mode.map(|mode| matches!(mode, crate::api::model::GraceMode::Hold)),
)
.await
{
@@ -1112,6 +1218,7 @@ pub async fn stream_response(
connection_permission: UserConnectionPermission,
connection_kind: crate::api::model::ConnectionKind,
allow_exhausted_shared_reconnect: bool,
grace_mode: Option<crate::api::model::GraceMode>,
) -> impl IntoResponse + Send {
let request_log_stream_url = resolve_request_url_for_logging(input, stream_url);
if log_enabled!(log::Level::Trace) {
@@ -1209,6 +1316,7 @@ pub async fn stream_response(
connection_priority_for_kind(user, connection_kind),
connection_kind,
Some(session_token),
grace_mode.map(|m| matches!(m, crate::api::model::GraceMode::Hold)),
)
.await
{
@@ -1257,7 +1365,7 @@ pub async fn stream_response(
)
.await;
let mut is_stream_shared = share_stream && !deferred_grace_hold_stream;
let mut is_stream_shared = share_stream && !stream_details.has_deferred_provider_open();
if let Some((_header, _status_code, _url, Some(_custom_video))) = stream_details.stream_info.as_ref() {
if stream_details.stream.is_some() {
is_stream_shared = false;
@@ -2556,6 +2664,26 @@ mod tests {
Fingerprint::new(format!("fp-{addr}"), addr.ip().to_string(), addr)
}
fn create_test_app_state_with_stream_config(stream: crate::model::StreamConfig) -> Arc<AppState> {
let mut config = Config::default();
config.reverse_proxy = Some(crate::model::ReverseProxyConfig {
resource_rewrite_disabled: false,
rewrite_secret: [0; 16],
resource_retry: crate::model::ResourceRetryConfig::default(),
disabled_header: None,
stream: Some(stream),
cache: None,
rate_limit: None,
geoip: None,
stream_history: None,
qos_aggregation: None,
});
let mut app_cfg = create_test_app_config();
app_cfg.config = Arc::new(ArcSwap::from_pointee(config));
create_test_app_state_for_config(Arc::new(app_cfg))
}
fn create_test_local_channel(url: &str) -> StreamChannel {
StreamChannel {
target_id: 1,
@@ -2631,6 +2759,132 @@ mod tests {
assert!(admission_failure_video_type(ConnectFailureReason::ProviderError).is_none());
}
#[tokio::test]
async fn effective_admission_strategies_use_legacy_grace_when_field_missing() {
let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig {
retry: true,
metrics_enabled: true,
buffer: None,
grace_period_millis: 2_000,
grace_period_timeout_secs: 8,
grace_period_hold_stream: true,
hls_session_ttl_secs: 10,
catchup_session_ttl_secs: 10,
throttle_str: None,
throttle_kbps: 0,
shared_burst_buffer_mb: 1,
admission_strategies: None,
});
assert_eq!(
get_effective_admission_strategies(&app_state),
vec![crate::model::AdmissionStrategy::GraceHoldStream]
);
}
#[tokio::test]
async fn effective_admission_strategies_respect_explicit_empty_list() {
let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig {
retry: true,
metrics_enabled: true,
buffer: None,
grace_period_millis: 2_000,
grace_period_timeout_secs: 8,
grace_period_hold_stream: true,
hls_session_ttl_secs: 10,
catchup_session_ttl_secs: 10,
throttle_str: None,
throttle_kbps: 0,
shared_burst_buffer_mb: 1,
admission_strategies: Some(vec![]),
});
assert!(get_effective_admission_strategies(&app_state).is_empty());
}
#[tokio::test]
async fn resolve_admission_with_strategies_falls_through_after_failed_grace_grant() {
let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig {
retry: true,
metrics_enabled: true,
buffer: None,
grace_period_millis: 2_000,
grace_period_timeout_secs: 8,
grace_period_hold_stream: true,
hls_session_ttl_secs: 10,
catchup_session_ttl_secs: 10,
throttle_str: None,
throttle_kbps: 0,
shared_burst_buffer_mb: 1,
admission_strategies: Some(vec![
crate::model::AdmissionStrategy::GraceHoldStream,
crate::model::AdmissionStrategy::EvictUserSameIpOldest,
]),
});
let first_addr: std::net::SocketAddr = "127.0.0.1:55151".parse().unwrap_or_else(|_| unreachable!());
let second_addr: std::net::SocketAddr = "127.0.0.1:55152".parse().unwrap_or_else(|_| unreachable!());
let first_fingerprint = create_test_fingerprint(first_addr);
let second_fingerprint = create_test_fingerprint(second_addr);
app_state.connection_manager.add_connection(&first_addr).await;
app_state.connection_manager.add_connection(&second_addr).await;
app_state
.connection_manager
.update_connection(crate::api::model::ConnectionParams {
meter_uid: 1,
username: "fallthrough",
max_connections: 1,
soft_connections: 1,
connection_kind: crate::api::model::ConnectionKind::Normal,
priority: 0,
soft_priority: 10,
fingerprint: &first_fingerprint,
provider: "provider-a",
stream_channel: &create_test_live_channel("http://provider-1.example/live/1.ts"),
user_agent: std::borrow::Cow::Borrowed("ua"),
session_token: Some("tok-first"),
})
.await;
assert!(app_state.active_users.grant_grace("fallthrough").await);
app_state
.connection_manager
.update_connection(crate::api::model::ConnectionParams {
meter_uid: 2,
username: "fallthrough",
max_connections: 1,
soft_connections: 1,
connection_kind: crate::api::model::ConnectionKind::Soft,
priority: 0,
soft_priority: 10,
fingerprint: &second_fingerprint,
provider: "provider-a",
stream_channel: &create_test_live_channel("http://provider-1.example/live/2.ts"),
user_agent: std::borrow::Cow::Borrowed("ua"),
session_token: Some("tok-second"),
})
.await;
let (admission, grace_mode) = resolve_admission_with_strategies(
&app_state,
"fallthrough",
1,
1,
"127.0.0.1",
&create_test_fingerprint("127.0.0.1:55153".parse().unwrap_or_else(|_| unreachable!())),
true,
Some("tok-third"),
)
.await;
assert_eq!(admission.permission, UserConnectionPermission::Allowed);
assert_eq!(admission.kind, Some(crate::api::model::ConnectionKind::Soft));
assert_eq!(grace_mode, None);
}
#[tokio::test]
async fn local_stream_response_registers_active_local_stream() {
let app_state = create_test_app_state();
@@ -2765,6 +3019,7 @@ mod tests {
admission.permission,
admission.kind.unwrap_or(crate::api::model::ConnectionKind::Normal),
false,
None,
)
.await
.into_response();
+21 -13
View File
@@ -454,6 +454,7 @@ async fn hls_api_stream(
user: &user,
session_reservation_ttl_secs: get_hls_session_ttl_secs(&app_state),
},
None,
)
.await
.into_response();
@@ -462,22 +463,28 @@ async fn hls_api_stream(
return axum::http::StatusCode::BAD_REQUEST.into_response();
}
let connection_admission = if (user.max_connections > 0 || user.soft_connections > 0)
let (connection_admission, grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0)
&& app_state.app_config.config.load().user_access_control
{
app_state
.get_connection_admission_for_session(
&user.username,
user.max_connections,
user.soft_connections,
&session.token,
)
.await
crate::api::api_utils::resolve_admission_with_strategies(
&app_state,
&user.username,
user.max_connections,
user.soft_connections,
&fingerprint.client_ip,
&fingerprint,
true,
Some(&session.token),
)
.await
} else {
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: session.connection_kind,
}
(
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: session.connection_kind,
},
None,
)
};
let connection_permission = connection_admission.permission;
let connection_kind = connection_admission
@@ -545,6 +552,7 @@ async fn hls_api_stream(
user: &user,
session_reservation_ttl_secs: get_hls_session_ttl_secs(&app_state),
},
grace_mode,
)
.await
.into_response()
+44 -16
View File
@@ -132,10 +132,30 @@ async fn m3u_api_stream(
}
if pli.item_type.is_local() {
let admission = app_state
.get_connection_admission(&user.username, user.max_connections, user.soft_connections)
.await;
let playback_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id);
let (admission, _grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0)
&& app_state.app_config.config.load().user_access_control
{
crate::api::api_utils::resolve_admission_with_strategies(
app_state,
&user.username,
user.max_connections,
user.soft_connections,
&fingerprint.client_ip,
fingerprint,
true,
Some(playback_session_token.as_str()),
)
.await
} else {
(
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: Some(crate::api::model::ConnectionKind::Normal),
},
None,
)
};
return local_stream_response(
fingerprint,
app_state,
@@ -202,6 +222,7 @@ async fn m3u_api_stream(
user: &user,
session_reservation_ttl_secs: get_session_reservation_ttl_secs(app_state, pli.item_type),
},
None,
)
.await
.into_response();
@@ -211,22 +232,28 @@ async fn m3u_api_stream(
pli.url.clone()
};
let connection_admission = if (user.max_connections > 0 || user.soft_connections > 0)
let (connection_admission, grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0)
&& app_state.app_config.config.load().user_access_control
{
app_state
.get_connection_admission_for_session(
&user.username,
user.max_connections,
user.soft_connections,
&session_key,
)
.await
crate::api::api_utils::resolve_admission_with_strategies(
app_state,
&user.username,
user.max_connections,
user.soft_connections,
&fingerprint.client_ip,
fingerprint,
true,
Some(&session_key),
)
.await
} else {
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: user_session.as_ref().and_then(|session| session.connection_kind),
}
(
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: user_session.as_ref().and_then(|session| session.connection_kind),
},
None,
)
};
let connection_permission = connection_admission.permission;
let connection_kind = connection_admission
@@ -306,6 +333,7 @@ async fn m3u_api_stream(
connection_permission,
connection_kind,
allow_exhausted_shared_reconnect,
grace_mode,
)
.await
.into_response()
+42 -22
View File
@@ -283,19 +283,30 @@ async fn xtream_player_api_stream(
}
if pli.item_type.is_local() {
let admission = if (user.max_connections > 0 || user.soft_connections > 0)
let playback_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id);
let (admission, _grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0)
&& app_state.app_config.config.load().user_access_control
{
app_state
.get_connection_admission(&user.username, user.max_connections, user.soft_connections)
.await
crate::api::api_utils::resolve_admission_with_strategies(
app_state,
&user.username,
user.max_connections,
user.soft_connections,
&fingerprint.client_ip,
fingerprint,
true,
Some(playback_session_token.as_str()),
)
.await
} else {
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: Some(crate::api::model::ConnectionKind::Normal),
}
(
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: Some(crate::api::model::ConnectionKind::Normal),
},
None,
)
};
let playback_session_token = create_session_fingerprint(fingerprint, &user.username, virtual_id);
return local_stream_response(
fingerprint,
app_state,
@@ -369,6 +380,7 @@ async fn xtream_player_api_stream(
user: &user,
session_reservation_ttl_secs: get_session_reservation_ttl_secs(app_state, item_type),
},
None,
)
.await
.into_response();
@@ -379,22 +391,28 @@ async fn xtream_player_api_stream(
pli.url.clone()
};
let connection_admission = if (user.max_connections > 0 || user.soft_connections > 0)
let (connection_admission, grace_mode) = if (user.max_connections > 0 || user.soft_connections > 0)
&& app_state.app_config.config.load().user_access_control
{
app_state
.get_connection_admission_for_session(
&user.username,
user.max_connections,
user.soft_connections,
&session_key,
)
.await
crate::api::api_utils::resolve_admission_with_strategies(
app_state,
&user.username,
user.max_connections,
user.soft_connections,
&fingerprint.client_ip,
fingerprint,
true,
Some(&session_key),
)
.await
} else {
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: user_session.as_ref().and_then(|session| session.connection_kind),
}
(
crate::api::model::ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: user_session.as_ref().and_then(|session| session.connection_kind),
},
None,
)
};
let connection_permission = connection_admission.permission;
let connection_kind = connection_admission
@@ -483,6 +501,7 @@ async fn xtream_player_api_stream(
connection_permission,
connection_kind,
allow_exhausted_shared_reconnect,
grace_mode,
)
.await
.into_response()
@@ -630,6 +649,7 @@ async fn xtream_player_api_stream_with_token(
UserConnectionPermission::Allowed,
crate::api::model::ConnectionKind::Normal,
false,
None,
)
.await
.into_response()
+414 -26
View File
@@ -489,7 +489,7 @@ impl ActiveUserManager {
removed_stream
}
pub async fn release_connection(&self, addr: &SocketAddr) -> ReleasedConnection {
async fn release_connection_inner(&self, addr: &SocketAddr, preserve_session_streams: bool) -> ReleasedConnection {
let (addr_removed, disconnected_user, removed_streams, expiry_entries, promotions) = {
let mut user_connections = self.connections.write().await;
@@ -501,9 +501,10 @@ impl ActiveUserManager {
if let Some(connection_data) = user_connections.by_key.get_mut(&username) {
let mut remaining_streams = Vec::with_capacity(connection_data.streams.len());
let mut released_kinds = Vec::new();
let mut removed_session_tokens = HashSet::new();
for mut stream_info in connection_data.streams.drain(..) {
if stream_info.addr == *addr {
if Self::should_preserve_session_stream(&stream_info) {
if preserve_session_streams && Self::should_preserve_session_stream(&stream_info) {
if let Some(entry) =
self.build_preserved_stream_expiry(&username, &stream_info, &connection_data.sessions)
{
@@ -522,6 +523,9 @@ impl ActiveUserManager {
released_kinds.push(kind);
}
connection_data.stream_normal_priorities.remove(&stream_info.uid);
if let Some(token) = stream_info.session_token.as_ref() {
removed_session_tokens.insert(token.clone());
}
removed_streams.push(stream_info);
}
} else {
@@ -529,6 +533,11 @@ impl ActiveUserManager {
}
}
connection_data.streams = remaining_streams;
if !preserve_session_streams && !removed_session_tokens.is_empty() {
connection_data
.sessions
.retain(|session| !removed_session_tokens.contains(&session.token));
}
for kind in released_kinds {
connection_data.decrement_kind(kind);
}
@@ -574,6 +583,14 @@ impl ActiveUserManager {
ReleasedConnection { addr_removed, removed_streams }
}
pub async fn release_connection(&self, addr: &SocketAddr) -> ReleasedConnection {
self.release_connection_inner(addr, true).await
}
pub async fn release_connection_as_kicked(&self, addr: &SocketAddr) -> ReleasedConnection {
self.release_connection_inner(addr, false).await
}
pub fn update_config(&self, config: &Config) {
let log_active_user = config.log.as_ref().is_some_and(|l| l.log_active_user);
let (grace_period_millis, grace_period_timeout_secs) = get_grace_options(config);
@@ -596,7 +613,6 @@ impl ActiveUserManager {
username: &str,
connection_data: &mut UserConnectionData,
) -> ConnectionAdmission {
let current_connections = connection_data.connections;
let selected_kind = decide_connection_kind(
connection_data.counts,
connection_data.max_connections,
@@ -604,9 +620,11 @@ impl ActiveUserManager {
);
if let Some(kind) = selected_kind {
// Reset grace period because the user is back under max_connections
connection_data.granted_grace = false;
connection_data.grace_ts = 0;
// Reset grace only once the user is back below the hard limit.
if connection_data.connections < connection_data.max_connections {
connection_data.granted_grace = false;
connection_data.grace_ts = 0;
}
return ConnectionAdmission {
permission: UserConnectionPermission::Allowed,
kind: Some(kind),
@@ -616,7 +634,7 @@ impl ActiveUserManager {
let now = get_current_timestamp();
// Check if user already used a grace period
if connection_data.granted_grace {
if current_connections > connection_data.max_connections
if connection_data.connections >= connection_data.max_connections
&& now - connection_data.grace_ts <= self.grace_period_timeout_secs.load(Ordering::Relaxed)
{
// Grace timeout, still active, deny connection
@@ -627,26 +645,12 @@ impl ActiveUserManager {
};
}
// Grace timeout expired, reset grace counters
connection_data.granted_grace = false;
connection_data.grace_ts = 0;
if connection_data.connections < connection_data.max_connections {
connection_data.granted_grace = false;
connection_data.grace_ts = 0;
}
}
if self.grace_period_millis.load(Ordering::Relaxed) > 0 && current_connections == connection_data.max_connections {
// Intentional asymmetry: grace is granted when current == max (AT limit), while
// Exhausted is returned when current > max (OVER limit after the grace window).
// This allows exactly one extra connection during the grace window — the new
// connection is accepted now, and a background check evicts it if the count is
// still over max after the grace period elapses.
connection_data.granted_grace = true;
connection_data.grace_ts = now;
debug!("Granted a grace period for user access: {username}");
return ConnectionAdmission {
permission: UserConnectionPermission::GracePeriod,
kind: Some(ConnectionKind::Normal),
};
}
// Too many connections, no grace allowed
debug!("User access denied, too many connections: {username}");
ConnectionAdmission {
permission: UserConnectionPermission::Exhausted,
@@ -745,6 +749,66 @@ impl ActiveUserManager {
.permission
}
pub(crate) async fn get_eviction_candidates(
&self,
username: &str,
client_ip: &str,
) -> Vec<crate::api::model::EvictionCandidate> {
let connections = self.connections.read().await;
let Some(connection_data) = connections.by_key.get(username) else {
return Vec::new();
};
let mut addr_counts = HashMap::new();
for stream in connection_data.streams.iter().filter(|stream| !stream.preserved) {
addr_counts
.entry(stream.addr)
.and_modify(|count: &mut u8| *count = count.saturating_add(1))
.or_insert(1_u8);
}
connection_data
.streams
.iter()
.filter(|stream| !stream.preserved && stream.client_ip == client_ip)
.filter(|stream| matches!(addr_counts.get(&stream.addr), Some(1)))
.map(|s| crate::api::model::EvictionCandidate {
addr: s.addr,
client_ip: s.client_ip.clone(),
ts: s.ts,
})
.collect()
}
pub(crate) async fn grant_grace(&self, username: &str) -> bool {
if self.grace_period_millis.load(Ordering::Relaxed) == 0 {
debug!("Grace grant denied, grace_period_millis is zero for {username}");
return false;
}
let mut connections = self.connections.write().await;
if let Some(connection_data) = connections.by_key.get_mut(username) {
let now = get_current_timestamp();
if connection_data.connections != connection_data.max_connections {
debug!(
"Grace grant denied for {username}, expected connections == max_connections but got {}/{}",
connection_data.connections,
connection_data.max_connections
);
return false;
}
if connection_data.granted_grace
&& connection_data.connections >= connection_data.max_connections
&& now - connection_data.grace_ts <= self.grace_period_timeout_secs.load(Ordering::Relaxed)
{
debug!("Grace grant denied, still within active grace timeout for {username}");
return false;
}
connection_data.granted_grace = true;
connection_data.grace_ts = now;
debug!("Granted a grace period for user access: {username}");
return true;
}
false
}
pub async fn active_users_and_connections(&self) -> (usize, usize) {
self.gc();
let user_connections = self.connections.read().await;
@@ -1529,6 +1593,13 @@ mod tests {
}
}
fn test_adaptive_channel(virtual_id: u32) -> StreamChannel {
StreamChannel {
item_type: PlaylistItemType::LiveHls,
..test_channel(virtual_id)
}
}
#[tokio::test]
async fn test_multi_session_same_addr_counts_and_releases_individually() {
let config = Config::default();
@@ -1563,7 +1634,7 @@ mod tests {
assert_eq!(manager.user_connections(username).await, 1);
assert_eq!(
manager.connection_permission(username, 1, 0).await,
UserConnectionPermission::GracePeriod
UserConnectionPermission::Exhausted
);
let second = manager
@@ -1593,6 +1664,323 @@ mod tests {
assert_eq!(manager.user_connections(username).await, 0);
}
#[tokio::test]
async fn test_grant_grace_only_at_limit() {
let config = Config::default();
let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
let event_manager = Arc::new(EventManager::new());
let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
let at_limit_addr: SocketAddr = "127.0.0.1:55011".parse().unwrap();
let at_limit_fingerprint = Fingerprint::new("fp-limit".to_string(), "127.0.0.1".to_string(), at_limit_addr);
let over_limit_addr: SocketAddr = "127.0.0.1:55012".parse().unwrap();
let over_limit_fingerprint = Fingerprint::new("fp-over".to_string(), "127.0.0.1".to_string(), over_limit_addr);
manager.add_connection(&at_limit_addr).await;
manager
.update_connection(ActiveUserConnectionParams {
uid: 10,
meter_uid: 0,
username: "at-limit",
max_connections: 1,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &at_limit_fingerprint,
provider: "provider-a",
stream_channel: &test_channel(1010),
user_agent: Cow::Borrowed("ua"),
session_token: Some("tok-limit"),
})
.await;
assert!(manager.grant_grace("at-limit").await);
manager.add_connection(&over_limit_addr).await;
manager
.update_connection(ActiveUserConnectionParams {
uid: 11,
meter_uid: 0,
username: "over-limit",
max_connections: 1,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &over_limit_fingerprint,
provider: "provider-a",
stream_channel: &test_channel(1011),
user_agent: Cow::Borrowed("ua"),
session_token: Some("tok-over-1"),
})
.await;
manager.add_connection(&"127.0.0.1:55013".parse().unwrap()).await;
let second_fingerprint =
Fingerprint::new("fp-over-2".to_string(), "127.0.0.1".to_string(), "127.0.0.1:55013".parse().unwrap());
manager
.update_connection(ActiveUserConnectionParams {
uid: 12,
meter_uid: 0,
username: "over-limit",
max_connections: 1,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &second_fingerprint,
provider: "provider-a",
stream_channel: &test_channel(1012),
user_agent: Cow::Borrowed("ua"),
session_token: Some("tok-over-2"),
})
.await;
assert!(!manager.grant_grace("over-limit").await);
}
#[tokio::test]
async fn eviction_candidates_ignore_ambiguous_socket_addrs() {
let config = Config::default();
let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
let event_manager = Arc::new(EventManager::new());
let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
let shared_addr: SocketAddr = "127.0.0.1:55031".parse().unwrap();
let unique_addr: SocketAddr = "127.0.0.1:55032".parse().unwrap();
let shared_fp = Fingerprint::new("fp-shared".to_string(), "127.0.0.1".to_string(), shared_addr);
let unique_fp = Fingerprint::new("fp-unique".to_string(), "127.0.0.1".to_string(), unique_addr);
manager.add_connection(&shared_addr).await;
manager.add_connection(&unique_addr).await;
for (uid, token, channel_id) in [(31, "tok-31", 1031), (32, "tok-32", 1032)] {
manager
.update_connection(ActiveUserConnectionParams {
uid,
meter_uid: 0,
username: "same-user",
max_connections: 3,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &shared_fp,
provider: "provider-a",
stream_channel: &test_channel(channel_id),
user_agent: Cow::Borrowed("ua"),
session_token: Some(token),
})
.await;
}
manager
.update_connection(ActiveUserConnectionParams {
uid: 33,
meter_uid: 0,
username: "same-user",
max_connections: 3,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &unique_fp,
provider: "provider-a",
stream_channel: &test_channel(1033),
user_agent: Cow::Borrowed("ua"),
session_token: Some("tok-33"),
})
.await;
let candidates = manager.get_eviction_candidates("same-user", "127.0.0.1").await;
assert_eq!(candidates.len(), 1);
assert_eq!(candidates[0].addr, unique_addr);
}
#[tokio::test]
async fn test_kicked_release_does_not_preserve_adaptive_stream() {
let config = Config::default();
let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
let event_manager = Arc::new(EventManager::new());
let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
let addr: SocketAddr = "127.0.0.1:55014".parse().unwrap();
let fingerprint = Fingerprint::new("fp-adaptive".to_string(), "127.0.0.1".to_string(), addr);
let mut user = ProxyUserCredentials::default();
user.username = String::from("user-adaptive");
user.max_connections = 1;
manager.add_connection(&addr).await;
manager
.create_user_session(CreateUserSessionParams {
user: &user,
session_token: "tok-adaptive",
virtual_id: 2014,
provider: "provider-a",
stream_url: "http://localhost/live.m3u8",
addr: &addr,
connection_permission: UserConnectionPermission::Allowed,
connection_kind: Some(ConnectionKind::Normal),
})
.await;
manager
.update_connection(ActiveUserConnectionParams {
uid: 14,
meter_uid: 0,
username: &user.username,
max_connections: 1,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &fingerprint,
provider: "provider-a",
stream_channel: &test_adaptive_channel(2014),
user_agent: Cow::Borrowed("ua"),
session_token: Some("tok-adaptive"),
})
.await;
let removed = manager.release_connection_as_kicked(&addr).await;
assert!(removed.addr_removed);
assert_eq!(removed.removed_streams.len(), 1);
assert!(manager.active_streams().await.is_empty());
}
#[tokio::test]
async fn test_kicked_release_invalidates_removed_session_tokens() {
let config = Config::default();
let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
let event_manager = Arc::new(EventManager::new());
let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
let kicked_addr: SocketAddr = "127.0.0.1:55015".parse().unwrap();
let survivor_addr: SocketAddr = "127.0.0.1:55016".parse().unwrap();
let kicked_fingerprint = Fingerprint::new("fp-kicked".to_string(), "127.0.0.1".to_string(), kicked_addr);
let survivor_fingerprint = Fingerprint::new("fp-survivor".to_string(), "127.0.0.1".to_string(), survivor_addr);
let mut user = ProxyUserCredentials::default();
user.username = String::from("kicked-user");
user.max_connections = 1;
manager.add_connection(&kicked_addr).await;
manager.add_connection(&survivor_addr).await;
manager
.create_user_session(CreateUserSessionParams {
user: &user,
session_token: "tok-kicked",
virtual_id: 2015,
provider: "provider-a",
stream_url: "http://localhost/live-1.ts",
addr: &kicked_addr,
connection_permission: UserConnectionPermission::Allowed,
connection_kind: Some(ConnectionKind::Normal),
})
.await;
manager
.create_user_session(CreateUserSessionParams {
user: &user,
session_token: "tok-survivor",
virtual_id: 2016,
provider: "provider-a",
stream_url: "http://localhost/live-2.ts",
addr: &survivor_addr,
connection_permission: UserConnectionPermission::Allowed,
connection_kind: Some(ConnectionKind::Normal),
})
.await;
manager
.update_connection(ActiveUserConnectionParams {
uid: 15,
meter_uid: 0,
username: &user.username,
max_connections: 1,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &kicked_fingerprint,
provider: "provider-a",
stream_channel: &test_channel(2015),
user_agent: Cow::Borrowed("ua"),
session_token: Some("tok-kicked"),
})
.await;
manager
.update_connection(ActiveUserConnectionParams {
uid: 16,
meter_uid: 0,
username: &user.username,
max_connections: 1,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &survivor_fingerprint,
provider: "provider-a",
stream_channel: &test_channel(2016),
user_agent: Cow::Borrowed("ua"),
session_token: Some("tok-survivor"),
})
.await;
let removed = manager.release_connection_as_kicked(&kicked_addr).await;
assert!(removed.addr_removed);
assert_eq!(removed.removed_streams.len(), 1);
assert_eq!(
manager
.connection_admission_for_session(&user.username, 1, 0, "tok-kicked")
.await
.permission,
UserConnectionPermission::Exhausted
);
assert_eq!(
manager
.connection_admission_for_session(&user.username, 1, 0, "tok-survivor")
.await
.permission,
UserConnectionPermission::Allowed
);
}
#[tokio::test]
async fn test_grace_at_limit_remains_active_until_connections_drop_below_limit() {
let config = Config::default();
let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
let event_manager = Arc::new(EventManager::new());
let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
let addr: SocketAddr = "127.0.0.1:55017".parse().unwrap();
let fingerprint = Fingerprint::new("fp-grace".to_string(), "127.0.0.1".to_string(), addr);
manager.add_connection(&addr).await;
manager
.update_connection(ActiveUserConnectionParams {
uid: 17,
meter_uid: 0,
username: "grace-at-limit",
max_connections: 1,
soft_connections: 0,
connection_kind: ConnectionKind::Normal,
priority: 0,
soft_priority: 0,
fingerprint: &fingerprint,
provider: "provider-a",
stream_channel: &test_channel(2017),
user_agent: Cow::Borrowed("ua"),
session_token: Some("tok-grace"),
})
.await;
assert!(manager.grant_grace("grace-at-limit").await);
assert_eq!(
manager.connection_admission("grace-at-limit", 1, 0).await.permission,
UserConnectionPermission::Exhausted
);
assert!(!manager.grant_grace("grace-at-limit").await);
}
#[tokio::test]
async fn test_same_session_token_on_new_addr_reuses_logical_connection() {
let config = Config::default();
+290
View File
@@ -0,0 +1,290 @@
use crate::model::AdmissionStrategy;
use log::debug;
use std::net::SocketAddr;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AdmissionDecision {
NoMatch,
Evict(EvictionTarget),
Grace(GraceMode),
#[cfg_attr(not(test), allow(dead_code))]
Deny,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GraceMode {
Instant,
Hold,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EvictionTarget {
pub addr: SocketAddr,
}
#[derive(Debug, Clone)]
pub struct StrategyContext<'a> {
pub username: &'a str,
pub client_ip: &'a str,
#[cfg_attr(not(test), allow(dead_code))]
pub strategies: &'a [AdmissionStrategy],
}
#[cfg_attr(not(test), allow(dead_code))]
pub fn evaluate_admission_strategies(
ctx: &StrategyContext<'_>,
candidates: &[EvictionCandidate],
) -> AdmissionDecision {
if ctx.strategies.is_empty() {
debug!("No admission strategies configured, denying request for user {}", ctx.username);
return AdmissionDecision::Deny;
}
debug!(
"Evaluating {} admission strategies for user {}",
ctx.strategies.len(),
ctx.username
);
for strategy in ctx.strategies {
let decision = evaluate_strategy(*strategy, ctx, candidates);
match decision {
AdmissionDecision::NoMatch => {}
AdmissionDecision::Evict(target) => {
debug!(
"Strategy {:?} selected eviction target {} for user {}",
strategy, target.addr, ctx.username
);
return AdmissionDecision::Evict(target);
}
AdmissionDecision::Grace(mode) => {
debug!(
"Strategy {:?} selected grace mode {:?} for user {} (blocking later strategies)",
strategy, mode, ctx.username
);
return AdmissionDecision::Grace(mode);
}
AdmissionDecision::Deny => {
debug!("Strategy {:?} denied request for user {}", strategy, ctx.username);
return AdmissionDecision::Deny;
}
}
}
debug!(
"No admission strategy matched for user {}, denying",
ctx.username
);
AdmissionDecision::Deny
}
pub(in crate::api) fn evaluate_strategy(
strategy: AdmissionStrategy,
ctx: &StrategyContext<'_>,
candidates: &[EvictionCandidate],
) -> AdmissionDecision {
match strategy {
AdmissionStrategy::EvictUserSameIpOldest => {
evaluate_evict_same_ip(ctx, candidates, EvictionOrder::Oldest)
}
AdmissionStrategy::EvictUserSameIpLatest => {
evaluate_evict_same_ip(ctx, candidates, EvictionOrder::Latest)
}
AdmissionStrategy::GraceInstantStream => AdmissionDecision::Grace(GraceMode::Instant),
AdmissionStrategy::GraceHoldStream => AdmissionDecision::Grace(GraceMode::Hold),
}
}
#[derive(Clone, Copy)]
enum EvictionOrder {
Oldest,
Latest,
}
fn evaluate_evict_same_ip(
ctx: &StrategyContext<'_>,
candidates: &[EvictionCandidate],
order: EvictionOrder,
) -> AdmissionDecision {
let mut selected: Option<&EvictionCandidate> = None;
for candidate in candidates {
if candidate.client_ip != ctx.client_ip {
continue;
}
let should_replace = match selected {
None => true,
Some(current) => match order {
EvictionOrder::Oldest => candidate.ts < current.ts,
EvictionOrder::Latest => candidate.ts > current.ts,
},
};
if should_replace {
selected = Some(candidate);
}
}
if selected.is_none() {
debug!(
"Strategy EvictUserSameIp{:?}: no same-IP candidates for user {}",
match order {
EvictionOrder::Oldest => "Oldest",
EvictionOrder::Latest => "Latest",
},
ctx.username
);
return AdmissionDecision::NoMatch;
}
match selected {
Some(candidate) => AdmissionDecision::Evict(EvictionTarget {
addr: candidate.addr,
}),
None => AdmissionDecision::NoMatch,
}
}
#[derive(Debug, Clone)]
pub struct EvictionCandidate {
pub addr: SocketAddr,
pub client_ip: String,
pub ts: u64,
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::{IpAddr, Ipv4Addr};
fn addr(port: u16) -> SocketAddr {
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), port)
}
fn candidate(port: u16, ip: &str, ts: u64) -> EvictionCandidate {
EvictionCandidate {
addr: addr(port),
client_ip: ip.to_string(),
ts,
}
}
#[test]
fn empty_strategy_list_returns_deny() {
let ctx = StrategyContext {
username: "user1",
client_ip: "1.1.1.1",
strategies: &[],
};
assert_eq!(evaluate_admission_strategies(&ctx, &[]), AdmissionDecision::Deny);
}
#[test]
fn evaluator_stops_at_first_matching_strategy() {
let ctx = StrategyContext {
username: "user1",
client_ip: "1.1.1.1",
strategies: &[
AdmissionStrategy::GraceInstantStream,
AdmissionStrategy::EvictUserSameIpOldest,
],
};
let candidates = vec![candidate(1000, "1.1.1.1", 100)];
let result = evaluate_admission_strategies(&ctx, &candidates);
assert_eq!(result, AdmissionDecision::Grace(GraceMode::Instant));
}
#[test]
fn grace_is_blocking() {
let ctx = StrategyContext {
username: "user1",
client_ip: "1.1.1.1",
strategies: &[
AdmissionStrategy::GraceHoldStream,
AdmissionStrategy::EvictUserSameIpOldest,
],
};
let candidates = vec![candidate(1000, "1.1.1.1", 100)];
let result = evaluate_admission_strategies(&ctx, &candidates);
assert_eq!(result, AdmissionDecision::Grace(GraceMode::Hold));
}
#[test]
fn evict_oldest_same_ip() {
let ctx = StrategyContext {
username: "user1",
client_ip: "1.1.1.1",
strategies: &[AdmissionStrategy::EvictUserSameIpOldest],
};
let candidates = vec![
candidate(1000, "1.1.1.1", 100),
candidate(1001, "1.1.1.1", 200),
candidate(1002, "2.2.2.2", 50),
];
let result = evaluate_admission_strategies(&ctx, &candidates);
match result {
AdmissionDecision::Evict(target) => assert_eq!(target.addr, addr(1000)),
other => panic!("Expected Evict, got {other:?}"),
}
}
#[test]
fn evict_latest_same_ip() {
let ctx = StrategyContext {
username: "user1",
client_ip: "1.1.1.1",
strategies: &[AdmissionStrategy::EvictUserSameIpLatest],
};
let candidates = vec![
candidate(1000, "1.1.1.1", 100),
candidate(1001, "1.1.1.1", 200),
candidate(1002, "2.2.2.2", 50),
];
let result = evaluate_admission_strategies(&ctx, &candidates);
match result {
AdmissionDecision::Evict(target) => assert_eq!(target.addr, addr(1001)),
other => panic!("Expected Evict, got {other:?}"),
}
}
#[test]
fn no_same_ip_candidate_returns_no_match_then_deny() {
let ctx = StrategyContext {
username: "user1",
client_ip: "1.1.1.1",
strategies: &[AdmissionStrategy::EvictUserSameIpOldest],
};
let candidates = vec![candidate(1000, "2.2.2.2", 100)];
let result = evaluate_admission_strategies(&ctx, &candidates);
assert_eq!(result, AdmissionDecision::Deny);
}
#[test]
fn eviction_falls_through_to_grace() {
let ctx = StrategyContext {
username: "user1",
client_ip: "1.1.1.1",
strategies: &[
AdmissionStrategy::EvictUserSameIpOldest,
AdmissionStrategy::GraceHoldStream,
],
};
let candidates = vec![candidate(1000, "2.2.2.2", 100)];
let result = evaluate_admission_strategies(&ctx, &candidates);
assert_eq!(result, AdmissionDecision::Grace(GraceMode::Hold));
}
#[test]
fn empty_candidates_with_eviction_then_grace() {
let ctx = StrategyContext {
username: "user1",
client_ip: "1.1.1.1",
strategies: &[
AdmissionStrategy::EvictUserSameIpLatest,
AdmissionStrategy::GraceInstantStream,
],
};
let result = evaluate_admission_strategies(&ctx, &[]);
assert_eq!(result, AdmissionDecision::Grace(GraceMode::Instant));
}
}
+22 -1
View File
@@ -139,7 +139,11 @@ async fn release_connection_parts(
reason: &DisconnectReason,
send_shared_stop_signal: bool,
) {
let removed = user_manager.release_connection(addr).await;
let removed = if matches!(reason, DisconnectReason::ClientKicked) {
user_manager.release_connection_as_kicked(addr).await
} else {
user_manager.release_connection(addr).await
};
for stream_info in &removed.removed_streams {
let (bytes_sent, first_byte_latency_ms) = event_manager.read_meter_qos(stream_info.meter_uid).await;
event_manager.unregister_meter_client(stream_info.uid).await;
@@ -675,6 +679,7 @@ impl ConnectionManager {
}
pub async fn release_connection_as_kicked(&self, addr: &SocketAddr) {
let _ = self.close_connection_with_reason(addr, DisconnectReason::ClientKicked);
release_connection_with_reason(self, addr, &DisconnectReason::ClientKicked, true).await;
}
@@ -1023,6 +1028,22 @@ mod tests {
);
}
#[tokio::test]
async fn release_connection_as_kicked_sends_kick_close_signal() {
let manager = create_test_connection_manager();
let mut rx = manager.get_close_connection_channel();
let addr: SocketAddr = "127.0.0.1:2234".parse().unwrap_or_else(|_| unreachable!());
manager.release_connection_as_kicked(&addr).await;
assert_eq!(
tokio::time::timeout(Duration::from_millis(100), rx.recv())
.await
.ok()
.and_then(Result::ok),
Some(CloseConnectionSignal::WithReason(addr, DisconnectReason::ClientKicked))
);
}
#[tokio::test]
async fn close_connection_signal_sends_generic_close_signal() {
let manager = create_test_connection_manager();
+4
View File
@@ -1,5 +1,6 @@
mod active_provider_manager;
mod active_user_manager;
mod admission_strategy;
mod app_state;
mod connection_manager;
mod download;
@@ -28,5 +29,8 @@ pub(in crate::api) use self::{
active_user_manager::*, download::*, model_utils::*, provider_config::*, qos_aggregation_manager::*,
recording_worker::*, request::*, stream_error::*, xtream::*,
};
pub(in crate::api) use self::admission_strategy::{
AdmissionDecision, EvictionCandidate, GraceMode, StrategyContext, evaluate_strategy,
};
mod batch_result_collector;
pub use self::batch_result_collector::*;
@@ -16,7 +16,7 @@ use crate::{
use axum::http::{header::USER_AGENT, HeaderMap};
use bytes::Bytes;
use futures::{task::AtomicWaker, Future, Stream, StreamExt};
use log::{error, info, trace};
use log::{error, info};
use shared::{
model::{StreamChannel, UserConnectionPermission, VirtualId},
utils::sanitize_sensitive_info,
@@ -932,8 +932,6 @@ fn stream_grace_period(request: GracePeriodParams<'_>) -> (Option<Arc<AtomicU8>>
None
};
trace!("grace hold stream {hold_stream}");
if provider_grace_check.is_some() || user_grace_check.is_some() {
let stream_strategy_flag = Arc::new(AtomicU8::new(
if hold_stream { StreamMode::GracePending as u8 } else { StreamMode::Inner as u8 },
+101 -1
View File
@@ -1,8 +1,48 @@
use shared::model::{StreamBufferConfigDto, StreamConfigDto};
use shared::model::{AdmissionStrategyDto, StreamBufferConfigDto, StreamConfigDto};
use shared::utils::parse_to_kbps;
use crate::api::model::TransportStreamBuffer;
use crate::model::macros;
#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
pub enum AdmissionStrategy {
EvictUserSameIpOldest,
EvictUserSameIpLatest,
GraceInstantStream,
GraceHoldStream,
}
impl From<&AdmissionStrategyDto> for AdmissionStrategy {
fn from(dto: &AdmissionStrategyDto) -> Self {
match dto {
AdmissionStrategyDto::EvictUserSameIpOldest => Self::EvictUserSameIpOldest,
AdmissionStrategyDto::EvictUserSameIpLatest => Self::EvictUserSameIpLatest,
AdmissionStrategyDto::GraceInstantStream => Self::GraceInstantStream,
AdmissionStrategyDto::GraceHoldStream => Self::GraceHoldStream,
}
}
}
impl From<&AdmissionStrategy> for AdmissionStrategyDto {
fn from(domain: &AdmissionStrategy) -> Self {
match domain {
AdmissionStrategy::EvictUserSameIpOldest => Self::EvictUserSameIpOldest,
AdmissionStrategy::EvictUserSameIpLatest => Self::EvictUserSameIpLatest,
AdmissionStrategy::GraceInstantStream => Self::GraceInstantStream,
AdmissionStrategy::GraceHoldStream => Self::GraceHoldStream,
}
}
}
impl AdmissionStrategy {
pub fn is_grace(&self) -> bool {
matches!(self, Self::GraceInstantStream | Self::GraceHoldStream)
}
pub fn is_grace_hold(&self) -> bool {
matches!(self, Self::GraceHoldStream)
}
}
#[derive(Debug, Clone)]
pub struct StreamBufferConfig {
pub enabled: bool,
@@ -41,6 +81,7 @@ pub struct StreamConfig {
pub throttle_str: Option<String>,
pub throttle_kbps: u64,
pub shared_burst_buffer_mb: u64,
pub admission_strategies: Option<Vec<AdmissionStrategy>>,
}
macros::from_impl!(StreamConfig);
@@ -58,6 +99,10 @@ impl From<&StreamConfigDto> for StreamConfig {
throttle_str: dto.throttle.clone(),
throttle_kbps: dto.throttle.as_ref().map_or(0u64, |throttle| parse_to_kbps(throttle).unwrap_or(0u64)),
shared_burst_buffer_mb: dto.shared_burst_buffer_mb,
admission_strategies: dto
.admission_strategies
.as_ref()
.map(|entries| entries.iter().map(AdmissionStrategy::from).collect()),
}
}
}
@@ -76,10 +121,65 @@ impl From<&StreamConfig> for StreamConfigDto {
throttle: instance.throttle_str.clone(),
throttle_kbps: instance.throttle_kbps,
shared_burst_buffer_mb: instance.shared_burst_buffer_mb,
admission_strategies: instance
.admission_strategies
.as_ref()
.map(|entries| entries.iter().map(AdmissionStrategyDto::from).collect()),
}
}
}
#[cfg(test)]
mod tests {
use super::{AdmissionStrategy, StreamConfig};
use shared::model::{AdmissionStrategyDto, StreamConfigDto};
#[test]
fn stream_config_preserves_missing_admission_strategies() {
let domain = StreamConfig::from(&StreamConfigDto::default());
assert_eq!(domain.admission_strategies, None);
}
#[test]
fn stream_config_preserves_explicit_empty_admission_strategies() {
let domain = StreamConfig::from(&StreamConfigDto {
admission_strategies: Some(vec![]),
..StreamConfigDto::default()
});
assert_eq!(domain.admission_strategies, Some(vec![]));
}
#[test]
fn stream_config_roundtrips_admission_strategies() {
let domain = StreamConfig {
retry: true,
metrics_enabled: true,
buffer: None,
grace_period_millis: 2_000,
grace_period_timeout_secs: 10,
grace_period_hold_stream: true,
hls_session_ttl_secs: 5,
catchup_session_ttl_secs: 5,
throttle_str: None,
throttle_kbps: 0,
shared_burst_buffer_mb: 1,
admission_strategies: Some(vec![
AdmissionStrategy::EvictUserSameIpOldest,
AdmissionStrategy::GraceHoldStream,
]),
};
let dto = StreamConfigDto::from(&domain);
assert_eq!(
dto.admission_strategies,
Some(vec![
AdmissionStrategyDto::EvictUserSameIpOldest,
AdmissionStrategyDto::GraceHoldStream,
])
);
}
}
#[derive(Debug, Clone)]
pub struct CustomStreamResponse {
pub channel_unavailable: Option<TransportStreamBuffer>,
+3 -3
View File
@@ -13,12 +13,12 @@ FROM mwader/static-ffmpeg:7.1 AS ffmpeg-static
# -----------------------------------------------------------------
# Stage 1: Build the Rust binary for production
# -----------------------------------------------------------------
FROM rust:bookworm AS rust-build
FROM rust@sha256:e8e2bb5ff27ad3b369a4f667392464e6ec399cfe81c1230ae78edb1036b9bd74 AS rust-build
# Get target architecture from build argument
ARG RUST_TARGET=x86_64-unknown-linux-musl
# Install build dependencies for Debian (bookworm)
# Install build dependencies for Debian (trixie)
RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
pkg-config libssl-dev musl-tools && \
apt-get clean && \
@@ -51,7 +51,7 @@ RUN cargo build -p tuliprox --target ${RUST_TARGET} --release
# -----------------------------------------------------------------
# Stage 2: Build the rust frontend
# -----------------------------------------------------------------
FROM rust:bookworm AS trunk-build
FROM rust@sha256:e8e2bb5ff27ad3b369a4f667392464e6ec399cfe81c1230ae78edb1036b9bd74 AS trunk-build
# Install dependencies for Trunk and WebAssembly
RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
+4
View File
@@ -8,6 +8,10 @@
- [Configuration & Setup (Overview)](./configuration/overview.md)
- [config.yml (Core System)](./configuration/config.md)
- [Streaming & Proxy Behavior](./configuration/reverse-proxy.md)
- [Connection Handling](./configuration/connection-handling.md)
- [Priorities, Soft Connections and Preemption](./configuration/connection-handling-priorities-and-preemption.md)
- [Sessions, HLS, Catchup and Reconnects](./configuration/connection-handling-sessions-and-reconnects.md)
- [Failures and User-Visible Behavior](./configuration/connection-handling-failures-and-user-visible-behavior.md)
- [Metadata Updater & FFprobe](./configuration/metadata-update.md)
- [Local Media Library](./configuration/local-library.md)
- [source.yml (Inputs, Panel API & Targets)](./configuration/source.md)
@@ -0,0 +1,250 @@
# Connection Handling: Failures and User-Visible Behavior
This page explains what operators and end users will see when connection handling rules take effect.
The most important point:
Not every "problem" is a defect. Very often Tuliprox is reacting exactly as configured:
- a stream is intentionally rejected
- a soft connection is intentionally preempted
- a shared stream is intentionally terminated as a whole
- a request intentionally waits in a short grace phase
## The most common visible outcomes
### 1. The user gets "too many connections"
This usually means:
- from the user-rule perspective, no further legal stream can be admitted
Possible causes:
- `max_connections` is reached
- `soft_connections` are not available or already consumed
- admission strategies could not help
- an earlier grace is still active
- a session is still counted
What to check:
- `max_connections`
- `soft_connections`
- `admission_strategies`
- session TTLs
- grace settings
### 2. The user gets provider-exhausted behavior
This means:
- the user might still be allowed to stream on the user side
- but the provider currently has no free slot left
Possible causes:
- real provider exhaustion
- no reusable shared stream was available
- no valid preemption target existed
- equal priority prevented preemption
What to check:
- provider capacity
- priorities of the affected users
- whether shared streams are in use
- whether a higher-priority user currently has precedence
### 3. The user sees `low_priority_preempted`
This means:
- the stream started legally
- but later a more important request arrived
- and the current stream became the valid victim
Important:
- this is not a random interruption
- it is an intentional result of the priority model
If a shared stream was affected, several viewers may see the same result at once.
### 4. A stream starts with a short delay
This is often not a bug, but one of these behaviors:
- `GraceHoldStream`
- provider-side grace
- intentional holding so Tuliprox does not make a wrong decision too early
In plain language:
- Tuliprox is waiting briefly to see whether the transition resolves cleanly
### 5. A stream starts first and is terminated later
This can happen when:
- a grace phase allows the start temporarily
- but the situation does not become legal in time
Then the behavior is:
- the start was provisionally allowed
- the later final decision was negative
### 6. A kick feels "harder" than a normal disconnect
This is intentional.
With a kick:
- the stream ends
- the related session is also invalidated
That prevents the same session from immediately undoing the kick.
## Typical operator scenarios
### Scenario A: "The user says they only switched channels"
Likely causes:
- the old and new stream briefly overlapped
- grace was too short or not well suited
- the provider still looked full during the switch moment
Check:
- `grace_period_millis`
- `grace_period_hold_stream`
- priority
- provider limits
### Scenario B: "The user says they only seeked"
Likely causes:
- the seek created overlapping requests
- the session window was too short
- the user limit was very tight
Check:
- `hls_session_ttl_secs`
- `catchup_session_ttl_secs`
- grace configuration
### Scenario C: "Two users watched the same channel and then both dropped at once"
Likely cause:
- they were attached to a shared stream
- that shared stream was preempted as a whole
Check:
- the priority of the later requester
- whether the shared stream only had low-priority viewers
- whether `low_priority_preempted` fallback content is configured
### Scenario D: "The same group of users sometimes switches successfully and sometimes not"
Likely causes:
- sometimes a shared stream is reused and sometimes not
- sometimes a slot is freed in time and sometimes not
- equal priority blocks preemption
- the session is sometimes still valid and sometimes already expired
Check:
- whether it was truly the same channel or a different source
- session TTLs
- provider load
- shared-stream usage
## How to classify failures correctly
### First ask: user side or provider side?
User side:
- user limits
- soft connections
- admission strategies
- grace
- sessions
Provider side:
- no free slots
- priorities
- shared streams
- preemption
Many incidents become easier to understand once you separate those two layers.
### Then ask: immediate or delayed?
Immediate:
- admission or provider allocation was not possible right away
Delayed:
- grace expired
- a higher-priority request arrived later
- a shared stream was preempted later
## What operators often misread
### "If something ended later, it must have been a network issue"
Not necessarily. It can just as easily have been a later priority or grace decision.
### "If several users are affected at once, Tuliprox must be unstable"
Not necessarily. During shared-stream preemption that is exactly the expected behavior.
### "If a user was allowed to start, they should never be stopped later"
Not necessarily. Soft connections and low-priority streams can be sacrificed later.
## Operational recommendations
### If you want as few complaints as possible
- use priorities deliberately and sparingly
- use soft connections only where preemption is acceptable
- tune session TTLs for the actual player types in use
- do not set grace too aggressively low
### If you want hard precedence rules
- give critical users clearly stronger priority values
- do not rely on "it will probably distribute fairly"
### If you rely heavily on shared streams
- document internally that one shared upstream connection can affect all attached viewers together
## Quick incident checklist
When an operator wants to classify an incident, these questions are often enough:
1. Was it the same session or a truly new start?
2. Was the user already at the normal limit?
3. Was any soft capacity still available?
4. Was the provider full?
5. Were multiple viewers attached to a shared stream?
6. Did someone with higher priority arrive later?
7. Was a grace phase active?
## Further reading
- [Connection Handling Overview](./connection-handling.md)
- [Priorities, Soft Connections and Preemption](./connection-handling-priorities-and-preemption.md)
- [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md)
@@ -0,0 +1,275 @@
# Connection Handling: Priorities, Soft Connections and Preemption
This page explains how Tuliprox decides which stream may stay alive when no provider capacity is left.
This is usually the hardest part for operators to reason about because several rules interact here:
- normal connections
- soft connections
- priorities
- shared streams
- real provider limits
## The most important rule first
Tuliprox does not preempt at random. It follows fixed precedence rules.
The general idea is:
- important streams should remain stable
- less important streams may be sacrificed
- shared streams save capacity, but they are not automatically protected
## How priority is read
In Tuliprox:
- lower numbers mean higher priority
Examples:
- `-10` is stronger than `0`
- `0` is stronger than `5`
- `5` is stronger than `20`
Many operators intuitively expect the opposite. In Tuliprox, the smaller number is the stronger priority.
## Two types of connections
### Normal connection
This is a stream within the user's normal limit.
Normal connections are the stronger class.
### Soft connection
This is an extra stream above `max_connections` when `soft_connections` are enabled.
Soft connections are the weaker class. They are intentionally easier to preempt.
## Who may preempt whom
### Normal against soft
A normal request may preempt an existing soft connection.
It does not need a better numeric priority value for that. The difference between `Normal` and `Soft` is already enough.
In plain language:
- a real normal slot is always stronger than a bonus slot
### Soft against normal
A soft request may not preempt a normal connection.
In plain language:
- a bonus slot may not push out a real standard slot
### Normal against normal
Here priority decides.
Rule:
- only if the new request truly has higher priority
- equal priority does not preempt
### Soft against soft
Here priority also decides.
Rule:
- only a soft request with higher priority may preempt another soft connection
## What happens with equal priority
Equal priority does not preempt.
This is intentional so the behavior remains stable and predictable.
Example:
- User A has priority `0`
- User B has priority `0`
- The provider is full
- User B starts a new stream
Result:
- User B may not simply throw out A
Instead, Tuliprox needs:
- a free slot
- a shared-stream reuse case
- a grace-based solution
- or the request is rejected
## What happens in an exact tie
If several possible victims share the same weak priority, Tuliprox chooses the oldest victim first.
In plain language:
- the longest-running stream in that weakest group is removed first
This avoids unpredictable switching between candidates.
## How shared streams fit into these rules
A shared stream is only one upstream connection from the provider's perspective.
That means:
- 2, 5, or 20 viewers may be attached to the same live stream
- on the provider side this still consumes just one shared slot
## Which priority a shared stream has
A shared stream does not simply keep "the priority of the first viewer".
Instead, Tuliprox derives an effective shared priority from the active viewers.
Simplified:
- within the currently effective connection class, Tuliprox uses the strongest remaining subscriber priority
- if at least one subscriber is `Normal`, the shared stream is treated as `Normal`
- only if all subscribers are `Soft`, the shared stream is treated as `Soft`
What this means for operators:
- a later high-priority subscriber can make the shared stream harder to preempt
- if that important subscriber leaves, the shared stream can become weaker again
## Can a shared stream be preempted completely?
Yes.
This is important:
- if no provider capacity is free
- and a shared stream is the best valid victim
- Tuliprox does not remove only one viewer
- it terminates the shared provider connection as a whole
Consequence:
- all viewers on that shared stream lose their upstream at the same time
If a `low_priority_preempted` fallback output is configured, those viewers may see the fallback content instead of a hard disconnect.
## Example: several low-priority users share the same stream
Situation:
- several users with low priority share channel A
- an important user wants channel B
- no free provider connection is available
Result:
- if channel A and channel B compete for the same provider lineup
- and the new user really has higher priority
- the shared stream for A may be preempted as a whole
Then the following happens:
- the shared provider connection is released
- the new important stream receives the slot
- all viewers on the old shared stream are affected at once
## Example: an important user joins the same already shared channel
Situation:
- low-priority users are already sharing channel A
- an important user also starts channel A
Result:
- in this case Tuliprox often does not need a new provider connection
- the user can join the existing shared stream
- therefore no preemption may be necessary at all
## Provider grace during switching
There is not only user grace, but also provider-side transition handling.
This matters in situations such as:
- a quick channel switch
- the old stream is just about to close
- the new stream has equal importance
- the provider appears full for a very short moment
In those cases Tuliprox does not always reject immediately. It may wait briefly to see whether the old slot is released in time.
This is especially important so equal-priority switch operations do not fail unnecessarily.
## What soft connections are good for
Soft connections are useful when you want to give operators or premium users more flexibility without weakening the overall protection model too much.
They are useful for:
- short-term overbooking
- comfort during quiet periods
- bonus slots for users who do not need a hard guarantee
They are not suitable for:
- hard guarantees
- critical primary streams
- user groups that must never be preempted
## Practical operator rules
### If important users should remain stable
- give them a smaller priority number
- do not rely on soft connections for critical streams
### If many equal users should be treated fairly
- give them the same priority
- then they do not preempt each other
### If shared streams are used heavily
- remember that shared streams save provider slots
- but they can still be preempted as a whole in a conflict
### If soft connections trigger complaints
That is often not a bug, but the intended design:
- soft is supposed to help
- but soft is not supposed to be as strong as normal
## Common misunderstandings
### "More viewers on a shared stream should make it safer"
No. The decisive factors are not the number of viewers, but:
- the connection class
- the effective priority
- provider scarcity
### "Equal priority means the newer request wins"
No. Equal priority does not preempt.
### "Soft is just another fully normal slot"
No. Soft is intentionally weak and remains preemptable.
## Next step
If you want to understand why many players create multiple requests for what users perceive as a single stream, read:
- [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md)
@@ -0,0 +1,246 @@
# Connection Handling: Sessions, HLS, Catchup and Reconnects
This page explains why Tuliprox does not treat every request as a completely new stream and why sessions are so important for streaming stability.
This matters for operators because many seemingly strange connection patterns are actually normal player behavior.
## Why sessions are needed at all
Many operators initially assume this model:
- a user starts one stream
- then one connection stays open until the end
In practice this is often not how players behave.
Many players behave more like this:
- several short follow-up requests
- segment changes
- reconnects
- brief overlap during seeks
- channel switching with an overlap window
Without session logic, Tuliprox would often count these operations incorrectly as "yet another new stream".
## What a session means in Tuliprox
A session is the logical playback identity of a user for one stream activity.
It helps Tuliprox decide:
- is this the same playback?
- should Tuliprox try to stay on the same provider account?
- may this reconnect continue the same logical connection?
- or should this be treated as a completely new start?
## HLS and why it is especially noticeable
HLS is not "one file, one connection".
Typical behavior:
- the player downloads a playlist
- then continuously fetches many small segments
- there are short idle gaps in between
- channel switches or quality changes may trigger new requests
For operators this can look like many connections in the logs. Tuliprox therefore tries to treat these follow-up requests as one logical session.
## Catchup and why it can look even more unstable
Catchup and archive playback often produce:
- jumps
- backward and forward moves
- repeated requests after timeline changes
That is why catchup usually needs a longer session TTL than HLS.
## What the session TTL does
Tuliprox keeps session information for a limited time so that a short interruption does not immediately look like a completely new stream.
Important values in the reverse proxy block:
- `hls_session_ttl_secs`
- `catchup_session_ttl_secs`
In plain language:
- HLS gets a short recognition window
- catchup usually gets a somewhat longer one
## What the session TTL does not mean
The TTL does not automatically mean:
- that a real provider stream always stays open the entire time
Instead, the TTL is about recognition, continuity, and provider affinity.
## Provider affinity for sessions
When a session continues, Tuliprox tries to stay on the same provider context whenever possible.
This matters because some upstream services react badly if one logical playback keeps jumping between different provider accounts or slots.
For operators this means:
- session logic is not only a comfort feature
- it also protects against unstable provider-side behavior
## What happens during a quick channel switch
During a channel switch, two things can be true at the same time:
- the old stream is not fully released yet
- the new stream is already being requested
To prevent unnecessary failures, Tuliprox uses transition tools such as grace and session reuse.
The goal is:
- ordinary switching should work
- real parallel viewing should still remain controlled
## What happens during a seek
Seek is one of the most common reasons for apparently duplicated connections.
Why?
- the player requests the new position already
- the old connection is often not fully closed yet
- for a brief moment both sides exist at the same time
Without grace and session logic, this would often end as:
- "too many connections"
- or "provider is full"
even though the user only performed a normal seek.
## User grace and session behavior
Tuliprox uses user grace as a short transition mechanism when a new request would otherwise be blocked by the user limit for a short moment.
Important for operators:
- grace is not a permanent extra entitlement
- grace is meant to bridge short transition moments
- grace is only evaluated after normal and soft admission are no longer sufficient
There are two broad modes:
- start immediately and verify later
- hold the start briefly and only continue once the decision is clear
Which one is better depends on the player and the behavior you want.
## Provider grace and session switches
Besides the user side, there is also a provider-side grace view.
This is especially important during:
- equal-priority switching
- moments where provider slots are still briefly occupied
Here Tuliprox tries not to turn the new request into an immediate failure if the old slot is likely to be released in time.
## Why HLS and catchup are still not "free"
Session reuse does not mean that HLS or catchup are ignored indefinitely.
If Tuliprox recognizes that this is genuinely a new situation, the request is counted as a new activity again.
Examples:
- a different channel
- a different session
- the session has expired
- a kick or another hard termination happened
## What happens on a kick
When a connection is deliberately kicked, not only the current stream ends.
Important for operators:
- the related session is also invalidated
- the same session should not immediately be accepted as "still valid"
This prevents a deliberately terminated stream from slipping right back through via the same session identity.
## What happens on a normal disconnect
A normal disconnect is not automatically a fault.
Depending on timing, Tuliprox may:
- keep the session briefly
- allow a resume
- or discard the session once the window expires
## What matters when shared streams and sessions meet
Shared streams and sessions solve different problems:
- shared streams save provider slots
- sessions help with recognition and continuity
So a user can simultaneously:
- be part of a shared stream
- and still rely on session logic for HLS or catchup continuity
## Common operator questions
### "Why do I see many requests even though the user only watches one stream?"
Usually because of:
- HLS segments
- catchup jumps
- reconnects
- normal player behavior
### "Why does a session remain for a short time after the stream is gone?"
So that brief reconnects are not punished as a completely new playback.
### "Why was the same session no longer accepted after a kick?"
Because a kick is intentionally supposed to invalidate the session side as well.
## Recommendations for operators
### If HLS users often have trouble while switching
Check:
- `hls_session_ttl_secs`
- `grace_period_millis`
- `grace_period_hold_stream`
### If catchup users often fail during timeline jumps
Check:
- `catchup_session_ttl_secs`
- whether the player performs very aggressive seeks
### If users complain about "too many connections" even though they are only seeking
Check:
- admission strategies
- grace configuration
- session TTLs
## Next step
If you want to understand which user-visible symptoms result from these rules, continue with:
- [Failures and User-Visible Behavior](./connection-handling-failures-and-user-visible-behavior.md)
@@ -0,0 +1,266 @@
# Connection Handling
This handbook explains how Tuliprox decides whether a stream is started, reused, briefly tolerated, preempted, or rejected.
It is written for operators rather than developers. The goal is to explain:
- what Tuliprox does in everyday operation
- why a user can sometimes start streaming immediately and sometimes cannot
- which rules apply to priorities, sessions, shared streams, and provider limits
- which visible effects operators and end users will notice
If you want to go deeper into individual topics, continue with:
- [Priorities, Soft Connections and Preemption](./connection-handling-priorities-and-preemption.md)
- [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md)
- [Failures and User-Visible Behavior](./connection-handling-failures-and-user-visible-behavior.md)
## Why this matters
Every stream request in Tuliprox is evaluated against several limits at the same time:
- user connection limits
- soft connections
- provider capacity
- session reuse
- shared streams
- priorities
- grace behavior during switching and seeking
Because of that, a stream can fail or be terminated later for very different reasons. Without a clear model,
the behavior can look random. In reality it follows fixed rules.
## The most important terms in plain language
### User limit
This is the normal number of simultaneous streams a user is allowed to have.
Example:
- `max_connections: 2`
- The user may normally watch 2 streams in parallel.
### Soft connection
A soft connection is an extra stream above the normal user limit.
Important:
- It is a bonus, not a guaranteed right.
- It can be preempted more easily than a normal connection.
### Priority
Priority decides which stream is protected when there is a conflict.
Important:
- lower numbers mean higher priority
- `0` is stronger than `5`
- `-10` is stronger than `0`
### Provider capacity
Tuliprox does not only enforce user limits. It also respects the actual free slots available from the upstream provider.
That means:
- a user may still be allowed to open another stream on the user side
- but the request can still be blocked
- if the provider currently has no capacity left
### Shared stream
Multiple viewers can reuse the same running live stream.
In that case:
- for the provider this is only one shared upstream connection
- for viewers, more than one client is attached to the same source
### Session
A session is Tuliprox's way of recognizing that several requests belong to the same logical playback.
This matters especially for:
- HLS
- catchup
- quick reconnects
- channel switches with a brief overlap
### Grace
Grace is a short, controlled transition window.
It exists so a new request does not fail immediately just because:
- the old stream is still shutting down
- a player sends overlapping requests during a seek
- a provider slot has not been released yet during a switch
## The simple overall flow
When a user starts a stream, Tuliprox roughly thinks in this order:
1. Is this an existing session or a truly new start?
2. Does the user still have a free normal connection?
3. If not, is a soft connection still available?
4. If not, do configured admission strategies or grace rules apply?
5. If provider capacity is missing, can a shared stream be reused or can a lower-priority connection be preempted?
6. If no legal solution exists, the request is rejected.
This is the short version. In practice, user rules and provider rules run in parallel:
- user side: "Is this user allowed to have another stream?"
- provider side: "Is there actually a free upstream connection for it?"
Both sides must succeed.
## What is checked first
### 1. Same session or new stream?
Tuliprox first tries to recognize whether this is the same logical playback.
This matters because many players do not use "one long connection", but many short follow-up requests.
Examples:
- HLS constantly fetches new segments
- catchup performs frequent jumps
- some players briefly overlap old and new requests during seeks or channel switches
If Tuliprox recognizes the same playback, not every follow-up request is treated like a completely new stream.
### 2. Normal connection
If the user is still below the normal limit, the stream is admitted as a normal connection.
This is the preferred case.
### 3. Soft connection
If the normal user limit is already reached, Tuliprox checks whether a soft connection is still allowed.
Soft connections are intentionally weaker. They help with short-term overbooking in quiet periods, but they are the first candidates for later preemption.
### 4. Admission strategies and user grace
If neither a normal nor a soft connection is possible, Tuliprox can still evaluate configured admission strategies.
Examples:
- `GraceInstantStream`
- `GraceHoldStream`
- `EvictUserSameIpOldest`
- `EvictUserSameIpLatest`
These rules are only evaluated after ordinary user admission is already exhausted.
### 5. Provider side
Even if the user is still allowed to stream, the upstream provider must also have room.
Tuliprox then checks:
- is there still a free provider slot?
- can an already running shared stream be reused?
- may a lower-priority connection be preempted?
- can a short provider grace transition help?
## The two major protection layers
### User rules
These rules protect the operator-defined usage limits:
- normal connections
- soft connections
- sessions
- admission strategies
### Provider rules
These rules protect the real upstream:
- free provider slots
- shared streams
- priorities
- preemption
- provider grace during switch moments
A request must pass both layers.
## Typical everyday situations
### Case 1: A user simply starts another channel
- If there is still room in both the user limit and provider capacity, the stream starts normally.
### Case 2: A user reaches the limit
- If soft connections are allowed, a soft connection may still be created.
- If that is no longer possible either, only admission strategies or rejection remain.
### Case 3: A user performs a seek
- The player may briefly request the old and the new position in parallel.
- Tuliprox tries not to treat this as abuse immediately.
- This is where grace and session recognition matter.
### Case 4: Many users share the same live stream
- On the provider side this counts as one shared upstream connection.
- If preemption happens later, this shared connection can be affected as a whole.
### Case 5: An important user needs a slot
- If no provider capacity is free, Tuliprox may preempt lower-priority streams.
- This also applies to shared streams.
## Important operator rules
### Equal priority does not preempt
If two requests are equally important, Tuliprox does not simply sacrifice the existing stream.
### Higher priority wins
If a genuinely more important request arrives while no provider capacity is free, a less important stream may be preempted.
### Soft is weaker than normal
Normal connections are protected more strongly than soft connections.
### Shared streams are not a magic shield
A shared stream saves provider slots, but it is not automatically untouchable.
If it is the best valid victim, it can be terminated completely.
### A kick ends both the stream and its session
When a connection is deliberately kicked, Tuliprox should not immediately treat the same session as still valid.
## Operator checklist
If connection handling looks "strange" for your users, check these first:
- How high is `max_connections`?
- Are `soft_connections` configured?
- What `priority` does the affected user have?
- Are `admission_strategies` configured?
- Is `grace_period_millis` set sensibly?
- Are `hls_session_ttl_secs` and `catchup_session_ttl_secs` appropriate for the player in use?
- Is `share_live_streams` enabled and are multiple users watching the same channel?
- Is the actual provider already at capacity?
## Recommended reading order
If you want to fully understand the system, read these next:
1. [Priorities, Soft Connections and Preemption](./connection-handling-priorities-and-preemption.md)
2. [Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md)
3. [Failures and User-Visible Behavior](./connection-handling-failures-and-user-visible-behavior.md)
+20
View File
@@ -13,6 +13,26 @@ It also contains the optional telemetry pipeline for stream reliability analysis
* `qos_aggregation` periodically condenses those raw events into compact QoS snapshots.
* The Web UI can then inspect both the raw history view and the aggregated QoS summaries.
## Read This First: Connection Handling Handbook
The `reverse_proxy.stream` block now interacts with a broader connection-handling model:
* user limits
* soft connections
* priorities
* admission strategies
* grace periods
* HLS and catchup sessions
* shared streams
* provider-side preemption
If you want to understand why a stream starts, waits, gets reused, gets preempted, or is rejected, read these pages first:
* [Connection Handling](./connection-handling.md)
* [Connection Handling: Priorities, Soft Connections and Preemption](./connection-handling-priorities-and-preemption.md)
* [Connection Handling: Sessions, HLS, Catchup and Reconnects](./connection-handling-sessions-and-reconnects.md)
* [Connection Handling: Failures and User-Visible Behavior](./connection-handling-failures-and-user-visible-behavior.md)
## Top-level entries
```yaml
+7 -1
View File
@@ -435,7 +435,7 @@
"INPUT_TYPE": "Defines the streaming protocol or source format (e.g. xtream, m3u).",
"LIVE_SOURCE": "Source selector for the Live cluster when staged input is enabled (`default`, `staged`, `input`, `skip`).",
"PROVIDERS": "Provider profiles define reusable upstream URLs and DNS behavior that this input can reference."
,
,
"SERIES_SOURCE": "Source selector for the Series cluster when staged input is enabled (`default`, `staged`, `input`, `skip`).",
"VOD_SOURCE": "Source selector for the VOD cluster when staged input is enabled (`default`, `staged`, `input`, `skip`)."
},
@@ -611,6 +611,7 @@
"URL": "Target URL that receives the notification webhook."
},
"REVERSE_PROXY_CONFIG": {
"ADMISSION_STRATEGIES": "Defines what Tuliprox may still try after the normal user admission is already exhausted.\n\nUse this to control short transition grace during switching or to evict an existing same-IP stream instead of rejecting the new request immediately.\n\nStrategies are evaluated from top to bottom.\n\nGrace strategies only work when `grace_period_millis` is greater than 0. Only one grace strategy can be active at a time.",
"FAILOVER_REDIRECT_PATTERNS": "Regex patterns determining which HTTP error responses trigger an automatic failover rotation.",
"RESOURCE_REWRITE_DISABLED": "If enabled, URL rewriting for logos and EPG is disabled (also disables caching).",
"REWRITE_SECRET": "Persistent secret for stable resource URLs across restarts. \nExample:\n```\nreverse_proxy:\n rewrite_secret: A1B2C3D4E5F60718293A4B5C6D7E8F90\n```",
@@ -805,6 +806,10 @@
"ADD_TRAKT_LIST": "Add a Trakt list",
"ADD_URL": "Add Url",
"ADD_WATCH": "Add Watch",
"ADMISSION_STRATEGY_EVICT_USER_SAME_IP_LATEST": "Evict same-IP latest stream",
"ADMISSION_STRATEGY_EVICT_USER_SAME_IP_OLDEST": "Evict same-IP oldest stream",
"ADMISSION_STRATEGY_GRACE_HOLD_STREAM": "Grace hold stream",
"ADMISSION_STRATEGY_GRACE_INSTANT_STREAM": "Grace instant stream",
"ADVANCED": "Advanced",
"ALIAS": "Alias",
"ALIASES": "Aliases",
@@ -1387,6 +1392,7 @@
"VIEW_TYPE_PLAYLIST_EXPLORER": "Playlist Explorer",
"VIEW_TYPE_PLAYLIST_EPG": "Playlist EPG",
"VIEW_TYPE_RBAC": "WebUI Users",
"ADMISSION_STRATEGIES": "Admission Strategies",
"RUNTIME_CONFIG_REPORT": "Runtime Config Report",
"RUNTIME_CONFIG_REPORT_FORMAT": "Runtime Config Report Format"
},
+14
View File
@@ -148,6 +148,20 @@ li {
}
}
.tp__toolbar {
display: flex;
flex-flow: row wrap;
gap: var(--gap-small);
box-sizing: border-box;
.tp__icon-button {
&:hover {
background-color: var(--icon-button-hover-background-color);
border-radius: 50%;
}
}
}
::selection {
background-color: #0078d4;
color: #ffffff;
+1
View File
@@ -35,6 +35,7 @@ body {
--scrollbar-border-radius: 4px;
--action-card-width: 380px;
--gap-micro: 0.3rem;
--gap-mini: 0.3rem;
--gap-small: 0.3rem;
--gap-default: 0.5rem;
+5 -4
View File
@@ -40,10 +40,9 @@
background-color: var(--table-header-background-color);
box-shadow: inset 0 -1px 0 var(--border-color);
// if you want cell border
//&:not(:last-child) {
// border-right: 1px solid var(--border-color);
//}
&:not(:last-child) {
border-right: 1px solid var(--border-color);
}
svg,
.svg-icon {
@@ -84,9 +83,11 @@
vertical-align: top;
padding: var(--padding-mini);
/* cell border
&:not(:last-child) {
border-right: 1px solid var(--border-color);
}
*/
svg,
.svg-icon {
@@ -60,7 +60,6 @@
display: flex;
flex-flow: row wrap;
gap: var(--gap-default);
height: 2.5rem;
align-items: center;
li {
@@ -8,20 +8,21 @@ use crate::{
config_view_context::ConfigViewContext,
use_emit_mapped_option,
},
Card, Chip,
Card, Chip, IconButton, TextButton,
},
context::ConfigContext,
},
config_field, config_field_bool, config_field_child, config_field_custom, config_field_hide, config_field_optional,
edit_field_bool, edit_field_list, edit_field_number, edit_field_number_f64, edit_field_number_u16,
edit_field_number_u64, edit_field_number_usize, edit_field_text, edit_field_text_option, generate_form_reducer,
i18n::use_translation,
i18n::{use_translation, YewI18n},
utils::t_safe,
};
use shared::{
model::{
CacheConfigDto, GeoIpConfigDto, QosAggregationConfigDto, RateLimitConfigDto, ResourceRetryConfigDto,
ReverseProxyConfigDto, ReverseProxyDisabledHeaderConfigDto, StreamBufferConfigDto, StreamConfigDto,
StreamHistoryConfigDto,
AdmissionStrategyDto, CacheConfigDto, GeoIpConfigDto, QosAggregationConfigDto, RateLimitConfigDto,
ResourceRetryConfigDto, ReverseProxyConfigDto, ReverseProxyDisabledHeaderConfigDto, StreamBufferConfigDto,
StreamConfigDto, StreamHistoryConfigDto,
},
utils::{default_secret, format_float_localized},
};
@@ -31,7 +32,6 @@ const LABEL_CACHE: &str = "LABEL.CACHE";
const LABEL_ENABLED: &str = "LABEL.ENABLED";
const LABEL_SIZE: &str = "LABEL.SIZE";
const LABEL_DIRECTORY: &str = "LABEL.DIRECTORY";
const LABEL_STREAM: &str = "LABEL.STREAM";
const LABEL_STREAM_GRACE: &str = "LABEL.STREAM_GRACE";
const LABEL_STREAM_SESSION: &str = "LABEL.STREAM_SESSION";
@@ -53,6 +53,8 @@ const LABEL_PERIOD_MILLIS: &str = "LABEL.PERIOD_MILLIS";
const LABEL_BURST_SIZE: &str = "LABEL.BURST_SIZE";
const LABEL_SHARED_BURST_BUFFER_MB: &str = "LABEL.SHARED_BURST_BUFFER_BYTES";
const LABEL_ADMISSION_STRATEGIES: &str = "LABEL.ADMISSION_STRATEGIES";
const LABEL_SETTINGS: &str = "LABEL.SETTINGS";
const LABEL_RESOURCE_REWRITE_DISABLED: &str = "LABEL.RESOURCE_REWRITE_DISABLED";
const LABEL_REWRITE_SECRET: &str = "LABEL.REWRITE_SECRET";
@@ -78,6 +80,10 @@ const LABEL_STREAM_HISTORY_RETENTION_DAYS: &str = "LABEL.STREAM_HISTORY_RETENTIO
const LABEL_QOS_AGGREGATION: &str = "LABEL.QOS_AGGREGATION";
const LABEL_QOS_AGGREGATION_ENABLED: &str = "LABEL.QOS_AGGREGATION_ENABLED";
const LABEL_INTERVAL_SECS: &str = "LABEL.INTERVAL_SECS";
const LABEL_ADMISSION_STRATEGY_EVICT_USER_SAME_IP_OLDEST: &str = "LABEL.ADMISSION_STRATEGY_EVICT_USER_SAME_IP_OLDEST";
const LABEL_ADMISSION_STRATEGY_EVICT_USER_SAME_IP_LATEST: &str = "LABEL.ADMISSION_STRATEGY_EVICT_USER_SAME_IP_LATEST";
const LABEL_ADMISSION_STRATEGY_GRACE_INSTANT_STREAM: &str = "LABEL.ADMISSION_STRATEGY_GRACE_INSTANT_STREAM";
const LABEL_ADMISSION_STRATEGY_GRACE_HOLD_STREAM: &str = "LABEL.ADMISSION_STRATEGY_GRACE_HOLD_STREAM";
generate_form_reducer!(
state: CacheConfigFormState { form: CacheConfigDto },
@@ -119,6 +125,171 @@ impl FailoverPatternsDto {
pub fn is_empty(&self) -> bool { self.patterns.is_empty() }
}
#[derive(Debug, Clone, Default, PartialEq)]
pub struct AdmissionStrategiesDto {
pub strategies: Option<Vec<String>>,
}
fn admission_strategy_tag(strategy: AdmissionStrategyDto) -> &'static str {
match strategy {
AdmissionStrategyDto::EvictUserSameIpOldest => "evict_user_same_ip_oldest",
AdmissionStrategyDto::EvictUserSameIpLatest => "evict_user_same_ip_latest",
AdmissionStrategyDto::GraceInstantStream => "grace_instant_stream",
AdmissionStrategyDto::GraceHoldStream => "grace_hold_stream",
}
}
fn parse_admission_strategy_tag(tag: &str) -> Option<AdmissionStrategyDto> {
match tag.trim() {
"evict_user_same_ip_oldest" => Some(AdmissionStrategyDto::EvictUserSameIpOldest),
"evict_user_same_ip_latest" => Some(AdmissionStrategyDto::EvictUserSameIpLatest),
"grace_instant_stream" => Some(AdmissionStrategyDto::GraceInstantStream),
"grace_hold_stream" => Some(AdmissionStrategyDto::GraceHoldStream),
_ => None,
}
}
fn admission_strategy_label_key(strategy: AdmissionStrategyDto) -> &'static str {
match strategy {
AdmissionStrategyDto::EvictUserSameIpOldest => LABEL_ADMISSION_STRATEGY_EVICT_USER_SAME_IP_OLDEST,
AdmissionStrategyDto::EvictUserSameIpLatest => LABEL_ADMISSION_STRATEGY_EVICT_USER_SAME_IP_LATEST,
AdmissionStrategyDto::GraceInstantStream => LABEL_ADMISSION_STRATEGY_GRACE_INSTANT_STREAM,
AdmissionStrategyDto::GraceHoldStream => LABEL_ADMISSION_STRATEGY_GRACE_HOLD_STREAM,
}
}
fn admission_strategy_label(translate: &YewI18n, strategy: AdmissionStrategyDto) -> String {
t_safe(translate, admission_strategy_label_key(strategy)).unwrap_or_else(|| match strategy {
AdmissionStrategyDto::EvictUserSameIpOldest => "Evict same-IP oldest stream".to_string(),
AdmissionStrategyDto::EvictUserSameIpLatest => "Evict same-IP latest stream".to_string(),
AdmissionStrategyDto::GraceInstantStream => "Grace instant stream".to_string(),
AdmissionStrategyDto::GraceHoldStream => "Grace hold stream".to_string(),
})
}
fn is_grace_strategy(strategy: AdmissionStrategyDto) -> bool {
matches!(strategy, AdmissionStrategyDto::GraceInstantStream | AdmissionStrategyDto::GraceHoldStream)
}
fn is_grace_strategy_tag(tag: &str) -> bool {
let tag = tag.trim();
tag.starts_with("grace_") || parse_admission_strategy_tag(tag).is_some_and(is_grace_strategy)
}
fn admission_strategy_tags(strategies: Option<&Vec<AdmissionStrategyDto>>) -> Option<Vec<String>> {
strategies.map(|entries| entries.iter().map(|entry| admission_strategy_tag(*entry).to_string()).collect())
}
fn parse_admission_strategy_tags(tags: Option<&[String]>) -> Option<Vec<AdmissionStrategyDto>> {
let tags = tags?;
let mut parsed = Vec::new();
for tag in tags {
if let Some(strategy) = parse_admission_strategy_tag(tag) {
if !parsed.contains(&strategy) {
parsed.push(strategy);
}
}
}
Some(parsed)
}
fn filter_disabled_grace_strategy_tags(tags: Vec<String>, grace_period_millis: u64) -> Vec<String> {
if grace_period_millis == 0 {
tags.into_iter().filter(|tag| !is_grace_strategy_tag(tag)).collect()
} else {
tags
}
}
fn filter_disabled_grace_strategies(
strategies: Option<Vec<AdmissionStrategyDto>>,
grace_period_millis: u64,
) -> Option<Vec<AdmissionStrategyDto>> {
strategies.map(|entries| {
if grace_period_millis == 0 {
entries.into_iter().filter(|strategy| !is_grace_strategy(*strategy)).collect()
} else {
entries
}
})
}
fn admission_strategy_tag_label(translate: &YewI18n, tag: &str) -> String {
parse_admission_strategy_tag(tag)
.map(|strategy| admission_strategy_label(translate, strategy))
.unwrap_or_else(|| tag.to_string())
}
fn legacy_admission_strategy_tags(stream: &StreamConfigDto) -> Vec<String> {
if stream.grace_period_millis == 0 {
Vec::new()
} else {
vec![admission_strategy_tag(if stream.grace_period_hold_stream {
AdmissionStrategyDto::GraceHoldStream
} else {
AdmissionStrategyDto::GraceInstantStream
})
.to_string()]
}
}
fn displayed_admission_strategy_tags(state: &AdmissionStrategiesDto, stream: &StreamConfigDto) -> Vec<String> {
filter_disabled_grace_strategy_tags(
state.strategies.clone().unwrap_or_else(|| {
admission_strategy_tags(stream.admission_strategies.as_ref())
.unwrap_or_else(|| legacy_admission_strategy_tags(stream))
}),
stream.grace_period_millis,
)
}
fn available_admission_strategies(selected_tags: &[String], grace_period_millis: u64) -> Vec<AdmissionStrategyDto> {
let has_grace = selected_tags.iter().filter_map(|tag| parse_admission_strategy_tag(tag)).any(is_grace_strategy);
[
AdmissionStrategyDto::EvictUserSameIpOldest,
AdmissionStrategyDto::EvictUserSameIpLatest,
AdmissionStrategyDto::GraceInstantStream,
AdmissionStrategyDto::GraceHoldStream,
]
.into_iter()
.filter(|strategy| {
let tag = admission_strategy_tag(*strategy);
let grace_available = grace_period_millis > 0 || !is_grace_strategy(*strategy);
!selected_tags.iter().any(|selected| selected == tag)
&& grace_available
&& (!has_grace || !is_grace_strategy(*strategy))
})
.collect()
}
fn add_admission_strategy_tag(current: &[String], strategy: AdmissionStrategyDto) -> Vec<String> {
let mut next = current.to_vec();
let tag = admission_strategy_tag(strategy).to_string();
if !next.iter().any(|selected| selected == &tag) {
next.push(tag);
}
next
}
fn remove_admission_strategy_tag(current: &[String], index: usize) -> Vec<String> {
let mut next = current.to_vec();
if index < next.len() {
next.remove(index);
}
next
}
fn move_admission_strategy_tag(current: &[String], index: usize, delta: isize) -> Vec<String> {
let mut next = current.to_vec();
if let Some(target_index) = index.checked_add_signed(delta) {
if index < next.len() && target_index < next.len() {
next.swap(index, target_index);
}
}
next
}
generate_form_reducer!(
state: FailoverPatternsFormState { form: FailoverPatternsDto },
action_name: FailoverPatternsFormAction,
@@ -127,6 +298,14 @@ generate_form_reducer!(
}
);
generate_form_reducer!(
state: AdmissionStrategiesFormState { form: AdmissionStrategiesDto },
action_name: AdmissionStrategiesFormAction,
fields {
Strategies => strategies: Option<Vec<String>>,
}
);
generate_form_reducer!(
state: StreamConfigFormState { form: StreamConfigDto },
action_name: StreamConfigFormAction,
@@ -235,6 +414,8 @@ pub fn ReverseProxyConfigView() -> Html {
let failover_patterns_state: UseReducerHandle<FailoverPatternsFormState> =
use_reducer(|| FailoverPatternsFormState { form: FailoverPatternsDto::default(), modified: false });
let admission_strategies_state: UseReducerHandle<AdmissionStrategiesFormState> =
use_reducer(|| AdmissionStrategiesFormState { form: AdmissionStrategiesDto::default(), modified: false });
let stream_history_state: UseReducerHandle<StreamHistoryConfigFormState> =
use_reducer(|| StreamHistoryConfigFormState { form: StreamHistoryConfigDto::default(), modified: false });
let qos_aggregation_state: UseReducerHandle<QosAggregationConfigFormState> =
@@ -251,6 +432,7 @@ pub fn ReverseProxyConfigView() -> Html {
let geoip_state = geoip_state.clone();
let stream_buffer_state = stream_buffer_state.clone();
let failover_patterns_state = failover_patterns_state.clone();
let admission_strategies_state = admission_strategies_state.clone();
let stream_history_state = stream_history_state.clone();
let qos_aggregation_state = qos_aggregation_state.clone();
let last_emitted_form = last_emitted_form.clone();
@@ -267,6 +449,7 @@ pub fn ReverseProxyConfigView() -> Html {
geoip_state.form.clone(),
stream_buffer_state.form.clone(),
failover_patterns_state.form.clone(),
admission_strategies_state.form.clone(),
stream_history_state.form.clone(),
qos_aggregation_state.form.clone(),
),
@@ -280,6 +463,7 @@ pub fn ReverseProxyConfigView() -> Html {
geoip_state.modified,
stream_buffer_state.modified,
failover_patterns_state.modified,
admission_strategies_state.modified,
stream_history_state.modified,
qos_aggregation_state.modified,
),
@@ -296,6 +480,7 @@ pub fn ReverseProxyConfigView() -> Html {
geoip,
stream_buffer,
failover_patterns,
admission_strategies,
stream_history,
qos_aggregation,
),
@@ -309,6 +494,7 @@ pub fn ReverseProxyConfigView() -> Html {
geoip_modified,
stream_buffer_modified,
failover_patterns_modified,
admission_strategies_modified,
stream_history_modified,
qos_aggregation_modified,
),
@@ -316,6 +502,10 @@ pub fn ReverseProxyConfigView() -> Html {
let mut form = rp.clone();
let mut stream_form = stream.clone();
stream_form.buffer = if stream_buffer.is_empty() { None } else { Some(stream_buffer.clone()) };
stream_form.admission_strategies = filter_disabled_grace_strategies(
parse_admission_strategy_tags(admission_strategies.strategies.as_deref()),
stream_form.grace_period_millis,
);
form.cache = Some(cache.clone());
form.rate_limit = Some(rl.clone());
@@ -338,6 +528,7 @@ pub fn ReverseProxyConfigView() -> Html {
|| geoip_modified
|| stream_buffer_modified
|| failover_patterns_modified
|| admission_strategies_modified
|| stream_history_modified
|| qos_aggregation_modified;
let next_form = ConfigForm::ReverseProxy(modified, form);
@@ -362,6 +553,7 @@ pub fn ReverseProxyConfigView() -> Html {
let geoip_state = geoip_state.clone();
let stream_buffer_state = stream_buffer_state.clone();
let failover_patterns_state = failover_patterns_state.clone();
let admission_strategies_state = admission_strategies_state.clone();
let stream_history_state = stream_history_state.clone();
let qos_aggregation_state = qos_aggregation_state.clone();
@@ -424,6 +616,16 @@ pub fn ReverseProxyConfigView() -> Html {
failover_patterns_state.dispatch(FailoverPatternsFormAction::SetAll(target_failover_patterns));
}
let target_admission_strategies = AdmissionStrategiesDto {
strategies: admission_strategy_tags(
rp.stream.as_ref().and_then(|stream| stream.admission_strategies.as_ref()),
),
};
if admission_strategies_state.form != target_admission_strategies {
admission_strategies_state
.dispatch(AdmissionStrategiesFormAction::SetAll(target_admission_strategies));
}
let target_stream_history =
rp.stream_history.as_ref().map_or_else(StreamHistoryConfigDto::default, |s| s.clone());
if stream_history_state.form != target_stream_history {
@@ -482,6 +684,12 @@ pub fn ReverseProxyConfigView() -> Html {
failover_patterns_state.dispatch(FailoverPatternsFormAction::SetAll(target_failover_patterns));
}
let target_admission_strategies = AdmissionStrategiesDto::default();
if admission_strategies_state.form != target_admission_strategies {
admission_strategies_state
.dispatch(AdmissionStrategiesFormAction::SetAll(target_admission_strategies));
}
let target_stream_history = StreamHistoryConfigDto::default();
if stream_history_state.form != target_stream_history {
stream_history_state.dispatch(StreamHistoryConfigFormAction::SetAll(target_stream_history));
@@ -507,6 +715,7 @@ pub fn ReverseProxyConfigView() -> Html {
}
};
let render_stream = || {
let strategy_tags = displayed_admission_strategy_tags(&admission_strategies_state.form, &stream_state.form);
html! {
<>
<Card class="tp__config-view__card">
@@ -528,6 +737,22 @@ pub fn ReverseProxyConfigView() -> Html {
{ config_field!(stream_state.form, translate.t(LABEL_HLS_SESSION_TTL_SECS), hls_session_ttl_secs) }
{ config_field!(stream_state.form, translate.t(LABEL_CATCHUP_SESSION_TTL_SECS), catchup_session_ttl_secs) }
</Card>
<Card class="tp__config-view__card">
<h1>{translate.t(LABEL_ADMISSION_STRATEGIES)}</h1>
{ config_field_child!(translate.t(LABEL_ADMISSION_STRATEGIES), "REVERSE_PROXY_CONFIG.ADMISSION_STRATEGIES", {
html! {
<div class="tp__config-view__tags">
if strategy_tags.is_empty() {
<Chip label="-" />
} else {
{ for strategy_tags.iter().map(|strategy| {
html! { <Chip label={admission_strategy_tag_label(&translate, strategy)} /> }
}) }
}
</div>
}
})}
</Card>
</>
}
};
@@ -684,6 +909,9 @@ pub fn ReverseProxyConfigView() -> Html {
};
let render_stream_edit = || {
let strategy_tags = displayed_admission_strategy_tags(&admission_strategies_state.form, &stream_state.form);
let available_strategies =
available_admission_strategies(&strategy_tags, stream_state.form.grace_period_millis);
html! {
<>
<Card class="tp__config-view__card">
@@ -705,6 +933,86 @@ pub fn ReverseProxyConfigView() -> Html {
{ edit_field_number_u64!(stream_state, translate.t(LABEL_HLS_SESSION_TTL_SECS), hls_session_ttl_secs, StreamConfigFormAction::HlsSessionTtlSecs) }
{ edit_field_number_u64!(stream_state, translate.t(LABEL_CATCHUP_SESSION_TTL_SECS), catchup_session_ttl_secs, StreamConfigFormAction::CatchupSessionTtlSecs) }
</Card>
<Card class="tp__config-view__card">
<h1>{translate.t(LABEL_ADMISSION_STRATEGIES)}</h1>
<div class="tp__config-view__tags">
if strategy_tags.is_empty() {
<Chip label="-" />
} else {
{ for strategy_tags.iter().enumerate().map(|(index, strategy)| {
let remove_state = admission_strategies_state.clone();
let remove_tags = strategy_tags.clone();
let move_up_state = admission_strategies_state.clone();
let move_up_tags = strategy_tags.clone();
let move_down_state = admission_strategies_state.clone();
let move_down_tags = strategy_tags.clone();
let label = admission_strategy_tag_label(&translate, strategy);
html! {
<div class="tp__inline-toolbar">
<Chip label={label} />
if index > 0 {
<IconButton
name={format!("move_up_{index}")}
icon="ChevronUp"
hint="Move up"
onclick={Callback::from(move |_| {
move_up_state.dispatch(AdmissionStrategiesFormAction::Strategies(Some(
move_admission_strategy_tag(&move_up_tags, index, -1)
)));
})}
/>
}
if index + 1 < strategy_tags.len() {
<IconButton
name={format!("move_down_{index}")}
icon="ChevronDown"
hint="Move down"
onclick={Callback::from(move |_| {
move_down_state.dispatch(AdmissionStrategiesFormAction::Strategies(Some(
move_admission_strategy_tag(&move_down_tags, index, 1)
)));
})}
/>
}
<IconButton
class="secondary"
name={format!("remove_{index}")}
icon="Delete"
hint="Remove"
onclick={Callback::from(move |_| {
remove_state.dispatch(AdmissionStrategiesFormAction::Strategies(Some(
remove_admission_strategy_tag(&remove_tags, index)
)));
})}
/>
</div>
}
}) }
}
</div>
<div class="tp__toolbar">
{
for available_strategies.into_iter().map(|strategy| {
let add_state = admission_strategies_state.clone();
let add_tags = strategy_tags.clone();
let strategy_name = admission_strategy_label(&translate, strategy);
html! {
<TextButton
class="primary"
name={strategy_name.clone()}
icon="Add"
title={strategy_name}
onclick={Callback::from(move |_| {
add_state.dispatch(AdmissionStrategiesFormAction::Strategies(Some(
add_admission_strategy_tag(&add_tags, strategy)
)));
})}
/>
}
})
}
</div>
</Card>
</>
}
};
@@ -809,3 +1117,81 @@ pub fn ReverseProxyConfigView() -> Html {
</div>
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn admission_strategy_tags_roundtrip() {
let tags = admission_strategy_tags(Some(&vec![
AdmissionStrategyDto::EvictUserSameIpOldest,
AdmissionStrategyDto::GraceHoldStream,
]))
.unwrap_or_default();
assert_eq!(
parse_admission_strategy_tags(Some(&tags)),
Some(vec![AdmissionStrategyDto::EvictUserSameIpOldest, AdmissionStrategyDto::GraceHoldStream,])
);
}
#[test]
fn invalid_admission_strategy_tags_are_ignored() {
let tags = vec![
"evict_user_same_ip_latest".to_string(),
"not-a-strategy".to_string(),
"evict_user_same_ip_latest".to_string(),
];
assert_eq!(parse_admission_strategy_tags(Some(&tags)), Some(vec![AdmissionStrategyDto::EvictUserSameIpLatest]));
}
#[test]
fn displayed_admission_strategies_fall_back_to_legacy_grace() {
let state = AdmissionStrategiesDto::default();
let stream = StreamConfigDto {
grace_period_millis: 2_000,
grace_period_hold_stream: true,
..StreamConfigDto::default()
};
assert_eq!(displayed_admission_strategy_tags(&state, &stream), vec!["grace_hold_stream".to_string()]);
}
#[test]
fn available_admission_strategies_hide_second_grace_option() {
let available = available_admission_strategies(&["grace_hold_stream".to_string()], 2_000);
assert!(!available.contains(&AdmissionStrategyDto::GraceInstantStream));
assert!(!available.contains(&AdmissionStrategyDto::GraceHoldStream));
assert!(available.contains(&AdmissionStrategyDto::EvictUserSameIpOldest));
}
#[test]
fn available_admission_strategies_hide_grace_when_disabled() {
let available = available_admission_strategies(&[], 0);
assert!(!available.contains(&AdmissionStrategyDto::GraceInstantStream));
assert!(!available.contains(&AdmissionStrategyDto::GraceHoldStream));
assert!(available.contains(&AdmissionStrategyDto::EvictUserSameIpOldest));
assert!(available.contains(&AdmissionStrategyDto::EvictUserSameIpLatest));
}
#[test]
fn displayed_admission_strategies_hide_disabled_grace_tags() {
let state = AdmissionStrategiesDto { strategies: Some(vec!["grace_hold_stream".to_string()]) };
let stream = StreamConfigDto { grace_period_millis: 0, ..StreamConfigDto::default() };
assert_eq!(displayed_admission_strategy_tags(&state, &stream), Vec::<String>::new());
}
#[test]
fn filtered_admission_strategies_drop_grace_when_disabled() {
let parsed = parse_admission_strategy_tags(Some(&[
"evict_user_same_ip_oldest".to_string(),
"grace_hold_stream".to_string(),
]));
assert_eq!(
filter_disabled_grace_strategies(parsed, 0),
Some(vec![AdmissionStrategyDto::EvictUserSameIpOldest])
);
}
}
+40 -3
View File
@@ -21,6 +21,8 @@ use yew::prelude::*;
use yew_hooks::{use_async_with_options, UseAsyncOptions};
use yew_router::prelude::*;
const STATIC_ASSET_VERSION: &str = env!("CARGO_PKG_VERSION");
/// App routes
#[derive(Routable, Debug, Clone, PartialEq, Eq)]
pub enum AppRoute {
@@ -41,6 +43,13 @@ pub fn switch(route: AppRoute) -> Html {
}
}
fn versioned_static_asset_url(path: &str) -> String {
let separator = if path.contains('?') { '&' } else { '?' };
format!("{path}{separator}v={STATIC_ASSET_VERSION}")
}
fn versioned_config_url() -> String { versioned_static_asset_url("config.json") }
#[component]
pub fn App() -> Html {
let supported_languages = vec!["en"];
@@ -56,7 +65,7 @@ pub fn App() -> Html {
let futures = languages
.iter()
.map(|lang| async move {
let url = format!("assets/i18n/{lang}.json");
let url = versioned_static_asset_url(&format!("assets/i18n/{lang}.json"));
let result: Result<Option<Value>, Error> = request_get(&url, None, None).await;
(lang.to_string(), result)
})
@@ -79,7 +88,8 @@ pub fn App() -> Html {
let config_state = configuration_state.clone();
use_async_with_options::<_, (), Error>(
async move {
match request_get::<WebConfig>("config.json", None, None).await {
let config_url = versioned_config_url();
match request_get::<WebConfig>(&config_url, None, None).await {
Ok(Some(cfg)) => {
if let Some(tab_title) = cfg.tab_title.as_deref() {
if let Some(win) = window() {
@@ -108,7 +118,8 @@ pub fn App() -> Html {
let icon_state = icon_state.clone();
use_async_with_options::<_, (), Error>(
async move {
match request_get("assets/icons.json", None, None).await {
let icons_url = versioned_static_asset_url("assets/icons.json");
match request_get(&icons_url, None, None).await {
Ok(Some(icons)) => icon_state.set(Some(icons)),
Ok(None) => icon_state.set(Some(Vec::new())),
Err(err) => {
@@ -150,3 +161,29 @@ pub fn App() -> Html {
pub(in crate::app) struct CardContext {
pub custom_class: UseStateHandle<String>,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn versioned_static_asset_url_appends_release_version() {
assert_eq!(
versioned_static_asset_url("assets/i18n/en.json"),
format!("assets/i18n/en.json?v={}", env!("CARGO_PKG_VERSION"))
);
}
#[test]
fn versioned_static_asset_url_keeps_existing_query_params() {
assert_eq!(
versioned_static_asset_url("assets/i18n/en.json?lang=en"),
format!("assets/i18n/en.json?lang=en&v={}", env!("CARGO_PKG_VERSION"))
);
}
#[test]
fn versioned_config_url_uses_release_version() {
assert_eq!(versioned_config_url(), format!("config.json?v={}", env!("CARGO_PKG_VERSION")));
}
}
+141
View File
@@ -12,6 +12,18 @@ use crate::{
const STREAM_QUEUE_SIZE: usize = 1024; // mpsc channel holding messages. with 8192byte chunks and 2Mbit/s approx 8MB
const MIN_SHARED_BURST_BUFFER_MB: u64 = 1;
#[derive(Debug, Copy, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
pub enum AdmissionStrategyDto {
#[serde(rename = "evict_user_same_ip_oldest")]
EvictUserSameIpOldest,
#[serde(rename = "evict_user_same_ip_latest")]
EvictUserSameIpLatest,
#[serde(rename = "grace_instant_stream")]
GraceInstantStream,
#[serde(rename = "grace_hold_stream")]
GraceHoldStream,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct StreamBufferConfigDto {
@@ -59,6 +71,8 @@ pub struct StreamConfigDto {
pub throttle_kbps: u64,
#[serde(default = "default_shared_burst_buffer_mb", skip_serializing_if = "is_default_shared_burst_buffer_mb")]
pub shared_burst_buffer_mb: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub admission_strategies: Option<Vec<AdmissionStrategyDto>>,
}
impl Default for StreamConfigDto {
@@ -75,6 +89,7 @@ impl Default for StreamConfigDto {
grace_period_hold_stream: true,
hls_session_ttl_secs: default_hls_session_ttl_secs(),
catchup_session_ttl_secs: default_catchup_session_ttl_secs(),
admission_strategies: None,
}
}
}
@@ -92,6 +107,7 @@ impl StreamConfigDto {
&& self.grace_period_hold_stream
&& self.hls_session_ttl_secs == default_hls_session_ttl_secs()
&& self.catchup_session_ttl_secs == default_catchup_session_ttl_secs()
&& self.admission_strategies.is_none()
}
pub(crate) fn prepare(&mut self) -> Result<(), TuliproxError> {
@@ -122,6 +138,131 @@ impl StreamConfigDto {
)));
}
if let Some(strategies) = &self.admission_strategies {
if let Err(err) = validate_admission_strategies(strategies, self.grace_period_millis) {
return Err(TuliproxError::ConfigStream(err));
}
}
Ok(())
}
}
fn validate_admission_strategies(strategies: &[AdmissionStrategyDto], grace_period_millis: u64) -> Result<(), String> {
use std::collections::HashSet;
use AdmissionStrategyDto::*;
let mut seen = HashSet::new();
for s in strategies {
if !seen.insert(*s) {
return Err(format!(
"Duplicate admission strategy: {}",
serde_json::to_string(s).unwrap_or_default().trim_matches('"')
));
}
}
let has_instant = strategies.iter().any(|s| matches!(s, GraceInstantStream));
let has_hold = strategies.iter().any(|s| matches!(s, GraceHoldStream));
if has_instant && has_hold {
return Err("admission_strategies: grace_instant_stream and grace_hold_stream are mutually exclusive".into());
}
if grace_period_millis == 0 && (has_instant || has_hold) {
return Err("admission_strategies: grace strategies require grace_period_millis > 0".into());
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_empty_list_accepted() {
let mut dto = StreamConfigDto::default();
dto.admission_strategies = Some(vec![]);
assert!(dto.prepare().is_ok());
}
#[test]
fn test_none_strategies_accepted() {
let mut dto = StreamConfigDto::default();
assert!(dto.prepare().is_ok());
}
#[test]
fn test_duplicate_rejected() {
let mut dto = StreamConfigDto::default();
dto.admission_strategies =
Some(vec![AdmissionStrategyDto::EvictUserSameIpOldest, AdmissionStrategyDto::EvictUserSameIpOldest]);
let err = dto.prepare().unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("Duplicate admission strategy"), "msg: {msg}");
}
#[test]
fn test_mutually_exclusive_grace_rejected() {
let mut dto = StreamConfigDto::default();
dto.admission_strategies =
Some(vec![AdmissionStrategyDto::GraceInstantStream, AdmissionStrategyDto::GraceHoldStream]);
let err = dto.prepare().unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("mutually exclusive"), "msg: {msg}");
}
#[test]
fn test_valid_ordered_subset_accepted() {
let mut dto = StreamConfigDto::default();
dto.admission_strategies =
Some(vec![AdmissionStrategyDto::EvictUserSameIpOldest, AdmissionStrategyDto::GraceHoldStream]);
assert!(dto.prepare().is_ok());
}
#[test]
fn test_grace_strategy_requires_positive_grace_period() {
let mut dto = StreamConfigDto::default();
dto.grace_period_millis = 0;
dto.admission_strategies = Some(vec![AdmissionStrategyDto::GraceHoldStream]);
let err = dto.prepare().unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("grace_period_millis"), "msg: {msg}");
}
#[test]
fn test_single_strategy_accepted() {
for s in [
AdmissionStrategyDto::EvictUserSameIpOldest,
AdmissionStrategyDto::EvictUserSameIpLatest,
AdmissionStrategyDto::GraceInstantStream,
AdmissionStrategyDto::GraceHoldStream,
] {
let mut dto = StreamConfigDto::default();
dto.admission_strategies = Some(vec![s]);
assert!(dto.prepare().is_ok(), "Failed for {s:?}");
}
}
#[test]
fn test_is_empty_with_strategies() {
let mut dto = StreamConfigDto::default();
dto.admission_strategies = Some(vec![AdmissionStrategyDto::GraceInstantStream]);
assert!(!dto.is_empty());
}
#[test]
fn test_is_empty_with_explicit_empty_strategy_list() {
let mut dto = StreamConfigDto::default();
dto.admission_strategies = Some(vec![]);
assert!(!dto.is_empty());
}
#[test]
fn test_explicit_empty_strategy_list_serializes() {
let dto = StreamConfigDto { admission_strategies: Some(vec![]), ..StreamConfigDto::default() };
let serialized = serde_json::to_string(&dto).unwrap_or_default();
assert!(serialized.contains("admission_strategies"), "serialized: {serialized}");
}
}