fix(live): ending an HLS session leaves a reader others still use

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<Option<JoinHandle<()>>>,
+    /// 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<std::collections::HashSet<u32>>,
 }
 
 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,15 +529,56 @@ 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);
 
         info!("Stopped live streaming session for monitor {}", monitor_id);
         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<u32> = 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;

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
     }

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();