opendeviationbar-streaming 13.70.3

//! Live bar engine for real-time open deviation bar construction (Issue #91)
//! # FILE-SIZE-OK
//!
//! Multiplexes Binance WebSocket streams across symbols and fans out trades
//! to canonical `OpenDeviationBarProcessor` instances (full 58-column microstructure).
//!
//! Architecture:
//! - One tokio task per symbol (independent reconnection via barter-rs pattern)
//! - One `OpenDeviationBarProcessor` per (symbol, threshold) pair
//! - Completed bars emitted to a bounded channel for Python consumption
//! - Graceful shutdown via `CancellationToken`

mod puck;
mod symbol_task;
mod trade_dispatch;
mod types;

// Re-export public types
pub use types::{
    CompletedBar, FormingBar, FormingBarWatches, LiveEngineConfig, LiveEngineMetrics,
    LiveEngineMetricsSnapshot, OuroborosMode, WsMode, split_symbols_by_volume,
};

// Internal re-exports used within this module
use symbol_task::symbol_task;
use trade_dispatch::{ExtraSinks, maybe_reset_at_midnight, maybe_reset_at_week_gap};
use types::FormingBarKey;

use opendeviationbar_core::Tick;
use opendeviationbar_core::IntraBarConfig;
use opendeviationbar_core::checkpoint::Checkpoint;
use opendeviationbar_core::interbar::InterBarConfig;
use opendeviationbar_core::processor::{OpenDeviationBarProcessor, ProcessingError};
use opendeviationbar_providers::binance::{BinanceWebSocketStream, CombinedWebSocketStream};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tokio::sync::{mpsc, watch};
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;

// Issue #96 Task #9: Ring buffer replaces mpsc channel for memory efficiency
use crate::ring_buffer::ConcurrentRingBuffer;

// Issue #257: Gap detection infrastructure
use crate::gap::{GapEvent, GapFillSender};

// (#286) No MAX_GAP_FILL_TRADES limit -- Puck fills any gap regardless of size.
// The rate limiter paces REST calls. Clean-slate startup may create gaps
// spanning hours (midnight to now) that must be filled unconditionally.

/// Live bar engine: multiplexes WebSocket streams -> canonical processors -> completed bars.
///
/// Issue #91: Uses `OpenDeviationBarProcessor` (NOT `ExportOpenDeviationBarProcessor`) for full
/// 3-step feature finalization producing all 58 columns.
pub struct LiveBarEngine {
    config: LiveEngineConfig,
    /// Issue #96 Task #9: Ring buffer replaces mpsc for 10-20% memory savings
    bar_buffer: ConcurrentRingBuffer<CompletedBar>,
    shutdown: CancellationToken,
    metrics: Arc<LiveEngineMetrics>,
    started: bool,
    /// Channel for receiving processor checkpoints on shutdown
    checkpoint_rx: Option<mpsc::Receiver<(String, u32, Checkpoint)>>,
    checkpoint_tx: mpsc::Sender<(String, u32, Checkpoint)>,
    /// Issue #214: Watch senders for forming bar snapshots (one per symbol x threshold).
    /// Passed into symbol tasks on start(). Senders are moved, receivers extracted via
    /// take_forming_bar_watches() before start().
    forming_bar_txs: HashMap<FormingBarKey, watch::Sender<Option<FormingBar>>>,
    /// Issue #214: Watch receivers for forming bar snapshots.
    /// Extracted before start() via take_forming_bar_watches().
    forming_bar_watches: Option<FormingBarWatches>,
    /// Issue #257: Channel for emitting gap events to Python consumers.
    /// Created in `new()`, passed to symbol tasks in `start()`.
    gap_event_tx: mpsc::Sender<GapEvent>,
    /// Issue #257: Receiver half, extracted via `take_gap_event_receiver()`.
    gap_event_rx: Option<mpsc::Receiver<GapEvent>>,
    /// Issue #257: Per-symbol gap-fill command senders for on-demand fill_gap().
    /// Cloneable -- Python can call fill_gap(&self) without borrow conflicts.
    gap_fill_senders: HashMap<String, GapFillSender>,
    /// Issue #257: Per-symbol gap-fill receivers, passed to symbol_tasks on start().
    gap_fill_receivers: Option<HashMap<String, crate::gap::GapFillReceiver>>,
    /// Issue #286: Pre-spawned WS trade receivers for buffered start.
    /// When `start_ws()` is called, WS connections are spawned and trades
    /// buffer in these channels. `start()` then drains the buffer seamlessly.
    ws_trade_receivers: Option<HashMap<String, mpsc::Receiver<Tick>>>,
    /// Issue #286: Pre-created processors from fill_from_rest().
    /// When `fill_from_rest()` runs, processors are created, REST trades are
    /// fed through them, and they're stored here with accumulated state.
    /// `start()` reuses them instead of creating fresh processors.
    pre_created_processors: Option<HashMap<String, Vec<(u32, OpenDeviationBarProcessor)>>>,
    /// Issue #318: Optional BarSink vector for fan-out dispatch (e.g., ClickHouseWriterSink).
    /// Shared across all symbol tasks via Arc<Mutex<...>>.
    /// When `clickhouse-sink` feature is enabled and OPENDEVIATIONBAR_CH_HOSTS is set,
    /// this contains the ClickHouseWriterSink. Otherwise empty.
    extra_sinks: ExtraSinks,
    /// Issue #318: Metrics from the ClickHouseWriterSink flush thread (if created).
    #[cfg(feature = "clickhouse-sink")]
    ch_writer_metrics: Option<Arc<crate::clickhouse_writer::flush_thread::FlushThreadMetrics>>,
    /// Per-symbol checkpoint snapshot request senders.
    /// When collect_checkpoints() fires, it sends () to each sender,
    /// and each symbol task responds by pushing checkpoints to checkpoint_tx.
    checkpoint_snapshot_txs: Vec<mpsc::Sender<()>>,
}

impl LiveBarEngine {
    /// Create a new live bar engine.
    ///
    /// Does NOT start streaming -- call `start()` after creation.
    pub fn new(config: LiveEngineConfig) -> Self {
        // Issue #96 Task #9: Initialize ring buffer instead of mpsc channel
        let bar_buffer = ConcurrentRingBuffer::new(config.bar_channel_capacity);
        // Channel for extracting checkpoints on shutdown (one per symbol x threshold).
        // Phase 59: total processor count is the SUM of per-symbol threshold counts,
        // NOT the Cartesian product.
        let max_checkpoints: usize = config
            .symbol_thresholds
            .values()
            .map(|v| v.len())
            .sum::<usize>();
        let (checkpoint_tx, checkpoint_rx) = mpsc::channel(max_checkpoints.max(1));

        // Issue #214: Create watch channels for forming bar snapshots.
        // One (tx, rx) pair per (symbol, threshold). Created here so receivers
        // can be extracted before start() (avoiding borrow conflicts with next_bar).
        let num_pairs: usize = config
            .symbol_thresholds
            .values()
            .map(|v| v.len())
            .sum::<usize>();
        let mut forming_bar_txs = HashMap::with_capacity(num_pairs);
        let mut forming_bar_rxs = HashMap::with_capacity(num_pairs);
        for (symbol, thresholds) in &config.symbol_thresholds {
            let sym_arc: Arc<str> = Arc::from(symbol.to_uppercase().as_str());
            for &threshold in thresholds {
                let (tx, rx) = watch::channel(None);
                let key = (sym_arc.clone(), threshold);
                forming_bar_txs.insert(key.clone(), tx);
                forming_bar_rxs.insert(key, rx);
            }
        }

        // Issue #257: Gap event channel (bounded, non-blocking sends from hot path)
        let (gap_event_tx, gap_event_rx) = mpsc::channel(256);

        // Issue #257: Per-symbol gap-fill command channels
        let num_symbols = config.symbol_thresholds.len();
        let mut gap_fill_senders = HashMap::with_capacity(num_symbols);
        let mut gap_fill_receivers = HashMap::with_capacity(num_symbols);
        for symbol in config.symbol_thresholds.keys() {
            let sym = symbol.to_uppercase();
            let (tx, rx) = crate::gap::gap_fill_channel(16);
            gap_fill_senders.insert(sym.clone(), tx);
            gap_fill_receivers.insert(sym, rx);
        }

        // Issue #318: Conditionally create ClickHouseWriterSink from env vars
        #[cfg(feature = "clickhouse-sink")]
        let (extra_sinks, ch_writer_metrics) = {
            if std::env::var("OPENDEVIATIONBAR_CH_HOSTS").is_ok() {
                let ch_config = crate::clickhouse_writer::config::ClickHouseWriterConfig::from_env();
                let sink = crate::clickhouse_writer::sink::ClickHouseWriterSink::new(ch_config);
                // Clone the Arc<FlushThreadMetrics> before boxing the sink
                let metrics_arc = sink.shared_metrics();
                let sinks: Vec<Box<dyn crate::engine::traits::BarSink>> = vec![Box::new(sink)];
                tracing::info!("clickhouse writer sink created (OPENDEVIATIONBAR_CH_HOSTS set)");
                (Some(Arc::new(std::sync::Mutex::new(sinks))), Some(metrics_arc))
            } else {
                tracing::info!("clickhouse writer sink skipped (OPENDEVIATIONBAR_CH_HOSTS not set)");
                (None, None)
            }
        };
        #[cfg(not(feature = "clickhouse-sink"))]
        let extra_sinks: ExtraSinks = None;

        Self {
            config,
            bar_buffer,
            shutdown: CancellationToken::new(),
            metrics: Arc::new(LiveEngineMetrics::default()),
            started: false,
            checkpoint_rx: Some(checkpoint_rx),
            checkpoint_tx,
            forming_bar_txs,
            forming_bar_watches: Some(forming_bar_rxs),
            gap_event_tx,
            gap_event_rx: Some(gap_event_rx),
            gap_fill_senders,
            gap_fill_receivers: Some(gap_fill_receivers),
            ws_trade_receivers: None,
            pre_created_processors: None,
            extra_sinks,
            #[cfg(feature = "clickhouse-sink")]
            ch_writer_metrics,
            checkpoint_snapshot_txs: Vec::new(),
        }
    }

