rusty_bin/monitor/

optimized_collection_manager.rs

//! Optimized collection manager using lock-free data structures
//!
//! This module provides a high-performance collection manager that leverages
//! lock-free data structures for minimal latency in high-frequency trading environments.

use crate::monitor::{
    collector::{DataType, ExchangeClient, MarketDataEvent, Result},
    lockfree_buffer_pool::{LockFreeBufferPool, LockFreeBufferPoolConfig},
    lockfree_exchange_manager::{ExchangeManagerStats, LockFreeExchangeManager},
    lockfree_stats::{LockFreeStatsCollector, StatsSnapshot},
    storage::manager::StorageManager,
    zerocopy_pipeline::{ZerocopyDataPipeline, ZerocopyPipelineConfig},
};
use flume::Sender;
use quanta::{Clock, Instant as QuantaInstant};
use smartstring::alias::String as SmartString;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;

/// Configuration for the optimized collection manager
#[derive(Debug, Clone)]
pub struct OptimizedCollectionConfig {
    /// Exchange connection timeout in milliseconds
    ///
    /// Controls how long to wait for an exchange connection to be established
    /// before timing out. Default is 30 seconds (30000ms).
    pub connection_timeout_ms: u64,
    /// Health check interval in milliseconds
    ///
    /// Determines how frequently the system performs health checks on exchange
    /// connections to ensure they remain active. Default is 30 seconds (30000ms).
    pub health_check_interval_ms: u64,
    /// Retry interval for failed connections in milliseconds
    ///
    /// Specifies the delay between retry attempts when reconnecting to
    /// exchanges after connection failures. Default is 60 seconds (60000ms).
    pub retry_interval_ms: u64,
    /// Maximum number of retry attempts
    ///
    /// Limits the number of times the system will attempt to reconnect to
    /// an exchange before marking it as permanently failed. Default is 10.
    pub max_retry_attempts: usize,
    /// Buffer pool configuration for lock-free operations
    ///
    /// Configures the lock-free buffer pool used for high-performance
    /// memory management during market data processing.
    pub buffer_pool_config: LockFreeBufferPoolConfig,
    /// Zero-copy pipeline configuration
    ///
    /// Configures the zero-copy data pipeline for ultra-low-latency
    /// processing of market data streams without unnecessary allocations.
    pub pipeline_config: ZerocopyPipelineConfig,
    /// Enable automatic cleanup of old statistics
    ///
    /// When enabled, the system will periodically clean up old statistical
    /// data to prevent memory bloat. Default is true.
    pub enable_stats_cleanup: bool,
    /// Statistics cleanup interval in milliseconds
    ///
    /// Determines how frequently the system cleans up old statistical data
    /// when automatic cleanup is enabled. Default is 5 minutes (300000ms).
    pub stats_cleanup_interval_ms: u64,
}

/// Default configuration values for the optimized collection manager
///
/// Provides sensible defaults for high-frequency trading environments,
/// with conservative timeouts and aggressive performance optimizations.
impl Default for OptimizedCollectionConfig {
    fn default() -> Self {
        Self {
            connection_timeout_ms: 30000,    // 30 seconds
            health_check_interval_ms: 30000, // 30 seconds
            retry_interval_ms: 60000,        // 60 seconds
            max_retry_attempts: 10,
            buffer_pool_config: LockFreeBufferPoolConfig::default(),
            pipeline_config: ZerocopyPipelineConfig::default(),
            enable_stats_cleanup: true,
            stats_cleanup_interval_ms: 300000, // 5 minutes
        }
    }
}

/// Optimized collection manager with lock-free optimizations
pub struct OptimizedCollectionManager {
    /// Configuration
    config: OptimizedCollectionConfig,

    /// Lock-free exchange manager
    exchange_manager: Arc<LockFreeExchangeManager>,

    /// Lock-free statistics collector
    stats_collector: Arc<LockFreeStatsCollector>,

    /// Lock-free buffer pool
    buffer_pool: Arc<LockFreeBufferPool>,

    /// Zero-copy data pipeline
    zerocopy_pipeline: Option<ZerocopyDataPipeline>,

    /// Storage manager
    storage_manager: Arc<StorageManager>,

    /// Management task handles
    management_handles: Vec<JoinHandle<()>>,

