rangebar_streaming/

processor.rs

use futures::Stream;
/// Production-ready streaming architecture with bounded memory and backpressure
/// # FILE-SIZE-OK
///
/// This module implements true infinite streaming capabilities addressing critical failures:
/// - Eliminates Vec<RangeBar> accumulation (unbounded memory growth)
/// - Implements proper backpressure with bounded channels
/// - Provides circuit breaker resilience patterns
/// - Maintains temporal integrity for financial data
use rangebar_core::processor::ExportRangeBarProcessor;
use rangebar_core::{AggTrade, RangeBar};
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::task::{Context, Poll};
use tokio::sync::mpsc;
use tokio::time::{Duration, Instant};

/// Configuration for production streaming
#[derive(Debug, Clone)]
pub struct StreamingProcessorConfig {
    /// Channel capacity for trade input
    pub trade_channel_capacity: usize,
    /// Channel capacity for completed bars
    pub bar_channel_capacity: usize,
    /// Memory usage threshold in bytes
    pub memory_threshold_bytes: usize,
    /// Backpressure timeout
    pub backpressure_timeout: Duration,
    /// Circuit breaker error rate threshold (0.0-1.0)
    pub circuit_breaker_threshold: f64,
    /// Circuit breaker timeout before retry
    pub circuit_breaker_timeout: Duration,
}

impl StreamingProcessorConfig {
    /// Get bar channel capacity from environment or use default (10K)
    /// Issue #96 Task #6: RANGEBAR_MAX_PENDING_BARS env var support
    fn get_bar_channel_capacity() -> usize {
        std::env::var("RANGEBAR_MAX_PENDING_BARS")
            .ok()
            .and_then(|v| v.parse::<usize>().ok())
            .unwrap_or(10_000)
    }
}

impl Default for StreamingProcessorConfig {
    fn default() -> Self {
        Self {
            trade_channel_capacity: 5_000,                              // Based on consensus analysis
            bar_channel_capacity: StreamingProcessorConfig::get_bar_channel_capacity(), // Issue #96: 10K backpressure bound
            memory_threshold_bytes: 100_000_000,                        // 100MB limit
            backpressure_timeout: Duration::from_millis(100),
            circuit_breaker_threshold: 0.5, // 50% error rate
            circuit_breaker_timeout: Duration::from_secs(30),
        }
    }
}

/// Production streaming processor with bounded memory
pub struct StreamingProcessor {
    /// Range bar processor (single instance, no accumulation)
    processor: ExportRangeBarProcessor,

    /// Threshold in decimal basis points for recreating processor
    #[allow(dead_code)]
    threshold_decimal_bps: u32,

    /// Bounded channel for incoming trades
    trade_sender: Option<mpsc::Sender<AggTrade>>,
    trade_receiver: mpsc::Receiver<AggTrade>,

    /// Bounded channel for outgoing bars
    bar_sender: mpsc::Sender<RangeBar>,
    bar_receiver: Option<mpsc::Receiver<RangeBar>>,

    /// Configuration
    config: StreamingProcessorConfig,

    /// Metrics
    metrics: Arc<StreamingMetrics>,

    /// Circuit breaker state
    circuit_breaker: CircuitBreaker,
}

/// Circuit breaker implementation
#[derive(Debug)]
struct CircuitBreaker {
    state: CircuitBreakerState,
    failure_count: u64,
    success_count: u64,
    last_failure_time: Option<Instant>,
    threshold: f64,
    timeout: Duration,
}

#[derive(Debug, PartialEq)]
enum CircuitBreakerState {
    Closed,
    Open,
    HalfOpen,
}

/// Streaming metrics for observability
/// Issue #96 Task #6: Extended with queue depth and block time tracking
#[derive(Debug, Default)]
pub struct StreamingMetrics {
    pub trades_processed: AtomicU64,
    pub bars_generated: AtomicU64,
    pub errors_total: AtomicU64,
    pub backpressure_events: AtomicU64,
    pub circuit_breaker_trips: AtomicU64,
    pub memory_usage_bytes: AtomicU64,
    pub max_queue_depth: AtomicU64,       // Issue #96 Task #6: Max observed queue depth
    pub total_block_time_ms: AtomicU64,   // Issue #96 Task #6: Accumulated block time
}

