liminal-rs 0.10.2

//! LIM-002: the channel actor as a REAL supervised beamr process.
//!
//! [`ChannelActorCore`] is the shared state (`Arc<ChannelActorCore>`) owned
//! jointly by the host-side handle and the beamr process's NIF command loop. The
//! process is a bytecode `trap_exit` process whose single NIF (`beam.rs`) either
//! drains one queued command or handles a trapped `{EXIT, pid, reason}` signal
//! from a dead subscriber. This is the same machinery the conversation actor
//! uses (`conversation/actor/{beam,core}.rs`) — chosen over a `NativeHandler`
//! because R2 requires LINKING to *already-existing* subscriber processes, which
//! only `ProcessContext::link_facility().link(actor, subscriber)` can do; the
//! `NativeContext` given to a `NativeHandler` can only link a freshly-spawned
//! child.
//!
//! Owns its subscriber list in-memory (R1). Subscriber death is detected via the
//! real link/EXIT path, not weak-pointer polling (R2). Predicates are owned by
//! the actor and evaluated at delivery (R3). Nothing on the delivery path
//! touches persistence or serialisation, and while Ephemeral no haematite type
//! is referenced (R5).

pub(crate) mod beam;
pub(crate) mod queue;
mod wait;

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Mutex, MutexGuard};

use beamr::atom::Atom;
use beamr::native::ProcessContext;
use beamr::scheduler::Scheduler;
use serde_json::Value;

use crate::causal::CausalContext;
use crate::channel::admission::PressureAggregate;
use crate::channel::schema::{Schema, SchemaId, SchemaValidationError};
use crate::channel::subscription::SubscriberRegistration;
use crate::envelope::{Envelope, PublisherId};
use crate::error::LiminalError;

/// Hard cap on the actor's pending command queue (A1 §9 buffers inventory).
///
/// The queue is naturally bounded by the number of concurrent callers — each
/// blocks on its own rendezvous reply — so this is defense, not the primary
/// bound. It exists so that "no queue anywhere grows without bound when a
/// consumer stalls" is true of EVERY row of §9's table and not merely of the
/// ones A1 rewrites. Overflow is an enqueue error, which surfaces as
/// `DeliveryFailed`/`PublishError` — an error, not a pressure signal, exactly
/// as §9 classifies it.
const MAX_PENDING_COMMANDS: usize = 4_096;

/// A publish that is already in the actor's command queue, with the rendezvous
/// its fan-out will answer on. Its ORDER against every other publish on this
/// channel is fixed the moment it exists; see [`ChannelActorCore::enqueue_publish`].
pub(crate) struct PendingPublish {
    response: mpsc::Receiver<Result<PublishOutcome, LiminalError>>,
    pid: u64,
}

pub(crate) use beam::{ActorRuntime, actor_module, private_data};
use queue::{ChannelCommand, ChannelCommandKind, SubscriberSummary};
pub(crate) use queue::{PublishOutcome, predicate_from};

/// Shared channel-actor state: schema, subscriber fan-out list, closed flag,
/// the command queue, and the wiring needed to wake the process.
pub(crate) struct ChannelActorCore {
    scheduler: Arc<Scheduler>,
    command_atom: Atom,
    schema: Mutex<Schema>,
    subscribers: Mutex<Vec<SubscriberRegistration>>,
    closed: Mutex<bool>,
    commands: Mutex<VecDeque<ChannelCommand>>,
    current_pid: Mutex<Option<u64>>,
    /// Serialises the supervisor's dead-check + respawn so two concurrent callers
    /// can never both observe the dead pid and both spawn a replacement actor
    /// (the restart TOCTOU). Held across the double-checked liveness test and the
    /// spawn in [`crate::channel::supervisor::ChannelSupervisor::ensure_running`].
    restart_lock: Mutex<()>,
    next_command_id: AtomicU64,
}

impl std::fmt::Debug for ChannelActorCore {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("ChannelActorCore")
            .field("current_pid", &self.current_pid.lock().ok().map(|pid| *pid))
            .finish_non_exhaustive()
    }
}

impl ChannelActorCore {
    pub(crate) const fn new(scheduler: Arc<Scheduler>, command_atom: Atom, schema: Schema) -> Self {
        Self {
            scheduler,
            command_atom,
            schema: Mutex::new(schema),
            subscribers: Mutex::new(Vec::new()),
            closed: Mutex::new(false),
            commands: Mutex::new(VecDeque::new()),
            current_pid: Mutex::new(None),
            restart_lock: Mutex::new(()),
            next_command_id: AtomicU64::new(1),
        }
    }