    /// Shutdown coordination
    shutdown_sender: Option<oneshot::Sender<()>>,
    shutdown_requested: Arc<AtomicBool>,

    /// High-resolution timing
    clock: Clock,
    start_time: QuantaInstant,

    /// Atomic counters for ultra-fast metrics
    events_processed_total: Arc<AtomicU64>,
    last_health_check: Arc<AtomicU64>,
    last_stats_cleanup: Arc<AtomicU64>,
}

impl OptimizedCollectionManager {
    /// Create a new optimized collection manager
    pub async fn new(
        config: OptimizedCollectionConfig,
        storage_manager: Arc<StorageManager>,
    ) -> Result<Self> {
        let exchange_manager = Arc::new(LockFreeExchangeManager::new());
        let stats_collector = Arc::new(LockFreeStatsCollector::new());
        let buffer_pool = Arc::new(LockFreeBufferPool::new(config.buffer_pool_config.clone()));

        let clock = Clock::new();
        let start_time = clock.now();

        Ok(Self {
            config,
            exchange_manager,
            stats_collector,
            buffer_pool,
            zerocopy_pipeline: None,
            storage_manager,
            management_handles: Vec::new(),
            shutdown_sender: None,
            shutdown_requested: Arc::new(AtomicBool::new(false)),
            clock,
            start_time,
            events_processed_total: Arc::new(AtomicU64::new(0)),
            last_health_check: Arc::new(AtomicU64::new(0)),
            last_stats_cleanup: Arc::new(AtomicU64::new(0)),
        })
    }

