palladium_actor/

actor.rs

use crate::bridge::{CachedSendFn, RuntimeBridge};
use crate::envelope::Envelope;
use crate::errors::{ActorError, SendError};
use crate::message::MessagePayload;
use crate::path::{ActorPath, AddrHash};
use crate::policy::NamespacePolicy;
use crate::spec::ChildSpec;
use std::collections::HashMap;
use std::fmt;
use std::sync::Arc;
use std::time::Duration;

// ── AddrCache ─────────────────────────────────────────────────────────────────

/// Per-actor cache of direct `MailboxSender` wrappers for frequently-contacted
/// destinations.
///
/// Populated on the first successful `send_raw` to a local actor; evicted when
/// delivery returns `SendError::ActorStopped`.  Cloned contexts receive a fresh
/// empty cache so parent/child actors don't share stale entries.
#[derive(Default)]
pub(crate) struct AddrCache {
    entries: HashMap<AddrHash, CachedSendFn>,
}

impl Clone for AddrCache {
    fn clone(&self) -> Self {
        Self::default()
    }
}

impl AddrCache {
    pub(crate) fn get(&self, addr: AddrHash) -> Option<&CachedSendFn> {
        self.entries.get(&addr)
    }

    pub(crate) fn insert(&mut self, addr: AddrHash, f: CachedSendFn) {
        self.entries.insert(addr, f);
    }

    pub(crate) fn remove(&mut self, addr: AddrHash) {
        self.entries.remove(&addr);
    }
}

// ── StopReason ────────────────────────────────────────────────────────────────

/// The reason supplied to [`Actor::on_stop`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StopReason {
    /// Actor exited cleanly under normal conditions.
    Normal,
    /// Stop was requested explicitly (e.g., `ActorContext::stop`).
    Requested,
    /// Stop was initiated by the actor's supervisor.
    Supervisor,
    /// Actor was stopped because its parent stopped (REQ-004).
    Hierarchical,
    /// Engine is shutting down.
    Shutdown,
    /// Actor was force-killed (e.g., `ShutdownPolicy::Brutal`).
    Killed,
    /// Actor failed with an error that caused it to stop.
    Error(ActorError),
}

// ── Actor ─────────────────────────────────────────────────────────────────────

/// The core actor trait.
///
/// Implementations provide lifecycle hooks.  The runtime calls `on_start`
/// before the first `on_message`, and `on_stop` after the last.
pub trait Actor<R: crate::bridge::Reactor>: Send + 'static {
    fn on_start(&mut self, _ctx: &mut ActorContext<R>) -> Result<(), ActorError> {
        Ok(())
    }

    fn on_message(
        &mut self,
        _ctx: &mut ActorContext<R>,
        _envelope: &Envelope,
        _payload: MessagePayload,
    ) -> Result<(), ActorError>;

    fn on_stop(&mut self, _ctx: &mut ActorContext<R>, _reason: StopReason) {}
}

// ── ActorContext ──────────────────────────────────────────────────────────────

/// Runtime handle provided to each lifecycle hook.
///
/// `bridge` is `None` in test/stub contexts (Phase 1).  `pd-runtime` injects
/// a real bridge via `ActorContext::with_bridge` when spawning actors so that
/// `spawn`, `stop`, `send_raw`, and `send_after` are fully functional.
#[derive(Clone)]
pub struct ActorContext<R: crate::bridge::Reactor> {
    pub(crate) path: ActorPath,
    pub(crate) addr_hash: AddrHash,
    pub(crate) bridge: Option<Arc<dyn RuntimeBridge<R>>>,
    /// Optional namespace policy enforced on every `send_raw` call.
    ///
    /// `None` means no enforcement (test/stub contexts; actors spawned without
    /// a ChildSpec).  When `Some`, destinations whose path violates the policy
    /// cause `send_raw` to return `Err(SendError::PolicyViolation)`.
    pub(crate) namespace_policy: Option<NamespacePolicy>,
    /// Per-actor cache of direct mailbox senders for hot destinations.
    ///
    /// Bypasses `TransportRegistry::try_route` (DashMap lookup) for locally
    /// known actors.  See [`AddrCache`] for eviction rules.
    pub(crate) addr_cache: AddrCache,
    pub(crate) _phantom: std::marker::PhantomData<R>,
}