    pub(crate) const fn scheduler(&self) -> &Arc<Scheduler> {
        &self.scheduler
    }

    pub(crate) const fn restart_lock(&self) -> &Mutex<()> {
        &self.restart_lock
    }

    pub(crate) fn set_current_pid(&self, pid: u64) -> Result<(), LiminalError> {
        *lock(&self.current_pid)? = Some(pid);
        Ok(())
    }

    pub(crate) fn current_pid(&self) -> Result<Option<u64>, LiminalError> {
        Ok(*lock(&self.current_pid)?)
    }

    // ---- host-side request surface (blocks on a per-command reply) ----------

    /// Drives the freshly-spawned actor process to re-establish its links to
    /// every surviving subscriber pid. Called by the supervisor immediately
    /// after a spawn/restart (with `current_pid` already set to the new pid), so
    /// EXIT detection for subscribers that outlived a crash is restored before
    /// the channel is used again (R2/R4).
    pub(crate) fn boot(&self) -> Result<(), LiminalError> {
        let (reply, response) = mpsc::sync_channel(1);
        let pid = self.enqueue(ChannelCommandKind::Boot { reply })?;
        self.wait_live(&response, pid)?
    }

    /// Hands one publish to the actor's FIFO command queue and returns the
    /// rendezvous to wait on, WITHOUT waiting on it (A1 §4).
    ///
    /// Split out from the wait so a durable publish can enqueue while it still
    /// holds the append lock that assigned its position — see
    /// [`ChannelHandle::publish_with_delivery`](crate::channel::ChannelHandle::publish_with_delivery).
    /// The queue is a `VecDeque` pushed under one mutex and drained in order by
    /// the actor's single command slice, so enqueue order IS fan-out order:
    /// enqueuing under the append lock is what makes "live pushes reach a
    /// subscriber in durable-position order" a property of the code rather than
    /// a hope about thread scheduling.
    pub(crate) fn enqueue_publish(
        &self,
        payload: Vec<u8>,
        publisher_id: PublisherId,
        causal_context: Option<CausalContext>,
        durable_position: Option<u64>,
    ) -> Result<PendingPublish, LiminalError> {
        let (reply, response) = mpsc::sync_channel(1);
        let pid = self.enqueue(ChannelCommandKind::Publish {
            payload,
            publisher_id,
            causal_context,
            durable_position,
            reply,
        })?;
        Ok(PendingPublish { response, pid })
    }

    /// Waits for the fan-out of a publish handed over by
    /// [`Self::enqueue_publish`]. Called with every append lock RELEASED: the
    /// ordering the enqueue established is already fixed by then, so nothing is
    /// gained by holding a lock across the actor rendezvous and a slow
    /// subscriber's notifier would otherwise stall the next append.
    /// Taken BY VALUE, and destructured: a queued publish has exactly one
    /// rendezvous and is awaited exactly once.
    pub(crate) fn await_publish(
        &self,
        pending: PendingPublish,
    ) -> Result<PublishOutcome, LiminalError> {
        let PendingPublish { response, pid } = pending;
        self.wait_live(&response, pid)?
    }

    /// The channel-aggregate live-buffer occupancy and its summed bound, read
    /// HOST-SIDE for the durable pre-append hard watermark (A1 §4, graft §0.3).
    ///
    /// Off the command queue on purpose, exactly as
    /// [`Self::pre_publish_gate`] is: the watermark runs BEFORE the durable
    /// append, so routing it through the actor would cost a second rendezvous
    /// per publish and put the check behind the very queue it exists to keep
    /// from growing. It reads the subscriber list under its own mutex and asks
    /// each inbox for two numbers; no predicate is evaluated and no envelope is
    /// touched, which is what makes it "cheap and coarse" in the design's
    /// words.
    pub(crate) fn live_buffer_occupancy(&self) -> Result<(usize, usize), LiminalError> {
        let mut total_queued: usize = 0;
        let mut total_bound: usize = 0;
        let subscribers = lock(&self.subscribers)?;
        for subscriber in subscribers.iter() {
            let (queued, bound) = subscriber.occupancy();
            total_queued = total_queued.saturating_add(queued);
            total_bound = total_bound.saturating_add(bound);
        }
        drop(subscribers);
        Ok((total_queued, total_bound))
    }

    pub(crate) fn subscribe(
        &self,
        registration: SubscriberRegistration,
    ) -> Result<(), LiminalError> {
        let (reply, response) = mpsc::sync_channel(1);
        let pid = self.enqueue(ChannelCommandKind::Subscribe {
            registration,
            reply,
        })?;
        self.wait_live(&response, pid)?
    }