    /// Inject a checkpoint for a specific (symbol, threshold) pair.
    /// Must be called before `start()`.
    pub fn set_initial_checkpoint(&mut self, symbol: &str, threshold: u32, checkpoint: Checkpoint) {
        self.config
            .initial_checkpoints
            .insert((symbol.to_uppercase(), threshold), checkpoint);
    }

    /// Update gap detector seeds (used by fill_from_rest to determine REST start point).
    /// Called by StreamManager to propagate runtime trade_id_state before fill.
    pub fn set_gap_detector_seeds(&mut self, seeds: HashMap<String, i64>) {
        self.config.gap_detector_seeds = seeds;
    }

    /// Start WebSocket connections ONLY -- buffer trades without processing (#286).
    ///
    /// Spawns WS connections immediately so trades start buffering in the
    /// mpsc channels (capacity 50,000 per symbol). Call this BEFORE loading
    /// checkpoints or seeding gap detectors. Then call `start()` to begin
    /// processing the buffered + live trades seamlessly.
    ///
    /// Safe to call multiple times (idempotent) -- if WS is already started
    /// or `start()` was called, this is a no-op.
    pub fn start_ws(&mut self) {
        if self.ws_trade_receivers.is_some() || self.started {
            return; // Already started WS or fully started
        }

        let mut receivers = HashMap::new();
        let mut senders = HashMap::new();

        // Create per-symbol channels -- same capacity, same Receiver type as before
        for symbol in self.config.symbol_thresholds.keys() {
            let symbol = symbol.to_uppercase();
            let (trade_tx, trade_rx) = mpsc::channel::<Tick>(50_000);
            senders.insert(symbol.clone(), trade_tx);
            receivers.insert(symbol, trade_rx);
        }

        match self.config.ws_mode {
            WsMode::Combined => {
                // Split symbols into 2 groups by volume for fault isolation (WS-01).
                // Each CombinedWebSocketStream has independent Antaeus backoff (WS-02).
                let symbols: Vec<String> = self
                    .config
                    .symbol_thresholds
                    .keys()
                    .map(|s| s.to_uppercase())
                    .collect();
                let (group_a, group_b) = split_symbols_by_volume(&symbols);

                let group_a_len = group_a.len();
                let group_b_len = group_b.len();

                // Spawn connection A (high-volume group)
                if !group_a.is_empty() {
                    let senders_a: HashMap<String, mpsc::Sender<Tick>> = group_a
                        .iter()
                        .filter_map(|s| senders.remove(s).map(|tx| (s.clone(), tx)))
                        .collect();
                    let ws_shutdown_a = self.shutdown.child_token();
                    let policy_a = self.config.reconnection_policy.clone();
                    let ws_url_a = self.config.ws_base_url.clone();
                    let ws_params_a = self.config.ws_params.clone();
                    tokio::spawn(async move {
                        CombinedWebSocketStream::run_with_reconnect(
                            group_a,
                            senders_a,
                            policy_a,
                            ws_shutdown_a,
                            ws_url_a,
                            ws_params_a,
                        )
                        .await;
                    });
                }

                // Spawn connection B (lower-volume group)
                if !group_b.is_empty() {
                    let senders_b: HashMap<String, mpsc::Sender<Tick>> = group_b
                        .iter()
                        .filter_map(|s| senders.remove(s).map(|tx| (s.clone(), tx)))
                        .collect();
                    let ws_shutdown_b = self.shutdown.child_token();
                    let policy_b = self.config.reconnection_policy.clone();
                    let ws_url_b = self.config.ws_base_url.clone();
                    let ws_params_b = self.config.ws_params.clone();
                    tokio::spawn(async move {
                        CombinedWebSocketStream::run_with_reconnect(
                            group_b,
                            senders_b,
                            policy_b,
                            ws_shutdown_b,
                            ws_url_b,
                            ws_params_b,
                        )
                        .await;
                    });
                }

                tracing::info!(
                    group_a = group_a_len,
                    group_b = group_b_len,
                    "started 2 combined WS connections (group_a={}, group_b={}) -- trades buffering until start()",
                    group_a_len,
                    group_b_len,
                );
            }
            WsMode::PerSymbol => {
                // Spawn N individual WS connections, one per symbol (D-01 rollback path).
                let policy = self.config.reconnection_policy.clone();
                for symbol in self.config.symbol_thresholds.keys() {
                    let symbol = symbol.to_uppercase();
                    if let Some(tx) = senders.remove(&symbol) {
                        let ws_shutdown = self.shutdown.child_token();
                        let policy = policy.clone();
                        let sym = symbol.clone();
                        let ws_url = self.config.ws_base_url.clone();
                        let ws_params = self.config.ws_params.clone();
                        tokio::spawn(async move {
                            BinanceWebSocketStream::run_with_reconnect(
                                &sym,
                                tx,
                                policy,
                                ws_shutdown,
                                ws_url,
                                ws_params,
                            )
                            .await;
                        });
                    }
                }

                let n_syms = self.config.symbol_thresholds.len();
                tracing::info!(
                    symbols = n_syms,
                    "started {} per-symbol WS connections -- trades buffering until start()",
                    n_syms,
                );
            }
        }

        self.ws_trade_receivers = Some(receivers);
    }

