liminal-rs 0.16.3

//! Real conversation participant: a beamr-supervised native process that
//! genuinely processes the messages a conversation forwards to it.
//!
//! Seam #2 spawned an INERT participant (`Scheduler::spawn_test_process`): a real
//! supervised, linked process — so crash detection worked — but it processed NO
//! messages. This module replaces that stand-in with a [`ParticipantProcess`], a
//! real [`NativeHandler`] (mirroring the connection-process pattern in
//! `liminal-server`). When the conversation actor forwards an envelope, the
//! participant is woken, drains the forwarded envelope from a shared Rust-side
//! queue (the same shared-queue + atom-wakeup mechanism `ActorCore` uses to move
//! payloads to a beamr process), runs application logic behind the clean
//! [`ParticipantBehaviour`] trait, and delivers any reply back into the
//! conversation so it flows to the caller.

use std::collections::HashMap;
use std::collections::VecDeque;
use std::collections::hash_map::Entry;
use std::sync::{Arc, Mutex, Weak};

use beamr::atom::Atom;
use beamr::native::native_process::{NativeContext, NativeHandler, NativeOutcome};
use beamr::scheduler::Scheduler;

use super::actor::ActorCore;
use super::handoff::ConsumptionReceipt;
use crate::conversation::types::ParticipantPid;
use crate::envelope::Envelope;
use crate::error::LiminalError;

/// Application-supplied behaviour for a conversation participant.
///
/// This is the clean extension point: a participant receives each forwarded
/// request [`Envelope`] and returns an optional reply [`Envelope`]. Returning
/// `Some` makes this a request-reply participant (the reply flows back to the
/// caller through the conversation); returning `None` consumes the message
/// without replying. Implementations run on a beamr scheduler worker, so they
/// must be `Send + Sync` and should not block.
pub trait ParticipantBehaviour: Send + Sync + std::fmt::Debug {
    /// Processes one forwarded request, optionally producing a reply.
    fn process(&self, request: &Envelope) -> Option<Envelope>;
}

/// Reference request-reply participant.
///
/// Echoes each request straight back as the reply, preserving the payload and
/// schema. Concrete, real processing — not a stub — and the natural first
/// pattern from LIM-005 (request-reply).
#[derive(Clone, Copy, Debug, Default)]
pub struct EchoBehaviour;

impl ParticipantBehaviour for EchoBehaviour {
    fn process(&self, request: &Envelope) -> Option<Envelope> {
        Some(request.clone())
    }
}

/// Shared state bridging the conversation actor and its native participants.
///
/// Owns, per participant pid: the forwarded-request queue the actor pushes into,
/// the application [`ParticipantBehaviour`], and a weak handle to the owning
/// [`ActorCore`] so a produced reply can be delivered back into the conversation.
/// This is the participant-side analogue of `liminal-server`'s `ConnectionRuntime`
/// — a shared Rust-side queue read by a native handler that is woken by an atom
/// message, never an invented side channel.
#[derive(Debug, Default)]
pub(super) struct ParticipantRuntime {
    registrations: Mutex<HashMap<u64, ParticipantRegistration>>,
}

/// Where a participant-produced reply belongs.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum ReplyExpectation {
    /// Ordinary liminal `send`/`receive` usage, outside the server correlation table.
    Local,
    /// A server reply-requested operation, correlated by its connection-local id.
    Operation(u64),
    /// Fire-and-forget traffic: behaviour may produce a value, but the runner drops it.
    None,
}

/// One participant request with its in-process reply-routing marker.
#[derive(Debug)]
pub(super) struct ForwardedRequest {
    envelope: Envelope,
    reply_expectation: ReplyExpectation,
    handoff: Option<Arc<ConsumptionReceipt>>,
}

/// Shared forwarding queue of requests awaiting processing by a participant.
pub(super) type Mailbox = Mutex<VecDeque<ForwardedRequest>>;
type RequestQueue = Arc<Mailbox>;

/// A participant snapshot: its forwarding queue, behaviour, and owning core.
type ParticipantSnapshot = (RequestQueue, Arc<dyn ParticipantBehaviour>, Weak<ActorCore>);

