diff --git a/src/streaming/live/coordinator.rs b/src/streaming/live/coordinator.rs index 88591c7..2d85cf9 100644 --- a/src/streaming/live/coordinator.rs +++ b/src/streaming/live/coordinator.rs @@ -130,6 +130,10 @@ pub struct LiveStreamCoordinator { /// Handle to the idle-session reaper task, retained so it is aborted when /// the coordinator is dropped (REVIEW_FIXES_PLAN ยง3.2). idle_watchdog: std::sync::Mutex>>, + /// Readers `stop_session` left running because something else still read + /// them. The watchdog stops each once nothing does and no session has + /// taken it back (#256). + kept_readers: std::sync::Mutex>, } impl LiveStreamCoordinator { @@ -143,6 +147,7 @@ impl LiveStreamCoordinator { hls_manager, sessions: Arc::new(RwLock::new(HashMap::new())), idle_watchdog: std::sync::Mutex::new(None), + kept_readers: Default::default(), } } @@ -185,6 +190,7 @@ impl LiveStreamCoordinator { warn!("Failed to reap idle session {}: {}", monitor_id, e); } } + coord.stop_unread_kept_readers().await; // Drop the strong ref before sleeping so it never pins the // coordinator across the idle interval. drop(coord); @@ -509,9 +515,11 @@ impl LiveStreamCoordinator { session.status = SessionStatus::Stopping; - // Abort the processing task + // Abort the processing task, and wait for it to go so its + // subscriptions to the source are dropped before they're counted. if let Some(handle) = session.task_handle.take() { handle.abort(); + let _ = handle.await; } // Stop HLS session if enabled @@ -521,8 +529,24 @@ impl LiveStreamCoordinator { } } - // Stop the source reader - let _ = self.source_router.stop_reader(monitor_id).await; + // Stop the source reader, unless something else still reads it: a + // WebRTC or WHEP viewer, or an RTSP client, shares this reader, and + // the idle reaper ends a session nobody fetches HLS from (#256). + let others = self + .source_router + .get_existing_source(monitor_id) + .map_or(0, |s| { + s.video_subscriber_count() + s.audio_subscriber_count() + }); + if others == 0 { + let _ = self.source_router.stop_reader(monitor_id).await; + } else { + info!( + "Leaving monitor {}'s reader running for {} other subscriber(s)", + monitor_id, others + ); + self.kept_readers.lock().unwrap().insert(monitor_id); + } sessions.remove(&monitor_id); @@ -530,6 +554,31 @@ impl LiveStreamCoordinator { Ok(()) } + /// Stop the readers [`stop_session`](Self::stop_session) left running + /// for other subscribers, once none are left. A reader a new session has + /// taken back is that session's again. Called by the idle watchdog each + /// tick. + pub async fn stop_unread_kept_readers(&self) { + let kept: Vec = self.kept_readers.lock().unwrap().iter().copied().collect(); + if kept.is_empty() { + return; + } + let sessions = self.sessions.read().await; + for monitor_id in kept { + let unread = self + .source_router + .get_existing_source(monitor_id) + .is_none_or(|s| s.video_subscriber_count() + s.audio_subscriber_count() == 0); + if sessions.contains_key(&monitor_id) { + self.kept_readers.lock().unwrap().remove(&monitor_id); + } else if unread { + let _ = self.source_router.stop_reader(monitor_id).await; + self.kept_readers.lock().unwrap().remove(&monitor_id); + info!("Stopped monitor {}'s reader: nothing reads it", monitor_id); + } + } + } + /// Check if a session exists pub async fn has_session(&self, monitor_id: u32) -> bool { let sessions = self.sessions.read().await; diff --git a/tests/common/harness.rs b/tests/common/harness.rs index 9b16b66..f1122df 100644 --- a/tests/common/harness.rs +++ b/tests/common/harness.rs @@ -54,6 +54,16 @@ impl TestApp { /// reading stream sockets from `socks_dir` (a /// [`FakeZmc`](super::fake_zmc::FakeZmc)'s). pub async fn spawn_live(socks_dir: &std::path::Path) -> Self { + Self::spawn_live_with(socks_dir, |_| {}).await + } + + /// [`spawn_live`](Self::spawn_live), then `adjust` on the live state + /// (to keep a handle on its coordinator, say). + #[allow(dead_code)] + pub async fn spawn_live_with( + socks_dir: &std::path::Path, + adjust: impl FnOnce(&mut AppState), + ) -> Self { let socks = socks_dir.to_string_lossy().into_owned(); Self::spawn_with(|state| { let mut config = (*state.config).clone(); @@ -65,6 +75,7 @@ impl TestApp { config.streaming.zoneminder.socks_path = socks; state.config = std::sync::Arc::new(config); state.enable_live_streaming(); + adjust(state); }) .await } diff --git a/tests/it_live_streaming.rs b/tests/it_live_streaming.rs index bde3556..d73e9b5 100644 --- a/tests/it_live_streaming.rs +++ b/tests/it_live_streaming.rs @@ -161,6 +161,77 @@ async fn live_hls_plays() { assert_eq!(r.status(), StatusCode::NO_CONTENT, "{}", r.text()); } +/// Ending a monitor's HLS session (the idle reaper does this 90 s after the +/// last playlist fetch) leaves the stream-socket reader running while +/// something else still reads from it, such as a WebRTC viewer (#256). +#[tokio::test] +#[ignore = "requires the test database (APP_PROFILE=test-db)"] +async fn ending_hls_keeps_a_reader_others_still_use() { + let app_db = TestApp::spawn().await; + let m = insert_monitor(&app_db.db, "LiveReap") + .await + .expect("monitor"); + let _m = RowGuard::monitor(m.id); + let zmc = FakeZmc::start(m.id); + let mut live = None; + let app = TestApp::spawn_live_with(zmc.socks_dir(), |state| { + live = Some(( + state.live_coordinator.clone().expect("coordinator"), + state.source_router.clone().expect("router"), + )); + }) + .await; + let (coordinator, router) = live.unwrap(); + let token = superuser_token(); + + let r = app + .post_json( + &format!("/api/v3/live/{}/start", m.id), + &token, + &json!({ "enable_hls": true }), + ) + .await; + assert_eq!(r.status(), StatusCode::OK, "{}", r.text()); + + // A second reader of the same source, as a WebRTC peer is. + let source = router.get_existing_source(m.id).expect("source"); + let mut viewer = source.subscribe_video(); + tokio::time::timeout(Duration::from_secs(10), viewer.recv()) + .await + .expect("video before the reap") + .expect("open channel"); + + coordinator.stop_session(m.id).await.expect("stop HLS"); + + // Skip what was sent before the stop; new frames must keep coming. + tokio::time::sleep(Duration::from_millis(500)).await; + viewer = viewer.resubscribe(); + tokio::time::timeout(Duration::from_secs(3), viewer.recv()) + .await + .expect("the reader stopped under a viewer") + .expect("open channel"); + + // When that viewer leaves, the watchdog's next pass stops the reader. + drop(viewer); + coordinator.stop_unread_kept_readers().await; + assert!(!source.is_active().await, "kept reader left running unread"); + + // With nobody left, stopping a session does stop the reader. + let r = app + .post_json( + &format!("/api/v3/live/{}/start", m.id), + &token, + &json!({ "enable_hls": true }), + ) + .await; + assert_eq!(r.status(), StatusCode::OK, "{}", r.text()); + coordinator.stop_session(m.id).await.expect("stop HLS"); + assert!( + !source.is_active().await, + "reader left running with no one reading" + ); +} + /// The video size of `path` and how many frames decode from it. fn decode(path: &std::path::Path) -> (u32, u32, usize) { ffmpeg_next::init().unwrap();