impl<R: crate::bridge::Reactor> fmt::Debug for ActorContext<R> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ActorContext")
            .field("path", &self.path)
            .field("addr_hash", &self.addr_hash)
            .field("bridge", &self.bridge.as_ref().map(|_| "<bridge>"))
            .finish()
    }
}

impl<R: crate::bridge::Reactor> ActorContext<R> {
    /// Create a context without a runtime bridge (Phase 1 / test use).
    pub fn new(path: ActorPath, addr_hash: AddrHash) -> Self {
        Self {
            path,
            addr_hash,
            bridge: None,
            namespace_policy: None,
            addr_cache: AddrCache::default(),
            _phantom: std::marker::PhantomData,
        }
    }

    /// Create a context with a runtime bridge.  Called by `pd-runtime` at
    /// spawn time.
    pub fn with_bridge(
        path: ActorPath,
        addr_hash: AddrHash,
        bridge: Arc<dyn RuntimeBridge<R>>,
    ) -> Self {
        Self {
            path,
            addr_hash,
            bridge: Some(bridge),
            namespace_policy: None,
            addr_cache: AddrCache::default(),
            _phantom: std::marker::PhantomData,
        }
    }

    /// Builder-style method to attach a namespace policy.
    ///
    /// Called by the supervisor when spawning actors from a `ChildSpec`.
    pub fn with_namespace_policy(mut self, policy: NamespacePolicy) -> Self {
        self.namespace_policy = Some(policy);
        self
    }

    pub fn path(&self) -> &ActorPath {
        &self.path
    }

    pub fn addr_hash(&self) -> AddrHash {
        self.addr_hash
    }

    /// Spawn a child actor under this actor's supervision tree.
    ///
    /// Returns the child's `AddrHash` (derived immediately from path);
    /// the runtime processes the actual spawn asynchronously.
    ///
    /// # Panics
    /// Panics when called without a runtime bridge (Phase 1 / stub contexts).
    pub fn spawn(&mut self, spec: ChildSpec<R>) -> AddrHash {
        self.bridge
            .as_ref()
            .expect("ActorContext::spawn requires a runtime bridge")
            .spawn_child(spec, &self.path)
    }