struct ParticipantRegistration {
    inbox: RequestQueue,
    behaviour: Arc<dyn ParticipantBehaviour>,
    core: Weak<ActorCore>,
    #[cfg(test)]
    consume_gate: Option<Arc<ConsumeGate>>,
}

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

impl ParticipantRuntime {
    /// Registers a participant pid with its forwarding queue, behaviour, and the
    /// actor core that owns the conversation, so forwarded requests can be
    /// processed and replies delivered back.
    ///
    /// # Errors
    /// Returns [`LiminalError`] when the registry lock is poisoned or when the
    /// pid is already registered to a live conversation.
    pub(super) fn register(
        &self,
        pid: ParticipantPid,
        inbox: RequestQueue,
        behaviour: Arc<dyn ParticipantBehaviour>,
        core: Weak<ActorCore>,
    ) -> Result<(), LiminalError> {
        let registration = ParticipantRegistration {
            inbox,
            behaviour,
            core,
            #[cfg(test)]
            consume_gate: None,
        };
        match lock(&self.registrations)?.entry(pid.get()) {
            Entry::Occupied(mut entry) => {
                // A colliding key whose owning core is LIVE is a supervision
                // defect (beamr pids are unique among live processes): refuse it
                // loudly rather than silently rebinding forwarding/reply routing
                // to a different conversation.
                if entry.get().core.strong_count() > 0 {
                    return Err(LiminalError::ConversationFailed {
                        message: format!(
                            "participant pid {} is already registered to a live conversation",
                            pid.get()
                        ),
                    });
                }
                // Dead-core colliding key: beamr recycled the pid to this new
                // spawn, so the old PROCESS is provably gone and the stale entry
                // cannot belong to any live conversation — replace it. Only the
                // COLLIDING key is swept here: the global dead-key sweep stays
                // with `reap_orphans`, which also terminates each orphan's still
                // parked process — something register (no scheduler handle)
                // cannot do, so a blanket retain here would leak those parked
                // processes by hiding their entries from the reaper.
                entry.insert(registration);
                Ok(())
            }
            Entry::Vacant(entry) => {
                entry.insert(registration);
                Ok(())
            }
        }
    }

    /// Drops a participant registration (used when a participant process exits).
    pub(super) fn deregister(&self, pid: ParticipantPid) {
        if let Ok(mut registrations) = self.registrations.lock() {
            registrations.remove(&pid.get());
        }
    }

    /// Drops every registration whose owning actor core is gone, terminating the
    /// registered participant process first so an unreachable conversation (core
    /// dropped without close) cannot leave its participants parked. Called from
    /// the actor-exit watcher, so removal is exit-driven, not touch-driven.
    pub(super) fn reap_orphans(&self, scheduler: &Scheduler) {
        if let Ok(mut registrations) = self.registrations.lock() {
            registrations.retain(|&pid, registration| {
                if registration.core.strong_count() > 0 {
                    return true;
                }
                scheduler.terminate_process(pid, beamr::process::ExitReason::Normal);
                false
            });
        }
    }

    /// Number of live participant registrations. A closed conversation must leave
    /// none, so the churn gate pins this bounded across open/close cycles.
    pub(super) fn registration_count(&self) -> usize {
        self.registrations
            .lock()
            .map_or(0, |registrations| registrations.len())
    }

