liminal-rs 0.14.1

//! Conversation-mediated dispatch (ADR-007).
//!
//! A dispatch conversation replaces Kafka-style consumer group rebalancing with
//! per-message, fault-aware consumer selection. For each message the dispatch
//! conversation evaluates the group's routing function to select a target
//! consumer, links the dispatch process to that consumer via a beamr process
//! link, forwards the message, and observes completion. If the linked consumer
//! crashes, the link fires immediately (no polling, no heartbeat) and the
//! conversation re-routes the message to another available consumer, excluding
//! the one that crashed. Selection is always via the routing function — never
//! random — and adding or removing consumers takes effect on the next dispatch
//! without a stop-the-world rebalance.

#[cfg(test)]
use std::sync::mpsc;
use std::time::{Duration, Instant};

use crate::channel::ChannelMode;
use crate::conversation::handoff::{AttemptDecision, ConsumptionReceipt};
use crate::conversation::{ConversationConfig, ConversationSupervisor, CrashPolicy};
use crate::error::LiminalError;
use crate::routing::group::{ConsumerGroup, ConsumerRegistration};
use crate::routing::{
    ConsumerId, ConsumerStateView, FieldValue, FunctionError, RoutingDecision, RoutingMessage,
    SupervisedExecutor,
};

/// Outcome of a successful dispatch: the consumer the message was delivered to
/// and the chain of consumers that crashed and were re-routed past, in order.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DispatchOutcome {
    delivered_to: ConsumerId,
    rerouted_from: Vec<ConsumerId>,
    reroute_timings: Vec<RerouteTiming>,
}

impl DispatchOutcome {
    /// Consumer the message was ultimately delivered to.
    #[must_use]
    pub const fn delivered_to(&self) -> &ConsumerId {
        &self.delivered_to
    }

    /// Consumers that crashed mid-dispatch and were re-routed past, in order.
    #[must_use]
    pub fn rerouted_from(&self) -> &[ConsumerId] {
        &self.rerouted_from
    }

    /// Real crash-to-reroute timings, one per consumer in [`Self::rerouted_from`]
    /// and in the same order. Each spans the consumer's EXIT instant (captured
    /// in the link handler) through to re-route initiation, so callers can
    /// verify sub-millisecond, event-driven detection (R3 / CN7).
    #[must_use]
    pub fn reroute_timings(&self) -> &[RerouteTiming] {
        &self.reroute_timings
    }

    /// True when the message was delivered without any consumer crash.
    #[must_use]
    pub fn delivered_first_try(&self) -> bool {
        self.rerouted_from.is_empty()
    }
}

/// Failure surfaced when a message cannot be dispatched to any consumer.
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
pub enum DispatchError {
    /// The routing function selected no consumer from the available set.
    #[error("routing function selected no consumer from the available group set")]
    NoConsumerAvailable,
    /// The routing function selected a consumer that is not in the group.
    #[error("routing function selected unknown consumer '{0}'")]
    UnknownConsumerSelected(String),
    /// Evaluating the group's routing function failed under supervision.
    #[error("routing function evaluation failed: {0}")]
    Evaluation(#[from] FunctionError),
    /// The dispatch conversation infrastructure could not be driven.
    #[error("dispatch conversation failed: {0}")]
    Conversation(String),
}

impl From<LiminalError> for DispatchError {
    fn from(error: LiminalError) -> Self {
        Self::Conversation(error.to_string())
    }
}

/// Real crash-to-reroute latency along the event path.
///
/// Records the instant the consumer's trapped EXIT signal was observed inside
/// the conversation actor's link handler and the instant the dispatcher begins
/// re-routing after releasing the failed attempt.
///
/// `crash_observed` is captured *inside* the EXIT handler the moment the beamr
/// process link fires (see `ActorCore::handle_participant_exit`); it is not a
/// post-detection sample. `reroute_initiated` is captured before the next
/// selection, after observation and attempt cleanup.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RerouteTiming {
    crash_observed: Instant,
    reroute_initiated: Instant,
}