    /// Fill historical gap from REST before starting WS processing (#286).
    ///
    /// Creates processors (same as `start()`), loads checkpoints, feeds REST
    /// trades through them, and stores the processors with accumulated state.
    /// When `start()` is called next, it reuses these processors instead of
    /// creating fresh ones -- zero forming-bar loss, zero transition gap.
    ///
    /// Call sequence: `start_ws()` -> `fill_from_rest()` -> `start()`
    ///
    /// Constraints respected (from 15 failed attempts in git history):
    /// - Runs BEFORE engine is live (not Asclepius -- #284)
    /// - Same processors transition to WS (no boundary mismatch -- #290)
    /// - No DELETE operations (no clean-slate -- #286)
    /// - Per-symbol sequential (~200MB each, no OOM -- #285)
    pub async fn fill_from_rest(&mut self) -> Result<u64, ProcessingError> {
        if self.started {
            return Ok(0); // Engine already running, skip
        }

        let fill_start = std::time::Instant::now();

        // Phase 4 (PARA-01): Raise rate limiter ceiling during parallel fill
        if let Some(ref limiter) = self.config.rate_limiter {
            limiter.set_ceiling(4800);
            tracing::info!("fill_from_rest: rate limiter ceiling raised to 4800 for parallel fill");
        }

        // Extract shared state for spawned tasks. initial_checkpoints uses remove()
        // so we take ownership of the whole map and partition per-symbol before spawning.
        let mut initial_checkpoints = std::mem::take(&mut self.config.initial_checkpoints);
        let gap_detector_seeds = self.config.gap_detector_seeds.clone();
        // Phase 59: thresholds are per-symbol. Snapshot the whole map; per-symbol slices
        // are cloned out inside the loop below.
        let symbol_thresholds_snapshot = self.config.symbol_thresholds.clone();
        let include_microstructure = self.config.include_microstructure;
        let inter_bar_config = InterBarConfig {
            compute_tier2: self.config.compute_tier2,
            compute_tier3: self.config.compute_tier3,
            compute_hurst: self.config.compute_hurst,
            compute_permutation_entropy: self.config.compute_permutation_entropy,
            ..Default::default()
        };
        let intra_hurst = self
            .config
            .compute_hurst
            .unwrap_or(self.config.compute_tier3);
        let intra_pe = self
            .config
            .compute_permutation_entropy
            .unwrap_or(self.config.compute_tier3);
        let intra_bar_config = IntraBarConfig {
            compute_hurst: intra_hurst,
            compute_permutation_entropy: intra_pe,
        };
        let rate_limiter = self.config.rate_limiter.clone();
        let bar_buffer = self.bar_buffer.clone_ref();
        let metrics = Arc::clone(&self.metrics);

        // Phase 45: Endpoint config for REST fill
        let rest_base_url = self.config.rest_base_url.clone();
        let rest_headers = self.config.rest_headers.clone();

        // Phase 16: Per-symbol ouroboros modes for fill_from_rest
        let symbol_modes = self.config.symbol_modes.clone();

        // Phase 4 (PARA-01): Parallel fill with JoinSet + Semaphore(4)
        let semaphore = Arc::new(tokio::sync::Semaphore::new(4));
        let mut join_set = tokio::task::JoinSet::new();

        for (symbol, sym_thresholds) in &symbol_thresholds_snapshot {
            let symbol = symbol.to_uppercase();
            let thresholds: Vec<u32> = sym_thresholds.clone();

            // Extract this symbol's checkpoints from the shared map
            let mut symbol_checkpoints: HashMap<
                (String, u32),
                opendeviationbar_core::checkpoint::Checkpoint,
            > = HashMap::new();
            for &threshold in &thresholds {
                let key = (symbol.clone(), threshold);
                if let Some(cp) = initial_checkpoints.remove(&key) {
                    symbol_checkpoints.insert(key, cp);
                }
            }

            let ch_seed_tid = gap_detector_seeds.get(&symbol).copied();
            let inter_bar_config = inter_bar_config.clone();
            let intra_bar_config = intra_bar_config.clone();
            let rate_limiter = rate_limiter.clone();
            let bar_buffer = bar_buffer.clone_ref();
            let metrics = Arc::clone(&metrics);
            let sem = Arc::clone(&semaphore);
            // Phase 45: Endpoint config for REST fill
            let rest_base_url = rest_base_url.clone();
            let rest_headers = rest_headers.clone();
            // Phase 16: Per-symbol ouroboros mode for REST fill
            let ouroboros_mode = symbol_modes.get(&symbol).copied().unwrap_or_default();
            // Issue #318: Pass extra_sinks for fill_from_rest fan-out.
            // Type is Option<Arc<Mutex<Vec<Box<dyn BarSink>>>>> — NOT Copy.
            // Arc clone is a cheap refcount bump; required because we move into the async block.
            #[allow(unused_variables)]
            let extra_sinks = self.extra_sinks.clone();

            join_set.spawn(async move {
                // Acquire semaphore permit -- at most 4 symbols fill concurrently
                let _permit = sem.acquire().await.expect("semaphore closed unexpectedly");

                // Create processors per threshold (same as start())
                let mut processors: Vec<(u32, OpenDeviationBarProcessor)> =
                    Vec::with_capacity(thresholds.len());
                for &threshold in &thresholds {
                    let key = (symbol.clone(), threshold);
                    let p = if let Some(cp) = symbol_checkpoints.remove(&key) {
                        match OpenDeviationBarProcessor::from_checkpoint(cp) {
                            Ok(mut restored) => {
                                if include_microstructure {
                                    restored = restored
                                        .with_inter_bar_config(inter_bar_config.clone())
                                        .with_intra_bar_features()
                                        .with_intra_bar_config(intra_bar_config.clone());
                                }
                                tracing::info!(
                                    %symbol, threshold,
                                    has_incomplete_bar = restored.get_incomplete_bar().is_some(),
                                    "fill_from_rest: processor restored from checkpoint"
                                );
                                restored
                            }
                            Err(e) => {
                                tracing::warn!(
                                    %symbol, threshold, ?e,
                                    "fill_from_rest: checkpoint restore failed, starting fresh"
                                );
                                OpenDeviationBarProcessor::new(threshold)?
                            }
                        }
                    } else {
                        let mut fresh = OpenDeviationBarProcessor::new(threshold)?;
                        if include_microstructure {
                            fresh = fresh
                                .with_inter_bar_config(inter_bar_config.clone())
                                .with_intra_bar_features()
                                .with_intra_bar_config(intra_bar_config.clone());
                        }
                        fresh
                    };
                    processors.push((threshold, p));
                }

                // Determine REST start point: min across all thresholds' checkpoints.
                // Issue #345: processors.first() only checked @100 (highest TID),
                // causing @250/@500/@750 to miss trades between their checkpoint
                // and @100's checkpoint, producing overlapping bars at restart.
                // Fix: use min() so ALL thresholds get all needed trades.
                let checkpoint_tid = processors
                    .iter()
                    .filter_map(|(_, p)| p.last_agg_trade_id())
                    .min();
                let from_tid = match (checkpoint_tid, ch_seed_tid) {
                    (Some(cp), Some(ch)) => Some(cp.min(ch)),
                    (Some(cp), None) => Some(cp),
                    (None, Some(ch)) => Some(ch),
                    (None, None) => None,
                };

                let mut symbol_bars: u64 = 0;
                let mut symbol_trades_total: u64 = 0;

                if let Some(from_id) = from_tid {
                    let symbol_fill_start = std::time::Instant::now();

                    // (#297) Dynamic chunk sizing: probe latest TID to bound the fill range.
                    // Costs 1 API weight per symbol. Eliminates empty-chunk waste (~42% of
                    // all API weight was burned on empty pages before this change).
                    let latest_tid: Option<i64> = {
                        let mut loader = opendeviationbar_providers::binance::historical::HistoricalDataLoader::new(&symbol)
                            .with_rest_base_url(rest_base_url.clone())
                            .with_rest_headers(rest_headers.clone());
                        if let Some(ref rl) = rate_limiter {
                            loader = loader.with_rate_limiter(Arc::clone(rl));
                        }
                        loader.fetch_latest_aggtrade().await.ok().map(|t| t.ref_id)
                    };

                    if let Some(lt) = latest_tid {
                        if lt <= from_id {
                            tracing::info!(
                                %symbol, from_id, latest_tid = lt,
                                "fill_from_rest: already caught up (latest_tid <= from_id)"
                            );
                            // Return processors for reuse -- symbol is current
                            return Ok((symbol, processors, 0, 0));
                        }
                        tracing::info!(
                            %symbol, from_id, latest_tid = lt,
                            actual_gap = lt - from_id,
                            "fill_from_rest: fetching trades from REST (bounded)"
                        );
                    } else {
                        tracing::info!(
                            %symbol, from_id,
                            "fill_from_rest: fetching trades from REST (unbounded -- latest TID probe failed)"
                        );
                    }

                    // Chunked REST pagination: fetch 50K trades at a time to avoid OOM.
                    const CHUNK_SIZE: i64 = 50_000;
                    const MAX_EMPTY_RETRIES: u32 = 3;
                    let mut current_from = from_id + 1;
                    let mut consecutive_errors: u32 = 0;
                    let mut consecutive_empty: u32 = 0;
                    let symbol_arc: Arc<str> = Arc::from(symbol.as_str());
                    let mut last_days: Vec<i64> =
                        processors.iter().map(|_| -1i64).collect();
                    // Phase 52: Track last trade timestamp per processor for Week gap detection
                    let mut last_trade_timestamps_us: Vec<i64> =
                        processors.iter().map(|_| 0i64).collect();

                    // Upper bound: latest_tid + 1 (if known) or open-ended
                    // to_id is inclusive — ceiling is the last TID we want to fetch.
                    let fill_ceiling = latest_tid;

                    loop {
                        // (#297) If we have a known ceiling and current_from exceeds it,
                        // we've covered the entire range -- no need for consecutive-empty heuristic.
                        if let Some(ceiling) = fill_ceiling && current_from > ceiling {
                            tracing::info!(
                                %symbol, symbol_trades_total, current_from, ceiling,
                                "fill_from_rest: reached known ceiling -- fill complete"
                            );
                            break;
                        }
                        // to_id is inclusive — subtract 1 from chunk range to keep same page size.
                        let chunk_to = if let Some(ceiling) = fill_ceiling {
                            (current_from + CHUNK_SIZE - 1).min(ceiling)
                        } else {
                            current_from + CHUNK_SIZE - 1
                        };
                        let trades = match opendeviationbar_providers::binance::parallel_fetch::fetch_aggtrades_parallel(
                            &symbol,
                            current_from,
                            chunk_to,
                            rate_limiter.clone(),
                            None,
                            Some(rest_base_url.clone()),
                            Some(rest_headers.clone()),
                        )
                        .await
                        {
                            Ok(t) => {
                                consecutive_errors = 0;
                                t
                            }
                            Err(e) => {
                                consecutive_errors += 1;
                                tracing::warn!(
                                    %symbol, ?e, chunk_from = current_from,
                                    chunk_to, consecutive_errors,
                                    symbol_trades_total,
                                    "fill_from_rest: REST chunk failed"
                                );
                                if consecutive_errors >= 3 {
                                    tracing::error!(
                                        %symbol, consecutive_errors, symbol_trades_total,
                                        chunk_from = current_from,
                                        "fill_from_rest: ABORTING after 3 consecutive errors -- \
                                         gap from tid {} will remain until kintsugi repairs",
                                        current_from
                                    );
                                    break;
                                }
                                // Advance past this chunk range to try the next one.
                                // chunk_to is inclusive, so +1 to avoid re-fetching the
                                // same boundary trade and to not permanently skip trades.
                                current_from = chunk_to + 1;
                                continue;
                            }
                        };

                        if trades.is_empty() {
                            consecutive_empty += 1;
                            if consecutive_empty >= MAX_EMPTY_RETRIES {
                                tracing::info!(
                                    %symbol, symbol_trades_total, current_from,
                                    "fill_from_rest: caught up (3 consecutive empty chunks)"
                                );
                                break;
                            }
                            tracing::debug!(
                                %symbol, current_from, chunk_to, consecutive_empty,
                                "fill_from_rest: empty chunk, skipping ahead"
                            );
                            current_from = chunk_to + 1;
                            continue;
                        }
                        consecutive_empty = 0;

                        let chunk_len = trades.len();
                        let last_tid = trades[chunk_len - 1].ref_id;

                        // Process this chunk through all threshold processors
                        for trade in &trades {
                            for (idx, (threshold, processor)) in
                                processors.iter_mut().enumerate()
                            {
                                // Phase 52: Week gap check before midnight check
                                if let OuroborosMode::Week { max_gap_us } = ouroboros_mode
                                    && let Some(orphan) = maybe_reset_at_week_gap(
                                        processor,
                                        trade.timestamp,
                                        &mut last_trade_timestamps_us[idx],
                                        max_gap_us,
                                    )
                                {
                                    metrics
                                        .bars_emitted
                                        .fetch_add(1, Ordering::Relaxed);
                                    let completed = CompletedBar {
                                        symbol: symbol_arc.clone(),
                                        threshold_decimal_bps: *threshold,
                                        bar: orphan,
                                    };
                                    if !bar_buffer.push(completed.clone()) {
                                        metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                                        tracing::warn!(%symbol, threshold = *threshold, "fill_from_rest: ring buffer full, bar dropped (week-gap orphan)");
                                    }
                                    #[cfg(feature = "clickhouse-sink")]
                                    if let Some(ref sinks) = extra_sinks {
                                        crate::clickhouse_writer::guards::dispatch_to_sinks(&completed, sinks);
                                    }
                                    symbol_bars += 1;
                                }

                                if let Some(orphan) = maybe_reset_at_midnight(
                                    processor,
                                    trade.timestamp,
                                    &mut last_days[idx],
                                    ouroboros_mode,
                                ) {
                                    metrics
                                        .bars_emitted
                                        .fetch_add(1, Ordering::Relaxed);
                                    let completed = CompletedBar {
                                        symbol: symbol_arc.clone(),
                                        threshold_decimal_bps: *threshold,
                                        bar: orphan,
                                    };
                                    if !bar_buffer.push(completed.clone()) {
                                        metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                                        tracing::warn!(%symbol, threshold = *threshold, "fill_from_rest: ring buffer full, bar dropped (midnight orphan)");
                                    }
                                    // Issue #318: Fan-out to extra sinks
                                    #[cfg(feature = "clickhouse-sink")]
                                    if let Some(ref sinks) = extra_sinks {
                                        crate::clickhouse_writer::guards::dispatch_to_sinks(&completed, sinks);
                                    }
                                    symbol_bars += 1;
                                }

                                match processor.process_single_trade(trade) {
                                    Ok(Some(bar)) => {
                                        metrics
                                            .bars_emitted
                                            .fetch_add(1, Ordering::Relaxed);
                                        let completed = CompletedBar {
                                            symbol: symbol_arc.clone(),
                                            threshold_decimal_bps: *threshold,
                                            bar,
                                        };
                                        if !bar_buffer.push(completed.clone()) {
                                            metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
                                            tracing::warn!(%symbol, threshold = *threshold, "fill_from_rest: ring buffer full, bar dropped");
                                        }
                                        // Issue #318: Fan-out to extra sinks
                                        #[cfg(feature = "clickhouse-sink")]
                                        if let Some(ref sinks) = extra_sinks {
                                            crate::clickhouse_writer::guards::dispatch_to_sinks(&completed, sinks);
                                        }
                                        symbol_bars += 1;
                                    }
                                    Ok(None) => {}
                                    Err(e) => {
                                        tracing::warn!(
                                            %symbol, threshold = *threshold, ?e,
                                            "fill_from_rest: trade processing error"
                                        );
                                    }
                                }
                            }
                        }

                        symbol_trades_total += chunk_len as u64;
                        current_from = last_tid + 1;

                        tracing::debug!(
                            %symbol, chunk_len, last_tid, symbol_trades_total,
                            "fill_from_rest: chunk processed"
                        );
                    }

                    if symbol_trades_total > 0 {
                        tracing::info!(
                            %symbol,
                            duration_ms = symbol_fill_start.elapsed().as_millis() as u64,
                            trade_count = symbol_trades_total,
                            bars_produced = symbol_bars,
                            last_tid = current_from - 1,
                            "fill_from_rest.symbol_summary"
                        );
                        metrics
                            .gap_trades_recovered
                            .fetch_add(symbol_trades_total, Ordering::Relaxed);
                    } else {
                        tracing::info!(
                            %symbol,
                            duration_ms = symbol_fill_start.elapsed().as_millis() as u64,
                            trade_count = 0u64,
                            bars_produced = 0u64,
                            "fill_from_rest.symbol_summary"
                        );
                    }
                } else {
                    tracing::info!(%symbol, "fill_from_rest: no CH seed -- skipping (first-ever start)");
                }

                Ok::<_, ProcessingError>((symbol, processors, symbol_bars, symbol_trades_total))
            });
        }

        // Collect results from all parallel tasks
        let mut total_bars = 0u64;
        let mut all_processors: HashMap<String, Vec<(u32, OpenDeviationBarProcessor)>> =
            HashMap::new();

        while let Some(result) = join_set.join_next().await {
            match result {
                Ok(Ok((symbol, processors, bars, _trades))) => {
                    total_bars += bars;
                    all_processors.insert(symbol, processors);
                }
                Ok(Err(e)) => {
                    tracing::error!(
                        ?e,
                        "fill_from_rest: symbol task failed with processing error"
                    );
                }
                Err(e) => {
                    tracing::error!(?e, "fill_from_rest: symbol task panicked");
                }
            }
        }

        // Phase 4 (PARA-03): Lower rate limiter ceiling after fill completes
        if let Some(ref limiter) = self.config.rate_limiter {
            limiter.set_ceiling(3000);
            tracing::info!("fill_from_rest: rate limiter ceiling lowered to 3000 after fill");
        }

        let total_duration_ms = fill_start.elapsed().as_millis() as u64;
        tracing::info!(
            total_bars,
            total_duration_ms,
            symbols = all_processors.len(),
            "fill_from_rest: complete -- processors preserved for start()"
        );
        self.pre_created_processors = Some(all_processors);
        Ok(total_bars)
    }