    /// Drains and processes every request currently queued for `pid`, delivering
    /// each produced reply back into the owning conversation. Returns the number
    /// of requests processed this slice.
    fn run_slice(&self, pid: u64) -> Result<usize, LiminalError> {
        let Some((inbox, behaviour, core)) = self.snapshot(pid) else {
            return Ok(0);
        };
        #[cfg(test)]
        let gate = self.consume_gate(pid)?;
        #[cfg(test)]
        if let Some(gate) = &gate {
            if !gate.permitted.load(std::sync::atomic::Ordering::Acquire) {
                if !gate.parked.swap(true, std::sync::atomic::Ordering::AcqRel) {
                    if let Err(error) = gate.parked_signal.send(()) {
                        tracing::error!(%error, "consumer parked signal disconnected");
                    }
                }
                return Ok(0);
            }
        }
        let mut processed = 0;
        loop {
            #[cfg(test)]
            if let Some(gate) = &gate {
                if gate.taken.load(std::sync::atomic::Ordering::Acquire)
                    >= gate.limit.load(std::sync::atomic::Ordering::Acquire)
                {
                    return Ok(processed);
                }
            }
            let (request, notification) = {
                let mut queue = lock(&inbox)?;
                let receipt = queue.front().and_then(|request| request.handoff.clone());
                let result = if let Some(receipt) = receipt {
                    if let Some((request, notification)) =
                        receipt.consume(ParticipantPid::new(pid), || {
                            let request = queue.pop_front();
                            #[cfg(test)]
                            if request.is_some() {
                                if let Some(gate) = &gate {
                                    gate.taken
                                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                                }
                            }
                            request
                        })?
                    {
                        (Some(request), Some(notification))
                    } else {
                        // A terminal attempt's queued request is discarded
                        // without crossing the application consumption boundary.
                        drop(queue.pop_front());
                        drop(queue);
                        continue;
                    }
                } else {
                    let request = queue.pop_front();
                    #[cfg(test)]
                    if request.is_some() {
                        if let Some(gate) = &gate {
                            gate.taken
                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        }
                    }
                    (request, None)
                };
                drop(queue);
                result
            };
            let Some(request) = request else { break };
            processed += 1;
            if let Some(notification) = notification {
                notification.publish();
            }
            if let Some(reply) = behaviour.process(&request.envelope) {
                if request.reply_expectation == ReplyExpectation::None {
                    continue;
                }
                if let Some(core) = core.upgrade() {
                    // Deliver the reply into the conversation. A failure here means
                    // the conversation is already closed/failed; the reply is then
                    // genuinely undeliverable and dropping it is correct.
                    let _ = core.deliver_participant_reply(request.reply_expectation, reply);
                }
            }
        }
        Ok(processed)
    }

    #[cfg(test)]
    pub(super) fn hold_consumption(
        &self,
        pid: ParticipantPid,
        scheduler: Arc<Scheduler>,
        wakeup: Atom,
    ) -> Result<(ConsumptionPermit, std::sync::mpsc::Receiver<()>), LiminalError> {
        let (parked_signal, parked) = std::sync::mpsc::channel();
        let gate = Arc::new(ConsumeGate {
            permitted: std::sync::atomic::AtomicBool::new(false),
            parked: std::sync::atomic::AtomicBool::new(false),
            taken: std::sync::atomic::AtomicU64::new(0),
            limit: std::sync::atomic::AtomicU64::new(0),
            parked_signal,
        });
        let mut registrations = lock(&self.registrations)?;
        let registration =
            registrations
                .get_mut(&pid.get())
                .ok_or_else(|| LiminalError::DeliveryFailed {
                    message: format!(
                        "participant {} has no registered forwarding queue",
                        pid.get()
                    ),
                })?;
        registration.consume_gate = Some(Arc::clone(&gate));
        drop(registrations);
        Ok((
            ConsumptionPermit {
                pid,
                scheduler,
                wakeup,
                gate,
            },
            parked,
        ))
    }

    #[cfg(test)]
    fn consume_gate(&self, pid: u64) -> Result<Option<Arc<ConsumeGate>>, LiminalError> {
        Ok(lock(&self.registrations)?
            .get(&pid)
            .and_then(|entry| entry.consume_gate.clone()))
    }

    pub(super) fn forwarding_channel(
        &self,
        participant: ParticipantPid,
    ) -> Result<ParticipantChannel, LiminalError> {
        let registrations = lock(&self.registrations)?;
        let registration =
            registrations
                .get(&participant.get())
                .ok_or_else(|| LiminalError::DeliveryFailed {
                    message: format!(
                        "participant {} has no registered forwarding queue",
                        participant.get()
                    ),
                })?;
        let owner = registration
            .core
            .upgrade()
            .ok_or_else(|| LiminalError::DeliveryFailed {
                message: format!(
                    "participant {} forwarding conversation is no longer live",
                    participant.get()
                ),
            })?;
        if owner.is_finalized() {
            return Err(LiminalError::DeliveryFailed {
                message: format!(
                    "participant {} forwarding conversation is closed",
                    participant.get()
                ),
            });
        }
        let channel = ParticipantChannel::new(participant, Arc::clone(&registration.inbox));
        drop(registrations);
        Ok(channel)
    }