    pub(crate) fn unsubscribe(&self, pid: u64) -> Result<(), LiminalError> {
        let (reply, response) = mpsc::sync_channel(1);
        let target = self.enqueue(ChannelCommandKind::Unsubscribe { pid, reply })?;
        self.wait_live(&response, target)?
    }

    pub(crate) fn evolve(
        &self,
        name: String,
        field_schema: Value,
        default: Value,
    ) -> Result<SchemaId, SchemaValidationError> {
        let (reply, response) = mpsc::sync_channel(1);
        let pid = self
            .enqueue(ChannelCommandKind::Evolve {
                name,
                field_schema,
                default,
                reply,
            })
            .map_err(|error| SchemaValidationError::InvalidSchema {
                message: error.to_string(),
            })?;
        self.wait_schema_live(&response, pid)?
    }

    pub(crate) fn schema_id(&self) -> Result<SchemaId, LiminalError> {
        let (reply, response) = mpsc::sync_channel(1);
        let pid = self.enqueue(ChannelCommandKind::SchemaId { reply })?;
        self.wait_live(&response, pid)?
    }

    /// Validates a payload against the channel's current schema and closed
    /// state WITHOUT publishing — the pre-persistence gate for durable
    /// channels (G4, docs/design/wire-durability-gaps-20260831.md): a payload
    /// the channel would refuse must never burn a durable store sequence.
    ///
    /// The actor's own validation in `apply_publish` stays authoritative; a
    /// schema evolve landing between this gate and that validation narrows
    /// the burn to exactly that race instead of every invalid publish.
    pub(crate) fn pre_publish_gate(&self, payload: &[u8]) -> Result<(), LiminalError> {
        if *lock(&self.closed)? {
            return Err(LiminalError::ChannelClosed {
                message: "channel is closed".to_owned(),
            });
        }
        lock(&self.schema)?
            .validate(payload)
            .map_err(|error| schema_mismatch(&error))
    }

    pub(crate) fn list_subscribers(&self) -> Result<SubscriberSummary, LiminalError> {
        let (reply, response) = mpsc::sync_channel(1);
        let pid = self.enqueue(ChannelCommandKind::ListSubscribers { reply })?;
        self.wait_live(&response, pid)?
    }

    pub(crate) fn close(&self) -> Result<(), LiminalError> {
        let (reply, response) = mpsc::sync_channel(1);
        let pid = self.enqueue(ChannelCommandKind::Close { reply })?;
        self.wait_live(&response, pid)?
    }

    fn wait_live<T>(
        &self,
        response: &mpsc::Receiver<Result<T, LiminalError>>,
        pid: u64,
    ) -> Result<Result<T, LiminalError>, LiminalError> {
        wait::wait_live(&self.scheduler, response, pid)
    }

    fn wait_schema_live(
        &self,
        response: &mpsc::Receiver<Result<SchemaId, SchemaValidationError>>,
        pid: u64,
    ) -> Result<Result<SchemaId, SchemaValidationError>, SchemaValidationError> {
        wait::wait_schema_live(&self.scheduler, response, pid)
    }

    fn enqueue(&self, kind: ChannelCommandKind) -> Result<u64, LiminalError> {
        let pid = lock(&self.current_pid)?.ok_or_else(|| LiminalError::DeliveryFailed {
            message: "channel actor has no live pid".to_owned(),
        })?;
        let id = self.next_command_id.fetch_add(1, Ordering::Relaxed);
        {
            let mut commands = lock(&self.commands)?;
            if commands.len() >= MAX_PENDING_COMMANDS {
                return Err(LiminalError::DeliveryFailed {
                    message: format!(
                        "channel actor command queue is full ({MAX_PENDING_COMMANDS} pending)"
                    ),
                });
            }
            commands.push_back(ChannelCommand { id, kind });
        }
        if self.scheduler.enqueue_atom_message(pid, self.command_atom) {
            Ok(pid)
        } else {
            self.remove_command(id)?;
            Err(LiminalError::DeliveryFailed {
                message: format!("channel actor pid {pid} is not live"),
            })
        }
    }

    fn remove_command(&self, id: u64) -> Result<(), LiminalError> {
        lock(&self.commands)?.retain(|command| command.id != id);
        Ok(())
    }

    // ---- process-side command loop (runs inside the beamr process) ----------

