fix(streaming): wire worker control planes for real graceful shutdown

Follow-up to a0bf294, which tolerated dead control channels; this makes
them functional:

- AdaptiveProcessor: mpsc control channel (single consumer behind a
  mutex, broke on ANY message including Start) replaced with broadcast;
  all three workers (resource monitor, batch optimizer, pressure
  monitor) subscribe and exit only on ControlCommand::Stop
- EdgeComputingManager / MonitoringSystem: their 7 interval-loop workers
  now subscribe to the existing broadcast control channels and exit on
  Stop instead of looping forever
- stop() in all three: graceful join with 5s timeout, abort only as a
  backstop (previously unconditional abort mid-tick)
- benches: criterion needs async_tokio for Bencher::to_async — bench
  target now compiles (clippy --all-targets clean)

cargo test -p rtx-streaming: 55 lib + 8 integration + 6 aux, all green.

Co-Authored-By: Claude Fable 5 <[email protected]>

crates/production/rtx-streaming/Cargo.toml

@@ -78,7 +78,7 @@ rand = "0.8"
 
 # Development dependencies
 [dev-dependencies]
-criterion = { version = "0.5", features = ["html_reports"] }
+criterion = { version = "0.5", features = ["html_reports", "async_tokio"] }
 tokio-test = "0.4"
 proptest = "1.5"
 approx = "0.5"

crates/production/rtx-streaming/src/adaptive_processing/processor.rs

