behavior_actors/routing/

order_gate.rs

//! Explicit monotonic release gate for ordered keys.

use std::collections::BTreeMap;

use super::DeliveryOutcomes;

use behavior::{
    Actions, Address, Behavior, BehaviorActed, BehaviorBase, Never, NoBirths, Protocol,
    SendEffects, User,
};

use crate::DeliveryRoute;

/// Complete observable state of an [`OrderGate`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OrderGateState<K> {
    /// Greatest explicitly opened key, or absence before the first opening.
    pub watermark: Option<K>,
    held: usize,
}

impl<K> OrderGateState<K> {
    /// Number of values retained above the watermark.
    #[must_use]
    pub fn held(&self) -> usize {
        self.held
    }
}

/// Factual result of one gate operation.
#[derive(Debug, PartialEq, Eq)]
pub enum OrderGateOutcome<K, T> {
    /// A value is retained until the watermark reaches its key.
    Held {
        /// Accepted key.
        key: K,
        /// Current retained count.
        held: usize,
    },
    /// The key is already open, so the value was delivered immediately.
    Delivered {
        /// Accepted key.
        key: K,
    },
    /// Another retained value already owns this key.
    Duplicate {
        /// Rejected key.
        key: K,
        /// Rejected owned value.
        value: T,
    },
    /// One monotonic opening was committed.
    Opened {
        /// New watermark.
        through: K,
        /// Number of retained values released in key order.
        released: usize,
        /// Number still held.
        held: usize,
    },
    /// An opening attempted to move the watermark backwards or repeat it.
    StaleOpening {
        /// Rejected watermark.
        requested: K,
        /// Current watermark.
        current: K,
    },
}

/// Commands accepted by [`OrderGate`].
pub enum OrderGateMessage<K, T, TargetRoute, ReplyRoute> {
    /// Submit one keyed value.
    Hold {
        /// Ordered release key.
        key: K,
        /// Owned value.
        value: T,
        /// Typed delivery destination.
        to: TargetRoute,
        /// Typed outcome recipient.
        reply_to: ReplyRoute,
    },
    /// Monotonically open every key through the supplied bound.
    OpenThrough {
        /// Inclusive new watermark.
        through: K,
        /// Typed outcome recipient.
        reply_to: ReplyRoute,
    },
}

struct Held<T, TargetRoute> {
    value: T,
    to: TargetRoute,
}

/// Deterministic explicit ordered-release policy.
///
/// Before a key is open, one value may be retained for it. Opening is strictly
/// monotonic and atomically releases all retained keys at or below the new
/// watermark in key order. Values submitted at an open key are delivered
/// immediately. Duplicate holds return ownership and never overwrite. A stale
/// opening changes nothing. Initialization is empty, no actors are created,
/// and the host never terminates by policy. Inclusive watermark semantics and
/// deterministic release order are Bombay policy; physical delivery remains a
/// runtime responsibility. Transitions have no semantic panic condition.
pub struct OrderGate<
    A: Address,
    K,
    T,
    TargetRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = T>>,
    ReplyRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = OrderGateOutcome<K, T>>>,
> {
    watermark: Option<K>,
    held: BTreeMap<K, Held<T, TargetRoute>>,
    marker: core::marker::PhantomData<fn() -> (A, ReplyRoute)>,
}

