use std::error::Error;
use std::panic::{AssertUnwindSafe, catch_unwind};
use std::sync::OnceLock;
use beamr::module::ModuleRegistry;
use beamr::scheduler::{NativeBifs, SchedulerConfig};
use super::*;
const SCHEDULER_UNAVAILABLE: &str = "channel actor scheduler is no longer available";
const PROCESS_INFO_UNAVAILABLE: &str = "channel actor process-info facility is unavailable";
fn orphaned_core() -> Result<ChannelActorCore, Box<dyn Error>> {
static FIXTURE: OnceLock<Result<(Weak<Scheduler>, Schema), String>> = OnceLock::new();
let (scheduler, schema) = FIXTURE
.get_or_init(|| {
let schema = Schema::new(serde_json::json!({"type": "object"}))
.map_err(|error| error.to_string())?;
let owner = Arc::new(Scheduler::new(
SchedulerConfig {
thread_count: Some(1),
..SchedulerConfig::default()
},
Arc::new(ModuleRegistry::new()),
NativeBifs::none(),
)?);
let scheduler = Arc::downgrade(&owner);
drop(owner);
Ok((scheduler, schema))
})
.as_ref()
.map_err(Clone::clone)?;
assert!(scheduler.upgrade().is_none());
Ok(ChannelActorCore {
scheduler: scheduler.clone(),
command_atom: Atom::OK,
schema: Mutex::new(schema.clone()),
subscribers: Mutex::new(Vec::new()),
closed: Mutex::new(false),
commands: Mutex::new(VecDeque::new()),
current_pid: Mutex::new(Some(37)),
restart_lock: Mutex::new(()),
next_command_id: AtomicU64::new(1),
departures: Departures {
channel: String::new(),
observer: Arc::new(OnceLock::new()),
},
})
}
fn assert_delivery_error<T>(
result: Result<T, LiminalError>,
expected: &str,
) -> Result<(), Box<dyn Error>> {
let error = result.err().ok_or("expected a delivery refusal")?;
match error {
LiminalError::DeliveryFailed { message } => assert_eq!(message, expected),
error => return Err(format!("wrong refusal: {error}").into()),
}
Ok(())
}
fn assert_unchanged(core: &ChannelActorCore) -> Result<(), Box<dyn Error>> {
assert!(lock(&core.commands)?.is_empty());
assert_eq!(core.current_pid()?, Some(37));
assert_eq!(core.next_command_id.load(Ordering::Relaxed), 1);
Ok(())
}
#[test]
fn expired_scheduler_getter_refuses() -> Result<(), Box<dyn Error>> {
let core = orphaned_core()?;
assert_delivery_error(core.scheduler(), SCHEDULER_UNAVAILABLE)?;
assert_unchanged(&core)
}
#[test]
fn expired_scheduler_enqueue_refuses_before_mutation() -> Result<(), Box<dyn Error>> {
let core = orphaned_core()?;
let (reply, response) = mpsc::sync_channel(1);
let result = core.enqueue(ChannelCommandKind::Boot { reply });
assert_unchanged(&core)?;
assert_delivery_error(result, SCHEDULER_UNAVAILABLE)?;
assert!(matches!(
response.try_recv(),
Err(mpsc::TryRecvError::Disconnected)
));
Ok(())
}
#[test]
fn expired_scheduler_wait_refuses() -> Result<(), Box<dyn Error>> {
let core = orphaned_core()?;
let (reply, response) = mpsc::sync_channel::<Result<(), LiminalError>>(1);
reply.send(Ok(()))?;
assert_delivery_error(core.wait_live(&response, 37), SCHEDULER_UNAVAILABLE)?;
assert!(matches!(response.try_recv(), Ok(Ok(()))));
assert_unchanged(&core)
}
#[test]
fn expired_scheduler_schema_wait_refuses() -> Result<(), Box<dyn Error>> {
let core = orphaned_core()?;
let schema_id = lock(&core.schema)?.id();
let (reply, response) = mpsc::sync_channel(1);
reply.send(Ok(schema_id))?;
let error = core
.wait_schema_live(&response, 37)
.err()
.ok_or("expected a schema wait refusal")?;
let expected = LiminalError::DeliveryFailed {
message: SCHEDULER_UNAVAILABLE.to_owned(),
}
.to_string();
assert!(
matches!(error, SchemaValidationError::InvalidSchema { message } if message == expected)
);
assert!(matches!(response.try_recv(), Ok(Ok(id)) if id == schema_id));
assert_unchanged(&core)
}
#[test]
fn boot_without_process_info_refuses_before_registration_access() -> Result<(), Box<dyn Error>> {
let core = orphaned_core()?;
let guard = lock(&core.subscribers)?;
let poisoned = catch_unwind(AssertUnwindSafe(move || {
let guard = guard;
assert_eq!(guard.len(), 1, "poison the registration lock");
}));
assert!(poisoned.is_err());
assert!(core.subscribers.is_poisoned());
assert_delivery_error(
core.apply_boot(&ProcessContext::new()),
PROCESS_INFO_UNAVAILABLE,
)
}