liminal-rs 0.16.0

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex, Weak};
use std::time::Instant;

use super::ParticipantPid;
use super::actor::ActorCore;
use super::participant::Mailbox;
use crate::error::LiminalError;

static NEXT_ATTEMPT: AtomicU64 = AtomicU64::new(1);

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ConsumptionAcceptance {
    pub(crate) attempt: u64,
    pub(crate) participant: ParticipantPid,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum AttemptDecision {
    Pending,
    Accepted(ConsumptionAcceptance),
    Exited(Instant),
    Closed,
    Abandoned,
}

#[cfg(test)]
#[derive(Clone, Copy, Debug)]
pub enum ReceiptEvent {
    Acceptance(ConsumptionAcceptance),
    Exit,
    Terminal,
    Registration,
    Release,
    /// The decision waiter woke and found the attempt still pending: what
    /// an armed timer or a polling loop produces, and nothing else does.
    IdleWakeup,
    /// The decision waiter parked again after an idle wakeup: a re-armed
    /// wait.
    Rearm,
}

#[cfg(test)]
#[derive(Debug, Default)]
pub struct ReceiptMetrics {
    pub(crate) acceptances: AtomicU64,
    pub(crate) exits: AtomicU64,
    pub(crate) terminals: AtomicU64,
    pub(crate) registrations: AtomicU64,
    pub(crate) releases: AtomicU64,
    /// Wakeups of [`ConsumptionReceipt::wait_decision`] that found the
    /// attempt still pending, counted at the wait itself.
    pub(crate) idle_wakeups: AtomicU64,
    /// Parks of [`ConsumptionReceipt::wait_decision`] after an idle wakeup.
    pub(crate) rearms: AtomicU64,
}

#[cfg(test)]
type ReceiptObserver = Arc<dyn Fn(ReceiptEvent) + Send + Sync>;

pub struct ConsumptionReceipt {
    attempt: u64,
    participant: ParticipantPid,
    decision: Mutex<AttemptDecision>,
    ready: Condvar,
    forwarding: Mutex<Option<Weak<Mailbox>>>,
    #[cfg(test)]
    metrics: std::sync::OnceLock<Arc<ReceiptMetrics>>,
    #[cfg(test)]
    observer: Mutex<Option<ReceiptObserver>>,
}

impl std::fmt::Debug for ConsumptionReceipt {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ConsumptionReceipt")
            .field("attempt", &self.attempt)
            .field("participant", &self.participant)
            .finish_non_exhaustive()
    }
}