impl<A, K, T, TargetRoute, ReplyRoute> OrderGate<A, K, T, TargetRoute, ReplyRoute>
where
    A: Address,
    K: Clone + Ord,
    TargetRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = T>>,
    ReplyRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = OrderGateOutcome<K, T>>>,
{
    /// Construct a closed gate with no retained values.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            watermark: None,
            held: BTreeMap::new(),
            marker: core::marker::PhantomData,
        }
    }

    /// Return the complete observable gate state.
    #[must_use]
    pub fn state(&self) -> OrderGateState<K> {
        OrderGateState {
            watermark: self.watermark.clone(),
            held: self.held.len(),
        }
    }

    fn sends(
        deliveries: TargetRoute::Sends,
        reply_to: ReplyRoute,
        outcome: OrderGateOutcome<K, T>,
    ) -> Actions<A, Never, DeliveryOutcomes<TargetRoute::Sends, ReplyRoute::Sends>, NoBirths> {
        Actions::send(DeliveryOutcomes {
            deliveries,
            outcomes: reply_to.deliver(outcome),
        })
    }

    fn hold(
        &mut self,
        key: K,
        value: T,
        to: TargetRoute,
        reply_to: ReplyRoute,
    ) -> Actions<A, Never, DeliveryOutcomes<TargetRoute::Sends, ReplyRoute::Sends>, NoBirths> {
        if self
            .watermark
            .as_ref()
            .is_some_and(|watermark| key <= *watermark)
        {
            return Self::sends(
                to.deliver(value),
                reply_to,
                OrderGateOutcome::Delivered { key },
            );
        }
        if self.held.contains_key(&key) {
            return Self::sends(
                TargetRoute::Sends::empty(),
                reply_to,
                OrderGateOutcome::Duplicate { key, value },
            );
        }
        self.held.insert(key.clone(), Held { value, to });
        Self::sends(
            TargetRoute::Sends::empty(),
            reply_to,
            OrderGateOutcome::Held {
                key,
                held: self.held.len(),
            },
        )
    }

    fn open(
        &mut self,
        through: K,
        reply_to: ReplyRoute,
    ) -> Actions<A, Never, DeliveryOutcomes<TargetRoute::Sends, ReplyRoute::Sends>, NoBirths> {
        if let Some(current) = &self.watermark
            && through <= *current
        {
            return Self::sends(
                TargetRoute::Sends::empty(),
                reply_to,
                OrderGateOutcome::StaleOpening {
                    requested: through,
                    current: current.clone(),
                },
            );
        }
        let keys = self
            .held
            .range(..=through.clone())
            .map(|(key, _)| key.clone())
            .collect::<Vec<_>>();
        let released = keys.len();
        let mut deliveries = TargetRoute::Sends::empty();
        for key in keys {
            if let Some(held) = self.held.remove(&key) {
                deliveries.append(held.to.deliver(held.value));
            }
        }
        self.watermark = Some(through.clone());
        Self::sends(
            deliveries,
            reply_to,
            OrderGateOutcome::Opened {
                through,
                released,
                held: self.held.len(),
            },
        )
    }
}

impl<A, K, T, TargetRoute, ReplyRoute> Default for OrderGate<A, K, T, TargetRoute, ReplyRoute>
where
    A: Address,
    K: Clone + Ord,
    TargetRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = T>>,
    ReplyRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = OrderGateOutcome<K, T>>>,
{
    fn default() -> Self {
        Self::new()
    }
}

impl<A, K, T, TargetRoute, ReplyRoute> BehaviorBase for OrderGate<A, K, T, TargetRoute, ReplyRoute>
where
    A: Address,
    TargetRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = T>>,
    ReplyRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = OrderGateOutcome<K, T>>>,
{
    type Base = Self;
    fn base(&self) -> &Self {
        self
    }
}

impl<A, K, T, TargetRoute, ReplyRoute> behavior::Protocol
    for OrderGate<A, K, T, TargetRoute, ReplyRoute>
where
    A: Address,
    TargetRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = T>>,
    ReplyRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = OrderGateOutcome<K, T>>>,
{
    type Addr = A;
    type Msg = OrderGateMessage<K, T, TargetRoute, ReplyRoute>;
}

