frame-core 0.2.0

//! The composed component runtime and its opaque support-module handles.

use std::sync::Arc;
use std::time::Duration;

use beamr::atom::Atom;
use beamr::module::ModuleRegistry;
use beamr::namespace::NamespaceId;
use beamr::scheduler::{Scheduler, SchedulerConfig, SchedulerServices};

use crate::composition::{compose_scheduler, deferred_erlang_imports, resolve_atom};
use crate::registry::ComponentRegistry;
use crate::supervision::LifecycleConfig;

use super::error::RuntimeError;

/// Stated-by-the-application runtime policy. No field has a default: frame
/// supplies no lifecycle values, so the embedding application declares each
/// one explicitly.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct RuntimePolicy {
    /// Scheduler worker threads for the component tree. Must be positive —
    /// the wrapper refuses zero rather than letting the runtime silently
    /// substitute the machine's parallelism.
    pub scheduler_threads: usize,
    /// Deadline bounding each spawn, mailbox round-trip, or tombstone
    /// observation. Must be nonzero.
    pub operation_timeout: Duration,
}

/// Opaque, single-use handle to a loaded support module.
///
/// Returned by [`ComponentRuntime::load_support_module`] and consumed by
/// [`ComponentRuntime::unload_support_module`]; because it is neither `Clone`
/// nor `Copy`, a double unload is unrepresentable.
#[derive(Debug)]
#[must_use = "an unconsumed handle leaves its module loaded; unload it through the runtime"]
pub struct SupportModuleHandle {
    module: Atom,
    name: String,
}

/// The composed component runtime: scheduler + registry, correctly assembled.
///
/// Composition happens once, in [`ComponentRuntime::compose`], with the
/// built-in-function registry populated before any bytecode can load (see the
/// [module docs](crate::runtime) for the full composition story). The
/// scheduler itself stays private: the host's authority surface is
/// [`ComponentRuntime::registry`], support modules travel as opaque handles,
/// and teardown is the residue-checked [`ComponentRuntime::shutdown`].
pub struct ComponentRuntime {
    scheduler: Arc<Scheduler>,
    registry: ComponentRegistry,
}

impl ComponentRuntime {
    /// Composes the runtime with its built-in-function registry populated
    /// before any bytecode can load (the composition
    /// [`crate::composition::compose_scheduler`] performs, absorbed).
    ///
    /// # Errors
    ///
    /// Refuses a zero thread count or zero operation timeout with typed
    /// policy errors, and propagates every typed scheduler-composition
    /// failure.
    pub fn compose(policy: RuntimePolicy) -> Result<Self, RuntimeError> {
        if policy.scheduler_threads == 0 {
            return Err(RuntimeError::ZeroSchedulerThreads);
        }
        if policy.operation_timeout.is_zero() {
            return Err(RuntimeError::ZeroOperationTimeout);
        }
        let scheduler = Arc::new(
            compose_scheduler(
                SchedulerConfig {
                    thread_count: Some(policy.scheduler_threads),
                    ..SchedulerConfig::default()
                },
                SchedulerServices::minimal(),
                Arc::new(ModuleRegistry::new()),
            )
            .map_err(|source| RuntimeError::Composition { source })?,
        );
        let registry = ComponentRegistry::new(
            Arc::clone(&scheduler),
            LifecycleConfig {
                operation_timeout: policy.operation_timeout,
            },
        );
        Ok(Self {
            scheduler,
            registry,
        })
    }

    /// Borrows the host's authority surface (register, grant, start, probe,
    /// stop — the unchanged [`ComponentRegistry`] API).
    #[must_use]
    pub const fn registry(&self) -> &ComponentRegistry {
        &self.registry
    }

    /// Loads a support module (e.g. a component's FFI bytecode) and returns
    /// an opaque handle for ordered unload.
    ///
    /// The freshly committed module passes the same deferred-`erlang:*` wall
    /// the registry applies to component bytecode: a module whose import
    /// table still defers a built-in would die at first dispatch, so it is
    /// refused — and removed again — up front.
    ///
    /// # Errors
    ///
    /// Returns typed failures for loader refusal, a module absent immediately
    /// after load, or deferred `erlang:*` imports.
    pub fn load_support_module(
        &self,
        bytecode: &[u8],
    ) -> Result<SupportModuleHandle, RuntimeError> {
        let loaded = self.scheduler.hot_load_module(bytecode).map_err(|error| {
            RuntimeError::SupportModuleLoad {
                detail: error.to_string(),
            }
        })?;
        let module = loaded.module_name;
        let name = resolve_atom(&self.scheduler, module);
        let Some(deferred) = deferred_erlang_imports(&self.scheduler, module) else {
            return Err(RuntimeError::SupportModuleAbsent { module: name });
        };
        if !deferred.is_empty() {
            // The just-loaded module is unusable; take it back out before
            // refusing so the refusal does not leak a loaded generation. A
            // cleanup failure is logged loudly but never masks the refusal
            // itself.
            if !self.scheduler.delete_module(module) {
                tracing::error!(
                    module = %name,
                    "failed to delete support module after deferred-BIF refusal"
                );
            }
            return Err(RuntimeError::SupportModuleDeferredBifImports {
                module: name,
                imports: deferred.join(", "),
            });
        }
        Ok(SupportModuleHandle { module, name })
    }

    /// Ordered unload with verification: purge any retained old generation,
    /// delete the current one, then verify the module is actually gone.
    ///
    /// # Errors
    ///
    /// Returns typed failures for an unsafe purge, a failed delete (a stale
    /// handle to an already-unloaded module lands here), or post-delete
    /// lookup residue.
    pub fn unload_support_module(&self, handle: SupportModuleHandle) -> Result<(), RuntimeError> {
        let SupportModuleHandle { module, name } = handle;
        if self.scheduler.check_old_code(module) {
            self.scheduler.purge_module(module).map_err(|error| {
                RuntimeError::SupportModulePurge {
                    module: name.clone(),
                    detail: error.to_string(),
                }
            })?;
        }
        if !self.scheduler.delete_module(module) {
            return Err(RuntimeError::SupportModuleDelete { module: name });
        }
        if self
            .scheduler
            .lookup_module_in(NamespaceId::DEFAULT, module)
            .is_some()
        {
            return Err(RuntimeError::SupportModuleStillLoaded { module: name });
        }
        Ok(())
    }

    /// Live process count — the zero-residue check after ordered stop.
    #[must_use]
    pub fn live_process_count(&self) -> usize {
        self.scheduler.process_count()
    }

    /// Final teardown; refuses if the process tree still has residue.
    ///
    /// `Scheduler::shutdown` is never cleanup: every component must have
    /// completed its ordered stop first. A refusal deliberately does NOT stop
    /// the scheduler — live processes are the embedder's unfinished ordered
    /// stop, and reaping them here would hide it.
    ///
    /// # Errors
    ///
    /// Returns the live process count as a typed residue refusal.
    pub fn shutdown(self) -> Result<(), RuntimeError> {
        let residue = self.scheduler.process_count();
        if residue != 0 {
            return Err(RuntimeError::ProcessResidue { count: residue });
        }
        self.scheduler.shutdown();
        Ok(())
    }
}