impl ConsumptionReceipt {
    pub(crate) fn new(participant: ParticipantPid) -> Result<Arc<Self>, LiminalError> {
        let attempt = NEXT_ATTEMPT
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |id| id.checked_add(1))
            .map_err(|_| failure("dispatch attempt identity exhausted"))?;
        Ok(Arc::new(Self {
            attempt,
            participant,
            decision: Mutex::new(AttemptDecision::Pending),
            ready: Condvar::new(),
            forwarding: Mutex::new(None),
            #[cfg(test)]
            metrics: std::sync::OnceLock::new(),
            #[cfg(test)]
            observer: Mutex::new(None),
        }))
    }

    pub(crate) const fn attempt(&self) -> u64 {
        self.attempt
    }

    pub(crate) const fn participant(&self) -> ParticipantPid {
        self.participant
    }

    // Both admission and consumption hold inbox before decision. Cleanup takes
    // forwarding metadata alone, releases it, then takes inbox; it never holds
    // decision while taking inbox. Actor state and registry never enter inbox.
    pub(super) fn enqueue(
        &self,
        inbox: &Arc<Mailbox>,
        forward: impl FnOnce() -> Result<(), LiminalError>,
    ) -> Result<(), LiminalError> {
        let decision = self
            .decision
            .lock()
            .map_err(|error| failure(format!("dispatch decision lock poisoned: {error}")))?;
        if *decision != AttemptDecision::Pending {
            return Err(failure("dispatch attempt settled before forwarding"));
        }
        let mut forwarding = self
            .forwarding
            .lock()
            .map_err(|error| failure(format!("dispatch forwarding lock poisoned: {error}")))?;
        if forwarding.is_some() {
            return Err(failure("dispatch attempt was already forwarded"));
        }
        *forwarding = Some(Arc::downgrade(inbox));
        drop(forwarding);
        let result = forward();
        drop(decision);
        result
    }

    pub(super) fn consume<T>(
        self: &Arc<Self>,
        participant: ParticipantPid,
        take: impl FnOnce() -> Option<T>,
    ) -> Result<Option<(T, ConsumptionNotification)>, LiminalError> {
        if participant != self.participant {
            return Err(failure(
                "dispatch acceptance participant does not match its attempt",
            ));
        }
        let mut decision = self
            .decision
            .lock()
            .map_err(|error| failure(format!("dispatch decision lock poisoned: {error}")))?;
        if *decision != AttemptDecision::Pending {
            return Ok(None);
        }
        let Some(request) = take() else {
            return Ok(None);
        };
        let acceptance = ConsumptionAcceptance {
            attempt: self.attempt,
            participant,
        };
        let terminal = AttemptDecision::Accepted(acceptance);
        *decision = terminal;
        #[cfg(test)]
        {
            self.count(ReceiptEvent::Acceptance(acceptance));
            self.count(ReceiptEvent::Terminal);
        }
        drop(decision);
        Ok(Some((
            request,
            ConsumptionNotification {
                receipt: Arc::clone(self),
                terminal,
                changed: true,
            },
        )))
    }

    pub(crate) fn decide(
        self: &Arc<Self>,
        terminal: AttemptDecision,
    ) -> Result<ConsumptionNotification, LiminalError> {
        if matches!(
            terminal,
            AttemptDecision::Pending | AttemptDecision::Accepted(_)
        ) {
            return Err(failure("dispatch external terminal decision is invalid"));
        }
        let mut decision = self
            .decision
            .lock()
            .map_err(|error| failure(format!("dispatch decision lock poisoned: {error}")))?;
        let changed = *decision == AttemptDecision::Pending;
        #[cfg(test)]
        if matches!(terminal, AttemptDecision::Exited(_)) {
            self.count(ReceiptEvent::Exit);
        }
        if changed {
            *decision = terminal;
            #[cfg(test)]
            self.count(ReceiptEvent::Terminal);
        }
        Ok(ConsumptionNotification {
            receipt: Arc::clone(self),
            terminal: *decision,
            changed,
        })
    }

    pub(crate) fn is_pending(&self) -> Result<bool, LiminalError> {
        self.decision
            .lock()
            .map(|decision| *decision == AttemptDecision::Pending)
            .map_err(|error| failure(format!("dispatch decision lock poisoned: {error}")))
    }

    pub(crate) fn wake_failure(&self) {
        self.ready.notify_all();
    }

    #[cfg(test)]
    pub(crate) fn instrument(&self, metrics: Arc<ReceiptMetrics>) -> Result<(), LiminalError> {
        self.metrics
            .set(metrics)
            .map_err(|_| failure("dispatch receipt metrics are already installed"))
    }

    #[cfg(test)]
    pub(crate) fn count(&self, event: ReceiptEvent) {
        if let Some(metrics) = self.metrics.get() {
            let counter = match event {
                ReceiptEvent::Acceptance(_) => &metrics.acceptances,
                ReceiptEvent::Exit => &metrics.exits,
                ReceiptEvent::Terminal => &metrics.terminals,
                ReceiptEvent::Registration => &metrics.registrations,
                ReceiptEvent::Release => &metrics.releases,
                ReceiptEvent::IdleWakeup => &metrics.idle_wakeups,
                ReceiptEvent::Rearm => &metrics.rearms,
            };
            counter.fetch_add(1, Ordering::Relaxed);
        }
    }

    pub(crate) fn wait_decision(&self) -> Result<AttemptDecision, LiminalError> {
        let mut decision = self
            .decision
            .lock()
            .map_err(|error| failure(format!("dispatch decision lock poisoned: {error}")))?;
        #[cfg(test)]
        let mut woken_idle = false;
        while *decision == AttemptDecision::Pending {
            #[cfg(test)]
            if woken_idle {
                self.count(ReceiptEvent::Rearm);
            }
            decision = self
                .ready
                .wait(decision)
                .map_err(|error| failure(format!("dispatch decision wait poisoned: {error}")))?;
            #[cfg(test)]
            {
                woken_idle = *decision == AttemptDecision::Pending;
                if woken_idle {
                    self.count(ReceiptEvent::IdleWakeup);
                }
            }
        }
        Ok(*decision)
    }

    pub(crate) fn discard_request(&self) -> Result<(), LiminalError> {
        let inbox = self
            .forwarding
            .lock()
            .map_err(|error| failure(format!("dispatch forwarding lock poisoned: {error}")))?
            .take()
            .and_then(|inbox| inbox.upgrade());
        if let Some(inbox) = inbox {
            super::participant::discard_dispatch(&inbox, self.attempt)?;
        }
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn observe(&self, observer: ReceiptObserver) -> Result<(), LiminalError> {
        let mut slot = self
            .observer
            .lock()
            .map_err(|error| failure(format!("dispatch observer lock poisoned: {error}")))?;
        if slot.is_some() {
            return Err(failure(
                "dispatch acceptance observer is already registered",
            ));
        }
        *slot = Some(observer);
        drop(slot);
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn record(&self, event: ReceiptEvent) {
        let observer = match self.observer.lock() {
            Ok(observer) => observer.clone(),
            Err(error) => {
                tracing::error!(%error, "dispatch observer lock poisoned");
                return;
            }
        };
        if let Some(observer) = observer {
            observer(event);
        }
    }

    #[cfg(test)]
    pub(crate) fn snapshot(&self) -> Result<Option<ConsumptionAcceptance>, LiminalError> {
        let decision = self
            .decision
            .lock()
            .map_err(|error| failure(format!("dispatch decision lock poisoned: {error}")))?;
        Ok(match *decision {
            AttemptDecision::Accepted(acceptance) => Some(acceptance),
            _ => None,
        })
    }

    #[cfg(test)]
    pub(crate) fn wait(&self) -> Result<ConsumptionAcceptance, LiminalError> {
        match self.wait_decision()? {
            AttemptDecision::Accepted(acceptance) => Ok(acceptance),
            _ => Err(failure("dispatch ended before consumption acceptance")),
        }
    }
}

pub struct ConsumptionNotification {
    receipt: Arc<ConsumptionReceipt>,
    terminal: AttemptDecision,
    changed: bool,
}

impl ConsumptionNotification {
    pub(crate) fn publish(self) {
        if self.changed {
            tracing::trace!(attempt = self.receipt.attempt, ?self.terminal, "dispatch attempt settled");
            #[cfg(test)]
            if let AttemptDecision::Accepted(acceptance) = self.terminal {
                self.receipt.record(ReceiptEvent::Acceptance(acceptance));
            }
            self.receipt.ready.notify_all();
        }
    }
}

pub struct DispatchRegistration {
    core: Weak<ActorCore>,
    receipt: Arc<ConsumptionReceipt>,
    released: bool,
}

impl DispatchRegistration {
    pub(crate) fn new(core: &Arc<ActorCore>, receipt: Arc<ConsumptionReceipt>) -> Self {
        Self {
            core: Arc::downgrade(core),
            receipt,
            released: false,
        }
    }

    pub(crate) fn release(&mut self) -> Result<(), LiminalError> {
        if self.released {
            return Ok(());
        }
        self.released = true;
        let removal = if let Some(core) = self.core.upgrade() {
            core.remove_dispatch_receipt(self.receipt.attempt)
        } else {
            Ok(())
        };
        let terminal = self.receipt.decide(AttemptDecision::Abandoned);
        let discard = self.receipt.discard_request();
        let mut failure = None;
        for result in [removal, discard] {
            if let Err(error) = result {
                tracing::error!(%error, "dispatch request or registration release failed");
                if failure.is_none() {
                    failure = Some(error);
                }
            }
        }
        match terminal {
            Ok(notification) => notification.publish(),
            Err(error) => {
                self.receipt.wake_failure();
                tracing::error!(%error, "abandoning dispatch terminal decision failed");
                if failure.is_none() {
                    failure = Some(error);
                }
            }
        }
        #[cfg(test)]
        self.receipt.count(ReceiptEvent::Release);
        failure.map_or(Ok(()), Err)
    }
}

impl Drop for DispatchRegistration {
    fn drop(&mut self) {
        if let Err(error) = self.release() {
            tracing::error!(%error, "dispatch registration cleanup failed");
        }
    }
}

fn failure(message: impl Into<String>) -> LiminalError {
    LiminalError::DeliveryFailed {
        message: message.into(),
    }
}