    /// Send a pre-built envelope through the local transport layer.
    ///
    /// If a namespace policy is set, the destination is resolved to an
    /// `ActorPath` and checked against the policy.  Unknown destinations
    /// (hash not found in the registry) are allowed through — the runtime
    /// will deliver or drop as appropriate.
    ///
    /// **Fast path**: On the second and subsequent sends to the same local
    /// actor, the cached `CachedSendFn` is used directly — bypassing the
    /// global `TransportRegistry` DashMap lookup.  The cache is populated on
    /// the first successful delivery and evicted on `ActorStopped`.
    ///
    /// Response envelopes (`FLAG_RESPONSE` set) always bypass the cache and go
    /// through `RuntimeBridge::route` so the `ResponseRegistry` can complete
    /// pending `ask()` futures.
    ///
    /// Returns `Err(SendError::PolicyViolation)` if the destination path
    /// is not permitted by the current namespace policy.
    /// Returns `Err(SendError::ActorStopped)` if called without a bridge.
    pub fn send_raw(
        &mut self,
        envelope: Envelope,
        payload: MessagePayload,
    ) -> Result<(), SendError> {
        // Policy check before routing.
        if let (Some(policy), Some(bridge)) = (&self.namespace_policy, &self.bridge) {
            if let Some(dest_path) = bridge.resolve_addr(envelope.destination) {
                if !policy.allows(&dest_path) {
                    return Err(SendError::PolicyViolation);
                }
            }
        }

        // Response envelopes complete a pending ask() — must go through route()
        // so the ResponseRegistry sees them, not the destination's mailbox.
        if envelope.is_response() {
            return match &self.bridge {
                Some(b) => b.route(envelope, payload),
                None => Err(SendError::ActorStopped),
            };
        }

        let dest = envelope.destination;

        // Fast path: cached direct sender (no DashMap lookup).
        if let Some(cached) = self.addr_cache.get(dest).cloned() {
            let result = cached(envelope, payload);
            if matches!(result, Err(SendError::ActorStopped)) {
                self.addr_cache.remove(dest);
            }
            return result;
        }

        // Cache miss: try to get a direct sender from the bridge.
        let cached_opt = self.bridge.as_ref().and_then(|b| b.cached_sender(dest));
        match cached_opt {
            Some(cached_fn) => {
                let result = cached_fn(envelope, payload);
                if result.is_ok() {
                    self.addr_cache.insert(dest, cached_fn);
                }
                result
            }
            None => {
                // No local mailbox (remote / federated / no bridge).
                match &self.bridge {
                    Some(b) => b.route(envelope, payload),
                    None => Err(SendError::ActorStopped),
                }
            }
        }
    }

    /// Request that this actor stop after the current message handler returns.
    ///
    /// # Panics
    /// Panics when called without a runtime bridge (Phase 1 / stub contexts).
    pub fn stop(&mut self) {
        self.bridge
            .as_ref()
            .expect("ActorContext::stop requires a runtime bridge")
            .stop_self();
    }

    /// Access the reactor for spawning, time, and async primitives.
    ///
    /// # Panics
    /// Panics when called without a runtime bridge (Phase 1 / stub contexts).
    pub fn reactor(&self) -> &R {
        self.bridge
            .as_ref()
            .expect("ActorContext::reactor requires a runtime bridge")
            .reactor()
    }

    /// Return the current monotonic time.
    ///
    /// Phase 3 routes this through the `Reactor` trait for deterministic
    /// simulation.
    pub fn now(&self) -> std::time::Instant {
        match &self.bridge {
            Some(b) => b.now(),
            None => std::time::Instant::now(),
        }
    }

    /// Schedule a message to be delivered after `delay`.
    ///
    /// # Panics
    /// Panics when called without a runtime bridge.
    pub fn send_after(&self, delay: Duration, envelope: Envelope, payload: MessagePayload) {
        self.bridge
            .as_ref()
            .expect("ActorContext::send_after requires a runtime bridge")
            .send_after(delay, envelope, payload);
    }

    /// Resolve an actor's current address by its path.
    ///
    /// Returns `None` if the actor is not currently running or if no bridge
    /// is available (test/stub contexts).
    pub fn lookup_path(&self, path: &ActorPath) -> Option<AddrHash> {
        self.bridge.as_ref()?.lookup_path(path)
    }

    /// Return a cached direct sender for `addr`, populating the cache on first call.
    ///
    /// Checks `addr_cache` first; on a miss, delegates to the bridge's
    /// `cached_sender()` and inserts the result.  Returns `None` when the
    /// destination is not local (remote / federated / no bridge).
    ///
    /// Used by `FfiBridge::cached_sender` to wire the fresh FFI context's
    /// addr_cache through to the real `RuntimeBridgeImpl`.
    pub fn cached_sender_for(&mut self, addr: AddrHash) -> Option<CachedSendFn> {
        if let Some(cached) = self.addr_cache.get(addr) {
            return Some(cached.clone());
        }
        let cached_fn = self.bridge.as_ref()?.cached_sender(addr)?;
        self.addr_cache.insert(addr, cached_fn.clone());
        Some(cached_fn)
    }
}