Session handling refactor

Cargo.lock

@@ -4905,7 +4905,7 @@ dependencies = [
 
 [[package]]
 name = "tuliprox-testkit"
-version = "3.3.110"
+version = "3.3.113"
 dependencies = [
  "axum",
  "base64 0.23.1",

Cargo.toml

@@ -29,7 +29,6 @@ shared = { version = "3", path = "shared" }
 tuliprox-hls = { version = "3", path = "backend/hls" }
 tuliprox-auth = { version = "3", path = "backend/auth" }
 tuliprox-dvr = { version = "3", path = "backend/dvr" }
-tuliprox = { version = "3", path = "backend/app" }
 tuliprox-mpegts = { version = "3", path = "backend/mpegts" }
 tuliprox-iptv = { version = "3", path = "backend/iptv" }
 tuliprox-parser = { version = "3", path = "backend/parser" }

Makefile

@@ -186,6 +186,30 @@ testkit-e2e: ## Run testkit E2E scenario suite against freshly built SUT
 	@echo "==> Executing testkit scenario suite"
 	@./bin/run-testkit-scenarios.sh
 
+.PHONY: testkit-scenario
+testkit-scenario: ## Run one testkit scenario: make testkit-scenario SCENARIO=<name> [RUN_ID=<id>]
+	@if [ -z "$(SCENARIO)" ]; then \
+		echo "❌ Error: SCENARIO is required."; \
+		echo "Usage: make testkit-scenario SCENARIO=<scenario-name>"; \
+		echo "Available scenarios:"; \
+		ls test/fixtures/testkit/scenarios/*.yml | xargs -n1 basename | sed 's/\.yml$$//' | sed 's/^/  - /'; \
+		exit 1; \
+	fi
+	@if [ ! -f "test/fixtures/testkit/scenarios/$(SCENARIO).yml" ]; then \
+		echo "❌ Unknown scenario: $(SCENARIO)"; \
+		exit 1; \
+	fi
+	@echo "==> Building tuliprox and tuliprox-testkit"
+	@$(CARGO_STABLE) build --package tuliprox --package tuliprox-testkit
+	@mkdir -p testkit-report/$(SCENARIO)
+	@echo "==> Running scenario $(SCENARIO)"
+	@TULIPROX_TESTKIT_SUT_BINARY="$(PROJECT_DIR)/target/debug/tuliprox" \
+		./target/debug/tuliprox-testkit controller \
+		--scenario test/fixtures/testkit/scenarios/$(SCENARIO).yml \
+		--report-directory testkit-report/$(SCENARIO) \
+		$(if $(RUN_ID),--run-id $(RUN_ID),)
+	@echo "==> Report: testkit-report/$(SCENARIO)/summary.txt"
+
 .PHONY: build
 build: ## Build the entire workspace in parallel using detected CPU count
 	@echo "==> Building workspace with $(CARGO_BUILD_JOBS) jobs"

backend/app/Cargo.toml

@@ -122,5 +122,5 @@ proxy-auth-regression = []
 
 
 [[bin]]
-name = "flags_builder"
+name = "flags-builder"
 path = "src/tools/flags_builder.rs"

backend/app/src/api/api_utils/mod.rs

@@ -84,12 +84,12 @@ use tuliprox_hls::api::MAX_HLS_MANIFEST_BYTES;
 use url::Url;
 
 pub(crate) fn resolve_request_url_for_logging<'a>(input: &ConfigInput, stream_url: &'a str) -> Cow<'a, str> {
+    if !is_sanitize_sensitive_info_enabled() {
+        return Cow::Borrowed(stream_url);
+    }
     if is_media_server_playback_url(input, stream_url) {
         return Cow::Borrowed("media-server://<redacted>");
     }
-    if is_sanitize_sensitive_info_enabled() {
-        return Cow::Borrowed(stream_url);
-    }
 
     let provider = input.get_resolve_provider(stream_url);
     if let Ok(url) = Url::parse(stream_url) {
@@ -197,6 +197,9 @@ fn stream_admission_rejected_response(error: StreamAdmissionError, username: &st
         .unwrap_or_else(|_| StatusCode::SERVICE_UNAVAILABLE.into_response())
 }
 
+/// Produces a quiet termination response when a recently evicted stream retries.
+pub fn reentry_suppressed_response() -> axum::response::Response { StatusCode::NO_CONTENT.into_response() }
+
 #[macro_export]
 macro_rules! try_option_bad_request {
     ($option:expr, $msg_is_error:expr, $msg:expr) => {
@@ -517,7 +520,7 @@ async fn activate_session_before_stream_open(
     // Classify based on current session state, not the pre-computed value.
     // If caller passes FollowUp, verify the session is still counted under the guard.
     // A stale FollowUp would bypass admission — reclassify to catch this.
-    let effective_request_class = if let Some(request_class) = request_class {
+    let (effective_request_class, loaded_session) = if let Some(request_class) = request_class {
         if matches!(request_class, PlaybackRequestClass::FollowUp | PlaybackRequestClass::Activate) {
             // Re-read session under the guard to ensure the counted lease is still held or acquired.
             // If it is no longer counted, classify it from the current lifecycle so
@@ -525,21 +528,23 @@ async fn activate_session_before_stream_open(
             // If it became counted, classify it so stale Activate requests don't double count.
             let current_session =
                 app_state.active_users.get_and_update_user_session(&user.username, session_token).await;
-            classify_playback_request(PlaybackRequestFacts {
+            let classified = classify_playback_request(PlaybackRequestFacts {
                 existing_session: current_session.as_ref(),
                 prepare_only: false,
                 terminate: false,
-            })
+            });
+            (classified, Some(current_session))
         } else {
-            request_class
+            (request_class, None)
         }
     } else {
         let existing_session = app_state.active_users.get_and_update_user_session(&user.username, session_token).await;
-        classify_playback_request(PlaybackRequestFacts {
+        let classified = classify_playback_request(PlaybackRequestFacts {
             existing_session: existing_session.as_ref(),
             prepare_only: false,
             terminate: false,
-        })
+        });
+        (classified, Some(existing_session))
     };
     let limits_enabled = app_state.app_config.config.load().user_access_control
         && (user.max_connections > 0 || user.soft_connections > 0);
@@ -553,7 +558,10 @@ async fn activate_session_before_stream_open(
     if connection_permission == UserConnectionPermission::GracePeriod {
         // Materialize grace lifecycle under the guard so the session state is consistent.
         // Determine which grace mode applies by checking the current session state.
-        let current_session = app_state.active_users.get_and_update_user_session(&user.username, session_token).await;
+        let current_session = match loaded_session {
+            Some(session) => session,
+            None => app_state.active_users.get_and_update_user_session(&user.username, session_token).await,
+        };
         let (_, resolved_grace) = match current_session.as_ref().map(|s| &s.lifecycle) {
             Some(crate::api::model::PlaybackLifecycle::PendingProvider { .. }) => {
                 // Session already in PendingProvider — refresh deadline.
@@ -623,10 +631,10 @@ async fn activate_session_before_stream_open(
             }
         };
         return PlaybackActivationResult {
-            admission: crate::api::model::ConnectionAdmission {
-                permission: connection_permission,
-                kind: Some(connection_kind),
-            },
+            admission: crate::api::model::ConnectionAdmission::from_permission(
+                connection_permission,
+                Some(connection_kind),
+            ),
             grace_mode: resolved_grace,
             grace_context: None,
             placeholder_transition_version: None,
@@ -638,10 +646,10 @@ async fn activate_session_before_stream_open(
         || effective_request_class == PlaybackRequestClass::Prepare
     {
         return PlaybackActivationResult {
-            admission: crate::api::model::ConnectionAdmission {
-                permission: connection_permission,
-                kind: Some(connection_kind),
-            },
+            admission: crate::api::model::ConnectionAdmission::from_permission(
+                connection_permission,
+                Some(connection_kind),
+            ),
             grace_mode: None,
             grace_context: None,
             placeholder_transition_version: None,
@@ -688,7 +696,7 @@ async fn activate_session_before_stream_open(
     let grace_mode = result.grace_mode;
     let grace_context = result.grace_context;
 
-    if admission.permission == UserConnectionPermission::GracePeriod {
+    if admission.permission() == UserConnectionPermission::GracePeriod {
         if matches!(grace_mode, Some(crate::api::model::GraceMode::Hold)) {
             // Hold: session waits for provider slot. Does not count until provider is acquired.
             let deadline = current_time_secs().saturating_add(app_state.get_grace_options().timeout_secs);
@@ -2304,8 +2312,8 @@ pub(crate) async fn stream_response(
     )
     .await;
     let grace_mode = activation.grace_mode.or(grace_mode);
-    connection_permission = activation.admission.permission;
-    connection_kind = activation.admission.kind.unwrap_or(connection_kind);
+    connection_permission = activation.admission.permission();
+    connection_kind = activation.admission.kind().unwrap_or(connection_kind);
 
     let allow_shared_reuse =
         connection_permission != UserConnectionPermission::Exhausted || allow_exhausted_shared_reconnect;
@@ -2325,6 +2333,7 @@ pub(crate) async fn stream_response(
                 stream_channel.clone(),
                 session_token,
                 req_headers,
+                activation.placeholder_transition_version,
             )
             .await
             {
@@ -2346,6 +2355,7 @@ pub(crate) async fn stream_response(
                 stream_channel.clone(),
                 session_token,
                 req_headers,
+                activation.placeholder_transition_version,
             )
             .await
             {
@@ -2366,6 +2376,9 @@ pub(crate) async fn stream_response(
                 activation.placeholder_transition_version.is_some(),
             )
             .await;
+        if activation.admission.is_reentry_suppressed() {
+            return reentry_suppressed_response();
+        }
         record_connect_failed_attempt(ConnectFailedAttempt {
             app_state,
             fingerprint,
@@ -3074,7 +3087,7 @@ pub(crate) fn should_pin_provider_for_session(
     )
 }
 
-#[allow(clippy::too_many_arguments)]
+#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
 async fn try_shared_stream_response_if_any(
     app_state: &Arc<AppState>,
     stream_url: &str,
@@ -3085,6 +3098,7 @@ async fn try_shared_stream_response_if_any(
     mut stream_channel: StreamChannel,
     session_token: &str,
     req_headers: &HeaderMap,
+    placeholder_transition_version: Option<u64>,
 ) -> Option<impl IntoResponse> {
     let subscriber_id =
         tuliprox_core::model::SharedSubscriberId::from_stream_uid(app_state.connection_manager.next_stream_uid());
@@ -3105,7 +3119,18 @@ async fn try_shared_stream_response_if_any(
     let (stream, provider, cleanup_capability) = match shared_subscription {
         Ok(Some(subscription)) => subscription,
         Ok(None) => return None,
-        Err(reason) => return Some(stream_admission_rejected_response(reason.into(), &user.username)),
+        Err(reason) => {
+            app_state
+                .active_users
+                .release_unbound_session_reservation(
+                    &user.username,
+                    session_token,
+                    placeholder_transition_version,
+                    placeholder_transition_version.is_some(),
+                )
+                .await;
+            return Some(stream_admission_rejected_response(reason.into(), &user.username));
+        }
     };
     debug_if_enabled!("Using shared stream {}", sanitize_sensitive_info(stream_url));
     if let Some(headers) = app_state.shared_stream_manager.get_shared_state_headers(stream_url).await {
@@ -3120,22 +3145,25 @@ async fn try_shared_stream_response_if_any(
         stream_details.provider_name = provider;
         let socket_bound =
             is_socket_bound_playback_session(stream_channel.item_type, extract_extension_from_url(stream_url));
-        if let Some(provider_name) = stream_details.provider_name.as_deref() {
-            let _ = app_state
-                .active_users
-                .create_user_session(crate::api::model::CreateUserSessionParams {
-                    user,
-                    session_token,
-                    virtual_id: stream_channel.virtual_id,
-                    provider: provider_name,
-                    stream_url,
-                    addr: &fingerprint.addr,
-                    connection_permission: connect_permission,
-                    connection_kind: Some(connection_kind),
-                    socket_bound,
-                })
-                .await;
-        }
+        // A shared origin may have no provider allocation (e.g. a local/unknown owner), in
+        // which case `provider` is `None`. The subscriber must still be tracked as a user
+        // session, so fall back to a stable placeholder name rather than skipping session
+        // creation; `is_over_limit` on an unknown provider is a no-op.
+        let provider_name = stream_details.provider_name.as_deref().unwrap_or("shared");
+        let _ = app_state
+            .active_users
+            .create_user_session(crate::api::model::CreateUserSessionParams {
+                user,
+                session_token,
+                virtual_id: stream_channel.virtual_id,
+                provider: provider_name,
+                stream_url,
+                addr: &fingerprint.addr,
+                connection_permission: connect_permission,
+                connection_kind: Some(connection_kind),
+                socket_bound,
+            })
+            .await;
         stream_channel.shared = true;
         stream_channel.shared_joined_existing = Some(true);
         // Joining an existing origin reuses its meter identity. When metrics are disabled
@@ -3377,8 +3405,8 @@ pub(crate) async fn local_stream_response(
         )
         .await;
         grace_mode = activation.grace_mode;
-        connection_permission = activation.admission.permission;
-        connection_kind = activation.admission.kind.unwrap_or(connection_kind);
+        connection_permission = activation.admission.permission();
+        connection_kind = activation.admission.kind().unwrap_or(connection_kind);
 
         if connection_permission == UserConnectionPermission::Exhausted {
             app_state
@@ -3390,6 +3418,9 @@ pub(crate) async fn local_stream_response(
                     activation.placeholder_transition_version.is_some(),
                 )
                 .await;
+            if activation.admission.is_reentry_suppressed() {
+                return reentry_suppressed_response();
+            }
             return create_custom_video_stream_response(
                 &app_state.provider_stream_ctx(),
                 &fingerprint.addr,

backend/app/src/api/endpoints/hls_api/catchup.rs

@@ -215,8 +215,8 @@ pub(super) async fn admit_recovered_archive_stream(
         false,
     )
     .await;
-    let connection_permission = connection_admission.permission;
-    let connection_kind = connection_admission.kind.or(session.connection_kind);
+    let connection_permission = connection_admission.permission();
+    let connection_kind = connection_admission.kind().or(session.connection_kind);
     session.permission = connection_permission;
     if let Some(connection_kind) = connection_kind {
         session.connection_kind = Some(connection_kind);
@@ -224,6 +224,9 @@ pub(super) async fn admit_recovered_archive_stream(
     if connection_permission == UserConnectionPermission::Exhausted
         || (connection_permission == UserConnectionPermission::GracePeriod && connection_kind.is_none())
     {
+        if connection_admission.is_reentry_suppressed() {
+            return Err(Box::new(crate::api::api_utils::reentry_suppressed_response()));
+        }
         let provider = if session.provider.is_empty() { input.name.clone() } else { session.provider.clone() };
         return Err(Box::new(
             hls_admission_failure_manifest_response(

backend/app/src/api/endpoints/hls_api/manifest.rs

@@ -728,7 +728,7 @@ pub(super) async fn try_reserve_hls_virtual_entry_origin_account_for_redirect(
         false,
     )
     .await;
-    if connection_admission.permission == UserConnectionPermission::Exhausted {
+    if connection_admission.permission() == UserConnectionPermission::Exhausted {
         return false;
     }
 
@@ -743,7 +743,7 @@ pub(super) async fn try_reserve_hls_virtual_entry_origin_account_for_redirect(
     let Some(hls_cache_origin) = build_hls_origin_resolution(input, &origin_playlist_url) else {
         return false;
     };
-    let Some(connection_kind) = connection_admission.kind else {
+    let Some(connection_kind) = connection_admission.kind() else {
         return false;
     };
     let (shared_hls_session_owner, reservation_ttl_secs) = if hls_cache_enabled_for_target(app_state, target) {
@@ -770,7 +770,7 @@ pub(super) async fn try_reserve_hls_virtual_entry_origin_account_for_redirect(
         session_owner,
         PlaybackKind::LiveHls,
         reservation_ttl_secs,
-        connection_admission.permission,
+        connection_admission.permission(),
         connection_kind,
         false,
     )

backend/app/src/api/endpoints/hls_api/segment.rs

@@ -139,16 +139,6 @@ pub(super) async fn prepare_hls_resource_access(
             return Err(Box::new(hls_resource_access_lease_validation_response(&err)));
         }
     };
-    app_state
-        .hls_proxy
-        .sync_session_access_lease_count_and_detach_if_needed(
-            &app_state.active_users,
-            &app_state.active_provider,
-            &session,
-            proxy_session_id,
-            now_ms,
-        )
-        .await;
     reclaim_hls_account_overlap_if_needed(app_state, &session, now_ms).await;
     let Some(lease) =
         app_state.hls_proxy.access_lease_response_snapshot(&access_context.lease_id, proxy_session_id, now_ms).await
@@ -2985,8 +2975,8 @@ pub(super) async fn hls_api_stream_resolved(
                 false,
             )
             .await;
-        let connection_permission = connection_admission.permission;
-        let connection_kind = connection_admission.kind.or(session.connection_kind);
+        let connection_permission = connection_admission.permission();
+        let connection_kind = connection_admission.kind().or(session.connection_kind);
         session.permission = connection_permission;
         if let Some(connection_kind) = connection_kind {
             session.connection_kind = Some(connection_kind);
@@ -2994,6 +2984,9 @@ pub(super) async fn hls_api_stream_resolved(
         if connection_permission == UserConnectionPermission::Exhausted
             || (connection_permission == UserConnectionPermission::GracePeriod && connection_kind.is_none())
         {
+            if connection_admission.is_reentry_suppressed() {
+                return crate::api::api_utils::reentry_suppressed_response();
+            }
             let provider = if session.provider.is_empty() { input.name.clone() } else { session.provider.clone() };
             let stream_channel = resolve_stream_channel(
                 &app_state,

backend/app/src/api/endpoints/hls_api/session.rs

@@ -1022,6 +1022,22 @@ pub(super) async fn prepare_hls_origin_policy_preempt_runtime(
         return Err(HlsOriginRuntimeAcquireError::Fatal(StatusCode::SERVICE_UNAVAILABLE));
     };
 
+    let is_valid_victim = {
+        let victim = candidate.session.read().await;
+        let has_no_active_origin_work =
+            victim.activity.active_origin_work_count.load(std::sync::atomic::Ordering::Acquire) == 0;
+        victim.origin_account_binding.as_ref().is_some_and(|b| {
+            b.account_name == candidate.account_name
+                && b.session_owner == candidate.session_owner
+                && matches!(b.binding_mode, HlsOriginAccountBindingMode::Active)
+                && has_no_active_origin_work
+        })
+    };
+    if !is_valid_victim {
+        debug!("HLS origin policy preemption denied: reason=stale-candidate");
+        return Err(HlsOriginRuntimeAcquireError::Fatal(StatusCode::SERVICE_UNAVAILABLE));
+    }
+
     app_state.active_provider.clear_identified_provider_reservation(
         &candidate.session_owner,
         &candidate.account_name,

backend/app/src/api/endpoints/m3u_api.rs

@@ -5,8 +5,8 @@ use crate::{
             create_playback_session_fingerprint, create_session_fingerprint, force_provider_stream_response,
             get_session_reservation_ttl_secs, get_user_target, get_user_target_by_credentials,
             is_seekable_media_request, is_session_based_playback, is_stream_share_enabled, local_stream_response,
-            redirect, redirect_response, resolve_initial_stalker_playback_url, resource_response,
-            separate_number_and_remainder, should_allow_exhausted_shared_reconnect, stream_response,
+            redirect, redirect_response, reentry_suppressed_response, resolve_initial_stalker_playback_url,
+            resource_response, separate_number_and_remainder, should_allow_exhausted_shared_reconnect, stream_response,
             try_option_bad_request, try_result_bad_request, try_result_not_found, try_unwrap_body, RedirectParams,
             ResourceFetchPolicy,
         },
@@ -226,6 +226,9 @@ pub(in crate::api) async fn m3u_api_stream_loaded(
             false,
         )
         .await;
+        if admission.is_reentry_suppressed() {
+            return reentry_suppressed_response();
+        }
         return local_stream_response(
             fingerprint,
             app_state,
@@ -234,8 +237,8 @@ pub(in crate::api) async fn m3u_api_stream_loaded(
             &input,
             &target,
             &user,
-            admission.permission,
-            admission.kind.unwrap_or(crate::api::model::ConnectionKind::Normal),
+            admission.permission(),
+            admission.kind().unwrap_or(crate::api::model::ConnectionKind::Normal),
             Some(playback_session_token.as_str()),
             Some(request_class),
             true,
@@ -379,9 +382,9 @@ pub(in crate::api) async fn m3u_api_stream_loaded(
         false,
     )
     .await;
-    let connection_permission = connection_admission.permission;
+    let connection_permission = connection_admission.permission();
     let connection_kind = connection_admission
-        .kind
+        .kind()
         .or(user_session.as_ref().and_then(|session| session.connection_kind))
         .unwrap_or(crate::api::model::ConnectionKind::Normal);
     let allow_exhausted_shared_reconnect = should_allow_exhausted_shared_reconnect(
@@ -391,6 +394,9 @@ pub(in crate::api) async fn m3u_api_stream_loaded(
         session_url.as_ref(),
     );
     if connection_permission == UserConnectionPermission::Exhausted && !allow_exhausted_shared_reconnect {
+        if connection_admission.is_reentry_suppressed() {
+            return reentry_suppressed_response();
+        }
         if extension == HLS_EXT {
             return hls_admission_failure_manifest_response(
                 app_state,

backend/app/src/api/endpoints/v1_api.rs

@@ -71,6 +71,7 @@ pub async fn create_status_check(app_state: &Arc<AppState>) -> StatusCheck {
         active_provider_connections,
         active_user_streams,
         cache,
+        reentry_suppressed_total: app_state.active_users.reentry_suppressed_total(),
     }
 }
 async fn status(axum::extract::State(app_state): axum::extract::State<Arc<AppState>>) -> axum::response::Response {

backend/app/src/api/endpoints/xtream_api/mod.rs

@@ -10,8 +10,8 @@ use crate::{
             empty_json_response_as_array, empty_json_response_as_object, force_provider_stream_response,
             get_session_reservation_ttl_secs, get_user_target, get_user_target_by_credentials, internal_server_error,
             is_seekable_media_request, is_session_based_playback, is_stream_share_enabled, local_stream_response,
-            redirect, redirect_response, resolve_initial_stalker_playback_url, resource_response,
-            separate_number_and_remainder, should_allow_exhausted_shared_reconnect, stream_response,
+            redirect, redirect_response, reentry_suppressed_response, resolve_initial_stalker_playback_url,
+            resource_response, separate_number_and_remainder, should_allow_exhausted_shared_reconnect, stream_response,
             try_option_bad_request, try_result_bad_request, try_unwrap_body, RedirectParams, ResourceFetchPolicy,
         },
         endpoints::{
@@ -408,6 +408,9 @@ async fn xtream_player_api_stream(
             false,
         )
         .await;
+        if admission.is_reentry_suppressed() {
+            return reentry_suppressed_response();
+        }
         return local_stream_response(
             fingerprint,
             app_state,
@@ -416,8 +419,8 @@ async fn xtream_player_api_stream(
             &input,
             &target,
             &user,
-            admission.permission,
-            admission.kind.unwrap_or(crate::api::model::ConnectionKind::Normal),
+            admission.permission(),
+            admission.kind().unwrap_or(crate::api::model::ConnectionKind::Normal),
             Some(playback_session_token.as_str()),
             Some(request_class),
             true,
@@ -581,9 +584,9 @@ async fn xtream_player_api_stream(
         false,
     )
     .await;
-    let connection_permission = connection_admission.permission;
+    let connection_permission = connection_admission.permission();
     let connection_kind = connection_admission
-        .kind
+        .kind()
         .or(user_session.as_ref().and_then(|session| session.connection_kind))
         .unwrap_or(crate::api::model::ConnectionKind::Normal);
     let allow_exhausted_shared_reconnect = should_allow_exhausted_shared_reconnect(
@@ -593,6 +596,9 @@ async fn xtream_player_api_stream(
         session_url.as_ref(),
     );
     if connection_permission == UserConnectionPermission::Exhausted && !allow_exhausted_shared_reconnect {
+        if connection_admission.is_reentry_suppressed() {
+            return reentry_suppressed_response();
+        }
         let stream_channel = create_stream_channel_with_type(target.id, &pli, item_type);
         if playback_ext == HLS_EXT {
             return hls_admission_failure_manifest_response(
@@ -671,7 +677,7 @@ async fn xtream_player_api_stream(
             &input,
             req_headers,
             connection_permission,
-            connection_admission.kind,
+            connection_admission.kind(),
             &original_hls_entry_path,
         )
         .await

backend/app/src/api/model/streams/active_client_stream.rs

@@ -1,7 +1,9 @@
 use crate::{
     api::{
         model::{
-            connection_manager::{PROVIDER_END_CLOSED, PROVIDER_END_ERROR, PROVIDER_END_NOT_SET},
+            connection_manager::{
+                PROVIDER_END_CLOSED, PROVIDER_END_ERROR, PROVIDER_END_NOT_SET, PROVIDER_END_PREEMPTED,
+            },
             open_provider_stream_with_lifecycle, uses_direct_body_idle_timeout, AppState, BoxedProviderStream,
             CleanupEvent, ConnectionManager, CustomVideoStreamType, EventManager, MeteringStream,
             PendingProviderWakeSource, ProviderStreamFactoryOptions, ProviderStreamOpenLifecycle, StreamDetails,
@@ -42,39 +44,58 @@ const DIRECT_BODY_SOCKET_ACTIVITY_TOUCH_SECS: u64 = 1;
 /// Stored as `u8` in an `AtomicU8` for lock-free access inside `poll_next`.
 /// Lower numeric values correspond to a live or custom stream; `GracePending`
 /// (255) is a transient sentinel that parks the poll until the grace task resolves.
+///
+/// Discriminants are implicit (declaration order) except for the `255` sentinel.
+/// The byte mapping lives once, in [`StreamMode::try_from`], and is pinned by
+/// `test_stream_mode_byte_values`.
 #[repr(u8)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum StreamMode {
     /// Forward bytes directly from the upstream provider.
-    Inner = 0,
+    Inner,
     /// Show the "user connections exhausted" custom video.
-    UserExhausted = 1,
+    UserExhausted,
     /// Show the "provider connections exhausted" custom video.
-    ProviderExhausted = 2,
+    ProviderExhausted,
     /// Show the "channel unavailable" custom video.
-    ChannelUnavailable = 3,
+    ChannelUnavailable,
     /// Show the provisioning/placeholder custom video while probing for capacity.
-    Provisioning = 4,
+    Provisioning,
     /// Show the "low-priority preempted" custom video.
-    LowPriorityPreempted = 5,
+    LowPriorityPreempted,
+    /// A recently evicted playback retried while this stream held a grace slot and
+    /// every remaining eviction candidate is reentry-protected. The body must end
+    /// without painting a user-visible "connections exhausted" error video.
+    ReentrySuppressed,
     /// Transient: grace-period check is still in progress; `poll_next` must park.
     GracePending = 255,
 }
 
-impl StreamMode {
-    fn from_u8(v: u8) -> Self {
-        match v {
-            0 => Self::Inner,
-            1 => Self::UserExhausted,
-            2 => Self::ProviderExhausted,
-            3 => Self::ChannelUnavailable,
-            4 => Self::Provisioning,
-            5 => Self::LowPriorityPreempted,
-            _ => Self::GracePending,
+impl TryFrom<u8> for StreamMode {
+    type Error = u8;
+
+    /// Decodes the atomic mode flag. Returns `Err(value)` for a byte that was never a
+    /// valid [`StreamMode`]; callers must fail safe instead of inventing a live state
+    /// such as `GracePending`.
+    fn try_from(value: u8) -> Result<Self, Self::Error> {
+        match value {
+            0 => Ok(Self::Inner),
+            1 => Ok(Self::UserExhausted),
+            2 => Ok(Self::ProviderExhausted),
+            3 => Ok(Self::ChannelUnavailable),
+            4 => Ok(Self::Provisioning),
+            5 => Ok(Self::LowPriorityPreempted),
+            6 => Ok(Self::ReentrySuppressed),
+            255 => Ok(Self::GracePending),
+            other => Err(other),
         }
     }
 }
 
+/// Publishes a new mode to the lock-free flag with release ordering. All writers go
+/// through this so the `as u8` cast and the memory ordering live in one place.
+fn store_stream_mode(flag: &AtomicU8, mode: StreamMode) { flag.store(mode as u8, Ordering::Release); }
+
 /// Holds the optional custom video buffers for each error/placeholder scenario.
 /// Using named fields avoids the positional-indexing confusion of a 4-tuple.
 struct CustomVideoBuffers {
@@ -288,15 +309,28 @@ impl ActiveClientStreamState {
         }
     }
 
-    fn custom_video_type_for_mode(mode: StreamMode) -> CustomVideoStreamType {
+    /// Maps a mode to the custom video it serves, if any. `ReentrySuppressed` and the
+    /// transparent forwarding modes deliberately have no custom video: a suppressed
+    /// reentry must never surface a user-visible error clip.
+    fn custom_video_type_for_mode(mode: StreamMode) -> Option<CustomVideoStreamType> {
         match mode {
-            StreamMode::UserExhausted => CustomVideoStreamType::UserConnectionsExhausted,
-            StreamMode::ProviderExhausted => CustomVideoStreamType::ProviderConnectionsExhausted,
-            StreamMode::Provisioning => CustomVideoStreamType::Provisioning,
-            StreamMode::LowPriorityPreempted => CustomVideoStreamType::LowPriorityPreempted,
-            StreamMode::ChannelUnavailable | StreamMode::Inner | StreamMode::GracePending => {
-                CustomVideoStreamType::ChannelUnavailable
-            }
+            StreamMode::UserExhausted => Some(CustomVideoStreamType::UserConnectionsExhausted),
+            StreamMode::ProviderExhausted => Some(CustomVideoStreamType::ProviderConnectionsExhausted),
+            StreamMode::Provisioning => Some(CustomVideoStreamType::Provisioning),
+            StreamMode::LowPriorityPreempted => Some(CustomVideoStreamType::LowPriorityPreempted),
+            StreamMode::ChannelUnavailable => Some(CustomVideoStreamType::ChannelUnavailable),
+            StreamMode::Inner | StreamMode::GracePending | StreamMode::ReentrySuppressed => None,
+        }
+    }
+
+    /// Drops the provider handle at most once (releasing the provider slot
+    /// synchronously) and detaches the provider body stream. The stream claim itself
+    /// is released separately through [`Self::release_user_stream`].
+    fn release_provider_handle_and_detach_body(&mut self) {
+        self.inner = None;
+        if !self.provider_handle_released {
+            self.provider_handle_released = true;
+            drop(self.provider_handle.take());
         }
     }
 
@@ -343,26 +377,8 @@ impl ActiveClientStreamState {
 
     fn release_stream_and_provider_handle_once(&mut self) {
         self.stop_grace_task();
-        // The provider slot is owned by a managed handle and released synchronously on
-        // drop. The stream claim is released through the reserved cleanup right, never a
-        // lossy cleanup message, with the actual provider outcome.
-        if !self.provider_handle_released {
-            self.provider_handle_released = true;
-            drop(self.provider_handle.take());
-        }
-        if self.user_stream_released {
-            return;
-        }
-        self.user_stream_released = true;
-        if let Some(cleanup) = self.request_cleanup.as_mut() {
-            cleanup.finish(
-                self.lease_request_id,
-                self.provider_end_reason.load(Ordering::Relaxed),
-                self.provider_reconnect_count.load(Ordering::Relaxed),
-                self.provider_error_class,
-                self.provider_http_status,
-            );
-        }
+        self.release_provider_handle_and_detach_body();
+        self.release_user_stream();
     }
 
     fn stop_direct_body_idle_timeout(&mut self) {
@@ -420,6 +436,8 @@ impl ActiveClientStreamState {
         self.provider_stopped = true;
         self.preempt_cancelled = None;
         self.stop_grace_task();
+        self.provider_end_reason.store(PROVIDER_END_PREEMPTED, Ordering::Relaxed);
+        self.provider_error_class = Some("preempted");
 
         let is_superseded = self
             .provider_handle
@@ -437,14 +455,14 @@ impl ActiveClientStreamState {
             if !is_superseded && self.custom_video.low_priority_preempted.is_some() {
                 serve_preempted_custom = true;
                 if let Some(flag) = &self.send_custom_stream_flag {
-                    flag.store(StreamMode::LowPriorityPreempted as u8, Ordering::Release);
+                    store_stream_mode(flag, StreamMode::LowPriorityPreempted);
                 } else {
                     // Fallback: create_active_client_stream usually initializes this via stream_grace_period.
                     self.send_custom_stream_flag =
                         Some(Arc::new(AtomicU8::new(StreamMode::LowPriorityPreempted as u8)));
                 }
             } else if let Some(flag) = &self.send_custom_stream_flag {
-                flag.store(StreamMode::Inner as u8, Ordering::Release);
+                store_stream_mode(flag, StreamMode::Inner);
             }
 
             if let Some(waker) = &self.waker {
@@ -480,11 +498,9 @@ impl ActiveClientStreamState {
         self.stop_grace_task();
 
         if self.provider_handle.is_some() {
-            let managed = self.provider_handle.take();
-            self.provider_handle_released = true;
             // Synchronous provider-slot release; the custom-video detail update is a
             // separate, best-effort UI effect.
-            drop(managed);
+            self.release_provider_handle_and_detach_body();
 
             if mode == StreamMode::ChannelUnavailable {
                 if let Some(flag) = &self.send_custom_stream_flag {
@@ -502,32 +518,45 @@ impl ActiveClientStreamState {
             }
 
             let addr = self.fingerprint.addr;
-            // Drop the provider stream immediately instead of replacing with an
-            // allocated empty stream — avoids a heap allocation on every mode switch.
-            self.inner = None;
-
-            let video_type = Self::custom_video_type_for_mode(mode);
             let reason = match mode {
                 StreamMode::ChannelUnavailable => "unavailable provider channel",
                 StreamMode::UserExhausted => "user grace period exhaustion",
                 StreamMode::ProviderExhausted => "provider grace period exhaustion",
                 StreamMode::Provisioning => "provider grace period provisioning",
                 StreamMode::LowPriorityPreempted => "low-priority preemption",
+                StreamMode::ReentrySuppressed => "reentry suppression",
                 StreamMode::Inner | StreamMode::GracePending => "stream mode transition",
             };
             debug_if_enabled!(
                 "Provider stream stopped due to {reason} for {}",
                 sanitize_sensitive_info(&addr.to_string())
             );
-            self.connection_manager.send_cleanup(CleanupEvent::UpdateDetailAndReleaseProvider {
-                addr,
-                stream_uid: self.stream_uid,
-                video_type,
-                handle: None,
-            });
+            if let Some(video_type) = Self::custom_video_type_for_mode(mode) {
+                self.connection_manager.send_cleanup(CleanupEvent::UpdateDetailAndReleaseProvider {
+                    addr,
+                    stream_uid: self.stream_uid,
+                    video_type,
+                    handle: None,
+                });
+            } else {
+                // No custom video for this mode: release the request without publishing
+                // a detail update so no error clip is shown.
+                self.release_user_stream();
+            }
         }
     }
 
+    /// Ends the response body without switching to a custom error video. Used for a
+    /// suppressed reentry retry, where a visible "connections exhausted" video would be
+    /// misleading: the request was declined by the reentry guard, not by a real limit.
+    fn terminate_quietly(&mut self) {
+        self.provider_stopped = true;
+        self.preempt_cancelled = None;
+        self.stop_grace_task();
+        self.release_provider_handle_and_detach_body();
+        self.release_user_stream();
+    }
+
     fn reset_custom_video_timeout(&mut self) {
         self.custom_video_timeout_mode = None;
         self.custom_video_timeout_sleep = None;
@@ -686,9 +715,19 @@ impl Stream for ActiveClientStream {
             }
 
             // 3. Read atomic mode flag (set by grace task or stop_provider_stream)
-            let mode = match &self.state.send_custom_stream_flag {
-                Some(flag) => StreamMode::from_u8(flag.load(Ordering::Acquire)),
-                None => StreamMode::Inner,
+            let Some(mode) = (match &self.state.send_custom_stream_flag {
+                Some(flag) => StreamMode::try_from(flag.load(Ordering::Acquire)).ok(),
+                None => Some(StreamMode::Inner),
+            }) else {
+                // An unknown mode can only come from a corrupted flag. Never map it
+                // onto a live state such as `GracePending`; end the body safely.
+                debug_assert!(false, "unknown stream mode flag");
+                error!(
+                    "Unknown stream mode for {}, terminating stream",
+                    sanitize_sensitive_info(&self.state.fingerprint.addr.to_string())
+                );
+                self.state.terminate_quietly();
+                return Poll::Ready(None);
             };
 
             // Dispatch based on the current streaming phase.
@@ -827,6 +866,17 @@ impl Stream for ActiveClientStream {
                     }
                 }
 
+                // Quiet termination: the reentry guard declined a retry of a recently
+                // evicted playback. End the body without a user-visible error video.
+                StreamMode::ReentrySuppressed => {
+                    info!(
+                        "Suppressing reentry retry for {}, terminating stream",
+                        sanitize_sensitive_info(&self.state.fingerprint.addr.to_string())
+                    );
+                    self.state.terminate_quietly();
+                    return Poll::Ready(None);
+                }
+
                 // Custom video modes: serve the appropriate buffer
                 video_mode => {
                     if self.state.custom_video_timeout_mode != Some(video_mode) {
@@ -1341,32 +1391,46 @@ fn stream_grace_period(request: GracePeriodParams) -> (Option<Arc<AtomicU8>>, Op
                                     grace_kind,
                                 )
                                 .await;
-                            match remaining_result.admission.permission {
+                            match remaining_result.admission.permission() {
                                 shared::model::UserConnectionPermission::Allowed
                                 | shared::model::UserConnectionPermission::GracePeriod => {
                                     // Remaining strategy succeeded — proceed to Inner.
-                                    stream_strategy_flag_copy.store(StreamMode::Inner as u8, Ordering::Release);
+                                    store_stream_mode(&stream_strategy_flag_copy, StreamMode::Inner);
                                     // updated stays false
                                 }
                                 shared::model::UserConnectionPermission::Exhausted => {
-                                    // Remaining strategies exhausted — final UserExhausted.
-                                    stream_strategy_flag_copy.store(StreamMode::UserExhausted as u8, Ordering::Release);
-                                    connection_manager
-                                        .update_stream_detail(
-                                            &fingerprint.addr,
-                                            CustomVideoStreamType::UserConnectionsExhausted,
-                                        )
-                                        .await;
+                                    let suppressed = remaining_result.admission.is_reentry_suppressed();
+                                    if suppressed {
+                                        // Every remaining eviction candidate is reentry-protected,
+                                        // so this is a suppressed retry of a recently evicted
+                                        // playback. End the body quietly: a visible
+                                        // connections-exhausted video would be misleading because
+                                        // the request was declined by the reentry guard, not by a
+                                        // real connection limit.
+                                        store_stream_mode(&stream_strategy_flag_copy, StreamMode::ReentrySuppressed);
+                                        info!(
+                                            "Suppressing reentry retry for recently evicted playback of user {username}"
+                                        );
+                                    } else {
+                                        // Remaining strategies exhausted — final UserExhausted.
+                                        store_stream_mode(&stream_strategy_flag_copy, StreamMode::UserExhausted);
+                                        connection_manager
+                                            .update_stream_detail(
+                                                &fingerprint.addr,
+                                                CustomVideoStreamType::UserConnectionsExhausted,
+                                            )
+                                            .await;
+                                        info!("User connections exhausted for active clients: {username}");
+                                    }
                                     if let Some(id) = shared_subscriber_id {
                                         connection_manager.shared_stream_manager.release_subscriber(id).await;
                                     }
-                                    info!("User connections exhausted for active clients: {username}");
                                     updated = true;
                                 }
                             }
                         } else {
                             // No grace context — immediate UserExhausted.
-                            stream_strategy_flag_copy.store(StreamMode::UserExhausted as u8, Ordering::Release);
+                            store_stream_mode(&stream_strategy_flag_copy, StreamMode::UserExhausted);
                             connection_manager
                                 .update_stream_detail(
                                     &fingerprint.addr,
@@ -1386,7 +1450,7 @@ fn stream_grace_period(request: GracePeriodParams) -> (Option<Arc<AtomicU8>>, Op
                     if let Some(provider_name) = provider_grace_check {
                         if provider_manager.is_over_limit(&provider_name) {
                             if let Some(provisioning_info) = provisioning_info {
-                                stream_strategy_flag_copy.store(StreamMode::Provisioning as u8, Ordering::Release);
+                                store_stream_mode(&stream_strategy_flag_copy, StreamMode::Provisioning);
                                 connection_manager
                                     .update_stream_detail(&fingerprint.addr, CustomVideoStreamType::Provisioning)
                                     .await;
@@ -1411,7 +1475,7 @@ fn stream_grace_period(request: GracePeriodParams) -> (Option<Arc<AtomicU8>>, Op
                                     }
                                 });
                             } else {
-                                stream_strategy_flag_copy.store(StreamMode::ProviderExhausted as u8, Ordering::Release);
+                                store_stream_mode(&stream_strategy_flag_copy, StreamMode::ProviderExhausted);
                                 connection_manager
                                     .update_stream_detail(
                                         &fingerprint.addr,
@@ -1430,7 +1494,7 @@ fn stream_grace_period(request: GracePeriodParams) -> (Option<Arc<AtomicU8>>, Op
                 }
 
                 if !updated {
-                    stream_strategy_flag_copy.store(StreamMode::Inner as u8, Ordering::Release);
+                    store_stream_mode(&stream_strategy_flag_copy, StreamMode::Inner);
                 }
 
                 // Resolve session lifecycle transitions.
@@ -2128,26 +2192,54 @@ mod tests {
 
     #[test]
     fn test_custom_video_type_mapping_for_grace_modes() {
-        assert!(matches!(
+        assert_eq!(
             ActiveClientStreamState::custom_video_type_for_mode(StreamMode::UserExhausted),
-            CustomVideoStreamType::UserConnectionsExhausted
-        ));
-        assert!(matches!(
+            Some(CustomVideoStreamType::UserConnectionsExhausted)
+        );
+        assert_eq!(
             ActiveClientStreamState::custom_video_type_for_mode(StreamMode::ProviderExhausted),
-            CustomVideoStreamType::ProviderConnectionsExhausted
-        ));
-        assert!(matches!(
+            Some(CustomVideoStreamType::ProviderConnectionsExhausted)
+        );
+        assert_eq!(
             ActiveClientStreamState::custom_video_type_for_mode(StreamMode::Provisioning),
-            CustomVideoStreamType::Provisioning
-        ));
-        assert!(matches!(
+            Some(CustomVideoStreamType::Provisioning)
+        );
+        assert_eq!(
             ActiveClientStreamState::custom_video_type_for_mode(StreamMode::LowPriorityPreempted),
-            CustomVideoStreamType::LowPriorityPreempted
-        ));
-        assert!(matches!(
+            Some(CustomVideoStreamType::LowPriorityPreempted)
+        );
+        assert_eq!(
             ActiveClientStreamState::custom_video_type_for_mode(StreamMode::ChannelUnavailable),
-            CustomVideoStreamType::ChannelUnavailable
-        ));
+            Some(CustomVideoStreamType::ChannelUnavailable)
+        );
+        // A suppressed reentry must never map to a user-visible error clip.
+        assert_eq!(ActiveClientStreamState::custom_video_type_for_mode(StreamMode::ReentrySuppressed), None);
+        assert_eq!(ActiveClientStreamState::custom_video_type_for_mode(StreamMode::Inner), None);
+        assert_eq!(ActiveClientStreamState::custom_video_type_for_mode(StreamMode::GracePending), None);
+    }
+
+    #[test]
+    fn test_stream_mode_byte_values_match_the_implicit_discriminants() {
+        // Discriminants are implicit (declaration order) except the 255 sentinel, so
+        // this pins the wire mapping against accidental reordering.
+        assert_eq!(StreamMode::Inner as u8, 0);
+        assert_eq!(StreamMode::UserExhausted as u8, 1);
+        assert_eq!(StreamMode::ProviderExhausted as u8, 2);
+        assert_eq!(StreamMode::ChannelUnavailable as u8, 3);
+        assert_eq!(StreamMode::Provisioning as u8, 4);
+        assert_eq!(StreamMode::LowPriorityPreempted as u8, 5);
+        assert_eq!(StreamMode::ReentrySuppressed as u8, 6);
+        assert_eq!(StreamMode::GracePending as u8, 255);
+    }
+
+    #[test]
+    fn test_stream_mode_try_from_rejects_unknown_values() {
+        assert_eq!(StreamMode::try_from(0), Ok(StreamMode::Inner));
+        assert_eq!(StreamMode::try_from(6), Ok(StreamMode::ReentrySuppressed));
+        assert_eq!(StreamMode::try_from(255), Ok(StreamMode::GracePending));
+        // Unknown values must not be silently mapped onto GracePending.
+        assert_eq!(StreamMode::try_from(7), Err(7));
+        assert_eq!(StreamMode::try_from(200), Err(200));
     }
 
     #[test]
@@ -2217,7 +2309,7 @@ mod tests {
             start_deferred_provider_grace_resolution(&app_state, &provider_name, deferred_addr, None).await;
 
         assert_eq!(
-            StreamMode::from_u8(flag.load(Ordering::Acquire)),
+            StreamMode::try_from(flag.load(Ordering::Acquire)).unwrap(),
             StreamMode::GracePending,
             "provider grace resolution must begin in GracePending while provider capacity is exhausted"
         );
@@ -2230,7 +2322,7 @@ mod tests {
             "provider grace resolution stayed pending after capacity_notify should have fired"
         );
         assert_eq!(
-            StreamMode::from_u8(flag.load(Ordering::Acquire)),
+            StreamMode::try_from(flag.load(Ordering::Acquire)).unwrap(),
             StreamMode::Inner,
             "capacity-notify should resolve provider grace from GracePending to Inner before the deadline"
         );
@@ -2706,7 +2798,7 @@ mod tests {
         .await;
 
         assert_eq!(
-            StreamMode::from_u8(flag.load(Ordering::Acquire)),
+            StreamMode::try_from(flag.load(Ordering::Acquire)).unwrap(),
             StreamMode::GracePending,
             "provider grace resolution must begin in GracePending while provider capacity is exhausted"
         );
@@ -2719,7 +2811,7 @@ mod tests {
             "grace-resolution task should complete once the deadline expires without capacity becoming available"
         );
         assert_eq!(
-            StreamMode::from_u8(flag.load(Ordering::Acquire)),
+            StreamMode::try_from(flag.load(Ordering::Acquire)).unwrap(),
             StreamMode::ProviderExhausted,
             "provider grace resolution should transition from GracePending to ProviderExhausted when the deadline expires"
         );
@@ -2766,6 +2858,7 @@ mod tests {
             shared_burst_buffer_mb: 1,
             shared_subscriber_idle_timeout_secs: 300,
             cleanup_queue_capacity: 4096,
+            recent_eviction_reentry_ttl: std::time::Duration::from_millis(1500),
             admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream]),
         });
 
@@ -2972,7 +3065,7 @@ mod tests {
 
         // Grace should be pending initially.
         assert_eq!(
-            StreamMode::from_u8(flag.as_ref().unwrap().load(Ordering::Acquire)),
+            StreamMode::try_from(flag.as_ref().unwrap().load(Ordering::Acquire)).unwrap(),
             StreamMode::GracePending,
             "user grace should start in GracePending"
         );
@@ -2984,7 +3077,7 @@ mod tests {
         // Remaining strategies are exhausted (only GraceHoldStream was configured, no eviction).
         // The session should expire with UserExhausted.
         assert_eq!(
-            StreamMode::from_u8(flag.as_ref().unwrap().load(Ordering::Acquire)),
+            StreamMode::try_from(flag.as_ref().unwrap().load(Ordering::Acquire)).unwrap(),
             StreamMode::UserExhausted,
             "exhausted remaining strategies should result in UserExhausted"
         );
@@ -3020,6 +3113,233 @@ mod tests {
         );
     }
 
+    #[tokio::test(start_paused = true)]
+    #[allow(clippy::too_many_lines)]
+    async fn test_user_grace_failure_reentry_suppression_terminates_quietly() {
+        // GraceHoldStream admits the request; the remaining EvictUserOldest candidate
+        // is reentry-protected, so the grace task must resolve to ReentrySuppressed
+        // instead of painting the user-visible connections-exhausted video.
+        let app_state = create_test_app_state_with_stream_config(crate::model::StreamConfig {
+            retry: true,
+            metrics_enabled: true,
+            buffer: None,
+            grace_period_millis: 100,
+            grace_period_timeout_secs: 8,
+            grace_period_hold_stream: true,
+            hls_session_ttl_secs: 10,
+            catchup_session_ttl_secs: 10,
+            throttle_str: None,
+            throttle_kbps: 0,
+            shared_burst_buffer_mb: 1,
+            shared_subscriber_idle_timeout_secs: 300,
+            cleanup_queue_capacity: 4096,
+            recent_eviction_reentry_ttl: std::time::Duration::from_millis(1500),
+            admission_strategies: Some(vec![AdmissionStrategy::GraceHoldStream, AdmissionStrategy::EvictUserOldest]),
+        });
+
+        let provider_name = "provider_1".intern();
+        let first_addr: std::net::SocketAddr = "127.0.0.1:55211".parse().unwrap_or_else(|_| unreachable!());
+        let second_addr: std::net::SocketAddr = "127.0.0.1:55212".parse().unwrap_or_else(|_| unreachable!());
+        let preload_addr: std::net::SocketAddr = "127.0.0.1:55213".parse().unwrap_or_else(|_| unreachable!());
+        let first_fingerprint = create_test_fingerprint(first_addr);
+        let second_fingerprint = create_test_fingerprint(second_addr);
+        let preload_fingerprint = create_test_fingerprint(preload_addr);
+
+        let mut user = create_test_user("grace-reentry-user");
+        user.max_connections = 1;
+        user.soft_connections = 0;
+
+        // Two counted Normal streams make the user over limit when grace expires.
+        app_state
+            .active_users
+            .create_user_session(CreateUserSessionParams {
+                user: &user,
+                session_token: "tok-first",
+                virtual_id: 1,
+                provider: provider_name.as_ref(),
+                stream_url: "http://provider-1.example/live/1.ts",
+                addr: &first_addr,
+                connection_permission: UserConnectionPermission::Allowed,
+                connection_kind: Some(crate::api::model::ConnectionKind::Normal),
+                socket_bound: false,
+            })
+            .await;
+        app_state
+            .connection_manager
+            .update_connection(crate::api::model::ConnectionParams {
+                meter_uid: 1,
+                username: "grace-reentry-user",
+                max_connections: 1,
+                soft_connections: 0,
+                connection_kind: crate::api::model::ConnectionKind::Normal,
+                priority: 0,
+                soft_priority: 10,
+                fingerprint: &first_fingerprint,
+                provider: provider_name.clone(),
+                stream_channel: &create_test_stream_channel(1, "http://provider-1.example/live/1.ts"),
+                user_agent: std::borrow::Cow::Borrowed("ua"),
+                session_token: Some("tok-first"),
+            })
+            .await;
+        app_state
+            .active_users
+            .create_user_session(CreateUserSessionParams {
+                user: &user,
+                session_token: "tok-preload",
+                virtual_id: 3,
+                provider: provider_name.as_ref(),
+                stream_url: "http://provider-1.example/live/3.ts",
+                addr: &preload_addr,
+                connection_permission: UserConnectionPermission::Allowed,
+                connection_kind: Some(crate::api::model::ConnectionKind::Normal),
+                socket_bound: false,
+            })
+            .await;
+        app_state
+            .connection_manager
+            .update_connection(crate::api::model::ConnectionParams {
+                meter_uid: 3,
+                username: "grace-reentry-user",
+                max_connections: 1,
+                soft_connections: 0,
+                connection_kind: crate::api::model::ConnectionKind::Normal,
+                priority: 0,
+                soft_priority: 10,
+                fingerprint: &preload_fingerprint,
+                provider: provider_name.clone(),
+                stream_channel: &create_test_stream_channel(3, "http://provider-1.example/live/3.ts"),
+                user_agent: std::borrow::Cow::Borrowed("ua"),
+                session_token: Some("tok-preload"),
+            })
+            .await;
+
+        // Grace session on a distinct channel; its socket-bound stream row is what
+        // `mark_recent_eviction_guard_for_addr` keys the reentry guard on.
+        app_state
+            .active_users
+            .create_user_session(CreateUserSessionParams {
+                user: &user,
+                session_token: "tok-second",
+                virtual_id: 2,
+                provider: provider_name.as_ref(),
+                stream_url: "http://provider-1.example/live/2.ts",
+                addr: &second_addr,
+                connection_permission: UserConnectionPermission::GracePeriod,
+                connection_kind: Some(crate::api::model::ConnectionKind::Soft),
+                socket_bound: false,
+            })
+            .await;
+        app_state
+            .connection_manager
+            .update_connection(crate::api::model::ConnectionParams {
+                meter_uid: 2,
+                username: "grace-reentry-user",
+                max_connections: 1,
+                soft_connections: 0,
+                connection_kind: crate::api::model::ConnectionKind::Normal,
+                priority: 0,
+                soft_priority: 10,
+                fingerprint: &second_fingerprint,
+                provider: provider_name.clone(),
+                stream_channel: &create_test_stream_channel(2, "http://provider-1.example/live/2.ts"),
+                user_agent: std::borrow::Cow::Borrowed("ua"),
+                session_token: Some("tok-second"),
+            })
+            .await;
+
+        // Protect the oldest candidate (channel 1) against a retry of this channel 2.
+        app_state
+            .active_users
+            .mark_recent_eviction_guard_for_addr(&second_addr, first_addr, Duration::from_secs(3))
+            .await;
+
+        let _pending_version = app_state
+            .active_users
+            .mark_pending_provider(
+                "grace-reentry-user",
+                "tok-second",
+                crate::api::model::PendingProviderReason::GraceHold,
+                9_999,
+            )
+            .await
+            .expect("pending version must be created for tok-second");
+
+        let grace_context = GraceResolutionContext {
+            strategy_index: 0,
+            strategies: [AdmissionStrategy::GraceHoldStream, AdmissionStrategy::EvictUserOldest].into(),
+            kind: Some(crate::api::model::ConnectionKind::Soft),
+        };
+        let pending_version = app_state
+            .active_users
+            .pending_provider_version("grace-reentry-user", "tok-second")
+            .await
+            .expect("pending version must be created for tok-second");
+
+        let stream_details = StreamDetails {
+            shared_subscriber_id: None,
+            stream: None,
+            stream_info: None,
+            provider_name: Some(provider_name),
+            request_url: Some("http://provider-1.example/live/2.ts".intern()),
+            session_headers: None,
+            provider_session_headers: HashMap::new(),
+            user_agent_stream_index: None,
+            grace_period: GracePeriodOptions { period_millis: 100, timeout_secs: 0, hold_stream: true },
+            provider_grace_active: false,
+            disable_provider_grace: false,
+            reconnect_flag: None,
+            provider_handle: None,
+            content_representation: ProviderContentRepresentationMode::PreserveOrigin,
+            grace_resolution_context: Some(grace_context.clone()),
+            custom_reason: None,
+        };
+
+        let (flag, grace_task) = stream_grace_period(GracePeriodParams {
+            app_state: Arc::clone(&app_state),
+            stream_details,
+            user_grace_period: true,
+            user: user.clone(),
+            fingerprint: second_fingerprint.clone(),
+            virtual_id: VirtualId::new(2),
+            session_token: Some("tok-second".to_string()),
+            provisioning_info: None,
+            waker: None,
+            hold_stream: true,
+            capacity_notify: app_state.connection_manager.capacity_notified(),
+            pending_provider_version: Some(pending_version),
+            grace_active_version: None,
+            grace_resolution_context: Some(grace_context),
+            grace_kind: Some(crate::api::model::ConnectionKind::Soft),
+            socket_bound: true,
+            shared_subscriber_id: None,
+        });
+
+        assert_eq!(
+            StreamMode::try_from(flag.as_ref().unwrap().load(Ordering::Acquire)).unwrap(),
+            StreamMode::GracePending,
+            "user grace should start in GracePending"
+        );
+
+        tokio::time::advance(Duration::from_millis(101)).await;
+        let _ = grace_task.expect("grace task should be spawned").await;
+
+        assert_eq!(
+            StreamMode::try_from(flag.as_ref().unwrap().load(Ordering::Acquire)).unwrap(),
+            StreamMode::ReentrySuppressed,
+            "reentry-protected remaining candidate must resolve to quiet termination"
+        );
+
+        let session = app_state
+            .active_users
+            .get_and_update_user_session("grace-reentry-user", "tok-second")
+            .await
+            .expect("session must exist after grace failure");
+        assert!(
+            matches!(session.lifecycle, crate::api::model::PlaybackLifecycle::Expired),
+            "session lifecycle should be Expired after reentry-suppressed grace failure"
+        );
+    }
+
     #[tokio::test(start_paused = true)]
     async fn test_active_client_stream_immediate_provider_stream_emits_meter_batches() {
         let app_state = create_test_app_state();

backend/app/src/api/panel_api/mod.rs

@@ -1242,10 +1242,7 @@ pub(crate) fn can_provision_on_exhausted(app_state: &AppState, input: &ConfigInp
         debug_if_enabled!("panel_api config invalid: {}", sanitize_sensitive_info(&err.to_string()));
         return false;
     }
-    if is_alias_pool_max_reached(app_state, input) {
-        return false;
-    }
-    true
+    !is_alias_pool_max_reached(app_state, input)
 }
 
 pub(crate) fn find_input_by_provider_name(app_state: &AppState, provider_name: &str) -> Option<Arc<ConfigInput>> {

backend/core/src/model/config/stream.rs

@@ -1,8 +1,10 @@
 use crate::model::macros;
 use shared::{
+    defaults::DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS,
     model::{AdmissionStrategy, StreamBufferConfigDto, StreamConfigDto},
     utils::parse_to_kbps,
 };
+use std::time::Duration;
 use tuliprox_mpegts::transport_stream_buffer::TransportStreamBuffer;
 
 #[derive(Debug, Clone)]
@@ -41,6 +43,7 @@ pub struct StreamConfig {
     pub shared_burst_buffer_mb: u64,
     pub shared_subscriber_idle_timeout_secs: u64,
     pub cleanup_queue_capacity: usize,
+    pub recent_eviction_reentry_ttl: Duration,
     pub admission_strategies: Option<Vec<AdmissionStrategy>>,
 }
 
@@ -61,6 +64,7 @@ impl Default for StreamConfig {
             shared_burst_buffer_mb: 12,
             shared_subscriber_idle_timeout_secs: 300,
             cleanup_queue_capacity: 4096,
+            recent_eviction_reentry_ttl: Duration::from_millis(DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS),
             admission_strategies: None,
         }
     }
@@ -81,6 +85,7 @@ impl From<&StreamConfigDto> for StreamConfig {
             shared_burst_buffer_mb: dto.shared_burst_buffer_mb,
             shared_subscriber_idle_timeout_secs: dto.shared_subscriber_idle_timeout_secs,
             cleanup_queue_capacity: dto.cleanup_queue_capacity,
+            recent_eviction_reentry_ttl: Duration::from_millis(dto.recent_eviction_reentry_ttl_ms),
             admission_strategies: dto.admission_strategies.clone(),
         }
     }
@@ -102,6 +107,11 @@ impl From<&StreamConfig> for StreamConfigDto {
             shared_burst_buffer_mb: instance.shared_burst_buffer_mb,
             shared_subscriber_idle_timeout_secs: instance.shared_subscriber_idle_timeout_secs,
             cleanup_queue_capacity: instance.cleanup_queue_capacity,
+            recent_eviction_reentry_ttl_ms: instance
+                .recent_eviction_reentry_ttl
+                .as_millis()
+                .try_into()
+                .unwrap_or(DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS),
             admission_strategies: instance.admission_strategies.clone(),
         }
     }
@@ -112,6 +122,18 @@ mod tests {
     use super::StreamConfig;
     use shared::model::{AdmissionStrategy, StreamConfigDto};
 
+    #[test]
+    fn stream_config_defaults_reentry_ttl_to_the_shared_default() {
+        let config = StreamConfig::default();
+        assert_eq!(
+            config.recent_eviction_reentry_ttl,
+            std::time::Duration::from_millis(shared::defaults::DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS)
+        );
+        // The documented default is 3000 ms for backward compatibility.
+        assert_eq!(shared::defaults::DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS, 3000);
+        assert_eq!(shared::defaults::default_recent_eviction_reentry_ttl_ms(), 3000);
+    }
+
     #[test]
     fn stream_config_preserves_missing_admission_strategies() {
         let domain = StreamConfig::from(&StreamConfigDto::default());
@@ -141,6 +163,7 @@ mod tests {
             shared_burst_buffer_mb: 1,
             shared_subscriber_idle_timeout_secs: 300,
             cleanup_queue_capacity: 4096,
+            recent_eviction_reentry_ttl: std::time::Duration::from_millis(2_500),
             admission_strategies: Some(vec![
                 AdmissionStrategy::EvictUserOldest,
                 AdmissionStrategy::GraceHoldStream,
@@ -149,6 +172,7 @@ mod tests {
         };
 
         let dto = StreamConfigDto::from(&domain);
+        assert_eq!(dto.recent_eviction_reentry_ttl_ms, 2_500);
         assert_eq!(
             dto.admission_strategies,
             Some(vec![

backend/core/src/model/provider.rs

@@ -391,11 +391,7 @@ impl ProviderConfig {
         if max == 0 {
             return false;
         }
-        let mut guard = self.write_connection();
-        if guard.current_connections < self.max_connections {
-            guard.grace_started_at = None;
-        }
-
+        let guard = self.read_connection();
         if guard.current_connections > max {
             if Self::grace_is_active(&guard, Instant::now(), grace_period_timeout_secs) {
                 // Grace timeout still active, deny connection
@@ -403,6 +399,13 @@ impl ProviderConfig {
             }
             return true;
         }
+        if guard.current_connections < max && guard.grace_started_at.is_some() {
+            drop(guard);
+            let mut write_guard = self.write_connection();
+            if write_guard.current_connections < max {
+                write_guard.grace_started_at = None;
+            }
+        }
         false
     }
 

backend/hls/src/origin.rs

@@ -167,11 +167,22 @@ impl HlsOriginAccountIoLease {
         if self.acquiring.load(std::sync::atomic::Ordering::Acquire) {
             return false;
         }
-        self.active_io_count
-            .fetch_update(std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire, |count| {
-                (count > 0).then(|| count.checked_add(1)).flatten()
-            })
-            .is_ok()
+        let mut count = self.active_io_count.load(std::sync::atomic::Ordering::Acquire);
+        loop {
+            if count == 0 {
+                return false;
+            }
+            let Some(next) = count.checked_add(1) else { return false };
+            match self.active_io_count.compare_exchange_weak(
+                count,
+                next,
+                std::sync::atomic::Ordering::AcqRel,
+                std::sync::atomic::Ordering::Acquire,
+            ) {
+                Ok(_) => return true,
+                Err(observed) => count = observed,
+            }
+        }
     }
 
     fn take_resource(&self) -> Option<(ProviderHandle, tokio::sync::mpsc::OwnedPermit<CleanupEvent>)> {

backend/processing/src/processor/xtream_series.rs

@@ -120,10 +120,7 @@ async fn playlist_resolve_series_info<E: EventSink + Clone + 'static, M: Metadat
     target: &ConfigTarget,
 ) {
     let filter = |pli: &PlaylistItem| {
-        if pli.header.xtream_cluster != XtreamCluster::Series || pli.header.item_type != PlaylistItemType::SeriesInfo {
-            return false;
-        }
-        true
+        pli.header.xtream_cluster == XtreamCluster::Series && pli.header.item_type == PlaylistItemType::SeriesInfo
     };
 
     // Skip if nothing to do

backend/processing/src/processor/xtream_vod.rs

@@ -118,10 +118,7 @@ async fn playlist_resolve_vod_info<E: EventSink + Clone + 'static, M: MetadataUp
     do_probe: bool,
 ) {
     let filter = |pli: &PlaylistItem| {
-        if pli.header.xtream_cluster != XtreamCluster::Video || pli.header.item_type != PlaylistItemType::Video {
-            return false;
-        }
-        true
+        pli.header.xtream_cluster == XtreamCluster::Video && pli.header.item_type == PlaylistItemType::Video
     };
 
     let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb);

backend/session/src/active_provider_manager.rs

@@ -259,6 +259,12 @@ pub struct ActiveProviderManagerCore {
 }
 
 impl ActiveProviderManagerCore {
+    /// Serialises capacity transitions (acquire/release/reclassify) so a reclassify
+    /// cannot race an acquire or release.
+    ///
+    /// Lock-order invariant: this lock must be acquired **before** the connections
+    /// lock (`read_connections`/`write_connections`). No path may hold the
+    /// connections lock and then call this; doing so can deadlock.
     fn lock_capacity_transition(&self) -> std::sync::MutexGuard<'_, ()> {
         match self.capacity_transition.lock() {
             Ok(guard) => guard,
@@ -2344,6 +2350,7 @@ impl ActiveProviderManager {
         kind: ConnectionKind,
         priority: i8,
     ) -> bool {
+        let _transition = self.lock_capacity_transition();
         let mut connections = self.write_connections();
 
         let Some(alloc_ids) = connections.single_by_addr.get(addr).cloned() else {
@@ -2386,6 +2393,7 @@ impl ActiveProviderManager {
         kind: ConnectionKind,
         priority: i8,
     ) -> bool {
+        let _transition = self.lock_capacity_transition();
         let mut connections = self.write_connections();
         let shared_key = connections.shared.key_by_subscriber.get(&subscriber_id).cloned();
         let Some(shared_key) = shared_key else {

backend/session/src/active_user_manager/mod.rs

@@ -152,10 +152,54 @@ struct UserConnectionCounts {
     soft: u16,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AdmissionRejectionReason {
+    UserConnectionsExhausted,
+    RecentEvictionReentry,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct ConnectionAdmission {
-    pub permission: UserConnectionPermission,
-    pub kind: Option<ConnectionKind>,
+    permission: UserConnectionPermission,
+    kind: Option<ConnectionKind>,
+    rejection_reason: Option<AdmissionRejectionReason>,
+}
+
+impl ConnectionAdmission {
+    pub fn allowed(kind: Option<ConnectionKind>) -> Self {
+        Self { permission: UserConnectionPermission::Allowed, kind, rejection_reason: None }
+    }
+
+    pub fn grace_period(kind: Option<ConnectionKind>) -> Self {
+        Self { permission: UserConnectionPermission::GracePeriod, kind, rejection_reason: None }
+    }
+
+    pub fn exhausted(reason: AdmissionRejectionReason, kind: Option<ConnectionKind>) -> Self {
+        Self { permission: UserConnectionPermission::Exhausted, kind, rejection_reason: Some(reason) }
+    }
+
+    /// Maps a stored session permission onto an admission. An exhausted session is
+    /// classified as a genuine connection-limit rejection so the
+    /// `Exhausted ⇒ rejection_reason.is_some()` invariant always holds.
+    pub fn from_permission(permission: UserConnectionPermission, kind: Option<ConnectionKind>) -> Self {
+        match permission {
+            UserConnectionPermission::Allowed => Self::allowed(kind),
+            UserConnectionPermission::GracePeriod => Self::grace_period(kind),
+            UserConnectionPermission::Exhausted => {
+                Self::exhausted(AdmissionRejectionReason::UserConnectionsExhausted, kind)
+            }
+        }
+    }
+
+    pub fn permission(&self) -> UserConnectionPermission { self.permission }
+
+    pub fn kind(&self) -> Option<ConnectionKind> { self.kind }
+
+    pub fn rejection_reason(&self) -> Option<AdmissionRejectionReason> { self.rejection_reason }
+
+    pub fn is_reentry_suppressed(&self) -> bool {
+        self.rejection_reason == Some(AdmissionRejectionReason::RecentEvictionReentry)
+    }
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -475,7 +519,7 @@ fn uses_session_reentry_guard(stream: &StreamInfo) -> bool {
 #[derive(Clone, Copy, Debug)]
 struct RecentWinnerProtection {
     protected_addr: SocketAddr,
-    expires_at: u64,
+    expires_at: Instant,
 }
 
 #[derive(Debug, Default)]
@@ -644,6 +688,7 @@ pub struct ActiveUserManager {
     provider_manager: tokio::sync::OnceCell<Arc<ActiveProviderManager>>,
     transition_gates: Mutex<HashMap<String, Arc<Mutex<()>>>>,
     pub dropped_cleanup_events: AtomicU64,
+    reentry_suppressed_total: AtomicU64,
     divergence_cache: Mutex<LruCache<String, DivergenceEntry>>,
     divergence_cooldown_secs: u64,
 }
@@ -744,11 +789,19 @@ impl ActiveUserManager {
             provider_manager: tokio::sync::OnceCell::new(),
             transition_gates: Mutex::new(HashMap::new()),
             dropped_cleanup_events: AtomicU64::new(0),
+            reentry_suppressed_total: AtomicU64::new(0),
             divergence_cache: Mutex::new(LruCache::new(DIVERGENCE_CACHE_CAPACITY)),
             divergence_cooldown_secs: 300,
         }
     }
 
+    /// Cumulative count of admission requests quietly suppressed by the recent-eviction
+    /// reentry guard. A diagnostic counter, not a failure metric.
+    pub fn reentry_suppressed_total(&self) -> u64 { self.reentry_suppressed_total.load(Ordering::Relaxed) }
+
+    /// Records one suppressed reentry retry.
+    pub fn record_reentry_suppressed(&self) { self.reentry_suppressed_total.fetch_add(1, Ordering::Relaxed); }
+
     fn transition_gate_key(username: &str, token: &str) -> String {
         let mut key = String::with_capacity(username.len() + token.len() + 1);
         key.push_str(username);
@@ -1333,7 +1386,7 @@ impl ActiveUserManager {
                 connection_data.granted_grace = false;
                 connection_data.grace_ts = 0;
             }
-            return ConnectionAdmission { permission: UserConnectionPermission::Allowed, kind: Some(kind) };
+            return ConnectionAdmission::allowed(Some(kind));
         }
 
         let now = get_current_timestamp();
@@ -1344,7 +1397,7 @@ impl ActiveUserManager {
             {
                 // Grace timeout, still active, deny connection
                 debug!("User access denied, grace exhausted, too many connections: {username}");
-                return ConnectionAdmission { permission: UserConnectionPermission::Exhausted, kind: None };
+                return ConnectionAdmission::exhausted(AdmissionRejectionReason::UserConnectionsExhausted, None);
             }
             // Grace timeout expired, reset grace counters
             if effective_connections < connection_data.max_connections {
@@ -1354,7 +1407,7 @@ impl ActiveUserManager {
         }
 
         debug!("User access denied, too many connections: {username}");
-        ConnectionAdmission { permission: UserConnectionPermission::Exhausted, kind: None }
+        ConnectionAdmission::exhausted(AdmissionRejectionReason::UserConnectionsExhausted, None)
     }
 
     fn check_connection_admission(
@@ -1382,7 +1435,7 @@ impl ActiveUserManager {
                 return self.check_connection_admission(username, connection_data);
             }
         }
-        ConnectionAdmission { permission: UserConnectionPermission::Allowed, kind: Some(ConnectionKind::Normal) }
+        ConnectionAdmission::allowed(Some(ConnectionKind::Normal))
     }
 
     pub async fn connection_permission(
@@ -1402,18 +1455,12 @@ impl ActiveUserManager {
         session_token: &str,
     ) -> ConnectionAdmission {
         if max_connections == 0 && soft_connections == 0 {
-            return ConnectionAdmission {
-                permission: UserConnectionPermission::Allowed,
-                kind: Some(ConnectionKind::Normal),
-            };
+            return ConnectionAdmission::allowed(Some(ConnectionKind::Normal));
         }
 
         let mut connections = self.connections.write().await;
         let Some(connection_data) = connections.by_key.get_mut(username) else {
-            return ConnectionAdmission {
-                permission: UserConnectionPermission::Allowed,
-                kind: Some(ConnectionKind::Normal),
-            };
+            return ConnectionAdmission::allowed(Some(ConnectionKind::Normal));
         };
         connection_data.max_connections = max_connections;
         connection_data.soft_connections = soft_connections;
@@ -1424,10 +1471,9 @@ impl ActiveUserManager {
         };
 
         if connection_data.sessions[session_index].lifecycle.is_counted() {
-            return ConnectionAdmission {
-                permission: UserConnectionPermission::Allowed,
-                kind: connection_data.sessions[session_index].connection_kind.or(Some(ConnectionKind::Normal)),
-            };
+            return ConnectionAdmission::allowed(
+                connection_data.sessions[session_index].connection_kind.or(Some(ConnectionKind::Normal)),
+            );
         }
 
         self.check_connection_admission_with_counts(
@@ -1446,7 +1492,7 @@ impl ActiveUserManager {
     ) -> UserConnectionPermission {
         self.connection_admission_for_session(username, max_connections, soft_connections, session_token)
             .await
-            .permission
+            .permission()
     }
 
     pub async fn refresh_session_connection_kind_for_origin_policy(
@@ -1479,15 +1525,15 @@ impl ActiveUserManager {
                     connection_data,
                     connection_data.effective_counts_for_admission(Some(session_token)),
                 );
-                if admission.permission == UserConnectionPermission::Allowed {
-                    if let Some(kind) = admission.kind {
+                if admission.permission() == UserConnectionPermission::Allowed {
+                    if let Some(kind) = admission.kind() {
                         Self::update_session_admission(
                             &mut connection_data.sessions[session_index],
-                            admission.permission,
+                            admission.permission(),
                             Some(kind),
                         );
                     }
-                    admission.kind
+                    admission.kind()
                 } else {
                     None
                 }
@@ -2075,18 +2121,12 @@ impl ActiveUserManager {
         session_token: &str,
     ) -> ConnectionAdmission {
         if max_connections == 0 && soft_connections == 0 {
-            return ConnectionAdmission {
-                permission: UserConnectionPermission::Allowed,
-                kind: Some(ConnectionKind::Normal),
-            };
+            return ConnectionAdmission::allowed(Some(ConnectionKind::Normal));
         }
 
         let mut connections = self.connections.write().await;
         let Some(connection_data) = connections.by_key.get_mut(username) else {
-            return ConnectionAdmission {
-                permission: UserConnectionPermission::Allowed,
-                kind: Some(ConnectionKind::Normal),
-            };
+            return ConnectionAdmission::allowed(Some(ConnectionKind::Normal));
         };
         connection_data.max_connections = max_connections;
         connection_data.soft_connections = soft_connections;
@@ -2103,10 +2143,9 @@ impl ActiveUserManager {
         if connection_data.sessions[session_index].lifecycle.is_counted()
             || Self::session_has_stream(connection_data, session_token)
         {
-            return ConnectionAdmission {
-                permission: UserConnectionPermission::Allowed,
-                kind: connection_data.sessions[session_index].connection_kind.or(Some(ConnectionKind::Normal)),
-            };
+            return ConnectionAdmission::allowed(
+                connection_data.sessions[session_index].connection_kind.or(Some(ConnectionKind::Normal)),
+            );
         }
 
         // Uncounted session with no stream row: run normal admission.
@@ -2117,9 +2156,9 @@ impl ActiveUserManager {
             connection_data,
             connection_data.effective_counts_for_admission(Some(session_token)),
         );
-        if admission.permission == UserConnectionPermission::Allowed {
+        if admission.permission() == UserConnectionPermission::Allowed {
             let session = &mut connection_data.sessions[session_index];
-            Self::update_session_admission(session, admission.permission, admission.kind);
+            Self::update_session_admission(session, admission.permission(), admission.kind());
         }
         admission
     }
@@ -3032,11 +3071,19 @@ impl ActiveUserManager {
 
     /// Allocates the next process-local, globally unique User-Agent stream index.
     pub fn next_user_agent_stream_index(&self) -> u64 {
-        self.next_user_agent_stream_index
-            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
-                Some(if current == 0 || current == u64::MAX { 1 } else { current + 1 })
-            })
-            .map_or(1, |val| if val == 0 { 1 } else { val })
+        let mut current = self.next_user_agent_stream_index.load(Ordering::Relaxed);
+        loop {
+            let next = if current == 0 || current == u64::MAX { 1 } else { current + 1 };
+            match self.next_user_agent_stream_index.compare_exchange_weak(
+                current,
+                next,
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+            ) {
+                Ok(previous) => return if previous == 0 { 1 } else { previous },
+                Err(observed) => current = observed,
+            }
+        }
     }
 
     /// Persists a previously allocated index without replacing an existing session identity.
@@ -3080,9 +3127,9 @@ impl ActiveUserManager {
             && !matches!(connection_data.sessions[session_index].lifecycle, PlaybackLifecycle::PendingProvider { .. })
         {
             let admission = self.check_connection_admission(username, connection_data);
-            connection_data.sessions[session_index].permission = admission.permission;
-            if admission.kind.is_some() {
-                connection_data.sessions[session_index].connection_kind = admission.kind;
+            connection_data.sessions[session_index].permission = admission.permission();
+            if admission.kind().is_some() {
+                connection_data.sessions[session_index].connection_kind = admission.kind();
             }
         }
 
@@ -3167,9 +3214,8 @@ impl ActiveUserManager {
 
     pub async fn recently_evicted_session_protected_addr(&self, session_token: &str) -> Option<SocketAddr> {
         let connections = self.connections.read().await;
-        let now = current_time_secs();
         let protection = connections.recently_evicted_sessions.get(session_token)?;
-        if protection.expires_at > now {
+        if protection.expires_at > Instant::now() {
             return Some(protection.protected_addr);
         }
 
@@ -3190,18 +3236,13 @@ impl ActiveUserManager {
         virtual_id: VirtualId,
     ) -> Option<SocketAddr> {
         let connections = self.connections.read().await;
-        let now = current_time_secs();
         let key = create_socket_reentry_guard_key(username, client_ip, virtual_id);
         let protection = connections.recent_socket_reentry_guards.get(&key)?;
-        if protection.expires_at > now {
-            return Some(protection.protected_addr);
+        if protection.expires_at > Instant::now() {
+            Some(protection.protected_addr)
+        } else {
+            None
         }
-
-        connections
-            .key_by_addr
-            .get(&protection.protected_addr)
-            .filter(|registration| registration.usernames.contains(username))
-            .map(|_| protection.protected_addr)
     }
 
     pub async fn block_user_for_stream(&self, addr: &SocketAddr, virtual_id: VirtualId, blocked_secs: u64) {
@@ -3250,14 +3291,14 @@ impl ActiveUserManager {
         &self,
         addr: &SocketAddr,
         protected_addr: SocketAddr,
-        ttl_secs: u64,
+        ttl: Duration,
     ) {
-        if ttl_secs == 0 {
+        if ttl.is_zero() {
             return;
         }
 
         let mut connections = self.connections.write().await;
-        let now = current_time_secs();
+        let now = Instant::now();
         connections.recently_evicted_sessions.retain(|_, protection| protection.expires_at > now);
         connections.recent_socket_reentry_guards.retain(|_, protection| protection.expires_at > now);
 
@@ -3266,7 +3307,7 @@ impl ActiveUserManager {
             return;
         }
 
-        let protection = RecentWinnerProtection { protected_addr, expires_at: now + ttl_secs };
+        let protection = RecentWinnerProtection { protected_addr, expires_at: now + ttl };
         let mut session_tokens = Vec::new();
         let mut socket_guard_keys = Vec::new();
 
@@ -3611,8 +3652,13 @@ impl ActiveUserManager {
             {
                 if let Ok(mut user_connections) = self.connections.try_write() {
                     user_connections.kicked.retain(|_, (expires_at, _)| *expires_at > now);
-                    user_connections.recently_evicted_sessions.retain(|_, protection| protection.expires_at > now);
-                    user_connections.recent_socket_reentry_guards.retain(|_, protection| protection.expires_at > now);
+                    let now_instant = Instant::now();
+                    user_connections
+                        .recently_evicted_sessions
+                        .retain(|_, protection| protection.expires_at > now_instant);
+                    user_connections
+                        .recent_socket_reentry_guards
+                        .retain(|_, protection| protection.expires_at > now_instant);
                     for connection_data in user_connections.by_key.values_mut() {
                         Self::release_expired_session_reservations(connection_data, now);
                         connection_data.sessions.retain(|s| now.saturating_sub(s.ts) < USER_CON_TTL);

backend/session/src/active_user_manager/tests.rs

@@ -2066,7 +2066,7 @@ async fn recently_evicted_session_guard_survives_ttl_while_protected_addr_is_sti
         })
         .await;
 
-    manager.mark_recent_eviction_guard_for_addr(&evicted_addr, protected_addr, 1).await;
+    manager.mark_recent_eviction_guard_for_addr(&evicted_addr, protected_addr, Duration::from_secs(1)).await;
     {
         let mut connections = manager.connections.write().await;
         if let Some(registration) = connections.key_by_addr.get_mut(&protected_addr) {
@@ -2076,7 +2076,7 @@ async fn recently_evicted_session_guard_survives_ttl_while_protected_addr_is_sti
             .recently_evicted_sessions
             .get_mut("tok-guard-session")
             .expect("recent eviction guard should exist");
-        protection.expires_at = current_time_secs().saturating_sub(1);
+        protection.expires_at = Instant::now().checked_sub(Duration::from_secs(1)).unwrap_or_else(Instant::now);
     }
 
     assert_eq!(manager.recently_evicted_session_protected_addr("tok-guard-session").await, Some(protected_addr));
@@ -2133,7 +2133,7 @@ async fn recently_evicted_vod_uses_session_reentry_guard() {
         })
         .await;
 
-    manager.mark_recent_eviction_guard_for_addr(&evicted_addr, protected_addr, 10).await;
+    manager.mark_recent_eviction_guard_for_addr(&evicted_addr, protected_addr, Duration::from_secs(10)).await;
 
     assert_eq!(manager.recently_evicted_session_protected_addr("tok-guard-vod").await, Some(protected_addr));
     let connections = manager.connections.read().await;
@@ -2177,7 +2177,7 @@ async fn provider_affine_stream_without_session_token_uses_socket_reentry_fallba
         })
         .await;
 
-    manager.mark_recent_eviction_guard_for_addr(&evicted_addr, protected_addr, 10).await;
+    manager.mark_recent_eviction_guard_for_addr(&evicted_addr, protected_addr, Duration::from_secs(10)).await;
 
     assert_eq!(
         manager
@@ -2191,6 +2191,56 @@ async fn provider_affine_stream_without_session_token_uses_socket_reentry_fallba
     );
 }
 
+#[tokio::test]
+async fn socket_reentry_guard_expires_at_ttl() {
+    let config = Config::default();
+    let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
+    let event_manager = Arc::new(EventManager::new());
+    let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
+
+    let evicted_addr: SocketAddr = "127.0.0.1:55117".parse().unwrap();
+    let protected_addr: SocketAddr = "127.0.0.1:55118".parse().unwrap();
+    let fingerprint = Fingerprint::new("fp-socket-ttl".to_string(), "127.0.0.1".to_string(), evicted_addr);
+    let channel = test_channel(2021);
+
+    manager.add_connection(&evicted_addr).await;
+    manager
+        .update_connection(ActiveUserConnectionParams {
+            uid: 21,
+            meter_uid: 0,
+            username: "socket-ttl-user",
+            max_connections: 1,
+            soft_connections: 0,
+            connection_kind: ConnectionKind::Normal,
+            priority: 0,
+            soft_priority: 0,
+            fingerprint: &fingerprint,
+            provider: "provider-a".intern(),
+            stream_channel: &channel,
+            user_agent: Cow::Borrowed("ua"),
+            session_token: None,
+        })
+        .await;
+
+    manager.mark_recent_eviction_guard_for_addr(&evicted_addr, protected_addr, Duration::from_secs(10)).await;
+    let virtual_id = shared::model::VirtualId::new(channel.virtual_id);
+    assert_eq!(
+        manager.recent_socket_reentry_protected_addr("socket-ttl-user", "127.0.0.1", virtual_id).await,
+        Some(protected_addr)
+    );
+
+    // Expire only the guard while the protected address stays registered: the TTL is
+    // authoritative and must not fall back to the still-present socket registration.
+    {
+        let mut connections = manager.connections.write().await;
+        let key = create_socket_reentry_guard_key("socket-ttl-user", "127.0.0.1", virtual_id);
+        let protection =
+            connections.recent_socket_reentry_guards.get_mut(&key).expect("socket reentry guard should exist");
+        protection.expires_at = Instant::now().checked_sub(Duration::from_secs(1)).unwrap_or_else(Instant::now);
+    }
+    assert_eq!(manager.recent_socket_reentry_protected_addr("socket-ttl-user", "127.0.0.1", virtual_id).await, None);
+}
+
 #[tokio::test]
 async fn test_reused_logical_stream_refreshes_normal_priority() {
     let config = Config::default();
@@ -4510,9 +4560,9 @@ async fn session_activation_keeps_first_hls_slot_uncommitted_before_stream_regis
         .connection_admission_for_session_activation(&user.username, user.max_connections, 0, "tok-second")
         .await;
 
-    assert_eq!(first_admission.permission, UserConnectionPermission::Allowed);
-    assert_eq!(first_admission.kind, Some(ConnectionKind::Normal));
-    assert_eq!(second_admission.permission, UserConnectionPermission::Allowed);
+    assert_eq!(first_admission.permission(), UserConnectionPermission::Allowed);
+    assert_eq!(first_admission.kind(), Some(ConnectionKind::Normal));
+    assert_eq!(second_admission.permission(), UserConnectionPermission::Allowed);
 
     let connections = manager.connections.read().await;
     let connection_data = connections.by_key.get(&user.username).expect("user connection data");
@@ -4587,8 +4637,8 @@ async fn binding_reserved_sessions_keeps_hard_and_soft_counts_stable() {
             "tok-normal",
         )
         .await;
-    assert_eq!(first_admission.permission, UserConnectionPermission::Allowed);
-    assert_eq!(first_admission.kind, Some(ConnectionKind::Normal));
+    assert_eq!(first_admission.permission(), UserConnectionPermission::Allowed);
+    assert_eq!(first_admission.kind(), Some(ConnectionKind::Normal));
 
     manager
         .update_connection(ActiveUserConnectionParams {
@@ -4617,8 +4667,8 @@ async fn binding_reserved_sessions_keeps_hard_and_soft_counts_stable() {
             "tok-soft",
         )
         .await;
-    assert_eq!(second_admission.permission, UserConnectionPermission::Allowed);
-    assert_eq!(second_admission.kind, Some(ConnectionKind::Soft));
+    assert_eq!(second_admission.permission(), UserConnectionPermission::Allowed);
+    assert_eq!(second_admission.kind(), Some(ConnectionKind::Soft));
 
     manager
         .update_connection(ActiveUserConnectionParams {
@@ -4712,8 +4762,8 @@ async fn origin_policy_refresh_promotes_counted_soft_session_when_hard_slot_is_a
             "tok-normal",
         )
         .await;
-    assert_eq!(normal_admission.permission, UserConnectionPermission::Allowed);
-    assert_eq!(normal_admission.kind, Some(ConnectionKind::Normal));
+    assert_eq!(normal_admission.permission(), UserConnectionPermission::Allowed);
+    assert_eq!(normal_admission.kind(), Some(ConnectionKind::Normal));
     manager
         .update_connection(ActiveUserConnectionParams {
             uid: 411,
@@ -4741,8 +4791,8 @@ async fn origin_policy_refresh_promotes_counted_soft_session_when_hard_slot_is_a
             "tok-soft",
         )
         .await;
-    assert_eq!(soft_admission.permission, UserConnectionPermission::Allowed);
-    assert_eq!(soft_admission.kind, Some(ConnectionKind::Soft));
+    assert_eq!(soft_admission.permission(), UserConnectionPermission::Allowed);
+    assert_eq!(soft_admission.kind(), Some(ConnectionKind::Soft));
     manager
         .update_connection(ActiveUserConnectionParams {
             uid: 412,
@@ -4916,7 +4966,7 @@ async fn release_unbound_session_reservation_frees_reserved_slot() {
     let admission = manager
         .connection_admission_for_session_activation(&user.username, user.max_connections, 0, "tok-release")
         .await;
-    assert_eq!(admission.permission, UserConnectionPermission::Allowed);
+    assert_eq!(admission.permission(), UserConnectionPermission::Allowed);
 
     manager.release_unbound_session_reservation(&user.username, "tok-release", None, false).await;
 
@@ -4949,7 +4999,7 @@ async fn preserved_reactivation_admission_does_not_create_ownerless_counted_slot
 
     let virtual_admission =
         manager.connection_admission(&user.username, user.max_connections, user.soft_connections).await;
-    assert_eq!(virtual_admission.permission, UserConnectionPermission::Exhausted);
+    assert_eq!(virtual_admission.permission(), UserConnectionPermission::Exhausted);
 
     let admission = manager
         .connection_admission_for_session_activation(
@@ -4960,8 +5010,8 @@ async fn preserved_reactivation_admission_does_not_create_ownerless_counted_slot
         )
         .await;
 
-    assert_eq!(admission.permission, UserConnectionPermission::Allowed);
-    assert_eq!(admission.kind, Some(ConnectionKind::Normal));
+    assert_eq!(admission.permission(), UserConnectionPermission::Allowed);
+    assert_eq!(admission.kind(), Some(ConnectionKind::Normal));
     let connections = manager.connections.read().await;
     let connection_data = connections.by_key.get(&user.username).expect("user connection data");
     assert_preserved_session_is_uncounted(connection_data, session_token, stream_uid);
@@ -4990,7 +5040,7 @@ async fn preserved_reactivation_admission_then_kicked_release_removes_state_with
             session_token,
         )
         .await;
-    assert_eq!(admission.permission, UserConnectionPermission::Allowed);
+    assert_eq!(admission.permission(), UserConnectionPermission::Allowed);
     {
         let connections = manager.connections.read().await;
         let connection_data = connections.by_key.get(&user.username).expect("user connection data");
@@ -5038,7 +5088,7 @@ async fn preserved_reactivation_admission_then_lease_idle_cleanup_leaves_counter
             session_token,
         )
         .await;
-    assert_eq!(admission.permission, UserConnectionPermission::Allowed);
+    assert_eq!(admission.permission(), UserConnectionPermission::Allowed);
     {
         let connections = manager.connections.read().await;
         let connection_data = connections.by_key.get(&user.username).expect("user connection data");
@@ -5087,7 +5137,7 @@ async fn repeated_preserved_reactivation_cleanup_does_not_accumulate_connections
                 session_token,
             )
             .await;
-        assert_eq!(admission.permission, UserConnectionPermission::Allowed);
+        assert_eq!(admission.permission(), UserConnectionPermission::Allowed);
 
         let counter_changed =
             manager.release_session_streams_and_counted_reservation(&user.username, session_token).await;
@@ -5125,7 +5175,7 @@ async fn dashboard_counts_only_real_slots_during_preserved_reactivation_admissio
             session_token,
         )
         .await;
-    assert_eq!(admission.permission, UserConnectionPermission::Allowed);
+    assert_eq!(admission.permission(), UserConnectionPermission::Allowed);
     assert_eq!(manager.active_users_and_connections().await, (0, 0));
 
     let connections = manager.connections.read().await;
@@ -5199,8 +5249,8 @@ async fn preserved_soft_reactivation_and_cleanup_leave_normal_slot_unchanged() {
             session_token,
         )
         .await;
-    assert_eq!(admission.permission, UserConnectionPermission::Allowed);
-    assert_eq!(admission.kind, Some(ConnectionKind::Soft));
+    assert_eq!(admission.permission(), UserConnectionPermission::Allowed);
+    assert_eq!(admission.kind(), Some(ConnectionKind::Soft));
 
     {
         let connections = manager.connections.read().await;
@@ -5515,7 +5565,7 @@ async fn connection_admission_treats_preserved_stream_as_reserved_capacity() {
 
     let admission = manager.connection_admission(username, user.max_connections, 0).await;
     assert_eq!(
-        admission.permission,
+        admission.permission(),
         UserConnectionPermission::Exhausted,
         "a preserved stream should still reserve capacity against unrelated playback admissions"
     );
@@ -5593,7 +5643,7 @@ async fn connection_admission_for_session_evaluates_admission_for_uncounted_sess
     // freeing a slot for the uncounted session to reactivate
     let admission = manager.connection_admission_for_session(username, 1, 0, "tok-uncounted").await;
     assert_eq!(
-        admission.permission,
+        admission.permission(),
         UserConnectionPermission::Exhausted,
         "uncounted session should not bypass admission when user is at limit; \
              bug: session exists -> Allowed -> strategy evaluation skipped"
@@ -5961,7 +6011,7 @@ async fn arm_eviction_protection_protects_all_users_on_addr() {
         .await
         .expect("register user_b");
 
-    manager.mark_recent_eviction_guard_for_addr(&addr, protected_addr, 60).await;
+    manager.mark_recent_eviction_guard_for_addr(&addr, protected_addr, Duration::from_secs(60)).await;
 
     let connections = manager.connections.read().await;
     let key_a = create_socket_reentry_guard_key("user_a", "192.168.1.100", shared::model::VirtualId::new(2001));

backend/session/src/admission.rs

@@ -32,9 +32,21 @@ pub struct AdmissionCtx {
     pub connection_manager: Arc<ConnectionManager>,
 }
 
-/// How long an eviction is remembered so the evicted client's immediate retry
-/// does not evict its own replacement.
-pub const RECENT_EVICTION_REENTRY_TTL_SECS: u64 = 3;
+/// Default duration an eviction is remembered if not configured.
+pub const DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS: u64 = shared::defaults::DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS;
+
+/// Reentry cooldown from a resolved stream config, falling back to the default when
+/// no `reverse_proxy.stream` block is configured.
+fn reentry_ttl_for_stream(stream: Option<&tuliprox_core::model::StreamConfig>) -> std::time::Duration {
+    stream.map_or(std::time::Duration::from_millis(DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS), |stream| {
+        stream.recent_eviction_reentry_ttl
+    })
+}
+
+pub(crate) fn get_reentry_ttl(adm: &AdmissionCtx) -> std::time::Duration {
+    let config = adm.app_config.config.load();
+    reentry_ttl_for_stream(config.reverse_proxy.as_ref().and_then(|rp| rp.stream.as_ref()))
+}
 
 #[derive(Clone, Copy)]
 pub enum EvictionReentryGuard<'a> {
@@ -120,10 +132,10 @@ pub async fn resolve_playback_request_admission(
             adm.active_users.terminate_session(&user.username, session.token.as_str()).await;
         }
         return (
-            crate::ConnectionAdmission {
-                permission: UserConnectionPermission::Exhausted,
-                kind: user_session.and_then(|session| session.connection_kind).or(Some(crate::ConnectionKind::Normal)),
-            },
+            crate::ConnectionAdmission::exhausted(
+                crate::AdmissionRejectionReason::UserConnectionsExhausted,
+                user_session.and_then(|session| session.connection_kind).or(Some(crate::ConnectionKind::Normal)),
+            ),
             None,
             request_class,
         );
@@ -133,10 +145,9 @@ pub async fn resolve_playback_request_admission(
     // running strategies or modifying counted state. Caller handles the actual activation.
     if request_class == PlaybackRequestClass::Prepare {
         return (
-            crate::ConnectionAdmission {
-                permission: UserConnectionPermission::Allowed,
-                kind: user_session.and_then(|session| session.connection_kind).or(Some(crate::ConnectionKind::Normal)),
-            },
+            crate::ConnectionAdmission::allowed(
+                user_session.and_then(|session| session.connection_kind).or(Some(crate::ConnectionKind::Normal)),
+            ),
             None,
             request_class,
         );
@@ -144,10 +155,10 @@ pub async fn resolve_playback_request_admission(
 
     if request_class == PlaybackRequestClass::FollowUp || !limits_enabled {
         return (
-            crate::ConnectionAdmission {
-                permission: user_session.map_or(UserConnectionPermission::Allowed, |session| session.permission),
-                kind: user_session.and_then(|session| session.connection_kind).or(Some(crate::ConnectionKind::Normal)),
-            },
+            crate::ConnectionAdmission::from_permission(
+                user_session.map_or(UserConnectionPermission::Allowed, |session| session.permission),
+                user_session.and_then(|session| session.connection_kind).or(Some(crate::ConnectionKind::Normal)),
+            ),
             None,
             request_class,
         );
@@ -177,7 +188,12 @@ pub async fn resolve_playback_request_admission(
     //
     // Only the strategy path emits. An explicit `Terminate` also resolves to
     // `Exhausted`, but that is a requested teardown, not a denial.
-    if result.admission.permission == UserConnectionPermission::Exhausted {
+    //
+    // Background retry of a recently evicted stream is quietly suppressed to avoid
+    // playback ping-pong. Count it as a diagnostic, never as a connection denial.
+    if result.admission.is_reentry_suppressed() {
+        adm.active_users.record_reentry_suppressed();
+    } else if result.admission.permission() == UserConnectionPermission::Exhausted {
         adm.active_users.events().send_event(EventMessage::ConnectionDenied(ConnectionDenied::new(
             Arc::from(user.username.as_str()),
             Arc::from(fingerprint.client_ip.as_str()),
@@ -266,14 +282,14 @@ pub fn get_effective_admission_strategies(adm: &AdmissionCtx) -> Arc<[AdmissionS
     }
 }
 
+enum StrategyLoopResult {
+    Admitted(AdmissionStrategyResolution),
+    Rejected(crate::AdmissionRejectionReason),
+}
+
 /// Shared strategy-evaluation loop used by both the initial admission path
 /// (`resolve_admission_with_strategies`) and the remaining-strategies path
 /// (`evaluate_remaining_strategies_after_grace`).
-///
-/// Returns `Some(resolution)` when a Grace or a successful Eviction+Retry is found.
-/// Returns `None` when every strategy in `strategies` returns `NoMatch` — the caller
-/// is then responsible for constructing the final exhausted result with the correct
-/// `kind` (preserved from the original admission).
 async fn evaluate_admission_strategy_loop<F>(
     adm: &AdmissionCtx,
     request: &AdmissionRequest<'_>,
@@ -281,7 +297,7 @@ async fn evaluate_admission_strategy_loop<F>(
     base_idx: usize,
     admission: crate::ConnectionAdmission,
     build_grace_ctx: F,
-) -> Option<AdmissionStrategyResolution>
+) -> StrategyLoopResult
 where
     F: Fn(usize) -> GraceResolutionContext,
 {
@@ -295,6 +311,12 @@ where
     // kick that frees nothing is taken as evidence that the next one would not
     // help either, and later eviction strategies are skipped.
     let mut evictions_ineffective = false;
+    // Set when a candidate was skipped because the reentry guard protects it.
+    let mut suppressed_candidate = false;
+    // Set only after a *non-protected* candidate was actually evicted. A genuine
+    // eviction means the final rejection is real resource exhaustion, not a
+    // reentry suppression, even if an earlier candidate was protected.
+    let mut performed_legitimate_eviction = false;
 
     // `enumerate` rather than a manual counter: the suppressed-eviction arm below
     // uses `continue`, which used to skip a trailing `idx += 1` and hand every
@@ -310,11 +332,8 @@ where
                     // Return a FRESH admission with GracePeriod permission (not the admission
                     // parameter, which may have Exhausted permission). The kind is preserved from
                     // the original admission.
-                    return Some(AdmissionStrategyResolution {
-                        admission: crate::ConnectionAdmission {
-                            permission: UserConnectionPermission::GracePeriod,
-                            kind: admission.kind,
-                        },
+                    return StrategyLoopResult::Admitted(AdmissionStrategyResolution {
+                        admission: crate::ConnectionAdmission::grace_period(admission.kind()),
                         grace_mode: Some(mode),
                         grace_context: Some(build_grace_ctx(base_idx + idx)),
                     });
@@ -333,17 +352,20 @@ where
                         "Skipping eviction strategy {strategy:?} for recently evicted request of user {username} targeting {}",
                         target.addr
                     );
+                    suppressed_candidate = true;
+                    // Exclude this specific stream candidate so subsequent strategies can evaluate other candidates
+                    candidates.retain(|c| c.uid != target.uid);
                     continue;
                 }
                 debug!("Evicting connection {} for user {username}", target.addr);
                 let connections_before = adm.active_users.user_connections(username).await;
-                adm.active_users
-                    .mark_recent_eviction_guard_for_addr(&target.addr, *request_addr, RECENT_EVICTION_REENTRY_TTL_SECS)
-                    .await;
+                let ttl = get_reentry_ttl(adm);
+                adm.active_users.mark_recent_eviction_guard_for_addr(&target.addr, *request_addr, ttl).await;
                 adm.connection_manager.release_connection_as_kicked(&target.addr).await;
+                performed_legitimate_eviction = true;
                 let retry_admission = get_admission_for_request(adm, request).await;
-                if retry_admission.permission == UserConnectionPermission::Allowed {
-                    return Some(AdmissionStrategyResolution {
+                if retry_admission.permission() == UserConnectionPermission::Allowed {
+                    return StrategyLoopResult::Admitted(AdmissionStrategyResolution {
                         admission: retry_admission,
                         grace_mode: None,
                         grace_context: None,
@@ -365,8 +387,24 @@ where
         }
     }
 
-    // All strategies returned NoMatch — caller constructs the final exhausted result.
-    None
+    StrategyLoopResult::Rejected(rejection_after_strategy_loop(suppressed_candidate, performed_legitimate_eviction))
+}
+
+/// Classifies the rejection after the strategy loop.
+///
+/// `RecentEvictionReentry` is only correct when the request failed *solely* because
+/// every candidate was reentry-protected and no legitimate eviction was performed.
+/// Once a non-protected candidate was evicted and admission still fails, the request
+/// hit genuine resource exhaustion and must be reported as such.
+fn rejection_after_strategy_loop(
+    suppressed_candidate: bool,
+    performed_legitimate_eviction: bool,
+) -> crate::AdmissionRejectionReason {
+    if suppressed_candidate && !performed_legitimate_eviction {
+        crate::AdmissionRejectionReason::RecentEvictionReentry
+    } else {
+        crate::AdmissionRejectionReason::UserConnectionsExhausted
+    }
 }
 
 pub async fn resolve_admission_with_strategies(
@@ -378,14 +416,21 @@ pub async fn resolve_admission_with_strategies(
     let username = request.username;
     let admission = get_admission_for_request(adm, &request).await;
 
-    if admission.permission != UserConnectionPermission::Exhausted {
+    if admission.permission() != UserConnectionPermission::Exhausted {
         return AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None };
     }
 
     let strategies = get_effective_admission_strategies(adm);
     if strategies.is_empty() {
         debug!("No admission strategies configured, denying request for user {username}");
-        return AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None };
+        return AdmissionStrategyResolution {
+            admission: crate::ConnectionAdmission::exhausted(
+                crate::AdmissionRejectionReason::UserConnectionsExhausted,
+                admission.kind(),
+            ),
+            grace_mode: None,
+            grace_context: None,
+        };
     }
 
     let _admission_guard = adm.active_users.acquire_user_admission(username).await;
@@ -397,7 +442,7 @@ pub async fn resolve_admission_with_strategies(
     // free.
     let admission = get_admission_for_request(adm, &request).await;
 
-    if admission.permission != UserConnectionPermission::Exhausted {
+    if admission.permission() != UserConnectionPermission::Exhausted {
         debug!("Admission became available while waiting on the admission gate for user {username}");
         return AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None };
     }
@@ -405,17 +450,20 @@ pub async fn resolve_admission_with_strategies(
     let build_grace_ctx = |global_idx: usize| GraceResolutionContext {
         strategy_index: global_idx,
         strategies: strategies.clone(),
-        kind: admission.kind,
+        kind: admission.kind(),
     };
 
-    if let Some(resolution) =
-        evaluate_admission_strategy_loop(adm, &request, &strategies, 0, admission, build_grace_ctx).await
-    {
-        return resolution;
+    match evaluate_admission_strategy_loop(adm, &request, &strategies, 0, admission, build_grace_ctx).await {
+        StrategyLoopResult::Admitted(resolution) => resolution,
+        StrategyLoopResult::Rejected(reason) => {
+            debug!("No admission strategy could admit user {username}: {reason:?}");
+            AdmissionStrategyResolution {
+                admission: crate::ConnectionAdmission::exhausted(reason, admission.kind()),
+                grace_mode: None,
+                grace_context: None,
+            }
+        }
     }
-
-    debug!("No admission strategy could admit user {username}");
-    AdmissionStrategyResolution { admission, grace_mode: None, grace_context: None }
 }
 
 /// Evaluates only the strategies that come AFTER the already-used grace strategy.
@@ -436,18 +484,16 @@ pub async fn evaluate_remaining_strategies_after_grace(
     grace_context: &GraceResolutionContext,
     original_kind: Option<crate::ConnectionKind>,
 ) -> AdmissionStrategyResolution {
-    use shared::model::UserConnectionPermission;
-
     let username = request.username;
     let remaining = grace_context.strategy_index + 1;
     let strategies = &grace_context.strategies;
     if remaining >= strategies.len() {
         debug!("No remaining strategies after grace for user {username}");
         return AdmissionStrategyResolution {
-            admission: crate::ConnectionAdmission {
-                permission: UserConnectionPermission::Exhausted,
-                kind: original_kind,
-            },
+            admission: crate::ConnectionAdmission::exhausted(
+                crate::AdmissionRejectionReason::UserConnectionsExhausted,
+                original_kind,
+            ),
             grace_mode: None,
             grace_context: None,
         };
@@ -457,7 +503,8 @@ pub async fn evaluate_remaining_strategies_after_grace(
     // returned `ConnectionAdmission`, and `build_grace_ctx` copies it onto the
     // `GraceResolutionContext`. Seeding it with `original_kind` keeps every exit from this
     // function reporting the kind the original admission decided.
-    let admission = crate::ConnectionAdmission { permission: UserConnectionPermission::Exhausted, kind: original_kind };
+    let admission =
+        crate::ConnectionAdmission::exhausted(crate::AdmissionRejectionReason::UserConnectionsExhausted, original_kind);
     let build_grace_ctx = |global_idx: usize| GraceResolutionContext {
         strategy_index: global_idx,
         strategies: strategies.clone(),
@@ -466,17 +513,73 @@ pub async fn evaluate_remaining_strategies_after_grace(
 
     let _admission_guard = adm.active_users.acquire_user_admission(username).await;
 
-    if let Some(resolution) =
-        evaluate_admission_strategy_loop(adm, &request, &strategies[remaining..], remaining, admission, build_grace_ctx)
-            .await
+    match evaluate_admission_strategy_loop(
+        adm,
+        &request,
+        &strategies[remaining..],
+        remaining,
+        admission,
+        build_grace_ctx,
+    )
+    .await
     {
-        return resolution;
-    }
-
-    debug!("No remaining strategy could admit user {username}");
-    AdmissionStrategyResolution {
-        admission: crate::ConnectionAdmission { permission: UserConnectionPermission::Exhausted, kind: original_kind },
-        grace_mode: None,
-        grace_context: None,
+        StrategyLoopResult::Admitted(resolution) => resolution,
+        StrategyLoopResult::Rejected(reason) => {
+            debug!("No remaining strategy could admit user {username}: {reason:?}");
+            AdmissionStrategyResolution {
+                admission: crate::ConnectionAdmission::exhausted(reason, original_kind),
+                grace_mode: None,
+                grace_context: None,
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{AdmissionRejectionReason, ConnectionAdmission};
+    use shared::model::UserConnectionPermission;
+
+    #[test]
+    fn reentry_ttl_uses_configured_value_and_falls_back_to_default() {
+        assert_eq!(
+            reentry_ttl_for_stream(None),
+            std::time::Duration::from_millis(DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS)
+        );
+        let stream = tuliprox_core::model::StreamConfig {
+            recent_eviction_reentry_ttl: std::time::Duration::from_millis(2_500),
+            ..tuliprox_core::model::StreamConfig::default()
+        };
+        assert_eq!(reentry_ttl_for_stream(Some(&stream)), std::time::Duration::from_millis(2_500));
+    }
+
+    #[test]
+    fn rejection_classification_distinguishes_reentry_from_exhaustion() {
+        // Pure reentry: a protected candidate was skipped and nothing was evicted.
+        assert_eq!(rejection_after_strategy_loop(true, false), AdmissionRejectionReason::RecentEvictionReentry);
+        // Mixed case: a protected candidate was skipped, but a legitimate eviction was
+        // performed and admission still failed. That residual failure is genuine
+        // exhaustion, not a quiet suppression.
+        assert_eq!(rejection_after_strategy_loop(true, true), AdmissionRejectionReason::UserConnectionsExhausted);
+        assert_eq!(rejection_after_strategy_loop(false, true), AdmissionRejectionReason::UserConnectionsExhausted);
+        assert_eq!(rejection_after_strategy_loop(false, false), AdmissionRejectionReason::UserConnectionsExhausted);
+    }
+
+    #[test]
+    fn exhausted_admission_always_carries_a_reason() {
+        let exhausted = ConnectionAdmission::from_permission(UserConnectionPermission::Exhausted, None);
+        assert_eq!(exhausted.permission(), UserConnectionPermission::Exhausted);
+        assert_eq!(exhausted.rejection_reason(), Some(AdmissionRejectionReason::UserConnectionsExhausted));
+        assert!(!exhausted.is_reentry_suppressed());
+
+        assert_eq!(
+            ConnectionAdmission::from_permission(UserConnectionPermission::Allowed, None).rejection_reason(),
+            None
+        );
+        assert_eq!(
+            ConnectionAdmission::from_permission(UserConnectionPermission::GracePeriod, None).permission(),
+            UserConnectionPermission::GracePeriod
+        );
     }
 }

backend/session/src/admission_strategy.rs

@@ -19,6 +19,7 @@ pub enum GraceMode {
 pub struct EvictionTarget {
     pub addr: SocketAddr,
     pub virtual_id: VirtualId,
+    pub uid: u32,
 }
 
 #[derive(Debug, Clone)]
@@ -89,18 +90,22 @@ fn evaluate_evict_same_ip(
     }
 
     match selected {
-        Some(candidate) => {
-            AdmissionDecision::Evict(EvictionTarget { addr: candidate.addr, virtual_id: candidate.virtual_id })
-        }
+        Some(candidate) => AdmissionDecision::Evict(EvictionTarget {
+            addr: candidate.addr,
+            virtual_id: candidate.virtual_id,
+            uid: candidate.uid,
+        }),
         None => AdmissionDecision::NoMatch,
     }
 }
 
 fn evaluate_evict_user(candidates: &[EvictionCandidate], order: EvictionOrder) -> AdmissionDecision {
     match select_candidate(candidates.iter(), order) {
-        Some(candidate) => {
-            AdmissionDecision::Evict(EvictionTarget { addr: candidate.addr, virtual_id: candidate.virtual_id })
-        }
+        Some(candidate) => AdmissionDecision::Evict(EvictionTarget {
+            addr: candidate.addr,
+            virtual_id: candidate.virtual_id,
+            uid: candidate.uid,
+        }),
         None => AdmissionDecision::NoMatch,
     }
 }

backend/session/src/connection_manager.rs

@@ -44,6 +44,7 @@ const CONTROL_CLEANUP_CAPACITY: usize = 64;
 pub const PROVIDER_END_NOT_SET: u8 = 0;
 pub const PROVIDER_END_CLOSED: u8 = 1; // Provider EOF
 pub const PROVIDER_END_ERROR: u8 = 2; // Provider Err
+pub const PROVIDER_END_PREEMPTED: u8 = 3; // Preempted by higher priority
 const PREEMPT_REENTRY_BLOCK_SECS: u64 = 3;
 // Bounded wait for a mandatory cleanup admission right before a request registration.
 const CLEANUP_ADMISSION_TIMEOUT: Duration = Duration::from_secs(5);
@@ -1621,6 +1622,7 @@ fn resolve_disconnect_reason(provider_end_reason: u8, stream_info: &StreamInfo)
     match provider_end_reason {
         PROVIDER_END_CLOSED => DisconnectReason::ProviderClosed,
         PROVIDER_END_ERROR => DisconnectReason::ProviderError,
+        PROVIDER_END_PREEMPTED => DisconnectReason::Preempted,
         _ => DisconnectReason::ClientClosed,
     }
 }
@@ -1631,6 +1633,7 @@ fn resolve_disconnect_reason_from_provider_end(provider_end_reason: u8) -> Disco
     match provider_end_reason {
         PROVIDER_END_CLOSED => DisconnectReason::ProviderClosed,
         PROVIDER_END_ERROR => DisconnectReason::ProviderError,
+        PROVIDER_END_PREEMPTED => DisconnectReason::Preempted,
         _ => DisconnectReason::ClientClosed,
     }
 }
@@ -2162,6 +2165,17 @@ mod tests {
         assert_eq!(reason, DisconnectReason::Preempted);
     }
 
+    #[test]
+    fn test_provider_preempted_atomic_maps_to_preempted() {
+        let info = make_stream_info("some_provider", "Some Channel");
+        let reason = resolve_disconnect_reason(PROVIDER_END_PREEMPTED, &info);
+        assert_eq!(reason, DisconnectReason::Preempted);
+
+        let reason_direct = resolve_disconnect_reason_from_provider_end(PROVIDER_END_PREEMPTED);
+        assert_eq!(reason_direct, DisconnectReason::Preempted);
+        assert_eq!(playback_outcome_for_reason(reason_direct), PlaybackRequestOutcome::Preempted);
+    }
+
     #[test]
     fn test_channel_unavailable_with_eof_maps_to_provider_closed() {
         let info = make_stream_info("tuliprox", "channel_unavailable");

backend/session/src/provider_leases.rs

@@ -235,6 +235,10 @@ impl ProviderLeaseTable {
                     lease.created_at = now;
                     lease.state =
                         ProviderLeaseState::Starting { expires_at: Self::startup_expiry(now, lease.idle_ttl_secs) };
+                } else if !lease.state.is_confirmed() {
+                    lease.state = ProviderLeaseState::Starting {
+                        expires_at: Self::startup_expiry(lease.created_at, lease.idle_ttl_secs),
+                    };
                 }
                 // The kind is authoritative from the first acquire of this playback and
                 // is never overwritten, so a compatibility renewal cannot downgrade a
@@ -242,11 +246,6 @@ impl ProviderLeaseTable {
                 lease.request_id = request_id;
                 lease.request_ids.insert(request_id);
                 lease.last_activity_at = now;
-                if !lease.state.is_confirmed() && !lease.state.is_idle() {
-                    lease.state = ProviderLeaseState::Starting {
-                        expires_at: Self::startup_expiry(lease.created_at, lease.idle_ttl_secs),
-                    };
-                }
                 self.expirations.insert((lease.state.expires_at(), id));
                 return id;
             }
@@ -486,7 +485,7 @@ impl ProviderLeaseTable {
     ) -> Option<ProviderSlotLease> {
         let id = self.by_owner.get(owner).copied()?;
         let now = TokioInstant::now();
-        let (keep_idle, remove) = {
+        let keep_idle = {
             let lease = self.leases.get_mut(&id)?;
             if !lease.request_ids.remove(&request_id) {
                 return None;
@@ -511,15 +510,13 @@ impl ProviderLeaseTable {
                 lease.state = ProviderLeaseState::Idle { expires_at: now + Duration::from_secs(ttl) };
                 self.expirations.insert((lease.state.expires_at(), id));
             }
-            (keep, !keep)
+            keep
         };
         if keep_idle {
-            return None;
+            None
+        } else {
+            self.remove_lease(id)
         }
-        if remove {
-            return self.remove_lease(id);
-        }
-        None
     }
 
     /// Releases one request claim. A stale generation cannot clear its successor.
@@ -556,6 +553,8 @@ impl ProviderLeaseTable {
         if lease.request_id == request_id {
             if let Some(next) = lease.request_ids.iter().next().copied() {
                 lease.request_id = next;
+            } else {
+                lease.request_id = PlaybackRequestId::default();
             }
         }
         true

docs/src/configuration/reverse-proxy.md

@@ -110,6 +110,7 @@ reverse_proxy:
     catchup_session_ttl_secs: 45
     shared_burst_buffer_mb: 12
     cleanup_queue_capacity: 4096
+    recent_eviction_reentry_ttl_ms: 3000
     metrics_enabled: false
 ```
 
@@ -131,6 +132,7 @@ reverse_proxy:
 | `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. |
 | `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`). |
 
 ### 1.1 `retry` & `buffer` (Deep Dive)
 

shared/src/defaults/network.rs

@@ -1,6 +1,11 @@
 //! Generic networking defaults: resolve/probe delays, grace periods, retry,
 //! interner GC, port, user-agent, background flags.
 
+/// Default reentry cooldown after an eviction: how long the evicted playback's
+/// immediate retry is suppressed so it cannot immediately evict its replacement.
+/// This is the single source of truth for the default.
+pub const DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS: u64 = 3000;
+
 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;
@@ -18,6 +23,7 @@ default_eq_fns!(
     default_custom_stream_response_error_status, is_default_custom_stream_response_error_status, u16, 502;
     default_event_channel_capacity, is_default_event_channel_capacity, u32, 256;
     default_cleanup_queue_capacity, is_default_cleanup_queue_capacity, usize, 4096;
+    default_recent_eviction_reentry_ttl_ms, is_default_recent_eviction_reentry_ttl_ms, u64, DEFAULT_RECENT_EVICTION_REENTRY_TTL_MS;
 );
 
 pub const fn default_resource_retry_backoff_multiplier() -> f64 { 1.0 }

shared/src/model/config/stream.rs

@@ -1,10 +1,11 @@
 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_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_shared_burst_buffer_mb,
+        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,
+        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,
     },
     error::TuliproxError,
@@ -117,6 +118,11 @@ pub struct StreamConfigDto {
     /// growing memory without limit.
     #[serde(default = "default_cleanup_queue_capacity", skip_serializing_if = "is_default_cleanup_queue_capacity")]
     pub cleanup_queue_capacity: usize,
+    #[serde(
+        default = "default_recent_eviction_reentry_ttl_ms",
+        skip_serializing_if = "is_default_recent_eviction_reentry_ttl_ms"
+    )]
+    pub recent_eviction_reentry_ttl_ms: u64,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub admission_strategies: Option<Vec<AdmissionStrategy>>,
 }
@@ -134,6 +140,7 @@ impl Default for StreamConfigDto {
             shared_burst_buffer_mb: default_shared_burst_buffer_mb(),
             shared_subscriber_idle_timeout_secs: default_shared_subscriber_idle_timeout_secs(),
             cleanup_queue_capacity: default_cleanup_queue_capacity(),
+            recent_eviction_reentry_ttl_ms: default_recent_eviction_reentry_ttl_ms(),
             grace_period_hold_stream: true,
             hls_session_ttl_secs: default_hls_session_ttl_secs(),
             catchup_session_ttl_secs: default_catchup_session_ttl_secs(),
@@ -154,6 +161,7 @@ impl StreamConfigDto {
             && self.shared_burst_buffer_mb == default_shared_burst_buffer_mb()
             && self.shared_subscriber_idle_timeout_secs == default_shared_subscriber_idle_timeout_secs()
             && self.cleanup_queue_capacity == default_cleanup_queue_capacity()
+            && self.recent_eviction_reentry_ttl_ms == default_recent_eviction_reentry_ttl_ms()
             && 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()

tools/tuliprox-testkit/Cargo.toml

@@ -21,7 +21,6 @@ serde_json.workspace = true
 serde-saphyr.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "process", "signal", "sync", "time"] }
-tokio-stream = "0.1.19"
 tokio-tungstenite = "0.29.0"
 tempfile = "3.27.0"
 pin-project-lite = "0.2.17"
@@ -32,3 +31,4 @@ tower-service = "0.3.3"
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }
+tokio-stream = "0.1.19"

tools/tuliprox-testkit/src/bootstrap.rs

@@ -99,6 +99,9 @@ fn render_config(
                 }
             }
         }
+        if let Some(ttl_ms) = policy.recent_eviction_reentry_ttl_ms {
+            let _ = writeln!(config, "    recent_eviction_reentry_ttl_ms: {ttl_ms}");
+        }
     }
     let _ = writeln!(
         config,
@@ -458,6 +461,7 @@ mod tests {
                 crate::config::UserPolicy { max_connections: 1, soft_connections: 0 },
             )]),
             admission_strategies: Some(vec![crate::oracle::AdmissionStrategy::EvictUserSameIpLatest]),
+            recent_eviction_reentry_ttl_ms: None,
             grace: None,
             provider_max_connections: None,
             expected_provider_slots: None,

tools/tuliprox-testkit/src/config.rs

@@ -47,6 +47,11 @@ pub struct PolicyContract {
     /// from an explicit empty admission-strategy list.
     #[serde(default)]
     pub admission_strategies: Option<Vec<AdmissionStrategy>>,
+    /// Overrides the reentry cooldown applied after an eviction. When set, a retry of
+    /// the evicted playback inside this window is suppressed instead of evicting its
+    /// replacement. Used by the reentry-suppression scenario.
+    #[serde(default)]
+    pub recent_eviction_reentry_ttl_ms: Option<u64>,
     #[serde(default)]
     pub grace: Option<GraceContract>,
     #[serde(default)]
@@ -224,6 +229,9 @@ pub enum ExpectedPlayback {
     Streaming,
     Rejected,
     RejectedWith(ExpectedRejection),
+    /// A recently evicted playback retried and the SUT terminated it quietly
+    /// (HTTP 204, no custom error video, no connection-denied event).
+    Suppressed,
 }
 
 impl ExpectedPlayback {
@@ -231,14 +239,19 @@ impl ExpectedPlayback {
     pub fn is_streaming(&self) -> bool { matches!(self, Self::Streaming) }
 
     #[must_use]
-    pub fn is_rejected(&self) -> bool { matches!(self, Self::Rejected | Self::RejectedWith(_)) }
+    pub fn is_rejected(&self) -> bool { matches!(self, Self::Rejected | Self::RejectedWith(_) | Self::Suppressed) }
 
     #[must_use]
     pub fn matches_outcome(&self, outcome: &crate::protocol::PlaybackOutcome) -> bool {
         use crate::protocol::{PlaybackOutcome, RejectionReason};
         match (self, outcome) {
             (Self::Streaming, PlaybackOutcome::Streaming { .. })
-            | (Self::Rejected, PlaybackOutcome::AdmissionRejected { .. }) => true,
+            | (Self::Rejected, PlaybackOutcome::AdmissionRejected { .. })
+            // A quiet suppression is exactly the 204 the SUT returns for a
+            // reentry-protected retry. Any other rejection reason must fail.
+            | (Self::Suppressed, PlaybackOutcome::AdmissionRejected { reason: RejectionReason::HttpStatus(204) }) => {
+                true
+            }
             (Self::RejectedWith(expected), PlaybackOutcome::AdmissionRejected { reason }) => match reason {
                 RejectionReason::CustomVideo(kind) => {
                     expected.custom_video.is_none_or(|expected_kind| expected_kind == *kind)
@@ -412,7 +425,7 @@ impl Scenario {
                 return Err(TestkitError::Configuration(format!("duplicate playback ID {}", start.playback_id)));
             }
             playback_ids.insert(start.playback_id.as_str());
-            if step.expect == ExpectedPlayback::Rejected
+            if step.expect.is_rejected()
                 && step
                     .await_frames
                     .or_else(|| step.await_condition.as_ref().and_then(|condition| condition.valid_frames))
@@ -567,6 +580,7 @@ mod tests {
             "vod-range-reopen-strict-cap.yml",
             "vod-reopen-backpressured-body.yml",
             "live-ts-same-channel-retry-latest-wins.yml",
+            "reentry-suppresses-evicted-retry.yml",
         ] {
             assert!(Scenario::from_path(&root.join(filename)).is_ok(), "{filename}");
         }
@@ -656,6 +670,26 @@ steps:
         assert!(!expected_rejected.matches_outcome(&PlaybackOutcome::IdleTimeout));
     }
 
+    #[test]
+    fn suppressed_matches_only_204_quiet_termination() {
+        use crate::{
+            custom_video::CustomVideoKind,
+            protocol::{PlaybackOutcome, RejectionReason},
+        };
+
+        let suppressed = ExpectedPlayback::Suppressed;
+        assert!(suppressed
+            .matches_outcome(&PlaybackOutcome::AdmissionRejected { reason: RejectionReason::HttpStatus(204) }));
+        // A custom error video (the regression this scenario guards against) must fail.
+        assert!(!suppressed.matches_outcome(&PlaybackOutcome::AdmissionRejected {
+            reason: RejectionReason::CustomVideo(CustomVideoKind::UserConnectionsExhausted),
+        }));
+        assert!(!suppressed
+            .matches_outcome(&PlaybackOutcome::AdmissionRejected { reason: RejectionReason::HttpStatus(503) }));
+        // An empty body without the 204 classification (bare EOF) must fail too.
+        assert!(!suppressed.matches_outcome(&PlaybackOutcome::UnexpectedEof { frames: 0, bytes: 0 }));
+    }
+
     #[test]
     fn playback_endpoint_defaults_to_m3u() {
         let cfg: TuliproxConfig = serde_saphyr::from_str("base_url: 'http://example.invalid'").unwrap();

tools/tuliprox-testkit/src/main.rs

@@ -56,6 +56,13 @@ fn is_rejection_http_status(status: u16, headers: &reqwest::header::HeaderMap) -
     matches!(status, 403 | 429) || ((status == 502 || status == 503) && headers.get("x-tuliprox-rejection").is_some())
 }
 
+/// Tuliprox answers a suppressed reentry retry with `204 No Content`: a success status
+/// and an empty body. Classify it as an admission rejection so a scenario can assert the
+/// quiet termination instead of observing a bare end-of-body.
+fn quiet_suppression_outcome(status: u16) -> Option<PlaybackOutcome> {
+    (status == 204).then_some(PlaybackOutcome::AdmissionRejected { reason: RejectionReason::HttpStatus(status) })
+}
+
 /// Build the final playback URL for a live channel, dispatching between M3U-discovered URLs and
 /// Xtream live endpoint URLs constructed from the virtual ID embedded in the M3U playlist entry.
 fn resolve_playback_url(
@@ -1018,6 +1025,9 @@ async fn run_agent(
             PlaybackOutcome::HttpError { status }
         });
     }
+    if let Some(outcome) = quiet_suppression_outcome(status) {
+        return Ok(outcome);
+    }
     let mut body = response.bytes_stream();
     let mut decoder = FrameDecoder::default();
     let mut validator = FrameValidator::new(expected_run_id, expected_marker);
@@ -1101,6 +1111,9 @@ async fn run_agent_until_released(
             PlaybackOutcome::HttpError { status }
         });
     }
+    if let Some(outcome) = quiet_suppression_outcome(status) {
+        return Ok(outcome);
+    }
     let mut body = response.bytes_stream();
     let mut decoder = FrameDecoder::default();
     let mut validator = FrameValidator::new(&expected_run_id, expected_marker);
@@ -2306,7 +2319,11 @@ async fn execute_scenario_steps<'a>(
                     scenario.policy_contract.as_ref().and_then(|contract| contract.provider_max_connections).is_some()
                         && step.expect.is_rejected()
                         && oracle_expected == ExpectedPlayback::Streaming;
-                if step.expect != oracle_expected && !provider_limited_rejection {
+                // The static oracle does not model the time-based reentry guard, so a
+                // suppressed retry intentionally diverges from its eviction prediction.
+                // The concrete SUT outcome is asserted through `step.expect`.
+                let reentry_suppression = matches!(step.expect, ExpectedPlayback::Suppressed);
+                if step.expect != oracle_expected && !provider_limited_rejection && !reentry_suppression {
                     return Err(TestkitError::Configuration(format!(
                         "step {} expects {:?}, but policy oracle expects {:?}",
                         step.command_id, step.expect, oracle_expected

tools/tuliprox-testkit/src/policy.rs

@@ -208,6 +208,7 @@ mod tests {
             user_access_control: true,
             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,
             grace: None,
             provider_max_connections: None,
             expected_provider_slots: None,