    /// Start optimized collection with lock-free coordination
    pub async fn start_collection(
        &mut self,
        exchange_filter: Option<Vec<SmartString>>,
        symbol_filter: Option<Vec<SmartString>>,
        event_sender: Sender<MarketDataEvent>,
    ) -> Result<()> {
        log::info!("Starting optimized collection manager with lock-free optimizations");

        // Determine exchanges and symbols
        let exchanges = exchange_filter.unwrap_or_else(|| vec!["binance".into()]);
        let symbols = symbol_filter.unwrap_or_else(|| {
            vec![
                "BTCUSDT".into(),
                "ETHUSDT".into(),
                "ADAUSDT".into(),
                "SOLUSDT".into(),
            ]
        });

        // Initialize zero-copy pipeline
        let mut pipeline = ZerocopyDataPipeline::new(
            self.config.pipeline_config.clone(),
            self.storage_manager.clone(),
        );

        // Create processing channels with optimal buffer sizes
        let (processing_sender, processing_receiver) =
            flume::bounded(self.config.pipeline_config.worker_buffer_size);

        // Create internal event routing channels
        let (internal_sender, internal_receiver) =
            flume::bounded(self.config.pipeline_config.worker_buffer_size);

        // Start the zero-copy pipeline
        pipeline.start(processing_receiver).await?;
        self.zerocopy_pipeline = Some(pipeline);

        // Start management tasks
        self.start_management_tasks(internal_receiver, processing_sender.clone())
            .await?;

        // Define data types to collect
        let data_types = vec![DataType::Trades, DataType::OrderBook];

        // Initialize exchanges with parallel connection attempts
        let mut connection_futures = Vec::new();
        for exchange_name in exchanges {
            let manager = self.exchange_manager.clone();
            let stats = self.stats_collector.clone();
            let exchange_name_clone = exchange_name.clone();
            let symbols_clone = symbols.clone();
            let data_types_clone = data_types.clone();
            let processing_sender_clone = processing_sender.clone();

            let future = async move {
                // Create ExchangeClient for market data collection (not trading)
                let mut exchange_client = ExchangeClient::new(exchange_name_clone.clone());

                // Connect to exchange
                match exchange_client.connect().await {
                    Ok(()) => {
                        log::info!("Successfully connected to exchange: {exchange_name_clone}");

                        // Add exchange to manager
                        if let Err(e) = manager
                            .add_exchange(exchange_name_clone.clone(), exchange_client)
                            .await
                        {
                            log::error!(
                                "Failed to add exchange {exchange_name_clone} to manager: {e}"
                            );
                            stats.record_connection_error();
                            return Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>);
                        }

                        // Start collection for symbols
                        if let Err(e) = manager
                            .start_collection(
                                &exchange_name_clone,
                                symbols_clone,
                                data_types_clone,
                                processing_sender_clone,
                            )
                            .await
                        {
                            log::error!(
                                "Failed to start collection for exchange {exchange_name_clone}: {e}"
                            );
                            stats.record_connection_error();
                            return Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>);
                        }

                        log::info!(
                            "Successfully started collection for exchange: {exchange_name_clone}"
                        );
                    }
                    Err(e) => {
                        log::error!("Failed to connect to exchange {exchange_name_clone}: {e}");
                        stats.record_connection_error();
                        return Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>);
                    }
                }

                Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
            };

            connection_futures.push(tokio::spawn(future));
        }

        // Wait for all connections to complete
        for future in connection_futures {
            if let Err(e) = future
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?
            {
                log::error!("Failed to initialize exchange connection: {e}");
                self.stats_collector.record_connection_error();
            }
        }

        log::info!("Optimized collection manager started successfully");
        Ok(())
    }

    /// Start management tasks using lock-free coordination
    async fn start_management_tasks(
        &mut self,
        event_receiver: flume::Receiver<MarketDataEvent>,
        processing_sender: Sender<MarketDataEvent>,
    ) -> Result<()> {
        let (shutdown_sender, mut shutdown_receiver) = oneshot::channel();
        self.shutdown_sender = Some(shutdown_sender);

        // Event routing task (high-frequency, lock-free)
        {
            let stats = self.stats_collector.clone();
            let exchange_manager = self.exchange_manager.clone();
            let shutdown_requested = self.shutdown_requested.clone();
            let events_processed = self.events_processed_total.clone();

            let routing_handle = tokio::spawn(async move {
                let mut event_batch = Vec::with_capacity(1000);
                let mut last_batch_time = QuantaInstant::now();
                const BATCH_TIMEOUT_NANOS: u64 = 1_000_000; // 1ms

                loop {
                    tokio::select! {
                        // Receive events
                        event_result = event_receiver.recv_async() => {
                            match event_result {
                                Ok(event) => {
                                    // Record statistics (lock-free)
                                    match &event {
                                        MarketDataEvent::Trade(_) => {
                                            stats.record_trades(1);
                                            stats.record_event();
                                        }
                                        MarketDataEvent::OrderBook(_) => {
                                            stats.record_orderbooks(1);
                                            stats.record_event();
                                        }
                                        MarketDataEvent::ConnectionStatus { exchange, connected, .. } => {
                                            if *connected {
                                                log::debug!("Exchange {exchange} connected");
                                            } else {
                                                exchange_manager.record_error(exchange);
                                            }
                                        }
                                        MarketDataEvent::Error { exchange, .. } => {
                                            exchange_manager.record_error(exchange);
                                            stats.record_parsing_error();
                                        }
                                    }

                                    event_batch.push(event);
                                    events_processed.fetch_add(1, Ordering::Relaxed);

                                    // Send batch when full or timeout reached
                                    let should_send_batch = event_batch.len() >= 1000 ||
                                        last_batch_time.elapsed().as_nanos() as u64 >= BATCH_TIMEOUT_NANOS;

                                    if should_send_batch && !event_batch.is_empty() {
                                        for event in event_batch.drain(..) {
                                            if processing_sender.send(event).is_err() {
                                                log::warn!("Processing channel closed");
                                                return;
                                            }
                                        }
                                        last_batch_time = QuantaInstant::now();
                                        stats.record_buffer_reuse();
                                    }
                                }
                                Err(_) => {
                                    log::info!("Event channel closed, stopping routing task");
                                    break;
                                }
                            }
                        }

                        // Check for shutdown
                        _ = &mut shutdown_receiver => {
                            log::info!("Event routing task received shutdown signal");
                            break;
                        }
                    }

                    // Break if shutdown requested
                    if shutdown_requested.load(Ordering::Acquire) {
                        break;
                    }
                }

                // Send remaining events
                for event in event_batch {
                    let _ = processing_sender.send(event);
                }

                log::info!("Event routing task stopped");
            });

            self.management_handles.push(routing_handle);
        }

        // Health check task (periodic, lock-free coordination)
        {
            let exchange_manager = self.exchange_manager.clone();
            let stats = self.stats_collector.clone();
            let shutdown_requested = self.shutdown_requested.clone();
            let last_health_check = self.last_health_check.clone();
            let health_check_interval = Duration::from_millis(self.config.health_check_interval_ms);

            let health_handle = tokio::spawn(async move {
                let mut interval = tokio::time::interval(health_check_interval);

                loop {
                    tokio::select! {
                        _ = interval.tick() => {
                            let start_time = QuantaInstant::now();
                            let health_status = exchange_manager.health_check().await;
                            let health_check_latency = start_time.elapsed().as_nanos() as u64;

                            // Update statistics (lock-free)
                            stats.update_latency(health_check_latency);
                            last_health_check.store(start_time.elapsed().as_nanos() as u64, Ordering::Relaxed);

                            // Log unhealthy exchanges
                            for (exchange, healthy) in health_status {
                                if !healthy {
                                    log::warn!("Exchange {exchange} health check failed");
                                }
                            }
                        }
                    }

                    if shutdown_requested.load(Ordering::Acquire) {
                        break;
                    }
                }

                log::info!("Health check task stopped");
            });

            self.management_handles.push(health_handle);
        }

        // Statistics cleanup task (periodic, low frequency)
        if self.config.enable_stats_cleanup {
            let stats = self.stats_collector.clone();
            let shutdown_requested = self.shutdown_requested.clone();
            let last_cleanup = self.last_stats_cleanup.clone();
            let cleanup_interval = Duration::from_millis(self.config.stats_cleanup_interval_ms);

            let cleanup_handle = tokio::spawn(async move {
                let mut interval = tokio::time::interval(cleanup_interval);

                loop {
                    tokio::select! {
                        _ = interval.tick() => {
                            // Perform periodic maintenance (e.g., log statistics)
                            let snapshot = stats.get_snapshot();
                            last_cleanup.store(
                                std::time::SystemTime::now()
                                    .duration_since(std::time::UNIX_EPOCH)
                                    .unwrap_or_default()
                                    .as_nanos() as u64,
                                Ordering::Relaxed
                            );

                            log::info!(
                                "Collection stats: {} events/sec, {} avg latency (ns), {} trades, {} orderbooks",
                                snapshot.events_per_second,
                                snapshot.avg_latency_nanos,
                                snapshot.trades_processed,
                                snapshot.orderbooks_processed
                            );
                        }
                    }

                    if shutdown_requested.load(Ordering::Acquire) {
                        break;
                    }
                }

                log::info!("Statistics cleanup task stopped");
            });

            self.management_handles.push(cleanup_handle);
        }

        Ok(())
    }

    /// Get comprehensive statistics (mostly lock-free reads)
    pub fn get_comprehensive_stats(&self) -> ComprehensiveStats {
        let stats_snapshot = self.stats_collector.get_snapshot();
        let exchange_stats = self.exchange_manager.get_statistics();
        let buffer_stats = self.buffer_pool.get_stats();
        let buffer_counts = self.buffer_pool.get_available_counts();

        let pipeline_stats = self
            .zerocopy_pipeline
            .as_ref()
            .map(|p| p.get_stats())
            .unwrap_or_default();

        let uptime_nanos = self.start_time.elapsed().as_nanos() as u64;

        ComprehensiveStats {
            stats_snapshot,
            exchange_stats,
            buffer_stats,
            buffer_counts,
            pipeline_stats,
            uptime_nanos,
            events_processed_total: self.events_processed_total.load(Ordering::Relaxed),
            last_health_check_nanos: self.last_health_check.load(Ordering::Relaxed),
            last_stats_cleanup_nanos: self.last_stats_cleanup.load(Ordering::Relaxed),
        }
    }

    /// Record a high-frequency event (ultra-fast path)
    #[inline(always)]
    pub fn record_hf_event(&self, exchange: &str, bytes: u64, latency_nanos: u64) {
        // All lock-free operations
        self.stats_collector.record_event();
        self.stats_collector.update_latency(latency_nanos);
        self.exchange_manager.record_message(exchange, bytes);
        self.events_processed_total.fetch_add(1, Ordering::Relaxed);
    }

    /// Request shutdown (lock-free)
    pub fn request_shutdown(&self) {
        self.shutdown_requested.store(true, Ordering::Release);
        self.exchange_manager.request_shutdown();
    }

    /// Check if shutdown was requested (lock-free)
    pub fn is_shutdown_requested(&self) -> bool {
        self.shutdown_requested.load(Ordering::Acquire)
    }

    /// Stop all collection operations
    pub async fn stop_all(&mut self) -> Result<()> {
        log::info!("Stopping optimized collection manager");

        // Request shutdown
        self.request_shutdown();

        // Send shutdown signal to management tasks
        if let Some(sender) = self.shutdown_sender.take() {
            let _ = sender.send(());
        }

        // Stop zerocopy pipeline
        if let Some(mut pipeline) = self.zerocopy_pipeline.take() {
            pipeline.stop().await?;
        }

        // Wait for management tasks to finish
        for handle in self.management_handles.drain(..) {
            handle.abort();
            let _ = handle.await;
        }

        // Shutdown exchange manager
        self.exchange_manager.shutdown().await?;

        log::info!("Optimized collection manager stopped");
        Ok(())
    }
}