    /// Start all WebSocket connections and processing loops.
    ///
    /// If `start_ws()` was called first, reuses the already-buffered WS
    /// channels (zero-gap transition). Otherwise, creates WS connections
    /// inline (legacy behavior).
    ///
    /// Spawns one tokio task per symbol. Each task:
    /// 1. Connects via `run_with_reconnect()` (independent backoff per symbol)
    /// 2. Fans trades out to N `OpenDeviationBarProcessor` instances (one per threshold)
    /// 3. Sends completed bars to the shared output channel
    pub fn start(&mut self) -> Result<(), ProcessingError> {
        if self.started {
            return Ok(());
        }

        // #347: Serialize junction fills across symbols to prevent rate limiter
        // exhaustion. Semaphore(1) means only one symbol does puck_fill at a time.
        // With 16 symbols and typical 100-1000 trade gaps, total serialized time
        // is ~16s — acceptable since WS trades continue buffering.
        let junction_semaphore = Arc::new(tokio::sync::Semaphore::new(1));

        // Phase 59: snapshot per-symbol threshold list into a local Vec so the
        // loop body can freely call `&mut self` methods (remove/push) on the
        // various internal maps without borrow conflicts.
        let symbol_thresholds_list: Vec<(String, Vec<u32>)> = self
            .config
            .symbol_thresholds
            .iter()
            .map(|(s, t)| (s.clone(), t.clone()))
            .collect();

        for (symbol, sym_thresholds) in symbol_thresholds_list {
            let symbol = symbol.to_uppercase();
            let thresholds = sym_thresholds;
            let include_microstructure = self.config.include_microstructure;
            // Issue #128: Build tier-aware InterBarConfig + IntraBarConfig
            let inter_bar_config = InterBarConfig {
                compute_tier2: self.config.compute_tier2,
                compute_tier3: self.config.compute_tier3,
                compute_hurst: self.config.compute_hurst,
                compute_permutation_entropy: self.config.compute_permutation_entropy,
                ..Default::default()
            };
            let intra_hurst = self
                .config
                .compute_hurst
                .unwrap_or(self.config.compute_tier3);
            let intra_pe = self
                .config
                .compute_permutation_entropy
                .unwrap_or(self.config.compute_tier3);
            let intra_bar_config = IntraBarConfig {
                compute_hurst: intra_hurst,
                compute_permutation_entropy: intra_pe,
            };
            // Issue #96 Task #9: Clone ring buffer reference for symbol task
            let bar_buffer = self.bar_buffer.clone_ref();
            let shutdown = self.shutdown.clone();
            let policy = self.config.reconnection_policy.clone();
            let metrics = Arc::clone(&self.metrics);
            let checkpoint_tx = self.checkpoint_tx.clone();

            // Issue #286: Reuse pre-created processors from fill_from_rest() if available.
            // These processors already have accumulated state from REST trades --
            // the SAME processors transition seamlessly to WS mode.
            let processors: Vec<(u32, OpenDeviationBarProcessor)> = if let Some(pre) = self
                .pre_created_processors
                .as_mut()
                .and_then(|m| m.remove(&symbol))
            {
                tracing::info!(
                    %symbol,
                    thresholds = pre.len(),
                    "reusing pre-created processors from fill_from_rest()"
                );
                pre
            } else {
                // Legacy path: create fresh processors (no fill_from_rest was called)
                let mut procs: Vec<(u32, OpenDeviationBarProcessor)> =
                    Vec::with_capacity(thresholds.len());
                for &threshold in &thresholds {
                    // Try to restore from checkpoint, fall back to fresh processor
                    let key = (symbol.clone(), threshold);
                    let p = if let Some(cp) = self.config.initial_checkpoints.remove(&key) {
                        match OpenDeviationBarProcessor::from_checkpoint(cp) {
                            Ok(mut restored) => {
                                // Re-enable microstructure features after checkpoint restore
                                if include_microstructure {
                                    restored = restored
                                        .with_inter_bar_config(inter_bar_config.clone())
                                        .with_intra_bar_features()
                                        .with_intra_bar_config(intra_bar_config.clone());
                                }
                                tracing::info!(
                                    %symbol, threshold,
                                    has_incomplete_bar = restored.get_incomplete_bar().is_some(),
                                    "processor restored from checkpoint"
                                );
                                restored
                            }
                            Err(e) => {
                                tracing::warn!(
                                    %symbol, threshold, ?e,
                                    "checkpoint restore failed, starting fresh"
                                );
                                let mut fresh = OpenDeviationBarProcessor::new(threshold)?;
                                if include_microstructure {
                                    fresh = fresh
                                        .with_inter_bar_config(inter_bar_config.clone())
                                        .with_intra_bar_features()
                                        .with_intra_bar_config(intra_bar_config.clone());
                                }
                                fresh
                            }
                        }
                    } else {
                        let mut fresh = OpenDeviationBarProcessor::new(threshold)?;
                        if include_microstructure {
                            fresh = fresh
                                .with_inter_bar_config(inter_bar_config.clone())
                                .with_intra_bar_features()
                                .with_intra_bar_config(intra_bar_config.clone());
                        }
                        fresh
                    };
                    procs.push((threshold, p));
                }
                procs
            };

            // Issue #214: Extract forming bar senders for this symbol's thresholds
            let symbol_arc: Arc<str> = Arc::from(symbol.as_str());
            let mut forming_txs: Vec<(u32, watch::Sender<Option<FormingBar>>)> = Vec::new();
            for &threshold in &thresholds {
                let key = (symbol_arc.clone(), threshold);
                if let Some(tx) = self.forming_bar_txs.remove(&key) {
                    forming_txs.push((threshold, tx));
                }
            }

            // Create per-symbol checkpoint snapshot channel
            let (snapshot_tx, snapshot_rx) = mpsc::channel::<()>(1);
            self.checkpoint_snapshot_txs.push(snapshot_tx);

            // Spawn per-symbol task
            let rate_limiter = self.config.rate_limiter.clone();
            let gap_event_tx = self.gap_event_tx.clone();
            // Issue #257: Extract per-symbol gap-fill receiver for command channel
            let gap_fill_rx = self
                .gap_fill_receivers
                .as_mut()
                .and_then(|m| m.remove(&symbol));
            // ClickHouse seed for gap detector (max trade ID across thresholds)
            let ch_seed_tid = self.config.gap_detector_seeds.get(&symbol).copied();
            // Issue #286: Reuse pre-buffered WS receiver if start_ws() was called
            let pre_buffered_rx = self
                .ws_trade_receivers
                .as_mut()
                .and_then(|m| m.remove(&symbol));
            // Phase 16: Per-symbol ouroboros mode (defaults to Day if not in map)
            let ouroboros_mode = self.config.symbol_modes.get(&symbol).copied().unwrap_or_default();
            // Issue #318: Pass extra_sinks for this symbol task.
            // Type is Option<Arc<Mutex<Vec<Box<dyn BarSink>>>>> — NOT Copy.
            // Arc clone is a cheap refcount bump; required because we move into the spawned task.
            let extra_sinks = self.extra_sinks.clone();
            let junction_sem = Arc::clone(&junction_semaphore);
            tokio::spawn(symbol_task(
                symbol,
                processors,
                bar_buffer,
                checkpoint_tx,
                policy,
                shutdown,
                metrics,
                rate_limiter,
                forming_txs,
                gap_event_tx,
                gap_fill_rx,
                ch_seed_tid,
                pre_buffered_rx,
                ouroboros_mode,
                extra_sinks,
                snapshot_rx,
                junction_sem,
            ));
        }

        self.started = true;
        tracing::info!(
            symbol_thresholds = ?self.config.symbol_thresholds,
            microstructure = self.config.include_microstructure,
            "live bar engine started"
        );
        Ok(())
    }

