Session handling refactor

CHANGELOG.md

@@ -957,6 +957,18 @@
   carrying the reason, current and maximum connections, foreign reserved slots, and the active/starting/idle slot split,
   so a fallback decision can be audited without inferring it from a socket address.
 
+  Playback cleanup is now request-specific. Parallel segment, range and reconnect requests carry independent request
+  identities, while a provider-binding generation prevents delayed cleanup from releasing a newer successor binding.
+  Provider allocations are owned by drop guards until they are explicitly transferred, closing cancellation gaps during
+  provider open, grace handling and HLS origin refresh.
+
+  Shared MPEG-TS subscribers now have identities independent of their transport socket, so clients sharing one reverse
+  proxy connection cannot replace or cancel one another. Subscriber queues enforce byte and chunk budgets, slow clients
+  have progress deadlines, and incomplete burst replay ends only the affected subscriber instead of skipping into live
+  delivery. Cleanup admission is bounded by the new `reverse_proxy.stream.cleanup_queue_capacity` setting (default
+  `4096`); mandatory HLS cleanup uses a reserved control lane. Graceful shutdown closes admission, releases active
+  claims and provider leases, and waits for owned streaming and cleanup workers to finish.
+
 - **Empty playlist updates no longer replace previously published input or target data.** A completely empty refresh is
   treated as a failed update and keeps the last usable playlist and its virtual-ID mapping intact. This prevents
   transient provider/download failures from making channels disappear or assigning different IDs when service

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

@@ -14,7 +14,12 @@ use crate::{
     utils::FileLockManager,
 };
 use arc_swap::{ArcSwap, ArcSwapOption};
-use axum::http::{HeaderMap, Response, StatusCode};
+use axum::{
+    extract::{ConnectInfo, State},
+    http::{HeaderMap, Response, StatusCode},
+    routing::get,
+    Router,
+};
 use bytes::Bytes;
 use futures::stream;
 use http_body_util::BodyExt;
