liminal-rs 0.16.4

pub(super) mod fixtures;

use std::error::Error;
use std::sync::Arc;
use std::time::Duration;

use beamr::process::ExitReason;

use super::{DispatchConversation, DispatchError, select_first};
use crate::conversation::ParticipantPid;
use crate::routing::function::loader::{ModuleLoader, RoutingModule};
use crate::routing::group::{ConsumerGroup, ConsumerRegistration};
use crate::routing::{
    ConsumerId, ConsumerStateView, RoutingDecision, RoutingFunction, RoutingMessage,
};
use fixtures::ControlledConsumer;

/// Routing function that selects the consumer named by the message's "target"
/// field, proving the function receives the message and consumer state.
fn select_by_target() -> RoutingFunction {
    ModuleLoader::new().load(RoutingModule::new(
        b"dispatch-by-target",
        |message, consumers| {
            let Some(crate::routing::FieldValue::Text(target)) = message.get("target") else {
                return RoutingDecision::none();
            };
            consumers
                .iter()
                .find(|state| state.consumer.as_str() == target)
                .map_or_else(RoutingDecision::none, |state| {
                    RoutingDecision::select(state.consumer.clone())
                })
        },
    ))
}

fn registration(id: &str, pid: u64) -> ConsumerRegistration {
    ConsumerRegistration::new(
        ParticipantPid::new(pid),
        ConsumerStateView::new(ConsumerId::new(id), 0, 4, 0, Vec::new()),
    )
}

/// True when some other live process is linked to `consumer_pid` (the dispatch
/// conversation actor). The actor pid is allocated after the consumer, so the
/// scan covers a window on both sides of the consumer pid.
fn has_link_to(scheduler: &beamr::scheduler::Scheduler, consumer_pid: u64) -> bool {
    (consumer_pid.saturating_sub(256)..consumer_pid + 256)
        .filter(|candidate| *candidate != consumer_pid && *candidate != 0)
        .any(|candidate| scheduler.is_linked(candidate, consumer_pid))
}

struct LinkReady {
    hook: beamr::hook::Hook,
    ready: std::sync::mpsc::Receiver<()>,
}

impl LinkReady {
    fn wait(&self) -> Result<(), String> {
        self.ready
            .recv()
            .map_err(|error| format!("dispatch link-ready signal disconnected: {error}"))
    }
}

impl Drop for LinkReady {
    fn drop(&mut self) {
        self.hook.unregister();
    }
}

fn arm_link_signal(scheduler: &Arc<beamr::scheduler::Scheduler>, consumer_pid: u64) -> LinkReady {
    let (ready_tx, ready) = std::sync::mpsc::channel();
    let hook = scheduler.hook().clone();
    let scheduler = Arc::clone(scheduler);
    let fired = std::sync::atomic::AtomicBool::new(false);
    hook.register(move |event| {
        if scheduler.is_linked(event.pid, consumer_pid)
            && !fired.swap(true, std::sync::atomic::Ordering::AcqRel)
        {
            if let Err(error) = ready_tx.send(()) {
                tracing::error!(%error, "dispatch link-ready signal disconnected");
            }
        }
        beamr::hook::HookDecision::Continue
    });
    LinkReady { hook, ready }
}