impl StreamingProcessor {
    /// Create new production streaming processor
    pub fn new(
        threshold_decimal_bps: u32,
    ) -> Result<Self, rangebar_core::processor::ProcessingError> {
        Self::with_config(threshold_decimal_bps, StreamingProcessorConfig::default())
    }

    /// Create with custom configuration
    pub fn with_config(
        threshold_decimal_bps: u32,
        config: StreamingProcessorConfig,
    ) -> Result<Self, rangebar_core::processor::ProcessingError> {
        let (trade_sender, trade_receiver) = mpsc::channel(config.trade_channel_capacity);
        let (bar_sender, bar_receiver) = mpsc::channel(config.bar_channel_capacity);

        let circuit_breaker_threshold = config.circuit_breaker_threshold;
        let circuit_breaker_timeout = config.circuit_breaker_timeout;

        Ok(Self {
            processor: ExportRangeBarProcessor::new(threshold_decimal_bps)?,
            threshold_decimal_bps,
            trade_sender: Some(trade_sender),
            trade_receiver,
            bar_sender,
            bar_receiver: Some(bar_receiver),
            config,
            metrics: Arc::new(StreamingMetrics::default()),
            circuit_breaker: CircuitBreaker::new(
                circuit_breaker_threshold,
                circuit_breaker_timeout,
            ),
        })
    }

    /// Get trade sender for external components
    pub fn trade_sender(&mut self) -> Option<mpsc::Sender<AggTrade>> {
        self.trade_sender.take()
    }

    /// Get bar receiver for external components
    pub fn bar_receiver(&mut self) -> Option<mpsc::Receiver<RangeBar>> {
        self.bar_receiver.take()
    }

    /// Start processing loop (bounded memory, infinite capability)
    pub async fn start_processing(&mut self) -> Result<(), StreamingError> {
        loop {
            // Check circuit breaker state
            if !self.circuit_breaker.can_process() {
                tokio::time::sleep(Duration::from_millis(100)).await;
                continue;
            }

            // Receive trade with timeout (prevents blocking forever)
            let trade = match tokio::time::timeout(
                self.config.backpressure_timeout,
                self.trade_receiver.recv(),
            )
            .await
            {
                Ok(Some(trade)) => trade,
                Ok(None) => {
                    // Channel closed - send final incomplete bar if exists
                    if let Some(final_bar) = self.processor.get_incomplete_bar()
                        && let Err(e) = self.send_bar_with_backpressure(final_bar).await
                    {
                        println!("Failed to send final incomplete bar: {:?}", e);
                    }
                    break;
                }
                Err(_) => continue, // Timeout, check circuit breaker again
            };

            // Process single trade (use borrowed reference per Issue #96 Task #78)
            match self.process_single_trade(&trade).await {
                Ok(bar_opt) => {
                    self.circuit_breaker.record_success();

                    // If bar completed, send with backpressure handling
                    if let Some(bar) = bar_opt
                        && let Err(e) = self.send_bar_with_backpressure(bar).await
                    {
                        println!("Failed to send bar: {:?}", e);
                        self.circuit_breaker.record_failure();
                    }
                }
                Err(e) => {
                    println!("Trade processing error: {:?}", e);
                    self.circuit_breaker.record_failure();
                    self.metrics.errors_total.fetch_add(1, Ordering::Relaxed);
                }
            }
        }

        Ok(())
    }

    /// Process single trade - extracts completed bars without accumulation
    // Issue #96 Task #78: Accept borrowed AggTrade reference
    async fn process_single_trade(
        &mut self,
        trade: &AggTrade,
    ) -> Result<Option<RangeBar>, StreamingError> {
        // Update metrics
        self.metrics
            .trades_processed
            .fetch_add(1, Ordering::Relaxed);

        // Process trade using existing algorithm (single trade at a time)
        // Note: Clone is acceptable here since this is the bounded-memory streaming processor
        // (not the main hot path; LiveBarEngine uses non-cloning path via live_engine.rs)
        self.processor.process_trades_continuously(&[trade.clone()]);

        // Extract completed bars immediately (prevents accumulation)
        let mut completed_bars = self.processor.get_all_completed_bars();

        if !completed_bars.is_empty() {
            // Bounded memory: only return first completed bar
            // Additional bars would be rare edge cases but must be handled
            let completed_bar = completed_bars.remove(0);

            // Handle rare case of multiple completions
            if !completed_bars.is_empty() {
                println!(
                    "Warning: {} additional bars completed, dropping for bounded memory",
                    completed_bars.len()
                );
                self.metrics
                    .backpressure_events
                    .fetch_add(completed_bars.len() as u64, Ordering::Relaxed);
            }

            self.metrics.bars_generated.fetch_add(1, Ordering::Relaxed);
            Ok(Some(completed_bar))
        } else {
            Ok(None)
        }
    }

