ek-core 0.5.0

//! forwarder(转发器):宿主为插件每条 route 注册的 handler 闭包,
//! 只做「总线事件 → 插件邮箱 send」,零业务、零阻塞。
//!
//! 0.5.0 口径反转(D5 改案):邮箱**无界** → 默认**不丢**;深度实时计数
//! (`pending`/`peak`)常年在统计面可见。发送前先查深度:**到软顶才丢**
//! (丢新帧 + `spilled` 计数 + 节流 stderr),这是"不丢"的唯一出口。
//! 分发线程永不阻塞这条不变(无界 send 不阻塞;软顶检查是一次原子读)。
//! 断连(worker 自退/邮箱关闭):alive 置 false 后短路,不再白调 send,
//! 正式注销留给 close 第①步(不在 handler 内 unregister,避免逆序复杂度)。

use crate::registry::EntryStats;
use ek_event::{Event, HandlerFn};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

use crossbeam_channel::Sender;

/// 溢出告警节流:累计溢出每越过这么多条,报一条(stderr;不进总线)。
pub const SPILL_REPORT_EVERY: u64 = 128;

pub(crate) struct Forwarder {
    tx: Sender<Event>,
    alive: AtomicBool,
    stats: Arc<EntryStats>,
    reported: AtomicU64,
    /// 该插件的软顶(发送前的裁决线)。
    soft_cap: u64,
    target: String,
}

impl Forwarder {
    pub(crate) fn new(
        tx: Sender<Event>,
        stats: Arc<EntryStats>,
        soft_cap: u64,
        target: String,
    ) -> Arc<Self> {
        Arc::new(Self {
            tx,
            alive: AtomicBool::new(true),
            stats,
            reported: AtomicU64::new(0),
            soft_cap,
            target,
        })
    }

    /// 分发线程上执行;永不阻塞。
    pub fn forward(&self, event: &Event) {
        if !self.alive.load(Ordering::Acquire) {
            return;
        }
        // 软顶裁决:深度到顶才丢(默认软顶极高 → 常态这条永不为真)。丢的是新到的那一条。
        if self.stats.pending.load(Ordering::Acquire) >= self.soft_cap {
            self.on_spill();
            return;
        }
        // ek-event 0.1.0 的 Event 不实现 Clone:手工复制(data 走 Arc 引用计数,载荷零拷贝)
        let event2 = Event::with_data(event.event_type.clone(), Arc::clone(&event.data));
        // 深度计数在**发送之前**占位(reserve-then-send):发送后自增会与 worker 的
        // 自减打架 —— worker 收到就减,减在加前 → 无符号下绕成天文数字
        // (debug 的溢出检查当场把它抓出来:0.5.0 首跑现场就是这条)。
        let depth = self
            .stats
            .pending
            .fetch_add(1, Ordering::AcqRel)
            .saturating_add(1);
        self.stats.peak.fetch_max(depth, Ordering::AcqRel);
        match self.tx.try_send(event2) {
            Ok(()) => {}
            // 无界邮箱不会"满":到这里只可能是 worker 自退(rx drop)或邮箱关闭(tx 消费尽)。
            // 占位要**原路退回**:这一帧没进队,深度计数不能留下它。
            Err(_) => {
                self.alive.store(false, Ordering::Release);
                let _ = self
                    .stats
                    .pending
                    .fetch_update(Ordering::AcqRel, Ordering::Acquire, |v| {
                        Some(v.saturating_sub(1))
                    });
            }
        }
    }

    fn on_spill(&self) {
        let total = self.stats.spilled.fetch_add(1, Ordering::AcqRel) + 1;
        let last = self.reported.load(Ordering::Acquire);
        if total >= last + SPILL_REPORT_EVERY
            && self
                .reported
                .compare_exchange(last, total, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
        {
            // 0.3.0:溢出诊断走 stderr(节流);结构化事实是 stats.spilled,查 mail_stats()
            // 0.5.0:措辞跟着语义走 —— 到软顶才丢,不是邮箱满即丢
            eprintln!(
                "ek-core: mailbox soft-cap spill: plugin {:?} dropped {} events in total (depth ≥ soft cap;pending={})",
                self.target,
                total,
                self.stats.pending.load(Ordering::Acquire)
            );
        }
    }
}

/// 把 Forwarder 包成 ek-event 的 handler。
pub(crate) fn make_forwarder(fwd: Arc<Forwarder>) -> HandlerFn {
    Arc::new(move |event: &Event| fwd.forward(event))
}