    /// Pop and service exactly one queued command. Returns `true` when the
    /// command requested the process stop (a successful `Close`).
    pub(crate) fn process_next_command(&self, context: &ProcessContext<'_>) -> bool {
        let Some(command) = self.pop_command() else {
            return false;
        };
        match command.kind {
            ChannelCommandKind::Boot { reply } => {
                let _ = reply.send(self.apply_boot(context));
                false
            }
            ChannelCommandKind::Publish {
                payload,
                publisher_id,
                causal_context,
                durable_position,
                reply,
            } => {
                let _ = reply.send(self.apply_publish(
                    &payload,
                    publisher_id,
                    causal_context,
                    durable_position,
                ));
                false
            }
            ChannelCommandKind::Subscribe {
                registration,
                reply,
            } => {
                let _ = reply.send(self.apply_subscribe(registration, context));
                false
            }
            ChannelCommandKind::Unsubscribe { pid, reply } => {
                let _ = reply.send(self.apply_unsubscribe(pid, context));
                false
            }
            ChannelCommandKind::Evolve {
                name,
                field_schema,
                default,
                reply,
            } => {
                let _ = reply.send(self.apply_evolve(name, field_schema, default));
                false
            }
            ChannelCommandKind::SchemaId { reply } => {
                let _ = reply.send(self.apply_schema_id());
                false
            }
            ChannelCommandKind::ListSubscribers { reply } => {
                let _ = reply.send(self.apply_list_subscribers());
                false
            }
            ChannelCommandKind::Close { reply } => {
                let result = self.apply_close();
                let stop = result.is_ok();
                let _ = reply.send(result);
                stop
            }
        }
    }

    fn pop_command(&self) -> Option<ChannelCommand> {
        self.commands.lock().ok()?.pop_front()
    }

    /// Re-link the actor process to every surviving subscriber. Runs inside the
    /// freshly-spawned process context (the only place a link to an existing pid
    /// can be created). On the initial spawn `subscribers` is empty so this is a
    /// no-op; after a restart it re-establishes the links a crash tore down, so
    /// subscriber-death EXIT detection works again (R2/R4) — exactly mirroring
    /// the conversation actor's `link_participants` boot step.
    fn apply_boot(&self, context: &ProcessContext<'_>) -> Result<(), LiminalError> {
        // Take the registrations out under a short lock, then re-link outside it.
        let registrations = std::mem::take(&mut *lock(&self.subscribers)?);
        // Re-link only to subscribers whose process is still alive; a subscriber
        // that did not outlive the crash is pruned rather than failing boot (its
        // pid no longer exists). Subscribers trap exits (see `subscription.rs`)
        // so in practice they survive, but pruning keeps boot robust either way.
        let mut live = Vec::with_capacity(registrations.len());
        for subscriber in registrations {
            if self
                .scheduler
                .process_table()
                .get(subscriber.pid())
                .is_none()
            {
                continue;
            }
            beam::link_subscriber(context, subscriber.pid())?;
            live.push(subscriber);
        }
        *lock(&self.subscribers)? = live;
        Ok(())
    }

    fn apply_publish(
        &self,
        payload: &[u8],
        publisher_id: PublisherId,
        causal_context: Option<CausalContext>,
        durable_position: Option<u64>,
    ) -> Result<PublishOutcome, LiminalError> {
        if *lock(&self.closed)? {
            return Err(LiminalError::ChannelClosed {
                message: "channel is closed".to_owned(),
            });
        }
        let envelope = {
            let schema = lock(&self.schema)?;
            schema
                .validate(payload)
                .map_err(|error| schema_mismatch(&error))?;
            let normalized = schema
                .validate_and_apply_defaults(payload)
                .map_err(|error| schema_mismatch(&error))?;
            Envelope::new(normalized, causal_context, schema.id(), publisher_id)
        };
        let subscribers = lock(&self.subscribers)?;
        let mut delivered_count = 0;
        // A1 §2/§3: the decision runs on the actor's single command slice,
        // AFTER the predicate, and every matching subscriber's outcome folds
        // into ONE producer-visible signal. Non-matching subscribers return
        // `None` and are never recorded, so a heavily filtered channel cannot
        // manufacture false backpressure.
        let mut aggregate = PressureAggregate::new();
        for subscriber in subscribers.iter() {
            // `deliver` is infallible (R3: an inbox overflow marks the
            // subscription for shedding rather than erroring), so an envelope
            // that did not enter an inbox simply is not counted.
            let Some(outcome) = subscriber.deliver(&envelope, durable_position) else {
                continue;
            };
            if outcome.is_queued() {
                delivered_count += 1;
            }
            // A §5 budget/fairness shed carries no PACING signal: it is a
            // memory-safety refusal, not a verdict about a slow consumer, and
            // recording it as one would report the wrong cause.
            //
            // It is still a delivered-to-nobody outcome for its subscriber,
            // though, and the producer's aggregate answers exactly one
            // question — did the bus take custody. So it is recorded through
            // the separate `record_dropped` door, which counts it toward
            // `matching` and folds in the Reject it actually was. Without that,
            // a publish EVERY subscriber dropped left `matching == 0` and
            // resolved to the zero-subscriber `Accept` sentinel: `is_admitted()`
            // read true and `services.rs`'s `complete_receipt` permanently
            // suppressed the producer's retry for a message nobody received.
            if let Some(signal) = outcome.signal() {
                aggregate.record(*signal);
            } else if let Some(shed) = outcome.shed_signal() {
                aggregate.record_dropped(*shed);
            }
        }
        drop(subscribers);
        // Return the normalised envelope (so the host can fan the SAME message out
        // to remote subscribers via the cluster observer, SRV-005), the genuine
        // local delivery count that backs the delivery-ack signal, and the
        // aggregated pressure signal that backs the A1 producer contract.
        Ok(PublishOutcome {
            envelope,
            delivered_count,
            signal: aggregate.resolve(),
            every_match_dropped: aggregate.every_match_dropped(),
        })
    }

