stochastic-routing-extended 1.0.2

//! Channel management: multiplexing, fallback, health monitoring.
//!
//! A [`Channel`] is an active transport connection coupled with a
//! [`HealthMonitor`].  The channel tracks its own health, can be
//! promoted from shadow (backup) to primary, and exposes a single
//! `score()` for routing decisions.

pub mod fallback;
pub mod health;

pub use fallback::FallbackController;
pub use health::HealthMonitor;

use std::time::{Duration, Instant};

use crate::transport::TransportKind;

/// Default block threshold (consecutive failures before marking blocked).
const DEFAULT_BLOCK_THRESHOLD: u32 = 5;

/// Represents an active communication channel over a specific transport.
pub struct Channel {
    /// Which transport this channel uses.
    pub transport: TransportKind,
    /// Whether this is a shadow (backup) channel.
    is_shadow: bool,
    /// Per-channel health monitor.
    monitor: HealthMonitor,
    /// When the channel was opened.
    opened_at: Instant,
    /// Total bytes sent through this channel.
    bytes_sent: u64,
    /// Total bytes received through this channel.
    bytes_recv: u64,
}

impl Channel {
    /// Create a new primary channel.
    pub fn new(transport: TransportKind) -> Self {
        Self {
            transport,
            is_shadow: false,
            monitor: HealthMonitor::new(DEFAULT_BLOCK_THRESHOLD),
            opened_at: Instant::now(),
            bytes_sent: 0,
            bytes_recv: 0,
        }
    }

    /// Create a new shadow (backup) channel.
    pub fn shadow(transport: TransportKind) -> Self {
        Self {
            is_shadow: true,
            ..Self::new(transport)
        }
    }

    /// Create a channel with a custom block threshold.
    pub fn with_block_threshold(transport: TransportKind, threshold: u32) -> Self {
        Self {
            monitor: HealthMonitor::new(threshold),
            ..Self::new(transport)
        }
    }

    // ── Health delegation ───────────────────────────────────────────────

    /// Record a successful send/recv with the measured RTT.
    pub fn record_success(&mut self, rtt: Duration) {
        self.monitor.record_success(rtt);
    }

    /// Record a failure (timeout, connection reset, etc.).
    pub fn record_failure(&mut self) {
        self.monitor.record_failure();
    }

    /// Whether this channel appears to be blocked by DPI.
    pub fn is_blocked(&self) -> bool {
        self.monitor.is_blocked()
    }

    /// Whether this channel is healthy (not blocked).
    pub fn is_healthy(&self) -> bool {
        !self.monitor.is_blocked()
    }

    /// Composite health score 0.0 (dead) – 1.0 (perfect).
    pub fn score(&self) -> f64 {
        self.monitor.score()
    }

    /// Smoothed RTT estimate.
    pub fn srtt(&self) -> Duration {
        self.monitor.srtt
    }

    /// Number of consecutive failures.
    pub fn consecutive_failures(&self) -> u32 {
        self.monitor.consecutive_failures
    }

    /// Direct access to the underlying monitor.
    pub fn monitor(&self) -> &HealthMonitor {
        &self.monitor
    }

    /// Mutable access to the underlying monitor.
    pub fn monitor_mut(&mut self) -> &mut HealthMonitor {
        &mut self.monitor
    }

    // ── Shadow / primary lifecycle ──────────────────────────────────────

    /// Whether this is a shadow (backup) channel.
    pub fn is_shadow(&self) -> bool {
        self.is_shadow
    }

    /// Promote shadow channel to primary.
    pub fn promote(&mut self) {
        self.is_shadow = false;
    }

    /// Demote primary channel to shadow.
    pub fn demote(&mut self) {
        self.is_shadow = true;
    }

    // ── Stats ───────────────────────────────────────────────────────────

    /// Record bytes sent.
    pub fn add_bytes_sent(&mut self, n: u64) {
        self.bytes_sent += n;
    }

    /// Record bytes received.
    pub fn add_bytes_recv(&mut self, n: u64) {
        self.bytes_recv += n;
    }

    /// Total bytes sent.
    pub fn bytes_sent(&self) -> u64 {
        self.bytes_sent
    }

    /// Total bytes received.
    pub fn bytes_recv(&self) -> u64 {
        self.bytes_recv
    }

    /// How long this channel has been open.
    pub fn uptime(&self) -> Duration {
        self.opened_at.elapsed()
    }
}

// ── Tests ───────────────────────────────────────────────────────────────────

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

    #[test]
    fn new_channel_is_healthy() {
        let ch = Channel::new(TransportKind::Tcp);
        assert!(ch.is_healthy());
        assert!(!ch.is_blocked());
        assert!(!ch.is_shadow());
        assert!(ch.score() > 0.0);
    }

    #[test]
    fn shadow_channel_flag() {
        let ch = Channel::shadow(TransportKind::Udp);
        assert!(ch.is_shadow());
        assert!(ch.is_healthy());
    }

    #[test]
    fn record_success_updates_srtt() {
        let mut ch = Channel::new(TransportKind::Tcp);
        let before = ch.srtt();
        ch.record_success(Duration::from_millis(200));
        // SRTT should move toward 200ms.
        assert!(ch.srtt() > before);
    }

    #[test]
    fn blocked_after_threshold() {
        let mut ch = Channel::with_block_threshold(TransportKind::Quic, 3);
        ch.record_failure();
        ch.record_failure();
        assert!(ch.is_healthy());
        ch.record_failure();
        assert!(ch.is_blocked());
        assert_eq!(ch.score(), 0.0);
    }

    #[test]
    fn success_resets_failures() {
        let mut ch = Channel::with_block_threshold(TransportKind::Tcp, 2);
        ch.record_failure();
        ch.record_success(Duration::from_millis(10));
        ch.record_failure();
        assert!(ch.is_healthy());
    }

    #[test]
    fn promote_demote_lifecycle() {
        let mut ch = Channel::shadow(TransportKind::WebSocket);
        assert!(ch.is_shadow());
        ch.promote();
        assert!(!ch.is_shadow());
        ch.demote();
        assert!(ch.is_shadow());
    }

    #[test]
    fn bytes_tracking() {
        let mut ch = Channel::new(TransportKind::Tcp);
        assert_eq!(ch.bytes_sent(), 0);
        assert_eq!(ch.bytes_recv(), 0);
        ch.add_bytes_sent(1024);
        ch.add_bytes_recv(512);
        ch.add_bytes_sent(256);
        assert_eq!(ch.bytes_sent(), 1280);
        assert_eq!(ch.bytes_recv(), 512);
    }

    #[test]
    fn uptime_increases() {
        let ch = Channel::new(TransportKind::Tcp);
        // Uptime should be very small but non-negative.
        assert!(ch.uptime() < Duration::from_secs(1));
    }
}