liminal-rs 0.16.3

//! The channel actor's beamr bytecode body, command NIF, and link plumbing.
//!
//! Mirrors `conversation/actor/beam.rs`. The actor runs a tiny bytecode loop
//! (`LoopRec` -> NIF -> tail-call) so it is a real `trap_exit` process: its NIF
//! receives every mailbox message, which is either the command-wakeup atom
//! (drain one queued command) or a trapped `{EXIT, pid, reason}` signal from a
//! dead subscriber (prune it from the fan-out list). Linking to *existing*
//! subscriber pids requires the `ProcessContext::link_facility()` available only
//! on this bytecode-NIF path — which is exactly why the channel actor is a
//! bytecode process and not a `NativeHandler`.

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

use beamr::atom::Atom;
use beamr::constant_pool::ConstantPool;
use beamr::loader::Instruction;
use beamr::loader::decode::Operand;
use beamr::module::{Module, ModuleOrigin, ResolvedImport, ResolvedImportTarget};
use beamr::native::{Capability, NativeEntry, ProcessContext};
use beamr::term::Term;
use beamr::term::boxed::Tuple;

use super::ChannelActorCore;
use crate::error::LiminalError;

/// Per-scheduler registry mapping a channel actor's pid to its shared core, plus
/// the command-wakeup atom the NIF recognises. Stored as the scheduler's
/// `nif_private_data` so the NIF can recover the right core for the running pid.
pub struct ActorRuntime {
    command_atom: Atom,
    actors: Mutex<HashMap<u64, Arc<ActorRegistration>>>,
    #[cfg(test)]
    entry_hold: EntryHold,
    /// Test-only: told the pid of every retirement that took effect.
    #[cfg(test)]
    retirements: Mutex<Option<std::sync::mpsc::Sender<u64>>>,
    /// Test-only: slices the exit watcher has run (LIM-009 R3, idle cost).
    #[cfg(test)]
    watcher_wakes: std::sync::atomic::AtomicU64,
}

struct ActorRegistration {
    core: Weak<ChannelActorCore>,
    /// Retirement must let an executing command finish before clearing its queue.
    retired: Mutex<bool>,
}

/// Test-only: holds the next slice of the command NIF at its entry, before
/// it resolves its core, until the test releases it. One-shot: arming it
/// holds exactly one slice. A test uses it to keep a command's wake inside
/// an executing slot while the actor is terminated (the stale-command red).
#[cfg(test)]
#[derive(Debug, Default)]
pub struct EntryHold {
    state: Mutex<EntryHoldState>,
    changed: std::sync::Condvar,
}

#[cfg(test)]
#[derive(Debug, Default)]
struct EntryHoldState {
    armed: bool,
    parked: usize,
    released: bool,
}

#[cfg(test)]
impl EntryHold {
    /// Holds the next slice that enters the command NIF.
    pub(crate) fn arm(&self) -> Result<(), LiminalError> {
        let mut state = self.lock()?;
        state.armed = true;
        state.released = false;
        drop(state);
        Ok(())
    }

    /// Blocks until the armed hold has parked a slice.
    pub(crate) fn await_parked(&self) -> Result<(), LiminalError> {
        let mut state = self.lock()?;
        while state.parked == 0 {
            state = self.changed.wait(state).map_err(|error| poisoned(&error))?;
        }
        drop(state);
        Ok(())
    }

    /// Lets the parked slice go on.
    pub(crate) fn release(&self) -> Result<(), LiminalError> {
        let mut state = self.lock()?;
        state.released = true;
        drop(state);
        self.changed.notify_all();
        Ok(())
    }

    /// Called by the NIF: parks this slice when the hold is armed, and
    /// returns once the test releases it.
    fn pass(&self) -> Result<(), LiminalError> {
        let mut state = self.lock()?;
        if !state.armed {
            return Ok(());
        }
        state.armed = false;
        state.parked += 1;
        drop(state);
        self.changed.notify_all();
        let mut state = self.lock()?;
        while !state.released {
            state = self.changed.wait(state).map_err(|error| poisoned(&error))?;
        }
        drop(state);
        Ok(())
    }

