fix: leak-fixes — stop pipeline on delete + overlay renderer cleanup (#530)

* fix: release pipeline resources on flow delete and overlay teardown

- delete_flow now stops the active pipeline before removing the flow
  record. Previously the PipelineManager stayed in AppState.pipelines
  after the flow record was gone, with no API path left to reach it.

- stop_flow now also unregisters the vision-mixer overlay renderer
  alongside the overlay state. Both share the same lifecycle and must
  be cleared together; otherwise the renderer entry keeps a background
  thread alive past pipeline shutdown.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* test: cover delete-while-running and overlay-registry teardown

- test_delete_running_flow_releases_pipeline starts a flow, captures
  weak refs to the pipeline and its elements, calls delete_flow without
  a prior stop, and asserts the pipeline is removed from
  AppState.pipelines and that no element survives.

- overlay_registries_round_trip exercises the register/unregister pair
  for both overlay_states and overlay_renderers, locking in the
  shared-lifecycle invariant so a future per-block registry isn't
  forgotten in stop_flow.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix: log delete-time stop_flow failure as error

A failed stop_flow during delete_flow leaves the PipelineManager orphaned
(the very leak the prior commit fixes), so it deserves error-level
visibility, not warn.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Per Enstedt <per.enstedt@svt.se>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

backend/src/blocks/builtin/vision_mixer/tests.rs

@@ -111,3 +111,75 @@ fn test_parse_initial_pgm_clamped() {
     props.insert("initial_pgm_input".to_string(), PropertyValue::UInt(99));
     assert_eq!(properties::parse_initial_pgm(&props, 4), 3); // max index = 3
 }
+
+/// `overlay_states` and `overlay_renderers` share the same lifecycle:
+/// they are populated together in `build_overlay` and must be cleared
+/// together by the cleanup branch in `state.rs::stop_flow`. If you add
+/// another per-block registry in this module, mirror its unregister call
+/// in that branch and extend this test.
+#[test]
+fn overlay_registries_round_trip() {
+    use super::overlay::{
+        get_overlay_renderer, get_overlay_state, register_overlay_renderer, register_overlay_state,
+        unregister_overlay_renderer, unregister_overlay_state, OverlayRenderer,
+        VisionMixerOverlayState,
+    };
+    use gstreamer as gst;
+    use gstreamer_app as gst_app;
+    use std::sync::{Arc, Mutex};
+
+    gst::init().unwrap();
+
+    let block_id = "test-vm-overlay-cleanup-block-id";
+
+    let lo = layout::compute_layout(1280, 720, 4);
+    let state = Arc::new(VisionMixerOverlayState::new(
+        4,
+        0,
+        0,
+        1,
+        vec!["A".into(), "B".into(), "C".into(), "D".into()],
+        lo,
+        false,
+    ));
+
+    let caps = gst::Caps::builder("video/x-raw")
+        .field("format", "BGRA")
+        .field("width", 1280i32)
+        .field("height", 720i32)
+        .field("framerate", gst::Fraction::new(50, 1))
+        .build();
+    let appsrc = gst_app::AppSrc::builder().caps(&caps).build();
+
+    let renderer = Arc::new(Mutex::new(OverlayRenderer::new(
+        appsrc,
+        caps,
+        Arc::clone(&state),
+        1280,
+        720,
+    )));
+
+    register_overlay_state(block_id, Arc::clone(&state));
+    register_overlay_renderer(block_id, Arc::clone(&renderer));
+
+    assert!(
+        get_overlay_state(block_id).is_some(),
+        "state should be registered"
+    );
+    assert!(
+        get_overlay_renderer(block_id).is_some(),
+        "renderer should be registered"
+    );
+
+    unregister_overlay_state(block_id);
+    unregister_overlay_renderer(block_id);
+
+    assert!(
+        get_overlay_state(block_id).is_none(),
+        "state must be cleaned (otherwise API still sees stale block)"
+    );
+    assert!(
+        get_overlay_renderer(block_id).is_none(),
+        "renderer must be cleaned (otherwise overlay-timer-* thread leaks)"
+    );
+}

backend/src/state.rs

@@ -631,6 +631,22 @@ impl AppState {
             return Ok(false);
         }
 
+        // Stop the pipeline first if it is still running. Without this, the
+        // PipelineManager stays in self.inner.pipelines after the flow record
+        // is gone, with no API path left to reach it.
+        let pipeline_active = {
+            let pipelines = self.inner.pipelines.read().await;
+            pipelines.contains_key(id)
+        };
+        if pipeline_active {
+            if let Err(e) = self.stop_flow(id).await {
+                error!(
+                    "Failed to stop flow {} before delete: {} — pipeline resources may leak",
+                    id, e
+                );
+            }
+        }
+
         // Delete from storage first (skip for ephemeral flows)
         let is_ephemeral = {
             let flows = self.inner.flows.read().await;
@@ -1365,6 +1381,12 @@ impl AppState {
                     crate::blocks::builtin::vision_mixer::overlay::unregister_overlay_state(
                         &block.id,
                     );
+                    // Without this, the overlay-timer-* thread keeps polling the
+                    // renderer registry, holds a strong AppSrc ref, and prevents
+                    // the pipeline (and its NiceAgent) from finalizing.
+                    crate::blocks::builtin::vision_mixer::overlay::unregister_overlay_renderer(
+                        &block.id,
+                    );
                 }
             }
         }

