beamr 0.4.5

//! Remote process monitor control plane for cross-node MONITOR_P/DEMONITOR_P/MONITOR_P_EXIT.
//!
//! This module owns the state needed to correlate outbound `monitor/2` calls
//! with later `DEMONITOR_P`/`MONITOR_P_EXIT` traffic, plus inbound registrations
//! from remote nodes that must be notified when a local process exits. Encoding
//! onto the distribution transport is intentionally represented as an injectable
//! sender seam until the lower-level distribution frame codec is present.

use std::collections::HashMap;
use std::fmt;
use std::sync::{Arc, Mutex};

use crate::atom::Atom;
use crate::process::ExitReason;
use crate::term::Term;

const REMOTE_MONITOR_REFERENCE_START: u64 = 1 << 56;
const REMOTE_MONITOR_REFERENCE_MAX: u64 = Term::SMALL_INT_MAX as u64;

/// Copyable representation of a PID that belongs to a particular node.
#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
pub struct RemotePid {
    /// Node atom that owns this PID.
    pub node: Atom,
    /// Numeric PID component.
    pub pid_number: u64,
    /// PID serial component.
    pub serial: u64,
}

impl RemotePid {
    /// Build a node-qualified PID record.
    #[must_use]
    pub const fn new(node: Atom, pid_number: u64, serial: u64) -> Self {
        Self {
            node,
            pid_number,
            serial,
        }
    }
}

/// Distribution control messages needed for remote process monitors.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum MonitorControlMessage {
    /// Establish a monitor from `watcher` to `target`.
    MonitorP {
        /// Monitor reference generated by the watcher side.
        reference: u64,
        /// Remote watcher PID as seen by the target node.
        watcher: RemotePid,
        /// Target PID as seen by its owning node.
        target: RemotePid,
    },
    /// Remove a previously established cross-node monitor.
    DemonitorP {
        /// Monitor reference generated by the watcher side.
        reference: u64,
        /// Remote watcher PID as seen by the target node.
        watcher: RemotePid,
        /// Target PID as seen by its owning node.
        target: RemotePid,
    },
    /// Report that the target process exited.
    MonitorPExit {
        /// Monitor reference generated by the watcher side.
        reference: u64,
        /// PID that was monitored.
        target: RemotePid,
        /// Exit reason to place in the DOWN message.
        reason: ExitReason,
    },
}

impl MonitorControlMessage {
    /// Operation code for this message.
    #[must_use]
    pub const fn opcode(self) -> u8 {
        match self {
            Self::MonitorP { .. } => 19,
            Self::DemonitorP { .. } => 20,
            Self::MonitorPExit { .. } => 21,
        }
    }
}

/// A control message after routing to its destination node.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct OutboundMonitorControl {
    /// Destination node atom.
    pub node: Atom,
    /// Routed control message.
    pub message: MonitorControlMessage,
}

/// Error returned when a control message cannot be sent.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum ControlSendError {
    /// No distribution route or connection exists for the destination node.
    NoConnection,
    /// The small-integer remote monitor reference space has been exhausted.
    ReferenceExhausted,
}

impl fmt::Display for ControlSendError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NoConnection => formatter.write_str("no distribution connection"),
            Self::ReferenceExhausted => {
                formatter.write_str("remote monitor reference space exhausted")
            }
        }
    }
}

impl std::error::Error for ControlSendError {}

/// Transport seam used by monitor control logic.
pub trait MonitorControlSender: Send + Sync {
    /// Send a control message to `node`.
    fn send_control(
        &self,
        node: Atom,
        message: MonitorControlMessage,
    ) -> Result<(), ControlSendError>;
}

/// In-memory sender used until the distribution frame codec is wired in.
#[derive(Default)]
pub struct RecordingMonitorSender {
    sent: Mutex<Vec<OutboundMonitorControl>>,
}

impl RecordingMonitorSender {
    /// Construct an empty recording sender.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Drain all recorded outbound messages in send order.
    pub fn drain(&self) -> Vec<OutboundMonitorControl> {
        let mut sent = self.sent.lock().unwrap_or_else(|error| error.into_inner());
        sent.drain(..).collect()
    }
}