impl<A, K, T, TargetRoute, ReplyRoute> Behavior for OrderGate<A, K, T, TargetRoute, ReplyRoute>
where
    A: Address,
    K: Clone + Ord,
    TargetRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = T>>,
    ReplyRoute: DeliveryRoute<Protocol: Protocol<Addr = A, Msg = OrderGateOutcome<K, T>>>,
    TargetRoute::Sends:
        behavior::SendsFor<User<A, OrderGateMessage<K, T, TargetRoute, ReplyRoute>>>,
    ReplyRoute::Sends: behavior::SendsFor<User<A, OrderGateMessage<K, T, TargetRoute, ReplyRoute>>>,
{
    type Protocol = Self;
    type Event = User<A, behavior::BehaviorMessage<Self>>;
    type Sends = DeliveryOutcomes<TargetRoute::Sends, ReplyRoute::Sends>;
    type Ph = Never;
    type Error = Never;
    type Birth = NoBirths;
    fn transition(&mut self, _: behavior::ActiveTurn, event: Self::Event) -> BehaviorActed<Self> {
        Ok(match event.message {
            OrderGateMessage::Hold {
                key,
                value,
                to,
                reply_to,
            } => self.hold(key, value, to, reply_to),
            OrderGateMessage::OpenThrough { through, reply_to } => self.open(through, reply_to),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Activate as _;
    use behavior::{Delivery, MailAddr, Recipient};
    struct Target;
    struct Reply;
    macro_rules! leaf {
        ($n:ident,$m:ty) => {
            impl behavior::Protocol for $n {
                type Addr = MailAddr;
                type Msg = $m;
            }

            impl Behavior for $n {
                type Protocol = Self;
                type Event = User<MailAddr, $m>;
                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())
                }
            }
        };
    }
    leaf!(Target, u8);
    leaf!(Reply,OrderGateOutcome<u8,u8>);
    type Subject = OrderGate<MailAddr, u8, u8, Recipient<Target>, Recipient<Reply>>;
    fn hold(
        s: &mut crate::Active<Subject>,
        key: u8,
        value: u8,
    ) -> Actions<
        MailAddr,
        Never,
        DeliveryOutcomes<Vec<Delivery<Target>>, Vec<Delivery<Reply>>>,
        NoBirths,
    > {
        s.receive(
            MailAddr(9),
            OrderGateMessage::Hold {
                key,
                value,
                to: Recipient::global(MailAddr(1)),
                reply_to: Recipient::global(MailAddr(2)),
            },
        )
        .unwrap()
    }
    fn open(
        s: &mut crate::Active<Subject>,
        through: u8,
    ) -> Actions<
        MailAddr,
        Never,
        DeliveryOutcomes<Vec<Delivery<Target>>, Vec<Delivery<Reply>>>,
        NoBirths,
    > {
        s.receive(
            MailAddr(9),
            OrderGateMessage::OpenThrough {
                through,
                reply_to: Recipient::global(MailAddr(2)),
            },
        )
        .unwrap()
    }
    #[test]
    fn opening_releases_in_key_order_and_future_open_keys_deliver_immediately() {
        let mut s = (Subject::new()).initialize().unwrap().behavior;
        let held_three = hold(&mut s, 3, 30);
        assert!(matches!(
            held_three.sends.outcomes[0].message,
            OrderGateOutcome::Held { key: 3, held: 1 }
        ));
        let held_one = hold(&mut s, 1, 10);
        assert!(matches!(
            held_one.sends.outcomes[0].message,
            OrderGateOutcome::Held { key: 1, held: 2 }
        ));
        let a = open(&mut s, 2);
        assert_eq!(
            a.sends
                .deliveries
                .iter()
                .map(|d| d.message)
                .collect::<Vec<_>>(),
            vec![10]
        );
        let delivered = hold(&mut s, 2, 20);
        assert_eq!(delivered.sends.deliveries.len(), 1);
        assert_eq!(s.state().watermark, Some(2));
        assert_eq!(s.state().held(), 1);
    }
    #[test]
    fn duplicate_and_stale_opening_are_atomic() {
        let mut s = (Subject::new()).initialize().unwrap().behavior;
        let held = hold(&mut s, 2, 20);
        assert!(matches!(
            held.sends.outcomes[0].message,
            OrderGateOutcome::Held { key: 2, held: 1 }
        ));
        let duplicate = hold(&mut s, 2, 21);
        assert!(matches!(
            duplicate.sends.outcomes[0].message,
            OrderGateOutcome::Duplicate { value: 21, .. }
        ));
        let opened = open(&mut s, 1);
        assert!(matches!(
            opened.sends.outcomes[0].message,
            OrderGateOutcome::Opened {
                through: 1,
                released: 0,
                held: 1
            }
        ));
        let stale = open(&mut s, 1);
        assert!(matches!(
            stale.sends.outcomes[0].message,
            OrderGateOutcome::StaleOpening { .. }
        ));
        assert_eq!(s.state().held, 1);
    }
}