#[test]
fn dispatch_selects_consumer_via_routing_function() -> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(select_by_target()))?;

    let a_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    a_fixture.permit.release()?;
    let a = a_fixture.pid;
    let b_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    b_fixture.permit.release()?;
    let b = b_fixture.pid;
    let c_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    c_fixture.permit.release()?;
    let c = c_fixture.pid;
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        a,
        ConsumerStateView::new(ConsumerId::new("A"), 0, 4, 0, Vec::new()),
    ));
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        b,
        ConsumerStateView::new(ConsumerId::new("B"), 0, 4, 0, Vec::new()),
    ));
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        c,
        ConsumerStateView::new(ConsumerId::new("C"), 0, 4, 0, Vec::new()),
    ));

    let message =
        RoutingMessage::new().with("target", crate::routing::FieldValue::Text("B".to_owned()));
    let outcome = dispatch.dispatch(&message)?;

    assert_eq!(outcome.delivered_to(), &ConsumerId::new("B"));
    assert!(outcome.delivered_first_try());
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn dispatch_links_to_selected_consumer_before_completion() -> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(select_first()))?;
    let scheduler = dispatch.supervisor().scheduler();
    let consumer_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    let consumer = consumer_fixture.pid;
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        consumer,
        ConsumerStateView::new(ConsumerId::new("solo"), 0, 4, 0, Vec::new()),
    ));

    // Drive a dispatch on a background thread; while it holds the link in
    // flight, observe that the conversation actor is linked to the consumer.
    let dispatch_for_thread = dispatch.clone();
    let link_ready = arm_link_signal(&scheduler, consumer.get());
    let worker = std::thread::spawn(move || dispatch_for_thread.dispatch(&RoutingMessage::new()));
    link_ready.wait()?;
    let linked = has_link_to(&scheduler, consumer.get());
    assert!(linked, "dispatch must link to the selected consumer");
    consumer_fixture.permit.release()?;

    let outcome = worker.join().map_err(|_| "dispatch thread panicked")??;
    assert_eq!(outcome.delivered_to(), &ConsumerId::new("solo"));
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn consumer_crash_reroutes_to_backup_with_real_sub_millisecond_latency()
-> Result<(), Box<dyn Error>> {
    let mut dispatch = DispatchConversation::new(ConsumerGroup::new(select_first()))?;
    let scheduler = dispatch.supervisor().scheduler();
    let primary_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    let primary = primary_fixture.pid;
    let backup_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    backup_fixture.permit.release()?;
    let backup = backup_fixture.pid;
    // The group orders consumers by id; "a-primary" sorts first, so
    // `select_first` selects it before "b-backup".
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        primary,
        ConsumerStateView::new(ConsumerId::new("a-primary"), 0, 4, 0, Vec::new()),
    ));
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        backup,
        ConsumerStateView::new(ConsumerId::new("b-backup"), 0, 4, 0, Vec::new()),
    ));

    // (a) The primary is a real spawned consumer process. Once the dispatch
    // conversation has linked to it, (b) kill it. The link fires a trapped EXIT
    // into the conversation actor, which wakes the blocked dispatcher.
    let link_ready = super::acceptance_tests::arm_waiter_signal(&mut dispatch, primary);
    let crasher = std::thread::spawn(move || -> Result<(), String> {
        link_ready.wait()?;
        scheduler.terminate_process(primary.get(), ExitReason::Error);
        Ok(())
    });

    let outcome = dispatch.dispatch(&RoutingMessage::new())?;
    crasher.join().map_err(|_| "crasher thread panicked")??;

    assert_eq!(primary_fixture.permit.messages_taken(), 0);
    // (d) Re-routing excluded the crashed primary and selected the backup.
    assert_eq!(outcome.delivered_to(), &ConsumerId::new("b-backup"));
    assert_eq!(outcome.rerouted_from(), &[ConsumerId::new("a-primary")]);

    // (c) Exactly one real crash-to-reroute timing was recorded, measured along
    // the event path: from the EXIT instant captured inside the conversation
    // actor's link handler to the instant the dispatcher woke and re-routed.
    // This bound is what a polling regression would blow: the prior poll loop
    // slept in 50µs ticks and, worse, captured both timestamps back-to-back
    // *after* detection — so a regression to sampling `state()` between sleeps
    // would push the real EXIT→reroute span past this bound (or, if the metric
    // were faked back-to-back again, the span would no longer track the EXIT
    // instant the handler stamps and this assertion would not be meaningful).
    let timings = outcome.reroute_timings();
    assert_eq!(timings.len(), 1, "exactly one reroute timing expected");
    let latency = timings[0].detection_to_reroute();
    eprintln!(
        "quiet crash detection_to_reroute_us={}",
        latency.as_micros()
    );
    let bound = reroute_bound()?;
    assert!(
        latency < bound,
        "crash-to-reroute latency {latency:?} exceeded the box's bound {bound:?}"
    );
    // The EXIT instant must precede (or equal) re-route initiation: proof the
    // span is a real ordered interval, not two back-to-back `Instant::now()`s.
    assert!(
        timings[0].crash_observed() <= std::time::Instant::now(),
        "crash instant must lie in the past"
    );
    dispatch.supervisor().shutdown();
    Ok(())
}

