frame_core/

registry.rs

//! Linearized component registration and lifecycle operations.

use std::collections::{HashMap, HashSet};
use std::num::NonZeroUsize;
use std::sync::{Arc, Mutex};

use beamr::atom::Atom;
use beamr::namespace::NamespaceId;
use beamr::scheduler::Scheduler;

use crate::capability::{
    Capability, CapabilityCheckError, CapabilityChecker, CapabilityMutationError, CapabilityTable,
    CheckVerdict, HostCapabilityFacade,
};
use crate::component::{ComponentId, ComponentMeta};
use crate::error::{FailureReason, RegistryError};
use crate::event::{EventHub, LifecycleState, LifecycleSubscription};
use crate::supervision::{LifecycleConfig, SharedStatus, StatusState, TreeHandle, start_tree};

/// Result of an idempotent stop operation.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StopOutcome {
    /// A live process tree completed the ordered drain.
    Stopped,
    /// No process tree was live, so no transition was needed.
    AlreadyStopped,
}

/// Result of an idempotent remove operation.
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum RemoveOutcome {
    /// Code and registry state were removed after an orderly drain.
    Removed,
    /// Code and registry state were removed after host-forced failed-tree cleanup.
    RemovedAfterForcedCleanup(ForcedCleanupReport),
    /// No registration remained for the identity.
    AlreadyRemoved,
}

/// Registry for component definitions and their native process trees.
pub struct ComponentRegistry {
    scheduler: Arc<Scheduler>,
    config: LifecycleConfig,
    records: Mutex<HashMap<ComponentId, Arc<ComponentRecord>>>,
    capabilities: CapabilityTable,
    events: Arc<EventHub>,
    /// Per-IDENTITY monotonic incarnation history, surviving remove: every
    /// committed entry into Running continues that identity's count, so a
    /// remove + re-register cannot restart it and hand a replacement
    /// incarnation an ordinal some stale handle already holds (F-2a's
    /// held-vs-current rule needs the ordinals distinguishable — the F-5a
    /// R2 red evidence pinned exactly that gap). Keyed per identity rather
    /// than registry-wide so an identity's ordinals depend only on its OWN
    /// lifecycle history — R6's determinism: permuting unrelated
    /// components' start order never changes a published generation.
    ///
    /// Growth bound: one `u64` per distinct component identity EVER
    /// registered in this registry's lifetime — configuration-scale (an
    /// installation's component roster), never traffic-scale (no entry is
    /// added by starts, stops, updates, or messages; re-registering a
    /// known identity reuses its entry). Entries are deliberately never
    /// pruned: dropping one on remove is exactly the reset that re-mints
    /// a stale handle's ordinal, so pruning would reopen the fence.
    incarnation_history: Mutex<HashMap<ComponentId, u64>>,
}

impl ComponentRegistry {
    /// Creates a registry over an embedding-owned scheduler.
    ///
    /// The scheduler is never shut down by the registry. The caller must remove
    /// or stop every component before shutting it down.
    #[must_use]
    pub fn new(scheduler: Arc<Scheduler>, config: LifecycleConfig) -> Self {
        Self {
            scheduler,
            config,
            records: Mutex::new(HashMap::new()),
            capabilities: CapabilityTable::new(),
            events: Arc::new(EventHub::default()),
            incarnation_history: Mutex::new(HashMap::new()),
        }
    }

    /// Adds a bounded lifecycle subscriber.
    ///
    /// Each subscriber drops its own oldest event on overflow and increments
    /// its lag counter; publication never waits for a consumer.
    ///
    /// # Errors
    ///
    /// Returns a synchronization failure if a prior panic poisoned the stream.
    pub fn subscribe(
        &self,
        capacity: NonZeroUsize,
    ) -> Result<LifecycleSubscription, RegistryError> {
        self.events
            .subscribe(capacity)
            .map_err(|_| RegistryError::SynchronizationPoisoned)
    }

