liminal-rs 0.16.0

use std::error::Error;
use std::sync::atomic::Ordering;
use std::sync::{Arc, mpsc};
use std::time::Instant;

use beamr::process::ExitReason;

use super::acceptance_tests::{AttemptProbe, register};
use super::tests::fixtures::ControlledConsumer;
use super::{DispatchConversation, DispatchError};
use crate::conversation::ParticipantPid;
use crate::conversation::handoff::{AttemptDecision, ConsumptionReceipt, ReceiptMetrics};
use crate::routing::group::{ConsumerGroup, ConsumerRegistration};
use crate::routing::{ConsumerId, ConsumerStateView, RoutingMessage};

#[test]
fn missing_registered_forwarding_queue_is_not_delivery() -> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let pid = dispatch.supervisor().scheduler().spawn_test_process(false);
    let _ = dispatch.group().add_consumer(ConsumerRegistration::new(
        ParticipantPid::new(pid),
        ConsumerStateView::new(ConsumerId::new("inert"), 0, 4, 0, Vec::new()),
    ));
    let result = dispatch.dispatch(&RoutingMessage::new());
    dispatch
        .supervisor()
        .scheduler()
        .terminate_process(pid, ExitReason::Normal);
    dispatch.supervisor().shutdown();
    assert!(matches!(result, Err(DispatchError::Conversation(message))
        if message.contains("no registered forwarding queue")));
    Ok(())
}

#[test]
fn private_receipts_signal_consumption_and_keep_attempts_distinct() -> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::new(dispatch.supervisor())?;
    let first = ConsumptionReceipt::new(consumer.pid)?;
    let second = ConsumptionReceipt::new(consumer.pid)?;
    consumer.owner.send_dispatch(
        super::dispatch_envelope(&RoutingMessage::new())?,
        Arc::clone(&first),
    )?;
    consumer.parked.recv()?;
    assert_eq!(consumer.permit.messages_taken(), 0);
    assert_eq!(first.snapshot()?, None);
    assert_eq!(second.snapshot()?, None);
    consumer.permit.release()?;
    let first_accepted = first.wait()?;
    assert_eq!(first_accepted.participant, consumer.pid);
    assert_eq!(second.snapshot()?, None);
    consumer.owner.send_dispatch(
        super::dispatch_envelope(&RoutingMessage::new())?,
        Arc::clone(&second),
    )?;
    let second_accepted = second.wait()?;
    assert_eq!(second_accepted.participant, consumer.pid);
    assert_ne!(first_accepted.attempt, second_accepted.attempt);
    assert_eq!(first.snapshot()?, Some(first_accepted));
    assert_eq!(consumer.permit.messages_taken(), 2);
    first.discard_request()?;
    second.discard_request()?;
    consumer.owner.finalize();
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn dispatch_keeps_replies_and_lifetime_with_the_registered_conversation()
-> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::echo(dispatch.supervisor())?;
    register(&dispatch, &consumer, "echo");
    consumer.permit.release()?;
    let message = RoutingMessage::new().with("value", crate::routing::FieldValue::Integer(42));
    let expected = super::dispatch_envelope(&message)?;
    let result = dispatch.dispatch(&message)?;
    assert_eq!(result.delivered_to(), &ConsumerId::new("echo"));
    let reply = consumer.owner.handle().receive()?;
    assert_eq!(reply.payload, expected.payload);
    assert!(
        dispatch
            .supervisor()
            .scheduler()
            .process_table()
            .get(consumer.pid.get())
            .is_some()
    );
    let ordinary = super::dispatch_envelope(&RoutingMessage::new())?;
    consumer.owner.handle().send(ordinary.clone())?;
    assert_eq!(consumer.owner.handle().receive()?.payload, ordinary.payload);
    consumer.owner.finalize();
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn acceptance_before_exit_must_preserve_the_delivered_result() -> Result<(), Box<dyn Error>> {
    let mut dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::new(dispatch.supervisor())?;
    register(&dispatch, &consumer, "selected");
    let (probe, armed) = AttemptProbe::new();
    let (accepted, acceptance) = mpsc::channel();
    *probe
        .accepted_signal
        .lock()
        .map_err(|error| error.to_string())? = Some(accepted);
    let (observe, permission) = 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()?, consumer.pid);
    consumer.permit.release()?;
    let event = acceptance.recv()?;
    dispatch
        .supervisor()
        .scheduler()
        .terminate_process(consumer.pid.get(), ExitReason::Error);
    let exited = probe
        .captured_exit
        .lock()
        .map_err(|error| error.to_string())?
        .take()
        .ok_or("actual EXIT signal was not captured")?;
    exited.recv()?;
    observe.send(())?;
    let result = worker
        .join()
        .map_err(|_| "acceptance-first dispatch worker panicked")?;
    dispatch.supervisor().shutdown();
    assert_eq!(event.participant, consumer.pid);
    assert_eq!(probe.receipts.acceptances.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.exits.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.terminals.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.registrations.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.releases.load(Ordering::Acquire), 1);
    assert_eq!(probe.reroutes.load(Ordering::Acquire), 0);
    assert!(
        matches!(result, Ok(outcome) if outcome.delivered_to() == &ConsumerId::new("selected"))
    );
    Ok(())
}

