liminal-rs 0.16.0

use std::collections::HashMap;
use std::sync::{Arc, Mutex, mpsc};
use std::time::Instant;

use super::sync::lock;
use crate::conversation::handoff::{AttemptDecision, ConsumptionNotification, ConsumptionReceipt};
use crate::conversation::types::{ParticipantHealth, ParticipantPid, ParticipantStatus};
use crate::error::LiminalError;

/// One-shot notifiers waiting on participant EXIT signals.
///
/// A dispatcher registers a notifier keyed by the participant it linked to, then
/// blocks on the matching receiver. When that participant's trapped EXIT signal
/// is processed by the actor, [`ExitNotifierRegistry::signal`] hands every
/// matching waiter the instant the link fired and drops it. This is the
/// event-driven crash-detection path: waiters are woken by the EXIT, never by
/// polling consumer liveness.
#[derive(Debug, Default)]
pub(super) struct ExitNotifierRegistry {
    notifiers: Mutex<Vec<(ParticipantPid, mpsc::SyncSender<Instant>)>>,
    dispatches: Mutex<HashMap<u64, Arc<ConsumptionReceipt>>>,
}

impl ExitNotifierRegistry {
    /// Registers `notifier` for `participant`'s EXIT, or replays an already-
    /// observed crash immediately.
    ///
    /// `participants` is the actor's recorded participant state, read by the
    /// caller under the state lock and held across this call. If `participant`
    /// is already dead there, the recorded EXIT instant is sent now and nothing
    /// is queued — closing the crash-before-register race; otherwise the
    /// notifier is queued for a future [`ExitNotifierRegistry::signal`]. Exactly
    /// one notification is delivered.
    ///
    /// # Errors
    /// Returns [`LiminalError`] when the registry lock is poisoned.
    pub(super) fn register(
        &self,
        participant: ParticipantPid,
        notifier: mpsc::SyncSender<Instant>,
        participants: &[ParticipantStatus],
    ) -> Result<Option<(mpsc::SyncSender<Instant>, Instant)>, LiminalError> {
        let already_exited_at = participants.iter().find_map(|status| {
            (status.participant == participant && status.health == ParticipantHealth::Dead)
                .then(|| status.exited_at.unwrap_or_else(Instant::now))
        });
        if let Some(exited_at) = already_exited_at {
            return Ok(Some((notifier, exited_at)));
        }
        lock(&self.notifiers, "actor exit notifiers")?.push((participant, notifier));
        Ok(None)
    }

    pub(super) fn register_dispatch(
        &self,
        receipt: Arc<ConsumptionReceipt>,
        participants: &[ParticipantStatus],
        closed: bool,
    ) -> Result<Option<ConsumptionNotification>, LiminalError> {
        if closed {
            return receipt.decide(AttemptDecision::Closed).map(Some);
        }
        if let Some(exited_at) = participants.iter().find_map(|status| {
            (status.participant == receipt.participant()
                && status.health == ParticipantHealth::Dead)
                .then(|| status.exited_at.unwrap_or_else(Instant::now))
        }) {
            return receipt.decide(AttemptDecision::Exited(exited_at)).map(Some);
        }
        let mut dispatches = lock(&self.dispatches, "actor dispatch notifiers")?;
        if dispatches.contains_key(&receipt.attempt()) {
            return Err(LiminalError::DeliveryFailed {
                message: "dispatch EXIT registration already exists".to_owned(),
            });
        }
        dispatches.insert(receipt.attempt(), receipt);
        drop(dispatches);
        Ok(None)
    }

    pub(super) fn remove_dispatch(&self, attempt: u64) -> Result<(), LiminalError> {
        lock(&self.dispatches, "actor dispatch notifiers")?.remove(&attempt);
        Ok(())
    }

