ek-core 0.5.6

//! fault_isolation:熔断检疫 / 慢邻居隔离 / 邮箱背压 / close 超时兜底
//!
//! 0.3.0:宿主零日志 —— quarantine 观测只走结构化通道
//! (ePlugQDown.<名> 广播 + PluginState::Quarantined + stats.panics)。

mod common;

use common::wait_until;
use ek_core::{
    AskError, EVENT_PLUG_DOWN_PREFIX, Envelope, MAILBOX_SOFT_CAP, MainEngine, PluginContext,
    PluginState,
};
use ek_event::Event;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

#[derive(Debug)]
pub struct Ping {
    pub n: u64,
}
#[derive(Debug, PartialEq)]
pub struct Pong {
    pub n: u64,
}

// FI-1 连续 3 次 panic → Quarantined;ePlugQDown.p3 结构化广播;邻居不受影响
#[test]
fn fi1_quarantine() {
    let eng = MainEngine::new(None);

    let good = common::counter();
    {
        let g = Arc::clone(&good);
        eng.add_plugin(
            common::test_plugin("good")
                .routes_str(&["eTick."])
                .mail(move |_e: &Event| {
                    common::bump(&g);
                })
                .build(),
        )
        .unwrap();
    }

    eng.add_plugin(
        common::test_plugin("p3")
            .routes_str(&["eTick."])
            .mail(move |_e: &Event| panic!("handler panic"))
            .build(),
    )
    .unwrap();

    let down = common::counter();
    {
        let d = Arc::clone(&down);
        eng.add_plugin(
            common::test_plugin("watch")
                .routes_owned(vec![format!("{EVENT_PLUG_DOWN_PREFIX}p3")])
                .mail(move |_e: &Event| {
                    common::bump(&d);
                })
                .build(),
        )
        .unwrap();
    }

    eng.start().unwrap();
    for i in 0..6u64 {
        eng.event_engine()
            .put(Event::with_data("eTick.", Arc::new(i)))
            .unwrap();
    }

    assert!(
        wait_until(Duration::from_secs(3), || common::load(&good) == 6),
        "good plugin keeps flowing"
    );
    assert_eq!(eng.get_state("p3"), Some(PluginState::Quarantined));
    let stats = eng.mail_stats("p3").unwrap();
    assert_eq!(stats.panics, 3);
    // 0.4.0 补钉(环节自审):广播入邮箱要过一拍路由,固定值直断撞到
    // 「状态已到、广播在路上」的窄缝 —— 本文件其余断言同款 wait_until 守卫。
    assert!(
        wait_until(Duration::from_secs(2), || common::load(&down) == 1),
        "one ePlugQDown.p3 broadcast"
    );
    // 0.3.0:无第二通道 —— 宿主零日志,quarantine 只此结构化广播 + 状态位

    // 检疫后再来事件:good 收满 6,p3 不再收(邮箱断连短路)
    for i in 6..10u64 {
        eng.event_engine()
            .put(Event::with_data("eTick.", Arc::new(i)))
            .unwrap();
    }
    std::thread::sleep(Duration::from_millis(300));
    assert!(
        wait_until(Duration::from_secs(2), || common::load(&good) == 10),
        "good keeps receiving post-quarantine"
    );
    eng.close_all(Duration::from_secs(2));
}

// FI-2 慢插件(每事件 80ms)不拖累其他插件的 tick 流(A1 隔离语义)
#[test]
fn fi2_slow_neighbor() {
    let eng = MainEngine::new(None);
    let slow = common::counter();
    let fast = common::counter();
    {
        let s = Arc::clone(&slow);
        eng.add_plugin(
            common::test_plugin("slow")
                .routes_str(&["eSlowIn."])
                .mail(move |_e: &Event| {
                    std::thread::sleep(Duration::from_millis(80));
                    common::bump(&s);
                })
                .build(),
        )
        .unwrap();
    }
    {
        let f = Arc::clone(&fast);
        eng.add_plugin(
            common::test_plugin("fast")
                .routes_str(&["eFastIn."])
                .mail(move |_e: &Event| {
                    common::bump(&f);
                })
                .build(),
        )
        .unwrap();
    }
    eng.start().unwrap();

    eng.event_engine().put(Event::new("eSlowIn.")).unwrap();
    std::thread::sleep(Duration::from_millis(20));
    let _t0 = Instant::now();
    for i in 0..3u64 {
        eng.event_engine()
            .put(Event::with_data("eFastIn.", Arc::new(i)))
            .unwrap();
    }
    assert!(wait_until(Duration::from_millis(300), || common::load(
        &fast
    ) == 3));
    assert_eq!(
        common::load(&slow),
        0,
        "slow worker still churning event #1"
    );
    assert!(wait_until(Duration::from_secs(2), || common::load(&slow) == 1));
    eng.close_all(Duration::from_secs(3));
}

