mirror of
https://github.com/euzu/tuliprox.git
synced 2026-10-27 09:12:35 +01:00
fix(hls): keep provider affinity after reconnect lease expiry (#894)
* fix(hls): keep provider affinity after reconnect lease expiry - Remember the provider that last delivered media for reconnect-capable playback (HLS, DASH, Catchup, VOD) beyond the lease window. The preference holds no slot, never forces grace and falls back to the lineup when the provider is full. - End it on provider failure, preemption, kick, timeout or admin terminate; stale cleanups are rejected by binding tag, request ID and session token. Retry token claims stay bounded. - Terminate and kick now release the HLS binding they acquired. - Add reverse_proxy.stream.provider_affinity_ttl_secs (default 120, max 86400) with Web UI field, docs and translations. - Testkit: pause_millis steps, TTL overrides and the m3u-hls-provider-affinity-after-lease-expiry scenario. * fix(hls): keep provider affinity after reconnect lease expiry - Remember the provider that last delivered media for reconnect-capable playback (HLS, DASH, Catchup, VOD) beyond the lease window. The preference holds no slot, never forces grace and falls back to the lineup when the provider is full. - End it on provider failure, preemption, kick, timeout or admin terminate; stale cleanups are rejected by binding tag, request ID and session token. Retry token claims stay bounded. - Terminate and kick now release the HLS binding they acquired. - Add reverse_proxy.stream.provider_affinity_ttl_secs (default 120, max 86400) with Web UI field, docs and translations. - Testkit: pause_millis steps, TTL overrides and the m3u-hls-provider-affinity-after-lease-expiry scenario. * fix(hls): keep provider affinity after reconnect lease expiry - Remember the provider that last delivered media for reconnect-capable playback (HLS, DASH, Catchup, VOD) beyond the lease window. The preference holds no slot, never forces grace and falls back to the lineup when the provider is full. - End it on provider failure, preemption, kick, timeout or admin terminate; stale cleanups are rejected by binding tag, request ID and session token. Retry token claims stay bounded. - Terminate and kick now release the HLS binding they acquired. - Add reverse_proxy.stream.provider_affinity_ttl_secs (default 120, max 86400) with Web UI field, docs and translations. - Testkit: pause_millis steps, TTL overrides and the m3u-hls-provider-affinity-after-lease-expiry scenario. --------- Co-authored-by: euzu <euzu@proton.me>
This commit is contained in:
euzu
@@ -1025,6 +1025,18 @@
|
||||
|
||||
## 🐛 Fixes
|
||||
|
||||
- **A continuing playback keeps its provider after its reconnect window ends.** When an HLS player paused requests
|
||||
longer than `hls_session_ttl_secs`, its lease expired and the next request ran priority selection again, so a playback
|
||||
that had fallen back to an alias moved back to the primary account and then back to the alias. A playback now returns
|
||||
to the provider that last delivered its media for `provider_affinity_ttl_secs` after the reconnect window. This
|
||||
preference holds no connection slot and never causes a grace over-allocation while another provider has a free slot:
|
||||
a full provider still falls back to the lineup. Provider errors on the preferred provider, preemption, kicks and
|
||||
timeouts end it immediately; a failure on a fallback provider keeps it. It applies to reconnect-capable playback
|
||||
(HLS, DASH, Catchup, VOD), not to one-shot live MPEG-TS responses. A delayed failure of an older binding cannot end the preference of its
|
||||
successor. Terminating an HLS session or kicking its client now also releases the provider lease and preference of
|
||||
the binding that session acquired, which the per-retry public HLS token previously left in place. Terminating an
|
||||
unknown or outdated session token leaves a newer playback of the same client, user and channel untouched.
|
||||
|
||||
- **Banned, disabled or expired provider accounts are excluded from allocation and remembered.** A login or expiry
|
||||
response that reports `Banned`, `Disabled` or `Expired` excludes the account immediately and stores the exclusion in
|
||||
`source.yml` (`account_disabled: true` for a root account) or the alias CSV (`enabled` = `0`). `Pending` and other
|
||||
@@ -1451,6 +1463,11 @@
|
||||
|
||||
## ⚙️ New Settings
|
||||
|
||||
- **config.yml (`reverse_proxy.stream`)**:
|
||||
- `provider_affinity_ttl_secs` (default `120`): seconds a playback keeps preferring its last provider after the
|
||||
reconnect window has ended. It reserves no capacity. `0` ends the preference together with the reconnect window;
|
||||
values above `86400` are rejected. Editable in the Web UI under Reverse Proxy → Stream.
|
||||
|
||||
- **Runtime diagnostics (environment variables)**:
|
||||
- `TULIPROX_WATCHDOG` (default unset = off) is a mode selector: `1` (`true`/`on`/`yes`/`enabled`) observes and logs
|
||||
stalls, `2` (`restart`) additionally exits the process after the stall persists so a supervisor restarts it.
|
||||
@@ -1599,6 +1616,11 @@
|
||||
|
||||
## 🛠 Maintenance
|
||||
|
||||
- **Testkit scenarios can pause and tune reconnect windows.** A step with only `pause_millis` (1 to 300000) idles the
|
||||
scenario, and `policy_contract` accepts `hls_session_ttl_secs` and `provider_affinity_ttl_secs`. The new scenario
|
||||
`m3u-hls-provider-affinity-after-lease-expiry` uses both to prove that a returning HLS playback stays on its
|
||||
fallback account after its reconnect window lapsed and falls back to priority selection once the affinity ended.
|
||||
|
||||
- **The testkit can now drive a Stalker/Ministra input.** The fixture origin emulates a portal
|
||||
(`handshake`, `get_profile`, `get_genres`, `get_ordered_list`, `create_link`), a scenario selects it with
|
||||
`input_type: stalker` plus a `stalker:` block (MAC, MAG preset, scripted `create_link` refusals), and the
|
||||
|
||||
@@ -5513,6 +5513,7 @@ async fn activate_session_before_stream_open_skips_placeholder_for_follow_up_ses
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -5603,6 +5604,7 @@ async fn activate_session_before_stream_open_revalidates_precomputed_follow_up_r
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -5678,6 +5680,7 @@ async fn activate_session_before_stream_open_stale_follow_up_reclassified_on_cou
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -5756,6 +5759,7 @@ async fn activate_session_before_stream_open_pre_resolved_grace_period_materiali
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -5852,6 +5856,7 @@ async fn activate_session_before_stream_open_skips_placeholder_for_prepare() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -5908,6 +5913,7 @@ async fn resolve_playback_request_admission_prepare_only_returns_prepare_class()
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -5954,6 +5960,7 @@ async fn resolve_playback_request_admission_terminate_returns_terminate_class()
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6033,6 +6040,7 @@ async fn activate_session_before_stream_open_marks_pending_provider_for_grace_ho
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6142,6 +6150,7 @@ async fn activate_session_before_stream_open_does_not_commit_user_lease_before_p
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6268,6 +6277,7 @@ async fn effective_admission_strategies_use_legacy_grace_when_field_missing() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6294,6 +6304,7 @@ async fn effective_admission_strategies_respect_explicit_empty_list() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6319,6 +6330,7 @@ async fn grace_context_is_populated_when_grace_strategy_is_actually_granted() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6425,6 +6437,7 @@ async fn evaluate_remaining_strategies_evicts_after_used_grace() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6597,6 +6610,7 @@ async fn evaluate_remaining_strategies_skips_no_match_and_uses_later_eviction()
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6698,6 +6712,7 @@ async fn evaluate_remaining_strategies_empty_slice_denies() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6752,6 +6767,7 @@ async fn evaluate_remaining_strategies_preserves_soft_kind_on_exhausted() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6813,6 +6829,7 @@ async fn evaluate_remaining_strategies_does_not_retry_used_prefix() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6916,6 +6933,7 @@ async fn evaluate_remaining_strategies_empty_slice_uses_original_kind_not_contex
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -6980,6 +6998,7 @@ async fn evaluate_remaining_strategies_later_grace_uses_original_kind_not_contex
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -7079,6 +7098,7 @@ async fn resolve_admission_with_strategies_falls_through_after_failed_grace_gran
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -7280,6 +7300,7 @@ async fn resolve_admission_with_strategies_prevents_recently_evicted_playback_pi
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -7462,6 +7483,7 @@ async fn resolve_admission_with_strategies_allows_other_channel_after_recent_evi
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -7591,6 +7613,7 @@ async fn resolve_admission_with_strategies_does_not_suppress_different_session_o
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -7720,6 +7743,7 @@ async fn resolve_admission_with_strategies_allows_recently_evicted_playback_when
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -8790,6 +8814,7 @@ async fn resolve_admission_with_strategies_evicts_preserved_hls_session_for_same
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -9084,6 +9109,7 @@ async fn xtream_hls_then_ts_uses_distinct_tokens_and_evicts_old_hls_session() {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -9962,6 +9988,7 @@ async fn test_eviction_strategy_falls_back_when_candidate_protected_by_reentry()
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -10163,6 +10190,7 @@ async fn test_mixed_suppressed_and_legitimate_eviction_reports_exhaustion() {
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -10333,6 +10361,7 @@ async fn test_reentry_suppression_does_not_emit_connection_denied() {
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -10456,6 +10485,7 @@ async fn test_real_exhaustion_emits_connection_denied() {
|
||||
grace_period_hold_stream: false,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
|
||||
@@ -2158,6 +2158,9 @@ pub(super) async fn terminate_failed_hls_manifest_session(
|
||||
);
|
||||
cleared = true;
|
||||
} else if let (Some(provider_name), Some(binding_tag)) = (provider_name, binding_tag) {
|
||||
// A failed manifest lets the next entry fall back to another provider, but only
|
||||
// when the preference still belongs to this binding.
|
||||
app_state.active_provider.forget_identified_provider_affinity(session_token, provider_name, binding_tag);
|
||||
app_state.active_provider.clear_identified_provider_reservation(
|
||||
session_token,
|
||||
provider_name,
|
||||
|
||||
@@ -12567,6 +12567,118 @@ async fn terminate_failed_hls_manifest_session_releases_identified_provider_rese
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn delayed_manifest_failure_of_older_binding_keeps_provider_affinity() {
|
||||
let input = single_hls_provider_input("affinity-hls-input");
|
||||
let app_state = test_app_state_with_inputs(vec![Arc::new(input.clone())]);
|
||||
let provider_name = Arc::clone(&input.name);
|
||||
let old_token = "testuser|stream-1|hls|0123456789abcdef";
|
||||
let new_token = "testuser|stream-1|hls|fedcba9876543210";
|
||||
let acquire = |token: &'static str, port: u16| {
|
||||
app_state
|
||||
.active_provider
|
||||
.acquire_connection_with_lease_for_session(
|
||||
&provider_name,
|
||||
&test_addr_with_port(port),
|
||||
false,
|
||||
0,
|
||||
ConnectionKind::Normal,
|
||||
Some(tuliprox_session::PlaybackLeaseRef::new(token, crate::model::PlaybackKind::LiveHls)),
|
||||
)
|
||||
.expect("handle should be acquired")
|
||||
};
|
||||
|
||||
// The older binding never produced media and ended.
|
||||
let old = acquire(old_token, 55311);
|
||||
let (old_tag, old_request) = (old.binding_tag, old.playback_request_id);
|
||||
let old_request_id = old_request.expect("old request id");
|
||||
app_state.connection_manager.release_provider_handle(Some(old));
|
||||
app_state.active_provider.finish_identified_playback_request(
|
||||
old_token,
|
||||
old_request_id,
|
||||
tuliprox_core::model::PlaybackRequestOutcome::FailedBeforeMedia,
|
||||
);
|
||||
|
||||
// The successor binding confirms media and owns the provider preference.
|
||||
let current = acquire(new_token, 55312);
|
||||
let current_request = current.playback_request_id.expect("current request id");
|
||||
app_state.active_provider.refresh_adaptive_playback_lease(
|
||||
&provider_name,
|
||||
new_token,
|
||||
crate::model::PlaybackKind::LiveHls,
|
||||
15,
|
||||
);
|
||||
app_state.active_provider.confirm_identified_playback_activity(new_token, current_request);
|
||||
app_state.connection_manager.release_provider_handle(Some(current));
|
||||
assert_eq!(app_state.active_provider.provider_affinity_for_owner(new_token), Some(Arc::clone(&provider_name)));
|
||||
|
||||
for (tag, request) in [(old_tag, old_request), (old_tag, None)] {
|
||||
super::segment::terminate_failed_hls_manifest_session(
|
||||
&app_state,
|
||||
"testuser",
|
||||
old_token,
|
||||
Some(&provider_name),
|
||||
tag,
|
||||
request,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
app_state.active_provider.provider_affinity_for_owner(new_token),
|
||||
Some(Arc::clone(&provider_name)),
|
||||
"a delayed failure of the older binding must not end the successor's provider preference"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn terminating_unknown_old_hls_session_keeps_newer_provider_lease() {
|
||||
use axum::response::IntoResponse;
|
||||
let input = single_hls_provider_input("terminate-hls-input");
|
||||
let app_state = test_app_state_with_inputs(vec![Arc::new(input.clone())]);
|
||||
let provider_name = Arc::clone(&input.name);
|
||||
let old_token = "testuser|stream-1|hls|0123456789abcdef";
|
||||
let new_token = "testuser|stream-1|hls|fedcba9876543210";
|
||||
let current = app_state
|
||||
.active_provider
|
||||
.acquire_connection_with_lease_for_session(
|
||||
&provider_name,
|
||||
&test_addr_with_port(55321),
|
||||
false,
|
||||
0,
|
||||
ConnectionKind::Normal,
|
||||
Some(tuliprox_session::PlaybackLeaseRef::new(new_token, crate::model::PlaybackKind::LiveHls)),
|
||||
)
|
||||
.expect("handle should be acquired");
|
||||
let request_id = current.playback_request_id.expect("request id");
|
||||
app_state.active_provider.refresh_adaptive_playback_lease(
|
||||
&provider_name,
|
||||
new_token,
|
||||
crate::model::PlaybackKind::LiveHls,
|
||||
15,
|
||||
);
|
||||
app_state.active_provider.confirm_identified_playback_activity(new_token, request_id);
|
||||
app_state.connection_manager.release_provider_handle(Some(current));
|
||||
app_state.active_provider.finish_identified_playback_request(
|
||||
new_token,
|
||||
request_id,
|
||||
tuliprox_core::model::PlaybackRequestOutcome::Completed,
|
||||
);
|
||||
let binding_tag = app_state.active_provider.binding_tag_for_owner(new_token);
|
||||
assert!(binding_tag.is_some(), "the newer playback keeps an idle lease");
|
||||
|
||||
let response = crate::api::endpoints::v1_api_user::terminate_user_session(
|
||||
axum::extract::State(Arc::clone(&app_state)),
|
||||
axum::extract::Path(("testuser".to_string(), old_token.to_string())),
|
||||
)
|
||||
.await
|
||||
.into_response();
|
||||
|
||||
assert_eq!(response.status(), StatusCode::NO_CONTENT);
|
||||
assert_eq!(app_state.active_provider.binding_tag_for_owner(new_token), binding_tag);
|
||||
assert_eq!(app_state.active_provider.provider_affinity_for_owner(new_token), Some(provider_name));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn terminate_failed_hls_manifest_session_preserves_shared_lease_when_request_id_available() {
|
||||
let mut input = single_hls_provider_input("shared-hls-input");
|
||||
|
||||
@@ -19,7 +19,7 @@ mod user_visibility;
|
||||
pub(in crate::api) mod v1_api;
|
||||
mod v1_api_config;
|
||||
pub(in crate::api) mod v1_api_playlist;
|
||||
mod v1_api_user;
|
||||
pub(in crate::api) mod v1_api_user;
|
||||
pub(in crate::api) mod web_index;
|
||||
pub(in crate::api) mod websocket_api;
|
||||
pub(in crate::api) mod xmltv_api;
|
||||
|
||||
@@ -288,12 +288,16 @@ async fn delete_config_api_proxy_user(
|
||||
/// - Removes the session and all associated streams
|
||||
/// - Releases the counted lease if held
|
||||
/// - Sets lifecycle to `Expired`
|
||||
async fn terminate_user_session(
|
||||
pub(in crate::api) async fn terminate_user_session(
|
||||
axum::extract::State(app_state): axum::extract::State<Arc<AppState>>,
|
||||
axum::extract::Path((username, session_token)): axum::extract::Path<(String, String)>,
|
||||
) -> impl axum::response::IntoResponse {
|
||||
app_state.active_users.terminate_session(&username, &session_token).await;
|
||||
app_state.active_provider.clear_provider_reservation(&session_token);
|
||||
let _transition_guard = app_state.active_users.acquire_playback_transition(&username, &session_token).await;
|
||||
// Only a session that actually existed may end provider state, and only the binding
|
||||
// its token acquired: a stale token must not end a newer playback of the same owner.
|
||||
if app_state.active_users.terminate_session(&username, &session_token).await {
|
||||
app_state.active_provider.terminate_identified_playback_owner(&session_token);
|
||||
}
|
||||
(axum::http::StatusCode::NO_CONTENT).into_response()
|
||||
}
|
||||
|
||||
|
||||
@@ -2868,6 +2868,7 @@ mod tests {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -3143,6 +3144,7 @@ mod tests {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 10,
|
||||
catchup_session_ttl_secs: 10,
|
||||
provider_affinity_ttl_secs: 120,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
|
||||
@@ -295,6 +295,13 @@ impl Config {
|
||||
self.reverse_proxy.as_ref().and_then(|r| r.disabled_header.clone())
|
||||
}
|
||||
|
||||
pub fn get_provider_affinity_ttl_secs(&self) -> u64 {
|
||||
self.reverse_proxy
|
||||
.as_ref()
|
||||
.and_then(|r| r.stream.as_ref())
|
||||
.map_or_else(shared::defaults::default_provider_affinity_ttl_secs, |s| s.provider_affinity_ttl_secs)
|
||||
}
|
||||
|
||||
pub fn get_grace_options(&self) -> GracePeriodOptions {
|
||||
self.reverse_proxy.as_ref().and_then(|r| r.stream.as_ref()).map_or_else(GracePeriodOptions::default, |s| {
|
||||
GracePeriodOptions {
|
||||
|
||||
@@ -38,6 +38,7 @@ pub struct StreamConfig {
|
||||
pub grace_period_hold_stream: bool,
|
||||
pub hls_session_ttl_secs: u64,
|
||||
pub catchup_session_ttl_secs: u64,
|
||||
pub provider_affinity_ttl_secs: u64,
|
||||
pub throttle_str: Option<String>,
|
||||
pub throttle_kbps: u64,
|
||||
pub shared_burst_buffer_mb: u64,
|
||||
@@ -59,6 +60,7 @@ impl Default for StreamConfig {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 15,
|
||||
catchup_session_ttl_secs: 45,
|
||||
provider_affinity_ttl_secs: shared::defaults::default_provider_affinity_ttl_secs(),
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 12,
|
||||
@@ -80,6 +82,7 @@ impl From<&StreamConfigDto> for StreamConfig {
|
||||
grace_period_hold_stream: dto.grace_period_hold_stream,
|
||||
hls_session_ttl_secs: dto.hls_session_ttl_secs,
|
||||
catchup_session_ttl_secs: dto.catchup_session_ttl_secs,
|
||||
provider_affinity_ttl_secs: dto.provider_affinity_ttl_secs,
|
||||
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,
|
||||
@@ -102,6 +105,7 @@ impl From<&StreamConfig> for StreamConfigDto {
|
||||
grace_period_hold_stream: instance.grace_period_hold_stream,
|
||||
hls_session_ttl_secs: instance.hls_session_ttl_secs,
|
||||
catchup_session_ttl_secs: instance.catchup_session_ttl_secs,
|
||||
provider_affinity_ttl_secs: instance.provider_affinity_ttl_secs,
|
||||
throttle: instance.throttle_str.clone(),
|
||||
throttle_kbps: instance.throttle_kbps,
|
||||
shared_burst_buffer_mb: instance.shared_burst_buffer_mb,
|
||||
@@ -158,6 +162,7 @@ mod tests {
|
||||
grace_period_hold_stream: true,
|
||||
hls_session_ttl_secs: 5,
|
||||
catchup_session_ttl_secs: 5,
|
||||
provider_affinity_ttl_secs: 30,
|
||||
throttle_str: None,
|
||||
throttle_kbps: 0,
|
||||
shared_burst_buffer_mb: 1,
|
||||
@@ -173,6 +178,8 @@ mod tests {
|
||||
|
||||
let dto = StreamConfigDto::from(&domain);
|
||||
assert_eq!(dto.recent_eviction_reentry_ttl_ms, 2_500);
|
||||
assert_eq!(dto.provider_affinity_ttl_secs, 30);
|
||||
assert_eq!(StreamConfig::from(&dto).provider_affinity_ttl_secs, 30);
|
||||
assert_eq!(
|
||||
dto.admission_strategies,
|
||||
Some(vec![
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
use crate::{
|
||||
provider_leases::{ProviderLeaseTable, ProviderLeaseUsage},
|
||||
provider_leases::{ProviderAffinityTtl, ProviderLeaseTable, ProviderLeaseUsage},
|
||||
provider_lineup_manager::ProviderLineupManager,
|
||||
EventManager, SharedStreamManager,
|
||||
};
|
||||
@@ -94,6 +94,14 @@ fn playback_lease_owner(owner: &str) -> &str {
|
||||
}
|
||||
}
|
||||
|
||||
/// How an acquisition of a playback owner is tied to a provider.
|
||||
enum OwnerProviderPin {
|
||||
/// A confirmed or active lease pins the playback to exactly this provider.
|
||||
Reserved(Arc<str>),
|
||||
/// A recently confirmed playback prefers this provider; it holds no capacity.
|
||||
Preferred(Arc<str>),
|
||||
}
|
||||
|
||||
/// Playback identity of one provider acquisition.
|
||||
///
|
||||
/// `owner` carries the public session identity so transport cleanup can distinguish
|
||||
@@ -632,7 +640,7 @@ impl ActiveProviderManager {
|
||||
shutdown_token: CancellationToken::new(),
|
||||
providers: ProviderLineupManager::new(inputs, grace_period_options, event_manager),
|
||||
connections: std::sync::RwLock::new(Connections::default()),
|
||||
leases: std::sync::RwLock::new(ProviderLeaseTable::default()),
|
||||
leases: std::sync::RwLock::new(ProviderLeaseTable::with_affinity_ttl(Self::get_affinity_ttl(cfg))),
|
||||
next_allocation_id: AtomicU64::new(1),
|
||||
}),
|
||||
shared_stream_manager: OnceLock::new(),
|
||||
@@ -700,10 +708,15 @@ impl ActiveProviderManager {
|
||||
|
||||
fn get_grace_options(cfg: &AppConfig) -> GracePeriodOptions { cfg.config.load().get_grace_options() }
|
||||
|
||||
fn get_affinity_ttl(cfg: &AppConfig) -> ProviderAffinityTtl {
|
||||
ProviderAffinityTtl(cfg.config.load().get_provider_affinity_ttl_secs())
|
||||
}
|
||||
|
||||
pub fn update_config(&self, cfg: &AppConfig) {
|
||||
let grace_period_options = Self::get_grace_options(cfg);
|
||||
let inputs = Self::get_config_inputs(cfg);
|
||||
self.providers.update_config(inputs, &grace_period_options);
|
||||
self.write_leases().set_affinity_ttl(Self::get_affinity_ttl(cfg));
|
||||
self.reconcile_connections();
|
||||
}
|
||||
|
||||
@@ -742,16 +755,36 @@ impl ActiveProviderManager {
|
||||
leases.has_foreign_reserved_lease(provider_name, session_owner)
|
||||
}
|
||||
|
||||
fn get_reserved_provider_for_owner(&self, input_name: &Arc<str>, session_owner: &str) -> Option<Arc<str>> {
|
||||
/// Resolves how a playback owner is bound to a provider of `input_name`, with a
|
||||
/// single lease-table lookup per acquisition.
|
||||
fn owner_provider_pin(&self, input_name: &Arc<str>, session_owner: &str) -> Option<OwnerProviderPin> {
|
||||
let (lease_pin, affinity_provider) = {
|
||||
let mut leases = self.write_leases();
|
||||
Self::prune_expired_leases(&mut leases);
|
||||
let lease_pin = leases
|
||||
.lease_of_owner(session_owner)
|
||||
.map(|lease| (Arc::clone(&lease.provider_name), lease.state.is_confirmed()));
|
||||
(lease_pin, leases.affinity_provider_for_owner(session_owner))
|
||||
};
|
||||
// The lease lock is released before the connection lock is taken.
|
||||
if let Some((provider_name, confirmed)) = lease_pin {
|
||||
if self.providers.is_provider_for_input(&provider_name, input_name)
|
||||
&& (confirmed || self.has_active_owner_for_provider(&provider_name, session_owner))
|
||||
{
|
||||
return Some(OwnerProviderPin::Reserved(provider_name));
|
||||
}
|
||||
}
|
||||
affinity_provider
|
||||
.filter(|provider_name| self.providers.is_provider_for_input(provider_name, input_name))
|
||||
.map(OwnerProviderPin::Preferred)
|
||||
}
|
||||
|
||||
/// Provider a playback currently prefers on re-entry, for diagnostics and tests.
|
||||
pub fn provider_affinity_for_owner(&self, session_owner: &str) -> Option<Arc<str>> {
|
||||
let session_owner = playback_lease_owner(session_owner);
|
||||
let mut leases = self.write_leases();
|
||||
Self::prune_expired_leases(&mut leases);
|
||||
let lease = leases.lease_of_owner(session_owner)?;
|
||||
let provider_name = Arc::clone(&lease.provider_name);
|
||||
let confirmed = lease.state.is_confirmed();
|
||||
drop(leases);
|
||||
(self.providers.is_provider_for_input(&provider_name, input_name)
|
||||
&& (confirmed || self.has_active_owner_for_provider(&provider_name, session_owner)))
|
||||
.then_some(provider_name)
|
||||
leases.affinity_provider_for_owner(session_owner)
|
||||
}
|
||||
|
||||
/// A provisional lease only pins its provider while its allocation is active.
|
||||
@@ -1028,9 +1061,42 @@ impl ActiveProviderManager {
|
||||
}
|
||||
let _transition = self.lock_capacity_transition();
|
||||
let mut leases = self.write_leases();
|
||||
leases.forget_affinity(session_owner);
|
||||
leases.release_owner(session_owner);
|
||||
}
|
||||
|
||||
/// Administrative end of the playback behind a session that was actually terminated
|
||||
/// (terminate endpoint, kick).
|
||||
///
|
||||
/// A public HLS token shares its stable owner with the retries of the same
|
||||
/// client/user/channel, so it only ends the binding it acquired itself. A stale
|
||||
/// token therefore never removes the lease or provider preference of a newer
|
||||
/// binding. A bare owner keeps the owner-wide release semantics.
|
||||
pub fn terminate_identified_playback_owner(&self, session_token: &str) -> bool {
|
||||
let owner = playback_lease_owner(session_token);
|
||||
let _transition = self.lock_capacity_transition();
|
||||
let mut leases = self.write_leases();
|
||||
if owner == session_token {
|
||||
leases.forget_affinity(owner);
|
||||
return leases.release_owner(owner).is_some();
|
||||
}
|
||||
leases.terminate_identified_token(owner, session_token)
|
||||
}
|
||||
|
||||
/// Ends the provider preference of a failed playback, but only while it still
|
||||
/// belongs to the exact binding that failed. A delayed failure of an older binding
|
||||
/// leaves the successor's preference intact.
|
||||
pub fn forget_identified_provider_affinity(
|
||||
&self,
|
||||
session_owner: &str,
|
||||
provider_name: &Arc<str>,
|
||||
binding_tag: ProviderBindingTag,
|
||||
) -> bool {
|
||||
let session_owner = playback_lease_owner(session_owner);
|
||||
let _transition = self.lock_capacity_transition();
|
||||
self.write_leases().forget_identified_affinity(session_owner, provider_name, binding_tag)
|
||||
}
|
||||
|
||||
pub fn clear_identified_provider_reservation(
|
||||
&self,
|
||||
session_owner: &str,
|
||||
@@ -1365,8 +1431,33 @@ impl ActiveProviderManager {
|
||||
params: &AcquireProviderParams<'_>,
|
||||
) -> Option<ProviderHandle> {
|
||||
if let Some(owner) = params.session_owner() {
|
||||
if let Some(reserved_provider) = self.get_reserved_provider_for_owner(provider_or_input_name, owner) {
|
||||
return self.acquire_exact_connection_inner_no_preempt(&reserved_provider, allow_grace, params);
|
||||
match self.owner_provider_pin(provider_or_input_name, owner) {
|
||||
Some(OwnerProviderPin::Reserved(reserved_provider)) => {
|
||||
return self.acquire_exact_connection_inner_no_preempt(&reserved_provider, allow_grace, params);
|
||||
}
|
||||
// A continuing playback whose reconnect lease lapsed during a request gap
|
||||
// returns to its provider before priority selection, but only into free
|
||||
// capacity: the lineup prefers any free slot over a grace over-allocation,
|
||||
// so the preference must not force one either.
|
||||
Some(OwnerProviderPin::Preferred(affinity_provider)) => {
|
||||
if let Some(handle) =
|
||||
self.acquire_exact_connection_inner_no_preempt(&affinity_provider, false, params)
|
||||
{
|
||||
debug_if_enabled!(
|
||||
"Provider affinity kept: provider={} owner={} playback_kind={}",
|
||||
sanitize_sensitive_info(&affinity_provider),
|
||||
sanitize_sensitive_info(owner),
|
||||
params.lease.map_or_else(|| "-".to_string(), |lease| lease.kind.to_string())
|
||||
);
|
||||
return Some(handle);
|
||||
}
|
||||
debug_if_enabled!(
|
||||
"Provider affinity unavailable, using lineup: provider={} owner={}",
|
||||
sanitize_sensitive_info(&affinity_provider),
|
||||
sanitize_sensitive_info(owner)
|
||||
);
|
||||
}
|
||||
None => {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1452,6 +1543,9 @@ impl ActiveProviderManager {
|
||||
let mut leases = self.write_leases();
|
||||
Self::prune_expired_leases(&mut leases);
|
||||
let id = leases.begin_owner(lease.provider_owner(), &provider_name, lease.kind, lease.request_id);
|
||||
if lease.owner != lease.provider_owner() {
|
||||
leases.attach_request_token(id, lease.request_id, lease.owner);
|
||||
}
|
||||
let generation = leases.lease(id).map_or(1, |lease| lease.binding_generation);
|
||||
debug_if_enabled!(
|
||||
"Playback lease began: provider={} owner={} kind={} request_id={} lease_id={} generation={} state=starting",
|
||||
@@ -4850,6 +4944,482 @@ mod tests {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Primary `provider_1` with an alias `provider_2` behind it.
|
||||
fn alias_pool(primary_max: u16, alias_max: u16) -> (ActiveProviderManager, Arc<str>, Arc<str>) {
|
||||
let app_cfg = create_test_app_config_with_pool(primary_max, alias_max);
|
||||
let manager = ActiveProviderManager::new(&app_cfg, &Arc::new(EventManager::new()));
|
||||
(manager, "provider_1".intern(), "provider_2".intern())
|
||||
}
|
||||
|
||||
fn acquire_live_hls_from_lineup(
|
||||
manager: &ActiveProviderManager,
|
||||
input: &Arc<str>,
|
||||
token: &str,
|
||||
port: u16,
|
||||
) -> Result<tuliprox_core::model::ProviderHandle, String> {
|
||||
manager
|
||||
.acquire_connection_with_lease_for_session(
|
||||
input,
|
||||
&SocketAddr::from(([172, 18, 0, 9], port)),
|
||||
false,
|
||||
default_user_priority(),
|
||||
ConnectionKind::Normal,
|
||||
Some(PlaybackLeaseRef::new(token, PlaybackKind::LiveHls)),
|
||||
)
|
||||
.ok_or_else(|| format!("no provider for {token}"))
|
||||
}
|
||||
|
||||
/// Starts `token` on the alias while the primary is full, confirms media, ends the
|
||||
/// request cleanly, frees the primary and lets the reconnect lease lapse.
|
||||
async fn live_hls_playback_on_alias_with_lapsed_lease(
|
||||
manager: &ActiveProviderManager,
|
||||
input: &Arc<str>,
|
||||
alias: &Arc<str>,
|
||||
token: &str,
|
||||
outcome: PlaybackRequestOutcome,
|
||||
) -> Result<(), String> {
|
||||
let busy_1 = acquire_live_hls_from_lineup(manager, input, "busy-1|bob|7|hls|0123456789abcdef", 50_001)?;
|
||||
let busy_2 = acquire_live_hls_from_lineup(manager, input, "busy-2|carol|8|hls|0123456789abcdef", 50_002)?;
|
||||
assert_eq!(busy_1.allocation.get_provider_name().as_ref(), Some(input));
|
||||
assert_eq!(busy_2.allocation.get_provider_name().as_ref(), Some(input));
|
||||
|
||||
let handle = acquire_live_hls_from_lineup(manager, input, token, 50_010)?;
|
||||
let provider = handle.allocation.get_provider_name().ok_or("provider")?;
|
||||
assert_eq!(&provider, alias, "full primary must fall back to the alias");
|
||||
let request_id = handle.playback_request_id.ok_or("identified request")?;
|
||||
manager.refresh_adaptive_playback_lease(&provider, token, PlaybackKind::LiveHls, 15);
|
||||
manager.confirm_identified_playback_activity(token, request_id).ok_or("confirmation")?;
|
||||
manager.release_handle(&handle);
|
||||
manager.finish_identified_playback_request(token, request_id, outcome);
|
||||
|
||||
manager.release_handle(&busy_1);
|
||||
manager.release_handle(&busy_2);
|
||||
tokio::time::advance(Duration::from_secs(20)).await;
|
||||
assert_eq!(manager.provider_lease_usage(alias).total(), 0, "reconnect lease must have lapsed");
|
||||
assert_eq!(manager.provider_lease_usage(input).total(), 0);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn live_hls_reentry_keeps_alias_affinity_after_lease_expiry() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
live_hls_playback_on_alias_with_lapsed_lease(
|
||||
&manager,
|
||||
&input,
|
||||
&alias,
|
||||
"client|alice|42|hls|0123456789abcdef",
|
||||
PlaybackRequestOutcome::Completed,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let reentry = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|fedcba9876543210", 50_011)?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&alias)));
|
||||
// Affinity reserves nothing: another playback still gets the free primary slot.
|
||||
let other = acquire_live_hls_from_lineup(&manager, &input, "client|dave|42|hls|0123456789abcdef", 50_012)?;
|
||||
assert_eq!(other.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
manager.release_handle(&reentry);
|
||||
manager.release_handle(&other);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn live_hls_alias_affinity_ends_after_its_window() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
live_hls_playback_on_alias_with_lapsed_lease(
|
||||
&manager,
|
||||
&input,
|
||||
&alias,
|
||||
"client|alice|42|hls|0123456789abcdef",
|
||||
PlaybackRequestOutcome::Completed,
|
||||
)
|
||||
.await?;
|
||||
tokio::time::advance(Duration::from_secs(shared::defaults::default_provider_affinity_ttl_secs())).await;
|
||||
|
||||
let reentry = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|fedcba9876543210", 50_011)?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
manager.release_handle(&reentry);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Starts `token` on the alias while the primary is full and confirms media. The
|
||||
/// request stays attached; the primary is freed again before returning.
|
||||
fn confirmed_alias_request(
|
||||
manager: &ActiveProviderManager,
|
||||
input: &Arc<str>,
|
||||
alias: &Arc<str>,
|
||||
token: &str,
|
||||
) -> Result<tuliprox_core::model::ProviderHandle, String> {
|
||||
let busy_1 = acquire_live_hls_from_lineup(manager, input, "busy-1|bob|7|hls|0123456789abcdef", 50_001)?;
|
||||
let busy_2 = acquire_live_hls_from_lineup(manager, input, "busy-2|carol|8|hls|0123456789abcdef", 50_002)?;
|
||||
let handle = acquire_live_hls_from_lineup(manager, input, token, 50_010)?;
|
||||
assert_eq!(handle.allocation.get_provider_name().as_ref(), Some(alias));
|
||||
let request_id = handle.playback_request_id.ok_or("identified request")?;
|
||||
manager.refresh_adaptive_playback_lease(alias, token, PlaybackKind::LiveHls, 15);
|
||||
manager.confirm_identified_playback_activity(token, request_id).ok_or("confirmation")?;
|
||||
manager.release_handle(&busy_1);
|
||||
manager.release_handle(&busy_2);
|
||||
Ok(handle)
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn provider_failure_after_lease_expiry_ends_alias_affinity() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
let token = "client|alice|42|hls|0123456789abcdef";
|
||||
let handle = confirmed_alias_request(&manager, &input, &alias, token)?;
|
||||
let request_id = handle.playback_request_id.ok_or("identified request")?;
|
||||
manager.release_handle(&handle);
|
||||
tokio::time::advance(Duration::from_secs(20)).await;
|
||||
manager.prune_expired_leases_now();
|
||||
assert_eq!(manager.provider_lease_usage(&alias).total(), 0, "lease must have expired");
|
||||
|
||||
manager.finish_identified_playback_request(token, request_id, PlaybackRequestOutcome::ProviderFailed);
|
||||
|
||||
let reentry = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|fedcba9876543210", 50_011)?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
manager.release_handle(&reentry);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn delayed_failure_of_older_binding_keeps_successor_affinity() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
let old_token = "client|alice|42|hls|0123456789abcdef";
|
||||
let new_token = "client|alice|42|hls|fedcba9876543210";
|
||||
|
||||
// An older binding on the primary that never confirmed media.
|
||||
let old = manager
|
||||
.acquire_exact_connection_with_lease_for_session(
|
||||
&input,
|
||||
&SocketAddr::from(([172, 18, 0, 9], 50_000)),
|
||||
false,
|
||||
default_user_priority(),
|
||||
ConnectionKind::Normal,
|
||||
Some(PlaybackLeaseRef::new(old_token, PlaybackKind::LiveHls)),
|
||||
)
|
||||
.ok_or("old binding")?;
|
||||
let old_request = old.playback_request_id.ok_or("old request")?;
|
||||
let old_tag = old.binding_tag.ok_or("old tag")?;
|
||||
manager.release_handle(&old);
|
||||
|
||||
let current = confirmed_alias_request(&manager, &input, &alias, new_token)?;
|
||||
let current_request = current.playback_request_id.ok_or("current request")?;
|
||||
manager.release_handle(¤t);
|
||||
manager.finish_identified_playback_request(new_token, current_request, PlaybackRequestOutcome::Completed);
|
||||
|
||||
// Delayed manifest failures of the older binding arrive after the rebind.
|
||||
assert!(!manager.forget_identified_provider_affinity(old_token, &input, old_tag));
|
||||
manager.finish_identified_playback_request(old_token, old_request, PlaybackRequestOutcome::ProviderFailed);
|
||||
tokio::time::advance(Duration::from_secs(20)).await;
|
||||
assert_eq!(manager.provider_lease_usage(&alias).total(), 0, "lease must have expired");
|
||||
manager.finish_identified_playback_request(old_token, old_request, PlaybackRequestOutcome::ProviderFailed);
|
||||
|
||||
let reentry = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|00000000000000aa", 50_011)?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&alias)));
|
||||
manager.release_handle(&reentry);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn administrative_termination_of_hls_token_ends_affinity_and_lease() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
let token = "client|alice|42|hls|0123456789abcdef";
|
||||
let handle = confirmed_alias_request(&manager, &input, &alias, token)?;
|
||||
let request_id = handle.playback_request_id.ok_or("identified request")?;
|
||||
manager.release_handle(&handle);
|
||||
manager.finish_identified_playback_request(token, request_id, PlaybackRequestOutcome::Completed);
|
||||
assert_eq!(manager.provider_lease_usage(&alias).idle, 1);
|
||||
|
||||
// The bare-owner clear has no delete right for public HLS tokens.
|
||||
manager.clear_provider_reservation(token);
|
||||
assert_eq!(manager.provider_lease_usage(&alias).idle, 1);
|
||||
manager.terminate_identified_playback_owner(token);
|
||||
assert_eq!(manager.provider_lease_usage(&alias).total(), 0);
|
||||
|
||||
let reentry = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|fedcba9876543210", 50_011)?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
manager.release_handle(&reentry);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn stale_hls_token_termination_keeps_newer_binding() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
let old_token = "client|alice|42|hls|0123456789abcdef";
|
||||
let new_token = "client|alice|42|hls|fedcba9876543210";
|
||||
|
||||
let old = confirmed_alias_request(&manager, &input, &alias, old_token)?;
|
||||
let old_request = old.playback_request_id.ok_or("old request")?;
|
||||
manager.release_handle(&old);
|
||||
manager.finish_identified_playback_request(old_token, old_request, PlaybackRequestOutcome::Completed);
|
||||
|
||||
// The retry re-enters the idle lease and starts a new binding incarnation.
|
||||
let new = acquire_live_hls_from_lineup(&manager, &input, new_token, 50_011)?;
|
||||
assert_eq!(new.allocation.get_provider_name(), Some(Arc::clone(&alias)));
|
||||
assert_ne!(new.binding_tag, old.binding_tag);
|
||||
let new_request = new.playback_request_id.ok_or("new request")?;
|
||||
manager.refresh_adaptive_playback_lease(&alias, new_token, PlaybackKind::LiveHls, 15);
|
||||
manager.confirm_identified_playback_activity(new_token, new_request).ok_or("confirmation")?;
|
||||
manager.release_handle(&new);
|
||||
manager.finish_identified_playback_request(new_token, new_request, PlaybackRequestOutcome::Completed);
|
||||
assert_eq!(manager.provider_lease_usage(&alias).idle, 1);
|
||||
|
||||
assert!(!manager.terminate_identified_playback_owner(old_token));
|
||||
assert_eq!(manager.provider_lease_usage(&alias).idle, 1, "stale token must keep the newer lease");
|
||||
assert_eq!(manager.provider_affinity_for_owner(new_token), Some(Arc::clone(&alias)));
|
||||
|
||||
assert!(manager.terminate_identified_playback_owner(new_token));
|
||||
assert_eq!(manager.provider_lease_usage(&alias).total(), 0);
|
||||
assert_eq!(manager.provider_affinity_for_owner(new_token), None);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn terminating_token_of_expired_binding_ends_its_affinity_only() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
let token = "client|alice|42|hls|0123456789abcdef";
|
||||
let handle = confirmed_alias_request(&manager, &input, &alias, token)?;
|
||||
manager.release_handle(&handle);
|
||||
tokio::time::advance(Duration::from_secs(20)).await;
|
||||
assert_eq!(manager.provider_lease_usage(&alias).total(), 0, "lease must have expired");
|
||||
|
||||
assert!(!manager.terminate_identified_playback_owner("client|alice|42|hls|00000000000000aa"));
|
||||
assert_eq!(manager.provider_affinity_for_owner(token), Some(Arc::clone(&alias)));
|
||||
assert!(manager.terminate_identified_playback_owner(token));
|
||||
assert_eq!(manager.provider_affinity_for_owner(token), None);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn finished_retry_tokens_do_not_accumulate_on_an_active_binding() -> Result<(), String> {
|
||||
let app_cfg = create_test_app_config_with_pool(2, 0);
|
||||
let event_manager = Arc::new(EventManager::new());
|
||||
let manager = ActiveProviderManager::new(&app_cfg, &event_manager);
|
||||
let input = "provider_1".intern();
|
||||
let first = "client|alice|42|hls|0000000000000000";
|
||||
// One request stays in flight for the whole run and keeps the binding active.
|
||||
let anchor = acquire_live_hls_from_lineup(&manager, &input, first, 50_000)?;
|
||||
let anchor_request = anchor.playback_request_id.ok_or("anchor request")?;
|
||||
manager.refresh_adaptive_playback_lease(&input, first, PlaybackKind::LiveHls, 15);
|
||||
manager.confirm_identified_playback_activity(first, anchor_request).ok_or("confirmation")?;
|
||||
|
||||
let mut last_token = String::new();
|
||||
for attempt in 1..=1_000_u32 {
|
||||
last_token = format!("client|alice|42|hls|{attempt:016x}");
|
||||
let handle = acquire_live_hls_from_lineup(&manager, &input, &last_token, 50_001)?;
|
||||
assert_eq!(handle.binding_tag, anchor.binding_tag, "retries share the active binding");
|
||||
let request_id = handle.playback_request_id.ok_or("retry request")?;
|
||||
manager.refresh_adaptive_playback_lease(&input, &last_token, PlaybackKind::LiveHls, 15);
|
||||
manager.release_handle(&handle);
|
||||
manager.finish_identified_playback_request(&last_token, request_id, PlaybackRequestOutcome::Completed);
|
||||
tokio::time::advance(Duration::from_secs(1)).await;
|
||||
}
|
||||
|
||||
let claims = manager.write_leases().request_token_claims(super::playback_lease_owner(first));
|
||||
assert!(claims <= 17, "only the anchor and recently finished retries may hold a claim, got {claims}");
|
||||
// The anchor keeps its claim while its request runs; a recent retry still has its cleanup right.
|
||||
assert!(manager.write_leases().request_token_claims(super::playback_lease_owner(first)) >= 2);
|
||||
assert!(manager.terminate_identified_playback_owner(&last_token));
|
||||
manager.release_handle(&anchor);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn affinity_never_forces_grace_over_allocation_while_primary_is_free() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 1);
|
||||
live_hls_playback_on_alias_with_lapsed_lease(
|
||||
&manager,
|
||||
&input,
|
||||
&alias,
|
||||
"client|alice|42|hls|0123456789abcdef",
|
||||
PlaybackRequestOutcome::Completed,
|
||||
)
|
||||
.await?;
|
||||
let alias_holder = manager
|
||||
.acquire_exact_connection_with_lease_for_session(
|
||||
&alias,
|
||||
&SocketAddr::from(([172, 18, 0, 9], 50_020)),
|
||||
false,
|
||||
default_user_priority(),
|
||||
ConnectionKind::Normal,
|
||||
Some(PlaybackLeaseRef::new("client|erin|9|hls|0123456789abcdef", PlaybackKind::LiveHls)),
|
||||
)
|
||||
.ok_or("alias holder")?;
|
||||
|
||||
// Grace is enabled by default and allowed for this request.
|
||||
let reentry = manager
|
||||
.acquire_connection_with_lease_for_session(
|
||||
&input,
|
||||
&SocketAddr::from(([172, 18, 0, 9], 50_011)),
|
||||
true,
|
||||
default_user_priority(),
|
||||
ConnectionKind::Normal,
|
||||
Some(PlaybackLeaseRef::new("client|alice|42|hls|fedcba9876543210", PlaybackKind::LiveHls)),
|
||||
)
|
||||
.ok_or("re-entry")?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
assert!(
|
||||
matches!(reentry.allocation, ProviderAllocation::Available(_)),
|
||||
"a free primary slot beats a grace allocation"
|
||||
);
|
||||
manager.release_handle(&reentry);
|
||||
manager.release_handle(&alias_holder);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn failure_on_fallback_provider_keeps_preference_for_preferred_provider() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 1);
|
||||
let token = "client|alice|42|hls|0123456789abcdef";
|
||||
live_hls_playback_on_alias_with_lapsed_lease(
|
||||
&manager,
|
||||
&input,
|
||||
&alias,
|
||||
token,
|
||||
PlaybackRequestOutcome::Completed,
|
||||
)
|
||||
.await?;
|
||||
let alias_holder = manager
|
||||
.acquire_exact_connection_with_lease_for_session(
|
||||
&alias,
|
||||
&SocketAddr::from(([172, 18, 0, 9], 50_020)),
|
||||
false,
|
||||
default_user_priority(),
|
||||
ConnectionKind::Normal,
|
||||
Some(PlaybackLeaseRef::new("client|erin|9|hls|0123456789abcdef", PlaybackKind::LiveHls)),
|
||||
)
|
||||
.ok_or("alias holder")?;
|
||||
|
||||
// The full alias sends the re-entry to the primary, which then fails before media.
|
||||
let fallback = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|fedcba9876543210", 50_011)?;
|
||||
assert_eq!(fallback.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
let fallback_request = fallback.playback_request_id.ok_or("fallback request")?;
|
||||
manager.release_handle(&fallback);
|
||||
manager.finish_identified_playback_request(
|
||||
"client|alice|42|hls|fedcba9876543210",
|
||||
fallback_request,
|
||||
PlaybackRequestOutcome::FailedBeforeMedia,
|
||||
);
|
||||
|
||||
assert_eq!(manager.provider_affinity_for_owner(token), Some(Arc::clone(&alias)));
|
||||
manager.release_handle(&alias_holder);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn live_ts_playback_records_no_provider_affinity() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
let busy_1 = acquire_live_hls_from_lineup(&manager, &input, "busy-1|bob|7|hls|0123456789abcdef", 50_001)?;
|
||||
let busy_2 = acquire_live_hls_from_lineup(&manager, &input, "busy-2|carol|8|hls|0123456789abcdef", 50_002)?;
|
||||
let owner = "ts-session-token";
|
||||
let handle = manager
|
||||
.acquire_connection_with_lease_for_session(
|
||||
&input,
|
||||
&SocketAddr::from(([172, 18, 0, 9], 50_010)),
|
||||
false,
|
||||
default_user_priority(),
|
||||
ConnectionKind::Normal,
|
||||
Some(PlaybackLeaseRef::new(owner, PlaybackKind::LiveTs)),
|
||||
)
|
||||
.ok_or("ts playback")?;
|
||||
assert_eq!(handle.allocation.get_provider_name(), Some(Arc::clone(&alias)));
|
||||
let request_id = handle.playback_request_id.ok_or("ts request")?;
|
||||
manager.confirm_identified_playback_activity(owner, request_id).ok_or("confirmation")?;
|
||||
manager.release_handle(&handle);
|
||||
manager.finish_identified_playback_request(owner, request_id, PlaybackRequestOutcome::ClientClosed);
|
||||
manager.release_handle(&busy_1);
|
||||
manager.release_handle(&busy_2);
|
||||
|
||||
assert_eq!(manager.provider_affinity_for_owner(owner), None, "live TS has nothing to return to");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn huge_reconnect_and_affinity_windows_never_panic() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
manager.write_leases().set_affinity_ttl(crate::provider_leases::ProviderAffinityTtl(u64::MAX));
|
||||
let token = "client|alice|42|hls|0123456789abcdef";
|
||||
let handle = acquire_live_hls_from_lineup(&manager, &input, token, 50_010)?;
|
||||
let provider = handle.allocation.get_provider_name().ok_or("provider")?;
|
||||
let request_id = handle.playback_request_id.ok_or("request")?;
|
||||
manager.refresh_adaptive_playback_lease(&provider, token, PlaybackKind::LiveHls, u64::MAX);
|
||||
manager.confirm_identified_playback_activity(token, request_id).ok_or("confirmation")?;
|
||||
manager.release_handle(&handle);
|
||||
manager.finish_identified_playback_request(token, request_id, PlaybackRequestOutcome::Completed);
|
||||
assert_eq!(manager.provider_affinity_for_owner(token), Some(provider));
|
||||
assert_eq!(manager.provider_lease_usage(&alias).total(), 0);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn zero_provider_affinity_ttl_ends_affinity_with_reconnect_window() -> Result<(), String> {
|
||||
let app_cfg = create_test_app_config_with_pool(2, 0);
|
||||
let event_manager = Arc::new(EventManager::new());
|
||||
let manager = ActiveProviderManager::new(&app_cfg, &event_manager);
|
||||
manager.write_leases().set_affinity_ttl(crate::provider_leases::ProviderAffinityTtl(0));
|
||||
let input = "provider_1".intern();
|
||||
let alias = "provider_2".intern();
|
||||
live_hls_playback_on_alias_with_lapsed_lease(
|
||||
&manager,
|
||||
&input,
|
||||
&alias,
|
||||
"client|alice|42|hls|0123456789abcdef",
|
||||
PlaybackRequestOutcome::Completed,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let reentry = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|fedcba9876543210", 50_011)?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
manager.release_handle(&reentry);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn live_hls_provider_failure_ends_alias_affinity() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 0);
|
||||
live_hls_playback_on_alias_with_lapsed_lease(
|
||||
&manager,
|
||||
&input,
|
||||
&alias,
|
||||
"client|alice|42|hls|0123456789abcdef",
|
||||
PlaybackRequestOutcome::ProviderFailed,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let reentry = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|fedcba9876543210", 50_011)?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
manager.release_handle(&reentry);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn live_hls_full_affinity_provider_falls_back_to_lineup() -> Result<(), String> {
|
||||
let (manager, input, alias) = alias_pool(2, 1);
|
||||
live_hls_playback_on_alias_with_lapsed_lease(
|
||||
&manager,
|
||||
&input,
|
||||
&alias,
|
||||
"client|alice|42|hls|0123456789abcdef",
|
||||
PlaybackRequestOutcome::Completed,
|
||||
)
|
||||
.await?;
|
||||
let alias_holder = manager
|
||||
.acquire_exact_connection_with_lease_for_session(
|
||||
&alias,
|
||||
&SocketAddr::from(([172, 18, 0, 9], 50_020)),
|
||||
false,
|
||||
default_user_priority(),
|
||||
ConnectionKind::Normal,
|
||||
Some(PlaybackLeaseRef::new("client|erin|9|hls|0123456789abcdef", PlaybackKind::LiveHls)),
|
||||
)
|
||||
.ok_or("alias holder")?;
|
||||
|
||||
let reentry = acquire_live_hls_from_lineup(&manager, &input, "client|alice|42|hls|fedcba9876543210", 50_011)?;
|
||||
assert_eq!(reentry.allocation.get_provider_name(), Some(Arc::clone(&input)));
|
||||
manager.release_handle(&reentry);
|
||||
manager.release_handle(&alias_holder);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn live_hls_socket_cleanup_only_cancels_its_public_session() -> Result<(), String> {
|
||||
let app_cfg = create_test_app_config_with_pool(2, 4);
|
||||
@@ -4891,11 +5461,7 @@ mod tests {
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn live_hls_old_token_cleanup_cannot_mutate_rebound_retry() -> Result<(), String> {
|
||||
let app_cfg = create_test_app_config_with_pool(2, 4);
|
||||
let event_manager = Arc::new(EventManager::new());
|
||||
let manager = ActiveProviderManager::new(&app_cfg, &event_manager);
|
||||
let input = "provider_1".intern();
|
||||
let alias = "provider_2".intern();
|
||||
let (manager, input, alias) = alias_pool(2, 4);
|
||||
let old = "client|alice|42|hls|0123456789abcdef";
|
||||
let new = "client|alice|42|hls|fedcba9876543210";
|
||||
let addr = SocketAddr::from(([172, 18, 0, 9], 55_000));
|
||||
|
||||
@@ -2821,17 +2821,18 @@ impl ActiveUserManager {
|
||||
///
|
||||
/// Unlike `release_unbound_session_reservation`, this terminates regardless of
|
||||
/// whether streams are currently active, and always removes associated streams.
|
||||
pub async fn terminate_session(&self, username: &str, session_token: &str) {
|
||||
/// Returns `true` only when the session existed and was removed by this call.
|
||||
pub async fn terminate_session(&self, username: &str, session_token: &str) -> bool {
|
||||
let (connection_changed, removed_count, promotions) = {
|
||||
let mut user_connections = self.connections.write().await;
|
||||
let Some(connection_data) = user_connections.by_key.get_mut(username) else {
|
||||
return;
|
||||
return false;
|
||||
};
|
||||
|
||||
let Some(session_index) =
|
||||
connection_data.sessions.iter().position(|session| session.token == session_token)
|
||||
else {
|
||||
return;
|
||||
return false;
|
||||
};
|
||||
|
||||
let counted_kind = connection_data.sessions[session_index]
|
||||
@@ -2868,6 +2869,7 @@ impl ActiveUserManager {
|
||||
self.emit_promotion_update(username, action).await;
|
||||
}
|
||||
debug!("Terminated session {session_token} for user {username}, released {removed_count} streams");
|
||||
true
|
||||
}
|
||||
|
||||
/// Terminates all sessions associated with a given socket address for a user.
|
||||
|
||||
@@ -339,7 +339,7 @@ async fn release_connection_parts(
|
||||
if matches!(reason, DisconnectReason::ClientKicked) {
|
||||
for stream_info in &removed.removed_streams {
|
||||
if let Some(session_token) = stream_info.session_token.as_deref() {
|
||||
deps.provider_manager.clear_provider_reservation(session_token);
|
||||
deps.provider_manager.terminate_identified_playback_owner(session_token);
|
||||
}
|
||||
}
|
||||
// Explicitly terminate all sessions for the kicked addr. This expires them
|
||||
@@ -1294,7 +1294,7 @@ impl ConnectionManager {
|
||||
// Provider release and capacity notification are deferred via `release_provider_deferred`.
|
||||
for stream_info in &removed.removed_streams {
|
||||
if let Some(session_token) = stream_info.session_token.as_deref() {
|
||||
self.provider_manager.clear_provider_reservation(session_token);
|
||||
self.provider_manager.terminate_identified_playback_owner(session_token);
|
||||
}
|
||||
}
|
||||
for username in &removed.disconnected_users {
|
||||
|
||||
@@ -39,6 +39,164 @@ use tuliprox_core::model::{
|
||||
/// capacity, independently of the configured reconnect TTL.
|
||||
pub const STARTING_LEASE_TTL_SECS: u64 = 5;
|
||||
|
||||
/// How long a playback keeps preferring its provider after its last confirmed
|
||||
/// media activity, on top of the lease's reconnect window.
|
||||
///
|
||||
/// Affinity reserves no capacity. It only orders the provider candidates of a
|
||||
/// re-entry, so a player that pauses requests longer than the reconnect TTL
|
||||
/// returns to the same account instead of rerunning priority selection.
|
||||
/// Configured by `reverse_proxy.stream.provider_affinity_ttl_secs`.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct ProviderAffinityTtl(pub u64);
|
||||
|
||||
impl Default for ProviderAffinityTtl {
|
||||
fn default() -> Self { Self(shared::defaults::default_provider_affinity_ttl_secs()) }
|
||||
}
|
||||
|
||||
/// Provider preference of one playback owner that outlives its capacity lease.
|
||||
///
|
||||
/// The binding tag and the requests still attached when the lease ended identify
|
||||
/// which cleanups may end the preference, so a delayed failure of an older binding
|
||||
/// cannot erase the preference of its successor.
|
||||
#[derive(Debug, Clone)]
|
||||
struct ProviderAffinity {
|
||||
provider_name: Arc<str>,
|
||||
binding_tag: ProviderBindingTag,
|
||||
orphaned_request_ids: HashSet<PlaybackRequestId>,
|
||||
orphaned_request_tokens: HashSet<Arc<str>>,
|
||||
expires_at: TokioInstant,
|
||||
}
|
||||
|
||||
/// Upper bound for any affinity or token deadline offset, so a large configured TTL
|
||||
/// can never overflow monotonic instant arithmetic.
|
||||
const MAX_DEADLINE_OFFSET_SECS: u64 = 86_400 * 2;
|
||||
|
||||
/// `now + secs`, capped so the addition can never panic.
|
||||
fn deadline_after(now: TokioInstant, secs: u64) -> TokioInstant {
|
||||
now + Duration::from_secs(secs.min(MAX_DEADLINE_OFFSET_SECS))
|
||||
}
|
||||
|
||||
/// Provider preferences of playback owners that outlive their capacity leases.
|
||||
#[derive(Debug, Default)]
|
||||
struct ProviderAffinities {
|
||||
entries: HashMap<Arc<str>, ProviderAffinity>,
|
||||
// Earliest deadline, so pruning scans the map only when an entry can have expired.
|
||||
sweep_at: Option<TokioInstant>,
|
||||
ttl: ProviderAffinityTtl,
|
||||
}
|
||||
|
||||
impl ProviderAffinities {
|
||||
fn with_ttl(ttl: ProviderAffinityTtl) -> Self { Self { ttl, ..Self::default() } }
|
||||
|
||||
/// Records that real media of `lease.owner` flowed from `lease.provider_name`.
|
||||
///
|
||||
/// Only reconnect-capable playback (HLS, DASH, catchup, VOD) returns through
|
||||
/// separate requests; a one-shot live TS response has nothing to return to.
|
||||
fn remember(&mut self, lease: &ProviderSlotLease, now: TokioInstant) {
|
||||
if !lease.kind.is_reconnect_capable() {
|
||||
return;
|
||||
}
|
||||
let expires_at = deadline_after(now, lease.idle_ttl_secs.saturating_add(self.ttl.0));
|
||||
let binding_tag = ProviderBindingTag::new(lease.id, lease.binding_generation);
|
||||
match self.entries.get_mut(&lease.owner) {
|
||||
Some(affinity) => {
|
||||
if affinity.binding_tag != binding_tag {
|
||||
affinity.provider_name = Arc::clone(&lease.provider_name);
|
||||
affinity.binding_tag = binding_tag;
|
||||
affinity.orphaned_request_ids.clear();
|
||||
affinity.orphaned_request_tokens.clear();
|
||||
}
|
||||
affinity.expires_at = expires_at;
|
||||
}
|
||||
None => {
|
||||
self.entries.insert(
|
||||
Arc::clone(&lease.owner),
|
||||
ProviderAffinity {
|
||||
provider_name: Arc::clone(&lease.provider_name),
|
||||
binding_tag,
|
||||
orphaned_request_ids: HashSet::new(),
|
||||
orphaned_request_tokens: HashSet::new(),
|
||||
expires_at,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
if self.sweep_at.is_none_or(|sweep_at| expires_at < sweep_at) {
|
||||
self.sweep_at = Some(expires_at);
|
||||
}
|
||||
}
|
||||
|
||||
/// Keeps the cleanup rights of requests and tokens that were still attached when
|
||||
/// the lease behind the preference ended.
|
||||
fn orphan(&mut self, lease: &ProviderSlotLease) {
|
||||
let Some(affinity) = self.entries.get_mut(&lease.owner) else {
|
||||
return;
|
||||
};
|
||||
if affinity.binding_tag != ProviderBindingTag::new(lease.id, lease.binding_generation) {
|
||||
return;
|
||||
}
|
||||
affinity.orphaned_request_ids.clone_from(&lease.request_ids);
|
||||
// A token released just before the lease ended still held its right then.
|
||||
let ended_at = lease.state.expires_at().min(TokioInstant::now());
|
||||
affinity.orphaned_request_tokens = lease.request_tokens.live_tokens_at(ended_at);
|
||||
}
|
||||
|
||||
fn forget(&mut self, owner: &str) { self.entries.remove(owner); }
|
||||
|
||||
/// Ends the preference only while it points at `provider_name`: a failure on a
|
||||
/// fallback provider says nothing about the preferred one.
|
||||
fn forget_for_provider(&mut self, owner: &str, provider_name: &Arc<str>) {
|
||||
if self.entries.get(owner).is_some_and(|affinity| affinity.provider_name == *provider_name) {
|
||||
self.entries.remove(owner);
|
||||
}
|
||||
}
|
||||
|
||||
fn forget_identified(&mut self, owner: &str, provider_name: &Arc<str>, binding_tag: ProviderBindingTag) -> bool {
|
||||
let matches = self
|
||||
.entries
|
||||
.get(owner)
|
||||
.is_some_and(|affinity| affinity.provider_name == *provider_name && affinity.binding_tag == binding_tag);
|
||||
if matches {
|
||||
self.entries.remove(owner);
|
||||
}
|
||||
matches
|
||||
}
|
||||
|
||||
/// Applies the outcome of a request whose lease already ended. Only a request that
|
||||
/// was still attached to the binding behind the preference may end it.
|
||||
fn finish_orphaned_request(&mut self, owner: &str, request_id: PlaybackRequestId, outcome: PlaybackRequestOutcome) {
|
||||
let Some(affinity) = self.entries.get_mut(owner) else {
|
||||
return;
|
||||
};
|
||||
if affinity.orphaned_request_ids.remove(&request_id) && !outcome.keeps_idle_lease() {
|
||||
self.entries.remove(owner);
|
||||
}
|
||||
}
|
||||
|
||||
fn terminate_orphaned_token(&mut self, owner: &str, token: &str) -> bool {
|
||||
let orphaned = self.entries.get(owner).is_some_and(|affinity| affinity.orphaned_request_tokens.contains(token));
|
||||
if orphaned {
|
||||
self.entries.remove(owner);
|
||||
}
|
||||
orphaned
|
||||
}
|
||||
|
||||
fn provider_for(&self, owner: &str, now: TokioInstant) -> Option<Arc<str>> {
|
||||
self.entries
|
||||
.get(owner)
|
||||
.filter(|affinity| affinity.expires_at > now)
|
||||
.map(|affinity| Arc::clone(&affinity.provider_name))
|
||||
}
|
||||
|
||||
fn prune(&mut self, now: TokioInstant) {
|
||||
if self.sweep_at.is_none_or(|sweep_at| sweep_at > now) {
|
||||
return;
|
||||
}
|
||||
self.entries.retain(|_, affinity| affinity.expires_at > now);
|
||||
self.sweep_at = self.entries.values().map(|affinity| affinity.expires_at).min();
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum ProviderLeaseState {
|
||||
Starting { expires_at: TokioInstant },
|
||||
@@ -72,6 +230,79 @@ impl ProviderLeaseState {
|
||||
}
|
||||
}
|
||||
|
||||
/// Delete right of one public session token on a binding incarnation.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
struct TokenClaim {
|
||||
active_requests: usize,
|
||||
/// Deadline after the token's last request finished. An idle session keeps its
|
||||
/// cleanup right for the reconnect window; a token without requests and past this
|
||||
/// deadline is dropped so retries cannot grow the set without bound.
|
||||
expires_at: TokioInstant,
|
||||
}
|
||||
|
||||
impl TokenClaim {
|
||||
#[inline]
|
||||
fn is_live_at(&self, at: TokioInstant) -> bool { self.active_requests > 0 || self.expires_at > at }
|
||||
}
|
||||
|
||||
/// Public session tokens that acquired a binding incarnation, keyed by the requests
|
||||
/// that hold them.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
struct RequestTokenClaims {
|
||||
claims: HashMap<Arc<str>, TokenClaim>,
|
||||
by_request: HashMap<PlaybackRequestId, Arc<str>>,
|
||||
}
|
||||
|
||||
impl RequestTokenClaims {
|
||||
fn clear(&mut self) {
|
||||
self.claims.clear();
|
||||
self.by_request.clear();
|
||||
}
|
||||
|
||||
fn attach(&mut self, request_id: PlaybackRequestId, token: &str, now: TokioInstant) {
|
||||
self.prune(now);
|
||||
if self.by_request.contains_key(&request_id) {
|
||||
return;
|
||||
}
|
||||
let token = match self.claims.get_key_value(token) {
|
||||
Some((key, _)) => Arc::clone(key),
|
||||
None => Arc::from(token),
|
||||
};
|
||||
self.claims
|
||||
.entry(Arc::clone(&token))
|
||||
.or_insert(TokenClaim { active_requests: 0, expires_at: now })
|
||||
.active_requests += 1;
|
||||
self.by_request.insert(request_id, token);
|
||||
}
|
||||
|
||||
fn release(&mut self, request_id: PlaybackRequestId, now: TokioInstant, idle_ttl_secs: u64) {
|
||||
let Some(token) = self.by_request.remove(&request_id) else {
|
||||
return;
|
||||
};
|
||||
if let Some(claim) = self.claims.get_mut(&token) {
|
||||
claim.active_requests = claim.active_requests.saturating_sub(1);
|
||||
if claim.active_requests == 0 {
|
||||
claim.expires_at = deadline_after(now, idle_ttl_secs.max(STARTING_LEASE_TTL_SECS));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn prune(&mut self, now: TokioInstant) { self.claims.retain(|_, claim| claim.is_live_at(now)); }
|
||||
|
||||
fn holds(&self, token: &str, now: TokioInstant) -> bool {
|
||||
self.claims.get(token).is_some_and(|claim| claim.is_live_at(now))
|
||||
}
|
||||
|
||||
/// Tokens whose claim was still valid when the lease ended at `ended_at`.
|
||||
fn live_tokens_at(&self, ended_at: TokioInstant) -> HashSet<Arc<str>> {
|
||||
self.claims
|
||||
.iter()
|
||||
.filter(|(_, claim)| claim.active_requests > 0 || claim.expires_at >= ended_at)
|
||||
.map(|(token, _)| Arc::clone(token))
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ProviderSlotLease {
|
||||
pub id: PlaybackLeaseId,
|
||||
@@ -86,6 +317,10 @@ pub struct ProviderSlotLease {
|
||||
/// Requests attached to the current binding incarnation. Parallel segment and
|
||||
/// range requests remain valid until that incarnation is explicitly rebound.
|
||||
request_ids: HashSet<PlaybackRequestId>,
|
||||
/// Public session tokens (for example per-retry HLS tokens) that acquired the
|
||||
/// current binding incarnation. An administrative end of one of these tokens may
|
||||
/// end this binding; a token of an older incarnation may not.
|
||||
request_tokens: RequestTokenClaims,
|
||||
/// Reconnect window configured for this playback; `0` means the endpoint wants no
|
||||
/// reservation, so the lease never blocks foreign capacity.
|
||||
pub idle_ttl_secs: u64,
|
||||
@@ -152,6 +387,7 @@ pub struct ProviderLeaseTable {
|
||||
// Exactly one deadline per lease; renewals replace entries rather than leaving
|
||||
// stale heap entries that grow with the number of segment requests.
|
||||
expirations: BTreeSet<(TokioInstant, PlaybackLeaseId)>,
|
||||
affinity: ProviderAffinities,
|
||||
}
|
||||
|
||||
impl ProviderLeaseTable {
|
||||
@@ -162,7 +398,14 @@ impl ProviderLeaseTable {
|
||||
pub fn len(&self) -> usize { self.leases.len() }
|
||||
|
||||
/// Removes every lease during terminal server shutdown.
|
||||
pub fn clear(&mut self) { *self = Self::default(); }
|
||||
pub fn clear(&mut self) { *self = Self::with_affinity_ttl(self.affinity.ttl); }
|
||||
|
||||
pub fn with_affinity_ttl(affinity_ttl: ProviderAffinityTtl) -> Self {
|
||||
Self { affinity: ProviderAffinities::with_ttl(affinity_ttl), ..Self::default() }
|
||||
}
|
||||
|
||||
/// Applies a reloaded affinity window to affinities recorded from now on.
|
||||
pub fn set_affinity_ttl(&mut self, affinity_ttl: ProviderAffinityTtl) { self.affinity.ttl = affinity_ttl; }
|
||||
|
||||
pub fn lease(&self, id: PlaybackLeaseId) -> Option<&ProviderSlotLease> { self.leases.get(&id) }
|
||||
|
||||
@@ -191,6 +434,7 @@ impl ProviderLeaseTable {
|
||||
let lease = self.leases.remove(&id)?;
|
||||
self.expirations.remove(&(lease.state.expires_at(), id));
|
||||
Self::deindex(&mut self.by_owner, &mut self.by_provider, &lease);
|
||||
self.affinity.orphan(&lease);
|
||||
Some(lease)
|
||||
}
|
||||
|
||||
@@ -207,13 +451,65 @@ impl ProviderLeaseTable {
|
||||
}
|
||||
}
|
||||
|
||||
/// Ends the provider preference of an owner whose playback was ended administratively.
|
||||
pub fn forget_affinity(&mut self, owner: &str) { self.affinity.forget(owner); }
|
||||
|
||||
/// Ends the preference only if it still belongs to the exact provider binding.
|
||||
pub fn forget_identified_affinity(
|
||||
&mut self,
|
||||
owner: &str,
|
||||
provider_name: &Arc<str>,
|
||||
binding_tag: ProviderBindingTag,
|
||||
) -> bool {
|
||||
self.affinity.forget_identified(owner, provider_name, binding_tag)
|
||||
}
|
||||
|
||||
/// Records the public session token that acquired the current binding of a lease.
|
||||
pub fn attach_request_token(&mut self, id: PlaybackLeaseId, request_id: PlaybackRequestId, token: &str) {
|
||||
if let Some(lease) = self.leases.get_mut(&id) {
|
||||
lease.request_tokens.attach(request_id, token, TokioInstant::now());
|
||||
}
|
||||
}
|
||||
|
||||
/// Number of public tokens holding a claim on the owner's current binding.
|
||||
#[cfg(test)]
|
||||
pub(crate) fn request_token_claims(&self, owner: &str) -> usize {
|
||||
self.lease_of_owner(owner).map_or(0, |lease| lease.request_tokens.claims.len())
|
||||
}
|
||||
|
||||
/// Administrative end of the playback that `token` belongs to.
|
||||
///
|
||||
/// Only the binding the token actually acquired is ended: the current lease when
|
||||
/// the token is attached to it, otherwise the provider preference left behind by
|
||||
/// the token's expired binding. A token of an older incarnation never touches a
|
||||
/// newer binding of the same owner.
|
||||
pub fn terminate_identified_token(&mut self, owner: &str, token: &str) -> bool {
|
||||
if let Some(id) = self.by_owner.get(owner).copied() {
|
||||
let now = TokioInstant::now();
|
||||
if self.leases.get(&id).is_some_and(|lease| lease.request_tokens.holds(token, now)) {
|
||||
self.forget_affinity(owner);
|
||||
self.remove_lease(id);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
self.affinity.terminate_orphaned_token(owner, token)
|
||||
}
|
||||
|
||||
/// The provider a recently confirmed playback of `owner` should return to.
|
||||
///
|
||||
/// Unlike [`Self::provider_for_owner`] this survives lease expiry and never
|
||||
/// implies a capacity reservation.
|
||||
pub fn affinity_provider_for_owner(&self, owner: &str) -> Option<Arc<str>> {
|
||||
self.affinity.provider_for(owner, TokioInstant::now())
|
||||
}
|
||||
|
||||
fn startup_expiry(now: TokioInstant, idle_ttl_secs: u64) -> TokioInstant {
|
||||
let ttl = if idle_ttl_secs == 0 { STARTING_LEASE_TTL_SECS } else { STARTING_LEASE_TTL_SECS.min(idle_ttl_secs) };
|
||||
now + Duration::from_secs(ttl)
|
||||
}
|
||||
|
||||
fn confirmed_expiry(now: TokioInstant, idle_ttl_secs: u64) -> TokioInstant {
|
||||
now + Duration::from_secs(idle_ttl_secs.max(1))
|
||||
deadline_after(now, idle_ttl_secs.max(1))
|
||||
}
|
||||
|
||||
/// Claims a slot for one playback at allocation time, before any media flows.
|
||||
@@ -239,12 +535,14 @@ impl ProviderLeaseTable {
|
||||
Self::rebind_lease_provider(&mut self.by_owner, &mut self.by_provider, lease, provider_name);
|
||||
lease.binding_generation = lease.binding_generation.saturating_add(1);
|
||||
lease.request_ids.clear();
|
||||
lease.request_tokens.clear();
|
||||
lease.created_at = now;
|
||||
lease.idle_ttl_secs = 0;
|
||||
lease.state = ProviderLeaseState::Starting { expires_at: Self::startup_expiry(now, 0) };
|
||||
} else if lease.state.is_idle() {
|
||||
lease.binding_generation = lease.binding_generation.saturating_add(1);
|
||||
lease.request_ids.clear();
|
||||
lease.request_tokens.clear();
|
||||
lease.created_at = now;
|
||||
lease.state =
|
||||
ProviderLeaseState::Starting { expires_at: Self::startup_expiry(now, lease.idle_ttl_secs) };
|
||||
@@ -273,6 +571,7 @@ impl ProviderLeaseTable {
|
||||
request_id,
|
||||
binding_generation: 1,
|
||||
request_ids: HashSet::from([request_id]),
|
||||
request_tokens: RequestTokenClaims::default(),
|
||||
idle_ttl_secs: 0,
|
||||
created_at: now,
|
||||
last_activity_at: now,
|
||||
@@ -329,6 +628,9 @@ impl ProviderLeaseTable {
|
||||
ProviderLeaseState::Starting { expires_at: Self::startup_expiry(lease.created_at, idle_ttl_secs) }
|
||||
};
|
||||
self.expirations.insert((lease.state.expires_at(), id));
|
||||
if was_confirmed {
|
||||
self.affinity.remember(lease, now);
|
||||
}
|
||||
return Some(id);
|
||||
}
|
||||
None
|
||||
@@ -418,6 +720,7 @@ impl ProviderLeaseTable {
|
||||
expires_at: Self::confirmed_expiry(now, lease.idle_ttl_secs),
|
||||
};
|
||||
self.expirations.insert((lease.state.expires_at(), id));
|
||||
self.affinity.remember(lease, now);
|
||||
Some(id)
|
||||
}
|
||||
|
||||
@@ -434,7 +737,7 @@ impl ProviderLeaseTable {
|
||||
) -> Option<ProviderSlotLease> {
|
||||
let id = self.by_owner.get(owner).copied()?;
|
||||
let now = TokioInstant::now();
|
||||
let keep_idle = {
|
||||
let (keep_idle, provider_name) = {
|
||||
let lease = self.leases.get_mut(&id)?;
|
||||
lease.last_activity_at = now;
|
||||
let ttl = idle_ttl_secs.unwrap_or(lease.idle_ttl_secs);
|
||||
@@ -445,11 +748,16 @@ impl ProviderLeaseTable {
|
||||
if keep {
|
||||
self.expirations.remove(&(lease.state.expires_at(), id));
|
||||
lease.idle_ttl_secs = ttl;
|
||||
lease.state = ProviderLeaseState::Idle { expires_at: now + Duration::from_secs(ttl) };
|
||||
lease.state = ProviderLeaseState::Idle { expires_at: deadline_after(now, ttl) };
|
||||
self.expirations.insert((lease.state.expires_at(), id));
|
||||
self.affinity.remember(lease, now);
|
||||
}
|
||||
keep
|
||||
(keep, Arc::clone(&lease.provider_name))
|
||||
};
|
||||
if !outcome.keeps_idle_lease() {
|
||||
// A failed or forcibly ended playback may fall back to another provider.
|
||||
self.affinity.forget_for_provider(owner, &provider_name);
|
||||
}
|
||||
if keep_idle {
|
||||
return None;
|
||||
}
|
||||
@@ -496,13 +804,22 @@ impl ProviderLeaseTable {
|
||||
outcome: PlaybackRequestOutcome,
|
||||
idle_ttl_secs: Option<u64>,
|
||||
) -> Option<ProviderSlotLease> {
|
||||
let id = self.by_owner.get(owner).copied()?;
|
||||
let Some(id) = self.by_owner.get(owner).copied() else {
|
||||
self.affinity.finish_orphaned_request(owner, request_id, outcome);
|
||||
return None;
|
||||
};
|
||||
if !self.leases.get(&id).is_some_and(|lease| lease.contains_request(request_id)) {
|
||||
// A request of the binding that preceded the current lease.
|
||||
self.affinity.finish_orphaned_request(owner, request_id, outcome);
|
||||
return None;
|
||||
}
|
||||
let now = TokioInstant::now();
|
||||
let keep_idle = {
|
||||
let (keep_idle, provider_name) = {
|
||||
let lease = self.leases.get_mut(&id)?;
|
||||
if !lease.request_ids.remove(&request_id) {
|
||||
return None;
|
||||
}
|
||||
lease.request_tokens.release(request_id, now, idle_ttl_secs.unwrap_or(lease.idle_ttl_secs));
|
||||
if !lease.request_ids.is_empty() {
|
||||
if lease.request_id == request_id {
|
||||
if let Some(next) = lease.request_ids.iter().next().copied() {
|
||||
@@ -520,11 +837,16 @@ impl ProviderLeaseTable {
|
||||
if keep {
|
||||
self.expirations.remove(&(lease.state.expires_at(), id));
|
||||
lease.idle_ttl_secs = ttl;
|
||||
lease.state = ProviderLeaseState::Idle { expires_at: now + Duration::from_secs(ttl) };
|
||||
lease.state = ProviderLeaseState::Idle { expires_at: deadline_after(now, ttl) };
|
||||
self.expirations.insert((lease.state.expires_at(), id));
|
||||
self.affinity.remember(lease, now);
|
||||
}
|
||||
keep
|
||||
(keep, Arc::clone(&lease.provider_name))
|
||||
};
|
||||
if !outcome.keeps_idle_lease() {
|
||||
// A failed or forcibly ended playback may fall back to another provider.
|
||||
self.affinity.forget_for_provider(owner, &provider_name);
|
||||
}
|
||||
if keep_idle {
|
||||
None
|
||||
} else {
|
||||
@@ -543,6 +865,7 @@ impl ProviderLeaseTable {
|
||||
if !lease.request_ids.remove(&request_id) {
|
||||
return None;
|
||||
}
|
||||
lease.request_tokens.release(request_id, TokioInstant::now(), lease.idle_ttl_secs);
|
||||
if let Some(next) = lease.request_ids.iter().next().copied() {
|
||||
if lease.request_id == request_id {
|
||||
lease.request_id = next;
|
||||
@@ -563,6 +886,7 @@ impl ProviderLeaseTable {
|
||||
if !lease.request_ids.remove(&request_id) {
|
||||
return false;
|
||||
}
|
||||
lease.request_tokens.release(request_id, TokioInstant::now(), lease.idle_ttl_secs);
|
||||
if lease.request_id == request_id {
|
||||
if let Some(next) = lease.request_ids.iter().next().copied() {
|
||||
lease.request_id = next;
|
||||
@@ -626,8 +950,9 @@ impl ProviderLeaseTable {
|
||||
.any(|lease| session_owner != Some(lease.owner.as_ref()))
|
||||
}
|
||||
|
||||
/// Removes expired leases of any state.
|
||||
/// Removes expired leases of any state and expired provider affinities.
|
||||
pub fn prune(&mut self, now: TokioInstant) -> Vec<ProviderSlotLease> {
|
||||
self.affinity.prune(now);
|
||||
if self.leases.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
|
||||
@@ -142,6 +142,7 @@ Check:
|
||||
|
||||
- `hls_session_ttl_secs`
|
||||
- `catchup_session_ttl_secs`
|
||||
- `provider_affinity_ttl_secs`
|
||||
- grace configuration
|
||||
|
||||
### Scenario C: "Two users watched the same channel and then both dropped at once"
|
||||
|
||||
@@ -168,6 +168,16 @@ TTL mapping (the reconnect window):
|
||||
Lease end is outcome-driven: provider failure, preemption, kick and timeout release capacity immediately; a clean finish
|
||||
of a reconnect-capable playback keeps the idle window.
|
||||
|
||||
Provider affinity is separate from the lease and reserves no capacity:
|
||||
|
||||
- confirmed media of a reconnect-capable playback records the provider for the reconnect window plus
|
||||
`provider_affinity_ttl_secs`
|
||||
- a re-entry without a live lease pin tries that provider first, without grace; when it has no free slot the normal
|
||||
lineup selection runs
|
||||
- a failure on the preferred provider, preemption, kick, timeout or an administrative terminate ends the preference;
|
||||
a delayed failure or terminate of an older binding is ignored through its binding tag, request IDs and session
|
||||
tokens
|
||||
|
||||
Important:
|
||||
|
||||
- VOD/movie/series still have strict provider affinity on follow-up requests
|
||||
|
||||
@@ -81,11 +81,14 @@ Important values in the reverse proxy block:
|
||||
|
||||
- `hls_session_ttl_secs`
|
||||
- `catchup_session_ttl_secs`
|
||||
- `provider_affinity_ttl_secs`
|
||||
|
||||
In plain language:
|
||||
|
||||
- HLS gets a short recognition window
|
||||
- catchup usually gets a somewhat longer one
|
||||
- after either window, a playback that comes back still prefers the provider account that last delivered its media for
|
||||
`provider_affinity_ttl_secs` (default `120`); this preference holds no connection slot
|
||||
|
||||
## What the session TTL does not mean
|
||||
|
||||
@@ -253,6 +256,13 @@ Check:
|
||||
- `grace_period_millis`
|
||||
- `grace_period_hold_stream`
|
||||
|
||||
### If HLS playback moves between provider accounts after a player pause
|
||||
|
||||
Check:
|
||||
|
||||
- `provider_affinity_ttl_secs` (it must exceed the player's longest request pause)
|
||||
- whether the preferred account is full, which sends the playback to the normal provider order
|
||||
|
||||
### If catchup users often fail during timeline jumps
|
||||
|
||||
Check:
|
||||
|
||||
@@ -306,6 +306,8 @@ If connection handling looks "strange" for your users, check these first:
|
||||
- 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?
|
||||
- Does `provider_affinity_ttl_secs` cover the longest request pause of the player, so a returning playback stays on its
|
||||
provider account?
|
||||
- Is `share_live_streams.mpeg_ts` enabled and are multiple users watching the same channel?
|
||||
- Is the actual provider already at capacity?
|
||||
|
||||
|
||||
@@ -134,6 +134,7 @@ reverse_proxy:
|
||||
grace_period_hold_stream: true
|
||||
hls_session_ttl_secs: 15
|
||||
catchup_session_ttl_secs: 45
|
||||
provider_affinity_ttl_secs: 120
|
||||
shared_burst_buffer_mb: 12
|
||||
cleanup_queue_capacity: 4096
|
||||
recent_eviction_reentry_ttl_ms: 3000
|
||||
@@ -156,6 +157,7 @@ reverse_proxy:
|
||||
| `grace_period_hold_stream` | Bool | `true` | Tuliprox artificially holds back video data to the client, waiting for grace check to finish, so it doesn't trigger provider prematurely. |
|
||||
| `hls_session_ttl_secs` | Int | `15` | Keeps virtual provider slot open between HLS segment (`.ts`) requests to prevent provider bans for "Account Hopping". |
|
||||
| `catchup_session_ttl_secs` | Int | `45` | Same session-holding principle applied to Archive/Catchup TV. See notes on section [Session TTLs for HLS & Catchup](#session-ttls-for-hls-m3u8--catchup) for details. |
|
||||
| `provider_affinity_ttl_secs` | Int | `120` | How long a reconnect-capable playback (HLS, DASH, Catchup, VOD) keeps returning to its last provider account after the reconnect window has ended. It holds no connection slot and never causes a grace over-allocation. `0` ends the preference together with the reconnect window; the maximum is `86400`. See [Session TTLs for HLS & Catchup](#session-ttls-for-hls-m3u8--catchup). |
|
||||
| `shared_burst_buffer_mb` | Int | `12` | Minimum burst buffer size (in MB) used for shared live streams to immediately synchronize new clients without Keyframe dropouts. See notes on section [Shared Live Streams](#shared-live-streams) for details. |
|
||||
| `cleanup_queue_capacity` | Int | `4096` | Maximum number of concurrent cleanup permits available to active response bodies and shared subscribers. Must be at least `1`. If all permits remain held, new stream admission waits for a bounded interval and then returns `503 Service Unavailable` instead of growing memory without limit. |
|
||||
| `recent_eviction_reentry_ttl_ms` | Int | `3000` | Time window after an eviction during which a retry of the evicted playback must not evict its replacement. The guard is scoped to the user, client address and channel, so unrelated clients are not blocked. Suppressed retries end quietly (no `user_connections_exhausted` video and no `ConnectionDenied` event), since they are not a real connection-limit refusal. Raise this for players with slow automatic retries; `0` disables the guard. The cumulative suppression count is exposed as `reentry_suppressed_total` in the server status (`GET /api/v1/status`). |
|
||||
@@ -598,6 +600,14 @@ hopping. Tuliprox therefore keeps a logical, confirmed slot lease for the playba
|
||||
create a capacity reservation.
|
||||
* Archive/Catchup playback uses the same confirmed-lease principle with `catchup_session_ttl_secs: 45`.
|
||||
* Provider errors, preemption, kicks and timeouts release the lease immediately rather than keeping the reconnect window.
|
||||
* `provider_affinity_ttl_secs: 120`: When a player pauses its requests longer than the reconnect window (for example
|
||||
while buffering), the lease expires but the playback still prefers the provider account that last delivered its
|
||||
media for another 120 seconds. This preference holds no slot: other playbacks can still use that account, and if it
|
||||
is full, the playback falls back to the normal provider order. A full preferred account never receives a grace
|
||||
over-allocation while another account has a free slot. Provider errors on the preferred account, preemption, kicks
|
||||
and timeouts end the preference immediately; a failure on a fallback account keeps it. The preference applies to
|
||||
reconnect-capable playback (HLS, DASH, Catchup, VOD), not to one-shot live MPEG-TS responses. `0` ends it together
|
||||
with the reconnect window; values above `86400` are rejected.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -747,6 +747,7 @@
|
||||
"GRACE_PERIOD_TIMEOUT_SECS": "كم ثانية للانتظار قبل إنهاء الاتصالات الزائدة بعد انتهاء فترة السماح (الافتراضي 5).",
|
||||
"HLS_SESSION_TTL_SECS": "كم من الوقت يُحافظ على حجز حساب مزود HLS نشطًا بين طلبات قائمة التشغيل أو المقطع حتى يمكن إعادة استخدام حساب المزود نفسه.",
|
||||
"METRICS_ENABLED": "يُفعّل عدادات النطاق الترددي وبيانات البث المباشر في جدول البث. عطّل هذا لإزالة عبء قياس البث بالكامل.",
|
||||
"PROVIDER_AFFINITY_TTL_SECS": "المدة التي يستمر فيها التشغيل القابل لإعادة الاتصال (HLS وDASH وCatchup وVOD) في العودة إلى المزود الذي قدّم الوسائط آخر مرة بعد انتهاء نافذة إعادة الاتصال. لا يحجز فتحة اتصال ولا يتسبب أبدًا في تجاوز الحد خلال فترة السماح؛ ويمكن للتشغيلات الأخرى استخدام المزود. القيمة 0 تنهي الارتباط مع TTL جلسة HLS/Catchup (الافتراضي 120، الحد الأقصى 86400).",
|
||||
"RETRY": "عدد المرات لإعادة محاولة الاتصال بالمزود عند فشل البث الأولي.",
|
||||
"SHARED_BURST_BUFFER_MB": "حجم المخزن المؤقت الأولي بالميجابايت لتوفير بيانات فورية للعملاء الجدد المنضمين إلى بث مشترك.",
|
||||
"THROTTLE": "يُفعّل تحديد النطاق الترددي للبث الصادر.",
|
||||
@@ -1148,6 +1149,7 @@
|
||||
"INCLUDE_TYPE_IN_URL": "تضمين النوع في URL",
|
||||
"INFINITE": "لا نهائي",
|
||||
"INPUTS": "الإدخالات",
|
||||
"PROVIDER_AFFINITY_TTL_SECS": "TTL ارتباط المزود (ثواني)",
|
||||
"SELECTION_MODE": "وضع التحديد",
|
||||
"WHITELIST_INCLUDE_SELECTED": "القائمة البيضاء (تضمين المحدد)",
|
||||
"BLACKLIST_EXCLUDE_SELECTED": "القائمة السوداء (استبعاد المحدد)",
|
||||
|
||||
@@ -751,6 +751,7 @@
|
||||
"GRACE_PERIOD_TIMEOUT_SECS": "How many seconds to wait before terminating excess connections after grace period ends (default 5).",
|
||||
"HLS_SESSION_TTL_SECS": "How long to keep an HLS provider-account reservation alive between playlist or segment requests so the same provider account can be reused.",
|
||||
"METRICS_ENABLED": "Enables live stream bandwidth and transferred counters in the Streams table. Disable this to remove stream metering overhead entirely.",
|
||||
"PROVIDER_AFFINITY_TTL_SECS": "How long a reconnect-capable playback (HLS, DASH, Catchup, VOD) keeps returning to the provider that last delivered its media after the reconnect window has ended. Holds no connection slot and never causes a grace over-allocation; other playbacks can still use the provider. Set 0 to end affinity together with the HLS/Catchup session TTL (default 120, maximum 86400).",
|
||||
"RETRY": "Number of times to retry connecting to the provider on initial stream failure.",
|
||||
"SHARED_BURST_BUFFER_MB": "Initial buffer size in MB to provide immediate data to new clients joining a shared stream.",
|
||||
"THROTTLE": "Enables bandwidth limiting for outgoing streams.",
|
||||
@@ -1155,6 +1156,7 @@
|
||||
"INCLUDE_TYPE_IN_URL": "Include type in url",
|
||||
"INFINITE": "Infinite",
|
||||
"INPUTS": "Inputs",
|
||||
"PROVIDER_AFFINITY_TTL_SECS": "Provider Affinity TTL (secs)",
|
||||
"SELECTION_MODE": "Selection mode",
|
||||
"WHITELIST_INCLUDE_SELECTED": "Whitelist (include selected)",
|
||||
"BLACKLIST_EXCLUDE_SELECTED": "Blacklist (exclude selected)",
|
||||
|
||||
@@ -710,6 +710,7 @@
|
||||
"GRACE_PERIOD_TIMEOUT_SECS": "Сколько секунд ждать перед завершением избыточных соединений после окончания льготного периода (по умолчанию 5).",
|
||||
"HLS_SESSION_TTL_SECS": "Как долго сохранять резервирование учетной записи провайдера HLS активным между запросами плейлиста или сегмента, чтобы одна и та же учетная запись провайдера могла быть повторно использована.",
|
||||
"METRICS_ENABLED": "Включает счетчики пропускной способности потока и переданных данных в таблице Streams. Отключите это, чтобы полностью убрать накладные расходы на измерение потока.",
|
||||
"PROVIDER_AFFINITY_TTL_SECS": "Как долго воспроизведение с поддержкой переподключения (HLS, DASH, Catchup, VOD) продолжает возвращаться к провайдеру, который последним доставлял медиаданные, после окончания окна переподключения. Не занимает слот подключения и никогда не вызывает превышение лимита в льготный период; другие воспроизведения по-прежнему могут использовать провайдера. 0 завершает привязку вместе с TTL сессии HLS/Catchup (по умолчанию 120, максимум 86400).",
|
||||
"RETRY": "Количество попыток повторного подключения к провайдеру при первоначальном сбое потока.",
|
||||
"SHARED_BURST_BUFFER_MB": "Начальный размер буфера в МБ для предоставления немедленных данных новым клиентам, присоединяющимся к общему потоку.",
|
||||
"THROTTLE": "Включает ограничение пропускной способности для исходящих потоков.",
|
||||
@@ -1098,6 +1099,7 @@
|
||||
"INCLUDE_TYPE_IN_URL": "Включать тип в url",
|
||||
"INFINITE": "Бесконечно",
|
||||
"INPUTS": "Источники",
|
||||
"PROVIDER_AFFINITY_TTL_SECS": "TTL привязки к провайдеру (сек)",
|
||||
"SELECTION_MODE": "Режим выбора",
|
||||
"WHITELIST_INCLUDE_SELECTED": "Белый список (включить выбранные)",
|
||||
"BLACKLIST_EXCLUDE_SELECTED": "Чёрный список (исключить выбранные)",
|
||||
|
||||
@@ -56,6 +56,7 @@ const LABEL_GRACE_PERIOD_TIMEOUT_SECS: &str = "LABEL.GRACE_PERIOD_TIMEOUT_SECS";
|
||||
const LABEL_GRACE_PERIOD_HOLD_STREAM: &str = "LABEL.GRACE_PERIOD_HOLD_STREAM";
|
||||
const LABEL_HLS_SESSION_TTL_SECS: &str = "LABEL.HLS_SESSION_TTL_SECS";
|
||||
const LABEL_CATCHUP_SESSION_TTL_SECS: &str = "LABEL.CATCHUP_SESSION_TTL_SECS";
|
||||
const LABEL_PROVIDER_AFFINITY_TTL_SECS: &str = "LABEL.PROVIDER_AFFINITY_TTL_SECS";
|
||||
const LABEL_THROTTLE_KBPS: &str = "LABEL.THROTTLE_KBPS";
|
||||
const LABEL_STREAM_BUFFER: &str = "LABEL.STREAM_BUFFER";
|
||||
const LABEL_BUFFER_ENABLED: &str = "LABEL.BUFFER_ENABLED";
|
||||
@@ -188,6 +189,7 @@ generate_form_reducer!(
|
||||
GracePeriodHoldStream => grace_period_hold_stream: bool,
|
||||
HlsSessionTtlSecs => hls_session_ttl_secs: u64,
|
||||
CatchupSessionTtlSecs => catchup_session_ttl_secs: u64,
|
||||
ProviderAffinityTtlSecs => provider_affinity_ttl_secs: u64,
|
||||
}
|
||||
);
|
||||
|
||||
@@ -1171,6 +1173,7 @@ pub fn ReverseProxyConfigView() -> Html {
|
||||
<h1>{translate.t(LABEL_STREAM_SESSION)}</h1>
|
||||
{ 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) }
|
||||
{ config_field!(stream_state.form, translate.t(LABEL_PROVIDER_AFFINITY_TTL_SECS), provider_affinity_ttl_secs) }
|
||||
</Card>
|
||||
<Card class="tp__config-view__card">
|
||||
<h1>{translate.t(LABEL_ADMISSION_STRATEGIES)}</h1>
|
||||
@@ -1394,6 +1397,7 @@ pub fn ReverseProxyConfigView() -> Html {
|
||||
<h1>{translate.t(LABEL_STREAM_SESSION)}</h1>
|
||||
{ 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) }
|
||||
{ edit_field_number_u64!(stream_state, translate.t(LABEL_PROVIDER_AFFINITY_TTL_SECS), provider_affinity_ttl_secs, StreamConfigFormAction::ProviderAffinityTtlSecs) }
|
||||
</Card>
|
||||
<Card class="tp__config-view__card">
|
||||
<h1>{translate.t(LABEL_ADMISSION_STRATEGIES)}</h1>
|
||||
|
||||
@@ -6,6 +6,10 @@
|
||||
/// This is the single source of truth for the default.
|
||||
pub const DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS: u64 = 3000;
|
||||
|
||||
/// Upper bound for `provider_affinity_ttl_secs` (one day). Larger windows serve no
|
||||
/// playback purpose and would overflow monotonic deadline arithmetic.
|
||||
pub const MAX_PROVIDER_AFFINITY_TTL_SECS: u64 = 86_400;
|
||||
|
||||
default_eq_fns!(
|
||||
default_resolve_delay_secs, is_default_resolve_delay_secs, u16, 2;
|
||||
default_probe_delay_secs, is_default_probe_delay_secs, u16, 2;
|
||||
@@ -15,6 +19,7 @@ default_eq_fns!(
|
||||
default_shared_subscriber_idle_timeout_secs, is_default_shared_subscriber_idle_timeout_secs, u64, 300;
|
||||
default_grace_period_timeout_secs, is_default_grace_period_timeout_secs, u64, 4;
|
||||
default_catchup_session_ttl_secs, is_default_catchup_session_ttl_secs, u64, 45;
|
||||
default_provider_affinity_ttl_secs, is_default_provider_affinity_ttl_secs, u64, 120;
|
||||
default_connect_timeout_secs, is_default_connect_timeout_secs, u32, 6;
|
||||
default_resource_retry_attempts, is_default_resource_retry_attempts, u32, 3;
|
||||
default_resource_retry_backoff_ms, is_default_resource_retry_backoff_ms, u64, 250;
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
use crate::{
|
||||
defaults::{
|
||||
default_as_true, default_catchup_session_ttl_secs, default_cleanup_queue_capacity, default_grace_period_millis,
|
||||
default_grace_period_timeout_secs, default_hls_session_ttl_secs, default_recent_eviction_reentry_ttl_ms,
|
||||
default_shared_burst_buffer_mb, default_shared_subscriber_idle_timeout_secs,
|
||||
default_stream_buffer_max_bytes_mb, is_default_catchup_session_ttl_secs, is_default_cleanup_queue_capacity,
|
||||
is_default_grace_period_millis, is_default_grace_period_timeout_secs, is_default_hls_session_ttl_secs,
|
||||
default_grace_period_timeout_secs, default_hls_session_ttl_secs, default_provider_affinity_ttl_secs,
|
||||
default_recent_eviction_reentry_ttl_ms, default_shared_burst_buffer_mb,
|
||||
default_shared_subscriber_idle_timeout_secs, default_stream_buffer_max_bytes_mb,
|
||||
is_default_catchup_session_ttl_secs, is_default_cleanup_queue_capacity, is_default_grace_period_millis,
|
||||
is_default_grace_period_timeout_secs, is_default_hls_session_ttl_secs, is_default_provider_affinity_ttl_secs,
|
||||
is_default_recent_eviction_reentry_ttl_ms, is_default_shared_burst_buffer_mb,
|
||||
is_default_shared_subscriber_idle_timeout_secs, is_default_stream_buffer_max_bytes_mb, is_false, is_true,
|
||||
MAX_PROVIDER_AFFINITY_TTL_SECS,
|
||||
},
|
||||
error::TuliproxError,
|
||||
utils::{is_blank_optional_string, parse_to_kbps},
|
||||
@@ -103,6 +105,13 @@ pub struct StreamConfigDto {
|
||||
pub hls_session_ttl_secs: u64,
|
||||
#[serde(default = "default_catchup_session_ttl_secs", skip_serializing_if = "is_default_catchup_session_ttl_secs")]
|
||||
pub catchup_session_ttl_secs: u64,
|
||||
/// Seconds a playback keeps preferring its provider after its last confirmed media,
|
||||
/// on top of the reconnect window. Holds no capacity; `0` ties affinity to the lease.
|
||||
#[serde(
|
||||
default = "default_provider_affinity_ttl_secs",
|
||||
skip_serializing_if = "is_default_provider_affinity_ttl_secs"
|
||||
)]
|
||||
pub provider_affinity_ttl_secs: u64,
|
||||
#[serde(default, skip)]
|
||||
pub throttle_kbps: u64,
|
||||
#[serde(default = "default_shared_burst_buffer_mb", skip_serializing_if = "is_default_shared_burst_buffer_mb")]
|
||||
@@ -144,6 +153,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(),
|
||||
provider_affinity_ttl_secs: default_provider_affinity_ttl_secs(),
|
||||
admission_strategies: None,
|
||||
}
|
||||
}
|
||||
@@ -165,6 +175,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.provider_affinity_ttl_secs == default_provider_affinity_ttl_secs()
|
||||
&& self.admission_strategies.is_none()
|
||||
}
|
||||
|
||||
@@ -206,6 +217,12 @@ impl StreamConfigDto {
|
||||
return Err(TuliproxError::ConfigStream("`cleanup_queue_capacity` must be at least 1".to_string()));
|
||||
}
|
||||
|
||||
if self.provider_affinity_ttl_secs > MAX_PROVIDER_AFFINITY_TTL_SECS {
|
||||
return Err(TuliproxError::ConfigStream(format!(
|
||||
"`provider_affinity_ttl_secs` must not exceed {MAX_PROVIDER_AFFINITY_TTL_SECS} seconds"
|
||||
)));
|
||||
}
|
||||
|
||||
if let Some(strategies) = &self.admission_strategies {
|
||||
if let Err(err) = validate_admission_strategies(strategies, self.grace_period_millis) {
|
||||
return Err(TuliproxError::ConfigStream(err));
|
||||
@@ -287,6 +304,31 @@ mod tests {
|
||||
assert!(dto.prepare().is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn provider_affinity_ttl_defaults_and_round_trips() -> Result<(), serde_json::Error> {
|
||||
let dto: StreamConfigDto = serde_json::from_str("{}")?;
|
||||
assert_eq!(dto.provider_affinity_ttl_secs, 120);
|
||||
assert!(dto.is_empty());
|
||||
assert!(!serde_json::to_string(&dto)?.contains("provider_affinity_ttl_secs"));
|
||||
|
||||
let dto: StreamConfigDto = serde_json::from_str(r#"{"provider_affinity_ttl_secs":0}"#)?;
|
||||
assert_eq!(dto.provider_affinity_ttl_secs, 0);
|
||||
assert!(!dto.is_empty());
|
||||
assert!(serde_json::to_string(&dto)?.contains("\"provider_affinity_ttl_secs\":0"));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn provider_affinity_ttl_is_bounded() {
|
||||
let mut dto = StreamConfigDto { provider_affinity_ttl_secs: u64::MAX, ..StreamConfigDto::default() };
|
||||
assert!(dto.prepare().is_err());
|
||||
let mut dto = StreamConfigDto {
|
||||
provider_affinity_ttl_secs: MAX_PROVIDER_AFFINITY_TTL_SECS,
|
||||
..StreamConfigDto::default()
|
||||
};
|
||||
assert!(dto.prepare().is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_none_strategies_accepted() {
|
||||
let mut dto = StreamConfigDto::default();
|
||||
|
||||
+87
@@ -0,0 +1,87 @@
|
||||
schema_version: 1
|
||||
name: m3u-hls-provider-affinity-after-lease-expiry
|
||||
tuliprox:
|
||||
base_url: http://fixture.invalid
|
||||
execution_mode: isolated_fixture
|
||||
fixture_stream: { share_live_hls: false, share_live_mpeg_ts: false }
|
||||
bootstrap:
|
||||
command: "${env:TULIPROX_TESTKIT_SUT_BINARY}"
|
||||
arguments: ["-s", "-c", "{config_file}", "-i", "{source_file}", "-a", "{api_proxy_file}"]
|
||||
readiness_url: "{base_url}/healthcheck"
|
||||
origin:
|
||||
run_id: m3u-hls-provider-affinity-after-lease-expiry
|
||||
account_limit: 6
|
||||
limit_mode: reject_new
|
||||
actors:
|
||||
- id: viewer
|
||||
agent: local
|
||||
username: viewer
|
||||
user_agent: tuliprox-testkit/viewer
|
||||
client_ip: { mode: forwarded, value: 10.33.0.10 }
|
||||
- id: holder-1
|
||||
agent: local
|
||||
username: holder1
|
||||
user_agent: tuliprox-testkit/holder-1
|
||||
client_ip: { mode: forwarded, value: 10.33.0.11 }
|
||||
- id: holder-2
|
||||
agent: local
|
||||
username: holder2
|
||||
user_agent: tuliprox-testkit/holder-2
|
||||
client_ip: { mode: forwarded, value: 10.33.0.12 }
|
||||
channels:
|
||||
viewer-channel: { origin_marker: 17, protocol: hls }
|
||||
holder-channel-1: { origin_marker: 19, protocol: hls }
|
||||
holder-channel-2: { origin_marker: 23, protocol: hls }
|
||||
steps:
|
||||
# Two other playbacks fill the primary account, so the viewer falls back to the alias.
|
||||
- command_id: holder-1-fills-primary
|
||||
start: { actor: holder-1, playback_id: holder-1, session_group: holder-1, channel: holder-channel-1 }
|
||||
await: { valid_frames: 2 }
|
||||
assert_origin: { latest_request_account: account-a, no_limit_rejections: true }
|
||||
- command_id: holder-2-fills-primary
|
||||
start: { actor: holder-2, playback_id: holder-2, session_group: holder-2, channel: holder-channel-2 }
|
||||
await: { valid_frames: 2 }
|
||||
assert_origin: { latest_request_account: account-a, no_limit_rejections: true }
|
||||
- command_id: viewer-falls-back-to-alias
|
||||
start: { actor: viewer, playback_id: viewer-first, session_group: viewer, channel: viewer-channel }
|
||||
await: { valid_frames: 2 }
|
||||
assert_origin: { latest_request_account: account-b, no_limit_rejections: true }
|
||||
- command_id: viewer-pauses
|
||||
stop: { playback_id: viewer-first }
|
||||
- command_id: stop-holder-1
|
||||
stop: { playback_id: holder-1 }
|
||||
- command_id: stop-holder-2
|
||||
stop: { playback_id: holder-2 }
|
||||
assert_runtime: { active_users: 0, active_user_connections: 0 }
|
||||
# The 2 s reconnect window lapses while the 6 s provider affinity is still alive.
|
||||
- command_id: request-gap-longer-than-reconnect-window
|
||||
pause_millis: 4000
|
||||
- command_id: viewer-returns-to-alias-although-primary-is-free
|
||||
start: { actor: viewer, playback_id: viewer-return, session_group: viewer, channel: viewer-channel }
|
||||
await: { valid_frames: 2 }
|
||||
assert_origin: { latest_request_account: account-b, no_limit_rejections: true }
|
||||
- command_id: viewer-ends-playback
|
||||
stop: { playback_id: viewer-return }
|
||||
assert_runtime: { active_users: 0, active_user_connections: 0 }
|
||||
# Once reconnect window and affinity have both lapsed, priority selection applies again.
|
||||
- command_id: playback-genuinely-ended
|
||||
pause_millis: 10000
|
||||
- command_id: new-playback-uses-free-primary
|
||||
start: { actor: viewer, playback_id: viewer-new, session_group: viewer, channel: viewer-channel }
|
||||
await: { valid_frames: 2 }
|
||||
assert_origin: { latest_request_account: account-a, no_limit_rejections: true }
|
||||
- command_id: stop-new-playback
|
||||
stop: { playback_id: viewer-new }
|
||||
assert_runtime: { active_users: 0, active_user_connections: 0 }
|
||||
policy_contract:
|
||||
user_access_control: true
|
||||
users:
|
||||
viewer: { max_connections: 4 }
|
||||
holder1: { max_connections: 1 }
|
||||
holder2: { max_connections: 1 }
|
||||
admission_strategies: []
|
||||
hls_session_ttl_secs: 2
|
||||
provider_affinity_ttl_secs: 6
|
||||
provider_pool:
|
||||
- { name: account-a, priority: -1, max_connections: 2 }
|
||||
- { name: account-b, max_connections: 4 }
|
||||
@@ -115,6 +115,12 @@ fn render_config(
|
||||
if let Some(ttl_ms) = policy.recent_eviction_reentry_ttl_ms {
|
||||
let _ = writeln!(config, " recent_eviction_reentry_ttl_ms: {ttl_ms}");
|
||||
}
|
||||
if let Some(ttl_secs) = policy.hls_session_ttl_secs {
|
||||
let _ = writeln!(config, " hls_session_ttl_secs: {ttl_secs}");
|
||||
}
|
||||
if let Some(ttl_secs) = policy.provider_affinity_ttl_secs {
|
||||
let _ = writeln!(config, " provider_affinity_ttl_secs: {ttl_secs}");
|
||||
}
|
||||
}
|
||||
let _ = writeln!(
|
||||
config,
|
||||
@@ -560,6 +566,8 @@ mod tests {
|
||||
)]),
|
||||
admission_strategies: Some(vec![crate::oracle::AdmissionStrategy::EvictUserSameIpLatest]),
|
||||
recent_eviction_reentry_ttl_ms: None,
|
||||
hls_session_ttl_secs: Some(2),
|
||||
provider_affinity_ttl_secs: Some(6),
|
||||
grace: None,
|
||||
provider_max_connections: None,
|
||||
provider_pool: Vec::new(),
|
||||
@@ -587,6 +595,8 @@ mod tests {
|
||||
let api_proxy = std::fs::read_to_string(&fixture.api_proxy_file)?;
|
||||
|
||||
assert!(config.contains(fixture.root.to_string_lossy().as_ref()));
|
||||
assert!(config.contains(" hls_session_ttl_secs: 2\n"), "{config}");
|
||||
assert!(config.contains(" provider_affinity_ttl_secs: 6\n"), "{config}");
|
||||
assert!(source.contains("fixture-run"));
|
||||
assert!(api_proxy.contains("fixture-user"));
|
||||
let expected_paths = [
|
||||
|
||||
@@ -52,6 +52,12 @@ pub struct PolicyContract {
|
||||
/// replacement. Used by the reentry-suppression scenario.
|
||||
#[serde(default)]
|
||||
pub recent_eviction_reentry_ttl_ms: Option<u64>,
|
||||
/// Overrides `reverse_proxy.stream.hls_session_ttl_secs`, the HLS reconnect window.
|
||||
#[serde(default)]
|
||||
pub hls_session_ttl_secs: Option<u64>,
|
||||
/// Overrides `reverse_proxy.stream.provider_affinity_ttl_secs`.
|
||||
#[serde(default)]
|
||||
pub provider_affinity_ttl_secs: Option<u64>,
|
||||
#[serde(default)]
|
||||
pub grace: Option<GraceContract>,
|
||||
#[serde(default)]
|
||||
@@ -269,6 +275,10 @@ pub struct Channel {
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct Step {
|
||||
pub command_id: String,
|
||||
/// Idles the scenario for this many milliseconds, for example to let a
|
||||
/// reconnect window lapse between two requests of one playback.
|
||||
#[serde(default)]
|
||||
pub pause_millis: Option<u64>,
|
||||
#[serde(default)]
|
||||
pub expect: ExpectedPlayback,
|
||||
pub start: Option<Start>,
|
||||
@@ -500,6 +510,19 @@ impl Scenario {
|
||||
if !command_ids.insert(step.command_id.as_str()) || step.command_id.is_empty() {
|
||||
return Err(TestkitError::Configuration("command IDs must be unique and non-empty".to_owned()));
|
||||
}
|
||||
if step.pause_millis.is_some() {
|
||||
if !step.expect.is_streaming()
|
||||
|| step.await_frames.is_some()
|
||||
|| step.await_condition.is_some()
|
||||
|| step.assert_origin.is_some()
|
||||
|| step.assert_runtime.is_some()
|
||||
{
|
||||
return Err(TestkitError::Configuration(
|
||||
"pause_millis cannot be combined with playback expectations or checks".to_owned(),
|
||||
));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if step.start.is_some() == step.stop.is_some() {
|
||||
return Err(TestkitError::Configuration(
|
||||
"each step must contain exactly one of start or stop".to_owned(),
|
||||
@@ -682,10 +705,27 @@ fn validate_provider_pool_contract(contract: &PolicyContract) -> Result<(), Test
|
||||
|
||||
fn expand_steps(steps: &[Step], prefix: &str, expanded: &mut Vec<Step>) -> Result<(), TestkitError> {
|
||||
const MAX_REPEAT_COUNT: usize = 1_000;
|
||||
const MAX_PAUSE_MILLIS: u64 = 300_000;
|
||||
for step in steps {
|
||||
if step.command_id.is_empty() {
|
||||
return Err(TestkitError::Configuration("command IDs must be unique and non-empty".to_owned()));
|
||||
}
|
||||
if let Some(pause_millis) = step.pause_millis {
|
||||
if step.start.is_some() || step.stop.is_some() || step.repeat.is_some() {
|
||||
return Err(TestkitError::Configuration(
|
||||
"pause_millis cannot be combined with start, stop or repeat".to_owned(),
|
||||
));
|
||||
}
|
||||
if pause_millis == 0 || pause_millis > MAX_PAUSE_MILLIS {
|
||||
return Err(TestkitError::Configuration(format!(
|
||||
"pause_millis must be between 1 and {MAX_PAUSE_MILLIS}"
|
||||
)));
|
||||
}
|
||||
let mut child = step.clone();
|
||||
child.command_id = format!("{prefix}{}", step.command_id);
|
||||
expanded.push(child);
|
||||
continue;
|
||||
}
|
||||
if let Some(repeat) = &step.repeat {
|
||||
if step.start.is_some() || step.stop.is_some() {
|
||||
return Err(TestkitError::Configuration("repeat cannot be combined with start or stop".to_owned()));
|
||||
@@ -792,6 +832,7 @@ steps:
|
||||
steps: (0..2)
|
||||
.map(|number| Step {
|
||||
command_id: number.to_string(),
|
||||
pause_millis: None,
|
||||
expect: ExpectedPlayback::Streaming,
|
||||
start: Some(Start {
|
||||
actor: "a".to_owned(),
|
||||
@@ -824,6 +865,7 @@ steps:
|
||||
"evict-user-latest.yml",
|
||||
"provider-hard-limit.yml",
|
||||
"m3u-hls-retries-stable-provider-lease.yml",
|
||||
"m3u-hls-provider-affinity-after-lease-expiry.yml",
|
||||
"shared-stream-single-provider-slot.yml",
|
||||
"vod-range-reopen-preserves-three-live.yml",
|
||||
"vod-range-reopen-strict-cap.yml",
|
||||
@@ -838,6 +880,43 @@ steps:
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pause_step_expands_alone_and_rejects_combinations() {
|
||||
let scenario: Scenario = serde_saphyr::from_str(
|
||||
r"
|
||||
schema_version: 1
|
||||
name: pause-validation
|
||||
tuliprox: { base_url: http://example.invalid, execution_mode: existing_instance }
|
||||
actors: [{ id: a, agent: local }]
|
||||
steps:
|
||||
- command_id: wait
|
||||
pause_millis: 1500
|
||||
",
|
||||
)
|
||||
.unwrap();
|
||||
assert!(scenario.validate().is_ok());
|
||||
assert_eq!(scenario.expanded_steps().unwrap()[0].pause_millis, Some(1_500));
|
||||
|
||||
for invalid_step in [
|
||||
"{ command_id: wait, pause_millis: 0 }",
|
||||
"{ command_id: wait, pause_millis: 300001 }",
|
||||
"{ command_id: wait, pause_millis: 10, stop: { playback_id: missing } }",
|
||||
"{ command_id: wait, pause_millis: 10, expect: rejected }",
|
||||
"{ command_id: wait, pause_millis: 10, expect: suppressed }",
|
||||
"{ command_id: wait, pause_millis: 10, await_frames: 0 }",
|
||||
"{ command_id: wait, pause_millis: 10, await: { valid_frames: 1 } }",
|
||||
"{ command_id: wait, pause_millis: 10, await: {} }",
|
||||
"{ command_id: wait, pause_millis: 10, assert_origin: {} }",
|
||||
"{ command_id: wait, pause_millis: 10, assert_runtime: {} }",
|
||||
] {
|
||||
let yaml = format!(
|
||||
"schema_version: 1\nname: pause-invalid\ntuliprox: {{ base_url: http://example.invalid, execution_mode: existing_instance }}\nactors: [{{ id: a, agent: local }}]\nsteps: [{invalid_step}]\n"
|
||||
);
|
||||
let scenario: Scenario = serde_saphyr::from_str(&yaml).unwrap();
|
||||
assert!(scenario.validate().is_err(), "{invalid_step}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stop_must_reference_an_active_playback() {
|
||||
let scenario: Scenario = serde_saphyr::from_str(
|
||||
|
||||
@@ -2674,6 +2674,11 @@ async fn execute_scenario_steps<'a>(
|
||||
}
|
||||
let execution_result = async {
|
||||
for step in steps {
|
||||
if let Some(pause_millis) = step.pause_millis {
|
||||
tokio::time::sleep(Duration::from_millis(pause_millis)).await;
|
||||
events.push((step.command_id.as_str(), "paused"));
|
||||
continue;
|
||||
}
|
||||
if let Some(stop) = &step.stop {
|
||||
if let Some(oracles) = admission_oracles.as_mut() {
|
||||
if let Some(username) = playback_users.remove(&stop.playback_id) {
|
||||
|
||||
@@ -252,6 +252,8 @@ mod tests {
|
||||
users: HashMap::from([("alice".to_owned(), UserPolicy { max_connections: 2, soft_connections: 1 })]),
|
||||
admission_strategies: Some(vec![AdmissionStrategy::EvictUserSameIpLatest]),
|
||||
recent_eviction_reentry_ttl_ms: None,
|
||||
hls_session_ttl_secs: None,
|
||||
provider_affinity_ttl_secs: None,
|
||||
grace: None,
|
||||
provider_max_connections: None,
|
||||
provider_pool: Vec::new(),
|
||||
|
||||
Reference in New Issue
Block a user