    /// Send bar with backpressure handling
    async fn send_bar_with_backpressure(&self, bar: RangeBar) -> Result<(), StreamingError> {
        // Use try_send for immediate check, then send for blocking
        match self.bar_sender.try_send(bar.clone()) {
            Ok(()) => Ok(()),
            Err(mpsc::error::TrySendError::Full(_)) => {
                // Apply backpressure - channel is full
                println!("Bar channel full, applying backpressure");
                self.metrics
                    .backpressure_events
                    .fetch_add(1, Ordering::Relaxed);

                // Wait for capacity with blocking send
                self.bar_sender
                    .send(bar)
                    .await
                    .map_err(|_| StreamingError::ChannelClosed)
            }
            Err(mpsc::error::TrySendError::Closed(_)) => Err(StreamingError::ChannelClosed),
        }
    }

    /// Get current metrics
    pub fn metrics(&self) -> &StreamingMetrics {
        &self.metrics
    }

    /// Extract final incomplete bar when stream ends (for algorithmic consistency)
    pub fn get_final_incomplete_bar(&mut self) -> Option<RangeBar> {
        self.processor.get_incomplete_bar()
    }

    /// Check memory usage against threshold
    pub fn check_memory_usage(&self) -> bool {
        let current_usage = self.metrics.memory_usage_bytes.load(Ordering::Relaxed);
        current_usage < self.config.memory_threshold_bytes as u64
    }
}

impl CircuitBreaker {
    fn new(threshold: f64, timeout: Duration) -> Self {
        Self {
            state: CircuitBreakerState::Closed,
            failure_count: 0,
            success_count: 0,
            last_failure_time: None,
            threshold,
            timeout,
        }
    }

    fn can_process(&mut self) -> bool {
        match self.state {
            CircuitBreakerState::Closed => true,
            CircuitBreakerState::Open => {
                if let Some(last_failure) = self.last_failure_time {
                    if last_failure.elapsed() > self.timeout {
                        self.state = CircuitBreakerState::HalfOpen;
                        true
                    } else {
                        false
                    }
                } else {
                    true
                }
            }
            CircuitBreakerState::HalfOpen => true,
        }
    }

    fn record_success(&mut self) {
        self.success_count += 1;

        if self.state == CircuitBreakerState::HalfOpen {
            // Successful request in half-open, close circuit
            self.state = CircuitBreakerState::Closed;
            self.failure_count = 0;
        }
    }

    fn record_failure(&mut self) {
        self.failure_count += 1;
        self.last_failure_time = Some(Instant::now());

        let total_requests = self.failure_count + self.success_count;
        if total_requests >= 10 {
            // Minimum sample size
            let failure_rate = self.failure_count as f64 / total_requests as f64;

            if failure_rate >= self.threshold {
                self.state = CircuitBreakerState::Open;
            }
        }
    }
}

/// Stream implementation for range bars (true streaming)
pub struct RangeBarStream {
    receiver: mpsc::Receiver<RangeBar>,
}

impl RangeBarStream {
    pub fn new(receiver: mpsc::Receiver<RangeBar>) -> Self {
        Self { receiver }
    }
}