#[test]
fn supervisor_shutdown_settles_a_pending_dispatch_by_name_once() -> Result<(), Box<dyn Error>> {
    let mut dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::new(dispatch.supervisor())?;
    register(&dispatch, &consumer, "held");
    let (probe, armed) = AttemptProbe::new();
    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()?, consumer.pid);
    assert_eq!(consumer.permit.messages_taken(), 0);
    dispatch.supervisor().shutdown();
    let result = worker
        .join()
        .map_err(|_| "shutdown dispatch worker panicked")?;
    assert!(matches!(result, Err(DispatchError::Conversation(message))
        if message.contains("shutdown")));
    assert_eq!(probe.waiter_resumes.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.terminals.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.registrations.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.releases.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.acceptances.load(Ordering::Acquire), 0);
    assert_eq!(consumer.permit.messages_taken(), 0);
    consumer.owner.finalize();
    Ok(())
}

fn linked_actor(
    dispatch: &DispatchConversation,
    participant: ParticipantPid,
) -> Result<crate::conversation::ConversationActor, crate::error::LiminalError> {
    dispatch
        .supervisor()
        .spawn(crate::conversation::ConversationConfig::new(
            vec![participant],
            None,
            crate::channel::ChannelMode::Ephemeral,
            crate::conversation::CrashPolicy::RouteToNext,
        ))
}