impl RerouteTiming {
    /// Elapsed time between the EXIT signal firing and re-route initiation.
    #[must_use]
    pub fn detection_to_reroute(&self) -> Duration {
        self.reroute_initiated
            .saturating_duration_since(self.crash_observed)
    }

    /// The instant the consumer's EXIT signal was observed in the link handler.
    #[must_use]
    pub const fn crash_observed(&self) -> Instant {
        self.crash_observed
    }
}

/// Drives conversation-mediated dispatch for one consumer group.
///
/// Each call to [`DispatchConversation::dispatch`] spawns a supervised
/// conversation actor (a beamr process), selects a consumer via the group's
/// routing function, links the conversation to the consumer, forwards the
/// message, and re-routes on consumer crash.
#[derive(Clone, Debug)]
pub struct DispatchConversation {
    group: ConsumerGroup,
    executor: SupervisedExecutor,
    supervisor: ConversationSupervisor,
    #[cfg(test)]
    probe: Option<std::sync::Arc<acceptance_tests::AttemptProbe>>,
}

impl DispatchConversation {
    /// Creates a dispatch driver for `group` using a fresh conversation supervisor.
    ///
    /// # Errors
    ///
    /// Returns [`DispatchError::Conversation`] if the beamr scheduler backing the
    /// conversation supervisor cannot start.
    pub fn new(group: ConsumerGroup) -> Result<Self, DispatchError> {
        let supervisor = ConversationSupervisor::new()?;
        let executor = SupervisedExecutor::with_default_timeout(supervisor.scheduler());
        Ok(Self {
            group,
            executor,
            supervisor,
            #[cfg(test)]
            probe: None,
        })
    }

    /// Creates a dispatch driver reusing an existing conversation supervisor and
    /// supervision executor.
    #[must_use]
    pub const fn with_supervisor(
        group: ConsumerGroup,
        executor: SupervisedExecutor,
        supervisor: ConversationSupervisor,
    ) -> Self {
        Self {
            group,
            executor,
            supervisor,
            #[cfg(test)]
            probe: None,
        }
    }

    /// Returns the consumer group this dispatch conversation routes over.
    #[must_use]
    pub const fn group(&self) -> &ConsumerGroup {
        &self.group
    }

    /// Returns the conversation supervisor backing dispatch processes.
    #[must_use]
    pub const fn supervisor(&self) -> &ConversationSupervisor {
        &self.supervisor
    }

    /// Dispatches `message` to a consumer selected by the group's routing function.
    ///
    /// Snapshots the group, evaluates the routing function over the live consumer
    /// state, links a dispatch conversation process to the selected consumer,
    /// forwards the message, and observes completion. On consumer crash the
    /// message is re-routed to another consumer, excluding any that have crashed.
    ///
    /// # Errors
    ///
    /// Returns [`DispatchError::NoConsumerAvailable`] if the routing function
    /// selects no consumer, [`DispatchError::UnknownConsumerSelected`] if it
    /// selects a consumer outside the group, [`DispatchError::Evaluation`] if the
    /// supervised routing function fails, and [`DispatchError::Conversation`] if
    /// the dispatch conversation process cannot be driven.
    pub fn dispatch(&self, message: &RoutingMessage) -> Result<DispatchOutcome, DispatchError> {
        let mut excluded: Vec<ConsumerId> = Vec::new();
        let mut reroute_timings: Vec<RerouteTiming> = Vec::new();
        loop {
            let selected = self.select_consumer(message, &excluded)?;
            match self.run_attempt(message, &selected)? {
                AttemptResult::Delivered => {
                    return Ok(DispatchOutcome {
                        delivered_to: selected.consumer().clone(),
                        rerouted_from: excluded,
                        reroute_timings,
                    });
                }
                AttemptResult::Crashed(mut timing) => {
                    timing.reroute_initiated = Instant::now();
                    #[cfg(test)]
                    if let Some(probe) = &self.probe {
                        probe
                            .reroutes
                            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    }
                    // Re-routing is initiated here, the instant the dispatcher
                    // woke from the EXIT notification. CN7 / R3: this is
                    // sub-millisecond because detection is the beamr link firing
                    // (no polling) and the wakeup is the very next step. The
                    // timing spans the real EXIT instant captured in the link
                    // handler through to this re-route initiation.
                    debug_assert!(
                        timing.detection_to_reroute() < Duration::from_millis(1),
                        "crash-to-reroute exceeded one millisecond"
                    );
                    reroute_timings.push(timing);
                    excluded.push(selected.consumer().clone());
                }
            }
        }
    }

