liminal-rs 0.16.2

//! LIM-009: a channel actor's death retires its queue, told by the exit.
//!
//! A command's reply sender sits in the channel core's queue, which outlives
//! the actor process, so a waiter on a dead actor is only answered once
//! something drops that sender. The [`ChannelExitWatcher`] is that something:
//! one native process per channel supervisor, parked on its mailbox. The
//! supervisor monitors every channel actor it spawns from this watcher, so an
//! actor's exit delivers `{'DOWN', Ref, process, Pid, Reason}` here, and the
//! watcher retires that pid through [`ActorRuntime::retire_exited`], the one
//! retirement path a restart also takes. Retiring drops the abandoned reply
//! senders, so each waiter's receive returns at once. Nothing polls: with no
//! exit there is no message, no slice and no allocation.

use std::sync::Weak;

use beamr::atom::Atom;
use beamr::native::native_process::{NativeContext, NativeHandler, NativeOutcome};
use beamr::process::ExitReason;
use beamr::term::Term;
use beamr::term::boxed::Tuple;

use crate::channel::actor::beam::ActorRuntime;

pub struct ChannelExitWatcher {
    runtime: Weak<ActorRuntime>,
}

impl ChannelExitWatcher {
    pub const fn new(runtime: Weak<ActorRuntime>) -> Self {
        Self { runtime }
    }
}

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

impl NativeHandler for ChannelExitWatcher {
    /// One slice per delivered batch: every `DOWN` retires its actor. A watcher
    /// whose runtime is gone has no supervisor left to serve, so it stops.
    fn handle(&mut self, ctx: &mut NativeContext<'_>) -> NativeOutcome {
        let Some(runtime) = self.runtime.upgrade() else {
            return NativeOutcome::Stop(ExitReason::Normal);
        };
        #[cfg(test)]
        runtime.count_watcher_wake();
        while let Some(message) = ctx.recv() {
            if let Some(pid) = down_source(message) {
                if let Err(error) = runtime.retire_exited(pid) {
                    // The waiters on this actor are then answered by their own
                    // deadline; the fault is said where an operator reads it.
                    tracing::error!(pid, %error, "channel actor exit could not retire its queue");
                }
            }
        }
        NativeOutcome::Wait
    }
}

/// The exited pid of a `{'DOWN', Ref, process, Pid, Reason}` monitor message.
fn down_source(message: Term) -> Option<u64> {
    let tuple = Tuple::new(message)?;
    if tuple.arity() == 5 && tuple.get(0) == Some(Term::atom(Atom::DOWN)) {
        tuple.get(3)?.as_pid()
    } else {
        None
    }
}