opendeviationbar-streaming 13.78.0

//! ClickHouseWriterSink: BarSink implementation for direct CH writes.
//!
//! Issue #318: Rust-native ClickHouse writer -- sink.
//!
//! Non-blocking `on_bar()` sends rows to the flush thread via bounded
//! `SyncSender`. Channel full returns `SinkError::Recoverable`, channel
//! closed returns `SinkError::Unrecoverable`.

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use crate::engine::traits::{BarSink, SinkError};
use crate::live_engine::CompletedBar;

use super::config::ClickHouseWriterConfig;
use super::flush_thread::{FlushCommand, FlushThreadMetrics, spawn_flush_thread};
use super::row::ClickHouseBarRow;

/// Sink that writes completed bars to ClickHouse via a dedicated flush thread.
///
/// Implements `BarSink` with non-blocking `on_bar()`. Bars are serialized to
/// `ClickHouseBarRow` and sent to the flush thread via a bounded channel.
pub struct ClickHouseWriterSink {
    tx: std::sync::mpsc::SyncSender<FlushCommand>,
    flush_thread: Option<std::thread::JoinHandle<()>>,
    metrics: Arc<FlushThreadMetrics>,
    bars_sent: AtomicU64,
    bars_dropped: AtomicU64,
    /// #363 R0.1: Bars refused because `first_agg_trade_id == 0` or
    /// `last_agg_trade_id == 0`. A live-engine bar with TID=0 is a
    /// cloaked ghost flush from a frozen forming-bar state and must
    /// not reach the ReplacingMergeTree dedup key — otherwise every
    /// ghost collapses under `(symbol, threshold, ouroboros, 0)`
    /// and cloaks the dark-symbol outage from every monitor.
    /// Parquet prefill writes via a separate Python path
    /// (`ch_cache.store_bars_batch`) and is unaffected.
    bars_rejected_ghost: AtomicU64,
    /// #641 P0.2: Bars rejected due to mega-span (TID gap > 1M trades).
    /// Indicates stale threshold floor or Puck gap-fill bypass, requiring
    /// investigation of committed_floors initialization and checkpoint restore.
    bars_rejected_megaspan: AtomicU64,
}

impl ClickHouseWriterSink {
    /// Create a new sink, spawning the flush thread immediately.
    pub fn new(config: ClickHouseWriterConfig) -> Self {
        let (tx, rx) = std::sync::mpsc::sync_channel(config.channel_capacity);
        let metrics = Arc::new(FlushThreadMetrics::default());
        let flush_thread = spawn_flush_thread(rx, config, Arc::clone(&metrics));
        Self {
            tx,
            flush_thread: Some(flush_thread),
            metrics,
            bars_sent: AtomicU64::new(0),
            bars_dropped: AtomicU64::new(0),
            bars_rejected_ghost: AtomicU64::new(0),
            bars_rejected_megaspan: AtomicU64::new(0),
        }
    }

    /// Access flush thread metrics (atomic counters).
    pub fn metrics(&self) -> &FlushThreadMetrics {
        &self.metrics
    }

    /// Get a shared reference to the flush thread metrics Arc (Issue #318).
    ///
    /// Used by LiveBarEngine to extract the Arc before the sink is boxed
    /// into `dyn BarSink`, enabling metrics access from Python bindings.
    pub fn shared_metrics(&self) -> Arc<FlushThreadMetrics> {
        Arc::clone(&self.metrics)
    }

    /// Total bars successfully sent to the channel.
    pub fn bars_sent(&self) -> u64 {
        self.bars_sent.load(Ordering::Relaxed)
    }

    /// Total bars dropped due to backpressure.
    pub fn bars_dropped(&self) -> u64 {
        self.bars_dropped.load(Ordering::Relaxed)
    }

    /// #363 R0.1: Total bars refused because `first_agg_trade_id == 0`
    /// or `last_agg_trade_id == 0`. Distinct from `bars_dropped` (which
    /// is backpressure-only) so dashboards do not confuse cloaked-ghost
    /// rejects with channel saturation events.
    pub fn bars_rejected_ghost(&self) -> u64 {
        self.bars_rejected_ghost.load(Ordering::Relaxed)
    }

    /// #641 P0.2: Total bars rejected due to mega-span (TID gap > 1M trades).
    pub fn bars_rejected_megaspan(&self) -> u64 {
        self.bars_rejected_megaspan.load(Ordering::Relaxed)
    }
}