    pub(super) fn close_dispatches(&self) -> ReceiptNotifications {
        let (receipts, failure) = match self.dispatches.lock() {
            Ok(mut dispatches) => (
                dispatches
                    .drain()
                    .map(|(_, receipt)| receipt)
                    .collect::<Vec<_>>(),
                None,
            ),
            Err(error) => {
                let failure = LiminalError::DeliveryFailed {
                    message: format!("actor dispatch notifier lock poisoned during close: {error}"),
                };
                let receipts = error
                    .into_inner()
                    .drain()
                    .map(|(_, receipt)| receipt)
                    .collect::<Vec<_>>();
                (receipts, Some(failure))
            }
        };
        ReceiptNotifications::new(receipts, AttemptDecision::Closed, failure)
    }

    /// Wakes every waiter registered for `participant`, handing each the instant
    /// the EXIT signal was observed, and drops them so each fires once. A waiter
    /// whose receiver was already dropped is silently discarded.
    ///
    /// # Errors
    /// Returns [`LiminalError`] when the registry lock is poisoned.
    pub(super) fn signal(
        &self,
        participant: ParticipantPid,
        observed_at: Instant,
    ) -> Result<ExitNotifications, LiminalError> {
        // Partition under the lock, then send after releasing it so the lock is
        // never held across the wakeups.
        let matched = {
            let mut notifiers = lock(&self.notifiers, "actor exit notifiers")?;
            let mut matched = Vec::new();
            let mut retained = Vec::with_capacity(notifiers.len());
            for (registered, notifier) in notifiers.drain(..) {
                if registered == participant {
                    matched.push(notifier);
                } else {
                    retained.push((registered, notifier));
                }
            }
            *notifiers = retained;
            matched
        };
        let receipts = {
            let mut dispatches = lock(&self.dispatches, "actor dispatch notifiers")?;
            let mut receipts = Vec::new();
            dispatches.retain(|_, receipt| {
                if receipt.participant() == participant {
                    receipts.push(Arc::clone(receipt));
                    false
                } else {
                    true
                }
            });
            drop(dispatches);
            receipts
        };
        let decisions =
            ReceiptNotifications::new(receipts, AttemptDecision::Exited(observed_at), None);
        Ok(ExitNotifications {
            matched,
            decisions,
            observed_at,
        })
    }
}

pub(super) struct ExitNotifications {
    matched: Vec<mpsc::SyncSender<Instant>>,
    decisions: ReceiptNotifications,
    observed_at: Instant,
}

impl ExitNotifications {
    pub(super) fn publish(self) -> Result<(), LiminalError> {
        let decisions = self.decisions.publish();
        for notifier in self.matched {
            if let Err(error) = notifier.try_send(self.observed_at) {
                tracing::debug!(%error, "participant EXIT receiver no longer accepts its notification");
            }
        }
        decisions
    }
}

type ReceiptSettlement = (
    Arc<ConsumptionReceipt>,
    Result<ConsumptionNotification, LiminalError>,
);

pub(super) struct ReceiptNotifications {
    decisions: Vec<ReceiptSettlement>,
    failure: Option<LiminalError>,
}

impl ReceiptNotifications {
    fn new(
        receipts: Vec<Arc<ConsumptionReceipt>>,
        terminal: AttemptDecision,
        failure: Option<LiminalError>,
    ) -> Self {
        let decisions = receipts
            .into_iter()
            .map(|receipt| {
                let decision = receipt.decide(terminal);
                (receipt, decision)
            })
            .collect();
        Self { decisions, failure }
    }

    pub(super) fn publish(self) -> Result<(), LiminalError> {
        let mut failure = self.failure;
        for (receipt, decision) in self.decisions {
            match decision {
                Ok(notification) => notification.publish(),
                Err(error) => {
                    tracing::error!(%error, "dispatch terminal decision failed");
                    receipt.wake_failure();
                    if failure.is_none() {
                        failure = Some(error);
                    }
                }
            }
        }
        failure.map_or(Ok(()), Err)
    }
}