    fn snapshot(&self, pid: u64) -> Option<ParticipantSnapshot> {
        let registrations = self.registrations.lock().ok()?;
        let registration = registrations.get(&pid)?;
        let snapshot = (
            Arc::clone(&registration.inbox),
            Arc::clone(&registration.behaviour),
            registration.core.clone(),
        );
        drop(registrations);
        Some(snapshot)
    }
}

/// Handle to one registered participant: its pid, the queue the actor forwards
/// into, and the shared runtime that wakes it. Held by [`ActorCore`] so a `Send`
/// command can forward an envelope to the real participant process.
#[derive(Clone, Debug)]
pub(super) struct ParticipantChannel {
    pid: ParticipantPid,
    inbox: RequestQueue,
}

impl ParticipantChannel {
    pub(super) const fn new(pid: ParticipantPid, inbox: RequestQueue) -> Self {
        Self { pid, inbox }
    }

    pub(super) const fn pid(&self) -> ParticipantPid {
        self.pid
    }

    /// Returns a clone of the shared forwarding queue, so the participant runtime
    /// and the conversation actor read and write the SAME queue.
    pub(super) fn inbox_arc(&self) -> RequestQueue {
        Arc::clone(&self.inbox)
    }

    /// Forwards `request` to the participant by enqueuing it on the shared queue
    /// and waking the participant process with `wakeup_atom`. Returns
    /// [`LiminalError::DeliveryFailed`] when the participant process is not live.
    pub(super) fn forward(
        &self,
        request: Envelope,
        reply_expectation: ReplyExpectation,
        scheduler: &Scheduler,
        wakeup_atom: Atom,
    ) -> Result<(), LiminalError> {
        lock(&self.inbox)?.push_back(ForwardedRequest {
            envelope: request,
            reply_expectation,
            handoff: None,
        });
        if scheduler.enqueue_atom_message(self.pid.get(), wakeup_atom) {
            Ok(())
        } else {
            // Roll back the enqueue so a dead participant does not leave an
            // orphaned request that a later live pid could mistakenly drain.
            lock(&self.inbox)?.pop_back();
            Err(LiminalError::DeliveryFailed {
                message: format!("conversation participant {} is not live", self.pid.get()),
            })
        }
    }
    pub(super) fn forward_dispatch(
        &self,
        request: Envelope,
        receipt: &Arc<ConsumptionReceipt>,
        scheduler: &Scheduler,
        wakeup_atom: Atom,
    ) -> Result<(), LiminalError> {
        if receipt.participant() != self.pid {
            return Err(LiminalError::DeliveryFailed {
                message: "dispatch forwarding target does not match its acceptance receipt"
                    .to_owned(),
            });
        }
        let mut queue = lock(&self.inbox)?;
        receipt.enqueue(&self.inbox, || {
            queue.push_back(ForwardedRequest {
                envelope: request,
                reply_expectation: ReplyExpectation::Local,
                handoff: Some(Arc::clone(receipt)),
            });
            if scheduler.enqueue_atom_message(self.pid.get(), wakeup_atom) {
                Ok(())
            } else {
                // The same inbox guard spans admission and rollback.
                queue.pop_back();
                Err(LiminalError::DeliveryFailed {
                    message: format!("conversation participant {} is not live", self.pid.get()),
                })
            }
        })
    }
}