/// Comprehensive statistics combining all subsystems
#[derive(Debug, Clone)]
pub struct ComprehensiveStats {
    /// Lock-free statistics snapshot containing event counts and latency metrics
    ///
    /// Provides a consistent view of the system's processing statistics
    /// including events per second, average latency, and throughput metrics.
    pub stats_snapshot: StatsSnapshot,
    /// Exchange manager statistics for connection health and performance
    ///
    /// Contains detailed statistics about exchange connections, including
    /// connection counts, error rates, and exchange-specific metrics.
    pub exchange_stats: ExchangeManagerStats,
    /// Lock-free buffer pool statistics for memory management efficiency
    ///
    /// Tracks buffer pool hits, misses, and allocation patterns to measure
    /// memory management efficiency in high-frequency scenarios.
    pub buffer_stats: crate::monitor::lockfree_buffer_pool::BufferPoolStats,
    /// Current buffer availability counts across all buffer types
    ///
    /// Provides real-time counts of available buffers for serialization,
    /// compression, and SIMD operations to monitor resource utilization.
    pub buffer_counts: crate::monitor::lockfree_buffer_pool::BufferCounts,
    /// Zero-copy pipeline statistics for ultra-low-latency processing
    ///
    /// Tracks zero-copy operations, memory mappings, and pipeline efficiency
    /// metrics to optimize for minimal allocation overhead.
    pub pipeline_stats: crate::monitor::zerocopy_pipeline::ZerocopyPipelineStats,
    /// System uptime in nanoseconds since manager initialization
    ///
    /// High-resolution uptime measurement used for calculating throughput
    /// and efficiency metrics over the system's operational lifetime.
    pub uptime_nanos: u64,
    /// Total number of events processed since system start
    ///
    /// Atomic counter tracking all market data events processed by the
    /// collection manager, used for throughput calculations.
    pub events_processed_total: u64,
    /// Timestamp of the last health check operation in nanoseconds
    ///
    /// Records when the most recent health check was performed to monitor
    /// the frequency and timing of system health assessments.
    pub last_health_check_nanos: u64,
    /// Timestamp of the last statistics cleanup operation in nanoseconds
    ///
    /// Tracks when the system last performed cleanup of old statistical
    /// data to ensure efficient memory usage over time.
    pub last_stats_cleanup_nanos: u64,
}