impl BarSink for ClickHouseWriterSink {
    fn on_bar(&mut self, bar: &CompletedBar) -> Result<(), SinkError> {
        // #363 R0.1: Refuse ghost forming-bar flushes. A live-engine bar
        // with `first_agg_trade_id == 0` or `last_agg_trade_id == 0` is
        // always a cloaked-ghost write: the live processor sets both
        // TIDs from `trade.ref_id` on `OpenDeviationBar::new()` and
        // legitimate Binance aggTrade IDs are > 0. Allowing these rows
        // into ReplacingMergeTree collapses them under the single key
        // `(symbol, threshold, ouroboros, 0)`, cloaking dark-symbol
        // outages from Stathera, `data_freshness`, and every downstream
        // consumer. The Parquet-prefill path writes via Python
        // (`ch_cache.store_bars_batch`) and is unaffected by this guard.
        if bar.bar.first_agg_trade_id == 0 || bar.bar.last_agg_trade_id == 0 {
            self.bars_rejected_ghost.fetch_add(1, Ordering::Relaxed);
            // #363 forensic signal: log as ERROR with the full bar shape
            // so the next occurrence is self-describing in the journal.
            // The accompanying `volume == 0` + valid `open` price is the
            // fingerprint of the live data-corruption bug observed on
            // bigblack at 2026-04-12 19:xx UTC (see incident #363).
            tracing::error!(
                symbol = %bar.symbol,
                threshold_decimal_bps = bar.threshold_decimal_bps,
                first_agg_trade_id = bar.bar.first_agg_trade_id,
                last_agg_trade_id = bar.bar.last_agg_trade_id,
                open_time_us = bar.bar.open_time,
                close_time_us = bar.bar.close_time,
                open = bar.bar.open.to_f64(),
                high = bar.bar.high.to_f64(),
                low = bar.bar.low.to_f64(),
                close = bar.bar.close.to_f64(),
                volume_raw = bar.bar.volume,
                individual_trade_count = bar.bar.individual_trade_count,
                agg_record_count = bar.bar.agg_record_count,
                "#363 ghost bar rejected at CH sink boundary (first_agg_trade_id or last_agg_trade_id == 0 — live engine must never emit TID=0)"
            );
            // Return Ok so the live engine does not mark the pair unhealthy
            // and does not dead-letter the row. The reject is intentional.
            return Ok(());
        }

        // #641 P0.2: Reject mega-span bars (TID gap > 1M trades).
        // Indicates stale threshold floor or Puck gap-fill bypass.
        // Likely root cause: checkpoint restore before seed override,
        // or per-threshold committed_floors not initialized from CH.
        const MAX_BAR_SPAN_TRADES: i64 = 1_000_000;
        let span = bar.bar.last_agg_trade_id - bar.bar.first_agg_trade_id;
        if span > MAX_BAR_SPAN_TRADES {
            self.bars_rejected_megaspan.fetch_add(1, Ordering::Relaxed);
            tracing::error!(
                symbol = %bar.symbol,
                threshold_decimal_bps = bar.threshold_decimal_bps,
                first_agg_trade_id = bar.bar.first_agg_trade_id,
                last_agg_trade_id = bar.bar.last_agg_trade_id,
                span = span,
                open_time_us = bar.bar.open_time,
                close_time_us = bar.bar.close_time,
                "#641 mega-span bar rejected at CH sink (span > 1M trades — stale floor suspected, checkpoint restore before seed order?)"
            );
            // Dead-letter for forensics
            if let Err(dl_err) = super::dead_letter::write_dead_letter_single(
                ClickHouseBarRow::from_completed_bar(bar)
            ) {
                tracing::error!(error = %dl_err, "dead-letter write failed for mega-span bar");
            }
            // Return Ok so the live engine does not mark the pair unhealthy
            return Ok(());
        }

        let row = ClickHouseBarRow::from_completed_bar(bar);
        match self.tx.try_send(FlushCommand::Bar(Box::new(row))) {
            Ok(()) => {
                self.bars_sent.fetch_add(1, Ordering::Relaxed);
                Ok(())
            }
            Err(std::sync::mpsc::TrySendError::Full(cmd)) => {
                self.bars_dropped.fetch_add(1, Ordering::Relaxed);
                // Extract the row from the command and dead-letter it to Parquet
                if let FlushCommand::Bar(boxed_row) = cmd {
                    match super::dead_letter::write_dead_letter_single(*boxed_row) {
                        Ok(path) => {
                            tracing::warn!(
                                path = %path.display(),
                                "backpressure - bar dead-lettered to Parquet"
                            );
                        }
                        Err(dl_err) => {
                            tracing::error!(
                                error = %dl_err,
                                "CRITICAL: backpressure AND dead-letter write failed -- bar lost"
                            );
                        }
                    }
                } else {
                    tracing::warn!(
                        "clickhouse flush thread backpressure - non-bar command dropped"
                    );
                }
                Err(SinkError::Recoverable(
                    "clickhouse flush thread backpressure".into(),
                ))
            }
            Err(std::sync::mpsc::TrySendError::Disconnected(_)) => Err(SinkError::Unrecoverable(
                "clickhouse flush thread died".into(),
            )),
        }
    }

    fn flush(&mut self) -> Result<(), SinkError> {
        self.tx
            .send(FlushCommand::Flush)
            .map_err(|_| SinkError::Unrecoverable("clickhouse flush thread died".into()))
    }