@@ -11,9 +11,9 @@ use std::{
     time::{Duration, Instant},
 };
 use tokio::{
-    sync::{Mutex as TokioMutex, mpsc},
+    sync::{Mutex as TokioMutex, broadcast},
     task::JoinHandle,
-    time::interval,
+    time::{interval, timeout},
 };
 
 use super::{
@@ -53,9 +53,8 @@ pub struct AdaptiveProcessor {
     /// Metrics collector
     metrics: Arc<AdaptiveProcessingMetrics>,
 
-    /// Control channels
+    /// Control channel — every worker subscribes and exits on `Stop`
-    control_tx: mpsc::UnboundedSender<ControlCommand>,
+    control_tx: broadcast::Sender<ControlCommand>,
-    control_rx: Arc<TokioMutex<mpsc::UnboundedReceiver<ControlCommand>>>,
 
     /// Worker handles
     worker_handles: Arc<TokioMutex<Vec<JoinHandle<()>>>>,
@@ -162,7 +161,7 @@ pub struct AdaptiveProcessingStatistics {
 impl AdaptiveProcessor {
     /// Create a new adaptive processor
     pub async fn new(config: AdaptiveProcessingConfig) -> StreamingResult<Self> {
-        let (control_tx, control_rx) = mpsc::unbounded_channel();
+        let (control_tx, _) = broadcast::channel(16);
 
         let batch_controller =
             Arc::new(DynamicBatchController::new(config.batching_config.clone()));
@@ -190,7 +189,6 @@ impl AdaptiveProcessor {
             config,
             metrics,
             control_tx,
-            control_rx: Arc::new(TokioMutex::new(control_rx)),
             worker_handles: Arc::new(TokioMutex::new(Vec::new())),
         })
     }
@@ -290,9 +288,9 @@ impl AdaptiveProcessor {
 
     /// Start adaptive processing
     pub async fn start(&self) -> StreamingResult<()> {
-        self.control_tx
+        // broadcast::send errors only when there are zero receivers, which is
-            .send(ControlCommand::Start)
+        // the normal state before workers spawn below — not a failure.
-            .map_err(|e| StreamingError::Config(format!("Failed to send start command: {e}")))?;
+        let _ = self.control_tx.send(ControlCommand::Start);
 
         // Start worker tasks
         let mut handles = self.worker_handles.lock().await;
@@ -314,20 +312,22 @@ impl AdaptiveProcessor {
 
     /// Stop adaptive processing
     pub async fn stop(&self) -> StreamingResult<()> {
-        self.control_tx
+        let _ = self.control_tx.send(ControlCommand::Stop);
-            .send(ControlCommand::Stop)
-            .map_err(|e| StreamingError::Config(format!("Failed to send stop command: {e}")))?;
 
-        // The optimizer/pressure workers run unconditional interval loops with
+        // Workers exit on Stop; abort as a backstop if one fails to.
-        // no control channel, so they must be aborted rather than awaited.
         let mut handles = self.worker_handles.lock().await;
-        for handle in handles.drain(..) {
+        for mut handle in handles.drain(..) {
-            handle.abort();
+            match timeout(Duration::from_secs(5), &mut handle).await {
-            if let Err(e) = handle.await
+                Ok(Err(e)) if !e.is_cancelled() => {
-                && !e.is_cancelled()
-            {
                     tracing::warn!("Worker task failed during shutdown: {}", e);
                 }
+                Ok(_) => {}
+                Err(_) => {
+                    tracing::warn!("Worker did not stop within 5s; aborting");
+                    handle.abort();
+                    let _ = handle.await;
+                }
+            }
         }
 
         Ok(())
@@ -347,7 +347,7 @@ impl AdaptiveProcessor {
     // Helper methods for spawning worker tasks
     async fn spawn_resource_monitor(&self) -> StreamingResult<JoinHandle<()>> {
         let resource_monitor = Arc::clone(&self.resource_monitor);
-        let control_rx = Arc::clone(&self.control_rx);
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             // Resource monitoring loop
@@ -359,12 +359,11 @@ impl AdaptiveProcessor {
                         // Collect metrics doesn't return an error
                         let _ = resource_monitor.collect_metrics().await;
                     }
-                    _ = async {
+                    cmd = control_rx.recv() => {
-                        let mut rx = control_rx.lock().await;
+                        match cmd {
-                        rx.recv().await
+                            Ok(ControlCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
-                    } => {
+                            _ => {} // Start/Pause/Resume/lagged: keep running
-                        // Check for stop command
+                        }
-                        break;
                     }
                 }
             }
@@ -376,16 +375,25 @@ impl AdaptiveProcessor {
     async fn spawn_batch_optimizer(&self) -> StreamingResult<JoinHandle<()>> {
         let batch_controller = Arc::clone(&self.batch_controller);
         let config = self.config.batching_config.clone();
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(Duration::from_millis(config.optimization_interval_ms));
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         // Optimize all streams doesn't return an error
                         batch_controller.optimize_all_streams().await;
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(ControlCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)
@@ -393,16 +401,25 @@ impl AdaptiveProcessor {
 
     async fn spawn_pressure_monitor(&self) -> StreamingResult<JoinHandle<()>> {
         let backpressure_manager = Arc::clone(&self.backpressure_manager);
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(Duration::from_millis(50)); // High frequency
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         // Monitor pressure for default stream
                         let _ = backpressure_manager.monitor_pressure("default").await;
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(ControlCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)

crates/production/rtx-streaming/src/edge_computing_original.rs

@@ -1912,8 +1912,8 @@ impl EdgeComputingManager {
 
     /// Start edge computing operations
     pub async fn start(&self) -> StreamingResult<()> {
-        // broadcast::send errors only when there are zero receivers; workers
+        // broadcast::send errors only when there are zero receivers, which is
-        // don't subscribe to the control channel, so that's not a failure.
+        // the normal state before workers subscribe below — not a failure.
         let _ = self.control_tx.send(EdgeControlCommand::Start);
 
         // Start worker tasks
@@ -1938,16 +1938,20 @@ impl EdgeComputingManager {
     pub async fn stop(&self) -> StreamingResult<()> {
         let _ = self.control_tx.send(EdgeControlCommand::Stop);
 
-        // Worker loops are unconditional interval loops with no control-channel
+        // Workers exit on Stop; abort as a backstop if one fails to.
-        // subscription, so they must be aborted rather than awaited.
         let mut handles = self.worker_handles.lock().await;
-        for handle in handles.drain(..) {
+        for mut handle in handles.drain(..) {
-            handle.abort();
+            match tokio::time::timeout(Duration::from_secs(5), &mut handle).await {
-            if let Err(e) = handle.await
+                Ok(Err(e)) if !e.is_cancelled() => {
-                && !e.is_cancelled()
-            {
                     tracing::warn!("Worker task failed during shutdown: {}", e);
                 }
+                Ok(_) => {}
+                Err(_) => {
+                    tracing::warn!("Worker did not stop within 5s; aborting");
+                    handle.abort();
+                    let _ = handle.await;
+                }
+            }
         }
 
         Ok(())
@@ -1956,17 +1960,26 @@ impl EdgeComputingManager {
     // Helper methods for spawning worker tasks
     async fn spawn_resource_monitor(&self) -> StreamingResult<JoinHandle<()>> {
         let resource_optimizer = Arc::clone(&self.resource_optimizer);
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(Duration::from_millis(1000));
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         if let Err(e) = resource_optimizer.monitor_and_optimize() {
                             tracing::error!("Resource monitoring error: {}", e);
                         }
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(EdgeControlCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)
@@ -1974,16 +1987,25 @@ impl EdgeComputingManager {
 
     async fn spawn_connectivity_monitor(&self) -> StreamingResult<JoinHandle<()>> {
         let connectivity_manager = Arc::clone(&self.connectivity_manager);
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(Duration::from_millis(5000));
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         let connectivity_status = connectivity_manager.monitor_connectivity().await;
                         tracing::debug!("Connectivity status: {:?}", connectivity_status);
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(EdgeControlCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)
@@ -1991,17 +2013,26 @@ impl EdgeComputingManager {
 
     async fn spawn_power_manager(&self) -> StreamingResult<JoinHandle<()>> {
         let device_manager = Arc::clone(&self.device_manager);
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(Duration::from_millis(10000));
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         if let Err(e) = device_manager.manage_power("default", PowerMode::Balanced) {
                             tracing::error!("Power management error: {}", e);
                         }
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(EdgeControlCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)

crates/production/rtx-streaming/src/monitoring_original.rs

@@ -1649,8 +1649,8 @@ impl MonitoringSystem {
 
     /// Start monitoring system
     pub async fn start(&self) -> StreamingResult<()> {
-        // broadcast::send errors only when there are zero receivers; workers
+        // broadcast::send errors only when there are zero receivers, which is
-        // don't subscribe to the control channel, so that's not a failure.
+        // the normal state before workers subscribe below — not a failure.
         let _ = self.control_tx.send(MonitoringCommand::Start);
 
         // Start worker tasks
@@ -1679,16 +1679,20 @@ impl MonitoringSystem {
     pub async fn stop(&self) -> StreamingResult<()> {
         let _ = self.control_tx.send(MonitoringCommand::Stop);
 
-        // Worker loops are unconditional interval loops with no control-channel
+        // Workers exit on Stop; abort as a backstop if one fails to.
-        // subscription, so they must be aborted rather than awaited.
         let mut handles = self.worker_handles.lock().await;
-        for handle in handles.drain(..) {
+        for mut handle in handles.drain(..) {
-            handle.abort();
+            match tokio::time::timeout(Duration::from_secs(5), &mut handle).await {
-            if let Err(e) = handle.await
+                Ok(Err(e)) if !e.is_cancelled() => {
-                && !e.is_cancelled()
-            {
                     tracing::warn!("Worker task failed during shutdown: {}", e);
                 }
+                Ok(_) => {}
+                Err(_) => {
+                    tracing::warn!("Worker did not stop within 5s; aborting");
+                    handle.abort();
+                    let _ = handle.await;
+                }
+            }
         }
 
         Ok(())
@@ -1710,17 +1714,26 @@ impl MonitoringSystem {
     async fn spawn_metrics_collector(&self) -> StreamingResult<JoinHandle<()>> {
         let metrics_collector = Arc::clone(&self.metrics_collector);
         let config = self.config.metrics_config.clone();
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(Duration::from_millis(config.collection_interval_ms));
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         if let Err(e) = metrics_collector.collect_metrics().await {
                             tracing::error!("Metrics collection error: {}", e);
                         }
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(MonitoringCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)
@@ -1728,17 +1741,26 @@ impl MonitoringSystem {
 
     async fn spawn_alerting_engine(&self) -> StreamingResult<JoinHandle<()>> {
         let alerting_engine = Arc::clone(&self.alerting_engine);
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(Duration::from_millis(1000));
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         if let Err(e) = alerting_engine.evaluate_rules(HashMap::new()) {
                             tracing::error!("Alert evaluation error: {}", e);
                         }
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(MonitoringCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)
@@ -1747,17 +1769,26 @@ impl MonitoringSystem {
     async fn spawn_health_checker(&self) -> StreamingResult<JoinHandle<()>> {
         let health_checker = Arc::clone(&self.health_checker);
         let config = self.config.health_config.clone();
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(config.interval);
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         if let Err(e) = health_checker.run_health_checks() {
                             tracing::error!("Health check error: {}", e);
                         }
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(MonitoringCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)
@@ -1766,17 +1797,26 @@ impl MonitoringSystem {
     async fn spawn_performance_analyzer(&self) -> StreamingResult<JoinHandle<()>> {
         let performance_analyzer = Arc::clone(&self.performance_analyzer);
         let _config = self.config.performance_config.clone();
+        let mut control_rx = self.control_tx.subscribe();
 
         let handle = tokio::spawn(async move {
             let mut interval = interval(Duration::from_millis(5000));
 
             loop {
-                interval.tick().await;
+                tokio::select! {
-
+                    _ = interval.tick() => {
                         if let Err(e) = performance_analyzer.run_analysis() {
                             tracing::error!("Performance analysis error: {}", e);
                         }
                     }
+                    cmd = control_rx.recv() => {
+                        match cmd {
+                            Ok(MonitoringCommand::Stop) | Err(broadcast::error::RecvError::Closed) => break,
+                            _ => {}
+                        }
+                    }
+                }
+            }
         });
 
         Ok(handle)