/// The bound a crash-to-reroute span is held under: one millisecond, or four of
/// this box's own thread hand-offs when the box is slower than that.
///
/// This is a latency promise, not a structural one: the span is two scheduler
/// hand-offs (the crasher to the conversation actor's link handler, the actor to
/// the dispatcher it wakes) and nothing else, so it is bounded by what a hand-off
/// costs on the box at the moment, measured right here. What the bound catches is
/// a regression to a sleeping poll of a millisecond or more between samples of the
/// consumer's state; a poll finer than a hand-off was never visible to the fixed
/// millisecond either. The ordering assertions beside it are what prove the
/// timing flows from the EXIT handler's own instant. Measured on an idle Mac the
/// span is tens of microseconds and the bound is the millisecond; on a Mac running
/// three builds (load 55, 17 October 2026) the span was 3,006 µs and the hand-off
/// about one millisecond, so the box is measured against itself.
fn reroute_bound() -> Result<Duration, Box<dyn Error>> {
    let (tx, rx) = std::sync::mpsc::channel();
    let started = std::time::Instant::now();
    let pong = std::thread::spawn(move || tx.send(std::time::Instant::now()));
    let arrived = rx.recv().map_or(started, |sent| sent.max(started));
    let handoff = started
        .elapsed()
        .max(arrived.saturating_duration_since(started));
    pong.join().map_err(|_| "hand-off thread panicked")??;
    Ok(Duration::from_millis(1).max(handoff * 4))
}

#[test]
fn reroute_timing_spans_real_exit_to_reroute_via_event_path() -> Result<(), Box<dyn Error>> {
    // Drive a genuine crash and assert the recorded timing is a real, ordered
    // span — the EXIT instant captured in the link handler precedes re-route
    // initiation — rather than two `Instant::now()` calls measuring nothing.
    let mut dispatch = DispatchConversation::new(ConsumerGroup::new(select_first()))?;
    let scheduler = dispatch.supervisor().scheduler();
    let primary_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    let primary = primary_fixture.pid;
    let backup_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    backup_fixture.permit.release()?;
    let backup = backup_fixture.pid;
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        primary,
        ConsumerStateView::new(ConsumerId::new("a-primary"), 0, 4, 0, Vec::new()),
    ));
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        backup,
        ConsumerStateView::new(ConsumerId::new("b-backup"), 0, 4, 0, Vec::new()),
    ));

    let before_dispatch = std::time::Instant::now();
    let link_ready = super::acceptance_tests::arm_waiter_signal(&mut dispatch, primary);
    let crasher = std::thread::spawn(move || -> Result<(), String> {
        link_ready.wait()?;
        scheduler.terminate_process(primary.get(), ExitReason::Error);
        Ok(())
    });
    let outcome = dispatch.dispatch(&RoutingMessage::new())?;
    crasher.join().map_err(|_| "crasher thread panicked")??;

    assert_eq!(primary_fixture.permit.messages_taken(), 0);
    let timing = outcome
        .reroute_timings()
        .first()
        .ok_or("a reroute timing must be recorded after a crash")?;
    // The EXIT was observed after dispatch began and the detection-to-reroute
    // span is non-negative — both impossible to satisfy with the old faked
    // metric had it not flowed from the actual EXIT handler instant.
    assert!(
        timing.crash_observed() >= before_dispatch,
        "EXIT instant must fall after dispatch started"
    );
    eprintln!(
        "quiet EXIT detection_to_reroute_us={}",
        timing.detection_to_reroute().as_micros()
    );
    let bound = reroute_bound()?;
    assert!(
        timing.detection_to_reroute() < bound,
        "EXIT-to-reroute span {:?} exceeded the box's bound {bound:?}",
        timing.detection_to_reroute()
    );
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn no_remaining_consumer_after_crash_is_an_error() -> Result<(), Box<dyn Error>> {
    let mut dispatch = DispatchConversation::new(ConsumerGroup::new(select_first()))?;
    let scheduler = dispatch.supervisor().scheduler();
    let only_fixture = ControlledConsumer::new(dispatch.supervisor())?;
    let only = only_fixture.pid;
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        only,
        ConsumerStateView::new(ConsumerId::new("only"), 0, 4, 0, Vec::new()),
    ));

    let link_ready = super::acceptance_tests::arm_waiter_signal(&mut dispatch, only);
    let crasher = std::thread::spawn(move || -> Result<(), String> {
        link_ready.wait()?;
        scheduler.terminate_process(only.get(), ExitReason::Error);
        Ok(())
    });

    let result = dispatch.dispatch(&RoutingMessage::new());
    crasher.join().map_err(|_| "crasher thread panicked")??;
    assert_eq!(only_fixture.permit.messages_taken(), 0);

    assert_eq!(result, Err(DispatchError::NoConsumerAvailable));
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn empty_group_yields_no_consumer_available() -> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(select_first()))?;
    let result = dispatch.dispatch(&RoutingMessage::new());
    assert_eq!(result, Err(DispatchError::NoConsumerAvailable));
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn registration_helper_constructs_consumer_state() {
    let registration = registration("helper", 1);
    assert_eq!(registration.consumer(), &ConsumerId::new("helper"));
}