// FI-3 软顶裁决(0.5.0 口径):soft_cap=32 + 200 突发 → **到顶才丢**,
// 受理量守恒(processed+spilled==200);峰值/软顶在统计面显形
#[test]
fn fi3_mailbox_backpressure() {
    let eng = MainEngine::new(None);
    let processed = common::counter();
    {
        let p = Arc::clone(&processed);
        eng.add_plugin(
            common::test_plugin("sleeper")
                .routes_str(&["eTick."])
                .cap(32)
                .mail(move |_e: &Event| {
                    std::thread::sleep(Duration::from_millis(2));
                    common::bump(&p);
                })
                .build(),
        )
        .unwrap();
    }
    eng.start().unwrap();
    for i in 0..200u64 {
        let _ = eng
            .event_engine()
            .put(Event::with_data("eTick.", Arc::new(i)));
    }
    assert!(
        wait_until(Duration::from_secs(3), || {
            let spilled = eng.mail_stats("sleeper").map(|s| s.spilled).unwrap_or(0);
            common::load(&processed) + spilled == 200
        }),
        "conservation: processed + spilled == total"
    );
    let spilled = eng.mail_stats("sleeper").unwrap().spilled;
    assert!(spilled >= 200 - 32 - 5, "majority spilled, got {spilled}");
    let s = eng.mail_stats("sleeper").unwrap();
    // 峰值 = 软顶那一格(裁决线是 `pending >= soft_cap` 才丢 → 深度不会越过它)
    assert_eq!(s.peak, 32, "峰值要顶到软顶那一格并留档,got {}", s.peak);
    assert_eq!(
        s.soft_cap, 32,
        "统计面要说清它自己的软顶(消费侧判据吃它自己的线)"
    );
    eng.close_all(Duration::from_secs(2));
}

// FI-3b 默认软顶下**不丢**(本轮核心口径):200 突发全收,spilled==0,峰值有读数
#[test]
fn fi3b_default_soft_cap_does_not_drop() {
    let eng = MainEngine::new(None);
    let processed = common::counter();
    {
        let p = Arc::clone(&processed);
        eng.add_plugin(
            common::test_plugin("drain")
                .routes_str(&["eTick."])
                .mail(move |_e: &Event| {
                    common::bump(&p);
                })
                .build(),
        )
        .unwrap();
    }
    eng.start().unwrap();
    for i in 0..200u64 {
        let _ = eng
            .event_engine()
            .put(Event::with_data("eTick.", Arc::new(i)));
    }
    assert!(
        wait_until(Duration::from_secs(5), || common::load(&processed) == 200),
        "默认软顶下 200 条必须全收到(一条都不许丢)"
    );
    let s = eng.mail_stats("drain").unwrap();
    assert_eq!(s.spilled, 0, "默认不丢:spilled 只在到软顶时才涨");
    assert!(s.peak >= 1, "峰值读数要有东西(至少排过一帧),got {}", s.peak);
    assert_eq!(s.soft_cap, MAILBOX_SOFT_CAP, "吃缺省的插件要看得见缺省软顶");
    assert!(
        s.pending <= 1,
        "收完深度要归零(容 1 帧在途),got {}",
        s.pending
    );
    eng.close_all(Duration::from_secs(2));
}

