acktor 1.0.14

//! Traits and type definitions for actors.
//!
//! This module defines the [`Actor`] trait and its [`ActorContext`] companion, which together
//! describe how an actor is started, processes messages, and shuts down.
//!

use std::any::Any;
use std::fmt::Display;
use std::panic::{self, AssertUnwindSafe};

use futures_util::FutureExt;
use tracing::{Instrument, Span, debug, error, error_span, field, warn};

#[cfg(feature = "ipc")]
use crate::address::RemoteMailbox;
use crate::address::{Address, Mailbox, Recipient, Sender};
use crate::error::{BoxError, ErrorReport};
use crate::supervisor::SupervisionEvent;
use crate::utils::panic_info_to_string;

mod index;
pub use index::ActorId;

#[cfg(feature = "ipc")]
mod remote;
#[cfg(feature = "ipc")]
#[cfg_attr(docsrs, doc(cfg(feature = "ipc")))]
pub use remote::{RemoteAddressable, RemoteSpawnable};

pub use tokio::task::JoinHandle;

/// State of an actor.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum ActorState {
    Unstarted,
    Starting,
    Running,
    Stopping,
    Stopped,
}

impl TryFrom<u8> for ActorState {
    type Error = ();

    fn try_from(value: u8) -> Result<Self, Self::Error> {
        match value {
            0 => Ok(ActorState::Unstarted),
            1 => Ok(ActorState::Starting),
            2 => Ok(ActorState::Running),
            3 => Ok(ActorState::Stopping),
            4 => Ok(ActorState::Stopped),
            _ => Err(()),
        }
    }
}

/// Return value of [`Actor::stopping`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum Stopping {
    /// The actor could not resume by itself. Stop the actor.
    Stop,
    /// The actor could resume by itself.
    Continue,
}

/// An actor.
pub trait Actor: Sized + Send + 'static {
    /// The execution context type for this actor.
    type Context: ActorContext<Self>;
    // NOTE: this bound is chosen to be compatible with `StdError`, `Box<dyn Error>` and
    // `anyhow::Error`
    /// The error type returned by lifecycle hooks and message handlers.
    type Error: Into<BoxError> + Display + Send + 'static;

    /// Invoked before an actor is spawned into the tokio runtime. The actor should be in
    /// [`Unstarted`][ActorState::Unstarted] state.
    ///
    /// This method is used to perform initialization tasks or spawn child actors. In the default
    /// [`Context`][crate::context::Context] implementation, it is not spawned into the tokio
    /// runtime and it is outside of the processing loop. Thus it will be invoked only once
    /// synchronously. The actor will enter the [`Starting`][ActorState::Starting] state after
    /// this method returns.
    ///
    /// Panics in this method propagate to the caller of [`start`][Actor::start] or
    /// [`create`][Actor::create].
    #[allow(unused_variables)]
    fn pre_start(&mut self, ctx: &mut Self::Context) -> Result<(), Self::Error> {
        Ok(())
    }

    /// Invoked after an actor is spawned into the tokio runtime. The actor should be in
    /// [`Starting`][ActorState::Starting] state.
    ///
    /// This method is used to perform additional initialization. In the default
    /// [`Context`][crate::context::Context] implementation, it is spawned into the tokio runtime
    /// and it is outside of the processing loop, which means it will be invoked once and only
    /// once, asynchronously. The actor will enter the [`Running`][ActorState::Running] state
    /// after this method returns.
    ///
    /// Panics in this method terminates the actor immediately and will be notified to the
    /// supervisor if there is one.
    #[allow(unused_variables)]
    fn post_start(
        &mut self,
        ctx: &mut Self::Context,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        std::future::ready(Ok(()))
    }

    /// Invoked when an actor is being stopped. The actor should be in
    /// [`Stopping`][ActorState::Stopping] state.
    ///
    /// This method is used to make decisions about whether to stop or to restart the actor.
    #[allow(unused_variables)]
    fn stopping(
        &mut self,
        ctx: &mut Self::Context,
    ) -> impl Future<Output = Result<Stopping, Self::Error>> + Send {
        std::future::ready(Ok(Stopping::Stop))
    }

    /// Invoked after an actor is stopped. The actor should be in [`Stopped`][ActorState::Stopped]
    /// state.
    ///
    /// This method is used to perform cleanup tasks or spawn new actors.
    #[allow(unused_variables)]
    fn post_stop(
        &mut self,
        ctx: &mut Self::Context,
    ) -> impl Future<Output = Result<(), Self::Error>> + Send {
        std::future::ready(Ok(()))
    }

    /// Starts an actor, returns its address and the join handle.
    fn start<S>(self, label: S) -> Result<(Address<Self>, JoinHandle<()>), Self::Error>
    where
        S: AsRef<str>,
    {
        let ctx = Self::Context::new(label.as_ref().to_string());
        let id = field::display(ctx.index());
        let label = ctx.label();
        let span = error_span!("Actor", id = id, label = label);
        ctx.spawn(self, span)
    }

    /// Creates a new actor, starts it and returns its address and the join handle.
    fn create<S, F>(label: S, f: F) -> Result<(Address<Self>, JoinHandle<()>), Self::Error>
    where
        S: AsRef<str>,
        F: FnOnce(&mut Self::Context) -> Result<Self, Self::Error>,
    {
        let mut ctx = Self::Context::new(label.as_ref().to_string());
        let id = field::display(ctx.index());
        let label = ctx.label();
        let span = error_span!("Actor", id = id, label = label);
        let actor = {
            let _enter = span.enter();
            f(&mut ctx)?
        };
        ctx.spawn(actor, span)
    }

    /// Returns a remote mailbox of this actor if it is a remote addressable actor.
    ///
    /// # Implementation
    ///
    /// **Do not implement this method yourself!** Instead, use the
    /// [`#[remote]`][acktor_derive::remote] attribute macro to annotate the
    /// `impl Actor for MyActor` block, it will generate the proper implementation for you.
    ///
    /// This is a temporary workaround since specialization is not yet stable in Rust.
    #[doc(hidden)]
    #[cfg(feature = "ipc")]
    #[allow(unused_variables)]
    fn remote_mailbox(address: Address<Self>) -> Option<RemoteMailbox> {
        // the default implementation for a remote addressable actor looks like this:
        // ```ignore
        //  Some(address.into())
        // ```
        None
    }
}

