ek_core/

main_engine.rs

//! `MainEngine` —— 纯插件宿主:单表注册、(typed)获取、(deps)拓扑 start、
//! 逆序五步 close、单插件 remove(允许+警告)、运行统计查询。零业务方法。
//!
//! close 五步(每插件,逆依赖序依次执行):
//! ①卸 forwarder(unregister 全部 HandlerId)
//! ②on_stop 钩子(业务清理)
//! ③关邮箱(宿主侧 Sender 置 None;全部 send 端消费尽 → 信道断开)
//! ④worker 排空(crossbeam recv 先取尽缓冲再报 Disconnected)
//! ⑤join 带超时:超时则告警 + 剥离句柄(线程自然排空收尾,Rust 不杀线程)

use crate::context::PluginContext;
use crate::error::PluginError;
use crate::forwarder::{Forwarder, make_forwarder};
use crate::lifecycle::{
    EVENT_PLUG_DOWN, EVENT_PLUG_DOWN_EVENT_PREFIX, EVENT_PLUG_UP, EVENT_PLUG_UP_PREFIX,
};
use crate::plugin::{CLOSE_JOIN_TIMEOUT, Plugin, valid_plugin_name};
use crate::registry::{EntryStats, PluginEntry, PluginStateCell, PluginStateCode, Registry};
use crate::service::PlugNotice;
use crate::worker::spawn_worker;
use ek_event::{EventEngine, HandlerId};
use std::any::Any;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant};

/// 插件生命周期状态(worker 状态机,AtomicU8 CAS 迁移)。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PluginState {
    Ready,
    Running,
    Stopping,
    Exited,
    Quarantined,
}

impl PluginState {
    pub(crate) fn from_code(c: u8) -> Self {
        match c {
            0 => Self::Ready,
            1 => Self::Running,
            2 => Self::Stopping,
            3 => Self::Exited,
            _ => Self::Quarantined,
        }
    }
    fn from_code_enum(code: PluginStateCode) -> Self {
        Self::from_code(code as u8)
    }
}

/// 插件信息快照(get_plugin / ctx.plugins())。
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PluginInfo {
    pub name: String,
    /// 插件申报版本(空串 = 未声明);A 侧版本策略据此选轨(F5:单一披露面)。
    pub version: String,
    pub state: PluginState,
    pub deps: Vec<String>,
}

/// 邮箱运行统计(mail_stats)。
///
/// 0.5.0 加两深度读数:`pending`(当前积压)与 `peak`(历史峰值)。
/// 口径 —— 不丢是默认,堵不堵要**看得见**:`spilled` 只在到软顶时才涨。
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct MailStats {
    /// 软顶触发的丢弃累计(默认无界不丢 → 常态应为 0)。
    pub spilled: u64,
    /// handler panic 累计。
    pub panics: u64,
    /// 当前邮箱深度(发送 +1 / 工人取走 −1)。
    pub pending: u64,
    /// 历史峰值(只涨不跌)。
    pub peak: u64,
    /// 该插件的软顶(消费侧判"溢满"要吃它自己的线,不能拿缺省去比)。
    pub soft_cap: u64,
}

pub struct MainEngine {
    event: Arc<EventEngine>,
    registry: Arc<Registry>,
    /// start() 装配顺序纪录(close 反向使用);None = 未 start。
    started_order: Mutex<Option<Vec<String>>>,
    closed: RwLock<bool>,
}

impl MainEngine {
    /// 注入式创建:`Some(bus)` 复用外部引擎;`None` 自建(start() 时拉起)。
    pub fn new(event: Option<Arc<EventEngine>>) -> Self {
        Self {
            event: event.unwrap_or_else(|| Arc::new(EventEngine::new())),
            registry: Arc::new(Registry::new()),
            started_order: Mutex::new(None),
            closed: RwLock::new(false),
        }
    }

    /// 宿主级引擎句柄(测试/装配工具可用;插件不可见——ctx 不暴露)。
    pub fn event_engine(&self) -> &Arc<EventEngine> {
        &self.event
    }

    fn closed_check(&self) -> Result<(), PluginError> {
        if *self.closed.read().unwrap() {
            Err(PluginError::Stopped)
        } else {
            Ok(())
        }
    }