// FI-3d 水位计数器不许下绕(0.5.0 首跑现场抓出的真 bug):发送与取走并发抢序时,
// 计数必须始终落在 [0, 已发总数] 里 —— 旧实现「发送后自增」与 worker 的自减抢先后,
// 无符号下绕成 u64::MAX(debug 溢出检查当场 panic;release 静默变乱数)。
#[test]
fn fi3d_water_counters_never_wrap_under_race() {
    let eng = MainEngine::new(None);
    let processed = common::counter();
    {
        let p = Arc::clone(&processed);
        eng.add_plugin(
            common::test_plugin("fast")
                .routes_str(&["eTick."])
                .mail(move |_e: &Event| {
                    common::bump(&p);
                })
                .build(),
        )
        .unwrap();
    }
    eng.start().unwrap();
    const N: u64 = 8;
    const M: u64 = 3_000;
    let mut hs = Vec::new();
    for k in 0..N {
        let ev = Arc::clone(eng.event_engine());
        hs.push(
            std::thread::Builder::new()
                .name(format!("fi3d-producer-{k}"))
                .spawn(move || {
                    for i in 0..M {
                        let _ = ev.put(Event::with_data("eTick.", Arc::new((k, i))));
                    }
                })
                .expect("spawn producer"),
        );
    }
    for h in hs {
        let _ = h.join();
    }
    let total = N * M;
    assert!(
        wait_until(Duration::from_secs(25), || common::load(&processed)
            >= total),
        "无界默认不丢:{} 条要全收,实收 {}",
        total,
        common::load(&processed)
    );
    let s = eng.mail_stats("fast").unwrap();
    assert_eq!(s.spilled, 0, "默认软顶下不许丢");
    assert_eq!(s.pending, 0, "收完深度归零");
    assert!(
        s.peak <= total,
        "峰值不可能超过已发总数(超过就是计数下绕),got {}",
        s.peak
    );
    assert!(s.peak >= 1, "峰值要有读数,got {}", s.peak);
    eng.close_all(Duration::from_secs(4));
}

