diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c4d3e9c5..6b174259a 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/backend/app/src/api/api_utils/tests.rs b/backend/app/src/api/api_utils/tests.rs index a9bddb3e0..ae0fa24cf 100644 --- a/backend/app/src/api/api_utils/tests.rs +++ b/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, + input: Arc, + origin_addr: SocketAddr, + } + + async fn stream_handler( + State(state): State, + ConnectInfo(addr): ConnectInfo, + ) -> 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::()).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(()); diff --git a/backend/app/src/api/model/streams/active_client_stream.rs b/backend/app/src/api/model/streams/active_client_stream.rs index 3826f5ef5..90fb4d19b 100644 --- a/backend/app/src/api/model/streams/active_client_stream.rs +++ b/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, } 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>, 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>, 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>, 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>, 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. diff --git a/backend/hls/src/origin.rs b/backend/hls/src/origin.rs index c58d0a1a3..67fde169b 100644 --- a/backend/hls/src/origin.rs +++ b/backend/hls/src/origin.rs @@ -797,7 +797,7 @@ async fn reserve_cleanup_permit( origin_io: &HlsOriginIoContext, deadline: Option, ) -> Result, 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 { diff --git a/backend/session/src/connection_manager.rs b/backend/session/src/connection_manager.rs index e6a79d174..2c7c45445 100644 --- a/backend/session/src/connection_manager.rs +++ b/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, close_socket_signal_tx: tokio::sync::broadcast::Sender, cleanup_sender: BackpressureSender, + control_cleanup_tx: mpsc::Sender, socket_activity_tracker: SocketActivityTracker, capacity_notify: Arc, stream_uid_counter: AtomicU32, history_writer: Arc>, is_shutting_down: AtomicBool, admission_gate: RwLock<()>, + shutdown_token: CancellationToken, + worker_handles: std::sync::Mutex>>, } 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, cleanup_tx: mpsc::Sender, - ) { + shutdown_token: CancellationToken, + ) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { let mut expiry_queue: BinaryHeap> = BinaryHeap::new(); let mut expiry_index: HashMap = 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, + mut control_rx: mpsc::Receiver, user_manager: Arc, provider_manager: Arc, shared_stream_manager: Arc, event_manager: Arc, capacity_notify: Arc, history_writer: Arc>, - ) { + 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 { 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 { 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] diff --git a/backend/session/src/stream.rs b/backend/session/src/stream.rs index 78cb92767..11a5fcbde 100644 --- a/backend/session/src/stream.rs +++ b/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(), diff --git a/backend/session/src/streams/shared_stream_manager.rs b/backend/session/src/streams/shared_stream_manager.rs index ea3ea71e1..7374cbcfe 100644 --- a/backend/session/src/streams/shared_stream_manager.rs +++ b/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, } +/// 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, + armed: bool, +} + +impl PendingJoinGuard { + fn new(state: &Arc) -> 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, @@ -389,9 +420,13 @@ pub struct SharedStreamState { stop_token: CancellationToken, burst_buffer: Arc>, live_notification: Arc, - task_handles: RwLock>>, + task_handles: std::sync::Mutex>>, subscriber_idle_timeout_secs: u64, subscriber_max_duration: Option, + /// 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>> { + 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, mut pending_cleanup: PendingSharedSubscriberCleanup, + pending_join: PendingJoinGuard, ) -> (BoxedProviderStream, Option>, SharedCleanupCapability) { let (client_tx, client_rx) = mpsc::channel::(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> { + 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(); diff --git a/docs/src/configuration/connection-handling-runtime-internals.md b/docs/src/configuration/connection-handling-runtime-internals.md index 782a52988..1d7629a9b 100644 --- a/docs/src/configuration/connection-handling-runtime-internals.md +++ b/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 diff --git a/docs/src/configuration/connection-handling-session-implementation-notes.md b/docs/src/configuration/connection-handling-session-implementation-notes.md index 73ae7eb40..04a98e734 100644 --- a/docs/src/configuration/connection-handling-session-implementation-notes.md +++ b/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. diff --git a/docs/src/configuration/reverse-proxy.md b/docs/src/configuration/reverse-proxy.md index fea400b68..badc252b0 100644 --- a/docs/src/configuration/reverse-proxy.md +++ b/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