/// The execution context of an actor.
///
/// Each actor is associated with a context which manages its lifecycle and communication
/// channels. The actor's associated type [`Context`][Actor::Context] defines the specific context
/// type to use. A context type must implement this trait.
pub trait ActorContext<A>: Sized + Send + 'static
where
    A: Actor<Context = Self>,
{
    // required methods

    /// Constructs a new actor context.
    fn new(label: String) -> Self;

    /// Returns the index of the actor.
    fn index(&self) -> ActorId;

    /// Returns the label of the actor.
    fn label(&self) -> &str;

    /// Returns the address of the actor.
    fn address(&self) -> Address<A>;

    /// Moves the mailbox of the actor out of the context, leaving `None` in its place.
    ///
    /// Typically the address and the mailbox are created together in the constructor of the
    /// context. However, since the [`run`][Self::run] method consumes the mailbox, the context
    /// needs to provide a way to move the mailbox out of itself so that it can be passed into the
    /// [`run`][Self::run] method.
    ///
    /// # Example
    ///
    /// A typical implementation stores the mailbox as an `Option<Mailbox<A>>` field and
    /// delegates to [`Option::take`]:
    ///
    /// ```ignore
    /// struct MyContext<A: Actor<Context = Self>> {
    ///     mailbox: Option<Mailbox<A>>,
    ///     // ... other fields (address, state, etc.)
    /// }
    ///
    /// impl<A: Actor<Context = Self>> ActorContext<A> for MyContext<A> {
    ///     fn take_mailbox(&mut self) -> Option<Mailbox<A>> {
    ///         self.mailbox.take()
    ///     }
    ///
    ///     // ... other trait methods
    /// }
    /// ```
    ///
    /// The first call returns `Some(mailbox)`; subsequent calls return `None`.
    fn take_mailbox(&mut self) -> Option<Mailbox<A>>;

    /// Returns the state of the actor.
    fn state(&self) -> ActorState;

    /// Sets the state of the actor.
    fn set_state(&mut self, state: ActorState);

    /// The message handling loop of the actor.
    ///
    /// Called by [`run`][Self::run] after [`post_start`][Actor::post_start] completes. Runs until
    /// the actor stops, repeatedly pulling envelopes from the `mailbox` and dispatching them to
    /// the appropriate [`Handler`][crate::message::Handler] implementation on `actor`.
    ///
    /// Implementors are responsible for checking [`ActorContext::state`][Self::state] and
    /// honoring [`Actor::stopping`].
    fn run_loop(
        &mut self,
        actor: &mut A,
        mailbox: &mut Mailbox<A>,
    ) -> impl Future<Output = Result<(), A::Error>> + Send;

    // provided methods

    /// Stops the actor.
    ///
    /// This method will switch the actor to the [`Stopping`][ActorState::Stopping] state.
    fn stop(&mut self) {
        self.set_state(ActorState::Stopping);
    }

    /// Terminates the actor.
    ///
    /// This method will switch the actor to the [`Stopped`][ActorState::Stopped] state.
    fn terminate(&mut self) {
        self.set_state(ActorState::Stopped);
    }

    /// Returns a reference to the supervisor of the actor, if any.
    ///
    /// Override the [`supervisor`][ActorContext::supervisor] method and the
    /// [`set_supervisor`][ActorContext::set_supervisor] method to opt-in the supervisor feature.
    fn supervisor(&self) -> Option<&Recipient<SupervisionEvent<A>>> {
        None
    }

    /// Sets a supervisor.
    ///
    /// Override the [`supervisor`][ActorContext::supervisor] method and the
    /// [`set_supervisor`][ActorContext::set_supervisor] method to opt-in the supervisor feature.
    #[allow(unused_variables)]
    fn set_supervisor(&mut self, supervisor: Option<Recipient<SupervisionEvent<A>>>) {}

    /// Notifies the supervisor for an event.
    ///
    /// This method will wait until there is capacity in the mailbox of the supervisor.
    fn notify_supervisor(&mut self, event: SupervisionEvent<A>) -> impl Future<Output = ()> + Send {
        async move {
            if let Some(supervisor) = self.supervisor() {
                let _ = supervisor.do_send(event).await;
            } else {
                match event {
                    SupervisionEvent::Warn(actor, e) => {
                        warn!("Actor {} error: {}", actor.index(), e.into().report());
                    }
                    SupervisionEvent::Terminated(actor, Some(e)) => {
                        error!("Actor {} error: {}", actor.index(), e.into().report());
                    }
                    _ => {}
                }
            }
        }
    }

    /// Notifies the supervisor for an event.
    ///
    /// This method will return immediately if there is no capacity in the mailbox of the
    /// supervisor.
    fn try_notify_supervisor(&mut self, event: SupervisionEvent<A>) {
        if let Some(supervisor) = self.supervisor() {
            let _ = supervisor.try_do_send(event);
        } else {
            match event {
                SupervisionEvent::Warn(actor, e) => {
                    warn!("Actor {} error: {}", actor.index(), e.into().report());
                }
                SupervisionEvent::Terminated(actor, Some(e)) => {
                    error!("Actor {} error: {}", actor.index(), e.into().report());
                }
                _ => {}
            }
        }
    }

    /// Spawns the actor into the tokio runtime and returns its address and the join handle.
    fn spawn(mut self, mut actor: A, span: Span) -> Result<(Address<A>, JoinHandle<()>), A::Error> {
        let address = self.address();

        let mailbox = self.take_mailbox().expect(
            "ActorContext::take_mailbox() returned None on first call to spawn(); \
             custom ActorContext implementations must provide a mailbox in new()",
        );

        let index = self.index();
        #[cfg(feature = "tokio-tracing")]
        let label = self.label().to_string();

        {
            let _enter = span.enter();
            let result = panic::catch_unwind(AssertUnwindSafe(|| actor.pre_start(&mut self)));
            match result {
                Ok(r) => r?,
                Err(info) => {
                    let msg: String = panic_info_to_string(&*info);
                    error!("Actor {} panicked in pre_start: {}", index, msg);
                    panic::resume_unwind(info);
                }
            }
            self.set_state(ActorState::Starting);
        }

        let future = async move {
            match self.run(actor, mailbox).await {
                Ok(Ok(_)) => {
                    self.try_notify_supervisor(SupervisionEvent::Terminated(self.address(), None));
                }
                Ok(Err(e)) => {
                    self.try_notify_supervisor(SupervisionEvent::Terminated(
                        self.address(),
                        Some(e),
                    ));
                }
                Err(e) => {
                    let msg: String = panic_info_to_string(&*e);
                    self.try_notify_supervisor(SupervisionEvent::Panicked(self.address(), msg));
                }
            }

            debug!("Actor {} is stopped", index);
        }
        .instrument(span.or_current())
        .boxed();

        #[cfg(not(feature = "tokio-tracing"))]
        let join_handle = tokio::spawn(future);
        #[cfg(feature = "tokio-tracing")]
        let join_handle = tokio::task::Builder::new()
            .name(&label)
            .spawn(future)
            .unwrap();

        Ok((address, join_handle))
    }

    /// The main lifecycle of the actor.
    ///
    /// This method is invoked by [`spawn`][Self::spawn]. It is responsible for invoking the
    /// [`post_start`][Actor::post_start] and the [`post_stop`][Actor::post_stop] lifecycle
    /// hooks, and running [`run_loop`][Self::run_loop] for handling messages. The default
    /// implementation of [`run_loop`][Self::run_loop] invokes the [`stopping`][Actor::stopping]
    /// lifecycle hook when the actor is about to stop, but users can change this behavior by
    /// overriding it.
    ///
    /// # Return value
    ///
    /// The nested `Result<Result<(), A::Error>, Box<dyn Any + Send>>` encodes three outcomes:
    ///
    /// - `Ok(Ok(()))` — the actor started, ran, and stopped cleanly.
    /// - `Ok(Err(error))` — a lifecycle method returned an [`Actor::Error`] (from
    ///   [`post_start`][Actor::post_start], [`run_loop`][Self::run_loop], or
    ///   [`post_stop`][Actor::post_stop]).
    /// - `Err(Box<dyn Any + Send>>)` — a lifecycle method panicked. The payload is the value
    ///   caught by [`catch_unwind`][panic::catch_unwind]. When
    ///   [`run_loop`][Self::run_loop] panics, [`post_stop`][Actor::post_stop] is skipped since
    ///   the actor's state is not assumed to be safe to observe after a panic.
    ///
    /// If both [`run_loop`][Self::run_loop] and [`post_stop`][Actor::post_stop] return
    /// errors (neither panics), the `run_loop` error is returned and the `post_stop` error
    /// is discarded with a debug log.
    fn run(
        &mut self,
        mut actor: A,
        mut mailbox: Mailbox<A>,
    ) -> impl Future<Output = Result<Result<(), A::Error>, Box<dyn Any + Send>>> + Send {
        async move {
            let result = AssertUnwindSafe(actor.post_start(self))
                .catch_unwind()
                .await
                .inspect_err(|e| {
                    let msg: String = panic_info_to_string(e);
                    error!("Actor {} panicked in post_start: {}", self.index(), msg);
                });

            if !matches!(result, Ok(Ok(()))) {
                self.set_state(ActorState::Stopped);
                return result;
            }

            debug!("Actor {} is started", self.index());
            self.set_state(ActorState::Running);

            let result = AssertUnwindSafe(self.run_loop(&mut actor, &mut mailbox))
                .catch_unwind()
                .await
                .inspect_err(|e| {
                    let msg: String = panic_info_to_string(e);
                    error!("Actor {} panicked in run_loop: {}", self.index(), msg);
                });

            if self.state() != ActorState::Stopped {
                self.set_state(ActorState::Stopped);
            }

            // drop mailbox so any actor holds the address of this actor will not be able to send
            // messages
            drop(mailbox);

            // if the process_loop panicked, post_stop is skipped since the actor's state is not
            // predictable once a panic happens, and we don't want to make it worse by invoking
            // user code in post_stop
            let result = result?;

            let result_post_stop = AssertUnwindSafe(actor.post_stop(self))
                .catch_unwind()
                .await
                .inspect_err(|e| {
                    let msg: String = panic_info_to_string(e);
                    error!("Actor {} panicked in post_stop: {}", self.index(), msg);
                })?;

            match (result, result_post_stop) {
                (Err(e), Err(post_stop_err)) => {
                    debug!(
                        "Actor {} also failed in post_stop: {}",
                        self.index(),
                        post_stop_err,
                    );
                    Ok(Err(e))
                }
                (Err(e), Ok(())) => Ok(Err(e)),
                (Ok(()), result_post_stop) => Ok(result_post_stop),
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use pretty_assertions::assert_eq;

    use super::*;

    #[test]
    fn test_actor_state() {
        assert_eq!(ActorState::try_from(0), Ok(ActorState::Unstarted));
        assert_eq!(ActorState::try_from(1), Ok(ActorState::Starting));
        assert_eq!(ActorState::try_from(2), Ok(ActorState::Running));
        assert_eq!(ActorState::try_from(3), Ok(ActorState::Stopping));
        assert_eq!(ActorState::try_from(4), Ok(ActorState::Stopped));
        assert_eq!(ActorState::try_from(5), Err(()));
    }
}