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::context::PluginContext;
8use crate::log_plugin::{EVENT_PLUG_DOWN_PREFIX, LOG_LEVEL_ERROR, LOG_LEVEL_WARN};
9use crate::plugin::{PANIC_LIMIT, Plugin};
10use crate::registry::{EntryStats, PluginStateCell, PluginStateCode};
11use crossbeam_channel::Receiver;
12use ek_event::{Event, EventEngine};
13use std::panic::{AssertUnwindSafe, catch_unwind};
14use std::sync::atomic::Ordering;
15use std::sync::Arc;
16use std::thread::JoinHandle;
17
18pub(crate) fn spawn_worker(
19    plugin: Arc<dyn Plugin>,
20    rx: Receiver<Event>,
21    state: PluginStateCell,
22    stats: Arc<EntryStats>,
23    ctx: Arc<PluginContext>,
24    bus: Arc<EventEngine>,
25) -> JoinHandle<()> {
26    let name = plugin.name().to_string();
27    std::thread::Builder::new()
28        .name(format!("ek-worker.{name}"))
29        .spawn(move || worker_loop(plugin, rx, state, stats, ctx, bus))
30        .expect("ek-core: spawn worker thread")
31}
32
33fn worker_loop(
34    plugin: Arc<dyn Plugin>,
35    rx: Receiver<Event>,
36    state: PluginStateCell,
37    stats: Arc<EntryStats>,
38    ctx: Arc<PluginContext>,
39    bus: Arc<EventEngine>,
40) {
41    // D3 = 连续 panic×3 熔断(复审 F3 修正):处理成功即复位;
42    // stats.panics 为累计统计面,不参与熔断判定。
43    let mut consecutive: u64 = 0;
44    loop {
45        match rx.recv() {
46            Ok(event) => {
47                let result = catch_unwind(AssertUnwindSafe(|| plugin.on_mail(&event)));
48                match result {
49                    Ok(()) => consecutive = 0,
50                    Err(_panic) => {
51                        consecutive += 1;
52                        let _ = stats.panics.fetch_add(1, Ordering::AcqRel);
53                        if consecutive >= PANIC_LIMIT as u64 {
54                            if state.cas(PluginStateCode::Running, PluginStateCode::Quarantined) {
55                                ctx.write_log(
56                                    LOG_LEVEL_ERROR,
57                                    &format!(
58                                        "plugin {:?} quarantined after {} consecutive handler panics",
59                                        plugin.name(),
60                                        consecutive
61                                    ),
62                                );
63                                // 双通道告警:通用日志 + 熔断广播(可被监控插件订阅)
64                                let _ = bus.put(Event::new(format!(
65                                    "{EVENT_PLUG_DOWN_PREFIX}{}",
66                                    plugin.name()
67                                )));
68                            }
69                            // 检疫即退出:邮箱内残留邮件随之丢弃(drop rx → forwarder 断连短路)
70                            break;
71                        }
72                        ctx.write_log(
73                            LOG_LEVEL_WARN,
74                            &format!(
75                                "plugin {:?} handler panicked ({consecutive}/{PANIC_LIMIT})",
76                                plugin.name()
77                            ),
78                        );
79                    }
80                }
81                // event 随本轮作用域 drop → 寻址信封 reply drop → 请求方即时感知
82                drop(event);
83            }
84            // 排空 + 断连:close 已把全部 send 端消费尽
85            Err(_disconnected) => {
86                state.finish();
87                break;
88            }
89        }
90    }
91    stats.done.store(true, Ordering::Release);
92}
ek_core/worker.rs

ek_core/
worker.rs