    /// Receive next completed bar. Returns `None` on timeout or shutdown.
    /// Issue #96 Task #9: Poll ring buffer with timeout, replacing async channel
    /// Memory v2: Exponential backoff 100us->10ms reduces idle CPU ~100x
    pub async fn next_bar(&mut self, timeout: Duration) -> Option<CompletedBar> {
        let start = tokio::time::Instant::now();
        let mut poll_interval = Duration::from_micros(100);
        let max_interval = Duration::from_millis(10);

        loop {
            // Try to pop from ring buffer (non-blocking)
            if let Some(bar) = self.bar_buffer.pop() {
                return Some(bar);
            }

            // Check timeout
            if start.elapsed() >= timeout {
                return None;
            }

            // Check shutdown
            if self.shutdown.is_cancelled() {
                return None;
            }

            // Sleep with exponential backoff (resets on successful pop above)
            tokio::time::sleep(poll_interval).await;
            poll_interval = (poll_interval * 2).min(max_interval);
        }
    }

    /// Drain all completed bars from the ring buffer synchronously.
    ///
    /// Returns bars in FIFO order. Purely synchronous -- no async, no polling.
    /// Used by sync_flush to bypass the async next_bar() polling loop which
    /// can fail when called via block_on() before engine.start().
    pub fn drain_bars(&self) -> Vec<CompletedBar> {
        self.bar_buffer.drain_all()
    }

    /// Graceful shutdown -- cancels all WebSocket tasks.
    pub fn stop(&self) {
        tracing::info!("live bar engine stopping");
        self.shutdown.cancel();
    }

    /// Get engine metrics snapshot.
    pub fn metrics(&self) -> &LiveEngineMetrics {
        &self.metrics
    }

    /// Whether the engine has been started.
    pub fn is_started(&self) -> bool {
        self.started
    }

    /// Get the ClickHouse writer flush thread metrics (Issue #318).
    ///
    /// Returns `Some` when `clickhouse-sink` feature is enabled and
    /// `OPENDEVIATIONBAR_CH_HOSTS` was set at engine creation time.
    /// Returns `None` otherwise.
    #[cfg(feature = "clickhouse-sink")]
    pub fn ch_writer_metrics(&self) -> Option<&Arc<crate::clickhouse_writer::flush_thread::FlushThreadMetrics>> {
        self.ch_writer_metrics.as_ref()
    }

    /// Collect checkpoints from all processors.
    ///
    /// Non-blocking: sends a snapshot request to all symbol tasks via
    /// `checkpoint_snapshot_txs`, then drains responses from `checkpoint_rx`.
    /// Works during live streaming (proactive) and after `stop()` (shutdown path).
    /// Returns a map of `"SYMBOL:THRESHOLD"` -> `Checkpoint`.
    pub async fn collect_checkpoints(&mut self, timeout: Duration) -> HashMap<String, Checkpoint> {
        let mut result = HashMap::new();

        // Send snapshot request to all symbol tasks (non-blocking, best-effort)
        let expected_count: usize = self
            .config
            .symbol_thresholds
            .values()
            .map(|v| v.len())
            .sum::<usize>();
        let mut sent = 0usize;
        for tx in &self.checkpoint_snapshot_txs {
            if tx.try_send(()).is_ok() {
                sent += 1;
            }
        }
        if sent > 0 {
            tracing::debug!(sent, expected_count, "checkpoint snapshot requests sent");
        }

        // If checkpoint_rx was already taken (by take_checkpoint_receiver),
        // we cannot collect here — the external owner is responsible.
        let rx = match self.checkpoint_rx.as_mut() {
            Some(rx) => rx,
            None => return result,
        };

        // Drain all available checkpoints within timeout
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            tokio::select! {
                item = rx.recv() => {
                    match item {
                        Some((symbol, threshold, cp)) => {
                            let key = format!("{symbol}:{threshold}");
                            result.insert(key, cp);
                            // Stop early if we have all expected checkpoints
                            if result.len() >= expected_count {
                                break;
                            }
                        }
                        None => break, // Channel closed, all senders dropped
                    }
                }
                () = tokio::time::sleep_until(deadline) => {
                    tracing::warn!(
                        collected = result.len(),
                        expected = expected_count,
                        "checkpoint collection timed out"
                    );
                    break;
                }
            }
        }

        tracing::info!(count = result.len(), "checkpoints collected");
        result
    }

    /// Get the shutdown token (for external coordination).
    pub fn shutdown_token(&self) -> CancellationToken {
        self.shutdown.clone()
    }

    /// Take the checkpoint receiver for external collection (Issue #178).
    ///
    /// Used by OdbEngine to extract the receiver before the engine is moved
    /// into a tokio task. The caller is responsible for draining the receiver
    /// after shutdown.
    pub fn take_checkpoint_receiver(
        &mut self,
    ) -> Option<mpsc::Receiver<(String, u32, Checkpoint)>> {
        self.checkpoint_rx.take()
    }

    /// Take the forming bar watch receivers (Issue #214).
    ///
    /// Must be called BEFORE `start()`. Returns watch receivers keyed by
    /// (symbol, threshold). The caller reads these at 1Hz to get the latest
    /// forming bar snapshot for each pair -- zero contention with the trade
    /// processing hot path.
    ///
    /// Follows the same pattern as `take_checkpoint_receiver()` (Issue #178).
    pub fn take_forming_bar_watches(&mut self) -> Option<FormingBarWatches> {
        self.forming_bar_watches.take()
    }

    /// Get a shared reference to the metrics (Issue #178).
    pub fn shared_metrics(&self) -> Arc<LiveEngineMetrics> {
        Arc::clone(&self.metrics)
    }

    /// Take the gap event receiver (Issue #257).
    ///
    /// Returns the receiver half of the gap event channel. Gap events are
    /// emitted by `TradeIdGapDetector` in each symbol task when trade-ID
    /// discontinuities are detected. The caller (StreamManager/Python)
    /// polls this to react to gaps.
    ///
    /// Can only be called once -- subsequent calls return None.
    pub fn take_gap_event_receiver(&mut self) -> Option<mpsc::Receiver<GapEvent>> {
        self.gap_event_rx.take()
    }

    /// Get a clone of the per-symbol gap-fill senders (Issue #257).
    ///
    /// Returns a map of symbol -> `GapFillSender`. Each sender can be used
    /// with `&self` (no borrow conflict) to submit on-demand gap-fill
    /// commands that are processed inline by the symbol's `tokio::select!` loop.
    pub fn gap_fill_senders(&self) -> HashMap<String, GapFillSender> {
        self.gap_fill_senders.clone()
    }
}