#[test]
fn explicit_abandonment_removes_only_its_request_and_keeps_the_consumer()
-> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::new(dispatch.supervisor())?;
    let actor = linked_actor(&dispatch, consumer.pid)?;
    let first = ConsumptionReceipt::new(consumer.pid)?;
    let second = ConsumptionReceipt::new(consumer.pid)?;
    let metrics = Arc::new(ReceiptMetrics::default());
    first.instrument(Arc::clone(&metrics))?;
    let mut first_registration = actor.register_dispatch_receipt(Arc::clone(&first))?;
    let mut second_registration = actor.register_dispatch_receipt(Arc::clone(&second))?;
    actor.send_dispatch(
        super::dispatch_envelope(&RoutingMessage::new())?,
        Arc::clone(&first),
    )?;
    actor.send_dispatch(
        super::dispatch_envelope(&RoutingMessage::new())?,
        Arc::clone(&second),
    )?;
    consumer.parked.recv()?;
    assert_eq!(consumer.permit.messages_taken(), 0);
    first_registration.release()?;
    first_registration.release()?;
    assert_eq!(first.wait_decision()?, AttemptDecision::Abandoned);
    assert_eq!(second.snapshot()?, None);
    consumer.permit.release()?;
    let accepted = second.wait()?;
    assert_eq!(accepted.participant, consumer.pid);
    assert_eq!(consumer.permit.messages_taken(), 1);
    assert_eq!(first.snapshot()?, None);
    assert_eq!(metrics.terminals.load(Ordering::Acquire), 1);
    assert_eq!(metrics.registrations.load(Ordering::Acquire), 1);
    assert_eq!(metrics.releases.load(Ordering::Acquire), 1);
    second_registration.release()?;
    actor.finalize();
    assert!(
        dispatch
            .supervisor()
            .scheduler()
            .process_table()
            .get(consumer.pid.get())
            .is_some()
    );
    consumer.owner.finalize();
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn already_observed_exit_replays_the_exact_instant_to_its_receipt() -> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::new(dispatch.supervisor())?;
    let actor = linked_actor(&dispatch, consumer.pid)?;
    let (signal, exited) = mpsc::sync_channel(1);
    actor.notify_on_participant_exit(consumer.pid, signal)?;
    dispatch
        .supervisor()
        .scheduler()
        .terminate_process(consumer.pid.get(), ExitReason::Error);
    let original = exited.recv()?;
    let receipt = ConsumptionReceipt::new(consumer.pid)?;
    let metrics = Arc::new(ReceiptMetrics::default());
    receipt.instrument(Arc::clone(&metrics))?;
    let mut registration = actor.register_dispatch_receipt(Arc::clone(&receipt))?;
    assert_eq!(receipt.wait_decision()?, AttemptDecision::Exited(original));
    receipt.decide(AttemptDecision::Closed)?.publish();
    assert_eq!(receipt.wait_decision()?, AttemptDecision::Exited(original));
    assert_eq!(receipt.snapshot()?, None);
    assert_eq!(consumer.permit.messages_taken(), 0);
    registration.release()?;
    assert_eq!(metrics.exits.load(Ordering::Acquire), 1);
    assert_eq!(metrics.terminals.load(Ordering::Acquire), 1);
    assert_eq!(metrics.registrations.load(Ordering::Acquire), 1);
    assert_eq!(metrics.releases.load(Ordering::Acquire), 1);
    actor.finalize();
    consumer.owner.finalize();
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn stale_terminal_events_do_not_settle_a_successive_attempt() -> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::new(dispatch.supervisor())?;
    let actor = linked_actor(&dispatch, consumer.pid)?;
    let earlier = ConsumptionReceipt::new(consumer.pid)?;
    let current = ConsumptionReceipt::new(consumer.pid)?;
    let mut earlier_registration = actor.register_dispatch_receipt(Arc::clone(&earlier))?;
    let metrics = Arc::new(ReceiptMetrics::default());
    current.instrument(Arc::clone(&metrics))?;
    let mut current_registration = actor.register_dispatch_receipt(Arc::clone(&current))?;
    actor.send_dispatch(
        super::dispatch_envelope(&RoutingMessage::new())?,
        Arc::clone(&earlier),
    )?;
    actor.send_dispatch(
        super::dispatch_envelope(&RoutingMessage::new())?,
        Arc::clone(&current),
    )?;
    consumer.parked.recv()?;
    earlier_registration.release()?;
    earlier
        .decide(AttemptDecision::Exited(Instant::now()))?
        .publish();
    assert_eq!(current.snapshot()?, None);
    assert!(current.is_pending()?);
    consumer.permit.release()?;
    let accepted = current.wait()?;
    earlier.decide(AttemptDecision::Closed)?.publish();
    assert_eq!(
        current.wait_decision()?,
        AttemptDecision::Accepted(accepted)
    );
    assert_ne!(earlier.attempt(), accepted.attempt);
    assert_eq!(accepted.participant, consumer.pid);
    assert_eq!(consumer.permit.messages_taken(), 1);
    assert_eq!(metrics.acceptances.load(Ordering::Acquire), 1);
    assert_eq!(metrics.terminals.load(Ordering::Acquire), 1);
    assert_eq!(metrics.exits.load(Ordering::Acquire), 0);
    current_registration.release()?;
    assert_eq!(metrics.registrations.load(Ordering::Acquire), 1);
    assert_eq!(metrics.releases.load(Ordering::Acquire), 1);
    actor.finalize();
    consumer.owner.finalize();
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn conversation_finalization_closes_pending_and_late_receipts_once() -> Result<(), Box<dyn Error>> {
    let dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::new(dispatch.supervisor())?;
    let actor = linked_actor(&dispatch, consumer.pid)?;
    let receipt = ConsumptionReceipt::new(consumer.pid)?;
    let metrics = Arc::new(ReceiptMetrics::default());
    receipt.instrument(Arc::clone(&metrics))?;
    let mut registration = actor.register_dispatch_receipt(Arc::clone(&receipt))?;
    actor.send_dispatch(
        super::dispatch_envelope(&RoutingMessage::new())?,
        Arc::clone(&receipt),
    )?;
    consumer.parked.recv()?;
    actor.finalize();
    actor.finalize();
    assert_eq!(receipt.wait_decision()?, AttemptDecision::Closed);
    assert_eq!(metrics.terminals.load(Ordering::Acquire), 1);
    registration.release()?;
    assert_eq!(metrics.releases.load(Ordering::Acquire), 1);
    assert_eq!(consumer.permit.messages_taken(), 0);
    let late = ConsumptionReceipt::new(consumer.pid)?;
    let mut late_registration = actor.register_dispatch_receipt(Arc::clone(&late))?;
    assert_eq!(late.wait_decision()?, AttemptDecision::Closed);
    late_registration.release()?;
    assert!(
        dispatch
            .supervisor()
            .scheduler()
            .process_table()
            .get(consumer.pid.get())
            .is_some()
    );
    consumer.owner.finalize();
    dispatch.supervisor().shutdown();
    Ok(())
}