/// The native participant process body.
///
/// A real [`NativeHandler`] — a first-class, scheduler-supervised beamr process
/// with a pid and mailbox, spawned through the same machinery as any other beamr
/// process. On each slice it drains its atom wakeups, then runs the shared
/// runtime slice that processes every forwarded request through the application
/// behaviour. It parks (`NativeOutcome::Wait`) when idle, so it consumes no CPU
/// until the actor forwards the next request — and stays linked to the actor for
/// structural crash detection.
pub(super) struct ParticipantProcess {
    runtime: Arc<ParticipantRuntime>,
    wakeup_atom: Atom,
}

impl ParticipantProcess {
    pub(super) const fn new(runtime: Arc<ParticipantRuntime>, wakeup_atom: Atom) -> Self {
        Self {
            runtime,
            wakeup_atom,
        }
    }
}

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

impl NativeHandler for ParticipantProcess {
    fn handle(&mut self, ctx: &mut NativeContext<'_>) -> NativeOutcome {
        let pid = ctx.self_pid();
        // Drain the atom wakeups the actor enqueued. The payloads themselves
        // travel through the shared Rust-side queue (the `ActorCore` pattern):
        // beamr's host enqueue API moves only atoms, so the atom is just the
        // wakeup signal and `run_slice` does the real work against the shared
        // queue. Participant failure is surfaced structurally via the beamr
        // link/EXIT to the supervised actor, not through a mailbox message, so
        // there is no message-driven shutdown branch here.
        while ctx.recv().is_some() {}
        match self.runtime.run_slice(pid) {
            Ok(_) => NativeOutcome::Wait,
            Err(error) => {
                tracing::error!(pid, %error, "participant request consumption failed");
                NativeOutcome::Stop(beamr::process::ExitReason::Error)
            }
        }
    }
}

fn lock<T>(mutex: &Mutex<T>) -> Result<std::sync::MutexGuard<'_, T>, LiminalError> {
    mutex.lock().map_err(|error| LiminalError::DeliveryFailed {
        message: format!("participant queue lock poisoned: {error}"),
    })
}

#[cfg(test)]
#[derive(Debug)]
struct ConsumeGate {
    permitted: std::sync::atomic::AtomicBool,
    parked: std::sync::atomic::AtomicBool,
    taken: std::sync::atomic::AtomicU64,
    limit: std::sync::atomic::AtomicU64,
    parked_signal: std::sync::mpsc::Sender<()>,
}

#[cfg(test)]
pub(crate) struct ConsumptionPermit {
    pid: ParticipantPid,
    scheduler: Arc<Scheduler>,
    wakeup: Atom,
    gate: Arc<ConsumeGate>,
}

#[cfg(test)]
impl std::fmt::Debug for ConsumptionPermit {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ConsumptionPermit")
            .field("pid", &self.pid)
            .finish_non_exhaustive()
    }
}

#[cfg(test)]
impl ConsumptionPermit {
    pub(crate) fn release(&self) -> Result<(), LiminalError> {
        self.permit_until(u64::MAX)
    }

    pub(crate) fn release_one(&self) -> Result<(), LiminalError> {
        let limit =
            self.messages_taken()
                .checked_add(1)
                .ok_or_else(|| LiminalError::DeliveryFailed {
                    message: "controlled consumer consumption count exhausted".to_owned(),
                })?;
        self.permit_until(limit)
    }

    fn permit_until(&self, limit: u64) -> Result<(), LiminalError> {
        self.gate
            .limit
            .store(limit, std::sync::atomic::Ordering::Release);
        self.gate
            .permitted
            .store(true, std::sync::atomic::Ordering::Release);
        if self
            .scheduler
            .enqueue_atom_message(self.pid.get(), self.wakeup)
        {
            Ok(())
        } else {
            Err(LiminalError::DeliveryFailed {
                message: "controlled consumer exited before consumption was permitted".to_owned(),
            })
        }
    }

    pub(crate) fn messages_taken(&self) -> u64 {
        self.gate.taken.load(std::sync::atomic::Ordering::Acquire)
    }
}

pub(super) fn discard_dispatch(inbox: &Mailbox, attempt: u64) -> Result<(), LiminalError> {
    lock(inbox)?.retain(|request| {
        request
            .handoff
            .as_ref()
            .is_none_or(|receipt| receipt.attempt() != attempt)
    });
    Ok(())
}