#[cfg(test)]
mod tests {
    use super::trade_dispatch::{ExtraSinks, process_trade_through_processors};
    use super::types::OuroborosMode;
    use super::*;
    use crate::ring_buffer::ConcurrentRingBuffer;
    use opendeviationbar_core::{Tick, FixedPoint, OpenDeviationBar};
    use opendeviationbar_providers::binance::AdaptiveRateLimiter;
    use std::sync::atomic::Ordering;

    fn make_trade(id: i64, price: f64, timestamp_ms: u64, is_buyer_maker: bool) -> Tick {
        let price_str = format!("{price:.8}");
        Tick {
            ref_id: id,
            price: FixedPoint::from_str(&price_str).unwrap(),
            volume: FixedPoint::from_str("1.00000000").unwrap(),
            first_sub_id: id,
            last_sub_id: id,
            timestamp: opendeviationbar_core::normalize_timestamp(timestamp_ms),
            is_buyer_maker,
            is_best_match: None,
            best_bid: None,
            best_ask: None,
        }
    }

    fn sym_thresh(pairs: &[(&str, &[u32])]) -> HashMap<String, Vec<u32>> {
        let mut m = HashMap::new();
        for (sym, thrs) in pairs {
            m.insert((*sym).to_string(), thrs.to_vec());
        }
        m
    }

    #[test]
    fn test_live_engine_config_defaults() {
        let config = LiveEngineConfig::new(sym_thresh(&[
            ("BTCUSDT", &[250, 500, 1000]),
            ("ETHUSDT", &[250, 500, 1000]),
        ]));
        assert_eq!(config.symbol_thresholds.len(), 2);
        assert_eq!(config.symbol_thresholds["BTCUSDT"].len(), 3);
        assert!(config.include_microstructure);
        // Default from get_bar_channel_capacity() when OPENDEVIATIONBAR_MAX_PENDING_BARS is unset
        assert_eq!(config.bar_channel_capacity, 10_000);
        // Issue #128: Verify tier config defaults
        assert!(config.compute_tier2);
        assert!(!config.compute_tier3);
        assert_eq!(config.compute_hurst, None);
        assert_eq!(config.compute_permutation_entropy, None);
    }

    #[test]
    fn test_live_engine_config_rate_limiter() {
        let config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        assert!(config.rate_limiter.is_none());

        let limiter = Arc::new(AdaptiveRateLimiter::binance_default());
        let config = config.with_rate_limiter(limiter);
        assert!(config.rate_limiter.is_some());
        assert_eq!(config.rate_limiter.as_ref().unwrap().remaining(), 4800);
    }