impl MonitorControlSender for RecordingMonitorSender {
    fn send_control(
        &self,
        node: Atom,
        message: MonitorControlMessage,
    ) -> Result<(), ControlSendError> {
        self.sent
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .push(OutboundMonitorControl { node, message });
        Ok(())
    }
}

/// Local metadata for a monitor established by a local process against a remote PID.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct OutboundRemoteMonitor {
    /// Local watcher process that owns the reference and receives DOWN.
    pub watcher_pid: u64,
    /// Reference returned by `monitor/2`.
    pub reference: u64,
    /// Remote target PID.
    pub target: RemotePid,
}

/// Metadata for a remote watcher observing a local process.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct InboundRemoteMonitor {
    /// Remote watcher PID.
    pub watcher: RemotePid,
    /// Reference generated on the remote watcher node.
    pub reference: u64,
    /// Local target PID being observed.
    pub target_pid: u64,
}

struct RemoteMonitorState {
    next_reference: u64,
    outbound_by_ref: HashMap<u64, OutboundRemoteMonitor>,
    inbound: Vec<InboundRemoteMonitor>,
}

impl Default for RemoteMonitorState {
    fn default() -> Self {
        Self {
            next_reference: REMOTE_MONITOR_REFERENCE_START,
            outbound_by_ref: HashMap::new(),
            inbound: Vec::new(),
        }
    }
}

/// Shared cross-node monitor control plane.
pub struct ControlPlane {
    local_node: Atom,
    state: Mutex<RemoteMonitorState>,
    sender: Arc<dyn MonitorControlSender>,
}

impl ControlPlane {
    /// Create a control plane for `local_node` with an injectable sender.
    #[must_use]
    pub fn new(local_node: Atom, sender: Arc<dyn MonitorControlSender>) -> Self {
        Self {
            local_node,
            state: Mutex::new(RemoteMonitorState::default()),
            sender,
        }
    }

    /// Register a local watcher of a remote target and emit MONITOR_P.
    pub fn monitor_remote(
        &self,
        watcher_pid: u64,
        target: RemotePid,
    ) -> Result<u64, ControlSendError> {
        let reference = {
            let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
            if state.next_reference > REMOTE_MONITOR_REFERENCE_MAX {
                return Err(ControlSendError::ReferenceExhausted);
            }
            let reference = state.next_reference;
            state.next_reference = state.next_reference.saturating_add(1);
            state.outbound_by_ref.insert(
                reference,
                OutboundRemoteMonitor {
                    watcher_pid,
                    reference,
                    target,
                },
            );
            reference
        };
        let watcher = RemotePid::new(self.local_node, watcher_pid, 0);
        let message = MonitorControlMessage::MonitorP {
            reference,
            watcher,
            target,
        };
        if let Err(error) = self.sender.send_control(target.node, message) {
            let mut state = self
                .state
                .lock()
                .unwrap_or_else(|poison| poison.into_inner());
            state.outbound_by_ref.remove(&reference);
            return Err(error);
        }
        Ok(reference)
    }

    /// Remove a local watcher of a remote target and emit DEMONITOR_P when known.
    pub fn demonitor_remote(
        &self,
        watcher_pid: u64,
        reference: u64,
    ) -> Result<bool, ControlSendError> {
        let monitor = {
            let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
            match state.outbound_by_ref.get(&reference).copied() {
                Some(monitor) if monitor.watcher_pid == watcher_pid => {
                    state.outbound_by_ref.remove(&reference);
                    Some(monitor)
                }
                _ => None,
            }
        };
        let Some(monitor) = monitor else {
            return Ok(false);
        };
        let watcher = RemotePid::new(self.local_node, watcher_pid, 0);
        self.sender.send_control(
            monitor.target.node,
            MonitorControlMessage::DemonitorP {
                reference,
                watcher,
                target: monitor.target,
            },
        )?;
        Ok(true)
    }

