behavior_actors/routing/

correlator.rs

//! Keyed request/result lifecycle correlation.

use behavior::{Actions, Address, Behavior, BehaviorActed, BehaviorBase, Never, NoBirths, User};
#[cfg(test)]
use behavior::{Delivery, Recipient};
use thiserror::Error;

use crate::DeliveryRoute;

/// Result delivered to the reply recipient retained by [`Correlator`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CorrelationResult<K, V> {
    /// A matching pending key accepted one value.
    Resolved { key: K, value: V },
    /// A matching pending key was explicitly cancelled.
    Cancelled { key: K },
}

/// Complete retained lifecycle of one correlation key.
pub enum CorrelationState<K, Route> {
    /// The key owns one reply recipient and may accept resolution or cancel.
    Pending {
        /// Correlation key.
        key: K,
        /// Typed recipient for the terminal result.
        reply_to: Route,
    },
    /// The key resolved and later replies are stale.
    Completed { key: K },
    /// The key was cancelled and later replies are stale.
    Cancelled { key: K },
}

impl<K, Route> CorrelationState<K, Route> {
    fn key(&self) -> &K {
        match self {
            Self::Pending { key, .. } | Self::Completed { key } | Self::Cancelled { key } => key,
        }
    }
}

/// Commands accepted by [`Correlator`].
pub enum CorrelatorMessage<K, V, Route> {
    /// Establish a unique pending key and its terminal reply recipient.
    Begin { key: K, reply_to: Route },
    /// Resolve one pending key with an owned value.
    Resolve { key: K, value: V },
    /// Cancel one pending key.
    Cancel { key: K },
}

/// A rejected correlation transition.
#[derive(Debug, Error, Clone, PartialEq, Eq)]
pub enum CorrelatorError<K, V, Route> {
    /// The key is already pending.
    #[error("correlation key is already pending")]
    AlreadyPending { key: K, reply_to: Route },
    /// A completed key cannot be opened again without an explicit retention policy.
    #[error("correlation key is already completed")]
    ReopenCompleted { key: K, reply_to: Route },
    /// A cancelled key cannot be opened again without an explicit retention policy.
    #[error("correlation key is already cancelled")]
    ReopenCancelled { key: K, reply_to: Route },
    /// An operation targeted an already completed lifecycle.
    #[error("correlation key is already completed")]
    AlreadyCompleted(K),
    /// An operation targeted an already cancelled lifecycle.
    #[error("correlation key is already cancelled")]
    AlreadyCancelled(K),
    /// No lifecycle fact exists for the supplied key.
    #[error("correlation key is unknown")]
    Unknown(K),
    /// No lifecycle fact exists for a reply; the owned value is returned.
    #[error("correlation reply key is unknown")]
    UnknownReply { key: K, value: V },
    /// A reply arrived after successful completion; the owned value is returned.
    #[error("correlation reply is stale because the key completed")]
    StaleCompleted { key: K, value: V },
    /// A reply arrived after cancellation; the owned value is returned.
    #[error("correlation reply is stale because the key was cancelled")]
    StaleCancelled { key: K, value: V },
}

/// Deterministic keyed request/result lifecycle behavior.
///
/// Each key is exactly pending, completed, or cancelled. `Begin` owns the
/// reply recipient. `Resolve` atomically marks a pending key completed and
/// emits one typed result; `Cancel` marks it cancelled and emits one typed
/// cancellation. Terminal keys are retained, making duplicate and stale input
/// explicit rather than indistinguishable from never-seen input. Reopening and
/// retention expiry require a separate explicit policy and are not inferred
/// from time or key reuse. Initialization is empty and the behavior does not
/// terminate itself. These lifecycle and retention choices are Bombay policy;
/// delivery remains an Address/Communication capability. No method has a
/// semantic panic condition.
pub struct Correlator<
    A: Address,
    K,
    V,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = CorrelationResult<K, V>>>,