    /// 统一格式注册(D2:名/重名/同名自依赖;add 期【不校验 deps 存在性】——
    /// 新 D1 注册自由,悬空 deps 由 start 期把关)。
    ///
    /// **运行期动态 add(0.2.0)**:start 之后仍可随时注入新插件——登记 → 织入
    /// 能力索引 → 立即走 launch_one 装配上线 → 记入 started_order(close 反序覆盖);
    /// 并广播 `ePlugUp` / `ePlugUp.<名>`,迟绑定依赖方据此自主发现。
    pub fn add_plugin(&self, plugin: Arc<impl Plugin + 'static>) -> Result<(), PluginError> {
        self.closed_check()?;
        let name = plugin.name().to_string();
        if !valid_plugin_name(&name) {
            return Err(PluginError::IllegalName(name));
        }
        let deps: Vec<String> = plugin.deps().iter().map(|d| d.to_string()).collect();
        if deps.iter().any(|d| d == &name) {
            return Err(PluginError::DepCycle(vec![name]));
        }
        let soft_cap = plugin.mailbox_soft_cap();
        let any: Arc<dyn Any + Send + Sync> = Arc::clone(&plugin) as Arc<dyn Any + Send + Sync>;
        let dyn_plugin: Arc<dyn Plugin> = plugin;
        // F8 纪律:services() 为纯声明;在拿锁前读一次抽 id
        let svc_ids: Vec<String> = {
            let mut ids = Vec::new();
            for d in dyn_plugin.services() {
                if !ids.contains(&d.id) {
                    ids.push(d.id);
                }
            }
            ids
        };

        let mut w = self.registry.state_of.write().unwrap();
        if w.entries.contains_key(&name) {
            return Err(PluginError::Duplicate(name));
        }
        let entry = Arc::new(PluginEntry {
            name: name.clone(),
            plugin: dyn_plugin,
            any,
            state: PluginStateCell::new(PluginStateCode::Ready),
            soft_cap,
            deps,
            mailbox: Mutex::new(None),
            forwarder_ids: Mutex::new(Vec::new()),
            worker: Mutex::new(None),
            stats: Arc::new(EntryStats::default()),
        });
        w.order.push(entry.name.clone());
        w.entries.insert(entry.name.clone(), entry);
        drop(w);
        self.registry.index_put(&name, &svc_ids);
        self.registry.invalidate(&name);

        // 动态 add:start 已发生 → 立即装配上线(与 start 共用 launch_one)
        let mut started = self.started_order.lock().unwrap();
        if started.is_some() {
            let entry = self.registry.get(&name).expect("just inserted");
            self.launch_one(&entry);
            started.as_mut().expect("checked Some").push(name);
        }
        Ok(())
    }

    /// 状态查询。
    pub fn get_state(&self, name: &str) -> Option<PluginState> {
        self.registry.get(name).map(|e| e.state.get())
    }

    /// 信息快照。
    pub fn get_plugin(&self, name: &str) -> Option<PluginInfo> {
        self.registry.info_of(name)
    }