impl Stream for RangeBarStream {
    type Item = Result<RangeBar, StreamingError>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match self.receiver.poll_recv(cx) {
            Poll::Ready(Some(bar)) => Poll::Ready(Some(Ok(bar))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

/// Streaming errors
#[derive(Debug, thiserror::Error)]
pub enum StreamingError {
    #[error("Channel closed")]
    ChannelClosed,

    #[error("Backpressure timeout")]
    BackpressureTimeout,

    #[error("Circuit breaker open")]
    CircuitBreakerOpen,

    #[error("Memory threshold exceeded")]
    MemoryThresholdExceeded,

    #[error("Processing error: {0}")]
    ProcessingError(String),
}

impl StreamingMetrics {
    /// Get metrics summary
    pub fn summary(&self) -> MetricsSummary {
        MetricsSummary {
            trades_processed: self.trades_processed.load(Ordering::Relaxed),
            bars_generated: self.bars_generated.load(Ordering::Relaxed),
            errors_total: self.errors_total.load(Ordering::Relaxed),
            backpressure_events: self.backpressure_events.load(Ordering::Relaxed),
            circuit_breaker_trips: self.circuit_breaker_trips.load(Ordering::Relaxed),
            memory_usage_bytes: self.memory_usage_bytes.load(Ordering::Relaxed),
        }
    }
}

/// Metrics snapshot
#[derive(Debug, Clone)]
pub struct MetricsSummary {
    pub trades_processed: u64,
    pub bars_generated: u64,
    pub errors_total: u64,
    pub backpressure_events: u64,
    pub circuit_breaker_trips: u64,
    pub memory_usage_bytes: u64,
}

impl MetricsSummary {
    /// Calculate bars per aggTrade ratio
    pub fn bars_per_aggtrade(&self) -> f64 {
        if self.trades_processed > 0 {
            self.bars_generated as f64 / self.trades_processed as f64
        } else {
            0.0
        }
    }

    /// Calculate error rate
    pub fn error_rate(&self) -> f64 {
        if self.trades_processed > 0 {
            self.errors_total as f64 / self.trades_processed as f64
        } else {
            0.0
        }
    }

    /// Format memory usage
    pub fn memory_usage_mb(&self) -> f64 {
        self.memory_usage_bytes as f64 / 1_000_000.0
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rangebar_core::FixedPoint;

    fn create_test_trade(id: u64, price: f64, timestamp: u64) -> AggTrade {
        let price_str = format!("{:.8}", price);
        AggTrade {
            agg_trade_id: id as i64,
            price: FixedPoint::from_str(&price_str).unwrap(),
            volume: FixedPoint::from_str("1.0").unwrap(),
            first_trade_id: id as i64,
            last_trade_id: id as i64,
            timestamp: timestamp as i64,
            is_buyer_maker: false,
            is_best_match: None,
        }
    }

    #[tokio::test]
    async fn test_bounded_memory_streaming() {
        let mut processor = StreamingProcessor::new(25).unwrap(); // 0.25% threshold

        // Test that memory remains bounded
        let initial_metrics = processor.metrics().summary();

        // Send 1000 trades
        for i in 0..1000 {
            let trade = create_test_trade(i, 23000.0 + (i as f64), 1659312000000 + i);
            if let Ok(bar_opt) = processor.process_single_trade(&trade).await {
                // Verify no accumulation - at most one bar per aggTrade
                assert!(bar_opt.is_none() || bar_opt.is_some());
            }
        }

        let final_metrics = processor.metrics().summary();
        assert!(final_metrics.trades_processed >= initial_metrics.trades_processed);
        assert!(final_metrics.trades_processed <= 1000);
    }

    #[tokio::test]
    async fn test_circuit_breaker() {
        let mut circuit_breaker = CircuitBreaker::new(0.5, Duration::from_millis(100));

        // Initially closed
        assert!(circuit_breaker.can_process());

        // Record failures
        for _ in 0..20 {
            circuit_breaker.record_failure();
        }

        // Should open after 50% failure rate
        assert_eq!(circuit_breaker.state, CircuitBreakerState::Open);
        assert!(!circuit_breaker.can_process());

        // Wait for timeout
        tokio::time::sleep(Duration::from_millis(150)).await;

        // Should transition to half-open
        assert!(circuit_breaker.can_process());

        // Record success
        circuit_breaker.record_success();

        // Should close
        assert_eq!(circuit_breaker.state, CircuitBreakerState::Closed);
    }

    // === Circuit Breaker State Machine Tests ===

    #[test]
    fn test_circuit_breaker_stays_closed_below_threshold() {
        let mut cb = CircuitBreaker::new(0.5, Duration::from_secs(10));

        // 8 successes, 2 failures = 20% failure rate, below 50% threshold
        for _ in 0..8 {
            cb.record_success();
        }
        for _ in 0..2 {
            cb.record_failure();
        }

        // Should remain closed (20% < 50%)
        assert_eq!(cb.state, CircuitBreakerState::Closed);
        assert!(cb.can_process());
    }

    #[test]
    fn test_circuit_breaker_minimum_sample_size() {
        let mut cb = CircuitBreaker::new(0.5, Duration::from_secs(10));

        // 9 failures, 0 successes = 100% failure rate, but only 9 requests (< 10 minimum)
        for _ in 0..9 {
            cb.record_failure();
        }

        // Should remain closed (minimum sample size not met)
        assert_eq!(cb.state, CircuitBreakerState::Closed);
        assert!(cb.can_process());

        // 10th failure triggers open
        cb.record_failure();
        assert_eq!(cb.state, CircuitBreakerState::Open);
    }

    #[test]
    fn test_circuit_breaker_halfopen_failure_reopens() {
        let mut cb = CircuitBreaker::new(0.5, Duration::from_secs(0));

        // Trip the breaker: 10 failures opens it
        for _ in 0..10 {
            cb.record_failure();
        }
        assert_eq!(cb.state, CircuitBreakerState::Open);

        // Zero-second timeout → immediately transitions to HalfOpen on can_process
        assert!(cb.can_process());
        assert_eq!(cb.state, CircuitBreakerState::HalfOpen);

        // Record failure in HalfOpen → should re-open
        // (failure_count accumulates, total >= 10, rate >= threshold)
        cb.record_failure();
        assert_eq!(cb.state, CircuitBreakerState::Open);
    }

    #[test]
    fn test_circuit_breaker_closed_resets_failure_count() {
        let mut cb = CircuitBreaker::new(0.5, Duration::from_secs(0));

        // Trip the breaker
        for _ in 0..10 {
            cb.record_failure();
        }
        assert_eq!(cb.state, CircuitBreakerState::Open);

        // Transition to HalfOpen
        assert!(cb.can_process());
        assert_eq!(cb.state, CircuitBreakerState::HalfOpen);

        // Record success → closes and resets failure_count
        cb.record_success();
        assert_eq!(cb.state, CircuitBreakerState::Closed);
        assert_eq!(cb.failure_count, 0);
    }

    #[test]
    fn test_circuit_breaker_open_blocks_until_timeout() {
        let mut cb = CircuitBreaker::new(0.5, Duration::from_secs(3600)); // 1 hour timeout

        // Trip the breaker
        for _ in 0..10 {
            cb.record_failure();
        }

        // Should be blocked — timeout hasn't elapsed
        assert!(!cb.can_process());
        assert_eq!(cb.state, CircuitBreakerState::Open);
    }

    #[test]
    fn test_metrics_zero_trades() {
        let metrics = MetricsSummary {
            trades_processed: 0,
            bars_generated: 0,
            errors_total: 0,
            backpressure_events: 0,
            circuit_breaker_trips: 0,
            memory_usage_bytes: 0,
        };

        // Division by zero guarded
        assert_eq!(metrics.bars_per_aggtrade(), 0.0);
        assert_eq!(metrics.error_rate(), 0.0);
        assert_eq!(metrics.memory_usage_mb(), 0.0);
    }

    #[test]
    fn test_metrics_calculations() {
        let metrics = MetricsSummary {
            trades_processed: 1000,
            bars_generated: 50,
            errors_total: 5,
            backpressure_events: 2,
            circuit_breaker_trips: 1,
            memory_usage_bytes: 50_000_000,
        };

        assert_eq!(metrics.bars_per_aggtrade(), 0.05);
        assert_eq!(metrics.error_rate(), 0.005);
        assert_eq!(metrics.memory_usage_mb(), 50.0);
    }

    // === Metrics Snapshot & Take-Once Tests (Issue #96 Task #114) ===

    #[test]
    fn test_streaming_metrics_summary_snapshot() {
        let metrics = StreamingMetrics::default();
        metrics.trades_processed.store(500, Ordering::Relaxed);
        metrics.bars_generated.store(25, Ordering::Relaxed);
        metrics.errors_total.store(3, Ordering::Relaxed);
        metrics.backpressure_events.store(1, Ordering::Relaxed);
        metrics.circuit_breaker_trips.store(0, Ordering::Relaxed);
        metrics.memory_usage_bytes.store(42_000_000, Ordering::Relaxed);

        let summary = metrics.summary();

        assert_eq!(summary.trades_processed, 500);
        assert_eq!(summary.bars_generated, 25);
        assert_eq!(summary.errors_total, 3);
        assert_eq!(summary.backpressure_events, 1);
        assert_eq!(summary.circuit_breaker_trips, 0);
        assert_eq!(summary.memory_usage_bytes, 42_000_000);
    }

    #[test]
    fn test_memory_usage_mb_conversion() {
        // Exact MB boundary
        let m1 = MetricsSummary {
            trades_processed: 0, bars_generated: 0, errors_total: 0,
            backpressure_events: 0, circuit_breaker_trips: 0,
            memory_usage_bytes: 1_000_000,
        };
        assert_eq!(m1.memory_usage_mb(), 1.0);

        // Fractional MB
        let m2 = MetricsSummary {
            trades_processed: 0, bars_generated: 0, errors_total: 0,
            backpressure_events: 0, circuit_breaker_trips: 0,
            memory_usage_bytes: 1_500_000,
        };
        assert_eq!(m2.memory_usage_mb(), 1.5);

        // Large value (4 GB)
        let m3 = MetricsSummary {
            trades_processed: 0, bars_generated: 0, errors_total: 0,
            backpressure_events: 0, circuit_breaker_trips: 0,
            memory_usage_bytes: 4_000_000_000,
        };
        assert_eq!(m3.memory_usage_mb(), 4000.0);
    }

    #[test]
    fn test_trade_sender_take_once() {
        let mut processor = StreamingProcessor::new(25).unwrap();

        // First call returns Some
        let sender = processor.trade_sender();
        assert!(sender.is_some(), "First trade_sender() call must return Some");

        // Second call returns None (already taken)
        let sender2 = processor.trade_sender();
        assert!(sender2.is_none(), "Second trade_sender() call must return None");
    }

    #[test]
    fn test_bar_receiver_take_once() {
        let mut processor = StreamingProcessor::new(25).unwrap();

        // First call returns Some
        let receiver = processor.bar_receiver();
        assert!(receiver.is_some(), "First bar_receiver() call must return Some");

        // Second call returns None (already taken)
        let receiver2 = processor.bar_receiver();
        assert!(receiver2.is_none(), "Second bar_receiver() call must return None");
    }

    #[test]
    fn test_check_memory_usage_below_threshold() {
        let processor = StreamingProcessor::new(25).unwrap();

        // Default: memory_usage_bytes = 0, threshold = 100MB → within bounds
        assert!(processor.check_memory_usage(), "Zero memory usage should be within threshold");
    }

    #[test]
    fn test_check_memory_usage_above_threshold() {
        let processor = StreamingProcessor::new(25).unwrap();

        // Simulate exceeding threshold (100MB default)
        processor.metrics.memory_usage_bytes.store(200_000_000, Ordering::Relaxed);
        assert!(!processor.check_memory_usage(), "200MB should exceed 100MB threshold");
    }

    #[test]
    fn test_get_final_incomplete_bar_empty() {
        let mut processor = StreamingProcessor::new(25).unwrap();

        // No trades processed → no incomplete bar
        let bar = processor.get_final_incomplete_bar();
        assert!(bar.is_none(), "No incomplete bar before any trades");
    }

    #[test]
    fn test_bars_per_aggtrade_ratio() {
        let metrics = MetricsSummary {
            trades_processed: 200,
            bars_generated: 10,
            errors_total: 0,
            backpressure_events: 0,
            circuit_breaker_trips: 0,
            memory_usage_bytes: 0,
        };

        assert_eq!(metrics.bars_per_aggtrade(), 0.05);
        assert_eq!(metrics.error_rate(), 0.0);
    }
}