> {
    states: Vec<CorrelationState<K, Route>>,
    address: core::marker::PhantomData<fn() -> A>,
}

impl<A, K, V, Route> Correlator<A, K, V, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = CorrelationResult<K, V>>>,
{
    /// Construct an empty correlator definition.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            states: Vec::new(),
            address: core::marker::PhantomData,
        }
    }

    /// Borrow every retained lifecycle in first-begin order.
    #[must_use]
    pub fn states(&self) -> &[CorrelationState<K, Route>] {
        &self.states
    }
}

impl<A, K, V, Route> Default for Correlator<A, K, V, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = CorrelationResult<K, V>>>,
{
    fn default() -> Self {
        Self::new()
    }
}

impl<A, K, V, Route> BehaviorBase for Correlator<A, K, V, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = CorrelationResult<K, V>>>,
{
    type Base = Self;

    fn base(&self) -> &Self {
        self
    }
}

impl<A, K, V, Route> behavior::Protocol for Correlator<A, K, V, Route>
where
    A: Address,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = CorrelationResult<K, V>>>,
{
    type Addr = A;
    type Msg = CorrelatorMessage<K, V, Route>;
}

impl<A, K, V, Route> Behavior for Correlator<A, K, V, Route>
where
    A: Address,
    K: Clone + Eq,
    Route: DeliveryRoute<Protocol: behavior::Protocol<Addr = A, Msg = CorrelationResult<K, V>>>
        + Clone,
    Route::Sends: behavior::SendsFor<User<A, CorrelatorMessage<K, V, Route>>>,
{
    type Protocol = Self;
    type Event = User<A, behavior::BehaviorMessage<Self>>;
    type Sends = Route::Sends;
    type Ph = Never;
    type Error = CorrelatorError<K, V, Route>;
    type Birth = NoBirths;

    fn transition(&mut self, _: behavior::ActiveTurn, event: Self::Event) -> BehaviorActed<Self> {
        match event.message {
            CorrelatorMessage::Begin { key, reply_to } => {
                if let Some(existing) = self.states.iter().find(|state| state.key() == &key) {
                    return Err(match existing {
                        CorrelationState::Pending { .. } => {
                            CorrelatorError::AlreadyPending { key, reply_to }
                        }
                        CorrelationState::Completed { .. } => {
                            CorrelatorError::ReopenCompleted { key, reply_to }
                        }
                        CorrelationState::Cancelled { .. } => {
                            CorrelatorError::ReopenCancelled { key, reply_to }
                        }
                    });
                }
                self.states
                    .push(CorrelationState::Pending { key, reply_to });
                Ok(Actions::cont())
            }
            CorrelatorMessage::Resolve { key, value } => {
                let Some(index) = self.states.iter().position(|state| state.key() == &key) else {
                    return Err(CorrelatorError::UnknownReply { key, value });
                };
                let reply_to = match &self.states[index] {
                    CorrelationState::Pending { reply_to, .. } => reply_to.clone(),
                    CorrelationState::Completed { .. } => {
                        return Err(CorrelatorError::StaleCompleted { key, value });
                    }
                    CorrelationState::Cancelled { .. } => {
                        return Err(CorrelatorError::StaleCancelled { key, value });
                    }
                };
                self.states[index] = CorrelationState::Completed { key: key.clone() };
                Ok(Actions::send(
                    reply_to.deliver(CorrelationResult::Resolved { key, value }),
                ))
            }
            CorrelatorMessage::Cancel { key } => {
                let Some(index) = self.states.iter().position(|state| state.key() == &key) else {
                    return Err(CorrelatorError::Unknown(key));
                };
                let reply_to = match &self.states[index] {
                    CorrelationState::Pending { reply_to, .. } => reply_to.clone(),
                    CorrelationState::Completed { .. } => {
                        return Err(CorrelatorError::AlreadyCompleted(key));
                    }
                    CorrelationState::Cancelled { .. } => {
                        return Err(CorrelatorError::AlreadyCancelled(key));
                    }
                };
                self.states[index] = CorrelationState::Cancelled { key: key.clone() };
                Ok(Actions::send(
                    reply_to.deliver(CorrelationResult::Cancelled { key }),
                ))
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Activate as _;
    use behavior::MailAddr;

    struct Reply;

    impl behavior::Protocol for Reply {
        type Addr = MailAddr;
        type Msg = CorrelationResult<u8, u16>;
    }

    impl Behavior for Reply {
        type Protocol = Self;
        type Event = User<MailAddr, behavior::BehaviorMessage<Self>>;
        type Sends = Vec<Never>;
        type Ph = Never;
        type Error = Never;
        type Birth = NoBirths;

        fn transition(&mut self, _: behavior::ActiveTurn, _: Self::Event) -> BehaviorActed<Self> {
            Ok(Actions::cont())
        }
    }

    type TestCorrelator = Correlator<MailAddr, u8, u16, Recipient<Reply>>;

    #[test]
    fn resolve_commits_before_one_terminal_delivery_and_marks_duplicates_stale() {
        let reply = Recipient::<Reply>::global(MailAddr(8));
        let mut correlator = (TestCorrelator::new()).initialize().unwrap().behavior;
        let begun = correlator
            .receive(
                MailAddr(9),
                CorrelatorMessage::Begin {
                    key: 1,
                    reply_to: reply,
                },
            )
            .unwrap();
        assert!(begun.sends.is_empty());
        assert!(begun.creates.is_empty());
        assert_eq!(begun.become_, behavior::Step::Continue);
        let resolved = correlator
            .receive(
                MailAddr(9),
                CorrelatorMessage::Resolve { key: 1, value: 42 },
            )
            .unwrap();
        assert!(
            resolved.sends
                == vec![Delivery::new(
                    reply,
                    CorrelationResult::Resolved { key: 1, value: 42 },
                )]
        );
        assert!(resolved.creates.is_empty());
        assert!(matches!(
            correlator.states(),
            [CorrelationState::Completed { key: 1 }]
        ));
        let rejection = correlator.receive(
            MailAddr(9),
            CorrelatorMessage::Resolve { key: 1, value: 43 },
        );
        assert!(matches!(
            rejection,
            Err(CorrelatorError::StaleCompleted { key: 1, value: 43 })
        ));
    }

    #[test]
    fn cancel_is_terminal_and_unknown_reply_preserves_the_value() {
        let reply = Recipient::<Reply>::global(MailAddr(8));
        let mut correlator = (TestCorrelator::new()).initialize().unwrap().behavior;
        let rejection = correlator.receive(
            MailAddr(9),
            CorrelatorMessage::Resolve { key: 7, value: 99 },
        );
        assert!(matches!(
            rejection,
            Err(CorrelatorError::UnknownReply { key: 7, value: 99 })
        ));
        let begun = correlator
            .receive(
                MailAddr(9),
                CorrelatorMessage::Begin {
                    key: 2,
                    reply_to: reply,
                },
            )
            .unwrap();
        assert!(begun.sends.is_empty());
        assert!(begun.creates.is_empty());
        assert_eq!(begun.become_, behavior::Step::Continue);
        let cancelled = correlator
            .receive(MailAddr(9), CorrelatorMessage::Cancel { key: 2 })
            .unwrap();
        assert!(
            cancelled.sends
                == vec![Delivery::new(
                    reply,
                    CorrelationResult::Cancelled { key: 2 },
                )]
        );
        let rejection = correlator.receive(
            MailAddr(9),
            CorrelatorMessage::Resolve { key: 2, value: 100 },
        );
        assert!(matches!(
            rejection,
            Err(CorrelatorError::StaleCancelled { key: 2, value: 100 })
        ));
    }
}