#[test]
fn delayed_exit_observation_records_scheduling_without_refusing_reroute()
-> Result<(), Box<dyn Error>> {
    let mut dispatch = DispatchConversation::new(ConsumerGroup::new(select_first()))?;
    let primary = ControlledConsumer::new(dispatch.supervisor())?;
    let backup = ControlledConsumer::new(dispatch.supervisor())?;
    backup.permit.release()?;
    let _ = dispatch
        .group()
        .add_consumer(registration("a-primary", primary.pid.get()));
    let _ = dispatch
        .group()
        .add_consumer(registration("b-backup", backup.pid.get()));
    let (probe, armed) = super::acceptance_tests::AttemptProbe::new();
    let (observe, permission) = std::sync::mpsc::channel();
    *probe
        .observe_permission
        .lock()
        .map_err(|error| error.to_string())? = Some(permission);
    dispatch.probe = Some(Arc::clone(&probe));
    let worker_dispatch = dispatch.clone();
    let worker = std::thread::spawn(move || worker_dispatch.dispatch(&RoutingMessage::new()));
    assert_eq!(armed.recv()?, primary.pid);
    dispatch
        .supervisor()
        .scheduler()
        .terminate_process(primary.pid.get(), ExitReason::Error);
    let captured = probe
        .captured_exit
        .lock()
        .map_err(|error| error.to_string())?
        .take()
        .ok_or("EXIT capture missing")?;
    let exited = captured.recv()?;
    // Observation stays signal-gated while independent CPU work occupies its
    // caller. No timer advances the dispatch or supplies a replacement instant.
    let checksum = (0..1_000_000_u64).fold(0_u64, |sum, value| {
        sum.wrapping_add(std::hint::black_box(value).rotate_left(13))
    });
    assert_eq!(checksum, 499_999_500_000_u64 << 13);
    let held = exited.elapsed();
    eprintln!(
        "delayed observer detection_to_release_us={}",
        held.as_micros()
    );
    assert!(
        held >= Duration::from_millis(1),
        "independent work did not exercise delayed observation: {held:?}"
    );
    observe.send(())?;
    // The backup also arms the probe, so retain its receiver until completion.
    let result = worker
        .join()
        .map_err(|_| "delayed observer dispatch worker panicked")?;
    dispatch.supervisor().shutdown();
    let outcome = result?;
    assert_eq!(outcome.delivered_to(), &ConsumerId::new("b-backup"));
    assert_eq!(outcome.rerouted_from(), &[ConsumerId::new("a-primary")]);
    assert_eq!(outcome.reroute_timings().len(), 1);
    let timing = outcome.reroute_timings()[0];
    assert_eq!(timing.crash_observed(), exited);
    assert!(timing.detection_to_reroute() >= held);
    eprintln!(
        "delayed observer detection_to_reroute_us={}",
        timing.detection_to_reroute().as_micros()
    );
    assert_eq!(primary.permit.messages_taken(), 0);
    assert_eq!(backup.permit.messages_taken(), 1);
    assert_eq!(probe.reroutes.load(std::sync::atomic::Ordering::Acquire), 1);
    Ok(())
}