    /// Register an inbound remote monitor for a live local target.
    pub fn register_inbound_monitor(&self, reference: u64, watcher: RemotePid, target_pid: u64) {
        let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
        state.inbound.retain(|monitor| {
            !(monitor.reference == reference
                && monitor.watcher == watcher
                && monitor.target_pid == target_pid)
        });
        state.inbound.push(InboundRemoteMonitor {
            watcher,
            reference,
            target_pid,
        });
    }

    /// Remove an inbound remote monitor. Missing monitors are ignored.
    pub fn remove_inbound_monitor(&self, reference: u64, watcher: RemotePid, target_pid: u64) {
        let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
        state.inbound.retain(|monitor| {
            !(monitor.reference == reference
                && monitor.watcher == watcher
                && monitor.target_pid == target_pid)
        });
    }

    /// Drain inbound remote watchers of a local target after process exit.
    pub fn collect_inbound_for_target(&self, target_pid: u64) -> Vec<InboundRemoteMonitor> {
        let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
        let mut drained = Vec::new();
        state.inbound.retain(|monitor| {
            if monitor.target_pid == target_pid {
                drained.push(*monitor);
                false
            } else {
                true
            }
        });
        drained
    }

    /// Take local outbound metadata after MONITOR_P_EXIT arrives.
    pub fn take_outbound_for_exit(&self, reference: u64) -> Option<OutboundRemoteMonitor> {
        self.state
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .outbound_by_ref
            .remove(&reference)
    }

    /// Remove all outbound monitors connected to `node` for node-down DOWN delivery.
    pub fn collect_outbound_for_node(&self, node: Atom) -> Vec<OutboundRemoteMonitor> {
        let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
        let references: Vec<u64> = state
            .outbound_by_ref
            .iter()
            .filter_map(|(reference, monitor)| (monitor.target.node == node).then_some(*reference))
            .collect();
        references
            .into_iter()
            .filter_map(|reference| state.outbound_by_ref.remove(&reference))
            .collect()
    }

    /// Remove all inbound registrations owned by watchers on `node`.
    pub fn remove_inbound_for_watcher_node(&self, node: Atom) {
        let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner());
        state.inbound.retain(|monitor| monitor.watcher.node != node);
    }

    /// Emit MONITOR_P_EXIT for a remote watcher.
    pub fn send_monitor_exit(
        &self,
        monitor: InboundRemoteMonitor,
        reason: ExitReason,
    ) -> Result<(), ControlSendError> {
        let target = RemotePid::new(self.local_node, monitor.target_pid, 0);
        self.sender.send_control(
            monitor.watcher.node,
            MonitorControlMessage::MonitorPExit {
                reference: monitor.reference,
                target,
                reason,
            },
        )
    }

    /// Return an Arc to the injected sender for external wiring.
    #[must_use]
    pub fn sender(&self) -> Arc<dyn MonitorControlSender> {
        Arc::clone(&self.sender)
    }
}

/// Facility exposed to process BIFs for remote monitor control.
pub trait DistributionMonitorFacility: Send + Sync {
    /// Establish a cross-node process monitor and send MONITOR_P.
    fn monitor_remote(&self, watcher_pid: u64, target: RemotePid) -> Result<u64, ControlSendError>;

    /// Remove a known cross-node monitor and send DEMONITOR_P.
    fn demonitor_remote(&self, watcher_pid: u64, reference: u64) -> Result<bool, ControlSendError>;
}

impl DistributionMonitorFacility for ControlPlane {
    fn monitor_remote(&self, watcher_pid: u64, target: RemotePid) -> Result<u64, ControlSendError> {
        self.monitor_remote(watcher_pid, target)
    }

    fn demonitor_remote(&self, watcher_pid: u64, reference: u64) -> Result<bool, ControlSendError> {
        self.demonitor_remote(watcher_pid, reference)
    }
}

#[cfg(test)]
#[path = "control_monitor_tests.rs"]
mod tests;