@@ -3633,6 +3638,106 @@ async fn http_provider_lease_smoke_test() {
     assert_eq!(app.active_provider.provider_lease_usage(&input.name).total(), 0, "smoke lease baseline");
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn http_provider_lease_downstream_smoke_test() {
+    const TS_PACKET: [u8; 188] = [0x47; 188];
+    let origin_addr = spawn_load_test_origin(&TS_PACKET).await;
+
+    let input = Arc::new(ConfigInput {
+        id: 1,
+        name: "provider_downstream".intern(),
+        input_type: InputType::Xtream,
+        headers: HashMap::default(),
+        url: format!("http://{origin_addr}"),
+        enabled: true,
+        priority: 0,
+        max_connections: 0,
+        ..ConfigInput::default()
+    });
+    let mut config = create_test_provider_app_config();
+    config.sources =
+        Arc::new(ArcSwap::from_pointee(SourcesConfig { inputs: vec![Arc::clone(&input)], ..SourcesConfig::default() }));
+    let app = create_test_app_state_for_config(Arc::new(config));
+
+    #[derive(Clone)]
+    struct DownstreamState {
+        app: Arc<AppState>,
+        input: Arc<ConfigInput>,
+        origin_addr: SocketAddr,
+    }
+
+    async fn stream_handler(
+        State(state): State<DownstreamState>,
+        ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    ) -> axum::response::Response {
+        let fingerprint = create_test_fingerprint(addr);
+        let user = load_test_user("downstream-user");
+        let session = load_test_session(&state.input, state.origin_addr, "downstream-token", addr);
+        let channel = load_test_channel(&state.input, state.origin_addr);
+        force_provider_stream_response(
+            &fingerprint,
+            &state.app,
+            &session,
+            channel,
+            ForceStreamRequestContext {
+                req_headers: &HeaderMap::new(),
+                input: &state.input,
+                user: &user,
+                session_reservation_ttl_secs: 0,
+                content_representation: crate::api::model::ProviderContentRepresentationMode::Identity,
+            },
+            None,
+        )
+        .await
+        .into_response()
+    }
+
+    let router = Router::new().route("/stream", get(stream_handler)).with_state(DownstreamState {
+        app: Arc::clone(&app),
+        input: Arc::clone(&input),
+        origin_addr,
+    });
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("downstream listener binds");
+    let downstream_addr = listener.local_addr().expect("downstream address");
+    let server = tokio::spawn(async move {
+        let _ = axum::serve(listener, router.into_make_service_with_connect_info::<SocketAddr>()).await;
+    });
+
+    let client = reqwest::Client::new();
+    for concurrency in [1usize, 2] {
+        let mut tasks = tokio::task::JoinSet::new();
+        for task_idx in 0..concurrency {
+            let client = client.clone();
+            let url = format!("http://{downstream_addr}/stream");
+            tasks.spawn(async move {
+                for round in 0..3 {
+                    let response = client.get(&url).send().await.expect("downstream response");
+                    let status = response.status();
+                    let body = response.bytes().await.expect("downstream body");
+                    validate_load_response(status, &body, &TS_PACKET)
+                        .unwrap_or_else(|err| panic!("downstream task {task_idx} round {round}: {err}"));
+                }
+            });
+        }
+        while let Some(res) = tasks.join_next().await {
+            res.expect("downstream task must succeed");
+        }
+    }
+
+    for _ in 0..50 {
+        if app.active_provider.get_provider_connections_count() == 0
+            && app.active_provider.provider_lease_usage(&input.name).total() == 0
+        {
+            break;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+    }
+    assert_eq!(app.active_provider.get_provider_connections_count(), 0, "downstream slots baseline");
+    assert_eq!(app.active_provider.provider_lease_usage(&input.name).total(), 0, "downstream lease baseline");
+
+    server.abort();
+}
+
 #[test]
 fn test_regular_response_keeps_compression_enabled() {
     let response = Response::new(());

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

@@ -143,6 +143,7 @@ struct GracePeriodParams {
     // Whether the session is `socket_bound`. Used to construct the correct
     // `EvictionReentryGuard`.
     socket_bound: bool,
+    shared_subscriber_id: Option<tuliprox_core::model::SharedSubscriberId>,
 }
 
 enum DeferredProviderOpenOutcome {
@@ -930,9 +931,13 @@ pub(crate) async fn create_active_client_stream(
     // before this registration starts. That terminal event also removes this exact user request, so reserving
     // another permit from the same bounded queue would create a circular admission
     // dependency under saturation.
-    let request_uid = stream_details
-        .shared_subscriber_id
-        .map_or_else(|| app_state.connection_manager.next_stream_uid(), |capability| capability.stream_uid());
+    let shared_cleanup_capability = stream_details.shared_subscriber_id.take();
+    let shared_subscriber_id =
+        shared_cleanup_capability.as_ref().map(tuliprox_session::SharedCleanupCapability::subscriber_id);
+    let request_uid = shared_subscriber_id.map_or_else(
+        || app_state.connection_manager.next_stream_uid(),
+        tuliprox_core::model::SharedSubscriberId::stream_uid,
+    );
     let provider_request_id = stream_details
         .provider_handle
         .as_ref()
@@ -952,28 +957,27 @@ pub(crate) async fn create_active_client_stream(
         user_agent,
         session_token,
     };
-    let mut registered_request =
-        if let Some(capability) = stream_details.shared_subscriber_id.filter(|_| stream_channel.shared) {
-            app_state
-                .connection_manager
-                .update_connection_with_uid_using_shared_cleanup(
-                    connection,
-                    tuliprox_session::ConnectionHistoryMode::EmitConnect,
-                    capability,
-                    provider_request_id,
-                )
-                .await
-        } else {
-            app_state
-                .connection_manager
-                .update_connection_with_uid(
-                    connection,
-                    tuliprox_session::ConnectionHistoryMode::EmitConnect,
-                    request_uid,
-                    provider_request_id,
-                )
-                .await
-        };
+    let mut registered_request = if let Some(capability) = shared_cleanup_capability.filter(|_| stream_channel.shared) {
+        app_state
+            .connection_manager
+            .update_connection_with_uid_using_shared_cleanup(
+                connection,
+                tuliprox_session::ConnectionHistoryMode::EmitConnect,
+                capability,
+                provider_request_id,
+            )
+            .await
+    } else {
+        app_state
+            .connection_manager
+            .update_connection_with_uid(
+                connection,
+                tuliprox_session::ConnectionHistoryMode::EmitConnect,
+                request_uid,
+                provider_request_id,
+            )
+            .await
+    };
     let request_uid = registered_request.request_uid;
     let display_stream_uid = registered_request.display_uid();
     let stream_uid = Some(request_uid);
@@ -1078,6 +1082,7 @@ pub(crate) async fn create_active_client_stream(
         grace_resolution_context: owned_grace_ctx,
         grace_kind: Some(connection_kind),
         socket_bound,
+        shared_subscriber_id,
     });
 
     let cfg = &app_state.app_config;
@@ -1208,6 +1213,7 @@ fn stream_grace_period(request: GracePeriodParams) -> (Option<Arc<AtomicU8>>, Op
         grace_resolution_context,
         grace_kind,
         socket_bound,
+        shared_subscriber_id,
         ..
     } = request;
     let grace_period = stream_details.grace_period;
@@ -1339,11 +1345,8 @@ fn stream_grace_period(request: GracePeriodParams) -> (Option<Arc<AtomicU8>>, Op
                                             CustomVideoStreamType::UserConnectionsExhausted,
                                         )
                                         .await;
-                                    if let Some(id) = stream_details.shared_subscriber_id {
-                                        connection_manager
-                                            .shared_stream_manager
-                                            .release_subscriber(id.subscriber_id())
-                                            .await;
+                                    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;
@@ -1358,8 +1361,8 @@ fn stream_grace_period(request: GracePeriodParams) -> (Option<Arc<AtomicU8>>, Op
                                     CustomVideoStreamType::UserConnectionsExhausted,
                                 )
                                 .await;
-                            if let Some(id) = stream_details.shared_subscriber_id {
-                                connection_manager.shared_stream_manager.release_subscriber(id.subscriber_id()).await;
+                            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;
@@ -1404,11 +1407,8 @@ fn stream_grace_period(request: GracePeriodParams) -> (Option<Arc<AtomicU8>>, Op
                                     )
                                     .await;
                                 // Release the shared stream subscription to stop the subscriber loop
-                                if let Some(id) = stream_details.shared_subscriber_id {
-                                    connection_manager
-                                        .shared_stream_manager
-                                        .release_subscriber(id.subscriber_id())
-                                        .await;
+                                if let Some(id) = shared_subscriber_id {
+                                    connection_manager.shared_stream_manager.release_subscriber(id).await;
                                 }
                                 info!("Provider connections exhausted for active clients: {provider_name}");
                             }
@@ -2053,6 +2053,7 @@ mod tests {
             grace_resolution_context: None,
             grace_kind: None,
             socket_bound: false,
+            shared_subscriber_id: None,
         });
         (
             flag.expect("provider grace should install a mode flag"),
@@ -2952,6 +2953,7 @@ mod tests {
             grace_resolution_context: Some(grace_context),
             grace_kind: Some(crate::api::model::ConnectionKind::Soft),
             socket_bound: false,
+            shared_subscriber_id: None,
         });
 
         // Grace should be pending initially.

backend/hls/src/origin.rs

@@ -797,7 +797,7 @@ async fn reserve_cleanup_permit(
     origin_io: &HlsOriginIoContext,
     deadline: Option<Instant>,
 ) -> Result<tokio::sync::mpsc::OwnedPermit<CleanupEvent>, HlsBoundAccountAcquireErrorKind> {
-    let reserve = origin_io.ctx.connection_manager.cleanup_tx().reserve_owned();
+    let reserve = origin_io.ctx.connection_manager.control_cleanup_tx().reserve_owned();
     match deadline {
         Some(deadline) => timeout(deadline.saturating_duration_since(Instant::now()), reserve)
             .await
@@ -1577,12 +1577,12 @@ mod tests {
         session.origin_account_binding = Some(binding.clone());
         let session = Arc::new(tokio::sync::RwLock::new(session));
 
-        // Saturate the cleanup permit queue so `reserve_cleanup_permit` blocks until
+        // Saturate the control cleanup lane so `reserve_cleanup_permit` blocks until
         // the bounded deadline fires.
         let mut held_permits = Vec::new();
         while let Ok(permit) = tokio::time::timeout(
             std::time::Duration::from_millis(1),
-            ctx.connection_manager.cleanup_tx().reserve_owned(),
+            ctx.connection_manager.control_cleanup_tx().reserve_owned(),
         )
         .await
         {

backend/session/src/connection_manager.rs

@@ -26,6 +26,7 @@ use std::{
     time::Duration,
 };
 use tokio::sync::{mpsc, Notify, RwLock, RwLockReadGuard};
+use tokio_util::sync::CancellationToken;
 use tuliprox_core::{
     model::{
         DisconnectQos, Fingerprint, PlaybackRequestOutcome, ProviderHandle, SharedSubscriberId, StreamHistoryConfig,
@@ -37,6 +38,9 @@ use tuliprox_repository::{recover_pending_files, StreamHistoryWriter};
 
 // Maximum number of deferred cleanup actions buffered before producers must wait/drop.
 const CLEANUP_QUEUE_CAPACITY: usize = 4096;
+// Reserved lane for control work (HLS origin I/O, shutdown) so a saturated body-cleanup
+// queue cannot starve the control path that releases provider handles.
+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
@@ -52,7 +56,9 @@ fn notify_capacity(capacity_notify: &Notify) { capacity_notify.notify_waiters();
 /// Proof that a shared-stream response owns the guaranteed terminal cleanup permit for
 /// its subscriber. The constructor is crate-private, so callers outside this crate cannot
 /// fabricate the claim; it is only produced when a cleanup permit is actually reserved.
-#[derive(Debug, Clone, Copy)]
+/// Each registration consumes one capability, so a stale capability cannot be duplicated
+/// and reused after its owner has released.
+#[derive(Debug)]
 pub struct SharedCleanupCapability {
     subscriber_id: SharedSubscriberId,
 }
@@ -606,12 +612,15 @@ pub struct ConnectionManager {
     event_manager: Arc<EventManager>,
     close_socket_signal_tx: tokio::sync::broadcast::Sender<CloseConnectionSignal>,
     cleanup_sender: BackpressureSender<CleanupEvent>,
+    control_cleanup_tx: mpsc::Sender<CleanupEvent>,
     socket_activity_tracker: SocketActivityTracker,
     capacity_notify: Arc<Notify>,
     stream_uid_counter: AtomicU32,
     history_writer: Arc<ArcSwapOption<StreamHistoryWriter>>,
     is_shutting_down: AtomicBool,
     admission_gate: RwLock<()>,
+    shutdown_token: CancellationToken,
+    worker_handles: std::sync::Mutex<Vec<tokio::task::JoinHandle<()>>>,
 }
 
 pub struct ConnectionParams<'a> {
@@ -815,11 +824,13 @@ impl ConnectionManager {
         let history_writer = Arc::new(ArcSwapOption::new(build_history_writer(history_config)));
         let (close_socket_signal_tx, _) = tokio::sync::broadcast::channel(256);
         let (cleanup_tx, cleanup_rx) = mpsc::channel(cleanup_capacity);
+        let (control_cleanup_tx, control_cleanup_rx) = mpsc::channel(CONTROL_CLEANUP_CAPACITY);
         user_manager.set_cleanup_sender(cleanup_tx.clone());
         user_manager.set_provider_manager(Arc::clone(provider_manager));
         let socket_cleanup_tx = cleanup_tx.clone();
         let socket_activity_tracker = SocketActivityTracker::new();
         let capacity_notify = Arc::new(Notify::new());
+        let shutdown_token = CancellationToken::new();
         let mgr = Self {
             user_manager: Arc::clone(user_manager),
             provider_manager: Arc::clone(provider_manager),
@@ -827,24 +838,38 @@ impl ConnectionManager {
             event_manager: Arc::clone(event_manager),
             close_socket_signal_tx,
             cleanup_sender: BackpressureSender::new(cleanup_tx, "cleanup", cleanup_capacity),
+            control_cleanup_tx: control_cleanup_tx.clone(),
             socket_activity_tracker: socket_activity_tracker.clone(),
             capacity_notify: Arc::clone(&capacity_notify),
             stream_uid_counter: AtomicU32::new(1),
             history_writer: Arc::clone(&history_writer),
             is_shutting_down: AtomicBool::new(false),
             admission_gate: RwLock::new(()),
+            shutdown_token: shutdown_token.clone(),
+            worker_handles: std::sync::Mutex::new(Vec::new()),
         };
 
-        Self::spawn_cleanup_worker(
+        let cleanup_handle = Self::spawn_cleanup_worker(
             cleanup_rx,
+            control_cleanup_rx,
             Arc::clone(user_manager),
             Arc::clone(provider_manager),
             Arc::clone(shared_stream_manager),
             Arc::clone(event_manager),
             Arc::clone(&capacity_notify),
             history_writer,
+            shutdown_token.clone(),
         );
-        Self::spawn_socket_activity_worker(socket_activity_tracker, Arc::clone(user_manager), socket_cleanup_tx);
+        let socket_handle = Self::spawn_socket_activity_worker(
+            socket_activity_tracker,
+            Arc::clone(user_manager),
+            socket_cleanup_tx,
+            shutdown_token,
+        );
+        mgr.worker_handles
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .extend([cleanup_handle, socket_handle]);
 
         mgr
     }
@@ -869,7 +894,8 @@ impl ConnectionManager {
         activity_tracker: SocketActivityTracker,
         user_manager: Arc<ActiveUserManager>,
         cleanup_tx: mpsc::Sender<CleanupEvent>,
-    ) {
+        shutdown_token: CancellationToken,
+    ) -> tokio::task::JoinHandle<()> {
         tokio::spawn(async move {
             let mut expiry_queue: BinaryHeap<Reverse<SocketExpiryEntry>> = BinaryHeap::new();
             let mut expiry_index: HashMap<SocketAddr, u64> = HashMap::new();
@@ -900,14 +926,18 @@ impl ConnectionManager {
 
                     tokio::select! {
                         biased;
+                        () = shutdown_token.cancelled() => break,
                         () = activity_tracker.notified() => {}
                         () = tokio::time::sleep(Duration::from_secs(expires_at.saturating_sub(now))) => {}
                     }
                 } else {
-                    activity_tracker.notified().await;
+                    tokio::select! {
+                        () = shutdown_token.cancelled() => break,
+                        () = activity_tracker.notified() => {}
+                    }
                 }
             }
-        });
+        })
     }
 
     async fn drain_pending_socket_activity(
@@ -1007,15 +1037,18 @@ impl ConnectionManager {
             .collect();
     }
 
+    #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
     fn spawn_cleanup_worker(
         mut rx: mpsc::Receiver<CleanupEvent>,
+        mut control_rx: mpsc::Receiver<CleanupEvent>,
         user_manager: Arc<ActiveUserManager>,
         provider_manager: Arc<ActiveProviderManager>,
         shared_stream_manager: Arc<SharedStreamManager>,
         event_manager: Arc<EventManager>,
         capacity_notify: Arc<Notify>,
         history_writer: Arc<ArcSwapOption<StreamHistoryWriter>>,
-    ) {
+        shutdown_token: CancellationToken,
+    ) -> tokio::task::JoinHandle<()> {
         let deps = CleanupWorkerDeps {
             user_manager,
             provider_manager,
@@ -1025,7 +1058,14 @@ impl ConnectionManager {
             history_writer,
         };
         tokio::spawn(async move {
-            while let Some(event) = rx.recv().await {
+            loop {
+                let event = tokio::select! {
+                    biased;
+                    () = shutdown_token.cancelled() => break,
+                    control_event = control_rx.recv() => control_event,
+                    event = rx.recv() => event,
+                };
+                let Some(event) = event else { break };
                 match event {
                     CleanupEvent::ReleaseSharedSubscriber { addr, subscriber_id, request_id, owner } => {
                         deps.shared_stream_manager.release_subscriber(subscriber_id).await;
@@ -1114,13 +1154,17 @@ impl ConnectionManager {
                 }
             }
             debug!("Cleanup worker exiting");
-        });
+        })
     }
 
     pub fn send_cleanup(&self, event: CleanupEvent) { self.cleanup_sender.enqueue(event); }
 
     pub fn cleanup_tx(&self) -> mpsc::Sender<CleanupEvent> { self.cleanup_sender.tx.clone() }
 
+    /// Reserved control lane for cleanup permits that must never be starved by a
+    /// saturated body-cleanup queue (HLS origin I/O, shutdown).
+    pub fn control_cleanup_tx(&self) -> mpsc::Sender<CleanupEvent> { self.control_cleanup_tx.clone() }
+
     pub fn dropped_cleanup_events(&self) -> u64 {
         self.cleanup_sender.dropped_count() + self.user_manager.dropped_cleanup_events.load(Ordering::Relaxed)
     }
@@ -1351,6 +1395,15 @@ impl ConnectionManager {
         if let Some(w) = self.history_writer.load_full() {
             w.shutdown().await;
         }
+        // Stop and join the cleanup and socket workers so no background task outlives
+        // the manager. Active claims were released directly above; later guard drops are
+        // idempotent against the emptied indices.
+        self.shutdown_token.cancel();
+        let handles =
+            std::mem::take(&mut *self.worker_handles.lock().unwrap_or_else(std::sync::PoisonError::into_inner));
+        for handle in handles {
+            let _ = handle.await;
+        }
     }
 
     pub async fn add_connection(&self, addr: &SocketAddr) { self.user_manager.add_connection(addr).await; }
@@ -2376,6 +2429,56 @@ mod tests {
         assert_eq!(registered.rejection_reason(), Some(ConnectionRejectionReason::CleanupAdmissionTimeout));
     }
 
+    #[tokio::test]
+    async fn control_cleanup_lane_precedes_a_ready_body_cleanup_backlog() {
+        let manager = create_test_connection_manager();
+        let release_blocker = Arc::new(Notify::new());
+        let (blocker_started_tx, blocker_started_rx) = tokio::sync::oneshot::channel();
+        let order = Arc::new(std::sync::Mutex::new(Vec::new()));
+
+        manager.send_cleanup(CleanupEvent::Defer(Box::pin({
+            let release_blocker = Arc::clone(&release_blocker);
+            let order = Arc::clone(&order);
+            async move {
+                let _ = blocker_started_tx.send(());
+                release_blocker.notified().await;
+                order.lock().unwrap_or_else(std::sync::PoisonError::into_inner).push('B');
+            }
+        })));
+        blocker_started_rx.await.expect("cleanup blocker must start");
+
+        manager.send_cleanup(CleanupEvent::Defer(Box::pin({
+            let order = Arc::clone(&order);
+            async move { order.lock().unwrap_or_else(std::sync::PoisonError::into_inner).push('N') }
+        })));
+        manager
+            .control_cleanup_tx()
+            .send(CleanupEvent::Defer(Box::pin({
+                let order = Arc::clone(&order);
+                async move { order.lock().unwrap_or_else(std::sync::PoisonError::into_inner).push('C') }
+            })))
+            .await
+            .expect("control cleanup receiver must be open");
+
+        release_blocker.notify_one();
+        tokio::time::timeout(Duration::from_secs(1), async {
+            loop {
+                if order.lock().unwrap_or_else(std::sync::PoisonError::into_inner).len() == 3 {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("both queued cleanup events must run");
+
+        assert_eq!(
+            *order.lock().unwrap_or_else(std::sync::PoisonError::into_inner),
+            vec!['B', 'C', 'N'],
+            "a ready control cleanup must not be starved by the body cleanup backlog"
+        );
+    }
+
     /// A joined shared body already holds the terminal cleanup permit. Registration must
     /// reuse that ownership instead of waiting for a second permit from the same full queue.
     #[tokio::test]

backend/session/src/stream.rs

@@ -92,7 +92,7 @@ pub struct StreamDetails {
 impl Clone for StreamDetails {
     fn clone(&self) -> Self {
         Self {
-            shared_subscriber_id: self.shared_subscriber_id,
+            shared_subscriber_id: None,
             stream: None,
             stream_info: self.stream_info.clone(),
             provider_name: self.provider_name.clone(),

backend/session/src/streams/shared_stream_manager.rs

@@ -13,7 +13,10 @@ use std::{
     future::Future,
     net::SocketAddr,
     pin::Pin,
-    sync::Arc,
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    },
     task::{Context, Poll},
 };
 use tokio::{
@@ -74,6 +77,34 @@ struct SharedMeterEntry {
     owner: Option<AllocationId>,
 }
 
+/// RAII guard for a join that committed its registry entry but has not yet registered
+/// its subscriber. It decrements the origin's pending-join counter exactly once, either
+/// on successful registration (`commit`) or when the join future is dropped.
+struct PendingJoinGuard {
+    state: Arc<SharedStreamState>,
+    armed: bool,
+}
+
+impl PendingJoinGuard {
+    fn new(state: &Arc<SharedStreamState>) -> Self {
+        state.increment_pending_joins();
+        Self { state: Arc::clone(state), armed: true }
+    }
+
+    fn commit(mut self) {
+        self.armed = false;
+        self.state.decrement_pending_joins();
+    }
+}
+
+impl Drop for PendingJoinGuard {
+    fn drop(&mut self) {
+        if self.armed {
+            self.state.decrement_pending_joins();
+        }
+    }
+}
+
 impl PendingSharedSubscriberCleanup {
     fn new(
         permit: tokio::sync::mpsc::OwnedPermit<CleanupEvent>,
@@ -389,9 +420,13 @@ pub struct SharedStreamState {
     stop_token: CancellationToken,
     burst_buffer: Arc<Mutex<BurstBuffer>>,
     live_notification: Arc<Notify>,
-    task_handles: RwLock<Vec<tokio::task::JoinHandle<()>>>,
+    task_handles: std::sync::Mutex<Vec<tokio::task::JoinHandle<()>>>,
     subscriber_idle_timeout_secs: u64,
     subscriber_max_duration: Option<Duration>,
+    /// Number of joins that committed their registry entry but have not yet registered
+    /// their subscriber in `subscribers`. Teardown must not remove an origin while this is
+    /// non-zero, otherwise a join in flight would land on a stopped state.
+    pending_joins: AtomicUsize,
 }
 
 impl SharedStreamState {
@@ -415,9 +450,10 @@ impl SharedStreamState {
             stop_token: CancellationToken::new(),
             burst_buffer: Arc::new(Mutex::new(BurstBuffer::new(burst_buffer_size_in_bytes))),
             live_notification: Arc::new(Notify::new()),
-            task_handles: RwLock::new(Vec::new()),
+            task_handles: std::sync::Mutex::new(Vec::new()),
             subscriber_idle_timeout_secs: DEFAULT_SUBSCRIBER_IDLE_TIMEOUT_SECS,
             subscriber_max_duration: None,
+            pending_joins: AtomicUsize::new(0),
         }
     }
 
@@ -428,6 +464,16 @@ impl SharedStreamState {
         self
     }
 
+    fn increment_pending_joins(&self) { self.pending_joins.fetch_add(1, Ordering::AcqRel); }
+
+    fn decrement_pending_joins(&self) { self.pending_joins.fetch_sub(1, Ordering::AcqRel); }
+
+    fn has_pending_joins(&self) -> bool { self.pending_joins.load(Ordering::Acquire) > 0 }
+
+    fn lock_task_handles(&self) -> std::sync::MutexGuard<'_, Vec<tokio::task::JoinHandle<()>>> {
+        self.task_handles.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+
     async fn register_subscriber(&self, id: SubscriberId, addr: &SocketAddr, cancel_token: CancellationToken) {
         self.subscribers.write().await.insert(id, SharedSubscriber { addr: *addr, cancel_token });
     }
@@ -445,6 +491,7 @@ impl SharedStreamState {
         subscriber_id: SubscriberId,
         connection_manager: Arc<ConnectionManager>,
         mut pending_cleanup: PendingSharedSubscriberCleanup,
+        pending_join: PendingJoinGuard,
     ) -> (BoxedProviderStream, Option<Arc<str>>, SharedCleanupCapability) {
         let (client_tx, client_rx) = mpsc::channel::<BudgetedChunk>(self.buf_size);
         let cancel_token = CancellationToken::new();
@@ -452,11 +499,12 @@ impl SharedStreamState {
         let byte_budget = Arc::new(Semaphore::new(queue_byte_budget));
 
         {
-            let mut handles = self.task_handles.write().await;
+            let mut handles = self.lock_task_handles();
             handles.retain(|h| !h.is_finished());
         }
 
         self.register_subscriber(subscriber_id, addr, cancel_token.clone()).await;
+        pending_join.commit();
         let (start_tx, start_rx) = oneshot::channel();
         let cleanup_manager = Arc::clone(&connection_manager);
 
@@ -495,6 +543,9 @@ impl SharedStreamState {
                 Ok(sent_burst_chunks) => {
                     drop(snapshot);
                     if sent_burst_chunks > 0 {
+                        // The replay renewed its own internal progress time; carry that
+                        // forward so the first live send does not immediately expire.
+                        last_active = Instant::now();
                         debug_if_enabled!(
                             "Shared stream subscriber {} replayed {sent_burst_chunks} burst chunks after {} ms",
                             sanitize_sensitive_info(&address.to_string()),
@@ -682,7 +733,7 @@ impl SharedStreamState {
             });
         });
 
-        self.task_handles.write().await.push(handle);
+        self.lock_task_handles().push(handle);
 
         let provider = self.provider_guard.as_ref().and_then(|h| h.allocation.get_provider_name());
         let (permit, request_id, owner) = pending_cleanup.take_cleanup();
@@ -857,9 +908,9 @@ impl SharedStreamState {
         });
 
         // Keep the broadcast handle so shutdown can join it instead of leaving a detached task.
-        if let Ok(mut handles) = self.task_handles.try_write() {
-            handles.push(broadcast_handle);
-        }
+        // Registration is guaranteed: a short-lived synchronous lock avoids the try_write
+        // failure that would otherwise drop the handle under contention.
+        self.lock_task_handles().push(broadcast_handle);
     }
 }
 
@@ -1070,7 +1121,7 @@ impl SharedStreamManager {
                     if let Some(subscriber) = subscribers.remove(&subscriber_id) {
                         subscriber.cancel_token.cancel();
                     }
-                    if subscribers.is_empty() {
+                    if subscribers.is_empty() && !state.has_pending_joins() {
                         register.by_key.remove(&url);
                         drop(subscribers);
                         Some((url, state))
@@ -1102,7 +1153,7 @@ impl SharedStreamManager {
             // Join the broadcast and forwarder tasks for this origin so teardown is
             // complete before the manager (and its runtime) is dropped. Cancellation has
             // already fired, so these tasks only finish their terminal cleanup.
-            let handles = std::mem::take(&mut *state.task_handles.write().await);
+            let handles = std::mem::take(&mut *state.lock_task_handles());
             for handle in handles {
                 let _ = handle.await;
             }
@@ -1140,7 +1191,7 @@ impl SharedStreamManager {
         // Keep the global registry lock only for the lookup and the key_by_subscriber
         // commit. The origin's own subscriber/task locks are taken during subscribe, so a
         // slow origin must not block joins on other URLs.
-        let state = {
+        let (state, pending_join) = {
             let mut register = self.shared_streams.write().await;
             let Some((key, state)) =
                 register.by_key.get_key_value(stream_url).map(|(key, state)| (Arc::clone(key), Arc::clone(state)))
@@ -1160,9 +1211,12 @@ impl SharedStreamManager {
                 return Ok(None);
             }
             register.key_by_subscriber.insert(subscriber_id, key);
-            state
+            let pending_join = PendingJoinGuard::new(&state);
+            (state, pending_join)
         };
-        Ok(Some(state.subscribe(addr, subscriber_id, Arc::clone(&connection_manager), pending_cleanup).await))
+        Ok(Some(
+            state.subscribe(addr, subscriber_id, Arc::clone(&connection_manager), pending_cleanup, pending_join).await,
+        ))
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -1203,9 +1257,11 @@ impl SharedStreamManager {
                 return None;
             }
             register.key_by_subscriber.insert(subscriber_id, key);
+            let pending_join = PendingJoinGuard::new(&existing);
             drop(register);
-            let response =
-                existing.subscribe(addr, subscriber_id, Arc::clone(ctx.connection_manager), pending_cleanup).await;
+            let response = existing
+                .subscribe(addr, subscriber_id, Arc::clone(ctx.connection_manager), pending_cleanup, pending_join)
+                .await;
             return Some(response);
         }
         let handle = provider_handle.as_ref().and_then(|managed| managed.handle())?;
@@ -1234,9 +1290,11 @@ impl SharedStreamManager {
         register.by_key.insert(Arc::clone(&stream_key), Arc::clone(&shared_state));
         register.key_by_subscriber.insert(subscriber_id, stream_key);
         ctx.shared_stream_manager.adopt_meter_uid(stream_url, allocation_id);
+        let pending_join = PendingJoinGuard::new(&shared_state);
         drop(register);
-        let subscribed_stream =
-            shared_state.subscribe(addr, subscriber_id, Arc::clone(ctx.connection_manager), pending_cleanup).await;
+        let subscribed_stream = shared_state
+            .subscribe(addr, subscriber_id, Arc::clone(ctx.connection_manager), pending_cleanup, pending_join)
+            .await;
         debug_if_enabled!(
             "Shared stream startup register+subscribe completed for {} in {} ms",
             sanitize_sensitive_info(stream_url),
@@ -1651,7 +1709,7 @@ mod tests {
         tokio::time::advance(SHARED_CLEANUP_ADMISSION_TIMEOUT + Duration::from_millis(1)).await;
         assert!(subscribe.await?.is_err(), "saturated cleanup admission must reject the subscriber");
         assert!(state.subscribers.read().await.is_empty(), "rejected admission must not register a subscriber");
-        assert!(state.task_handles.read().await.is_empty(), "rejected admission must not spawn a forwarding task");
+        assert!(state.lock_task_handles().is_empty(), "rejected admission must not spawn a forwarding task");
 
         drop(permits);
         Ok(())
@@ -1928,8 +1986,9 @@ mod tests {
         else {
             panic!("shared subscriber admission failed");
         };
-        let (mut stream, _provider, _capability) =
-            state.subscribe(&addr, subscriber_id, connection_manager, pending_cleanup).await;
+        let (mut stream, _provider, _capability) = state
+            .subscribe(&addr, subscriber_id, connection_manager, pending_cleanup, super::PendingJoinGuard::new(&state))
+            .await;
 
         state.preempted_token.cancel();
         drop(state);
@@ -2467,6 +2526,40 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn pending_join_keeps_origin_alive_when_last_subscriber_leaves() -> Result<(), Box<dyn std::error::Error>> {
+        let app_cfg = Arc::new(create_test_app_config());
+        let events = Arc::new(EventManager::new());
+        let (_providers, _users, manager, connections) = create_test_connection_manager(&app_cfg, &events);
+        let addr = "127.0.0.1:41019".parse()?;
+        let first_id = SharedSubscriberId::from_stream_uid(connections.next_stream_uid());
+        let url = "https://example.invalid/live/pending_join.ts";
+
+        let state = Arc::new(SharedStreamState::new(Vec::new(), CHANNEL_SIZE, None, 1024, None));
+        state.register_subscriber(first_id, &addr, CancellationToken::new()).await;
+        {
+            let mut reg = manager.shared_streams.write().await;
+            reg.by_key.insert(Arc::from(url), Arc::clone(&state));
+            reg.key_by_subscriber.insert(first_id, Arc::from(url));
+        }
+
+        // Simulate a join that committed its registry entry but has not yet registered
+        // its subscriber in the origin.
+        let pending_join = super::PendingJoinGuard::new(&state);
+        assert!(state.has_pending_joins());
+
+        // The last (and only) subscriber leaves; the origin must not be torn down because
+        // the pending join is about to land on it.
+        manager.release_subscriber(first_id).await;
+        assert!(manager.get_shared_state(url).await.is_some(), "origin must survive a pending join");
+
+        // The join completes; the origin has no subscribers and no pending joins.
+        pending_join.commit();
+        assert!(!state.has_pending_joins());
+        assert!(state.subscribers.read().await.is_empty(), "origin has no subscribers after release");
+        Ok(())
+    }
+
     #[test]
     fn uncommitted_shared_meter_registration_is_rolled_back() {
         let app_cfg = create_test_app_config();

docs/src/configuration/connection-handling-runtime-internals.md

@@ -277,11 +277,15 @@ Playback becomes visible/active when the runtime updates connection state throug
 That is the point where:
 
 - logical playback
+- the individual HTTP request claim
 - stream visibility
 - socket/provider tracking
 
 become externally visible as an active running stream.
 
+The visible stream and the request claim are deliberately separate. Segment, range and reconnect requests can reuse one
+logical playback without allowing completion of one body to remove its siblings.
+
 ## 9. Cleanup paths
 
 When the stream ends or is kicked, cleanup may:
@@ -294,9 +298,14 @@ When the stream ends or is kicked, cleanup may:
 
 Important cleanup behaviors:
 
+- registration reserves a bounded cleanup permit before publishing mutable state
+- the response body owns that permit and releases the exact request on EOF, error, cancellation or drop
+- provider cleanup uses the request identity and binding incarnation captured during acquire instead of rediscovering
+  ownership from the socket address
 - kicked cleanup can terminate the associated session immediately
 - adaptive/HLS-style playback may remain preserved for a TTL
 - preserved does not mean counted
+- server shutdown closes new admission, drains active state directly and joins its owned workers
 
 ## 10. Background probe and resolve tasks
 

docs/src/configuration/connection-handling-session-implementation-notes.md

@@ -189,6 +189,52 @@ When multiple users connect through the same reverse proxy socket:
 - Stream details and QoS fallbacks are updated directly by `stream_uid` (`update_stream_detail_by_uid`) rather than
   relying on ambiguous socket lookups.
 
+### Request ownership and binding generations
+
+The runtime uses separate identities for separate ownership scopes:
+
+| Identity | Scope |
+| :--- | :--- |
+| session token | one logical user playback |
+| `PlaybackLeaseId` | one logical provider-capacity lease |
+| `PlaybackRequestId` | one provider request attached to that lease |
+| `ProviderBindingTag` | one generation of the owner's provider-account binding |
+| `SharedSubscriberId` | one viewer attached to a shared MPEG-TS origin |
+
+Parallel range, segment and reconnect requests may belong to the same logical stream while retaining independent request
+claims. Completing one request removes only that claim. The logical stream and provider lease remain until their own
+terminal conditions are met.
+
+A provider binding increments its generation when the same playback moves to a different provider or reactivates an
+idle lease. Cleanup created for an older generation cannot clear the successor binding. This protects new playback from
+delayed body drops, provider errors and asynchronous HLS cleanup belonging to the preceding request.
+
+### Cleanup ownership and overload behavior
+
+Connection registration reserves a cleanup permit before publishing mutable user or provider state. Ownership then
+moves into the response body or shared-subscriber body. Dropping an unpolled body, cancelling an await, reaching EOF or
+reporting an error all converge on request-specific, idempotent cleanup.
+
+The regular cleanup queue is bounded by `reverse_proxy.stream.cleanup_queue_capacity`. Admission waits for a bounded
+interval when all permits are held and then rejects the request instead of registering a stream without guaranteed
+cleanup. HLS origin I/O uses a separate bounded control lane that is preferred over ordinary body cleanup, preventing a
+saturated subscriber workload from starving release of an origin provider handle.
+
+Graceful shutdown first closes the admission gate, then removes shared origins and active user claims, releases provider
+leases, flushes stream history and cancels and joins the owned background workers. Cleanup emitted by a later body drop
+is harmless because the terminal indices have already been emptied.
+
+### Shared MPEG-TS subscriber isolation
+
+A shared origin is keyed by stream URL, but its viewers are keyed by `SharedSubscriberId`, never by `SocketAddr`. This
+allows several users behind one reverse proxy connection to consume independent response bodies without replacing one
+another.
+
+Each subscriber has both a chunk limit and a byte semaphore for queued payload. Burst replay and live delivery use a
+progress deadline that is renewed only after a successful send. Timeout, cancellation and a closed receiver are distinct
+terminal outcomes; an incomplete replay never advances directly to the live tail. A pending-join guard also prevents
+the last existing subscriber from tearing down an origin while another subscriber is committing.
+
 ## Where session tokens come from
 
 ### TS live, VOD, series
@@ -323,11 +369,13 @@ Main entry point:
 
 - `ActiveUserManager::update_connection(...)`
 
-This creates or reuses the tracked logical stream entry.
+This creates or reuses the tracked logical stream entry and attaches a request claim.
 
 Important behavior:
 
 - same `session_token` reuses the logical stream
+- every response receives its own request UID even when it reuses that stream
+- ending one response detaches only its request claim; the stream is removed when its final claim ends
 - stream metrics and duration stay tied to the logical session
 - for non-socket-bound sessions, the session remembers multiple active addresses
 
@@ -386,6 +434,10 @@ For non-socket-bound sessions:
 
 This is the key behavior that keeps VOD and local playback stable across multiple sockets.
 
+Queued provider cleanup carries the `PlaybackRequestId` and owner captured during acquire, while provider handles retain
+their `ProviderBindingTag`. The lease table validates both against the current binding, so cleanup does not need to
+rediscover ownership from a possibly reused socket address.
+
 ## URL persistence rules
 
 Session URL handling is intentionally different by stream type.
@@ -410,14 +462,16 @@ For live playback, following the redirected provider URL is still acceptable and
 
 Read these files together before changing session logic:
 
-- `backend/app/src/api/api_utils.rs`
-- `backend/app/src/api/endpoints/m3u_api.rs`
-- `backend/app/src/api/endpoints/xtream_api.rs`
-- `backend/app/src/api/endpoints/hls_api.rs`
-- `backend/app/src/api/model/active_user_manager.rs`
-- `backend/app/src/api/model/active_provider_manager.rs`
-- `backend/app/src/api/model/metadata_update_manager.rs`
-- `shared/src/model/playlist.rs`
+- `backend/core/src/model/playback.rs`
+- `backend/session/src/provider_leases.rs`
+- `backend/session/src/active_user_manager/mod.rs`
+- `backend/session/src/active_provider_manager.rs`
+- `backend/session/src/connection_manager.rs`
+- `backend/session/src/streams/shared_stream_manager.rs`
+- `backend/app/src/api/api_utils/mod.rs`
+- `backend/app/src/api/model/streams/active_client_stream.rs`
+- `backend/app/src/api/endpoints/hls_api/session.rs`
+- `backend/hls/src/origin.rs`
 
 ## Tests that should stay green
 
@@ -432,6 +486,12 @@ These tests cover the most fragile parts of the current logic:
 - `recently_evicted_vod_uses_session_reentry_guard`
 - `update_session_addr_prunes_previous_registration_for_socket_bound_session`
 - `test_adaptive_session_release_connection_preserves_logical_stream_and_start_time`
+- `stale_direct_cleanup_after_rebind_preserves_successor_request`
+- `subscribers_on_same_socket_do_not_replace_each_other`
+- `send_client_chunk_respects_byte_budget`
+- `control_cleanup_lane_precedes_a_ready_body_cleanup_backlog`
+- `shutdown_with_all_cleanup_permits_held_completes`
+- `http_provider_lease_downstream_smoke_test`
 
 If you change session code and one of these assumptions no longer holds, update the documentation and the tests in the same change.
 

docs/src/configuration/reverse-proxy.md

@@ -109,6 +109,7 @@ reverse_proxy:
     hls_session_ttl_secs: 15
     catchup_session_ttl_secs: 45
     shared_burst_buffer_mb: 12
+    cleanup_queue_capacity: 4096
     metrics_enabled: false
 ```
 
@@ -129,6 +130,7 @@ reverse_proxy:
 | `hls_session_ttl_secs` | Int | `15` | Keeps virtual provider slot open between HLS segment (`.ts`) requests to prevent provider bans for "Account Hopping". |
 | `catchup_session_ttl_secs` | Int | `45` | Same session-holding principle applied to Archive/Catchup TV. See notes on section [Session TTLs for HLS & Catchup](#session-ttls-for-hls-m3u8--catchup) for details. |
 | `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. |
 
 ### 1.1 `retry` & `buffer` (Deep Dive)
 
@@ -196,8 +198,23 @@ to maintain account affinity:
 Important boundary:
 
 * These TTLs are for HLS/catchup style session continuity.
-* Regular TS/VOD/local playback is socket-bound for admission and user-connection counting.
-* Opening the same TS/VOD stream on a second socket consumes another connection or soft slot.
+* Plain TS live playback is socket-bound. A new TS socket is a separate playback attempt.
+* VOD, series and local playback use logical session identities so size, range, seek and reopen requests can continue the
+  same playback across sockets. A genuinely separate playback still consumes another connection or soft slot.
+* A provider reservation becomes capacity-relevant only after Tuliprox forwards real media bytes. Manifests, HEAD
+  requests, keys, maps and error responses do not confirm a provider slot lease.
+
+### 1.5 Cleanup admission and overload behavior
+
+Every admitted streaming response must own a terminal cleanup permit before it can publish user or provider state. The
+permit follows the response body and releases the exact request when the body completes, is dropped without being
+polled, or fails while streaming. Shared subscribers use the same ownership rule without reserving a second permit for
+the same body.
+
+`cleanup_queue_capacity` bounds how many of these cleanup owners may exist at once. If the queue remains saturated for
+the admission deadline, Tuliprox rejects the new request with `503 Service Unavailable`; it does not admit an
+unprotected stream. HLS origin-control cleanup has reserved capacity so ordinary body churn cannot prevent a provider
+handle from being released. Changing the setting requires a server restart.
 
 ---
 
@@ -545,14 +562,14 @@ reverse_proxy:
 
 HLS streams do not consist of an endless TCP pipe. Instead, the player downloads small `.ts` segments every few seconds (e.g., `seg1.ts`, `seg2.ts`).
 
-If Tuliprox released and re-acquired the provider slot for every single segment,
-providers would block the account for "Account Hopping" or spam. Tuliprox simulates a continuous session:
+If Tuliprox selected a different provider account for every segment, providers could treat the traffic as account
+hopping. Tuliprox therefore keeps a logical, confirmed slot lease for the playback:
 
-* `hls_session_ttl_secs: 15`: After a `.ts` segment finishes downloading, the physical slot to the provider is closed,
-  but the "Virtual Slot" for this specific user remains reserved for 15 seconds. No other user can steal this slot
-  during this window. Channel switches from the same client can immediately take over the reservation.
-* The same principle applies to Archive/Catchup TV (`catchup_session_ttl_secs: 45`),
-  which shares the same fragmentation and seeking issues.
+* `hls_session_ttl_secs: 15`: After Tuliprox has forwarded actual media bytes, a cleanly finished HLS request may keep
+  its provider account reserved for this playback for up to 15 seconds. A manifest-only or abandoned start does not
+  create a capacity reservation.
+* Archive/Catchup playback uses the same confirmed-lease principle with `catchup_session_ttl_secs: 45`.
+* Provider errors, preemption, kicks and timeouts release the lease immediately rather than keeping the reconnect window.
 
 ---
 
@@ -577,6 +594,12 @@ To ensure a user who tunes in 10 seconds later doesn't get player errors due to
 Tuliprox continuously keeps the last X Megabytes (`shared_burst_buffer_mb`, default `12`) in RAM.
 It fires this burst buffer at new subscribers so their decoders can instantly synchronize.
 
+Each viewer has a distinct subscriber identity even when several viewers arrive through the same reverse-proxy socket.
+The per-subscriber delivery queue is bounded by both chunk count and retained bytes. A client that stops making progress
+is disconnected after `shared_subscriber_idle_timeout_secs`; it cannot indefinitely retain the burst buffer or stall
+other viewers. If initial burst replay cannot complete, that subscriber ends instead of silently skipping missing data
+and continuing with the live tail.
+
 ---
 
 ### The "VLC Seek Problem" & Grace Periods