    fn apply_subscribe(
        &self,
        registration: SubscriberRegistration,
        context: &ProcessContext<'_>,
    ) -> Result<(), LiminalError> {
        if *lock(&self.closed)? {
            return Err(LiminalError::ChannelClosed {
                message: "channel is closed".to_owned(),
            });
        }
        beam::link_subscriber(context, registration.pid())?;
        lock(&self.subscribers)?.push(registration);
        Ok(())
    }

    fn apply_unsubscribe(
        &self,
        pid: u64,
        context: &ProcessContext<'_>,
    ) -> Result<(), LiminalError> {
        beam::unlink_subscriber(context, pid)?;
        lock(&self.subscribers)?.retain(|subscriber| subscriber.pid() != pid);
        Ok(())
    }

    fn apply_evolve(
        &self,
        name: String,
        field_schema: Value,
        default: Value,
    ) -> Result<SchemaId, SchemaValidationError> {
        let mut schema =
            self.schema
                .lock()
                .map_err(|error| SchemaValidationError::InvalidSchema {
                    message: format!("channel schema lock poisoned: {error}"),
                })?;
        let evolved = schema.evolve_add_field(name, field_schema, default)?;
        let schema_id = evolved.id();
        *schema = evolved;
        drop(schema);
        Ok(schema_id)
    }

    fn apply_schema_id(&self) -> Result<SchemaId, LiminalError> {
        Ok(lock(&self.schema)?.id())
    }

    fn apply_list_subscribers(&self) -> Result<SubscriberSummary, LiminalError> {
        Ok(lock(&self.subscribers)?
            .iter()
            .map(SubscriberRegistration::pid)
            .collect())
    }

    /// Closes the channel and stops the actor process.
    ///
    /// Subscriber-process lifetime is deliberately owned by the host-side
    /// [`crate::channel::SubscriptionHandle`] (its `Drop` terminates the beamr
    /// subscriber process), NOT by the channel actor. This is the correct model:
    /// a subscriber may outlive any single actor incarnation (it survives an
    /// actor crash + restart, which is precisely what MAJOR-1's re-link path
    /// depends on — the surviving subscriber pids are the ones `apply_boot`
    /// re-links to). If `close` terminated the subscriber processes here it would
    /// (a) race the handle's own `Drop`-driven terminate and (b) destroy
    /// processes the actor does not own. So `close` only releases the actor's
    /// *references* to its subscribers (dropping the inbox `Arc`s and predicates);
    /// each subscriber process is reclaimed when its owning handle drops. The
    /// trapped-EXIT pruning path means the actor never leaks a stale link either:
    /// a subscriber that dies first is removed from the fan-out before close.
    fn apply_close(&self) -> Result<(), LiminalError> {
        *lock(&self.closed)? = true;
        lock(&self.subscribers)?.clear();
        Ok(())
    }

    /// Handle a trapped EXIT from a dead subscriber: drop it from the fan-out
    /// list. This is the structural unsubscribe-on-death path (R2).
    pub(crate) fn handle_subscriber_exit(&self, pid: u64) -> Result<(), LiminalError> {
        lock(&self.subscribers)?.retain(|subscriber| subscriber.pid() != pid);
        Ok(())
    }
}

fn schema_mismatch(error: &SchemaValidationError) -> LiminalError {
    LiminalError::SchemaMismatch {
        message: error.to_string(),
    }
}

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