// FI-3c ask 超时带深度(0.5.0):目标忙 → Timeout.pending > 0,把"慢"变成读数
#[test]
fn fi3c_ask_timeout_carries_depth() {
    let eng = MainEngine::new(None);
    // busy:每帧睡 15ms(制造积压);eReq.busy 照常契约应答(只有排到它才应)
    eng.add_plugin(
        common::test_plugin("busy")
            .routes_str(&["eTick.", "eReq.busy"])
            .mail(|e: &Event| {
                if e.event_type == "eTick." {
                    std::thread::sleep(Duration::from_millis(15));
                } else {
                    Envelope::handle::<Ping, Pong>(e, |q| Pong { n: q.n });
                }
            })
            .build(),
    )
    .unwrap();

    let ctx_slot: Arc<Mutex<Option<PluginContext>>> = Arc::new(Mutex::new(None));
    let init_slot = Arc::clone(&ctx_slot);
    let kick = common::counter();
    let seen = common::records::<String>();
    let mail_seen = Arc::clone(&seen);
    eng.add_plugin(
        common::test_plugin("probe")
            .routes_str(&["eTick."])
            .init_hook(move |ctx| {
                *init_slot.lock().unwrap() = Some(ctx.clone());
            })
            .mail(move |e: &Event| {
                if e.event_type != "eTick." {
                    return;
                }
                // 第一帧时开一条问数线程:让 busy 那边先积压,再来问它
                if common::bump(&kick) == 1 {
                    let ctx = ctx_slot.lock().unwrap().as_ref().expect("ctx").clone();
                    let out = Arc::clone(&mail_seen);
                    std::thread::spawn(move || {
                        std::thread::sleep(Duration::from_millis(40));
                        let r =
                            ctx.ask::<Ping, Pong>("busy", Ping { n: 7 }, Duration::from_millis(60));
                        match r {
                            Err(AskError::Timeout { target, pending }) => {
                                common::push_record(&out, format!("timeout:{target}:{pending}"));
                            }
                            other => common::push_record(&out, format!("other:{other:?}")),
                        }
                    });
                }
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    for i in 0..40u64 {
        let _ = eng
            .event_engine()
            .put(Event::with_data("eTick.", Arc::new(i)));
    }
    let line = wait_until(Duration::from_secs(6), || {
        common::snapshot(&seen)
            .iter()
            .any(|l| l.starts_with("timeout:busy:"))
    });
    assert!(
        line,
        "probe 的 ask 要超时并带回深度,seen={:?}",
        common::snapshot(&seen)
    );
    let got = common::snapshot(&seen)
        .into_iter()
        .find(|l| l.starts_with("timeout:busy:"))
        .expect("checked");
    let pending: u64 = got
        .rsplit(':')
        .next()
        .expect("depth 段")
        .parse()
        .expect("深度是数字");
    assert!(pending > 0, "超时回包要带得出积压读数,got {got}");
    eng.close_all(Duration::from_secs(3));
}

// FI-4 close 超时兜底(worker 忙则不硬等):close 快速返回,worker 稍后排空自退
#[test]
fn fi4_close_join_deadline() {
    let eng = MainEngine::new(None);
    let entered = Arc::new(Mutex::new(false));
    let finished = Arc::new(Mutex::new(false));
    {
        let e2 = Arc::clone(&entered);
        let f2 = Arc::clone(&finished);
        eng.add_plugin(
            common::test_plugin("busy")
                .routes_str(&["eTick."])
                .mail(move |_e: &Event| {
                    *e2.lock().unwrap() = true;
                    std::thread::sleep(Duration::from_millis(3000));
                    *f2.lock().unwrap() = true;
                })
                .build(),
        )
        .unwrap();
    }
    eng.start().unwrap();
    eng.event_engine().put(Event::new("eTick.")).unwrap();
    assert!(wait_until(Duration::from_secs(1), || *entered
        .lock()
        .unwrap()));

    let t0 = Instant::now();
    eng.close(Duration::from_millis(200));
    let elapsed = t0.elapsed();
    assert!(
        elapsed < Duration::from_millis(1500),
        "close respects deadline; got {elapsed:?}"
    );
    assert_eq!(eng.get_state("busy"), None, "registry cleared on close");

    // 被剥离的 worker 随后自行排空收尾
    assert!(wait_until(Duration::from_secs(5), || *finished
        .lock()
        .unwrap()));
}

// 溢出诊断在 0.3.0 已不进总线(stderr);FI-3 的守恒断言即结构化事实面。

// FI-5 复审 F1 修正:Quarantined 插件 close 时同样调用 on_stop(曾装配即需清理)
#[test]
fn fi5_on_stop_after_quarantine() {
    let eng = MainEngine::new(None);
    let stopped = common::counter();
    eng.add_plugin(
        common::test_plugin("p3")
            .routes_str(&["eTick."])
            .mail(move |_e: &Event| panic!("on purpose"))
            .stop_hook({
                let s = Arc::clone(&stopped);
                move || {
                    common::bump(&s);
                }
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();
    for i in 0..4u64 {
        let _ = eng
            .event_engine()
            .put(Event::with_data("eTick.", Arc::new(i)));
    }
    assert!(
        wait_until(Duration::from_secs(3), || {
            eng.get_state("p3") == Some(PluginState::Quarantined)
        }),
        "quarantined after 3 panics"
    );
    assert_eq!(common::load(&stopped), 0, "no stop before close");
    eng.close_all(Duration::from_secs(2));
    assert_eq!(
        common::load(&stopped),
        1,
        "on_stop called even for quarantined"
    );
}

// FI-6 复审 F3 修正:熔断计数 = 连续(成功处理即复位),不是累计终生
#[test]
fn fi6_consecutive_panic_semantics() {
    let eng = MainEngine::new(None);
    // id 10,11,12 处 panic;其他 id 正常
    eng.add_plugin(
        common::test_plugin("pc")
            .routes_str(&["eTick."])
            .mail(move |e: &Event| {
                let id = e.data.downcast_ref::<u64>().copied().unwrap_or(0);
                if (10..13).contains(&id) {
                    panic!("isolated panic on purpose");
                }
            })
            .build(),
    )
    .unwrap();
    eng.start().unwrap();

    let put = |id: u64| {
        let _ = eng
            .event_engine()
            .put(Event::with_data("eTick.", Arc::new(id)));
    };

    // 第一波:三次 panic 各被一次成功处理隔开 → 连续计数反复归零,不熔断
    for id in [1u64, 10, 2, 11, 3, 12, 4] {
        put(id);
        std::thread::sleep(Duration::from_millis(15));
    }
    assert!(
        wait_until(Duration::from_secs(2), || {
            eng.mail_stats("pc").map(|s| s.panics).unwrap_or(0) >= 3
        }),
        "three lifetime panics observed"
    );
    assert_eq!(
        eng.get_state("pc"),
        Some(PluginState::Running),
        "consecutive semantics: isolated panics must not quarantine"
    );

    // 第二波:连续第三个 panic → 立即熔断
    put(10);
    put(11);
    put(12);
    assert!(
        wait_until(Duration::from_secs(2), || {
            eng.get_state("pc") == Some(PluginState::Quarantined)
        }),
        "3 consecutive panics quarantine"
    );
    let stats = eng.mail_stats("pc").unwrap();
    assert_eq!(stats.panics, 6, "cumulative stats 3 + 3");
    eng.close_all(Duration::from_secs(2));
}