    /// Selects a consumer for `message`, excluding `excluded`, via the routing
    /// function evaluated over the current group snapshot.
    fn select_consumer(
        &self,
        message: &RoutingMessage,
        excluded: &[ConsumerId],
    ) -> Result<ConsumerRegistration, DispatchError> {
        let snapshot = self.group.snapshot();
        let available: Vec<&ConsumerRegistration> = snapshot
            .consumers()
            .iter()
            .filter(|registration| !excluded.contains(registration.consumer()))
            .collect();
        if available.is_empty() {
            return Err(DispatchError::NoConsumerAvailable);
        }

        let state_views: Vec<ConsumerStateView> = available
            .iter()
            .map(|registration| registration.state().clone())
            .collect();
        let decision: RoutingDecision =
            self.executor
                .execute(snapshot.routing_function(), message.clone(), state_views)?;

        let Some(selected_id) = decision.selected() else {
            return Err(DispatchError::NoConsumerAvailable);
        };
        available
            .into_iter()
            .find(|registration| registration.consumer() == selected_id)
            .cloned()
            .ok_or_else(|| DispatchError::UnknownConsumerSelected(selected_id.as_str().to_owned()))
    }

    /// Links and arms one attempt before forwarding into the selected queue.
    /// Consumption and EXIT commit to the same receipt; only its owned signal
    /// wakes observation, and teardown removes its borrowed request metadata.
    fn run_attempt(
        &self,
        message: &RoutingMessage,
        selected: &ConsumerRegistration,
    ) -> Result<AttemptResult, DispatchError> {
        #[cfg(test)]
        if let Some(probe) = &self.probe {
            probe
                .attempts
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
        let consumer_pid = selected.participant();
        let actor = AttemptActor(self.supervisor.spawn(ConversationConfig::new(
            vec![consumer_pid],
            None,
            ChannelMode::Ephemeral,
            CrashPolicy::RouteToNext,
        ))?);
        actor.0.pid()?;
        #[cfg(test)]
        if let Some(probe) = &self.probe {
            let capture = probe
                .observe_permission
                .lock()
                .map_err(|error| {
                    DispatchError::Conversation(format!(
                        "observe permission lock poisoned: {error}"
                    ))
                })?
                .is_some();
            if capture {
                let (captured, receiver) = mpsc::sync_channel(1);
                actor.0.notify_on_participant_exit(consumer_pid, captured)?;
                *probe.captured_exit.lock().map_err(|error| {
                    DispatchError::Conversation(format!("EXIT capture lock poisoned: {error}"))
                })? = Some(receiver);
            }
        }
        let receipt = ConsumptionReceipt::new(consumer_pid)?;
        #[cfg(test)]
        if let Some(probe) = &self.probe {
            receipt.instrument(std::sync::Arc::clone(&probe.receipts))?;
            let probe = std::sync::Arc::clone(probe);
            receipt.observe(std::sync::Arc::new(move |event| {
                probe.record_event(event);
            }))?;
        }
        let mut registration = actor
            .0
            .register_dispatch_receipt(std::sync::Arc::clone(&receipt))?;
        if receipt.is_pending()? {
            if let Err(error) = actor
                .0
                .send_dispatch(dispatch_envelope(message)?, std::sync::Arc::clone(&receipt))
            {
                if receipt.is_pending()? {
                    return Err(error.into());
                }
            }
        }
        #[cfg(test)]
        if let Some(probe) = &self.probe {
            probe.arm_waiter(consumer_pid)?;
        }
        let result = observe_attempt(&receipt);
        #[cfg(test)]
        if let Some(probe) = &self.probe {
            probe
                .waiter_resumes
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        }
        registration.release()?;
        result
    }
}

struct AttemptActor(crate::conversation::ConversationActor);

impl Drop for AttemptActor {
    fn drop(&mut self) {
        self.0.finalize();
    }
}

fn observe_attempt(receipt: &ConsumptionReceipt) -> Result<AttemptResult, DispatchError> {
    match receipt.wait_decision()? {
        AttemptDecision::Accepted(acceptance)
            if acceptance.attempt == receipt.attempt()
                && acceptance.participant == receipt.participant() =>
        {
            Ok(AttemptResult::Delivered)
        }
        AttemptDecision::Exited(crash_observed) => Ok(AttemptResult::Crashed(RerouteTiming {
            crash_observed,
            reroute_initiated: Instant::now(),
        })),
        AttemptDecision::Closed => Err(DispatchError::Conversation(
            "dispatch closed by conversation finalization or supervisor shutdown".to_owned(),
        )),
        AttemptDecision::Abandoned => Err(DispatchError::Conversation(
            "dispatch attempt explicitly abandoned before acceptance".to_owned(),
        )),
        AttemptDecision::Pending | AttemptDecision::Accepted(_) => {
            Err(DispatchError::Conversation(
                "dispatch receipt returned an invalid or unrelated terminal decision".to_owned(),
            ))
        }
    }
}

/// Internal result of a single dispatch attempt.
#[derive(Debug)]
enum AttemptResult {
    /// The selected consumer committed correlated consumption acceptance.
    Delivered,
    /// The linked consumer's EXIT signal fired; re-route, excluding it. Carries
    /// the real crash-to-reroute timing along the event path.
    Crashed(RerouteTiming),
}

/// Builds the envelope forwarded to the selected consumer, carrying the real
/// dispatched message content (the routing message's fields), not a placeholder.
fn dispatch_envelope(message: &RoutingMessage) -> Result<crate::envelope::Envelope, DispatchError> {
    let payload = encode_message(message)?;
    Ok(crate::envelope::Envelope::new(
        payload,
        None,
        crate::channel::SchemaId::new(),
        crate::envelope::PublisherId::default(),
    ))
}

/// Serializes the dispatched message's fields to JSON bytes so the forwarded
/// envelope genuinely carries the message, not a synthetic substitute.
fn encode_message(message: &RoutingMessage) -> Result<Vec<u8>, DispatchError> {
    let map: serde_json::Map<String, serde_json::Value> = message
        .fields()
        .map(|(name, value)| (name.to_owned(), field_to_json(value)))
        .collect();
    serde_json::to_vec(&serde_json::Value::Object(map)).map_err(|error| {
        DispatchError::Conversation(format!("failed to encode dispatched message: {error}"))
    })
}

/// Maps a routing [`FieldValue`] to its JSON representation. A non-finite float
/// has no JSON number form and is encoded as null, matching `serde_json`'s own
/// handling of non-finite numbers.
fn field_to_json(value: &FieldValue) -> serde_json::Value {
    match value {
        FieldValue::Text(text) => serde_json::Value::String(text.clone()),
        FieldValue::Integer(integer) => serde_json::Value::from(*integer),
        FieldValue::Float(float) => serde_json::Number::from_f64(*float)
            .map_or(serde_json::Value::Null, serde_json::Value::Number),
        FieldValue::Boolean(boolean) => serde_json::Value::Bool(*boolean),
        FieldValue::Null => serde_json::Value::Null,
    }
}

/// Routing function that selects the first available consumer (the routing
/// function genuinely drives selection — never random).
#[cfg(test)]
fn select_first() -> crate::routing::RoutingFunction {
    use crate::routing::function::loader::{ModuleLoader, RoutingModule};
    ModuleLoader::new().load(RoutingModule::new(
        b"dispatch-first",
        |_message, consumers| {
            consumers
                .first()
                .map_or_else(RoutingDecision::none, |state| {
                    RoutingDecision::select(state.consumer.clone())
                })
        },
    ))
}

#[cfg(test)]
mod tests;

#[cfg(test)]
mod acceptance_tests;