    fn lock(&self) -> Result<std::sync::MutexGuard<'_, EntryHoldState>, LiminalError> {
        self.state.lock().map_err(|error| poisoned(&error))
    }
}

#[cfg(test)]
fn poisoned(error: &dyn std::fmt::Display) -> LiminalError {
    LiminalError::DeliveryFailed {
        message: format!("channel actor entry hold poisoned: {error}"),
    }
}

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

impl ActorRuntime {
    pub fn new(command_atom: Atom) -> Self {
        Self {
            command_atom,
            actors: Mutex::new(HashMap::new()),
            #[cfg(test)]
            entry_hold: EntryHold::default(),
            #[cfg(test)]
            retirements: Mutex::new(None),
            #[cfg(test)]
            watcher_wakes: std::sync::atomic::AtomicU64::new(0),
        }
    }

    /// Test-only: counts one exit-watcher slice.
    #[cfg(test)]
    pub(crate) fn count_watcher_wake(&self) {
        self.watcher_wakes
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    }

    /// Test-only: the exit-watcher slices run so far.
    #[cfg(test)]
    pub(crate) fn watcher_wakes(&self) -> u64 {
        self.watcher_wakes
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Test-only: a receiver told the pid of every retirement that takes effect from now on.
    #[cfg(test)]
    pub(crate) fn retirements(&self) -> Result<std::sync::mpsc::Receiver<u64>, LiminalError> {
        let (sender, receiver) = std::sync::mpsc::channel();
        *self
            .retirements
            .lock()
            .map_err(|error| LiminalError::DeliveryFailed {
                message: format!("channel actor retirement probe poisoned: {error}"),
            })? = Some(sender);
        Ok(receiver)
    }

    /// The test-only hold at the command NIF's entry.
    #[cfg(test)]
    pub(crate) const fn entry_hold(&self) -> &EntryHold {
        &self.entry_hold
    }

    pub const fn command_atom(&self) -> Atom {
        self.command_atom
    }

    /// Registers a channel actor pid with its shared core.
    ///
    /// # Errors
    ///
    /// Returns [`LiminalError`] when the registry lock is poisoned or when the
    /// pid is already registered to a live core.
    pub fn register(&self, pid: u64, core: Weak<ChannelActorCore>) -> Result<(), LiminalError> {
        let mut actors = self
            .actors
            .lock()
            .map_err(|error| LiminalError::DeliveryFailed {
                message: format!("channel actor runtime lock poisoned: {error}"),
            })?;
        // Drop keys whose actor core is already gone before inserting (mirrors
        // the conversation actor runtime), so an actor that exited without a
        // restart deregistering it cannot accumulate a dead key across spawns.
        actors.retain(|_, actor| actor.core.strong_count() > 0);
        // Every key left is owned by a LIVE core, so an occupied slot for a
        // freshly spawned pid is a supervision defect (beamr pids are unique
        // among live processes). Refuse it loudly instead of silently rebinding
        // the pid's NIF dispatch to a different channel core.
        match actors.entry(pid) {
            Entry::Occupied(_) => Err(LiminalError::DeliveryFailed {
                message: format!(
                    "channel actor pid {pid} is already registered to a live actor core"
                ),
            }),
            Entry::Vacant(entry) => {
                entry.insert(Arc::new(ActorRegistration {
                    core,
                    retired: Mutex::new(false),
                }));
                Ok(())
            }
        }
    }

    /// Ends the dead actor's queue ownership before a replacement is published.
    /// An executing slice keeps its reply; commands whose wakes died disconnect,
    /// which is what answers their waiters (LIM-009).
    ///
    /// One path, idempotent (LIM-009 R2): the exit watcher and a restart may
    /// both reach it, in either order. A pid already retired, or no longer
    /// registered, is left as it is. A core that has moved on to a newer pid
    /// keeps its queue: only the queue this pid still owns is taken.
    ///
    /// # Errors
    /// Returns [`LiminalError`] when a registry, execution or core lock is poisoned.
    pub fn retire(&self, pid: u64, core: &ChannelActorCore) -> Result<(), LiminalError> {
        let Some(actor) = self.registered(pid)? else {
            return Ok(());
        };
        let mut retired = actor
            .retired
            .lock()
            .map_err(|error| LiminalError::DeliveryFailed {
                message: format!("channel actor execution lock poisoned: {error}"),
            })?;
        if *retired {
            return Ok(());
        }
        let abandoned = {
            let mut current_pid = super::lock(&core.current_pid)?;
            if *current_pid == Some(pid) {
                let mut commands = super::lock(&core.commands)?;
                *current_pid = None;
                drop(current_pid);
                std::mem::take(&mut *commands)
            } else {
                std::collections::VecDeque::new()
            }
        };
        *retired = true;
        drop(retired);
        self.actors
            .lock()
            .map_err(|error| LiminalError::DeliveryFailed {
                message: format!("channel actor runtime lock poisoned: {error}"),
            })?
            .remove(&pid);
        drop(abandoned);
        #[cfg(test)]
        {
            let mut probe =
                self.retirements
                    .lock()
                    .map_err(|error| LiminalError::DeliveryFailed {
                        message: format!("channel actor retirement probe poisoned: {error}"),
                    })?;
            // a test that stopped listening is let go, not told again
            if probe
                .as_ref()
                .is_some_and(|sender| sender.send(pid).is_err())
            {
                *probe = None;
            }
            drop(probe);
        }
        Ok(())
    }

    /// Retires `pid` on its exit, told by the exit watcher's `DOWN` (LIM-009).
    /// An actor whose core is already gone has no queue left to answer from:
    /// its registration is removed and nothing else is owed.
    ///
    /// # Errors
    /// Returns [`LiminalError`] when a registry, execution or core lock is poisoned.
    pub(crate) fn retire_exited(&self, pid: u64) -> Result<(), LiminalError> {
        let Some(actor) = self.registered(pid)? else {
            return Ok(());
        };
        if let Some(core) = actor.core.upgrade() {
            return self.retire(pid, &core);
        }
        self.actors
            .lock()
            .map_err(|error| LiminalError::DeliveryFailed {
                message: format!("channel actor runtime lock poisoned: {error}"),
            })?
            .remove(&pid);
        Ok(())
    }

    fn registered(&self, pid: u64) -> Result<Option<Arc<ActorRegistration>>, LiminalError> {
        Ok(self
            .actors
            .lock()
            .map_err(|error| LiminalError::DeliveryFailed {
                message: format!("channel actor runtime lock poisoned: {error}"),
            })?
            .get(&pid)
            .cloned())
    }
}

/// Build the channel actor's bytecode module: an endless `receive` loop that
/// hands each message to the `process_command` NIF. Identical in shape to the
/// conversation actor module.
pub fn actor_module(module_name: Atom, entry_function: Atom, command_function: Atom) -> Module {
    let code = vec![
        Instruction::Label { label: 1 },
        Instruction::LoopRec {
            fail: Operand::Label(2),
            destination: Operand::X(0),
        },
        Instruction::RemoveMessage,
        Instruction::CallExt {
            arity: Operand::Unsigned(1),
            import: Operand::Unsigned(0),
        },
        Instruction::CallOnly {
            arity: Operand::Unsigned(0),
            label: Operand::Label(1),
        },
        Instruction::Label { label: 2 },
        Instruction::Wait {
            fail: Operand::Label(1),
        },
    ];
    let label_index = code
        .iter()
        .enumerate()
        .filter_map(|(ip, instruction)| match instruction {
            Instruction::Label { label } => Some((*label, ip)),
            _ => None,
        })
        .collect();
    let mut exports = HashMap::new();
    exports.insert((entry_function, 0), 1);
    Module {
        name: module_name,
        generation: 0,
        origin: ModuleOrigin::Preloaded,
        exports,
        label_index,
        code,
        function_table: Vec::new(),
        line_table: Vec::new(),
        literals: Vec::new(),
        constant_pool: ConstantPool::default(),
        resolved_imports: vec![ResolvedImport {
            module: module_name,
            function: command_function,
            arity: 1,
            target: ResolvedImportTarget::Native(NativeEntry {
                function: process_command_nif,
                dirty_kind: None,
                capability: Capability::ProcessLocal,
            }),
        }],
        lambdas: Vec::new(),
        string_table: Vec::new(),
        line_info: Vec::new(),
    }
}

/// Establish a bidirectional beamr link from the channel actor to `subscriber`,
/// so the subscriber's exit is trapped as an `{EXIT, pid, reason}` message.
pub(super) fn link_subscriber(
    context: &ProcessContext<'_>,
    subscriber: u64,
) -> Result<(), LiminalError> {
    let actor_pid = context
        .pid()
        .ok_or_else(|| LiminalError::SubscriptionFailed {
            message: "channel actor has no beamr pid".to_owned(),
        })?;
    let link_facility =
        context
            .link_facility()
            .ok_or_else(|| LiminalError::SubscriptionFailed {
                message: "beamr link facility is unavailable".to_owned(),
            })?;
    link_facility
        .link(actor_pid, subscriber)
        .map_err(|error| LiminalError::SubscriptionFailed {
            message: format!(
                "failed to link channel actor {actor_pid} to subscriber {subscriber}: {error}"
            ),
        })
}

/// Remove the bidirectional link to `subscriber` (caller-driven unsubscribe).
pub(super) fn unlink_subscriber(
    context: &ProcessContext<'_>,
    subscriber: u64,
) -> Result<(), LiminalError> {
    let actor_pid = context
        .pid()
        .ok_or_else(|| LiminalError::SubscriptionFailed {
            message: "channel actor has no beamr pid".to_owned(),
        })?;
    let link_facility =
        context
            .link_facility()
            .ok_or_else(|| LiminalError::SubscriptionFailed {
                message: "beamr link facility is unavailable".to_owned(),
            })?;
    link_facility
        .unlink(actor_pid, subscriber)
        .map_err(|error| LiminalError::SubscriptionFailed {
            message: format!(
                "failed to unlink channel actor {actor_pid} from subscriber {subscriber}: {error}"
            ),
        })
}

fn process_command_nif(args: &[Term], context: &mut ProcessContext<'_>) -> Result<Term, Term> {
    let [message] = args else {
        return Err(badarg());
    };
    let pid = context.pid().ok_or_else(badarg)?;
    let runtime = runtime(context)?;
    let actor = runtime
        .registered(pid)
        .map_err(|_| badarg())?
        .ok_or_else(badarg)?;
    // Keep retirement behind this whole command, including a parked NIF entry.
    let retired = actor.retired.lock().map_err(|_| badarg())?;
    if *retired {
        return Err(badarg());
    }
    #[cfg(test)]
    runtime.entry_hold.pass().map_err(|_| badarg())?;
    let core = actor.core.upgrade().ok_or_else(badarg)?;
    if message.as_atom() == Some(runtime.command_atom) {
        if core.process_next_command(context) {
            context.request_shutdown();
        }
    } else if let Some(subscriber) = exit_source(*message) {
        core.handle_subscriber_exit(subscriber)
            .map_err(|_| badarg())?;
    }
    drop(retired);
    Ok(Term::atom(Atom::OK))
}

fn runtime(context: &ProcessContext<'_>) -> Result<Arc<ActorRuntime>, Term> {
    let data = context.nif_private_data().ok_or_else(badarg)?;
    Arc::downcast::<ActorRuntime>(Arc::clone(data)).map_err(|_| badarg())
}

/// Decode a trapped `{EXIT, pid, reason}` tuple into the source pid.
pub fn exit_source(message: Term) -> Option<u64> {
    let tuple = Tuple::new(message)?;
    if tuple.arity() == 3 && tuple.get(0) == Some(Term::atom(Atom::EXIT)) {
        tuple.get(1)?.as_pid()
    } else {
        None
    }
}

const fn badarg() -> Term {
    Term::atom(Atom::BADARG)
}

/// Convenience for callers wiring a scheduler's private data.
pub fn private_data(runtime: Arc<ActorRuntime>) -> Arc<dyn Any + Send + Sync> {
    runtime
}