    fn name(&self) -> &str {
        "clickhouse"
    }
}

impl Drop for ClickHouseWriterSink {
    fn drop(&mut self) {
        let _ = self.tx.send(FlushCommand::Shutdown);
        if let Some(handle) = self.flush_thread.take() {
            let _ = handle.join();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use opendeviationbar_core::{FixedPoint, OpenDeviationBar};
    use std::sync::Arc;

    fn make_bar() -> CompletedBar {
        make_bar_with_ref_id(1)
    }

    fn make_bar_with_ref_id(ref_id: i64) -> CompletedBar {
        let trade = opendeviationbar_core::Tick {
            ref_id,
            price: FixedPoint::from_str("50000.0").unwrap(),
            volume: FixedPoint::from_str("1.0").unwrap(),
            first_sub_id: 1,
            last_sub_id: 1,
            timestamp: opendeviationbar_core::normalize_timestamp(1_700_000_000_000),
            is_buyer_maker: false,
            is_best_match: None,
            best_bid: None,
            best_ask: None,
        };
        CompletedBar {
            symbol: Arc::from("BTCUSDT"),
            threshold_decimal_bps: 250,
            bar: OpenDeviationBar::new(&trade),
        }
    }

    #[test]
    fn test_sink_name() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            channel_capacity: 10,
            max_retries: 0,
            flush_period_ms: 60_000,
            ..Default::default()
        };
        let mut sink = ClickHouseWriterSink::new(config);
        assert_eq!(sink.name(), "clickhouse");
        // Explicit flush to satisfy Drop
        let _ = sink.flush();
    }

    #[test]
    fn test_sink_on_bar_sends_to_channel() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            channel_capacity: 100,
            max_retries: 0,
            flush_period_ms: 60_000,
            ..Default::default()
        };
        let mut sink = ClickHouseWriterSink::new(config);

        let bar = make_bar();
        assert!(sink.on_bar(&bar).is_ok());
        assert_eq!(sink.bars_sent(), 1);
        assert_eq!(sink.bars_dropped(), 0);
    }

    #[test]
    fn test_sink_backpressure() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            channel_capacity: 1, // tiny channel
            max_retries: 0,
            flush_period_ms: 60_000,
            max_rows: 10_000, // prevent auto-flush
            ..Default::default()
        };
        let mut sink = ClickHouseWriterSink::new(config);

        let bar = make_bar();
        // First send fills the channel
        assert!(sink.on_bar(&bar).is_ok());
        // Second send should get backpressure (Recoverable)
        let result = sink.on_bar(&bar);
        assert!(matches!(result, Err(SinkError::Recoverable(_))));
        assert_eq!(sink.bars_dropped(), 1);
    }

    #[test]
    fn test_sink_graceful_shutdown() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            channel_capacity: 100,
            max_retries: 0,
            flush_period_ms: 60_000,
            ..Default::default()
        };
        let mut sink = ClickHouseWriterSink::new(config);

        let bar = make_bar();
        sink.on_bar(&bar).unwrap();
        sink.on_bar(&bar).unwrap();
        assert_eq!(sink.bars_sent(), 2);

        // Drop triggers Shutdown + join
        drop(sink);
        // If we reach here, the flush thread joined cleanly
    }

    /// #363 R0.1: ghost bars (TID=0) must be rejected before reaching
    /// the CH dedup key. The reject is silent (returns Ok) but is
    /// counted via `bars_rejected_ghost` and must NOT advance
    /// `bars_sent`.
    #[test]
    fn test_sink_rejects_ghost_bar_tid_zero() {
        let config = ClickHouseWriterConfig {
            url: "http://127.0.0.1:1".to_string(),
            channel_capacity: 100,
            max_retries: 0,
            flush_period_ms: 60_000,
            ..Default::default()
        };
        let mut sink = ClickHouseWriterSink::new(config);

        let ghost = make_bar_with_ref_id(0);
        assert_eq!(ghost.bar.first_agg_trade_id, 0);
        assert_eq!(ghost.bar.last_agg_trade_id, 0);

        // Reject: returns Ok (silent), increments ghost counter,
        // does NOT increment bars_sent.
        assert!(sink.on_bar(&ghost).is_ok());
        assert_eq!(sink.bars_rejected_ghost(), 1);
        assert_eq!(sink.bars_sent(), 0);
        assert_eq!(sink.bars_dropped(), 0);

        // A legitimate bar still passes.
        let real = make_bar_with_ref_id(42);
        assert!(sink.on_bar(&real).is_ok());
        assert_eq!(sink.bars_rejected_ghost(), 1);
        assert_eq!(sink.bars_sent(), 1);

        // A second ghost is counted and rejected independently.
        let ghost2 = make_bar_with_ref_id(0);
        assert!(sink.on_bar(&ghost2).is_ok());
        assert_eq!(sink.bars_rejected_ghost(), 2);
        assert_eq!(sink.bars_sent(), 1);

        let _ = sink.flush();
    }
}