ek_core/
worker.rs

1//! worker(插件串行 actor 线程):邮箱消费 + catch_unwind 隔离 + 熔断(D3)。
2//!
3//! 事件所有权在 worker 循环体内:handler panic unwind 被 catch_unwind 截住后,
4//! 事件随本轮循环作用域自然 drop —— 寻址信封的 reply 随之 drop、信道关闭,
5//! 发起方立即得到 TargetGone(信令依赖于此,勿改为引用投递)。
6
7use crate::lifecycle::EVENT_PLUG_DOWN_PREFIX;
8use crate::plugin::{PANIC_LIMIT, Plugin};
9use crate::registry::{EntryStats, PluginStateCell, PluginStateCode};
10use crossbeam_channel::Receiver;
11use ek_event::{Event, EventEngine};
12use std::panic::{AssertUnwindSafe, catch_unwind};
13use std::sync::Arc;
14use std::sync::atomic::Ordering;
15use std::thread::JoinHandle;
16
17pub(crate) fn spawn_worker(
18    plugin: Arc<dyn Plugin>,
19    rx: Receiver<Event>,
20    state: PluginStateCell,
21    stats: Arc<EntryStats>,
22    bus: Arc<EventEngine>,
23) -> JoinHandle<()> {
24    let name = plugin.name().to_string();
25    std::thread::Builder::new()
26        .name(format!("ek-worker.{name}"))
27        .spawn(move || worker_loop(plugin, rx, state, stats, bus))
28        .expect("ek-core: spawn worker thread")
29}
30
31fn worker_loop(
32    plugin: Arc<dyn Plugin>,
33    rx: Receiver<Event>,
34    state: PluginStateCell,
35    stats: Arc<EntryStats>,
36    bus: Arc<EventEngine>,
37) {
38    // D3 = 连续 panic×3 熔断(复审 F3 修正):处理成功即复位;
39    // stats.panics 为累计统计面,不参与熔断判定。
40    let mut consecutive: u64 = 0;
41    loop {
42        match rx.recv() {
43            Ok(event) => {
44                let result = catch_unwind(AssertUnwindSafe(|| plugin.on_mail(&event)));
45                match result {
46                    Ok(()) => consecutive = 0,
47                    Err(_panic) => {
48                        consecutive += 1;
49                        let _ = stats.panics.fetch_add(1, Ordering::AcqRel);
50                        if consecutive >= PANIC_LIMIT as u64 {
51                            if state.cas(PluginStateCode::Running, PluginStateCode::Quarantined) {
52                                // 运维诊断走 stderr(不进总线,不占类型);结构化通知走 ePlugQDown.<名>
53                                eprintln!(
54                                    "ek-core: plugin {:?} quarantined after {} consecutive handler panics",
55                                    plugin.name(),
56                                    consecutive
57                                );
58                                // 熔断广播(可被监控插件订阅;Q=quarantine)
59                                let _ = bus.put(Event::new(format!(
60                                    "{EVENT_PLUG_DOWN_PREFIX}{}",
61                                    plugin.name()
62                                )));
63                            }
64                            // 检疫即退出:邮箱内残留邮件随之丢弃(drop rx → forwarder 断连短路)
65                            break;
66                        }
67                        // 单次 panic:诊断 stderr;不熔断、不广播
68                        eprintln!(
69                            "ek-core: plugin {:?} handler panicked ({consecutive}/{PANIC_LIMIT})",
70                            plugin.name()
71                        );
72                    }
73                }
74                // event 随本轮作用域 drop → 寻址信封 reply drop → 请求方即时感知
75                drop(event);
76            }
77            // 排空 + 断连:close 已把全部 send 端消费尽
78            Err(_disconnected) => {
79                state.finish();
80                break;
81            }
82        }
83    }
84    stats.done.store(true, Ordering::Release);
85}
ek_core/worker.rs

ek_core/
worker.rs