impl ComprehensiveStats {
    /// Calculate overall system efficiency metrics
    pub fn calculate_efficiency_metrics(&self) -> EfficiencyMetrics {
        let uptime_seconds = self.uptime_nanos as f64 / 1_000_000_000.0;

        let overall_throughput = if uptime_seconds > 0.0 {
            self.events_processed_total as f64 / uptime_seconds
        } else {
            0.0
        };

        let memory_efficiency = if self.buffer_counts.total_capacity > 0 {
            (self.buffer_stats.pool_hits as f64)
                / ((self.buffer_stats.pool_hits + self.buffer_stats.pool_misses) as f64).max(1.0)
        } else {
            0.0
        };

        let processing_efficiency = if self.pipeline_stats.events_processed > 0 {
            (self.pipeline_stats.zero_copy_operations as f64)
                / (self.pipeline_stats.events_processed as f64)
        } else {
            0.0
        };

        EfficiencyMetrics {
            overall_throughput_eps: overall_throughput,
            memory_efficiency_ratio: memory_efficiency,
            processing_efficiency_ratio: processing_efficiency,
            avg_latency_microseconds: self.stats_snapshot.avg_latency_nanos as f64 / 1000.0,
            buffer_utilization_ratio: {
                let total_available = self.buffer_counts.serialization_available
                    + self.buffer_counts.compression_available
                    + self.buffer_counts.simd_available;
                let total_capacity = self.buffer_counts.total_capacity * 3; // 3 buffer types
                if total_capacity > 0 {
                    1.0 - (total_available as f64 / total_capacity as f64)
                } else {
                    0.0
                }
            },
        }
    }
}

