fix(streaming): sane DynamicBatchingConfig default; worker lifecycle regression tests

The new lifecycle tests caught that the batch-optimizer worker crashed at
spawn: DynamicBatchingConfig derived Default (all zeros), and tokio's
interval() panics on a zero period. Default is now a usable config
(batch 32 in [1,128], step 4, 1ms latency target, 100-sample window,
1s optimization interval, AIMD adaptation).

New regression tests in AdaptiveProcessor, EdgeComputingManager, and
MonitoringSystem assert that all workers are still alive shortly after
start() (catches workers dying at startup) and that stop() completes via
the graceful control-channel path, not the 5s abort backstop (catches
shutdown hangs).

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

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

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

@@ -197,7 +197,7 @@ impl DynamicBatchController {
 // ============================================================================
 
 /// Dynamic batching configuration
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DynamicBatchingConfig {
     /// Initial batch size
     pub initial_batch_size: usize,
@@ -224,6 +224,24 @@ pub struct DynamicBatchingConfig {
     pub optimization_interval_ms: u64,
 }
 
+impl Default for DynamicBatchingConfig {
+    fn default() -> Self {
+        // The derived all-zero Default was unusable: batch sizes of 0 and a
+        // 0ms optimization interval (tokio's `interval` panics on a zero
+        // period, killing the batch-optimizer worker at spawn).
+        Self {
+            initial_batch_size: 32,
+            min_batch_size: 1,
+            max_batch_size: 128,
+            batch_size_step: 4,
+            target_latency_micros: 1000,
+            adaptation_algorithm: BatchAdaptationAlgorithm::default(),
+            stats_window_size: 100,
+            optimization_interval_ms: 1000,
+        }
+    }
+}
+
 /// Batch adaptation algorithms
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum BatchAdaptationAlgorithm {

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

@@ -440,6 +440,37 @@ mod tests {
         assert!(processor.is_ok());
     }
 
+    /// Regression test for two lifecycle bugs: (1) the resource monitor used
+    /// to exit on ANY control message — including the Start sent by start() —
+    /// so it died moments after starting; (2) stop() used to await interval
+    /// loops that never exit, hanging forever. Workers must stay alive after
+    /// start() and exit gracefully on Stop (not via the 5s abort backstop).
+    #[tokio::test]
+    async fn test_worker_lifecycle_start_stop() {
+        let processor = AdaptiveProcessor::new(AdaptiveProcessingConfig::default())
+            .await
+            .unwrap();
+
+        processor.start().await.expect("start failed");
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        {
+            let handles = processor.worker_handles.lock().await;
+            assert_eq!(handles.len(), 3, "expected 3 workers running");
+            assert!(
+                handles.iter().all(|h| !h.is_finished()),
+                "a worker exited right after start"
+            );
+        }
+
+        let stop_started = Instant::now();
+        processor.stop().await.expect("stop failed");
+        assert!(
+            stop_started.elapsed() < Duration::from_secs(4),
+            "stop() fell back to the abort path instead of graceful exit"
+        );
+        assert!(processor.worker_handles.lock().await.is_empty());
+    }
+
     #[tokio::test]
     async fn test_adaptive_batch_processing() {
         let config = AdaptiveProcessingConfig::default();

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

@@ -2173,6 +2173,35 @@ mod tests {
         assert!(manager.is_ok());
     }
 
+    /// Regression test: workers must stay alive after start() and exit
+    /// gracefully on stop() (via the control channel, not the 5s abort
+    /// backstop).
+    #[tokio::test]
+    async fn test_worker_lifecycle_start_stop() {
+        let manager = EdgeComputingManager::new(EdgeComputingConfig::default())
+            .await
+            .unwrap();
+
+        manager.start().await.expect("start failed");
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        {
+            let handles = manager.worker_handles.lock().await;
+            assert_eq!(handles.len(), 3, "expected 3 workers running");
+            assert!(
+                handles.iter().all(|h| !h.is_finished()),
+                "a worker exited right after start"
+            );
+        }
+
+        let stop_started = std::time::Instant::now();
+        manager.stop().await.expect("stop failed");
+        assert!(
+            stop_started.elapsed() < Duration::from_secs(4),
+            "stop() fell back to the abort path instead of graceful exit"
+        );
+        assert!(manager.worker_handles.lock().await.is_empty());
+    }
+
     #[tokio::test]
     async fn test_lightweight_inference() {
         let config = EdgeComputingConfig::default();

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

@@ -1922,6 +1922,35 @@ mod tests {
         assert!(monitoring.is_ok());
     }
 
+    /// Regression test: workers must stay alive after start() and exit
+    /// gracefully on stop() (via the control channel, not the 5s abort
+    /// backstop).
+    #[tokio::test]
+    async fn test_worker_lifecycle_start_stop() {
+        let monitoring = MonitoringSystem::new(MonitoringConfig::default())
+            .await
+            .unwrap();
+
+        monitoring.start().await.expect("start failed");
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        {
+            let handles = monitoring.worker_handles.lock().await;
+            assert_eq!(handles.len(), 4, "expected 4 workers running");
+            assert!(
+                handles.iter().all(|h| !h.is_finished()),
+                "a worker exited right after start"
+            );
+        }
+
+        let stop_started = std::time::Instant::now();
+        monitoring.stop().await.expect("stop failed");
+        assert!(
+            stop_started.elapsed() < Duration::from_secs(4),
+            "stop() fell back to the abort path instead of graceful exit"
+        );
+        assert!(monitoring.worker_handles.lock().await.is_empty());
+    }
+
     #[tokio::test]
     async fn test_stream_metrics_recording() {
         let config = MonitoringConfig::default();