    /// 类型安全获取(get_typed):同一分配的 Arc<dyn Any> downcast。
    pub fn get_typed<T: Plugin + 'static>(&self, name: &str) -> Option<Arc<T>> {
        let entry = self.registry.get(name)?;
        Arc::downcast::<T>(Arc::clone(&entry.any)).ok()
    }

    /// 注册名列表(注册序)。
    pub fn plugin_names(&self) -> Vec<String> {
        self.registry.names()
    }

    /// 运行统计(软顶丢弃/panic/当前深度/峰值/软顶)。未注册 → None。
    pub fn mail_stats(&self, name: &str) -> Option<MailStats> {
        self.registry.get(name).map(|e| MailStats {
            spilled: e.stats.spilled.load(Ordering::Acquire),
            panics: e.stats.panics.load(Ordering::Acquire),
            pending: e.stats.pending.load(Ordering::Acquire),
            peak: e.stats.peak.load(Ordering::Acquire),
            soft_cap: e.soft_cap,
        })
    }

    /// 启动全部插件:总线 start → deps 拓扑序逐个装配
    /// (on_init → 建邮箱 → spawn worker → state=Running → 注册 forwarder → on_start
    /// → 广播 ePlugUp)。
    /// 依赖环/悬空 deps → DepCycle/UnknownDep;重复 start 幂等 Ok;已 close → Stopped。
    pub fn start(&self) -> Result<(), PluginError> {
        let mut started = self.started_order.lock().unwrap();
        if started.is_some() {
            return Ok(());
        }
        self.closed_check()?;
        self.event.start();
        let order = self.registry.topo_order()?;
        let mut launched = Vec::with_capacity(order.len());
        for entry in order {
            self.launch_one(&entry);
            launched.push(entry.name.clone());
        }
        *started = Some(launched);
        Ok(())
    }

    /// 单插件装配(start 与动态 add 共用的最小闭包):
    /// on_init → 建邮箱 → spawn worker → Running → 注册 forwarder → on_start
    /// → 广播 `ePlugUp.<名>`(后缀版,载荷=版本串)+ `ePlugUp`(固定版,
    /// 载荷=PlugNotice,供不认识插件名的巡检者订阅)。
    fn launch_one(&self, entry: &Arc<PluginEntry>) {
        let ctx = Arc::new(PluginContext::new(
            Arc::clone(&self.event),
            Arc::clone(&self.registry),
            entry.name.clone(),
        ));
        entry.plugin.on_init(&ctx);
        // 0.5.0:邮箱**无界**(D5 改案:默认不丢),软顶在 forwarder 发送前裁决。
        // close 第③步语义不变:卸完 forwarder + 宿主侧 Sender 置 None → 全部 send 端消费尽
        // → rx Disconnected → worker 自退(与 bounded 时同一条路径)。
        let (tx, rx) = crossbeam_channel::unbounded::<ek_event::Event>();
        *entry.mailbox.lock().unwrap() = Some(tx.clone());
        let handler = spawn_worker(
            Arc::clone(&entry.plugin),
            rx,
            entry.state.clone(),
            Arc::clone(&entry.stats),
            Arc::clone(&self.event),
        );
        *entry.worker.lock().unwrap() = Some(handler);
        // 先 Running 再装 forwarder:避免 ask 预检在装配窗口把目标误判 Ready
        let _ = entry
            .state
            .cas(PluginStateCode::Ready, PluginStateCode::Running);
        let mut ids: Vec<HandlerId> = Vec::new();
        for ty in entry.plugin.routes() {
            let fwd = Forwarder::new(
                tx.clone(),
                Arc::clone(&entry.stats),
                entry.soft_cap,
                entry.name.clone(),
            );
            let id = self.event.register(ty.into_owned(), make_forwarder(fwd));
            ids.push(id);
        }
        *entry.forwarder_ids.lock().unwrap() = ids;
        entry.plugin.on_start();

        let version = entry.plugin.version().into_owned();
        let _ = self.event.put(ek_event::Event::with_data(
            format!("{EVENT_PLUG_UP_PREFIX}{}", entry.name),
            Arc::new(version.clone()),
        ));
        let _ = self.event.put(ek_event::Event::with_data(
            EVENT_PLUG_UP,
            Arc::new(PlugNotice {
                name: entry.name.clone(),
                version,
            }),
        ));
    }

    /// D4 五步闭停单个插件(不摘表)。
    ///
    /// on_stop 语义(执行期复审 F1 修正):凡曾装配过的插件(Running/Stopping/Exited/
    /// Quarantined)close 时都调 on_stop —— 熔断或自然退出的插件也需要业务清理;
    /// 仅 Ready(从未 start)不调。
    fn close_one(&self, entry: &Arc<PluginEntry>, deadline: Instant) {
        let pre_state = entry.state.get();
        let call_stop = pre_state != PluginState::Ready;
        // 挂起迁移:Running → Stopping(容错其他状态)
        let _ = entry
            .state
            .cas(PluginStateCode::Running, PluginStateCode::Stopping);
        // ① 卸 forwarder
        let ids: Vec<HandlerId> = std::mem::take(&mut *entry.forwarder_ids.lock().unwrap());
        for id in ids {
            self.event.unregister(id);
        }
        // ② on_stop
        if call_stop {
            entry.plugin.on_stop();
        }
        // ③ 关邮箱(宿主侧 Sender drop → 全部 send 端消费尽)
        *entry.mailbox.lock().unwrap() = None;
        // ④⑤ 排空等待 + join(带超时,超时告警 + 剥离)
        while !entry.stats.done.load(Ordering::Acquire) && Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(2));
        }
        if let Some(handle) = entry.worker.lock().unwrap().take() {
            if entry.stats.done.load(Ordering::Acquire) {
                let _ = handle.join();
            } else {
                // 运维诊断走 stderr(0.3.0:不进总线,不占类型)
                eprintln!(
                    "ek-core: close timeout: worker of {:?} still busy; detach and keep draining",
                    entry.name
                );
            }
        }
        // 兜底:Stopping → Exited(worker 已退出但未来得及 CAS 的场合)
        let _ = entry
            .state
            .cas(PluginStateCode::Stopping, PluginStateCode::Exited);
    }

    /// 全局闭停:逆 start 序逐插件五步;然后清插件表。总线留给调用方(close_all 停总线)。
    pub fn close(&self, timeout: Duration) {
        let order = self.started_order.lock().unwrap().take();
        *self.closed.write().unwrap() = true;
        let deadline = Instant::now() + timeout.max(Duration::from_millis(1));
        if let Some(order) = order {
            for name in order.iter().rev() {
                if let Some(entry) = self.registry.get(name) {
                    self.close_one(&entry, deadline);
                }
            }
        }
        self.registry.clear();
    }

    /// close + 停总线(终态)。
    pub fn close_all(&self, timeout: Duration) {
        self.close(timeout);
        self.event.stop();
    }

    /// 移除单个插件:Running 者执行五步(UNKNOWN 表项报错);仍被其他插件 deps 引用
    /// 时 stderr 警告(定稿 policy:允许移除 + 警告;依赖方后续 ask 走预检/断连失败)。
    ///
    /// 0.2.0:同步修剪能力索引(F3:防幽灵提供方)+ 作废缓存,并发
    /// `ePlugDown`(固定版,PlugNotice)+ `ePlugDown.<名>`(后缀版,版本串)。
    /// 注:close()(宿主整体下架)不发逐插件 ePlugDown(审核 F7:避免拆除期事件风暴)。
    pub fn remove_plugin(&self, name: &str) -> Result<(), PluginError> {
        self.closed_check()?;
        let entry = self
            .registry
            .get(name)
            .ok_or_else(|| PluginError::UnknownName(name.to_string()))?;
        let dependents = self.registry.any_depends_on(name);
        if !dependents.is_empty() {
            eprintln!(
                "ek-core: remove_plugin {:?} still depended on by {:?}",
                name, dependents
            );
        }
        let deadline = Instant::now() + CLOSE_JOIN_TIMEOUT;
        if entry.state.get() == PluginState::from_code_enum(PluginStateCode::Running) {
            self.close_one(&entry, deadline);
        } else {
            // Ready / Exited / Quarantined:摘 forwarder(可能残留于熔断前)+ 关邮箱
            let ids: Vec<HandlerId> = std::mem::take(&mut *entry.forwarder_ids.lock().unwrap());
            for id in ids {
                self.event.unregister(id);
            }
            *entry.mailbox.lock().unwrap() = None;
        }
        let version = entry.plugin.version().into_owned();
        self.registry
            .remove(name)
            .ok_or_else(|| PluginError::UnknownName(name.to_string()))?;
        // 摘表成功才动索引(修剪幽灵;F3)+ 失效缓存 + 移除广播
        self.registry.index_remove(name);
        self.registry.invalidate(name);
        let _ = self.event.put(ek_event::Event::with_data(
            format!("{EVENT_PLUG_DOWN_EVENT_PREFIX}{name}"),
            Arc::new(version.clone()),
        ));
        let _ = self.event.put(ek_event::Event::with_data(
            EVENT_PLUG_DOWN,
            Arc::new(PlugNotice {
                name: name.to_string(),
                version,
            }),
        ));
        Ok(())
    }
}