    /// Borrows the embedding host's sole grant mutation and inspection facade.
    ///
    /// This Rust host value has no BEAM representation and is never passed to a
    /// component entrypoint; component-facing code receives check-only paths.
    #[must_use]
    pub fn host_capabilities(&self) -> HostCapabilityFacade<'_> {
        HostCapabilityFacade::new(&self.capabilities, Arc::clone(&self.events))
    }

    /// Creates a retained check-only path for one registered component.
    ///
    /// # Errors
    ///
    /// Refuses a missing component or poisoned capability table.
    pub fn capability_checker(
        &self,
        component_id: ComponentId,
    ) -> Result<CapabilityChecker, CapabilityMutationError> {
        self.host_capabilities().checker(component_id)
    }

    /// Performs a fresh capability check for one host-mediated consuming act.
    ///
    /// Denials are returned as typed verdicts and emitted exactly once on the
    /// lifecycle stream. No verdict or authority token is retained.
    ///
    /// # Errors
    ///
    /// Refuses a missing component or poisoned table/event synchronization.
    pub fn check_capability(
        &self,
        component_id: ComponentId,
        capability: &Capability,
    ) -> Result<CheckVerdict, CapabilityCheckError> {
        let checker = self
            .capability_checker(component_id)
            .map_err(|error| match error {
                CapabilityMutationError::NotRegistered { component_id } => {
                    CapabilityCheckError::NotRegistered { component_id }
                }
                CapabilityMutationError::UndeclaredNeed { .. }
                | CapabilityMutationError::AlreadyGranted { .. }
                | CapabilityMutationError::SynchronizationPoisoned => {
                    CapabilityCheckError::SynchronizationPoisoned
                }
            })?;
        checker.check(capability)
    }

    /// Validates and atomically registers metadata with its BEAM bytecode.
    ///
    /// # Errors
    ///
    /// Refuses duplicate identities, missing/cyclic dependency edges,
    /// capability collisions, duplicate child keys, and absent/invalid policy.
    pub fn register(&self, meta: ComponentMeta, bytecode: Vec<u8>) -> Result<(), RegistryError> {
        let mut records = self
            .records
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        validate_registration(&records, &meta, self.config.max_fragment_bytes)?;
        let id = meta.id;
        let needs = meta.needs.clone();
        let record = Arc::new(ComponentRecord::new(meta, bytecode));
        records.insert(id, record);
        if self.capabilities.register_component(id, needs).is_err() {
            records.remove(&id);
            return Err(RegistryError::SynchronizationPoisoned);
        }
        if self
            .events
            .publish_transition(id, None, LifecycleState::Registered)
            .is_err()
        {
            self.capabilities
                .remove_component(id)
                .map_err(|_| RegistryError::SynchronizationPoisoned)?;
            records.remove(&id);
            return Err(RegistryError::SynchronizationPoisoned);
        }
        Ok(())
    }

    /// Starts one component after checking every dependency is Running.
    ///
    /// Successful return means every child answered its declared mailbox probe.
    ///
    /// # Errors
    ///
    /// Returns typed state, dependency, loader, spawn, liveness, or protocol
    /// failures. A runtime failure leaves an observable Failed registration.
    pub fn start(&self, id: ComponentId) -> Result<(), RegistryError> {
        let record = self.record(id)?;
        claim_operation(&record, id, Operation::Start)?;
        if let Err(error) = self.check_running_dependencies(&record.meta) {
            clear_operation(&record)?;
            return Err(error);
        }
        transition(&record, &self.events, id, LifecycleState::Starting, None)?;
        let load_result = {
            let bytecode = record
                .bytecode
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?;
            self.scheduler.hot_load_module(&bytecode)
        };
        let loaded = match load_result {
            Ok(loaded) => loaded,
            Err(error) => {
                let reason = FailureReason {
                    child: None,
                    tombstone: None,
                    detail: error.to_string(),
                };
                transition(
                    &record,
                    &self.events,
                    id,
                    LifecycleState::Failed,
                    Some(reason),
                )?;
                clear_operation(&record)?;
                return Err(RegistryError::ModuleLoad {
                    id,
                    detail: error.to_string(),
                });
            }
        };
        record
            .modules
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .push(loaded.module_name);
        if let Err(error) = self.refuse_deferred_bif_imports(id, loaded.module_name) {
            transition(
                &record,
                &self.events,
                id,
                LifecycleState::Failed,
                Some(failure_from_error(&error)),
            )?;
            clear_operation(&record)?;
            return Err(error);
        }
        let policy = record
            .meta
            .supervision
            .clone()
            .ok_or(RegistryError::UndeclaredSupervision { id })?;
        match start_tree(
            id,
            Arc::clone(&self.scheduler),
            &record.meta.children,
            policy,
            self.config,
            &record.status,
            Arc::clone(&self.events),
        ) {
            Ok(tree) => {
                *record
                    .tree
                    .lock()
                    .map_err(|_| RegistryError::SynchronizationPoisoned)? = Some(tree);
                self.prepare_running_incarnation(id, &record)?;
                transition(&record, &self.events, id, LifecycleState::Running, None)?;
                clear_operation(&record)
            }
            Err(error) => {
                let reason = failure_from_error(&error);
                transition(
                    &record,
                    &self.events,
                    id,
                    LifecycleState::Failed,
                    Some(reason),
                )?;
                clear_operation(&record)?;
                Err(error)
            }
        }
    }

    /// Stops a running component by draining children in declaration order and
    /// then observing the host supervisor's Normal tombstone.
    ///
    /// # Errors
    ///
    /// Returns typed in-flight, protocol, timeout, or non-normal exit failures.
    pub fn stop(&self, id: ComponentId) -> Result<StopOutcome, RegistryError> {
        let record = self.record(id)?;
        claim_operation(&record, id, Operation::Stop)?;
        let current = state(&record)?;
        if matches!(
            current,
            LifecycleState::Registered | LifecycleState::Stopped | LifecycleState::Failed
        ) {
            clear_operation(&record)?;
            return Ok(StopOutcome::AlreadyStopped);
        }
        transition(&record, &self.events, id, LifecycleState::Stopping, None)?;
        let tree = record
            .tree
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .take()
            .ok_or(RegistryError::MonitorTerminated { id })?;
        match tree.stop(id) {
            Ok(()) => {
                clear_runtime(&record)?;
                transition(&record, &self.events, id, LifecycleState::Stopped, None)?;
                clear_operation(&record)?;
                Ok(StopOutcome::Stopped)
            }
            Err(error) => {
                transition(
                    &record,
                    &self.events,
                    id,
                    LifecycleState::Failed,
                    Some(failure_from_error(&error)),
                )?;
                clear_operation(&record)?;
                Err(error)
            }
        }
    }

    /// Stops if needed, unloads every loaded generation, and unregisters.
    ///
    /// # Errors
    ///
    /// Refuses concurrent operations, running dependents, abnormal drain, unsafe
    /// purge, failed deletion, or post-delete lookup residue.
    pub fn remove(&self, id: ComponentId) -> Result<RemoveOutcome, RegistryError> {
        let Some(record) = self.optional_record(id)? else {
            return Ok(RemoveOutcome::AlreadyRemoved);
        };
        claim_operation(&record, id, Operation::Remove)?;
        if let Err(error) = self.check_running_dependents(id) {
            clear_operation(&record)?;
            return Err(error);
        }
        let current = state(&record)?;
        let mut forced_cleanup = None;
        if current == LifecycleState::Running {
            transition(&record, &self.events, id, LifecycleState::Stopping, None)?;
            let tree = record
                .tree
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?
                .take()
                .ok_or(RegistryError::MonitorTerminated { id })?;
            match tree.stop(id) {
                Ok(()) => {
                    clear_runtime(&record)?;
                    transition(&record, &self.events, id, LifecycleState::Stopped, None)?;
                }
                Err(error) => {
                    transition(
                        &record,
                        &self.events,
                        id,
                        LifecycleState::Failed,
                        Some(failure_from_error(&error)),
                    )?;
                    clear_operation(&record)?;
                    return Err(error);
                }
            }
        } else if current == LifecycleState::Failed {
            forced_cleanup = self.recover_failed_tree(id, &record)?;
        }
        self.unload_modules(id, &record)?;
        let from = state(&record)?;
        self.capabilities
            .remove_component(id)
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        self.events
            .publish_transition(id, Some(from), LifecycleState::Removed)
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        self.records
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .remove(&id);
        Ok(forced_cleanup.map_or(
            RemoveOutcome::Removed,
            RemoveOutcome::RemovedAfterForcedCleanup,
        ))
    }

    /// Replaces the registered bytecode a component's NEXT start hot-loads
    /// — the F-7b hot-reload staging seam. The component identity, its
    /// registered metadata, grants, and durable branch are untouched: the
    /// branch outlives the bytes.
    ///
    /// Only legal while the component is Registered or Stopped: the
    /// upgrade barrier drains and stops the old incarnation first, so new
    /// bytes never stage under a live tree.
    ///
    /// Staging PURGES the retained old module generation (the C5 law: the
    /// wall is retention-based, so drain alone never re-opens loading —
    /// without this purge the third stop→stage→start lap of one
    /// registration refuses `OldCodeStillRunning`). The purge is the SAFE
    /// one: after an ordered stop nothing references old code, and a
    /// `StillReferenced` refusal here is the typed drain-failure signal,
    /// surfaced, never forced over.
    ///
    /// # Errors
    ///
    /// Refuses concurrent operations and non-stopped states; surfaces a
    /// purge refusal as the drain failure it is. All typed.
    pub fn upgrade_bytecode(
        &self,
        id: ComponentId,
        bytecode: Vec<u8>,
    ) -> Result<(), RegistryError> {
        let record = self.record(id)?;
        claim_operation(&record, id, Operation::Upgrade)?;
        let current = state(&record)?;
        if !matches!(
            current,
            LifecycleState::Registered | LifecycleState::Stopped
        ) {
            clear_operation(&record)?;
            return Err(RegistryError::UpgradeRequiresStopped { id, state: current });
        }
        if let Err(error) = self.purge_retained_old(id, &record) {
            clear_operation(&record)?;
            return Err(error);
        }
        *record
            .bytecode
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)? = bytecode;
        clear_operation(&record)
    }

    /// Purges the retained old generation of every module this component
    /// has loaded, keeping the current generation live for rollback reads.
    /// Distinct from [`Self::unload_modules`]: staging keeps the
    /// registration and the current code; only the retained OLD version
    /// goes.
    fn purge_retained_old(
        &self,
        id: ComponentId,
        record: &ComponentRecord,
    ) -> Result<(), RegistryError> {
        let modules = record
            .modules
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .clone();
        for module in unique_modules(modules) {
            if self.scheduler.check_old_code(module) {
                self.scheduler.purge_module(module).map_err(|error| {
                    RegistryError::ModulePurge {
                        id,
                        module: self.module_name(module),
                        detail: error.to_string(),
                    }
                })?;
            }
        }
        Ok(())
    }

    /// The component's live incarnation ordinal — the per-identity mint,
    /// strictly increasing across every entry into Running (the
    /// registry-side freshness witness of the F-7b dev status report).
    ///
    /// # Errors
    ///
    /// Refuses unknown identities and non-running components, typed.
    pub fn component_incarnation(&self, id: ComponentId) -> Result<u64, RegistryError> {
        let record = self.record(id)?;
        require_running(&record, id)?;
        Ok(record
            .incarnation
            .load(std::sync::atomic::Ordering::Acquire))
    }

    /// The current VM code generation of the component's own module — the
    /// VM-side freshness witness of the F-7b dev status report (beamr's
    /// per-name monotonic counter, surfaced not interpreted).
    ///
    /// # Errors
    ///
    /// Refuses unknown identities, components that never loaded, and a
    /// module absent from the VM, all typed.
    pub fn component_module_generation(&self, id: ComponentId) -> Result<u64, RegistryError> {
        let record = self.record(id)?;
        let module = record
            .modules
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .last()
            .copied()
            .ok_or_else(|| RegistryError::ModuleLoad {
                id,
                detail: "component has never loaded a module".to_owned(),
            })?;
        self.scheduler
            .lookup_module_in(NamespaceId::DEFAULT, module)
            .map(|loaded| loaded.generation())
            .ok_or_else(|| RegistryError::ModuleLoad {
                id,
                detail: format!("module {} is absent from the VM", self.module_name(module)),
            })
    }

    /// Sends an integer mailbox command to a named live child.
    ///
    /// # Errors
    ///
    /// Refuses non-running components, unknown children, and monitor failures.
    pub fn send_child(
        &self,
        id: ComponentId,
        child: impl Into<String>,
        message: i64,
    ) -> Result<(), RegistryError> {
        let record = self.record(id)?;
        require_running(&record, id)?;
        record
            .tree
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .as_ref()
            .ok_or(RegistryError::MonitorTerminated { id })?
            .send(id, child.into(), message)
    }

    /// Runs the child's declared liveness mailbox round-trip and returns its
    /// integer state reply.
    ///
    /// # Errors
    ///
    /// Refuses non-running components, unknown children, timeout, or failure.
    pub fn probe_child(
        &self,
        id: ComponentId,
        child: impl Into<String>,
    ) -> Result<i64, RegistryError> {
        let record = self.record(id)?;
        require_running(&record, id)?;
        record
            .tree
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?
            .as_ref()
            .ok_or(RegistryError::MonitorTerminated { id })?
            .probe(id, child.into())
    }

    fn record(&self, id: ComponentId) -> Result<Arc<ComponentRecord>, RegistryError> {
        self.optional_record(id)?
            .ok_or(RegistryError::NotRegistered { id })
    }

    fn optional_record(
        &self,
        id: ComponentId,
    ) -> Result<Option<Arc<ComponentRecord>>, RegistryError> {
        self.records
            .lock()
            .map(|records| records.get(&id).cloned())
            .map_err(|_| RegistryError::SynchronizationPoisoned)
    }

    fn check_running_dependencies(&self, meta: &ComponentMeta) -> Result<(), RegistryError> {
        for required in &meta.requires {
            let record = self.record(*required)?;
            let required_state = state(&record)?;
            if required_state != LifecycleState::Running {
                return Err(RegistryError::DependencyNotRunning {
                    component: meta.id,
                    required: *required,
                    state: required_state,
                });
            }
        }
        Ok(())
    }

    fn check_running_dependents(&self, id: ComponentId) -> Result<(), RegistryError> {
        let records = self
            .records
            .lock()
            .map_err(|_| RegistryError::SynchronizationPoisoned)?;
        for (dependent, record) in records.iter() {
            if record.meta.requires.contains(&id) && state(record)? == LifecycleState::Running {
                return Err(RegistryError::RunningDependent {
                    dependent: *dependent,
                    required: id,
                });
            }
        }
        Ok(())
    }

    fn unload_modules(
        &self,
        id: ComponentId,
        record: &ComponentRecord,
    ) -> Result<(), RegistryError> {
        let modules = std::mem::take(
            &mut *record
                .modules
                .lock()
                .map_err(|_| RegistryError::SynchronizationPoisoned)?,
        );
        for module in unique_modules(modules) {
            let name = self.module_name(module);
            if self.scheduler.check_old_code(module) {
                self.scheduler.purge_module(module).map_err(|error| {
                    RegistryError::ModulePurge {
                        id,
                        module: name.clone(),
                        detail: error.to_string(),
                    }
                })?;
            }
            if !self.scheduler.delete_module(module) {
                return Err(RegistryError::ModuleDelete { id, module: name });
            }
            if self
                .scheduler
                .lookup_module_in(NamespaceId::DEFAULT, module)
                .is_some()
            {
                return Err(RegistryError::ModuleStillLoaded { id, module: name });
            }
        }
        Ok(())
    }

    fn module_name(&self, module: Atom) -> String {
        crate::composition::resolve_atom(&self.scheduler, module)
    }

    /// Refuses a freshly hot-loaded module whose committed import table still
    /// defers any `erlang:*` entry (the scan itself, shared with the runtime
    /// wrapper's support-module path, is
    /// [`crate::composition::deferred_erlang_imports`] — see its docs for why
    /// a deferred built-in is process-fatal at first dispatch).
    fn refuse_deferred_bif_imports(
        &self,
        id: ComponentId,
        module: Atom,
    ) -> Result<(), RegistryError> {
        let Some(deferred) = crate::composition::deferred_erlang_imports(&self.scheduler, module)
        else {
            return Err(RegistryError::ModuleLoad {
                id,
                detail: "committed module was absent immediately after hot load".to_owned(),
            });
        };
        if deferred.is_empty() {
            return Ok(());
        }
        Err(RegistryError::DeferredBifImports {
            id,
            module: self.module_name(module),
            imports: deferred.join(", "),
        })
    }
}

mod fragments;
mod query;
mod recovery;
mod support;

pub use recovery::{ChildCleanup, ForcedCleanupReport, ProcessCleanup};
use support::{
    ComponentRecord, Operation, claim_operation, clear_operation, clear_runtime,
    failure_from_error, require_running, state, transition, unique_modules, validate_registration,
};