#[test]
fn an_unrelated_acceptance_keeps_the_other_dispatch_pending() -> Result<(), Box<dyn Error>> {
    let mut dispatch = DispatchConversation::new(ConsumerGroup::new(super::select_first()))?;
    let consumer = ControlledConsumer::new(dispatch.supervisor())?;
    register(&dispatch, &consumer, "shared");
    let (probe, armed) = AttemptProbe::new();
    let (accepted, acceptance) = mpsc::channel();
    *probe
        .accepted_signal
        .lock()
        .map_err(|error| error.to_string())? = Some(accepted);
    dispatch.probe = Some(Arc::clone(&probe));
    let first_dispatch = dispatch.clone();
    let first = std::thread::spawn(move || {
        first_dispatch.dispatch(
            &RoutingMessage::new().with("request", crate::routing::FieldValue::Integer(1)),
        )
    });
    assert_eq!(armed.recv()?, consumer.pid);
    let second_dispatch = dispatch.clone();
    let (settled, second_result) = mpsc::channel();
    let second = std::thread::spawn(move || {
        let result = second_dispatch.dispatch(
            &RoutingMessage::new().with("request", crate::routing::FieldValue::Integer(2)),
        );
        settled.send(()).map_err(|error| error.to_string())?;
        Ok::<_, String>(result)
    });
    assert_eq!(armed.recv()?, consumer.pid);
    consumer.permit.release_one()?;
    let earlier = acceptance.recv()?;
    let first_outcome = first
        .join()
        .map_err(|_| "first unrelated dispatch worker panicked")??;
    assert_eq!(first_outcome.delivered_to(), &ConsumerId::new("shared"));
    assert_eq!(earlier.participant, consumer.pid);
    assert_eq!(consumer.permit.messages_taken(), 1);
    assert_eq!(probe.receipts.acceptances.load(Ordering::Acquire), 1);
    assert_eq!(probe.receipts.terminals.load(Ordering::Acquire), 1);
    assert!(matches!(
        second_result.try_recv(),
        Err(mpsc::TryRecvError::Empty)
    ));
    consumer.permit.release_one()?;
    let current = acceptance.recv()?;
    let second_outcome = second
        .join()
        .map_err(|_| "second unrelated dispatch worker panicked")???;
    assert_ne!(earlier.attempt, current.attempt);
    assert_eq!(current.participant, consumer.pid);
    assert_eq!(second_outcome.delivered_to(), &ConsumerId::new("shared"));
    assert_eq!(consumer.permit.messages_taken(), 2);
    assert_eq!(probe.receipts.acceptances.load(Ordering::Acquire), 2);
    assert_eq!(probe.receipts.terminals.load(Ordering::Acquire), 2);
    assert_eq!(probe.receipts.registrations.load(Ordering::Acquire), 2);
    assert_eq!(probe.receipts.releases.load(Ordering::Acquire), 2);
    assert_eq!(probe.waiter_resumes.load(Ordering::Acquire), 2);
    consumer.owner.finalize();
    dispatch.supervisor().shutdown();
    Ok(())
}