backend/tests/pipeline_lifecycle_test.rs

@@ -8,6 +8,8 @@ use std::collections::HashMap;
 use strom::blocks::BlockRegistry;
 use strom::events::EventBroadcaster;
 use strom::gst::pipeline::PipelineManager;
+use strom::state::AppState;
+use strom::storage::JsonFileStorage;
 use strom_types::{Flow, Link};
 use tempfile::NamedTempFile;
 
@@ -162,3 +164,81 @@ async fn test_leak_detection_catches_circular_reference() {
          leak detection would miss real leaks!"
     );
 }
+
+/// Regression test for the delete-without-stop leak: deleting a flow that has
+/// an active pipeline must fully release the pipeline (encoders, sockets,
+/// element refs). Before the fix in delete_flow, the PipelineManager stayed
+/// in AppState.pipelines after deletion and orphaned NVENC sessions / sockets
+/// would accumulate until process exit.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn test_delete_running_flow_releases_pipeline() {
+    gstreamer::init().unwrap();
+
+    let storage_file = NamedTempFile::new().unwrap();
+    let blocks_file = NamedTempFile::new().unwrap();
+    let storage = JsonFileStorage::new(storage_file.path());
+
+    let state = AppState::new(
+        storage,
+        blocks_file.path(),
+        std::env::temp_dir(),
+        vec![],
+        "all".to_string(),
+        vec![],
+    );
+
+    let flow = build_test_flow("delete_running_flow_test");
+    let flow_id = flow.id;
+    state.upsert_flow(flow).await.expect("upsert_flow failed");
+
+    state.start_flow(&flow_id).await.expect("start_flow failed");
+
+    // Capture weak refs to the running pipeline before deletion
+    let (pipeline_weak, element_weak_refs) = {
+        let pipelines = state.pipelines_read().await;
+        let manager = pipelines
+            .get(&flow_id)
+            .expect("Pipeline should be active after start_flow");
+        (manager.pipeline_weak(), manager.element_weak_refs())
+    };
+    assert!(
+        pipeline_weak.upgrade().is_some(),
+        "Pipeline should be alive before delete"
+    );
+    assert!(
+        !element_weak_refs.is_empty(),
+        "Pipeline should have elements"
+    );
+
+    // Delete WITHOUT calling stop_flow first — this is the path the UI takes
+    let deleted = state
+        .delete_flow(&flow_id)
+        .await
+        .expect("delete_flow failed");
+    assert!(deleted, "delete_flow should report the flow was deleted");
+
+    // Pipeline must no longer be tracked in AppState
+    {
+        let pipelines = state.pipelines_read().await;
+        assert!(
+            !pipelines.contains_key(&flow_id),
+            "Pipeline still in AppState.pipelines after delete — orphaned manager"
+        );
+    }
+
+    // And every GStreamer object must be finalized — surviving refs would
+    // mean a leaked encoder / socket / thread that no API path can release.
+    assert!(
+        pipeline_weak.upgrade().is_none(),
+        "Pipeline still alive after delete_flow — resources leaked"
+    );
+    let leaked: Vec<_> = element_weak_refs
+        .iter()
+        .filter_map(|(name, weak)| weak.upgrade().map(|_| name.clone()))
+        .collect();
+    assert!(
+        leaked.is_empty(),
+        "Elements still alive after delete_flow: {:?}",
+        leaked
+    );
+}