use std::sync::mpsc::Sender;
use std::sync::{Arc, Barrier, mpsc};
use std::time::Duration;
use beamr::module::ModuleRegistry;
use beamr::native::native_process::{NativeContext, NativeHandler, NativeOutcome};
use beamr::process::ExitReason;
use beamr::scheduler::{Scheduler, SchedulerConfig, SchedulerServices};
use super::LifecycleConfig;
use super::observation::ExitObservation;
use crate::component::{
ChildArgument, ChildSpec, ComponentId, ComponentMeta, ServiceCapability, SupervisionPolicy,
};
use crate::registry::ComponentRegistry;
const WORKER: &[u8] = include_bytes!("../../tests/fixtures/spike_worker_v1.beam");
const FFI: &[u8] = include_bytes!("../../tests/fixtures/spike_worker_ffi.beam");
const OPERATION_DEADLINE: Duration = Duration::from_secs(3);
const IDLE_WITNESS_HOLD: Duration = Duration::from_millis(25);
const OUTSIDE_WINDOW_RESTART_WINDOW: Duration = Duration::from_millis(40);
fn hold_without_polling(duration: Duration) {
let (_sender, receiver) = mpsc::channel::<()>();
assert_eq!(
receiver.recv_timeout(duration),
Err(mpsc::RecvTimeoutError::Timeout)
);
}
fn assert_timer_wheel_empty(scheduler: &Scheduler, phase: &str) {
let timers = match scheduler.timers().lock() {
Ok(timers) => timers,
Err(poisoned) => poisoned.into_inner(),
};
assert!(timers.is_empty(), "timer scheduled {phase}");
}
fn id(name: &str) -> ComponentId {
ComponentId::derive("frame.tests.event-driven", name)
}
fn policy() -> SupervisionPolicy {
SupervisionPolicy {
max_restarts: 3,
window: Duration::from_secs(30),
}
}
fn ffi_meta() -> ComponentMeta {
ComponentMeta {
id: id("ffi"),
name: "event witness ffi".to_owned(),
version: "1.0.0".to_owned(),
requires: Vec::new(),
provides: vec![ServiceCapability {
id: "event.witness.ffi".to_owned(),
description: "fixture mailbox implementation".to_owned(),
}],
needs: Vec::new(),
children: Vec::new(),
supervision: Some(policy()),
}
}
fn worker_meta() -> ComponentMeta {
ComponentMeta {
id: id("worker"),
name: "event witness worker".to_owned(),
version: "1.0.0".to_owned(),
requires: vec![id("ffi")],
provides: vec![ServiceCapability {
id: "event.witness.worker".to_owned(),
description: "event-driven supervision witness".to_owned(),
}],
needs: Vec::new(),
children: vec![ChildSpec {
name: "child".to_owned(),
module: "spike_worker".to_owned(),
function: "run".to_owned(),
arguments: vec![ChildArgument::SupervisorPid],
liveness_message: 0,
stop_message: 2,
}],
supervision: Some(policy()),
}
}
#[test]
fn idle_tree_command_and_exit_wakes_schedule_zero_timer_work()
-> Result<(), Box<dyn std::error::Error>> {
let scheduler = Arc::new(crate::composition::compose_scheduler(
SchedulerConfig {
thread_count: Some(2),
..SchedulerConfig::default()
},
SchedulerServices::minimal(),
Arc::new(ModuleRegistry::new()),
)?);
let registry = ComponentRegistry::new(
Arc::clone(&scheduler),
LifecycleConfig {
operation_timeout: OPERATION_DEADLINE,
},
);
registry.register(ffi_meta(), FFI.to_vec())?;
registry.start(id("ffi"))?;
registry.register(worker_meta(), WORKER.to_vec())?;
registry.start(id("worker"))?;
assert_timer_wheel_empty(&scheduler, "after tree start");
let old_pid = registry
.status(id("worker"))?
.ok_or_else(|| std::io::Error::other("worker status missing"))?
.children[0]
.pid;
hold_without_polling(IDLE_WITNESS_HOLD);
assert_timer_wheel_empty(&scheduler, "during the idle hold");
registry.send_child(id("worker"), "child", 7)?;
assert_eq!(registry.probe_child(id("worker"), "child")?, 1);
let exit = ExitObservation::register(&scheduler, old_pid, OPERATION_DEADLINE)
.map_err(|_| std::io::Error::other("exit observation registration failed"))?;
registry.send_child(id("worker"), "child", 1)?;
exit.wait()
.map_err(|_| std::io::Error::other("child EXIT observation failed"))?;
assert_eq!(registry.probe_child(id("worker"), "child")?, 0);
let new_pid = registry
.status(id("worker"))?
.ok_or_else(|| std::io::Error::other("worker status missing after restart"))?
.children[0]
.pid;
assert_ne!(new_pid, old_pid);
assert_timer_wheel_empty(&scheduler, "after the restart storm");
registry.remove(id("worker"))?;
registry.remove(id("ffi"))?;
assert_eq!(scheduler.process_count(), 0);
scheduler.shutdown();
Ok(())
}
#[test]
fn equal_crash_counts_outside_restart_window_do_not_escalate()
-> Result<(), Box<dyn std::error::Error>> {
let scheduler = observation_scheduler()?;
let registry = ComponentRegistry::new(
Arc::clone(&scheduler),
LifecycleConfig {
operation_timeout: OPERATION_DEADLINE,
},
);
registry.register(ffi_meta(), FFI.to_vec())?;
registry.start(id("ffi"))?;
let mut worker = worker_meta();
worker.supervision = Some(SupervisionPolicy {
max_restarts: 1,
window: OUTSIDE_WINDOW_RESTART_WINDOW,
});
registry.register(worker, WORKER.to_vec())?;
registry.start(id("worker"))?;
for crash in 0..2 {
let pid = registry
.status(id("worker"))?
.ok_or_else(|| std::io::Error::other("worker status missing"))?
.children[0]
.pid;
let exit = ExitObservation::register(&scheduler, pid, OPERATION_DEADLINE)
.map_err(|error| std::io::Error::other(format!("register exit: {error:?}")))?;
registry.send_child(id("worker"), "child", 1)?;
await_exit(exit)?;
assert_eq!(registry.probe_child(id("worker"), "child")?, 0);
if crash == 0 {
hold_without_polling(OUTSIDE_WINDOW_RESTART_WINDOW.saturating_mul(2));
}
}
assert_eq!(
registry
.status(id("worker"))?
.ok_or_else(|| std::io::Error::other("worker status missing after crashes"))?
.state,
crate::event::LifecycleState::Running
);
registry.remove(id("worker"))?;
registry.remove(id("ffi"))?;
assert_eq!(scheduler.process_count(), 0);
scheduler.shutdown();
Ok(())
}
struct Parked;
impl NativeHandler for Parked {
fn handle(&mut self, _context: &mut NativeContext<'_>) -> NativeOutcome {
NativeOutcome::Wait
}
}
struct StopsOnce {
dropped: Sender<()>,
}
impl NativeHandler for StopsOnce {
fn handle(&mut self, _context: &mut NativeContext<'_>) -> NativeOutcome {
NativeOutcome::Stop(ExitReason::Normal)
}
}
impl Drop for StopsOnce {
fn drop(&mut self) {
let _undelivered = self.dropped.send(());
}
}
fn observation_scheduler() -> Result<Arc<Scheduler>, Box<dyn std::error::Error>> {
Ok(Arc::new(crate::composition::compose_scheduler(
SchedulerConfig {
thread_count: Some(2),
..SchedulerConfig::default()
},
SchedulerServices::minimal(),
Arc::new(ModuleRegistry::new()),
)?))
}
fn await_exit(observation: ExitObservation) -> Result<(), Box<dyn std::error::Error>> {
observation.wait().map(|_| ()).map_err(|error| {
std::io::Error::other(format!("exit observation failed: {error:?}")).into()
})
}
#[test]
fn tombstone_observation_closes_before_during_and_after_registration_edges()
-> Result<(), Box<dyn std::error::Error>> {
const REGISTRATION_RACE_CASES: usize = 16;
let scheduler = observation_scheduler()?;
let (dropped_tx, dropped_rx) = mpsc::channel();
let already_dead = scheduler.spawn_native(Box::new(move || {
Box::new(StopsOnce {
dropped: dropped_tx.clone(),
})
}))?;
dropped_rx.recv_timeout(OPERATION_DEADLINE)?;
await_exit(
ExitObservation::register(&scheduler, already_dead, OPERATION_DEADLINE)
.map_err(|error| std::io::Error::other(format!("register dead: {error:?}")))?,
)?;
let after_registration = scheduler.spawn_native(Box::new(|| Box::new(Parked)))?;
let observation = ExitObservation::register(&scheduler, after_registration, OPERATION_DEADLINE)
.map_err(|error| std::io::Error::other(format!("register live: {error:?}")))?;
scheduler.terminate_process(after_registration, ExitReason::Kill);
await_exit(observation)?;
for _ in 0..REGISTRATION_RACE_CASES {
let target = scheduler.spawn_native(Box::new(|| Box::new(Parked)))?;
let barrier = Arc::new(Barrier::new(2));
let terminating_scheduler = Arc::clone(&scheduler);
let terminating_barrier = Arc::clone(&barrier);
let terminator = std::thread::spawn(move || {
terminating_barrier.wait();
terminating_scheduler.terminate_process(target, ExitReason::Kill);
});
barrier.wait();
let observation = ExitObservation::register(&scheduler, target, OPERATION_DEADLINE)
.map_err(|error| std::io::Error::other(format!("register race: {error:?}")))?;
terminator
.join()
.map_err(|_| std::io::Error::other("terminator thread panicked"))?;
await_exit(observation)?;
}
assert_eq!(scheduler.process_count(), 0);
scheduler.shutdown();
Ok(())
}