/// System efficiency metrics
#[derive(Debug, Clone)]
pub struct EfficiencyMetrics {
    /// Overall system throughput measured in events per second
    ///
    /// Calculated as total events processed divided by system uptime,
    /// providing a key performance indicator for the collection manager.
    pub overall_throughput_eps: f64,
    /// Memory efficiency ratio from buffer pool hit rate
    ///
    /// Ratio of buffer pool hits to total buffer requests, indicating
    /// how effectively the system is reusing memory resources.
    pub memory_efficiency_ratio: f64,
    /// Processing efficiency ratio for zero-copy operations
    ///
    /// Proportion of operations that used zero-copy techniques versus
    /// traditional copying, measuring optimization effectiveness.
    pub processing_efficiency_ratio: f64,
    /// Average processing latency in microseconds
    ///
    /// Mean latency for processing market data events, converted from
    /// nanoseconds to microseconds for easier interpretation.
    pub avg_latency_microseconds: f64,
    /// Buffer utilization ratio across all buffer types
    ///
    /// Proportion of buffers currently in use versus total available,
    /// indicating resource utilization and potential bottlenecks.
    pub buffer_utilization_ratio: f64,
}

// Thread safety
unsafe impl Send for OptimizedCollectionManager {}
unsafe impl Sync for OptimizedCollectionManager {}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::monitor::config::storage::StorageConfig;
    use tempfile::TempDir;

    #[tokio::test]
    async fn test_optimized_collection_manager_creation() {
        let temp_dir = TempDir::new().unwrap();
        let storage_config = StorageConfig {
            market_data_path: temp_dir.path().join("market_data"),
            ..Default::default()
        };
        let storage_manager = Arc::new(StorageManager::new(storage_config).await.unwrap());

        let config = OptimizedCollectionConfig::default();
        let manager = OptimizedCollectionManager::new(config, storage_manager)
            .await
            .unwrap();

        assert!(!manager.is_shutdown_requested());
        assert_eq!(manager.events_processed_total.load(Ordering::Relaxed), 0);
    }

    #[tokio::test]
    async fn test_optimized_collection_manager_stats() {
        let temp_dir = TempDir::new().unwrap();
        let storage_config = StorageConfig {
            market_data_path: temp_dir.path().join("market_data"),
            ..Default::default()
        };
        let storage_manager = Arc::new(StorageManager::new(storage_config).await.unwrap());

        let config = OptimizedCollectionConfig::default();
        let manager = OptimizedCollectionManager::new(config, storage_manager)
            .await
            .unwrap();

        // Test high-frequency event recording
        manager.record_hf_event("test_exchange", 1024, 500);
        manager.record_hf_event("test_exchange", 2048, 750);

        let stats = manager.get_comprehensive_stats();
        assert_eq!(stats.events_processed_total, 2);
        assert!(stats.uptime_nanos > 0);

        let efficiency = stats.calculate_efficiency_metrics();
        assert!(efficiency.overall_throughput_eps >= 0.0);
        assert!(efficiency.avg_latency_microseconds >= 0.0);
    }

    #[tokio::test]
    async fn test_optimized_collection_manager_shutdown() {
        let temp_dir = TempDir::new().unwrap();
        let storage_config = StorageConfig {
            market_data_path: temp_dir.path().join("market_data"),
            ..Default::default()
        };
        let storage_manager = Arc::new(StorageManager::new(storage_config).await.unwrap());

        let config = OptimizedCollectionConfig::default();
        let mut manager = OptimizedCollectionManager::new(config, storage_manager)
            .await
            .unwrap();

        // Test shutdown request
        manager.request_shutdown();
        assert!(manager.is_shutdown_requested());

        // Test stop_all
        manager.stop_all().await.unwrap();
    }
}