    #[test]
    fn test_completed_bar_metadata() {
        let trade = make_trade(1, 50000.0, 1700000000000, false);
        let bar = OpenDeviationBar::new(&trade);
        let completed = CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar,
        };
        assert_eq!(&*completed.symbol, "BTCUSDT");
        assert_eq!(completed.threshold_decimal_bps, 250);
    }

    #[test]
    fn test_live_engine_creation() {
        let config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        let engine = LiveBarEngine::new(config);
        assert!(!engine.is_started());
    }

    #[test]
    fn test_metrics_snapshot() {
        let metrics = LiveEngineMetrics::default();
        metrics.trades_received.store(100, Ordering::Relaxed);
        metrics.bars_emitted.store(5, Ordering::Relaxed);
        metrics.reconnections.store(2, Ordering::Relaxed);
        metrics.dropped_bars.store(1, Ordering::Relaxed);
        metrics.max_queue_depth.store(50, Ordering::Relaxed);
        metrics.backpressure_events.store(3, Ordering::Relaxed);
        metrics.gap_fills.store(2, Ordering::Relaxed);
        metrics.gap_trades_recovered.store(150, Ordering::Relaxed);

        let snap = metrics.snapshot();
        assert_eq!(snap.trades_received, 100);
        assert_eq!(snap.bars_emitted, 5);
        assert_eq!(snap.reconnections, 2);
        assert_eq!(snap.dropped_bars, 1);
        assert_eq!(snap.max_queue_depth, 50);
        assert_eq!(snap.backpressure_events, 3);
        assert_eq!(snap.gap_fills, 2);
        assert_eq!(snap.gap_trades_recovered, 150);
    }

    #[test]
    fn test_ring_buffer_metrics() {
        // Issue #96 Task #12: Verify ring buffer metrics tracking
        let metrics = LiveEngineMetrics::default();

        // Initially all zeros
        assert_eq!(metrics.dropped_bars.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.backpressure_events.load(Ordering::Relaxed), 0);
        assert_eq!(metrics.max_queue_depth.load(Ordering::Relaxed), 0);

        // Simulate ring buffer drops
        metrics.dropped_bars.fetch_add(5, Ordering::Relaxed);
        metrics.backpressure_events.fetch_add(5, Ordering::Relaxed);
        let _ = metrics.max_queue_depth.fetch_max(100, Ordering::Relaxed);

        assert_eq!(metrics.dropped_bars.load(Ordering::Relaxed), 5);
        assert_eq!(metrics.backpressure_events.load(Ordering::Relaxed), 5);
        assert_eq!(metrics.max_queue_depth.load(Ordering::Relaxed), 100);

        // Verify snapshot captures all metrics
        let snap = metrics.snapshot();
        assert_eq!(snap.dropped_bars, 5);
        assert_eq!(snap.backpressure_events, 5);
        assert_eq!(snap.max_queue_depth, 100);
    }

    #[tokio::test]
    async fn test_engine_start_and_stop() {
        let config = LiveEngineConfig::new(sym_thresh(&[("BTCUSDT", &[250])]));
        let mut engine = LiveBarEngine::new(config);

        // Start should succeed (spawns tasks but WS won't connect in test)
        assert!(engine.start().is_ok());
        assert!(engine.is_started());

        // Double-start is idempotent
        assert!(engine.start().is_ok());

        // Stop
        engine.stop();

        // next_bar should return None after stop
        let bar = engine.next_bar(Duration::from_millis(50)).await;
        assert!(bar.is_none());
    }

    #[tokio::test]
    async fn test_processor_fan_out() {
        // Verify that trades fan out to all threshold processors correctly
        let thresholds = vec![250u32, 500, 750];
        let mut processors: Vec<(u32, OpenDeviationBarProcessor)> = thresholds
            .iter()
            .map(|&t| (t, OpenDeviationBarProcessor::new(t).unwrap()))
            .collect();

        // Process a series of trades -- each processor should maintain independent state
        let base_price = 50000.0;
        for i in 0..10 {
            let trade = make_trade(
                i,
                base_price + (i as f64) * 0.01,
                1700000000000 + (i as u64) * 1000,
                i % 2 == 0,
            );
            for (_, processor) in &mut processors {
                let _ = processor.process_single_trade(&trade);
            }
        }

        // All processors should have processed trades independently
        for (threshold, processor) in &processors {
            let incomplete = processor.get_incomplete_bar();
            assert!(
                incomplete.is_some(),
                "threshold {threshold} should have an incomplete bar"
            );
        }
    }

    #[tokio::test]
    async fn test_microstructure_processor_creation() {
        // Verify processors can be created with full microstructure features
        let threshold = 250u32;
        let p = OpenDeviationBarProcessor::new(threshold)
            .unwrap()
            .with_inter_bar_config(InterBarConfig::default())
            .with_intra_bar_features();

        assert!(p.inter_bar_enabled());
        assert!(p.intra_bar_enabled());
    }

    #[test]
    fn test_ring_buffer_drop_tracking() {
        // Issue #96 Task #12: Verify ring buffer correctly tracks dropped bars

        let ring_buf = ConcurrentRingBuffer::new(3);
        let metrics = LiveEngineMetrics::default();

        // Fill the buffer
        let bar1 = CompletedBar {
            symbol: Arc::from("BTC"),
            threshold_decimal_bps: 250,
            bar: { let t = make_trade(1, 50000.0, 1700000000000, false); OpenDeviationBar::new(&t) },
        };
        let bar2 = CompletedBar {
            symbol: Arc::from("BTC"),
            threshold_decimal_bps: 250,
            bar: { let t = make_trade(2, 50001.0, 1700000000001, false); OpenDeviationBar::new(&t) },
        };
        let bar3 = CompletedBar {
            symbol: Arc::from("BTC"),
            threshold_decimal_bps: 250,
            bar: { let t = make_trade(3, 50002.0, 1700000000002, false); OpenDeviationBar::new(&t) },
        };
        let bar4 = CompletedBar {
            symbol: Arc::from("BTC"),
            threshold_decimal_bps: 250,
            bar: { let t = make_trade(4, 50003.0, 1700000000003, false); OpenDeviationBar::new(&t) },
        };

        assert!(ring_buf.push(bar1));
        assert!(ring_buf.push(bar2));
        assert!(ring_buf.push(bar3));

        // Buffer is full (capacity=3), so next push should fail and drop oldest
        assert!(!ring_buf.push(bar4));

        // Simulate metrics tracking (as symbol_task would do)
        metrics.dropped_bars.fetch_add(1, Ordering::Relaxed);
        metrics.backpressure_events.fetch_add(1, Ordering::Relaxed);
        let current_depth = ring_buf.len() as u64;
        let _ = metrics
            .max_queue_depth
            .fetch_max(current_depth, Ordering::Relaxed);

        assert_eq!(metrics.dropped_bars.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.backpressure_events.load(Ordering::Relaxed), 1);
        assert_eq!(metrics.max_queue_depth.load(Ordering::Relaxed), 3);

        // Verify we can still retrieve bars
        assert_eq!(ring_buf.pop().is_some(), true); // bar2
        assert_eq!(ring_buf.pop().is_some(), true); // bar3
        assert_eq!(ring_buf.pop().is_some(), true); // bar4
        assert_eq!(ring_buf.pop().is_none(), true); // empty
    }

    #[test]
    fn test_live_engine_metrics_estimate_queue_depth() {
        // Issue #96 Task #12: Verify queue depth estimation
        let metrics = LiveEngineMetrics::default();

        metrics.trades_received.store(1000, Ordering::Relaxed);
        metrics.bars_emitted.store(500, Ordering::Relaxed);
        metrics.dropped_bars.store(0, Ordering::Relaxed);

        let estimated = metrics.estimate_queue_depth();
        assert_eq!(estimated, 500); // 1000 - 500 - 0
    }

    #[test]
    fn test_forming_bar_watch_channels_created() {
        // Issue #214: Verify watch channels are created for each (symbol, threshold) pair
        let config = LiveEngineConfig::new(sym_thresh(&[
            ("BTCUSDT", &[250, 500]),
            ("ETHUSDT", &[250, 500]),
        ]));
        let mut engine = LiveBarEngine::new(config);

        // Should have 2 symbols x 2 thresholds = 4 watch channels
        let watches = engine.take_forming_bar_watches();
        assert!(watches.is_some());
        let watches = watches.unwrap();
        assert_eq!(watches.len(), 4);

        // Verify all expected keys exist
        let btc: Arc<str> = Arc::from("BTCUSDT");
        let eth: Arc<str> = Arc::from("ETHUSDT");
        assert!(watches.contains_key(&(btc.clone(), 250)));
        assert!(watches.contains_key(&(btc, 500)));
        assert!(watches.contains_key(&(eth.clone(), 250)));
        assert!(watches.contains_key(&(eth, 500)));

        // Second call returns None (already taken)
        assert!(engine.take_forming_bar_watches().is_none());
    }

    #[test]
    fn test_forming_bar_watch_send_receive() {
        // Issue #214: Verify watch channel send/receive semantics
        let (tx, rx) = watch::channel::<Option<FormingBar>>(None);

        // Initially None
        assert!(rx.borrow().is_none());

        // Send a forming bar
        let trade = make_trade(1, 50000.0, 1700000000000, false);
        let bar = OpenDeviationBar::new(&trade);
        let forming = FormingBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar: bar.clone(),
            last_trade_timestamp_us: 1700000000000000,
        };
        tx.send(Some(forming)).unwrap();

        // Receiver sees the latest value
        let snapshot = rx.borrow();
        assert!(snapshot.is_some());
        let fb = snapshot.as_ref().unwrap();
        assert_eq!(&*fb.symbol, "BTCUSDT");
        assert_eq!(fb.threshold_decimal_bps, 250);
        assert_eq!(fb.last_trade_timestamp_us, 1700000000000000);
        drop(snapshot);

        // Clear on bar completion (send None)
        tx.send(None).unwrap();
        assert!(rx.borrow().is_none());
    }

    #[test]
    fn test_process_trade_through_processors_basic() {
        // Issue #273: Verify extracted process_trade_through_processors works correctly
        let thresholds = vec![250u32, 500];
        let mut processors: Vec<(u32, OpenDeviationBarProcessor)> = thresholds
            .iter()
            .map(|&t| (t, OpenDeviationBarProcessor::new(t).unwrap()))
            .collect();
        let mut last_days: Vec<i64> = processors.iter().map(|_| -1i64).collect();
        let mut last_trade_timestamps_us: Vec<i64> = processors.iter().map(|_| 0i64).collect();
        let bar_buffer = ConcurrentRingBuffer::new(100);
        let metrics = LiveEngineMetrics::default();
        let symbol_arc: Arc<str> = Arc::from("BTCUSDT");
        let forming_tx_map: HashMap<u32, watch::Sender<Option<FormingBar>>> = HashMap::new();
        let mut committed_floors: HashMap<u32, i64> = HashMap::new();
        let mut last_forming_update: HashMap<u32, i64> = HashMap::new();
        let extra_sinks: ExtraSinks = None;

        // Process a trade -- should be absorbed (no bar completed yet)
        let trade = make_trade(1, 50000.0, 1700000000000, false);
        process_trade_through_processors(
            &trade,
            &mut processors,
            &mut last_days,
            &mut last_trade_timestamps_us,
            &bar_buffer,
            &metrics,
            &symbol_arc,
            "BTCUSDT",
            &forming_tx_map,
            &mut committed_floors,
            &mut last_forming_update,
            OuroborosMode::Day,
            &extra_sinks,
        );

        // No bars should be emitted yet (single trade can't breach threshold)
        assert!(bar_buffer.pop().is_none());
        // Both processors should have an incomplete bar
        for (_, processor) in &processors {
            assert!(processor.get_incomplete_bar().is_some());
        }
    }

    #[test]
    fn test_burst_atomic_frame_same_ms_trades() {
        // Issue #273: Verify that a burst of same-ms trades is processed
        // atomically through all processors. Simulates the scenario where
        // 19 trades arrive with the same millisecond timestamp.
        let thresholds = vec![250u32];
        let mut processors: Vec<(u32, OpenDeviationBarProcessor)> = thresholds
            .iter()
            .map(|&t| (t, OpenDeviationBarProcessor::new(t).unwrap()))
            .collect();
        let mut last_days: Vec<i64> = processors.iter().map(|_| -1i64).collect();
        let mut last_trade_timestamps_us: Vec<i64> = processors.iter().map(|_| 0i64).collect();
        let bar_buffer = ConcurrentRingBuffer::new(100);
        let metrics = LiveEngineMetrics::default();
        let symbol_arc: Arc<str> = Arc::from("BTCUSDT");
        let forming_tx_map: HashMap<u32, watch::Sender<Option<FormingBar>>> = HashMap::new();
        let mut committed_floors: HashMap<u32, i64> = HashMap::new();
        let mut last_forming_update: HashMap<u32, i64> = HashMap::new();
        let extra_sinks: ExtraSinks = None;

        let base_price = 50000.0;
        let same_ms = 1700000000000u64; // All trades at same millisecond

        // Create a burst of 20 trades at the same millisecond
        let mut frame: Vec<Tick> = Vec::with_capacity(20);
        for i in 0..20 {
            // Vary price slightly to stay within threshold (not breach)
            let price = base_price + (i as f64) * 0.01;
            frame.push(make_trade(100 + i, price, same_ms, i % 2 == 0));
        }

        // Process all trades in the frame atomically
        for trade in &frame {
            process_trade_through_processors(
                trade,
                &mut processors,
                &mut last_days,
                &mut last_trade_timestamps_us,
                &bar_buffer,
                &metrics,
                &symbol_arc,
                "BTCUSDT",
                &forming_tx_map,
                &mut committed_floors,
                &mut last_forming_update,
                OuroborosMode::Day,
                &extra_sinks,
            );
        }

        // All 20 trades should have been processed
        // The processor should have an incomplete bar containing all trades
        let (_, processor) = &processors[0];
        let incomplete = processor.get_incomplete_bar().unwrap();
        assert_eq!(incomplete.first_agg_trade_id, 100);
        assert_eq!(incomplete.last_agg_trade_id, 119);
    }

    #[test]
    fn test_max_burst_drain_constant() {
        // Issue #273: Verify the burst drain limit is sensible
        // MAX_BURST_DRAIN is in symbol_task module, test the value indirectly
        // by verifying the concept: 1024 provides generous headroom for same-ms bursts.
        assert!(
            1024 > 50,
            "burst drain limit must exceed typical burst size"
        );
    }

    #[test]
    fn test_junction_fill_zero_stathera_gaps() {
        // Simulate REST->WS junction scenario
        let (gap_tx, mut gap_rx) = mpsc::channel::<crate::gap::GapEvent>(16);
        let mut gap_detector =
            crate::gap::TradeIdGapDetector::new(Arc::from("TESTUSDT"), Some(gap_tx));

        // Simulate fill_from_rest: process trades 100-104
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        for i in 0..5i64 {
            let trade = make_trade(
                100 + i,
                50000.0 + (i as f64) * 0.01,
                1700000000000 + (i as u64) * 1000,
                i % 2 == 0,
            );
            let _ = processor.process_single_trade(&trade);
        }
        assert_eq!(processor.last_agg_trade_id(), Some(104));
        assert!(
            processor.get_incomplete_bar().is_some(),
            "forming bar must exist after REST fill"
        );

        // WS first trade arrives at ID 110 (junction gap: 105-109)
        let first_ws_tid = 110i64;

        // JUNC-03: Re-seed gap detector past junction
        gap_detector.seed(first_ws_tid);

        // Subsequent WS trades should NOT trigger gap detection
        // check(111) -> None because 111 <= 110+1
        assert!(
            gap_detector.check(111).is_none(),
            "no gap after junction re-seed"
        );
        // Must update_last_tid before next check (check doesn't update state)
        gap_detector.update_last_tid(111);
        // check(112) -> None because 112 <= 111+1
        assert!(
            gap_detector.check(112).is_none(),
            "consecutive trade also no gap"
        );

        // No gap events should have been emitted
        assert!(
            gap_rx.try_recv().is_err(),
            "no gap events after junction re-seed"
        );

        // Processor forming bar is still intact (JUNC-01)
        assert!(
            processor.get_incomplete_bar().is_some(),
            "forming bar preserved across junction"
        );
    }

    #[test]
    fn test_junction_zero_gap_no_fill_needed() {
        // Zero-gap junction: first WS trade is exactly last_rest_tid + 1
        let (gap_tx, mut gap_rx) = mpsc::channel::<crate::gap::GapEvent>(16);
        let mut gap_detector =
            crate::gap::TradeIdGapDetector::new(Arc::from("TESTUSDT"), Some(gap_tx));

        // REST fill: trades 100-104
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();
        for i in 0..5i64 {
            let trade = make_trade(
                100 + i,
                50000.0 + (i as f64) * 0.01,
                1700000000000 + (i as u64) * 1000,
                i % 2 == 0,
            );
            let _ = processor.process_single_trade(&trade);
        }
        assert_eq!(processor.last_agg_trade_id(), Some(104));

        // First WS trade at ID 105 -- zero gap
        let first_ws_tid = 105i64;
        let last_rest_tid = processor.last_agg_trade_id().unwrap();
        let junction_gap = first_ws_tid - last_rest_tid - 1;
        assert_eq!(
            junction_gap, 0,
            "no junction gap when WS continues exactly from REST"
        );

        // JUNC-03: Still re-seed gap detector
        gap_detector.seed(first_ws_tid);

        // Next WS trade should not trigger gap
        assert!(
            gap_detector.check(106).is_none(),
            "no gap for consecutive trade after zero-gap junction"
        );

        // No gap events emitted
        assert!(
            gap_rx.try_recv().is_err(),
            "no gap events for zero-gap junction"
        );

        // Forming bar intact
        assert!(
            processor.get_incomplete_bar().is_some(),
            "forming bar preserved in zero-gap junction"
        );
    }

    #[test]
    fn test_junction_monotonicity_guard() {
        // JUNC-04: Overlapping WS trades must be silently dropped
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Simulate REST + junction fill: trades 100-109
        for i in 0..10i64 {
            let trade = make_trade(
                100 + i,
                50000.0 + (i as f64) * 0.01,
                1700000000000 + (i as u64) * 1000,
                i % 2 == 0,
            );
            let _ = processor.process_single_trade(&trade);
        }
        assert_eq!(processor.last_agg_trade_id(), Some(109));

        // Overlapping WS trades (IDs 107-109) should be dropped
        let last_tid = processor.last_agg_trade_id().unwrap();
        for overlap_id in [107i64, 108, 109] {
            assert!(
                overlap_id <= last_tid,
                "trade {overlap_id} should be dropped (monotonicity guard)"
            );
        }

        // Non-overlapping trade (ID 110) should be processed
        assert!(110 > last_tid, "trade 110 should pass monotonicity guard");
        let trade_110 = make_trade(110, 50000.10, 1700000010000, true);
        let _ = processor.process_single_trade(&trade_110);
        assert_eq!(processor.last_agg_trade_id(), Some(110));
    }

    #[test]
    fn test_junction_forming_bar_trade_count_preserved() {
        // JUNC-01: Forming bar must carry trades across REST->junction->WS
        let mut processor = OpenDeviationBarProcessor::new(250).unwrap();

        // Phase 1: REST fill -- 20 trades (no breach at 250 dbps = 2.5%)
        for i in 0..20i64 {
            let trade = make_trade(
                i + 1,
                50000.0 + (i as f64) * 0.001, // tiny increments, no breach
                1700000000000 + (i as u64) * 1000,
                i % 2 == 0,
            );
            let result = processor.process_single_trade(&trade).unwrap();
            assert!(result.is_none(), "should not breach at tiny price changes");
        }
        let bar_after_rest = processor
            .get_incomplete_bar()
            .expect("forming bar after REST");
        assert!(
            bar_after_rest.individual_trade_count >= 20,
            "forming bar has REST trades"
        );

        // Phase 2: Junction fill -- 5 more trades through SAME processor
        for i in 20..25i64 {
            let trade = make_trade(
                i + 1,
                50000.0 + (i as f64) * 0.001,
                1700000000000 + (i as u64) * 1000,
                i % 2 == 0,
            );
            let result = processor.process_single_trade(&trade).unwrap();
            assert!(result.is_none(), "still no breach during junction fill");
        }
        let bar_after_junction = processor
            .get_incomplete_bar()
            .expect("forming bar after junction");
        assert!(
            bar_after_junction.individual_trade_count >= 25,
            "forming bar accumulated junction trades: got {}",
            bar_after_junction.individual_trade_count
        );

        // Phase 3: First WS trade -- processed through SAME processor
        let ws_trade = make_trade(26, 50000.025, 1700000025000, true);
        let result = processor.process_single_trade(&ws_trade).unwrap();
        assert!(result.is_none(), "still no breach");
        let bar_after_ws = processor
            .get_incomplete_bar()
            .expect("forming bar after WS start");
        assert!(
            bar_after_ws.individual_trade_count >= 26,
            "forming bar spans REST+junction+WS: got {}",
            bar_after_ws.individual_trade_count
        );
    }

    #[cfg(feature = "clickhouse-sink")]
    #[test]
    fn test_fan_out_with_counting_sink() {
        // Issue #318: Verify fan-out dispatch calls on_bar() on extra sinks.
        use crate::engine::traits::{BarSink, SinkError};
        use std::sync::atomic::AtomicU64;

        struct CountingSink {
            count: Arc<AtomicU64>,
        }
        impl BarSink for CountingSink {
            fn on_bar(&mut self, _bar: &CompletedBar) -> Result<(), SinkError> {
                self.count.fetch_add(1, Ordering::Relaxed);
                Ok(())
            }
            fn flush(&mut self) -> Result<(), SinkError> {
                Ok(())
            }
            fn name(&self) -> &str {
                "counting"
            }
        }

        let count = Arc::new(AtomicU64::new(0));
        let sink = CountingSink {
            count: Arc::clone(&count),
        };
        let sinks: Vec<Box<dyn BarSink>> = vec![Box::new(sink)];
        let extra_sinks: ExtraSinks =
            Some(Arc::new(std::sync::Mutex::new(sinks)));

        let thresholds = vec![250u32];
        let mut processors: Vec<(u32, OpenDeviationBarProcessor)> = thresholds
            .iter()
            .map(|&t| (t, OpenDeviationBarProcessor::new(t).unwrap()))
            .collect();
        let mut last_days: Vec<i64> = processors.iter().map(|_| -1i64).collect();
        let mut last_trade_timestamps_us: Vec<i64> = processors.iter().map(|_| 0i64).collect();
        let bar_buffer = ConcurrentRingBuffer::new(100);
        let metrics = LiveEngineMetrics::default();
        let symbol_arc: Arc<str> = Arc::from("BTCUSDT");
        let forming_tx_map: HashMap<u32, watch::Sender<Option<FormingBar>>> = HashMap::new();
        let mut committed_floors: HashMap<u32, i64> = HashMap::new();
        let mut last_forming_update: HashMap<u32, i64> = HashMap::new();

        // Feed trades until a bar is emitted (250 dbps = 0.25% threshold)
        // Open at 50000, breach at 50000 * 1.0025 = 50125
        let base_price = 50000.0;
        let trade1 = make_trade(1, base_price, 1700000000000, false);
        process_trade_through_processors(
            &trade1,
            &mut processors,
            &mut last_days,
            &mut last_trade_timestamps_us,
            &bar_buffer,
            &metrics,
            &symbol_arc,
            "BTCUSDT",
            &forming_tx_map,
            &mut committed_floors,
            &mut last_forming_update,
            OuroborosMode::Aion,
            &extra_sinks,
        );
        // No bar emitted yet
        assert_eq!(count.load(Ordering::Relaxed), 0);

        // Breach trade (different timestamp required by timestamp gate)
        let breach_price = base_price * 1.003; // > 0.25% threshold
        let trade2 = make_trade(2, breach_price, 1700000001000, false);
        process_trade_through_processors(
            &trade2,
            &mut processors,
            &mut last_days,
            &mut last_trade_timestamps_us,
            &bar_buffer,
            &metrics,
            &symbol_arc,
            "BTCUSDT",
            &forming_tx_map,
            &mut committed_floors,
            &mut last_forming_update,
            OuroborosMode::Aion,
            &extra_sinks,
        );
        // Bar emitted -- sink should have been called
        assert_eq!(
            count.load(Ordering::Relaxed),
            1,
            "counting sink should have received 1 bar via fan-out"
        );
        // Ring buffer also got the bar
        assert!(bar_buffer.pop().is_some(), "ring buffer should have the bar too");
    }
}