ek-core 0.1.0

//! worker(插件串行 actor 线程):邮箱消费 + catch_unwind 隔离 + 熔断(D3)。
//!
//! 事件所有权在 worker 循环体内:handler panic unwind 被 catch_unwind 截住后,
//! 事件随本轮循环作用域自然 drop —— 寻址信封的 reply 随之 drop、信道关闭,
//! 发起方立即得到 TargetGone(信令依赖于此,勿改为引用投递)。

use crate::context::PluginContext;
use crate::log_plugin::{EVENT_PLUG_DOWN_PREFIX, LOG_LEVEL_ERROR, LOG_LEVEL_WARN};
use crate::plugin::{PANIC_LIMIT, Plugin};
use crate::registry::{EntryStats, PluginStateCell, PluginStateCode};
use crossbeam_channel::Receiver;
use ek_event::{Event, EventEngine};
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::thread::JoinHandle;

pub(crate) fn spawn_worker(
    plugin: Arc<dyn Plugin>,
    rx: Receiver<Event>,
    state: PluginStateCell,
    stats: Arc<EntryStats>,
    ctx: Arc<PluginContext>,
    bus: Arc<EventEngine>,
) -> JoinHandle<()> {
    let name = plugin.name().to_string();
    std::thread::Builder::new()
        .name(format!("ek-worker.{name}"))
        .spawn(move || worker_loop(plugin, rx, state, stats, ctx, bus))
        .expect("ek-core: spawn worker thread")
}

fn worker_loop(
    plugin: Arc<dyn Plugin>,
    rx: Receiver<Event>,
    state: PluginStateCell,
    stats: Arc<EntryStats>,
    ctx: Arc<PluginContext>,
    bus: Arc<EventEngine>,
) {
    // D3 = 连续 panic×3 熔断(复审 F3 修正):处理成功即复位;
    // stats.panics 为累计统计面,不参与熔断判定。
    let mut consecutive: u64 = 0;
    loop {
        match rx.recv() {
            Ok(event) => {
                let result = catch_unwind(AssertUnwindSafe(|| plugin.on_mail(&event)));
                match result {
                    Ok(()) => consecutive = 0,
                    Err(_panic) => {
                        consecutive += 1;
                        let _ = stats.panics.fetch_add(1, Ordering::AcqRel);
                        if consecutive >= PANIC_LIMIT as u64 {
                            if state.cas(PluginStateCode::Running, PluginStateCode::Quarantined) {
                                ctx.write_log(
                                    LOG_LEVEL_ERROR,
                                    &format!(
                                        "plugin {:?} quarantined after {} consecutive handler panics",
                                        plugin.name(),
                                        consecutive
                                    ),
                                );
                                // 双通道告警:通用日志 + 熔断广播(可被监控插件订阅)
                                let _ = bus.put(Event::new(format!(
                                    "{EVENT_PLUG_DOWN_PREFIX}{}",
                                    plugin.name()
                                )));
                            }
                            // 检疫即退出:邮箱内残留邮件随之丢弃(drop rx → forwarder 断连短路)
                            break;
                        }
                        ctx.write_log(
                            LOG_LEVEL_WARN,
                            &format!(
                                "plugin {:?} handler panicked ({consecutive}/{PANIC_LIMIT})",
                                plugin.name()
                            ),
                        );
                    }
                }
                // event 随本轮作用域 drop → 寻址信封 reply drop → 请求方即时感知
                drop(event);
            }
            // 排空 + 断连:close 已把全部 send 端消费尽
            Err(_disconnected) => {
                state.finish();
                break;
            }
        }
    }
    stats.done.store(true, Ordering::Release);
}